From a9841d92b7d9aa1f506bfd18b462999e0dfb0234 Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Sun, 19 Jul 2026 10:58:03 +0200 Subject: [PATCH] research: sector-resid deep test (deepen shallow + one masked PASS/FAIL) Terminal follow-up for Task 1: detect/refetch shallow two-tier symbols and sector ETFs at 5000d, regenerate manifest, run ONE liquid-1500 harness with era split, grade mom_12_1_sector_resid mechanically. Bundled as run_tier1_macbook.sh --sector-resid-deep. --- scripts/run_sector_resid_deep_test.py | 998 ++++++++++++++++++++++++++ scripts/run_tier1_macbook.sh | 19 +- 2 files changed, 1016 insertions(+), 1 deletion(-) create mode 100644 scripts/run_sector_resid_deep_test.py diff --git a/scripts/run_sector_resid_deep_test.py b/scripts/run_sector_resid_deep_test.py new file mode 100644 index 0000000..54de8bf --- /dev/null +++ b/scripts/run_sector_resid_deep_test.py @@ -0,0 +1,998 @@ +#!/usr/bin/env python3 +"""Terminal sector-residual deep test: deepen shallow symbols → ONE masked run → PASS/FAIL. + +Repairs the two-tier history-depth defect (prod/ETF names left at ~5y while breadth +got 5000d), then runs a single liquid-breadth signal harness and grades +``mom_12_1_sector_resid`` against the pre-registered rule. + +Local research only. No production changes. + +MacBook +------- + # On deep research.sqlite from the prior history-depth rebuild: + python scripts/run_sector_resid_deep_test.py \\ + --snapshot backtest_snapshots/research.sqlite \\ + --workers 8 --allow-spawn + + # Skip re-fetch if Step-1 already done and sanity-check passes: + python scripts/run_sector_resid_deep_test.py --skip-deepen --workers 8 --allow-spawn +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import math +import os +import sys +import time +from collections import defaultdict +from datetime import date, datetime, timedelta, timezone +from pathlib import Path +from typing import Any + +from sqlalchemy import create_engine, text +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)) + +from app.ssl_bootstrap import bootstrap_ssl # noqa: E402 + +bootstrap_ssl() + +from app.services.sector_map import ( # noqa: E402 + DEFAULT_SECTOR_MAP_PATH, + SECTOR_ETFS, + load_ticker_sector_map, +) +from scripts.research_snapshot_manifest import ( # noqa: E402 + assert_research_snapshot_complete, + clear_manifest, + write_completion_manifest, +) + +ERA_SPLIT = date(2021, 1, 1) +IRON_IC = 0.03 +# "weeks ≫ 35 (expect ~80)" — mechanical floor for "data fix worked" +MIN_WEEKS_DEEP = 50 +SANITY_MEGACAPS = ("AAPL", "MSFT", "JPM", "XOM", "JNJ") +SURVIVORSHIP = ( + "SURVIVORSHIP BIAS: today's constituents backfilled. Relative IC only — not levels." +) + + +def _parse_args() -> argparse.Namespace: + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--snapshot", default="backtest_snapshots/research.sqlite") + p.add_argument("--history-days", type=int, default=5000) + p.add_argument("--sleep", type=float, default=0.15) + p.add_argument("--workers", type=int, default=8) + p.add_argument("--allow-spawn", action="store_true") + p.add_argument( + "--skip-deepen", + action="store_true", + help="Skip Step-1 re-fetch; only sanity-check + harness.", + ) + p.add_argument( + "--sector-map", + default=str(DEFAULT_SECTOR_MAP_PATH), + ) + p.add_argument("--liquid-breadth", type=int, default=1500) + p.add_argument("--min-price", type=float, default=5.0) + p.add_argument("--quiet", action="store_true") + p.add_argument("--out", default=None) + return p.parse_args() + + +def _sqlite_url(path: Path) -> str: + return f"sqlite+aiosqlite:///{path.resolve().as_posix()}" + + +def _symbol_depth(snapshot: Path) -> list[dict[str, Any]]: + engine = create_engine( + f"sqlite:///{snapshot.resolve().as_posix()}", + future=True, + ) + try: + with engine.connect() as conn: + rows = conn.execute( + text( + """ + SELECT t.symbol, COUNT(*) AS n, MIN(o.date), MAX(o.date) + FROM ohlcv_records o + JOIN tickers t ON t.id = o.ticker_id + GROUP BY t.symbol + """ + ) + ).fetchall() + finally: + engine.dispose() + out = [] + for sym, n, d0, d1 in rows: + out.append({ + "symbol": str(sym).upper(), + "bars": int(n), + "min_date": str(d0)[:10] if d0 else None, + "max_date": str(d1)[:10] if d1 else None, + }) + return out + + +def _derive_shallow( + depths: list[dict[str, Any]], + *, + lag_days: int = 400, +) -> tuple[list[str], dict[str, Any]]: + """Symbols whose earliest bar starts materially later than the deep cohort.""" + starts: list[tuple[str, date]] = [] + for row in depths: + if not row.get("min_date"): + continue + starts.append((row["symbol"], date.fromisoformat(row["min_date"]))) + if not starts: + return [], {"error": "no symbols with min_date"} + + # Deep cohort start ≈ 10th percentile of earliest dates (early = deep). + ordered = sorted(d for _, d in starts) + p10 = ordered[max(0, int(0.10 * (len(ordered) - 1)))] + cutoff = p10 + timedelta(days=lag_days) + shallow = sorted({sym for sym, d in starts if d > cutoff}) + meta = { + "n_symbols": len(starts), + "deep_cohort_p10_start": p10.isoformat(), + "shallow_cutoff": cutoff.isoformat(), + "lag_days": lag_days, + "n_shallow": len(shallow), + "shallow_start_histogram": _year_hist( + [d for sym, d in starts if sym in set(shallow)] + ), + "deep_start_histogram": _year_hist( + [d for sym, d in starts if sym not in set(shallow)] + ), + "shallow_sample": shallow[:30], + } + return shallow, meta + + +def _year_hist(dates: list[date]) -> dict[str, int]: + h: dict[str, int] = defaultdict(int) + for d in dates: + h[str(d.year)] += 1 + return dict(sorted(h.items())) + + +async def _fetch_and_replace_ohlcv( + engine, + provider, + symbol: str, + start: date, + end: date, + *, + sleep_s: float, + max_retries: int = 5, +) -> int: + from app.exceptions import ProviderError, RateLimitError + + bars = [] + for attempt in range(max_retries): + try: + bars = await provider.fetch_ohlcv(symbol, start, end) + break + except RateLimitError: + wait = min(60.0, 2.0 ** attempt) + print(f" rate limited {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 + if sleep_s > 0: + await asyncio.sleep(sleep_s) + if not bars: + return 0 + + with engine.begin() as write: + tid = write.execute( + text("SELECT id FROM tickers WHERE symbol = :s"), + {"s": symbol}, + ).scalar_one_or_none() + if tid is None: + write.execute( + text( + "INSERT INTO tickers (symbol, name, created_at) " + "VALUES (:s, NULL, :c)" + ), + {"s": symbol, "c": datetime.now(timezone.utc).isoformat()}, + ) + tid = write.execute( + text("SELECT id FROM tickers WHERE symbol = :s"), + {"s": symbol}, + ).scalar_one() + # Full replace for this symbol so shallow tails cannot linger. + write.execute( + text("DELETE FROM ohlcv_records WHERE ticker_id = :tid"), + {"tid": int(tid)}, + ) + now = datetime.now(timezone.utc).replace(tzinfo=None).isoformat() + write.execute( + text( + """ + INSERT INTO ohlcv_records + (ticker_id, date, open, high, low, close, volume, created_at) + VALUES + (:ticker_id, :date, :open, :high, :low, :close, :volume, :created_at) + """ + ), + [ + { + "ticker_id": int(tid), + "date": b.date.isoformat() + if hasattr(b.date, "isoformat") + else str(b.date), + "open": float(b.open), + "high": float(b.high), + "low": float(b.low), + "close": float(b.close), + "volume": int(b.volume), + "created_at": now, + } + for b in bars + ], + ) + return len(bars) + + +async def _deepen_sector_etfs( + snapshot: Path, *, history_days: int, sleep_s: float +) -> dict[str, Any]: + """Refresh SPY + 11 sector ETFs in benchmark_prices to full depth.""" + # Reuse the existing CLI helper for consistency. + from scripts.fetch_sector_etfs_to_snapshot import _fetch_and_upsert + from app.config import settings + from app.providers.alpaca import AlpacaOHLCVProvider + + if not settings.alpaca_api_key or not settings.alpaca_api_secret: + raise SystemExit("ALPACA keys required to deepen sector ETFs") + + provider = AlpacaOHLCVProvider(settings.alpaca_api_key, settings.alpaca_api_secret) + end = date.today() + start = end - timedelta(days=history_days) + engine = create_engine( + f"sqlite:///{snapshot.resolve().as_posix()}", + future=True, + ) + symbols = ["SPY", *SECTOR_ETFS] + written: dict[str, int] = {} + try: + for sym in symbols: + n = await _fetch_and_upsert( + engine, provider, sym, start, end, sleep_s=sleep_s + ) + written[sym] = n + finally: + engine.dispose() + + engine = create_engine( + f"sqlite:///{snapshot.resolve().as_posix()}", + future=True, + ) + try: + with engine.connect() as conn: + rows = conn.execute( + text( + "SELECT symbol, COUNT(*), MIN(date), MAX(date) " + "FROM benchmark_prices GROUP BY symbol ORDER BY symbol" + ) + ).fetchall() + finally: + engine.dispose() + return { + "written": written, + "benchmark_summary": [ + {"symbol": s, "n": n, "min": d0, "max": d1} for s, n, d0, d1 in rows + ], + } + + +def _sanity_check(snapshot: Path, *, history_days: int) -> dict[str, Any]: + depths = {r["symbol"]: r for r in _symbol_depth(snapshot)} + end = date.today() + target_start = end - timedelta(days=history_days) + # Allow ~1 year slack for IPO/listing limits (not a hard fail for all names). + megacap_deadline = target_start + timedelta(days=400) + + megacap = {} + ok_mega = True + for sym in SANITY_MEGACAPS: + row = depths.get(sym) + megacap[sym] = row + if row is None or not row.get("min_date"): + ok_mega = False + continue + if date.fromisoformat(row["min_date"]) > megacap_deadline: + ok_mega = False + + engine = create_engine( + f"sqlite:///{snapshot.resolve().as_posix()}", + future=True, + ) + try: + with engine.connect() as conn: + etf_rows = conn.execute( + text( + "SELECT symbol, COUNT(*), MIN(date), MAX(date) " + "FROM benchmark_prices WHERE symbol IN " + f"({','.join(repr(s) for s in SECTOR_ETFS)}) " + "GROUP BY symbol" + ) + ).fetchall() + finally: + engine.dispose() + + etf_info = { + s: {"n": n, "min": d0, "max": d1} for s, n, d0, d1 in etf_rows + } + deep_etfs = 0 + for sym in SECTOR_ETFS: + info = etf_info.get(sym) + if not info or not info["min"]: + continue + # XLC lists mid-2018 — accept that floor. + floor = date(2018, 6, 1) if sym == "XLC" else megacap_deadline + if date.fromisoformat(str(info["min"])[:10]) <= floor + timedelta(days=60): + deep_etfs += 1 + elif date.fromisoformat(str(info["min"])[:10]) <= date(2019, 1, 1): + # moderately deep still counts for non-XLC if near 2018-19 + if sym != "XLC": + deep_etfs += 1 + + # Require 10 of 11 sector ETFs deep (XLC may be the exception with mid-2018 start). + ok_etf = deep_etfs >= 10 + + # Shallow residual after deepen: few names should still start after 2020. + still_shallow, _ = _derive_shallow(list(depths.values()), lag_days=400) + # After fix, shallow set should shrink dramatically vs ~500. + note_xlc = ( + "XLC lists mid-2018 → Communication Services residual coverage from ~mid-2019." + ) + + passed = bool(ok_mega and ok_etf) + return { + "passed": passed, + "megacap": megacap, + "megacap_ok": ok_mega, + "megacap_deadline": megacap_deadline.isoformat(), + "sector_etfs": etf_info, + "sector_etfs_deep_count": deep_etfs, + "sector_etfs_ok": ok_etf, + "still_shallow_count": len(still_shallow), + "still_shallow_sample": still_shallow[:20], + "xlc_note": note_xlc, + "target_history_days": history_days, + } + + +async def _step1_deepen( + snapshot: Path, + *, + history_days: int, + sleep_s: float, + quiet: bool, +) -> dict[str, Any]: + from app.config import settings + 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") + + clear_manifest(snapshot) + depths = _symbol_depth(snapshot) + shallow, shallow_meta = _derive_shallow(depths) + print( + f"Shallow symbols to deepen: {len(shallow)} " + f"(p10 deep start={shallow_meta.get('deep_cohort_p10_start')}, " + f"cutoff={shallow_meta.get('shallow_cutoff')})" + ) + if not shallow: + print("WARNING: no shallow symbols detected — snapshot may already be uniform") + + provider = AlpacaOHLCVProvider(settings.alpaca_api_key, settings.alpaca_api_secret) + end = date.today() + start = end - timedelta(days=history_days) + engine = create_engine( + f"sqlite:///{snapshot.resolve().as_posix()}", + future=True, + ) + + ok = fail = 0 + t0 = time.monotonic() + try: + for i, sym in enumerate(shallow, 1): + try: + n = await _fetch_and_replace_ohlcv( + engine, provider, sym, start, end, sleep_s=sleep_s + ) + if n <= 0: + fail += 1 + if not quiet: + print(f" [{i}/{len(shallow)}] {sym} empty") + continue + ok += 1 + if not quiet and (i % 25 == 0 or i == len(shallow)): + print( + f" progress {i}/{len(shallow)} ok={ok} fail={fail} " + f"last={sym} bars={n} elapsed={(time.monotonic()-t0)/60:.1f}m" + ) + except Exception as exc: + fail += 1 + print(f" [{i}/{len(shallow)}] {sym} FAIL {exc}") + finally: + engine.dispose() + + print("Deepening SPY + sector ETFs in benchmark_prices…") + etf_result = await _deepen_sector_etfs( + snapshot, history_days=history_days, sleep_s=sleep_s + ) + + # Manifest: full completion after deepen (no --limit). + from scripts.research_snapshot_manifest import _count_snapshot + + counts = _count_snapshot(snapshot) + manifest_path = write_completion_manifest( + snapshot, + complete=True, + sources={"deepen": "sector_resid_deep_test step1"}, + history_days=history_days, + min_bars=None, + fetch_ok=ok, + fetch_fail=fail, + limit=None, + extra={ + "shallow_meta": shallow_meta, + "shallow_fetched_ok": ok, + "shallow_fetched_fail": fail, + "etf_refresh": etf_result.get("written"), + "counts_after": counts, + }, + ) + print(f"Manifest written: {manifest_path}") + + sanity = _sanity_check(snapshot, history_days=history_days) + return { + "shallow_meta": shallow_meta, + "shallow_list_n": len(shallow), + "fetch_ok": ok, + "fetch_fail": fail, + "etf_refresh": etf_result, + "sanity": sanity, + "manifest_path": str(manifest_path), + } + + +async def _one_masked_run( + snapshot: Path, + *, + sector_map_path: Path, + liquid_breadth: int, + min_price: float, + workers: int, + quiet: bool, +) -> dict[str, Any]: + """Single collection under liquid mask; full + era IC from the same series.""" + from app.config import settings + from app.services import backtest_service as bt + from app.services.benchmark_service import load_benchmark_closes + from app.models.ticker import Ticker + from sqlalchemy import select + + os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1" + os.environ["BACKTEST_SIGNAL_EVAL_ONLY"] = "1" + os.environ["BACKTEST_LIQUID_BREADTH"] = str(int(liquid_breadth)) + os.environ["BACKTEST_LIQUID_MIN_PRICE"] = str(float(min_price)) + os.environ["BACKTEST_SECTOR_MAP_PATH"] = str(sector_map_path.resolve()) + if workers: + settings.backtest_workers = workers + + # One collection pass (not run_backtest twice). + engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True) + Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + collected: dict = defaultdict(lambda: defaultdict(list)) + symbol_to_sector = load_ticker_sector_map(sector_map_path) + + try: + async with Session() as db: + tickers = list( + (await db.execute(select(Ticker).order_by(Ticker.symbol))).scalars() + ) + spy = await load_benchmark_closes(db, "SPY") + sector_etf: dict[str, dict] = {} + for etf in SECTOR_ETFS: + series = await load_benchmark_closes(db, etf) + if series: + sector_etf[etf] = series + + total = len(tickers) + for idx, t in enumerate(tickers): + if not quiet and idx % 100 == 0: + print(f" collect {idx}/{total}", end="\r", flush=True) + cols = await bt._fetch_columns(db, t.symbol) + if cols is None: + continue + records = [ + type( + "R", + (), + { + "date": date.fromordinal(int(cols[0][i])), + "close": cols[4][i], + "high": cols[2][i], + "volume": cols[5][i] if len(cols) > 5 else 0, + }, + )() + for i in range(len(cols[0])) + ] + etf_closes = bt._sector_etf_closes_for_symbol( + t.symbol, symbol_to_sector, sector_etf + ) + series = bt._signal_series( + records, + spy, + symbol=t.symbol, + sector_etf_closes=etf_closes, + ) + for name, weeks in series.items(): + for wk, pairs in weeks.items(): + collected[name][wk].extend(pairs) + finally: + await engine.dispose() + if not quiet: + print() + + if symbol_to_sector: + bt._inject_sector_demeaned_momentum(collected, symbol_to_sector) + + full_eval = bt._signal_evaluation(dict(collected)) + + def _filter_era(coll: dict, *, pre: bool) -> dict: + out: dict = defaultdict(lambda: defaultdict(list)) + for name, weeks in coll.items(): + for wk, recs in weeks.items(): + year = int(wk[0]) if isinstance(wk, tuple) else int(str(wk)[:4]) + if pre and year >= ERA_SPLIT.year: + continue + if not pre and year < ERA_SPLIT.year: + continue + out[name][wk].extend(recs) + return out + + pre_eval = bt._signal_evaluation(_filter_era(collected, pre=True)) + post_eval = bt._signal_evaluation(_filter_era(collected, pre=False)) + + # Identical-subset: resid IC only where sector_resid exists (same CS for t rule). + identical = _identical_subset_eval(collected, bt) + + def _idx(rows: list[dict]) -> dict[str, dict]: + return {r["signal"]: r for r in rows} + + # Mask bind diagnostics from liquid-aware rows if present. + mask_diag = _mask_diagnostics(full_eval) + + return { + "liquid_breadth_top_n": liquid_breadth, + "liquid_min_price": min_price, + "survivorship_banner": SURVIVORSHIP, + "signal_eval": full_eval, + "signal_eval_by_name": _idx(full_eval), + "era_split": { + "era_split_date": ERA_SPLIT.isoformat(), + "note": "Diagnostic only — not a tuning input.", + "pre_2021": _idx(pre_eval), + "post_2021": _idx(post_eval), + }, + "identical_subset_sector_cs": identical, + "mask_diagnostics": mask_diag, + "sector_map_size": len(symbol_to_sector), + "sector_etfs_loaded": sorted(sector_etf), + "spy_bars": len(spy), + } + + +def _identical_subset_eval(collected: dict, bt) -> dict[str, Any]: + """Re-score mom_12_1_resid on the same (week, symbol) cells as sector_resid.""" + sector_weeks = collected.get("mom_12_1_sector_resid") or {} + resid_weeks = collected.get("mom_12_1_resid") or {} + demean_weeks = collected.get("mom_12_1_sector_demeaned") or {} + mom_weeks = collected.get("mom_12_1") or {} + + restricted: dict = defaultdict(lambda: defaultdict(list)) + for wk, recs in sector_weeks.items(): + syms = set() + for rec in recs: + if isinstance(rec, dict) and rec.get("symbol"): + syms.add(str(rec["symbol"]).upper()) + restricted["mom_12_1_sector_resid"][wk].append(rec) + for name, source in ( + ("mom_12_1_resid", resid_weeks), + ("mom_12_1", mom_weeks), + ("mom_12_1_sector_demeaned", demean_weeks), + ): + for rec in source.get(wk) or []: + if not isinstance(rec, dict): + continue + sym = rec.get("symbol") + if sym and str(sym).upper() in syms: + restricted[name][wk].append(rec) + + rows = bt._signal_evaluation(dict(restricted)) + return {r["signal"]: r for r in rows} + + +def _mask_diagnostics(signal_eval: list[dict]) -> dict[str, Any]: + # Prefer a dense signal for mask stats. + for name in ("vol_6m", "mom_12_1", "fip_id"): + for row in signal_eval: + if row.get("signal") == name and row.get("mask_binds_pct") is not None: + return { + "reference_signal": name, + "avg_cross_section": row.get("avg_cross_section"), + "avg_raw_pool": row.get("avg_raw_pool"), + "avg_eligible_pre_mask": row.get("avg_eligible_pre_mask"), + "mask_binds_pct": row.get("mask_binds_pct"), + "weeks": row.get("weeks"), + } + # Fallback: any row with liquid fields + for row in signal_eval: + if row.get("liquid_breadth_top_n"): + return { + "reference_signal": row.get("signal"), + "avg_cross_section": row.get("avg_cross_section"), + "mask_binds_pct": row.get("mask_binds_pct"), + "weeks": row.get("weeks"), + } + return {"note": "no liquid mask diagnostics on rows (mask may be off)"} + + +def _grade(harness: dict[str, Any]) -> dict[str, Any]: + """Pre-registered PASS/FAIL for mom_12_1_sector_resid — mechanical.""" + by = harness.get("signal_eval_by_name") or {} + era = harness.get("era_split") or {} + identical = harness.get("identical_subset_sector_cs") or {} + + sector = by.get("mom_12_1_sector_resid") + # Prefer identical-subset resid for t comparison; fall back to full-table resid. + resid = identical.get("mom_12_1_resid") or by.get("mom_12_1_resid") + pre = (era.get("pre_2021") or {}).get("mom_12_1_sector_resid") + post = (era.get("post_2021") or {}).get("mom_12_1_sector_resid") + + checks: dict[str, Any] = { + "sector_row": sector, + "resid_row_for_t": resid, + "resid_t_source": ( + "identical_subset" if identical.get("mom_12_1_resid") else "full_table" + ), + "pre_2021": pre, + "post_2021": post, + } + + if sector is None: + return { + "verdict": "FAIL", + "reason": "mom_12_1_sector_resid missing from signal_eval", + "checks": checks, + "headline": "Task 1 CLOSED — sector residual dead on deep evidence.", + } + + mean_ic = sector.get("mean_ic") + t_stat = sector.get("ic_t_stat") + weeks = int(sector.get("weeks") or 0) + reliable = bool(sector.get("reliable")) + resid_t = resid.get("ic_t_stat") if resid else None + + mag_ok = mean_ic is not None and abs(float(mean_ic)) >= IRON_IC + sign_ok = mean_ic is not None and float(mean_ic) > 0 + reliable_ok = reliable and weeks >= 12 + weeks_ok = weeks >= MIN_WEEKS_DEEP + t_ok = ( + t_stat is not None + and resid_t is not None + and float(t_stat) >= float(resid_t) + ) + + pre_ic = pre.get("mean_ic") if pre else None + post_ic = post.get("mean_ic") if post else None + era_sign_ok = ( + pre_ic is not None + and post_ic is not None + and float(pre_ic) > 0 + and float(post_ic) > 0 + ) + # If pre era has no row, data fix failed for depth / era coverage. + era_present = pre is not None and post is not None + + checks.update({ + "abs_mean_ic_ge_0_03": mag_ok, + "sign_positive": sign_ok, + "reliable": reliable_ok, + "weeks_ge_50": weeks_ok, + "weeks": weeks, + "t_ge_resid_same_cs": t_ok, + "sector_t": t_stat, + "resid_t": resid_t, + "era_both_present": era_present, + "era_sign_consistent_positive": era_sign_ok, + "pre_ic": pre_ic, + "post_ic": post_ic, + "avg_cross_section": sector.get("avg_cross_section"), + }) + + if not weeks_ok: + return { + "verdict": "FAIL", + "reason": ( + f"weeks={weeks} did not extend (need ≥{MIN_WEEKS_DEEP}) — " + "data fix did not work or sector residual still shallow" + ), + "checks": checks, + "headline": "Task 1 CLOSED — sector residual dead on deep evidence.", + } + + passed = ( + mag_ok + and sign_ok + and reliable_ok + and weeks_ok + and t_ok + and era_present + and era_sign_ok + ) + if passed: + return { + "verdict": "PASS", + "reason": ( + "iron bar + weeks extended + t≥resid on same CS + era sign consistent" + ), + "checks": checks, + "headline": ( + "PROMOTE case strengthened — portfolio A/B is the next human decision." + ), + } + return { + "verdict": "FAIL", + "reason": "failed one or more pre-registered checks (see checks)", + "checks": checks, + "headline": "Task 1 CLOSED — sector residual dead on deep evidence.", + } + + +def _write_reports(payload: dict, out_json: Path, doc_path: Path) -> None: + out_json.parent.mkdir(parents=True, exist_ok=True) + out_json.write_text( + json.dumps(payload, indent=2, default=str) + "\n", encoding="utf-8" + ) + + grade = payload.get("grade") or {} + harness = payload.get("harness") or {} + by = harness.get("signal_eval_by_name") or {} + era = harness.get("era_split") or {} + lines = [ + "# Sector-residual deep test (masked, repaired snapshot)", + "", + f"Generated: `{payload.get('generated_at')}`", + "", + f"> **{SURVIVORSHIP}**", + "", + "## Pre-registered grade (mechanical)", + "", + f"**Verdict: {grade.get('verdict')}**", + "", + f"{grade.get('headline')}", + "", + f"Reason: {grade.get('reason')}", + "", + f"```json\n{json.dumps(grade.get('checks') or {}, indent=2, default=str)}\n```", + "", + "## Step-1 sanity", + "", + f"```json\n{json.dumps(payload.get('step1') or {}, indent=2, default=str)}\n```", + "", + "## Mask diagnostics", + "", + f"```json\n{json.dumps(harness.get('mask_diagnostics') or {}, indent=2, default=str)}\n```", + "", + "## Signal table (rows only — no narrative for non-sector signals)", + "", + "| signal | mean_ic | t | weeks | avg_N | reliable |", + "|---|---:|---:|---:|---:|---|", + ] + for name in sorted(by): + r = by[name] + lines.append( + f"| {name} | {r.get('mean_ic')} | {r.get('ic_t_stat')} | " + f"{r.get('weeks')} | {r.get('avg_cross_section')} | {r.get('reliable')} |" + ) + lines.extend([ + "", + "### Era split — mom_12_1_sector_resid only (for grade)", + "", + f"| era | IC | t | weeks | N |", + f"|---|---:|---:|---:|---:|", + ]) + for label in ("pre_2021", "post_2021"): + r = (era.get(label) or {}).get("mom_12_1_sector_resid") or {} + lines.append( + f"| {label} | {r.get('mean_ic', '')} | {r.get('ic_t_stat', '')} | " + f"{r.get('weeks', '')} | {r.get('avg_cross_section', '')} |" + ) + lines.extend([ + "", + "### Identical-subset baselines (sector CS)", + "", + f"```json\n{json.dumps(harness.get('identical_subset_sector_cs') or {}, indent=2, default=str)}\n```", + "", + "## Status", + "", + "PENDING_HUMAN beyond the mechanical PASS/FAIL above. " + "Nothing merged into production docs or prod code.", + "", + f"JSON: `{out_json.as_posix()}`", + "", + ]) + md_path = out_json.with_suffix(".md") + md_path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + # Update history-depth-extension.md + _update_history_doc(doc_path, payload, out_json) + + +def _update_history_doc(doc_path: Path, payload: dict, out_json: Path) -> None: + grade = payload.get("grade") or {} + banner = ( + "\n\n---\n\n" + "## Supersession notice (2026-07-19 sector-resid deep test)\n\n" + "The table and interpretation from **`history-depth-20260719-103315`** are " + "**UNMASKED, TWO-TIER SNAPSHOT — superseded, directional only, do not cite**. " + "Prod-universe names (and sector residual coverage) were left shallow while " + "breadth names were deepened; sector residual weeks=35 was a data gap.\n\n" + f"### Sector-residual deep test outcome: **{grade.get('verdict')}**\n\n" + f"{grade.get('headline')}\n\n" + f"- Reason: {grade.get('reason')}\n" + f"- Artifact: `{out_json.as_posix()}`\n" + f"- Mechanical checks: see that report.\n\n" + "**Future snapshot rebuilds must verify per-symbol depth** (earliest-bar " + "uniformity across the intended universe) — guard is a to-do, not part of " + "this order.\n" + ) + if doc_path.exists(): + text = doc_path.read_text(encoding="utf-8") + # Insert supersession after status line / near top results if not already there. + marker = "## Supersession notice (2026-07-19 sector-resid deep test)" + if marker in text: + # Replace from marker to end of that section or append fresh block at end. + pre = text.split(marker)[0].rstrip() + text = pre + banner + else: + # Mark 103315 in place if mentioned. + text = text.replace( + "Authoritative artifact:** `reports/history-depth-20260719-103315.json`", + "Superseded artifact (do not cite):** `reports/history-depth-20260719-103315.json` " + "— **UNMASKED, TWO-TIER SNAPSHOT**", + ) + text = text.rstrip() + banner + # Soften old PARK-only language if present — leave body but status at top. + if text.startswith("#"): + lines = text.splitlines() + for i, line in enumerate(lines[:15]): + if line.startswith("**Status:**"): + lines[i] = ( + f"**Status:** sector-resid deep test **{grade.get('verdict')}** " + f"— see supersession section. PENDING_HUMAN beyond PASS/FAIL." + ) + break + text = "\n".join(lines) + doc_path.write_text(text if text.endswith("\n") else text + "\n", encoding="utf-8") + else: + doc_path.write_text( + "# History-depth extension\n" + banner, encoding="utf-8" + ) + + +async def _main() -> None: + args = _parse_args() + snapshot = Path(args.snapshot) + if not snapshot.exists(): + raise SystemExit(f"Snapshot missing: {snapshot}") + if args.allow_spawn: + os.environ["BACKTEST_ALLOW_SPAWN"] = "1" + + sector_map = Path(args.sector_map) + if not sector_map.exists(): + raise SystemExit(f"Sector map missing: {sector_map}") + + step1: dict[str, Any] + if args.skip_deepen: + print("Skip deepen — race guard + sanity only…") + assert_research_snapshot_complete(snapshot) + sanity = _sanity_check(snapshot, history_days=args.history_days) + step1 = {"skipped": True, "sanity": sanity} + if not sanity["passed"]: + raise SystemExit( + "Sanity check FAILED with --skip-deepen. " + f"Details: {json.dumps(sanity, default=str)}" + ) + else: + print("Step 1 — deepen shallow symbols…") + step1 = await _step1_deepen( + snapshot, + history_days=args.history_days, + sleep_s=args.sleep, + quiet=args.quiet, + ) + if not step1["sanity"]["passed"]: + print("SANITY CHECK FAILED — refusing harness.") + print(json.dumps(step1["sanity"], indent=2, default=str)) + stamp = datetime.now().strftime("%Y%m%d-%H%M%S") + fail_path = Path("reports") / f"sector-resid-deep-{stamp}-SANITY-FAIL.json" + fail_path.parent.mkdir(parents=True, exist_ok=True) + fail_path.write_text( + json.dumps({"step1": step1, "harness": None}, indent=2, default=str) + + "\n", + encoding="utf-8", + ) + raise SystemExit( + f"Stop: sanity failed. Wrote {fail_path}. Do not run harness on two-tier data." + ) + print("Sanity check PASSED.") + assert_research_snapshot_complete(snapshot) + + print( + f"Step 2 — ONE masked harness " + f"(top {args.liquid_breadth}, min_price={args.min_price})…" + ) + harness = await _one_masked_run( + snapshot, + sector_map_path=sector_map, + liquid_breadth=args.liquid_breadth, + min_price=args.min_price, + workers=args.workers, + quiet=args.quiet, + ) + grade = _grade(harness) + print(f"GRADE: {grade['verdict']} — {grade['headline']}") + + stamp = datetime.now().strftime("%Y%m%d-%H%M%S") + out = Path(args.out) if args.out else Path("reports") / f"sector-resid-deep-{stamp}.json" + payload = { + "generated_at": datetime.now().isoformat(), + "snapshot": str(snapshot.resolve()), + "pre_registration": { + "iron_ic": IRON_IC, + "min_weeks_deep": MIN_WEEKS_DEEP, + "liquid_breadth": args.liquid_breadth, + "min_price": args.min_price, + "rule": ( + "PASS = |IC|>=0.03, +sign, reliable, weeks>=50, " + "t>=resid on same CS, era signs both +" + ), + }, + "step1": step1, + "harness": harness, + "grade": grade, + "pending_human": True, + "note": "Nothing merged into production. Thread ends at PASS/FAIL.", + } + _write_reports( + payload, + out, + Path("docs/research/history-depth-extension.md"), + ) + print(f"Wrote {out}") + print(f"Wrote {out.with_suffix('.md')}") + print("Updated docs/research/history-depth-extension.md") + + +if __name__ == "__main__": + asyncio.run(_main()) diff --git a/scripts/run_tier1_macbook.sh b/scripts/run_tier1_macbook.sh index 1472cf4..72f8aef 100755 --- a/scripts/run_tier1_macbook.sh +++ b/scripts/run_tier1_macbook.sh @@ -15,6 +15,7 @@ # ./scripts/run_tier1_macbook.sh --earnings-only # multi-day FMP backfill + re-run 2a/2b # ./scripts/run_tier1_macbook.sh --harness-only # skip rebuild; race-guard + IC only # ./scripts/run_tier1_macbook.sh --coverage-only # bars-per-year probe only +# ./scripts/run_tier1_macbook.sh --sector-resid-deep # deepen shallow + ONE masked grade # # Does NOT touch production Postgres, scheduler, gates, or prod config. @@ -35,7 +36,7 @@ FMP_SLEEP="${FMP_SLEEP:-0.35}" PYTHON="${PYTHON:-python3}" USE_CORP_PROXY="${USE_CORP_PROXY:-0}" -PHASE="depth" # depth | all | earnings | harness | coverage | ssl-check +PHASE="depth" # depth | all | earnings | harness | coverage | ssl | sector-resid-deep usage() { sed -n '2,25p' "$0" | sed 's/^# \?//' @@ -59,6 +60,7 @@ while [[ $# -gt 0 ]]; do --coverage-only) PHASE=coverage; shift ;; --depth) PHASE=depth; shift ;; --ssl-check) PHASE=ssl; shift ;; + --sector-resid-deep) PHASE=sector_resid_deep; shift ;; --corp-proxy) USE_CORP_PROXY=1; shift ;; --prod-snap) PROD_SNAP="$2"; shift 2 ;; --research-snap) RESEARCH_SNAP="$2"; shift 2 ;; @@ -223,6 +225,18 @@ run_harness() { --allow-spawn } +run_sector_resid_deep() { + need_file "$RESEARCH_SNAP" + log "Sector-resid deep test: deepen shallow symbols + ONE liquid-1500 masked grade" + log "Pre-registered PASS/FAIL only — thread ends after this run" + "$PYTHON" scripts/run_sector_resid_deep_test.py \ + --snapshot "$RESEARCH_SNAP" \ + --history-days "$HISTORY_DAYS" \ + --sleep "$ALPACA_SLEEP" \ + --workers "$WORKERS" \ + --allow-spawn +} + log "cwd=$ROOT python=$PYTHON phase=$PHASE workers=$WORKERS" setup_ssl @@ -230,6 +244,9 @@ case "$PHASE" in ssl) ssl_check ;; + sector_resid_deep) + run_sector_resid_deep + ;; coverage) run_coverage ;;