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.
This commit is contained in:
@@ -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())
|
||||
@@ -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
|
||||
;;
|
||||
|
||||
Reference in New Issue
Block a user