用于自适应交易退出的防泄漏机器学习

# ML-Based Exit Signals: Two-Model Architecture


# ML-Based Exit Signals: Two-Model Architecture
**Docker image**: `ml4t`

## Purpose
Demonstrate a two-model exit architecture in which entry-model confidence becomes an
explicit feature of the exit model, and judge the resulting decision rule against a
basic exit baseline using both AUC and realized trade-path metrics.

## Learning Objectives
After completing this notebook, you will be able to:
- Train separate entry and exit prediction models on a shared feature set.
- Inject entry-probability into the exit model and read its feature importance.
- Evaluate exit logic by AUC *and* by realized trade-path metrics, not AUC alone.
- Recognize when an architectural change does not pay off on AUC but still matters operationally.

## Book reference
§19.7 Adaptive risk controls without leakage, Figure 19.5
(signal-strength-conditioned barrier outcomes).

## Prerequisites
Complete [`02_exit_strategies`](02_exit_strategies.ipynb) first for the rule-based exit
baseline (fixed stops, trailing stops, volatility-adjusted exits, whipsaw analysis) that
this notebook extends with an ML-driven exit policy.

## Setup

```python
"""Train a causal two-model architecture where entry confidence drives exit timing."""

from datetime import UTC, datetime, timedelta

import lightgbm as lgb
import numpy as np
import plotly.graph_objects as go
import polars as pl
from IPython.display import Markdown, display
from matplotlib import pyplot as plt
from plotly.subplots import make_subplots
from sklearn.metrics import roc_auc_score

from data import load_crypto_perps
from utils.paths import get_output_dir
from utils.reproducibility import set_global_seeds
from utils.style import COLORS, show_plotly_with_alt, show_with_alt
```

```python
SEED = 42
LGB_DEVICE = "cpu"
FORWARD_HOURS = 24
N_OOF_FOLDS = 5
N_IMPORTANCE_REPEATS = 5
TEST_FRACTION = 0.20
ENTRY_QUANTILE = 0.95
ENTRY_PROBABILITY_THRESHOLD = 0.5
PROBABILITY_BIN_WIDTH = 0.02
ENTRY_CONFIDENCE_DROP = 0.30
```

```python
set_global_seeds(SEED)
OUTPUT_DIR = get_output_dir(19, "ml_exit_signals")
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
print(
    f"LightGBM device_type requested: {LGB_DEVICE}. A build without that tree learner raises "
    "at fit(), and the booster is checked against this value below in case the parameter "
    "never reached it."
)
```

## 1. Data Loading

Three crypto perpetuals (BTC, ETH, SOL) at hourly frequency from 2023-01-01. High
volatility and round-the-clock trading make exit timing the dominant risk control
rather than a corner-case concern.

```python
ohlcv = load_crypto_perps(frequency="1h")

SYMBOLS = ["BTCUSDT", "ETHUSDT", "SOLUSDT"]
START_DATE = "2023-01-01"

# Filter data
timestamp_dtype = ohlcv.schema["timestamp"]
start_ts = datetime.fromisoformat(f"{START_DATE}T00:00:00")
if getattr(timestamp_dtype, "time_zone", None):
    start_ts = start_ts.replace(tzinfo=UTC)

df = ohlcv.filter(
    (pl.col("symbol").is_in(SYMBOLS))
    & (pl.col("timestamp") >= pl.lit(start_ts, dtype=timestamp_dtype))
)

print(f"Loaded {len(df):,} rows for {df['symbol'].n_unique()} symbols")
print(f"Date range: {df['timestamp'].min()} to {df['timestamp'].max()}")
```

**Interpretation**: The sample is intentionally short-horizon and volatile, so the
exit model has enough adverse moves to learn from. That makes crypto perpetuals a
good stress test for conviction-driven exits.

## 2. Feature Engineering

Create technical features for both entry and exit models.

The feature builder combines momentum, volatility, mean-reversion, and volume
signals into a single tabular dataset for both the entry and exit models.

```python
def create_features(df: pl.DataFrame, forward_hours: int = 24) -> pl.DataFrame:
    """
    Create features for entry/exit models.

    Features focus on momentum, volatility, and mean reversion signals.
    """
    close = pl.col("close")
    returns = close.pct_change()
    feature_exprs = [
        returns.over("symbol").alias("ret_1h"),
        close.pct_change(4).over("symbol").alias("ret_4h"),
        close.pct_change(8).over("symbol").alias("ret_8h"),
        close.pct_change(24).over("symbol").alias("ret_24h"),
        close.pct_change(72).over("symbol").alias("ret_72h"),
        returns.rolling_std(24).over("symbol").alias("vol_24h"),
        returns.rolling_std(72).over("symbol").alias("vol_72h"),
        (
            returns.rolling_mean(14).over("symbol") / returns.abs().rolling_mean(14).over("symbol")
        ).alias("rsi_proxy"),
        (close / close.rolling_mean(24).over("symbol") - 1).alias("dist_ma24"),
        (close / close.rolling_mean(72).over("symbol") - 1).alias("dist_ma72"),
        (pl.col("volume") / pl.col("volume").rolling_mean(24).over("symbol")).alias("vol_ratio"),
        ((pl.col("high") - pl.col("low")) / close).alias("hl_range"),
        (((pl.col("high") - pl.col("low")) / close).rolling_mean(24).over("symbol")).alias(
            "hl_range_ma"
        ),
        close.pct_change(forward_hours).shift(-forward_hours).over("symbol").alias("fwd_return"),
        # True one-step realized return: close[t] -> close[t+1], per symbol.
        # Used for the trade-path backtest so realized PnL is summed from
        # actual next-bar returns rather than a forward label divided by h.
        close.pct_change().shift(-1).over("symbol").alias("realized_ret_1step"),
    ]
    return df.sort(["symbol", "timestamp"]).with_columns(feature_exprs).drop_nulls()
```

```python
# Create features
df_feat = create_features(df, forward_hours=FORWARD_HOURS)

# Define feature columns
FEATURE_COLS = [
    "ret_1h",
    "ret_4h",
    "ret_8h",
    "ret_24h",
    "ret_72h",
    "vol_24h",
    "vol_72h",
    "rsi_proxy",
    "dist_ma24",
    "dist_ma72",
    "vol_ratio",
    "hl_range",
    "hl_range_ma",
]

print(f"Feature dataset: {len(df_feat):,} samples")
df_feat.select(
    pl.len().alias("Samples"),
    pl.col("fwd_return").mean().alias("Mean forward return"),
    pl.col("fwd_return").std().alias("Forward-return volatility"),
    pl.col("fwd_return").quantile(0.05).alias("5th percentile"),
    pl.col("fwd_return").quantile(ENTRY_QUANTILE).alias("entry quantile"),
)
```

**Interpretation**: The forward-return distribution is heavy enough in both tails to
justify separate entry and exit labels. A symmetric Gaussian world would not make
the exit model especially useful.

## 3. Chronological Split and Train-Only Labels

The split is chronological and the last part of the sample is the test interval. Between the two
sits a purge of `FORWARD_HOURS`: each label reads that far ahead, so without it the last training
rows would carry outcomes from inside the test window.

The entry threshold is estimated on the training interval alone and then applied unchanged. That
ordering is what makes the test interval a test - a threshold picked to look good on the test
rows would report the fit of a choice made after seeing them.

This is a single fixed interval used once for demonstration. Nothing here is selected on it, and
it is not large enough to distinguish two models whose scores are close.

```python
timestamps = df_feat["timestamp"].unique().sort()
test_start = timestamps[int((1 - TEST_FRACTION) * len(timestamps))]
purge_start = test_start - timedelta(hours=FORWARD_HOURS)

train_df = df_feat.filter(pl.col("timestamp") < purge_start).sort(["timestamp", "symbol"])
test_df = df_feat.filter(pl.col("timestamp") >= test_start).sort(["timestamp", "symbol"])
embargo_rows = df_feat.filter(
    (pl.col("timestamp") >= purge_start) & (pl.col("timestamp") < test_start)
).height
entry_return_threshold = train_df["fwd_return"].quantile(ENTRY_QUANTILE)
```

The same train-only threshold creates both label frames. Exit labels need no fitted
threshold: they indicate whether the next 24-hour return is negative.

```python
def add_labels(frame: pl.DataFrame, entry_threshold: float) -> pl.DataFrame:
    """Attach entry and exit labels using a threshold learned outside the frame."""

    return frame.with_columns(
        (pl.col("fwd_return") >= entry_threshold).cast(pl.Int8).alias("y_entry"),
        (pl.col("fwd_return") < 0).cast(pl.Int8).alias("y_exit"),
    )
```

```python
train_df = add_labels(train_df, entry_return_threshold)
test_df = add_labels(test_df, entry_return_threshold)

X_train = train_df.select(FEATURE_COLS).to_numpy()
X_test = test_df.select(FEATURE_COLS).to_numpy()
y_entry_train = train_df["y_entry"].to_numpy()
y_entry_test = test_df["y_entry"].to_numpy()
y_exit_train = train_df["y_exit"].to_numpy()
y_exit_test = test_df["y_exit"].to_numpy()

split_summary = pl.DataFrame(
    {
        "Interval": ["Purged train", "Embargo", "Demonstration test"],
        "Start": [train_df["timestamp"].min(), purge_start, test_df["timestamp"].min()],
        "End": [train_df["timestamp"].max(), test_start, test_df["timestamp"].max()],
        "Rows": [train_df.height, embargo_rows, test_df.height],
    }
)
split_summary
```

```python
label_summary = pl.DataFrame(
    {
        "Sample": ["Train", "Test"],
        "Entry positive share": [train_df["y_entry"].mean(), test_df["y_entry"].mean()],
        "Exit positive share": [train_df["y_exit"].mean(), test_df["y_exit"].mean()],
        "Train-only entry threshold": [entry_return_threshold, entry_return_threshold],
    }
)
label_summary
```

The split is chronological by timestamp across all three symbols. The 24-hour gap
prevents a training label from reaching into the test interval; the threshold and
every fitted model state are learned without test rows.

## 4. Train Entry Model

The entry model predicts an exceptional positive return - the top slice of the forward-return
distribution set by `ENTRY_QUANTILE`. Because the class is rare by construction, the model spends
most of its capacity learning what an ordinary bar looks like, and its probabilities are
correspondingly low almost everywhere.

The exit model that follows will take the entry probability as one of its inputs. That creates a
trap: a probability produced by a model that was fitted on the same row is not the probability
that model would have produced in production, and an exit model trained on it learns a
relationship that will not exist live. The expanding-window out-of-fold pass below produces each
training row's entry probability from a model that never saw that row, with a purge at every
fold boundary for the same reason as the main split.

```python
def make_classifier(seed: int) -> lgb.LGBMClassifier:
    """Create the pinned LightGBM classifier used throughout the notebook."""

    return lgb.LGBMClassifier(
        n_estimators=100,
        learning_rate=0.05,
        num_leaves=31,
        max_depth=5,
        min_child_samples=20,
        subsample=0.8,
        subsample_freq=1,
        colsample_bytree=0.8,
        max_bin=255,
        importance_type="gain",
        device_type=LGB_DEVICE,
        n_jobs=1,
        verbosity=-1,
        random_state=seed,
        data_random_seed=seed,
        feature_fraction_seed=seed,
        bagging_seed=seed,
        extra_seed=seed,
    )
```

Probability extraction uses the fitted LightGBM booster directly so inference follows
the same pinned one-thread path without sklearn feature-name conversion warnings.

```python
def predict_positive_probability(model: lgb.LGBMClassifier, features: np.ndarray) -> np.ndarray:
    """Predict binary positive-class probabilities without sklearn name warnings."""

    probability = np.asarray(model.booster_.predict(features, num_threads=1))
    if probability.shape != (len(features),) or not np.isfinite(probability).all():
        raise ValueError("binary LightGBM prediction must return one finite value per row")
    if np.any((probability < 0) | (probability > 1)):
        raise ValueError("binary LightGBM probabilities must lie in [0, 1]")
    return probability
```

Expanding validation blocks preserve order. Each fold model stops at least one
label horizon before its validation block, so the entry-probability feature is
point-in-time for every row used to train the enhanced exit model.

Each validation block is contiguous in timestamp space. This helper types its boundaries
to the frame schema and returns purged fit rows plus the bounded validation interval.

```python
def chronological_fold_rows(
    ordered: pl.DataFrame,
    validation_times: list[datetime],
    label_horizon_hours: int,
) -> tuple[np.ndarray, np.ndarray]:
    """Return purged fit rows and dtype-safe validation rows for one fold."""

    timestamp_dtype = ordered.schema["timestamp"]
    validation_values = pl.Series("validation_timestamp", validation_times, dtype=timestamp_dtype)
    validation_start, validation_end = validation_values[0], validation_values[-1]
    fit_end = validation_start - timedelta(hours=label_horizon_hours)
    fit_rows = ordered.filter(pl.col("timestamp") < pl.lit(fit_end, dtype=timestamp_dtype))[
        "row_id"
    ].to_numpy()
    validation_rows = ordered.filter(
        (pl.col("timestamp") >= pl.lit(validation_start, dtype=timestamp_dtype))
        & (pl.col("timestamp") <= pl.lit(validation_end, dtype=timestamp_dtype))
    )["row_id"].to_numpy()
    if not len(fit_rows) or not len(validation_rows):
        raise ValueError("each OOF fold requires nonempty fit and validation rows")
    return fit_rows, validation_rows
```

Each OOF model defines its rare-event label from that fold's purged fit rows. Later
outer-training returns therefore cannot move an earlier fold's cutoff or fit labels.

```python
def fold_entry_labels(ordered: pl.DataFrame, fit_rows: np.ndarray) -> tuple[np.ndarray, float]:
    """Build binary entry labels from one fold's purged fit-return distribution."""

    fit_returns = ordered[fit_rows, "fwd_return"]
    threshold = float(fit_returns.quantile(ENTRY_QUANTILE))
    labels = (fit_returns >= threshold).cast(pl.Int8).to_numpy()
    if len(np.unique(labels)) != 2:
        raise ValueError("each OOF fit requires both entry-label classes")
    return labels, threshold
```

The OOF driver stays focused on expanding folds, local label fits, model fits, and
prediction placement.

```python
def chronological_oof_entry_probabilities(
    frame: pl.DataFrame,
    feature_cols: list[str],
    n_folds: int,
    label_horizon_hours: int,
    seed: int,
) -> tuple[np.ndarray, list[lgb.LGBMClassifier], list[float]]:
    """Generate purged expanding-window entry probabilities for stacking."""

    ordered = frame.sort(["timestamp", "symbol"]).with_row_index("row_id")
    unique_times = ordered["timestamp"].unique().sort().to_list()
    initial = max(int(0.4 * len(unique_times)), label_horizon_hours + 1)
    validation_blocks = [
        [unique_times[index] for index in block]
        for block in np.array_split(np.arange(initial, len(unique_times)), n_folds)
    ]
    features = ordered.select(feature_cols).to_numpy()
    oof = np.full(ordered.height, np.nan)
    models: list[lgb.LGBMClassifier] = []
    thresholds: list[float] = []

    for fold, validation_times in enumerate(validation_blocks):
        if not validation_times:
            continue
        fit_rows, validation_rows = chronological_fold_rows(
            ordered, validation_times, label_horizon_hours
        )
        fit_labels, fold_threshold = fold_entry_labels(ordered, fit_rows)
        model = make_classifier(seed + fold)
        model.fit(features[fit_rows], fit_labels)
        oof[validation_rows] = predict_positive_probability(model, features[validation_rows])
        models.append(model)
        thresholds.append(fold_threshold)

    if not models or np.isnan(oof).all():
        raise ValueError("chronological OOF construction produced no predictions")
    return oof, models, thresholds
```

```python
entry_proba_oof, entry_fold_models, entry_fold_thresholds = chronological_oof_entry_probabilities(
    train_df,
    FEATURE_COLS,
    n_folds=N_OOF_FOLDS,
    label_horizon_hours=FORWARD_HOURS,
    seed=SEED,
)
meta_train_mask = np.isfinite(entry_proba_oof)
print(
    f"Purged chronological OOF coverage: {meta_train_mask.sum():,}/{len(meta_train_mask):,} "
    f"rows across {len(entry_fold_models)} folds"
)
print(
    f"Fold-local entry cutoffs: {min(entry_fold_thresholds):.6f} to "
    f"{max(entry_fold_thresholds):.6f}"
)
```

```python
entry_model = make_classifier(SEED)
entry_model.fit(X_train, y_entry_train)
entry_model_device = str(entry_model.booster_.params.get("device_type", "")).lower()
if entry_model_device != LGB_DEVICE:
    raise RuntimeError(
        f"LightGBM device mismatch: required {LGB_DEVICE}, observed {entry_model_device or 'unset'}"
    )
print(
    f"Device check: LightGBM {lgb.__version__} trained the booster with "
    f"device_type={entry_model_device}, the value that was requested; helper threads=1"
)
entry_proba_test = predict_positive_probability(entry_model, X_test)

entry_auc = roc_auc_score(y_entry_test, entry_proba_test)
print(f"Entry Model AUC-ROC: {entry_auc:.4f}")
```

**Interpretation**: The entry model sets the benchmark for conviction. Its AUC and
feature ranking tell us whether the signal has enough structure to be useful as an
input to the exit model.

## 5. Train Exit Model (Basic)

Predicts negative returns from technical features. For a fair comparison, it uses
exactly the OOF-covered training rows available to the enhanced model.

```python
exit_model_basic = make_classifier(SEED + 100)
exit_model_basic.fit(X_train[meta_train_mask], y_exit_train[meta_train_mask])

exit_proba_basic = predict_positive_probability(exit_model_basic, X_test)

exit_auc_basic = roc_auc_score(y_exit_test, exit_proba_basic)
print(f"Basic Exit Model AUC-ROC: {exit_auc_basic:.4f}")
```

**Interpretation**: The basic exit model shows how much adverse-move prediction is
already present in the raw feature set before we feed in entry conviction.

## 6. Train Enhanced Exit Model

Add the entry-model probability as an extra feature. The hypothesis is that fading
conviction is informative about adverse moves beyond the raw technicals. Training
uses only OOF entry probabilities; the final entry model supplies test probabilities.

```python
X_train_enhanced = np.column_stack([X_train[meta_train_mask], entry_proba_oof[meta_train_mask]])
X_test_enhanced = np.column_stack([X_test, entry_proba_test])

FEATURE_COLS_ENHANCED = FEATURE_COLS + ["entry_prediction"]

exit_model_enhanced = make_classifier(SEED + 200)
exit_model_enhanced.fit(X_train_enhanced, y_exit_train[meta_train_mask])

# Predictions
exit_proba_enhanced = predict_positive_probability(exit_model_enhanced, X_test_enhanced)

# Evaluate
exit_auc_enhanced = roc_auc_score(y_exit_test, exit_proba_enhanced)
improvement = (exit_auc_enhanced - exit_auc_basic) / exit_auc_basic * 100

print(f"Enhanced Exit Model AUC-ROC: {exit_auc_enhanced:.4f}")
print(f"AUC delta vs basic:          {improvement:+.2f}%")
```

```python
display(
    Markdown(
        f"""**Interpretation**: The enhanced exit AUC is {exit_auc_enhanced:.3f} versus
{exit_auc_basic:.3f} for the basic model, a relative change of {improvement:+.2f}%. These values
describe one fixed test interval. Neither model was selected on it, and an interval this size
cannot separate two AUCs this close.
The realized trade-path comparison below is a separate operational diagnostic."""
    )
)
```

## 7. Model Comparison

The summary table puts the three models on one page so we can judge whether the
extra architecture earns its complexity.

```python
delta_label = f"AUC delta: {improvement:+.2f}%"
comparison_summary = pl.DataFrame(
    {
        "Model": ["Entry (Top 5%)", "Exit (Basic)", "Exit (Enhanced)"],
        "AUC-ROC": [entry_auc, exit_auc_basic, exit_auc_enhanced],
        "Features": [
            "Technical indicators",
            "Technical indicators only",
            "Technical + Entry Prediction",
        ],
        "Notes": [
            "Identify exceptional opportunities",
            "Baseline exit model",
            delta_label,
        ],
    }
)
comparison_summary
```

```python
display(
    Markdown(
        f"""**Interpretation**: Entry AUC is {entry_auc:.3f}; basic and enhanced exit AUC are
{exit_auc_basic:.3f} and {exit_auc_enhanced:.3f}. AUC measures ranking on the fixed test interval,
while Section 9 asks a different question: how the predeclared exit rules change realized holding
periods and compounded trade returns."""
    )
)
```

## 8. Signal Analysis

How do entry and exit signals interact?

The entry label marks a rare event, so demanding a probability above one half before entering is
a severe filter: it admits only the bars where the model thinks a rare move is more likely than
not. Almost no bars clear it, which is the intent - the policy buys precision with coverage, and
spends most of the sample flat rather than trying to hold a position most of the time.

### Three Ways to Say "Get Out"

The exit rules compared below differ in what evidence they act on. One exits when the exit model
calls for it. One exits when the *entry* model's confidence has fallen far enough that the reason
for holding has gone, even if the exit model has said nothing. The third fires on either.

`ENTRY_CONFIDENCE_DROP` is the level the entry probability must fall below for the second rule.
It sits well under the threshold required to open a position, so the rule asks whether the case
for the trade has collapsed rather than merely weakened.

```python
entry_threshold = ENTRY_PROBABILITY_THRESHOLD
exit_threshold = ENTRY_PROBABILITY_THRESHOLD
entry_confidence_drop = ENTRY_CONFIDENCE_DROP

# Classify signals
entry_signal = entry_proba_test > entry_threshold
exit_signal_model = exit_proba_enhanced > exit_threshold
exit_signal_confidence = entry_proba_test < entry_confidence_drop

# Combined exit: model says exit OR entry confidence too low
exit_signal_combined = exit_signal_model | exit_signal_confidence
```

```python
# Bound once: both rates are read by name in the table, the reading under it, and the takeaways.
confidence_fire_rate = exit_signal_confidence.mean()
combined_fire_rate = exit_signal_combined.mean()

signal_counts = pl.DataFrame(
    {
        "Rule": [
            f"Entry signal (p_entry > {entry_threshold})",
            f"Exit model (p_exit > {exit_threshold})",
            f"Confidence drop (p_entry < {entry_confidence_drop})",
            "Combined exit (OR)",
        ],
        "Count": [
            int(entry_signal.sum()),
            int(exit_signal_model.sum()),
            int(exit_signal_confidence.sum()),
            int(exit_signal_combined.sum()),
        ],
        "Share of bars": [
            f"{entry_signal.mean():.1%}",
            f"{exit_signal_model.mean():.1%}",
            f"{confidence_fire_rate:.1%}",
            f"{combined_fire_rate:.1%}",
        ],
    }
)
signal_counts
```

```python
display(
    Markdown(
        f"""**Interpretation**: The confidence-drop clause fires on
{confidence_fire_rate:.1%} of test bars and the combined rule fires on
{combined_fire_rate:.1%}. Because the rare-event entry model usually assigns low
probability, most bars sit below the confidence-drop level already, so that clause fires on nearly
everything and the combined rule inherits its behaviour rather than the exit model's. The next
section reads the holding periods and trade returns instead of assuming more exits are better."""
    )
)
```

## 9. Backtest Simulation

Compare exit strategies on actual returns.

Signals are observed at an hourly close. An entry at close t earns the return from
close t to close t+1. An exit at close u does not earn the return from u to u+1.
This explicit ordering makes a k-bar holding period contain exactly k one-step returns.

```python
def build_trade(
    symbol: str,
    entry_idx: int,
    exit_idx: int,
    timestamps: np.ndarray,
    returns: np.ndarray,
    exit_reason: str,
) -> dict:
    """Create one close-to-close trade with compounded simple returns."""

    path = returns[entry_idx:exit_idx]
    if not len(path) or np.any(path <= -1) or not np.isfinite(path).all():
        raise ValueError("a trade requires finite simple returns above -100%")
    return {
        "symbol": symbol,
        "entry_timestamp": timestamps[entry_idx],
        "exit_timestamp": timestamps[exit_idx],
        "entry_idx": entry_idx,
        "exit_idx": exit_idx,
        "return": np.prod(1 + path) - 1,
        "holding_bars": exit_idx - entry_idx,
        "exit_reason": exit_reason,
    }
```

The simulator processes each symbol independently, so a position can never cross an
asset boundary. Open positions close at the last observed close without inventing a
return beyond the evaluation sample.

A separate summary step keeps variable-duration trade accounting explicit and avoids
burying the reported statistics inside the event loop.

```python
def summarize_trades(trades: list[dict]) -> dict:
    """Summarize compounded trade paths without annualizing unequal horizons."""

    if not trades:
        return {
            "n_trades": 0,
            "mean_return": 0.0,
            "median_return": 0.0,
            "win_rate": 0.0,
            "avg_holding": 0.0,
            "signal_exits": 0,
            "timeout_exits": 0,
            "sample_end_exits": 0,
            "trades": [],
        }
    trade_returns = np.array([trade["return"] for trade in trades])
    return {
        "n_trades": len(trades),
        "mean_return": float(trade_returns.mean()),
        "median_return": float(np.median(trade_returns)),
        "win_rate": float((trade_returns > 0).mean()),
        "avg_holding": float(np.mean([trade["holding_bars"] for trade in trades])),
        "signal_exits": sum(trade["exit_reason"] == "signal" for trade in trades),
        "timeout_exits": sum(trade["exit_reason"] == "timeout" for trade in trades),
        "sample_end_exits": sum(trade["exit_reason"] == "sample_end" for trade in trades),
        "trades": trades,
    }
```

The event loop handles only entry, signal/timeout exits, and sample-end closure by symbol.

```python
def simulate_strategy(
    symbols: np.ndarray,
    timestamps: np.ndarray,
    returns: np.ndarray,
    entry_signals: np.ndarray,
    exit_signals: np.ndarray,
    max_hold: int = 24,
) -> dict:
    """Simulate independent symbol paths with explicit close-time event ordering."""

    lengths = {len(symbols), len(timestamps), len(returns), len(entry_signals), len(exit_signals)}
    if lengths != {len(returns)}:
        raise ValueError("strategy arrays must have equal length")
    trades: list[dict] = []

    for symbol in dict.fromkeys(symbols):
        symbol_rows = np.flatnonzero(symbols == symbol)
        position = False
        entry_idx = -1
        for i in symbol_rows:
            if not position and entry_signals[i]:
                position = True
                entry_idx = i
                continue
            if not position:
                continue
            holding_time = i - entry_idx
            if not (exit_signals[i] or holding_time >= max_hold):
                continue
            exit_reason = "signal" if exit_signals[i] else "timeout"
            trades.append(build_trade(symbol, entry_idx, i, timestamps, returns, exit_reason))
            position = False

        final_idx = symbol_rows[-1]
        if position and final_idx > entry_idx:
            trades.append(
                build_trade(symbol, entry_idx, final_idx, timestamps, returns, "sample_end")
            )

    return summarize_trades(trades)
```

The model's predictions are in the order `test_df` was built in, and the simulator needs the rows
regrouped so each symbol forms an independent path. Carrying an explicit row index through the
regrouping is what keeps every prediction attached to the bar it was made for; re-sorting without
it would silently pair predictions with other symbols' bars.

```python
bt_frame = test_df.with_row_index("model_row").sort(["symbol", "timestamp"])
bt_pos = bt_frame["model_row"].to_numpy()
bt_symbols = bt_frame["symbol"].to_numpy()
bt_timestamps = bt_frame["timestamp"].to_numpy()
actual_returns = bt_frame["realized_ret_1step"].to_numpy()
entry_signal_bt = entry_signal[bt_pos]
exit_basic_bt = (exit_proba_basic > ENTRY_PROBABILITY_THRESHOLD)[bt_pos]
exit_model_bt = exit_signal_model[bt_pos]
exit_confidence_bt = exit_signal_confidence[bt_pos]
exit_combined_bt = exit_signal_combined[bt_pos]

# Strategy variants
strategies = {
    "No Exit Signal": simulate_strategy(
        bt_symbols,
        bt_timestamps,
        actual_returns,
        entry_signal_bt,
        np.zeros_like(entry_signal_bt, dtype=bool),
    ),
    "Basic Exit Model": simulate_strategy(
        bt_symbols, bt_timestamps, actual_returns, entry_signal_bt, exit_basic_bt
    ),
    "Enhanced Exit Model": simulate_strategy(
        bt_symbols, bt_timestamps, actual_returns, entry_signal_bt, exit_model_bt
    ),
    "Entry Confidence Drop": simulate_strategy(
        bt_symbols, bt_timestamps, actual_returns, entry_signal_bt, exit_confidence_bt
    ),
    "Combined Exit": simulate_strategy(
        bt_symbols, bt_timestamps, actual_returns, entry_signal_bt, exit_combined_bt
    ),
}
```

```python
results_df = pl.DataFrame(
    [
        {
            "Strategy": name,
            **{key: value for key, value in result.items() if key != "trades"},
        }
        for name, result in strategies.items()
    ]
)
results_df
```

```python
best_mean_row = results_df.sort("mean_return", descending=True).row(0, named=True)
shortest_row = results_df.sort("avg_holding").row(0, named=True)
display(
    Markdown(
        f"""**Interpretation**: {best_mean_row["Strategy"]} has the highest mean compounded trade
return ({best_mean_row["mean_return"]:.2%}) on this fixed test interval. {shortest_row["Strategy"]}
has the shortest average holding period ({shortest_row["avg_holding"]:.1f} hourly bars). These are
descriptive trade-level metrics over a single test interval. They are not annualized, carry no
cost, and are not an estimate of what the strategy would earn."""
    )
)
```

## 10. Visualization

The figures test whether the architecture's behavior is explainable. Gain importance
depends on the row and feature subsamples each fit happens to draw, so it is summarized
across seeded repeats rather than read off a single fit.

```python
def repeated_gain_importance(
    features: np.ndarray,
    labels: np.ndarray,
    feature_names: list[str],
    seed: int,
    repeats: int,
) -> pl.DataFrame:
    """Estimate normalized gain importance dispersion across seeded repeats."""

    records = []
    for repeat in range(repeats):
        model = make_classifier(seed + repeat)
        model.fit(features, labels)
        gain = model.feature_importances_.astype(float)
        if not np.isfinite(gain).all() or gain.sum() <= 0:
            raise ValueError("gain importance must be finite with positive total")
        for feature, importance in zip(feature_names, 100 * gain / gain.sum(), strict=True):
            records.append({"feature": feature, "repeat": repeat, "importance": importance})

    return (
        pl.DataFrame(records)
        .group_by("feature")
        .agg(
            pl.col("importance").mean().alias("mean_importance"),
            pl.col("importance").std().fill_null(0).alias("std_importance"),
        )
        .sort("mean_importance")
    )
```

```python
entry_importance = repeated_gain_importance(
    X_train,
    y_entry_train,
    FEATURE_COLS,
    seed=SEED + 300,
    repeats=N_IMPORTANCE_REPEATS,
)
enhanced_importance = repeated_gain_importance(
    X_train_enhanced,
    y_exit_train[meta_train_mask],
    FEATURE_COLS_ENHANCED,
    seed=SEED + 400,
    repeats=N_IMPORTANCE_REPEATS,
)
entry_prediction_rank = (
    enhanced_importance.sort("mean_importance", descending=True)
    .with_row_index("rank", offset=1)
    .filter(pl.col("feature") == "entry_prediction")["rank"]
    .item()
)
```

```python
fig = make_subplots(
    rows=1, cols=2, subplot_titles=["Entry Model Features", "Enhanced Exit Model Features"]
)

entry_imp_plot = entry_importance.to_pandas()
_ = fig.add_trace(
    go.Bar(
        x=entry_imp_plot["mean_importance"],
        y=entry_imp_plot["feature"],
        error_x={"type": "data", "array": entry_imp_plot["std_importance"], "visible": True},
        orientation="h",
        marker_color=COLORS["blue"],
        name="Entry",
    ),
    row=1,
    col=1,
)
```

The second panel uses the same normalized-gain scale and highlights only the stacked
entry-prediction feature before both panels receive a common range and title.

```python
exit_imp_plot = enhanced_importance.to_pandas()
colors = [
    COLORS["amber"] if feature == "entry_prediction" else COLORS["blue"]
    for feature in exit_imp_plot["feature"]
]
fig.add_trace(
    go.Bar(
        x=exit_imp_plot["mean_importance"],
        y=exit_imp_plot["feature"],
        error_x={"type": "data", "array": exit_imp_plot["std_importance"], "visible": True},
        orientation="h",
        marker_color=colors,
        name="Exit",
    ),
    row=1,
    col=2,
)

importance_xmax = 1.1 * max(
    (entry_importance["mean_importance"] + entry_importance["std_importance"]).max(),
    (enhanced_importance["mean_importance"] + enhanced_importance["std_importance"]).max(),
)
fig.update_layout(
    title=(
        "Feature importance in the entry and enhanced exit models"
        f"<br><sup>Mean normalized gain across {N_IMPORTANCE_REPEATS} seeded repeats; "
        "error bars show ±1 SD</sup>"
    ),
    width=950,
    height=500,
    margin={"l": 120, "r": 40, "t": 110, "b": 70},
    showlegend=False,
)
fig.update_xaxes(title_text="Mean normalized gain importance (%)", range=[0, importance_xmax])
show_plotly_with_alt(
    fig,
    "Two panels of horizontal bars on a shared importance scale, the entry model on the left "
    "and the enhanced exit model on the right, each feature carrying an error bar for its "
    "repeat-to-repeat spread and the bars ordered from most important down. Both panels fall "
    "away steeply from their top feature. The stacked entry-prediction feature is highlighted "
    "in the right-hand panel, in the middle of that ordering rather than at either end.",
)
```

```python
display(
    Markdown(
        f"""**Interpretation**: Across {N_IMPORTANCE_REPEATS} seeded fits,
`entry_prediction` ranks #{entry_prediction_rank} by mean normalized gain in the enhanced exit
model. Gain importance is an impurity-based allocation measure, not a signed or causal effect; the
error bars show its repeat-to-repeat dispersion."""
    )
)
```

```python
# Signal distribution
fig = make_subplots(
    rows=1, cols=2, subplot_titles=["Entry Signal Distribution", "Exit Signal Distribution"]
)

fig.add_trace(
    go.Histogram(
        x=entry_proba_test,
        xbins={"start": 0, "end": 1, "size": PROBABILITY_BIN_WIDTH},
        name="Entry probability",
        marker_color=COLORS["blue"],
    ),
    row=1,
    col=1,
)
fig.add_vline(x=entry_threshold, line_dash="dash", line_color=COLORS["amber"], row=1, col=1)

fig.add_trace(
    go.Histogram(
        x=exit_proba_enhanced,
        xbins={"start": 0, "end": 1, "size": PROBABILITY_BIN_WIDTH},
        name="Exit probability",
        marker_color=COLORS["slate"],
    ),
    row=1,
    col=2,
)
fig.add_vline(x=exit_threshold, line_dash="dash", line_color=COLORS["amber"], row=1, col=2)

fig.update_layout(
    title="Predicted probability by model, with each decision threshold",
    height=400,
    showlegend=False,
)
fig.update_xaxes(title_text="Predicted probability", range=[0, 1])
fig.update_yaxes(title_text="Test bars (count)")
show_plotly_with_alt(
    fig,
    "Two histograms of predicted probability sharing a scale and bin width, the entry model on "
    "the left and the exit model on the right, each with a dashed vertical line at its "
    "decision threshold. The entry distribution is pressed hard against the low end and decays "
    "away long before its threshold; the exit distribution is a broad hump centred close to "
    "its own threshold, so the line cuts through the bulk of it rather than past the tail.",
)
```

**Interpretation**: Both panels share a bin width and a common probability scale. Letting each
panel choose its own bins would make two distributions look different because their histograms
were built differently, which is the wrong reason for a reader to see a difference.

```python
signal_quintile_breaks = np.quantile(entry_proba_test, [0.2, 0.4, 0.6, 0.8])
signal_quintile_labels = ["Q1 (Weak)", "Q2", "Q3", "Q4", "Q5 (Strong)"]
publication_df = pl.DataFrame(
    {
        "signal": entry_proba_test,
        "fwd_return": test_df["fwd_return"],
        "signal_quintile": [
            signal_quintile_labels[index]
            for index in np.digitize(entry_proba_test, signal_quintile_breaks)
        ],
    }
).with_columns(
    pl.when(pl.col("fwd_return") < 0)
    .then(pl.lit("Adverse move"))
    .when(pl.col("fwd_return") >= entry_return_threshold)
    .then(pl.lit("Strong upside"))
    .otherwise(pl.lit("Neutral"))
    .alias("outcome")
)
```

Aggregate each ordered quintile into exhaustive outcome shares and persist the exact
table used by the publication figure.

```python
plot_data = (
    publication_df.group_by("signal_quintile", "outcome")
    .len()
    .pivot(on="outcome", index="signal_quintile", values="len")
    .join(
        pl.DataFrame({"signal_quintile": signal_quintile_labels, "quintile_order": range(5)}),
        on="signal_quintile",
        how="right",
    )
    .fill_null(0)
    .sort("quintile_order")
    .with_columns(pl.sum_horizontal("Adverse move", "Neutral", "Strong upside").alias("total"))
    .with_columns(
        (100 * pl.col(outcome) / pl.col("total")).alias(outcome)
        for outcome in ["Adverse move", "Neutral", "Strong upside"]
    )
)

# Persist quintile-conditioned outcome shares for the figure-19.5 publication script.
plot_data.drop("quintile_order", "total").write_parquet(
    OUTPUT_DIR / "signal_quintile_outcomes.parquet"
)
```

```python
strong_upside = plot_data["Strong upside"].to_numpy()
adverse_move = plot_data["Adverse move"].to_numpy()
strong_direction = (
    "rises monotonically" if np.all(np.diff(strong_upside) >= 0) else "is non-monotonic"
)
adverse_direction = "declines" if adverse_move[-1] < adverse_move[0] else "increases"
outcome_title = (
    "Stronger entry signals concentrate strong-upside outcomes"
    if np.all(np.diff(strong_upside) >= 0)
    else "Signal quintiles reveal non-monotonic outcome conditioning"
)

fig_pub, ax_pub = plt.subplots(figsize=(8.2, 4.4), constrained_layout=True)
bottom = np.zeros(plot_data.height)
for label, color in zip(
    ["Adverse move", "Neutral", "Strong upside"],
    [COLORS["negative"], COLORS["silver_muted"], COLORS["positive"]],
    strict=False,
):
    values = plot_data[label].to_numpy()
    ax_pub.bar(
        plot_data["signal_quintile"].to_list(),
        values,
        bottom=bottom,
        label=label,
        color=color,
        edgecolor="white",
        linewidth=0.6,
    )
    bottom = bottom + values
ax_pub.set_ylabel("Outcome share (%)")
ax_pub.set_xlabel("Signal quintile")
ax_pub.set_title(outcome_title)
ax_pub.set_ylim(0, 112)
ax_pub.set_yticks(range(0, 101, 20))
ax_pub.legend(loc="upper center", ncols=3, frameon=False)
show_with_alt(
    fig_pub,
    "Stacked bars of outcome share by signal quintile, each bar split into adverse move, neutral "
    "and strong upside. The strong-upside band grows steadily across the quintiles from almost "
    "nothing in the weakest. The adverse-move band does not follow it: it rises over the first "
    "four quintiles and then falls back in the strongest, so the two bands are not simply "
    "trading off against each other.",
)
```

```python
display(
    Markdown(
        f"""**Interpretation**: Across test-sample probability quintiles, the strong-upside share
{strong_direction} from {strong_upside[0]:.1f}% to {strong_upside[-1]:.1f}%. The adverse-move share
{adverse_direction} from {adverse_move[0]:.1f}% to {adverse_move[-1]:.1f}%. These are descriptive
test-sample bins whose cut points are estimated from the displayed test probabilities; they are not
an independently validated trading rule."""
    )
)
```

## 11. Key Takeaways

```python
takeaways = pl.DataFrame(
    {
        "Metric": [
            "Entry AUC",
            "Exit AUC (basic)",
            "Exit AUC (enhanced)",
            "AUC delta (enhanced − basic)",
            "Confidence-drop rule firing rate",
        ],
        "Value": [
            f"{entry_auc:.3f}",
            f"{exit_auc_basic:.3f}",
            f"{exit_auc_enhanced:.3f}",
            f"{(exit_auc_enhanced - exit_auc_basic):+.3f} ({improvement:+.2f}%)",
            f"{confidence_fire_rate:.1%} of bars",
        ],
    }
)
takeaways
```

```python
meta_train_rows = meta_train_mask.sum()
display(
    Markdown(
        f"The purged out-of-fold pass supplies an entry probability for "
        f"**{meta_train_rows:,} training rows** without any of them coming from a model that "
        f"saw its own row. Basic and enhanced exit AUC are **{exit_auc_basic:.3f}** and "
        f"**{exit_auc_enhanced:.3f}**. On the test interval, **{best_mean_row['Strategy']}** has "
        f"the highest mean compounded trade return at **{best_mean_row['mean_return']:.2%}**, and "
        f"**{shortest_row['Strategy']}** the shortest average hold at "
        f"**{shortest_row['avg_holding']:.1f} bars**."
    )
)
```

1. **A feature built from another model has to be produced out of sample.** The exit model takes
   the entry model's probability as an input. If that probability came from a model fitted on the
   same row, it is more accurate than anything production will ever supply, and the exit model
   learns to rely on an accuracy that will not be there. The expanding-window pass, with a purge
   at every fold boundary, is what makes the training feature resemble the live one.

2. **Purge wherever a forward-looking label meets a boundary.** Not only at the train-test split:
   every fold boundary inside the out-of-fold pass has the same problem, because the label on the
   last row of a fold reads into the next one.

3. **A better AUC is not a better strategy.** AUC scores the ranking of every bar. A trading rule
   acts on a threshold, holds a position for a while, and is judged on what the position earned.
   The two can move in opposite directions, which is why the comparison here reports both and
   reads the trade-level result rather than inferring it from the score.

4. **Check what a combined rule actually fires on.** Exiting when either of two conditions holds
   sounds like a compromise between them. When one condition is satisfied on almost every bar -
   as the confidence-drop clause is, because a rare-event model assigns low probability nearly
   everywhere - the combination is that condition, and the other contributes nothing.

5. **A threshold estimated on the training rows and applied unchanged is a test; one refitted on
   the test rows is not.** The entry cutoff here is a quantile of the training returns. Recomputing
   it on the test period would produce a better-looking result and measure nothing.

### Known limitations

- One fixed test interval, used once. It is not large enough to separate two models whose AUCs
  differ in the third decimal, and nothing here reports how much of that difference is sampling.
- No cost is charged. The rules differ mainly in how often they exit, which is exactly the
  dimension trading costs price, so a cost-aware comparison could reorder them.
- The quintile cut points in the signal analysis are estimated from the test probabilities being
  displayed, so those bins describe this sample rather than defining a rule that could be applied
  forward.
- The entry policy is deliberately sparse and spends most of the sample flat. Trade-level
  averages therefore rest on relatively few positions.

**Next**: `09_deep_hedging` replaces a rule that decides when to exit with a network that learns
how much to hold.

**Book reference**: Chapter 19, Section 19.7.
![notebook output](figures/p1_1.png)
![notebook output](figures/p1_2.png)
![notebook output](figures/p1_3.png)