fithia2/apps/intraday_bt/scripts/diag_orb_breadth.py

"""
V40 Diagnostic: Prior-Day Market Breadth Signal

Hypothesis: On days when broad market breadth was positive the day before an ORB gap-up,
follow-through should be better because the gap is "with the market" rather than isolated.

Features:
  breadth_up_pct    : fraction of universe that closed UP the prior day [0-1]
  breadth_adv_dec   : (advancers - decliners) / total — net breadth [-1 to +1]

This is a DAY-LEVEL signal (same value for all stocks on same day).
Unlike QQQ gap (already in V24) which is a single-stock measure, breadth captures
the DISTRIBUTION of market participation.

Source: daily bars for full midlarge universe, computed from parquet cache.
"""
from __future__ import annotations

import concurrent.futures
import datetime as dt
import json
import os
import sys
from pathlib import Path

import pyarrow.parquet as pq
import yaml

sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../..")))

from zoneinfo import ZoneInfo
from libs.common.time_utils import trading_days_between

_ET = ZoneInfo("America/New_York")
_MKT_OPEN = dt.time(9, 30)
_MKT_CLOSE = dt.time(16, 0)

INTRADAY_CACHE_DIR = "data/cache/intraday"
UNIVERSE_FILE = "configs/symbols_midlarge_snapshot_exact.yaml"
V24_400D_RUN = "runs/intraday_orb/intraday_20260422_011012_06f59ede.json"


def _parse_ts(ts_raw: object) -> dt.datetime:
    s = str(ts_raw)
    if s.endswith("Z"):
        s = s[:-1] + "+00:00"
    return dt.datetime.fromisoformat(s).astimezone(_ET)


def get_daily_close(ticker: str, date: str) -> float | None:
    path = Path(INTRADAY_CACHE_DIR) / ticker / f"{date}.parquet"
    if not path.exists():
        return None
    try:
        table = pq.read_table(str(path))
        rows = table.to_pydict()
    except Exception:
        return None
    closes = []
    for i, ts_raw in enumerate(rows.get("timestamp", [])):
        try:
            ts = _parse_ts(ts_raw)
        except Exception:
            continue
        if _MKT_OPEN <= ts.time() < _MKT_CLOSE:
            closes.append(float(rows["close"][i] or 0))
    return closes[-1] if closes and closes[-1] > 0 else None


def compute_breadth(universe: list[str], date: str, prev_date: str) -> dict | None:
    """Compute breadth on `date` using closes from `prev_date` and day before."""
    def _load(ticker: str) -> tuple[str, float | None, float | None]:
        return ticker, get_daily_close(ticker, prev_date), get_daily_close(ticker, _two_days_prior(date, prev_date))
    # Simplified: just compare prev_date vs prev-prev-date close
    return None


def get_two_closes(ticker: str, dates: list[str]) -> dict[str, float]:
    """Get close prices for a list of dates."""
    result = {}
    for d in dates:
        c = get_daily_close(ticker, d)
        if c is not None:
            result[d] = c
    return result


def pearson(xs: list[float], ys: list[float]) -> float | None:
    n = len(xs)
    if n < 2:
        return None
    xm, ym = sum(xs) / n, sum(ys) / n
    num = sum((xs[i] - xm) * (ys[i] - ym) for i in range(n))
    dx = sum((x - xm) ** 2 for x in xs) ** 0.5
    dy = sum((y - ym) ** 2 for y in ys) ** 0.5
    if dx <= 0 or dy <= 0:
        return None
    return num / (dx * dy)


def tercile_stats(vals: list[float], rs: list[float]) -> dict:
    if len(vals) < 9:
        return {}
    pairs = sorted(zip(vals, rs), key=lambda p: p[0])
    n = len(pairs)
    t = n // 3
    def stats(sub):
        ys = [p[1] for p in sub]
        return {"n": len(ys), "wr": sum(1 for y in ys if y > 0) / len(ys), "avg_r": sum(ys) / len(ys)}
    return {"low": stats(pairs[:t]), "mid": stats(pairs[t:2*t]), "high": stats(pairs[2*t:])}


def report_feature(label: str, vals: list[float], rs: list[float]) -> None:
    n = len(vals)
    p = pearson(vals, rs)
    ts = tercile_stats(vals, rs)
    if not ts or p is None:
        print(f"  {label}: n={n}, insufficient data")
        return
    low, mid, high = ts["low"], ts["mid"], ts["high"]
    avg_r_gap = abs(high["avg_r"] - low["avg_r"])
    wr_gap = abs(high["wr"] - low["wr"])
    g1 = abs(p) >= 0.07 and n >= 120
    g2 = avg_r_gap >= 0.30
    g3 = wr_gap >= 0.05
    print(f"\n  [{label}]  n={n}  Pearson={p:+.3f}")
    print(f"    G1: {'PASS' if g1 else 'FAIL'} (|{abs(p):.3f}| {'≥' if abs(p)>=0.07 else '<'} 0.07, n={n})")
    print(f"    G2: {'PASS' if g2 else 'FAIL'} (avg_R gap = {avg_r_gap:.3f}R  [threshold 0.30R])")
    print(f"    G3: {'PASS' if g3 else 'FAIL'} (WR gap = {wr_gap*100:.1f}pp  [threshold 5pp])")
    print(f"    Bottom tercile: WR {low['wr']*100:.1f}%  avg_R {low['avg_r']:+.3f}  n={low['n']}")
    print(f"    Middle tercile: WR {mid['wr']*100:.1f}%  avg_R {mid['avg_r']:+.3f}  n={mid['n']}")
    print(f"    Top tercile:    WR {high['wr']*100:.1f}%  avg_R {high['avg_r']:+.3f}  n={high['n']}")
    print(f"    → {'ALL GATES PASS ✓' if (g1 and g2 and g3) else 'FAIL'}")


def main() -> None:
    print("=== V40 Prior-Day Market Breadth Diagnostic ===\n")

    with open(V24_400D_RUN) as f:
        run_data = json.load(f)
    trades = run_data.get("trades", [])
    m = run_data.get("metrics", {})
    print(f"Loaded V24 400d: {m.get('total_trades')} trades, "
          f"{m.get('start_date')} → {m.get('end_date')}")

    trade_records = [
        {"ticker": t["ticker"], "date": t["date"][:10],
         "r_multiple": float(t["r_multiple_at_exit"])}
        for t in trades
        if t.get("r_multiple_at_exit") is not None
    ]

    with open(UNIVERSE_FILE) as f:
        udata = yaml.safe_load(f)
    universe = udata.get("symbols", udata) if isinstance(udata, dict) else list(udata)
    print(f"Universe: {len(universe)} tickers")

    # Get all unique trade dates and their prior trading days
    all_dates_set = set(r["date"] for r in trade_records)
    min_d = dt.date.fromisoformat(min(all_dates_set))
    max_d = dt.date.fromisoformat(max(all_dates_set))
    all_td = trading_days_between(min_d - dt.timedelta(days=30), max_d)
    td_list = [d.isoformat() for d in all_td]

    # Map each trade date to its prior trading day
    prior_day_map: dict[str, str] = {}
    for i, d in enumerate(td_list):
        if d in all_dates_set and i > 0:
            prior_day_map[d] = td_list[i - 1]

    print(f"Trade dates with prior day: {len(prior_day_map)} / {len(all_dates_set)}")

    # For each unique prior day, compute market breadth
    prior_days_needed = sorted(set(prior_day_map.values()))
    print(f"Loading daily closes for universe across {len(prior_days_needed)} prior days...")

    # Build date range for loading
    all_needed_dates: list[str] = []
    for pd in prior_days_needed:
        all_needed_dates.append(pd)

    def _load_ticker(ticker: str) -> tuple[str, dict[str, float]]:
        return ticker, get_two_closes(ticker, all_needed_dates)

    ticker_closes: dict[str, dict[str, float]] = {}
    with concurrent.futures.ThreadPoolExecutor(max_workers=12) as ex:
        for ticker, closes in ex.map(_load_ticker, universe):
            if closes:
                ticker_closes[ticker] = closes

    # Compute breadth per prior day
    # Breadth = fraction of universe UP on that day vs the day before
    # We need TWO prior days: prior_day and prior_prior_day
    all_dates_calendar = []
    d = min_d - dt.timedelta(days=40)
    while d <= max_d:
        all_dates_calendar.append(d.isoformat())
        d += dt.timedelta(days=1)

    # Get prior-prior days
    prior_prior_map: dict[str, str] = {}
    for i, d in enumerate(td_list):
        if d in prior_days_needed and i > 0:
            prior_prior_map[d] = td_list[i - 1]

    # Load prior-prior day closes
    prior_prior_days = sorted(set(prior_prior_map.values()))
    def _load_ticker2(ticker: str) -> tuple[str, dict[str, float]]:
        return ticker, get_two_closes(ticker, prior_prior_days)

    ticker_pp_closes: dict[str, dict[str, float]] = {}
    with concurrent.futures.ThreadPoolExecutor(max_workers=12) as ex:
        for ticker, closes in ex.map(_load_ticker2, universe):
            if closes:
                ticker_pp_closes[ticker] = closes

    # Compute breadth for each prior day
    breadth_by_day: dict[str, dict] = {}
    for pd in prior_days_needed:
        ppd = prior_prior_map.get(pd)
        if ppd is None:
            continue
        adv, dec, total = 0, 0, 0
        for ticker in universe:
            c_pd = ticker_closes.get(ticker, {}).get(pd)
            c_ppd = ticker_pp_closes.get(ticker, {}).get(ppd)
            if c_pd is None or c_ppd is None or c_ppd <= 0:
                continue
            total += 1
            if c_pd > c_ppd:
                adv += 1
            elif c_pd < c_ppd:
                dec += 1
        if total >= 50:
            breadth_by_day[pd] = {
                "up_pct": adv / total,
                "adv_dec": (adv - dec) / total,
                "n": total,
            }

    print(f"Breadth computed for {len(breadth_by_day)} prior days")
    if breadth_by_day:
        sample = list(breadth_by_day.values())[:5]
        up_pcts = [v["up_pct"] for v in breadth_by_day.values()]
        print(f"Breadth range: up_pct {min(up_pcts):.1%} - {max(up_pcts):.1%}  mean={sum(up_pcts)/len(up_pcts):.1%}")

    # Match to trades
    up_pct_vals, adv_dec_vals, r_mults = [], [], []
    missing = 0
    for rec in trade_records:
        pd = prior_day_map.get(rec["date"])
        if pd is None:
            missing += 1
            continue
        bdata = breadth_by_day.get(pd)
        if bdata is None:
            missing += 1
            continue
        up_pct_vals.append(bdata["up_pct"])
        adv_dec_vals.append(bdata["adv_dec"])
        r_mults.append(rec["r_multiple"])

    n_valid = len(r_mults)
    print(f"\nValid trades: {n_valid} / {len(trade_records)}  (missing: {missing})")

    print("\n" + "=" * 60)
    print("GATE RESULTS (G1: |P|≥0.07 & n≥120; G2: avg_R≥0.30R; G3: WR≥5pp)")
    print("NOTE: Day-level signal — effective n is unique trading days, not trades.")
    unique_days = len(set(r["date"] for r in trade_records if prior_day_map.get(r["date"]) in breadth_by_day))
    print(f"Unique trade days: {unique_days}\n")

    report_feature("breadth_up_pct (prior day)", up_pct_vals, r_mults)
    report_feature("breadth_adv_dec (adv-dec/total)", adv_dec_vals, r_mults)

    p_up_adv = pearson(up_pct_vals, adv_dec_vals)
    print(f"\n  ρ(up_pct, adv_dec) = {p_up_adv:.3f}")

    print("\n=== Summary ===")
    print("V24's QQQ gap filter already captures macro regime. This tests BREADTH depth.")


if __name__ == "__main__":
    main()