# src/training/ensemble.py import logging from typing import List, Dict from pathlib import Path import torch import numpy as np import mlflow from sklearn.metrics import roc_curve, roc_auc_score from config import ( SEQ_LEN, BASE_HORIZON, MIN_HORIZON, MAX_HORIZON, ENSEMBLE_SEEDS, DEVICE, MLFLOW_EXP_NAME, MODELS_DIR, USE_CONV, USE_ATTENTION, USE_CONTEXT, HIDDEN_DIM, PATIENCE, CALIB_USE_TEMP, CALIB_USE_PLATT, CALIB_USE_ISOTONIC, ENSEMBLE_WEIGHT_BY_AUC, CONTEXT_CAPACITY, CONTEXT_HIDDEN_DIM, TRAIN_LR, TRAIN_BATCH_SIZE, DEFAULT_DROPOUT, DEFAULT_NUM_LAYERS, ) from src.data.loader import fetch_ohlcv from src.data.features import engineer_features from src.models.lstm import DualHeadLSTMModel, FocalLoss from src.models.context import ContextEnhancedLSTMModel from src.models.calibration import MultiCalibrator from src.training.walk_forward import walk_forward_loop from src.reporting.generate_html_report import generate_html_report logger = logging.getLogger(__name__) def train_single_model(seed: int, df, feat_cols: list, save_path_base: str, rr_ratio: float): """Обучение одной модели ансамбля.""" logger.info(f"🔧 === Training seed {seed} ===") model_class = ContextEnhancedLSTMModel if USE_CONTEXT else DualHeadLSTMModel best_state, best_scaler, ( calib_logits_long, calib_labels_long, calib_logits_short, calib_labels_short ), fold_losses = walk_forward_loop( df, feat_cols, model_class, FocalLoss, lr=TRAIN_LR, batch_size=TRAIN_BATCH_SIZE, seed=seed, rr_ratio=rr_ratio, use_conv=USE_CONV, use_attention=USE_ATTENTION, use_context=USE_CONTEXT, hidden_dim=HIDDEN_DIM, context_capacity=CONTEXT_CAPACITY, context_hidden_dim=CONTEXT_HIDDEN_DIM, ) calib_long = MultiCalibrator(use_temp=CALIB_USE_TEMP, use_platt=CALIB_USE_PLATT, use_isotonic=CALIB_USE_ISOTONIC) calib_short = MultiCalibrator(use_temp=CALIB_USE_TEMP, use_platt=CALIB_USE_PLATT, use_isotonic=CALIB_USE_ISOTONIC) if len(calib_logits_long) >= 30: calib_long.fit(np.array(calib_logits_long), np.array(calib_labels_long)) if len(calib_logits_short) >= 30: calib_short.fit(np.array(calib_logits_short), np.array(calib_labels_short)) save_dir = Path(f"{MODELS_DIR}/{save_path_base}/seed_{seed}") save_dir.mkdir(parents=True, exist_ok=True) if best_state: temp_l = calib_long.temp.temperature if calib_long.temp and calib_long.temp.fitted else 1.0 temp_s = calib_short.temp.temperature if calib_short.temp and calib_short.temp.fitted else 1.0 checkpoint = { 'model': best_state, 'scaler': best_scaler, 'feat_cols': feat_cols, 'val_auc': best_state.get('val_auc', 0.0), 'temperature_long': temp_l, 'temperature_short': temp_s, 'rr_ratio': rr_ratio, 'use_conv': USE_CONV, 'use_attention': USE_ATTENTION, 'use_context': USE_CONTEXT, 'hidden_dim': HIDDEN_DIM, 'context_capacity': CONTEXT_CAPACITY, 'context_hidden_dim': CONTEXT_HIDDEN_DIM, 'input_dim': len(feat_cols) } torch.save(checkpoint, save_dir / 'best.pt') logger.info(f"✅ Seed {seed}: AUC={best_state.get('val_auc', 0):.3f}, " f"Temp L={temp_l:.2f}, S={temp_s:.2f}") # Return a structure suitable for reporting: (best_state, best_scaler, calib data, fold_losses) return best_state, best_scaler, ( calib_logits_long, calib_labels_long, calib_logits_short, calib_labels_short ), fold_losses def train_ensemble(df, args) -> None: """Обучение ансамбля.""" feat_cols = [c for c in df.columns if c not in ['timestamp', 'Label_Long', 'Label_Short']] base = f"{args.instrument.upper()}_{args.timeframe.upper()}" mlflow.set_tracking_uri(f"sqlite:///{args.mlflow_db}") mlflow.set_experiment(MLFLOW_EXP_NAME) with mlflow.start_run(run_name=f"train_{base}"): mlflow.log_params({ 'instrument': args.instrument, 'timeframe': args.timeframe, 'rr_ratio': args.rr, 'seq_len': SEQ_LEN, 'base_horizon': BASE_HORIZON, 'min_horizon': MIN_HORIZON, 'max_horizon': MAX_HORIZON, 'use_conv': USE_CONV, 'use_attention': USE_ATTENTION, 'use_context': USE_CONTEXT, 'hidden_dim': HIDDEN_DIM, 'context_capacity': CONTEXT_CAPACITY, 'context_hidden_dim': CONTEXT_HIDDEN_DIM, 'calib_temp': CALIB_USE_TEMP, 'calib_platt': CALIB_USE_PLATT, 'calib_isotonic': CALIB_USE_ISOTONIC, 'ensemble_weight_auc': ENSEMBLE_WEIGHT_BY_AUC }) logger.info(f"🚀 Ансамбль: {len(ENSEMBLE_SEEDS)} моделей (Dual-Head) | RR: {args.rr}") aucs = [] metrics: List[Dict] = [] calib_results = [] # per-seed calib data for ROC plotting all_loss_history = [] # per-seed loss curves for seed in ENSEMBLE_SEEDS: logger.info(f"\n📍 Обработка seed {seed}") result = train_single_model(seed, df, feat_cols, base, args.rr) if isinstance(result, tuple) and len(result) == 4: best_state, best_scaler, calib_tuple, fold_losses = result auc = best_state.get('val_auc', float('nan')) if best_state else float('nan') if not np.isnan(auc): aucs.append(auc) # Извлекаем последнюю train/val loss из фолдов last_train = None last_val = None if fold_losses: last_train_losses = [l for fold in fold_losses for l in fold[0]] last_val_losses = [l for fold in fold_losses for l in fold[1]] if last_train_losses: last_train = last_train_losses[-1] if last_val_losses: last_val = last_val_losses[-1] metrics.append({ 'model_id': seed, 'train_loss': last_train, 'val_loss': last_val, 'auc': auc, 'patience_left': PATIENCE # если early stopping сработал — будет другое }) calib_results.append(calib_tuple) all_loss_history.append(fold_losses) else: logger.warning(f"⚠ Seed {seed}: unexpected result format, trying fallback") try: if not np.isnan(result): aucs.append(result) metrics.append({ 'model_id': seed, 'train_loss': None, 'val_loss': None, 'auc': result, 'patience_left': None }) except Exception: continue if aucs: mean_auc = np.mean(aucs) mlflow.log_metric('val_auc_mean', mean_auc) logger.info(f"✅ Ансамбль обучен. Средний Val AUC: {mean_auc:.3f}") else: logger.error("❌ Обучение не удалось: нет валидных фолдов.") # Build ROC data from calib data for plots (LONG + SHORT combined) roc_data = [] for calib in calib_results: calib_logits_long, calib_labels_long, calib_logits_short, calib_labels_short = calib try: fpr_long, tpr_long, _ = roc_curve(calib_labels_long, calib_logits_long)[:3] auc_long = roc_auc_score(calib_labels_long, calib_logits_long) except Exception: fpr_long, tpr_long, auc_long = [], [], float('nan') try: fpr_short, tpr_short, _ = roc_curve(calib_labels_short, calib_logits_short)[:3] auc_short = roc_auc_score(calib_labels_short, calib_logits_short) except Exception: fpr_short, tpr_short, auc_short = [], [], float('nan') roc_data.append((fpr_long, tpr_long, [], [auc_long, auc_short])) # Flatten loss history: list of (train, val) per model flat_loss_history = [] for losses in all_loss_history: if losses: train_all = [l for fold in losses for l in fold[0]] val_all = [l for fold in losses for l in fold[1]] flat_loss_history.append((train_all, val_all)) # Generate HTML report try: generate_html_report( metrics=metrics, loss_history=flat_loss_history, roc_curves=roc_data, out_path="reports/training_report.html" ) logger.info("HTML report written to reports/training_report.html") except Exception as e: logger.exception("Failed to write HTML report: %s", e)