Loading...
Loading...
ML trading signal classifiers using XGBoost and LightGBM with walk-forward validation, SHAP feature importance, and threshold optimization
npx skill4agent add agiprolabs/claude-trading-skills signal-classificationimport numpy as np
def create_binary_labels(
prices: np.ndarray, horizon: int = 24, threshold: float = 0.01
) -> np.ndarray:
"""Create binary labels from forward returns.
Args:
prices: Array of prices.
horizon: Forward return lookback in bars.
threshold: Minimum return magnitude for a label.
Returns:
Array of labels: 1 (up), 0 (down), NaN (neutral).
"""
fwd_returns = np.roll(prices, -horizon) / prices - 1
fwd_returns[-horizon:] = np.nan
labels = np.where(fwd_returns > threshold, 1,
np.where(fwd_returns < -threshold, 0, np.nan))
return labels| Class | Condition | Typical threshold |
|---|---|---|
| Strong Up | fwd_return > +2% | High confidence long |
| Mild Up | +0.5% to +2% | Moderate confidence |
| Down | fwd_return < -0.5% | Avoid / short |
from sklearn.calibration import CalibratedClassifierCV
calibrated = CalibratedClassifierCV(base_model, cv=5, method="isotonic")
calibrated.fit(X_train, y_train)
probs = calibrated.predict_proba(X_test)[:, 1]Window 1: [===TRAIN===][GAP][=TEST=]
Window 2: [===TRAIN===][GAP][=TEST=]
Window 3: [===TRAIN===][GAP][=TEST=]
Window 4: [===TRAIN===][GAP][=TEST=]| Parameter | Value | Rationale |
|---|---|---|
| Train window | 30 days (720 hourly bars) | Enough data to learn, recent enough to be relevant |
| Test window | 7 days (168 hourly bars) | Enough predictions for statistical significance |
| Step size | 1 day (24 bars) | Overlap test windows for more data points |
| Gap (embargo) | Same as forward horizon | Prevents label leakage |
from typing import Iterator
def walk_forward_splits(
n_samples: int,
train_size: int = 720,
test_size: int = 168,
step_size: int = 24,
gap: int = 24,
) -> Iterator[tuple[np.ndarray, np.ndarray]]:
"""Generate walk-forward train/test index splits.
Args:
n_samples: Total number of samples.
train_size: Number of training samples per window.
test_size: Number of test samples per window.
step_size: Step between successive windows.
gap: Gap between train end and test start.
Yields:
Tuples of (train_indices, test_indices).
"""
start = 0
while start + train_size + gap + test_size <= n_samples:
train_idx = np.arange(start, start + train_size)
test_start = start + train_size + gap
test_idx = np.arange(test_start, test_start + test_size)
yield train_idx, test_idx
start += step_sizereferences/validation_methods.mdfeature-engineeringfrom xgboost import XGBClassifier
model = XGBClassifier(
n_estimators=200,
max_depth=4,
learning_rate=0.05,
subsample=0.8,
colsample_bytree=0.8,
eval_metric="logloss",
use_label_encoder=False,
random_state=42,
)
model.fit(
X_train, y_train,
eval_set=[(X_val, y_val)],
verbose=False,
)
probabilities = model.predict_proba(X_test)[:, 1]references/model_guide.mdimport shap
explainer = shap.TreeExplainer(model)
shap_values = explainer.shap_values(X_test)
# Summary plot (top 15 features)
shap.summary_plot(shap_values, X_test, max_display=15)# Explain a single prediction
shap.force_plot(explainer.expected_value, shap_values[0], X_test.iloc[0])def optimize_threshold(
probabilities: np.ndarray,
returns: np.ndarray,
thresholds: np.ndarray | None = None,
) -> tuple[float, float]:
"""Find threshold that maximizes profit factor.
Args:
probabilities: Model predicted probabilities.
returns: Actual forward returns.
thresholds: Thresholds to search over.
Returns:
Tuple of (best_threshold, best_profit_factor).
"""
if thresholds is None:
thresholds = np.arange(0.50, 0.85, 0.01)
best_threshold, best_pf = 0.5, 0.0
for t in thresholds:
signals = probabilities >= t
if signals.sum() < 10:
continue
signal_returns = returns[signals]
wins = signal_returns[signal_returns > 0].sum()
losses = abs(signal_returns[signal_returns < 0].sum())
pf = wins / losses if losses > 0 else 0.0
if pf > best_pf:
best_pf = pf
best_threshold = t
return best_threshold, best_pfscale_pos_weightnet_return = gross_return - 0.005 # 50 bps round-trip| Skill | Integration |
|---|---|
| Compute input features for the classifier |
| Backtest trading strategies from ML signals |
| Train separate models per regime, or use regime as a feature |
| Size positions based on classifier confidence |
| Apply portfolio-level risk limits to ML-generated signals |
references/model_guide.mdreferences/validation_methods.mdscripts/train_classifier.pyscripts/walk_forward_backtest.py# Core (required)
uv pip install pandas numpy scikit-learn
# Optional (recommended)
uv pip install xgboost lightgbm shap