#!/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]:
    """Pass if megacaps + sector ETFs sit at the *empirical feed floor*, not calendar 5000d.

    Alpaca daily history for this stack bottoms out around 2016-01-04 (~2649 bars)
    even when history_days=5000 is requested. That is feed coverage, not a two-tier
    snapshot bug. Fail only if megacaps are still stuck near the old ~2021 prod floor
    or if sector ETFs are missing / shorter than the SPY series (except XLC listing).
    """
    depths = {r["symbol"]: r for r in _symbol_depth(snapshot)}

    engine = create_engine(
        f"sqlite:///{snapshot.resolve().as_posix()}",
        future=True,
    )
    try:
        with engine.connect() as conn:
            spy_row = conn.execute(
                text(
                    "SELECT COUNT(*), MIN(date), MAX(date) FROM benchmark_prices "
                    "WHERE symbol = 'SPY'"
                )
            ).fetchone()
            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()

    spy_n, spy_min, spy_max = spy_row if spy_row else (0, None, None)
    feed_floor = (
        date.fromisoformat(str(spy_min)[:10])
        if spy_min
        else date(2016, 1, 4)
    )
    # Megacaps must match the feed floor within a few sessions (not calendar-5000).
    megacap_slack_days = 10
    # Old two-tier defect left prod names at ~2021-06; anything still after this fails.
    old_shallow_floor = date(2020, 1, 1)

    megacap = {}
    ok_mega = True
    mega_reasons: list[str] = []
    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
            mega_reasons.append(f"{sym}: missing")
            continue
        d0 = date.fromisoformat(row["min_date"])
        if d0 > old_shallow_floor:
            ok_mega = False
            mega_reasons.append(
                f"{sym}: min_date={d0} still after {old_shallow_floor} (two-tier unrepaired)"
            )
        elif d0 > feed_floor + timedelta(days=megacap_slack_days):
            ok_mega = False
            mega_reasons.append(
                f"{sym}: min_date={d0} later than SPY feed floor {feed_floor}"
            )

    etf_info = {
        s: {"n": n, "min": d0, "max": d1} for s, n, d0, d1 in etf_rows
    }
    deep_etfs = 0
    etf_reasons: list[str] = []
    for sym in SECTOR_ETFS:
        info = etf_info.get(sym)
        if not info or not info["min"]:
            etf_reasons.append(f"{sym}: missing")
            continue
        d0 = date.fromisoformat(str(info["min"])[:10])
        if sym == "XLC":
            # Listed 2018-06-18/19.
            if d0 <= date(2018, 7, 15):
                deep_etfs += 1
            else:
                etf_reasons.append(f"XLC: min_date={d0} later than listing floor")
        else:
            if d0 <= feed_floor + timedelta(days=megacap_slack_days):
                deep_etfs += 1
            else:
                etf_reasons.append(
                    f"{sym}: min_date={d0} later than SPY feed floor {feed_floor}"
                )

    ok_etf = deep_etfs >= 10
    still_shallow, _ = _derive_shallow(list(depths.values()), lag_days=400)
    note_xlc = (
        "XLC lists mid-2018 → Communication Services residual coverage from ~mid-2019."
    )
    note_feed = (
        f"Empirical Alpaca floor observed via SPY: {feed_floor.isoformat()} "
        f"(n={spy_n}). Calendar history_days={history_days} is a request cap, not a "
        "guarantee — sanity grades against the feed floor, not 5000 calendar days."
    )

    passed = bool(ok_mega and ok_etf)
    return {
        "passed": passed,
        "megacap": megacap,
        "megacap_ok": ok_mega,
        "megacap_reasons": mega_reasons,
        "feed_floor": feed_floor.isoformat(),
        "spy_benchmark": {"n": spy_n, "min": spy_min, "max": spy_max},
        "old_shallow_floor": old_shallow_floor.isoformat(),
        "sector_etfs": etf_info,
        "sector_etfs_deep_count": deep_etfs,
        "sector_etfs_ok": ok_etf,
        "sector_etf_reasons": etf_reasons,
        "still_shallow_count": len(still_shallow),
        "still_shallow_sample": still_shallow[:20],
        "still_shallow_note": (
            "Remaining 'shallow' names are mostly post-2017 IPOs/listings — expected, "
            "not a two-tier defect."
        ),
        "xlc_note": note_xlc,
        "feed_note": note_feed,
        "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())