#!/usr/bin/env python3 """ Extract Best Hyperparameters from Tuning Results Analyzes JSON result files and extracts optimal hyperparameters for each model """ import json import sys from pathlib import Path from typing import Dict, Any, List, Optional from datetime import datetime class HyperparameterExtractor: """Extract and analyze best hyperparameters from tuning results""" def __init__(self, results_dir: str = "results"): self.results_dir = Path(results_dir) self.models = ["DQN", "PPO", "TFT", "MAMBA2", "Liquid"] def extract_best_params(self, model: str) -> Optional[Dict[str, Any]]: """Extract best hyperparameters for a given model""" result_file = self.results_dir / f"{model.lower()}_tuning_50trials.json" if not result_file.exists(): print(f"⚠️ Result file not found: {result_file}") return None try: with open(result_file, 'r') as f: data = json.load(f) # Find best trial by Sharpe ratio best_trial = max(data.get('trials', []), key=lambda x: x.get('sharpe_ratio', -999)) return { 'model': model, 'best_trial_id': best_trial.get('trial_id'), 'sharpe_ratio': best_trial.get('sharpe_ratio'), 'loss': best_trial.get('loss'), 'training_time': best_trial.get('training_time'), 'hyperparameters': best_trial.get('hyperparameters', {}), 'total_trials': len(data.get('trials', [])), 'completed_trials': sum(1 for t in data.get('trials', []) if t.get('status') == 'completed') } except Exception as e: print(f"❌ Error extracting {model} parameters: {e}") return None def format_hyperparameters(self, params: Dict[str, Any]) -> str: """Format hyperparameters for display""" if not params: return "No hyperparameters available" lines = [] for key, value in params.items(): if isinstance(value, float): lines.append(f" {key}: {value:.6f}") else: lines.append(f" {key}: {value}") return "\n".join(lines) def generate_report(self, output_file: str = "HYPERPARAMETER_TUNING_EXECUTION_REPORT.md"): """Generate comprehensive report of all tuning results""" print("=" * 70) print("HYPERPARAMETER TUNING RESULTS EXTRACTION") print("=" * 70) print(f"Timestamp: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"Results directory: {self.results_dir}") print() all_results = {} for model in self.models: print(f"Processing {model}...") result = self.extract_best_params(model) if result: all_results[model] = result print(f" ✓ Found {result['completed_trials']}/{result['total_trials']} trials") print(f" ✓ Best Sharpe: {result['sharpe_ratio']:.4f}") else: print(f" ⚠️ No results available") print() # Generate markdown report report = self._generate_markdown_report(all_results) output_path = Path(output_file) with open(output_path, 'w') as f: f.write(report) print("=" * 70) print(f"Report generated: {output_path}") print("=" * 70) return all_results def _generate_markdown_report(self, results: Dict[str, Dict[str, Any]]) -> str: """Generate markdown report from results""" report = [] report.append("# Hyperparameter Tuning Execution Report") report.append("") report.append(f"**Generated**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") report.append(f"**Pipeline Status**: {'Complete' if len(results) == 5 else 'In Progress'}") report.append(f"**Models Completed**: {len(results)}/5") report.append("") report.append("---") report.append("") # Executive Summary report.append("## Executive Summary") report.append("") if results: total_trials = sum(r['total_trials'] for r in results.values()) completed_trials = sum(r['completed_trials'] for r in results.values()) avg_sharpe = sum(r['sharpe_ratio'] for r in results.values()) / len(results) report.append(f"- **Total Trials**: {completed_trials}/{total_trials}") report.append(f"- **Average Sharpe Ratio**: {avg_sharpe:.4f}") report.append(f"- **Models Optimized**: {', '.join(results.keys())}") else: report.append("*No results available yet*") report.append("") report.append("---") report.append("") # Individual Model Results for model, result in results.items(): report.append(f"## {model} Hyperparameter Tuning") report.append("") report.append(f"**Status**: ✅ Complete") report.append(f"**Trials**: {result['completed_trials']}/{result['total_trials']}") report.append(f"**Best Trial**: #{result['best_trial_id']}") report.append("") report.append("### Performance Metrics") report.append("") report.append(f"- **Sharpe Ratio**: {result['sharpe_ratio']:.4f}") report.append(f"- **Final Loss**: {result['loss']:.6f}") report.append(f"- **Training Time**: {result['training_time']:.1f}s") report.append("") report.append("### Best Hyperparameters") report.append("") report.append("```yaml") for key, value in result['hyperparameters'].items(): if isinstance(value, float): report.append(f"{key}: {value:.6f}") else: report.append(f"{key}: {value}") report.append("```") report.append("") report.append("---") report.append("") # Pending Models pending_models = [m for m in self.models if m not in results] if pending_models: report.append("## Pending Models") report.append("") for model in pending_models: report.append(f"- **{model}**: ⏳ In Progress or Not Started") report.append("") report.append("---") report.append("") # Next Steps report.append("## Next Steps") report.append("") if len(results) == 5: report.append("1. ✅ All models tuned successfully") report.append("2. 📝 Review hyperparameters for each model") report.append("3. 🔧 Update model configuration files") report.append("4. 🚀 Run production training with optimized hyperparameters") report.append("5. 📊 Validate models with backtesting") else: report.append(f"1. ⏳ Wait for remaining {len(pending_models)} models to complete") report.append("2. 📊 Monitor tuning progress with dashboard") report.append("3. 🔍 Check for CUDA OOM errors in logs") report.append("4. 🔄 Regenerate report when all models complete") report.append("") return "\n".join(report) def print_summary(self): """Print a quick summary of available results""" print("=" * 70) print("AVAILABLE TUNING RESULTS") print("=" * 70) for model in self.models: result_file = self.results_dir / f"{model.lower()}_tuning_50trials.json" if result_file.exists(): try: with open(result_file, 'r') as f: data = json.load(f) trials = len(data.get('trials', [])) completed = sum(1 for t in data.get('trials', []) if t.get('status') == 'completed') print(f"✅ {model:10} | {completed:2}/{trials:2} trials | {result_file}") except: print(f"❌ {model:10} | ERROR reading file | {result_file}") else: print(f"⏳ {model:10} | Not started | {result_file}") print("=" * 70) if __name__ == "__main__": extractor = HyperparameterExtractor() if len(sys.argv) > 1 and sys.argv[1] == "--summary": extractor.print_summary() else: extractor.generate_report()