research: prepare effective risk floor ab

docs/research/README.md

@@ -146,7 +146,7 @@ knobs.
 | **Broader universe** | Composition changes factor signs (fip tug-of-war); vol-tilt on breadth is only a **directional hypothesis** (auth. −0.048 / t −1.36) | Any prod broaden must re-validate 80/20 tilt; offline research only; research.sqlite requires completion manifest |
 | **Forward paper-trade record** | The only true out-of-sample evidence the snapshot cannot give | Time; mark entries at actual near-close fill once ops ships |
 | **Better target model for clear-air names** | The return is demonstrably there (#2 wins on raw CAGR in *both* train and test); it's the *flat* 3× ATR target that makes it too expensive in risk | Needs a per-name model, not a constant k×ATR |
-| **Minimum effective-risk floor** | In cap-never-bound paths, the confounded 0.5% floor arm removed about 8% of fills while EV rose from 0.328 to 0.399 R and PF from 1.60 to 1.75, with exposure nearly unchanged | Run a single-variable A/B: cap 10 control versus cap 10 plus `min_initial_risk_fraction=0.005`. [Capacity findings](portfolio-capacity-bracket-findings.md) |
+| **Minimum effective-risk floor** | In cap-never-bound paths, the confounded 0.5% floor arm removed about 8% of fills while EV rose from 0.328 to 0.399 R and PF from 1.60 to 1.75, with exposure nearly unchanged | Run the frozen single-variable cap-10 A/B. [Specification](effective-risk-floor-ab.md) / [capacity findings](portfolio-capacity-bracket-findings.md) |
 
 ---
 
@@ -206,7 +206,7 @@ only 0.0018 R/trade in paths where cap 10 bound. Weekly current-rank replacement
 reduced mean EV and created substantial churn. Keep cap 10 and do not build the
 replacement policy. See the [frozen specification](portfolio-capacity-bracket.md)
 and the separate [capacity findings](portfolio-capacity-bracket-findings.md).
-The only open follow-up from that run is the confound-free 0.5% minimum
+The only open follow-up from that run is the
-effective-risk-floor A/B.
+[frozen confound-free 0.5% minimum effective-risk-floor A/B](effective-risk-floor-ab.md).
 
 The next real evidence is **forward**, not backward: the live paper-trade record.

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

@@ -0,0 +1,124 @@
+# Effective initial-risk floor A/B - frozen specification
+
+Date frozen: 2026-08-05
+Branch: research/portfolio-capacity-rebalancing
+Runner: scripts/run_portfolio_construction_matrix.py
+Study ID: risk-floor-ab
+
+## Question
+
+Does rejecting an otherwise qualified cap-10 entry when its actual initial
+stop-risk after cash and notional sizing is below 0.5% of marked equity improve
+trade selection?
+
+The completed capacity bracket cannot answer this. Its cash_unbounded arm
+removed the count cap and applied the 0.5% floor simultaneously. In the 70 paths
+where the control cap never bound, that arm still raised mean EV from 0.328 to
+0.399 R and profit factor from 1.60 to 1.75 while trades fell about 8% and
+exposure stayed nearly flat. Capacity was a no-op in those paths, so the floor
+is the plausible cause, but the prior arm remains confounded.
+
+This A/B changes only the floor. It has no formal promotion gate and does not
+automatically change production.
+
+## Frozen arms
+
+1. cap10_incumbent: current production-style cap-10 control, with no minimum
+   effective-risk floor.
+2. cap10_min_risk_005: the same cap-10 strategy, rejecting an entry only when
+   actual initial stop-risk after cash/notional sizing is below 0.5% of marked
+   equity.
+
+Both arms have max_positions=10, weekly replacement disabled, 1% target risk
+per trade, and identical admission ordering. The only differing simulator
+argument is min_initial_risk_fraction: None versus 0.005.
+
+All other settings remain the frozen daily Phase A control: current production
+construction universe, full-universe residual-momentum/low-volatility 80/20
+rank, threshold 80, normal gate-reset re-entry, close fills, 3x ATR trail,
+30-session maximum hold, 20% per-position notional ceiling, no leverage, and
+costs of 0.10% and 0.20% per fill.
+
+Every priced symbol contributes to the daily cross-sectional rank. Rank-only
+symbols cannot submit trades. Validation retains the 450-600-symbol production
+construction guardrail and the legacy-snapshot column-scoped loader.
+
+## Frozen cohorts
+
+Reuse the completed bracket's point-in-time daily candidate/rank cache and
+cohort manifest:
+
+- Empty book: first eligible session of each month in 2019-2025, with 504 prior
+  scoring sessions and 252 measurement sessions. This is the primary start-date
+  evidence.
+- Warm book: weekly seeds 63-126 sessions before each 2019-2025 annual anchor,
+  with state carried into the same 252-session measurement window. This is a
+  state-carrying replication, not independent evidence.
+
+The expected realization is 78 empty-book paths, 97 warm paths, seven annual
+clusters in each protocol, two costs, two arms, and 700 cells.
+
+Do not use warm-seed IQR as evidence. Six of seven completed-bracket anchors
+were structurally degenerate because fractional sizing is scale invariant and
+the 30-session maximum hold washed out books before anchors. The 2023 exception
+shows that state carrying itself works.
+
+## Reporting and interpretation
+
+For every protocol and cost, pair identical paths. Report:
+
+- mean, median, P25, and P75 paired net-EV changes in R;
+- positive-path and bit-identical-path fractions;
+- the median paired delta within each year and the median across seven years;
+- simple 90% cluster-bootstrap context for EV and Calmar, with no CI gate;
+- mean paired PF, Gain-to-Pain, Sortino, Calmar/MAR, CAGR, maximum drawdown,
+  total return, and Sharpe changes;
+- trades, floor rejections, holding time, cash, gross exposure, average/peak
+  positions, turnover, and costs.
+
+Means and identical-path fractions must appear beside medians so inert cohorts
+cannot turn a left- or right-skewed treatment into a misleading zero headline.
+For these 252-session windows, the implementation's full-window Calmar is CAGR
+divided by maximum drawdown, the same numeric definition commonly called MAR;
+do not present the duplicate label as a second independent metric.
+
+Today's production membership is projected backward. Use paired differences
+for the treatment conclusion; absolute profitability remains descriptive and
+survivorship-biased. Empty and warm protocols cover the same seven market years
+and must not be interpreted as independent replications.
+
+Interpretation is deliberately simple:
+
+- a positive result means the isolated floor improves the paired EV
+  distribution without an economically important loss of total-return or
+  drawdown quality;
+- a negative result closes the floor;
+- mixed EV/portfolio-quality results are reported as a trade-off, not forced
+  through a composite score.
+
+## Reproducibility and macOS execution
+
+The authoritative run refuses a dirty worktree. Its fingerprint includes the
+implementation commit, this specification hash, snapshot hash, candidate-cache
+key, construction view, cohort manifest, arm definitions, costs, and study
+version. Cells checkpoint atomically and --resume verifies the fingerprint.
+
+From the repository root on macOS:
+
+    python3 -m venv .venv
+    ./.venv/bin/python -m pip install -e '.[dev]'
+
+Preflight, reusing the completed bracket's candidate/rank cache:
+
+    ./.venv/bin/python scripts/run_portfolio_construction_matrix.py +      backtest_snapshots/research.sqlite +      --study risk-floor-ab +      --run-id prod505-effective-risk-floor-ab-daily-v1 +      --candidate-cache reports/.cache/prod505-capacity-bracket-daily-v1-candidates.pkl +      --workers 8 +      --resume +      --validate-only
+
+Authoritative run:
+
+    ./.venv/bin/python scripts/run_portfolio_construction_matrix.py +      backtest_snapshots/research.sqlite +      --study risk-floor-ab +      --run-id prod505-effective-risk-floor-ab-daily-v1 +      --candidate-cache reports/.cache/prod505-capacity-bracket-daily-v1-candidates.pkl +      --workers 8 +      --resume
+
+On an M2 Pro, eight workers is the explicit high-utilization setting. Use six
+instead on a memory-constrained machine; auto intentionally caps itself at six.
+Changing worker count does not change the fingerprint or results.
+
+Commit only the compact final JSON and Markdown reports. Candidate caches,
+checkpoints, raw curves, and trade ledgers remain ignored.

scripts/portfolio_capacity_research.py

@@ -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]]:
+def build_cells(
-    paths = [*manifest['empty_book'], *manifest['warm_book']]
+    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:
+    path_distributions: list[dict[str, Any]] = []
-        for protocol in ('empty_book', 'warm_book'):
+    for cost in costs:
-            for arm in ARMS:
+        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 cost in costs:
-        for arm in ARMS:
+        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,

scripts/run_portfolio_construction_matrix.py

@@ -1,8 +1,8 @@
-'''Run the focused four-arm daily portfolio-capacity research matrix.
+'''Run focused daily portfolio-construction research matrices.
 
 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
+once and reused by the frozen capacity bracket and its focused effective-risk
-under cap 10, cap 15, cash-only unbounded, and weekly current-rank top 10.
+floor follow-up.
 '''
 
 from __future__ import annotations
@@ -37,6 +37,7 @@ from scripts.portfolio_capacity_research import (  # noqa: E402
     BOOTSTRAP_REPLICATES,
     BOOTSTRAP_SEED,
     COSTS_PER_SIDE_PCT,
+    RISK_FLOOR_ARMS,
     aggregate_results,
     build_cells,
     build_cohort_manifest,
@@ -54,13 +55,76 @@ MIN_PRODUCTION_UNIVERSE = 450
 MAX_PRODUCTION_UNIVERSE = 600
 SPEC_PATH = ROOT / 'docs' / 'research' / 'portfolio-capacity-bracket.md'
 DEFAULT_RUN_ID = 'prod505-capacity-bracket-daily-v1'
+RISK_FLOOR_SPEC_PATH = (
+    ROOT / 'docs' / 'research' / 'effective-risk-floor-ab.md'
+)
+STUDIES: dict[str, dict[str, Any]] = {
+    'capacity-bracket': {
+        'arms': ARMS,
+        'protocols': ('empty_book', 'warm_book'),
+        'include_warm_dispersion': True,
+        'runner_version': RUNNER_VERSION,
+        'spec_path': SPEC_PATH,
+        'default_run_id': DEFAULT_RUN_ID,
+        '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.'
+        ),
+        '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.'
+            ),
+        },
+    },
+    'risk-floor-ab': {
+        'arms': RISK_FLOOR_ARMS,
+        'protocols': ('empty_book', 'warm_book'),
+        'include_warm_dispersion': False,
+        'runner_version': 'effective-risk-floor-ab-v1',
+        'spec_path': RISK_FLOOR_SPEC_PATH,
+        'default_run_id': 'prod505-effective-risk-floor-ab-daily-v1',
+        'research_question': (
+            'Estimate the isolated effect of rejecting entries whose effective '
+            'initial stop risk is below 0.5% of marked equity.'
+        ),
+        'decision_rule': (
+            'No formal promotion gate. Attribute paired differences causally to '
+            'the floor, headline means and identical-path fractions beside '
+            'annual medians, and decide paper adoption after interpretation.'
+        ),
+        'motivation': {
+            'source': (
+                'portfolio-construction-prod505-capacity-bracket-daily-v1 '
+                'post-run decomposition'
+            ),
+            'finding': (
+                'The confounded floor arm improved EV most where the position '
+                'cap never bound; isolate the 0.5% floor at cap 10.'
+            ),
+        },
+    },
+}
 _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(
+        '--study',
+        choices=tuple(STUDIES),
+        default='capacity-bracket',
+    )
+    parser.add_argument('--run-id', default=None)
     parser.add_argument(
         '--workers',
         default='auto',
@@ -235,7 +299,7 @@ def _worker_run_cell(cell: dict[str, Any]) -> dict[str, Any]:
     from app.services import backtest_service as bt
 
     context = _WORKER_CONTEXT
-    arm = ARM_BY_ID[str(cell['arm_id'])]
+    arm = context.get('arm_by_id', 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(
@@ -600,6 +664,7 @@ def _checkpoint_state(
     fingerprint: str,
     *,
     resume: bool,
+    runner_version: str = RUNNER_VERSION,
 ) -> dict[str, dict[str, Any]]:
     manifest_path = checkpoint_dir / 'manifest.json'
     if checkpoint_dir.exists() and not resume:
@@ -620,7 +685,7 @@ def _checkpoint_state(
         _atomic_json(
             manifest_path,
             {
-                'runner_version': RUNNER_VERSION,
+                'runner_version': runner_version,
                 'fingerprint': fingerprint,
                 'created_at': datetime.now(timezone.utc).isoformat(),
             },
@@ -650,9 +715,13 @@ def _fmt(value: Any, digits: int = 3) -> str:
     return str(value)
 
 
-def _operational_summary(cells: list[dict[str, Any]]) -> list[dict[str, Any]]:
+def _operational_summary(
+    cells: list[dict[str, Any]],
+    *,
+    arms: tuple[dict[str, Any], ...] = ARMS,
+) -> list[dict[str, Any]]:
     rows: list[dict[str, Any]] = []
-    for arm in ARMS:
+    for arm in arms:
         arm_cells = [
             row
             for row in cells
@@ -676,6 +745,16 @@ def _operational_summary(cells: list[dict[str, Any]]) -> list[dict[str, Any]]:
             'median_avg_positions': median(
                 row['metrics'].get('avg_positions') for row in arm_cells
             ),
+            'median_avg_hold_days': median(
+                row['metrics'].get('avg_hold_days') for row in arm_cells
+            ),
+            'median_avg_cash_pct': median(
+                row['metrics'].get('avg_cash_pct') for row in arm_cells
+            ),
+            'median_avg_gross_exposure_pct': median(
+                row['metrics'].get('avg_gross_exposure_pct')
+                for row in arm_cells
+            ),
             'peak_positions': max(
                 (
                     int(row['metrics'].get('peak_positions') or 0)
@@ -865,19 +944,158 @@ def _markdown(report: dict[str, Any]) -> str:
     return '\n'.join(lines)
 
 
+def _risk_floor_markdown(report: dict[str, Any]) -> str:
+    treatment_id = str(report['arms'][1]['id'])
+    lines = [
+        '# Effective initial-risk floor A/B',
+        '',
+        'Generated: ' + str(report['generated_at']),
+        '',
+        '## Question',
+        '',
+        'Does rejecting an otherwise qualified cap-10 entry when actual initial '
+        'stop risk is below 0.5% of marked equity improve trade selection?',
+        '',
+        'The arms differ only by min_initial_risk_fraction=0.005. There is no '
+        'position-cap, ranking, exit, sizing-target, or execution confound.',
+        '',
+        'Use paired differences, not absolute profitability, because current '
+        'production membership is projected backward.',
+        '',
+    ]
+    paired = report['analysis']['paired_per_year']
+    distributions = report['analysis']['paired_path_distributions']
+    for cost in report['costs_per_side_pct']:
+        for protocol in report['protocols']:
+            annual = next(
+                row for row in paired
+                if row['arm_id'] == treatment_id
+                and row['protocol'] == protocol
+                and float(row['cost_per_side_pct']) == float(cost)
+            )
+            path = next(
+                row for row in distributions
+                if row['arm_id'] == treatment_id
+                and row['protocol'] == protocol
+                and float(row['cost_per_side_pct']) == float(cost)
+            )
+            metrics = path['metrics']
+            ev = metrics['ev_net_r']
+            annual_ev = annual['headline']['ev_net_r']
+            ev_ci = annual_ev['bootstrap_90']
+            separator = chr(124)
+            lines.append(
+                '## ' + protocol.replace('_', ' ').title()
+                + f' at {float(cost):.2f}% per fill'
+            )
+            lines.extend([
+                '',
+                separator + ' Paths ' + separator + ' Mean dEV '
+                + separator + ' Median dEV ' + separator + ' P25 '
+                + separator + ' P75 ' + separator + ' Positive '
+                + separator + ' Identical ' + separator + ' Annual median '
+                + separator + ' 90% context ' + separator,
+                separator.join(
+                    ['', '---:', '---:', '---:', '---:', '---:', '---:',
+                     '---:', '---:', '---:', '']
+                ),
+                separator + ' ' + str(ev['paired_paths']) + ' '
+                + separator + ' ' + _fmt(ev['paired_delta_mean']) + ' '
+                + separator + ' ' + _fmt(ev['paired_delta_median']) + ' '
+                + separator + ' ' + _fmt(ev['paired_delta_p25']) + ' '
+                + separator + ' ' + _fmt(ev['paired_delta_p75']) + ' '
+                + separator + ' '
+                + _fmt(ev['positive_fraction'] * 100.0, 1) + '% '
+                + separator + ' '
+                + _fmt(ev['identical_fraction'] * 100.0, 1) + '% '
+                + separator + ' '
+                + _fmt(annual_ev['paired_delta_median']) + ' '
+                + separator + ' [' + _fmt(ev_ci['p05']) + ', '
+                + _fmt(ev_ci['p95']) + '] '
+                + separator,
+                '',
+                separator + ' Mean dPF ' + separator + ' Mean dGtP '
+                + separator + ' Mean dSortino ' + separator
+                + ' Mean dCalmar/MAR '
+                + separator + ' Mean dCAGR pp ' + separator
+                + ' Mean dMaxDD pp ' + separator,
+                separator.join(
+                    ['', '---:', '---:', '---:', '---:', '---:', '---:', '']
+                ),
+                separator
+                + ' ' + _fmt(
+                    metrics['profit_factor']['paired_delta_mean']
+                ) + ' '
+                + separator
+                + ' ' + _fmt(
+                    metrics['gain_to_pain']['paired_delta_mean']
+                ) + ' '
+                + separator
+                + ' ' + _fmt(metrics['sortino']['paired_delta_mean']) + ' '
+                + separator
+                + ' ' + _fmt(metrics['calmar']['paired_delta_mean']) + ' '
+                + separator
+                + ' ' + _fmt(metrics['cagr_pct']['paired_delta_mean']) + ' '
+                + separator
+                + ' ' + _fmt(
+                    metrics['max_drawdown_pct']['paired_delta_mean']
+                ) + ' '
+                + separator,
+                '',
+            ])
+    lines.extend([
+        '## Operations at 0.10% per fill',
+        '',
+        separator + ' Arm ' + separator + ' Trades ' + separator + ' Hold '
+        + separator + ' Cash ' + separator + ' Gross ' + separator
+        + ' Positions ' + separator + ' Floor rejects ' + separator,
+        separator.join(
+            ['', '---', '---:', '---:', '---:', '---:', '---:', '---:', '']
+        ),
+    ])
+    for row in report['operational_summary']:
+        lines.append(
+            separator + ' ' + str(row['arm_id']) + ' '
+            + separator + ' ' + _fmt(row['median_trades'], 1) + ' '
+            + separator + ' ' + _fmt(row['median_avg_hold_days'], 1) + ' '
+            + separator + ' ' + _fmt(row['median_avg_cash_pct'], 1) + '% '
+            + separator
+            + ' ' + _fmt(row['median_avg_gross_exposure_pct'], 1) + '% '
+            + separator + ' ' + _fmt(row['median_avg_positions'], 2) + ' '
+            + separator + ' ' + str(row['min_risk_rejections']) + ' '
+            + separator
+        )
+    lines.extend([
+        '',
+        'Empty-book starts are primary. Warm-book paths are a state-carrying '
+        'replication over the same seven years, not independent evidence or an '
+        'initialization-dispersion test.',
+        '',
+        'The 90% intervals resample seven annual paired summaries. They are '
+        'descriptive context, not gates or population-confidence claims.',
+    ])
+    return '\n'.join(lines)
+
+
 async def _main() -> None:
     args = _parse_args()
+    study = STUDIES[str(args.study)]
+    args.run_id = args.run_id or study['default_run_id']
+    arms = tuple(study['arms'])
+    protocols = tuple(study['protocols'])
+    runner_version = str(study['runner_version'])
+    spec_path = Path(study['spec_path'])
     snapshot = Path(args.snapshot)
     if not snapshot.exists():
         raise SystemExit(f'Snapshot does not exist: {snapshot}')
-    if not SPEC_PATH.exists():
+    if not spec_path.exists():
-        raise SystemExit(f'Frozen specification is missing: {SPEC_PATH}')
+        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)
+    specification_sha256 = _sha256_file(spec_path)
 
     snapshot_data = await _load_snapshot(snapshot, quiet=bool(args.quiet))
     cache_path = (
@@ -906,9 +1124,14 @@ async def _main() -> None:
         snapshot_data['construction_universe_manifest']
     )
     validation_errors = [*universe_errors, *cohort_errors]
-    cells = build_cells(cohort_manifest)
+    cells = build_cells(
+        cohort_manifest,
+        arms=arms,
+        protocols=protocols,
+    )
     validation_payload = {
-        'runner_version': RUNNER_VERSION,
+        'study': args.study,
+        'runner_version': runner_version,
         'snapshot': str(snapshot.resolve()),
         'snapshot_sha256': snapshot_sha256,
         'snapshot_sessions': {
@@ -939,6 +1162,8 @@ async def _main() -> None:
         'empty_cluster_count': cohort_manifest['empty_cluster_count'],
         'warm_cluster_count': cohort_manifest['warm_cluster_count'],
         'expected_clusters': list(ANCHOR_YEARS),
+        'protocols': list(protocols),
+        'arms': [arm['id'] for arm in arms],
         'matrix_cells': len(cells),
         'cache_path': str(cache_path.resolve()),
         'cache_key_hash': _json_hash(candidate_cache['key']),
@@ -958,7 +1183,8 @@ async def _main() -> None:
     _assert_clean_worktree()
     git_commit = _git_output('rev-parse', 'HEAD')
     fingerprint_payload = {
-        'runner_version': RUNNER_VERSION,
+        'study': args.study,
+        'runner_version': runner_version,
         'git_commit': git_commit,
         'snapshot_sha256': snapshot_sha256,
         'specification_sha256': specification_sha256,
@@ -969,7 +1195,9 @@ async def _main() -> None:
             snapshot_data['construction_universe_manifest']
         ),
         'cohort_manifest': cohort_manifest,
-        'arms': list(ARMS),
+        'arms': list(arms),
+        'protocols': list(protocols),
+        'include_warm_dispersion': bool(study['include_warm_dispersion']),
         'costs_per_side_pct': list(COSTS_PER_SIDE_PCT),
         'bootstrap': {
             'replicates': BOOTSTRAP_REPLICATES,
@@ -983,12 +1211,13 @@ async def _main() -> None:
         else ROOT
         / 'reports'
         / '.cache'
-        / f'{args.run_id}-{RUNNER_VERSION}-checkpoint'
+        / f'{args.run_id}-{runner_version}-checkpoint'
     )
     completed = _checkpoint_state(
         checkpoint_dir,
         fingerprint,
         resume=bool(args.resume),
+        runner_version=runner_version,
     )
     expected_ids = {str(cell['cell_id']) for cell in cells}
     unknown = set(completed) - expected_ids
@@ -1006,6 +1235,7 @@ async def _main() -> None:
         )
 
     worker_context = {
+        'arm_by_id': {arm['id']: arm for arm in arms},
         'qualified_candidates': candidate_cache['qualified_candidates'],
         'daily_rank_map': candidate_cache['daily_rank_map'],
         'prices': snapshot_data['prices'],
@@ -1060,34 +1290,25 @@ async def _main() -> None:
         completed.values(),
         key=lambda row: str(row['cell_id']),
     )
-    analysis = aggregate_results(result_cells)
+    analysis = aggregate_results(
+        result_cells,
+        arms=arms,
+        protocols=protocols,
+        include_warm_dispersion=bool(study['include_warm_dispersion']),
+    )
     dependency_manifest = ROOT / 'pyproject.toml'
     report: dict[str, Any] = {
         'run_id': args.run_id,
+        'study': args.study,
         'status': 'complete',
         'generated_at': datetime.now(timezone.utc).isoformat(),
-        'research_question': (
+        'research_question': study['research_question'],
-            'Bracket the economic cost of the binding ten-position cap and '
+        'decision_rule': study['decision_rule'],
-            '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': {
+        'motivation': dict(study['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': {
@@ -1117,12 +1338,17 @@ async def _main() -> None:
             }
         },
         'cohort_manifest': cohort_manifest,
-        'arms': list(ARMS),
+        'arms': list(arms),
+        'protocols': list(protocols),
+        'include_warm_dispersion': bool(study['include_warm_dispersion']),
         '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),
+        'operational_summary': _operational_summary(
+            result_cells,
+            arms=arms,
+        ),
     }
     out_path = (
         Path(args.out)
@@ -1132,7 +1358,12 @@ async def _main() -> None:
         / f'portfolio-construction-{args.run_id}.json'
     )
     _atomic_json(out_path, report)
-    _atomic_text(out_path.with_suffix('.md'), _markdown(report))
+    markdown = (
+        _risk_floor_markdown(report)
+        if args.study == 'risk-floor-ab'
+        else _markdown(report)
+    )
+    _atomic_text(out_path.with_suffix('.md'), markdown)
     if not args.quiet:
         print(f'wrote {out_path}', flush=True)
         print(f'wrote {out_path.with_suffix(".md")}', flush=True)

tests/unit/test_portfolio_capacity_research.py

@@ -10,6 +10,7 @@ import pytest
 from app.services import backtest_service as bt
 from scripts.portfolio_capacity_research import (
     ANCHOR_YEARS,
+    RISK_FLOOR_ARMS,
     aggregate_results,
     bootstrap_median_interval,
     build_cells,
@@ -20,6 +21,7 @@ from scripts.portfolio_capacity_research import (
 )
 from scripts.run_portfolio_construction_matrix import (
     CACHE_VERSION,
+    STUDIES,
     _assert_clean_worktree,
     _build_candidate_cache,
     _checkpoint_state,
@@ -29,6 +31,7 @@ from scripts.run_portfolio_construction_matrix import (
     _load_snapshot,
     _markdown,
     _operational_summary,
+    _risk_floor_markdown,
     _worker_init,
     _worker_run_cell,
     _write_cell_checkpoint,
@@ -538,6 +541,29 @@ def test_cohort_manifest_realizes_seven_frozen_clusters():
     assert len(cells) == (
         len(manifest['empty_book']) + len(manifest['warm_book'])
     ) * 4 * 2
+    floor_cells = build_cells(manifest, arms=RISK_FLOOR_ARMS)
+    assert len(floor_cells) == (
+        len(manifest['empty_book']) + len(manifest['warm_book'])
+    ) * 2 * 2
+    assert {row['arm_id'] for row in floor_cells} == {
+        'cap10_incumbent',
+        'cap10_min_risk_005',
+    }
+
+
+def test_risk_floor_study_changes_only_the_effective_risk_floor():
+    control, treatment = RISK_FLOOR_ARMS
+
+    assert control['max_positions'] == treatment['max_positions'] == 10
+    assert (
+        control['weekly_top_n_rebalance']
+        == treatment['weekly_top_n_rebalance']
+        is False
+    )
+    assert control['min_initial_risk_fraction'] is None
+    assert treatment['min_initial_risk_fraction'] == 0.005
+    assert STUDIES['risk-floor-ab']['arms'] == RISK_FLOOR_ARMS
+    assert STUDIES['capacity-bracket']['arms'] != RISK_FLOOR_ARMS
 
 
 def test_zero_outcome_horizon_extends_rank_replay_to_last_session(monkeypatch):
@@ -700,6 +726,18 @@ def test_aggregate_reports_paired_years_and_separate_warm_iqrs():
     assert cash_empty['headline']['ev_net_r']['paired_delta_median'] == pytest.approx(
         0.2
     )
+    cash_paths = next(
+        row
+        for row in report['paired_path_distributions']
+        if row['arm_id'] == 'cash_unbounded'
+        and row['protocol'] == 'empty_book'
+        and row['cost_per_side_pct'] == 0.1
+    )
+    assert cash_paths['metrics']['ev_net_r']['paired_delta_mean'] == pytest.approx(
+        0.2
+    )
+    assert cash_paths['metrics']['ev_net_r']['positive_fraction'] == 1.0
+    assert cash_paths['metrics']['ev_net_r']['identical_fraction'] == 0.0
     cash_warm = next(
         row
         for row in report['warm_seed_dispersion']
@@ -738,6 +776,43 @@ def test_aggregate_reports_paired_years_and_separate_warm_iqrs():
     assert 'Rank-only qualified rows removed: 137000.' in markdown
     assert 'formal promotion gate' in markdown
 
+    focused_cells = [
+        row
+        for row in cells
+        if row['arm_id'] == 'cap10_incumbent'
+    ] + [
+        {
+            **row,
+            'arm_id': 'cap10_min_risk_005',
+        }
+        for row in cells
+        if row['arm_id'] == 'cash_unbounded'
+    ]
+    focused_analysis = aggregate_results(
+        focused_cells,
+        arms=RISK_FLOOR_ARMS,
+        include_warm_dispersion=False,
+    )
+    assert focused_analysis['warm_seed_dispersion'] == []
+    focused_markdown = _risk_floor_markdown({
+        'generated_at': '2026-08-05T00:00:00Z',
+        'arms': list(RISK_FLOOR_ARMS),
+        'protocols': ['empty_book', 'warm_book'],
+        'costs_per_side_pct': [0.1, 0.2],
+        'analysis': focused_analysis,
+        'operational_summary': _operational_summary(
+            focused_cells,
+            arms=RISK_FLOOR_ARMS,
+        ),
+    })
+    assert '# Effective initial-risk floor A/B' in focused_markdown
+    assert 'Mean dEV' in focused_markdown
+    assert 'Identical' in focused_markdown
+    assert 'Mean dGtP' in focused_markdown
+    assert 'Mean dCalmar/MAR' in focused_markdown
+    assert 'Floor rejects' in focused_markdown
+    assert 'not independent evidence' in focused_markdown
+
 
 def test_synthetic_worker_matrix_covers_four_arms_protocols_and_costs(monkeypatch):
     monkeypatch.setenv('BACKTEST_SNAPSHOT_OFFLINE', '0')