"""Sector-residual momentum research runner (local only).

Protocol
--------
1. Race-guard the research/prod snapshot (completion manifest when present).
2. Require sector map + sector ETFs in ``benchmark_prices``.
3. Run signal IC harness on the production ~505-name snapshot
   (``BACKTEST_SIGNAL_EVAL_ONLY=1``) with sector context loaded.
4. Grade candidates vs pre-registered iron rule + t-stat vs ``mom_12_1_resid``.
5. If a candidate promotes: portfolio A/B with candidate as momentum leg +
   gate percentile (``fill_mode=close``). Optional sector-cap arm.

Does not modify production DB, gate, scanner, or schedule.

Example
-------
  python scripts/run_sector_residual_research.py \\
      --snapshot backtest_snapshots/prod.sqlite \\
      --workers 6 --allow-spawn
"""

from __future__ import annotations

import argparse
import asyncio
import json
import math
import os
import sys
from collections import defaultdict
from copy import deepcopy
from datetime import date, datetime
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,
    coverage_stats,
    load_ticker_sector_map,
    normalise_symbol,
    sector_to_etf,
)

VALIDATION_SPLIT = date(2024, 7, 1)
IRON_IC_BAR = 0.03
MIN_RELIABLE = 12


def _sqlite_url(path: Path) -> str:
    return f"sqlite+aiosqlite:///{path.resolve().as_posix()}"


def _parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("--snapshot", default="backtest_snapshots/prod.sqlite")
    p.add_argument(
        "--sector-map",
        default=str(DEFAULT_SECTOR_MAP_PATH),
    )
    p.add_argument("--workers", type=int, default=6)
    p.add_argument("--allow-spawn", action="store_true")
    p.add_argument(
        "--skip-ab",
        action="store_true",
        help="IC only — never run portfolio A/B even if promotion fires.",
    )
    p.add_argument(
        "--force-ab",
        action="store_true",
        help="Run A/B for diagnostic even if IC bar fails (still reported as non-promote).",
    )
    p.add_argument(
        "--sector-cap",
        type=int,
        default=None,
        help="Optional max positions per sector (e.g. 3). Only used in A/B.",
    )
    p.add_argument("--quiet", action="store_true")
    p.add_argument(
        "--out",
        default=None,
        help="JSON report path (default reports/sector-residual-YYYYMMDD-HHMMSS.json)",
    )
    return p.parse_args()


def _assert_snapshot_ready(snapshot: Path) -> dict[str, Any]:
    """Race guard: prefer completion manifest; always check live bar sanity."""
    from scripts.research_snapshot_manifest import (  # type: ignore
        assert_research_snapshot_complete,
        load_manifest,
    )

    guard: dict[str, Any] = {"snapshot": str(snapshot.resolve())}
    manifest = load_manifest(snapshot)
    if manifest is not None:
        # Full assert when a manifest exists (research.sqlite path).
        try:
            m = assert_research_snapshot_complete(snapshot)
            guard["manifest"] = m
            guard["manifest_ok"] = True
        except SystemExit as exc:
            raise SystemExit(str(exc)) from exc
    else:
        guard["manifest"] = None
        guard["manifest_ok"] = None
        guard["note"] = (
            "No completion manifest (prod.sqlite is expected without one). "
            "Bar-count sanity still applied."
        )

    engine = create_engine(
        f"sqlite:///{snapshot.resolve().as_posix()}",
        future=True,
    )
    try:
        with engine.connect() as conn:
            ticker_n = int(conn.execute(text("SELECT COUNT(*) FROM tickers")).scalar_one())
            ohlcv_n = int(
                conn.execute(text("SELECT COUNT(*) FROM ohlcv_records")).scalar_one()
            )
            bar_stats = conn.execute(
                text(
                    """
                    SELECT MIN(c), AVG(c), MAX(c) FROM (
                      SELECT COUNT(*) AS c FROM ohlcv_records GROUP BY ticker_id
                    )
                    """
                )
            ).fetchone()
            bench = conn.execute(
                text(
                    "SELECT symbol, COUNT(*), MIN(date), MAX(date) "
                    "FROM benchmark_prices GROUP BY symbol ORDER BY symbol"
                )
            ).fetchall()
            d_range = conn.execute(
                text("SELECT MIN(date), MAX(date) FROM ohlcv_records")
            ).fetchone()
    finally:
        engine.dispose()

    guard["ticker_count"] = ticker_n
    guard["ohlcv_row_count"] = ohlcv_n
    guard["bars_min_avg_max"] = {
        "min": bar_stats[0],
        "avg": round(float(bar_stats[1]), 1) if bar_stats[1] is not None else None,
        "max": bar_stats[2],
    }
    guard["ohlcv_date_range"] = {"min": d_range[0], "max": d_range[1]}
    guard["benchmark_prices"] = [
        {"symbol": s, "n": n, "min": d0, "max": d1} for s, n, d0, d1 in bench
    ]

    # Sanity: a half-built snapshot would show many tickers with tiny bar counts.
    min_bars = int(bar_stats[0] or 0)
    avg_bars = float(bar_stats[1] or 0)
    if ticker_n < 400:
        raise SystemExit(
            f"Snapshot looks short: only {ticker_n} tickers (expected ~505 prod)."
        )
    if avg_bars < 200:
        raise SystemExit(
            f"Snapshot bar counts look short (avg={avg_bars:.0f}). Rebuild before research."
        )
    # Allow a few thin names; refuse if median path is collapsed.
    if min_bars < 10 and avg_bars < 500:
        raise SystemExit(
            f"Snapshot min bars={min_bars}, avg={avg_bars:.0f} — possible partial build."
        )

    present_etfs = {row[0] for row in bench}
    missing_etfs = [e for e in SECTOR_ETFS if e not in present_etfs]
    guard["missing_sector_etfs"] = missing_etfs
    if missing_etfs:
        raise SystemExit(
            "Sector ETFs missing from benchmark_prices: "
            f"{missing_etfs}. Run scripts/fetch_sector_etfs_to_snapshot.py first."
        )
    if "SPY" not in present_etfs:
        raise SystemExit("SPY missing from benchmark_prices")

    return guard


def _find_signal(rows: list[dict], name: str) -> dict | None:
    for row in rows or []:
        if row.get("signal") == name:
            return row
    return None


def _grade_ic(
    candidate: dict | None,
    resid: dict | None,
    *,
    expected_sign: float = 1.0,
) -> dict[str, Any]:
    """Iron rule + t-stat ≥ mom_12_1_resid."""
    if candidate is None:
        return {
            "promote_to_ab": False,
            "reason": "signal missing from signal_eval",
        }
    mean_ic = candidate.get("mean_ic")
    t_stat = candidate.get("ic_t_stat")
    reliable = bool(candidate.get("reliable"))
    weeks = int(candidate.get("weeks") or 0)
    if mean_ic is None or t_stat is None:
        return {"promote_to_ab": False, "reason": "missing mean_ic or t", "row": candidate}

    sign_ok = (float(mean_ic) * expected_sign) > 0
    mag_ok = abs(float(mean_ic)) >= IRON_IC_BAR
    reliable_ok = reliable and weeks >= MIN_RELIABLE
    resid_t = resid.get("ic_t_stat") if resid else None
    t_ok = resid_t is not None and float(t_stat) >= float(resid_t)

    promote = sign_ok and mag_ok and reliable_ok and t_ok
    return {
        "promote_to_ab": promote,
        "checks": {
            "sign_ok": sign_ok,
            "abs_mean_ic_ge_0_03": mag_ok,
            "reliable": reliable_ok,
            "t_ge_resid": t_ok,
            "mean_ic": mean_ic,
            "ic_t_stat": t_stat,
            "resid_ic_t_stat": resid_t,
            "weeks": weeks,
        },
        "reason": (
            "clears iron rule and t ≥ mom_12_1_resid — authorized for A/B only"
            if promote
            else "does not clear pre-registered IC promotion bar"
        ),
        "row": candidate,
    }


async def _run_signal_eval(
    snapshot: Path,
    *,
    workers: int,
    quiet: bool,
    sector_map_path: Path,
) -> dict:
    from app.config import settings
    from app.services.backtest_service import run_backtest

    os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1"
    os.environ["BACKTEST_SIGNAL_EVAL_ONLY"] = "1"
    os.environ["BACKTEST_SECTOR_MAP_PATH"] = str(sector_map_path.resolve())
    # Clear liquid-breadth — this is the 505-name prod IC, not breadth.
    os.environ.pop("BACKTEST_LIQUID_BREADTH", None)
    settings.backtest_workers = workers

    engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True)
    Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

    def progress(done: int, total: int, symbol: str) -> None:
        if quiet:
            return
        print(f"  progress {done}/{total} {symbol}", end="\r", flush=True)

    try:
        async with Session() as db:
            report = await run_backtest(db, progress_cb=progress, cadence="weekly")
    finally:
        await engine.dispose()
    if not quiet:
        print()
    return report


def _period_percentiles(rows: list[dict], value_key: str) -> dict[tuple, dict[str, float]]:
    by_period: dict[tuple, list[dict]] = defaultdict(list)
    for row in rows:
        if row.get(value_key) is None:
            continue
        period = row.get("ranking_period") or row.get("iso_week")
        by_period[period].append(row)
    out: dict[tuple, dict[str, float]] = {}
    for period, group in by_period.items():
        ordered = sorted(group, key=lambda r: float(r[value_key]))
        n = len(ordered)
        for rank, row in enumerate(ordered):
            key = (str(row["symbol"]), str(row["date"]))
            pct = (rank / (n - 1) * 100.0) if n > 1 else 100.0
            out.setdefault(key, {})[value_key] = float(row[value_key])
            out[key][f"{value_key}_percentile"] = pct
    return out


async def _load_prices_and_benchmarks(snapshot: Path) -> tuple[dict, dict, dict]:
    """Return (price_columns, spy_closes, sector_etf_closes)."""
    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

    engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True)
    Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    prices: dict[str, tuple] = {}
    try:
        async with Session() as db:
            tickers = list(
                (await db.execute(select(Ticker).order_by(Ticker.symbol))).scalars()
            )
            for t in tickers:
                cols = await bt._fetch_columns(db, t.symbol)
                if cols is not None:
                    prices[t.symbol] = cols
            spy = await load_benchmark_closes(db, "SPY")
            sector: dict[str, dict] = {}
            for etf in SECTOR_ETFS:
                series = await load_benchmark_closes(db, etf)
                if series:
                    sector[etf] = series
    finally:
        await engine.dispose()
    return prices, spy, sector


def _recompute_sector_residual_on_candidates(
    candidates: list[dict],
    prices: dict[str, tuple],
    spy_closes: dict,
    sector_etf_closes: dict[str, dict],
    symbol_to_sector: dict[str, str],
    *,
    momentum_field: str,
) -> list[dict]:
    """Attach alternative residual momentum on each candidate as-of date."""
    from app.services import backtest_service as bt
    from app.services.sector_map import etf_for_symbol

    # Index price series once.
    series_cache: dict[str, tuple[list, list, list]] = {}
    for sym, cols in prices.items():
        ords, _o, _h, _l, closes, _v = cols
        dates = [date.fromordinal(int(o)) for o in ords]
        series_cache[sym] = (dates, list(closes), list(ords))

    out: list[dict] = []
    for cand in candidates:
        c = dict(cand)
        sym = str(c["symbol"])
        if sym not in series_cache:
            out.append(c)
            continue
        dates, closes, ords = series_cache[sym]
        asof = date.fromisoformat(str(c["date"]))
        # Find as-of index.
        try:
            i = next(idx for idx, d in enumerate(dates) if d == asof)
        except StopIteration:
            # nearest on/before
            i = max((idx for idx, d in enumerate(dates) if d <= asof), default=-1)
        if i < 0:
            out.append(c)
            continue

        if momentum_field == "mom_12_1_sector_resid":
            etf = etf_for_symbol(sym, symbol_to_sector)
            etf_series = sector_etf_closes.get(etf or "")
            val = None
            if spy_closes and etf_series:
                val = bt._multi_factor_residual_momentum_12_1(
                    dates, closes, i, [spy_closes, etf_series]
                )
            c["residual_momentum"] = val
            c["_alt_momentum_signal"] = momentum_field
            c["_alt_momentum_value"] = val
        elif momentum_field == "mom_12_1_sector_demeaned":
            # Placeholder: demean requires cross-section; filled in a second pass.
            raw = None
            if i >= 252 and closes[i - 252] > 0:
                raw = closes[i - 21] / closes[i - 252] - 1.0
            c["_raw_mom_12_1"] = raw
            c["_alt_momentum_signal"] = momentum_field
        else:
            raise ValueError(momentum_field)
        out.append(c)

    if momentum_field == "mom_12_1_sector_demeaned":
        # Cross-sectional demean within ranking period × sector.
        by_period: dict[Any, list[dict]] = defaultdict(list)
        for c in out:
            if c.get("_raw_mom_12_1") is None:
                continue
            period = c.get("ranking_period") or c.get("iso_week")
            by_period[period].append(c)
        for period, group in by_period.items():
            by_sec: dict[str, list[float]] = defaultdict(list)
            for c in group:
                sec = symbol_to_sector.get(normalise_symbol(str(c["symbol"])))
                if sec:
                    by_sec[sec].append(float(c["_raw_mom_12_1"]))
            means = {
                s: sum(vs) / len(vs) for s, vs in by_sec.items() if len(vs) >= 2
            }
            for c in group:
                sec = symbol_to_sector.get(normalise_symbol(str(c["symbol"])))
                raw = float(c["_raw_mom_12_1"])
                if sec in means:
                    val = raw - means[sec]
                    c["residual_momentum"] = val
                    c["_alt_momentum_value"] = val
                else:
                    c["residual_momentum"] = None
                    c["_alt_momentum_value"] = None

    return out


def _assign_prod_ranks(candidates: list[dict]) -> None:
    from app.services import backtest_service as bt

    bt._assign_momentum_percentiles(candidates)
    bt._assign_residual_momentum_percentiles(candidates)
    bt._assign_low_volatility_percentiles(candidates)
    bt._assign_activation_momentum_percentiles(candidates)
    bt._assign_residual_high_vol_blend(candidates)
    for c in candidates:
        c["qualified"] = bt._momentum_qualifies(c, 80.0)


async def _run_ab(
    snapshot: Path,
    *,
    sector_map: dict[str, str],
    signal_name: str,
    sector_cap: int | None,
    quiet: bool,
    workers: int,
) -> dict[str, Any]:
    """Control vs treatment book with candidate as momentum residual."""
    from app.services import backtest_service as bt
    from app.config import settings
    from app.models.ticker import Ticker
    from app.services.admin_service import get_activation_config
    from app.services.recommendation_service import get_recommendation_config
    from app.services.paper_trade_service import get_exit_policy
    from app.services.benchmark_service import load_benchmark_closes
    from sqlalchemy import select

    os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1"
    os.environ.pop("BACKTEST_SIGNAL_EVAL_ONLY", None)
    settings.backtest_workers = max(1, int(workers))

    engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True)
    Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    try:
        async with Session() as db:
            config = await get_recommendation_config(db)
            activation = await get_activation_config(db)
            exit_config = await get_exit_policy(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

            prices: dict[str, tuple] = {}
            candidates: list[dict] = []
            for idx, t in enumerate(tickers):
                if not quiet and idx % 25 == 0:
                    print(f"  fetch {idx}/{len(tickers)}", end="\r", flush=True)
                cols = await bt._fetch_columns(db, t.symbol)
                if cols is None:
                    continue
                prices[t.symbol] = cols
                cands, _series = bt._replay_and_signals(
                    t.symbol,
                    cols,
                    config,
                    activation,
                    spy,
                    bt.PRODUCTION_GTL_TARGET_MODEL,
                    "weekly",
                    False,
                    sector_etf,
                    sector_map,
                )
                candidates.extend(cands)
    finally:
        await engine.dispose()
    if not quiet:
        print()

    # Control ranks (production residual).
    control = [dict(c) for c in candidates]
    _assign_prod_ranks(control)
    control_longs = [
        c for c in control if c.get("qualified") and c.get("direction") == "long"
    ]

    # Treatment: replace residual with sector signal, re-rank.
    treatment = _recompute_sector_residual_on_candidates(
        candidates,
        prices,
        spy,
        sector_etf,
        sector_map,
        momentum_field=signal_name,
    )
    _assign_prod_ranks(treatment)
    treatment_longs = [
        c for c in treatment if c.get("qualified") and c.get("direction") == "long"
    ]

    strategy = next(s for s in bt.PORTFOLIO_MONITOR_STRATEGIES if s.get("is_production"))
    entry_config = bt._entry_variant_config(str(strategy["entry_variant"]))
    assert entry_config is not None
    ranking_key = str(
        entry_config.get("ranking_key") or entry_config["percentile_key"]
    )
    # Production ranking key is residual_high_vol_blend_80_20.
    if ranking_key not in (bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY, bt.PRODUCTION_PERCENTILE_KEY):
        ranking_key = bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY

    exit_policy = bt.LIVE_EXIT_MODE_TO_SIM.get(
        str(exit_config.get("mode", "atr_trailing")), "atr_trail3"
    )
    hold_days = int(exit_config.get("hold_days", 30))
    trail = float(exit_config.get("atr_multiplier", bt.ATR_TRAIL_MULTIPLIER))
    risk = float(entry_config["risk_per_trade"])
    max_pos = int(entry_config["max_positions"])

    def _sim_book(longs: list[dict], *, label: str, cap: int | None = None) -> dict:
        reentry = bt._make_gate_reset_reentry_fn(
            longs, prices, cadence="weekly", ranking_key=ranking_key
        )
        windows = {}
        for wname, start, end in (
            ("train", None, VALIDATION_SPLIT),
            ("validation", VALIDATION_SPLIT, None),
            ("full", None, None),
        ):
            sim = bt._simulate_portfolio(
                longs,
                prices,
                spy,
                exit_policy,
                hold_days,
                ranking_key=ranking_key,
                max_positions=max_pos,
                risk_per_trade=risk,
                atr_trail_multiplier=trail,
                post_stop_reentry_fn=reentry,
                start_date=start,
                end_date=end,
                fill_mode=bt.FILL_MODE_CLOSE,
                include_trades=True,
            )
            if sim is None:
                windows[wname] = {"error": "no_trades"}
                continue
            # Optional sector cap: filter trade_details is post-hoc; real cap needs
            # simulator support. For research, re-sim with a wrapper ranking that
            # drops overflow sector names is approximate — we implement a simple
            # pre-filter on daily entry sets via max_positions only when cap is None.
            # When cap is set, apply a post-sim diagnostic on entries.
            payload = {
                k: sim.get(k)
                for k in (
                    "sharpe",
                    "sharpe_se",
                    "cagr_pct",
                    "max_drawdown_pct",
                    "total_return_pct",
                    "trades",
                    "win_rate_pct",
                    "avg_r",
                    "n_returns",
                    "return_skew",
                    "return_kurtosis",
                    "psr",
                )
            }
            details = sim.get("trade_details") or []
            rs = [
                float(t["realized_r"])
                for t in details
                if t.get("realized_r") is not None
            ]
            if rs:
                rs_sorted = sorted(rs)
                payload["r_p05"] = rs_sorted[max(0, int(0.05 * (len(rs_sorted) - 1)))]
                payload["r_p50"] = rs_sorted[len(rs_sorted) // 2]
                payload["r_p95"] = rs_sorted[min(len(rs_sorted) - 1, int(0.95 * (len(rs_sorted) - 1)))]
                payload["entry_count"] = len(rs)
            if cap is not None and details:
                # Diagnostic: count how often a calendar day would exceed cap.
                from collections import Counter

                # Use entry dates; sector from map.
                day_sector: dict[str, Counter] = defaultdict(Counter)
                for t in details:
                    sec = sector_map.get(normalise_symbol(str(t.get("symbol", "")))) or "?"
                    day_sector[str(t.get("entry_date") or t.get("date") or "")][sec] += 1
                breaches = sum(
                    1
                    for day, ctr in day_sector.items()
                    if any(v > cap for v in ctr.values())
                )
                payload["sector_cap"] = cap
                payload["entry_days_with_sector_over_cap"] = breaches
            windows[wname] = payload
        return {"label": label, "n_qualified_longs": len(longs), "windows": windows}

    control_result = _sim_book(control_longs, label="control_mom_12_1_resid")
    treatment_result = _sim_book(
        treatment_longs, label=f"treatment_{signal_name}", cap=None
    )
    out: dict[str, Any] = {
        "signal": signal_name,
        "ranking_key": ranking_key,
        "fill_mode": "close",
        "validation_split": VALIDATION_SPLIT.isoformat(),
        "control": control_result,
        "treatment": treatment_result,
        "promotion": _grade_ab(control_result, treatment_result),
    }
    if sector_cap is not None:
        # Approximate sector-cap book: when selecting, prefer higher rank but
        # refuse a 4th name in the same sector among concurrent opens.
        # Implemented by tagging candidates and using a custom sim is heavy;
        # instead report diagnostic on unconstrained treatment + a filtered
        # re-rank that zeros residual for overflow names within each period.
        capped = _apply_sector_cap_to_ranks(
            treatment, sector_map, cap=sector_cap, ranking_key=ranking_key
        )
        capped_longs = [
            c for c in capped if c.get("qualified") and c.get("direction") == "long"
        ]
        out["sector_cap_arm"] = _sim_book(
            capped_longs, label=f"treatment_{signal_name}_cap{sector_cap}", cap=sector_cap
        )
        out["sector_cap_promotion"] = _grade_ab(
            control_result, out["sector_cap_arm"]
        )
    return out


def _apply_sector_cap_to_ranks(
    candidates: list[dict],
    sector_map: dict[str, str],
    *,
    cap: int,
    ranking_key: str,
) -> list[dict]:
    """Within each ranking period, keep top `cap` per sector by ranking_key."""
    by_period: dict[Any, list[dict]] = defaultdict(list)
    for c in candidates:
        period = c.get("ranking_period") or c.get("iso_week")
        by_period[period].append(dict(c))
    out: list[dict] = []
    for period, group in by_period.items():
        ordered = sorted(
            group,
            key=lambda r: float(r.get(ranking_key) or r.get("residual_momentum") or -1e9),
            reverse=True,
        )
        sector_counts: dict[str, int] = defaultdict(int)
        for c in ordered:
            sec = sector_map.get(normalise_symbol(str(c["symbol"]))) or "_unknown"
            if sector_counts[sec] >= cap:
                # Push below gate by nulling activation percentile.
                c["qualified"] = False
                c["_sector_cap_blocked"] = True
            else:
                if c.get("qualified"):
                    sector_counts[sec] += 1
            out.append(c)
    return out


def _grade_ab(control: dict, treatment: dict) -> dict[str, Any]:
    """Pre-registered: val Sharpe ≥ control − 0.5·SE; full Sharpe & maxDD not worse."""
    def win(arm: dict, name: str) -> dict:
        return (arm.get("windows") or {}).get(name) or {}

    c_val = win(control, "validation")
    t_val = win(treatment, "validation")
    c_full = win(control, "full")
    t_full = win(treatment, "full")

    def _f(d: dict, k: str) -> float | None:
        v = d.get(k)
        return None if v is None else float(v)

    c_sh = _f(c_val, "sharpe")
    t_sh = _f(t_val, "sharpe")
    # Use treatment SE if present else control SE.
    se = _f(t_val, "sharpe_se")
    if se is None:
        se = _f(c_val, "sharpe_se")
    if se is None:
        se = 0.0

    val_ok = (
        c_sh is not None
        and t_sh is not None
        and t_sh >= (c_sh - 0.5 * se)
    )
    c_full_sh = _f(c_full, "sharpe")
    t_full_sh = _f(t_full, "sharpe")
    full_sh_ok = (
        c_full_sh is not None
        and t_full_sh is not None
        and t_full_sh >= c_full_sh
    )
    # max DD: higher absolute drawdown is worse; stored as positive pct typically.
    c_dd = _f(c_full, "max_drawdown_pct")
    t_dd = _f(t_full, "max_drawdown_pct")
    full_dd_ok = (
        c_dd is not None and t_dd is not None and abs(t_dd) <= abs(c_dd) + 1e-9
    )
    promote = bool(val_ok and full_sh_ok and full_dd_ok)
    return {
        "promote": promote,
        "checks": {
            "validation_sharpe_ge_control_minus_half_se": val_ok,
            "full_sharpe_not_worse": full_sh_ok,
            "full_maxdd_not_worse": full_dd_ok,
            "control_validation_sharpe": c_sh,
            "treatment_validation_sharpe": t_sh,
            "se_used": se,
            "control_full_sharpe": c_full_sh,
            "treatment_full_sharpe": t_full_sh,
            "control_full_maxdd": c_dd,
            "treatment_full_maxdd": t_dd,
        },
        "reason": (
            "clears pre-registered A/B bar — human decides wire-in"
            if promote
            else "fails pre-registered A/B bar"
        ),
    }


def _write_md(path: Path, payload: dict) -> None:
    """Refresh the results sections of the research doc (preserve pre-reg header)."""
    # Always write a standalone results companion + update the main doc's
    # results block by rewriting the full file with pre-reg + results.
    pre = Path("docs/research/sector-residual-momentum.md")
    # Keep pre-registration by reading until '## Results' if present.
    header = ""
    if pre.exists():
        text = pre.read_text(encoding="utf-8")
        marker = "## Results"
        idx = text.find(marker)
        header = text[:idx] if idx >= 0 else text.split("## Verdict")[0]

    guard = payload.get("snapshot_guard") or {}
    cov = payload.get("sector_coverage") or {}
    ic_rows = payload.get("signal_eval") or []
    grades = payload.get("ic_grades") or {}
    lines = [
        header.rstrip(),
        "",
        "## Results",
        "",
        f"Generated: `{payload.get('generated_at')}`",
        "",
        "### Snapshot race guard",
        "",
        f"- Snapshot: `{guard.get('snapshot')}`",
        f"- Tickers: **{guard.get('ticker_count')}**  OHLCV rows: **{guard.get('ohlcv_row_count')}**",
        f"- Bars min/avg/max: `{guard.get('bars_min_avg_max')}`",
        f"- OHLCV range: `{guard.get('ohlcv_date_range')}`",
        f"- Manifest ok: `{guard.get('manifest_ok')}`",
        f"- Missing sector ETFs at start: `{guard.get('missing_sector_etfs')}`",
        "",
        "### Sector label coverage",
        "",
        f"```json\n{json.dumps(cov, indent=2, default=str)}\n```",
        "",
        "### IC harness (identical cross-sections)",
        "",
        "| signal | mean_ic | ic_t_stat | weeks | avg_N | reliable | ic+_pct |",
        "|---|---:|---:|---:|---:|---|---:|",
    ]
    want = [
        "mom_12_1",
        "mom_12_1_resid",
        "mom_12_1_sector_resid",
        "mom_12_1_sector_demeaned",
    ]
    by_name = {r.get("signal"): r for r in ic_rows}
    for name in want:
        r = by_name.get(name) or {}
        lines.append(
            f"| {name} | {r.get('mean_ic', '')} | {r.get('ic_t_stat', '')} | "
            f"{r.get('weeks', '')} | {r.get('avg_cross_section', '')} | "
            f"{r.get('reliable', '')} | {r.get('ic_positive_pct', '')} |"
        )
    lines.extend(["", "### IC promotion grades", ""])
    for name, g in grades.items():
        lines.append(f"- **{name}**: promote_to_ab=`{g.get('promote_to_ab')}` — {g.get('reason')}")
        lines.append(f"  - checks: `{json.dumps(g.get('checks') or {}, default=str)}`")

    ab = payload.get("portfolio_ab")
    lines.extend(["", "### Portfolio A/B", ""])
    if not ab:
        lines.append("_Not run (IC bar not cleared, or --skip-ab)._")
    else:
        lines.append(f"```json\n{json.dumps(ab, indent=2, default=str)}\n```")

    lines.extend([
        "",
        "## Verdict",
        "",
        f"**{payload.get('verdict')}**",
        "",
        payload.get("verdict_detail") or "",
        "",
        "## What a human must decide next",
        "",
        payload.get("human_next") or "- Review numbers; do not merge into strategy docs without approval.",
        "",
        "## Artifacts",
        "",
        f"- JSON: `{payload.get('report_path')}`",
        "",
    ])
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")


async def _main() -> None:
    args = _parse_args()
    snapshot = Path(args.snapshot)
    sector_map_path = Path(args.sector_map)
    if not snapshot.exists():
        raise SystemExit(f"Snapshot missing: {snapshot}")
    if not sector_map_path.exists():
        raise SystemExit(
            f"Sector map missing: {sector_map_path}. "
            "Run scripts/build_ticker_sector_map.py first."
        )

    if args.allow_spawn:
        os.environ["BACKTEST_ALLOW_SPAWN"] = "1"

    print("Race-guarding snapshot…")
    guard = _assert_snapshot_ready(snapshot)
    print(
        f"  tickers={guard['ticker_count']} ohlcv={guard['ohlcv_row_count']} "
        f"bars={guard['bars_min_avg_max']}"
    )

    mapping = load_ticker_sector_map(sector_map_path)
    engine = create_engine(
        f"sqlite:///{snapshot.resolve().as_posix()}",
        future=True,
    )
    try:
        with engine.connect() as conn:
            symbols = [
                normalise_symbol(r[0])
                for r in conn.execute(text("SELECT symbol FROM tickers")).fetchall()
            ]
    finally:
        engine.dispose()
    cov = coverage_stats(symbols, mapping)
    print(
        f"Sector map: {cov['mapped']}/{cov['universe']} "
        f"({cov['mapped_pct']}%) with_etf={cov['with_etf']}"
    )
    if cov["mapped_pct"] < 90:
        print(
            f"WARNING: sector coverage {cov['mapped_pct']}% < 90%; "
            f"missing e.g. {cov['missing'][:20]}"
        )

    print("Running IC harness (signal-eval only)…")
    report = await _run_signal_eval(
        snapshot,
        workers=args.workers,
        quiet=args.quiet,
        sector_map_path=sector_map_path,
    )
    signal_eval = report.get("signal_eval") or report.get("signals") or []
    # Locate key in report — backtest uses "signal_edge" historically.
    if not signal_eval:
        for key in ("signal_edge", "signal_evaluation", "factor_ic"):
            if key in report and isinstance(report[key], list):
                signal_eval = report[key]
                break

    resid = _find_signal(signal_eval, "mom_12_1_resid")
    grades = {
        name: _grade_ic(_find_signal(signal_eval, name), resid)
        for name in ("mom_12_1_sector_resid", "mom_12_1_sector_demeaned")
    }
    for name, g in grades.items():
        print(
            f"  {name}: promote_to_ab={g['promote_to_ab']} "
            f"ic={((g.get('row') or {}).get('mean_ic'))} "
            f"t={((g.get('row') or {}).get('ic_t_stat'))}"
        )

    ab_results: dict[str, Any] | None = None
    promote_names = [n for n, g in grades.items() if g.get("promote_to_ab")]
    run_ab = (bool(promote_names) or args.force_ab) and not args.skip_ab
    if run_ab:
        # Prefer sector_resid if both; else the one that promoted / force resid.
        if "mom_12_1_sector_resid" in promote_names or (
            args.force_ab and not promote_names
        ):
            ab_signal = "mom_12_1_sector_resid"
        else:
            ab_signal = promote_names[0]
        print(f"Running portfolio A/B for {ab_signal}…")
        ab_results = await _run_ab(
            snapshot,
            sector_map=mapping,
            signal_name=ab_signal,
            sector_cap=args.sector_cap,
            quiet=args.quiet,
            workers=args.workers,
        )
        print(
            f"  A/B promote={ab_results.get('promotion', {}).get('promote')} "
            f"— {ab_results.get('promotion', {}).get('reason')}"
        )
    else:
        print("Skipping portfolio A/B (no IC promotion; use --force-ab to override).")

    # Verdict
    if ab_results and ab_results.get("promotion", {}).get("promote"):
        verdict = "PROMOTE"
        detail = (
            f"{ab_results['signal']} cleared IC + A/B bars. "
            "Human must design wire-in; do not ship from this branch."
        )
        human = (
            "- Approve or reject production residual swap vs dual-signal design.\n"
            "- If sector-cap arm ran, review tail-trim diagnostics before any cap."
        )
    elif any(g.get("promote_to_ab") for g in grades.values()):
        verdict = "PARK"
        detail = (
            "IC promotion bar cleared but A/B did not promote "
            "(or A/B skipped). Park for human review."
        )
        human = "- Inspect A/B windows; decide whether to re-run or park."
    elif any(
        (g.get("row") or {}).get("mean_ic") is not None
        and abs(float((g.get("row") or {}).get("mean_ic") or 0)) >= IRON_IC_BAR * 0.5
        for g in grades.values()
    ):
        verdict = "PARK"
        detail = "Weak / partial IC — not dead, not green. Machinery kept."
        human = "- No book change. Revisit after history-depth extension (Task 3)."
    else:
        verdict = "DEAD"
        detail = (
            "Neither sector residual nor sector demean cleared the iron-rule bar "
            "with t ≥ mom_12_1_resid on this window."
        )
        human = "- Do not wire sector residual. Optional: re-check after Task 3 depth."

    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    out_path = Path(args.out) if args.out else Path("reports") / f"sector-residual-{stamp}.json"
    payload = {
        "generated_at": datetime.now().isoformat(),
        "snapshot_guard": guard,
        "sector_coverage": cov,
        "sector_map_path": str(sector_map_path.resolve()),
        "signal_eval": signal_eval,
        "ic_grades": grades,
        "portfolio_ab": ab_results,
        "verdict": verdict,
        "verdict_detail": detail,
        "human_next": human,
        "report_path": str(out_path.as_posix()),
        "pre_registration": {
            "iron_ic_bar": IRON_IC_BAR,
            "validation_split": VALIDATION_SPLIT.isoformat(),
            "fill_mode": "close",
            "cost_per_side": 0.001,
            "ab_rule": "val Sharpe >= control - 0.5*SE; full Sharpe & maxDD not worse",
        },
    }
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(payload, indent=2, default=str) + "\n", encoding="utf-8")

    md_path = Path("docs/research/sector-residual-momentum.md")
    _write_md(md_path, payload)
    # Companion md under reports/
    md_report = out_path.with_suffix(".md")
    md_report.write_text(md_path.read_text(encoding="utf-8"), encoding="utf-8")

    print(f"Verdict: {verdict}")
    print(f"Wrote {out_path}")
    print(f"Wrote {md_path}")


if __name__ == "__main__":
    asyncio.run(_main())