feat: enhance GMM model handling and introduce spread efficiency filter in trading logic

main.py

@@ -95,6 +95,8 @@ def _train_for_tf(tf: str, balance: float, broker: str, params: dict):
         trans_cov=params.get("trans_cov"),
         save=True,
         tf=tf,
+        save_models=True,
+        broker=broker,
     )
     model_hmm, states, state_map = fit_hmm(
         df, n_states=params.get("n_states", TF_CONFIG[tf].get("n_states_default", 3))
@@ -295,7 +297,7 @@ def cmd_process(args):
             logger.error("Unknown timeframe '%s'. Valid: %s", tf, list(TF_CONFIG))
             continue
         try:
-            df = process_pipeline(save=True, tf=tf)
+            df = process_pipeline(save=True, tf=tf, save_models=True, broker=args.broker)
             logger.info(
                 "[%s] Done: %d bars | Range: %s to %s",
                 tf, len(df), df.index.min(), df.index.max(),

src/backtester.py

@@ -223,7 +223,13 @@ def vectorized_backtest(
         threshold=buy_th,
         short_threshold=short_threshold,   # None → symmetric default inside
     )
-    signals = raw_signals  # daily trade-count cap removed; DailyEquityGate governs live risk
+
+    # Spread efficiency filter: suppress signals where ATR / spread < 3.0.
+    # Both atr_norm and spread_frac are expressed as fractions of price so
+    # the ratio is scale-free and consistent across all timeframes.
+    _spread_frac = BROKER_CONFIGS.get(broker, BROKER_CONFIGS["standard"]).get("spread_frac", 0.0004)
+    _er_mask     = (atr_norm / _spread_frac) >= 3.0
+    signals = np.where(_er_mask, raw_signals, 0)
 
     # Determine pos_per_trade from the adaptive risk manager
     arm           = AdaptiveRiskManager(account_size, broker=broker)

src/mt5_trader.py

@@ -30,13 +30,15 @@ from src.processor import (
     compute_volatility,
     compute_rsi,
     compute_atr,
+    compute_gmm_vol_cluster,
+    load_gmm_model,
 )
 from src.engine_hmm import load_model as load_hmm, predict_states, get_model_path as hmm_model_path, MODEL_PATH as HMM_GENERIC_PATH
 from src.engine_xgb import (
     load_xgb_ensemble, get_predictions_ensemble, assign_vol_bucket, FEATURE_COLS,
     get_ensemble_path, ENSEMBLE_PKL_PATH as XGB_GENERIC_PATH,
 )
-from src.risk_manager import AdaptiveRiskManager, CENT_MULTIPLIER, DailyEquityGate
+from src.risk_manager import AdaptiveRiskManager, BROKER_CONFIGS, CENT_MULTIPLIER, DailyEquityGate
 
 logger = setup_logger(__name__)
 
@@ -576,6 +578,8 @@ def compute_live_features(
     trans_cov: float,
     feature_cols: list | None = None,
     mt5=None,
+    tf: str = "H1",
+    broker: str = "headway_cent",
 ):
     """Build the current-bar feature vector for XGBoost ensemble inference.
 
@@ -625,6 +629,17 @@ def compute_live_features(
             logger.debug("USDCHF live return: %.6f (source: MT5 bar)", usdchf_ret)
         feature_dict["usdchf_log_return"] = usdchf_ret
 
+    if "gmm_vol_cluster" in feature_cols:
+        try:
+            _gmm, _scaler = load_gmm_model(tf, broker)
+            _cluster = compute_gmm_vol_cluster(
+                df["volatility"].values, fitted_gmm=_gmm, fitted_scaler=_scaler
+            )
+            feature_dict["gmm_vol_cluster"] = float(_cluster[-1])
+        except FileNotFoundError:
+            logger.warning("GMM model missing for [%s/%s] — using cluster=0", tf, broker)
+            feature_dict["gmm_vol_cluster"] = 0.0
+
     features_df = pd.DataFrame([feature_dict])[feature_cols]
     atr_price   = atr_normalized * float(df["Close"].iloc[-1])
     return features_df, current_state, atr_price
@@ -922,10 +937,9 @@ def _run_loop_inner(tf: str, broker: str, account_size: float, mt5,
                                                 comment="GRX_Daily_Loss_Limit")
                         try:
                             send_telegram_msg(
-                                f"⚠️ <b>Daily Loss Limit Reached</b>\n"
-                                f"Loss: <b>${_loss_usd:.2f}</b> "
-                                f"({equity_gate.loss_pct*100:.0f}% of ${account_size:.2f})\n"
-                                "🔒 Trading suspended until next UTC day."
+                                f"🚨 UNIVERSAL SAFETY TRIGGERED [{tf}]: "
+                                f"Daily loss limit hit (${_loss_usd:.2f}). "
+                                "Bot locked for recovery."
                             )
                         except Exception:
                             pass
@@ -1004,12 +1018,25 @@ def _run_loop_inner(tf: str, broker: str, account_size: float, mt5,
             # ── 4. Compute live features + probability ────────────────────
             features_df, hmm_state, atr_price = compute_live_features(
                 DEFAULT_SYMBOL, tf_mt5, model_hmm, obs_cov, trans_cov,
-                feature_cols=feature_cols, mt5=mt5,
+                feature_cols=feature_cols, mt5=mt5, tf=tf, broker=broker,
             )
             signal_tracker["atr_price"] = atr_price   # cache for between-bar ATR trail
             _, _probs = get_predictions_ensemble(models_xgb, thresholds_xgb, features_df)
             prob = float(_probs[0])
 
+            # ── Spread efficiency filter ──────────────────────────────────────
+            # Skip signals where ATR / spread < 3.0 (market noise exceeds edge).
+            _spread_frac = BROKER_CONFIGS.get(broker, {}).get("spread_frac", 0.0004)
+            _atr_norm    = float(features_df["atr_normalized"].iloc[0])
+            _er          = _atr_norm / _spread_frac if _spread_frac > 0 else 999.0
+            if _er < 3.0:
+                logger.info(
+                    "Efficiency ratio %.2f < 3.0 (ATR=%.5f / spread=%.5f) — no signal.",
+                    _er, _atr_norm, _spread_frac,
+                )
+                time.sleep(POLL_INTERVAL_SEC)
+                continue
+
             # ── 5. Log bar info on every new bar (always visible) ─────────
             bar_str          = datetime.fromtimestamp(bar_time, timezone.utc).strftime("%Y-%m-%d %H:%M")
             state_lbl        = {0: "Bull", 1: "Bear", 2: "Chop"}.get(hmm_state, str(hmm_state))

src/optimizer.py

@@ -142,20 +142,22 @@ def make_objective(balance: float = 15.0, broker: str = "standard", tf: str = "H
         # so a more conservative range (0.55–0.60) ensures spread is covered.
         # H1/M15: wide range (0.50–0.85). Signal locking in the backtester ensures
         # trades are counted per entry (not per active bar), so the Recovery Factor
-        # naturally penalises low-conviction overtrading without artificial caps.
-        # short_threshold must stay below prob_threshold (no-trade zone must exist).
+        # Threshold ranges are TF-specific to match each timeframe's edge profile.
+        # The efficiency ratio filter (ATR/spread ≥ 3) now handles low-conviction
+        # bars at the backtest level, so these ranges can be set by market style.
+        # short_threshold must always stay below prob_threshold (no-trade zone).
         if tf.upper() == "M5":
-            if broker == "standard":
-                prob_threshold  = trial.suggest_float("prob_threshold",  0.55, 0.60)
-                short_threshold = trial.suggest_float("short_threshold", 0.40, 0.45)
-            else:
-                prob_threshold  = trial.suggest_float("prob_threshold",  0.50, 0.53)
-                short_threshold = trial.suggest_float("short_threshold", 0.44, 0.50)
+            # Scalp selectivity: wider range, efficiency filter gates low-ATR bars
+            prob_threshold  = trial.suggest_float("prob_threshold",  0.55, 0.80)
+            short_threshold = trial.suggest_float("short_threshold", 0.20, 0.45)
+        elif tf.upper() == "H1":
+            # Institutional selectivity: high-conviction entries only
+            prob_threshold  = trial.suggest_float("prob_threshold",  0.68, 0.88)
+            short_threshold = trial.suggest_float("short_threshold", 0.12, 0.32)
         else:
-            # H1/M15: wide range — Recovery Factor scoring penalises low-conviction
-            # overtrading naturally now that trade counting uses entries (not active bars).
-            prob_threshold  = trial.suggest_float("prob_threshold",  0.50, 0.85)
-            short_threshold = trial.suggest_float("short_threshold", 0.15, 0.50)
+            # M15 — swing selectivity
+            prob_threshold  = trial.suggest_float("prob_threshold",  0.62, 0.85)
+            short_threshold = trial.suggest_float("short_threshold", 0.15, 0.38)
 
         # Guard: thresholds must not overlap — a crossover means every bar gets
         # both a BUY and SELL signal simultaneously, which is nonsensical.
@@ -255,6 +257,14 @@ def make_objective(balance: float = 15.0, broker: str = "standard", tf: str = "H
                 oos_n = result.get("oos_n_trades", 0)
                 if oos_n < min_oos_trades:
                     return -10.0
+                # Equity kill switch: if OOS drawdown would bring $15 below $1, disqualify.
+                _kill_dd = (balance - 1.0) / balance   # e.g., 14/15 ≈ 0.933
+                if result.get("oos_max_drawdown", 0.0) > _kill_dd:
+                    logger.warning(
+                        "Trial %d: equity kill switch (OOS DD=%.3f > kill threshold=%.3f)",
+                        trial.number, result["oos_max_drawdown"], _kill_dd,
+                    )
+                    return -100.0
                 oos_result = {
                     "recovery_factor": result.get("oos_recovery_factor", 0.0),
                     "max_drawdown":    result["oos_max_drawdown"],

src/processor.py

@@ -1,11 +1,53 @@
+import pickle
 import numpy as np
 import pandas as pd
 from pathlib import Path
 from sklearn.mixture import GaussianMixture
+from sklearn.preprocessing import StandardScaler
 from src.logger import setup_logger
 
 logger = setup_logger(__name__)
 
+MODELS_DIR = Path("models")
+
+
+def get_gmm_paths(tf: str, broker: str = "headway_cent") -> tuple[Path, Path]:
+    """Return (gmm_path, scaler_path) for the given TF/broker pair."""
+    return (
+        MODELS_DIR / f"gmm_{tf.upper()}_{broker}.pkl",
+        MODELS_DIR / f"scaler_{tf.upper()}_{broker}.pkl",
+    )
+
+
+def save_gmm_model(gmm, scaler, tf: str, broker: str = "headway_cent") -> None:
+    """Persist the fitted GaussianMixture and StandardScaler to models/."""
+    gmm_path, scaler_path = get_gmm_paths(tf, broker)
+    MODELS_DIR.mkdir(parents=True, exist_ok=True)
+    with open(gmm_path, "wb") as f:
+        pickle.dump(gmm, f)
+    with open(scaler_path, "wb") as f:
+        pickle.dump(scaler, f)
+    logger.info("GMM saved: %s | Scaler: %s", gmm_path.name, scaler_path.name)
+
+
+def load_gmm_model(tf: str, broker: str = "headway_cent"):
+    """Load the fitted GaussianMixture + StandardScaler.
+
+    Raises FileNotFoundError if files are missing — run --mode train first.
+    """
+    gmm_path, scaler_path = get_gmm_paths(tf, broker)
+    if not gmm_path.exists() or not scaler_path.exists():
+        raise FileNotFoundError(
+            f"GMM/Scaler models not found for [{tf.upper()}/{broker}]. "
+            f"Run  python main.py --mode train --tf {tf.upper()} --broker {broker}  first."
+        )
+    with open(gmm_path, "rb") as f:
+        gmm = pickle.load(f)
+    with open(scaler_path, "rb") as f:
+        scaler = pickle.load(f)
+    logger.info("GMM + Scaler loaded [%s/%s]", tf.upper(), broker)
+    return gmm, scaler
+
 # Per-timeframe configuration ─────────────────────────────────────────────────
 # Kalman obs_cov controls smoothing: higher value = more smoothing (less trust
 # in raw observations). M15 is ~4× noisier than H1; M5 uses a low obs_cov
@@ -199,51 +241,66 @@ def compute_atr(df: pd.DataFrame, period: int = 14) -> pd.Series:
     return (atr / df["Close"]).rename("atr_normalized")
 
 
-def compute_gmm_vol_cluster(volatility: np.ndarray, n_components: int = 3) -> np.ndarray:
+def compute_gmm_vol_cluster(
+    volatility: np.ndarray,
+    n_components: int = 3,
+    fitted_gmm=None,
+    fitted_scaler=None,
+    return_models: bool = False,
+):
     """Cluster bars into Volatility Buckets (Low=0, Med=1, High=2) via GMM.
 
-    Fits a Gaussian Mixture Model on the rolling-volatility series and returns
-    integer bucket labels sorted by component mean.  This provides XGBoost a
-    second 'opinion' on market state — e.g. a breakout bar (High Vol, label 2)
-    versus genuine Chop (Low Vol, label 0) can look identical to the HMM but
-    the GMM cluster separates them reliably.
+    Training mode  (fitted_gmm=None):
+        Fit a StandardScaler then a GaussianMixture on non-NaN values.
+        Remap component labels so 0=lowest-volatility, n-1=highest.
+        Pass return_models=True to get (labels, gmm, scaler) back.
 
-    Falls back to quantile-based labelling if GMM collapses to fewer than
-    ``n_components`` unique labels (degenerate convergence on uniform data).
+    Inference mode (fitted_gmm + fitted_scaler provided):
+        Apply saved scaler.transform() then gmm.predict() — strictly no
+        re-fitting.  This prevents cluster drift between training and live data.
 
-    Args:
-        volatility: 1-D array of rolling-volatility values (from compute_volatility).
-        n_components: Number of GMM components (default 3 = Low/Med/High).
-
-    Returns:
-        Integer array of shape (N,) with values in {0, 1, …, n_components-1}.
+    Falls back to quantile buckets (training mode only) when GMM collapses.
     """
     valid_mask = ~np.isnan(volatility)
-    labels_out = np.zeros(len(volatility), dtype=np.int8)  # NaN rows → 0; dropped later by dropna
+    labels_out = np.zeros(len(volatility), dtype=np.int8)
+    vol_valid  = volatility[valid_mask].reshape(-1, 1)
 
-    vol_valid = volatility[valid_mask]
-    X = vol_valid.reshape(-1, 1)
-    gmm = GaussianMixture(n_components=n_components, random_state=42, n_init=5)
+    if fitted_gmm is not None and fitted_scaler is not None:
+        # ── Inference: use saved scaler + GMM, no fitting ──────────────────
+        X = fitted_scaler.transform(vol_valid)
+        raw_labels = fitted_gmm.predict(X)
+        order = np.argsort(fitted_gmm.means_.ravel())
+        remap = np.empty(fitted_gmm.n_components, dtype=int)
+        for rank, orig in enumerate(order):
+            remap[orig] = rank
+        labels_valid = remap[raw_labels].astype(np.int8)
+        labels_out[valid_mask] = labels_valid
+        return labels_out
+
+    # ── Training: fit scaler + GMM ──────────────────────────────────────────
+    scaler = StandardScaler()
+    X      = scaler.fit_transform(vol_valid)
+    gmm    = GaussianMixture(n_components=n_components, random_state=42, n_init=5)
     raw_labels = gmm.fit_predict(X)
 
-    # Remap: 0 = lowest-volatility component, n_components-1 = highest
     order = np.argsort(gmm.means_.ravel())
     remap = np.empty(n_components, dtype=int)
-    for rank, original_idx in enumerate(order):
-        remap[original_idx] = rank
+    for rank, orig in enumerate(order):
+        remap[orig] = rank
     labels_valid = remap[raw_labels].astype(np.int8)
 
-    # Fallback: if GMM collapsed to fewer unique labels, use quantile buckets
     if len(np.unique(labels_valid)) < n_components:
         logger.warning(
             "GMM collapsed to %d clusters (expected %d) — using quantile fallback.",
             len(np.unique(labels_valid)), n_components,
         )
-        boundaries = np.percentile(vol_valid, [100 / n_components * i
-                                               for i in range(1, n_components)])
-        labels_valid = np.digitize(vol_valid, boundaries).astype(np.int8)
+        boundaries = np.percentile(vol_valid.ravel(),
+                                   [100 / n_components * i for i in range(1, n_components)])
+        labels_valid = np.digitize(vol_valid.ravel(), boundaries).astype(np.int8)
 
     labels_out[valid_mask] = labels_valid
+    if return_models:
+        return labels_out, gmm, scaler
     return labels_out
 
 
@@ -252,6 +309,8 @@ def process_pipeline(
     trans_cov: float = None,
     save: bool = True,
     tf: str = "H1",
+    save_models: bool = False,
+    broker: str = "headway_cent",
 ) -> pd.DataFrame:
     """Run the full feature-engineering pipeline for the given timeframe.
 
@@ -279,7 +338,14 @@ def process_pipeline(
     df["rsi"] = compute_rsi(df["Close"])
     df["rsi_slope"] = df["rsi"].diff()
     df["atr_normalized"] = compute_atr(df)
-    df["gmm_vol_cluster"] = compute_gmm_vol_cluster(df["volatility"].values)
+    if save_models:
+        labels, _gmm, _scaler = compute_gmm_vol_cluster(
+            df["volatility"].values, return_models=True
+        )
+        df["gmm_vol_cluster"] = labels
+        save_gmm_model(_gmm, _scaler, tf=tf, broker=broker)
+    else:
+        df["gmm_vol_cluster"] = compute_gmm_vol_cluster(df["volatility"].values)
 
     # ── Optional USDCHF cross-asset feature ─────────────────────────────────
     # Each TF uses its own USDCHF master so bar frequencies stay in sync:

src/validator.py

@@ -33,6 +33,7 @@ from src.processor import (
     compute_rsi,
     compute_atr,
     compute_gmm_vol_cluster,
+    load_gmm_model,
 )
 from src.engine_hmm import load_model as load_hmm, predict_states, get_model_path as hmm_model_path, MODEL_PATH as HMM_GENERIC_PATH
 from src.engine_xgb import load_xgb_ensemble, prepare_features, get_predictions_ensemble, get_ensemble_path, ENSEMBLE_PKL_PATH as XGB_GENERIC_PATH
@@ -56,6 +57,7 @@ def _apply_features(
     tf: str,
     obs_cov: float,
     trans_cov: float,
+    broker: str = "headway_cent",
 ) -> pd.DataFrame:
     """Apply the same feature steps as process_pipeline() to an arbitrary frame.
 
@@ -75,7 +77,12 @@ def _apply_features(
     df["rsi"]            = compute_rsi(df["Close"])
     df["rsi_slope"]       = df["rsi"].diff()
     df["atr_normalized"]  = compute_atr(df)
-    df["gmm_vol_cluster"] = compute_gmm_vol_cluster(df["volatility"].values)
+
+    # Load the training GMM + scaler — strictly no re-fitting on live/validation data.
+    _gmm, _scaler = load_gmm_model(tf, broker)
+    df["gmm_vol_cluster"] = compute_gmm_vol_cluster(
+        df["volatility"].values, fitted_gmm=_gmm, fitted_scaler=_scaler
+    )
 
     # Mirror process_pipeline: add USDCHF if the matching TF master file exists
     usdchf_path = _USDCHF_PATH_BY_TF.get(tf.upper(), USDCHF_MASTER_PATH)
@@ -145,7 +152,7 @@ def run_validation(
     # Load and featurise data
     df = pd.read_csv(sync_data_path, index_col="Date", parse_dates=True)
     logger.info("Loaded %d bars from %s for validation", len(df), sync_data_path)
-    df = _apply_features(df, tf, obs_cov, trans_cov)
+    df = _apply_features(df, tf, obs_cov, trans_cov, broker=broker)
 
     # Load models — prefer broker+TF specific file; fall back to generic
     hmm_path = hmm_model_path(tf, broker)