"""Post-breadth diagnostics for fip_id (research branch only). Same research.sqlite as the liquid-breadth IC run. No production changes. Checks (pre-registered interpretation follow-ups) ------------------------------------------------ 1. **Lagged membership** — liquid top-N ranked on *prior* week's $vol (extra lag) so same-week liquidity explosion cannot pull a name into history. 2. **Liquidity tiers** — fip IC on ranks 1–800 vs 801–1500 (same-week mask). 3. **Prod-universe subset** — symbols present in prod.sqlite (~S&P-like large-cap book) inside the same breadth weeks — compositional vs temporal flip. 4. **Momentum-conditional fip** — among weekly top 20% by mom_12_1 (or resid when available) within the liquid top-N — the paper's actual claim and the only version a gate could consume. Also reports vol_6m / mom raw vs residual on the same panels for the log. Example (Windows) ----------------- .\\.venv\\Scripts\\python.exe scripts\\run_fip_breadth_diagnostics.py ^ --research-snapshot backtest_snapshots\\research.sqlite ^ --prod-snapshot backtest_snapshots\\prod.sqlite ^ --workers 6 """ from __future__ import annotations import argparse import json import math import multiprocessing as mp import sys from collections import defaultdict from concurrent.futures import ProcessPoolExecutor, as_completed from datetime import date, datetime from pathlib import Path from typing import Any from sqlalchemy import create_engine, text ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) HORIZON = 30 MIN_CROSS = 20 MIN_RELIABLE = 12 LIQUID_TOP = 1500 MIN_PRICE = 5.0 MOM_WINNER_PCT = 80.0 # top 20% within liquid cross-section def _parse_args() -> argparse.Namespace: p = argparse.ArgumentParser(description=__doc__) p.add_argument( "--research-snapshot", default="backtest_snapshots/research.sqlite", ) p.add_argument( "--prod-snapshot", default="backtest_snapshots/prod.sqlite", help="Symbols here define the large-cap / prod-like subset.", ) p.add_argument("--top-n", type=int, default=LIQUID_TOP) p.add_argument("--min-price", type=float, default=MIN_PRICE) p.add_argument("--workers", type=int, default=max(1, (mp.cpu_count() or 4) - 1)) p.add_argument("--out", default=None) p.add_argument("--quiet", action="store_true") return p.parse_args() def _week_key(d: date) -> tuple[int, int]: iso = d.isocalendar() return (int(iso[0]), int(iso[1])) def _week_ord(wk: tuple[int, int]) -> int: return wk[0] * 53 + wk[1] def _nonoverlap(weeks: list[tuple[int, int]], stride: int) -> list[tuple[int, int]]: kept: list[tuple[int, int]] = [] last: int | None = None for wk in sorted(weeks, key=_week_ord): o = _week_ord(wk) if last is None or o - last >= stride: kept.append(wk) last = o return kept def _rank(xs: list[float]) -> list[float]: order = sorted(range(len(xs)), key=lambda k: xs[k]) ranks = [0.0] * len(xs) i = 0 while i < len(xs): j = i while j + 1 < len(xs) and xs[order[j + 1]] == xs[order[i]]: j += 1 avg = (i + j) / 2.0 + 1.0 for k in range(i, j + 1): ranks[order[k]] = avg i = j + 1 return ranks def _pearson(a: list[float], b: list[float]) -> float | None: n = len(a) if n < 3: return None ma, mb = sum(a) / n, sum(b) / n va = sum((x - ma) ** 2 for x in a) vb = sum((y - mb) ** 2 for y in b) if va <= 0 or vb <= 0: return None cov = sum((a[k] - ma) * (b[k] - mb) for k in range(n)) return cov / math.sqrt(va * vb) def _spearman(xs: list[float], ys: list[float]) -> float | None: if len(xs) < 3: return None return _pearson(_rank(xs), _rank(ys)) def _ic_row(pairs: list[tuple[float, float]], *, label: str) -> dict[str, Any]: """pairs = (signal, fwd) over non-overlapping weeks aggregated… actually we pass per-week then aggregate outside. This helper is for multi-week IC.""" raise NotImplementedError def _ic_from_weekly( week_pairs: dict[tuple[int, int], list[tuple[float, float]]], ) -> dict[str, Any]: stride = max(1, round(HORIZON / 5)) usable = [wk for wk, ps in week_pairs.items() if len(ps) >= MIN_CROSS] kept = _nonoverlap(usable, stride) ics: list[float] = [] sizes: list[int] = [] for wk in kept: ps = week_pairs[wk] if len(ps) < MIN_CROSS: continue ic = _spearman([p[0] for p in ps], [p[1] for p in ps]) if ic is not None: ics.append(ic) sizes.append(len(ps)) if not ics: return { "mean_ic": None, "ic_t_stat": None, "weeks": 0, "avg_cross_section": None, "ic_positive_pct": None, "reliable": False, } mean_ic = sum(ics) / len(ics) if len(ics) > 1: var = sum((x - mean_ic) ** 2 for x in ics) / (len(ics) - 1) std = math.sqrt(var) if var > 0 else 0.0 t_stat = mean_ic / std * math.sqrt(len(ics)) if std > 0 else None else: t_stat = None return { "mean_ic": round(mean_ic, 4), "ic_t_stat": round(t_stat, 2) if t_stat is not None else None, "weeks": len(ics), "avg_cross_section": round(sum(sizes) / len(sizes), 1), "ic_positive_pct": round(sum(1 for x in ics if x > 0) / len(ics) * 100, 1), "reliable": len(ics) >= MIN_RELIABLE, } def _panel_worker(payload: tuple) -> list[dict]: """Build weekly observations for one ticker (picklable top-level).""" symbol, date_ords, opens, highs, lows, closes, volumes, spy = payload from types import SimpleNamespace from app.services.backtest_service import ( HORIZON as H, _median_dollar_vol_63, _signal_values, _weekly_asof_indices, ) dates = [date.fromordinal(int(o)) for o in date_ords] opens_f = [float(x) for x in opens] highs_f = [float(x) for x in highs] lows_f = [float(x) for x in lows] closes_f = [float(x) for x in closes] vols_f = [float(x) for x in volumes] n = len(closes_f) if n < H + 21: return [] # Match backtest_service bar objects exactly (weekly as-of + signal_values). bar_records = [ SimpleNamespace( date=dates[i], open=opens_f[i], high=highs_f[i], low=lows_f[i], close=closes_f[i], volume=vols_f[i], ) for i in range(n) ] out: list[dict] = [] for i in _weekly_asof_indices(bar_records): j = i + H if j >= n or closes_f[i] <= 0: continue sigs = _signal_values(dates, closes_f, highs_f, i, spy) fip = sigs.get("fip_id") mom = sigs.get("mom_12_1") mom_r = sigs.get("mom_12_1_resid") vol = sigs.get("vol_6m") if fip is None and mom is None: continue dvol = _median_dollar_vol_63(closes_f, vols_f, i) wk = _week_key(dates[i]) out.append({ "symbol": symbol, "week": wk, "fwd": closes_f[j] / closes_f[i] - 1.0, "close": closes_f[i], "dvol": dvol, "fip_id": fip, "mom_12_1": mom, "mom_12_1_resid": mom_r, "vol_6m": vol, }) return out def _load_spy(conn) -> dict[date, float]: rows = conn.execute( text("SELECT date, close FROM benchmark_prices WHERE symbol = 'SPY' ORDER BY date") ).fetchall() out: dict[date, float] = {} for d, c in rows: if isinstance(d, str): d = date.fromisoformat(d[:10]) out[d] = float(c) return out def _load_symbols(conn) -> list[str]: return [ str(r[0]) for r in conn.execute(text("SELECT symbol FROM tickers ORDER BY symbol")).fetchall() ] def _load_columns(conn, symbol: str) -> tuple | None: tid = conn.execute( text("SELECT id FROM tickers WHERE symbol = :s"), {"s": symbol} ).scalar() if tid is None: return None rows = conn.execute( text( "SELECT date, open, high, low, close, volume FROM ohlcv_records " "WHERE ticker_id = :t ORDER BY date" ), {"t": tid}, ).fetchall() if len(rows) < HORIZON + 60: return None ords: list[int] = [] opens: list[float] = [] highs: list[float] = [] lows: list[float] = [] closes: list[float] = [] vols: list[float] = [] for d, o, h, l, c, v in rows: if isinstance(d, str): d = date.fromisoformat(d[:10]) ords.append(d.toordinal()) opens.append(float(o)) highs.append(float(h)) lows.append(float(l)) closes.append(float(c)) vols.append(float(v or 0)) return (symbol, ords, opens, highs, lows, closes, vols) def _liquid_members( obs: list[dict], *, top_n: int, min_price: float, dvol_key: str = "dvol", ) -> list[dict]: eligible = [ o for o in obs if o.get("close") is not None and float(o["close"]) >= min_price and o.get(dvol_key) is not None and float(o[dvol_key]) > 0 ] eligible.sort(key=lambda o: float(o[dvol_key]), reverse=True) return eligible[:top_n] def _pairs(obs: list[dict], signal: str) -> list[tuple[float, float]]: out: list[tuple[float, float]] = [] for o in obs: v = o.get(signal) if v is None: continue out.append((float(v), float(o["fwd"]))) return out def main() -> None: args = _parse_args() research = Path(args.research_snapshot) prod = Path(args.prod_snapshot) if not research.exists(): raise SystemExit(f"Missing research snapshot: {research}") research_eng = create_engine(f"sqlite:///{research.resolve().as_posix()}") prod_symbols: set[str] = set() if prod.exists(): prod_eng = create_engine(f"sqlite:///{prod.resolve().as_posix()}") with prod_eng.connect() as c: prod_symbols = { str(r[0]) for r in c.execute(text("SELECT symbol FROM tickers")).fetchall() } prod_eng.dispose() with research_eng.connect() as conn: spy = _load_spy(conn) symbols = _load_symbols(conn) jobs: list[tuple] = [] for i, sym in enumerate(symbols, 1): cols = _load_columns(conn, sym) if cols is None: continue jobs.append((*cols, spy)) if not args.quiet and i % 500 == 0: print(f" queued {i}/{len(symbols)}", flush=True) if not args.quiet: print(f"Building weekly panel for {len(jobs)} tickers…", flush=True) # Panel: week -> list of obs by_week: dict[tuple[int, int], list[dict]] = defaultdict(list) workers = max(1, int(args.workers)) if workers == 1: for j, job in enumerate(jobs, 1): for row in _panel_worker(job): by_week[tuple(row["week"])].append(row) if not args.quiet and j % 200 == 0: print(f" panel {j}/{len(jobs)}", flush=True) else: with ProcessPoolExecutor(max_workers=workers) as pool: futs = {pool.submit(_panel_worker, job): job[0] for job in jobs} done = 0 for fut in as_completed(futs): done += 1 try: rows = fut.result() except Exception as exc: if not args.quiet: print(f" worker error {futs[fut]}: {exc}", flush=True) continue for row in rows: by_week[tuple(row["week"])].append(row) if not args.quiet and done % 200 == 0: print(f" panel {done}/{len(jobs)}", flush=True) if not args.quiet: print(f"Weeks with data: {len(by_week)}", flush=True) # Prior-week dvol map for lagged membership: (symbol, week) -> dvol dvol_by_sym_week: dict[tuple[str, tuple[int, int]], float] = {} for wk, obs in by_week.items(): for o in obs: if o.get("dvol") is not None: dvol_by_sym_week[(o["symbol"], wk)] = float(o["dvol"]) ordered_weeks = sorted(by_week.keys(), key=_week_ord) prev_week: dict[tuple[int, int], tuple[int, int]] = {} for i, wk in enumerate(ordered_weeks): if i > 0: prev_week[wk] = ordered_weeks[i - 1] top_n = int(args.top_n) min_price = float(args.min_price) # --- Panels for each check --- same_week_fip: dict[tuple[int, int], list[tuple[float, float]]] = defaultdict(list) lag_week_fip: dict[tuple[int, int], list[tuple[float, float]]] = defaultdict(list) tier_hi_fip: dict[tuple[int, int], list[tuple[float, float]]] = defaultdict(list) tier_lo_fip: dict[tuple[int, int], list[tuple[float, float]]] = defaultdict(list) prod_subset_fip: dict[tuple[int, int], list[tuple[float, float]]] = defaultdict(list) mom_cond_fip: dict[tuple[int, int], list[tuple[float, float]]] = defaultdict(list) liquid_vol: dict[tuple[int, int], list[tuple[float, float]]] = defaultdict(list) liquid_mom: dict[tuple[int, int], list[tuple[float, float]]] = defaultdict(list) liquid_mom_r: dict[tuple[int, int], list[tuple[float, float]]] = defaultdict(list) for wk, obs in by_week.items(): # Same-week liquid top-N among names that have fip (matches signal_eval mask: # membership is ranked within each signal's observation set). with_fip = [o for o in obs if o.get("fip_id") is not None] liq_fip = _liquid_members(with_fip, top_n=top_n, min_price=min_price) for rank, o in enumerate(liq_fip, 1): same_week_fip[wk].append((float(o["fip_id"]), float(o["fwd"]))) if rank <= 800: tier_hi_fip[wk].append((float(o["fip_id"]), float(o["fwd"]))) elif rank <= top_n: tier_lo_fip[wk].append((float(o["fip_id"]), float(o["fwd"]))) if o["symbol"] in prod_symbols: prod_subset_fip[wk].append((float(o["fip_id"]), float(o["fwd"]))) # Context signals: liquid among names that carry that signal with_vol = [o for o in obs if o.get("vol_6m") is not None] for o in _liquid_members(with_vol, top_n=top_n, min_price=min_price): liquid_vol[wk].append((float(o["vol_6m"]), float(o["fwd"]))) with_mom_all = [o for o in obs if o.get("mom_12_1") is not None] liq_mom = _liquid_members(with_mom_all, top_n=top_n, min_price=min_price) for o in liq_mom: liquid_mom[wk].append((float(o["mom_12_1"]), float(o["fwd"]))) with_mom_r = [o for o in obs if o.get("mom_12_1_resid") is not None] for o in _liquid_members(with_mom_r, top_n=top_n, min_price=min_price): liquid_mom_r[wk].append((float(o["mom_12_1_resid"]), float(o["fwd"]))) # Momentum-conditional: within liquid fip set, keep mom_12_1 ≥ P80 mom_key = "mom_12_1" with_mom = [ o for o in liq_fip if o.get(mom_key) is not None and o.get("fip_id") is not None ] if len(with_mom) >= MIN_CROSS: with_mom.sort(key=lambda o: float(o[mom_key])) n = len(with_mom) cut = int(math.floor(n * (MOM_WINNER_PCT / 100.0))) winners = with_mom[cut:] # upper tail for o in winners: mom_cond_fip[wk].append((float(o["fip_id"]), float(o["fwd"]))) # Lagged membership: rank by *previous* week's dvol among fip names pw = prev_week.get(wk) if pw is not None: lagged: list[dict] = [] for o in with_fip: if o.get("close") is None or float(o["close"]) < min_price: continue prev_dvol = dvol_by_sym_week.get((o["symbol"], pw)) if prev_dvol is None or prev_dvol <= 0: continue lagged.append({**o, "lag_dvol": prev_dvol}) lagged.sort(key=lambda o: float(o["lag_dvol"]), reverse=True) for o in lagged[:top_n]: lag_week_fip[wk].append((float(o["fip_id"]), float(o["fwd"]))) results = { "generated_at": datetime.now().isoformat(), "research_snapshot": str(research.resolve()), "prod_subset_n": len(prod_symbols), "panel_tickers": len(jobs), "top_n": top_n, "min_price": min_price, "checks": { "fip_same_week_liquid_1500": { "note": "Replication of main breadth run (same-week $vol mask)", **_ic_from_weekly(same_week_fip), }, "fip_lagged_membership_1w": { "note": ( "Liquid top-N ranked on *prior* week's median $vol — " "excludes same-week liquidity explosion leak" ), **_ic_from_weekly(lag_week_fip), }, "fip_tier_1_800": { "note": "Same-week liquid ranks 1–800 (senior liquid tier)", **_ic_from_weekly(tier_hi_fip), }, "fip_tier_801_1500": { "note": "Same-week liquid ranks 801–1500 (junior liquid tier)", **_ic_from_weekly(tier_lo_fip), }, "fip_prod_universe_subset": { "note": ( "Symbols in prod.sqlite (~S&P-like large-cap book) inside " "same-week liquid top-N — compositional control" ), **_ic_from_weekly(prod_subset_fip), }, "fip_momentum_conditional_top20pct": { "note": ( f"Among liquid top-N, keep mom_12_1 percentile ≥ {MOM_WINNER_PCT} " "(paper: ID modulates continuation among winners; gate-relevant)" ), **_ic_from_weekly(mom_cond_fip), }, "vol_6m_liquid_1500": { "note": "Context: low-vol anomaly strength on this pool", **_ic_from_weekly(liquid_vol), }, "mom_12_1_liquid_1500": { "note": "Context: raw momentum on liquid breadth", **_ic_from_weekly(liquid_mom), }, "mom_12_1_resid_liquid_1500": { "note": "Context: residual momentum on liquid breadth", **_ic_from_weekly(liquid_mom_r), }, }, } # Interpretations checks = results["checks"] lag = checks["fip_lagged_membership_1w"] same = checks["fip_same_week_liquid_1500"] hi = checks["fip_tier_1_800"] lo = checks["fip_tier_801_1500"] prod = checks["fip_prod_universe_subset"] cond = checks["fip_momentum_conditional_top20pct"] def _sign(x: float | None) -> str: if x is None: return "na" return "neg" if x < 0 else "pos" results["interpretation"] = { "leak_ruled_out": ( lag.get("mean_ic") is not None and same.get("mean_ic") is not None and _sign(lag["mean_ic"]) == _sign(same["mean_ic"]) and abs(float(lag["mean_ic"])) >= 0.02 ), "junior_tier_drives_positive": ( lo.get("mean_ic") is not None and float(lo["mean_ic"]) > 0 and (hi.get("mean_ic") is None or float(hi["mean_ic"]) < float(lo["mean_ic"])) ), "prod_subset_still_negative": ( prod.get("mean_ic") is not None and float(prod["mean_ic"]) < 0 ), "mom_conditional_negative_and_reliable": ( cond.get("mean_ic") is not None and float(cond["mean_ic"]) < 0 and abs(float(cond["mean_ic"])) >= 0.03 and bool(cond.get("reliable")) ), "compositional_flip_story": ( "If prod subset IC is negative while full liquid-1500 is positive, " "the sign flip is compositional (bleeders / Nasdaq junk), not a " "temporal regime change. Unconditional fip pools continuous winners " "(want neg IC) against continuous losers/bleeders (want pos IC)." ), "vol_tilt_warning": ( "vol_6m large negative IC on breadth: high-vol lottery names " "underperform. Production 80/20 high-vol tilt was validated on " "S&P-like names; must re-validate before any universe broaden." ), } # Gate-relevant summary line if results["interpretation"]["mom_conditional_negative_and_reliable"]: results["platform_verdict"] = ( "ALIVE as breadth-book tilt candidate among momentum winners only — " "still needs a book-level experiment; not a production wire-in." ) else: results["platform_verdict"] = ( "CLOSED for production use: momentum-conditional fip does not clear " "iron rule on this liquid-Nasdaq pool. Display card remains final resting place." ) stamp = datetime.now().strftime("%Y%m%d-%H%M%S") out = Path(args.out) if args.out else Path("reports") / f"fip-breadth-diagnostics-{stamp}.json" out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(results, indent=2, default=str), encoding="utf-8") # Append to research log md_path = Path("docs/research/fip-breadth-ic.md") _append_diagnostics_md(md_path, results, out) if not args.quiet: print(json.dumps(results["checks"], indent=2, default=str)) print() print("interpretation:", json.dumps(results["interpretation"], indent=2)) print("platform_verdict:", results["platform_verdict"]) print(f"Wrote {out}") print(f"Updated {md_path}") def _append_diagnostics_md(path: Path, results: dict, artifact: Path) -> None: checks = results["checks"] interp = results["interpretation"] lines = [ "", "---", "", f"## Follow-up diagnostics ({results['generated_at'][:10]})", "", "Compositional reading of the sign flip (before any 'jumpiness premium' story):", "", "`fip_id = sign(PRET) × (%neg − %pos)` pools two opposite continuous populations:", "", "- **Continuous winners** (PRET>0, mostly up days) → paper claim → **negative** IC contribution.", "- **Continuous losers / bleeders** (PRET<0, mostly down days) → momentum continuation down → **positive** IC contribution.", "", "Unconditional IC is a tug-of-war weighted by universe composition. S&P-like books " "have few steady bleeders → negative fip IC. Liquid Nasdaq has many → sign can flip " "without contradicting Da/Gurun/Warachka (claim was always **momentum-conditional**).", "", "### Artifact / composition checks", "", "| check | mean_ic | t | weeks | avg N | reliable |", "|---|---:|---:|---:|---:|---|", ] order = [ "fip_same_week_liquid_1500", "fip_lagged_membership_1w", "fip_tier_1_800", "fip_tier_801_1500", "fip_prod_universe_subset", "fip_momentum_conditional_top20pct", "vol_6m_liquid_1500", "mom_12_1_liquid_1500", "mom_12_1_resid_liquid_1500", ] for key in order: row = checks.get(key) or {} lines.append( f"| {key} | {row.get('mean_ic')} | {row.get('ic_t_stat')} | " f"{row.get('weeks')} | {row.get('avg_cross_section')} | {row.get('reliable')} |" ) lines.extend([ "", "### Flags", "", f"- Lagged mask keeps same sign / material |IC|: **{interp.get('leak_ruled_out')}**", f"- Junior tier (801–1500) drives more positive IC: **{interp.get('junior_tier_drives_positive')}**", f"- Prod-universe subset still negative: **{interp.get('prod_subset_still_negative')}**", f"- Mom-conditional (≥P80) negative + reliable: **{interp.get('mom_conditional_negative_and_reliable')}**", "", "### Platform verdict", "", results.get("platform_verdict", ""), "", "### Vol-tilt warning (any future breadth move)", "", interp.get("vol_tilt_warning", ""), "", f"Artifact: `{artifact.as_posix()}`", "", ]) # Replace previous diagnostics section if re-run, else append existing = path.read_text(encoding="utf-8") if path.exists() else "" marker = "## Follow-up diagnostics" if marker in existing: existing = existing.split(marker)[0].rstrip() + "\n" path.write_text(existing + "\n".join(lines), encoding="utf-8") if __name__ == "__main__": main()