Skip to content

Volatility Normalization

Section E: Vol-normalized metrics and regime analysis.

volnorm

Volatility normalization and regime-based analysis.

normalize_metrics(trades)

Convert % MFE/MAE to volatility-adjusted units (R-multiples).

Parameters:

Name Type Description Default
trades TradeSet

Trade geometry data (must have vol_at_entry)

 required

Returns:

Type Description
TradeSet

New TradeSet with normalized MFE/MAE values
Notes

Normalization: MFE_norm = MFE / vol_at_entry This converts percentage moves into "volatility units" or "R-multiples".

Source code in signal_analyzer/analysis/volnorm.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def normalize_metrics(trades: TradeSet) -> TradeSet:
    """
    Convert % MFE/MAE to volatility-adjusted units (R-multiples).

    Parameters
    ----------
    trades : TradeSet
        Trade geometry data (must have vol_at_entry)

    Returns
    -------
    TradeSet
        New TradeSet with normalized MFE/MAE values

    Notes
    -----
    Normalization: MFE_norm = MFE / vol_at_entry
    This converts percentage moves into "volatility units" or "R-multiples".
    """
    if trades.vol_at_entry is None:
        raise ValueError("TradeSet must have vol_at_entry for normalization")

    # Avoid division by zero
    vol_safe = np.where(trades.vol_at_entry > 0, trades.vol_at_entry, np.nan)

    mfe_norm = trades.mfe / vol_safe
    mae_norm = trades.mae / vol_safe

    # Filter out invalid normalized values
    valid = np.isfinite(mfe_norm) & np.isfinite(mae_norm)

    # Create new TradeSet with normalized values
    return TradeSet(
        side=trades.side,
        n_trades=np.sum(valid),
        entry_idx=trades.entry_idx[valid],
        entry_price=trades.entry_price[valid],
        mfe=mfe_norm[valid],
        mae=mae_norm[valid],
        t_mfe=trades.t_mfe[valid],
        t_mae=trades.t_mae[valid],
        vol_at_entry=trades.vol_at_entry[valid],
        mfe_path=trades.mfe_path[valid] / vol_safe[valid, np.newaxis]
        if trades.mfe_path is not None
        else None,
        mae_path=trades.mae_path[valid] / vol_safe[valid, np.newaxis]
        if trades.mae_path is not None
        else None,
    )

split_by_vol_regime(trades, n_regimes=3, regime_method='quantile')

Split trades by volatility regimes.

Parameters:

Name Type Description Default
trades TradeSet

Trade geometry data (must have vol_at_entry)

 required
n_regimes int

Number of regimes (e.g., 3 for low/mid/high)

 3
regime_method (quantile, kmeans)

Method for defining regimes

 'quantile'

Returns:

Type Description
dict

Dictionary mapping regime names to TradeSet objects e.g., {'low': TradeSet(...), 'mid': TradeSet(...), 'high': TradeSet(...)}
Source code in signal_analyzer/analysis/volnorm.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def split_by_vol_regime(
    trades: TradeSet,
    n_regimes: int = 3,
    regime_method: str = "quantile",
) -> Dict[str, TradeSet]:
    """
    Split trades by volatility regimes.

    Parameters
    ----------
    trades : TradeSet
        Trade geometry data (must have vol_at_entry)
    n_regimes : int
        Number of regimes (e.g., 3 for low/mid/high)
    regime_method : {'quantile', 'kmeans'}
        Method for defining regimes

    Returns
    -------
    dict
        Dictionary mapping regime names to TradeSet objects
        e.g., {'low': TradeSet(...), 'mid': TradeSet(...), 'high': TradeSet(...)}
    """
    if trades.vol_at_entry is None:
        raise ValueError("TradeSet must have vol_at_entry for regime splitting")

    vol = trades.vol_at_entry

    if regime_method == "quantile":
        # Split by quantiles
        quantiles = np.linspace(0, 100, n_regimes + 1)
        edges = np.percentile(vol, quantiles)

        regime_labels = np.digitize(vol, edges[1:-1])

        # Label regimes
        if n_regimes == 3:
            regime_names = ["low", "mid", "high"]
        else:
            regime_names = [f"regime_{i}" for i in range(n_regimes)]

    elif regime_method == "kmeans":
        from sklearn.cluster import KMeans

        kmeans = KMeans(n_clusters=n_regimes, random_state=42)
        regime_labels = kmeans.fit_predict(vol.reshape(-1, 1))

        # Sort by cluster center (low to high vol)
        centers = kmeans.cluster_centers_.flatten()
        order = np.argsort(centers)
        label_map = {old: new for new, old in enumerate(order)}
        regime_labels = np.array([label_map[label] for label in regime_labels])

        if n_regimes == 3:
            regime_names = ["low", "mid", "high"]
        else:
            regime_names = [f"regime_{i}" for i in range(n_regimes)]

    else:
        raise ValueError(f"Unknown regime_method: {regime_method}")

    # Split into separate TradeSets
    result = {}
    for i, name in enumerate(regime_names):
        mask = regime_labels == i

        if np.sum(mask) == 0:
            continue

        result[name] = TradeSet(
            side=trades.side,
            n_trades=np.sum(mask),
            entry_idx=trades.entry_idx[mask],
            entry_price=trades.entry_price[mask],
            mfe=trades.mfe[mask],
            mae=trades.mae[mask],
            t_mfe=trades.t_mfe[mask],
            t_mae=trades.t_mae[mask],
            vol_at_entry=trades.vol_at_entry[mask],
            mfe_path=trades.mfe_path[mask] if trades.mfe_path is not None else None,
            mae_path=trades.mae_path[mask] if trades.mae_path is not None else None,
        )

    return result

compare_percent_vs_volnorm(trades)

Compare key metrics in % space vs vol-normalized space.

Parameters:

Name Type Description Default
trades TradeSet

Trade geometry data (must have vol_at_entry)

 required

Returns:

Type Description
dict

Dictionary with 'percent' and 'volnorm' keys, each containing metrics: - median_mfe, median_mae, mfe_p95, mae_p5
Source code in signal_analyzer/analysis/volnorm.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def compare_percent_vs_volnorm(
    trades: TradeSet,
) -> Dict[str, Dict[str, float]]:
    """
    Compare key metrics in % space vs vol-normalized space.

    Parameters
    ----------
    trades : TradeSet
        Trade geometry data (must have vol_at_entry)

    Returns
    -------
    dict
        Dictionary with 'percent' and 'volnorm' keys, each containing metrics:
        - median_mfe, median_mae, mfe_p95, mae_p5
    """
    from ..analysis.geometry import geometry_metrics

    # Original % metrics
    pct_metrics = geometry_metrics(trades)

    # Normalized metrics
    norm_trades = normalize_metrics(trades)
    norm_metrics = geometry_metrics(norm_trades)

    return {
        "percent": {
            "median_mfe": pct_metrics["median_mfe"],
            "median_mae": pct_metrics["median_mae"],
            "mfe_p95": pct_metrics["mfe_p95"],
            "mae_p5": pct_metrics["mae_p5"],
        },
        "volnorm": {
            "median_mfe": norm_metrics["median_mfe"],
            "median_mae": norm_metrics["median_mae"],
            "mfe_p95": norm_metrics["mfe_p95"],
            "mae_p5": norm_metrics["mae_p5"],
        },
    }

regime_dependence_analysis(trades, n_regimes=3)

Analyze metrics per volatility regime.

Parameters:

Name Type Description Default
trades TradeSet

Trade geometry data (must have vol_at_entry)

 required
n_regimes int

Number of regimes

 3

Returns:

Type Description
dict

Dictionary mapping regime names to metrics
Source code in signal_analyzer/analysis/volnorm.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def regime_dependence_analysis(
    trades: TradeSet,
    n_regimes: int = 3,
) -> Dict[str, Dict[str, float]]:
    """
    Analyze metrics per volatility regime.

    Parameters
    ----------
    trades : TradeSet
        Trade geometry data (must have vol_at_entry)
    n_regimes : int
        Number of regimes

    Returns
    -------
    dict
        Dictionary mapping regime names to metrics
    """
    from ..analysis.geometry import geometry_metrics

    regimes = split_by_vol_regime(trades, n_regimes=n_regimes)

    result = {}
    for regime_name, regime_trades in regimes.items():
        metrics = geometry_metrics(regime_trades)
        result[regime_name] = {
            "n_trades": metrics["n_trades"],
            "median_mfe": metrics["median_mfe"],
            "median_mae": metrics["median_mae"],
            "win_rate": metrics["win_rate"],
            "mfe_mae_corr": metrics["mfe_mae_corr"],
            "avg_vol": float(np.mean(regime_trades.vol_at_entry)),
        }

    return result