Files
signal-platform/scripts/run_fip_breadth_diagnostics.py
T
dennisthiessen 7d60e54f5a research: single-source liquid mask; orphan +0.06 fip IC
Harness and diagnostics share _filter_liquid_breadth_week_rich. Recompute
shows unconditional liquid fip IC -0.017 (mask binds 97%); mom-conditional
-0.088/t-4.58 stands. Document +0.0575 as orphaned.
2026-07-19 00:06:04 +02:00

648 lines
25 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""fip_id breadth diagnostics — single-sourced through harness mask helpers.
Uses the same collection + ``_filter_liquid_breadth_week_rich`` as
``run_backtest`` signal_eval. No parallel mask implementation.
Reconciles the harness +0.0575 vs prior dual-path 0.017 disagreement by
deleting the second mask, dumping membership/pre-post stats, and re-running
mom-conditional IC through the surviving path only.
Research branch only. Example:
.\\.venv\\Scripts\\python.exe scripts\\run_fip_breadth_diagnostics.py ^
--research-snapshot backtest_snapshots\\research.sqlite ^
--prod-snapshot backtest_snapshots\\prod.sqlite ^
--workers 6 --allow-spawn
"""
from __future__ import annotations
import argparse
import json
import math
import multiprocessing as mp
import os
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))
# Match production signal_eval cadence / reliability bars.
MIN_CROSS = 20
MIN_RELIABLE = 12
MOM_WINNER_PCT = 80.0
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")
p.add_argument("--top-n", type=int, default=1500)
p.add_argument("--min-price", type=float, default=5.0)
p.add_argument("--workers", type=int, default=max(1, (mp.cpu_count() or 4) - 1))
p.add_argument("--allow-spawn", action="store_true")
p.add_argument("--dump-weeks", type=int, default=5, help="How many weeks to dump membership for")
p.add_argument("--out", default=None)
p.add_argument("--quiet", action="store_true")
return p.parse_args()
def _week_ord(wk: tuple[int, int]) -> int:
return int(wk[0]) * 53 + int(wk[1])
def _nonoverlap(weeks: list[tuple[int, int]], stride: int) -> list[tuple[int, int]]:
from app.services.backtest_service import _nonoverlapping_weeks
return _nonoverlapping_weeks(weeks, stride)
def _ic_from_weekly(
week_pairs: dict[tuple[int, int], list[tuple[float, float]]],
) -> dict[str, Any]:
from app.services.backtest_service import HORIZON, _spearman
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 _worker(payload: tuple) -> dict:
"""Return harness-style signal series for one ticker (liquid-mode dicts)."""
symbol, ords, opens, highs, lows, closes, volumes, spy = payload
from types import SimpleNamespace
from app.services.backtest_service import _signal_series
bars = [
SimpleNamespace(
date=date.fromordinal(int(o)),
open=float(op),
high=float(hi),
low=float(lo),
close=float(cl),
volume=float(vo),
)
for o, op, hi, lo, cl, vo in zip(ords, opens, highs, lows, closes, volumes)
]
return _signal_series(bars, spy, symbol=symbol)
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_job(conn, symbol: str, spy: dict) -> 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) < 90:
return None
ords, opens, highs, lows, closes, vols = [], [], [], [], [], []
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, spy)
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}")
# Force harness liquid-mode collection (same env as breadth run).
os.environ["BACKTEST_LIQUID_BREADTH"] = str(int(args.top_n))
os.environ["BACKTEST_LIQUID_MIN_PRICE"] = str(float(args.min_price))
if args.allow_spawn:
os.environ["BACKTEST_ALLOW_SPAWN"] = "1"
from app.services.backtest_service import (
HORIZON,
_filter_liquid_breadth_week_rich,
_liquid_breadth_week_stats,
_signal_evaluation,
)
eng = create_engine(f"sqlite:///{research.resolve().as_posix()}")
prod_symbols: set[str] = set()
if prod.exists():
peng = create_engine(f"sqlite:///{prod.resolve().as_posix()}")
with peng.connect() as c:
prod_symbols = {
str(r[0]) for r in c.execute(text("SELECT symbol FROM tickers"))
}
peng.dispose()
with eng.connect() as conn:
spy = _load_spy(conn)
symbols = [
str(r[0])
for r in conn.execute(text("SELECT symbol FROM tickers ORDER BY symbol"))
]
jobs = []
for i, sym in enumerate(symbols, 1):
job = _load_job(conn, sym, spy)
if job is not None:
jobs.append(job)
if not args.quiet and i % 500 == 0:
print(f" queued {i}/{len(symbols)}", flush=True)
if not args.quiet:
print(f"Collecting harness signal series for {len(jobs)} tickers…", flush=True)
collected: dict = defaultdict(lambda: defaultdict(list))
workers = max(1, int(args.workers))
def _merge(series: dict) -> None:
for name, weeks in series.items():
for wk, recs in weeks.items():
# week keys may arrive as lists after JSON; normalize to tuple
key = tuple(wk) if not isinstance(wk, tuple) else wk
collected[name][key].extend(recs)
if workers == 1:
for j, job in enumerate(jobs, 1):
_merge(_worker(job))
if not args.quiet and j % 200 == 0:
print(f" series {j}/{len(jobs)}", flush=True)
else:
ctx = mp.get_context("spawn") if args.allow_spawn or sys.platform == "win32" else None
with ProcessPoolExecutor(max_workers=workers, mp_context=ctx) as pool:
futs = [pool.submit(_worker, job) for job in jobs]
for j, fut in enumerate(as_completed(futs), 1):
try:
_merge(fut.result())
except Exception as exc:
if not args.quiet:
print(f" worker error: {exc}", flush=True)
if not args.quiet and j % 200 == 0:
print(f" series {j}/{len(jobs)}", flush=True)
# --- Harness signal_eval (authoritative unconditional ICs) ---
harness_rows = _signal_evaluation(dict(collected))
harness_by_name = {r["signal"]: r for r in harness_rows}
top_n = int(args.top_n)
min_price = float(args.min_price)
fip_weeks = collected.get("fip_id") or {}
mom_weeks = collected.get("mom_12_1") or {}
vol_weeks = collected.get("vol_6m") or {}
momr_weeks = collected.get("mom_12_1_resid") or {}
# Index mom/vol by (week, symbol) for joins
def _index(weeks_map: dict) -> dict[tuple, dict]:
out: dict[tuple, dict] = {}
for wk, recs in weeks_map.items():
key_wk = tuple(wk) if not isinstance(wk, tuple) else wk
for rec in recs:
if not isinstance(rec, dict):
continue
sym = rec.get("symbol")
if not sym:
continue
out[(key_wk, str(sym))] = rec
return out
mom_ix = _index(mom_weeks)
vol_ix = _index(vol_weeks)
momr_ix = _index(momr_weeks)
# Per-week membership + extended checks via shared rich filter
same_week: dict[tuple, list[tuple[float, float]]] = defaultdict(list)
lag_week: dict[tuple, list[tuple[float, float]]] = defaultdict(list)
tier_hi: dict[tuple, list[tuple[float, float]]] = defaultdict(list)
tier_lo: dict[tuple, list[tuple[float, float]]] = defaultdict(list)
prod_sub: dict[tuple, list[tuple[float, float]]] = defaultdict(list)
mom_cond: dict[tuple, list[tuple[float, float]]] = defaultdict(list)
vol_pairs: dict[tuple, list[tuple[float, float]]] = defaultdict(list)
mom_pairs: dict[tuple, list[tuple[float, float]]] = defaultdict(list)
momr_pairs: dict[tuple, list[tuple[float, float]]] = defaultdict(list)
ordered = sorted((tuple(w) for w in fip_weeks.keys()), key=_week_ord)
prev: dict[tuple, tuple] = {}
for i, wk in enumerate(ordered):
if i:
prev[wk] = ordered[i - 1]
# Prior-week dvol for lag: (symbol, week) from fip recs
dvol_sw: dict[tuple[str, tuple], float] = {}
for wk, recs in fip_weeks.items():
key_wk = tuple(wk) if not isinstance(wk, tuple) else wk
for rec in recs:
if isinstance(rec, dict) and rec.get("symbol") and rec.get("median_dvol_63"):
dvol_sw[(str(rec["symbol"]), key_wk)] = float(rec["median_dvol_63"])
membership_dumps: list[dict] = []
dump_count = 0
stride = max(1, round(HORIZON / 5))
dump_weeks = _nonoverlap(ordered, stride)[: max(0, int(args.dump_weeks))]
for wk_raw, recs in fip_weeks.items():
wk = tuple(wk_raw) if not isinstance(wk_raw, tuple) else wk_raw
stats = _liquid_breadth_week_stats(recs, top_n=top_n, min_price=min_price)
rich = _filter_liquid_breadth_week_rich(
recs, top_n=top_n, min_price=min_price
)
for rank, row in enumerate(rich, 1):
same_week[wk].append((float(row["val"]), float(row["fwd"])))
if rank <= 800:
tier_hi[wk].append((float(row["val"]), float(row["fwd"])))
elif rank <= top_n:
tier_lo[wk].append((float(row["val"]), float(row["fwd"])))
sym = row.get("symbol")
if sym and str(sym) in prod_symbols:
prod_sub[wk].append((float(row["val"]), float(row["fwd"])))
# Join mom for conditional
mrec = mom_ix.get((wk, str(sym))) if sym else None
if mrec is not None:
row["mom_12_1"] = mrec.get("val")
# Mom-conditional among liquid fip set
with_mom = [
r for r in rich
if r.get("mom_12_1") is not None or mom_ix.get((wk, str(r.get("symbol"))))
]
# ensure mom filled
for r in with_mom:
if r.get("mom_12_1") is None and r.get("symbol"):
m = mom_ix.get((wk, str(r["symbol"])))
if m is not None:
r["mom_12_1"] = m["val"]
with_mom = [r for r in rich if r.get("mom_12_1") is not None]
if len(with_mom) >= MIN_CROSS:
with_mom.sort(key=lambda r: float(r["mom_12_1"]))
cut = int(math.floor(len(with_mom) * (MOM_WINNER_PCT / 100.0)))
for r in with_mom[cut:]:
mom_cond[wk].append((float(r["val"]), float(r["fwd"])))
# Context signals via same shared filter on their own pools
for r in _filter_liquid_breadth_week_rich(
vol_weeks.get(wk_raw) or vol_weeks.get(wk) or [],
top_n=top_n,
min_price=min_price,
):
vol_pairs[wk].append((float(r["val"]), float(r["fwd"])))
for r in _filter_liquid_breadth_week_rich(
mom_weeks.get(wk_raw) or mom_weeks.get(wk) or [],
top_n=top_n,
min_price=min_price,
):
mom_pairs[wk].append((float(r["val"]), float(r["fwd"])))
for r in _filter_liquid_breadth_week_rich(
momr_weeks.get(wk_raw) or momr_weeks.get(wk) or [],
top_n=top_n,
min_price=min_price,
):
momr_pairs[wk].append((float(r["val"]), float(r["fwd"])))
# Lagged membership using prior week dvol on current fip pool
pw = prev.get(wk)
if pw is not None:
lagged_recs = []
for rec in recs:
if not isinstance(rec, dict) or not rec.get("symbol"):
continue
pdv = dvol_sw.get((str(rec["symbol"]), pw))
if pdv is None or pdv <= 0:
continue
# Clone with lag dvol for ranking
lagged_recs.append({
**rec,
"median_dvol_63": pdv,
})
for r in _filter_liquid_breadth_week_rich(
lagged_recs, top_n=top_n, min_price=min_price
):
lag_week[wk].append((float(r["val"]), float(r["fwd"])))
if wk in dump_weeks and dump_count < args.dump_weeks:
membership_dumps.append({
"week": list(wk),
"stats": stats,
"symbols": sorted(
str(r["symbol"]) for r in rich if r.get("symbol")
),
"n_symbols": len(rich),
})
dump_count += 1
# IC rows
checks = {
"fip_harness_signal_eval": {
"note": "Authoritative harness _signal_evaluation on collected fip_id",
**(harness_by_name.get("fip_id") or {}),
},
"fip_same_week_via_shared_filter": {
"note": "Same collected data, IC via shared _filter_liquid_breadth_week_rich",
**_ic_from_weekly(same_week),
},
"fip_lagged_membership_1w": {
"note": "Top-N by prior-week $vol on current fip pool (shared filter)",
**_ic_from_weekly(lag_week),
},
"fip_tier_1_800": {
"note": "Senior liquid ranks 1800",
**_ic_from_weekly(tier_hi),
},
"fip_tier_801_1500": {
"note": "Junior liquid ranks 801top_n",
**_ic_from_weekly(tier_lo),
},
"fip_prod_universe_subset": {
"note": "Prod.sqlite symbols inside liquid fip set",
**_ic_from_weekly(prod_sub),
},
"fip_momentum_conditional_top20pct": {
"note": (
f"Among liquid fip set, mom_12_1 ≥ P{MOM_WINNER_PCT:.0f} "
"(paper / gate-relevant)"
),
**_ic_from_weekly(mom_cond),
},
"vol_6m_liquid": {
"note": "vol_6m through shared filter",
**_ic_from_weekly(vol_pairs),
},
"mom_12_1_liquid": {
"note": "raw mom through shared filter",
**_ic_from_weekly(mom_pairs),
},
"mom_12_1_resid_liquid": {
"note": "residual mom through shared filter",
**_ic_from_weekly(momr_pairs),
},
}
h = checks["fip_harness_signal_eval"]
s = checks["fip_same_week_via_shared_filter"]
cond = checks["fip_momentum_conditional_top20pct"]
prod = checks["fip_prod_universe_subset"]
hi = checks["fip_tier_1_800"]
lo = checks["fip_tier_801_1500"]
lag = checks["fip_lagged_membership_1w"]
# Self-consistency: harness eval vs manual IC on same filter must match
harness_ic = h.get("mean_ic")
shared_ic = s.get("mean_ic")
consistent = (
harness_ic is not None
and shared_ic is not None
and abs(float(harness_ic) - float(shared_ic)) < 0.005
)
mom_alive = (
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"))
)
results = {
"generated_at": datetime.now().isoformat(),
"research_snapshot": str(research.resolve()),
"top_n": top_n,
"min_price": min_price,
"prod_subset_n": len(prod_symbols),
"panel_tickers": len(jobs),
"single_source": (
"diagnostics uses harness _signal_series + "
"_filter_liquid_breadth_week_rich only (no parallel mask)"
),
"avg_cross_section_semantics": (
"avg_cross_section = post-mask IC sample size. "
"avg_raw_pool = pre-filter observations. "
"avg_eligible_pre_mask = pass price+dvol before top-N. "
"mask_binds_pct = weeks where eligible_pre_mask > top_n."
),
"harness_self_consistent": consistent,
"checks": checks,
"membership_dumps": membership_dumps,
"interpretation": {
"harness_and_shared_filter_agree": consistent,
"mask_binds_pct": h.get("mask_binds_pct"),
"avg_eligible_pre_mask": h.get("avg_eligible_pre_mask"),
"avg_raw_pool": h.get("avg_raw_pool"),
"prod_subset_still_negative": (
prod.get("mean_ic") is not None and float(prod["mean_ic"]) < 0
),
"junior_tier_more_positive": (
lo.get("mean_ic") is not None
and hi.get("mean_ic") is not None
and float(lo["mean_ic"]) > float(hi["mean_ic"])
),
"lag_same_sign_as_same_week": (
lag.get("mean_ic") is not None
and s.get("mean_ic") is not None
and (float(lag["mean_ic"]) < 0) == (float(s["mean_ic"]) < 0)
),
"mom_conditional_negative_and_reliable": mom_alive,
"orphan_plus_five_sigma": (
"Prior report fip-breadth-20260718-211440-breadth.json listed "
"fip IC +0.0575 / t +5.12. This single-sourced recompute is the "
"authoritative number; if it disagrees, the +0.0575 row is orphaned."
),
"compositional_story": (
"fip_id pools continuous winners (neg IC) vs continuous bleeders "
"(pos IC). Prod-subset and senior liquid stay negative; junior "
"liquid is less negative / positive — composition, not jumpiness premium."
),
"vol_tilt_warning": (
"High-vol names underperform on breadth relative to S&P-like books. "
"Re-validate production 80/20 high-vol tilt before any universe broaden."
),
},
"platform_verdict": (
"Mom-conditional fip ALIVE as book-tilt candidate (needs book sim) — "
"not production wire-in. Unconditional fip not green."
if mom_alive
else (
"fip CLOSED for production: mom-conditional does not clear iron rule "
"on single-sourced path. Display card is the resting place."
)
),
}
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
out = Path(args.out) if args.out else Path("reports") / f"fip-reconcile-{stamp}.json"
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(json.dumps(results, indent=2, default=str), encoding="utf-8")
# Update research log
_update_md(Path("docs/research/fip-breadth-ic.md"), results, out)
if not args.quiet:
print("=== Harness fip_id (authoritative) ===")
print(json.dumps(h, indent=2, default=str))
print("=== Shared-filter same-week (must match) ===")
print(json.dumps(s, indent=2, default=str))
print("=== Mom-conditional ===")
print(json.dumps(cond, indent=2, default=str))
print("self_consistent:", consistent)
print("platform_verdict:", results["platform_verdict"])
print(f"Wrote {out}")
def _update_md(path: Path, results: dict, artifact: Path) -> None:
checks = results["checks"]
interp = results["interpretation"]
h = checks.get("fip_harness_signal_eval") or {}
lines = [
"",
"---",
"",
f"## Reconciliation ({results['generated_at'][:10]})",
"",
"### Problem",
"",
"Two implementations of the liquid-1500 fip IC disagreed on **sign**:",
"",
"- Harness report `fip-breadth-20260718-211440-breadth.json`: **+0.0575 / t +5.12**",
"- Dual-path diagnostics (since deleted): **0.017 / t 1.9**",
"",
"A static read cannot decide which is right without single-sourcing the mask.",
"",
"### Resolution",
"",
f"- **Single source:** {results.get('single_source')}",
f"- **avg_cross_section semantics:** {results.get('avg_cross_section_semantics')}",
f"- Harness `_signal_evaluation` vs shared-filter recompute agree: "
f"**{interp.get('harness_and_shared_filter_agree')}**",
"",
"### Authoritative unconditional fip (liquid top-N, post-mask)",
"",
f"| metric | value |",
f"|---|---|",
f"| mean_ic | {h.get('mean_ic')} |",
f"| ic_t_stat | {h.get('ic_t_stat')} |",
f"| weeks | {h.get('weeks')} |",
f"| avg_cross_section (post-mask) | {h.get('avg_cross_section')} |",
f"| avg_raw_pool | {h.get('avg_raw_pool')} |",
f"| avg_eligible_pre_mask | {h.get('avg_eligible_pre_mask')} |",
f"| mask_binds_pct | {h.get('mask_binds_pct')} |",
f"| reliable | {h.get('reliable')} |",
"",
"The **+0.0575 / +5.12** row is **orphaned** if the authoritative recompute "
"disagrees; do not cite it. Iron-rule unconditional green still requires "
"negative sign and |IC| ≳ 0.03 on this row.",
"",
"### Checks (single-sourced)",
"",
"| check | mean_ic | t | weeks | avg N | reliable |",
"|---|---:|---:|---:|---:|---|",
]
for key in [
"fip_harness_signal_eval",
"fip_same_week_via_shared_filter",
"fip_lagged_membership_1w",
"fip_tier_1_800",
"fip_tier_801_1500",
"fip_prod_universe_subset",
"fip_momentum_conditional_top20pct",
"vol_6m_liquid",
"mom_12_1_liquid",
"mom_12_1_resid_liquid",
]:
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"- Prod subset still negative: **{interp.get('prod_subset_still_negative')}**",
f"- Junior tier more positive than senior: **{interp.get('junior_tier_more_positive')}**",
f"- Lag same sign as same-week: **{interp.get('lag_same_sign_as_same_week')}**",
f"- Mom-conditional negative + reliable: **{interp.get('mom_conditional_negative_and_reliable')}**",
"",
"### Platform verdict (post-reconciliation)",
"",
results.get("platform_verdict", ""),
"",
"### Vol-tilt warning",
"",
interp.get("vol_tilt_warning", ""),
"",
f"Artifact: `{artifact.as_posix()}`",
"",
])
existing = path.read_text(encoding="utf-8") if path.exists() else ""
marker = "## Reconciliation"
if marker in existing:
existing = existing.split(marker)[0].rstrip() + "\n"
# Also strip old dual-path diagnostics section if present after reconciliation
if "## Follow-up diagnostics" in existing and marker not in path.read_text(encoding="utf-8") if path.exists() else "":
pass
path.write_text(existing.rstrip() + "\n" + "\n".join(lines), encoding="utf-8")
if __name__ == "__main__":
main()