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

research: prepare effective risk floor ab

Browse Source
This commit is contained in:
dennisthiessen
2026-08-05 22:49:50 +02:00
parent aa6cd5cac4
commit 6f1ee450f1
5 changed files with 573 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/portfolio_capacity_research.py
+102 -10
View File
@@ -43,6 +43,16 @@ ARMS: tuple[dict[str, Any], ...] = (
)
ARM_BY_ID = {arm['id']: arm for arm in ARMS}
RISK_FLOOR_ARMS: tuple[dict[str, Any], ...] = (
ARMS[0],
{
'id': 'cap10_min_risk_005',
'label': 'Cap 10, 0.5% minimum effective initial risk',
'max_positions': 10,
'min_initial_risk_fraction': 0.005,
'weekly_top_n_rebalance': False,
},
)
COSTS_PER_SIDE_PCT = (0.1, 0.2)
ANCHOR_YEARS = tuple(range(2019, 2026))
SCORING_SESSIONS = 504
@@ -190,12 +200,22 @@ def validate_cohort_manifest(manifest: dict[str, Any]) -> list[str]:
return errors
def build_cells(manifest: dict[str, Any]) -> list[dict[str, Any]]:
paths = [*manifest['empty_book'], *manifest['warm_book']]
def build_cells(
manifest: dict[str, Any],
*,
arms: tuple[dict[str, Any], ...] = ARMS,
protocols: tuple[str, ...] = ('empty_book', 'warm_book'),
costs: tuple[float, ...] = COSTS_PER_SIDE_PCT,
) -> list[dict[str, Any]]:
paths = [
path
for protocol in protocols
for path in manifest[protocol]
]
cells: list[dict[str, Any]] = []
for cost in COSTS_PER_SIDE_PCT:
for cost in costs:
for path in paths:
for arm in ARMS:
for arm in arms:
cell_id = (
f'{arm["id"]}|{path["protocol"]}|{path["path_id"]}'
f'|cost={cost:.1f}'
@@ -531,11 +551,26 @@ def _cluster_rows(
return summaries
def aggregate_results(cells: list[dict[str, Any]]) -> dict[str, Any]:
def aggregate_results(
cells: list[dict[str, Any]],
*,
arms: tuple[dict[str, Any], ...] = ARMS,
protocols: tuple[str, ...] = ('empty_book', 'warm_book'),
costs: tuple[float, ...] = COSTS_PER_SIDE_PCT,
include_warm_dispersion: bool = True,
) -> dict[str, Any]:
paired: list[dict[str, Any]] = []
for cost in COSTS_PER_SIDE_PCT:
for protocol in ('empty_book', 'warm_book'):
for arm in ARMS:
path_distributions: list[dict[str, Any]] = []
for cost in costs:
for protocol in protocols:
control_by_path = {
row['path_id']: row
for row in cells
if row['arm_id'] == 'cap10_incumbent'
and row['protocol'] == protocol
and float(row['cost_per_side_pct']) == float(cost)
}
for arm in arms:
arm_id = str(arm['id'])
clusters = _cluster_rows(
cells,
@@ -588,13 +623,66 @@ def aggregate_results(cells: list[dict[str, Any]]) -> dict[str, Any]:
'clusters': clusters,
'headline': headline,
})
treatment_by_path = {
row['path_id']: row
for row in cells
if row['arm_id'] == arm_id
and row['protocol'] == protocol
and float(row['cost_per_side_pct']) == float(cost)
}
shared_paths = sorted(
set(treatment_by_path) & set(control_by_path)
)
path_metrics: dict[str, Any] = {}
for metric in PAIRED_METRICS:
deltas = [
float(treatment_by_path[path_id]['metrics'][metric])
- float(control_by_path[path_id]['metrics'][metric])
for path_id in shared_paths
if treatment_by_path[path_id]['metrics'].get(metric)
is not None
and control_by_path[path_id]['metrics'].get(metric)
is not None
and math.isfinite(
float(treatment_by_path[path_id]['metrics'][metric])
)
and math.isfinite(
float(control_by_path[path_id]['metrics'][metric])
)
]
path_metrics[metric] = {
'paired_paths': len(deltas),
'paired_delta_mean': (
statistics.fmean(deltas) if deltas else None
),
'paired_delta_median': median(deltas),
'paired_delta_p25': percentile(deltas, 0.25),
'paired_delta_p75': percentile(deltas, 0.75),
'positive_fraction': (
sum(delta > 0.0 for delta in deltas) / len(deltas)
if deltas
else None
),
'identical_fraction': (
sum(abs(delta) <= 1e-12 for delta in deltas)
/ len(deltas)
if deltas
else None
),
}
path_distributions.append({
'arm_id': arm_id,
'protocol': protocol,
'cost_per_side_pct': cost,
'metrics': path_metrics,
})
warm_rows = [
row for row in cells if row['protocol'] == 'warm_book'
]
warm_dispersion: list[dict[str, Any]] = []
for cost in COSTS_PER_SIDE_PCT:
for arm in ARMS:
for cost in costs:
for arm in arms:
arm_id = str(arm['id'])
anchor_rows: list[dict[str, Any]] = []
for cluster in ANCHOR_YEARS:
@@ -660,8 +748,12 @@ def aggregate_results(cells: list[dict[str, Any]]) -> dict[str, Any]:
'headline': headline,
})
if not include_warm_dispersion:
warm_dispersion = []
return {
'paired_per_year': paired,
'paired_path_distributions': path_distributions,
'warm_seed_dispersion': warm_dispersion,
'bootstrap': {
'replicates': BOOTSTRAP_REPLICATES,