docs: land capacity-study evidence and share the rank-map helper

Brings the durable artifacts of research/portfolio-capacity-rebalancing onto
main so the rationale for raising the count cap lives with the code that cites
it. The matrix runner, the research simulator hooks and the study's unit tests
are deliberately left behind; they remain at tag research/portfolio-capacity-final.

Corrects conclusions that were reached on EV per trade and are now superseded:
the findings doc's decisions 1 (keep cap 10) and 4 (run the risk-floor A/B) are
struck through and answered in a new correction section, and the research README
and phase-A matrix entries are updated to match. The frozen specification itself
is untouched -- its recorded SHA-256 f1e37783 still verifies.

effective-risk-floor-ab.md is retained but marked CLOSED/NEGATIVE: the study it
proposes is already answered by cap15 vs cash_unbounded (-0.753pp CAGR while
EV/trade rises), and its EV-based pass rule would have shipped it.

scripts/research_rankings.py replaces a fourth copy of the historical rank-map
helper; run_research_matrix, run_execution_recovery_matrix and
run_daily_reentry_matrix now share it. The shared version adds a duplicate
observation guard and a deterministic symbol tie-break the copies lacked.

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

README.md

@@ -255,18 +255,11 @@ 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 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 |
+| 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) |
 | 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,7 +1320,6 @@ 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.
 
@@ -1344,13 +1343,10 @@ 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) - replay_horizon,
+        len(bars) - HORIZON,
         backtest_step_sessions(cadence),
     ):
         if bars[i].date < start_date:
@@ -1705,7 +1701,14 @@ def _gate_ablation(candidates: list[dict], activation: dict, threshold: float) -
 # the QUALIFIED setups at their detection close, best momentum first while
 # slots and cash allow.
 SIM_STARTING_CAPITAL = 10_000.0
-SIM_MAX_POSITIONS = 10
+# Headroom, not a target: the count cap should never bind. The capacity study
+# (reports/portfolio-construction-prod505-capacity-bracket-daily-v1) showed a book
+# that never hits the count cap earns +1.1pp CAGR over the old 10 (51 cohorts of 175
+# better, 2 worse) at unchanged drawdown, because the blocked entries were as good as
+# the taken ones — capacity costs trade COUNT, not trade quality. The real ceiling is
+# cash plus SIM_NOTIONAL_CAP, which saturates the book near 12 positions, so 15/20/None
+# are the same experiment. Judge any future change here on CAGR, never on EV per trade.
+SIM_MAX_POSITIONS = 15
 SIM_RISK_PER_TRADE = 0.01  # fraction of equity risked per position (entry→stop)
 SIM_NOTIONAL_CAP = 0.20    # max fraction of equity per position (no margin)
 _EULER_MASCHERONI = 0.5772156649015329
@@ -1946,7 +1949,6 @@ 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.
 
@@ -1964,18 +1966,11 @@ 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) - evaluation_horizon,
-                step_sessions,
-            )
+            for index in range(MIN_LOOKBACK - 1, len(ordinals) - HORIZON, step_sessions)
         }
 
     qualified_by_symbol_date: dict[tuple[str, int], dict] = {}
@@ -2022,7 +2017,7 @@ def _simulate_portfolio(
     *,
     qualified_fn: Callable[[dict], bool] | None = None,
     ranking_key: str = PRODUCTION_PERCENTILE_KEY,
-    max_positions: int | None = SIM_MAX_POSITIONS,
+    max_positions: int = 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,
@@ -2046,12 +2041,6 @@ 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,
@@ -2101,20 +2090,6 @@ 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")
@@ -2126,26 +2101,8 @@ 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
@@ -2154,8 +2111,6 @@ 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)
@@ -2168,12 +2123,7 @@ def _simulate_portfolio(
     }
 
     first_ord = start_ord if start_ord is not None else min(entries_by_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)
-    ]
+    calendar = sorted({o for cols in prices.values() for o in cols[0] if o >= first_ord})
     if not calendar:
         return None
 
@@ -2181,39 +2131,20 @@ 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).
-    if hard_end_ord is None:
-        last_signal_ord = max(entries_by_ord)
-        resolve_pad = hold_days + (1 if fill_mode in DELAYED_FILL_MODES else 0)
-        cut = bisect.bisect_left(calendar, last_signal_ord) + resolve_pad + 1
-        calendar = calendar[:cut]
+    last_signal_ord = max(entries_by_ord)
+    resolve_pad = hold_days + (1 if fill_mode in DELAYED_FILL_MODES else 0)
+    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] = {}
@@ -2228,12 +2159,6 @@ 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)
@@ -2303,13 +2228,6 @@ 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"],
@@ -2318,13 +2236,8 @@ def _simulate_portfolio(
             "initial_stop": pos["initial_stop"],
             "active_stop": pos["stop"],
             "fill": fill,
-            "shares": pos["shares"],
-            "initial_risk_dollars": initial_risk_dollars,
-            "pnl": net_pnl,
+            "pnl": proceeds - pos["shares"] * pos["entry"] - cost - pos["entry_cost"],
             "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"],
@@ -2339,13 +2252,6 @@ 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]
@@ -2459,82 +2365,6 @@ 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,
@@ -2543,11 +2373,7 @@ def _simulate_portfolio(
             )
             pending_delayed = []
         else:
-            fill_candidates = (
-                weekly_selected_entries
-                if weekly_selected_entries is not None
-                else signal_todays
-            )
+            fill_candidates = signal_todays
 
         def _corr_scale_for(sym: str, asof_idx: int) -> float | None:
             """1.0 ok, 0.5 half-size, None = skip. Missing history → uncorrelated."""
@@ -2592,21 +2418,15 @@ def _simulate_portfolio(
             corr_scale: float,
             fill_bar: Any | None,
         ) -> None:
-            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
+            nonlocal cash, equity, skipped_full, skipped_cooldown, post_stop_events
             sym = c["symbol"]
             if sym in positions:
                 return
             if calendar_index < cooldown_until_index.get(sym, -1):
                 skipped_cooldown += 1
                 return
-            if max_positions is not None and len(positions) >= max_positions:
+            if 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:
@@ -2623,16 +2443,6 @@ 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
@@ -2672,21 +2482,6 @@ 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).
@@ -2788,25 +2583,7 @@ def _simulate_portfolio(
             # Queue today's signals for the next session's fill.
             pending_delayed.extend(signal_todays)
 
-        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))
+        curve.append((o, _marked_equity()))
 
     # Close whatever is still open at its last mark so final equity is realized.
     for sym in list(positions):
@@ -2814,57 +2591,32 @@ def _simulate_portfolio(
     final_equity = cash
     curve[-1] = (calendar[-1], final_equity)
 
-    metric_start_ord = (
-        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
+    total_return_pct = (final_equity / SIM_STARTING_CAPITAL - 1.0) * 100.0
+    years = (calendar[-1] - calendar[0]) / 365.25
     cagr_pct = (
-        ((final_equity / metric_base_equity) ** (1.0 / years) - 1.0) * 100.0
+        ((final_equity / SIM_STARTING_CAPITAL) ** (1.0 / years) - 1.0) * 100.0
         if years > 0.25 and final_equity > 0
         else None
     )
 
     peak = float("-inf")
     max_dd = 0.0
-    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:
+    for _, eq in curve:
         peak = max(peak, eq)
         if peak > 0:
             max_dd = max(max_dd, (peak - eq) / peak)
 
-    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
-    ]
+    rets = [b / a - 1.0 for (_, a), (_, b) in zip(curve, curve[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 = metric_base_equity
-    cur_year = date.fromordinal(metric_start_ord).year
-    last_eq = metric_base_equity
-    for o, eq in metric_curve:
+    year_start_eq = curve[0][1]
+    cur_year = date.fromordinal(curve[0][0]).year
+    last_eq = curve[0][1]
+    for o, eq in curve:
         y = date.fromordinal(o).year
         if y != cur_year:
             yearly.append({
@@ -2883,29 +2635,24 @@ def _simulate_portfolio(
         ),
     })
 
-    metric_trades = [
-        trade for trade in trades if trade["entry_ord"] >= metric_start_ord
-    ]
-    pnls = [t["pnl"] for t in metric_trades]
+    pnls = [t["pnl"] for t in trades]
     wins = sum(1 for p in pnls if p > 0)
     reason_counts = {
-        reason: sum(1 for t in metric_trades if t["reason"] == reason)
-        for reason in sorted({t["reason"] for t in metric_trades})
+        reason: sum(1 for t in trades if t["reason"] == reason)
+        for reason in sorted({t["reason"] for t in 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(metric_start_ord),
-            date.fromordinal(calendar[-1]),
+            spy_closes, date.fromordinal(calendar[0]), date.fromordinal(calendar[-1])
         )
 
     curve_payload: list[dict] | None = None
     benchmark_payload: list[dict] | None = None
     if include_curve:
-        curve_base = metric_base_equity
+        curve_base = curve[0][1] if curve else SIM_STARTING_CAPITAL
         curve_payload = [
             {
                 "date": date.fromordinal(o).isoformat(),
@@ -2914,12 +2661,12 @@ def _simulate_portfolio(
                 if curve_base > 0
                 else None,
             }
-            for o, eq in metric_curve
+            for o, eq in curve
         ]
         if spy_closes:
             benchmark_payload = []
             base_spy = None
-            for o, _ in metric_curve:
+            for o, _ in curve:
                 d = date.fromordinal(o)
                 close = spy_closes.get(d)
                 if close is None or close <= 0:
@@ -2938,8 +2685,6 @@ 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),
@@ -2953,161 +2698,23 @@ def _simulate_portfolio(
         "n_returns": diag["n_returns"],
         "return_skew": diag["return_skew"],
         "return_kurtosis": diag["return_kurtosis"],
-        "trades": len(metric_trades),
-        "win_rate": (
-            round(wins / len(metric_trades) * 100.0, 1)
-            if metric_trades
-            else None
-        ),
+        "trades": len(trades),
+        "win_rate": round(wins / len(trades) * 100.0, 1) if trades else None,
         "avg_trade_pnl": round(sum(pnls) / len(pnls), 2) if pnls else None,
-        "best_trade_r": (
-            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_r": round(max(t["r"] for t in trades), 2) if trades else None,
+        "worst_trade_r": round(min(t["r"] for t in trades), 2) if 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 metric_trades) / len(metric_trades),
-                1,
-            )
-            if metric_trades
-            else None
+            round(sum(t["hold"] for t in trades) / len(trades), 1) if 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(metric_start_ord).isoformat(),
+        "start_date": date.fromordinal(calendar[0]).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)
@@ -3182,7 +2789,7 @@ def _simulate_portfolio(
                 "entry_date": date.fromordinal(trade["entry_ord"]).isoformat(),
                 "exit_date": date.fromordinal(trade["exit_ord"]).isoformat(),
             }
-            for trade in metric_trades
+            for trade in trades
         ]
     return result
 

app/services/shadow_book_service.py

@@ -40,10 +40,12 @@ KEY_CAPACITY = "shadow_book_capacity"
 KEY_RISK_PCT = "shadow_book_risk_pct"
 KEY_START_EQUITY = "shadow_book_start_equity"
 
-# Matches the validated configuration: 10-position book, 1% fixed-fractional
-# risk. Start equity is only a sizing base — comparisons are drawn in percent
-# and R-multiples, never in raw currency.
-DEFAULT_CAPACITY = 10
+# Matches the validated configuration: 1% fixed-fractional risk, and a count cap
+# set as headroom rather than a target — see backtest_service.SIM_MAX_POSITIONS,
+# which this must track. NOTIONAL_CAP below saturates the book near 12 positions,
+# so the count cap should simply never bind. Start equity is only a sizing base —
+# comparisons are drawn in percent and R-multiples, never in raw currency.
+DEFAULT_CAPACITY = 15
 DEFAULT_RISK_PCT = 1.0
 DEFAULT_START_EQUITY = 100_000.0
 

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 | 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) |
+| Max **15** concurrent positions, 1% risk per trade | Sizing | Raised from 10 (2026-08-05) so the count cap never binds: +1.075pp CAGR paired, 51 paths better / 2 worse, drawdown unchanged. Cash plus the 20% notional cap saturates the book near 12. [Findings](portfolio-capacity-bracket-findings.md#correction-2026-08-05-ev-per-trade-was-the-wrong-lens) |
 | 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** — the focused daily bracket found no meaningful gain from cap 15, while weekly rank replacement hurt. [Findings](portfolio-capacity-bracket-findings.md) |
+| Selection cutoff {70…90} × book size {10, 15, 20} | **Keep cutoff 80; book size now 15** — the focused daily bracket found cap 15 worth +1.075pp CAGR (the weekly replay's contrary reading was EV-per-trade). Weekly rank replacement hurt. [Findings](portfolio-capacity-bracket-findings.md#correction-2026-08-05-ev-per-trade-was-the-wrong-lens) |
 | 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,7 +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) |
+| **Minimum effective-risk floor** | ⛔ CLOSED NEGATIVE, not run. The floor lifts EV/trade (+0.032) and PF (+0.073) *by deleting trades* — 11.4 fewer per path, never one more — and costs **−0.753pp CAGR**, −0.047 Sharpe, −0.051 Calmar | Do not run the A/B; its EV-based pass rule would have shipped it. [Withdrawn specification](effective-risk-floor-ab.md) / [findings](portfolio-capacity-bracket-findings.md#correction-2026-08-05-ev-per-trade-was-the-wrong-lens) |
 
 ---
 
@@ -198,15 +198,19 @@ 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).
+Capacity is closed **positive**: the count cap was raised 10 → 15 so it no longer
+binds, worth **+1.075pp CAGR** paired across 175 paths (51 better, 2 worse) at
+unchanged drawdown. Fifteen is headroom, not a target — cap15 peaked at 12 with
+zero full-book skips, so cash plus the 20% notional cap is the real ceiling.
+
+An earlier reading of this run concluded "keep cap 10, added only 0.0018 R/trade."
+That was **EV per trade**, which is the wrong metric for a treatment that changes
+trade *count*: flat EV/trade means the blocked entries were as good as the taken
+ones, so refusing them cost their whole contribution to return. Weekly
+current-rank replacement remains rejected (−0.043 EV R, 24% churn). The 0.5%
+effective-risk-floor A/B is **closed negative** without being run — it costs
+0.75pp of CAGR while raising EV/trade, and its frozen pass rule would have shipped
+it. See the [frozen specification](portfolio-capacity-bracket.md) and the
+[capacity findings](portfolio-capacity-bracket-findings.md#correction-2026-08-05-ev-per-trade-was-the-wrong-lens).
 
 The next real evidence is **forward**, not backward: the live paper-trade record.

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

@@ -1,8 +1,27 @@
 # Effective initial-risk floor A/B - frozen specification
 
+> ## ⛔ CLOSED 2026-08-05 — NEGATIVE. DO NOT RUN.
+>
+> This A/B was never executed because the capacity-bracket run already contains
+> it. `cap15_incumbent` (peak 12, zero blocked, no floor) and `cash_unbounded`
+> (peak 12, floor) have the same effective capacity and differ essentially only
+> by `min_initial_risk_fraction`. Paired over 175 paths, the 0.5% floor gives
+> **EV/trade +0.032 and profit factor +0.073, but CAGR −0.753pp, total return
+> −0.765pp, Sharpe −0.047, Calmar −0.051**, and it removes 11.4 trades per path
+> while never adding one (174 worse / 0 better).
+>
+> **The pass rule below is unsafe.** It promotes on paired EV, and the floor
+> raises EV per trade *precisely by deleting trades* that were net positive
+> contributors — so this specification would have shipped a change costing
+> 0.75pp of CAGR. Any successor study must decide on CAGR/total return and treat
+> EV per trade as a diagnostic.
+>
+> See [portfolio-capacity-bracket-findings.md](portfolio-capacity-bracket-findings.md#correction-2026-08-05-ev-per-trade-was-the-wrong-lens).
+> Retained as a record of what was specified and why it was withdrawn.
+
 Date frozen: 2026-08-05
-Branch: research/portfolio-capacity-rebalancing
-Runner: scripts/run_portfolio_construction_matrix.py
+Branch: research/portfolio-capacity-rebalancing (deleted; tag `research/portfolio-capacity-final`)
+Runner: scripts/run_portfolio_construction_matrix.py (not on main; see tag)
 Study ID: risk-floor-ab
 
 ## Question

docs/research/phase-a-matrix.md

@@ -32,8 +32,13 @@ Mechanics guards confirmed before reading results: calendar truncation asserted
 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).
+gate-reset configuration. Capacity was isolated in the
+[focused bracket study](portfolio-capacity-bracket.md) and **resolved: the count
+cap was raised 10 → 15 so it no longer binds (+1.075pp CAGR paired, 51 paths
+better / 2 worse, drawdown unchanged).** Note that the blocked *count* was a poor
+guide in both directions — one path had 244 blocked entries and relieving all of
+them moved CAGR by −0.1pp. See the
+[findings correction](portfolio-capacity-bracket-findings.md#correction-2026-08-05-ev-per-trade-was-the-wrong-lens).
 
 Validation SE ≈ 0.72 — almost no arm clears a 1-SE delta.
 

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

@@ -2,8 +2,16 @@
 
 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.**
+Status: **SUPERSEDED IN PART — see [Correction](#correction-2026-08-05-ev-per-trade-was-the-wrong-lens)
+at the foot of this document before acting on anything here.** Weekly replacement
+is closed as a negative result and that still holds. The capacity decision below
+("keep cap 10") and the recommendation to run the effective-risk-floor A/B were
+both reached on EV per trade and are **reversed** by the correction: the count cap
+was raised so it no longer binds, and the floor A/B is closed as negative.
+
+> The runner (`scripts/run_portfolio_construction_matrix.py`), the research
+> simulator hooks, and the study's unit tests were deliberately not merged to
+> main. They live at tag `research/portfolio-capacity-final`.
 
 This document interprets the frozen v2 run without modifying its generated
 outputs:
@@ -116,9 +124,96 @@ 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.
+1. ~~Keep cap 10; its measured opportunity cost is negligible.~~ **REVERSED —
+   see the correction below.**
+2. Reject weekly rank replacement. *(Stands.)*
 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.
+   *(Stands — and it is not a floor effect worth having either; see below.)*
+4. ~~Run only the focused cap-10 effective-risk-floor A/B next.~~ **REVERSED —
+   that A/B is answered and negative; do not run it.**
 5. Report means, inert fractions, and absolute dispersion beside medians and
-   ratios in future sparse-treatment studies.
+   ratios in future sparse-treatment studies. *(Stands, and see below — the
+   metric itself matters as much as the summary statistic.)*
+
+## Correction 2026-08-05: EV per trade was the wrong lens
+
+Everything above judged the arms on **mean paired net EV per trade**. That is the
+wrong metric for any treatment that changes how many trades the book takes.
+Capacity does not change trade *quality*; it changes trade *count*. A flat EV/trade
+delta therefore does not mean "no benefit" — it means the blocked entries were
+**just as good** as the taken ones, so refusing them cost their entire
+contribution to return. Re-running the same paired comparison on CAGR inverts two
+conclusions.
+
+### Capacity: raise the cap (reverses decision 1)
+
+`cap15_incumbent` versus `cap10_incumbent`, paired, all 175 paths, 0.10% per fill:
+
+| Metric | Mean Δ | Worse / better |
+|---|---:|---:|
+| Trades | +1.00 | **0 / 76** (never fewer) |
+| **CAGR pp** | **+1.075** | 2 / 51 |
+| Total return pp | +1.079 | 1 / 51 |
+| Max drawdown pp | +0.007 | 1 / 2 |
+| Calmar | +0.062 | **1 / 51** |
+| Sharpe | +0.022 | 10 / 28 |
+| Net EV R/trade | +0.001 | 47 / 29 |
+
+Restricted to the 105 paths where the cap actually bound: **+1.791pp CAGR**.
+
+The honest tail: exactly one path was materially hurt — `empty-2023-04`, CAGR
+87.2 → 81.2 (−6.0pp), drawdown 13.0 → 14.3, from two extra trades. Second-worst
+was −0.1pp. The best paths (+6.6/+6.7/+6.9pp) came with *identical* drawdown. Best
+and worst magnitudes are symmetric at roughly ±6pp, but the frequency is 51:1.
+
+Blocked count is not lost value in either direction: `empty-2021-05` had **244**
+blocked entries under cap 10, and relieving every one of them moved CAGR by
+−0.1pp.
+
+**Shipped:** `SIM_MAX_POSITIONS` and `shadow_book_service.DEFAULT_CAPACITY` raised
+10 → 15. Fifteen is headroom, not a target — cap15 peaked at 12 with zero
+full-book skips, so cash plus the 20% notional cap is the real ceiling and
+15/20/None are the same experiment.
+
+### Effective-risk floor: closed negative (reverses decision 4)
+
+The floor A/B does not need running — this study already contains it.
+`cap15_incumbent` (peak 12, zero blocked, no floor) and `cash_unbounded` (peak 12,
+floor) have the same effective capacity and differ essentially only by
+`min_initial_risk_fraction`. Paired, n=175, 0.10% per fill, floor minus no-floor:
+
+| Metric | Mean Δ | Worse / better |
+|---|---:|---:|
+| Net EV R/trade | **+0.032** | 53 / 121 |
+| Profit factor | **+0.073** | 46 / 128 |
+| Trades | **−11.4** | **174 / 0** (never adds one) |
+| **CAGR pp** | **−0.753** | 105 / 68 |
+| Total return pp | −0.765 | 105 / 68 |
+| Sharpe | −0.047 | 108 / 65 |
+| Calmar | −0.051 | 103 / 71 |
+| Max drawdown pp | +0.333 (worse) | — |
+
+The same trap, mirrored: the floor raises per-trade quality *precisely by deleting
+trades*, and the deleted trades were net positive contributors. The frozen
+specification in [effective-risk-floor-ab.md](effective-risk-floor-ab.md) would
+have passed it on paired EV and shipped a change costing 0.75pp of CAGR.
+
+Genuinely open, low priority: 0.005 clearly over-cuts, but the sizing code's real
+floor is a **$1** minimum, which is no floor at all. Whether something near 0.001
+strips true dust without cutting real trades is untested, and only worth revisiting
+if live broker order minimums force it.
+
+### Start-date sensitivity is real but not a capacity artifact
+
+Within-year spread of EV across monthly start dates is ~0.672 R and is
+*identical* for `cap10` (0.672), `cap15` (0.672) and `cash_unbounded` (0.677). It
+is small-sample noise — roughly 84 trades per 252-session window drawn from a
+fat-tailed R distribution gives an EV standard error near 0.15–0.25 R — not a
+queueing artifact. No construction policy reduces it.
+
+### Rule for future studies
+
+Choose the metric from the treatment's mechanism before reading any table. If a
+treatment changes trade count, CAGR and total return are the decision metrics and
+EV per trade is a diagnostic. The generated report's headline tables lead with
+ΔEV net R, which is what made this error easy to make twice.

scripts/portfolio_capacity_research.py

@@ -1,764 +0,0 @@
-'''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',
-        },
-    }

tests/unit/test_portfolio_capacity_research.py

@@ -1,905 +0,0 @@
-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()