dennisthiessen/signal-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

research: add focused portfolio capacity matrix

Browse Source
This commit is contained in:
dennisthiessen
2026-08-05 08:28:30 +02:00
parent 07d864cf64
commit 1ace6688dd
12 changed files with 2893 additions and 236 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/services/backtest_service.py
+436 -36
View File
@@ -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