"""Training orchestration and utilities.""" import os from pathlib import Path from typing import Optional, Dict import torch import torch.nn as nn from torch.utils.data import DataLoader from tqdm import tqdm import json from .config import TrainingConfig from .evaluator import evaluate_model class EarlyStopping: """Early stopping to stop training when validation loss doesn't improve.""" def __init__(self, patience: int = 10, min_delta: float = 1e-4): """Initialize early stopping. Args: patience: Number of epochs to wait before stopping min_delta: Minimum change to qualify as improvement """ self.patience = patience self.min_delta = min_delta self.counter = 0 self.best_loss = None self.should_stop = False def __call__(self, val_loss: float) -> bool: """Check if training should stop. Args: val_loss: Current validation loss Returns: True if training should stop """ if self.best_loss is None: self.best_loss = val_loss elif val_loss > self.best_loss - self.min_delta: self.counter += 1 if self.counter >= self.patience: self.should_stop = True else: self.best_loss = val_loss self.counter = 0 return self.should_stop class Trainer: """Model trainer with support for MPS/CUDA/CPU.""" def __init__( self, model: nn.Module, config: TrainingConfig, task: str = 'classification', pos_weight: float = None # 添加类别权重参数 ): """Initialize trainer. Args: model: PyTorch model to train config: Training configuration task: Task type ('classification' or 'regression') pos_weight: Positive class weight for imbalanced data """ self.model = model self.config = config self.task = task self.pos_weight = pos_weight # Set device self.device = config.get_device() self.model = self.model.to(self.device) print(f"Training on device: {self.device}") # Loss function (with class weight if provided) if hasattr(model, 'get_loss_function'): self.criterion = model.get_loss_function(pos_weight=pos_weight, device=self.device) else: if task == 'classification': if pos_weight is not None: # 使用float32以兼容MPS pos_weight_tensor = torch.tensor([pos_weight], dtype=torch.float32).to(self.device) self.criterion = nn.BCEWithLogitsLoss(pos_weight=pos_weight_tensor) else: self.criterion = nn.BCELoss() else: self.criterion = nn.MSELoss() if pos_weight is not None: print(f"Using weighted loss - positive class weight: {pos_weight:.2f}") # Optimizer self.optimizer = torch.optim.AdamW( model.parameters(), lr=config.learning_rate, weight_decay=config.weight_decay ) # Learning rate scheduler if config.lr_scheduler == 'plateau': self.scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau( self.optimizer, mode='min', factor=config.lr_factor, patience=config.lr_patience, verbose=True ) elif config.lr_scheduler == 'cosine': self.scheduler = torch.optim.lr_scheduler.CosineAnnealingLR( self.optimizer, T_max=config.max_epochs ) else: self.scheduler = None # Early stopping self.early_stopping = EarlyStopping( patience=config.early_stopping_patience, min_delta=config.early_stopping_min_delta ) # Training history self.history = { 'train_loss': [], 'val_loss': [], 'learning_rates': [] } # Best model tracking self.best_val_loss = float('inf') self.best_model_path = None def train_epoch(self, train_loader: DataLoader) -> float: """Train for one epoch. Args: train_loader: Training data loader Returns: Average training loss """ self.model.train() total_loss = 0.0 num_batches = 0 pbar = tqdm(train_loader, desc='Training') for batch_X, batch_y in pbar: batch_X = batch_X.to(self.device) batch_y = batch_y.to(self.device) # Forward pass self.optimizer.zero_grad() outputs = self.model(batch_X).squeeze() # Handle different output shapes if self.task == 'classification': batch_y = batch_y.float() loss = self.criterion(outputs, batch_y) # Backward pass loss.backward() self.optimizer.step() # Track loss total_loss += loss.item() num_batches += 1 # Update progress bar pbar.set_postfix({'loss': f'{loss.item():.4f}'}) return total_loss / num_batches def validate(self, val_loader: DataLoader) -> float: """Validate the model. Args: val_loader: Validation data loader Returns: Average validation loss """ self.model.eval() total_loss = 0.0 num_batches = 0 with torch.no_grad(): for batch_X, batch_y in val_loader: batch_X = batch_X.to(self.device) batch_y = batch_y.to(self.device) outputs = self.model(batch_X).squeeze() if self.task == 'classification': batch_y = batch_y.float() loss = self.criterion(outputs, batch_y) total_loss += loss.item() num_batches += 1 return total_loss / num_batches def fit( self, train_loader: DataLoader, val_loader: DataLoader, save_dir: Optional[str] = None ) -> Dict: """Train the model. Args: train_loader: Training data loader val_loader: Validation data loader save_dir: Directory to save checkpoints Returns: Training history """ if save_dir is None: save_dir = self.config.checkpoint_dir save_dir = Path(save_dir) save_dir.mkdir(parents=True, exist_ok=True) print(f"Starting training for {self.config.max_epochs} epochs...") print(f"Model parameters: {self.model.count_parameters()}") for epoch in range(self.config.max_epochs): # Train train_loss = self.train_epoch(train_loader) # Validate val_loss = self.validate(val_loader) # Update history self.history['train_loss'].append(train_loss) self.history['val_loss'].append(val_loss) current_lr = self.optimizer.param_groups[0]['lr'] self.history['learning_rates'].append(current_lr) # Print progress print(f"Epoch {epoch + 1}/{self.config.max_epochs}") print(f" Train loss: {train_loss:.4f}") print(f" Val loss: {val_loss:.4f}") print(f" LR: {current_lr:.6f}") # Save best model if val_loss < self.best_val_loss: self.best_val_loss = val_loss if self.config.save_best: self.best_model_path = save_dir / 'best_model.pt' self.model.save(str(self.best_model_path)) print(f" Saved best model (val_loss: {val_loss:.4f})") # Learning rate scheduling if self.scheduler is not None: if self.config.lr_scheduler == 'plateau': self.scheduler.step(val_loss) else: self.scheduler.step() # Early stopping if self.early_stopping(val_loss): print(f"Early stopping at epoch {epoch + 1}") break # Save final model final_model_path = save_dir / 'final_model.pt' self.model.save(str(final_model_path)) # Save training history history_path = save_dir / 'training_history.json' with open(history_path, 'w') as f: json.dump(self.history, f, indent=2) print(f"\nTraining completed!") print(f"Best validation loss: {self.best_val_loss:.4f}") print(f"Best model saved to: {self.best_model_path}") return self.history def evaluate(self, test_loader: DataLoader) -> Dict[str, float]: """Evaluate the model on test data. Args: test_loader: Test data loader Returns: Dictionary of evaluation metrics """ # Load best model if available if self.best_model_path is not None and self.best_model_path.exists(): print(f"Loading best model from {self.best_model_path}") checkpoint = torch.load(self.best_model_path, map_location=self.device) self.model.load_state_dict(checkpoint['model_state_dict']) # Evaluate metrics = evaluate_model(self.model, test_loader, self.task, self.device) print("\nTest Metrics:") for key, value in metrics.items(): print(f" {key}: {value:.4f}") return metrics if __name__ == "__main__": # Test the trainer from ..models.dlinear.model import DLinear, DLinearConfig from ..data.fetcher import BinanceDataFetcher from ..data.preprocessor import CryptoPreprocessor, split_data from ..data.dataset import create_dataloaders # Fetch data print("Fetching data...") fetcher = BinanceDataFetcher() df = fetcher.fetch_ohlcv('BTC', '1h') # Preprocess print("\nPreprocessing...") preprocessor = CryptoPreprocessor(lookback=168) X, y, _ = preprocessor.preprocess_pipeline(df, task='classification') # Split train_data, val_data, test_data = split_data(X, y) # Create dataloaders train_loader, val_loader, test_loader = create_dataloaders( train_data, val_data, test_data, batch_size=32 ) # Create model print("\nCreating model...") config = DLinearConfig( input_dim=X.shape[-1], lookback=168, task='classification' ) model = DLinear(config) # Train train_config = TrainingConfig( batch_size=32, max_epochs=5, # Short test early_stopping_patience=3 ) trainer = Trainer(model, train_config, task='classification') history = trainer.fit(train_loader, val_loader, save_dir='/tmp/test_training') # Evaluate metrics = trainer.evaluate(test_loader)