gitignore: option-chain snapshots and broker state under data/

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

.gitignore

@@ -69,3 +69,10 @@ data/precomputed/
 
 # Research panels and results — fetched/rebuilt by research/lab
 data/research/
+
+# Option-chain snapshots (research.lab.fetch_options, ~11 MB/day)
+data/options/
+
+# Broker state and fills: Alpaca paper overnight book, IBKR live account
+data/overnight_paper/
+data/ibkr_live/

CLAUDE.md

@@ -102,6 +102,7 @@ PYTHONPATH=. uv run python -m research.lab.union_shield                 # SPY>MA
 PYTHONPATH=. uv run --with numba python -m research.lab.stream_portfolio   # risk-budgeted book of momentum/lowrisk/TrendRider/ETF TSMOM/BTC streams, 2006-26 (preregistered; REJECT 2026-10-08)
 PYTHONPATH=. uv run python -m research.lab.hedged_momentum           # momentum with its beta hedged by index futures (quick check; STOP 2026-10-08)
 PYTHONPATH=. uv run python -m research.lab.futures_trend [--no-proxy] # 24-market futures TSMOM on Yahoo =F + data checks (quick check; Sharpe 0.44, DATA 2026-10-08)
+uv run python -m research.lab.futures_stock_signals --output data/research/results/futures_stock_signals_new  # lagged futures vs ETF stock-exposure rules; 2026-10-17: 128 fixed runs + 12 availability diagnostics, retain baseline; report in futures_stock_signals_20261017/report.md
 PYTHONPATH=. uv run python -m research.lab.managed_futures_book [--no-proxy]  # three sleeves + managed futures (tsmom24 proxy -> DBMF), 2006-26 (preregistered; REJECT 2026-10-08)
 PYTHONPATH=. uv run python -m research.lab.small_cap_reversal       # 5-day reversal on S&P 400/600, EW-hedged, spread costs (quick check; STOP 2026-10-08)
 PYTHONPATH=. uv run python -m research.lab.turn_of_month            # turn-of-month premium 1928-2026, 3x in the window (quick check; STOP 2026-10-08)

research/lab/futures_stock_signals.py

@@ -0,0 +1,442 @@
+"""Fixed-budget, lagged futures signals for stock exposure; research only.
+
+uv run python -m research.lab.futures_stock_signals --output data/research/results/futures_stock_signals_20261017
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import math
+import pickle
+import platform
+import subprocess
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+
+import metrics
+from backtest_runner import ENGINE_VERSION, prepare_strategy, replay_prepared
+from research.lab.fetch_long_history import _cached
+from research.lab.overnight_only import legs, nightly, top_mask
+from research.lab.trials import append_trials
+from strategies.robust_momentum import signal_overnight_recovery
+
+START, END = "2006-10-02", "2026-09-24"
+BLOCKS = [(START, "2012-12-31"), ("2013-01-01", "2019-12-31"), ("2020-01-01", END)]
+POLICIES = ("equity_trend", "equity_shock", "rates_shock")
+SOURCES = {"futures": ("ES=F", "NQ=F", "ZN=F"), "etf": ("SPY", "QQQ", "IEF")}
+SETTINGS = ("default", "neighbor_low", "neighbor_high", "delay_two", "jump_masked")
+PANEL = Path("data/research/us.panel.pkl")
+SEED = 20261017
+
+
+def clean(value):
+    if isinstance(value, dict):
+        return {str(k): clean(v) for k, v in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [clean(v) for v in value]
+    if isinstance(value, (np.integer, np.floating)):
+        value = value.item()
+    if isinstance(value, float) and not math.isfinite(value):
+        return None
+    return value
+
+
+def write_json(path, value):
+    Path(path).write_text(json.dumps(clean(value), indent=2, allow_nan=False) + "\n")
+
+
+def configurations():
+    out = [{"name": "baseline"}, {"name": "static_75"}]
+    for policy in POLICIES:
+        for setting in SETTINGS:
+            for source in SOURCES:
+                look = ({"neighbor_low": 10, "neighbor_high": 40}.get(setting, 20)
+                        if policy == "equity_trend" else 20)
+                threshold = {"neighbor_low": 1.0, "neighbor_high": 2.0}.get(setting, 1.5)
+                out.append({"name": f"{source}_{policy}_{setting}", "source": source,
+                            "policy": policy, "setting": setting, "look": look,
+                            "threshold": threshold, "lag": 2 if setting == "delay_two" else 1,
+                            "mask_jumps": setting == "jump_masked"})
+    return out
+
+
+def risk_exposure(prices: pd.DataFrame, config: dict) -> tuple[pd.Series, pd.Series]:
+    """Row t trades at close t; the newest observable daily bar is t-lag.
+
+    Missing features leave the baseline unchanged and are counted as missing,
+    rather than interpreting an unavailable observation as a bearish signal.
+    """
+    if config["name"] in ("baseline", "static_75"):
+        value = 1.0 if config["name"] == "baseline" else .75
+        return pd.Series(value, index=prices.index), pd.Series(True, index=prices.index)
+    names = SOURCES[config["source"]]
+    ret = np.log(prices[list(names)] / prices[list(names)].shift(1))
+    if config["mask_jumps"]:
+        scale = ret.rolling(63, min_periods=63).std().shift(1)
+        ret = ret.mask(ret.abs() > 6 * scale)
+    equity = ret.iloc[:, :2].mean(axis=1).where(ret.iloc[:, :2].notna().all(axis=1))
+    trend = equity.rolling(config["look"], min_periods=config["look"]).sum()
+    if config["policy"] == "equity_trend":
+        valid = trend.notna()
+        risk = (trend < 0).where(valid)
+    else:
+        x = equity if config["policy"] == "equity_shock" else ret.iloc[:, 2]
+        scale = x.rolling(63, min_periods=63).std().shift(1) * np.sqrt(5)
+        shock = x.rolling(5, min_periods=5).sum() / scale.where(scale > 0)
+        valid = shock.notna()
+        risk = shock < -config["threshold"]
+        if config["policy"] == "rates_shock":
+            valid &= trend.notna()
+            risk &= trend < 0
+        risk = risk.where(valid)
+        # Hold the reduction for five sessions after a shock to limit churn.
+        risk = risk.astype(float).rolling(5, min_periods=1).max().where(valid)
+    risk = risk.shift(config["lag"])
+    known = risk.notna()
+    return 1.0 - .5 * risk.fillna(0).astype(float), known
+
+
+def overnight_exposure(entry_exposure: pd.Series) -> pd.Series:
+    # A return dated t is earned close t-1 -> open t, so the entry decision
+    # must already exist on t-1. This is separate from the source-data lag.
+    return entry_exposure.shift(1).fillna(1.0)
+
+
+def describe(returns: pd.Series) -> dict:
+    r = returns.dropna()
+    anchor = r.index[0] - pd.Timedelta(days=1)
+    equity = pd.concat([pd.Series([1.0], index=[anchor]), (1 + r).cumprod()])
+    s = metrics.raw_summary(equity)
+    return {"cagr": s["annualizedReturn"], "sharpe": s["sharpeRatio"],
+            "vol": s["annualizedVolatility"], "mdd": s["maxDrawdown"],
+            "worst_day": float(r.min()), "es_5pct": float(r[r <= r.quantile(.05)].mean()),
+            "sessions": len(r),
+            "blocks": [describe_basic(r.loc[a:b]) for a, b in BLOCKS],
+            "later_period": describe_basic(r.loc["2017-01-01":]),
+            "yearly": {str(y): float((1 + g).prod() - 1) for y, g in r.groupby(r.index.year)}}
+
+
+def describe_basic(r: pd.Series) -> dict:
+    anchor = r.index[0] - pd.Timedelta(days=1)
+    s = metrics.raw_summary(pd.concat([pd.Series([1.0], index=[anchor]), (1 + r).cumprod()]))
+    return {"cagr": s["annualizedReturn"], "sharpe": s["sharpeRatio"], "mdd": s["maxDrawdown"]}
+
+
+def paired_sharpe_interval(candidate, control, repetitions=2000):
+    pair = pd.concat([candidate, control], axis=1).dropna().to_numpy()
+    n = len(pair)
+    rng = np.random.default_rng(SEED)
+    differences = []
+    for _ in range(repetitions):
+        starts = rng.integers(0, n - 20 + 1, size=math.ceil(n / 20))
+        idx = (starts[:, None] + np.arange(20)).ravel()[:n]
+        sample = pair[idx]
+        sharpe = sample.mean(axis=0) / sample.std(axis=0, ddof=1) * np.sqrt(252)
+        differences.append(sharpe[0] - sharpe[1])
+    return {"lower": float(np.quantile(differences, .025)),
+            "upper": float(np.quantile(differences, .975)),
+            "block_sessions": 20, "repetitions": repetitions, "seed": SEED,
+            "role": "paired descriptive interval; dependent on historical stationarity, not independent OOS"}
+
+
+def data_inventory(prices, dates):
+    out = {}
+    for name, p in prices.items():
+        x = p.reindex(dates)
+        r = np.log(x / x.shift(1))
+        jump = r.abs() > 6 * r.rolling(63, min_periods=63).std().shift(1)
+        out[name] = {"first": str(p.first_valid_index().date()), "last": str(p.last_valid_index().date()),
+                     "coverage": float(x.loc[START:END].notna().mean()), "jump_days": int(jump.sum())}
+    return out
+
+
+def protocol(destination, configs):
+    source_paths = [Path(__file__), Path("execution.py"), Path("signals.py"), Path("backtest_runner.py"),
+                    Path("metrics.py"), Path("trader.py"), Path("research/lab/overnight_only.py"),
+                    Path("strategies/base.py"), Path("strategies/robust_momentum.py"),
+                    Path("tests/test_futures_stock_signals.py")]
+    data_paths = [PANEL] + [Path(_cached(t)) for tickers in SOURCES.values() for t in tickers]
+    missing = [str(p) for p in data_paths if not p.exists()]
+    if missing:
+        raise FileNotFoundError(f"Required cached data missing: {missing}")
+    record = {
+        "created_at": datetime.now(timezone.utc).isoformat(),
+        "claim": "Lagged futures information improves stock exposure beyond matched ETF signals and simple de-risking.",
+        "primary": "futures_equity_shock_default on overnight_recovery_top10",
+        "hypotheses": {"incremental_information": "Futures beats both baseline and the same ETF rule after costs.",
+                       "redundant_or_stale": "ETF control matches it, or costs/delay remove the benefit."},
+        "span": [START, END], "blocks": BLOCKS,
+        "role": "All historical periods previously inspected; fixed-rule development study, not independent OOS.",
+        "design_verdict": "APPROVED TO RUN: user authorized local research and backtests; one local process, no paid data.",
+        "baseline": "production RobustMomentumStrategy overnight_recovery_top10, PIT panel, unchanged target stock list",
+        "secondary": "nightly top10 by the same score through entry-day minus one; MOC -> next open, flat intraday",
+        "engine": ENGINE_VERSION, "capital": 10000, "integer_shares": True,
+        "daily_costs": [.001, .0025], "daily_fixed_fee": 2.0, "resize_band": .1,
+        "overnight_costs": [.0001, .0003], "overnight_fixed_fee": 0.0,
+        "cash_interest": 0.0, "sharpe": "mean daily return / sample standard deviation * sqrt(252), no rf subtraction",
+        "controlled": "identical stock selections, data, timing, capital and per-engine fees for all exposure rules",
+        "configurations": configs,
+        "budget": {"profiles": len(configs), "engines": 2, "costs_per_engine": 2,
+                   "backtests": len(configs) * 4, "bootstrap_repetitions": 2000},
+        "rules": {"trend": "mean ES/NQ (or SPY/QQQ) log return over 20 sessions < 0",
+                  "shock": "5-session return / (prior 63-session daily vol * sqrt(5)) < -1.5",
+                  "rates": "ZN (or IEF) shock AND equity trend < 0",
+                  "exposure": "0.5 in risk state, 1 otherwise; shock risk held five sessions",
+                  "neighbors": "trend 10/40 sessions; shock threshold 1.0/2.0",
+                  "timing": "entry close t uses source data <= t-1; overnight exit t additionally shifts exposure one session",
+                  "missing": "no backward fill; missing signal preserves baseline, coverage recorded; require >=99% usable signals",
+                  "jumps": "default untouched; sensitivity masks >6 times trailing63 volatility without using future observations"},
+        "gates": {"G1": "full-span Sharpe >= baseline +0.10 at both costs",
+                  "G2": "full-span Sharpe >= matched ETF +0.05 AND static75 at both costs",
+                  "G3": "CAGR >= baseline -2pp AND MaxDD improves >=3pp at both costs",
+                  "G4": "beats baseline and ETF Sharpe in >=2/3 blocks AND later2017+ at both costs",
+                  "G5": "base-cost later2017+ paired block-bootstrap 95% lower bound of Sharpe delta >0 vs baseline AND ETF",
+                  "G6": "base-cost Sharpe delta vs ETF positive for both neighbors, lag2 and jump-masked variants",
+                  "G7": "default source-feature availability >=99%"},
+        "decisions": "All gates pass -> candidate for forward validation only; else retain baseline. No live changes.",
+        "excluded": {"basis": "continuous contracts lack contract maturity, synchronous spot, rate and dividend inputs",
+                     "positions": "no release-time-verified historical futures OI/COT panel",
+                     "premarket": "daily open/close cache cannot reconstruct premarket decision-time quotes",
+                     "industry_selection": "separate next-stage test; this study only changes total stock exposure"},
+        "known_limits": ["Yahoo continuous contracts are not roll-adjusted; jump masking is sensitivity, not reconstruction.",
+                         "Futures/ETF timestamps and duration differ; lagged inputs do not isolate pure price discovery.",
+                         "Historical PIT member price coverage is incomplete, especially before 2010.",
+                         "Adjusted-share prices and delisting settlement assumptions follow the existing ledger.",
+                         "Overnight open fallback to same-day close is counted and tested with zero-return sensitivity.",
+                         "No independent forward sample; descriptive bootstrap does not resolve model-selection bias."],
+        "references": ["https://www.cmegroup.com/education/articles-and-reports/faq-daily-settlement-price-determination-time-change.html",
+                       "https://www.cmegroup.com/trading/equity-index/fairvalue.html",
+                       "https://www.cftc.gov/MarketReports/CommitmentsofTraders/index.htm"],
+        "environment": {"python": sys.version, "platform": platform.platform(), "numpy": np.__version__,
+                        "pandas": pd.__version__, "command": sys.argv,
+                        "git": subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip()},
+        "source_hashes": {str(p): hashlib.sha256(p.read_bytes()).hexdigest() for p in source_paths},
+        "data_hashes": {str(p): hashlib.sha256(p.read_bytes()).hexdigest() for p in data_paths},
+    }
+    destination.mkdir(parents=True, exist_ok=False)
+    write_json(destination / "protocol.json", record)
+    (destination / "experiment_card.md").write_text(
+        "# Experiment: Lagged Futures Information for Stock Exposure\n\n"
+        "> Status: Approved for the user-authorized local research run.\n\n"
+        "## Claim and Decision\n\n"
+        + record["claim"] + "\n\n"
+        "Futures-information hypothesis predicts an improvement over the unchanged stock strategy, "
+        "matched ETF rule and static 75% exposure. The redundancy hypothesis predicts that the ETF "
+        "rule matches it or costs and a further one-session lag remove the benefit. "
+        "All seven fixed gates must pass before recommending a forward experiment; no production promotion.\n\n"
+        "## Setup\n\n"
+        "Only exposure changes. Production stock targets, PIT panel, capital and fees stay fixed. "
+        "Separate production daily-close ledger and nightly MOC/MOO models have their own matched controls. "
+        "Three fixed signal families, two neighboring parameters, lag2 and causal jump masking; "
+        "128 runs total. All parameter values and gates are in protocol.json, written before any evaluation. "
+        "No fresh OOS label is assigned to previously inspected history.\n\n"
+        "## Expected Artifacts and Review\n\n"
+        "Daily returns, exposures, production-ledger trades, costs, subperiods, paired 20-session "
+        "block-bootstrap intervals and complete trial records. Results are tables; no result figure planned. "
+        "Local low-cost execution is authorized by the user's instruction to start research and backtests.\n\n"
+        "## Reproduction Details\n\n"
+        "protocol.json records the command, runtime versions, Git revision and source/data SHA256. "
+        "No downloads, paid sources, shared-host jobs or live-account writes.\n\n"
+        "## Benchmark Audit\n\n"
+        "| Issue | Verdict | Evidence |\n|---|---|---|\n"
+        "| Timing | PASS | Source lag1 and extra entry-to-exit shift for overnight returns; causal tests |\n"
+        "| Fair baselines | PASS | Same stocks, costs and execution; ETF and static-exposure controls |\n"
+        "| Search budget | PASS | Fixed 32 profiles and both costs before execution; all runs retained |\n"
+        "| Uncertainty | PASS | Paired block bootstrap and three chronological blocks |\n"
+        "| Independent OOS | NEEDS EVIDENCE | All historical periods previously inspected |\n"
+        "| Continuous-contract data | NEEDS EVIDENCE | No contract/roll history; use sensitivity, limit claims |\n"
+        "| Serving workload | N/A | Financial backtest, no serving system or request trace |\n\n"
+        "## Results\n\nRecorded after execution in results.json and report.md.\n")
+    return record
+
+
+def decisions(rows, curves, known):
+    out = []
+    for engine, costs in (("daily", (.001, .0025)), ("overnight", (.0001, .0003))):
+        indexed = {(r["profile"], r["cost"]): r for r in rows if r["engine"] == engine}
+        for policy in POLICIES:
+            name, control = f"futures_{policy}_default", f"etf_{policy}_default"
+            checks = {f"G{i}": True for i in range(1, 8)}
+            comparisons = []
+            for cost in costs:
+                f, e, b, s = (indexed[(k, cost)] for k in (name, control, "baseline", "static_75"))
+                checks["G1"] &= f["sharpe"] >= b["sharpe"] + .10
+                checks["G2"] &= f["sharpe"] >= e["sharpe"] + .05 and f["sharpe"] >= s["sharpe"]
+                checks["G3"] &= f["cagr"] >= b["cagr"] - .02 and f["mdd"] >= b["mdd"] + .03
+                blocks = sum(x["sharpe"] > max(y["sharpe"], z["sharpe"])
+                             for x, y, z in zip(f["blocks"], b["blocks"], e["blocks"]))
+                checks["G4"] &= blocks >= 2 and f["later_period"]["sharpe"] > max(
+                    b["later_period"]["sharpe"], e["later_period"]["sharpe"])
+                comparisons.append({"cost": cost, "sharpe_delta_baseline": f["sharpe"] - b["sharpe"],
+                                    "sharpe_delta_etf": f["sharpe"] - e["sharpe"], "blocks_better_than_both": blocks})
+            intervals = {}
+            for other in ("baseline", control):
+                intervals[other] = paired_sharpe_interval(
+                    curves[f"{engine}|{name}|{costs[0]}"] .loc["2017-01-01":],
+                    curves[f"{engine}|{other}|{costs[0]}"] .loc["2017-01-01":])
+            checks["G5"] = all(c["lower"] > 0 for c in intervals.values())
+            sensitivities = {}
+            for setting in SETTINGS[1:]:
+                f, e = (indexed[(f"{source}_{policy}_{setting}", costs[0])] for source in SOURCES)
+                sensitivities[setting] = f["sharpe"] - e["sharpe"]
+            checks["G6"] = all(v > 0 for v in sensitivities.values())
+            checks["G7"] = known[name] >= .99
+            out.append({"engine": engine, "policy": policy, "checks": checks,
+                        "comparisons": comparisons, "later_period_bootstrap": intervals,
+                        "sensitivity_sharpe_delta_etf": sensitivities,
+                        "decision": "forward_candidate_only" if all(checks.values()) else "retain_baseline"})
+    return out
+
+
+def report(destination, result):
+    labels = {"baseline": "原策略", "static_75": "固定75%仓位",
+              "futures_equity_trend_default": "股指期货20日趋势", "etf_equity_trend_default": "对应ETF20日趋势",
+              "futures_equity_shock_default": "股指期货5日急跌", "etf_equity_shock_default": "对应ETF5日急跌",
+              "futures_rates_shock_default": "国债期货急跌＋股指走弱", "etf_rates_shock_default": "对应ETF国债急跌＋股指走弱"}
+    passed = sum(d["decision"] == "forward_candidate_only" for d in result["decisions"])
+    text = ["# 期货信号辅助股票仓位：固定规则回测", "",
+            f"结论：三类期货信号、两种持有方式共6个默认方案，{passed}个通过预定标准。", "",
+            "本研究只改变股票总仓位，未交易期货，也未改变股票评分。",
+            "完成32种配置×2种持有方式×2档成本，共128组预先固定的回测。",
+            "全部历史区间已在过去研究中查看；结果属于开发证据，不能称为独立样本外或实盘收益。", "",
+            f"区间：{START} 至 {END}。指标中的回撤为负数；Sharpe 未减无风险利率。", ""]
+    for engine, cost in (("daily", .001), ("overnight", .0001)):
+        text += ["## " + ("连续持仓版：原股票策略" if engine == "daily" else "隔夜版：收盘买、次日开盘卖"), "",
+                 "| 版本 | 年化收益 | Sharpe | 最大回撤 | 平均目标仓位 |", "|---|---:|---:|---:|---:|"]
+        for row in result["rows"]:
+            if row["engine"] == engine and row["cost"] == cost and row["profile"] in labels:
+                text.append(f"| {labels[row['profile']]} | {row['cagr']:.2%} | {row['sharpe']:.3f} | "
+                            f"{row['mdd']:.2%} | {row['mean_target_exposure']:.1%} |")
+        text.append("")
+    text += ["连续持仓版：生产执行账本、$10,000、整股、单边10/25 bp及每单$2，10%调仓带。",
+             "隔夜版：每日全额往返、单边1/3 bp、无固定佣金、分数股；两种引擎分别比较，不互相排名。", "",
+             "## 验收", "", "| 引擎 | 信号 | 未通过门槛 | 结论 |", "|---|---|---|---|"]
+    for d in result["decisions"]:
+        failed = ", ".join(k for k, v in d["checks"].items() if not v) or "无"
+        text.append(f"| {d['engine']} | {d['policy']} | {failed} | {d['decision']} |")
+    text += ["", "G1：超越原策略；G2：超越ETF及简单降仓；G3：收益和回撤门槛；G4：分段稳定；",
+             "G5：2017年后配对分块bootstrap区间；G6：邻近参数、延迟及跳变敏感性；G7：数据覆盖。", "",
+             "bootstrap区间为名义95%区间，未作多重比较校正，不能代替独立前向验证。", "",
+             "## 数据与边界", "", "| 默认期货特征 | 可用率 |", "|---|---:|"]
+    for policy in POLICIES:
+        key = f"futures_{policy}_default"
+        text.append(f"| {labels[key]} | {result['audit']['signal_coverage'][key]:.2%} |")
+    text += ["", "原始日线有缺口，滚动窗口会进一步放大特征不可用范围；覆盖情况见data_audit.json。",
+             f"隔夜模型有{result['audit']['overnight_missing_open_nights']}个夜晚涉及个别选中股票缺开盘价；",
+             "已将这些股票的该夜收益设为0做敏感性检查，完整结果列于results.json。",
+             "早期指数成员价格覆盖不足；逐年观测成员数列于data_audit.json。", "",
+             "未测试真正期现基差、期货持仓/COT、盘前分钟信号和行业重排；对应数据缺口列于protocol.json。",
+             "下一步优先取得带明确时间戳的期货分钟数据、逐月合约及持仓发布记录，再检验信息增量。", "",
+             "原始日线、源代码及股票面板的SHA256列于protocol.json；完整参数与成本结果见results.csv、",
+             "results.json；逐日收益、目标仓位见daily_returns.csv.gz和exposure.csv；生产账本成交见trades.csv.gz。", ""]
+    ablation = destination / "availability_results.json"
+    if ablation.exists():
+        a = json.loads(ablation.read_text())
+        text += ["## 数据缺口匹配诊断", "",
+                 "主实验结束后另固定12组诊断：在期货特征不可用的日期，让ETF版本也保持原仓位。",
+                 "该诊断只解释数据可用性，不参与方案选择。下表是期货Sharpe减去此ETF对照的Sharpe。", "",
+                 "| 信号 | 连续持仓低成本档 | 隔夜低成本档 |", "|---|---:|---:|"]
+        for policy in POLICIES:
+            selected = [r for r in a["rows"] if r["profile"] == f"etf_{policy}_availability_matched"
+                        and r["cost"] == (.001 if r["engine"] == "daily" else .0001)]
+            delta = {r["engine"]: r["futures_sharpe_minus_matched_etf"] for r in selected}
+            text.append(f"| {labels[f'futures_{policy}_default']} | {delta['daily']:+.4f} | {delta['overnight']:+.4f} |")
+        text += ["", "完整两档成本及仅在可观测日期计算的信号一致率见availability_results.json。", ""]
+    if (destination / "execution_source.py").exists():
+        text += ["首次执行的脚本保存在execution_source.py，匹配预先登记的源代码SHA256；后续仅改善报告排版。", ""]
+    text += ["复现：", "", "```bash",
+             f"uv run python -m research.lab.futures_stock_signals --output {destination}_rerun", "```", ""]
+    (destination / "report.md").write_text("\n".join(text))
+
+
+def run(destination: Path):
+    configs = configurations()
+    record = protocol(destination, configs)
+    print(f"Protocol saved before evaluation: {destination}; budget {record['budget']}", flush=True)
+    with PANEL.open("rb") as fh:
+        p = pickle.load(fh)
+    prices = pd.DataFrame({t: pd.read_csv(_cached(t), index_col=0, parse_dates=True)["Close"]
+                           for tickers in SOURCES.values() for t in tickers})
+    prices = prices.reindex(p.dates)
+    if p.dates[-1] < pd.Timestamp(END):
+        raise ValueError("Stock panel ends before the fixed evaluation interval")
+    panels = {field: getattr(p, field) for field in ("open", "high", "low", "close", "volume")}
+    panels["close"] = p.close.join(p.extra["etf"][["SPY"]])
+    plan = prepare_strategy("overnight_recovery_top10", "us", panels, list(p.close.columns),
+                            member=p.member, start=START, end=END)
+    score = signal_overnight_recovery({"open": p.open, "close": p.close,
+                                      "eligible": p.member & p.close.notna()})
+    selected = top_mask(score.shift(2), 10)
+    overnight, _ = legs(p)
+    gross_night = nightly(selected, overnight).loc[START:END]
+    missing_opens = selected & p.open.isna()
+    zero_open_sensitivity = nightly(selected, overnight.mask(p.open.isna(), 0.0)).loc[START:END]
+    entries = {}
+    known = {}
+    for config in configs:
+        exposure, valid = risk_exposure(prices, config)
+        entries[config["name"]] = exposure
+        known[config["name"]] = float(valid.loc[START:END].mean())
+    pd.DataFrame(entries).loc[START:END].to_csv(destination / "exposure.csv")
+    audit = {"daily_bar_inventory": data_inventory(prices, p.dates), "signal_coverage": known,
+             "pit_mean_observed_members_by_year": p.member.sum(axis=1).groupby(p.dates.year).mean().to_dict(),
+             "overnight_selected_missing_opens": int(missing_opens.loc[START:END].sum().sum()),
+             "overnight_missing_open_nights": int(missing_opens.loc[START:END].any(axis=1).sum()),
+             "max_gross_return_change_zero_open_sensitivity": float((gross_night - zero_open_sensitivity).abs().max())}
+    write_json(destination / "data_audit.json", audit)
+    rows, curves, trades = [], {}, []
+    for config in configs:
+        name = config["name"]
+        entry = entries[name].loc[START:END]
+        daily_plan = dict(plan, targets=plan["targets"].mul(entry, axis=0))
+        for cost in record["daily_costs"]:
+            ledger = replay_prepared(daily_plan, 10000, cost, 2.0, integer_shares=True, resize_band=.1)
+            r = ledger["equity"].pct_change().dropna()
+            if ledger["cash"].min() < -1e-7:
+                raise AssertionError("Unfunded replay")
+            row = {"engine": "daily", "profile": name, "cost": cost, **describe(r),
+                   "mean_target_exposure": float(entry.mean()),
+                   "mean_actual_exposure": float((1 - ledger["cash"] / ledger["equity"]).dropna().mean()),
+                   "orders": len(ledger["trade_log"]),
+                   "fees": sum(t["commission"] for t in ledger["trade_log"]),
+                   "minimum_cash": float(ledger["cash"].min())}
+            rows.append(row)
+            curves[f"daily|{name}|{cost}"] = r
+            trades.extend({"profile": name, "cost": cost, **t} for t in ledger["trade_log"])
+            append_trials(destination / "trials.jsonl", [row], {"study": "futures_stock_signals", "config": config})
+            print(f"daily {name:43s} {cost:.4f}: CAGR {row['cagr']:.2%} Sh {row['sharpe']:.3f} DD {row['mdd']:.2%}", flush=True)
+        night_exp = overnight_exposure(entries[name]).reindex(gross_night.index)
+        for cost in record["overnight_costs"]:
+            r = night_exp * (gross_night - 2 * cost)
+            sensitivity = describe_basic(night_exp * (zero_open_sensitivity - 2 * cost))
+            row = {"engine": "overnight", "profile": name, "cost": cost, **describe(r),
+                   "mean_target_exposure": float(night_exp.mean()), "zero_open_sensitivity": sensitivity}
+            rows.append(row)
+            curves[f"overnight|{name}|{cost}"] = r
+            append_trials(destination / "trials.jsonl", [row], {"study": "futures_stock_signals", "config": config})
+    verdicts = decisions(rows, curves, known)
+    result = {"protocol": record, "audit": audit, "rows": rows, "decisions": verdicts}
+    write_json(destination / "results.json", result)
+    pd.DataFrame([{k: v for k, v in r.items() if k not in ("blocks", "later_period", "yearly", "zero_open_sensitivity")}
+                  for r in rows]).to_csv(destination / "results.csv", index=False)
+    pd.DataFrame(curves).to_csv(destination / "daily_returns.csv.gz", compression="gzip")
+    pd.DataFrame(trades).to_csv(destination / "trades.csv.gz", index=False, compression="gzip")
+    report(destination, result)
+    append_trials(Path("data/research/results/trials_us.jsonl"),
+                  [{"name": f"futures_stock_signals:{r['engine']}:{r['profile']}", "cost": r["cost"],
+                    "cagr": r["cagr"], "sharpe": r["sharpe"], "mdd": r["mdd"]} for r in rows],
+                  {"study": "futures_stock_signals", "role": record["role"], "path": str(destination)})
+    print(json.dumps(clean(verdicts), indent=2), flush=True)
+    print(f"Saved {destination / 'report.md'}", flush=True)
+    return result
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--output", type=Path, required=True)
+    run(parser.parse_args().output)

tests/test_futures_stock_signals.py

@@ -0,0 +1,64 @@
+import unittest
+
+import numpy as np
+import pandas as pd
+
+from research.lab.futures_stock_signals import configurations, overnight_exposure, risk_exposure
+
+
+class FuturesTimingTests(unittest.TestCase):
+    def setUp(self):
+        dates = pd.bdate_range("2020-01-01", periods=180)
+        t = np.arange(len(dates))
+        equity = 100 * np.exp(.001 * t + .002 * np.sin(t))
+        rates = 100 * np.exp(.0001 * t + .0002 * np.sin(t))
+        self.prices = pd.DataFrame({"ES=F": equity, "NQ=F": equity, "ZN=F": rates,
+                                    "SPY": equity, "QQQ": equity, "IEF": rates}, index=dates)
+
+    def test_future_prices_cannot_change_earlier_decisions(self):
+        changed = self.prices.copy()
+        changed.iloc[110:, :3] *= .7
+        for config in configurations():
+            original, known = risk_exposure(self.prices, config)
+            modified, modified_known = risk_exposure(changed, config)
+            last = 110 + config.get("lag", 1)
+            with self.subTest(config=config["name"]):
+                pd.testing.assert_series_equal(original.iloc[:last], modified.iloc[:last])
+                pd.testing.assert_series_equal(known.iloc[:last], modified_known.iloc[:last])
+
+    def test_same_day_futures_crash_is_not_usable_for_entry_or_preceding_night(self):
+        config = next(c for c in configurations() if c["name"] == "futures_equity_shock_default")
+        changed = self.prices.copy()
+        changed.iloc[110:, :2] *= .7
+        entry, _ = risk_exposure(changed, config)
+        self.assertEqual(entry.iloc[110], 1.0)
+        self.assertEqual(entry.iloc[111], .5)
+        night = overnight_exposure(entry)
+        self.assertEqual(night.iloc[111], 1.0)
+        self.assertEqual(night.iloc[112], .5)
+
+    def test_identical_etf_and_futures_inputs_have_identical_rules(self):
+        configs = {c["name"]: c for c in configurations()}
+        for name, config in configs.items():
+            if not name.startswith("futures_"):
+                continue
+            with self.subTest(config=name):
+                a, av = risk_exposure(self.prices, config)
+                b, bv = risk_exposure(self.prices, configs[name.replace("futures_", "etf_", 1)])
+                pd.testing.assert_series_equal(a, b, check_names=False)
+                pd.testing.assert_series_equal(av, bv, check_names=False)
+
+    def test_unavailable_prices_are_not_backward_filled_from_future(self):
+        config = next(c for c in configurations() if c["name"] == "futures_equity_shock_default")
+        missing = self.prices.copy()
+        missing.iloc[110, :2] = np.nan
+        exposure, known = risk_exposure(missing, config)
+        self.assertFalse(known.iloc[111])
+        self.assertEqual(exposure.iloc[111], 1.0)
+        prefix, prefix_known = risk_exposure(missing.iloc[:112], config)
+        pd.testing.assert_series_equal(exposure.iloc[:112], prefix)
+        pd.testing.assert_series_equal(known.iloc[:112], prefix_known)
+
+
+if __name__ == "__main__":
+    unittest.main()