feat: implement XRPUSDT drawdown-sensitive signal gate 15m strategy (closes #218)

This commit is contained in:
Stanley Isaac
2026-05-21 18:58:09 +00:00
parent 16f54e175e
commit 99a6929942
4 changed files with 272 additions and 1 deletions
@@ -41,4 +41,4 @@ XRPUSDT may show repeatable behavior when drawdown-sensitive signal gate conditi
This idea is intentionally Markdown-only. A future template can add `strategy.py`, `quant.config.json`, and a focused README once PPE results justify turning the idea into executable code.
Closes #29
Closes #218
@@ -0,0 +1,57 @@
# XRPUSDT Drawdown-sensitive signal gate 15m
This strategy implements a drawdown-sensitive signal gate approach for XRPUSDT on 15-minute candles using XGBoost.
## Overview
- **Pair**: XRPUSDT
- **Timeframe**: 15m
- **Model**: XGBoost with feature engineering focused on drawdown-sensitive signal gating
- **Goal**: Gate signals based on drawdown conditions to avoid trading during extreme negative periods
## Features Engineered
1. **Basic returns**: 1, 3, 6, 12 period returns
2. **Drawdown-sensitive features**:
- Running maximum (peak) and drawdown from peak
- Drawdown magnitude (absolute value)
- Drawdown state (in drawdown if >1% below peak)
- Drawdown recovery signals
- Recovery strength during drawdown periods
- Volatility-adjusted drawdown
- Drawdown percentiles (extreme drawdown detection)
3. **Range breakout features** (from original):
- Breakout signals above/below recent highs/lows
- Short and medium-term returns (4, 12 periods)
- Volatility measure
- Range percentage
- EMA slope
4. **ATR for normalization**: Average True Range for volatility measurement
5. **ATR-normalized candle range and close location value** (from idea)
6. **Distance from EMAs** (from idea): 20, 50, and 200 period EMAs
7. **Prior swing high and swing low distance** (from idea)
8. **Volume features**: Z-score and ratio to moving average
9. **Rolling volatility percentile** (from idea): Fast/slow volatility ratio, volatility percentile
## Configuration
See `quant.config.json` for hyperparameters:
- `lookback`: 100 candles for prediction
- `horizon`: 4 candles forward for labeling (1 hour for 15m timeframe)
- `threshold`: 0.003 (30 pips) for ATR-normalized breakout
- `min_confidence`: 0.52 minimum probability for signal generation
## Usage
This template follows the PyP Quant Mode contract:
```python
def train(data, config):
return model, metrics
def predict(model, market_data, config):
return {"signal": "UP|DOWN|HOLD", "confidence": 0.0, "metadata": {}}
```
## Disclaimer
Educational template only. Not financial advice. Past performance does not guarantee future results.
@@ -0,0 +1,30 @@
{
"pair": "XRPUSDT",
"timeframe": "15m",
"model_family": "XGBoost",
"runtime_target": "edge",
"artifact_format": "weights_bundle",
"parameters": {
"lookback": 100,
"horizon": 4,
"threshold": 0.003,
"min_confidence": 0.52
},
"training_requirements": [
"numpy",
"pandas",
"scikit-learn",
"xgboost",
"joblib"
],
"inference_requirements": [
"numpy",
"pandas",
"scikit-learn",
"xgboost",
"joblib"
],
"symbol": "XRPUSDT",
"description": "XRPUSDT drawdown-sensitive signal gate strategy using XGBoost",
"disclaimer": "Educational template only. Not financial advice."
}
@@ -0,0 +1,184 @@
import numpy as np
import pandas as pd
from xgboost import XGBClassifier
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
SYMBOL = "XRPUSDT"
MODEL_NAME = "xrpusdt-drawdown-sensitive-signal-gate-15m"
def _normalise(data):
df = data.copy()
df.columns = [str(c).lower() for c in df.columns]
if "volume" not in df.columns:
df["volume"] = 1.0
for col in ["open", "high", "low", "close", "volume"]:
df[col] = pd.to_numeric(df[col], errors="coerce")
return df.dropna(subset=["open", "high", "low", "close"]).reset_index(drop=True)
def _features(df):
c = df["close"]
h = df["high"]
l = df["low"]
v = df["volume"]
f = pd.DataFrame(index=df.index)
# Basic returns
for n in [1, 3, 6, 12]:
f[f"ret{n}"] = c.pct_change(n)
# Drawdown-sensitive features
# Calculate running maximum (peak)
rolling_max = c.rolling(window=50, min_periods=1).max()
# Drawdown from peak
drawdown = (c - rolling_max) / rolling_max
f["drawdown"] = drawdown
f["drawdown_magnitude"] = abs(drawdown) # Absolute drawdown
f["is_in_drawdown"] = (drawdown < -0.01).astype(int) # In drawdown if >1% below peak
# Drawdown recovery signals
f["drawdown_recovery"] = np.where(drawdown < 0, c.diff(), 0) # Positive movement during drawdown
f["recovery_strength"] = f["drawdown_recovery"].rolling(5).sum() / (abs(f["drawdown"]) + 1e-9)
# Volatility-adjusted drawdown
returns = c.pct_change()
volatility = returns.rolling(24).std()
f["vol_adjusted_drawdown"] = drawdown / (volatility + 1e-9)
# Drawdown percentiles (extreme drawdown detection)
f["drawdown_percentile"] = drawdown.rolling(100).apply(
lambda x: pd.Series(x).rank(pct=True).iloc[-1] if len(x) > 0 else 0.5, raw=False)
# Range breakout features (from original)
high_break = h.rolling(36).max().shift()
low_break = l.rolling(36).min().shift()
f["breakout_up"] = (c - high_break) / c
f["breakout_down"] = (c - low_break) / c
f["ret4"] = c.pct_change(4)
f["ret12"] = c.pct_change(12)
f["volatility"] = c.pct_change().rolling(24).std()
f["range_pct"] = (h - l) / c
f["ema_slope"] = c.ewm(span=12, adjust=False).mean().pct_change(6)
# ATR for normalization
tr = np.maximum(h - l, np.maximum(abs(h - c.shift(1)), abs(l - c.shift(1))))
atr = pd.Series(tr).rolling(14).mean()
f["atr"] = atr
# ATR-normalized candle range and close location value
f["range_pct"] = (h - l) / c
f["body_pct"] = (c - df["open"]) / (h - l).replace(0, np.nan)
f["close_pos"] = (c - l) / (h - l).replace(0, np.nan)
# Distance from EMAs (from idea)
f["ema20_dist"] = (c - c.ewm(span=20, adjust=False).mean()) / c
f["ema50_dist"] = (c - c.ewm(span=50, adjust=False).mean()) / c
f["ema200_dist"] = (c - c.ewm(span=200, adjust=False).mean()) / c
# Prior swing high and swing low distance (from idea)
swing_high = h.rolling(20, center=False).max().shift(1)
swing_low = l.rolling(20, center=False).min().shift(1)
f["dist_to_swing_high"] = (swing_high - c) / c
f["dist_to_swing_low"] = (c - swing_low) / c
# Volume features
f["volume_z"] = (v - v.rolling(48).mean()) / (v.rolling(48).std() + 1e-9)
f["volume_ratio"] = v / v.rolling(20).mean()
# Rolling volatility percentile (from idea)
returns = c.pct_change()
f["volatility_fast"] = returns.rolling(16).std()
f["volatility_slow"] = returns.rolling(64).std()
f["volatility_ratio"] = f["volatility_fast"] / (f["volatility_slow"] + 1e-9)
f["volatility_percentile"] = f["volatility_fast"].rolling(200).apply(
lambda x: pd.Series(x).rank(pct=True).iloc[-1] if len(x) > 0 else 0.5, raw=False)
return f.replace([np.inf, -np.inf], np.nan).dropna()
def _labels(close, index, horizon, threshold):
fwd = close.pct_change(horizon).shift(-horizon)
y = pd.Series(1, index=close.index)
y[fwd > threshold] = 2
y[fwd < -threshold] = 0
return y.reindex(index).fillna(1).astype(int)
def train(data, config):
params = config.get("parameters", {})
horizon = int(params.get("horizon", 4)) # 4 candles = 1 hour for 15m timeframe
threshold = float(params.get("threshold", 0.003))
df = _normalise(data)
feat = _features(df)
y = _labels(df["close"], feat.index, horizon, threshold)
# Create pipeline with StandardScaler and XGBoost
scaler = StandardScaler()
x_scaled = scaler.fit_transform(feat.values.astype(np.float32))
clf = XGBClassifier(
n_estimators=250,
max_depth=6,
learning_rate=0.04,
subsample=0.9,
colsample_bytree=0.9,
objective="multi:softprob",
num_class=3,
random_state=42,
n_jobs=-1
)
clf.fit(x_scaled, y.values)
# Create a pipeline-like object for consistency
model = {"scaler": scaler, "clf": clf}
preds = clf.predict(x_scaled)
metrics = {
"training_bars": int(len(feat)),
"feature_count": int(feat.shape[1]),
"buy_signals": int((preds == 2).sum()),
"sell_signals": int((preds == 0).sum()),
"hold_signals": int((preds == 1).sum()),
}
return {"model": model, "features": list(feat.columns), "symbol": SYMBOL}, metrics
def predict(model, market_data, config):
params = config.get("parameters", {})
lookback = int(params.get("lookback", 100))
min_conf = float(params.get("min_confidence", 0.52))
candles = market_data.get("candles", [])
if len(candles) < lookback:
return {"signal": "HOLD", "confidence": 0.0, "metadata": {"reason": "not_enough_candles", "model": MODEL_NAME}}
df = _normalise(pd.DataFrame(candles, columns=["open", "high", "low", "close", "volume"]))
feat = _features(df).tail(1)
if feat.empty:
return {"signal": "HOLD", "confidence": 0.0, "metadata": {"reason": "no_features", "model": MODEL_NAME}}
# Apply same preprocessing as in training
scaler = model["model"]["scaler"]
clf = model["model"]["clf"]
feat_scaled = scaler.transform(feat.values.astype(np.float32))
prob = clf.predict_proba(feat_scaled)[0]
klass = int(np.argmax(prob))
conf = float(np.max(prob))
signal = {0: "DOWN", 1: "HOLD", 2: "UP"}[klass]
if conf < min_conf:
signal = "HOLD"
return {"signal": signal, "confidence": round(conf, 4), "metadata": {
"p_sell": round(float(prob[0]), 4),
"p_hold": round(float(prob[1]), 4),
"p_buy": round(float(prob[2]), 4),
"model": MODEL_NAME,
"symbol": SYMBOL
}}