معايرة قواعد الخروج من المراكز من دون تسرب التقييم

# Systematic Risk Parameter Sweep


# Systematic Risk Parameter Sweep
**Chapter 19: Risk Management**

## Purpose
Calibrate position-level exit rules on a chronological calibration window, then compare the
selected rules once on a later evaluation window. One-dimensional sweeps show sensitivity to
each rule width, joint grids expose interactions, and MAE/MFE percentiles provide a second
calibration method.

## Learning Objectives
After completing this notebook, you will be able to:
- separate exit-rule calibration from chronological evaluation;
- run one-dimensional and joint StopLoss, TakeProfit, and TrailingStop sweeps;
- derive stop and target priors from closed-trade MAE/MFE paths;
- compare preselected configurations with CAGR-based Calmar ratios.

## Book reference
Section 19.7, "Adaptive Risk Controls Without Leakage," and Figures 19.6-19.7.

## Prerequisites
Complete [`02_exit_strategies`](02_exit_strategies.ipynb),
[`03_position_sizing_mae_mfe`](03_position_sizing_mae_mfe.ipynb), and
[`10_ml4t_backtest_risk_demo`](10_ml4t_backtest_risk_demo.ipynb) first.

```python
"""Calibrate and evaluate systematic position-exit rules without sample reuse."""

from datetime import date

import numpy as np
import plotly.graph_objects as go
import polars as pl
from IPython.display import Markdown, display
from ml4t.backtest import BacktestConfig, DataFeed, Engine, ExecutionMode, Strategy
from ml4t.backtest.analytics.trades import MAEMFEAnalyzer
from ml4t.backtest.execution.rebalancer import RebalanceConfig, TargetWeightExecutor
from ml4t.backtest.risk.position import RuleChain, StopLoss, TakeProfit, TrailingStop
from plotly.subplots import make_subplots

from data import load_etfs
from utils.paths import get_output_dir
from utils.style import COLORS, ml4t_diverging, show_plotly_with_alt
```

```python
START_DATE = "2019-01-02"
END_DATE = "2023-12-31"
CALIBRATION_END = "2021-12-31"
N_BARS = 1260
LOOKBACK_BARS = 60
CADENCE_BARS = 21
INITIAL_CASH = 100_000
```

```python
OUTPUT_DIR = get_output_dir(19, "risk_sweep")
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
SYMBOLS = ["SPY", "QQQ", "IWM", "XLF", "EEM"]
calibration_end = date.fromisoformat(CALIBRATION_END)
```

## 1. Point-in-Time ETF Panel and Split

The fixed five-ETF panel is an illustrative, present-day universe rather than a
point-in-time membership reconstruction. It therefore supports a rule-calibration lesson, not
an unbiased claim about historical ETF selection. Momentum uses each close at time *t*, and the
engine executes resulting targets on the next bar.

```python
prices_df = load_etfs(
    symbols=SYMBOLS,
    start_date=START_DATE,
    end_date=END_DATE,
).sort(["symbol", "timestamp"])
prices_df = (
    prices_df.with_columns(_bar=pl.col("timestamp").cum_count().over("symbol"))
    .filter(pl.col("_bar") <= N_BARS)
    .drop("_bar")
)
```

Before any backtest, fail closed on schema, duplicate keys, null OHLC data, and incomplete daily
snapshots. This makes a missing asset an input error instead of an exception to suppress inside
the executor.

```python
required_columns = {"symbol", "timestamp", "open", "high", "low", "close", "volume"}
assert required_columns <= set(prices_df.columns)
assert prices_df["timestamp"].dtype == pl.Date
assert set(prices_df["symbol"].unique()) == set(SYMBOLS)
assert prices_df.select("symbol", "timestamp").is_duplicated().sum() == 0
assert prices_df.select(sorted(required_columns)).null_count().row(0) == (0,) * len(
    required_columns
)
panel_counts = prices_df.group_by("timestamp").agg(pl.col("symbol").n_unique().alias("n_symbols"))
assert panel_counts.filter(pl.col("n_symbols") != len(SYMBOLS)).is_empty()

all_dates = prices_df["timestamp"].unique().sort().to_list()
calibration_dates = [ts for ts in all_dates if ts <= calibration_end]
evaluation_dates = [ts for ts in all_dates if ts > calibration_end]
assert len(calibration_dates) > LOOKBACK_BARS
assert len(evaluation_dates) > LOOKBACK_BARS
```

```python
calibration_prices = prices_df.filter(pl.col("timestamp") <= calibration_end)
evaluation_prices = prices_df.filter(pl.col("timestamp") > calibration_end)
calibration_schedule = calibration_dates[LOOKBACK_BARS::CADENCE_BARS]
evaluation_schedule = evaluation_dates[::CADENCE_BARS]

display(
    Markdown(
        f"Loaded **{prices_df.height:,} complete bars** for **{len(SYMBOLS)} ETFs**. "
        f"Calibration ends on **{calibration_dates[-1]}**; the untouched evaluation window runs "
        f"from **{evaluation_dates[0]}** through **{evaluation_dates[-1]}**. Rebalances are "
        f"scheduled every **{CADENCE_BARS} trading bars**."
    )
)
```

The weight builder computes trailing returns independently within each symbol. Passing an
explicit schedule prevents the strategy from mixing trading-bar schedules with calendar-day
comparisons.

```python
def build_momentum_weights(
    prices: pl.DataFrame,
    schedule: list[date],
    lookback: int,
    top_n: int = 3,
) -> dict[date, dict[str, float]]:
    """Return top-N equal weights for explicit decision timestamps."""
    momentum = (
        prices.sort(["symbol", "timestamp"])
        .with_columns(
            momentum=(pl.col("close") / pl.col("close").shift(lookback) - 1).over("symbol")
        )
        .select("timestamp", "symbol", "momentum")
        .drop_nulls()
    )
    weights: dict[date, dict[str, float]] = {}
    for timestamp in schedule:
        snapshot = momentum.filter(pl.col("timestamp") == timestamp).sort(
            "momentum", descending=True
        )
        assert snapshot.height == len(SYMBOLS)
        selected = snapshot.head(top_n)["symbol"].to_list()
        weights[timestamp] = {symbol: 1.0 / len(selected) for symbol in selected}
    return weights
```

```python
calibration_weights = build_momentum_weights(
    calibration_prices,
    calibration_schedule,
    LOOKBACK_BARS,
)
evaluation_weights = build_momentum_weights(
    prices_df,
    evaluation_schedule,
    LOOKBACK_BARS,
)
assert list(calibration_weights) == calibration_schedule
assert list(evaluation_weights) == evaluation_schedule
```

## 2. Backtest Contract

A target is submitted only on its scheduled trading bar. `NEXT_BAR` execution means the signal
formed from the current close cannot fill on that same close. The complete-panel assertion above
makes missing-symbol recovery unnecessary.

```python
class SweepStrategy(Strategy):
    """Apply a fixed rule chain and rebalance on explicit decision timestamps."""

    def __init__(self, weights: dict[date, dict[str, float]], rules=None):
        self.rules = rules
        self.weights = weights
        self.executor = TargetWeightExecutor(
            config=RebalanceConfig(
                min_trade_value=100.0,
                min_weight_change=0.005,
                allow_fractional=True,
            )
        )

    def on_start(self, broker):
        if self.rules is not None:
            broker.set_position_rules(self.rules)

    def on_data(self, timestamp, data, context, broker):
        del context
        if timestamp not in self.weights:
            return
        assert set(data) == set(SYMBOLS)
        self.executor.execute(self.weights[timestamp], data, broker)
```

The runner delegates annualization to `ml4t.backtest`: Calmar is the library's CAGR divided by
absolute maximum drawdown.

Trade statistics are a separate matter. A backtest that ends while positions are open produces
trades in three states - closed, partly unwound, and still open - and the last two carry a PnL
that is a mark-to-market rather than a result. The library's `num_trades` and `win_rate` count
all three, which inflates the win rate with positions that have not finished losing. Everything
trade-level below is therefore computed over the closed trades only, which is also the population
the MAE/MFE analysis reads.

```python
def run_risk_sweep(
    prices: pl.DataFrame,
    weights: dict[date, dict[str, float]],
    rules=None,
    initial_cash: float = 100_000,
) -> dict:
    """Run one exit-rule configuration and return closed-trade metrics."""
    config = BacktestConfig(
        initial_cash=initial_cash,
        commission_rate=0.001,
        slippage_rate=0.001,
        execution_mode=ExecutionMode.NEXT_BAR,
    )
    result = Engine(
        feed=DataFeed(prices_df=prices),
        strategy=SweepStrategy(weights, rules=rules),
        config=config,
    ).run()
    closed_trades = [trade for trade in result.trades if trade.status == "closed"]
    closed_wins = sum(1 for trade in closed_trades if trade.pnl > 0)
    return {
        # Equity-curve metrics come from the library and are unaffected by trade bookkeeping.
        "sharpe": float(result.metrics["sharpe"]),
        "max_dd": float(result.equity.max_dd),
        "calmar": float(result.metrics["calmar"]),
        "cagr": float(result.metrics["cagr"]),
        "total_return": float(result.metrics["total_return"]),
        # Trade-level statistics are computed over the closed trades only, matching the population
        # the MAE/MFE analysis reads. The library's own num_trades and win_rate count every trade
        # including the ones still open at the final bar, whose PnL is a mark rather than a result.
        "n_trades_all": int(result.metrics["num_trades"]),
        "n_trades": len(closed_trades),
        "win_rate": closed_wins / len(closed_trades) if closed_trades else float("nan"),
        "trades": closed_trades,
    }
```

```python
baseline_calibration = run_risk_sweep(
    calibration_prices,
    calibration_weights,
    initial_cash=INITIAL_CASH,
)
if not baseline_calibration["trades"]:
    raise ValueError("The calibration baseline closed no trades, so no trade statistics exist")

display(
    Markdown(
        f"The calibration baseline has Sharpe **{baseline_calibration['sharpe']:.2f}**, "
        f"CAGR **{baseline_calibration['cagr']:.1%}**, maximum drawdown "
        f"**{baseline_calibration['max_dd']:.1%}**, and Calmar "
        f"**{baseline_calibration['calmar']:.2f}**. Of the "
        f"**{baseline_calibration['n_trades_all']}** positions the backtest opened, "
        f"**{baseline_calibration['n_trades']}** closed before the window ended; the rest were "
        "still open or only partly unwound, and every trade statistic below counts the closed "
        "ones only."
    )
)
```

## 3. Calibration Sweeps

Every candidate below is fit and ranked on the calibration window only. The evaluation window is
not consulted until one candidate from each calibration method has been frozen.

This helper runs a one-dimensional rule family while retaining the raw metrics needed for the
diagnostic and publication-data outputs.

```python
def run_1d_sweep(
    rule_class,
    levels: list[float],
    name: str,
) -> list[dict]:
    """Evaluate one rule family on the calibration window."""
    results = []
    for level in levels:
        metrics = run_risk_sweep(
            calibration_prices,
            calibration_weights,
            rules=RuleChain([rule_class(pct=level)]),
            initial_cash=INITIAL_CASH,
        )
        results.append({**metrics, "rule": name, "param": level})
    return results
```

```python
sl_levels = [0.02, 0.03, 0.04, 0.05, 0.06, 0.08, 0.10, 0.12, 0.15, 0.20]
tp_levels = [0.03, 0.05, 0.07, 0.10, 0.12, 0.15, 0.20, 0.25, 0.30]
trail_levels = [0.02, 0.03, 0.04, 0.05, 0.06, 0.08, 0.10, 0.12, 0.15]

sl_results = run_1d_sweep(StopLoss, sl_levels, "StopLoss")
tp_results = run_1d_sweep(TakeProfit, tp_levels, "TakeProfit")
trail_results = run_1d_sweep(TrailingStop, trail_levels, "TrailingStop")
```

The solid line reports calibration Sharpe; the dotted line reports closed-trade count. Their
different axes make turnover changes visible without treating trade count as a performance
metric.

```python
def add_sweep_panel(
    fig,
    row: int,
    sweep: list[dict],
    name: str,
    color: str,
) -> None:
    """Add one calibrated sweep panel with a secondary trade-count axis."""
    fig.add_trace(
        go.Scatter(
            x=[row["param"] for row in sweep],
            y=[row["sharpe"] for row in sweep],
            mode="lines+markers",
            line={"color": color},
            name=f"{name} Sharpe",
        ),
        row=row,
        col=1,
        secondary_y=False,
    )
    fig.add_trace(
        go.Scatter(
            x=[row["param"] for row in sweep],
            y=[row["n_trades"] for row in sweep],
            mode="lines+markers",
            line={"color": COLORS["neutral"], "dash": "dot"},
            marker={"symbol": "square"},
            name=f"{name} closed trades",
        ),
        row=row,
        col=1,
        secondary_y=True,
    )
    fig.add_hline(
        y=baseline_calibration["sharpe"],
        line={"color": COLORS["slate"], "dash": "dash"},
        row=row,
        col=1,
        secondary_y=False,
    )
```

```python
fig = make_subplots(
    rows=3,
    cols=1,
    specs=[[{"secondary_y": True}] for _ in range(3)],
    subplot_titles=["StopLoss", "TakeProfit", "TrailingStop"],
    vertical_spacing=0.09,
)
for row, (sweep, name, color) in enumerate(
    [
        (sl_results, "StopLoss", COLORS["blue"]),
        (tp_results, "TakeProfit", COLORS["positive"]),
        (trail_results, "TrailingStop", COLORS["amber"]),
    ],
    start=1,
):
    add_sweep_panel(fig, row, sweep, name, color)
    fig.update_xaxes(
        title_text="Rule width" if row == 3 else None, tickformat=".0%", row=row, col=1
    )
    fig.update_yaxes(title_text="Sharpe ratio", secondary_y=False, row=row, col=1)
    fig.update_yaxes(title_text="Closed trades", secondary_y=True, row=row, col=1)
fig.update_layout(
    height=900,
    margin={"l": 75, "r": 85, "t": 105, "b": 70},
    title=(
        "Calibration Sharpe Depends on Exit Width"
        "<br><sup>2019-2021 calibration only; dashed line is the no-rule baseline</sup>"
    ),
    showlegend=False,
)
show_plotly_with_alt(
    fig,
    "Sharpe and trade count against threshold for each of the three one-dimensional sweeps, showing how each rule's parameter trades performance against how often it fires.",
)
```

```python
sweep_rows = [
    {
        "rule": row["rule"],
        "param": float(row["param"]),
        "sharpe": float(row["sharpe"]),
        "n_trades": int(row["n_trades"]),
        "sample": "calibration",
    }
    for sweep in (sl_results, tp_results, trail_results)
    for row in sweep
]
pl.DataFrame(sweep_rows).write_parquet(OUTPUT_DIR / "risk_rule_1d_sweeps.parquet")
```

Joint grids ask whether one rule's useful width depends on the other. The selection score is
calibration Calmar, but the later evaluation reports all candidates under one untouched window.

```python
def run_rule_grid(
    left_levels: list[float],
    right_levels: list[float],
    right_rule,
) -> tuple[np.ndarray, np.ndarray]:
    """Return calibration Sharpe and Calmar matrices for a two-rule grid."""
    sharpe = np.zeros((len(left_levels), len(right_levels)))
    calmar = np.zeros_like(sharpe)
    for row, stop in enumerate(left_levels):
        for column, right in enumerate(right_levels):
            metrics = run_risk_sweep(
                calibration_prices,
                calibration_weights,
                rules=RuleChain([StopLoss(pct=stop), right_rule(pct=right)]),
                initial_cash=INITIAL_CASH,
            )
            sharpe[row, column] = metrics["sharpe"]
            calmar[row, column] = metrics["calmar"]
    return sharpe, calmar
```

```python
sl_grid = [0.03, 0.05, 0.08, 0.10, 0.15]
tp_grid = [0.05, 0.10, 0.15, 0.20, 0.30]
trail_grid = [0.02, 0.04, 0.06, 0.08, 0.10]

sl_tp_sharpe, sl_tp_calmar = run_rule_grid(sl_grid, tp_grid, TakeProfit)
sl_trail_sharpe, sl_trail_calmar = run_rule_grid(sl_grid, trail_grid, TrailingStop)
assert np.isfinite(sl_tp_sharpe).all() and np.isfinite(sl_tp_calmar).all()
assert np.isfinite(sl_trail_sharpe).all() and np.isfinite(sl_trail_calmar).all()

sl_tp_sharpe_delta = sl_tp_sharpe - baseline_calibration["sharpe"]
sl_trail_sharpe_delta = sl_trail_sharpe - baseline_calibration["sharpe"]
sl_tp_calmar_delta = sl_tp_calmar - baseline_calibration["calmar"]
sl_trail_calmar_delta = sl_trail_calmar - baseline_calibration["calmar"]
sharpe_limit = float(
    np.max(np.abs(np.concatenate([sl_tp_sharpe_delta.ravel(), sl_trail_sharpe_delta.ravel()])))
)
calmar_limit = float(
    np.max(np.abs(np.concatenate([sl_tp_calmar_delta.ravel(), sl_trail_calmar_delta.ravel()])))
)
```

Each metric uses one pooled, symmetric scale across both rule pairs. Colors therefore encode the
change from the same no-rule baseline instead of re-normalizing every panel independently.

```python
def add_surface(
    fig,
    data: np.ndarray,
    x_labels: list[str],
    y_labels: list[str],
    row: int,
    column: int,
    limit: float,
    colorbar_title: str,
) -> None:
    """Add one baseline-relative heatmap with an explicit pooled range."""
    fig.add_trace(
        go.Heatmap(
            z=data,
            x=x_labels,
            y=y_labels,
            zmin=-limit,
            zmax=limit,
            zmid=0,
            colorscale=ml4t_diverging(),
            text=np.round(data, 2),
            texttemplate="%{text:+.2f}",
            showscale=row == 1,
            colorbar={
                "title": colorbar_title,
                "len": 0.8,
                "y": 0.5,
                "x": 1.03 if column == 1 else 1.17,
                "thickness": 12,
            },
        ),
        row=row,
        col=column,
    )
```

Explicit categorical ticks prevent Plotly's static renderer from dropping percentage signs.

```python
def apply_surface_axes(
    fig,
    stop_labels: list[str],
    take_profit_labels: list[str],
    trailing_labels: list[str],
) -> None:
    """Apply explicit percentage categories and units to every grid axis."""
    panels = [(1, 1, take_profit_labels), (1, 2, take_profit_labels)]
    panels += [(2, 1, trailing_labels), (2, 2, trailing_labels)]
    for row, column, labels in panels:
        fig.update_xaxes(
            type="category",
            tickmode="array",
            tickvals=labels,
            ticktext=labels,
            title_text="Second rule width (%)" if row == 2 else None,
            row=row,
            col=column,
        )
        fig.update_yaxes(
            type="category",
            tickmode="array",
            tickvals=stop_labels,
            ticktext=stop_labels,
            title_text="StopLoss width (%)" if column == 1 else None,
            row=row,
            col=column,
        )
```

```python
sl_labels = [f"{value:.0%}" for value in sl_grid]
tp_labels = [f"{value:.0%}" for value in tp_grid]
trail_labels = [f"{value:.0%}" for value in trail_grid]
fig = make_subplots(
    rows=2,
    cols=2,
    subplot_titles=[
        "SL x TP: Sharpe change",
        "SL x TP: Calmar change",
        "SL x Trail: Sharpe change",
        "SL x Trail: Calmar change",
    ],
    vertical_spacing=0.16,
    horizontal_spacing=0.14,
)
add_surface(fig, sl_tp_sharpe_delta, tp_labels, sl_labels, 1, 1, sharpe_limit, "Delta Sharpe")
add_surface(fig, sl_tp_calmar_delta, tp_labels, sl_labels, 1, 2, calmar_limit, "Delta Calmar")
add_surface(fig, sl_trail_sharpe_delta, trail_labels, sl_labels, 2, 1, sharpe_limit, "")
add_surface(fig, sl_trail_calmar_delta, trail_labels, sl_labels, 2, 2, calmar_limit, "")
apply_surface_axes(fig, sl_labels, tp_labels, trail_labels)
fig.update_layout(
    width=980,
    height=720,
    margin={"l": 80, "r": 190, "t": 100, "b": 75},
    title=(
        "Joint Exit Rules Move Calibration Performance Around the Baseline"
        "<br><sup>Pooled ranges make colors comparable within each metric</sup>"
    ),
)
show_plotly_with_alt(
    fig,
    "Four heatmap panels of change against the no-rule baseline: the stop-loss by take-profit grid on the top row and the stop-loss by trailing-stop grid on the bottom, with the Sharpe change in the left column and the Calmar change in the right. Each cell is labelled with its value, and each metric has one colour scale shared by both of its panels.",
)
```

```python
best_sl_tp_index = np.unravel_index(np.nanargmax(sl_tp_calmar), sl_tp_calmar.shape)
best_sl_trail_index = np.unravel_index(np.nanargmax(sl_trail_calmar), sl_trail_calmar.shape)
best_sl_tp = (sl_grid[best_sl_tp_index[0]], tp_grid[best_sl_tp_index[1]])
best_sl_trail = (sl_grid[best_sl_trail_index[0]], trail_grid[best_sl_trail_index[1]])

display(
    Markdown(
        f"Calibration selects **SL {best_sl_tp[0]:.0%} / TP {best_sl_tp[1]:.0%}** "
        f"at Calmar **{sl_tp_calmar[best_sl_tp_index]:.2f}**, and "
        f"**SL {best_sl_trail[0]:.0%} / Trail {best_sl_trail[1]:.0%}** at Calmar "
        f"**{sl_trail_calmar[best_sl_trail_index]:.2f}**. These are frozen before the "
        "evaluation window is opened."
    )
)
```

```python
grid_rows = [
    {
        "stop_loss": float(stop),
        "take_profit": float(target),
        "sharpe": float(sl_tp_sharpe[row, column]),
        "calmar": float(sl_tp_calmar[row, column]),
        "sample": "calibration",
    }
    for row, stop in enumerate(sl_grid)
    for column, target in enumerate(tp_grid)
]
pl.DataFrame(grid_rows).write_parquet(OUTPUT_DIR / "stoploss_takeprofit_grid.parquet")
```

## 4. MAE/MFE Calibration on Closed Trades

MAE/MFE percentiles are another fitted rule. They use only closed calibration trades, so neither
an end-of-window mark nor any evaluation path can influence the selected stop and target.

```python
calibration_trades = baseline_calibration["trades"]
mae_mfe = MAEMFEAnalyzer(calibration_trades)
suggested_sl = mae_mfe.suggest_stop_loss(percentile=90)
suggested_tp = mae_mfe.suggest_take_profit(percentile=75)
calibrated_sl_pct = abs(suggested_sl)
calibrated_tp_pct = abs(suggested_tp)

display(
    Markdown(
        f"Across **{mae_mfe.num_trades} closed calibration trades**, mean MAE is "
        f"**{mae_mfe.mae_mean:.1%}** and mean MFE is **{mae_mfe.mfe_mean:.1%}**. The frozen "
        f"percentile prior is **SL {calibrated_sl_pct:.1%} / TP {calibrated_tp_pct:.1%}**. "
        "The small trade sample makes this a deliberately low-confidence prior."
    )
)
```

```python
distribution = mae_mfe.distribution_data()
realized_returns = [trade.pnl_percent for trade in calibration_trades]
color_limit = max(abs(min(realized_returns)), abs(max(realized_returns)))
fig = go.Figure(
    go.Scatter(
        x=distribution["mae"],
        y=distribution["mfe"],
        mode="markers",
        marker={
            "size": 8,
            "color": realized_returns,
            "colorscale": ml4t_diverging(),
            "cmin": -color_limit,
            "cmax": color_limit,
            "colorbar": {"title": "Realized return", "tickformat": ".0%"},
            "line": {"width": 0.5, "color": COLORS["neutral"]},
        },
        text=[f"Realized return: {value:.1%}" for value in realized_returns],
        hovertemplate="MAE: %{x:.1%}<br>MFE: %{y:.1%}<br>%{text}<extra></extra>",
    )
)
fig.add_vline(x=suggested_sl, line_dash="dash", line_color=COLORS["negative"])
fig.add_hline(y=suggested_tp, line_dash="dash", line_color=COLORS["positive"])
fig.update_layout(
    height=480,
    title=(
        "Closed Calibration Trades Define the MAE/MFE Prior"
        "<br><sup>Dashed lines are the 90th-percentile stop and 75th-percentile target</sup>"
    ),
    xaxis={"title": "Maximum adverse excursion", "tickformat": ".0%"},
    yaxis={"title": "Maximum favorable excursion", "tickformat": ".0%"},
)
show_plotly_with_alt(
    fig,
    "A scatter of maximum favorable excursion against maximum adverse excursion for the closed calibration trades, each point coloured by its realized return. A dashed vertical line marks the stop percentile and a dashed horizontal line the target percentile. Most points sit near the origin with a few reaching far up the favorable axis.",
)
```

## 5. One-Time Chronological Evaluation

Only the no-rule baseline and the three configurations frozen above enter the 2022-2023 window.
This comparison estimates one realized-window outcome; it does not provide multiple-testing-
adjusted inference or establish that any rule is universally optimal.

```python
evaluation_configs = {
    "No rules": None,
    f"Grid SL/TP {best_sl_tp[0]:.0%}/{best_sl_tp[1]:.0%}": RuleChain(
        [StopLoss(pct=best_sl_tp[0]), TakeProfit(pct=best_sl_tp[1])]
    ),
    f"Grid SL/Trail {best_sl_trail[0]:.0%}/{best_sl_trail[1]:.0%}": RuleChain(
        [StopLoss(pct=best_sl_trail[0]), TrailingStop(pct=best_sl_trail[1])]
    ),
    f"MAE/MFE {calibrated_sl_pct:.0%}/{calibrated_tp_pct:.0%}": RuleChain(
        [StopLoss(pct=calibrated_sl_pct), TakeProfit(pct=calibrated_tp_pct)]
    ),
}
evaluation_results = {
    name: run_risk_sweep(
        evaluation_prices,
        evaluation_weights,
        rules=rules,
        initial_cash=INITIAL_CASH,
    )
    for name, rules in evaluation_configs.items()
}
```

```python
comparison = pl.DataFrame(
    [
        {
            "configuration": name,
            "sharpe": metrics["sharpe"],
            "cagr": metrics["cagr"],
            "max_dd": metrics["max_dd"],
            "calmar": metrics["calmar"],
            "closed_trades": metrics["n_trades"],
            "win_rate": metrics["win_rate"],
        }
        for name, metrics in evaluation_results.items()
    ]
).sort("calmar", descending=True)
winner = comparison.row(0, named=True)
baseline_evaluation = evaluation_results["No rules"]
comparison
```

The trade columns count closed trades and the win rate is computed over those same trades, so
both describe the positions that actually finished. Read them alongside the equity-curve columns
rather than instead of them: a configuration can close few trades and still shape the curve,
because the rules act on positions that remain open as well.

```python
plot_rows = comparison.sort("calmar", descending=True)
plot_labels = [name.replace(" ", "<br>", 1) for name in plot_rows["configuration"]]
fig = go.Figure(
    go.Bar(
        x=plot_labels,
        y=plot_rows["calmar"],
        marker_color=[
            COLORS["amber"] if name == winner["configuration"] else COLORS["blue"]
            for name in plot_rows["configuration"]
        ],
        text=[f"{value:.2f}" for value in plot_rows["calmar"]],
        textposition="outside",
        cliponaxis=False,
        customdata=np.column_stack([plot_rows["sharpe"], plot_rows["cagr"], plot_rows["max_dd"]]),
        hovertemplate=(
            "Calmar: %{y:.2f}<br>Sharpe: %{customdata[0]:.2f}<br>"
            "CAGR: %{customdata[1]:.1%}<br>Max drawdown: %{customdata[2]:.1%}<extra></extra>"
        ),
    )
)
fig.update_layout(
    width=850,
    height=480,
    margin={"l": 75, "r": 45, "t": 100, "b": 90},
    title=(
        f"{winner['configuration']} Leads the One-Time Evaluation"
        "<br><sup>2022-2023; all rule choices frozen on 2019-2021 calibration data</sup>"
    ),
    xaxis={"title": None, "automargin": True},
    yaxis={"title": "Calmar ratio (CAGR / |maximum drawdown|)", "zeroline": True},
    showlegend=False,
)
show_plotly_with_alt(
    fig,
    "Bars of evaluation-window Calmar per configuration, sorted, with the highest-ranked bar highlighted. Several bars sit below zero.",
)
```

```python
pl.DataFrame(
    [
        {
            "sample": "calibration",
            "sharpe": baseline_calibration["sharpe"],
            "n_trades": baseline_calibration["n_trades"],
        },
        {
            "sample": "evaluation",
            "sharpe": baseline_evaluation["sharpe"],
            "n_trades": baseline_evaluation["n_trades"],
        },
    ]
).write_parquet(OUTPUT_DIR / "risk_rule_baseline.parquet")
```

## Key Takeaways

```python
display(
    Markdown(
        f"""
1. **Calibrate before evaluating.** The grids and MAE/MFE percentiles use only 2019-2021 data;
   the 2022-2023 window is opened once after all three rule configurations are frozen.
2. **Use the metric you name.** The evaluation ranks configurations by CAGR-based Calmar. In this
   run, **{winner["configuration"]}** ranks first at **{winner["calmar"]:.2f}**, while the no-rule
   baseline records **{baseline_evaluation["calmar"]:.2f}**.
3. **Treat small-sample excursion percentiles as priors.** The MAE/MFE levels come from only
   **{mae_mfe.num_trades} closed calibration trades**, so they seed a search rather than settle it.
4. **Keep the operational contract explicit.** Signals use completed closes, orders execute on the
   next bar, schedules count trading bars, and incomplete panels fail before the backtest.

This capstone closes the chapter's progression from risk measurement to calibrated controls.
"""
    )
)
```
![notebook output](figures/p1_1.png)
![notebook output](figures/p1_2.png)
![notebook output](figures/p1_3.png)
![notebook output](figures/p1_4.png)