Machine Learning Specification¶

# Option A: Rule-based (transparent, no training)
def classify_vol_regime(atr, atr_90d_pct):
    if atr_90d_pct < 25:
        return 'low'
    elif atr_90d_pct < 75:
        return 'normal'
    elif atr_90d_pct < 95:
        return 'high'
    else:
        return 'extreme'

# Option B: HMM (data-driven)
from hmmlearn import hmm
model = hmm.GaussianHMM(n_components=4, covariance_type="diag")
model.fit(features)  # features: returns, vol, volume

from arch import arch_model

def forecast_volatility(returns, horizon=1):
    """
    Forecast volatility using GARCH(1,1).
    Returns annualized volatility forecast.
    """
    model = arch_model(returns * 100, vol='Garch', p=1, q=1)
    fit = model.fit(disp='off')
    forecast = fit.forecast(horizon=horizon)
    # Convert to annualized
    daily_var = forecast.variance.values[-1, 0]
    return np.sqrt(daily_var * 252) / 100

def calculate_position_size(account_value, risk_pct, entry, stop, vol_forecast, vol_target=0.15):
    """
    Volatility-adjusted position sizing.

    vol_forecast: annualized volatility
    vol_target: target annualized volatility (15% default)
    """
    # Base risk per trade
    risk_amount = account_value * risk_pct
    stop_distance = abs(entry - stop)
    base_size = risk_amount / stop_distance

    # Vol adjustment (scale down in high vol)
    vol_scalar = vol_target / vol_forecast
    vol_scalar = np.clip(vol_scalar, 0.5, 1.5)  # Limit adjustment

    return base_size * vol_scalar

def create_signal_features(df, signal_bar):
    """
    Create features for signal filtering.
    All features use data BEFORE signal_bar only.
    """
    features = {}

    # Price action features (before signal)
    features['atr_ratio'] = df.loc[:signal_bar-1, 'atr'].iloc[-1] / df.loc[:signal_bar-1, 'atr'].mean()
    features['trend_strength'] = df.loc[:signal_bar-1, 'adx'].iloc[-1]
    features['rsi'] = df.loc[:signal_bar-1, 'rsi'].iloc[-1]

    # Session features
    features['hour'] = df.loc[signal_bar, 'timestamp'].hour
    features['day_of_week'] = df.loc[signal_bar, 'timestamp'].dayofweek
    features['is_killzone'] = is_killzone(df.loc[signal_bar, 'timestamp'])

    # Recent performance (no leakage - uses closed trades only)
    features['recent_win_rate'] = get_recent_win_rate(closed_trades, lookback=20)

    return features

def create_label(entry_price, exit_price, direction):
    """
    Binary label: 1 if trade was profitable, 0 otherwise.

    CRITICAL: Label is determined AFTER trade closes.
    Never use label information in features.
    """
    if direction == 'long':
        return 1 if exit_price > entry_price else 0
    else:
        return 1 if exit_price < entry_price else 0

from sklearn.ensemble import GradientBoostingClassifier
from sklearn.calibration import CalibratedClassifierCV

def train_signal_filter(X_train, y_train):
    """
    Train calibrated classifier for probability estimation.
    """
    base_model = GradientBoostingClassifier(
        n_estimators=100,
        max_depth=3,
        min_samples_leaf=20,
        random_state=42
    )

    # Calibrate for accurate probabilities
    model = CalibratedClassifierCV(base_model, method='isotonic', cv=5)
    model.fit(X_train, y_train)

    return model

def should_take_signal(model, features, min_prob=0.55):
    """
    Filter signals based on predicted probability.
    """
    prob = model.predict_proba(features.reshape(1, -1))[0, 1]

    return {
        'take_signal': prob >= min_prob,
        'probability': prob,
        'size_multiplier': np.clip((prob - 0.5) * 4, 0.5, 1.5)  # Scale size by confidence
    }

import numpy as np
from scipy.optimize import minimize

def risk_parity_weights(cov_matrix):
    """
    Calculate risk parity weights.
    Each strategy contributes equally to portfolio risk.
    """
    n = cov_matrix.shape[0]

    def risk_contribution(weights):
        port_vol = np.sqrt(weights @ cov_matrix @ weights)
        marginal_contrib = cov_matrix @ weights
        risk_contrib = weights * marginal_contrib / port_vol
        return risk_contrib

    def objective(weights):
        rc = risk_contribution(weights)
        target = np.ones(n) / n  # Equal risk contribution
        return np.sum((rc - target) ** 2)

    constraints = {'type': 'eq', 'fun': lambda w: np.sum(w) - 1}
    bounds = [(0.05, 0.5) for _ in range(n)]  # Min 5%, max 50% per strategy

    result = minimize(objective, np.ones(n)/n, bounds=bounds, constraints=constraints)
    return result.x

from sklearn.model_selection import TimeSeriesSplit

def purged_cv(X, y, n_splits=5, purge_gap=10, embargo_gap=5):
    """
    Time series CV with purge and embargo.

    purge_gap: bars removed between train and test
    embargo_gap: bars removed after test before next train
    """
    splits = []
    tscv = TimeSeriesSplit(n_splits=n_splits)

    for train_idx, test_idx in tscv.split(X):
        # Purge: remove overlap
        train_idx = train_idx[train_idx < test_idx.min() - purge_gap]
        # Embargo: remove recent from test
        test_idx = test_idx[test_idx > train_idx.max() + embargo_gap]

        if len(train_idx) > 0 and len(test_idx) > 0:
            splits.append((train_idx, test_idx))

    return splits

def walk_forward_regime_aware(X, y, regimes, n_windows=6):
    """
    Walk-forward validation ensuring each window
    contains multiple regime types.
    """
    results = []

    for i in range(n_windows):
        train_end = int(len(X) * (0.5 + i * 0.1))
        test_end = int(len(X) * (0.6 + i * 0.1))

        X_train, y_train = X[:train_end], y[:train_end]
        X_test, y_test = X[train_end:test_end], y[train_end:test_end]

        # Check regime coverage
        train_regimes = set(regimes[:train_end])
        test_regimes = set(regimes[train_end:test_end])

        results.append({
            'window': i,
            'train_regimes': train_regimes,
            'test_regimes': test_regimes,
            'regime_overlap': len(train_regimes & test_regimes)
        })

    return results

def check_calibration(y_true, y_prob, n_bins=10):
    """
    Check if predicted probabilities are well-calibrated.
    """
    from sklearn.calibration import calibration_curve

    fraction_positives, mean_predicted = calibration_curve(
        y_true, y_prob, n_bins=n_bins
    )

    # Calibration error
    ece = np.mean(np.abs(fraction_positives - mean_predicted))

    return {
        'expected_calibration_error': ece,
        'pass': ece < 0.1,  # <10% error is acceptable
        'bins': list(zip(mean_predicted, fraction_positives))
    }