signal-platform/scripts/run_fip_breadth_research.py

"""Phase B: fip_id IC on liquid-breadth cross-section (local research only).

1. Fingerprint check on the unextended prod snapshot (must ≈ IC −0.045 / t −2.9).
2. Assert research.sqlite has a matching **completion manifest** (race guard).
3. Run signal_eval on research.sqlite with BACKTEST_LIQUID_BREADTH=1500 PIT mask.
4. Write a research report under docs/research/ and reports/.

Does not modify production DB, gate, scanner, or schedule.

Example
-------
  # After extend_snapshot_universe.py has built research.sqlite:
  python scripts/run_fip_breadth_research.py \\
      --prod-snapshot backtest_snapshots/prod.sqlite \\
      --research-snapshot backtest_snapshots/research.sqlite \\
      --workers 6 --allow-spawn
"""

from __future__ import annotations

import argparse
import asyncio
import json
import os
import sys
from datetime import datetime
from pathlib import Path

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

FINGERPRINT_IC = -0.045
FINGERPRINT_T = -2.9
FINGERPRINT_IC_TOL = 0.015
FINGERPRINT_T_TOL = 0.6


def _sqlite_url(path: Path) -> str:
    return f"sqlite+aiosqlite:///{path.resolve().as_posix()}"


def _parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("--prod-snapshot", default="backtest_snapshots/prod.sqlite")
    p.add_argument("--research-snapshot", default="backtest_snapshots/research.sqlite")
    p.add_argument("--workers", type=int, default=6)
    p.add_argument("--allow-spawn", action="store_true")
    p.add_argument("--skip-fingerprint", action="store_true")
    p.add_argument("--skip-research", action="store_true")
    p.add_argument("--liquid-breadth", type=int, default=1500)
    p.add_argument("--min-price", type=float, default=5.0)
    p.add_argument(
        "--out",
        default=None,
        help="JSON report path (default reports/fip-breadth-YYYYMMDD.json)",
    )
    p.add_argument("--quiet", action="store_true")
    return p.parse_args()


def _find_fip(signal_eval: list[dict]) -> dict | None:
    for row in signal_eval or []:
        if row.get("signal") == "fip_id":
            return row
    return None


def _verdict(row: dict | None) -> dict:
    if row is None:
        return {
            "green": False,
            "reason": "fip_id missing from signal_eval",
        }
    mean_ic = row.get("mean_ic")
    t_stat = row.get("ic_t_stat")
    reliable = bool(row.get("reliable"))
    if mean_ic is None or t_stat is None:
        return {"green": False, "reason": "missing mean_ic or ic_t_stat", "row": row}
    sign_ok = mean_ic < 0
    mag_ok = abs(float(mean_ic)) >= 0.03
    green = sign_ok and mag_ok and reliable
    return {
        "green": green,
        "reason": (
            "iron rule cleared — follow-up proposal only, not production wire-in"
            if green
            else "iron rule not met on liquid-breadth cross-section"
        ),
        "checks": {
            "mean_ic": mean_ic,
            "abs_mean_ic_ge_0_03": mag_ok,
            "sign_negative": sign_ok,
            "ic_t_stat": t_stat,
            "reliable": reliable,
            "weeks": row.get("weeks"),
            "avg_cross_section": row.get("avg_cross_section"),
        },
        "row": row,
    }


async def _run_signal_eval(snapshot: Path, *, workers: int, quiet: bool) -> dict:
    from app.config import settings
    from app.services.backtest_service import run_backtest

    settings.backtest_workers = workers
    engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True)
    Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

    def progress(done: int, total: int, symbol: str) -> None:
        if quiet:
            return
        print(f"  progress {done}/{total} {symbol}", end="\r")

    try:
        async with Session() as db:
            report = await run_backtest(db, progress_cb=progress, cadence="weekly")
    finally:
        await engine.dispose()
    if not quiet:
        print()
    return report


def _write_md(path: Path, payload: dict) -> None:
    fp = payload.get("fingerprint") or {}
    br = payload.get("breadth") or {}
    v = payload.get("verdict") or {}
    lines = [
        "# Broad-universe fip_id IC research (Phase B)",
        "",
        f"Generated: {payload.get('generated_at')}",
        "",
        "## Scope",
        "",
        "- **Research only** — production universe, gate, scanner, schedule unchanged.",
        "- Price-only signal harness; no sentiment/fundamentals on the broad tier.",
        "- Point-in-time liquidity mask: top "
        f"**{payload.get('liquid_breadth_top_n')}** by 63d median $vol, "
        f"price ≥ **${payload.get('liquid_min_price')}** at as-of.",
        "",
        "## Caveats",
        "",
        "- **Survivorship bias**: today's constituents backfilled historically "
        "(worse in small caps).",
        "- **IEX volume undercount**: relative $vol rank only, not absolute floors.",
        "- **Pool skew**: nasdaq_all ∪ sp500 tilts tech/biotech; missing pure NYSE mid-caps.",
        "",
        "## Fingerprint (505-name prod snapshot)",
        "",
        f"- Expected: IC ≈ {FINGERPRINT_IC}, t ≈ {FINGERPRINT_T}",
        f"- Observed: IC = {fp.get('mean_ic')}, t = {fp.get('ic_t_stat')}, "
        f"weeks = {fp.get('weeks')}, reliable = {fp.get('reliable')}",
        f"- Pass: **{fp.get('pass')}**",
        "",
        "## Liquid-breadth signal_eval (fip_id)",
        "",
    ]
    row = br.get("row") or br
    if row:
        lines.extend([
            "| metric | value |",
            "|---|---|",
            f"| mean_ic | {row.get('mean_ic')} |",
            f"| ic_t_stat | {row.get('ic_t_stat')} |",
            f"| ic_positive_pct | {row.get('ic_positive_pct')} |",
            f"| weeks | {row.get('weeks')} |",
            f"| avg_cross_section | {row.get('avg_cross_section')} |",
            f"| reliable | {row.get('reliable')} |",
            f"| mean_quintile_spread | {row.get('mean_quintile_spread')} |",
            "",
        ])
    else:
        lines.append("_No breadth result (run skipped or failed)._")
        lines.append("")
    lines.extend([
        "## Verdict (iron rule)",
        "",
        f"- **Green: {v.get('green')}**",
        f"- {v.get('reason')}",
        f"- Checks: `{json.dumps(v.get('checks') or {}, default=str)}`",
        "",
        "A green verdict authorizes a **follow-up proposal** only "
        "(two-tier universe / gate revalidation) — **not** production wire-in.",
        "",
        "## Artifacts",
        "",
        f"- Fingerprint report: `{payload.get('fingerprint_report_path')}`",
        f"- Breadth report: `{payload.get('breadth_report_path')}`",
        "",
    ])
    path.write_text("\n".join(lines), encoding="utf-8")


async def _main() -> None:
    args = _parse_args()
    prod = Path(args.prod_snapshot)
    research = Path(args.research_snapshot)
    if not prod.exists():
        raise SystemExit(f"Prod snapshot missing: {prod}")

    os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1"
    if args.allow_spawn:
        os.environ["BACKTEST_ALLOW_SPAWN"] = "1"
    os.environ["BACKTEST_SIGNAL_EVAL_ONLY"] = "1"

    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    out_json = Path(args.out) if args.out else Path("reports") / f"fip-breadth-{stamp}.json"
    out_json.parent.mkdir(parents=True, exist_ok=True)
    # Never clobber the curated research log (docs/research/fip-breadth-ic.md).
    # Machine summary goes next to the JSON report only.
    out_md = out_json.with_suffix(".md")

    payload: dict = {
        "generated_at": datetime.now().isoformat(),
        "liquid_breadth_top_n": args.liquid_breadth,
        "liquid_min_price": args.min_price,
        "fingerprint": None,
        "breadth": None,
        "verdict": None,
    }

    # --- 1) Fingerprint ---
    if not args.skip_fingerprint:
        # Clear liquid breadth for fingerprint
        os.environ.pop("BACKTEST_LIQUID_BREADTH", None)
        os.environ.pop("BACKTEST_LIQUID_MIN_PRICE", None)
        if not args.quiet:
            print(f"Fingerprint run on {prod}…")
        fp_report = await _run_signal_eval(prod, workers=args.workers, quiet=args.quiet)
        fp_path = out_json.with_name(out_json.stem + "-fingerprint.json")
        fp_path.write_text(json.dumps(fp_report, indent=2, default=str), encoding="utf-8")
        fip = _find_fip(fp_report.get("signal_eval") or [])
        if fip is None:
            raise SystemExit("ABORT: fip_id missing from fingerprint signal_eval")
        ic_ok = abs(float(fip["mean_ic"]) - FINGERPRINT_IC) <= FINGERPRINT_IC_TOL
        t_ok = abs(float(fip["ic_t_stat"]) - FINGERPRINT_T) <= FINGERPRINT_T_TOL
        passed = ic_ok and t_ok and bool(fip.get("reliable"))
        payload["fingerprint"] = {
            **fip,
            "pass": passed,
            "expected_ic": FINGERPRINT_IC,
            "expected_t": FINGERPRINT_T,
        }
        payload["fingerprint_report_path"] = str(fp_path)
        if not args.quiet:
            print(
                f"Fingerprint fip_id IC={fip.get('mean_ic')} t={fip.get('ic_t_stat')} "
                f"pass={passed}"
            )
        if not passed:
            out_json.write_text(json.dumps(payload, indent=2, default=str), encoding="utf-8")
            raise SystemExit(
                "ABORT: fingerprint mismatch — investigate before trusting breadth runs "
                f"(got IC={fip.get('mean_ic')} t={fip.get('ic_t_stat')})"
            )

    # --- 2) Breadth ---
    if not args.skip_research:
        # Refuse half-built research.sqlite (2026-07-18 21:14 race).
        scripts_dir = Path(__file__).resolve().parent
        if str(scripts_dir) not in sys.path:
            sys.path.insert(0, str(scripts_dir))
        from research_snapshot_manifest import (  # type: ignore[import-not-found]
            assert_research_snapshot_complete,
        )

        manifest = assert_research_snapshot_complete(research)
        payload["research_snapshot_manifest"] = {
            "finished_at": manifest.get("finished_at"),
            "ticker_count": manifest.get("ticker_count"),
            "ohlcv_row_count": manifest.get("ohlcv_row_count"),
            "rank_only_count": manifest.get("rank_only_count"),
            "complete": manifest.get("complete"),
        }
        os.environ["BACKTEST_LIQUID_BREADTH"] = str(int(args.liquid_breadth))
        os.environ["BACKTEST_LIQUID_MIN_PRICE"] = str(float(args.min_price))
        if not args.quiet:
            print(
                f"Breadth run on {research} "
                f"(top {args.liquid_breadth}, min_price={args.min_price}; "
                f"manifest ok tickers={manifest.get('ticker_count')} "
                f"finished_at={manifest.get('finished_at')})…"
            )
        br_report = await _run_signal_eval(
            research, workers=args.workers, quiet=args.quiet
        )
        br_path = out_json.with_name(out_json.stem + "-breadth.json")
        br_path.write_text(json.dumps(br_report, indent=2, default=str), encoding="utf-8")
        fip_b = _find_fip(br_report.get("signal_eval") or [])
        payload["breadth"] = fip_b or {"error": "fip_id missing"}
        payload["breadth_report_path"] = str(br_path)
        payload["breadth_tickers"] = br_report.get("tickers")
        payload["breadth_rank_only_tickers"] = br_report.get("rank_only_tickers")
        payload["verdict"] = _verdict(fip_b)
        if not args.quiet:
            print(
                f"Breadth fip_id IC={ (fip_b or {}).get('mean_ic') } "
                f"t={ (fip_b or {}).get('ic_t_stat') } "
                f"green={payload['verdict'].get('green')}"
            )

    out_json.write_text(json.dumps(payload, indent=2, default=str), encoding="utf-8")
    out_md.parent.mkdir(parents=True, exist_ok=True)
    _write_md(out_md, payload)
    if not args.quiet:
        print(f"Wrote {out_json}")
        print(f"Wrote {out_md}")


if __name__ == "__main__":
    asyncio.run(_main())