From 1ace6688dd05d0618d8ad0561a430e919839b70a Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Wed, 5 Aug 2026 08:28:30 +0200 Subject: [PATCH] research: add focused portfolio capacity matrix --- README.md | 9 +- app/services/backtest_service.py | 472 ++++++++- docs/research/README.md | 7 + docs/research/phase-a-matrix.md | 7 + docs/research/portfolio-capacity-bracket.md | 124 +++ scripts/portfolio_capacity_research.py | 665 ++++++++++++ scripts/research_rankings.py | 84 ++ scripts/run_daily_reentry_matrix.py | 84 +- scripts/run_execution_recovery_matrix.py | 65 +- scripts/run_portfolio_construction_matrix.py | 952 ++++++++++++++++++ scripts/run_research_matrix.py | 65 +- .../unit/test_portfolio_capacity_research.py | 595 +++++++++++ 12 files changed, 2893 insertions(+), 236 deletions(-) create mode 100644 docs/research/portfolio-capacity-bracket.md create mode 100644 scripts/portfolio_capacity_research.py create mode 100644 scripts/research_rankings.py create mode 100644 scripts/run_portfolio_construction_matrix.py create mode 100644 tests/unit/test_portfolio_capacity_research.py diff --git a/README.md b/README.md index fe90cf2..22c91d5 100644 --- a/README.md +++ b/README.md @@ -255,11 +255,18 @@ A systematic single-variable sweep (offline prod snapshot, production gate/rank/ | ATR trail multiple {1.5–4.0} | **Keep 3.0** | Return+Sharpe peak; ≤2.0 whipsaws out the momentum right tail; ≥2.5 is a plateau | | SPY 200d-MA regime overlay (block entries / go flat) | **Reject** | Halves return (315%→138%) with zero drawdown benefit — the ATR trail already manages downside, and the filter blocks the recovery-phase entries that make the money | | Momentum lookback: 6-1, 3-1, 12-7 (Novy-Marx), composites | **Keep residual 12-1** | 6-1/3-1 rank-IC ≈ 0; 12-7 IC 0.045 / t 1.58 — weaker than residual 12-1 (0.055 / t 1.98) | -| Selection cutoff {70, 75, 85, 90} × book size {10, 15, 20} | **Keep 80 × 10** | Monotonically worse in both directions from 80; the 10-slot cap never binds (<10 concurrent) | +| Selection cutoff {70, 75, 85, 90} × book size {10, 15, 20} | **Keep cutoff 80; capacity reopened** | The older weekly replay favored 80 × 10, but its no-cap-pressure conclusion is superseded by 519 book-full rejections versus 472 trades under the current daily gate-reset control | | Position sizing: equal-weight, inverse-vol, risk-% sweep | **Keep 1% fixed-fractional** | See the inverse-vol warning below | | Post-stop re-entry: immediate, fixed 2–5 sessions, gate resets, confirmation filters | **Keep normal gate reset for the 10-position production book** | Sharpe 1.77 vs 1.67 immediate and 1.47 cooldown 5; rerun before changing portfolio capacity | | FIP path-smoothness as an in-book tie-breaker/filter | **Reject** (but see the lead below) | Non-monotonic across FIP quintiles within the qualified set; either half of a median split underperforms the full book — thinning the entry stream costs more compounding than the tilt returns | +> **Capacity correction (2026-08-05):** the table's older weekly conclusion +> that the ten-slot cap never binds is superseded. Under the current daily +> gate-reset Phase A control, 472 trades were admitted and 519 qualified entries +> were rejected because the book was full (52.4% of admitted+blocked +> opportunities). Cutoff 80 remains the signal setting; portfolio capacity is +> reopened in the focused capacity-bracket study. + Two findings future sessions must not re-litigate: - **The "inverse-vol sizing win" (July 2026) was mis-attributed — do not resurrect.** The diagnostic sized `notional = equity × 1% / vol_6m`, and the 20% notional cap bound on 95% of entries, so it actually measured "~5 positions × 20% notional each" — a concentration/risk-appetite bump economically equivalent to raising risk to 1.5%, not vol-managed sizing. Genuine inverse-vol sizing (risk budget × median-vol/vol) cuts max drawdown to −18.2% but costs ~58pp total return at flat Sharpe: a risk-preference trade, not edge. diff --git a/app/services/backtest_service.py b/app/services/backtest_service.py index 1c06c87..e3d058d 100644 --- a/app/services/backtest_service.py +++ b/app/services/backtest_service.py @@ -1320,6 +1320,7 @@ def _replay_candidates_for_period( cadence: str = DEFAULT_BACKTEST_CADENCE, include_short_candidates: bool = False, include_universe_rank_observations: bool = False, + outcome_horizon_sessions: int = HORIZON, ) -> list[dict]: """Slim picklable replay used by local event studies. @@ -1343,10 +1344,13 @@ def _replay_candidates_for_period( ) ] cadence = validate_backtest_cadence(cadence) + replay_horizon = int(outcome_horizon_sessions) + if replay_horizon < 0: + raise ValueError('outcome_horizon_sessions must be non-negative') candidates: list[dict] = [] for i in range( MIN_LOOKBACK - 1, - len(bars) - HORIZON, + len(bars) - replay_horizon, backtest_step_sessions(cadence), ): if bars[i].date < start_date: @@ -1942,6 +1946,7 @@ def _make_gate_reset_reentry_fn( cadence: str, qualified_fn: Callable[[dict], bool] | None = None, ranking_key: str = PRODUCTION_PERCENTILE_KEY, + evaluation_horizon_sessions: int = HORIZON, ) -> Callable[[str, int, dict, Any], dict | None]: """Build the production post-stop gate-reset callback. @@ -1959,11 +1964,18 @@ def _make_gate_reset_reentry_fn( evaluation_ords: dict[str, set[int]] = {} step_sessions = backtest_step_sessions(cadence) + evaluation_horizon = int(evaluation_horizon_sessions) + if evaluation_horizon < 0: + raise ValueError('evaluation_horizon_sessions must be non-negative') for symbol, columns in prices.items(): ordinals = columns[0] evaluation_ords[symbol] = { int(ordinals[index]) - for index in range(MIN_LOOKBACK - 1, len(ordinals) - HORIZON, step_sessions) + for index in range( + MIN_LOOKBACK - 1, + len(ordinals) - evaluation_horizon, + step_sessions, + ) } qualified_by_symbol_date: dict[tuple[str, int], dict] = {} @@ -2010,7 +2022,7 @@ def _simulate_portfolio( *, qualified_fn: Callable[[dict], bool] | None = None, ranking_key: str = PRODUCTION_PERCENTILE_KEY, - max_positions: int = SIM_MAX_POSITIONS, + max_positions: int | None = SIM_MAX_POSITIONS, risk_per_trade: float = SIM_RISK_PER_TRADE, atr_trail_multiplier: float = ATR_TRAIL_MULTIPLIER, cost_per_side: float = COST_PER_SIDE, @@ -2034,6 +2046,12 @@ def _simulate_portfolio( corr_lookback: int = 120, corr_action: str = "skip", corr_min_overlap: int = 60, + min_initial_risk_fraction: float | None = None, + weekly_top_n_rebalance: bool = False, + daily_rank_map: dict[tuple[str, str], dict[str, float | None]] | None = None, + measurement_start_date: date | None = None, + hard_end_date: date | None = None, + include_capacity_diagnostics: bool = False, ) -> dict | None: """Replay the qualified setups as ONE capital-constrained book and report portfolio economics from the daily equity curve (return, CAGR, drawdown, @@ -2083,6 +2101,20 @@ def _simulate_portfolio( raise ValueError("corr_action must be 'skip' or 'half_size'") if vol_target is not None and vol_target <= 0: raise ValueError("vol_target must be positive when set") + if max_positions is not None and int(max_positions) <= 0: + raise ValueError("max_positions must be positive or None") + if min_initial_risk_fraction is not None and not ( + 0.0 < float(min_initial_risk_fraction) < 1.0 + ): + raise ValueError("min_initial_risk_fraction must be between 0 and 1") + if weekly_top_n_rebalance and ( + max_positions is None or daily_rank_map is None + ): + raise ValueError( + "weekly_top_n_rebalance requires max_positions and daily_rank_map" + ) + if weekly_top_n_rebalance and fill_mode != FILL_MODE_CLOSE: + raise ValueError("weekly_top_n_rebalance requires fill_mode=close") clamp_lo, clamp_hi = float(vol_clamp[0]), float(vol_clamp[1]) if clamp_lo <= 0 or clamp_hi < clamp_lo: raise ValueError("vol_clamp must satisfy 0 < lo <= hi") @@ -2094,8 +2126,26 @@ def _simulate_portfolio( entries_by_ord: dict[int, list[dict]] = defaultdict(list) start_ord = start_date.toordinal() if start_date is not None else None + measurement_start_ord = ( + measurement_start_date.toordinal() + if measurement_start_date is not None + else start_ord + ) + hard_end_ord = hard_end_date.toordinal() if hard_end_date is not None else None # Explicit simulator/holdout end dates are exclusive split boundaries. end_ord = end_date.toordinal() if end_date is not None else None + if ( + start_ord is not None + and measurement_start_ord is not None + and measurement_start_ord < start_ord + ): + raise ValueError("measurement_start_date cannot precede start_date") + if ( + hard_end_ord is not None + and measurement_start_ord is not None + and hard_end_ord <= measurement_start_ord + ): + raise ValueError("hard_end_date must follow measurement_start_date") for c in candidates: if not qualified_fn(c) or c.get("direction") != "long": continue @@ -2104,6 +2154,8 @@ def _simulate_portfolio( continue if end_ord is not None and entry_ord >= end_ord: continue # holdout/validation: entries strictly before the split + if hard_end_ord is not None and entry_ord >= hard_end_ord: + continue if not c.get("entry") or not c.get("stop"): continue entries_by_ord[entry_ord].append(c) @@ -2116,7 +2168,12 @@ def _simulate_portfolio( } first_ord = start_ord if start_ord is not None else min(entries_by_ord) - calendar = sorted({o for cols in prices.values() for o in cols[0] if o >= first_ord}) + full_calendar = sorted({o for cols in prices.values() for o in cols[0]}) + calendar = [ + o + for o in full_calendar + if o >= first_ord and (hard_end_ord is None or o < hard_end_ord) + ] if not calendar: return None @@ -2124,20 +2181,39 @@ def _simulate_portfolio( # fill lag). Prevents trailing flat-cash after the last resolvable entry — # the clear-air train-window bug — for train, validation, and full-period # books alike (including max-hold sweeps out to 90 days). - last_signal_ord = max(entries_by_ord) - 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 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] if not calendar: return None + weekly_rebalance_ords: set[int] = set() + for index, session_ord in enumerate(full_calendar): + session_date = date.fromordinal(session_ord) + iso = session_date.isocalendar() + if index + 1 < len(full_calendar): + next_iso = date.fromordinal(full_calendar[index + 1]).isocalendar() + if (iso.year, iso.week) != (next_iso.year, next_iso.week): + weekly_rebalance_ords.add(session_ord) + elif session_date.weekday() == 4: + weekly_rebalance_ords.add(session_ord) + cash = SIM_STARTING_CAPITAL positions: dict[str, dict] = {} curve: list[tuple[int, float]] = [] trades: list[dict] = [] skipped_full = 0 + measurement_skipped_full = 0 skipped_cooldown = 0 skipped_corr = 0 + skipped_min_initial_risk = 0 + measurement_skipped_min_initial_risk = 0 + opened_positions = 0 + measurement_opened_positions = 0 + weekly_rank_rejected_entries = 0 + measurement_weekly_rank_rejected_entries = 0 skipped_missing_fill = 0 skipped_gap_cap = 0 cooldown_until_index: dict[str, int] = {} @@ -2152,6 +2228,12 @@ def _simulate_portfolio( vol_scalars: list[float] = [] overnight_slippage_pct: list[float] = [] pending_delayed: list[dict] = [] + measurement_start_equity: float | None = None + measurement_start_position_count: int | None = None + capacity_samples: list[dict[str, float | int]] = [] + weekly_rebalance_events: list[dict] = [] + rebalance_exit_index: dict[str, tuple[int, int]] = {} + rebalance_reentry_events: list[dict] = [] def _bar(sym: str, o: int): idx = index_of.get(sym, {}).get(o) @@ -2221,6 +2303,13 @@ def _simulate_portfolio( cost = proceeds * cost_rate cash += proceeds - cost risk = pos["entry"] - pos["initial_stop"] + initial_risk_dollars = pos["shares"] * risk + net_pnl = ( + proceeds + - pos["shares"] * pos["entry"] + - cost + - pos["entry_cost"] + ) trades.append({ "symbol": sym, "entry_ord": pos["entry_ord"], @@ -2229,8 +2318,13 @@ def _simulate_portfolio( "initial_stop": pos["initial_stop"], "active_stop": pos["stop"], "fill": fill, - "pnl": proceeds - pos["shares"] * pos["entry"] - cost - pos["entry_cost"], + "shares": pos["shares"], + "initial_risk_dollars": initial_risk_dollars, + "pnl": net_pnl, "r": (fill - pos["entry"]) / risk if risk > 0 else 0.0, + "net_r": net_pnl / initial_risk_dollars + if initial_risk_dollars > 0 + else 0.0, "hold": pos["bars_held"], "reason": reason, "stop_refreshes": pos["stop_refreshes"], @@ -2245,6 +2339,13 @@ def _simulate_portfolio( cooldown_sessions = max(0, int(reentry_cooldown_sessions)) for calendar_index, o in enumerate(calendar): + in_measurement = ( + measurement_start_ord is None or o >= measurement_start_ord + ) + if in_measurement and measurement_start_equity is None: + measurement_start_equity = _marked_equity() + measurement_start_position_count = len(positions) + # 1) exits on today's bars (stop intraday, target intraday, time at close) for sym in list(positions): pos = positions[sym] @@ -2358,6 +2459,82 @@ def _simulate_portfolio( reverse=True, ) + weekly_selected_entries: list[dict] | None = None + if weekly_top_n_rebalance and o in weekly_rebalance_ords: + assert max_positions is not None + assert daily_rank_map is not None + asof = date.fromordinal(o).isoformat() + protected: set[str] = set() + ranked_pool: list[tuple[float, int, str, dict | None]] = [] + for sym in positions: + rank_row = daily_rank_map.get((sym, asof)) + current_rank = ( + rank_row.get("strategy_rank") if rank_row is not None else None + ) + if current_rank is None or _bar(sym, o) is None: + protected.add(sym) + continue + ranked_pool.append((float(current_rank), 0, sym, None)) + + entrants_by_symbol: dict[str, dict] = {} + for candidate in signal_todays: + sym = str(candidate["symbol"]) + if sym in positions or sym in entrants_by_symbol: + continue + entrants_by_symbol[sym] = candidate + eligible_entrants = 0 + for sym, candidate in entrants_by_symbol.items(): + rank_row = daily_rank_map.get((sym, asof)) + current_rank = ( + rank_row.get("strategy_rank") if rank_row is not None else None + ) + if current_rank is None: + continue + eligible_entrants += 1 + ranked_pool.append((float(current_rank), 1, sym, candidate)) + + available_slots = max(0, int(max_positions) - len(protected)) + ranked_pool.sort(key=lambda row: (-row[0], row[1], row[2])) + selected = ranked_pool[:available_slots] + selected_holding_symbols = { + sym for _rank, kind, sym, _candidate in selected if kind == 0 + } + weekly_selected_entries = [ + candidate + for _rank, kind, _sym, candidate in selected + if kind == 1 and candidate is not None + ] + selected_entrant_symbols = { + str(candidate["symbol"]) for candidate in weekly_selected_entries + } + rejected_now = max(0, eligible_entrants - len(selected_entrant_symbols)) + weekly_rank_rejected_entries += rejected_now + if in_measurement: + measurement_weekly_rank_rejected_entries += rejected_now + + exited_symbols: list[str] = [] + for sym in list(positions): + if sym in protected or sym in selected_holding_symbols: + continue + bar = _bar(sym, o) + if bar is None: + continue + _close_trade(sym, float(bar.close), "weekly_rebalance") + rebalance_exit_index[sym] = (calendar_index, o) + exited_symbols.append(sym) + + weekly_rebalance_events.append({ + "ord": o, + "fresh_entrant_pool": len(entrants_by_symbol), + "rank_eligible_entrant_pool": eligible_entrants, + "selected_entrants": len(selected_entrant_symbols), + "replacements": len(exited_symbols), + "exited_symbols": sorted(exited_symbols), + "selected_entrant_symbols": sorted(selected_entrant_symbols), + "measurement": in_measurement, + }) + equity = _marked_equity() + if fill_mode in DELAYED_FILL_MODES: fill_candidates = sorted( pending_delayed, @@ -2366,7 +2543,11 @@ def _simulate_portfolio( ) pending_delayed = [] else: - fill_candidates = signal_todays + fill_candidates = ( + weekly_selected_entries + if weekly_selected_entries is not None + else signal_todays + ) def _corr_scale_for(sym: str, asof_idx: int) -> float | None: """1.0 ok, 0.5 half-size, None = skip. Missing history → uncorrelated.""" @@ -2411,15 +2592,21 @@ def _simulate_portfolio( corr_scale: float, fill_bar: Any | None, ) -> None: - nonlocal cash, equity, skipped_full, skipped_cooldown, post_stop_events + nonlocal cash, equity, skipped_full, measurement_skipped_full + nonlocal skipped_cooldown, post_stop_events + nonlocal skipped_min_initial_risk + nonlocal measurement_skipped_min_initial_risk + nonlocal opened_positions, measurement_opened_positions sym = c["symbol"] if sym in positions: return if calendar_index < cooldown_until_index.get(sym, -1): skipped_cooldown += 1 return - if len(positions) >= max_positions: + if max_positions is not None and len(positions) >= max_positions: skipped_full += 1 + if in_measurement: + measurement_skipped_full += 1 return risk_ps = entry - stop if risk_ps <= 0 or entry <= 0: @@ -2436,6 +2623,16 @@ def _simulate_portfolio( (equity * SIM_NOTIONAL_CAP) / entry, max(cash, 0.0) / (entry * (1.0 + cost_rate)), ) + initial_risk_dollars = shares * risk_ps + if ( + min_initial_risk_fraction is not None + and initial_risk_dollars + < equity * float(min_initial_risk_fraction) + ): + skipped_min_initial_risk += 1 + if in_measurement: + measurement_skipped_min_initial_risk += 1 + return if shares * entry < 1.0: return entry_cost = shares * entry * cost_rate @@ -2475,6 +2672,21 @@ def _simulate_portfolio( "vol_scalar": scalar, "corr_scale": corr_scale, } + opened_positions += 1 + if in_measurement: + measurement_opened_positions += 1 + prior_rebalance_exit = rebalance_exit_index.pop(sym, None) + if prior_rebalance_exit is not None: + prior_exit_index, prior_exit_ord = prior_rebalance_exit + rebalance_reentry_events.append({ + "symbol": sym, + "exit_ord": prior_exit_ord, + "exit_calendar_index": prior_exit_index, + "reentry_calendar_index": calendar_index, + "wait_sessions": calendar_index - prior_exit_index, + "reentry_ord": entry_ord, + "measurement": in_measurement, + }) # next_open only: fill is at the open, so the rest of the bar can stop out. # stale_close fills at the close — same-day stop after entry does not apply. # bars_held stays 0 on the fill day (matches close-fill cadence). @@ -2576,7 +2788,25 @@ def _simulate_portfolio( # Queue today's signals for the next session's fill. pending_delayed.extend(signal_todays) - curve.append((o, _marked_equity())) + marked_equity = _marked_equity() + if in_measurement and include_capacity_diagnostics: + gross_notional = sum( + pos["shares"] * pos["last_close"] for pos in positions.values() + ) + capacity_samples.append({ + "positions": len(positions), + "cash_pct": cash / marked_equity * 100.0 + if marked_equity > 0 + else 0.0, + "gross_exposure_pct": gross_notional / marked_equity * 100.0 + if marked_equity > 0 + else 0.0, + "at_capacity": int( + max_positions is not None + and len(positions) >= max_positions + ), + }) + curve.append((o, marked_equity)) # Close whatever is still open at its last mark so final equity is realized. for sym in list(positions): @@ -2584,32 +2814,57 @@ def _simulate_portfolio( final_equity = cash curve[-1] = (calendar[-1], final_equity) - total_return_pct = (final_equity / SIM_STARTING_CAPITAL - 1.0) * 100.0 - years = (calendar[-1] - calendar[0]) / 365.25 + 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 cagr_pct = ( - ((final_equity / SIM_STARTING_CAPITAL) ** (1.0 / years) - 1.0) * 100.0 + ((final_equity / metric_base_equity) ** (1.0 / years) - 1.0) * 100.0 if years > 0.25 and final_equity > 0 else None ) peak = float("-inf") max_dd = 0.0 - for _, eq in curve: + drawdown_equities = ( + [metric_base_equity, *(eq for _, eq in metric_curve)] + if measurement_start_date is not None + else [eq for _, eq in metric_curve] + ) + for eq in drawdown_equities: peak = max(peak, eq) if peak > 0: max_dd = max(max_dd, (peak - eq) / peak) - rets = [b / a - 1.0 for (_, a), (_, b) in zip(curve, curve[1:]) if a > 0] + return_equities = ( + [metric_base_equity, *(eq for _, eq in metric_curve)] + if measurement_start_date is not None + else [eq for _, eq in metric_curve] + ) + rets = [ + b / a - 1.0 + for a, b in zip(return_equities, return_equities[1:]) + if a > 0 + ] diag = sharpe_diagnostics(rets) sharpe = diag["sharpe"] # Per-calendar-year returns off the equity curve — shows whether every year # contributed or one exceptional stretch carried the result. yearly: list[dict] = [] - year_start_eq = curve[0][1] - cur_year = date.fromordinal(curve[0][0]).year - last_eq = curve[0][1] - for o, eq in curve: + 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: y = date.fromordinal(o).year if y != cur_year: yearly.append({ @@ -2628,24 +2883,29 @@ def _simulate_portfolio( ), }) - pnls = [t["pnl"] for t in trades] + metric_trades = [ + trade for trade in trades if trade["entry_ord"] >= metric_start_ord + ] + pnls = [t["pnl"] for t in metric_trades] wins = sum(1 for p in pnls if p > 0) reason_counts = { - reason: sum(1 for t in trades if t["reason"] == reason) - for reason in sorted({t["reason"] for t in trades}) + reason: sum(1 for t in metric_trades if t["reason"] == reason) + for reason in sorted({t["reason"] for t in metric_trades}) } spy_pct = None if spy_closes: from app.services.benchmark_service import benchmark_return_pct spy_pct = benchmark_return_pct( - spy_closes, date.fromordinal(calendar[0]), date.fromordinal(calendar[-1]) + spy_closes, + date.fromordinal(metric_start_ord), + date.fromordinal(calendar[-1]), ) curve_payload: list[dict] | None = None benchmark_payload: list[dict] | None = None if include_curve: - curve_base = curve[0][1] if curve else SIM_STARTING_CAPITAL + curve_base = metric_base_equity curve_payload = [ { "date": date.fromordinal(o).isoformat(), @@ -2654,12 +2914,12 @@ def _simulate_portfolio( if curve_base > 0 else None, } - for o, eq in curve + for o, eq in metric_curve ] if spy_closes: benchmark_payload = [] base_spy = None - for o, _ in curve: + for o, _ in metric_curve: d = date.fromordinal(o) close = spy_closes.get(d) if close is None or close <= 0: @@ -2678,6 +2938,8 @@ def _simulate_portfolio( calmar = float(cagr_pct) / max_dd_pct result = { "starting_capital": SIM_STARTING_CAPITAL, + "measurement_start_equity": round(metric_base_equity, 2), + "measurement_start_positions": measurement_start_position_count or 0, "cost_per_side_pct": round(cost_rate * 100.0, 3), "fill_mode": fill_mode, "final_equity": round(final_equity, 2), @@ -2691,23 +2953,161 @@ def _simulate_portfolio( "n_returns": diag["n_returns"], "return_skew": diag["return_skew"], "return_kurtosis": diag["return_kurtosis"], - "trades": len(trades), - "win_rate": round(wins / len(trades) * 100.0, 1) if trades else None, + "trades": len(metric_trades), + "win_rate": ( + round(wins / len(metric_trades) * 100.0, 1) + if metric_trades + else None + ), "avg_trade_pnl": round(sum(pnls) / len(pnls), 2) if pnls else None, - "best_trade_r": round(max(t["r"] for t in trades), 2) if trades else None, - "worst_trade_r": round(min(t["r"] for t in trades), 2) if trades 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_pnl": round(max(pnls), 2) if pnls else None, "worst_trade_pnl": round(min(pnls), 2) if pnls else None, "avg_hold_days": ( - round(sum(t["hold"] for t in trades) / len(trades), 1) if trades else None + round( + sum(t["hold"] for t in metric_trades) / len(metric_trades), + 1, + ) + if metric_trades + else None ), "exit_reasons": reason_counts, "skipped_book_full": skipped_full, "spy_return_pct": round(spy_pct, 1) if spy_pct is not None else None, "yearly_returns": yearly, - "start_date": date.fromordinal(calendar[0]).isoformat(), + "start_date": date.fromordinal(metric_start_ord).isoformat(), "end_date": date.fromordinal(calendar[-1]).isoformat(), } + if measurement_start_date is not None: + result["simulation_start_date"] = date.fromordinal(calendar[0]).isoformat() + if hard_end_date is not None: + result["hard_end_date_exclusive"] = hard_end_date.isoformat() + if measurement_start_date is not None: + result["measurement_skipped_book_full"] = measurement_skipped_full + result["measurement_opened_positions"] = measurement_opened_positions + if min_initial_risk_fraction is not None: + result["min_initial_risk_fraction"] = float(min_initial_risk_fraction) + result["skipped_min_initial_risk"] = skipped_min_initial_risk + result["measurement_skipped_min_initial_risk"] = ( + measurement_skipped_min_initial_risk + ) + if include_capacity_diagnostics: + measured_opened = ( + measurement_opened_positions + if measurement_start_date is not None + else opened_positions + ) + measured_full = ( + measurement_skipped_full + if measurement_start_date is not None + else skipped_full + ) + capacity_opportunities = measured_opened + measured_full + result["opened_positions"] = measured_opened + result["capacity_opportunities"] = capacity_opportunities + result["blocked_fraction"] = ( + round(measured_full / capacity_opportunities, 6) + if capacity_opportunities + else 0.0 + ) + result["avg_positions"] = ( + round( + sum(float(sample["positions"]) for sample in capacity_samples) + / len(capacity_samples), + 4, + ) + if capacity_samples + else 0.0 + ) + result["peak_positions"] = ( + max(int(sample["positions"]) for sample in capacity_samples) + if capacity_samples + else 0 + ) + result["sessions_at_capacity"] = sum( + int(sample["at_capacity"]) for sample in capacity_samples + ) + result["sessions_measured"] = len(capacity_samples) + result["avg_cash_pct"] = ( + round( + sum(float(sample["cash_pct"]) for sample in capacity_samples) + / len(capacity_samples), + 4, + ) + if capacity_samples + else None + ) + result["avg_gross_exposure_pct"] = ( + round( + sum( + float(sample["gross_exposure_pct"]) + for sample in capacity_samples + ) + / len(capacity_samples), + 4, + ) + if capacity_samples + else None + ) + if weekly_top_n_rebalance: + measured_events = [ + event for event in weekly_rebalance_events if event["measurement"] + ] + measured_reentries = [ + event for event in rebalance_reentry_events if event["measurement"] + ] + result["weekly_rank_rejected_entries"] = ( + measurement_weekly_rank_rejected_entries + if measurement_start_date is not None + else weekly_rank_rejected_entries + ) + result["weekly_rebalance_events"] = [ + { + **{ + key: value + for key, value in event.items() + if key not in {"ord", "measurement"} + }, + "date": date.fromordinal(event["ord"]).isoformat(), + } + for event in measured_events + ] + result["rebalance_reentry_events"] = [ + { + **{ + key: value + for key, value in event.items() + if key + not in { + "exit_ord", + "reentry_ord", + "measurement", + "exit_calendar_index", + "reentry_calendar_index", + } + }, + "exit_date": date.fromordinal(event["exit_ord"]).isoformat(), + "reentry_date": date.fromordinal( + event["reentry_ord"] + ).isoformat(), + } + for event in measured_reentries + ] + for session_limit in (5, 10, 20): + result[f"rebalance_reentries_within_{session_limit}_sessions"] = sum( + 1 + for event in measured_reentries + if int(event["wait_sessions"]) <= session_limit + ) if vol_target is not None: result["vol_target"] = vol_target result["vol_lookback"] = int(vol_lookback) @@ -2782,7 +3182,7 @@ def _simulate_portfolio( "entry_date": date.fromordinal(trade["entry_ord"]).isoformat(), "exit_date": date.fromordinal(trade["exit_ord"]).isoformat(), } - for trade in trades + for trade in metric_trades ] return result diff --git a/docs/research/README.md b/docs/research/README.md index 0bd30b6..393e116 100644 --- a/docs/research/README.md +++ b/docs/research/README.md @@ -197,4 +197,11 @@ 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 itself is no longer considered settled. The current daily Phase A +control rejects 519 qualified entries because the ten-slot book is full versus +472 admitted trades. The older weekly “cap never binds” result is stale. The +[frozen focused capacity bracket](portfolio-capacity-bracket.md) compares cap +10, cap 15, cash-only unbounded, and weekly current-rank top 10 without tuning +replacement variants or using a formal promotion gate. + The next real evidence is **forward**, not backward: the live paper-trade record. diff --git a/docs/research/phase-a-matrix.md b/docs/research/phase-a-matrix.md index f925005..ff65af5 100644 --- a/docs/research/phase-a-matrix.md +++ b/docs/research/phase-a-matrix.md @@ -28,6 +28,13 @@ Mechanics guards confirmed before reading results: calendar truncation asserted | **Validation** | **1.68** | **0.72** | **41.6%** | **20.9%** | **1.99** | **239** | | Full (close-fill) | 1.77 | 0.50 | 48.3% | 21.6% | 2.23 | 472 | +**Capacity correction (2026-08-05):** the full close-fill control also records +skipped_book_full = 519 versus 472 admitted trades, so the ten-slot book +refuses 52.4% of admitted+blocked qualified opportunities. The older weekly +claim that the cap never bound is stale and does not apply to this daily +gate-reset configuration. Capacity is now isolated in the +[focused bracket study](portfolio-capacity-bracket.md). + Validation SE ≈ 0.72 — almost no arm clears a 1-SE delta. --- diff --git a/docs/research/portfolio-capacity-bracket.md b/docs/research/portfolio-capacity-bracket.md new file mode 100644 index 0000000..4585afa --- /dev/null +++ b/docs/research/portfolio-capacity-bracket.md @@ -0,0 +1,124 @@ +# Portfolio-capacity bracket — frozen specification + +Date frozen: 2026-08-05 +Branch: research/portfolio-capacity-rebalancing +Runner: scripts/run_portfolio_construction_matrix.py + +## Question and motivation + +The daily Phase A production control (a0_control: close fill, 30-session +maximum hold, 1% fixed-fractional risk, no correlation or volatility overlay) +recorded 472 trades and 519 otherwise qualified entries rejected because the +ten-position book was full. The blocked share is 519 / (519 + 472) = 52.4%. +The book is therefore materially arrival-order constrained. + +This supersedes the older statement that the ten-slot cap never bound. That +statement came from a shorter, weekly, pre-gate-reset replay and is not evidence +about the current daily strategy. + +The study brackets the value of capacity before tuning replacement details. It +does not contain a formal promotion rule or automatically change production. +Because the current ~505-name production membership is projected backward, +paired arm-versus-control differences are the primary evidence. Absolute +profitability is descriptive and survivorship-biased. + +## Frozen arms + +1. **cap10_incumbent:** exact production-style cap-10 control, no displacement. +2. **cash_unbounded:** no position-count cap; cash/no leverage and the existing + 20% per-position notional ceiling remain. Reject an entry if actual initial + stop-risk after cash/notional sizing is below 0.5% of marked equity. +3. **cap10_weekly_top10:** on the final trading session of each ISO week, rank + holdings plus fresh same-day qualified entrants and retain the top ten. +4. **cap15_incumbent:** cap 15, no displacement. + +All arms use the frozen Phase A control configuration: daily candidate replay, +live-like full-universe residual-momentum/low-volatility 80/20 rank, activation +threshold 80, normal gate-reset re-entry, close fill, 3×ATR trail, 30-session +maximum hold, 1% risk, and costs of 0.10% and 0.20% per fill. + +The daily replay uses zero outcome horizon: setup and rank observations continue +through the snapshot's last session because portfolio simulation, unlike outcome +grading, does not require 30 future bars. + +### Weekly-selection mechanics + +- Ordinary exits run before entries/rebalancing. +- Open slots may still fill from daily qualified entries during the week. +- On the final ISO-week session, current holdings and that day's fresh qualified + entrants use the full-universe strategy_rank for that same date. +- Stored entry-day rank is never used. +- Holdings with missing current rank/data are protected and consume a slot; + entrants missing rank are ineligible. +- Incumbents win exact rank ties; symbol is the deterministic final tie-breaker. +- Rebalance exits pay costs and bypass cooldown/post-stop state. +- Report entrant-pool sizes, replacements, turnover, and same-symbol re-entry + within 5/10/20 sessions. + +## Frozen cohorts + +research.sqlite is expected to cover 2016-01-04 through 2026-07-17. Residual +momentum requires 252 benchmark sessions. Empty-book starts additionally require +504 prior scoring sessions and 252 forward measurement sessions. + +- **Empty book:** first eligible session of each month, approximately January + 2019 through July 2025; start with no positions and measure 252 sessions. +- **Warm book:** first session of each year 2019–2025 is the measurement anchor. + Seed the portfolio on the first session of every ISO week falling 63–126 + trading sessions before the anchor, carry all positions and gate-reset state + forward, and measure the same 252-session anchor window. + +Warm portfolio returns reset to marked equity immediately before the anchor +session. P&L after the anchor from carried positions belongs to portfolio +returns, while trade EV includes only entries on or after the anchor. Remaining +positions liquidate at the last measurement close with costs. + +The validate-only mode must print realized cohort counts and fail unless both +protocols contain the seven annual clusters 2019–2025 and every warm anchor has +at least 12 seeds. + +## Reporting + +Primary reported measures: + +- net EV per trade in R, with costs and actual initial stop-risk dollars; +- Calmar (CAGR / max drawdown); +- profit factor on net trade R; +- Gain-to-Pain (sum of all monthly returns / absolute sum of negative months); +- Sortino using daily returns and zero target. + +Also report total return/CAGR, maximum drawdown, Sharpe, win rate, time +underwater, exposure, cash, average/peak positions, sessions at capacity, +turnover, costs, qualified/admitted/blocked opportunities, and minimum-risk +rejections. + +For each arm/protocol/cost/metric, pair identical paths with cap10_incumbent, +take the median paired delta within each start year or annual anchor, show all +seven cluster values, and headline their median. + +Initialization dispersion is reported separately for EV and Calmar: calculate +the seed-path IQR within each warm anchor, divide by the paired control IQR, show +all seven ratios, and headline their median. Do not combine them into a composite. + +For context only, run a deterministic 10,000-replicate cluster bootstrap over +the seven paired annual summaries and report the central 90% percentile interval +for median EV and Calmar deltas and warm IQR ratios. These intervals are not +promotion gates, independent-population confidence claims, or formal inference. + +## Reproducibility and execution + +Candidate replay/ranks cache under reports/.cache; each matrix cell checkpoints +atomically and resume verifies a fingerprint over the implementation commit, +this specification hash, snapshot SHA-256, cache key, arm definitions, costs, +and cohort manifest. An authoritative run refuses a dirty worktree. + +Preflight: + + python scripts/run_portfolio_construction_matrix.py backtest_snapshots/research.sqlite --run-id prod505-capacity-bracket-daily-v1 --validate-only + +Authoritative run: + + python scripts/run_portfolio_construction_matrix.py backtest_snapshots/research.sqlite --run-id prod505-capacity-bracket-daily-v1 --workers auto --resume + +Commit only the compact final JSON and Markdown reports. Raw curves, trades, +candidate caches, and checkpoints remain ignored. diff --git a/scripts/portfolio_capacity_research.py b/scripts/portfolio_capacity_research.py new file mode 100644 index 0000000..e0c7e98 --- /dev/null +++ b/scripts/portfolio_capacity_research.py @@ -0,0 +1,665 @@ +'''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} +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]) -> list[dict[str, Any]]: + paths = [*manifest['empty_book'], *manifest['warm_book']] + cells: list[dict[str, Any]] = [] + for cost in COSTS_PER_SIDE_PCT: + 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: + q25 = percentile(values, 0.25) + q75 = percentile(values, 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]]) -> dict[str, Any]: + paired: list[dict[str, Any]] = [] + for cost in COSTS_PER_SIDE_PCT: + for protocol in ('empty_book', 'warm_book'): + 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, + }) + + warm_rows = [ + row for row in cells if row['protocol'] == 'warm_book' + ] + warm_dispersion: list[dict[str, Any]] = [] + for cost in COSTS_PER_SIDE_PCT: + 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, + }) + + return { + 'paired_per_year': paired, + 'warm_seed_dispersion': warm_dispersion, + 'bootstrap': { + 'replicates': BOOTSTRAP_REPLICATES, + 'seed': BOOTSTRAP_SEED, + 'interval': 'central 90% percentile, context only', + 'resampling_unit': 'seven annual paired summaries', + }, + } diff --git a/scripts/research_rankings.py b/scripts/research_rankings.py new file mode 100644 index 0000000..b201774 --- /dev/null +++ b/scripts/research_rankings.py @@ -0,0 +1,84 @@ +'''Shared production-style historical ranking helpers for research runners.''' + +from __future__ import annotations + +from datetime import date + + +def _period_percentiles( + observations: list[dict], value_key: str +) -> dict[tuple[str, str], float]: + '''Rank one deterministic ticker observation per historical period.''' + by_period: dict[tuple, list[dict]] = {} + seen: set[tuple[str, str]] = set() + for row in observations: + identity = (str(row['symbol']), str(row['date'])) + if identity in seen: + raise ValueError(f'Duplicate universe rank observation: {identity}') + seen.add(identity) + if row.get(value_key) is None: + continue + period = tuple(row['ranking_period']) + by_period.setdefault(period, []).append(row) + + result: dict[tuple[str, str], float] = {} + for group in by_period.values(): + ordered = sorted( + group, + key=lambda row: (float(row[value_key]), str(row['symbol'])), + ) + denominator = len(ordered) - 1 + for rank, row in enumerate(ordered): + result[(str(row['symbol']), str(row['date']))] = round( + rank / denominator * 100.0 if denominator > 0 else 100.0, + 2, + ) + return result + + +def _live_universe_rank_map( + observations: list[dict], + benchmark_closes: dict[date, float], + momentum_weight: float, +) -> dict[tuple[str, str], dict[str, float | None]]: + '''Historical equivalent of production compute_activation_ranks. + + Every ticker contributes at most once per session. Residual momentum starts + only once 252 benchmark closes were point-in-time available; earlier dates + use the same raw-momentum fallback as production. + ''' + identities = [(str(row['symbol']), str(row['date'])) for row in observations] + if len(identities) != len(set(identities)): + raise ValueError('Universe ranking requires one observation per ticker/date') + + raw_pct = _period_percentiles(observations, 'momentum') + residual_pct = _period_percentiles(observations, 'residual_momentum') + vol_pct = _period_percentiles(observations, 'vol_6m') + benchmark_ords = sorted(value.toordinal() for value in benchmark_closes) + residual_start_ord = benchmark_ords[251] if len(benchmark_ords) >= 252 else None + + ranks: dict[tuple[str, str], dict[str, float | None]] = {} + for row in observations: + identity = (str(row['symbol']), str(row['date'])) + asof_ord = date.fromisoformat(identity[1]).toordinal() + momentum_pct = ( + residual_pct.get(identity) + if residual_start_ord is not None and asof_ord >= residual_start_ord + else raw_pct.get(identity) + ) + volatility_pct = vol_pct.get(identity) + strategy_rank = ( + round( + momentum_pct * momentum_weight + + volatility_pct * (1.0 - momentum_weight), + 2, + ) + if momentum_pct is not None and volatility_pct is not None + else momentum_pct + ) + ranks[identity] = { + 'momentum_percentile': momentum_pct, + 'volatility_percentile': volatility_pct, + 'strategy_rank': strategy_rank, + } + return ranks diff --git a/scripts/run_daily_reentry_matrix.py b/scripts/run_daily_reentry_matrix.py index 5da6e53..f71f79e 100644 --- a/scripts/run_daily_reentry_matrix.py +++ b/scripts/run_daily_reentry_matrix.py @@ -29,6 +29,11 @@ ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) +from scripts.research_rankings import ( # noqa: E402 + _live_universe_rank_map, + _period_percentiles, +) + POLICY_NAMES = ( "immediate", "next_session", @@ -107,85 +112,6 @@ def _default_output_path() -> Path: return Path("reports") / f"daily-reentry-matrix-{stamp}.json" -def _period_percentiles( - observations: list[dict], value_key: str -) -> dict[tuple[str, str], float]: - """Production-style percentiles, one deterministic symbol row per period.""" - by_period: dict[tuple, list[dict]] = {} - seen: set[tuple[str, str]] = set() - for row in observations: - identity = (str(row["symbol"]), str(row["date"])) - if identity in seen: - raise ValueError(f"Duplicate universe rank observation: {identity}") - seen.add(identity) - if row.get(value_key) is None: - continue - period = tuple(row["ranking_period"]) - by_period.setdefault(period, []).append(row) - - result: dict[tuple[str, str], float] = {} - for group in by_period.values(): - ordered = sorted( - group, - key=lambda row: (float(row[value_key]), str(row["symbol"])), - ) - denominator = len(ordered) - 1 - for rank, row in enumerate(ordered): - result[(str(row["symbol"]), str(row["date"]))] = round( - rank / denominator * 100.0 if denominator > 0 else 100.0, - 2, - ) - return result - - -def _live_universe_rank_map( - observations: list[dict], - benchmark_closes: dict[date, float], - momentum_weight: float, -) -> dict[tuple[str, str], dict[str, float | None]]: - """Historical equivalent of ``compute_activation_ranks``. - - Every ticker contributes at most once per session. Residual momentum starts - only once 252 benchmark closes were point-in-time available; earlier dates - use the same raw-momentum fallback as production. - """ - identities = [(str(row["symbol"]), str(row["date"])) for row in observations] - if len(identities) != len(set(identities)): - raise ValueError("Universe ranking requires one observation per ticker/date") - - raw_pct = _period_percentiles(observations, "momentum") - residual_pct = _period_percentiles(observations, "residual_momentum") - vol_pct = _period_percentiles(observations, "vol_6m") - benchmark_ords = sorted(value.toordinal() for value in benchmark_closes) - residual_start_ord = benchmark_ords[251] if len(benchmark_ords) >= 252 else None - - ranks: dict[tuple[str, str], dict[str, float | None]] = {} - for row in observations: - identity = (str(row["symbol"]), str(row["date"])) - asof_ord = date.fromisoformat(identity[1]).toordinal() - momentum_pct = ( - residual_pct.get(identity) - if residual_start_ord is not None and asof_ord >= residual_start_ord - else raw_pct.get(identity) - ) - volatility_pct = vol_pct.get(identity) - strategy_rank = ( - round( - momentum_pct * momentum_weight - + volatility_pct * (1.0 - momentum_weight), - 2, - ) - if momentum_pct is not None and volatility_pct is not None - else momentum_pct - ) - ranks[identity] = { - "momentum_percentile": momentum_pct, - "volatility_percentile": volatility_pct, - "strategy_rank": strategy_rank, - } - return ranks - - class PrecomputedDailyEngine: """Exact date/symbol lookup over the already-ranked production gate.""" diff --git a/scripts/run_execution_recovery_matrix.py b/scripts/run_execution_recovery_matrix.py index d1b9a02..07abdfa 100644 --- a/scripts/run_execution_recovery_matrix.py +++ b/scripts/run_execution_recovery_matrix.py @@ -55,6 +55,11 @@ ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) +from scripts.research_rankings import ( # noqa: E402 + _live_universe_rank_map, + _period_percentiles, +) + # Must match Phase A cache when reusing research-cands.pkl CACHE_VERSION = "research-matrix-v1-daily-prod" @@ -104,66 +109,6 @@ def _parse_args() -> argparse.Namespace: return p.parse_args() -def _period_percentiles( - observations: list[dict], value_key: str -) -> dict[tuple[str, str], float]: - by_period: dict[tuple, list[dict]] = {} - for row in observations: - if row.get(value_key) is None: - continue - period = tuple(row["ranking_period"]) - by_period.setdefault(period, []).append(row) - result: dict[tuple[str, str], float] = {} - for group in by_period.values(): - ordered = sorted( - group, key=lambda row: (float(row[value_key]), str(row["symbol"])) - ) - denominator = len(ordered) - 1 - for rank, row in enumerate(ordered): - result[(str(row["symbol"]), str(row["date"]))] = round( - rank / denominator * 100.0 if denominator > 0 else 100.0, - 2, - ) - return result - - -def _live_universe_rank_map( - observations: list[dict], - benchmark_closes: dict[date, float], - momentum_weight: float, -) -> dict[tuple[str, str], dict[str, float | None]]: - raw_pct = _period_percentiles(observations, "momentum") - residual_pct = _period_percentiles(observations, "residual_momentum") - vol_pct = _period_percentiles(observations, "vol_6m") - benchmark_ords = sorted(value.toordinal() for value in benchmark_closes) - residual_start_ord = benchmark_ords[251] if len(benchmark_ords) >= 252 else None - ranks: dict[tuple[str, str], dict[str, float | None]] = {} - for row in observations: - identity = (str(row["symbol"]), str(row["date"])) - asof_ord = date.fromisoformat(identity[1]).toordinal() - momentum_pct = ( - residual_pct.get(identity) - if residual_start_ord is not None and asof_ord >= residual_start_ord - else raw_pct.get(identity) - ) - volatility_pct = vol_pct.get(identity) - strategy_rank = ( - round( - momentum_pct * momentum_weight - + volatility_pct * (1.0 - momentum_weight), - 2, - ) - if momentum_pct is not None and volatility_pct is not None - else momentum_pct - ) - ranks[identity] = { - "momentum_percentile": momentum_pct, - "volatility_percentile": volatility_pct, - "strategy_rank": strategy_rank, - } - return ranks - - def _window(arm: dict, name: str) -> dict | None: for row in arm.get("windows") or []: if row.get("window") == name: diff --git a/scripts/run_portfolio_construction_matrix.py b/scripts/run_portfolio_construction_matrix.py new file mode 100644 index 0000000..84b133e --- /dev/null +++ b/scripts/run_portfolio_construction_matrix.py @@ -0,0 +1,952 @@ +'''Run the focused four-arm daily portfolio-capacity research matrix. + +The expensive point-in-time daily replay and full-universe ranks are cached +once. Empty-book monthly paths and warm-book weekly seeds are then evaluated +under cap 10, cap 15, cash-only unbounded, and weekly current-rank top 10. +''' + +from __future__ import annotations + +import argparse +import asyncio +import hashlib +import json +import multiprocessing +import os +import pickle +import platform +import subprocess +import sys +from collections import Counter +from concurrent.futures import ProcessPoolExecutor, as_completed +from datetime import date, datetime, timezone +from pathlib import Path +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from scripts.portfolio_capacity_research import ( # noqa: E402 + ANCHOR_YEARS, + ARM_BY_ID, + ARMS, + BOOTSTRAP_REPLICATES, + BOOTSTRAP_SEED, + COSTS_PER_SIDE_PCT, + aggregate_results, + build_cells, + build_cohort_manifest, + median, + summarize_simulation, + validate_cohort_manifest, +) +from scripts.research_rankings import _live_universe_rank_map # noqa: E402 + + +CACHE_VERSION = 'portfolio-capacity-candidates-v1-zero-horizon' +RUNNER_VERSION = 'portfolio-capacity-bracket-v1' +SPEC_PATH = ROOT / 'docs' / 'research' / 'portfolio-capacity-bracket.md' +DEFAULT_RUN_ID = 'prod505-capacity-bracket-daily-v1' +_WORKER_CONTEXT: dict[str, Any] | None = None + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('snapshot', help='SQLite backtest snapshot') + parser.add_argument('--run-id', default=DEFAULT_RUN_ID) + parser.add_argument( + '--workers', + default='auto', + help='Worker count or auto', + ) + parser.add_argument('--resume', action='store_true') + parser.add_argument('--validate-only', action='store_true') + parser.add_argument('--candidate-cache', default=None) + parser.add_argument('--checkpoint', default=None) + parser.add_argument('--out', default=None) + parser.add_argument('--quiet', action='store_true') + return parser.parse_args() + + +def _worker_count(raw: str) -> int: + if str(raw).lower() == 'auto': + return max(1, min(6, (multiprocessing.cpu_count() or 2) - 1)) + value = int(raw) + if value <= 0: + raise ValueError('--workers must be positive or auto') + return value + + +def _sqlite_url(path: Path) -> str: + return f'sqlite+aiosqlite:///{path.resolve().as_posix()}' + + +def _sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open('rb') as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b''): + digest.update(chunk) + return digest.hexdigest() + + +def _json_hash(value: Any) -> str: + payload = json.dumps( + value, + sort_keys=True, + separators=(',', ':'), + default=str, + ).encode('utf-8') + return hashlib.sha256(payload).hexdigest() + + +def _atomic_json(path: Path, value: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(path.suffix + '.tmp') + with temporary.open('w', encoding='utf-8', newline='\n') as handle: + json.dump(value, handle, indent=2, sort_keys=True, allow_nan=False) + handle.write('\n') + temporary.replace(path) + + +def _atomic_text(path: Path, value: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(path.suffix + '.tmp') + with temporary.open('w', encoding='utf-8', newline='\n') as handle: + handle.write(value.rstrip() + '\n') + temporary.replace(path) + + +def _atomic_pickle(path: Path, value: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(path.suffix + '.tmp') + with temporary.open('wb') as handle: + pickle.dump(value, handle, protocol=pickle.HIGHEST_PROTOCOL) + temporary.replace(path) + + +def _git_output(*args: str) -> str: + completed = subprocess.run( + ['git', *args], + cwd=ROOT, + check=True, + capture_output=True, + text=True, + ) + return completed.stdout.strip() + + +def _assert_clean_worktree() -> None: + dirty = _git_output('status', '--porcelain') + if dirty: + raise SystemExit( + 'Authoritative research refuses a dirty worktree; commit first:\n' + + dirty + ) + + +def _worker_init(context: dict[str, Any]) -> None: + global _WORKER_CONTEXT + os.environ['BACKTEST_SNAPSHOT_OFFLINE'] = '1' + os.environ['BACKTEST_ALLOW_SPAWN'] = '1' + from app.services import backtest_service as bt + + context = dict(context) + context['post_stop_reentry_fn'] = bt._make_gate_reset_reentry_fn( + context['qualified_candidates'], + context['prices'], + cadence='daily', + ranking_key=context['ranking_key'], + evaluation_horizon_sessions=0, + ) + _WORKER_CONTEXT = context + + +def _worker_run_cell(cell: dict[str, Any]) -> dict[str, Any]: + if _WORKER_CONTEXT is None: + raise RuntimeError('Portfolio-capacity worker was not initialized') + from app.services import backtest_service as bt + + context = _WORKER_CONTEXT + arm = ARM_BY_ID[str(cell['arm_id'])] + measurement_start = date.fromisoformat(str(cell['measurement_start'])) + hard_end = date.fromisoformat(str(cell['hard_end_exclusive'])) + sim = bt._simulate_portfolio( + context['qualified_candidates'], + context['prices'], + context['benchmark_closes'], + context['exit_policy'], + context['hold_days'], + ranking_key=context['ranking_key'], + max_positions=arm['max_positions'], + risk_per_trade=context['risk_per_trade'], + atr_trail_multiplier=context['atr_trail_multiplier'], + cost_per_side=float(cell['cost_per_side_pct']) / 100.0, + post_stop_reentry_fn=context['post_stop_reentry_fn'], + start_date=date.fromisoformat(str(cell['simulation_start'])), + end_date=hard_end, + measurement_start_date=measurement_start, + hard_end_date=hard_end, + fill_mode=bt.FILL_MODE_CLOSE, + min_initial_risk_fraction=arm['min_initial_risk_fraction'], + weekly_top_n_rebalance=bool(arm['weekly_top_n_rebalance']), + daily_rank_map=( + context['daily_rank_map'] + if arm['weekly_top_n_rebalance'] + else None + ), + include_curve=True, + include_trades=True, + include_capacity_diagnostics=True, + ) + if sim is None: + raise RuntimeError(f'Cell produced no trades: {cell["cell_id"]}') + return { + **cell, + 'metrics': summarize_simulation(sim), + } + + +async def _load_snapshot( + snapshot: Path, + *, + quiet: bool, +) -> dict[str, Any]: + from app.models.ticker import Ticker + from app.services import backtest_service as bt + from app.services.admin_service import get_activation_config + from app.services.paper_trade_service import get_exit_policy + from app.services.recommendation_service import get_recommendation_config + + engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True) + Session = async_sessionmaker( + engine, + class_=AsyncSession, + expire_on_commit=False, + ) + try: + async with Session() as db: + recommendation_config = await get_recommendation_config(db) + activation = await get_activation_config(db) + exit_config = await get_exit_policy(db) + benchmark_closes = await bt._load_benchmark_closes_for_backtest( + db, + days=None, + refresh=False, + ) + ticker_result = await db.execute( + select(Ticker).order_by(Ticker.symbol) + ) + symbols = [ + ticker.symbol for ticker in ticker_result.scalars().all() + ] + prices: dict[str, tuple] = {} + for index, symbol in enumerate(symbols, 1): + columns = await bt._fetch_columns(db, symbol) + if columns is not None: + prices[symbol] = columns + if not quiet and index % 50 == 0: + print( + f'loaded prices: {index}/{len(symbols)}', + flush=True, + ) + finally: + await engine.dispose() + + if not prices or not benchmark_closes: + raise SystemExit('Snapshot has no usable prices or benchmark history') + + strategy = next( + row + for row in bt.PORTFOLIO_MONITOR_STRATEGIES + if row.get('is_production') + ) + entry_config = bt._entry_variant_config(str(strategy['entry_variant'])) + if entry_config is None: + raise RuntimeError('Production entry configuration is missing') + ranking_key = str( + entry_config.get('ranking_key') or entry_config['percentile_key'] + ) + exit_policy = bt.LIVE_EXIT_MODE_TO_SIM.get( + str(exit_config.get('mode', 'atr_trailing')), + 'atr_trail3', + ) + hold_days = int(exit_config.get('hold_days', 30)) + risk_per_trade = float(entry_config['risk_per_trade']) + atr_trail_multiplier = float( + exit_config.get('atr_multiplier', bt.ATR_TRAIL_MULTIPLIER) + ) + threshold = float( + activation.get('min_momentum_percentile', 80.0) + ) + expected = { + 'ranking_key': bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY, + 'exit_policy': 'atr_trail3', + 'hold_days': 30, + 'risk_per_trade': 0.01, + 'max_positions': 10, + 'threshold': 80.0, + } + realized = { + 'ranking_key': ranking_key, + 'exit_policy': exit_policy, + 'hold_days': hold_days, + 'risk_per_trade': risk_per_trade, + 'max_positions': int(entry_config['max_positions']), + 'threshold': threshold, + } + if realized != expected: + raise SystemExit( + 'Snapshot runtime configuration is not the frozen Phase A control: ' + + json.dumps({'expected': expected, 'realized': realized}, sort_keys=True) + ) + + return { + 'recommendation_config': recommendation_config, + 'activation': activation, + 'exit_config': exit_config, + 'benchmark_closes': benchmark_closes, + 'prices': prices, + 'symbols': symbols, + 'universe_manifest': { + 'ticker_rows': len(symbols), + 'symbols_with_prices': len(prices), + 'symbols_sha256': _json_hash(sorted(symbols)), + }, + 'ranking_key': ranking_key, + 'exit_policy': exit_policy, + 'hold_days': hold_days, + 'risk_per_trade': risk_per_trade, + 'atr_trail_multiplier': atr_trail_multiplier, + 'threshold': threshold, + 'runtime_config': realized, + } + + +def _build_candidate_cache( + snapshot_data: dict[str, Any], + *, + snapshot: Path, + snapshot_sha256: str, + cache_path: Path, + workers: int, + quiet: bool, +) -> dict[str, Any]: + from app.services import backtest_service as bt + + cache_key = { + 'version': CACHE_VERSION, + 'snapshot': str(snapshot.resolve()), + 'snapshot_sha256': snapshot_sha256, + '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'], + } + if cache_path.exists(): + with cache_path.open('rb') as handle: + cached = pickle.load(handle) # noqa: S301 - trusted local cache + if cached.get('key') == cache_key: + if not quiet: + print(f'loaded candidate/rank cache: {cache_path}', flush=True) + return cached + if not quiet: + print(f'candidate cache mismatch; rebuilding: {cache_path}', flush=True) + + replay_rows: list[dict[str, Any]] = [] + prices = snapshot_data['prices'] + replay_args = [ + ( + symbol, + columns, + snapshot_data['recommendation_config'], + snapshot_data['activation'], + snapshot_data['benchmark_closes'], + date(1900, 1, 1), + 'daily', + True, + True, + 0, + ) + for symbol, columns in prices.items() + ] + if workers == 1: + for index, args in enumerate(replay_args, 1): + replay_rows.extend(bt._replay_candidates_for_period(*args)) + if not quiet and index % 25 == 0: + print( + f'daily replay: {index}/{len(replay_args)} tickers', + flush=True, + ) + else: + context = bt._mp_context() or multiprocessing.get_context('spawn') + with ProcessPoolExecutor( + max_workers=workers, + mp_context=context, + ) as pool: + futures = [ + pool.submit(bt._replay_candidates_for_period, *args) + for args in replay_args + ] + for index, future in enumerate(as_completed(futures), 1): + replay_rows.extend(future.result()) + if not quiet and index % 25 == 0: + print( + f'daily replay: {index}/{len(futures)} tickers', + flush=True, + ) + + setup_candidates = [ + row for row in replay_rows if not row.get('_rank_only') + ] + rank_observations = [ + row for row in replay_rows if row.get('_universe_rank_observation') + ] + daily_rank_map = _live_universe_rank_map( + rank_observations, + snapshot_data['benchmark_closes'], + bt.STRATEGY_RANK_MOMENTUM_WEIGHT, + ) + qualified: list[dict[str, Any]] = [] + for setup in setup_candidates: + if setup.get('direction') != 'long': + continue + identity = (str(setup['symbol']), str(setup['date'])) + rank = daily_rank_map.get(identity) + if rank is None: + continue + candidate = { + key: value + for key, value in setup.items() + if not key.startswith('_universe_') + } + candidate[bt.PRODUCTION_PERCENTILE_KEY] = rank['momentum_percentile'] + candidate[bt.VOL_PERCENTILE_KEY] = rank['volatility_percentile'] + candidate[bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY] = rank['strategy_rank'] + candidate['qualified'] = bt._momentum_qualifies( + candidate, + snapshot_data['threshold'], + ) + if candidate['qualified']: + qualified.append(candidate) + + qualified.sort( + key=lambda row: ( + str(row['date']), + -float(row.get(snapshot_data['ranking_key']) or 0.0), + str(row['symbol']), + float(row.get('entry') or 0.0), + float(row.get('stop') or 0.0), + float(row.get('target') or 0.0), + str(row.get('action') or ''), + ) + ) + rank_dates = sorted({identity[1] for identity in daily_rank_map}) + cached = { + 'key': cache_key, + 'qualified_candidates': qualified, + 'daily_rank_map': daily_rank_map, + 'setup_candidate_count': len(setup_candidates), + 'qualified_long_count': len(qualified), + 'entry_candidates_by_direction': dict( + Counter(str(row['direction']) for row in setup_candidates) + ), + 'rank_observation_count': len(rank_observations), + 'rank_first_date': rank_dates[0] if rank_dates else None, + 'rank_last_date': rank_dates[-1] if rank_dates else None, + } + _atomic_pickle(cache_path, cached) + if not quiet: + print(f'wrote candidate/rank cache: {cache_path}', flush=True) + return cached + + +def _checkpoint_state( + checkpoint_dir: Path, + fingerprint: str, + *, + resume: bool, +) -> dict[str, dict[str, Any]]: + manifest_path = checkpoint_dir / 'manifest.json' + if checkpoint_dir.exists() and not resume: + existing_cells = list(checkpoint_dir.glob('cell-*.json')) + if existing_cells: + raise SystemExit( + f'Checkpoint cells already exist at {checkpoint_dir}; use --resume ' + 'or a new --run-id' + ) + checkpoint_dir.mkdir(parents=True, exist_ok=True) + if manifest_path.exists(): + existing = json.loads(manifest_path.read_text(encoding='utf-8')) + if existing.get('fingerprint') != fingerprint: + raise SystemExit( + f'Checkpoint fingerprint mismatch at {checkpoint_dir}' + ) + else: + _atomic_json( + manifest_path, + { + 'runner_version': RUNNER_VERSION, + 'fingerprint': fingerprint, + 'created_at': datetime.now(timezone.utc).isoformat(), + }, + ) + + completed: dict[str, dict[str, Any]] = {} + if resume: + for path in sorted(checkpoint_dir.glob('cell-*.json')): + row = json.loads(path.read_text(encoding='utf-8')) + completed[str(row['cell_id'])] = row + return completed + + +def _write_cell_checkpoint( + checkpoint_dir: Path, + row: dict[str, Any], +) -> None: + name = hashlib.sha256(str(row['cell_id']).encode('utf-8')).hexdigest() + _atomic_json(checkpoint_dir / f'cell-{name}.json', row) + + +def _fmt(value: Any, digits: int = 3) -> str: + if value is None: + return 'n/a' + if isinstance(value, float): + return f'{value:.{digits}f}' + return str(value) + + +def _operational_summary(cells: list[dict[str, Any]]) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for arm in ARMS: + arm_cells = [ + row + for row in cells + if row['arm_id'] == arm['id'] + and float(row['cost_per_side_pct']) == 0.1 + ] + weekly = [ + row['metrics']['weekly_rebalance'] + for row in arm_cells + if row['metrics'].get('weekly_rebalance') + ] + rows.append({ + 'arm_id': arm['id'], + 'paths': len(arm_cells), + 'median_trades': median( + row['metrics'].get('trades') for row in arm_cells + ), + 'median_blocked_fraction': median( + row['metrics'].get('blocked_fraction') for row in arm_cells + ), + 'median_avg_positions': median( + row['metrics'].get('avg_positions') for row in arm_cells + ), + 'peak_positions': max( + ( + int(row['metrics'].get('peak_positions') or 0) + for row in arm_cells + ), + default=0, + ), + 'median_turnover_multiple': median( + row['metrics'].get('turnover_multiple') for row in arm_cells + ), + 'min_risk_rejections': sum( + int(row['metrics'].get('skipped_min_initial_risk') or 0) + for row in arm_cells + ), + 'weekly_zero_entrant_fraction': median( + item.get('zero_entrant_fraction') for item in weekly + ), + 'weekly_entrant_pool_median': median( + item.get('entrant_pool_median') for item in weekly + ), + 'weekly_replacements': sum( + int(item.get('replacements') or 0) for item in weekly + ), + 'weekly_reentries_within_10_sessions': sum( + int(item.get('reentries_within_10_sessions') or 0) + for item in weekly + ), + }) + return rows + + +def _markdown(report: dict[str, Any]) -> str: + lines = [ + '# Focused daily portfolio-capacity matrix', + '', + f'Generated: {report["generated_at"]}', + '', + '## Question', + '', + 'The current daily Phase A control admitted 472 trades and rejected 519 ' + 'qualified opportunities because the ten-slot book was full. This run ' + 'brackets the economic cost of that binding constraint; it has no formal ' + 'promotion gate.', + '', + '> Universe caveat: today\'s production membership is projected backward. ' + 'Use paired arm-versus-control differences, not absolute profitability, ' + 'for construction conclusions.', + '', + '## Paired annual medians', + '', + ] + paired = report['analysis']['paired_per_year'] + for protocol in ('empty_book', 'warm_book'): + lines.extend([ + f'### {protocol.replace("_", " ").title()} — 0.10% per fill', + '', + '| Arm | ΔEV net R | 90% context | ΔCalmar | 90% context |', + '|---|---:|---:|---:|---:|', + ]) + for arm in ARMS: + row = next( + item + for item in paired + if item['arm_id'] == arm['id'] + and item['protocol'] == protocol + and float(item['cost_per_side_pct']) == 0.1 + ) + ev = row['headline']['ev_net_r'] + calmar = row['headline']['calmar'] + ev_ci = ev['bootstrap_90'] + calmar_ci = calmar['bootstrap_90'] + lines.append( + f'| {arm["id"]} | {_fmt(ev["paired_delta_median"])} | ' + f'[{_fmt(ev_ci["p05"])}, {_fmt(ev_ci["p95"])}] | ' + f'{_fmt(calmar["paired_delta_median"])} | ' + f'[{_fmt(calmar_ci["p05"])}, {_fmt(calmar_ci["p95"])}] |' + ) + lines.append('') + lines.extend([ + '| Arm | ΔPF | ΔGain-to-Pain | ΔSortino | ΔCAGR pp | ΔMaxDD pp |', + '|---|---:|---:|---:|---:|---:|', + ]) + for arm in ARMS: + row = next( + item + for item in paired + if item['arm_id'] == arm['id'] + and item['protocol'] == protocol + and float(item['cost_per_side_pct']) == 0.1 + ) + headline = row['headline'] + lines.append( + f'| {arm["id"]} | ' + f'{_fmt(headline["profit_factor"]["paired_delta_median"])} | ' + f'{_fmt(headline["gain_to_pain"]["paired_delta_median"])} | ' + f'{_fmt(headline["sortino"]["paired_delta_median"])} | ' + f'{_fmt(headline["cagr_pct"]["paired_delta_median"])} | ' + f'{_fmt(headline["max_drawdown_pct"]["paired_delta_median"])} |' + ) + lines.append('') + + lines.extend([ + '## Warm-seed initialization dispersion', + '', + '| Arm | Cost/fill | Median EV IQR ratio | Median Calmar IQR ratio |', + '|---|---:|---:|---:|', + ]) + for row in report['analysis']['warm_seed_dispersion']: + lines.append( + f'| {row["arm_id"]} | {row["cost_per_side_pct"]:.2f}% | ' + f'{_fmt(row["headline"]["ev_net_r"]["median_iqr_ratio"])} | ' + f'{_fmt(row["headline"]["calmar"]["median_iqr_ratio"])} |' + ) + + lines.extend([ + '', + '## Capacity and operations — 0.10% per fill', + '', + '| Arm | Median trades | Median blocked | Median positions | Peak | ' + 'Turnover | Min-risk rejects |', + '|---|---:|---:|---:|---:|---:|---:|', + ]) + for row in report['operational_summary']: + blocked = row['median_blocked_fraction'] + blocked_text = ( + f'{float(blocked) * 100.0:.1f}%' if blocked is not None else 'n/a' + ) + lines.append( + f'| {row["arm_id"]} | {_fmt(row["median_trades"], 1)} | ' + f'{blocked_text} | {_fmt(row["median_avg_positions"], 2)} | ' + f'{row["peak_positions"]} | ' + f'{_fmt(row["median_turnover_multiple"], 2)} | ' + f'{row["min_risk_rejections"]} |' + ) + + weekly = next( + row + for row in report['operational_summary'] + if row['arm_id'] == 'cap10_weekly_top10' + ) + lines.extend([ + '', + '## Weekly-ranking opportunity set', + '', + f'- Median fresh entrant pool: ' + f'{_fmt(weekly["weekly_entrant_pool_median"], 1)}.', + f'- Median zero-entrant fraction: ' + f'{_fmt(weekly["weekly_zero_entrant_fraction"], 3)}.', + f'- Replacements across reported paths: {weekly["weekly_replacements"]}.', + f'- Same-symbol re-entries within 10 sessions: ' + f'{weekly["weekly_reentries_within_10_sessions"]}.', + '', + 'Bootstrap intervals above resample seven annual summaries and are ' + 'descriptive context only. They are not gates or independent-population ' + 'confidence claims.', + ]) + return '\n'.join(lines) + + +async def _main() -> None: + args = _parse_args() + snapshot = Path(args.snapshot) + if not snapshot.exists(): + raise SystemExit(f'Snapshot does not exist: {snapshot}') + if not SPEC_PATH.exists(): + raise SystemExit(f'Frozen specification is missing: {SPEC_PATH}') + + os.environ['BACKTEST_SNAPSHOT_OFFLINE'] = '1' + os.environ['BACKTEST_ALLOW_SPAWN'] = '1' + workers = _worker_count(str(args.workers)) + snapshot_sha256 = _sha256_file(snapshot) + specification_sha256 = _sha256_file(SPEC_PATH) + + snapshot_data = await _load_snapshot(snapshot, quiet=bool(args.quiet)) + cache_path = ( + Path(args.candidate_cache) + if args.candidate_cache + else ROOT / 'reports' / '.cache' / f'{args.run_id}-candidates.pkl' + ) + candidate_cache = _build_candidate_cache( + snapshot_data, + snapshot=snapshot, + snapshot_sha256=snapshot_sha256, + cache_path=cache_path, + workers=workers, + quiet=bool(args.quiet), + ) + + cohort_manifest = build_cohort_manifest( + snapshot_data['benchmark_closes'].keys() + ) + cohort_errors = validate_cohort_manifest(cohort_manifest) + cells = build_cells(cohort_manifest) + validation_payload = { + 'runner_version': RUNNER_VERSION, + 'snapshot': str(snapshot.resolve()), + 'snapshot_sha256': snapshot_sha256, + 'snapshot_sessions': { + 'count': cohort_manifest['session_count'], + 'first': cohort_manifest['snapshot_first_session'], + 'last': cohort_manifest['snapshot_last_session'], + }, + 'candidate_rank_coverage': { + 'first': candidate_cache['rank_first_date'], + 'last': candidate_cache['rank_last_date'], + 'observations': candidate_cache['rank_observation_count'], + 'qualified_longs': candidate_cache['qualified_long_count'], + }, + 'universe_manifest': snapshot_data['universe_manifest'], + 'empty_cluster_counts': cohort_manifest['empty_cluster_counts'], + 'warm_seed_counts': cohort_manifest['warm_seed_counts'], + 'empty_cluster_count': cohort_manifest['empty_cluster_count'], + 'warm_cluster_count': cohort_manifest['warm_cluster_count'], + 'expected_clusters': list(ANCHOR_YEARS), + 'matrix_cells': len(cells), + 'cache_path': str(cache_path.resolve()), + 'cache_key_hash': _json_hash(candidate_cache['key']), + 'errors': cohort_errors, + } + print(json.dumps(validation_payload, indent=2, sort_keys=True), flush=True) + if cohort_errors: + raise SystemExit( + 'Cohort validation failed; revise and re-hash the specification' + ) + if args.validate_only: + return + + _assert_clean_worktree() + git_commit = _git_output('rev-parse', 'HEAD') + fingerprint_payload = { + 'runner_version': RUNNER_VERSION, + 'git_commit': git_commit, + 'snapshot_sha256': snapshot_sha256, + 'specification_sha256': specification_sha256, + 'candidate_cache_key': candidate_cache['key'], + 'universe_manifest': snapshot_data['universe_manifest'], + 'cohort_manifest': cohort_manifest, + 'arms': list(ARMS), + 'costs_per_side_pct': list(COSTS_PER_SIDE_PCT), + 'bootstrap': { + 'replicates': BOOTSTRAP_REPLICATES, + 'seed': BOOTSTRAP_SEED, + }, + } + fingerprint = _json_hash(fingerprint_payload) + checkpoint_dir = ( + Path(args.checkpoint) + if args.checkpoint + else ROOT / 'reports' / '.cache' / f'{args.run_id}-checkpoint' + ) + completed = _checkpoint_state( + checkpoint_dir, + fingerprint, + resume=bool(args.resume), + ) + expected_ids = {str(cell['cell_id']) for cell in cells} + unknown = set(completed) - expected_ids + if unknown: + raise SystemExit( + f'Checkpoint contains {len(unknown)} unknown matrix cells' + ) + remaining = [ + cell for cell in cells if str(cell['cell_id']) not in completed + ] + if not args.quiet: + print( + f'matrix cells: {len(completed)} resumed, {len(remaining)} remaining', + flush=True, + ) + + worker_context = { + 'qualified_candidates': candidate_cache['qualified_candidates'], + 'daily_rank_map': candidate_cache['daily_rank_map'], + 'prices': snapshot_data['prices'], + 'benchmark_closes': snapshot_data['benchmark_closes'], + 'ranking_key': snapshot_data['ranking_key'], + 'exit_policy': snapshot_data['exit_policy'], + 'hold_days': snapshot_data['hold_days'], + 'risk_per_trade': snapshot_data['risk_per_trade'], + 'atr_trail_multiplier': snapshot_data['atr_trail_multiplier'], + } + if workers == 1: + _worker_init(worker_context) + for index, cell in enumerate(remaining, 1): + row = _worker_run_cell(cell) + completed[str(row['cell_id'])] = row + _write_cell_checkpoint(checkpoint_dir, row) + if not args.quiet: + print( + f'portfolio cells: {len(completed)}/{len(cells)} ' + f'({row["cell_id"]})', + flush=True, + ) + elif remaining: + context = multiprocessing.get_context('spawn') + with ProcessPoolExecutor( + max_workers=workers, + mp_context=context, + initializer=_worker_init, + initargs=(worker_context,), + ) as pool: + futures = { + pool.submit(_worker_run_cell, cell): str(cell['cell_id']) + for cell in remaining + } + for future in as_completed(futures): + row = future.result() + completed[str(row['cell_id'])] = row + _write_cell_checkpoint(checkpoint_dir, row) + if not args.quiet: + print( + f'portfolio cells: {len(completed)}/{len(cells)} ' + f'({row["cell_id"]})', + flush=True, + ) + + missing = expected_ids - set(completed) + if missing: + raise RuntimeError( + f'Incomplete matrix: {len(missing)} cells are missing' + ) + result_cells = sorted( + completed.values(), + key=lambda row: str(row['cell_id']), + ) + analysis = aggregate_results(result_cells) + requirements_path = ROOT / 'requirements.txt' + report: dict[str, Any] = { + 'run_id': args.run_id, + 'status': 'complete', + 'generated_at': datetime.now(timezone.utc).isoformat(), + 'research_question': ( + 'Bracket the economic cost of the binding ten-position cap and ' + 'test whether weekly current-rank selection beats arrival order.' + ), + 'decision_rule': ( + 'No formal promotion gate. Report paired annual medians, warm-seed ' + 'EV/Calmar IQR ratios, and simple bootstrap intervals as context.' + ), + 'survivorship_bias_caveat': ( + 'The current production universe is projected backward; construction ' + 'conclusions rely on paired relative comparisons, not absolute levels.' + ), + 'motivation': { + 'source': 'reports/research-matrix-phase-a.json a0_control full window', + 'trades': 472, + 'skipped_book_full': 519, + 'blocked_fraction': 519 / (519 + 472), + 'stale_claim_corrected': ( + 'The older weekly pre-gate-reset claim that cap 10 never bound ' + 'does not apply to the current daily configuration.' + ), + }, + 'fingerprint': fingerprint, + 'fingerprint_payload': fingerprint_payload, + 'environment': { + 'python': sys.version, + 'platform': platform.platform(), + 'requirements_sha256': ( + _sha256_file(requirements_path) + if requirements_path.exists() + else None + ), + 'command': [sys.executable, *sys.argv], + }, + 'validation': validation_payload, + 'runtime_config': snapshot_data['runtime_config'], + 'universe_manifest': snapshot_data['universe_manifest'], + 'candidate_cache': { + key: value + for key, value in candidate_cache.items() + if key + not in { + 'qualified_candidates', + 'daily_rank_map', + } + }, + 'cohort_manifest': cohort_manifest, + 'arms': list(ARMS), + 'costs_per_side_pct': list(COSTS_PER_SIDE_PCT), + 'cell_count': len(result_cells), + 'cells': result_cells, + 'analysis': analysis, + 'operational_summary': _operational_summary(result_cells), + } + out_path = ( + Path(args.out) + if args.out + else ROOT + / 'reports' + / f'portfolio-construction-{args.run_id}.json' + ) + _atomic_json(out_path, report) + _atomic_text(out_path.with_suffix('.md'), _markdown(report)) + if not args.quiet: + print(f'wrote {out_path}', flush=True) + print(f'wrote {out_path.with_suffix(".md")}', flush=True) + + +if __name__ == '__main__': + asyncio.run(_main()) diff --git a/scripts/run_research_matrix.py b/scripts/run_research_matrix.py index 2470fa7..d771a4b 100644 --- a/scripts/run_research_matrix.py +++ b/scripts/run_research_matrix.py @@ -68,6 +68,11 @@ ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) +from scripts.research_rankings import ( # noqa: E402 + _live_universe_rank_map, + _period_percentiles, +) + CACHE_VERSION = "research-matrix-v1-daily-prod" # Pre-registered arm catalogue (order is report order). Control is a0. @@ -210,66 +215,6 @@ def _sqlite_url(path: Path) -> str: return f"sqlite+aiosqlite:///{path.resolve().as_posix()}" -def _period_percentiles( - observations: list[dict], value_key: str -) -> dict[tuple[str, str], float]: - by_period: dict[tuple, list[dict]] = {} - for row in observations: - if row.get(value_key) is None: - continue - period = tuple(row["ranking_period"]) - by_period.setdefault(period, []).append(row) - result: dict[tuple[str, str], float] = {} - for group in by_period.values(): - ordered = sorted( - group, key=lambda row: (float(row[value_key]), str(row["symbol"])) - ) - denominator = len(ordered) - 1 - for rank, row in enumerate(ordered): - result[(str(row["symbol"]), str(row["date"]))] = round( - rank / denominator * 100.0 if denominator > 0 else 100.0, - 2, - ) - return result - - -def _live_universe_rank_map( - observations: list[dict], - benchmark_closes: dict[date, float], - momentum_weight: float, -) -> dict[tuple[str, str], dict[str, float | None]]: - raw_pct = _period_percentiles(observations, "momentum") - residual_pct = _period_percentiles(observations, "residual_momentum") - vol_pct = _period_percentiles(observations, "vol_6m") - benchmark_ords = sorted(value.toordinal() for value in benchmark_closes) - residual_start_ord = benchmark_ords[251] if len(benchmark_ords) >= 252 else None - ranks: dict[tuple[str, str], dict[str, float | None]] = {} - for row in observations: - identity = (str(row["symbol"]), str(row["date"])) - asof_ord = date.fromisoformat(identity[1]).toordinal() - momentum_pct = ( - residual_pct.get(identity) - if residual_start_ord is not None and asof_ord >= residual_start_ord - else raw_pct.get(identity) - ) - volatility_pct = vol_pct.get(identity) - strategy_rank = ( - round( - momentum_pct * momentum_weight - + volatility_pct * (1.0 - momentum_weight), - 2, - ) - if momentum_pct is not None and volatility_pct is not None - else momentum_pct - ) - ranks[identity] = { - "momentum_percentile": momentum_pct, - "volatility_percentile": volatility_pct, - "strategy_rank": strategy_rank, - } - return ranks - - def _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description=__doc__, diff --git a/tests/unit/test_portfolio_capacity_research.py b/tests/unit/test_portfolio_capacity_research.py new file mode 100644 index 0000000..53fc6b7 --- /dev/null +++ b/tests/unit/test_portfolio_capacity_research.py @@ -0,0 +1,595 @@ +from __future__ import annotations + +from datetime import date, timedelta + +import pytest + +from app.services import backtest_service as bt +from scripts.portfolio_capacity_research import ( + ANCHOR_YEARS, + aggregate_results, + bootstrap_median_interval, + build_cells, + build_cohort_manifest, + summarize_simulation, + validate_cohort_manifest, +) +from scripts.run_portfolio_construction_matrix import ( + _assert_clean_worktree, + _checkpoint_state, + _markdown, + _operational_summary, + _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_options_preserve_legacy_control_path(): + 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_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 + + +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_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_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 + markdown = _markdown({ + 'generated_at': '2026-08-05T00:00:00Z', + 'analysis': report, + 'operational_summary': _operational_summary(cells), + }) + assert 'ΔGain-to-Pain' in markdown + assert 'formal promotion gate' in markdown + + +def test_synthetic_worker_matrix_covers_four_arms_protocols_and_costs(): + 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()