import math import warnings import numpy as np import pandas as pd from sklearn.linear_model import LogisticRegression, Ridge from sklearn.ensemble import GradientBoostingClassifier, HistGradientBoostingClassifier from sklearn.preprocessing import StandardScaler from sklearn.pipeline import Pipeline from sklearn.metrics import accuracy_score, brier_score_loss, log_loss warnings.filterwarnings("ignore") try: from xgboost import XGBClassifier, XGBRegressor XGBOOST_AVAILABLE = True except Exception: XGBClassifier = None XGBRegressor = None XGBOOST_AVAILABLE = False def _safe_float(value, default=0.0): try: if value is None: return default value = float(value) if math.isnan(value) or math.isinf(value): return default return value except Exception: return default def _clip(value, low, high): return max(low, min(high, value)) def _returns(close): close = pd.Series(close).astype(float) return close.pct_change().replace([np.inf, -np.inf], np.nan) def _normal_cdf(x): try: return 0.5 * (1.0 + math.erf(float(x) / math.sqrt(2.0))) except Exception: return 0.5 def _rolling_zscore(series, window=72): series = pd.Series(series).astype(float) mean = series.rolling(window, min_periods=max(5, window // 4)).mean() std = series.rolling(window, min_periods=max(5, window // 4)).std() return (series - mean) / std.replace(0, np.nan) def _rsi_series(close, period=14): close = pd.Series(close).astype(float) delta = close.diff() gain = delta.clip(lower=0) loss = -delta.clip(upper=0) avg_gain = gain.ewm(alpha=1 / period, adjust=False, min_periods=period).mean() avg_loss = loss.ewm(alpha=1 / period, adjust=False, min_periods=period).mean() rs = avg_gain / avg_loss.replace(0, np.nan) rsi = 100 - (100 / (1 + rs)) return rsi.fillna(50) def _macd_hist_series(close): close = pd.Series(close).astype(float) ema12 = close.ewm(span=12, adjust=False).mean() ema26 = close.ewm(span=26, adjust=False).mean() macd = ema12 - ema26 signal = macd.ewm(span=9, adjust=False).mean() hist = macd - signal return hist.fillna(0) def _atr_series(high, low, close, period=14): high = pd.Series(high).astype(float) low = pd.Series(low).astype(float) close = pd.Series(close).astype(float) prev_close = close.shift(1) tr = pd.concat( [ high - low, (high - prev_close).abs(), (low - prev_close).abs(), ], axis=1, ).max(axis=1) atr = tr.ewm(alpha=1 / period, adjust=False, min_periods=period).mean() return atr.fillna(tr.rolling(period, min_periods=1).mean()) def _vwap_series(close, volume, window=24): close = pd.Series(close).astype(float) volume = pd.Series(volume).astype(float) pv = close * volume vwap = pv.rolling(window, min_periods=1).sum() / volume.rolling(window, min_periods=1).sum().replace(0, np.nan) return vwap.fillna(close) def get_quant_feature_columns(): return [ "ret_1", "ret_3", "ret_6", "ret_12", "ret_24", "mom_6", "mom_12", "mom_24", "volatility_12", "volatility_24", "volatility_72", "rsi_norm", "macd_hist_norm", "atr_pct", "vwap_gap", "ma20_gap", "ma50_gap", "ma100_gap", "rvol_24", "rvol_72", "rvol_120", "volume_z", "ofi", "ofi_6", "ofi_12", "ofi_24", "cvd_6_z", "cvd_12_z", "cvd_24_z", "range_position_72", "drawdown_72", "trend_strength", "volatility_regime", "bb_width_z", ] def build_quant_feature_frame(df, horizon=24): if df is None or len(df) < 80: return pd.DataFrame() work = df.copy() required = ["Open", "High", "Low", "Close", "Volume"] for col in required: if col not in work.columns: return pd.DataFrame() work = work[required].copy() work = work.replace([np.inf, -np.inf], np.nan).dropna() if len(work) < 80: return pd.DataFrame() open_ = work["Open"].astype(float) high = work["High"].astype(float) low = work["Low"].astype(float) close = work["Close"].astype(float) volume = work["Volume"].astype(float) ret = _returns(close) features = pd.DataFrame(index=work.index) features["ret_1"] = ret features["ret_3"] = close.pct_change(3) features["ret_6"] = close.pct_change(6) features["ret_12"] = close.pct_change(12) features["ret_24"] = close.pct_change(24) features["mom_6"] = close / close.shift(6) - 1 features["mom_12"] = close / close.shift(12) - 1 features["mom_24"] = close / close.shift(24) - 1 features["volatility_12"] = ret.rolling(12, min_periods=6).std() features["volatility_24"] = ret.rolling(24, min_periods=12).std() features["volatility_72"] = ret.rolling(72, min_periods=24).std() rsi = _rsi_series(close) macd_hist = _macd_hist_series(close) atr = _atr_series(high, low, close) vwap = _vwap_series(close, volume) features["rsi_norm"] = (rsi - 50) / 50 features["macd_hist_norm"] = macd_hist / close.replace(0, np.nan) features["atr_pct"] = atr / close.replace(0, np.nan) features["vwap_gap"] = (close - vwap) / vwap.replace(0, np.nan) ma20 = close.rolling(20, min_periods=5).mean() ma50 = close.rolling(50, min_periods=10).mean() ma100 = close.rolling(100, min_periods=20).mean() features["ma20_gap"] = (close - ma20) / ma20.replace(0, np.nan) features["ma50_gap"] = (close - ma50) / ma50.replace(0, np.nan) features["ma100_gap"] = (close - ma100) / ma100.replace(0, np.nan) volume_safe = volume.replace(0, np.nan).ffill().fillna(1.0) features["rvol_24"] = volume_safe / volume_safe.rolling(24, min_periods=6).mean() features["rvol_72"] = volume_safe / volume_safe.rolling(72, min_periods=12).mean() features["rvol_120"] = volume_safe / volume_safe.rolling(120, min_periods=20).mean() features["volume_z"] = _rolling_zscore(volume_safe, 72) candle_range = (high - low).replace(0, np.nan) close_position = ((close - low) / candle_range).clip(0, 1).fillna(0.5) buy_volume = volume_safe * close_position sell_volume = volume_safe - buy_volume volume_delta = buy_volume - sell_volume features["ofi"] = (volume_delta / volume_safe.replace(0, np.nan)).fillna(0) features["ofi_6"] = features["ofi"].rolling(6, min_periods=2).mean() features["ofi_12"] = features["ofi"].rolling(12, min_periods=3).mean() features["ofi_24"] = features["ofi"].rolling(24, min_periods=6).mean() cvd = volume_delta.cumsum() features["cvd_6_z"] = _rolling_zscore(cvd - cvd.shift(6), 72) features["cvd_12_z"] = _rolling_zscore(cvd - cvd.shift(12), 72) features["cvd_24_z"] = _rolling_zscore(cvd - cvd.shift(24), 72) rolling_high = high.rolling(72, min_periods=24).max() rolling_low = low.rolling(72, min_periods=24).min() features["range_position_72"] = ((close - rolling_low) / (rolling_high - rolling_low).replace(0, np.nan)).clip(0, 1) features["drawdown_72"] = close / rolling_high.replace(0, np.nan) - 1 features["trend_strength"] = (ma20 - ma50) / close.replace(0, np.nan) features["volatility_regime"] = features["volatility_24"] / features["volatility_72"].replace(0, np.nan) bb_mid = close.rolling(20, min_periods=10).mean() bb_std = close.rolling(20, min_periods=10).std() bb_width = (bb_std * 4) / bb_mid.replace(0, np.nan) features["bb_width_z"] = _rolling_zscore(bb_width, 72) future_close = close.shift(-horizon) future_return = future_close / close - 1 future_low_rows = [] low_values = low.values for i in range(len(low_values)): end = min(len(low_values), i + horizon + 1) if i + 1 >= end: future_low_rows.append(np.nan) else: future_low_rows.append(np.nanmin(low_values[i + 1:end])) future_low = pd.Series(future_low_rows, index=work.index) future_drawdown = future_low / close - 1 features["future_return"] = future_return features["target_up"] = (future_return > 0).astype(int) features["future_drawdown"] = future_drawdown feature_cols = get_quant_feature_columns() features = features.replace([np.inf, -np.inf], np.nan) features[feature_cols] = features[feature_cols].fillna(method="ffill").fillna(0) features = features.dropna(subset=["future_return", "target_up"]) return features def classify_quant_regime(feature_row): trend = _safe_float(feature_row.get("trend_strength", 0)) volatility = _safe_float(feature_row.get("volatility_regime", 1)) bb_width_z = _safe_float(feature_row.get("bb_width_z", 0)) range_position = _safe_float(feature_row.get("range_position_72", 0.5)) drawdown = _safe_float(feature_row.get("drawdown_72", 0)) if trend > 0.015: trend_label = "Uptrend" elif trend < -0.015: trend_label = "Downtrend" else: trend_label = "Range" if volatility > 1.35: volatility_label = "High Volatility" elif volatility < 0.75: volatility_label = "Low Volatility" else: volatility_label = "Normal Volatility" if bb_width_z < -1.0: compression_label = "Volatility Compression" elif bb_width_z > 1.0: compression_label = "Volatility Expansion" else: compression_label = "Normal Bandwidth" if drawdown < -0.10 and range_position < 0.25: setup_label = "Oversold Mean-Reversion Zone" elif range_position > 0.80 and trend_label == "Uptrend": setup_label = "Momentum Continuation Zone" elif range_position < 0.25 and trend_label == "Downtrend": setup_label = "Breakdown Risk Zone" else: setup_label = "Neutral Setup Zone" return { "trend_regime": trend_label, "volatility_regime_label": volatility_label, "compression_regime": compression_label, "setup_regime": setup_label, } def _make_classifier(model_type="xgboost"): model_type = str(model_type or "xgboost").lower() if model_type in ["xgboost", "xgb"] and XGBOOST_AVAILABLE: return XGBClassifier( n_estimators=140, max_depth=3, learning_rate=0.04, subsample=0.85, colsample_bytree=0.85, min_child_weight=3, reg_lambda=2.5, reg_alpha=0.05, objective="binary:logistic", eval_metric="logloss", random_state=42, n_jobs=1, ) if model_type in ["hist_gradient_boosting", "hist", "hgb"]: return HistGradientBoostingClassifier( max_iter=160, learning_rate=0.04, max_leaf_nodes=18, l2_regularization=0.25, random_state=42, ) if model_type in ["logistic", "logit"]: return Pipeline( steps=[ ("scaler", StandardScaler()), ("model", LogisticRegression(max_iter=1000, class_weight="balanced")), ] ) return GradientBoostingClassifier( n_estimators=140, learning_rate=0.04, max_depth=3, subsample=0.85, random_state=42, ) def _make_return_model(model_type="xgboost"): model_type = str(model_type or "xgboost").lower() if model_type in ["xgboost", "xgb"] and XGBOOST_AVAILABLE: return XGBRegressor( n_estimators=140, max_depth=3, learning_rate=0.04, subsample=0.85, colsample_bytree=0.85, min_child_weight=3, reg_lambda=2.5, reg_alpha=0.05, objective="reg:squarederror", random_state=42, n_jobs=1, ) return Pipeline( steps=[ ("scaler", StandardScaler()), ("model", Ridge(alpha=2.5)), ] ) def _predict_probability(model, x): try: if hasattr(model, "predict_proba"): return float(model.predict_proba(x)[0][1]) pred = model.predict(x) return float(pred[0]) except Exception: return 0.5 def _calibrate_probability(probability, validation_probs, validation_actuals): probability = _clip(_safe_float(probability, 0.5), 0.01, 0.99) if validation_probs is None or validation_actuals is None: return probability validation_probs = np.asarray(validation_probs, dtype=float) validation_actuals = np.asarray(validation_actuals, dtype=float) mask = np.isfinite(validation_probs) & np.isfinite(validation_actuals) validation_probs = validation_probs[mask] validation_actuals = validation_actuals[mask] if len(validation_probs) < 40: return probability bins = np.linspace(0, 1, 6) calibrated = probability for i in range(len(bins) - 1): low = bins[i] high = bins[i + 1] if i == len(bins) - 2: bucket = (validation_probs >= low) & (validation_probs <= high) else: bucket = (validation_probs >= low) & (validation_probs < high) if low <= probability <= high and bucket.sum() >= 8: empirical = validation_actuals[bucket].mean() calibrated = 0.60 * probability + 0.40 * empirical break return _clip(calibrated, 0.01, 0.99) def _build_calibration_table(probs, actuals): if probs is None or actuals is None: return [] probs = np.asarray(probs, dtype=float) actuals = np.asarray(actuals, dtype=float) mask = np.isfinite(probs) & np.isfinite(actuals) probs = probs[mask] actuals = actuals[mask] if len(probs) < 20: return [] rows = [] bins = np.linspace(0, 1, 6) for i in range(len(bins) - 1): low = bins[i] high = bins[i + 1] if i == len(bins) - 2: bucket = (probs >= low) & (probs <= high) else: bucket = (probs >= low) & (probs < high) count = int(bucket.sum()) if count == 0: continue rows.append({ "bucket": f"{int(low * 100)}-{int(high * 100)}%", "samples": count, "avg_predicted_probability": float(probs[bucket].mean() * 100), "actual_win_rate": float(actuals[bucket].mean() * 100), }) return rows def _expected_calibration_error(probs, actuals): table = _build_calibration_table(probs, actuals) if not table: return 0.0 total = sum(row["samples"] for row in table) if total <= 0: return 0.0 ece = 0.0 for row in table: pred = row["avg_predicted_probability"] / 100 actual = row["actual_win_rate"] / 100 weight = row["samples"] / total ece += weight * abs(pred - actual) return float(ece) def walk_forward_quant_validation(feature_frame, model_type="xgboost", min_train_size=140, test_size=24, step_size=24): feature_cols = get_quant_feature_columns() if feature_frame is None or feature_frame.empty: return { "available": False, "walk_forward_accuracy": 0.0, "high_confidence_accuracy": 0.0, "brier_score": 0.0, "log_loss": 0.0, "calibration_error": 0.0, "calibration_table": [], "fold_results": [], "validation_probs": [], "validation_actuals": [], "avg_win_return": 0.0, "avg_loss_return": 0.0, "avg_drawdown": 0.0, "samples": 0, } data = feature_frame.copy() if len(data) < min_train_size + test_size: return { "available": False, "walk_forward_accuracy": 0.0, "high_confidence_accuracy": 0.0, "brier_score": 0.0, "log_loss": 0.0, "calibration_error": 0.0, "calibration_table": [], "fold_results": [], "validation_probs": [], "validation_actuals": [], "avg_win_return": 0.0, "avg_loss_return": 0.0, "avg_drawdown": 0.0, "samples": 0, } predictions = [] probabilities = [] actuals = [] future_returns = [] future_drawdowns = [] fold_results = [] end = min_train_size while end + test_size <= len(data): train = data.iloc[:end] test = data.iloc[end:end + test_size] x_train = train[feature_cols] y_train = train["target_up"].astype(int) x_test = test[feature_cols] y_test = test["target_up"].astype(int) if y_train.nunique() < 2: end += step_size continue model = _make_classifier(model_type) try: model.fit(x_train, y_train) probs = model.predict_proba(x_test)[:, 1] if hasattr(model, "predict_proba") else model.predict(x_test) preds = (probs >= 0.5).astype(int) acc = accuracy_score(y_test, preds) fold_results.append({ "train_samples": int(len(train)), "test_samples": int(len(test)), "accuracy": float(acc), "avg_probability": float(np.mean(probs)), }) predictions.extend(preds.tolist()) probabilities.extend(probs.tolist()) actuals.extend(y_test.tolist()) future_returns.extend(test["future_return"].tolist()) future_drawdowns.extend(test["future_drawdown"].tolist()) except Exception: pass end += step_size if len(actuals) < 20: return { "available": False, "walk_forward_accuracy": 0.0, "high_confidence_accuracy": 0.0, "brier_score": 0.0, "log_loss": 0.0, "calibration_error": 0.0, "calibration_table": [], "fold_results": fold_results, "validation_probs": probabilities, "validation_actuals": actuals, "avg_win_return": 0.0, "avg_loss_return": 0.0, "avg_drawdown": 0.0, "samples": len(actuals), } probabilities = np.asarray(probabilities, dtype=float) actuals = np.asarray(actuals, dtype=int) predictions = np.asarray(predictions, dtype=int) future_returns = np.asarray(future_returns, dtype=float) future_drawdowns = np.asarray(future_drawdowns, dtype=float) walk_acc = accuracy_score(actuals, predictions) * 100 high_conf_mask = (probabilities >= 0.62) | (probabilities <= 0.38) if high_conf_mask.sum() >= 5: high_conf_acc = accuracy_score(actuals[high_conf_mask], predictions[high_conf_mask]) * 100 else: high_conf_acc = 0.0 try: brier = brier_score_loss(actuals, probabilities) except Exception: brier = 0.0 try: ll = log_loss(actuals, np.clip(probabilities, 0.01, 0.99)) except Exception: ll = 0.0 calibration_table = _build_calibration_table(probabilities, actuals) calibration_error = _expected_calibration_error(probabilities, actuals) win_returns = future_returns[actuals == 1] loss_returns = future_returns[actuals == 0] avg_win_return = float(np.nanmean(win_returns) * 100) if len(win_returns) else 0.0 avg_loss_return = float(np.nanmean(loss_returns) * 100) if len(loss_returns) else 0.0 avg_drawdown = float(np.nanmean(future_drawdowns) * 100) if len(future_drawdowns) else 0.0 return { "available": True, "walk_forward_accuracy": float(walk_acc), "high_confidence_accuracy": float(high_conf_acc), "brier_score": float(brier), "log_loss": float(ll), "calibration_error": float(calibration_error), "calibration_table": calibration_table, "fold_results": fold_results, "validation_probs": probabilities.tolist(), "validation_actuals": actuals.tolist(), "avg_win_return": avg_win_return, "avg_loss_return": avg_loss_return, "avg_drawdown": avg_drawdown, "samples": int(len(actuals)), } def _feature_importance(model, feature_cols): try: raw_model = model if hasattr(model, "named_steps"): raw_model = model.named_steps.get("model", model) if hasattr(raw_model, "feature_importances_"): values = raw_model.feature_importances_ elif hasattr(raw_model, "coef_"): values = abs(raw_model.coef_).ravel() else: return [] total = float(np.sum(np.abs(values))) if total <= 0: return [] rows = [] for name, value in zip(feature_cols, values): rows.append({ "feature": name, "importance": float(abs(value) / total), }) rows = sorted(rows, key=lambda x: x["importance"], reverse=True) return rows[:12] except Exception: return [] def build_quant_explanation(prob_up, expected_return, expected_risk, edge_ratio, regime, feature_importance): rows = [] rows.append( f"The meta-model estimates a {prob_up:.1f}% probability of upside over the selected horizon." ) rows.append( f"Expected return is {expected_return:+.2f}% versus estimated risk of {expected_risk:.2f}%." ) rows.append( f"Risk-adjusted edge ratio is {edge_ratio:.2f}." ) rows.append( f"Current regime: {regime.get('trend_regime', 'Unknown')} · {regime.get('volatility_regime_label', 'Unknown')} · {regime.get('setup_regime', 'Unknown')}." ) if feature_importance: top = feature_importance[:5] drivers = ", ".join([f"{x['feature']} ({x['importance'] * 100:.1f}%)" for x in top]) rows.append(f"Top model drivers: {drivers}.") return "\n".join(rows) def empty_quant_result(horizon=24): return { "available": False, "horizon": horizon, "model_type": "Unavailable", "quant_decision": "WAIT", "probability_up": 50.0, "raw_probability_up": 50.0, "probability_down": 50.0, "expected_return": 0.0, "expected_risk": 0.0, "expected_drawdown": 0.0, "edge_ratio": 0.0, "risk_adjusted_score": 0.0, "reliability": "Low", "walk_forward_accuracy": 0.0, "high_confidence_accuracy": 0.0, "brier_score": 0.0, "log_loss": 0.0, "calibration_error": 0.0, "samples_tested": 0, "avg_win_return": 0.0, "avg_loss_return": 0.0, "avg_drawdown": 0.0, "calibration_table": [], "fold_results": [], "trend_regime": "Unknown", "volatility_regime_label": "Unknown", "compression_regime": "Unknown", "setup_regime": "Unknown", "feature_importance": [], "quant_explanation": "Quant model unavailable. Not enough historical data or model training failed.", } def quant_meta_forecast(df, horizon=24, model_type="xgboost"): feature_cols = get_quant_feature_columns() frame = build_quant_feature_frame(df, horizon=horizon) if frame.empty or len(frame) < 120: return empty_quant_result(horizon) validation = walk_forward_quant_validation( frame, model_type=model_type, min_train_size=140, test_size=24, step_size=24, ) train = frame.copy() x = train[feature_cols] y = train["target_up"].astype(int) if y.nunique() < 2: return empty_quant_result(horizon) classifier = _make_classifier(model_type) return_model = _make_return_model(model_type) drawdown_model = _make_return_model(model_type) try: classifier.fit(x, y) return_model.fit(x, train["future_return"].astype(float)) drawdown_model.fit(x, train["future_drawdown"].astype(float)) except Exception: return empty_quant_result(horizon) latest = frame.iloc[[-1]][feature_cols] latest_row = frame.iloc[-1] raw_prob = _predict_probability(classifier, latest) calibrated_prob = _calibrate_probability( raw_prob, validation.get("validation_probs", []), validation.get("validation_actuals", []), ) try: expected_return = float(return_model.predict(latest)[0]) * 100 except Exception: expected_return = 0.0 try: expected_drawdown = float(drawdown_model.predict(latest)[0]) * 100 except Exception: expected_drawdown = 0.0 expected_risk = abs(min(expected_drawdown, 0.0)) if expected_risk <= 0: expected_risk = abs(_safe_float(latest_row.get("volatility_24", 0), 0.01) * math.sqrt(horizon) * 100) edge_ratio = expected_return / expected_risk if expected_risk > 0 else 0.0 risk_adjusted_score = ( ((calibrated_prob - 0.50) * 140) + _clip(expected_return * 4, -25, 25) + _clip(edge_ratio * 8, -20, 20) ) risk_adjusted_score = _clip(risk_adjusted_score, -100, 100) prob_pct = calibrated_prob * 100 if prob_pct >= 72 and edge_ratio >= 1.50 and expected_return > 0: decision = "ENTER NOW" elif prob_pct >= 62 and edge_ratio >= 1.10 and expected_return > 0: decision = "BUY" elif prob_pct >= 55 and expected_return > 0: decision = "LEAN BUY" elif prob_pct <= 38 and expected_return < 0: decision = "SELL" elif prob_pct <= 45: decision = "LEAN SELL" else: decision = "WAIT" wf_acc = validation.get("walk_forward_accuracy", 0.0) samples = validation.get("samples", 0) brier = validation.get("brier_score", 0.0) if samples >= 80 and wf_acc >= 58 and brier <= 0.24: reliability = "High" elif samples >= 40 and wf_acc >= 52: reliability = "Medium" else: reliability = "Low" regime = classify_quant_regime(latest_row) importance = _feature_importance(classifier, feature_cols) model_label = "XGBoost" if model_type in ["xgboost", "xgb"] and XGBOOST_AVAILABLE else "Sklearn Fallback" explanation = build_quant_explanation( prob_up=prob_pct, expected_return=expected_return, expected_risk=expected_risk, edge_ratio=edge_ratio, regime=regime, feature_importance=importance, ) return { "available": True, "horizon": horizon, "model_type": model_label, "quant_decision": decision, "probability_up": float(prob_pct), "raw_probability_up": float(raw_prob * 100), "probability_down": float((1 - calibrated_prob) * 100), "expected_return": float(expected_return), "expected_risk": float(expected_risk), "expected_drawdown": float(expected_drawdown), "edge_ratio": float(edge_ratio), "risk_adjusted_score": float(risk_adjusted_score), "reliability": reliability, "walk_forward_accuracy": float(wf_acc), "high_confidence_accuracy": float(validation.get("high_confidence_accuracy", 0.0)), "brier_score": float(validation.get("brier_score", 0.0)), "log_loss": float(validation.get("log_loss", 0.0)), "calibration_error": float(validation.get("calibration_error", 0.0)), "samples_tested": int(samples), "avg_win_return": float(validation.get("avg_win_return", 0.0)), "avg_loss_return": float(validation.get("avg_loss_return", 0.0)), "avg_drawdown": float(validation.get("avg_drawdown", 0.0)), "calibration_table": validation.get("calibration_table", []), "fold_results": validation.get("fold_results", []), "trend_regime": regime.get("trend_regime", "Unknown"), "volatility_regime_label": regime.get("volatility_regime_label", "Unknown"), "compression_regime": regime.get("compression_regime", "Unknown"), "setup_regime": regime.get("setup_regime", "Unknown"), "feature_importance": importance, "quant_explanation": explanation, } def quant_multi_horizon_forecast(df, horizons=(6, 12, 24), model_type="xgboost"): results = {} for h in horizons: try: results[f"{h}h"] = quant_meta_forecast(df, horizon=h, model_type=model_type) except Exception: results[f"{h}h"] = empty_quant_result(h) return results def fuse_rule_signal_with_quant(rule_score, quant_result): rule_score = _safe_float(rule_score, 0.0) if not quant_result or not quant_result.get("available", False): if rule_score >= 35: return { "final_score": rule_score, "final_label": "ENTER NOW", "quant_weight": 0.0, } if rule_score >= 18: return { "final_score": rule_score, "final_label": "BUY", "quant_weight": 0.0, } if rule_score >= 8: return { "final_score": rule_score, "final_label": "LEAN BUY", "quant_weight": 0.0, } if rule_score <= -18: return { "final_score": rule_score, "final_label": "SELL", "quant_weight": 0.0, } if rule_score <= -8: return { "final_score": rule_score, "final_label": "LEAN SELL", "quant_weight": 0.0, } return { "final_score": rule_score, "final_label": "WAIT", "quant_weight": 0.0, } reliability = quant_result.get("reliability", "Low") quant_score = _safe_float(quant_result.get("risk_adjusted_score", 0.0), 0.0) if reliability == "High": quant_weight = 0.65 elif reliability == "Medium": quant_weight = 0.50 else: quant_weight = 0.35 final_score = (rule_score * (1 - quant_weight)) + (quant_score * quant_weight) probability_up = _safe_float(quant_result.get("probability_up", 50.0), 50.0) edge_ratio = _safe_float(quant_result.get("edge_ratio", 0.0), 0.0) expected_return = _safe_float(quant_result.get("expected_return", 0.0), 0.0) if probability_up >= 72 and edge_ratio >= 1.50 and final_score >= 35 and expected_return > 0: label = "ENTER NOW" elif probability_up >= 62 and final_score >= 18 and expected_return > 0: label = "BUY" elif probability_up >= 55 and final_score >= 8: label = "LEAN BUY" elif probability_up <= 38 and final_score <= -18: label = "SELL" elif probability_up <= 45 and final_score <= -8: label = "LEAN SELL" else: if final_score >= 35: label = "ENTER NOW" elif final_score >= 18: label = "BUY" elif final_score >= 8: label = "LEAN BUY" elif final_score <= -18: label = "SELL" elif final_score <= -8: label = "LEAN SELL" else: label = "WAIT" return { "final_score": float(final_score), "final_label": label, "quant_weight": float(quant_weight), }