dennisthiessen
a9841d92b7
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.
999 lines
35 KiB
Python
999 lines
35 KiB
Python
#!/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())
|