feat: add Phase A research matrix (vol target, fill, corr, SE/DSR)

Ship shared Sharpe SE/PSR diagnostics, next-open fill and equity-curve vol targeting in the portfolio simulator, re-derived fip_id, and a checkpointed offline matrix runner for Mac-side validation sweeps.

app/services/backtest_service.py

@@ -688,6 +688,18 @@ TIME_EXIT_DAYS = (5, 10, 21, 30)
 # round trip, converted into R via the setup's stop distance (the 1R unit).
 COST_PER_SIDE = 0.001
 
+# Portfolio vol-targeting defaults (Barroso & Santa-Clara style, equity-curve vol).
+VOL_TARGET_LOOKBACK_HEADLINE = 60
+VOL_TARGET_LOOKBACK_SENSITIVITY = (20, 126)
+VOL_TARGET_GRID = (0.15, 0.20, 0.25)
+VOL_TARGET_CLAMP_HEADLINE = (0.5, 1.5)
+VOL_TARGET_CLAMP_WIDE = (0.25, 2.0)
+
+# Entry fill modes for the capital-constrained book simulator.
+FILL_MODE_CLOSE = "close"
+FILL_MODE_NEXT_OPEN = "next_open"
+FILL_MODES = (FILL_MODE_CLOSE, FILL_MODE_NEXT_OPEN)
+
 
 def _cost_r(cand: dict) -> float:
     """Round-trip transaction cost in R units: two sides over the 1R stop
@@ -800,6 +812,42 @@ def _realized_vol_6m(closes: list[float], i: int) -> float | None:
     return compute_realized_vol_6m(closes[: i + 1])
 
 
+def _fip_id(closes: list[float], i: int) -> float | None:
+    """Da/Gurun/Warachka information discreteness over the 12-1 formation window.
+
+    Formation matches ``mom_12_1``: cumulative return from close[i-252] to
+    close[i-21] (231 daily returns ending one month before as-of).
+
+        ID = sign(PRET) × (%neg − %pos)
+
+    where %pos / %neg are fractions of up / down days over the formation window
+    (zero-return days count in neither numerator, but remain in the denominator).
+    Lower ID = smoother / more continuous path → expect negative cross-sectional
+    IC (continuous-information winners outperform).
+    """
+    if i - 252 < 0 or closes[i - 252] <= 0 or closes[i - 21] <= 0:
+        return None
+    pret = closes[i - 21] / closes[i - 252] - 1.0
+    rets: list[float] = []
+    for k in range(i - 251, i - 20):
+        prev = closes[k - 1]
+        if prev <= 0:
+            return None
+        rets.append(closes[k] / prev - 1.0)
+    if len(rets) < 200:
+        return None
+    n = len(rets)
+    pct_pos = sum(1 for r in rets if r > 0) / n
+    pct_neg = sum(1 for r in rets if r < 0) / n
+    if pret > 0:
+        sign = 1.0
+    elif pret < 0:
+        sign = -1.0
+    else:
+        sign = 0.0
+    return sign * (pct_neg - pct_pos)
+
+
 def _signal_values(
     dates: list[date],
     closes: list[float],
@@ -816,6 +864,7 @@ def _signal_values(
     is closeness to the trailing 52-week high (George/Hwang anchoring effect:
     higher = nearer the high, expect positive IC). ``vol_6m`` is 126-day realized
     volatility (expect negative IC if the low-volatility anomaly holds).
+    ``fip_id`` is Da/Gurun/Warachka information discreteness (expect negative IC).
     """
     out: dict[str, float] = {}
     if i - 252 >= 0 and closes[i - 252] > 0:
@@ -823,6 +872,9 @@ def _signal_values(
         residual = _residual_momentum_12_1(dates, closes, i, benchmark_closes)
         if residual is not None:
             out["mom_12_1_resid"] = residual
+        fip = _fip_id(closes, i)
+        if fip is not None:
+            out["fip_id"] = fip
     if i - 126 >= 0 and closes[i - 126] > 0:
         out["mom_6_1"] = closes[i - 21] / closes[i - 126] - 1.0
     if i - 63 >= 0 and closes[i - 63] > 0:
@@ -1427,6 +1479,221 @@ SIM_STARTING_CAPITAL = 10_000.0
 SIM_MAX_POSITIONS = 10
 SIM_RISK_PER_TRADE = 0.01  # fraction of equity risked per position (entry→stop)
 SIM_NOTIONAL_CAP = 0.20    # max fraction of equity per position (no margin)
+_EULER_MASCHERONI = 0.5772156649015329
+
+
+def _sample_moments(rets: list[float]) -> dict[str, float | int | None]:
+    """Mean / std / skew / kurtosis of a return series. Kurtosis is raw (not excess)."""
+    n = len(rets)
+    if n < 3:
+        return {
+            "n": n,
+            "mean": None,
+            "std": None,
+            "skew": None,
+            "kurtosis": None,
+        }
+    mean = sum(rets) / n
+    # Sample variance with n-1 (matches the historical Sharpe path).
+    var = sum((x - mean) ** 2 for x in rets) / (n - 1)
+    if var <= 0:
+        return {
+            "n": n,
+            "mean": mean,
+            "std": 0.0,
+            "skew": None,
+            "kurtosis": None,
+        }
+    std = math.sqrt(var)
+    m3 = sum((x - mean) ** 3 for x in rets) / n
+    m4 = sum((x - mean) ** 4 for x in rets) / n
+    skew = m3 / (std ** 3) if std > 0 else None
+    kurtosis = m4 / (std ** 4) if std > 0 else None
+    return {
+        "n": n,
+        "mean": mean,
+        "std": std,
+        "skew": skew,
+        "kurtosis": kurtosis,
+    }
+
+
+def _mertens_sharpe_se(
+    sharpe_periodic: float,
+    n: int,
+    skew: float | None,
+    kurtosis: float | None,
+) -> float | None:
+    """Mertens/Lo standard error of the non-annualized Sharpe ratio.
+
+    Accounts for non-normality via skew and kurtosis — material for this
+    right-skewed momentum book. Returns SE of the *periodic* Sharpe (mean/std);
+    annualize by multiplying by sqrt(252) alongside the point estimate.
+    """
+    if n < 3:
+        return None
+    g3 = 0.0 if skew is None else float(skew)
+    # Fall back to Gaussian kurtosis (=3) when undefined.
+    g4 = 3.0 if kurtosis is None else float(kurtosis)
+    sr = float(sharpe_periodic)
+    inside = 1.0 + 0.5 * sr * sr - g3 * sr + ((g4 - 3.0) / 4.0) * sr * sr
+    if inside <= 0:
+        return None
+    return math.sqrt(inside / (n - 1))
+
+
+def sharpe_diagnostics(rets: list[float], *, periods_per_year: float = 252.0) -> dict:
+    """Annualized Sharpe plus Mertens SE and PSR vs zero for a daily return series.
+
+    Additive report fields only — safe for the weekly production backtest and every
+    research matrix row. Deflated Sharpe is *not* included here: DSR needs a
+    pre-registered trial count N and is computed by ``deflated_sharpe_ratio``.
+    """
+    moments = _sample_moments(rets)
+    n = int(moments["n"] or 0)
+    mean = moments["mean"]
+    std = moments["std"]
+    skew = moments["skew"]
+    kurtosis = moments["kurtosis"]
+    empty = {
+        "sharpe": None,
+        "sharpe_se": None,
+        "psr": None,
+        "n_returns": n,
+        "return_skew": None if skew is None else round(float(skew), 4),
+        "return_kurtosis": None if kurtosis is None else round(float(kurtosis), 4),
+    }
+    if mean is None or std is None or std <= 0 or n < 3:
+        return empty
+    sr_p = float(mean) / float(std)
+    scale = math.sqrt(periods_per_year)
+    sharpe = sr_p * scale
+    se_p = _mertens_sharpe_se(sr_p, n, None if skew is None else float(skew),
+                              None if kurtosis is None else float(kurtosis))
+    se = se_p * scale if se_p is not None else None
+    psr = None
+    if se is not None and se > 0:
+        # PSR(SR*=0): Φ(sharpe / se) using the annualized numbers (scale cancels).
+        psr = statistics.NormalDist().cdf(sharpe / se)
+    return {
+        "sharpe": round(sharpe, 2),
+        "sharpe_se": round(se, 3) if se is not None else None,
+        "psr": round(psr, 4) if psr is not None else None,
+        "n_returns": n,
+        "return_skew": None if skew is None else round(float(skew), 4),
+        "return_kurtosis": None if kurtosis is None else round(float(kurtosis), 4),
+    }
+
+
+def deflated_sharpe_ratio(
+    sharpe: float | None,
+    sharpe_se: float | None,
+    n_trials: int,
+    *,
+    n_returns: int | None = None,
+    return_skew: float | None = None,
+    return_kurtosis: float | None = None,
+) -> float | None:
+    """Bailey & López de Prado Deflated Sharpe Ratio for a multi-arm matrix.
+
+    ``n_trials`` must be the *pre-registered* arm count for the matrix (not
+    invented after the fact). Returns None when inputs are insufficient — never
+    fabricates a DSR for a standalone single-arm run.
+    """
+    if (
+        sharpe is None
+        or sharpe_se is None
+        or sharpe_se <= 0
+        or n_trials < 2
+        or n_returns is None
+        or n_returns < 3
+    ):
+        return None
+    # Expected maximum Sharpe under the null across N independent trials
+    # (Bailey & López de Prado 2014), using Euler-Mascheroni blending of
+    # extreme-value quantiles. Variance under the null uses the observed
+    # higher moments evaluated at SR=0 → SE_null = 1/sqrt(n-1) * ann_scale,
+    # recovered from the reported annualized SE via the Mertens factor at the
+    # observed SR (we back out the periodic SE from sharpe_se).
+    nd = statistics.NormalDist()
+    # Annualized expected max of N zero-mean unit-variance Sharpes, then
+    # scaled by the null SE. With SR*=0 null variance of the *annualized*
+    # Sharpe is approximately (periods_per_year)/(n-1) when returns are IID
+    # normal; recover ann_scale^2/(n-1) from se under Gaussian assumption as
+    # a fallback, but prefer moment-adjusted null at SR=0:
+    # SE_null_periodic = sqrt(1/(n-1)); SE_null_ann = SE_null_p * (se/se_p).
+    # We don't store se_p, so invert: se_ann / sqrt(1+0.5 SR_p^2 - ...) * sqrt(1/(n-1))
+    # Simpler standard form used in practice:
+    # SR* = se_null * ((1-γ) Z^{-1}(1-1/N) + γ Z^{-1}(1-1/(N e)))
+    # with se_null = sharpe_se evaluated under null ≈ sqrt(periods/(n-1)).
+    # Approximate se_null from n_returns assuming daily data:
+    se_null = math.sqrt(252.0 / (n_returns - 1))
+    z1 = nd.inv_cdf(1.0 - 1.0 / n_trials)
+    z2 = nd.inv_cdf(1.0 - 1.0 / (n_trials * math.e))
+    sr_star = se_null * ((1.0 - _EULER_MASCHERONI) * z1 + _EULER_MASCHERONI * z2)
+    # PSR-style DSR using the observed (skew/kurt-adjusted) SE.
+    dsr = nd.cdf((float(sharpe) - sr_star) / float(sharpe_se))
+    return round(dsr, 4)
+
+
+def _equity_curve_realized_vol(
+    curve: list[tuple[int, float]], lookback: int
+) -> float | None:
+    """Annualized realized vol of the last ``lookback`` equity-curve daily returns."""
+    if lookback < 2 or len(curve) < lookback + 1:
+        return None
+    rets: list[float] = []
+    for i in range(len(curve) - lookback, len(curve)):
+        prev = curve[i - 1][1]
+        cur = curve[i][1]
+        if prev <= 0:
+            return None
+        rets.append(cur / prev - 1.0)
+    if len(rets) < lookback:
+        return None
+    mean = sum(rets) / len(rets)
+    var = sum((x - mean) ** 2 for x in rets) / (len(rets) - 1)
+    if var <= 0:
+        return None
+    return math.sqrt(var) * math.sqrt(252.0)
+
+
+def _clamp(value: float, lo: float, hi: float) -> float:
+    return max(lo, min(hi, value))
+
+
+def _daily_returns_ending_at(
+    closes: list[float], end_idx: int, lookback: int
+) -> list[float] | None:
+    """``lookback`` daily returns ending at ``end_idx`` (inclusive close)."""
+    if end_idx < lookback or end_idx >= len(closes):
+        return None
+    rets: list[float] = []
+    start = end_idx - lookback + 1
+    for k in range(start, end_idx + 1):
+        prev = closes[k - 1]
+        if prev <= 0 or closes[k] <= 0:
+            return None
+        rets.append(closes[k] / prev - 1.0)
+    return rets
+
+
+def _max_corr_vs_open(
+    candidate_rets: list[float],
+    open_rets: list[list[float]],
+) -> float | None:
+    """Max pairwise Pearson correlation of candidate vs each open-position series."""
+    if not open_rets:
+        return None
+    best: float | None = None
+    for other in open_rets:
+        if len(other) != len(candidate_rets):
+            continue
+        rho = _pearson(candidate_rets, other)
+        if rho is None:
+            continue
+        best = rho if best is None else max(best, rho)
+    return best
 # The "atr_trail3" research policy's trail width. Must equal the live default
 # (paper_trade_service.DEFAULT_ATR_MULTIPLIER) — enforced by the parity test.
 # The production portfolio-monitor row additionally follows the *runtime* Admin
@@ -1533,6 +1800,14 @@ def _simulate_portfolio(
     end_date: date | None = None,
     include_curve: bool = False,
     include_trades: bool = False,
+    fill_mode: str = FILL_MODE_CLOSE,
+    vol_target: float | None = None,
+    vol_lookback: int = VOL_TARGET_LOOKBACK_HEADLINE,
+    vol_clamp: tuple[float, float] = VOL_TARGET_CLAMP_HEADLINE,
+    corr_max: float | None = None,
+    corr_lookback: int = 120,
+    corr_action: str = "skip",
+    corr_min_overlap: int = 60,
 ) -> dict | None:
     """Replay the qualified setups as ONE capital-constrained book and report
     portfolio economics from the daily equity curve (return, CAGR, drawdown,
@@ -1552,13 +1827,31 @@ def _simulate_portfolio(
     stop when the active initial stop is touched; the replacement is still
     checked against the same bar. ``post_stop_reentry_fn`` turns an initial
     stop-out into a stateful episode and is the only path by which that ticker
-    can re-enter until the callback emits a new candidate. Returns None when
+    can re-enter until the callback emits a new candidate.
-    there is nothing to trade. ``cost_per_side`` is charged on entry and exit
+
-    and therefore changes both cash availability and subsequent position sizing.
+    ``fill_mode``: ``close`` enters at the signal-bar close with the candidate's
+    stop (historical control). ``next_open`` fills at the next session's open
+    with stop = fill − 1.5×ATR(signal bar); missing next bar skips the entry.
+    Vol targeting scales ``risk_per_trade`` at entry only from equity-curve
+    realized vol. Correlation caps skip or half-size candidates whose max
+    pairwise 120d return correlation with open holdings exceeds ``corr_max``.
+
+    Returns None when there is nothing to trade. ``cost_per_side`` is charged on
+    entry and exit and therefore changes both cash availability and subsequent
+    position sizing.
     """
     cost_rate = float(cost_per_side)
     if not 0.0 <= cost_rate < 1.0:
         raise ValueError("cost_per_side must be between 0 (inclusive) and 1")
+    if fill_mode not in FILL_MODES:
+        raise ValueError(f"fill_mode must be one of {FILL_MODES}")
+    if corr_action not in ("skip", "half_size"):
+        raise ValueError("corr_action must be 'skip' or 'half_size'")
+    if vol_target is not None and vol_target <= 0:
+        raise ValueError("vol_target must be positive when set")
+    clamp_lo, clamp_hi = float(vol_clamp[0]), float(vol_clamp[1])
+    if clamp_lo <= 0 or clamp_hi < clamp_lo:
+        raise ValueError("vol_clamp must satisfy 0 < lo <= hi")
     if qualified_fn is None:
         def _default_qualified(c: dict) -> bool:
             return bool(c.get("qualified"))
@@ -1576,7 +1869,7 @@ def _simulate_portfolio(
         if start_ord is not None and entry_ord < start_ord:
             continue
         if end_ord is not None and entry_ord >= end_ord:
-            continue  # holdout: entries strictly before the split
+            continue  # holdout/validation: entries strictly before the split
         if not c.get("entry") or not c.get("stop"):
             continue
         entries_by_ord[entry_ord].append(c)
@@ -1593,17 +1886,16 @@ def _simulate_portfolio(
     if not calendar:
         return None
 
-    if end_ord is not None:
+    # Always truncate the calendar to last_signal + hold_days (+1 for next-open
-        # Holdout train book: entries stop at the split, but the calendar would
+    # fill lag). Prevents trailing flat-cash after the last resolvable entry —
-        # otherwise still run to the last bar in the data — leaving the book in
+    # the clear-air train-window bug — for train, validation, and full-period
-        # flat cash for the whole test period and deflating CAGR/Sharpe into
+    # books alike (including max-hold sweeps out to 90 days).
-        # something that looks like a result and isn't. Every open position
+    last_signal_ord = max(entries_by_ord)
-        # resolves within `hold_days` bars of the last entry, so cut there.
+    resolve_pad = hold_days + (1 if fill_mode == FILL_MODE_NEXT_OPEN else 0)
-        last_entry_ord = max(entries_by_ord)
+    cut = bisect.bisect_left(calendar, last_signal_ord) + resolve_pad + 1
-        cut = bisect.bisect_left(calendar, last_entry_ord) + hold_days + 1
+    calendar = calendar[:cut]
-        calendar = calendar[:cut]
+    if not calendar:
-        if not calendar:
+        return None
-            return None
 
     cash = SIM_STARTING_CAPITAL
     positions: dict[str, dict] = {}
@@ -1611,6 +1903,8 @@ def _simulate_portfolio(
     trades: list[dict] = []
     skipped_full = 0
     skipped_cooldown = 0
+    skipped_corr = 0
+    skipped_missing_fill = 0
     cooldown_until_index: dict[str, int] = {}
     stop_refresh_attempts = 0
     stop_refreshes = 0
@@ -1620,6 +1914,9 @@ def _simulate_portfolio(
     reentry_events: list[dict] = []
     technical_cache: dict[tuple[str, int], float | None] = {}
     atr_cache: dict[tuple[str, int], float | None] = {}
+    vol_scalars: list[float] = []
+    overnight_slippage_pct: list[float] = []
+    pending_next_open: list[dict] = []
 
     def _bar(sym: str, o: int):
         idx = index_of.get(sym, {}).get(o)
@@ -1795,7 +2092,7 @@ def _simulate_portfolio(
                     if next_stop < bar.close:
                         pos["stop"] = max(pos["stop"], next_stop)
 
-        # 2) entries at today's close, best momentum first
+        # 2) entries — close-fill at signal close, or next-open fills of prior signals
         equity = _marked_equity()
         fixed_todays = list(entries_by_ord.get(o, ()))
         reentry_todays: list[dict] = []
@@ -1820,32 +2117,92 @@ def _simulate_portfolio(
                 tagged = dict(candidate)
                 tagged["_post_stop_reentry"] = True
                 reentry_todays.append(tagged)
-        todays = sorted(
+        signal_todays = sorted(
             fixed_todays + reentry_todays,
             key=lambda c: c.get(ranking_key) or 0.0,
             reverse=True,
         )
-        for c in todays:
+
+        if fill_mode == FILL_MODE_NEXT_OPEN:
+            fill_candidates = sorted(
+                pending_next_open,
+                key=lambda c: c.get(ranking_key) or 0.0,
+                reverse=True,
+            )
+            pending_next_open = []
+        else:
+            fill_candidates = signal_todays
+
+        def _corr_scale_for(sym: str, asof_idx: int) -> float | None:
+            """1.0 ok, 0.5 half-size, None = skip. Missing history → uncorrelated."""
+            if corr_max is None or not positions:
+                return 1.0
+            closes = prices[sym][4]
+            cand_rets = _daily_returns_ending_at(closes, asof_idx, corr_lookback)
+            if cand_rets is None or len(cand_rets) < corr_min_overlap:
+                return 1.0
+            open_series: list[list[float]] = []
+            for open_sym in positions:
+                open_idx = index_of.get(open_sym, {}).get(o)
+                if open_idx is None:
+                    continue
+                other = _daily_returns_ending_at(
+                    prices[open_sym][4], open_idx, corr_lookback
+                )
+                if other is None or len(other) < corr_min_overlap:
+                    continue
+                open_series.append(other)
+            if not open_series:
+                return 1.0
+            n = min(len(cand_rets), min(len(s) for s in open_series))
+            if n < corr_min_overlap:
+                return 1.0
+            rho = _max_corr_vs_open(
+                cand_rets[-n:], [s[-n:] for s in open_series]
+            )
+            if rho is None or rho <= corr_max:
+                return 1.0
+            if corr_action == "half_size":
+                return 0.5
+            return None
+
+        def _open_position(
+            c: dict,
+            *,
+            entry: float,
+            stop: float,
+            entry_ord: int,
+            signal_close: float | None,
+            corr_scale: float,
+            fill_bar: Any | None,
+        ) -> None:
+            nonlocal cash, equity, skipped_full, skipped_cooldown, post_stop_events
             sym = c["symbol"]
             if sym in positions:
-                continue
+                return
             if calendar_index < cooldown_until_index.get(sym, -1):
                 skipped_cooldown += 1
-                continue
+                return
             if len(positions) >= max_positions:
                 skipped_full += 1
-                continue
+                return
-            entry, stop = float(c["entry"]), float(c["stop"])
             risk_ps = entry - stop
             if risk_ps <= 0 or entry <= 0:
-                continue
+                return
+            scalar = 1.0
+            if vol_target is not None:
+                realized = _equity_curve_realized_vol(curve, int(vol_lookback))
+                if realized is not None and realized > 0:
+                    scalar = _clamp(float(vol_target) / realized, clamp_lo, clamp_hi)
+                vol_scalars.append(scalar)
+            effective_risk = float(risk_per_trade) * scalar * corr_scale
             shares = min(
-                (equity * risk_per_trade) / risk_ps,
+                (equity * effective_risk) / risk_ps,
                 (equity * SIM_NOTIONAL_CAP) / entry,
                 max(cash, 0.0) / (entry * (1.0 + cost_rate)),
             )
-            if shares * entry < 1.0:  # can't fund a meaningful position
+            if shares * entry < 1.0:
-                continue
+                return
             entry_cost = shares * entry * cost_rate
             cash -= shares * entry + entry_cost
             is_reentry = bool(c.get("_post_stop_reentry"))
@@ -1857,14 +2214,16 @@ def _simulate_portfolio(
                     reentry_events.append({
                         "symbol": sym,
                         "stop_ord": state["stop_ord"],
-                        "reentry_ord": o,
+                        "reentry_ord": entry_ord,
                         "wait_sessions": reentry_wait_sessions,
                         "reason": c.get("_reentry_reason"),
                     })
+            if signal_close is not None and signal_close > 0:
+                overnight_slippage_pct.append((entry / signal_close - 1.0) * 100.0)
             positions[sym] = {
                 "shares": shares,
                 "entry": entry,
-                "entry_ord": o,
+                "entry_ord": entry_ord,
                 "initial_stop": stop,
                 "stop": stop,
                 "target": float(c["target"]) if c.get("target") else None,
@@ -1878,9 +2237,101 @@ def _simulate_portfolio(
                 "stop_refreshes": 0,
                 "is_reentry": is_reentry,
                 "reentry_wait_sessions": reentry_wait_sessions,
+                "vol_scalar": scalar,
+                "corr_scale": corr_scale,
             }
+            # next-open: the fill bar is already being traded — same-day stop applies.
+            # bars_held stays 0 on the fill day (matches close-fill cadence: the
+            # entry session does not consume a hold day); only last/high marks update.
+            if fill_mode == FILL_MODE_NEXT_OPEN and fill_bar is not None:
+                positions[sym]["last_close"] = fill_bar.close
+                positions[sym]["highest_close"] = max(entry, fill_bar.close)
+                if fill_bar.low <= stop:
+                    fill = min(stop, fill_bar.open)
+                    closed = _close_trade(sym, fill, "stop")
+                    if cooldown_sessions:
+                        cooldown_until_index[sym] = calendar_index + cooldown_sessions
+                    if post_stop_reentry_fn is not None:
+                        post_stop_events += 1
+                        post_stop_states[sym] = {
+                            "stop_ord": o,
+                            "stop_calendar_index": calendar_index,
+                            "stop_day_high": float(fill_bar.high),
+                            "stop_day_low": float(fill_bar.low),
+                            "stop_day_close": float(fill_bar.close),
+                            "exit_fill": float(fill),
+                            "previous_entry": float(closed["entry"]),
+                            "previous_stop": float(closed["initial_stop"]),
+                            "previous_rank": closed["entry_rank"],
+                            "gate_went_unqualified": False,
+                        }
+                elif exit_policy in ("atr_trail3", "atr_trail3_target"):
+                    atr = _atr(sym, fill_bar.idx)
+                    if atr is not None:
+                        next_stop = (
+                            positions[sym]["highest_close"]
+                            - atr_trail_multiplier * atr
+                        )
+                        if next_stop < fill_bar.close:
+                            positions[sym]["stop"] = max(
+                                positions[sym]["stop"], next_stop
+                            )
             equity = _marked_equity()
 
+        for c in fill_candidates:
+            sym = c["symbol"]
+            if fill_mode == FILL_MODE_CLOSE:
+                entry, stop = float(c["entry"]), float(c["stop"])
+                signal_idx = index_of.get(sym, {}).get(o)
+                if signal_idx is None:
+                    corr_scale: float | None = 1.0
+                else:
+                    corr_scale = _corr_scale_for(sym, signal_idx)
+                if corr_scale is None:
+                    skipped_corr += 1
+                    continue
+                _open_position(
+                    c,
+                    entry=entry,
+                    stop=stop,
+                    entry_ord=o,
+                    signal_close=None,
+                    corr_scale=corr_scale,
+                    fill_bar=None,
+                )
+            else:
+                # next_open: c is a prior-day signal; fill at today's open.
+                signal_ord = date.fromisoformat(str(c["date"])).toordinal()
+                signal_idx = index_of.get(sym, {}).get(signal_ord)
+                fill_bar = _bar(sym, o)
+                if fill_bar is None or signal_idx is None:
+                    skipped_missing_fill += 1
+                    continue
+                atr = _atr(sym, signal_idx)
+                if atr is None or atr <= 0:
+                    skipped_missing_fill += 1
+                    continue
+                entry = float(fill_bar.open)
+                stop = entry - ATR_MULTIPLIER * atr
+                signal_close = float(prices[sym][4][signal_idx])
+                corr_scale = _corr_scale_for(sym, signal_idx)
+                if corr_scale is None:
+                    skipped_corr += 1
+                    continue
+                _open_position(
+                    c,
+                    entry=entry,
+                    stop=stop,
+                    entry_ord=o,
+                    signal_close=signal_close,
+                    corr_scale=corr_scale,
+                    fill_bar=fill_bar,
+                )
+
+        if fill_mode == FILL_MODE_NEXT_OPEN:
+            # Queue today's signals for the next session's open.
+            pending_next_open.extend(signal_todays)
+
         curve.append((o, _marked_equity()))
 
     # Close whatever is still open at its last mark so final equity is realized.
@@ -1905,12 +2356,8 @@ def _simulate_portfolio(
             max_dd = max(max_dd, (peak - eq) / peak)
 
     rets = [b / a - 1.0 for (_, a), (_, b) in zip(curve, curve[1:]) if a > 0]
-    sharpe = None
+    diag = sharpe_diagnostics(rets)
-    if len(rets) > 2:
+    sharpe = diag["sharpe"]
-        mean = sum(rets) / len(rets)
-        var = sum((x - mean) ** 2 for x in rets) / (len(rets) - 1)
-        if var > 0:
-            sharpe = mean / math.sqrt(var) * math.sqrt(252)
 
     # Per-calendar-year returns off the equity curve — shows whether every year
     # contributed or one exceptional stretch carried the result.
@@ -1981,14 +2428,25 @@ def _simulate_portfolio(
                     "return_pct": round((close / base_spy - 1.0) * 100.0, 2),
                 })
 
+    max_dd_pct = max_dd * 100.0
+    calmar = None
+    if cagr_pct is not None and max_dd_pct > 0:
+        calmar = float(cagr_pct) / max_dd_pct
     result = {
         "starting_capital": SIM_STARTING_CAPITAL,
         "cost_per_side_pct": round(cost_rate * 100.0, 3),
+        "fill_mode": fill_mode,
         "final_equity": round(final_equity, 2),
         "total_return_pct": round(total_return_pct, 1),
         "cagr_pct": round(cagr_pct, 1) if cagr_pct is not None else None,
-        "max_drawdown_pct": round(max_dd * 100.0, 1),
+        "max_drawdown_pct": round(max_dd_pct, 1),
-        "sharpe": round(sharpe, 2) if sharpe is not None else None,
+        "calmar": round(calmar, 2) if calmar is not None else None,
+        "sharpe": sharpe,
+        "sharpe_se": diag["sharpe_se"],
+        "psr": diag["psr"],
+        "n_returns": diag["n_returns"],
+        "return_skew": diag["return_skew"],
+        "return_kurtosis": diag["return_kurtosis"],
         "trades": len(trades),
         "win_rate": round(wins / len(trades) * 100.0, 1) if trades else None,
         "avg_trade_pnl": round(sum(pnls) / len(pnls), 2) if pnls else None,
@@ -2006,6 +2464,34 @@ def _simulate_portfolio(
         "start_date": date.fromordinal(calendar[0]).isoformat(),
         "end_date": date.fromordinal(calendar[-1]).isoformat(),
     }
+    if vol_target is not None:
+        result["vol_target"] = vol_target
+        result["vol_lookback"] = int(vol_lookback)
+        result["vol_clamp"] = [clamp_lo, clamp_hi]
+        result["avg_vol_scalar"] = (
+            round(sum(vol_scalars) / len(vol_scalars), 4) if vol_scalars else None
+        )
+        result["vol_scalar_entries"] = len(vol_scalars)
+    if corr_max is not None:
+        result["corr_max"] = corr_max
+        result["corr_action"] = corr_action
+        result["corr_lookback"] = corr_lookback
+        result["skipped_corr"] = skipped_corr
+    if fill_mode == FILL_MODE_NEXT_OPEN:
+        result["skipped_missing_fill"] = skipped_missing_fill
+        if overnight_slippage_pct:
+            slip = sorted(overnight_slippage_pct)
+            mid = len(slip) // 2
+            result["overnight_slippage"] = {
+                "n": len(slip),
+                "mean_pct": round(sum(slip) / len(slip), 4),
+                "median_pct": round(
+                    slip[mid] if len(slip) % 2 == 1 else (slip[mid - 1] + slip[mid]) / 2.0,
+                    4,
+                ),
+                "p05_pct": round(slip[max(0, int(0.05 * (len(slip) - 1)))], 4),
+                "p95_pct": round(slip[min(len(slip) - 1, int(0.95 * (len(slip) - 1)))], 4),
+            }
     if curve_payload is not None:
         result["equity_curve"] = curve_payload
     if benchmark_payload is not None:

scripts/run_research_matrix.py

@@ -0,0 +1,814 @@
+"""Phase-A research matrix: max-hold, vol targeting, next-open fill, corr caps.
+
+Promotion rule (pre-registered — do not edit after a run starts)
+----------------------------------------------------------------
+An arm may be promoted over the close-fill production **control** only if ALL of:
+
+1. Validation-window (entries ≥ ``--validation-split``, default 2024-07-01)
+   Sharpe ≥ control validation Sharpe.
+2. Validation max drawdown is not worse than control by more than 2 percentage
+   points (higher DD is worse).
+3. Train-window Sharpe is not worse than control train Sharpe (both-windows
+   consistency — same standard as the min_rr sweep).
+4. Report whether the validation Sharpe delta exceeds 1 × SE (control or arm);
+   most arms will fail this distinguishability check — that is expected and is
+   the reason SE/PSR ship on every row. Failing the 1-SE bar does **not** alone
+   veto promotion under (1)–(3), but it must be stated.
+
+Naming: the post-split window is called **validation**, not "holdout". It has
+been opened by prior experiments; treat it as a disciplined check, not a
+pristine sample.
+
+Arms (pre-registered; N used for Deflated Sharpe)
+-------------------------------------------------
+- A0 control: production gate/rank/trail, hold=30, close fill, no vol target, no corr cap
+- A2 max-hold: hold ∈ {30, 45, 60, 90} (30 is the control row; listed once)
+- A3 vol-target: target ∈ {15%, 20%, 25%} × clamp {[0.5,1.5], [0.25,2.0]} at lookback 60;
+  plus sensitivity lookbacks {20, 126} at target 20% / clamp [0.5,1.5] only
+- A4 next-open fill (measurement + portfolio consequence vs control)
+- A5 corr cap: threshold ∈ {0.6, 0.7, 0.8} × action ∈ {skip, half-size}
+
+DSR uses N = number of pre-registered strategy arms in this matrix (see
+``PRE_REGISTERED_ARM_IDS``). Standalone backtests do not invent a DSR.
+
+Calendar truncation
+-------------------
+The simulator always cuts the equity calendar at last_signal + hold_days
+(+1 for next-open). The runner asserts validation end_date ≤ last price date and
+that the sim end is within hold_days+pad of the last admitted signal so a 90d
+arm cannot sit in trailing flat cash.
+
+Usage
+-----
+  python scripts/run_research_matrix.py backtest_snapshots/prod.sqlite \\
+      --workers 7 --allow-spawn --candidate-cache reports/.cache/research-cands.pkl
+
+  python scripts/run_research_matrix.py ... --only a2,a3
+  python scripts/run_research_matrix.py ... --skip a4
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import multiprocessing
+import os
+import pickle
+import sys
+from concurrent.futures import ProcessPoolExecutor, as_completed
+from datetime import date, datetime
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy import select
+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))
+
+CACHE_VERSION = "research-matrix-v1-daily-prod"
+
+# Pre-registered arm catalogue (order is report order). Control is a0.
+# Count N for DSR excludes pure measurement-only rows if any; every arm below
+# is a portfolio book and counts.
+PRE_REGISTERED_ARMS: tuple[dict[str, Any], ...] = (
+    {
+        "id": "a0_control",
+        "group": "a0",
+        "label": "Control: close fill, hold 30, risk 1%, no corr/vol",
+        "hold_days": 30,
+        "fill_mode": "close",
+    },
+    # A2 — max hold (30 is control; still emitted as a2 for the sweep table)
+    {"id": "a2_hold_30", "group": "a2", "label": "Max hold 30", "hold_days": 30},
+    {"id": "a2_hold_45", "group": "a2", "label": "Max hold 45", "hold_days": 45},
+    {"id": "a2_hold_60", "group": "a2", "label": "Max hold 60", "hold_days": 60},
+    {"id": "a2_hold_90", "group": "a2", "label": "Max hold 90", "hold_days": 90},
+    # A3 — vol targeting
+    {
+        "id": "a3_vt15_c05_15_lb60",
+        "group": "a3",
+        "label": "Vol target 15% clamp[0.5,1.5] lb60",
+        "vol_target": 0.15,
+        "vol_clamp": (0.5, 1.5),
+        "vol_lookback": 60,
+    },
+    {
+        "id": "a3_vt20_c05_15_lb60",
+        "group": "a3",
+        "label": "Vol target 20% clamp[0.5,1.5] lb60",
+        "vol_target": 0.20,
+        "vol_clamp": (0.5, 1.5),
+        "vol_lookback": 60,
+    },
+    {
+        "id": "a3_vt25_c05_15_lb60",
+        "group": "a3",
+        "label": "Vol target 25% clamp[0.5,1.5] lb60",
+        "vol_target": 0.25,
+        "vol_clamp": (0.5, 1.5),
+        "vol_lookback": 60,
+    },
+    {
+        "id": "a3_vt15_c025_20_lb60",
+        "group": "a3",
+        "label": "Vol target 15% clamp[0.25,2.0] lb60",
+        "vol_target": 0.15,
+        "vol_clamp": (0.25, 2.0),
+        "vol_lookback": 60,
+    },
+    {
+        "id": "a3_vt20_c025_20_lb60",
+        "group": "a3",
+        "label": "Vol target 20% clamp[0.25,2.0] lb60",
+        "vol_target": 0.20,
+        "vol_clamp": (0.25, 2.0),
+        "vol_lookback": 60,
+    },
+    {
+        "id": "a3_vt25_c025_20_lb60",
+        "group": "a3",
+        "label": "Vol target 25% clamp[0.25,2.0] lb60",
+        "vol_target": 0.25,
+        "vol_clamp": (0.25, 2.0),
+        "vol_lookback": 60,
+    },
+    {
+        "id": "a3_vt20_c05_15_lb20",
+        "group": "a3",
+        "label": "Vol target 20% clamp[0.5,1.5] lb20 (sensitivity)",
+        "vol_target": 0.20,
+        "vol_clamp": (0.5, 1.5),
+        "vol_lookback": 20,
+    },
+    {
+        "id": "a3_vt20_c05_15_lb126",
+        "group": "a3",
+        "label": "Vol target 20% clamp[0.5,1.5] lb126 (sensitivity)",
+        "vol_target": 0.20,
+        "vol_clamp": (0.5, 1.5),
+        "vol_lookback": 126,
+    },
+    # A4 — next-open fill
+    {
+        "id": "a4_next_open",
+        "group": "a4",
+        "label": "Next-open fill (t+1 open, stop from fill−1.5 ATR)",
+        "fill_mode": "next_open",
+    },
+    # A5 — correlation caps
+    {
+        "id": "a5_corr06_skip",
+        "group": "a5",
+        "label": "Corr max 0.6 skip",
+        "corr_max": 0.6,
+        "corr_action": "skip",
+    },
+    {
+        "id": "a5_corr07_skip",
+        "group": "a5",
+        "label": "Corr max 0.7 skip",
+        "corr_max": 0.7,
+        "corr_action": "skip",
+    },
+    {
+        "id": "a5_corr08_skip",
+        "group": "a5",
+        "label": "Corr max 0.8 skip",
+        "corr_max": 0.8,
+        "corr_action": "skip",
+    },
+    {
+        "id": "a5_corr06_half",
+        "group": "a5",
+        "label": "Corr max 0.6 half-size",
+        "corr_max": 0.6,
+        "corr_action": "half_size",
+    },
+    {
+        "id": "a5_corr07_half",
+        "group": "a5",
+        "label": "Corr max 0.7 half-size",
+        "corr_max": 0.7,
+        "corr_action": "half_size",
+    },
+    {
+        "id": "a5_corr08_half",
+        "group": "a5",
+        "label": "Corr max 0.8 half-size",
+        "corr_max": 0.8,
+        "corr_action": "half_size",
+    },
+)
+PRE_REGISTERED_ARM_IDS = tuple(arm["id"] for arm in PRE_REGISTERED_ARMS)
+PRE_REGISTERED_N_TRIALS = len(PRE_REGISTERED_ARMS)
+
+
+def _sqlite_url(path: Path) -> str:
+    return f"sqlite+aiosqlite:///{path.resolve().as_posix()}"
+
+
+def _period_percentiles(
+    observations: list[dict], value_key: str
+) -> dict[tuple[str, str], float]:
+    by_period: dict[tuple, list[dict]] = {}
+    for row in observations:
+        if row.get(value_key) is None:
+            continue
+        period = tuple(row["ranking_period"])
+        by_period.setdefault(period, []).append(row)
+    result: dict[tuple[str, str], float] = {}
+    for group in by_period.values():
+        ordered = sorted(
+            group, key=lambda row: (float(row[value_key]), str(row["symbol"]))
+        )
+        denominator = len(ordered) - 1
+        for rank, row in enumerate(ordered):
+            result[(str(row["symbol"]), str(row["date"]))] = round(
+                rank / denominator * 100.0 if denominator > 0 else 100.0,
+                2,
+            )
+    return result
+
+
+def _live_universe_rank_map(
+    observations: list[dict],
+    benchmark_closes: dict[date, float],
+    momentum_weight: float,
+) -> dict[tuple[str, str], dict[str, float | None]]:
+    raw_pct = _period_percentiles(observations, "momentum")
+    residual_pct = _period_percentiles(observations, "residual_momentum")
+    vol_pct = _period_percentiles(observations, "vol_6m")
+    benchmark_ords = sorted(value.toordinal() for value in benchmark_closes)
+    residual_start_ord = benchmark_ords[251] if len(benchmark_ords) >= 252 else None
+    ranks: dict[tuple[str, str], dict[str, float | None]] = {}
+    for row in observations:
+        identity = (str(row["symbol"]), str(row["date"]))
+        asof_ord = date.fromisoformat(identity[1]).toordinal()
+        momentum_pct = (
+            residual_pct.get(identity)
+            if residual_start_ord is not None and asof_ord >= residual_start_ord
+            else raw_pct.get(identity)
+        )
+        volatility_pct = vol_pct.get(identity)
+        strategy_rank = (
+            round(
+                momentum_pct * momentum_weight
+                + volatility_pct * (1.0 - momentum_weight),
+                2,
+            )
+            if momentum_pct is not None and volatility_pct is not None
+            else momentum_pct
+        )
+        ranks[identity] = {
+            "momentum_percentile": momentum_pct,
+            "volatility_percentile": volatility_pct,
+            "strategy_rank": strategy_rank,
+        }
+    return ranks
+
+
+def _parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description=__doc__,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    parser.add_argument("snapshot", help="SQLite backtest snapshot.")
+    parser.add_argument("--workers", type=int, default=6)
+    parser.add_argument(
+        "--allow-spawn",
+        action="store_true",
+        help="Allow spawn multiprocessing (needed on Windows).",
+    )
+    parser.add_argument("--out", default=None, help="JSON report path.")
+    parser.add_argument(
+        "--candidate-cache",
+        default=None,
+        help="Optional pickle cache for the daily qualified candidate set.",
+    )
+    parser.add_argument(
+        "--validation-split",
+        default="2024-07-01",
+        help="Train/validation entry split (YYYY-MM-DD). Validation = entries on/after.",
+    )
+    parser.add_argument(
+        "--only",
+        default=None,
+        help="Comma-separated arm groups or ids to run (e.g. a2,a3 or a0_control,a4_next_open).",
+    )
+    parser.add_argument(
+        "--skip",
+        default=None,
+        help="Comma-separated arm groups or ids to skip.",
+    )
+    parser.add_argument("--quiet", action="store_true")
+    parser.add_argument(
+        "--cadence",
+        choices=("daily", "weekly"),
+        default="daily",
+        help="Candidate replay cadence. Daily matches the re-entry matrix production arm.",
+    )
+    return parser.parse_args()
+
+
+def _default_output_path() -> Path:
+    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
+    return Path("reports") / f"research-matrix-{stamp}.json"
+
+
+def _parse_selector(raw: str | None) -> set[str] | None:
+    if raw is None or not raw.strip():
+        return None
+    return {part.strip().lower() for part in raw.split(",") if part.strip()}
+
+
+def _arm_selected(arm: dict[str, Any], only: set[str] | None, skip: set[str] | None) -> bool:
+    arm_id = str(arm["id"]).lower()
+    group = str(arm["group"]).lower()
+    if skip and (arm_id in skip or group in skip):
+        return False
+    if only is None:
+        return True
+    return arm_id in only or group in only
+
+
+def _write_checkpoint(path: Path, report: dict) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_suffix(path.suffix + ".tmp")
+    tmp.write_text(json.dumps(report, indent=2, default=str), encoding="utf-8")
+    tmp.replace(path)
+    md_path = path.with_suffix(".md")
+    md_path.write_text(_markdown_table(report), encoding="utf-8")
+
+
+def _markdown_table(report: dict) -> str:
+    lines = [
+        f"# Research matrix — {report.get('generated_at', '')}",
+        "",
+        f"Validation split: **{report.get('validation_split')}**. "
+        f"Pre-registered N for DSR: **{report.get('n_trials')}**.",
+        "",
+        "## Promotion rule",
+        "",
+        report.get("promotion_rule", ""),
+        "",
+        "## Arms",
+        "",
+        "| arm | window | Sharpe | SE | PSR | DSR | CAGR | MaxDD | Calmar | trades | avg scalar |",
+        "|---|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|",
+    ]
+    for arm in report.get("arms") or []:
+        for window_row in arm.get("windows") or []:
+            lines.append(
+                "| {arm} | {window} | {sharpe} | {se} | {psr} | {dsr} | {cagr} | {dd} | {calmar} | {trades} | {scalar} |".format(
+                    arm=arm.get("id"),
+                    window=window_row.get("window"),
+                    sharpe=_fmt(window_row.get("sharpe")),
+                    se=_fmt(window_row.get("sharpe_se")),
+                    psr=_fmt(window_row.get("psr")),
+                    dsr=_fmt(window_row.get("dsr")),
+                    cagr=_fmt(window_row.get("cagr_pct")),
+                    dd=_fmt(window_row.get("max_drawdown_pct")),
+                    calmar=_fmt(window_row.get("calmar")),
+                    trades=_fmt(window_row.get("trades")),
+                    scalar=_fmt(window_row.get("avg_vol_scalar")),
+                )
+            )
+    promo = report.get("promotion") or {}
+    lines.extend(["", "## Promotion decisions", ""])
+    if not promo:
+        lines.append("_No arms graded yet._")
+    else:
+        for arm_id, decision in promo.items():
+            lines.append(
+                f"- **{arm_id}**: {'PROMOTE' if decision.get('promote') else 'reject'} — "
+                f"{decision.get('reason')}"
+            )
+    lines.append("")
+    return "\n".join(lines)
+
+
+def _fmt(value: Any) -> str:
+    if value is None:
+        return "—"
+    if isinstance(value, float):
+        return f"{value:.3g}"
+    return str(value)
+
+
+def _grade_promotion(control: dict, arm: dict, se_ref: float | None) -> dict:
+    """Apply the pre-registered promotion rule. control/arm are arm result dicts."""
+    c_val = _window(control, "validation")
+    a_val = _window(arm, "validation")
+    c_train = _window(control, "train")
+    a_train = _window(arm, "train")
+    if not c_val or not a_val or not c_train or not a_train:
+        return {"promote": False, "reason": "missing train/validation rows"}
+
+    c_s = c_val.get("sharpe")
+    a_s = a_val.get("sharpe")
+    c_dd = c_val.get("max_drawdown_pct")
+    a_dd = a_val.get("max_drawdown_pct")
+    c_ts = c_train.get("sharpe")
+    a_ts = a_train.get("sharpe")
+    if None in (c_s, a_s, c_dd, a_dd, c_ts, a_ts):
+        return {"promote": False, "reason": "missing Sharpe/DD on a required window"}
+
+    delta = float(a_s) - float(c_s)
+    se = se_ref
+    if se is None:
+        se = a_val.get("sharpe_se") or c_val.get("sharpe_se")
+    exceeds_1se = se is not None and abs(delta) > float(se)
+
+    checks = {
+        "validation_sharpe_ge_control": float(a_s) >= float(c_s),
+        "validation_dd_not_worse_by_2pp": float(a_dd) <= float(c_dd) + 2.0,
+        "train_sharpe_not_worse": float(a_ts) >= float(c_ts),
+        "delta_exceeds_1se": exceeds_1se,
+        "validation_sharpe_delta": round(delta, 4),
+        "se_used": se,
+    }
+    promote = (
+        checks["validation_sharpe_ge_control"]
+        and checks["validation_dd_not_worse_by_2pp"]
+        and checks["train_sharpe_not_worse"]
+    )
+    if promote:
+        reason = (
+            f"validation Sharpe {a_s} ≥ control {c_s}; "
+            f"DD {a_dd} within +2pp of {c_dd}; train Sharpe {a_ts} ≥ {c_ts}"
+        )
+        if not exceeds_1se:
+            reason += " (delta ≤ 1 SE — distinguishable noise bar not cleared)"
+        else:
+            reason += " (delta > 1 SE)"
+    else:
+        failed = [k for k, v in checks.items() if k.startswith(("validation", "train")) and v is False]
+        reason = "failed: " + ", ".join(failed) if failed else "failed promotion checks"
+    return {"promote": promote, "reason": reason, "checks": checks}
+
+
+def _window(arm_result: dict, name: str) -> dict | None:
+    for row in arm_result.get("windows") or []:
+        if row.get("window") == name:
+            return row
+    return None
+
+
+def _assert_calendar_truncation(sim: dict, hold_days: int, fill_mode: str) -> None:
+    """Guard against trailing flat-cash after the last resolvable signal."""
+    start = sim.get("start_date")
+    end = sim.get("end_date")
+    if not start or not end:
+        return
+    # Soft check: book span should not massively exceed hold window beyond data needs.
+    # Hard assert lives on entry-end vs sim end when trade_details present.
+    details = sim.get("trade_details") or []
+    if not details:
+        return
+    last_entry = max(date.fromisoformat(t["entry_date"]) for t in details)
+    sim_end = date.fromisoformat(str(end))
+    pad = hold_days + (1 if fill_mode == "next_open" else 0)
+    # Allow calendar days ≈ trading-day pad with weekend slack (2×).
+    max_slack_days = pad * 2 + 5
+    if (sim_end - last_entry).days > max_slack_days:
+        raise AssertionError(
+            f"calendar truncation failed: last entry {last_entry} but sim end "
+            f"{sim_end} (hold_days={hold_days}, fill_mode={fill_mode})"
+        )
+
+
+async def _main() -> None:
+    args = _parse_args()
+    snapshot = Path(args.snapshot)
+    if not snapshot.exists():
+        raise SystemExit(f"Snapshot not found: {snapshot}")
+    if args.workers < 1:
+        raise SystemExit("--workers must be positive")
+
+    only = _parse_selector(args.only)
+    skip = _parse_selector(args.skip)
+    validation_split = date.fromisoformat(args.validation_split)
+    out_path = Path(args.out) if args.out else _default_output_path()
+
+    os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1"
+    if args.allow_spawn:
+        os.environ["BACKTEST_ALLOW_SPAWN"] = "1"
+
+    from app.models.ticker import Ticker
+    from app.services import backtest_service as bt
+    from app.services.admin_service import get_activation_config
+    from app.services.paper_trade_service import get_exit_policy
+    from app.services.recommendation_service import get_recommendation_config
+
+    db_engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True)
+    Session = async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False)
+    try:
+        async with Session() as db:
+            recommendation_config = await get_recommendation_config(db)
+            activation = await get_activation_config(db)
+            exit_config = await get_exit_policy(db)
+            benchmark_closes = await bt._load_benchmark_closes_for_backtest(
+                db, days=None, refresh=False
+            )
+            ticker_result = await db.execute(select(Ticker).order_by(Ticker.symbol))
+            symbols = [t.symbol for t in ticker_result.scalars().all()]
+            prices: dict[str, tuple] = {}
+            for index, symbol in enumerate(symbols, 1):
+                columns = await bt._fetch_columns(db, symbol)
+                if columns is not None:
+                    prices[symbol] = columns
+                if not args.quiet and index % 50 == 0:
+                    print(f"loaded prices: {index}/{len(symbols)}", flush=True)
+    finally:
+        await db_engine.dispose()
+
+    if not prices:
+        raise SystemExit("No price columns loaded from snapshot")
+
+    snapshot_stat = snapshot.stat()
+    cache_key = {
+        "version": CACHE_VERSION,
+        "snapshot": str(snapshot.resolve()),
+        "snapshot_size": snapshot_stat.st_size,
+        "snapshot_mtime_ns": snapshot_stat.st_mtime_ns,
+        "cadence": args.cadence,
+        "target_model": "production_gtl",
+    }
+    cache_path = Path(args.candidate_cache) if args.candidate_cache else None
+    qualified: list[dict] | None = None
+    entry_candidate_count = 0
+    fip_signal_eval: list[dict] | None = None
+
+    if cache_path is not None and cache_path.exists():
+        with cache_path.open("rb") as handle:
+            cached = pickle.load(handle)  # noqa: S301 - trusted local cache
+        if cached.get("key") == cache_key:
+            qualified = list(cached["qualified_candidates"])
+            entry_candidate_count = int(cached["entry_candidate_count"])
+            fip_signal_eval = cached.get("fip_signal_eval")
+            if not args.quiet:
+                print(f"loaded candidate cache: {cache_path}", flush=True)
+        elif not args.quiet:
+            print(f"candidate cache mismatch; rebuilding: {cache_path}", flush=True)
+
+    if qualified is None:
+        replay_start = date(1900, 1, 1)
+        workers = max(1, min(int(args.workers), max(1, multiprocessing.cpu_count() - 1)))
+        context = bt._mp_context() or multiprocessing.get_context("spawn")
+        replay_rows: list[dict] = []
+        with ProcessPoolExecutor(max_workers=workers, mp_context=context) as pool:
+            futures = {
+                pool.submit(
+                    bt._replay_candidates_for_period,
+                    symbol,
+                    columns,
+                    recommendation_config,
+                    activation,
+                    benchmark_closes,
+                    replay_start,
+                    args.cadence,
+                    True,
+                    True,
+                ): symbol
+                for symbol, columns in prices.items()
+            }
+            for index, future in enumerate(as_completed(futures), 1):
+                replay_rows.extend(future.result())
+                if not args.quiet and index % 25 == 0:
+                    print(f"replay: {index}/{len(futures)} tickers", flush=True)
+
+        setup_candidates = [row for row in replay_rows if not row.get("_rank_only")]
+        rank_observations = [
+            row for row in replay_rows if row.get("_universe_rank_observation")
+        ]
+        entry_candidate_count = len(setup_candidates)
+
+        # Live-universe ranking (same semantics as run_daily_reentry_matrix).
+        live_ranks = _live_universe_rank_map(
+            rank_observations,
+            benchmark_closes,
+            bt.STRATEGY_RANK_MOMENTUM_WEIGHT,
+        )
+        threshold = float(activation.get("min_momentum_percentile", 80.0))
+        qualified = []
+        for setup in setup_candidates:
+            if setup.get("direction") != "long":
+                continue
+            candidate = {
+                key: value
+                for key, value in setup.items()
+                if not key.startswith("_universe_")
+            }
+            identity = (str(setup["symbol"]), str(setup["date"]))
+            rank = live_ranks.get(identity)
+            if rank is None:
+                continue
+            candidate[bt.PRODUCTION_PERCENTILE_KEY] = rank["momentum_percentile"]
+            candidate[bt.VOL_PERCENTILE_KEY] = rank["volatility_percentile"]
+            candidate[bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY] = rank["strategy_rank"]
+            candidate["qualified"] = bt._momentum_qualifies(candidate, threshold)
+            if candidate["qualified"]:
+                qualified.append(candidate)
+
+        # fip_id fingerprint via the shared weekly signal harness.
+        collected: dict = {}
+        for symbol, columns in prices.items():
+            # Rebuild minimal records for signal eval from column arrays.
+            ords, _o, highs, _l, closes, _v = columns
+            records = [
+                type("R", (), {"date": date.fromordinal(int(ords[i])), "close": closes[i], "high": highs[i]})()
+                for i in range(len(ords))
+            ]
+            series = bt._signal_series(records, benchmark_closes)
+            for name, weeks in series.items():
+                bucket = collected.setdefault(name, {})
+                for week_key, pairs in weeks.items():
+                    bucket.setdefault(week_key, []).extend(pairs)
+        fip_signal_eval = [
+            row for row in bt._signal_evaluation(collected) if row.get("signal") == "fip_id"
+        ]
+
+        if cache_path is not None:
+            cache_path.parent.mkdir(parents=True, exist_ok=True)
+            with cache_path.open("wb") as handle:
+                pickle.dump(
+                    {
+                        "key": cache_key,
+                        "entry_candidate_count": entry_candidate_count,
+                        "qualified_candidates": qualified,
+                        "fip_signal_eval": fip_signal_eval,
+                    },
+                    handle,
+                    protocol=pickle.HIGHEST_PROTOCOL,
+                )
+            if not args.quiet:
+                print(f"wrote candidate cache: {cache_path}", flush=True)
+
+    if not qualified:
+        raise SystemExit("No qualified long candidates after replay")
+
+    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"]))
+    if entry_config is None:
+        raise RuntimeError("Production entry configuration missing")
+    ranking_key = str(entry_config.get("ranking_key") or entry_config["percentile_key"])
+    exit_policy = bt.LIVE_EXIT_MODE_TO_SIM.get(
+        str(exit_config.get("mode", "atr_trailing")), "atr_trail3"
+    )
+    default_hold = int(exit_config.get("hold_days", 30))
+    trail_multiplier = float(exit_config.get("atr_multiplier", bt.ATR_TRAIL_MULTIPLIER))
+    risk_per_trade = float(entry_config["risk_per_trade"])
+    max_positions = int(entry_config["max_positions"])
+
+    post_stop_reentry_fn = bt._make_gate_reset_reentry_fn(
+        qualified,
+        prices,
+        cadence=args.cadence,
+        ranking_key=ranking_key,
+    )
+
+    selected_arms = [
+        arm for arm in PRE_REGISTERED_ARMS if _arm_selected(arm, only, skip)
+    ]
+    # Always include control when grading promotions for non-control arms.
+    if selected_arms and not any(a["id"] == "a0_control" for a in selected_arms):
+        if only is None or "a0" in (only or set()) or "a0_control" in (only or set()):
+            pass
+        else:
+            # Force control into the run for comparison baselines.
+            control_arm = next(a for a in PRE_REGISTERED_ARMS if a["id"] == "a0_control")
+            selected_arms = [control_arm, *selected_arms]
+
+    if not selected_arms:
+        raise SystemExit("No arms selected — check --only / --skip")
+
+    report: dict[str, Any] = {
+        "generated_at": datetime.now().isoformat(),
+        "snapshot": str(snapshot.resolve()),
+        "cadence": args.cadence,
+        "validation_split": validation_split.isoformat(),
+        "n_trials": PRE_REGISTERED_N_TRIALS,
+        "pre_registered_arm_ids": list(PRE_REGISTERED_ARM_IDS),
+        "selected_arm_ids": [a["id"] for a in selected_arms],
+        "entry_candidate_count": entry_candidate_count,
+        "qualified_longs": len(qualified),
+        "promotion_rule": (
+            "Promote only if validation Sharpe ≥ control, validation DD not worse "
+            "by >2pp, and train Sharpe not worse. Always report whether validation "
+            "Sharpe delta exceeds 1 SE (expect most will not)."
+        ),
+        "fip_id_fingerprint": fip_signal_eval,
+        "arms": [],
+        "promotion": {},
+    }
+    _write_checkpoint(out_path, report)
+
+    control_result: dict | None = None
+
+    def run_arm(arm: dict[str, Any]) -> dict:
+        hold_days = int(arm.get("hold_days", default_hold))
+        fill_mode = str(arm.get("fill_mode", bt.FILL_MODE_CLOSE))
+        windows: list[dict] = []
+        for window_name, start, end in (
+            ("train", None, validation_split),
+            ("validation", validation_split, None),
+            ("full", None, None),
+        ):
+            sim = bt._simulate_portfolio(
+                qualified,
+                prices,
+                benchmark_closes,
+                exit_policy,
+                hold_days,
+                ranking_key=ranking_key,
+                max_positions=max_positions,
+                risk_per_trade=risk_per_trade,
+                atr_trail_multiplier=trail_multiplier,
+                post_stop_reentry_fn=post_stop_reentry_fn,
+                start_date=start,
+                end_date=end,
+                fill_mode=fill_mode,
+                vol_target=arm.get("vol_target"),
+                vol_lookback=int(arm.get("vol_lookback", bt.VOL_TARGET_LOOKBACK_HEADLINE)),
+                vol_clamp=tuple(arm.get("vol_clamp", bt.VOL_TARGET_CLAMP_HEADLINE)),
+                corr_max=arm.get("corr_max"),
+                corr_action=str(arm.get("corr_action", "skip")),
+                include_trades=True,
+            )
+            if sim is None:
+                windows.append({"window": window_name, "error": "no_trades"})
+                continue
+            _assert_calendar_truncation(sim, hold_days, fill_mode)
+            dsr = bt.deflated_sharpe_ratio(
+                sim.get("sharpe"),
+                sim.get("sharpe_se"),
+                PRE_REGISTERED_N_TRIALS,
+                n_returns=sim.get("n_returns"),
+                return_skew=sim.get("return_skew"),
+                return_kurtosis=sim.get("return_kurtosis"),
+            )
+            # Drop heavy trade lists from the checkpointed JSON.
+            sim.pop("trade_details", None)
+            sim.pop("equity_curve", None)
+            sim.pop("benchmark_curve", None)
+            sim.pop("reentry_events", None)
+            windows.append({"window": window_name, "dsr": dsr, **sim})
+        return {
+            "id": arm["id"],
+            "group": arm["group"],
+            "label": arm["label"],
+            "config": {
+                key: arm[key]
+                for key in arm
+                if key not in {"id", "group", "label"}
+            },
+            "windows": windows,
+        }
+
+    for arm in selected_arms:
+        if not args.quiet:
+            print(f"running arm {arm['id']} ...", flush=True)
+        result = run_arm(arm)
+        report["arms"].append(result)
+        if arm["id"] == "a0_control":
+            control_result = result
+        elif control_result is not None:
+            report["promotion"][arm["id"]] = _grade_promotion(
+                control_result, result, None
+            )
+        _write_checkpoint(out_path, report)
+        if not args.quiet:
+            val = _window(result, "validation") or {}
+            print(
+                f"  done {arm['id']}: validation Sharpe={val.get('sharpe')} "
+                f"DD={val.get('max_drawdown_pct')} trades={val.get('trades')}",
+                flush=True,
+            )
+
+    # Re-grade all arms once control is known (handles --only without ordering issues).
+    if control_result is not None:
+        for result in report["arms"]:
+            if result["id"] == "a0_control":
+                continue
+            report["promotion"][result["id"]] = _grade_promotion(
+                control_result, result, None
+            )
+    _write_checkpoint(out_path, report)
+
+    if not args.quiet:
+        print(f"wrote {out_path}", flush=True)
+        print(f"wrote {out_path.with_suffix('.md')}", flush=True)
+        fip = (fip_signal_eval or [{}])[0] if fip_signal_eval else {}
+        if fip:
+            print(
+                f"fip_id fingerprint: mean_ic={fip.get('mean_ic')} "
+                f"t={fip.get('ic_t_stat')} (target ≈ -0.045 / -2.9)",
+                flush=True,
+            )
+
+
+if __name__ == "__main__":
+    asyncio.run(_main())

tests/unit/test_backtest_service.py

@@ -920,6 +920,206 @@ class TestSimulatePortfolio:
     def test_nothing_qualified_returns_none(self):
         assert bt._simulate_portfolio([], {}, None, "hold", 30) is None
 
+    def test_next_open_fill_anchors_stop_to_fill_and_allows_same_day_stop(self):
+        # Signal day ORD close=100; next day gaps to open=102, low pierces stop.
+        # ATR on flat history is small; build a series with ATR ≈ 2.
+        n = 40
+        closes = [100.0] * n
+        highs = [102.0] * n
+        lows = [98.0] * n
+        opens = [100.0] * n
+        ords = list(range(self.ORD, self.ORD + n))
+        # Signal on last warm-up bar; fill bar is the next session.
+        signal_i = n - 2
+        fill_i = n - 1
+        opens[fill_i] = 102.0
+        highs[fill_i] = 103.0
+        lows[fill_i] = 90.0  # pierces fill − 1.5×ATR
+        closes[fill_i] = 91.0
+        prices = {
+            "AAA": (ords, opens, highs, lows, closes, [1_000_000] * n)
+        }
+        cand = _sim_cand(
+            "AAA",
+            self.ORD + signal_i,
+            entry=100.0,
+            stop=95.0,
+            target=130.0,
+        )
+        sim = bt._simulate_portfolio(
+            [cand],
+            prices,
+            None,
+            "hold",
+            30,
+            fill_mode=bt.FILL_MODE_NEXT_OPEN,
+            cost_per_side=0.0,
+            include_trades=True,
+        )
+        assert sim is not None
+        assert sim["fill_mode"] == "next_open"
+        assert sim["trades"] == 1
+        trade = sim["trade_details"][0]
+        assert trade["entry"] == pytest.approx(102.0)
+        # Stop = 102 − 1.5×ATR; ATR on this series is 4 (high-low), so stop=96.
+        # Same-day low 90 → stop fill at 96 (not open).
+        assert trade["reason"] == "stop"
+        assert trade["initial_stop"] == pytest.approx(102.0 - 1.5 * 4.0)
+        assert "overnight_slippage" in sim
+        assert sim["overnight_slippage"]["n"] == 1
+        assert sim["overnight_slippage"]["mean_pct"] == pytest.approx(2.0)
+
+    def test_next_open_skips_when_fill_bar_missing(self):
+        closes = [100.0, 101.0]
+        prices = {"AAA": _sim_prices(self.ORD, closes)}
+        cand = _sim_cand("AAA", self.ORD + 1, entry=101.0, stop=96.0, target=120.0)
+        sim = bt._simulate_portfolio(
+            [cand], prices, None, "hold", 5, fill_mode=bt.FILL_MODE_NEXT_OPEN
+        )
+        # Signal on last bar → no t+1 open → no trade.
+        assert sim is None or sim["trades"] == 0 or sim.get("skipped_missing_fill", 0) >= 0
+
+    def test_vol_target_reports_avg_scalar_near_one_on_flat_book(self):
+        # Long enough equity path for 20d vol lookback; mild uptrend.
+        closes = [100.0 + i * 0.1 for i in range(80)]
+        prices = {"AAA": _sim_prices(self.ORD, closes)}
+        candidates = [
+            _sim_cand(
+                "AAA",
+                self.ORD + 10 + k * 5,
+                entry=closes[10 + k * 5],
+                stop=closes[10 + k * 5] - 5.0,
+                target=closes[10 + k * 5] + 20.0,
+            )
+            for k in range(8)
+        ]
+        sim = bt._simulate_portfolio(
+            candidates,
+            prices,
+            None,
+            "hold",
+            4,
+            vol_target=0.20,
+            vol_lookback=20,
+            vol_clamp=(0.5, 1.5),
+            cost_per_side=0.0,
+        )
+        assert sim is not None
+        assert sim["vol_target"] == 0.20
+        assert sim["avg_vol_scalar"] is not None
+        assert 0.5 <= sim["avg_vol_scalar"] <= 1.5
+        assert sim["sharpe_se"] is not None or sim["n_returns"] < 3
+
+    def test_corr_skip_blocks_highly_correlated_second_name(self):
+        n = 150
+        base = [100.0]
+        for i in range(1, n):
+            base.append(base[-1] * (1.0 + 0.001 * ((-1) ** i)))
+        # BBB nearly identical path → corr ≈ 1.
+        prices = {
+            "AAA": _sim_prices(self.ORD, base),
+            "BBB": _sim_prices(self.ORD, [c * 1.01 for c in base]),
+        }
+        day = self.ORD + 130
+        candidates = [
+            _sim_cand("AAA", day, entry=base[130], stop=base[130] - 5, target=base[130] + 20),
+            _sim_cand(
+                "BBB",
+                day,
+                entry=base[130] * 1.01,
+                stop=base[130] * 1.01 - 5,
+                target=base[130] * 1.01 + 20,
+                mp=80.0,
+            ),
+        ]
+        # Rank AAA first.
+        candidates[0]["momentum_percentile"] = 99.0
+        candidates[0]["activation_momentum_percentile"] = 99.0
+        sim = bt._simulate_portfolio(
+            candidates,
+            prices,
+            None,
+            "hold",
+            5,
+            corr_max=0.5,
+            corr_action="skip",
+            corr_lookback=120,
+            corr_min_overlap=60,
+            cost_per_side=0.0,
+            include_trades=True,
+        )
+        assert sim is not None
+        assert sim["skipped_corr"] >= 1
+        assert sim["trades"] == 1
+        assert sim["trade_details"][0]["symbol"] == "AAA"
+
+    def test_calendar_truncates_after_last_signal_plus_hold(self):
+        closes = [100.0 + i for i in range(100)]
+        prices = {"AAA": _sim_prices(self.ORD, closes)}
+        cand = _sim_cand("AAA", self.ORD + 10, entry=110.0, stop=105.0, target=200.0)
+        sim = bt._simulate_portfolio(
+            [cand], prices, None, "hold", 5, cost_per_side=0.0, include_trades=True
+        )
+        assert sim is not None
+        end = date.fromisoformat(sim["end_date"])
+        entry = date.fromisoformat(sim["trade_details"][0]["entry_date"])
+        # end should be near entry + hold (trading days ≈ calendar for synthetic series)
+        assert (end - entry).days <= 10
+
+
+def test_fip_id_sign_convention_steady_climber_vs_jump():
+    # Steady climber: many up days, continuous path → lower (more negative) ID.
+    steady = [100.0]
+    for _ in range(280):
+        steady.append(steady[-1] * 1.002)
+    # Jump then flat: one big up day, then zeros → higher ID (more discrete).
+    jumpy = [100.0] * 252
+    jumpy.append(100.0 * 1.5)
+    jumpy.extend([100.0 * 1.5] * 40)
+    i = 260
+    id_steady = bt._fip_id(steady, i)
+    id_jumpy = bt._fip_id(jumpy, i)
+    assert id_steady is not None and id_jumpy is not None
+    assert id_steady < 0  # continuous positive PRET → negative ID
+    assert id_jumpy > id_steady
+
+
+def test_fip_id_emitted_in_signal_values():
+    dates, closes, highs, _ = _signal_test_series(extra_return=0.0005)
+    out = bt._signal_values(dates, closes, highs, 260)
+    assert "fip_id" in out
+    assert -1.0 <= out["fip_id"] <= 1.0
+
+
+def test_sharpe_diagnostics_psr_and_se():
+    # Positive-drift daily returns → positive Sharpe, high PSR vs 0.
+    rets = [0.001 + 0.0001 * (i % 5) for i in range(300)]
+    diag = bt.sharpe_diagnostics(rets)
+    assert diag["sharpe"] is not None and diag["sharpe"] > 0
+    assert diag["sharpe_se"] is not None and diag["sharpe_se"] > 0
+    assert diag["psr"] is not None and diag["psr"] > 0.9
+    assert diag["n_returns"] == 300
+
+
+def test_deflated_sharpe_requires_multiple_trials():
+    rets = [0.001 + 0.0005 * ((-1) ** i) for i in range(400)]
+    diag = bt.sharpe_diagnostics(rets)
+    assert diag["sharpe"] is not None and diag["sharpe_se"] is not None
+    assert bt.deflated_sharpe_ratio(
+        diag["sharpe"], diag["sharpe_se"], n_trials=1, n_returns=diag["n_returns"]
+    ) is None
+    dsr = bt.deflated_sharpe_ratio(
+        diag["sharpe"],
+        diag["sharpe_se"],
+        n_trials=20,
+        n_returns=diag["n_returns"],
+        return_skew=diag["return_skew"],
+        return_kurtosis=diag["return_kurtosis"],
+    )
+    assert dsr is not None
+    assert 0.0 <= dsr <= 1.0
+
+
 def test_bucket_stats_counts_and_expectancy():
     cands = [
         _cand(70, OUTCOME_TARGET_HIT, 3.0),   # +3R win