feat: Phase B fip_id liquid-breadth research tooling

Add research-only snapshot extender, PIT dollar-volume mask for signal IC,
rank-only harness path, fingerprint+breadth runner, and docs. Fingerprint
reproduced IC -0.045 / t -2.91 on prod.sqlite. No production gate/schedule changes.

scripts/extend_snapshot_universe.py

@@ -0,0 +1,329 @@
+"""Extend a *copy* of the production backtest snapshot with broad-universe OHLCV.
+
+Research only — never writes to production Postgres.
+
+Pipeline
+--------
+1. Copy ``--source`` snapshot (default ``backtest_snapshots/prod.sqlite``) to
+   ``--output`` (default ``backtest_snapshots/research.sqlite``).
+2. Resolve symbol pool = nasdaq_all ∪ sp500 via ``ticker_universe_service``.
+3. Fetch ~5y daily bars from Alpaca for symbols missing (or short) in the copy.
+4. Insert new tickers + OHLCV; mark them in side table ``research_rank_only``
+   so the harness can feed signal IC without GTL/candidate replay.
+
+Resume-friendly: re-running skips symbols that already have ≥ ``--min-bars``.
+
+Example
+-------
+  python scripts/extend_snapshot_universe.py \\
+      --source backtest_snapshots/prod.sqlite \\
+      --output backtest_snapshots/research.sqlite \\
+      --force-copy
+
+  # smoke: first 50 missing symbols only
+  python scripts/extend_snapshot_universe.py --limit 50
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import shutil
+import sys
+import time
+from datetime import date, datetime, timedelta, timezone
+from pathlib import Path
+
+from sqlalchemy import create_engine, select, text
+from sqlalchemy.orm import Session
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+
+def _parse_args() -> argparse.Namespace:
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument(
+        "--source",
+        default="backtest_snapshots/prod.sqlite",
+        help="Existing prod snapshot to copy (read-only after copy).",
+    )
+    p.add_argument(
+        "--output",
+        default="backtest_snapshots/research.sqlite",
+        help="Research snapshot path (created/updated).",
+    )
+    p.add_argument(
+        "--force-copy",
+        action="store_true",
+        help="Overwrite output by re-copying from source first.",
+    )
+    p.add_argument(
+        "--history-days",
+        type=int,
+        default=1825,
+        help="OHLCV lookback days (~5y). Default 1825.",
+    )
+    p.add_argument(
+        "--min-bars",
+        type=int,
+        default=260,
+        help="Skip re-fetch when a symbol already has this many bars.",
+    )
+    p.add_argument(
+        "--limit",
+        type=int,
+        default=None,
+        help="Max *new* symbols to fetch (smoke tests).",
+    )
+    p.add_argument(
+        "--sleep",
+        type=float,
+        default=0.15,
+        help="Seconds between Alpaca symbol requests (rate-limit cushion).",
+    )
+    p.add_argument(
+        "--max-retries",
+        type=int,
+        default=5,
+        help="Retries per symbol on RateLimitError.",
+    )
+    p.add_argument("--quiet", action="store_true")
+    return p.parse_args()
+
+
+def _ensure_rank_only_table(conn) -> None:
+    conn.execute(
+        text(
+            """
+            CREATE TABLE IF NOT EXISTS research_rank_only (
+                ticker_id INTEGER PRIMARY KEY,
+                symbol TEXT NOT NULL UNIQUE
+            )
+            """
+        )
+    )
+    conn.commit()
+
+
+async def _resolve_pool() -> tuple[list[str], dict[str, str]]:
+    """Return sorted unique symbols and source labels."""
+    from app.database import async_session_factory
+    from app.services.ticker_universe_service import fetch_universe_symbols
+
+    sources: dict[str, str] = {}
+    symbols: set[str] = set()
+    # Need a DB session for cache writes; use local async engine if configured,
+    # but public/FMP fetch works with any session. Prefer a throwaway sqlite.
+    from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+    try:
+        async with Session() as db:
+            for universe in ("nasdaq_all", "sp500"):
+                try:
+                    syms, src = await fetch_universe_symbols(db, universe)
+                except Exception as exc:
+                    print(f"WARNING: universe {universe} failed: {exc}")
+                    continue
+                sources[universe] = src
+                symbols.update(syms)
+                print(f"  {universe}: {len(syms)} symbols (source={src})")
+    finally:
+        await engine.dispose()
+    return sorted(symbols), sources
+
+
+async def _fetch_symbol_bars(
+    provider,
+    symbol: str,
+    start: date,
+    end: date,
+    *,
+    max_retries: int,
+    sleep_s: float,
+) -> list:
+    from app.exceptions import ProviderError, RateLimitError
+
+    for attempt in range(max_retries):
+        try:
+            bars = await provider.fetch_ohlcv(symbol, start, end)
+            if sleep_s > 0:
+                await asyncio.sleep(sleep_s)
+            return bars
+        except RateLimitError:
+            wait = min(60.0, 2.0 ** attempt)
+            print(f"  rate limited on {symbol}; sleep {wait:.0f}s")
+            await asyncio.sleep(wait)
+        except ProviderError as exc:
+            if attempt + 1 >= max_retries:
+                raise
+            await asyncio.sleep(1.0)
+            _ = exc
+    return []
+
+
+async def _main() -> None:
+    args = _parse_args()
+    source = Path(args.source)
+    output = Path(args.output)
+    if not source.exists():
+        raise SystemExit(f"Source snapshot not found: {source}")
+
+    if args.force_copy or not output.exists():
+        output.parent.mkdir(parents=True, exist_ok=True)
+        if output.exists():
+            output.unlink()
+        print(f"Copying {source} → {output}")
+        shutil.copy2(source, output)
+    else:
+        print(f"Updating existing research snapshot: {output}")
+
+    from app.config import settings
+    from app.models.ohlcv import OHLCVRecord
+    from app.models.ticker import Ticker
+    from app.providers.alpaca import AlpacaOHLCVProvider
+
+    if not settings.alpaca_api_key or not settings.alpaca_api_secret:
+        raise SystemExit("ALPACA_API_KEY / ALPACA_API_SECRET required in .env")
+
+    provider = AlpacaOHLCVProvider(settings.alpaca_api_key, settings.alpaca_api_secret)
+    end = date.today()
+    start = end - timedelta(days=int(args.history_days))
+
+    print("Resolving universe pool (nasdaq_all ∪ sp500)…")
+    pool, sources = await _resolve_pool()
+    print(f"Pool size: {len(pool)} (sources={sources})")
+
+    # Sync sqlite via sqlalchemy core (simpler than async for bulk insert)
+    engine = create_engine(f"sqlite:///{output.resolve().as_posix()}")
+    with Session(engine) as session:
+        _ensure_rank_only_table(session.connection())
+        existing = {
+            row.symbol: row
+            for row in session.execute(select(Ticker)).scalars().all()
+        }
+        prod_symbols = set(existing)
+
+        # Bar counts
+        bar_counts: dict[str, int] = {}
+        for sym, ticker in existing.items():
+            n = session.execute(
+                text("SELECT COUNT(*) FROM ohlcv_records WHERE ticker_id = :tid"),
+                {"tid": ticker.id},
+            ).scalar_one()
+            bar_counts[sym] = int(n)
+
+        to_fetch: list[str] = []
+        for sym in pool:
+            if sym in existing and bar_counts.get(sym, 0) >= args.min_bars:
+                # Existing production or previously extended — keep rank_only
+                # only for *new* research names, not original prod universe.
+                continue
+            to_fetch.append(sym)
+
+        if args.limit is not None:
+            to_fetch = to_fetch[: max(0, int(args.limit))]
+
+        print(f"Symbols to fetch/extend: {len(to_fetch)}")
+        ok = 0
+        fail = 0
+        t0 = time.monotonic()
+        for index, sym in enumerate(to_fetch, 1):
+            try:
+                bars = await _fetch_symbol_bars(
+                    provider,
+                    sym,
+                    start,
+                    end,
+                    max_retries=args.max_retries,
+                    sleep_s=args.sleep,
+                )
+            except Exception as exc:
+                fail += 1
+                if not args.quiet:
+                    print(f"  [{index}/{len(to_fetch)}] {sym} FAIL {exc}")
+                continue
+
+            if not bars:
+                fail += 1
+                if not args.quiet:
+                    print(f"  [{index}/{len(to_fetch)}] {sym} empty")
+                continue
+
+            ticker = existing.get(sym)
+            is_new = ticker is None
+            if ticker is None:
+                ticker = Ticker(symbol=sym, name=None, created_at=datetime.now(timezone.utc))
+                session.add(ticker)
+                session.flush()
+                existing[sym] = ticker
+
+            # Upsert bars (delete+insert range for simplicity on research path)
+            session.execute(
+                text(
+                    "DELETE FROM ohlcv_records WHERE ticker_id = :tid "
+                    "AND date >= :start AND date <= :end"
+                ),
+                {"tid": ticker.id, "start": start.isoformat(), "end": end.isoformat()},
+            )
+            now = datetime.utcnow()
+            session.bulk_insert_mappings(
+                OHLCVRecord,
+                [
+                    {
+                        "ticker_id": ticker.id,
+                        "date": b.date,
+                        "open": b.open,
+                        "high": b.high,
+                        "low": b.low,
+                        "close": b.close,
+                        "volume": b.volume,
+                        "created_at": now,
+                    }
+                    for b in bars
+                ],
+            )
+
+            # rank_only only for names that were NOT in the original production
+            # snapshot at copy time (or are newly introduced to this research DB).
+            if is_new or sym not in prod_symbols:
+                # Re-evaluate: if source copy already had the symbol, don't flag.
+                # Only new inserts get rank_only.
+                if is_new:
+                    session.execute(
+                        text(
+                            "INSERT OR REPLACE INTO research_rank_only "
+                            "(ticker_id, symbol) VALUES (:tid, :sym)"
+                        ),
+                        {"tid": ticker.id, "sym": sym},
+                    )
+
+            session.commit()
+            ok += 1
+            if not args.quiet and (index % 25 == 0 or index == len(to_fetch)):
+                elapsed = time.monotonic() - t0
+                print(
+                    f"  progress {index}/{len(to_fetch)} ok={ok} fail={fail} "
+                    f"elapsed={elapsed/60:.1f}m last={sym} bars={len(bars)}"
+                )
+
+        rank_only_n = session.execute(
+            text("SELECT COUNT(*) FROM research_rank_only")
+        ).scalar_one()
+        ticker_n = session.execute(text("SELECT COUNT(*) FROM tickers")).scalar_one()
+        ohlcv_n = session.execute(text("SELECT COUNT(*) FROM ohlcv_records")).scalar_one()
+
+    print("Done.")
+    print(f"  output: {output}")
+    print(f"  tickers: {ticker_n}")
+    print(f"  ohlcv rows: {ohlcv_n}")
+    print(f"  research_rank_only: {rank_only_n}")
+    print(f"  fetched ok/fail: {ok}/{fail}")
+
+
+if __name__ == "__main__":
+    asyncio.run(_main())

scripts/run_fip_breadth_research.py

@@ -0,0 +1,299 @@
+"""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. Run signal_eval on research.sqlite with BACKTEST_LIQUID_BREADTH=1500 PIT mask.
+3. 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([
+            f"| metric | value |",
+            f"|---|---|",
+            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)
+    out_md = Path("docs/research") / "fip-breadth-ic.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:
+        if not research.exists():
+            raise SystemExit(
+                f"Research snapshot missing: {research}\n"
+                "Build it with: python scripts/extend_snapshot_universe.py"
+            )
+        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})…"
+            )
+        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())