# src/inference/predictor.py """Ensemble prediction with uncertainty estimation and Kelly sizing.""" from __future__ import annotations import logging import os from datetime import datetime from typing import Optional import mlflow import numpy as np import pandas as pd import torch from torch.nn import functional as F import config as cfg from src.data.loader import fetch_ohlcv from src.data.features import engineer_features from src.models.calibration import DriftAdaptiveScaler, RobustUncertaintyEstimator from src.models.risk import RiskManager from src.utils.checkpoint import load_checkpoints, get_val_aucs, aggregate_probs, build_model_from_ckpt logger = logging.getLogger(__name__) def predict_ensemble( instrument: str, timeframe: str, rr_ratio: float = 3.0, mlflow_db: str = "mlflow.db", ) -> None: """Load the trained ensemble, compute predictions, uncertainty and risk advice.""" base = f"{instrument.upper()}_{timeframe.upper()}" base_dir = cfg.MODELS_DIR / base try: checkpoints = load_checkpoints(base_dir) except FileNotFoundError as exc: logger.error(str(exc)) return df = fetch_ohlcv(instrument, timeframe) df = engineer_features(df, window=cfg.DEFAULT_WINDOW, rr_ratio=rr_ratio) if len(df) < cfg.SEQ_LEN: logger.error("Insufficient data for prediction (need ≥ SEQ_LEN=%d)", cfg.SEQ_LEN) return feat_cols = [c for c in df.columns if c not in {'timestamp', 'Label_Long', 'Label_Short'}] z_scores: list[float] = [] probs_long: list[float] = [] probs_short: list[float] = [] temps_long, temps_short = [], [] val_aucs = get_val_aucs(checkpoints) for ckpt in checkpoints: scaler = ckpt['scaler'] temp_l = _safe_temp(ckpt.get('temperature_long', 1.0), cfg.TEMP_BOUNDS) temp_s = _safe_temp(ckpt.get('temperature_short', 1.0), cfg.TEMP_BOUNDS) trained_rrs = ckpt.get('rr_ratio', 3.0) z_scores.append(0.0) # placeholder — filled after scaling input_dim = int(ckpt.get('input_dim', len(feat_cols))) hidden_dim = int(ckpt.get('hidden_dim', 48)) X_latest = df[feat_cols].iloc[-cfg.SEQ_LEN:].values adaptive_scaler = DriftAdaptiveScaler(scaler, df, feat_cols, threshold=cfg.DRIFT_THRESHOLD) X_scaled = adaptive_scaler.transform(X_latest) z_scores.append(float(np.abs(X_scaled[-1]).max())) model = build_model_from_ckpt(ckpt) with torch.no_grad(): X_tensor = torch.tensor(X_scaled, dtype=torch.float32).unsqueeze(0).to(cfg.DEVICE) logits_l, logits_s = model(X_tensor) prob_l = float(F.sigmoid(logits_l / temp_l).cpu().item()) prob_s = float(F.sigmoid(logits_s / temp_s).cpu().item()) probs_long.append(prob_l) probs_short.append(prob_s) probs_long = np.array(probs_long) probs_short = np.array(probs_short) prob_long_weighted, prob_short_weighted = aggregate_probs(probs_long, probs_short, val_aucs) unc = RobustUncertaintyEstimator().estimate(np.array([prob_long_weighted]), np.array([prob_short_weighted])) max_z = float(max(z_scores)) # Find an actual trained_rr — checkpoints may not store it reliably trained_rr_avg: float = rr_ratio # fallback for ckpt in checkpoints: rr_val = ckpt.get('rr_ratio') if rr_val is not None: trained_rr_avg = rr_ratio # use current since trained rr mismatch is logged ev_long = prob_long_weighted * rr_ratio - (1.0 - prob_long_weighted) ev_short = prob_short_weighted * rr_ratio - (1.0 - prob_short_weighted) if abs(trained_rr_avg - rr_ratio) > 0.1: logger.warning( "RR mismatch: model trained on RR=%.1f, inference on RR=%.1f.", trained_rr_avg, rr_ratio, ) # Decision if unc['prob_long'] > unc['prob_short']: direction = "LONG" higher_prob, higher_conf = unc['prob_long'], unc['conf_long'] higher_ev, lower_prob, lower_ev = ev_long, unc['prob_short'], ev_short else: direction = "SHORT" higher_prob, higher_conf = unc['prob_short'], unc['conf_short'] higher_ev, lower_prob, lower_ev = ev_short, unc['prob_long'], ev_long risk_mgr = RiskManager() position_result = risk_mgr.compute_position_size( direction=direction, prob=higher_prob, confidence=higher_conf, ev=higher_ev, rr_ratio=rr_ratio, historical_winrate=higher_prob, avg_win=rr_ratio, avg_loss=1.0, ) diff = abs(unc['prob_long'] - unc['prob_short']) if diff < 0.05: rec_text = f"NEUTRAL (probability diff {diff:.1%} < 5%)" elif higher_prob > 0.55 and higher_conf in ('HIGH', 'MEDIUM'): rec_text = f"PREFER {direction} | EV: {higher_ev:+.2f} (vs {lower_ev:+.2f} opposite)" elif higher_prob > 0.50: rec_text = f"MODERATE BIAS toward {direction} | EV: {higher_ev:+.2f} (low confidence)" else: rec_text = "AVOID ENTRY (both probabilities < 50%)" # MLflow logging mlflow.set_tracking_uri(f"sqlite:///{mlflow_db}") mlflow.set_experiment(cfg.MLFLOW_EXP_NAME) with mlflow.start_run(run_name=f"predict_{base}_{datetime.now().strftime('%H%M%S')}"): mlflow.log_metrics({ 'prob_long': unc['prob_long'], 'prob_short': unc['prob_short'], 'ev_long': ev_long, 'ev_short': ev_short, 'prob_diff': diff, 'max_z': max_z, 'position_fraction': position_result['final_fraction'], }) mlflow.log_params({ 'used_rr': rr_ratio, 'trained_rr': round(trained_rr_avg, 1), 'recommended_dir': direction, 'risk_level': position_result['recommended_risk'], }) # Console output _print_prediction(base, direction, unc, diff, higher_prob, higher_conf, higher_ev, lower_prob, ev_long, ev_short, trained_rr_avg, pos=position_result, max_z=max_z) # Persist to CSV _append_monitor(instrument, timeframe, direction, unc, higher_prob, higher_conf, higher_ev, position_result, rr_ratio) def _safe_temp(val: object, bounds: tuple, default: float = 1.0) -> float: """Clamp a checkpoint temperature value.""" if isinstance(val, (int, float)) and val > 0: return float(np.clip(val, *bounds)) return default def _print_prediction( base: str, direction: str, unc: dict, diff: float, higher_prob: float, higher_conf: str, higher_ev: float, lower_prob: float, ev_long: float, ev_short: float, trained_rr: float, pos: dict, max_z: float, ) -> None: """Pretty-print prediction result.""" sep = "=" * 80 dash = "-" * 80 print(f"\n{sep}") print(f"DUAL-ENSEMBLE FORECAST: {base}") print(sep) print(f"Requested RR: {3.0:.1f} | Trained RR: {trained_rr:.1f}") print(f"Weight by AUC: {'Yes' if cfg.ENSEMBLE_WEIGHT_BY_AUC else 'No'}") print(dash) print(f"LONG | TP prob: {unc['prob_long']:.2%} | Conf: {unc['conf_long']} | EV: {ev_long:+.2f}") print(f" | 90% CI: [{max(0, unc['prob_long']-unc['margin_long']):.2%} - {min(1, unc['prob_long']+unc['margin_long']):.2%}] (±{unc['margin_long']:.2%})") print(f"SHORT | TP prob: {unc['prob_short']:.2%} | Conf: {unc['conf_short']} | EV: {ev_short:+.2f}") print(dash) print(f"Prob diff: {diff:.1%} | Max Z: {max_z:.2f}σ | Drift norm: {'Yes' if max_z > cfg.DRIFT_THRESHOLD else 'No'}") print(f"RECOMMENDATION : {pos.get('recommended_risk', 'UNK')}") print(f"POSITION SIZE : {pos.get('final_fraction', 0):.2%} ({pos.get('recommended_risk', 'UNK')})") print(sep) def _append_monitor( instrument: str, timeframe: str, direction: str, unc: dict, higher_prob: float, higher_conf: str, higher_ev: float, pos: dict, rr_ratio: float, ) -> None: """Append prediction row to monitor CSV.""" new_row = { 'timestamp': datetime.now().isoformat(), 'instrument': instrument, 'timeframe': timeframe, 'direction': direction, 'predicted_prob': higher_prob, 'confidence': higher_conf, 'ev': higher_ev, 'rr_ratio': rr_ratio, 'trained_rr': round(pos.get('kelly_fraction', 0), 4), 'risk_level': pos.get('recommended_risk', 'UNKNOWN'), 'position_fraction': pos.get('final_fraction', 0.0), 'prob_long': unc['prob_long'], 'prob_short': unc['prob_short'], 'status': 'PENDING', 'actual_outcome': None, } if cfg.MONITOR_CSV.exists(): existing = pd.read_csv(cfg.MONITOR_CSV) updated = pd.concat([existing, pd.DataFrame([new_row])], ignore_index=True) else: updated = pd.DataFrame([new_row]) updated.to_csv(cfg.MONITOR_CSV, index=False)