""" 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(), }