research: volatility-standardized residual momentum screen, failed

Strategy review (2026-09-08) and a frozen, paired signal screen of
residual 12-1 momentum divided by its formation-window residual
volatility. Across 73 non-overlapping windows the normalized signal's IC
is 0.0128 against 0.0145 for the matched control, below the frozen 0.03
floor, so the conditional portfolio replay was not run and nothing in
production changes.

Code-review fixes to the runner, before first commit: a missing OHLC
price stops the run instead of reading as 0; the snapshot connection is
closed explicitly (on Windows it otherwise kept the file locked); the
paired-coverage check raises instead of asserting, so python -O cannot
strip it; and source files are recorded as git blob ids with line
endings normalized, so the report can be checked against history. The
report was regenerated with the fixed runner; its screen section is
byte-identical to the original run.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

scripts/run_residual_normalization.py

@@ -0,0 +1,360 @@
+"""Frozen, paired residual-volatility normalization screen. Offline only.
+
+See docs/research/residual-normalization.md. A failed screen terminates the
+experiment; a passing screen allows the separately specified daily book replay.
+"""
+
+from __future__ import annotations
+
+import argparse
+import contextlib
+import hashlib
+import importlib.metadata
+import json
+import math
+import sqlite3
+import statistics
+import sys
+from collections import Counter
+from datetime import date, datetime, timezone
+from pathlib import Path
+
+import numpy as np
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from app.services.backtest_service import _quintile_spread, _spearman  # noqa: E402
+from app.services.momentum_service import compute_residual_12_1_momentum  # noqa: E402
+
+START = date(2016, 7, 1)
+ERA_SPLIT = date(2021, 1, 1)
+LOOKBACK = 252
+SKIP = 21
+PAIRS = LOOKBACK - SKIP
+HORIZON = 30
+MIN_CROSS = 20
+MIN_WINDOWS = 12
+IC_FLOOR = 0.03
+SEED = 20260908
+REPLICATES = 5000
+IC_BLOCK = 4
+ARMS = ("unrestricted_control", "paired_control", "normalized")
+PROTOCOL = ROOT / "docs/research/residual-normalization.md"
+
+
+def sha256(path: Path) -> str:
+    with path.open("rb") as source:
+        return hashlib.file_digest(source, "sha256").hexdigest()
+
+
+def git_blob_id(path: Path) -> str:
+    """The id ``git hash-object`` gives this file. Line endings are normalized
+    first, as git does on commit, so a CRLF checkout and an LF checkout of the
+    same source record the same id — one that can be looked up in history."""
+    data = path.read_bytes().replace(b"\r\n", b"\n")
+    return hashlib.sha1(b"blob %d\0" % len(data) + data).hexdigest()
+
+
+def read_snapshot(path: Path) -> tuple[dict, dict, dict]:
+    """Fail closed on incomplete data; never open a live application database."""
+    path = path.resolve(strict=True)
+    manifest = json.loads(Path(str(path) + ".manifest.json").read_text(encoding="utf-8"))
+    if manifest.get("complete") is not True:
+        raise ValueError("Snapshot completion manifest is not complete")
+    if any(Path(str(path) + suffix).exists() for suffix in ("-wal", "-journal")):
+        raise ValueError("Snapshot has a journal; use a completed, quiescent copy")
+    # sqlite3's own context manager only ends the transaction; closing releases
+    # the file handle, which on Windows otherwise pins the snapshot for the run.
+    with contextlib.closing(sqlite3.connect(path.as_uri() + "?mode=ro", uri=True)) as con:
+        con.execute("PRAGMA query_only=ON")
+        counts = {
+            key: con.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
+            for key, table in (
+                ("ticker_count", "tickers"),
+                ("ohlcv_row_count", "ohlcv_records"),
+                ("rank_only_count", "research_rank_only"),
+            )
+        }
+        if any(manifest.get(k) != v for k, v in counts.items()):
+            raise ValueError(f"Snapshot completion counts do not match: {counts}")
+        tickers = con.execute(
+            "SELECT id, symbol FROM tickers WHERE symbol NOT IN "
+            "(SELECT symbol FROM research_rank_only) ORDER BY symbol"
+        ).fetchall()
+        benchmark = {
+            date.fromisoformat(d[:10]): float(c)
+            for d, c in con.execute(
+                "SELECT date, close FROM benchmark_prices WHERE symbol='SPY' ORDER BY date"
+            )
+        }
+        if not benchmark or any(not math.isfinite(c) or c <= 0 for c in benchmark.values()):
+            raise ValueError("Missing or invalid SPY history")
+        prices = {}
+        for ticker_id, symbol in tickers:
+            rows = con.execute(
+                "SELECT date, open, high, low, close, volume FROM ohlcv_records "
+                "WHERE ticker_id=? ORDER BY date", (ticker_id,),
+            ).fetchall()
+            dates = [date.fromisoformat(r[0][:10]) for r in rows]
+            if len(dates) != len(set(dates)):
+                raise ValueError(f"Duplicate stock dates: {symbol}")
+            # A missing price must stop the run: read as 0 it would put a -100%
+            # return into the unrestricted control while the paired arms drop it.
+            if any(r[k] is None for r in rows for k in range(1, 5)):
+                raise ValueError(f"Missing OHLC price: {symbol}")
+            columns = (
+                ([d.toordinal() for d in dates],)
+                + tuple([float(r[k]) for r in rows] for k in range(1, 5))
+                + ([float(r[5] or 0) for r in rows],)
+            )
+            prices[symbol] = columns
+    return prices, benchmark, {**manifest, "verified_counts": counts}
+
+
+def signal_at(
+    dates: list[date], closes: list[float], i: int,
+    benchmark: dict[date, float], benchmark_index: dict[date, int],
+) -> dict:
+    """Production numerator, paired with a strictly aligned sample denominator."""
+    empty = {"unrestricted_control": None, "paired_control": None,
+             "normalized": None, "residual_sd": None, "reason": "short_history"}
+    if i < LOOKBACK:
+        return empty
+    window_dates = dates[i - LOOKBACK:i + 1]
+    window_closes = closes[i - LOOKBACK:i + 1]
+    raw = compute_residual_12_1_momentum(window_dates, window_closes, benchmark)
+    if raw is not None and not math.isfinite(raw):
+        raise ValueError("Non-finite production signal")
+    empty["unrestricted_control"] = raw
+    formation_dates = window_dates[:PAIRS + 1]
+    formation_closes = window_closes[:PAIRS + 1]
+    indices = [benchmark_index.get(d) for d in formation_dates]
+    if any(x is None for x in indices) or any(
+        b != a + 1 for a, b in zip(indices, indices[1:])
+    ):
+        return {**empty, "reason": "unaligned_formation"}
+    if any(not math.isfinite(c) or c <= 0 for c in formation_closes):
+        return {**empty, "reason": "invalid_formation_price"}
+    market = [benchmark[d] for d in formation_dates]
+    stock_returns = [b / a - 1 for a, b in zip(formation_closes, formation_closes[1:])]
+    market_returns = [b / a - 1 for a, b in zip(market, market[1:])]
+    mean_s, mean_m = statistics.fmean(stock_returns), statistics.fmean(market_returns)
+    var_m = sum((r - mean_m) ** 2 for r in market_returns)
+    if var_m <= 0 or raw is None:
+        return {**empty, "reason": "unavailable_beta"}
+    beta = sum((s - mean_s) * (m - mean_m)
+               for s, m in zip(stock_returns, market_returns)) / var_m
+    residuals = [s - beta * m for s, m in zip(stock_returns, market_returns)]
+    # Independent arithmetic verifies that the denominator uses the numerator's
+    # exact return window. It must never silently redefine production momentum.
+    if not math.isclose(sum(residuals), raw, rel_tol=1e-10, abs_tol=1e-12):
+        raise AssertionError("Production numerator parity failed")
+    sd = statistics.stdev(residuals)
+    # Treat machine-precision residuals of a perfect market tracker as zero,
+    # not as infinite signal strength. This is a numerical degeneracy check,
+    # not a tunable volatility floor.
+    numerical_zero = 64 * np.finfo(float).eps * max(1.0, max(map(abs, stock_returns)))
+    if sd <= numerical_zero:
+        return {**empty, "reason": "zero_residual_variance"}
+    normalized = raw / sd
+    if not math.isfinite(normalized):
+        raise ValueError("Non-finite normalized signal")
+    return {"unrestricted_control": raw, "paired_control": raw,
+            "normalized": normalized, "residual_sd": sd, "reason": "eligible"}
+
+
+def weekly_indices(calendar: list[date]) -> list[int]:
+    last = {}
+    for i, day in enumerate(calendar):
+        last[day.isocalendar()[:2]] = i
+    return sorted(i for i in last.values() if calendar[i] >= START and i + HORIZON < len(calendar))
+
+
+def nonoverlapping_indices(indices: list[int], horizon: int = HORIZON) -> list[int]:
+    kept = []
+    for i in sorted(indices):
+        if not kept or i >= kept[-1] + horizon:
+            kept.append(i)
+    return kept
+
+
+def interval(values) -> list[float] | None:
+    if len(values) == 0:
+        return None
+    return [float(v) for v in np.quantile(values, [0.025, 0.975])]
+
+
+def bootstrap_means(matrix: np.ndarray) -> dict:
+    """Shared moving-block indices preserve both the dates and arm pairing."""
+    n = len(matrix)
+    if n < IC_BLOCK:
+        return {"reason": "fewer_windows_than_block", "arms": {}, "paired_delta_ci95": None}
+    rng = np.random.default_rng(SEED)
+    starts = rng.integers(0, n - IC_BLOCK + 1, (REPLICATES, math.ceil(n / IC_BLOCK)))
+    indices = (starts[:, :, None] + np.arange(IC_BLOCK)).reshape(REPLICATES, -1)[:, :n]
+    means = matrix[indices].mean(axis=1)
+    return {
+        "arms": {arm: interval(means[:, i]) for i, arm in enumerate(ARMS)},
+        "paired_delta_ci95": interval(means[:, 2] - means[:, 1]),
+    }
+
+
+def summarize(windows: list[dict]) -> dict:
+    if not windows:
+        return {"windows": 0, "arms": {}, "paired_mean_ic_delta": None}
+    matrix = np.array([[w["arms"][arm]["ic"] for arm in ARMS] for w in windows])
+    uncertainty = bootstrap_means(matrix)
+    arms = {}
+    for i, arm in enumerate(ARMS):
+        values = matrix[:, i]
+        se = float(values.std(ddof=1) / math.sqrt(len(values))) if len(values) > 1 else None
+        arms[arm] = {
+            "mean_ic": float(values.mean()),
+            "ic_t_stat_descriptive": float(values.mean() / se) if se and se > 0 else None,
+            "ic_ci95": uncertainty["arms"].get(arm),
+            "positive_window_fraction": float((values > 0).mean()),
+            "avg_cross_section": statistics.fmean(w["arms"][arm]["n"] for w in windows),
+            "mean_quintile_spread": statistics.fmean(
+                w["arms"][arm]["quintile_spread"] for w in windows
+            ),
+            "reliable_window_count": len(windows) >= MIN_WINDOWS,
+        }
+    return {"windows": len(windows), "first_asof": windows[0]["asof"],
+            "last_asof": windows[-1]["asof"], "last_forward_end": windows[-1]["forward_end"],
+            "arms": arms, "paired_mean_ic_delta": float((matrix[:, 2] - matrix[:, 1]).mean()),
+            "paired_ic_delta_ci95": uncertainty["paired_delta_ci95"]}
+
+
+def screen_pass(summaries: dict) -> dict:
+    full = summaries["full"]
+    means = {era: summary.get("arms", {}).get("normalized", {}).get("mean_ic")
+             for era, summary in summaries.items()}
+    checks = {
+        "mean_ic_ge_0_03": means["full"] is not None and means["full"] >= IC_FLOOR,
+        "at_least_12_windows": full["windows"] >= MIN_WINDOWS,
+        "pre_2021_positive": means["pre_2021"] is not None and means["pre_2021"] > 0,
+        "post_2021_positive": means["post_2021"] is not None and means["post_2021"] > 0,
+    }
+    return {"passed": all(checks.values()), "checks": checks,
+            "failed_checks": [k for k, v in checks.items() if not v]}
+
+
+def run_screen(prices: dict, benchmark: dict) -> dict:
+    calendar = sorted(benchmark)
+    bench_index = {d: i for i, d in enumerate(calendar)}
+    weekly = weekly_indices(calendar)
+    observations = {i: [] for i in weekly}
+    reasons = Counter()
+    for count, (symbol, columns) in enumerate(sorted(prices.items()), 1):
+        dates = [date.fromordinal(int(o)) for o in columns[0]]
+        closes = columns[4]
+        index = {d: i for i, d in enumerate(dates)}
+        for asof in weekly:
+            i, forward = index.get(calendar[asof]), index.get(calendar[asof + HORIZON])
+            if i is None or forward is None:
+                reasons["missing_forward_or_asof"] += 1
+                continue
+            if any(not math.isfinite(closes[j]) or closes[j] <= 0 for j in (i, forward)):
+                reasons["invalid_forward_or_asof"] += 1
+                continue
+            signal = signal_at(dates, closes, i, benchmark, bench_index)
+            reasons[signal["reason"]] += 1
+            observations[asof].append({"symbol": symbol, **signal,
+                                       "forward": closes[forward] / closes[i] - 1})
+        if count % 50 == 0 or count == len(prices):
+            print(f"signal screen: {count}/{len(prices)} symbols", flush=True)
+    usable = [i for i in weekly if sum(o["normalized"] is not None for o in observations[i]) >= MIN_CROSS]
+    kept = nonoverlapping_indices(usable)
+    windows = []
+    for i in kept:
+        rows = observations[i]
+        arms = {}
+        for arm in ARMS:
+            pairs = [(o[arm], o["forward"]) for o in rows if o[arm] is not None]
+            ic = _spearman([p[0] for p in pairs], [p[1] for p in pairs])
+            if ic is None:
+                raise ValueError(f"Degenerate {arm} cross-section on {calendar[i]}")
+            arms[arm] = {"n": len(pairs), "ic": ic, "quintile_spread": _quintile_spread(pairs)}
+        # Raised, not asserted: python -O must not strip the pairing guarantee.
+        if arms["paired_control"]["n"] != arms["normalized"]["n"]:
+            raise AssertionError(f"Paired arms scored different stocks on {calendar[i]}")
+        windows.append({"asof": calendar[i].isoformat(),
+                        "forward_end": calendar[i + HORIZON].isoformat(), "arms": arms})
+    summaries = {
+        "full": summarize(windows),
+        "pre_2021": summarize([w for w in windows if w["forward_end"] < ERA_SPLIT.isoformat()]),
+        "post_2021": summarize([w for w in windows if w["asof"] >= ERA_SPLIT.isoformat()]),
+    }
+    return {"summaries": summaries, "verdict": screen_pass(summaries), "windows": windows,
+            "coverage": {"registry_symbols": len(prices), "weekly_asofs_considered": len(weekly),
+                         "usable_weekly_asofs": len(usable), "all_weekly_cell_reasons": dict(reasons),
+                         "retained_unrestricted_cells": sum(w["arms"][ARMS[0]]["n"] for w in windows),
+                         "retained_paired_cells": sum(w["arms"][ARMS[1]]["n"] for w in windows)},
+            "bootstrap": {"seed": SEED, "replicates": REPLICATES, "window_block": IC_BLOCK,
+                          "generator": "NumPy PCG64", "interval": "percentile 95%"}}
+
+
+def render_report(report: dict) -> str:
+    screen = report["screen"]
+    lines = ["# Residual normalization — signal screen", "",
+             f"Generated: {report['generated_at']}", "",
+             f"**Verdict: {'PASS' if screen['verdict']['passed'] else 'FAIL'}**", "",
+             "Previously inspected historical data; no production change or new out-of-sample claim.", "",
+             "| Era | Arm | Windows | Avg stocks | Mean IC | IC 95% interval | Quintile spread |",
+             "|---|---|---:|---:|---:|---|---:|"]
+    for era, summary in screen["summaries"].items():
+        for arm, row in summary["arms"].items():
+            ci = row["ic_ci95"]
+            ci_text = f"[{ci[0]:.4f}, {ci[1]:.4f}]" if ci else "unavailable"
+            lines.append(f"| {era} | {arm} | {summary['windows']} | {row['avg_cross_section']:.1f} "
+                         f"| {row['mean_ic']:.4f} | {ci_text} | {100 * row['mean_quintile_spread']:.2f}% |")
+    lines.extend(["", "Paired normalized-minus-control IC differences:", ""])
+    for era, summary in screen["summaries"].items():
+        if summary["windows"]:
+            lines.append(f"- {era}: {summary['paired_mean_ic_delta']:.4f}; "
+                         f"95% interval {summary['paired_ic_delta_ci95']}.")
+    lines.extend(["", f"Screen checks: `{json.dumps(screen['verdict']['checks'])}`", "",
+                  f"Portfolio stage: **{report['portfolio_stage']}**.", "",
+                  "Signal IC and quintile spreads are predictive diagnostics, not simulated strategy returns.",
+                  "Intervals resample paired blocks of four non-overlapping forward windows; they remain exploratory.",
+                  "The unrestricted control shows the effect of requiring complete formation history.", "",
+                  "[Frozen protocol](../docs/research/residual-normalization.md)", ""])
+    return "\n".join(lines)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--snapshot", type=Path, default=ROOT / "backtest_snapshots/research.sqlite")
+    parser.add_argument("--out", type=Path, default=ROOT / "reports/residual-normalization-20260908.json")
+    args = parser.parse_args()
+    snapshot = args.snapshot.resolve(strict=True)
+    before = sha256(snapshot)
+    code_paths = [Path(__file__).resolve(), PROTOCOL, ROOT / "app/services/momentum_service.py",
+                  ROOT / "app/services/backtest_service.py"]
+    hashes = {p.relative_to(ROOT).as_posix(): git_blob_id(p) for p in code_paths}
+    prices, benchmark, manifest = read_snapshot(snapshot)
+    print(f"manifest verified: {len(prices)} production-universe names; snapshot {before[:16]}", flush=True)
+    screen = run_screen(prices, benchmark)
+    if sha256(snapshot) != before or any(git_blob_id(ROOT / p) != h for p, h in hashes.items()):
+        raise RuntimeError("Snapshot, protocol, or implementation changed during the run")
+    report = {
+        "schema_version": 1, "generated_at": datetime.now(timezone.utc).isoformat(),
+        "status": "complete_signal_screen", "snapshot": str(snapshot), "snapshot_sha256": before,
+        "source_git_blobs": hashes, "manifest": manifest, "python": sys.version,
+        "packages": {p: importlib.metadata.version(p) for p in ("numpy", "sqlalchemy")},
+        "screen": screen,
+        "portfolio_stage": "eligible_pending_replay" if screen["verdict"]["passed"] else "not_run_failed_signal_screen",
+        "production_changed": False,
+    }
+    args.out.parent.mkdir(parents=True, exist_ok=True)
+    args.out.write_text(json.dumps(report, indent=2, allow_nan=False) + "\n", encoding="utf-8")
+    args.out.with_suffix(".md").write_text(render_report(report), encoding="utf-8")
+    print(json.dumps({"verdict": screen["verdict"], "summaries": screen["summaries"],
+                      "report": str(args.out)}, indent=2), flush=True)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())