gap-trading-system / analysis /quant_model.py
l2345623's picture
Upload analysis/quant_model.py with huggingface_hub
9c83564 verified
Raw
History Blame Contribute Delete
29 kB
"""
Quantitative Gap Prediction Model v7 — 3-Model Ensemble
═══════════════════════════════════════════════════════
Production model for REAL MONEY trading on XM Broker (GOLD symbol).
Ensemble: GradientBoosting + LogisticRegression + RandomForest
Decision: Average of 3 model probabilities (most robust from backtesting)
Lot management (anti-martingale):
- Base lot: 0.01 per $20 capital
- Below 0.20 lot: loss reduces by 0.01
- At/above 0.20 lot: loss reduces by 0.02
- After win: restore to calculated level
- 15% risk cap per trade
- Hard cap at 1.00 lot
Features: 25 (20 raw + 5 interaction features)
"""
from __future__ import annotations
import asyncio
import logging
import math
from pathlib import Path
from typing import Any
import numpy as np
import pandas as pd
logger = logging.getLogger("gap_system.analysis.quant_model")
try:
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import GradientBoostingClassifier, RandomForestClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import accuracy_score, log_loss, brier_score_loss
_sklearn_available = True
except ImportError:
_sklearn_available = False
logger.warning("scikit-learn not installed -- quant model disabled")
# ═══════════════════════════════════════════════════════════════════════════════
# FEATURE ENGINEERING
# ═══════════════════════════════════════════════════════════════════════════════
def _encode_cyclic(value: float, period: float) -> tuple[float, float]:
angle = 2 * math.pi * value / period
return math.sin(angle), math.cos(angle)
def _safe_float(val, default=0.0) -> float:
if val is None:
return default
try:
f = float(val)
return default if (math.isnan(f) or math.isinf(f)) else f
except (ValueError, TypeError):
return default
def _engineer_features_from_row(row: dict, asset: str = "") -> dict:
"""Transform a DB row into the 25-feature vector."""
features = {}
is_gold = "XAU" in asset.upper() or "GOLD" in asset.upper()
features["vix_close"] = _safe_float(row.get("vix_close"), 15.0)
features["vix_change_5d"] = _safe_float(row.get("vix_change_5d"))
features["volume_spike"] = _safe_float(row.get("volume_spike"), 1.0)
features["weekly_return"] = _safe_float(row.get("weekly_return"))
features["rsi_14"] = _safe_float(row.get("rsi_14"), 50.0)
features["macd_hist"] = _safe_float(row.get("macd_hist"))
features["ema_spread"] = _safe_float(row.get("ema_spread"))
features["bb_width"] = _safe_float(row.get("bb_width"), 0.02)
features["dxy_change"] = _safe_float(row.get("dxy_change"))
prev = row.get("prev_gap_dir", "NONE")
features["prev_gap_dir"] = 1.0 if prev == "BULLISH" else (-1.0 if prev == "BEARISH" else 0.0)
friday_date = row.get("friday_date", "2020-01-01")
try:
month = pd.Timestamp(friday_date).month
except Exception:
month = 1
sin_m, cos_m = _encode_cyclic(month, 12)
features["month_sin"] = sin_m
features["month_cos"] = cos_m
features["atr_14"] = _safe_float(row.get("atr_14"), 1.0)
features["gap_atr_ratio"] = _safe_float(row.get("gap_atr_ratio"))
dom = _safe_float(row.get("day_of_month"), 15)
features["day_of_month_norm"] = dom / 31.0
features["prev_3_gaps_mean"] = _safe_float(row.get("prev_3_gaps_mean"))
features["gap_fill_rate"] = _safe_float(row.get("gap_fill_rate"), 0.6)
features["gold_dxy_ratio"] = _safe_float(row.get("gold_dxy_ratio")) if is_gold else 0.0
features["gld_momentum"] = _safe_float(row.get("gld_momentum")) if is_gold else 0.0
features["real_yield_proxy"] = _safe_float(row.get("real_yield_proxy"))
# 5 interaction features
features["vix_x_dxy"] = features["vix_close"] * features["dxy_change"]
rsi_extreme = max(0, features["rsi_14"] - 70) + max(0, 30 - features["rsi_14"])
features["rsi_extreme_x_momentum"] = rsi_extreme * features["weekly_return"]
features["squeeze_x_trend"] = features["bb_width"] * abs(features["ema_spread"])
features["vix_fear_regime"] = 1.0 if features["vix_close"] > 25 else 0.0
features["month_end"] = 1.0 if dom >= 28 else 0.0
return features
FEATURE_COLUMNS = [
"vix_close", "vix_change_5d", "volume_spike", "weekly_return",
"rsi_14", "macd_hist", "ema_spread", "bb_width",
"dxy_change", "prev_gap_dir", "month_sin", "month_cos",
"atr_14", "gap_atr_ratio", "day_of_month_norm",
"prev_3_gaps_mean", "gap_fill_rate",
"gold_dxy_ratio", "gld_momentum", "real_yield_proxy",
"vix_x_dxy", "rsi_extreme_x_momentum", "squeeze_x_trend",
"vix_fear_regime", "month_end",
]
# ═══════════════════════════════════════════════════════════════════════════════
# LOT MANAGER — Anti-Martingale with Tiered Reduction
# ═══════════════════════════════════════════════════════════════════════════════
class LotManager:
"""
Manages lot sizing with anti-martingale (reduce on loss).
Rules:
- Base lot = 0.01 per $20 capital
- Below 0.20 lot: each loss reduces by 0.01
- At/above 0.20 lot: each loss reduces by 0.02
- After each win: restore to calculated level
- 15% risk cap per trade
- Hard cap at 1.00 lot
"""
def __init__(self, starting_capital: float = 50.0,
capital_per_step: float = 20.0,
base_lot: float = 0.01,
max_lot: float = 0.50,
max_risk_pct: float = 0.15,
max_loss_per_micro: float = 15.0):
self.capital = starting_capital
self.capital_per_step = capital_per_step
self.base_lot = base_lot
self.max_lot = max_lot
self.max_risk_pct = max_risk_pct
self.max_loss_per_micro = max_loss_per_micro
self._consecutive_losses = 0
self._last_result = None
self._last_lot = base_lot
def get_lot(self) -> float:
"""Calculate current lot size based on capital, loss streak, and risk cap."""
# Base lot from capital steps (+0.01 per $20)
calculated = self.base_lot * max(1, int(self.capital / self.capital_per_step))
# Tiered loss reduction (anti-martingale)
if calculated >= 0.20:
reduction_per_loss = 0.02 # bigger lots = bigger reduction
else:
reduction_per_loss = 0.01
reduction = reduction_per_loss * self._consecutive_losses
adjusted = max(self.base_lot, calculated - reduction)
# RISK CAP: max loss per trade must not exceed max_risk_pct of capital
risk_capped = adjusted
if self.capital > 0 and self.max_loss_per_micro > 0:
max_risk_dollars = self.capital * self.max_risk_pct
max_safe_lots = max_risk_dollars / self.max_loss_per_micro
risk_capped = max(self.base_lot, round(max_safe_lots * 100) / 100 * self.base_lot)
adjusted = min(adjusted, risk_capped)
# Apply hard cap
final = min(adjusted, self.max_lot)
self._last_lot = final
logger.info(
"LotManager: capital=$%.2f, calc=%.2f, losses=%d, risk_cap=%.2f, final=%.2f",
self.capital, calculated, self._consecutive_losses, risk_capped, final
)
return round(final, 2)
def record_result(self, profit: float):
"""Record trade result to adjust future lot sizing."""
self.capital += profit
self.capital = max(0, self.capital)
if profit > 0:
self._consecutive_losses = 0
self._last_result = "WIN"
elif profit < 0:
self._consecutive_losses += 1
self._last_result = "LOSS"
else:
self._last_result = "BREAKEVEN"
logger.info(
"LotManager: result=%s ($%.2f), capital=$%.2f, streak=%d",
self._last_result, profit, self.capital, self._consecutive_losses
)
def get_status(self) -> dict:
return {
"capital": self.capital,
"current_lot": self.get_lot(),
"consecutive_losses": self._consecutive_losses,
"last_result": self._last_result,
}
# ═══════════════════════════════════════════════════════════════════════════════
# QUANT MODEL — 3-Model Ensemble
# ═══════════════════════════════════════════════════════════════════════════════
class QuantGapModel:
"""
3-model ensemble: GradientBoosting + LogisticRegression + RandomForest.
Averages probabilities from all 3 for the final prediction.
"""
def __init__(self, db_path: Path | None = None):
self.db_path = db_path
self.models = {}
self.scaler = None
self.is_trained = False
self.train_accuracy = 0.0
self.train_samples = 0
self.asset_models: dict[str, dict] = {}
self._feat_cols = FEATURE_COLUMNS
self.lot_manager = LotManager()
async def train(self) -> dict:
"""Train 3-model ensemble from historical_gaps table."""
if not _sklearn_available:
return {"status": "fallback", "reason": "sklearn_not_installed"}
if self.db_path is None or not self.db_path.exists():
return {"status": "skipped", "reason": "no_db"}
import aiosqlite
try:
async with aiosqlite.connect(str(self.db_path)) as db:
db.row_factory = aiosqlite.Row
cursor = await db.execute("PRAGMA table_info(historical_gaps)")
columns = [row[1] for row in await cursor.fetchall()]
has_v3 = "atr_14" in columns
has_v2 = "rsi_14" in columns
if not has_v2:
return await self._train_basic(db)
cursor = await db.execute("""
SELECT * FROM historical_gaps
WHERE gap_direction IN ('BULLISH', 'BEARISH')
ORDER BY friday_date ASC
""")
rows = [dict(r) for r in await cursor.fetchall()]
if len(rows) < 30:
return {"status": "insufficient_data", "samples": len(rows)}
feat_cols = FEATURE_COLUMNS if has_v3 else [
c for c in FEATURE_COLUMNS
if c not in ("atr_14", "gap_atr_ratio", "day_of_month_norm",
"prev_3_gaps_mean", "gap_fill_rate",
"gold_dxy_ratio", "gld_momentum", "real_yield_proxy",
"vix_x_dxy", "rsi_extreme_x_momentum",
"squeeze_x_trend", "vix_fear_regime", "month_end")
]
self._feat_cols = feat_cols
X_rows, y = [], []
for row in rows:
asset = row.get("asset", "")
feats = _engineer_features_from_row(row, asset)
X_rows.append([feats.get(c, 0.0) for c in feat_cols])
y.append(1 if row["gap_direction"] == "BULLISH" else 0)
X = np.nan_to_num(np.array(X_rows, dtype=np.float64), nan=0.0, posinf=0.0, neginf=0.0)
y = np.array(y)
self.scaler = StandardScaler()
X_scaled = self.scaler.fit_transform(X)
# Train 3 global models
self.models["gbm"] = GradientBoostingClassifier(
n_estimators=200, max_depth=4, learning_rate=0.05,
subsample=0.8, min_samples_leaf=10, random_state=42)
self.models["gbm"].fit(X_scaled, y)
self.models["lr"] = LogisticRegression(
C=1.0, penalty="l2", solver="lbfgs",
max_iter=1000, class_weight="balanced", random_state=42)
self.models["lr"].fit(X_scaled, y)
self.models["rf"] = RandomForestClassifier(
n_estimators=300, max_depth=6,
min_samples_leaf=10, class_weight="balanced", random_state=42)
self.models["rf"].fit(X_scaled, y)
self.is_trained = True
probs = self._ensemble_predict(X_scaled)
y_pred = (probs > 0.5).astype(int)
self.train_accuracy = accuracy_score(y, y_pred)
self.train_samples = len(y)
try:
brier = brier_score_loss(y, probs)
logloss = log_loss(y, probs)
except Exception:
brier, logloss = 0.0, 0.0
# Per-asset 3-model ensembles
assets = sorted(set(r["asset"] for r in rows))
for asset in assets:
asset_rows = [r for r in rows if r["asset"] == asset]
if len(asset_rows) < 20:
continue
Xa = np.array([
[_engineer_features_from_row(r, asset).get(c, 0.0) for c in feat_cols]
for r in asset_rows
], dtype=np.float64)
Xa = np.nan_to_num(Xa, nan=0.0, posinf=0.0, neginf=0.0)
ya = np.array([1 if r["gap_direction"] == "BULLISH" else 0 for r in asset_rows])
scaler_a = StandardScaler()
Xa_s = scaler_a.fit_transform(Xa)
msl = max(5, len(ya) // 50)
asset_ens = {}
asset_ens["gbm"] = GradientBoostingClassifier(
n_estimators=200, max_depth=4, learning_rate=0.05,
subsample=0.8, min_samples_leaf=msl, random_state=42)
asset_ens["gbm"].fit(Xa_s, ya)
asset_ens["lr"] = LogisticRegression(
C=1.0, penalty="l2", solver="lbfgs",
max_iter=1000, class_weight="balanced", random_state=42)
asset_ens["lr"].fit(Xa_s, ya)
asset_ens["rf"] = RandomForestClassifier(
n_estimators=300, max_depth=6, min_samples_leaf=msl,
class_weight="balanced", random_state=42)
asset_ens["rf"].fit(Xa_s, ya)
probs_a = np.mean([m.predict_proba(Xa_s)[:, 1] for m in asset_ens.values()], axis=0)
acc_a = accuracy_score(ya, (probs_a > 0.5).astype(int))
self.asset_models[asset] = {
"models": asset_ens, "scaler": scaler_a,
"accuracy": acc_a, "samples": len(ya),
}
logger.info("Ensemble [%s]: %.1f%% on %d samples", asset, acc_a * 100, len(ya))
metrics = {
"status": "trained",
"model_type": "3-model ensemble (GBM + LR + RF)",
"global_accuracy": round(self.train_accuracy, 4),
"brier_score": round(brier, 4),
"log_loss": round(logloss, 4),
"samples": self.train_samples,
"features": len(feat_cols),
"assets": {a: {"acc": round(m["accuracy"], 4), "n": m["samples"]}
for a, m in self.asset_models.items()},
}
logger.info("Ensemble trained: %.1f%% accuracy (%d samples, %d features)",
self.train_accuracy * 100, self.train_samples, len(feat_cols))
return metrics
except Exception as e:
logger.error("Model training failed: %s", e, exc_info=True)
return {"status": "error", "error": str(e)}
def _ensemble_predict(self, X_scaled: np.ndarray) -> np.ndarray:
"""Average probabilities from all 3 models."""
probs = []
for name, model in self.models.items():
try:
probs.append(model.predict_proba(X_scaled)[:, 1])
except Exception:
pass
if not probs:
return np.full(X_scaled.shape[0], 0.5)
return np.mean(probs, axis=0)
async def _train_basic(self, db) -> dict:
try:
cursor = await db.execute("""
SELECT asset, gap_direction, COUNT(*) as cnt
FROM historical_gaps WHERE gap_direction IN ('BULLISH', 'BEARISH')
GROUP BY asset, gap_direction
""")
totals = await cursor.fetchall()
self._basic_priors = {}
for row in totals:
asset, direction, count = row[0], row[1], row[2]
if asset not in self._basic_priors:
self._basic_priors[asset] = {"BULLISH": 0, "BEARISH": 0}
self._basic_priors[asset][direction] = count
total = sum(v["BULLISH"] + v["BEARISH"] for v in self._basic_priors.values())
self.is_trained = True
self.train_samples = total
return {"status": "basic_fallback", "samples": total}
except Exception as e:
return {"status": "error", "error": str(e)}
async def predict_live(self, asset: str) -> dict:
"""Generate a live prediction using the 3-model ensemble."""
default = {
"bull_prob": 0.5, "bear_prob": 0.5, "confidence": 0.0,
"direction": "NEUTRAL", "samples": 0,
"explanation": "No trained model available",
"lot_size": self.lot_manager.get_lot(),
"model_agreement": {},
}
if not self.is_trained:
return default
model_asset = "XAUUSD" if asset == "GOLD" else asset
try:
features = await self._get_live_features(model_asset)
feat_cols = self._feat_cols
X = np.array([[features.get(c, 0.0) for c in feat_cols]], dtype=np.float64)
X = np.nan_to_num(X, nan=0.0, posinf=0.0, neginf=0.0)
if _sklearn_available and self.models and self.scaler is not None:
if model_asset in self.asset_models:
am = self.asset_models[model_asset]
X_s = am["scaler"].transform(X)
model_set = am["models"]
model_label = f"asset-specific ({am['samples']})"
else:
X_s = self.scaler.transform(X)
model_set = self.models
model_label = f"global ({self.train_samples})"
individual = {}
probs_list = []
for name, model in model_set.items():
try:
p = model.predict_proba(X_s)[0]
bp = float(p[1]) if len(p) > 1 else 0.5
individual[name] = {
"bull_prob": round(bp, 4),
"direction": "BULLISH" if bp > 0.5 else "BEARISH",
}
probs_list.append(bp)
except Exception:
pass
if not probs_list:
return default
bull_prob = float(np.mean(probs_list))
bear_prob = 1.0 - bull_prob
elif hasattr(self, "_basic_priors") and model_asset in self._basic_priors:
counts = self._basic_priors[model_asset]
total = counts["BULLISH"] + counts["BEARISH"]
bull_prob = (counts["BULLISH"] + 1) / (total + 2)
bear_prob = 1.0 - bull_prob
individual = {}
model_label = f"frequency ({total})"
else:
return default
if bull_prob > 0.50:
direction = "BULLISH"
confidence = bull_prob
else:
direction = "BEARISH"
confidence = bear_prob
agree_count = sum(1 for m in individual.values() if m["direction"] == direction)
total_models = len(individual)
samples = self.asset_models.get(model_asset, {}).get("samples", self.train_samples)
explanation = (
f"Ensemble v7 ({model_label}): P(bull)={bull_prob:.3f}, P(bear)={bear_prob:.3f}. "
f"Agreement: {agree_count}/{total_models} models. "
f"VIX={features.get('vix_close', 0):.1f}, RSI={features.get('rsi_14', 50):.1f}"
)
return {
"bull_prob": round(bull_prob, 4),
"bear_prob": round(bear_prob, 4),
"confidence": round(confidence, 4),
"direction": direction,
"samples": samples,
"explanation": explanation,
"lot_size": self.lot_manager.get_lot(),
"model_agreement": individual,
"agree_count": agree_count,
"total_models": total_models,
}
except Exception as e:
logger.error("Live prediction failed for %s: %s", asset, e)
default["explanation"] = f"Prediction error: {e}"
return default
async def _get_live_features(self, asset: str) -> dict:
"""Fetch current market features from yfinance."""
from data.price_data import fetch_candles, get_vix, get_volume_spike
from analysis.technical import ema, rsi, macd, bollinger_bands, atr as compute_atr
features = {c: 0.0 for c in FEATURE_COLUMNS}
is_gold = "XAU" in asset.upper() or "GOLD" in asset.upper()
try:
vix_val = await get_vix()
features["vix_close"] = vix_val
try:
import yfinance as yf
vix_hist = await asyncio.to_thread(
lambda: yf.download("^VIX", period="10d", interval="1d", progress=False))
if len(vix_hist) >= 6:
if isinstance(vix_hist.columns, pd.MultiIndex):
vix_hist.columns = vix_hist.columns.get_level_values(0)
vix_hist.columns = [c.lower() for c in vix_hist.columns]
features["vix_change_5d"] = _safe_float(
(vix_hist["close"].iloc[-1] - vix_hist["close"].iloc[-6]) / vix_hist["close"].iloc[-6] * 100)
except Exception:
pass
features["volume_spike"] = await get_volume_spike(asset)
df = await fetch_candles(asset, "1d", 50)
if not df.empty and len(df) >= 20:
close, high, low = df["close"], df["high"], df["low"]
if len(close) >= 6:
features["weekly_return"] = _safe_float(
(close.iloc[-1] - close.iloc[-6]) / close.iloc[-6] * 100)
rsi_series = rsi(close)
features["rsi_14"] = _safe_float(rsi_series.iloc[-1], 50.0)
_, _, hist = macd(close)
features["macd_hist"] = _safe_float(hist.iloc[-1])
ema10, ema20 = ema(close, 10), ema(close, 20)
price = float(close.iloc[-1])
if price > 0:
features["ema_spread"] = _safe_float((ema10.iloc[-1] - ema20.iloc[-1]) / price * 100)
upper, mid, lower = bollinger_bands(close)
mid_val = _safe_float(mid.iloc[-1])
if mid_val > 0:
features["bb_width"] = _safe_float((upper.iloc[-1] - lower.iloc[-1]) / mid_val, 0.02)
atr_series = compute_atr(high, low, close)
features["atr_14"] = _safe_float(atr_series.iloc[-1], 1.0)
# DXY
try:
import yfinance as yf
dxy_hist = await asyncio.to_thread(
lambda: yf.download("DX-Y.NYB", period="10d", interval="1d", progress=False))
if len(dxy_hist) >= 6:
if isinstance(dxy_hist.columns, pd.MultiIndex):
dxy_hist.columns = dxy_hist.columns.get_level_values(0)
dxy_hist.columns = [c.lower() for c in dxy_hist.columns]
features["dxy_change"] = _safe_float(
(dxy_hist["close"].iloc[-1] - dxy_hist["close"].iloc[-6]) / dxy_hist["close"].iloc[-6] * 100)
except Exception:
pass
# Time features
from datetime import datetime, timezone
now = datetime.now(timezone.utc)
sin_m, cos_m = _encode_cyclic(now.month, 12)
features["month_sin"] = sin_m
features["month_cos"] = cos_m
features["day_of_month_norm"] = now.day / 31.0
features["month_end"] = 1.0 if now.day >= 28 else 0.0
features["vix_fear_regime"] = 1.0 if features["vix_close"] > 25 else 0.0
# Previous gaps from DB
try:
import aiosqlite
if self.db_path and self.db_path.exists():
async with aiosqlite.connect(str(self.db_path)) as db:
cursor = await db.execute(
"SELECT gap_direction, gap_pct, gap_fill_rate FROM historical_gaps WHERE asset=? ORDER BY friday_date DESC LIMIT 3",
(asset,))
rows = await cursor.fetchall()
if rows:
features["prev_gap_dir"] = 1.0 if rows[0][0] == "BULLISH" else (-1.0 if rows[0][0] == "BEARISH" else 0.0)
features["prev_3_gaps_mean"] = _safe_float(np.mean([r[1] for r in rows if r[1]]))
features["gap_fill_rate"] = _safe_float(rows[0][2], 0.6)
except Exception:
pass
# Gold-specific
if is_gold:
try:
import yfinance as yf
gld = await asyncio.to_thread(
lambda: yf.download("GLD", period="10d", interval="1d", progress=False))
if len(gld) >= 6:
if isinstance(gld.columns, pd.MultiIndex):
gld.columns = gld.columns.get_level_values(0)
gld.columns = [c.lower() for c in gld.columns]
features["gld_momentum"] = _safe_float(
(gld["close"].iloc[-1] - gld["close"].iloc[-6]) / gld["close"].iloc[-6] * 100)
except Exception:
pass
try:
import yfinance as yf
tip = await asyncio.to_thread(
lambda: yf.download("TIP", period="10d", interval="1d", progress=False))
if len(tip) >= 6:
if isinstance(tip.columns, pd.MultiIndex):
tip.columns = tip.columns.get_level_values(0)
tip.columns = [c.lower() for c in tip.columns]
features["real_yield_proxy"] = _safe_float(
(tip["close"].iloc[-1] - tip["close"].iloc[-6]) / tip["close"].iloc[-6] * 100)
except Exception:
pass
# Interaction features
features["vix_x_dxy"] = features["vix_close"] * features["dxy_change"]
rsi_extreme = max(0, features["rsi_14"] - 70) + max(0, 30 - features["rsi_14"])
features["rsi_extreme_x_momentum"] = rsi_extreme * features["weekly_return"]
features["squeeze_x_trend"] = features["bb_width"] * abs(features["ema_spread"])
except Exception as e:
logger.error("Feature extraction failed for %s: %s", asset, e)
return features
def get_metrics(self) -> dict:
return {
"is_trained": self.is_trained,
"model_type": "3-model ensemble (GBM + LR + RF)",
"global_accuracy": round(self.train_accuracy, 4) if self.is_trained else None,
"train_samples": self.train_samples,
"feature_count": len(self._feat_cols),
"features": self._feat_cols,
"asset_models": {
a: {"accuracy": round(m["accuracy"], 4), "samples": m["samples"]}
for a, m in self.asset_models.items()
},
"lot_status": self.lot_manager.get_status(),
}