mirror of
https://github.com/PyP-Quant/quant-trading-strategy-templates.git
synced 2026-08-22 07:08:05 +00:00
177 lines
6.5 KiB
Python
177 lines
6.5 KiB
Python
import numpy as np
|
|
import pandas as pd
|
|
from sklearn.experimental import enable_hist_gradient_boosting # noqa
|
|
from sklearn.ensemble import HistGradientBoostingClassifier
|
|
from sklearn.pipeline import Pipeline
|
|
from sklearn.preprocessing import StandardScaler
|
|
|
|
|
|
SYMBOL = "EURGBP"
|
|
MODEL_NAME = "eurgbp-volume-proxy-divergence-5m"
|
|
|
|
|
|
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"]
|
|
|
|
# Basic returns
|
|
out = pd.DataFrame(index=df.index)
|
|
out["ret1"] = c.pct_change()
|
|
out["ret3"] = c.pct_change(3)
|
|
out["ret6"] = c.pct_change(6)
|
|
out["ret12"] = c.pct_change(12)
|
|
|
|
# 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()
|
|
out["atr"] = atr
|
|
|
|
# ATR-normalized candle range and close location value
|
|
out["range_pct"] = (h - l) / c
|
|
out["body_pct"] = (c - df["open"]) / (h - l).replace(0, np.nan)
|
|
out["close_pos"] = (c - l) / (h - l).replace(0, np.nan)
|
|
|
|
# Volume proxy divergence features
|
|
# Price-Volume divergence (similar to RSI but with volume)
|
|
# Calculate price changes
|
|
price_change = c.diff()
|
|
|
|
# Volume-weighted price change
|
|
vol_price_change = price_change * v
|
|
|
|
# RSI of price changes
|
|
gain = price_change.where(price_change > 0, 0)
|
|
loss = (-price_change.where(price_change < 0, 0))
|
|
avg_gain = gain.rolling(14).mean()
|
|
avg_loss = loss.rolling(14).mean()
|
|
rs = avg_gain / (avg_loss + 1e-9)
|
|
price_rsi = 100 - (100 / (1 + rs))
|
|
out["price_rsi"] = price_rsi
|
|
|
|
# RSI of volume-weighted price changes (volume proxy)
|
|
vol_gain = vol_price_change.where(vol_price_change > 0, 0)
|
|
vol_loss = (-vol_price_change.where(vol_price_change < 0, 0))
|
|
vol_avg_gain = vol_gain.rolling(14).mean()
|
|
vol_avg_loss = vol_loss.rolling(14).mean()
|
|
vol_rs = vol_avg_gain / (vol_avg_loss + 1e-9)
|
|
vol_price_rsi = 100 - (100 / (1 + vol_rs))
|
|
out["vol_price_rsi"] = vol_price_rsi
|
|
|
|
# Volume proxy divergence (difference between price RSI and volume-price RSI)
|
|
out["vol_price_divergence"] = out["price_rsi"] - out["vol_price_rsi"]
|
|
|
|
# Volume moving average convergence/divergence
|
|
vol_fast_ma = v.ewm(span=12, adjust=False).mean()
|
|
vol_slow_ma = v.ewm(span=26, adjust=False).mean()
|
|
out["vol_macd"] = vol_fast_ma - vol_slow_ma
|
|
out["vol_macd_signal"] = out["vol_macd"].ewm(span=9, adjust=False).mean()
|
|
out["vol_macd_hist"] = out["vol_macd"] - out["vol_macd_signal"]
|
|
|
|
# Distance from EMAs (from idea)
|
|
out["ema20_dist"] = (c - c.ewm(span=20, adjust=False).mean()) / c
|
|
out["ema50_dist"] = (c - c.ewm(span=50, adjust=False).mean()) / c
|
|
out["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)
|
|
out["dist_to_swing_high"] = (swing_high - c) / c
|
|
out["dist_to_swing_low"] = (c - swing_low) / c
|
|
|
|
# Volume features
|
|
out["volume_z"] = (v - v.rolling(48).mean()) / (v.rolling(48).std() + 1e-9)
|
|
out["volume_ratio"] = v / v.rolling(20).mean()
|
|
|
|
# Rolling volatility percentile (from idea)
|
|
returns = c.pct_change()
|
|
out["volatility_fast"] = returns.rolling(16).std()
|
|
out["volatility_slow"] = returns.rolling(64).std()
|
|
out["volatility_ratio"] = out["volatility_fast"] / (out["volatility_slow"] + 1e-9)
|
|
out["volatility_percentile"] = out["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 out.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", 5)) # 5m horizon for 5m timeframe
|
|
threshold = float(params.get("threshold", 0.0006))
|
|
df = _normalise(data)
|
|
feat = _features(df)
|
|
y = _labels(df["close"], feat.index, horizon, threshold)
|
|
|
|
model = Pipeline([
|
|
("scaler", StandardScaler()),
|
|
("clf", HistGradientBoostingClassifier(
|
|
learning_rate=0.08,
|
|
max_iter=120,
|
|
max_depth=6,
|
|
min_samples_leaf=12,
|
|
l2_regularization=0.1,
|
|
random_state=42
|
|
)),
|
|
])
|
|
model.fit(feat.values.astype(np.float32), y.values)
|
|
preds = model.predict(feat.values.astype(np.float32))
|
|
|
|
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.5))
|
|
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}}
|
|
|
|
proba = model["model"].predict_proba(feat.values.astype(np.float32))[0]
|
|
klass = int(np.argmax(proba))
|
|
conf = float(proba[klass])
|
|
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(proba[0]), 4),
|
|
"p_hold": round(float(proba[1]), 4),
|
|
"p_buy": round(float(proba[2]), 4),
|
|
"model": MODEL_NAME,
|
|
"symbol": SYMBOL
|
|
}} |