'''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: clean: list[float] = [] for value in values: if value is None: continue parsed = float(value) if math.isfinite(parsed): clean.append(parsed) q25 = percentile(clean, 0.25) q75 = percentile(clean, 0.75) if q25 is None or q75 is None: return None return q75 - q25 def median(values: Iterable[float | None]) -> float | None: clean = [float(value) for value in values if value is not None] return statistics.median(clean) if clean else None def _safe_ratio(numerator: float | None, denominator: float | None) -> float | None: if numerator is None or denominator is None: return None if abs(denominator) <= 1e-12: return 1.0 if abs(numerator) <= 1e-12 else None return numerator / denominator def _stable_seed(*parts: object) -> int: digest = hashlib.sha256('|'.join(map(str, parts)).encode('utf-8')).digest() return BOOTSTRAP_SEED + int.from_bytes(digest[:4], 'big') def bootstrap_median_interval( values: Iterable[float | None], *, seed_parts: tuple[object, ...], replicates: int = BOOTSTRAP_REPLICATES, ) -> dict[str, float | int | None]: clean = [float(value) for value in values if value is not None] if not clean: return {'n': 0, 'point': None, 'p05': None, 'p95': None} rng = random.Random(_stable_seed(*seed_parts)) draws = [ statistics.median(rng.choices(clean, k=len(clean))) for _ in range(replicates) ] return { 'n': len(clean), 'replicates': replicates, 'point': statistics.median(clean), 'p05': percentile(draws, 0.05), 'p95': percentile(draws, 0.95), } def _monthly_returns( equity_curve: list[dict[str, Any]], base_equity: float ) -> list[float]: month_ends: dict[tuple[int, int], float] = {} for point in equity_curve: point_date = date.fromisoformat(str(point['date'])) month_ends[(point_date.year, point_date.month)] = float(point['equity']) previous = float(base_equity) returns: list[float] = [] for month in sorted(month_ends): equity = month_ends[month] if previous > 0: returns.append(equity / previous - 1.0) previous = equity return returns def _time_underwater(equities: list[float]) -> tuple[int, float]: peak = float('-inf') current = 0 longest = 0 underwater = 0 for equity in equities: peak = max(peak, equity) if peak > 0 and equity < peak - 1e-9: current += 1 underwater += 1 longest = max(longest, current) else: current = 0 percentage = underwater / len(equities) * 100.0 if equities else 0.0 return longest, percentage def summarize_simulation(sim: dict[str, Any]) -> dict[str, Any]: trades = list(sim.get('trade_details') or []) equity_curve = list(sim.get('equity_curve') or []) net_rs = [float(trade['net_r']) for trade in trades] positive_rs = [value for value in net_rs if value > 0] negative_rs = [value for value in net_rs if value < 0] ev_net_r = statistics.fmean(net_rs) if net_rs else None profit_factor = ( sum(positive_rs) / abs(sum(negative_rs)) if negative_rs else None ) base_equity = float( sim.get('measurement_start_equity') or sim.get('starting_capital') or 0.0 ) curve_equities = [float(point['equity']) for point in equity_curve] daily_equities = [base_equity, *curve_equities] daily_returns = [ current / previous - 1.0 for previous, current in zip(daily_equities, daily_equities[1:]) if previous > 0 ] downside_deviation = ( math.sqrt( statistics.fmean(min(value, 0.0) ** 2 for value in daily_returns) ) if daily_returns else None ) sortino = ( statistics.fmean(daily_returns) / downside_deviation * math.sqrt(252.0) if downside_deviation is not None and downside_deviation > 0 else None ) monthly_returns = _monthly_returns(equity_curve, base_equity) negative_monthly = sum(value for value in monthly_returns if value < 0) gain_to_pain = ( sum(monthly_returns) / abs(negative_monthly) if negative_monthly < 0 else None ) longest_underwater, underwater_pct = _time_underwater(daily_equities) transaction_cost = sum( float(trade.get('transaction_cost') or 0.0) for trade in trades ) traded_notional = sum( float(trade.get('shares') or 0.0) * (float(trade.get('entry') or 0.0) + float(trade.get('fill') or 0.0)) for trade in trades ) turnover_multiple = ( traded_notional / base_equity if base_equity > 0 else None ) ordered_rs = sorted(net_rs, reverse=True) ev_without_best: dict[str, float | None] = {} for count in (1, 5, 10): remaining = ordered_rs[count:] ev_without_best[str(count)] = ( statistics.fmean(remaining) if remaining else None ) events = list(sim.get('weekly_rebalance_events') or []) entrant_sizes = [int(event['fresh_entrant_pool']) for event in events] eligible_sizes = [ int(event['rank_eligible_entrant_pool']) for event in events ] replacements = [int(event['replacements']) for event in events] capacity_skips = int( sim.get('measurement_skipped_book_full', sim.get('skipped_book_full', 0)) ) opened = int(sim.get('opened_positions', sim.get('trades', 0))) capacity_opportunities = opened + capacity_skips result = { 'start_date': sim.get('start_date'), 'end_date': sim.get('end_date'), 'simulation_start_date': sim.get('simulation_start_date'), 'measurement_start_equity': base_equity, 'measurement_start_positions': sim.get('measurement_start_positions', 0), 'trades': len(trades), 'ev_net_r': ev_net_r, 'profit_factor': profit_factor, 'gain_to_pain': gain_to_pain, 'sortino': sortino, 'ev_without_best': ev_without_best, 'total_return_pct': sim.get('total_return_pct'), 'cagr_pct': sim.get('cagr_pct'), 'max_drawdown_pct': sim.get('max_drawdown_pct'), 'calmar': sim.get('calmar'), 'sharpe': sim.get('sharpe'), 'win_rate': sim.get('win_rate'), 'avg_hold_days': sim.get('avg_hold_days'), 'longest_underwater_sessions': longest_underwater, 'underwater_pct': underwater_pct, 'transaction_cost': transaction_cost, 'turnover_multiple': turnover_multiple, 'skipped_book_full': capacity_skips, 'opened_positions': opened, 'capacity_opportunities': capacity_opportunities, 'blocked_fraction': ( capacity_skips / capacity_opportunities if capacity_opportunities else 0.0 ), 'skipped_min_initial_risk': int( sim.get('measurement_skipped_min_initial_risk', 0) ), 'avg_positions': sim.get('avg_positions'), 'peak_positions': sim.get('peak_positions'), 'sessions_at_capacity': sim.get('sessions_at_capacity'), 'sessions_measured': sim.get('sessions_measured'), 'avg_cash_pct': sim.get('avg_cash_pct'), 'avg_gross_exposure_pct': sim.get('avg_gross_exposure_pct'), 'exit_reasons': sim.get('exit_reasons'), } if events: result['weekly_rebalance'] = { 'events': len(events), 'zero_entrant_fraction': ( sum(1 for value in entrant_sizes if value == 0) / len(events) ), 'entrant_pool_mean': statistics.fmean(entrant_sizes), 'entrant_pool_median': statistics.median(entrant_sizes), 'entrant_pool_p90': percentile(entrant_sizes, 0.9), 'eligible_pool_mean': statistics.fmean(eligible_sizes), 'replacements': sum(replacements), 'weekly_rank_rejected_entries': int( sim.get('weekly_rank_rejected_entries', 0) ), 'reentries_within_5_sessions': int( sim.get('rebalance_reentries_within_5_sessions', 0) ), 'reentries_within_10_sessions': int( sim.get('rebalance_reentries_within_10_sessions', 0) ), 'reentries_within_20_sessions': int( sim.get('rebalance_reentries_within_20_sessions', 0) ), } return result def _cluster_rows( cells: list[dict[str, Any]], *, arm_id: str, protocol: str, cost: float, ) -> list[dict[str, Any]]: treatment = { row['path_id']: row for row in cells if row['arm_id'] == arm_id and row['protocol'] == protocol and float(row['cost_per_side_pct']) == cost } control = { row['path_id']: row for row in cells if row['arm_id'] == 'cap10_incumbent' and row['protocol'] == protocol and float(row['cost_per_side_pct']) == cost } shared_paths = sorted(set(treatment) & set(control)) by_cluster: dict[int, list[tuple[dict, dict]]] = defaultdict(list) for path_id in shared_paths: row = treatment[path_id] by_cluster[int(row['cluster'])].append((row, control[path_id])) summaries: list[dict[str, Any]] = [] for cluster, pairs in sorted(by_cluster.items()): metrics: dict[str, Any] = {} for metric in PAIRED_METRICS: arm_values = [ pair[0]['metrics'].get(metric) for pair in pairs if pair[0]['metrics'].get(metric) is not None and math.isfinite(float(pair[0]['metrics'][metric])) ] control_values = [ pair[1]['metrics'].get(metric) for pair in pairs if pair[1]['metrics'].get(metric) is not None and math.isfinite(float(pair[1]['metrics'][metric])) ] deltas = [ float(arm['metrics'][metric]) - float(base['metrics'][metric]) for arm, base in pairs if arm['metrics'].get(metric) is not None and base['metrics'].get(metric) is not None and math.isfinite(float(arm['metrics'][metric])) and math.isfinite(float(base['metrics'][metric])) ] arm_median = median(arm_values) control_median = median(control_values) metrics[metric] = { 'arm_median': arm_median, 'control_median': control_median, 'paired_delta_median': median(deltas), 'arm_control_ratio': _safe_ratio( arm_median, control_median ), 'paired_paths': len(deltas), } summaries.append({ 'cluster': cluster, 'paths': len(pairs), 'metrics': metrics, }) return summaries def aggregate_results(cells: list[dict[str, Any]]) -> 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', }, }