research: prepare effective risk floor ab

README.md

@@ -255,11 +255,18 @@ A systematic single-variable sweep (offline prod snapshot, production gate/rank/
 | ATR trail multiple {1.5–4.0} | **Keep 3.0** | Return+Sharpe peak; ≤2.0 whipsaws out the momentum right tail; ≥2.5 is a plateau |
 | SPY 200d-MA regime overlay (block entries / go flat) | **Reject** | Halves return (315%→138%) with zero drawdown benefit — the ATR trail already manages downside, and the filter blocks the recovery-phase entries that make the money |
 | Momentum lookback: 6-1, 3-1, 12-7 (Novy-Marx), composites | **Keep residual 12-1** | 6-1/3-1 rank-IC ≈ 0; 12-7 IC 0.045 / t 1.58 — weaker than residual 12-1 (0.055 / t 1.98) |
-| Selection cutoff {70, 75, 85, 90} × book size {10, 15, 20} | **Keep 80 × 10** | Monotonically worse in both directions from 80; the 10-slot cap never binds (<10 concurrent) |
+| Selection cutoff {70, 75, 85, 90} × book size {10, 15, 20} | **Keep cutoff 80; capacity reopened** | The older weekly replay favored 80 × 10, but its no-cap-pressure conclusion is superseded by 519 book-full rejections versus 472 trades under the current daily gate-reset control |
 | Position sizing: equal-weight, inverse-vol, risk-% sweep | **Keep 1% fixed-fractional** | See the inverse-vol warning below |
 | Post-stop re-entry: immediate, fixed 2–5 sessions, gate resets, confirmation filters | **Keep normal gate reset for the 10-position production book** | Sharpe 1.77 vs 1.67 immediate and 1.47 cooldown 5; rerun before changing portfolio capacity |
 | FIP path-smoothness as an in-book tie-breaker/filter | **Reject** (but see the lead below) | Non-monotonic across FIP quintiles within the qualified set; either half of a median split underperforms the full book — thinning the entry stream costs more compounding than the tilt returns |
 
+> **Capacity correction (2026-08-05):** the table's older weekly conclusion
+> that the ten-slot cap never binds is superseded. Under the current daily
+> gate-reset Phase A control, 472 trades were admitted and 519 qualified entries
+> were rejected because the book was full (52.4% of admitted+blocked
+> opportunities). Cutoff 80 remains the signal setting; portfolio capacity is
+> reopened in the focused capacity-bracket study.
+
 Two findings future sessions must not re-litigate:
 
 - **The "inverse-vol sizing win" (July 2026) was mis-attributed — do not resurrect.** The diagnostic sized `notional = equity × 1% / vol_6m`, and the 20% notional cap bound on 95% of entries, so it actually measured "~5 positions × 20% notional each" — a concentration/risk-appetite bump economically equivalent to raising risk to 1.5%, not vol-managed sizing. Genuine inverse-vol sizing (risk budget × median-vol/vol) cuts max drawdown to −18.2% but costs ~58pp total return at flat Sharpe: a risk-preference trade, not edge.

app/services/backtest_service.py

@@ -1320,6 +1320,7 @@ def _replay_candidates_for_period(
     cadence: str = DEFAULT_BACKTEST_CADENCE,
     include_short_candidates: bool = False,
     include_universe_rank_observations: bool = False,
+    outcome_horizon_sessions: int = HORIZON,
 ) -> list[dict]:
     """Slim picklable replay used by local event studies.
 
@@ -1343,10 +1344,13 @@ def _replay_candidates_for_period(
         )
     ]
     cadence = validate_backtest_cadence(cadence)
+    replay_horizon = int(outcome_horizon_sessions)
+    if replay_horizon < 0:
+        raise ValueError('outcome_horizon_sessions must be non-negative')
     candidates: list[dict] = []
     for i in range(
         MIN_LOOKBACK - 1,
-        len(bars) - HORIZON,
+        len(bars) - replay_horizon,
         backtest_step_sessions(cadence),
     ):
         if bars[i].date < start_date:
@@ -1942,6 +1946,7 @@ def _make_gate_reset_reentry_fn(
     cadence: str,
     qualified_fn: Callable[[dict], bool] | None = None,
     ranking_key: str = PRODUCTION_PERCENTILE_KEY,
+    evaluation_horizon_sessions: int = HORIZON,
 ) -> Callable[[str, int, dict, Any], dict | None]:
     """Build the production post-stop gate-reset callback.
 
@@ -1959,11 +1964,18 @@ def _make_gate_reset_reentry_fn(
 
     evaluation_ords: dict[str, set[int]] = {}
     step_sessions = backtest_step_sessions(cadence)
+    evaluation_horizon = int(evaluation_horizon_sessions)
+    if evaluation_horizon < 0:
+        raise ValueError('evaluation_horizon_sessions must be non-negative')
     for symbol, columns in prices.items():
         ordinals = columns[0]
         evaluation_ords[symbol] = {
             int(ordinals[index])
-            for index in range(MIN_LOOKBACK - 1, len(ordinals) - HORIZON, step_sessions)
+            for index in range(
+                MIN_LOOKBACK - 1,
+                len(ordinals) - evaluation_horizon,
+                step_sessions,
+            )
         }
 
     qualified_by_symbol_date: dict[tuple[str, int], dict] = {}
@@ -2010,7 +2022,7 @@ def _simulate_portfolio(
     *,
     qualified_fn: Callable[[dict], bool] | None = None,
     ranking_key: str = PRODUCTION_PERCENTILE_KEY,
-    max_positions: int = SIM_MAX_POSITIONS,
+    max_positions: int | None = SIM_MAX_POSITIONS,
     risk_per_trade: float = SIM_RISK_PER_TRADE,
     atr_trail_multiplier: float = ATR_TRAIL_MULTIPLIER,
     cost_per_side: float = COST_PER_SIDE,
@@ -2034,6 +2046,12 @@ def _simulate_portfolio(
     corr_lookback: int = 120,
     corr_action: str = "skip",
     corr_min_overlap: int = 60,
+    min_initial_risk_fraction: float | None = None,
+    weekly_top_n_rebalance: bool = False,
+    daily_rank_map: dict[tuple[str, str], dict[str, float | None]] | None = None,
+    measurement_start_date: date | None = None,
+    hard_end_date: date | None = None,
+    include_capacity_diagnostics: bool = False,
 ) -> dict | None:
     """Replay the qualified setups as ONE capital-constrained book and report
     portfolio economics from the daily equity curve (return, CAGR, drawdown,
@@ -2083,6 +2101,20 @@ def _simulate_portfolio(
         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")
+    if max_positions is not None and int(max_positions) <= 0:
+        raise ValueError("max_positions must be positive or None")
+    if min_initial_risk_fraction is not None and not (
+        0.0 < float(min_initial_risk_fraction) < 1.0
+    ):
+        raise ValueError("min_initial_risk_fraction must be between 0 and 1")
+    if weekly_top_n_rebalance and (
+        max_positions is None or daily_rank_map is None
+    ):
+        raise ValueError(
+            "weekly_top_n_rebalance requires max_positions and daily_rank_map"
+        )
+    if weekly_top_n_rebalance and fill_mode != FILL_MODE_CLOSE:
+        raise ValueError("weekly_top_n_rebalance requires fill_mode=close")
     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")
@@ -2094,8 +2126,26 @@ def _simulate_portfolio(
 
     entries_by_ord: dict[int, list[dict]] = defaultdict(list)
     start_ord = start_date.toordinal() if start_date is not None else None
+    measurement_start_ord = (
+        measurement_start_date.toordinal()
+        if measurement_start_date is not None
+        else start_ord
+    )
+    hard_end_ord = hard_end_date.toordinal() if hard_end_date is not None else None
     # Explicit simulator/holdout end dates are exclusive split boundaries.
     end_ord = end_date.toordinal() if end_date is not None else None
+    if (
+        start_ord is not None
+        and measurement_start_ord is not None
+        and measurement_start_ord < start_ord
+    ):
+        raise ValueError("measurement_start_date cannot precede start_date")
+    if (
+        hard_end_ord is not None
+        and measurement_start_ord is not None
+        and hard_end_ord <= measurement_start_ord
+    ):
+        raise ValueError("hard_end_date must follow measurement_start_date")
     for c in candidates:
         if not qualified_fn(c) or c.get("direction") != "long":
             continue
@@ -2104,6 +2154,8 @@ def _simulate_portfolio(
             continue
         if end_ord is not None and entry_ord >= end_ord:
             continue  # holdout/validation: entries strictly before the split
+        if hard_end_ord is not None and entry_ord >= hard_end_ord:
+            continue
         if not c.get("entry") or not c.get("stop"):
             continue
         entries_by_ord[entry_ord].append(c)
@@ -2116,7 +2168,12 @@ def _simulate_portfolio(
     }
 
     first_ord = start_ord if start_ord is not None else min(entries_by_ord)
-    calendar = sorted({o for cols in prices.values() for o in cols[0] if o >= first_ord})
+    full_calendar = sorted({o for cols in prices.values() for o in cols[0]})
+    calendar = [
+        o
+        for o in full_calendar
+        if o >= first_ord and (hard_end_ord is None or o < hard_end_ord)
+    ]
     if not calendar:
         return None
 
@@ -2124,20 +2181,39 @@ def _simulate_portfolio(
     # fill lag). Prevents trailing flat-cash after the last resolvable entry —
     # the clear-air train-window bug — for train, validation, and full-period
     # books alike (including max-hold sweeps out to 90 days).
-    last_signal_ord = max(entries_by_ord)
+    if hard_end_ord is None:
-    resolve_pad = hold_days + (1 if fill_mode in DELAYED_FILL_MODES else 0)
+        last_signal_ord = max(entries_by_ord)
-    cut = bisect.bisect_left(calendar, last_signal_ord) + resolve_pad + 1
+        resolve_pad = hold_days + (1 if fill_mode in DELAYED_FILL_MODES else 0)
-    calendar = calendar[:cut]
+        cut = bisect.bisect_left(calendar, last_signal_ord) + resolve_pad + 1
+        calendar = calendar[:cut]
     if not calendar:
         return None
 
+    weekly_rebalance_ords: set[int] = set()
+    for index, session_ord in enumerate(full_calendar):
+        session_date = date.fromordinal(session_ord)
+        iso = session_date.isocalendar()
+        if index + 1 < len(full_calendar):
+            next_iso = date.fromordinal(full_calendar[index + 1]).isocalendar()
+            if (iso.year, iso.week) != (next_iso.year, next_iso.week):
+                weekly_rebalance_ords.add(session_ord)
+        elif session_date.weekday() == 4:
+            weekly_rebalance_ords.add(session_ord)
+
     cash = SIM_STARTING_CAPITAL
     positions: dict[str, dict] = {}
     curve: list[tuple[int, float]] = []
     trades: list[dict] = []
     skipped_full = 0
+    measurement_skipped_full = 0
     skipped_cooldown = 0
     skipped_corr = 0
+    skipped_min_initial_risk = 0
+    measurement_skipped_min_initial_risk = 0
+    opened_positions = 0
+    measurement_opened_positions = 0
+    weekly_rank_rejected_entries = 0
+    measurement_weekly_rank_rejected_entries = 0
     skipped_missing_fill = 0
     skipped_gap_cap = 0
     cooldown_until_index: dict[str, int] = {}
@@ -2152,6 +2228,12 @@ def _simulate_portfolio(
     vol_scalars: list[float] = []
     overnight_slippage_pct: list[float] = []
     pending_delayed: list[dict] = []
+    measurement_start_equity: float | None = None
+    measurement_start_position_count: int | None = None
+    capacity_samples: list[dict[str, float | int]] = []
+    weekly_rebalance_events: list[dict] = []
+    rebalance_exit_index: dict[str, tuple[int, int]] = {}
+    rebalance_reentry_events: list[dict] = []
 
     def _bar(sym: str, o: int):
         idx = index_of.get(sym, {}).get(o)
@@ -2221,6 +2303,13 @@ def _simulate_portfolio(
         cost = proceeds * cost_rate
         cash += proceeds - cost
         risk = pos["entry"] - pos["initial_stop"]
+        initial_risk_dollars = pos["shares"] * risk
+        net_pnl = (
+            proceeds
+            - pos["shares"] * pos["entry"]
+            - cost
+            - pos["entry_cost"]
+        )
         trades.append({
             "symbol": sym,
             "entry_ord": pos["entry_ord"],
@@ -2229,8 +2318,13 @@ def _simulate_portfolio(
             "initial_stop": pos["initial_stop"],
             "active_stop": pos["stop"],
             "fill": fill,
-            "pnl": proceeds - pos["shares"] * pos["entry"] - cost - pos["entry_cost"],
+            "shares": pos["shares"],
+            "initial_risk_dollars": initial_risk_dollars,
+            "pnl": net_pnl,
             "r": (fill - pos["entry"]) / risk if risk > 0 else 0.0,
+            "net_r": net_pnl / initial_risk_dollars
+            if initial_risk_dollars > 0
+            else 0.0,
             "hold": pos["bars_held"],
             "reason": reason,
             "stop_refreshes": pos["stop_refreshes"],
@@ -2245,6 +2339,13 @@ def _simulate_portfolio(
 
     cooldown_sessions = max(0, int(reentry_cooldown_sessions))
     for calendar_index, o in enumerate(calendar):
+        in_measurement = (
+            measurement_start_ord is None or o >= measurement_start_ord
+        )
+        if in_measurement and measurement_start_equity is None:
+            measurement_start_equity = _marked_equity()
+            measurement_start_position_count = len(positions)
+
         # 1) exits on today's bars (stop intraday, target intraday, time at close)
         for sym in list(positions):
             pos = positions[sym]
@@ -2358,6 +2459,82 @@ def _simulate_portfolio(
             reverse=True,
         )
 
+        weekly_selected_entries: list[dict] | None = None
+        if weekly_top_n_rebalance and o in weekly_rebalance_ords:
+            assert max_positions is not None
+            assert daily_rank_map is not None
+            asof = date.fromordinal(o).isoformat()
+            protected: set[str] = set()
+            ranked_pool: list[tuple[float, int, str, dict | None]] = []
+            for sym in positions:
+                rank_row = daily_rank_map.get((sym, asof))
+                current_rank = (
+                    rank_row.get("strategy_rank") if rank_row is not None else None
+                )
+                if current_rank is None or _bar(sym, o) is None:
+                    protected.add(sym)
+                    continue
+                ranked_pool.append((float(current_rank), 0, sym, None))
+
+            entrants_by_symbol: dict[str, dict] = {}
+            for candidate in signal_todays:
+                sym = str(candidate["symbol"])
+                if sym in positions or sym in entrants_by_symbol:
+                    continue
+                entrants_by_symbol[sym] = candidate
+            eligible_entrants = 0
+            for sym, candidate in entrants_by_symbol.items():
+                rank_row = daily_rank_map.get((sym, asof))
+                current_rank = (
+                    rank_row.get("strategy_rank") if rank_row is not None else None
+                )
+                if current_rank is None:
+                    continue
+                eligible_entrants += 1
+                ranked_pool.append((float(current_rank), 1, sym, candidate))
+
+            available_slots = max(0, int(max_positions) - len(protected))
+            ranked_pool.sort(key=lambda row: (-row[0], row[1], row[2]))
+            selected = ranked_pool[:available_slots]
+            selected_holding_symbols = {
+                sym for _rank, kind, sym, _candidate in selected if kind == 0
+            }
+            weekly_selected_entries = [
+                candidate
+                for _rank, kind, _sym, candidate in selected
+                if kind == 1 and candidate is not None
+            ]
+            selected_entrant_symbols = {
+                str(candidate["symbol"]) for candidate in weekly_selected_entries
+            }
+            rejected_now = max(0, eligible_entrants - len(selected_entrant_symbols))
+            weekly_rank_rejected_entries += rejected_now
+            if in_measurement:
+                measurement_weekly_rank_rejected_entries += rejected_now
+
+            exited_symbols: list[str] = []
+            for sym in list(positions):
+                if sym in protected or sym in selected_holding_symbols:
+                    continue
+                bar = _bar(sym, o)
+                if bar is None:
+                    continue
+                _close_trade(sym, float(bar.close), "weekly_rebalance")
+                rebalance_exit_index[sym] = (calendar_index, o)
+                exited_symbols.append(sym)
+
+            weekly_rebalance_events.append({
+                "ord": o,
+                "fresh_entrant_pool": len(entrants_by_symbol),
+                "rank_eligible_entrant_pool": eligible_entrants,
+                "selected_entrants": len(selected_entrant_symbols),
+                "replacements": len(exited_symbols),
+                "exited_symbols": sorted(exited_symbols),
+                "selected_entrant_symbols": sorted(selected_entrant_symbols),
+                "measurement": in_measurement,
+            })
+            equity = _marked_equity()
+
         if fill_mode in DELAYED_FILL_MODES:
             fill_candidates = sorted(
                 pending_delayed,
@@ -2366,7 +2543,11 @@ def _simulate_portfolio(
             )
             pending_delayed = []
         else:
-            fill_candidates = signal_todays
+            fill_candidates = (
+                weekly_selected_entries
+                if weekly_selected_entries is not None
+                else 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."""
@@ -2411,15 +2592,21 @@ def _simulate_portfolio(
             corr_scale: float,
             fill_bar: Any | None,
         ) -> None:
-            nonlocal cash, equity, skipped_full, skipped_cooldown, post_stop_events
+            nonlocal cash, equity, skipped_full, measurement_skipped_full
+            nonlocal skipped_cooldown, post_stop_events
+            nonlocal skipped_min_initial_risk
+            nonlocal measurement_skipped_min_initial_risk
+            nonlocal opened_positions, measurement_opened_positions
             sym = c["symbol"]
             if sym in positions:
                 return
             if calendar_index < cooldown_until_index.get(sym, -1):
                 skipped_cooldown += 1
                 return
-            if len(positions) >= max_positions:
+            if max_positions is not None and len(positions) >= max_positions:
                 skipped_full += 1
+                if in_measurement:
+                    measurement_skipped_full += 1
                 return
             risk_ps = entry - stop
             if risk_ps <= 0 or entry <= 0:
@@ -2436,6 +2623,16 @@ def _simulate_portfolio(
                 (equity * SIM_NOTIONAL_CAP) / entry,
                 max(cash, 0.0) / (entry * (1.0 + cost_rate)),
             )
+            initial_risk_dollars = shares * risk_ps
+            if (
+                min_initial_risk_fraction is not None
+                and initial_risk_dollars
+                < equity * float(min_initial_risk_fraction)
+            ):
+                skipped_min_initial_risk += 1
+                if in_measurement:
+                    measurement_skipped_min_initial_risk += 1
+                return
             if shares * entry < 1.0:
                 return
             entry_cost = shares * entry * cost_rate
@@ -2475,6 +2672,21 @@ def _simulate_portfolio(
                 "vol_scalar": scalar,
                 "corr_scale": corr_scale,
             }
+            opened_positions += 1
+            if in_measurement:
+                measurement_opened_positions += 1
+            prior_rebalance_exit = rebalance_exit_index.pop(sym, None)
+            if prior_rebalance_exit is not None:
+                prior_exit_index, prior_exit_ord = prior_rebalance_exit
+                rebalance_reentry_events.append({
+                    "symbol": sym,
+                    "exit_ord": prior_exit_ord,
+                    "exit_calendar_index": prior_exit_index,
+                    "reentry_calendar_index": calendar_index,
+                    "wait_sessions": calendar_index - prior_exit_index,
+                    "reentry_ord": entry_ord,
+                    "measurement": in_measurement,
+                })
             # next_open only: fill is at the open, so the rest of the bar can stop out.
             # stale_close fills at the close — same-day stop after entry does not apply.
             # bars_held stays 0 on the fill day (matches close-fill cadence).
@@ -2576,7 +2788,25 @@ def _simulate_portfolio(
             # Queue today's signals for the next session's fill.
             pending_delayed.extend(signal_todays)
 
-        curve.append((o, _marked_equity()))
+        marked_equity = _marked_equity()
+        if in_measurement and include_capacity_diagnostics:
+            gross_notional = sum(
+                pos["shares"] * pos["last_close"] for pos in positions.values()
+            )
+            capacity_samples.append({
+                "positions": len(positions),
+                "cash_pct": cash / marked_equity * 100.0
+                if marked_equity > 0
+                else 0.0,
+                "gross_exposure_pct": gross_notional / marked_equity * 100.0
+                if marked_equity > 0
+                else 0.0,
+                "at_capacity": int(
+                    max_positions is not None
+                    and len(positions) >= max_positions
+                ),
+            })
+        curve.append((o, marked_equity))
 
     # Close whatever is still open at its last mark so final equity is realized.
     for sym in list(positions):
@@ -2584,32 +2814,57 @@ def _simulate_portfolio(
     final_equity = cash
     curve[-1] = (calendar[-1], final_equity)
 
-    total_return_pct = (final_equity / SIM_STARTING_CAPITAL - 1.0) * 100.0
+    metric_start_ord = (
-    years = (calendar[-1] - calendar[0]) / 365.25
+        measurement_start_ord if measurement_start_ord is not None else calendar[0]
+    )
+    metric_curve = [(day_ord, eq) for day_ord, eq in curve if day_ord >= metric_start_ord]
+    if not metric_curve:
+        return None
+    metric_base_equity = (
+        measurement_start_equity
+        if measurement_start_date is not None and measurement_start_equity is not None
+        else SIM_STARTING_CAPITAL
+    )
+    total_return_pct = (final_equity / metric_base_equity - 1.0) * 100.0
+    years = (calendar[-1] - metric_start_ord) / 365.25
     cagr_pct = (
-        ((final_equity / SIM_STARTING_CAPITAL) ** (1.0 / years) - 1.0) * 100.0
+        ((final_equity / metric_base_equity) ** (1.0 / years) - 1.0) * 100.0
         if years > 0.25 and final_equity > 0
         else None
     )
 
     peak = float("-inf")
     max_dd = 0.0
-    for _, eq in curve:
+    drawdown_equities = (
+        [metric_base_equity, *(eq for _, eq in metric_curve)]
+        if measurement_start_date is not None
+        else [eq for _, eq in metric_curve]
+    )
+    for eq in drawdown_equities:
         peak = max(peak, eq)
         if peak > 0:
             max_dd = max(max_dd, (peak - eq) / peak)
 
-    rets = [b / a - 1.0 for (_, a), (_, b) in zip(curve, curve[1:]) if a > 0]
+    return_equities = (
+        [metric_base_equity, *(eq for _, eq in metric_curve)]
+        if measurement_start_date is not None
+        else [eq for _, eq in metric_curve]
+    )
+    rets = [
+        b / a - 1.0
+        for a, b in zip(return_equities, return_equities[1:])
+        if a > 0
+    ]
     diag = sharpe_diagnostics(rets)
     sharpe = diag["sharpe"]
 
     # Per-calendar-year returns off the equity curve — shows whether every year
     # contributed or one exceptional stretch carried the result.
     yearly: list[dict] = []
-    year_start_eq = curve[0][1]
+    year_start_eq = metric_base_equity
-    cur_year = date.fromordinal(curve[0][0]).year
+    cur_year = date.fromordinal(metric_start_ord).year
-    last_eq = curve[0][1]
+    last_eq = metric_base_equity
-    for o, eq in curve:
+    for o, eq in metric_curve:
         y = date.fromordinal(o).year
         if y != cur_year:
             yearly.append({
@@ -2628,24 +2883,29 @@ def _simulate_portfolio(
         ),
     })
 
-    pnls = [t["pnl"] for t in trades]
+    metric_trades = [
+        trade for trade in trades if trade["entry_ord"] >= metric_start_ord
+    ]
+    pnls = [t["pnl"] for t in metric_trades]
     wins = sum(1 for p in pnls if p > 0)
     reason_counts = {
-        reason: sum(1 for t in trades if t["reason"] == reason)
+        reason: sum(1 for t in metric_trades if t["reason"] == reason)
-        for reason in sorted({t["reason"] for t in trades})
+        for reason in sorted({t["reason"] for t in metric_trades})
     }
     spy_pct = None
     if spy_closes:
         from app.services.benchmark_service import benchmark_return_pct
 
         spy_pct = benchmark_return_pct(
-            spy_closes, date.fromordinal(calendar[0]), date.fromordinal(calendar[-1])
+            spy_closes,
+            date.fromordinal(metric_start_ord),
+            date.fromordinal(calendar[-1]),
         )
 
     curve_payload: list[dict] | None = None
     benchmark_payload: list[dict] | None = None
     if include_curve:
-        curve_base = curve[0][1] if curve else SIM_STARTING_CAPITAL
+        curve_base = metric_base_equity
         curve_payload = [
             {
                 "date": date.fromordinal(o).isoformat(),
@@ -2654,12 +2914,12 @@ def _simulate_portfolio(
                 if curve_base > 0
                 else None,
             }
-            for o, eq in curve
+            for o, eq in metric_curve
         ]
         if spy_closes:
             benchmark_payload = []
             base_spy = None
-            for o, _ in curve:
+            for o, _ in metric_curve:
                 d = date.fromordinal(o)
                 close = spy_closes.get(d)
                 if close is None or close <= 0:
@@ -2678,6 +2938,8 @@ def _simulate_portfolio(
         calmar = float(cagr_pct) / max_dd_pct
     result = {
         "starting_capital": SIM_STARTING_CAPITAL,
+        "measurement_start_equity": round(metric_base_equity, 2),
+        "measurement_start_positions": measurement_start_position_count or 0,
         "cost_per_side_pct": round(cost_rate * 100.0, 3),
         "fill_mode": fill_mode,
         "final_equity": round(final_equity, 2),
@@ -2691,23 +2953,161 @@ def _simulate_portfolio(
         "n_returns": diag["n_returns"],
         "return_skew": diag["return_skew"],
         "return_kurtosis": diag["return_kurtosis"],
-        "trades": len(trades),
+        "trades": len(metric_trades),
-        "win_rate": round(wins / len(trades) * 100.0, 1) if trades else None,
+        "win_rate": (
+            round(wins / len(metric_trades) * 100.0, 1)
+            if metric_trades
+            else None
+        ),
         "avg_trade_pnl": round(sum(pnls) / len(pnls), 2) if pnls else None,
-        "best_trade_r": round(max(t["r"] for t in trades), 2) if trades else None,
+        "best_trade_r": (
-        "worst_trade_r": round(min(t["r"] for t in trades), 2) if trades else None,
+            round(max(t["r"] for t in metric_trades), 2)
+            if metric_trades
+            else None
+        ),
+        "worst_trade_r": (
+            round(min(t["r"] for t in metric_trades), 2)
+            if metric_trades
+            else None
+        ),
         "best_trade_pnl": round(max(pnls), 2) if pnls else None,
         "worst_trade_pnl": round(min(pnls), 2) if pnls else None,
         "avg_hold_days": (
-            round(sum(t["hold"] for t in trades) / len(trades), 1) if trades else None
+            round(
+                sum(t["hold"] for t in metric_trades) / len(metric_trades),
+                1,
+            )
+            if metric_trades
+            else None
         ),
         "exit_reasons": reason_counts,
         "skipped_book_full": skipped_full,
         "spy_return_pct": round(spy_pct, 1) if spy_pct is not None else None,
         "yearly_returns": yearly,
-        "start_date": date.fromordinal(calendar[0]).isoformat(),
+        "start_date": date.fromordinal(metric_start_ord).isoformat(),
         "end_date": date.fromordinal(calendar[-1]).isoformat(),
     }
+    if measurement_start_date is not None:
+        result["simulation_start_date"] = date.fromordinal(calendar[0]).isoformat()
+    if hard_end_date is not None:
+        result["hard_end_date_exclusive"] = hard_end_date.isoformat()
+    if measurement_start_date is not None:
+        result["measurement_skipped_book_full"] = measurement_skipped_full
+        result["measurement_opened_positions"] = measurement_opened_positions
+    if min_initial_risk_fraction is not None:
+        result["min_initial_risk_fraction"] = float(min_initial_risk_fraction)
+        result["skipped_min_initial_risk"] = skipped_min_initial_risk
+        result["measurement_skipped_min_initial_risk"] = (
+            measurement_skipped_min_initial_risk
+        )
+    if include_capacity_diagnostics:
+        measured_opened = (
+            measurement_opened_positions
+            if measurement_start_date is not None
+            else opened_positions
+        )
+        measured_full = (
+            measurement_skipped_full
+            if measurement_start_date is not None
+            else skipped_full
+        )
+        capacity_opportunities = measured_opened + measured_full
+        result["opened_positions"] = measured_opened
+        result["capacity_opportunities"] = capacity_opportunities
+        result["blocked_fraction"] = (
+            round(measured_full / capacity_opportunities, 6)
+            if capacity_opportunities
+            else 0.0
+        )
+        result["avg_positions"] = (
+            round(
+                sum(float(sample["positions"]) for sample in capacity_samples)
+                / len(capacity_samples),
+                4,
+            )
+            if capacity_samples
+            else 0.0
+        )
+        result["peak_positions"] = (
+            max(int(sample["positions"]) for sample in capacity_samples)
+            if capacity_samples
+            else 0
+        )
+        result["sessions_at_capacity"] = sum(
+            int(sample["at_capacity"]) for sample in capacity_samples
+        )
+        result["sessions_measured"] = len(capacity_samples)
+        result["avg_cash_pct"] = (
+            round(
+                sum(float(sample["cash_pct"]) for sample in capacity_samples)
+                / len(capacity_samples),
+                4,
+            )
+            if capacity_samples
+            else None
+        )
+        result["avg_gross_exposure_pct"] = (
+            round(
+                sum(
+                    float(sample["gross_exposure_pct"])
+                    for sample in capacity_samples
+                )
+                / len(capacity_samples),
+                4,
+            )
+            if capacity_samples
+            else None
+        )
+    if weekly_top_n_rebalance:
+        measured_events = [
+            event for event in weekly_rebalance_events if event["measurement"]
+        ]
+        measured_reentries = [
+            event for event in rebalance_reentry_events if event["measurement"]
+        ]
+        result["weekly_rank_rejected_entries"] = (
+            measurement_weekly_rank_rejected_entries
+            if measurement_start_date is not None
+            else weekly_rank_rejected_entries
+        )
+        result["weekly_rebalance_events"] = [
+            {
+                **{
+                    key: value
+                    for key, value in event.items()
+                    if key not in {"ord", "measurement"}
+                },
+                "date": date.fromordinal(event["ord"]).isoformat(),
+            }
+            for event in measured_events
+        ]
+        result["rebalance_reentry_events"] = [
+            {
+                **{
+                    key: value
+                    for key, value in event.items()
+                    if key
+                    not in {
+                        "exit_ord",
+                        "reentry_ord",
+                        "measurement",
+                        "exit_calendar_index",
+                        "reentry_calendar_index",
+                    }
+                },
+                "exit_date": date.fromordinal(event["exit_ord"]).isoformat(),
+                "reentry_date": date.fromordinal(
+                    event["reentry_ord"]
+                ).isoformat(),
+            }
+            for event in measured_reentries
+        ]
+        for session_limit in (5, 10, 20):
+            result[f"rebalance_reentries_within_{session_limit}_sessions"] = sum(
+                1
+                for event in measured_reentries
+                if int(event["wait_sessions"]) <= session_limit
+            )
     if vol_target is not None:
         result["vol_target"] = vol_target
         result["vol_lookback"] = int(vol_lookback)
@@ -2782,7 +3182,7 @@ def _simulate_portfolio(
                 "entry_date": date.fromordinal(trade["entry_ord"]).isoformat(),
                 "exit_date": date.fromordinal(trade["exit_ord"]).isoformat(),
             }
-            for trade in trades
+            for trade in metric_trades
         ]
     return result
 

docs/research/README.md

@@ -25,7 +25,7 @@ score, Structural S/R, the Gate Target Ladder, sentiment, fundamentals) is
 | 1.5× ATR initial stop | Real exit | Cuts losers fast |
 | 3× ATR trailing stop, 30-day max hold | Real exit | Best Sharpe of every exit tested |
 | Post-stop normal gate reset | Re-entry policy | Stop always closes; a later gate failure and subsequent fresh qualification define the next signal episode. The selected study arm reached Sharpe 1.77 / CAGR 48.3% at capacity 10; live scan-before-outcome timing is stricter (Sharpe 1.68 / CAGR 44.8% analogue). [Full study](post-stop-reentry.md) |
-| Max 10 concurrent positions, 1% risk per trade | Sizing | Cap never binds in practice |
+| Max 10 concurrent positions, 1% risk per trade | Sizing | The cap binds by signal count, but the focused bracket found negligible opportunity cost: cap 15 admitted every blocked setup and added only 0.0018 R/trade in affected paths. [Findings](portfolio-capacity-bracket-findings.md) |
 | Structural S/R | Human-facing product context | Clean, capped zones for charts and alerts; not read by the scanner |
 | Gate Target Ladder | Screening machinery | Volume-free transient proposals preserve the production candidate set exactly; never an exit |
 
@@ -61,7 +61,7 @@ invites overfitting.
 |---|---|
 | ATR trail multiple {1.5–4.0} | **Keep 3.0** — ≤2.0 whipsaws out the right tail; ≥2.5 is a plateau |
 | Momentum lookback (6-1, 3-1, 12-7 Novy-Marx, composites) | **Keep residual 12-1** — the others have IC ≈ 0 or weaker t-stats |
-| Selection cutoff {70…90} × book size {10, 15, 20} | **Keep 80 × 10** — monotonically worse in both directions |
+| Selection cutoff {70…90} × book size {10, 15, 20} | **Keep 80 × 10** — the focused daily bracket found no meaningful gain from cap 15, while weekly rank replacement hurt. [Findings](portfolio-capacity-bracket-findings.md) |
 | Position sizing (equal-weight, inverse-vol, risk-% sweep) | **Keep 1% fixed-fractional** |
 | Primary-target probability floor | **Keep 20%** — pruned lottery targets, 1,428 → 1,089 qualified, lifted Sharpe |
 | Primary-target R:R selector | **Keep 1.5** — target choice is intentionally independent of the later 2.0 activation floor |
@@ -146,6 +146,7 @@ knobs.
 | **Broader universe** | Composition changes factor signs (fip tug-of-war); vol-tilt on breadth is only a **directional hypothesis** (auth. −0.048 / t −1.36) | Any prod broaden must re-validate 80/20 tilt; offline research only; research.sqlite requires completion manifest |
 | **Forward paper-trade record** | The only true out-of-sample evidence the snapshot cannot give | Time; mark entries at actual near-close fill once ops ships |
 | **Better target model for clear-air names** | The return is demonstrably there (#2 wins on raw CAGR in *both* train and test); it's the *flat* 3× ATR target that makes it too expensive in risk | Needs a per-name model, not a constant k×ATR |
+| **Minimum effective-risk floor** | In cap-never-bound paths, the confounded 0.5% floor arm removed about 8% of fills while EV rose from 0.328 to 0.399 R and PF from 1.60 to 1.75, with exposure nearly unchanged | Run the frozen single-variable cap-10 A/B. [Specification](effective-risk-floor-ab.md) / [capacity findings](portfolio-capacity-bracket-findings.md) |
 
 ---
 
@@ -197,4 +198,15 @@ qualification. The [daily re-entry matrix](post-stop-reentry.md) supports this
 for the current 10-position book, but not as a universal rule for other
 portfolio capacities.
 
+Capacity is now closed as a negative result. The current daily Phase A control
+does reject 519 qualified entries because the ten-slot book is full versus 472
+admitted trades, so the older weekly “cap never binds” claim was stale. But the
+clean cap-15 arm admitted every opportunity the strategy requested and added
+only 0.0018 R/trade in paths where cap 10 bound. Weekly current-rank replacement
+reduced mean EV and created substantial churn. Keep cap 10 and do not build the
+replacement policy. See the [frozen specification](portfolio-capacity-bracket.md)
+and the separate [capacity findings](portfolio-capacity-bracket-findings.md).
+The only open follow-up from that run is the
+[frozen confound-free 0.5% minimum effective-risk-floor A/B](effective-risk-floor-ab.md).
+
 The next real evidence is **forward**, not backward: the live paper-trade record.

docs/research/effective-risk-floor-ab.md

@@ -0,0 +1,124 @@
+# Effective initial-risk floor A/B - frozen specification
+
+Date frozen: 2026-08-05
+Branch: research/portfolio-capacity-rebalancing
+Runner: scripts/run_portfolio_construction_matrix.py
+Study ID: risk-floor-ab
+
+## Question
+
+Does rejecting an otherwise qualified cap-10 entry when its actual initial
+stop-risk after cash and notional sizing is below 0.5% of marked equity improve
+trade selection?
+
+The completed capacity bracket cannot answer this. Its cash_unbounded arm
+removed the count cap and applied the 0.5% floor simultaneously. In the 70 paths
+where the control cap never bound, that arm still raised mean EV from 0.328 to
+0.399 R and profit factor from 1.60 to 1.75 while trades fell about 8% and
+exposure stayed nearly flat. Capacity was a no-op in those paths, so the floor
+is the plausible cause, but the prior arm remains confounded.
+
+This A/B changes only the floor. It has no formal promotion gate and does not
+automatically change production.
+
+## Frozen arms
+
+1. cap10_incumbent: current production-style cap-10 control, with no minimum
+   effective-risk floor.
+2. cap10_min_risk_005: the same cap-10 strategy, rejecting an entry only when
+   actual initial stop-risk after cash/notional sizing is below 0.5% of marked
+   equity.
+
+Both arms have max_positions=10, weekly replacement disabled, 1% target risk
+per trade, and identical admission ordering. The only differing simulator
+argument is min_initial_risk_fraction: None versus 0.005.
+
+All other settings remain the frozen daily Phase A control: current production
+construction universe, full-universe residual-momentum/low-volatility 80/20
+rank, threshold 80, normal gate-reset re-entry, close fills, 3x ATR trail,
+30-session maximum hold, 20% per-position notional ceiling, no leverage, and
+costs of 0.10% and 0.20% per fill.
+
+Every priced symbol contributes to the daily cross-sectional rank. Rank-only
+symbols cannot submit trades. Validation retains the 450-600-symbol production
+construction guardrail and the legacy-snapshot column-scoped loader.
+
+## Frozen cohorts
+
+Reuse the completed bracket's point-in-time daily candidate/rank cache and
+cohort manifest:
+
+- Empty book: first eligible session of each month in 2019-2025, with 504 prior
+  scoring sessions and 252 measurement sessions. This is the primary start-date
+  evidence.
+- Warm book: weekly seeds 63-126 sessions before each 2019-2025 annual anchor,
+  with state carried into the same 252-session measurement window. This is a
+  state-carrying replication, not independent evidence.
+
+The expected realization is 78 empty-book paths, 97 warm paths, seven annual
+clusters in each protocol, two costs, two arms, and 700 cells.
+
+Do not use warm-seed IQR as evidence. Six of seven completed-bracket anchors
+were structurally degenerate because fractional sizing is scale invariant and
+the 30-session maximum hold washed out books before anchors. The 2023 exception
+shows that state carrying itself works.
+
+## Reporting and interpretation
+
+For every protocol and cost, pair identical paths. Report:
+
+- mean, median, P25, and P75 paired net-EV changes in R;
+- positive-path and bit-identical-path fractions;
+- the median paired delta within each year and the median across seven years;
+- simple 90% cluster-bootstrap context for EV and Calmar, with no CI gate;
+- mean paired PF, Gain-to-Pain, Sortino, Calmar/MAR, CAGR, maximum drawdown,
+  total return, and Sharpe changes;
+- trades, floor rejections, holding time, cash, gross exposure, average/peak
+  positions, turnover, and costs.
+
+Means and identical-path fractions must appear beside medians so inert cohorts
+cannot turn a left- or right-skewed treatment into a misleading zero headline.
+For these 252-session windows, the implementation's full-window Calmar is CAGR
+divided by maximum drawdown, the same numeric definition commonly called MAR;
+do not present the duplicate label as a second independent metric.
+
+Today's production membership is projected backward. Use paired differences
+for the treatment conclusion; absolute profitability remains descriptive and
+survivorship-biased. Empty and warm protocols cover the same seven market years
+and must not be interpreted as independent replications.
+
+Interpretation is deliberately simple:
+
+- a positive result means the isolated floor improves the paired EV
+  distribution without an economically important loss of total-return or
+  drawdown quality;
+- a negative result closes the floor;
+- mixed EV/portfolio-quality results are reported as a trade-off, not forced
+  through a composite score.
+
+## Reproducibility and macOS execution
+
+The authoritative run refuses a dirty worktree. Its fingerprint includes the
+implementation commit, this specification hash, snapshot hash, candidate-cache
+key, construction view, cohort manifest, arm definitions, costs, and study
+version. Cells checkpoint atomically and --resume verifies the fingerprint.
+
+From the repository root on macOS:
+
+    python3 -m venv .venv
+    ./.venv/bin/python -m pip install -e '.[dev]'
+
+Preflight, reusing the completed bracket's candidate/rank cache:
+
+    ./.venv/bin/python scripts/run_portfolio_construction_matrix.py +      backtest_snapshots/research.sqlite +      --study risk-floor-ab +      --run-id prod505-effective-risk-floor-ab-daily-v1 +      --candidate-cache reports/.cache/prod505-capacity-bracket-daily-v1-candidates.pkl +      --workers 8 +      --resume +      --validate-only
+
+Authoritative run:
+
+    ./.venv/bin/python scripts/run_portfolio_construction_matrix.py +      backtest_snapshots/research.sqlite +      --study risk-floor-ab +      --run-id prod505-effective-risk-floor-ab-daily-v1 +      --candidate-cache reports/.cache/prod505-capacity-bracket-daily-v1-candidates.pkl +      --workers 8 +      --resume
+
+On an M2 Pro, eight workers is the explicit high-utilization setting. Use six
+instead on a memory-constrained machine; auto intentionally caps itself at six.
+Changing worker count does not change the fingerprint or results.
+
+Commit only the compact final JSON and Markdown reports. Candidate caches,
+checkpoints, raw curves, and trade ledgers remain ignored.

docs/research/phase-a-matrix.md

@@ -28,6 +28,13 @@ Mechanics guards confirmed before reading results: calendar truncation asserted
 | **Validation** | **1.68** | **0.72** | **41.6%** | **20.9%** | **1.99** | **239** |
 | Full (close-fill) | 1.77 | 0.50 | 48.3% | 21.6% | 2.23 | 472 |
 
+**Capacity correction (2026-08-05):** the full close-fill control also records
+skipped_book_full = 519 versus 472 admitted trades, so the ten-slot book
+refuses 52.4% of admitted+blocked qualified opportunities. The older weekly
+claim that the cap never bound is stale and does not apply to this daily
+gate-reset configuration. Capacity is now isolated in the
+[focused bracket study](portfolio-capacity-bracket.md).
+
 Validation SE ≈ 0.72 — almost no arm clears a 1-SE delta.
 
 ---

docs/research/portfolio-capacity-bracket-findings.md

@@ -0,0 +1,124 @@
+# Portfolio-capacity bracket — findings
+
+Date interpreted: 2026-08-05
+
+Status: **capacity and weekly replacement closed as negative results; the
+minimum effective-risk floor remains an open single-variable follow-up.**
+
+This document interprets the frozen v2 run without modifying its generated
+outputs:
+
+- result commit: `24482c6`;
+- simulation source commit: `6fc82ae8574de9104c83273e018391e75a5f8ac6`;
+- frozen specification SHA-256:
+  `f1e37783cf6d157ecc827d48211fa45da16f0a0ac19cd23686b3902d347a1898`;
+- JSON SHA-256:
+  `2435875667097db7416a0d96f412db81d2f2d09ba053748c9f2cfb8a0cba4417`;
+- Markdown SHA-256:
+  `dc3f5de25eb0a156ce51d0025c90e04ac0977e9502dec47bcf1b25bdcf609c81`.
+
+The run completed 78 empty-book paths, 97 warm-seed paths, seven annual
+clusters under both protocols, two cost levels, four arms, and 1,400 cells with
+no validation errors. The construction universe was 505 priced tradable
+symbols plus 4,149 priced rank-only symbols.
+
+## Capacity is economically free
+
+The clean capacity treatment is `cap15_incumbent`: it changes no sizing or
+admission rule. Its cap never bound in any cell (maximum observed position count
+12; zero full-book skips), so it absorbed every opportunity blocked by cap 10.
+
+At 0.10% per fill, split the 175 paths by whether the paired control recorded
+any `skipped_book_full`. Values below are mean paired changes in net EV per
+trade, in R:
+
+| Arm | Cap never bound (n=70) | Cap did bind (n=105) |
+|---|---:|---:|
+| `cap15_incumbent` | +0.0000 | +0.0018 |
+| `cash_unbounded` | +0.0714 | +0.0077 |
+| `cap10_weekly_top10` | -0.0246 | -0.0426 |
+
+The exact zero for cap15 in the never-bound stratum is also a harness validity
+check: when the treatment cannot act, results are identical. Where it does act,
+giving the strategy every slot it requested adds only 0.0018 R/trade. The old
+519-blocked-versus-472-admitted count was true, but it did not imply that the
+blocked opportunities were economically valuable.
+
+Decision: **keep the production cap at 10.** Do not remove it or raise it in the
+expectation of additional edge.
+
+## The positive arm measured the risk floor
+
+`cash_unbounded` combined two treatments: no count cap and a 0.5% minimum
+effective initial-risk fraction. Its EV effect is roughly nine times larger in
+the 70 paths where the control cap never bound, so capacity cannot explain the
+improvement.
+
+Within that never-bound stratum:
+
+| Measure | Control | `cash_unbounded` |
+|---|---:|---:|
+| Mean trades | 75.7 | 69.9 |
+| Mean cash | 27.8% | 28.2% |
+| Mean gross exposure | 72.2% | 71.8% |
+| Mean hold | 15.4 sessions | 15.6 sessions |
+| Mean EV | +0.328 R | +0.399 R |
+| Mean profit factor | 1.60 | 1.75 |
+
+The floor removes about 8% of fills while leaving exposure and holding time
+nearly unchanged. This is selection, not general de-risking: candidates that
+available sizing compresses below half the intended risk are worse on average.
+The report records repeated reject attempts, not the rejected candidates'
+ranks, so whether the effect is rank-mediated remains unknown.
+
+Next research: one single-variable A/B, `cap10_incumbent` versus cap 10 with
+`min_initial_risk_fraction=0.005`, with every other rule unchanged. Do not call
+the current `cash_unbounded` result causal evidence for that floor until this
+confound-free comparison is run.
+
+## Weekly replacement hurts
+
+Median paired deltas read zero because enough cohorts are inert. The distribution
+is not neutral:
+
+| Protocol | Mean ΔEV | P25 ΔEV | Identical paths |
+|---|---:|---:|---:|
+| Empty book | -0.0360 R | -0.0817 R | 27/78 (34.6%) |
+| Warm book | -0.0348 R | -0.1582 R | 14/97 (14.4%) |
+
+The arm made 2,170 replacements and 529 same-symbol re-entries within ten
+sessions, so 24% of replacements were associated with short-horizon churn.
+
+Decision: **reject weekly top-10 replacement.** Future reports should show mean
+paired effects and identical-path fractions beside medians whenever treatments
+are inert in a material share of cohorts.
+
+## Warm dispersion was mostly structurally degenerate
+
+For six of seven anchors, control EV IQR is numerical zero (approximately
+`1e-16`) and Calmar IQR is exactly zero. The displayed ratio `1.000` is therefore
+mostly the implementation's zero-over-zero convention, not evidence of equal
+nonzero dispersion.
+
+Two mechanics cause convergence: sizing and notional limits are fractions of
+equity, making R and ratio metrics scale-invariant; and the 30-session maximum
+hold is shorter than the 63-session minimum seed offset, allowing initial books
+to wash out before the anchor.
+
+The exception is 2023. Control measurement-start positions vary from 6 to 9,
+EV IQR is 0.0274 R, and Calmar IQR is 0.2675. The protocol therefore carries
+state correctly, but its chosen offsets usually erase the initialization effect
+it was intended to measure.
+
+Future initialization studies should use seed offsets shorter than maximum hold,
+approximately 5–25 sessions. The current empty-book cohorts remain the primary
+start-date evidence, but they necessarily mix initialization with market regime.
+
+## Final decisions
+
+1. Keep cap 10; its measured opportunity cost is negligible.
+2. Reject weekly rank replacement.
+3. Do not interpret the `cash_unbounded` improvement as a capacity effect.
+4. Run only the focused cap-10 effective-risk-floor A/B next.
+5. Report means, inert fractions, and absolute dispersion beside medians and
+   ratios in future sparse-treatment studies.

docs/research/portfolio-capacity-bracket.md

@@ -0,0 +1,169 @@
+# Portfolio-capacity bracket — frozen specification
+
+Date frozen: 2026-08-05
+Branch: research/portfolio-capacity-rebalancing
+Runner: scripts/run_portfolio_construction_matrix.py
+
+## Question and motivation
+
+The daily Phase A production control (a0_control: close fill, 30-session
+maximum hold, 1% fixed-fractional risk, no correlation or volatility overlay)
+recorded 472 trades and 519 otherwise qualified entries rejected because the
+ten-position book was full. The blocked share is 519 / (519 + 472) = 52.4%.
+The book is therefore materially arrival-order constrained.
+
+This supersedes the older statement that the ten-slot cap never bound. That
+statement came from a shorter, weekly, pre-gate-reset replay and is not evidence
+about the current daily strategy.
+
+The study brackets the value of capacity before tuning replacement details. It
+does not contain a formal promotion rule or automatically change production.
+Because the current ~505-name production membership is projected backward,
+paired arm-versus-control differences are the primary evidence. Absolute
+profitability is descriptive and survivorship-biased.
+
+Implementation correction: the first completed v1 artifact at commit `23fe39f`
+incorrectly allowed the snapshot's broad rank-only universe to submit trades.
+That artifact is invalid, is removed from the branch, and must not be used for
+strategy conclusions. Runner v2 fixes the construction/ranking partition below.
+
+## Frozen arms
+
+1. **cap10_incumbent:** exact production-style cap-10 control, no displacement.
+2. **cash_unbounded:** no position-count cap; cash/no leverage and the existing
+   20% per-position notional ceiling remain. Reject an entry if actual initial
+   stop-risk after cash/notional sizing is below 0.5% of marked equity.
+3. **cap10_weekly_top10:** on the final trading session of each ISO week, rank
+   holdings plus fresh same-day qualified entrants and retain the top ten.
+4. **cap15_incumbent:** cap 15, no displacement.
+
+All arms use the frozen Phase A control configuration: daily candidate replay,
+live-like full-universe residual-momentum/low-volatility 80/20 rank, activation
+threshold 80, normal gate-reset re-entry, close fill, 3×ATR trail, 30-session
+maximum hold, 1% risk, and costs of 0.10% and 0.20% per fill.
+
+Every priced symbol contributes to the daily cross-sectional rank. Only symbols
+not listed in the snapshot's `research_rank_only` side table may submit trade
+setups to any arm. The resulting construction universe must contain 450-600
+symbols (expected approximately 505); validation fails outside that frozen
+guardrail or when the side table references unknown ticker symbols.
+
+The daily replay uses zero outcome horizon: setup and rank observations continue
+through the snapshot's last session because portfolio simulation, unlike outcome
+grading, does not require 30 future bars.
+
+Control-parity note: a direct main-versus-branch comparison found identical
+total return, CAGR, maximum drawdown, and Sharpe. The branch intentionally
+changes only the first calendar year's `yearly_returns` convention: it starts
+from initial capital rather than equity after the first session, so day-one
+entry costs are now charged to year one. Older reports can therefore show a
+different first-year contextual return without a strategy-performance
+regression. New trade-detail and measurement-start fields are additive.
+
+### Weekly-selection mechanics
+
+- Ordinary exits run before entries/rebalancing.
+- Open slots may still fill from daily qualified entries during the week.
+- On the final ISO-week session, current holdings and that day's fresh qualified
+  entrants use the full-universe strategy_rank for that same date.
+- Stored entry-day rank is never used.
+- Holdings with missing current rank/data are protected and consume a slot;
+  entrants missing rank are ineligible.
+- Incumbents win exact rank ties; symbol is the deterministic final tie-breaker.
+- Rebalance exits pay costs and bypass cooldown/post-stop state.
+- Report entrant-pool sizes, replacements, turnover, and same-symbol re-entry
+  within 5/10/20 sessions.
+
+## Frozen cohorts
+
+research.sqlite is expected to cover 2016-01-04 through 2026-07-17. Residual
+momentum requires 252 benchmark sessions. Empty-book starts additionally require
+504 prior scoring sessions and 252 forward measurement sessions.
+
+- **Empty book:** first eligible session of each month, approximately January
+  2019 through July 2025; start with no positions and measure 252 sessions.
+- **Warm book:** first session of each year 2019–2025 is the measurement anchor.
+  Seed the portfolio on the first session of every ISO week falling 63–126
+  trading sessions before the anchor, carry all positions and gate-reset state
+  forward, and measure the same 252-session anchor window.
+
+Warm portfolio returns reset to marked equity immediately before the anchor
+session. P&L after the anchor from carried positions belongs to portfolio
+returns, while trade EV includes only entries on or after the anchor. Remaining
+positions liquidate at the last measurement close with costs.
+
+The validate-only mode must print realized cohort counts and fail unless both
+protocols contain the seven annual clusters 2019–2025 and every warm anchor has
+at least 12 seeds. It must also print ranking, rank-only, and tradable symbol
+counts plus the raw, removed, and retained qualified-long counts.
+
+## Reporting
+
+Primary reported measures:
+
+- net EV per trade in R, with costs and actual initial stop-risk dollars;
+- Calmar (CAGR / max drawdown);
+- profit factor on net trade R;
+- Gain-to-Pain (sum of all monthly returns / absolute sum of negative months);
+- Sortino using daily returns and zero target.
+
+Also report total return/CAGR, maximum drawdown, Sharpe, win rate, time
+underwater, exposure, cash, average/peak positions, sessions at capacity,
+turnover, costs, qualified/admitted/blocked opportunities, and minimum-risk
+rejections.
+
+For each arm/protocol/cost/metric, pair identical paths with cap10_incumbent,
+take the median paired delta within each start year or annual anchor, show all
+seven cluster values, and headline their median.
+
+Initialization dispersion is reported separately for EV and Calmar: calculate
+the seed-path IQR within each warm anchor, divide by the paired control IQR, show
+all seven ratios, and headline their median. Do not combine them into a composite.
+
+For context only, run a deterministic 10,000-replicate cluster bootstrap over
+the seven paired annual summaries and report the central 90% percentile interval
+for median EV and Calmar deltas and warm IQR ratios. These intervals are not
+promotion gates, independent-population confidence claims, or formal inference.
+
+## Reproducibility and execution
+
+Candidate replay/ranks cache under reports/.cache; each matrix cell checkpoints
+atomically and resume verifies a fingerprint over the implementation commit,
+this specification hash, snapshot SHA-256, cache key, arm definitions, costs,
+and cohort manifest. An authoritative run refuses a dirty worktree.
+
+The existing v1 candidate/rank cache is intentionally reusable: its
+full-universe current-day ranks are correct. Runner v2 derives a fingerprinted
+construction view by removing qualified rows whose symbols are rank-only. V2
+uses a versioned checkpoint directory, so invalid v1 portfolio cells are never
+resumed and the expensive daily rank replay does not need to run again.
+
+The loader reads only ticker ID/symbol and the OHLCV columns used by replay, so
+snapshots created before SEC metadata added `tickers.cik`, `tickers.sic`, and
+`tickers.sic_description` remain valid. Do not migrate or alter the research
+snapshot: its original SHA-256 is part of the run fingerprint.
+
+macOS environment setup from the repository root (zsh):
+
+    python3 -m venv .venv
+    ./.venv/bin/python -m pip install -e '.[dev]'
+
+Preflight:
+
+    ./.venv/bin/python scripts/run_portfolio_construction_matrix.py \
+      backtest_snapshots/research.sqlite \
+      --run-id prod505-capacity-bracket-daily-v1 \
+      --workers auto \
+      --resume \
+      --validate-only
+
+Authoritative run:
+
+    ./.venv/bin/python scripts/run_portfolio_construction_matrix.py \
+      backtest_snapshots/research.sqlite \
+      --run-id prod505-capacity-bracket-daily-v1 \
+      --workers auto \
+      --resume
+
+Commit only the compact final JSON and Markdown reports. Raw curves, trades,
+candidate caches, and checkpoints remain ignored.

reports/portfolio-construction-prod505-capacity-bracket-daily-v1.md

@@ -0,0 +1,114 @@
+# Focused daily portfolio-capacity matrix
+
+Generated: 2026-08-05T19:25:17.150472+00:00
+
+## Question
+
+The current daily Phase A control admitted 472 trades and rejected 519 qualified opportunities because the ten-slot book was full. This run brackets the economic cost of that binding constraint; it has no formal promotion gate.
+
+> Universe caveat: today's production membership is projected backward. Use paired arm-versus-control differences, not absolute profitability, for construction conclusions.
+
+## Validated universes
+
+- Tradable setup symbols with prices: 505.
+- Rank-only symbols with prices: 4149.
+- Full ranking symbols with prices: 4654.
+- Tradable qualified longs: 6118.
+- Rank-only qualified rows removed: 136286.
+
+## Paired annual medians
+
+### Empty Book — 0.10% per fill
+
+| Arm | ΔEV net R | 90% context | ΔCalmar | 90% context |
+|---|---:|---:|---:|---:|
+| cap10_incumbent | 0.000 | [0.000, 0.000] | 0.000 | [0.000, 0.000] |
+| cash_unbounded | 0.044 | [-0.011, 0.060] | 0.030 | [-0.030, 0.120] |
+| cap10_weekly_top10 | 0.000 | [-0.091, 0.000] | 0.000 | [-0.260, 0.000] |
+| cap15_incumbent | 0.000 | [0.000, 0.011] | 0.000 | [0.000, 0.130] |
+
+| Arm | ΔPF | ΔGain-to-Pain | ΔSortino | ΔCAGR pp | ΔMaxDD pp |
+|---|---:|---:|---:|---:|---:|
+| cap10_incumbent | 0.000 | 0.000 | 0.000 | 0.000 | 0.000 |
+| cash_unbounded | 0.079 | 0.047 | -0.013 | 1.350 | 0.000 |
+| cap10_weekly_top10 | 0.000 | 0.000 | 0.000 | 0.000 | 0.000 |
+| cap15_incumbent | 0.000 | 0.000 | 0.000 | 0.000 | 0.000 |
+
+### Warm Book — 0.10% per fill
+
+| Arm | ΔEV net R | 90% context | ΔCalmar | 90% context |
+|---|---:|---:|---:|---:|
+| cap10_incumbent | 0.000 | [0.000, 0.000] | 0.000 | [0.000, 0.000] |
+| cash_unbounded | 0.034 | [-0.014, 0.100] | 0.050 | [-0.160, 0.250] |
+| cap10_weekly_top10 | 0.000 | [-0.158, 0.065] | 0.000 | [-0.200, 0.330] |
+| cap15_incumbent | 0.000 | [-0.006, 0.000] | 0.000 | [0.000, 0.180] |
+
+| Arm | ΔPF | ΔGain-to-Pain | ΔSortino | ΔCAGR pp | ΔMaxDD pp |
+|---|---:|---:|---:|---:|---:|
+| cap10_incumbent | 0.000 | 0.000 | 0.000 | 0.000 | 0.000 |
+| cash_unbounded | 0.062 | 0.085 | -0.004 | 2.200 | 0.400 |
+| cap10_weekly_top10 | 0.000 | 0.012 | 0.018 | 0.300 | 0.000 |
+| cap15_incumbent | 0.000 | 0.000 | 0.000 | 0.000 | 0.000 |
+
+### Empty Book — 0.20% per fill
+
+| Arm | ΔEV net R | 90% context | ΔCalmar | 90% context |
+|---|---:|---:|---:|---:|
+| cap10_incumbent | 0.000 | [0.000, 0.000] | 0.000 | [0.000, 0.000] |
+| cash_unbounded | 0.041 | [-0.010, 0.052] | 0.030 | [-0.015, 0.100] |
+| cap10_weekly_top10 | 0.000 | [-0.090, 0.000] | 0.000 | [-0.260, 0.000] |
+| cap15_incumbent | 0.000 | [0.000, 0.010] | 0.000 | [0.000, 0.110] |
+
+| Arm | ΔPF | ΔGain-to-Pain | ΔSortino | ΔCAGR pp | ΔMaxDD pp |
+|---|---:|---:|---:|---:|---:|
+| cap10_incumbent | 0.000 | 0.000 | 0.000 | 0.000 | 0.000 |
+| cash_unbounded | 0.066 | 0.035 | -0.014 | 0.900 | 0.000 |
+| cap10_weekly_top10 | 0.000 | 0.000 | 0.000 | 0.000 | 0.000 |
+| cap15_incumbent | 0.000 | 0.000 | 0.000 | 0.000 | 0.000 |
+
+### Warm Book — 0.20% per fill
+
+| Arm | ΔEV net R | 90% context | ΔCalmar | 90% context |
+|---|---:|---:|---:|---:|
+| cap10_incumbent | 0.000 | [0.000, 0.000] | 0.000 | [0.000, 0.000] |
+| cash_unbounded | 0.034 | [-0.022, 0.102] | 0.040 | [-0.130, 0.230] |
+| cap10_weekly_top10 | 0.000 | [-0.158, 0.065] | 0.000 | [-0.190, 0.310] |
+| cap15_incumbent | 0.000 | [-0.006, 0.000] | 0.000 | [0.000, 0.170] |
+
+| Arm | ΔPF | ΔGain-to-Pain | ΔSortino | ΔCAGR pp | ΔMaxDD pp |
+|---|---:|---:|---:|---:|---:|
+| cap10_incumbent | 0.000 | 0.000 | 0.000 | 0.000 | 0.000 |
+| cash_unbounded | 0.060 | 0.083 | -0.003 | 2.100 | 0.300 |
+| cap10_weekly_top10 | 0.000 | 0.017 | 0.020 | 0.300 | 0.000 |
+| cap15_incumbent | 0.000 | 0.000 | 0.000 | 0.000 | 0.000 |
+
+## Warm-seed initialization dispersion
+
+| Arm | Cost/fill | Median EV IQR ratio | Median Calmar IQR ratio |
+|---|---:|---:|---:|
+| cap10_incumbent | 0.10% | 1.000 | 1.000 |
+| cash_unbounded | 0.10% | 1.000 | 1.000 |
+| cap10_weekly_top10 | 0.10% | 1.000 | 1.000 |
+| cap15_incumbent | 0.10% | 1.000 | 1.000 |
+| cap10_incumbent | 0.20% | 1.000 | 1.000 |
+| cash_unbounded | 0.20% | 1.000 | 1.000 |
+| cap10_weekly_top10 | 0.20% | 1.000 | 1.000 |
+| cap15_incumbent | 0.20% | 1.000 | 1.000 |
+
+## Capacity and operations — 0.10% per fill
+
+| Arm | Median trades | Median blocked | Median positions | Peak | Turnover | Min-risk rejects |
+|---|---:|---:|---:|---:|---:|---:|
+| cap10_incumbent | 76.0 | 21.6% | 4.98 | 10 | 26.36 | 0 |
+| cash_unbounded | 74.0 | 0.0% | 4.82 | 12 | 26.76 | 85517 |
+| cap10_weekly_top10 | 88.0 | 18.1% | 5.13 | 10 | 28.44 | 0 |
+| cap15_incumbent | 79.0 | 0.0% | 5.15 | 12 | 27.32 | 0 |
+
+## Weekly-ranking opportunity set
+
+- Median fresh entrant pool: 0.0.
+- Median zero-entrant fraction: 0.558.
+- Replacements across reported paths: 2170.
+- Same-symbol re-entries within 10 sessions: 529.
+
+Bootstrap intervals above resample seven annual summaries and are descriptive context only. They are not gates or independent-population confidence claims.

scripts/portfolio_capacity_research.py

@@ -0,0 +1,764 @@
+'''Pure helpers for the focused daily portfolio-capacity research matrix.'''
+
+from __future__ import annotations
+
+import hashlib
+import math
+import random
+import statistics
+from collections import defaultdict
+from datetime import date, timedelta
+from typing import Any, Iterable
+
+
+ARMS: tuple[dict[str, Any], ...] = (
+    {
+        'id': 'cap10_incumbent',
+        'label': 'Cap 10, arrival-order incumbents',
+        'max_positions': 10,
+        'min_initial_risk_fraction': None,
+        'weekly_top_n_rebalance': False,
+    },
+    {
+        'id': 'cash_unbounded',
+        'label': 'Cash-constrained, no count cap',
+        'max_positions': None,
+        'min_initial_risk_fraction': 0.005,
+        'weekly_top_n_rebalance': False,
+    },
+    {
+        'id': 'cap10_weekly_top10',
+        'label': 'Cap 10, weekly current-rank top 10',
+        'max_positions': 10,
+        'min_initial_risk_fraction': None,
+        'weekly_top_n_rebalance': True,
+    },
+    {
+        'id': 'cap15_incumbent',
+        'label': 'Cap 15, arrival-order incumbents',
+        'max_positions': 15,
+        'min_initial_risk_fraction': None,
+        'weekly_top_n_rebalance': False,
+    },
+)
+
+ARM_BY_ID = {arm['id']: arm for arm in ARMS}
+RISK_FLOOR_ARMS: tuple[dict[str, Any], ...] = (
+    ARMS[0],
+    {
+        'id': 'cap10_min_risk_005',
+        'label': 'Cap 10, 0.5% minimum effective initial risk',
+        'max_positions': 10,
+        'min_initial_risk_fraction': 0.005,
+        'weekly_top_n_rebalance': False,
+    },
+)
+COSTS_PER_SIDE_PCT = (0.1, 0.2)
+ANCHOR_YEARS = tuple(range(2019, 2026))
+SCORING_SESSIONS = 504
+MEASUREMENT_SESSIONS = 252
+RESIDUAL_BENCHMARK_SESSIONS = 252
+WARM_SEED_MIN_OFFSET = 63
+WARM_SEED_MAX_OFFSET = 126
+BOOTSTRAP_REPLICATES = 10_000
+BOOTSTRAP_SEED = 20260805
+PRIMARY_METRICS = (
+    'ev_net_r',
+    'calmar',
+    'profit_factor',
+    'gain_to_pain',
+    'sortino',
+)
+PAIRED_METRICS = (
+    *PRIMARY_METRICS,
+    'cagr_pct',
+    'max_drawdown_pct',
+    'total_return_pct',
+    'sharpe',
+)
+
+
+def _end_exclusive(
+    sessions: list[date], start_index: int, count: int
+) -> date:
+    end_index = start_index + count
+    if end_index < len(sessions):
+        return sessions[end_index]
+    return sessions[-1] + timedelta(days=1)
+
+
+def build_cohort_manifest(session_dates: Iterable[date]) -> dict[str, Any]:
+    sessions = sorted(set(session_dates))
+    minimum = RESIDUAL_BENCHMARK_SESSIONS + SCORING_SESSIONS
+    if len(sessions) <= minimum + MEASUREMENT_SESSIONS:
+        raise ValueError('Snapshot is too short for the frozen cohort design')
+
+    index_of = {session: index for index, session in enumerate(sessions)}
+    first_eligible_index = RESIDUAL_BENCHMARK_SESSIONS - 1 + SCORING_SESSIONS
+    last_eligible_index = len(sessions) - MEASUREMENT_SESSIONS
+
+    first_by_month: dict[tuple[int, int], date] = {}
+    for session in sessions:
+        first_by_month.setdefault((session.year, session.month), session)
+
+    empty: list[dict[str, Any]] = []
+    for (year, month), session in sorted(first_by_month.items()):
+        index = index_of[session]
+        if year not in ANCHOR_YEARS:
+            continue
+        if index < first_eligible_index or index > last_eligible_index:
+            continue
+        empty.append({
+            'protocol': 'empty_book',
+            'path_id': f'empty-{year:04d}-{month:02d}',
+            'cluster': year,
+            'simulation_start': session.isoformat(),
+            'measurement_start': session.isoformat(),
+            'hard_end_exclusive': _end_exclusive(
+                sessions, index, MEASUREMENT_SESSIONS
+            ).isoformat(),
+        })
+
+    first_by_year: dict[int, date] = {}
+    for session in sessions:
+        first_by_year.setdefault(session.year, session)
+
+    warm: list[dict[str, Any]] = []
+    warm_seed_counts: dict[str, int] = {}
+    for year in ANCHOR_YEARS:
+        anchor = first_by_year.get(year)
+        if anchor is None:
+            continue
+        anchor_index = index_of[anchor]
+        if (
+            anchor_index < WARM_SEED_MAX_OFFSET
+            or anchor_index > last_eligible_index
+        ):
+            continue
+        seed_window = sessions[
+            anchor_index - WARM_SEED_MAX_OFFSET:
+            anchor_index - WARM_SEED_MIN_OFFSET + 1
+        ]
+        first_by_iso_week: dict[tuple[int, int], date] = {}
+        for session in seed_window:
+            iso = session.isocalendar()
+            first_by_iso_week.setdefault((iso.year, iso.week), session)
+        seeds = sorted(first_by_iso_week.values())
+        warm_seed_counts[str(year)] = len(seeds)
+        for seed_index, seed in enumerate(seeds, 1):
+            warm.append({
+                'protocol': 'warm_book',
+                'path_id': f'warm-{year}-seed-{seed_index:02d}',
+                'cluster': year,
+                'simulation_start': seed.isoformat(),
+                'measurement_start': anchor.isoformat(),
+                'hard_end_exclusive': _end_exclusive(
+                    sessions, anchor_index, MEASUREMENT_SESSIONS
+                ).isoformat(),
+                'seed_offset_sessions': anchor_index - index_of[seed],
+            })
+
+    return {
+        'snapshot_first_session': sessions[0].isoformat(),
+        'snapshot_last_session': sessions[-1].isoformat(),
+        'session_count': len(sessions),
+        'expected_clusters': list(ANCHOR_YEARS),
+        'empty_book': empty,
+        'warm_book': warm,
+        'empty_cluster_counts': dict(
+            sorted(
+                (
+                    str(year),
+                    sum(1 for row in empty if row['cluster'] == year),
+                )
+                for year in {row['cluster'] for row in empty}
+            )
+        ),
+        'warm_seed_counts': warm_seed_counts,
+        'empty_cluster_count': len({row['cluster'] for row in empty}),
+        'warm_cluster_count': len({row['cluster'] for row in warm}),
+    }
+
+
+def validate_cohort_manifest(manifest: dict[str, Any]) -> list[str]:
+    errors: list[str] = []
+    expected = set(ANCHOR_YEARS)
+    empty_clusters = {row['cluster'] for row in manifest['empty_book']}
+    warm_clusters = {row['cluster'] for row in manifest['warm_book']}
+    if empty_clusters != expected:
+        errors.append(
+            f'empty-book clusters {sorted(empty_clusters)} != {sorted(expected)}'
+        )
+    if warm_clusters != expected:
+        errors.append(
+            f'warm-book clusters {sorted(warm_clusters)} != {sorted(expected)}'
+        )
+    for year in ANCHOR_YEARS:
+        seed_count = int(manifest['warm_seed_counts'].get(str(year), 0))
+        if seed_count < 12:
+            errors.append(f'warm anchor {year} has only {seed_count} seeds')
+    return errors
+
+
+def build_cells(
+    manifest: dict[str, Any],
+    *,
+    arms: tuple[dict[str, Any], ...] = ARMS,
+    protocols: tuple[str, ...] = ('empty_book', 'warm_book'),
+    costs: tuple[float, ...] = COSTS_PER_SIDE_PCT,
+) -> list[dict[str, Any]]:
+    paths = [
+        path
+        for protocol in protocols
+        for path in manifest[protocol]
+    ]
+    cells: list[dict[str, Any]] = []
+    for cost in costs:
+        for path in paths:
+            for arm in arms:
+                cell_id = (
+                    f'{arm["id"]}|{path["protocol"]}|{path["path_id"]}'
+                    f'|cost={cost:.1f}'
+                )
+                cells.append({
+                    **path,
+                    'cell_id': cell_id,
+                    'arm_id': arm['id'],
+                    'cost_per_side_pct': cost,
+                })
+    return cells
+
+
+def percentile(values: Iterable[float], probability: float) -> float | None:
+    ordered = sorted(float(value) for value in values if value is not None)
+    if not ordered:
+        return None
+    if len(ordered) == 1:
+        return ordered[0]
+    location = (len(ordered) - 1) * probability
+    lower = math.floor(location)
+    upper = math.ceil(location)
+    if lower == upper:
+        return ordered[lower]
+    weight = location - lower
+    return ordered[lower] * (1.0 - weight) + ordered[upper] * weight
+
+
+def iqr(values: Iterable[float]) -> float | None:
+    clean: list[float] = []
+    for value in values:
+        if value is None:
+            continue
+        parsed = float(value)
+        if math.isfinite(parsed):
+            clean.append(parsed)
+    q25 = percentile(clean, 0.25)
+    q75 = percentile(clean, 0.75)
+    if q25 is None or q75 is None:
+        return None
+    return q75 - q25
+
+
+def median(values: Iterable[float | None]) -> float | None:
+    clean = [float(value) for value in values if value is not None]
+    return statistics.median(clean) if clean else None
+
+
+def _safe_ratio(numerator: float | None, denominator: float | None) -> float | None:
+    if numerator is None or denominator is None:
+        return None
+    if abs(denominator) <= 1e-12:
+        return 1.0 if abs(numerator) <= 1e-12 else None
+    return numerator / denominator
+
+
+def _stable_seed(*parts: object) -> int:
+    digest = hashlib.sha256('|'.join(map(str, parts)).encode('utf-8')).digest()
+    return BOOTSTRAP_SEED + int.from_bytes(digest[:4], 'big')
+
+
+def bootstrap_median_interval(
+    values: Iterable[float | None],
+    *,
+    seed_parts: tuple[object, ...],
+    replicates: int = BOOTSTRAP_REPLICATES,
+) -> dict[str, float | int | None]:
+    clean = [float(value) for value in values if value is not None]
+    if not clean:
+        return {'n': 0, 'point': None, 'p05': None, 'p95': None}
+    rng = random.Random(_stable_seed(*seed_parts))
+    draws = [
+        statistics.median(rng.choices(clean, k=len(clean)))
+        for _ in range(replicates)
+    ]
+    return {
+        'n': len(clean),
+        'replicates': replicates,
+        'point': statistics.median(clean),
+        'p05': percentile(draws, 0.05),
+        'p95': percentile(draws, 0.95),
+    }
+
+
+def _monthly_returns(
+    equity_curve: list[dict[str, Any]], base_equity: float
+) -> list[float]:
+    month_ends: dict[tuple[int, int], float] = {}
+    for point in equity_curve:
+        point_date = date.fromisoformat(str(point['date']))
+        month_ends[(point_date.year, point_date.month)] = float(point['equity'])
+    previous = float(base_equity)
+    returns: list[float] = []
+    for month in sorted(month_ends):
+        equity = month_ends[month]
+        if previous > 0:
+            returns.append(equity / previous - 1.0)
+        previous = equity
+    return returns
+
+
+def _time_underwater(equities: list[float]) -> tuple[int, float]:
+    peak = float('-inf')
+    current = 0
+    longest = 0
+    underwater = 0
+    for equity in equities:
+        peak = max(peak, equity)
+        if peak > 0 and equity < peak - 1e-9:
+            current += 1
+            underwater += 1
+            longest = max(longest, current)
+        else:
+            current = 0
+    percentage = underwater / len(equities) * 100.0 if equities else 0.0
+    return longest, percentage
+
+
+def summarize_simulation(sim: dict[str, Any]) -> dict[str, Any]:
+    trades = list(sim.get('trade_details') or [])
+    equity_curve = list(sim.get('equity_curve') or [])
+    net_rs = [float(trade['net_r']) for trade in trades]
+    positive_rs = [value for value in net_rs if value > 0]
+    negative_rs = [value for value in net_rs if value < 0]
+    ev_net_r = statistics.fmean(net_rs) if net_rs else None
+    profit_factor = (
+        sum(positive_rs) / abs(sum(negative_rs))
+        if negative_rs
+        else None
+    )
+
+    base_equity = float(
+        sim.get('measurement_start_equity') or sim.get('starting_capital') or 0.0
+    )
+    curve_equities = [float(point['equity']) for point in equity_curve]
+    daily_equities = [base_equity, *curve_equities]
+    daily_returns = [
+        current / previous - 1.0
+        for previous, current in zip(daily_equities, daily_equities[1:])
+        if previous > 0
+    ]
+    downside_deviation = (
+        math.sqrt(
+            statistics.fmean(min(value, 0.0) ** 2 for value in daily_returns)
+        )
+        if daily_returns
+        else None
+    )
+    sortino = (
+        statistics.fmean(daily_returns) / downside_deviation * math.sqrt(252.0)
+        if downside_deviation is not None and downside_deviation > 0
+        else None
+    )
+    monthly_returns = _monthly_returns(equity_curve, base_equity)
+    negative_monthly = sum(value for value in monthly_returns if value < 0)
+    gain_to_pain = (
+        sum(monthly_returns) / abs(negative_monthly)
+        if negative_monthly < 0
+        else None
+    )
+    longest_underwater, underwater_pct = _time_underwater(daily_equities)
+
+    transaction_cost = sum(
+        float(trade.get('transaction_cost') or 0.0) for trade in trades
+    )
+    traded_notional = sum(
+        float(trade.get('shares') or 0.0)
+        * (float(trade.get('entry') or 0.0) + float(trade.get('fill') or 0.0))
+        for trade in trades
+    )
+    turnover_multiple = (
+        traded_notional / base_equity if base_equity > 0 else None
+    )
+
+    ordered_rs = sorted(net_rs, reverse=True)
+    ev_without_best: dict[str, float | None] = {}
+    for count in (1, 5, 10):
+        remaining = ordered_rs[count:]
+        ev_without_best[str(count)] = (
+            statistics.fmean(remaining) if remaining else None
+        )
+
+    events = list(sim.get('weekly_rebalance_events') or [])
+    entrant_sizes = [int(event['fresh_entrant_pool']) for event in events]
+    eligible_sizes = [
+        int(event['rank_eligible_entrant_pool']) for event in events
+    ]
+    replacements = [int(event['replacements']) for event in events]
+
+    capacity_skips = int(
+        sim.get('measurement_skipped_book_full', sim.get('skipped_book_full', 0))
+    )
+    opened = int(sim.get('opened_positions', sim.get('trades', 0)))
+    capacity_opportunities = opened + capacity_skips
+
+    result = {
+        'start_date': sim.get('start_date'),
+        'end_date': sim.get('end_date'),
+        'simulation_start_date': sim.get('simulation_start_date'),
+        'measurement_start_equity': base_equity,
+        'measurement_start_positions': sim.get('measurement_start_positions', 0),
+        'trades': len(trades),
+        'ev_net_r': ev_net_r,
+        'profit_factor': profit_factor,
+        'gain_to_pain': gain_to_pain,
+        'sortino': sortino,
+        'ev_without_best': ev_without_best,
+        'total_return_pct': sim.get('total_return_pct'),
+        'cagr_pct': sim.get('cagr_pct'),
+        'max_drawdown_pct': sim.get('max_drawdown_pct'),
+        'calmar': sim.get('calmar'),
+        'sharpe': sim.get('sharpe'),
+        'win_rate': sim.get('win_rate'),
+        'avg_hold_days': sim.get('avg_hold_days'),
+        'longest_underwater_sessions': longest_underwater,
+        'underwater_pct': underwater_pct,
+        'transaction_cost': transaction_cost,
+        'turnover_multiple': turnover_multiple,
+        'skipped_book_full': capacity_skips,
+        'opened_positions': opened,
+        'capacity_opportunities': capacity_opportunities,
+        'blocked_fraction': (
+            capacity_skips / capacity_opportunities
+            if capacity_opportunities
+            else 0.0
+        ),
+        'skipped_min_initial_risk': int(
+            sim.get('measurement_skipped_min_initial_risk', 0)
+        ),
+        'avg_positions': sim.get('avg_positions'),
+        'peak_positions': sim.get('peak_positions'),
+        'sessions_at_capacity': sim.get('sessions_at_capacity'),
+        'sessions_measured': sim.get('sessions_measured'),
+        'avg_cash_pct': sim.get('avg_cash_pct'),
+        'avg_gross_exposure_pct': sim.get('avg_gross_exposure_pct'),
+        'exit_reasons': sim.get('exit_reasons'),
+    }
+    if events:
+        result['weekly_rebalance'] = {
+            'events': len(events),
+            'zero_entrant_fraction': (
+                sum(1 for value in entrant_sizes if value == 0) / len(events)
+            ),
+            'entrant_pool_mean': statistics.fmean(entrant_sizes),
+            'entrant_pool_median': statistics.median(entrant_sizes),
+            'entrant_pool_p90': percentile(entrant_sizes, 0.9),
+            'eligible_pool_mean': statistics.fmean(eligible_sizes),
+            'replacements': sum(replacements),
+            'weekly_rank_rejected_entries': int(
+                sim.get('weekly_rank_rejected_entries', 0)
+            ),
+            'reentries_within_5_sessions': int(
+                sim.get('rebalance_reentries_within_5_sessions', 0)
+            ),
+            'reentries_within_10_sessions': int(
+                sim.get('rebalance_reentries_within_10_sessions', 0)
+            ),
+            'reentries_within_20_sessions': int(
+                sim.get('rebalance_reentries_within_20_sessions', 0)
+            ),
+        }
+    return result
+
+
+def _cluster_rows(
+    cells: list[dict[str, Any]],
+    *,
+    arm_id: str,
+    protocol: str,
+    cost: float,
+) -> list[dict[str, Any]]:
+    treatment = {
+        row['path_id']: row
+        for row in cells
+        if row['arm_id'] == arm_id
+        and row['protocol'] == protocol
+        and float(row['cost_per_side_pct']) == cost
+    }
+    control = {
+        row['path_id']: row
+        for row in cells
+        if row['arm_id'] == 'cap10_incumbent'
+        and row['protocol'] == protocol
+        and float(row['cost_per_side_pct']) == cost
+    }
+    shared_paths = sorted(set(treatment) & set(control))
+    by_cluster: dict[int, list[tuple[dict, dict]]] = defaultdict(list)
+    for path_id in shared_paths:
+        row = treatment[path_id]
+        by_cluster[int(row['cluster'])].append((row, control[path_id]))
+
+    summaries: list[dict[str, Any]] = []
+    for cluster, pairs in sorted(by_cluster.items()):
+        metrics: dict[str, Any] = {}
+        for metric in PAIRED_METRICS:
+            arm_values = [
+                pair[0]['metrics'].get(metric)
+                for pair in pairs
+                if pair[0]['metrics'].get(metric) is not None
+                and math.isfinite(float(pair[0]['metrics'][metric]))
+            ]
+            control_values = [
+                pair[1]['metrics'].get(metric)
+                for pair in pairs
+                if pair[1]['metrics'].get(metric) is not None
+                and math.isfinite(float(pair[1]['metrics'][metric]))
+            ]
+            deltas = [
+                float(arm['metrics'][metric])
+                - float(base['metrics'][metric])
+                for arm, base in pairs
+                if arm['metrics'].get(metric) is not None
+                and base['metrics'].get(metric) is not None
+                and math.isfinite(float(arm['metrics'][metric]))
+                and math.isfinite(float(base['metrics'][metric]))
+            ]
+            arm_median = median(arm_values)
+            control_median = median(control_values)
+            metrics[metric] = {
+                'arm_median': arm_median,
+                'control_median': control_median,
+                'paired_delta_median': median(deltas),
+                'arm_control_ratio': _safe_ratio(
+                    arm_median, control_median
+                ),
+                'paired_paths': len(deltas),
+            }
+        summaries.append({
+            'cluster': cluster,
+            'paths': len(pairs),
+            'metrics': metrics,
+        })
+    return summaries
+
+
+def aggregate_results(
+    cells: list[dict[str, Any]],
+    *,
+    arms: tuple[dict[str, Any], ...] = ARMS,
+    protocols: tuple[str, ...] = ('empty_book', 'warm_book'),
+    costs: tuple[float, ...] = COSTS_PER_SIDE_PCT,
+    include_warm_dispersion: bool = True,
+) -> dict[str, Any]:
+    paired: list[dict[str, Any]] = []
+    path_distributions: list[dict[str, Any]] = []
+    for cost in costs:
+        for protocol in protocols:
+            control_by_path = {
+                row['path_id']: row
+                for row in cells
+                if row['arm_id'] == 'cap10_incumbent'
+                and row['protocol'] == protocol
+                and float(row['cost_per_side_pct']) == float(cost)
+            }
+            for arm in arms:
+                arm_id = str(arm['id'])
+                clusters = _cluster_rows(
+                    cells,
+                    arm_id=arm_id,
+                    protocol=protocol,
+                    cost=float(cost),
+                )
+                headline: dict[str, Any] = {}
+                for metric in PAIRED_METRICS:
+                    deltas = [
+                        cluster['metrics'][metric]['paired_delta_median']
+                        for cluster in clusters
+                    ]
+                    arm_levels = [
+                        cluster['metrics'][metric]['arm_median']
+                        for cluster in clusters
+                    ]
+                    control_levels = [
+                        cluster['metrics'][metric]['control_median']
+                        for cluster in clusters
+                    ]
+                    arm_level = median(arm_levels)
+                    control_level = median(control_levels)
+                    metric_summary: dict[str, Any] = {
+                        'paired_delta_median': median(deltas),
+                        'arm_median': arm_level,
+                        'control_median': control_level,
+                        'arm_control_ratio': _safe_ratio(
+                            arm_level, control_level
+                        ),
+                    }
+                    if metric in ('ev_net_r', 'calmar'):
+                        metric_summary['bootstrap_90'] = (
+                            bootstrap_median_interval(
+                                deltas,
+                                seed_parts=(
+                                    arm_id,
+                                    protocol,
+                                    cost,
+                                    metric,
+                                    'paired-delta',
+                                ),
+                            )
+                        )
+                    headline[metric] = metric_summary
+                paired.append({
+                    'arm_id': arm_id,
+                    'protocol': protocol,
+                    'cost_per_side_pct': cost,
+                    'clusters': clusters,
+                    'headline': headline,
+                })
+                treatment_by_path = {
+                    row['path_id']: row
+                    for row in cells
+                    if row['arm_id'] == arm_id
+                    and row['protocol'] == protocol
+                    and float(row['cost_per_side_pct']) == float(cost)
+                }
+                shared_paths = sorted(
+                    set(treatment_by_path) & set(control_by_path)
+                )
+                path_metrics: dict[str, Any] = {}
+                for metric in PAIRED_METRICS:
+                    deltas = [
+                        float(treatment_by_path[path_id]['metrics'][metric])
+                        - float(control_by_path[path_id]['metrics'][metric])
+                        for path_id in shared_paths
+                        if treatment_by_path[path_id]['metrics'].get(metric)
+                        is not None
+                        and control_by_path[path_id]['metrics'].get(metric)
+                        is not None
+                        and math.isfinite(
+                            float(treatment_by_path[path_id]['metrics'][metric])
+                        )
+                        and math.isfinite(
+                            float(control_by_path[path_id]['metrics'][metric])
+                        )
+                    ]
+                    path_metrics[metric] = {
+                        'paired_paths': len(deltas),
+                        'paired_delta_mean': (
+                            statistics.fmean(deltas) if deltas else None
+                        ),
+                        'paired_delta_median': median(deltas),
+                        'paired_delta_p25': percentile(deltas, 0.25),
+                        'paired_delta_p75': percentile(deltas, 0.75),
+                        'positive_fraction': (
+                            sum(delta > 0.0 for delta in deltas) / len(deltas)
+                            if deltas
+                            else None
+                        ),
+                        'identical_fraction': (
+                            sum(abs(delta) <= 1e-12 for delta in deltas)
+                            / len(deltas)
+                            if deltas
+                            else None
+                        ),
+                    }
+                path_distributions.append({
+                    'arm_id': arm_id,
+                    'protocol': protocol,
+                    'cost_per_side_pct': cost,
+                    'metrics': path_metrics,
+                })
+
+    warm_rows = [
+        row for row in cells if row['protocol'] == 'warm_book'
+    ]
+    warm_dispersion: list[dict[str, Any]] = []
+    for cost in costs:
+        for arm in arms:
+            arm_id = str(arm['id'])
+            anchor_rows: list[dict[str, Any]] = []
+            for cluster in ANCHOR_YEARS:
+                arm_paths = [
+                    row
+                    for row in warm_rows
+                    if row['arm_id'] == arm_id
+                    and int(row['cluster']) == cluster
+                    and float(row['cost_per_side_pct']) == float(cost)
+                ]
+                control_by_path = {
+                    row['path_id']: row
+                    for row in warm_rows
+                    if row['arm_id'] == 'cap10_incumbent'
+                    and int(row['cluster']) == cluster
+                    and float(row['cost_per_side_pct']) == float(cost)
+                }
+                metric_rows: dict[str, Any] = {}
+                for metric in ('ev_net_r', 'calmar'):
+                    arm_spread = iqr(
+                        row['metrics'].get(metric) for row in arm_paths
+                    )
+                    control_spread = iqr(
+                        control_by_path[row['path_id']]['metrics'].get(metric)
+                        for row in arm_paths
+                        if row['path_id'] in control_by_path
+                    )
+                    metric_rows[metric] = {
+                        'arm_iqr': arm_spread,
+                        'control_iqr': control_spread,
+                        'iqr_ratio': _safe_ratio(
+                            arm_spread, control_spread
+                        ),
+                    }
+                anchor_rows.append({
+                    'cluster': cluster,
+                    'seeds': len(arm_paths),
+                    'metrics': metric_rows,
+                })
+
+            headline: dict[str, Any] = {}
+            for metric in ('ev_net_r', 'calmar'):
+                ratios = [
+                    row['metrics'][metric]['iqr_ratio']
+                    for row in anchor_rows
+                ]
+                headline[metric] = {
+                    'median_iqr_ratio': median(ratios),
+                    'bootstrap_90': bootstrap_median_interval(
+                        ratios,
+                        seed_parts=(
+                            arm_id,
+                            cost,
+                            metric,
+                            'warm-iqr-ratio',
+                        ),
+                    ),
+                }
+            warm_dispersion.append({
+                'arm_id': arm_id,
+                'cost_per_side_pct': cost,
+                'anchors': anchor_rows,
+                'headline': headline,
+            })
+
+    if not include_warm_dispersion:
+        warm_dispersion = []
+
+    return {
+        'paired_per_year': paired,
+        'paired_path_distributions': path_distributions,
+        'warm_seed_dispersion': warm_dispersion,
+        'bootstrap': {
+            'replicates': BOOTSTRAP_REPLICATES,
+            'seed': BOOTSTRAP_SEED,
+            'interval': 'central 90% percentile, context only',
+            'resampling_unit': 'seven annual paired summaries',
+        },
+    }

scripts/research_rankings.py

@@ -0,0 +1,84 @@
+'''Shared production-style historical ranking helpers for research runners.'''
+
+from __future__ import annotations
+
+from datetime import date
+
+
+def _period_percentiles(
+    observations: list[dict], value_key: str
+) -> dict[tuple[str, str], float]:
+    '''Rank one deterministic ticker observation per historical period.'''
+    by_period: dict[tuple, list[dict]] = {}
+    seen: set[tuple[str, str]] = set()
+    for row in observations:
+        identity = (str(row['symbol']), str(row['date']))
+        if identity in seen:
+            raise ValueError(f'Duplicate universe rank observation: {identity}')
+        seen.add(identity)
+        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]]:
+    '''Historical equivalent of production compute_activation_ranks.
+
+    Every ticker contributes at most once per session. Residual momentum starts
+    only once 252 benchmark closes were point-in-time available; earlier dates
+    use the same raw-momentum fallback as production.
+    '''
+    identities = [(str(row['symbol']), str(row['date'])) for row in observations]
+    if len(identities) != len(set(identities)):
+        raise ValueError('Universe ranking requires one observation per ticker/date')
+
+    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

scripts/run_daily_reentry_matrix.py

@@ -29,6 +29,11 @@ ROOT = Path(__file__).resolve().parents[1]
 if str(ROOT) not in sys.path:
     sys.path.insert(0, str(ROOT))
 
+from scripts.research_rankings import (  # noqa: E402
+    _live_universe_rank_map,
+    _period_percentiles,
+)
+
 POLICY_NAMES = (
     "immediate",
     "next_session",
@@ -107,85 +112,6 @@ def _default_output_path() -> Path:
     return Path("reports") / f"daily-reentry-matrix-{stamp}.json"
 
 
-def _period_percentiles(
-    observations: list[dict], value_key: str
-) -> dict[tuple[str, str], float]:
-    """Production-style percentiles, one deterministic symbol row per period."""
-    by_period: dict[tuple, list[dict]] = {}
-    seen: set[tuple[str, str]] = set()
-    for row in observations:
-        identity = (str(row["symbol"]), str(row["date"]))
-        if identity in seen:
-            raise ValueError(f"Duplicate universe rank observation: {identity}")
-        seen.add(identity)
-        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]]:
-    """Historical equivalent of ``compute_activation_ranks``.
-
-    Every ticker contributes at most once per session. Residual momentum starts
-    only once 252 benchmark closes were point-in-time available; earlier dates
-    use the same raw-momentum fallback as production.
-    """
-    identities = [(str(row["symbol"]), str(row["date"])) for row in observations]
-    if len(identities) != len(set(identities)):
-        raise ValueError("Universe ranking requires one observation per ticker/date")
-
-    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
-
-
 class PrecomputedDailyEngine:
     """Exact date/symbol lookup over the already-ranked production gate."""
 

scripts/run_execution_recovery_matrix.py

@@ -55,6 +55,11 @@ ROOT = Path(__file__).resolve().parents[1]
 if str(ROOT) not in sys.path:
     sys.path.insert(0, str(ROOT))
 
+from scripts.research_rankings import (  # noqa: E402
+    _live_universe_rank_map,
+    _period_percentiles,
+)
+
 # Must match Phase A cache when reusing research-cands.pkl
 CACHE_VERSION = "research-matrix-v1-daily-prod"
 
@@ -104,66 +109,6 @@ def _parse_args() -> argparse.Namespace:
     return p.parse_args()
 
 
-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 _window(arm: dict, name: str) -> dict | None:
     for row in arm.get("windows") or []:
         if row.get("window") == name:

scripts/run_research_matrix.py

@@ -68,6 +68,11 @@ ROOT = Path(__file__).resolve().parents[1]
 if str(ROOT) not in sys.path:
     sys.path.insert(0, str(ROOT))
 
+from scripts.research_rankings import (  # noqa: E402
+    _live_universe_rank_map,
+    _period_percentiles,
+)
+
 CACHE_VERSION = "research-matrix-v1-daily-prod"
 
 # Pre-registered arm catalogue (order is report order). Control is a0.
@@ -210,66 +215,6 @@ 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__,

tests/unit/test_portfolio_capacity_research.py

@@ -0,0 +1,905 @@
+from __future__ import annotations
+
+import asyncio
+import pickle
+import sqlite3
+from datetime import date, timedelta
+
+import pytest
+
+from app.services import backtest_service as bt
+from scripts.portfolio_capacity_research import (
+    ANCHOR_YEARS,
+    RISK_FLOOR_ARMS,
+    aggregate_results,
+    bootstrap_median_interval,
+    build_cells,
+    build_cohort_manifest,
+    iqr,
+    summarize_simulation,
+    validate_cohort_manifest,
+)
+from scripts.run_portfolio_construction_matrix import (
+    CACHE_VERSION,
+    STUDIES,
+    _assert_clean_worktree,
+    _build_candidate_cache,
+    _checkpoint_state,
+    _construction_candidate_view,
+    _construction_universe_errors,
+    _json_hash,
+    _load_snapshot,
+    _markdown,
+    _operational_summary,
+    _risk_floor_markdown,
+    _worker_init,
+    _worker_run_cell,
+    _write_cell_checkpoint,
+)
+
+
+def _prices(ords: list[int], close: float = 100.0) -> tuple:
+    closes = [close] * len(ords)
+    return (
+        ords,
+        list(closes),
+        [value + 1.0 for value in closes],
+        [value - 1.0 for value in closes],
+        list(closes),
+        [1_000_000] * len(ords),
+    )
+
+
+def _candidate(
+    symbol: str,
+    day: date,
+    *,
+    entry: float = 100.0,
+    stop: float = 80.0,
+    rank: float = 90.0,
+) -> dict:
+    return {
+        'qualified': True,
+        'direction': 'long',
+        'symbol': symbol,
+        'date': day.isoformat(),
+        'entry': entry,
+        'stop': stop,
+        'target': entry + 100.0,
+        'momentum_percentile': rank,
+        'activation_momentum_percentile': rank,
+        'residual_high_vol_blend_80_20': rank,
+    }
+
+
+def _business_days(start: date, end: date) -> list[date]:
+    days: list[date] = []
+    current = start
+    while current <= end:
+        if current.weekday() < 5:
+            days.append(current)
+        current += timedelta(days=1)
+    return days
+
+
+def test_new_simulator_option_defaults_match_explicit_defaults():
+    start = date(2025, 1, 6)
+    ords = [start.toordinal() + offset for offset in range(8)]
+    prices = {'AAA': _prices(ords)}
+    candidates = [_candidate('AAA', start)]
+
+    legacy = bt._simulate_portfolio(
+        candidates,
+        prices,
+        None,
+        'hold',
+        3,
+        include_trades=True,
+    )
+    explicit = bt._simulate_portfolio(
+        candidates,
+        prices,
+        None,
+        'hold',
+        3,
+        max_positions=10,
+        min_initial_risk_fraction=None,
+        weekly_top_n_rebalance=False,
+        measurement_start_date=None,
+        hard_end_date=None,
+        include_capacity_diagnostics=False,
+        include_trades=True,
+    )
+
+    assert legacy == explicit
+
+
+def test_load_snapshot_accepts_pre_sec_ticker_schema(tmp_path, monkeypatch):
+    snapshot = tmp_path / 'legacy-research.sqlite'
+    with sqlite3.connect(snapshot) as connection:
+        connection.executescript(
+            '''
+            CREATE TABLE tickers (
+                id INTEGER PRIMARY KEY,
+                symbol VARCHAR(10) NOT NULL UNIQUE,
+                name VARCHAR(120),
+                created_at DATETIME NOT NULL
+            );
+            CREATE TABLE ohlcv_records (
+                id INTEGER PRIMARY KEY,
+                ticker_id INTEGER NOT NULL,
+                date DATE NOT NULL,
+                open FLOAT NOT NULL,
+                high FLOAT NOT NULL,
+                low FLOAT NOT NULL,
+                close FLOAT NOT NULL,
+                volume BIGINT NOT NULL,
+                created_at DATETIME NOT NULL
+            );
+            CREATE TABLE research_rank_only (
+                symbol VARCHAR(10) PRIMARY KEY
+            );
+            INSERT INTO tickers VALUES
+                (1, 'LEGACY', 'Legacy Co', '2024-01-01 00:00:00'),
+                (2, 'RANK', 'Rank Only Co', '2024-01-01 00:00:00');
+            INSERT INTO ohlcv_records VALUES
+                (1, 1, '2024-01-02', 100, 102, 99, 101, 1000000,
+                 '2024-01-02 00:00:00'),
+                (2, 2, '2024-01-02', 50, 51, 49, 50, 500000,
+                 '2024-01-02 00:00:00');
+            INSERT INTO research_rank_only VALUES ('RANK');
+            '''
+        )
+
+    async def recommendation_config(_db):
+        return {}
+
+    async def activation_config(_db):
+        return {'min_momentum_percentile': 80.0}
+
+    async def exit_policy(_db):
+        return {'mode': 'atr_trailing', 'hold_days': 30, 'atr_multiplier': 3.0}
+
+    async def benchmark_closes(_db, *, days, refresh):
+        assert days is None
+        assert refresh is False
+        return {date(2024, 1, 2): 100.0}
+
+    monkeypatch.setattr(
+        'app.services.recommendation_service.get_recommendation_config',
+        recommendation_config,
+    )
+    monkeypatch.setattr(
+        'app.services.admin_service.get_activation_config',
+        activation_config,
+    )
+    monkeypatch.setattr(
+        'app.services.paper_trade_service.get_exit_policy',
+        exit_policy,
+    )
+    monkeypatch.setattr(
+        'app.services.backtest_service._load_benchmark_closes_for_backtest',
+        benchmark_closes,
+    )
+
+    loaded = asyncio.run(_load_snapshot(snapshot, quiet=True))
+
+    assert loaded['symbols'] == ['LEGACY', 'RANK']
+    assert loaded['construction_symbols'] == {'LEGACY'}
+    assert loaded['prices']['LEGACY'] == (
+        [date(2024, 1, 2).toordinal()],
+        [100.0],
+        [102.0],
+        [99.0],
+        [101.0],
+        [1_000_000],
+    )
+    assert loaded['prices']['RANK'][4] == [50.0]
+    assert loaded['construction_universe_manifest'][
+        'construction_ticker_rows'
+    ] == 1
+    assert loaded['construction_universe_manifest']['rank_only_ticker_rows'] == 1
+    with sqlite3.connect(snapshot) as connection:
+        columns = {
+            row[1] for row in connection.execute('PRAGMA table_info(tickers)')
+        }
+    assert {'cik', 'sic', 'sic_description'}.isdisjoint(columns)
+
+
+def test_construction_view_filters_rank_only_rows_without_rebuilding_cache():
+    manifest = {
+        'ranking_ticker_rows': 506,
+        'ranking_symbols_with_prices': 506,
+        'construction_ticker_rows': 505,
+        'construction_symbols_with_prices': 505,
+        'rank_only_ticker_rows': 1,
+        'rank_only_symbols_with_prices': 1,
+        'rank_only_unknown_symbols': 0,
+    }
+    cached = {
+        'key': {'version': 'existing-broad-cache'},
+        'qualified_candidates': [
+            {'symbol': 'PROD', 'date': '2025-01-02'},
+            {'symbol': 'RANK', 'date': '2025-01-02'},
+        ],
+        'qualified_long_count': 2,
+        'daily_rank_map': {
+            ('RANK', '2025-01-02'): {'strategy_rank': 99.0},
+        },
+    }
+
+    view = _construction_candidate_view(
+        cached,
+        {
+            'construction_symbols': {'PROD'},
+            'construction_universe_manifest': manifest,
+        },
+    )
+
+    assert [row['symbol'] for row in view['qualified_candidates']] == ['PROD']
+    assert view['raw_full_universe_qualified_long_count'] == 2
+    assert view['filtered_rank_only_qualified_long_count'] == 1
+    assert view['qualified_long_count'] == 1
+    assert ('RANK', '2025-01-02') in view['daily_rank_map']
+    assert len(cached['qualified_candidates']) == 2
+
+
+def test_existing_broad_candidate_cache_key_remains_reusable(tmp_path, monkeypatch):
+    snapshot = tmp_path / 'research.sqlite'
+    snapshot.write_bytes(b'snapshot-placeholder')
+    cache_path = tmp_path / 'broad-cache.pkl'
+    snapshot_data = {
+        'recommendation_config': {'rr': 3.0},
+        'activation': {'min_momentum_percentile': 80.0},
+        'runtime_config': {'ranking_key': 'test'},
+        'universe_manifest': {
+            'ticker_rows': 4655,
+            'symbols_with_prices': 4654,
+            'symbols_sha256': 'symbols',
+        },
+    }
+    key = {
+        'version': CACHE_VERSION,
+        'snapshot': str(snapshot.resolve()),
+        'snapshot_sha256': 'snapshot-hash',
+        'cadence': 'daily',
+        'outcome_horizon_sessions': 0,
+        'recommendation_config_hash': _json_hash(
+            snapshot_data['recommendation_config']
+        ),
+        'activation_hash': _json_hash(snapshot_data['activation']),
+        'runtime_config': snapshot_data['runtime_config'],
+        'universe_manifest': snapshot_data['universe_manifest'],
+    }
+    cached = {'key': key, 'qualified_candidates': [{'symbol': 'PROD'}]}
+    cache_path.write_bytes(pickle.dumps(cached))
+    monkeypatch.setattr(
+        bt,
+        '_replay_candidates_for_period',
+        lambda *_args: pytest.fail('existing cache should avoid replay'),
+    )
+
+    loaded = _build_candidate_cache(
+        snapshot_data,
+        snapshot=snapshot,
+        snapshot_sha256='snapshot-hash',
+        cache_path=cache_path,
+        workers=1,
+        quiet=True,
+    )
+
+    assert loaded == cached
+
+
+def test_construction_universe_guard_rejects_leaked_broad_book():
+    valid = {
+        'ranking_ticker_rows': 4655,
+        'construction_ticker_rows': 506,
+        'construction_symbols_with_prices': 506,
+        'rank_only_ticker_rows': 4149,
+        'rank_only_unknown_symbols': 0,
+    }
+    assert _construction_universe_errors(valid) == []
+
+    leaked = {
+        **valid,
+        'construction_ticker_rows': 4655,
+        'construction_symbols_with_prices': 4654,
+        'rank_only_ticker_rows': 0,
+    }
+    errors = _construction_universe_errors(leaked)
+    assert any('450-600' in error for error in errors)
+
+
+def test_unbounded_count_and_effective_risk_floor():
+    start = date(2025, 1, 6)
+    ords = [start.toordinal() + offset for offset in range(4)]
+    symbols = [f'S{index}' for index in range(25)]
+    prices = {symbol: _prices(ords) for symbol in symbols}
+    candidates = [
+        _candidate(symbol, start, stop=80.0, rank=100.0 - index)
+        for index, symbol in enumerate(symbols)
+    ]
+
+    capped = bt._simulate_portfolio(
+        candidates,
+        prices,
+        None,
+        'hold',
+        30,
+        max_positions=1,
+        hard_end_date=start + timedelta(days=4),
+        measurement_start_date=start,
+        include_capacity_diagnostics=True,
+    )
+    unbounded = bt._simulate_portfolio(
+        candidates,
+        prices,
+        None,
+        'hold',
+        30,
+        max_positions=None,
+        min_initial_risk_fraction=0.005,
+        hard_end_date=start + timedelta(days=4),
+        measurement_start_date=start,
+        include_capacity_diagnostics=True,
+    )
+
+    assert capped is not None and unbounded is not None
+    assert capped['peak_positions'] == 1
+    assert capped['measurement_skipped_book_full'] == 24
+    assert unbounded['peak_positions'] > 1
+    assert unbounded['measurement_skipped_book_full'] == 0
+    assert unbounded['skipped_min_initial_risk'] > 0
+    assert unbounded['peak_positions'] == unbounded['trades']
+
+
+def test_measurement_window_carries_state_but_excludes_pre_anchor_trade_ev():
+    start = date(2025, 1, 6)
+    anchor = start + timedelta(days=2)
+    hard_end = start + timedelta(days=7)
+    ords = [
+        start.toordinal() + offset
+        for offset in range((hard_end - start).days)
+    ]
+    prices = {
+        'AAA': _prices(ords, 100.0),
+        'BBB': _prices(ords, 100.0),
+    }
+    candidates = [
+        _candidate('AAA', start),
+        _candidate('BBB', anchor + timedelta(days=1)),
+    ]
+
+    sim = bt._simulate_portfolio(
+        candidates,
+        prices,
+        None,
+        'hold',
+        30,
+        start_date=start,
+        end_date=hard_end,
+        measurement_start_date=anchor,
+        hard_end_date=hard_end,
+        include_curve=True,
+        include_trades=True,
+    )
+
+    assert sim is not None
+    assert sim['simulation_start_date'] == start.isoformat()
+    assert sim['start_date'] == anchor.isoformat()
+    assert sim['measurement_start_positions'] == 1
+    assert sim['trades'] == 1
+    assert [trade['symbol'] for trade in sim['trade_details']] == ['BBB']
+    assert sim['equity_curve'][0]['date'] == anchor.isoformat()
+
+
+def test_weekly_top10_uses_current_rank_for_both_sides_not_entry_rank():
+    monday = date(2025, 1, 6)
+    friday = date(2025, 1, 10)
+    sessions = _business_days(monday, friday)
+    ords = [session.toordinal() for session in sessions]
+    prices = {
+        'AAA': _prices(ords),
+        'BBB': _prices(ords),
+    }
+    candidates = [
+        _candidate('AAA', monday, rank=99.0),
+        _candidate('BBB', friday, rank=10.0),
+    ]
+    rank_map = {
+        ('AAA', friday.isoformat()): {'strategy_rank': 10.0},
+        ('BBB', friday.isoformat()): {'strategy_rank': 90.0},
+    }
+
+    sim = bt._simulate_portfolio(
+        candidates,
+        prices,
+        None,
+        'hold',
+        30,
+        max_positions=1,
+        weekly_top_n_rebalance=True,
+        daily_rank_map=rank_map,
+        measurement_start_date=monday,
+        hard_end_date=friday + timedelta(days=1),
+        include_trades=True,
+        include_capacity_diagnostics=True,
+    )
+
+    assert sim is not None
+    assert [trade['symbol'] for trade in sim['trade_details']] == ['AAA', 'BBB']
+    assert sim['trade_details'][0]['reason'] == 'weekly_rebalance'
+    event = sim['weekly_rebalance_events'][0]
+    assert event['exited_symbols'] == ['AAA']
+    assert event['selected_entrant_symbols'] == ['BBB']
+
+
+def test_weekly_top10_incumbent_wins_exact_current_rank_tie():
+    monday = date(2025, 1, 6)
+    friday = date(2025, 1, 10)
+    sessions = _business_days(monday, friday)
+    ords = [session.toordinal() for session in sessions]
+    prices = {
+        'AAA': _prices(ords),
+        'BBB': _prices(ords),
+    }
+    candidates = [
+        _candidate('AAA', monday, rank=10.0),
+        _candidate('BBB', friday, rank=99.0),
+    ]
+    rank_map = {
+        ('AAA', friday.isoformat()): {'strategy_rank': 80.0},
+        ('BBB', friday.isoformat()): {'strategy_rank': 80.0},
+    }
+
+    sim = bt._simulate_portfolio(
+        candidates,
+        prices,
+        None,
+        'hold',
+        30,
+        max_positions=1,
+        weekly_top_n_rebalance=True,
+        daily_rank_map=rank_map,
+        measurement_start_date=monday,
+        hard_end_date=friday + timedelta(days=1),
+        include_trades=True,
+    )
+
+    assert sim is not None
+    assert [trade['symbol'] for trade in sim['trade_details']] == ['AAA']
+    assert sim['trade_details'][0]['reason'] == 'open_at_end'
+    assert sim['weekly_rebalance_events'][0]['replacements'] == 0
+
+
+def test_weekly_rebalance_exit_bypasses_cooldown_and_churn_is_counted():
+    first_monday = date(2025, 1, 6)
+    friday = date(2025, 1, 10)
+    next_monday = date(2025, 1, 13)
+    sessions = _business_days(first_monday, next_monday)
+    ords = [session.toordinal() for session in sessions]
+    prices = {
+        'AAA': _prices(ords),
+        'BBB': (
+            ords,
+            [100.0] * len(ords),
+            [101.0] * len(ords),
+            [99.0] * (len(ords) - 1) + [70.0],
+            [100.0] * len(ords),
+            [1_000_000] * len(ords),
+        ),
+    }
+    candidates = [
+        _candidate('AAA', first_monday, rank=99.0),
+        _candidate('BBB', friday, rank=10.0),
+        _candidate('AAA', next_monday, rank=99.0),
+    ]
+    rank_map = {
+        ('AAA', friday.isoformat()): {'strategy_rank': 10.0},
+        ('BBB', friday.isoformat()): {'strategy_rank': 90.0},
+    }
+
+    sim = bt._simulate_portfolio(
+        candidates,
+        prices,
+        None,
+        'hold',
+        30,
+        max_positions=1,
+        reentry_cooldown_sessions=5,
+        weekly_top_n_rebalance=True,
+        daily_rank_map=rank_map,
+        measurement_start_date=first_monday,
+        hard_end_date=next_monday + timedelta(days=1),
+        include_trades=True,
+    )
+
+    assert sim is not None
+    assert [trade['symbol'] for trade in sim['trade_details']] == [
+        'AAA',
+        'BBB',
+        'AAA',
+    ]
+    assert sim['trade_details'][0]['reason'] == 'weekly_rebalance'
+    assert sim['rebalance_reentries_within_5_sessions'] == 1
+    assert sim['skipped_cooldown'] == 0
+
+
+def test_cohort_manifest_realizes_seven_frozen_clusters():
+    sessions = _business_days(date(2016, 1, 4), date(2026, 7, 17))
+    manifest = build_cohort_manifest(sessions)
+
+    assert validate_cohort_manifest(manifest) == []
+    assert manifest['empty_cluster_count'] == 7
+    assert manifest['warm_cluster_count'] == 7
+    assert set(map(int, manifest['empty_cluster_counts'])) == set(ANCHOR_YEARS)
+    assert all(
+        int(count) >= 12 for count in manifest['warm_seed_counts'].values()
+    )
+    cells = build_cells(manifest)
+    assert len(cells) == (
+        len(manifest['empty_book']) + len(manifest['warm_book'])
+    ) * 4 * 2
+    floor_cells = build_cells(manifest, arms=RISK_FLOOR_ARMS)
+    assert len(floor_cells) == (
+        len(manifest['empty_book']) + len(manifest['warm_book'])
+    ) * 2 * 2
+    assert {row['arm_id'] for row in floor_cells} == {
+        'cap10_incumbent',
+        'cap10_min_risk_005',
+    }
+
+
+def test_risk_floor_study_changes_only_the_effective_risk_floor():
+    control, treatment = RISK_FLOOR_ARMS
+
+    assert control['max_positions'] == treatment['max_positions'] == 10
+    assert (
+        control['weekly_top_n_rebalance']
+        == treatment['weekly_top_n_rebalance']
+        is False
+    )
+    assert control['min_initial_risk_fraction'] is None
+    assert treatment['min_initial_risk_fraction'] == 0.005
+    assert STUDIES['risk-floor-ab']['arms'] == RISK_FLOOR_ARMS
+    assert STUDIES['capacity-bracket']['arms'] != RISK_FLOOR_ARMS
+
+
+def test_zero_outcome_horizon_extends_rank_replay_to_last_session(monkeypatch):
+    monkeypatch.setattr(bt, '_window_setups', lambda *_args, **_kwargs: [])
+    count = bt.MIN_LOOKBACK + bt.HORIZON
+    start = date(2025, 1, 1)
+    ords = [start.toordinal() + offset for offset in range(count)]
+    columns = _prices(ords)
+
+    legacy = bt._replay_candidates_for_period(
+        'AAA',
+        columns,
+        {},
+        {},
+        None,
+        date.min,
+        'daily',
+        True,
+        True,
+    )
+    zero_horizon = bt._replay_candidates_for_period(
+        'AAA',
+        columns,
+        {},
+        {},
+        None,
+        date.min,
+        'daily',
+        True,
+        True,
+        0,
+    )
+
+    assert len(zero_horizon) == len(legacy) + bt.HORIZON
+    assert zero_horizon[-1]['date'] == date.fromordinal(ords[-1]).isoformat()
+
+
+def test_gain_to_pain_uses_all_monthly_returns_and_net_r():
+    sim = {
+        'measurement_start_equity': 100.0,
+        'trade_details': [
+            {
+                'net_r': 1.0,
+                'pnl': 10.0,
+                'shares': 1.0,
+                'entry': 100.0,
+                'fill': 110.0,
+                'transaction_cost': 0.0,
+            },
+            {
+                'net_r': -0.5,
+                'pnl': -5.0,
+                'shares': 1.0,
+                'entry': 100.0,
+                'fill': 95.0,
+                'transaction_cost': 0.0,
+            },
+        ],
+        'equity_curve': [
+            {'date': '2025-01-31', 'equity': 110.0},
+            {'date': '2025-02-28', 'equity': 99.0},
+        ],
+        'trades': 2,
+        'skipped_book_full': 0,
+    }
+
+    summary = summarize_simulation(sim)
+
+    assert summary['ev_net_r'] == pytest.approx(0.25)
+    assert summary['profit_factor'] == pytest.approx(2.0)
+    # Monthly returns are +10% and -10%; all-return numerator is zero.
+    assert summary['gain_to_pain'] == pytest.approx(0.0)
+
+
+def test_simple_cluster_bootstrap_is_deterministic_and_not_a_gate():
+    first = bootstrap_median_interval(
+        [1, 2, 3, 4, 5, 6, 7],
+        seed_parts=('determinism',),
+        replicates=500,
+    )
+    second = bootstrap_median_interval(
+        [1, 2, 3, 4, 5, 6, 7],
+        seed_parts=('determinism',),
+        replicates=500,
+    )
+
+    assert first == second
+    assert first['point'] == 4
+    assert first['p05'] <= first['point'] <= first['p95']
+
+
+def test_iqr_materializes_generator_before_both_quantiles():
+    assert iqr(value for value in (0.0, 1.0, 2.0, 3.0)) == pytest.approx(1.5)
+
+
+def test_aggregate_reports_paired_years_and_separate_warm_iqrs():
+    cells: list[dict] = []
+    for cost in (0.1, 0.2):
+        for cluster in ANCHOR_YEARS:
+            for seed in range(3):
+                path_id = f'warm-{cluster}-{seed}'
+                for arm_id, shift in (
+                    ('cap10_incumbent', 0.0),
+                    ('cash_unbounded', 0.2),
+                    ('cap10_weekly_top10', 0.1),
+                    ('cap15_incumbent', 0.05),
+                ):
+                    cells.append({
+                        'arm_id': arm_id,
+                        'protocol': 'warm_book',
+                        'path_id': path_id,
+                        'cluster': cluster,
+                        'cost_per_side_pct': cost,
+                        'metrics': {
+                            'ev_net_r': seed + shift,
+                            'calmar': 1.0 + seed * 0.1 + shift,
+                            'profit_factor': 1.5 + shift,
+                            'gain_to_pain': 2.0 + shift,
+                            'sortino': 1.0 + shift,
+                            'cagr_pct': 10.0 + shift,
+                            'max_drawdown_pct': 5.0,
+                            'total_return_pct': 10.0 + shift,
+                            'sharpe': 1.0 + shift,
+                        },
+                    })
+            for arm_id, shift in (
+                ('cap10_incumbent', 0.0),
+                ('cash_unbounded', 0.2),
+                ('cap10_weekly_top10', 0.1),
+                ('cap15_incumbent', 0.05),
+            ):
+                cells.append({
+                    'arm_id': arm_id,
+                    'protocol': 'empty_book',
+                    'path_id': f'empty-{cluster}',
+                    'cluster': cluster,
+                    'cost_per_side_pct': cost,
+                    'metrics': {
+                        'ev_net_r': 1.0 + shift,
+                        'calmar': 2.0 + shift,
+                        'profit_factor': 1.5 + shift,
+                        'gain_to_pain': 2.0 + shift,
+                        'sortino': 1.0 + shift,
+                        'cagr_pct': 10.0 + shift,
+                        'max_drawdown_pct': 5.0,
+                        'total_return_pct': 10.0 + shift,
+                        'sharpe': 1.0 + shift,
+                    },
+                })
+
+    report = aggregate_results(cells)
+
+    cash_empty = next(
+        row
+        for row in report['paired_per_year']
+        if row['arm_id'] == 'cash_unbounded'
+        and row['protocol'] == 'empty_book'
+        and row['cost_per_side_pct'] == 0.1
+    )
+    assert cash_empty['headline']['ev_net_r']['paired_delta_median'] == pytest.approx(
+        0.2
+    )
+    cash_paths = next(
+        row
+        for row in report['paired_path_distributions']
+        if row['arm_id'] == 'cash_unbounded'
+        and row['protocol'] == 'empty_book'
+        and row['cost_per_side_pct'] == 0.1
+    )
+    assert cash_paths['metrics']['ev_net_r']['paired_delta_mean'] == pytest.approx(
+        0.2
+    )
+    assert cash_paths['metrics']['ev_net_r']['positive_fraction'] == 1.0
+    assert cash_paths['metrics']['ev_net_r']['identical_fraction'] == 0.0
+    cash_warm = next(
+        row
+        for row in report['warm_seed_dispersion']
+        if row['arm_id'] == 'cash_unbounded'
+        and row['cost_per_side_pct'] == 0.1
+    )
+    assert set(cash_warm['headline']) == {'ev_net_r', 'calmar'}
+    assert 'D' not in cash_warm
+    assert cash_warm['headline']['ev_net_r']['median_iqr_ratio'] == pytest.approx(
+        1.0
+    )
+    assert cash_warm['headline']['calmar']['median_iqr_ratio'] == pytest.approx(
+        1.0
+    )
+    assert cash_warm['headline']['ev_net_r']['bootstrap_90']['n'] == 7
+    markdown = _markdown({
+        'generated_at': '2026-08-05T00:00:00Z',
+        'analysis': report,
+        'operational_summary': _operational_summary(cells),
+        'validation': {
+            'construction_universe_manifest': {
+                'construction_symbols_with_prices': 506,
+                'rank_only_symbols_with_prices': 4148,
+                'ranking_symbols_with_prices': 4654,
+            },
+            'candidate_rank_coverage': {
+                'construction_qualified_longs': 5000,
+                'filtered_rank_only_qualified_longs': 137000,
+            },
+        },
+    })
+    assert 'ΔGain-to-Pain' in markdown
+    assert '0.10% per fill' in markdown
+    assert '0.20% per fill' in markdown
+    assert 'Tradable setup symbols with prices: 506.' in markdown
+    assert 'Rank-only qualified rows removed: 137000.' in markdown
+    assert 'formal promotion gate' in markdown
+
+    focused_cells = [
+        row
+        for row in cells
+        if row['arm_id'] == 'cap10_incumbent'
+    ] + [
+        {
+            **row,
+            'arm_id': 'cap10_min_risk_005',
+        }
+        for row in cells
+        if row['arm_id'] == 'cash_unbounded'
+    ]
+    focused_analysis = aggregate_results(
+        focused_cells,
+        arms=RISK_FLOOR_ARMS,
+        include_warm_dispersion=False,
+    )
+    assert focused_analysis['warm_seed_dispersion'] == []
+    focused_markdown = _risk_floor_markdown({
+        'generated_at': '2026-08-05T00:00:00Z',
+        'arms': list(RISK_FLOOR_ARMS),
+        'protocols': ['empty_book', 'warm_book'],
+        'costs_per_side_pct': [0.1, 0.2],
+        'analysis': focused_analysis,
+        'operational_summary': _operational_summary(
+            focused_cells,
+            arms=RISK_FLOOR_ARMS,
+        ),
+    })
+    assert '# Effective initial-risk floor A/B' in focused_markdown
+    assert 'Mean dEV' in focused_markdown
+    assert 'Identical' in focused_markdown
+    assert 'Mean dGtP' in focused_markdown
+    assert 'Mean dCalmar/MAR' in focused_markdown
+    assert 'Floor rejects' in focused_markdown
+    assert 'not independent evidence' in focused_markdown
+
+
+def test_synthetic_worker_matrix_covers_four_arms_protocols_and_costs(monkeypatch):
+    monkeypatch.setenv('BACKTEST_SNAPSHOT_OFFLINE', '0')
+    monkeypatch.setenv('BACKTEST_ALLOW_SPAWN', '0')
+    start = date(2025, 1, 6)
+    sessions = _business_days(start, date(2025, 1, 17))
+    ords = [session.toordinal() for session in sessions]
+    symbols = [f'S{index}' for index in range(12)]
+    prices = {symbol: _prices(ords) for symbol in symbols}
+    candidates = [
+        _candidate(symbol, start, rank=99.0 - index)
+        for index, symbol in enumerate(symbols[:11])
+    ]
+    friday = date(2025, 1, 10)
+    candidates.append(_candidate('S11', friday, rank=99.0))
+    rank_map = {
+        (symbol, friday.isoformat()): {
+            'strategy_rank': 100.0 if symbol == 'S11' else float(index)
+        }
+        for index, symbol in enumerate(symbols)
+    }
+    _worker_init({
+        'qualified_candidates': candidates,
+        'daily_rank_map': rank_map,
+        'prices': prices,
+        'benchmark_closes': None,
+        'ranking_key': 'residual_high_vol_blend_80_20',
+        'exit_policy': 'hold',
+        'hold_days': 30,
+        'risk_per_trade': 0.01,
+        'atr_trail_multiplier': 3.0,
+    })
+
+    rows = []
+    for protocol, measurement_start in (
+        ('empty_book', start),
+        ('warm_book', date(2025, 1, 8)),
+    ):
+        for cost in (0.1, 0.2):
+            for arm_id in (
+                'cap10_incumbent',
+                'cash_unbounded',
+                'cap10_weekly_top10',
+                'cap15_incumbent',
+            ):
+                rows.append(_worker_run_cell({
+                    'cell_id': f'{arm_id}|{protocol}|{cost}',
+                    'arm_id': arm_id,
+                    'protocol': protocol,
+                    'path_id': f'{protocol}-synthetic',
+                    'cluster': 2025,
+                    'simulation_start': start.isoformat(),
+                    'measurement_start': measurement_start.isoformat(),
+                    'hard_end_exclusive': date(2025, 1, 14).isoformat(),
+                    'cost_per_side_pct': cost,
+                }))
+
+    assert len(rows) == 16
+    assert {row['arm_id'] for row in rows} == {
+        'cap10_incumbent',
+        'cash_unbounded',
+        'cap10_weekly_top10',
+        'cap15_incumbent',
+    }
+    assert {row['protocol'] for row in rows} == {'empty_book', 'warm_book'}
+    assert {row['cost_per_side_pct'] for row in rows} == {0.1, 0.2}
+    assert all('ev_net_r' in row['metrics'] for row in rows)
+
+
+def test_checkpoint_resume_rejects_fingerprint_mismatch(tmp_path):
+    checkpoint = tmp_path / 'checkpoint'
+    completed = _checkpoint_state(checkpoint, 'fingerprint-a', resume=False)
+    assert completed == {}
+    _write_cell_checkpoint(
+        checkpoint,
+        {'cell_id': 'one', 'metrics': {'ev_net_r': 1.0}},
+    )
+    resumed = _checkpoint_state(checkpoint, 'fingerprint-a', resume=True)
+    assert set(resumed) == {'one'}
+    with pytest.raises(SystemExit, match='fingerprint mismatch'):
+        _checkpoint_state(checkpoint, 'fingerprint-b', resume=True)
+
+
+def test_dirty_worktree_guard(monkeypatch):
+    monkeypatch.setattr(
+        'scripts.run_portfolio_construction_matrix._git_output',
+        lambda *_args: ' M changed.py',
+    )
+    with pytest.raises(SystemExit, match='dirty worktree'):
+        _assert_clean_worktree()