MCP Hybrid Forecasting

if __name__ == "__main__": import argparse parser = argparse.ArgumentParser( description="ARIMA Optimization Workflow Example", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: python example_arima_workflow.py # Show available portfolios python example_arima_workflow.py --quick # Quick demo with AAPL python example_arima_workflow.py --portfolio tech_giants # Demo with specific portfolio python example_arima_workflow.py --full # Full workflow with first portfolio """ ) parser.add_argument('--quick', action='store_true', help='Run quick demo with one stock (AAPL)') parser.add_argument('--portfolio', type=str, help='Run demo with specific portfolio from config') parser.add_argument('--full', action='store_true', help='Run full workflow with first available portfolio') parser.add_argument('--list-portfolios', action='store_true', help='List available portfolios and exit') args = parser.parse_args() if args.list_portfolios: # Show available portfolios config = load_portfolio_config() print("📊 AVAILABLE PORTFOLIOS FOR DEMO") print("=" * 50) for i, (portfolio_name, tickers) in enumerate(config.get("tickers", {}).items(), 1): if isinstance(tickers, list): print(f" {i:2d}. {portfolio_name:<20} ({len(tickers):2d} stocks)") print(f"\n💡 Usage:") print(f" python example_arima_workflow.py --portfolio PORTFOLIO_NAME") print(f" python example_arima_workflow.py --quick") print(f" python example_arima_workflow.py --full") elif args.quick: quick_demo() elif args.portfolio: run_portfolio_optimization_demo(args.portfolio) elif args.full:# example_arima_workflow.py - Complete workflow example """ Example workflow showing how to: 1. Optimize ARIMA parameters for your portfolio 2. Use the optimized parameters in your trading system 3. Compare performance with and without optimization """ import pandas as pd import numpy as np from datetime import datetime import time # Import the optimization system from preprocess_arima_params import ARIMAParameterOptimizer, load_portfolio_config, get_all_unique_tickers # Import enhanced ARIMA model from models.arima_model import get_arima_forecast, get_enhanced_arima_forecast, get_optimization_summary def run_portfolio_optimization_demo(portfolio_name: str = None): """Run optimization demo for a specific portfolio.""" print("🚀 PORTFOLIO-SPECIFIC ARIMA OPTIMIZATION") print("=" * 60) # Step 1: Load portfolio configuration print("\n📋 Step 1: Loading Portfolio Configuration") config = load_portfolio_config() # Show available portfolios available_portfolios = list(config.get("tickers", {}).keys()) print(f" 📊 Available portfolios: {', '.join(available_portfolios)}") # Select portfolio if portfolio_name: if portfolio_name not in available_portfolios: print(f"❌ Portfolio '{portfolio_name}' not found") print(f" Available: {available_portfolios}") return None selected_portfolio = portfolio_name tickers = config["tickers"][portfolio_name] else: # Use first portfolio as default selected_portfolio = available_portfolios[0] tickers = config["tickers"][selected_portfolio] print(f" 🎯 Using default portfolio: {selected_portfolio}") print(f" 📊 Selected portfolio: {selected_portfolio}") print(f" 📈 Stocks in portfolio: {len(tickers)}") # Limit to first 5 for demo if len(tickers) > 5: demo_tickers = tickers[:5] print(f" 🎯 Demo using first 5 stocks: {demo_tickers}") else: demo_tickers = tickers print(f" 🎯 Demo using all stocks: {demo_tickers}") # Step 2: Run parameter optimization print(f"\n🔧 Step 2: Optimizing {selected_portfolio} Portfolio") print(" This will take a few minutes...") start_time = time.time() # Initialize optimizer with demo settings optimizer = ARIMAParameterOptimizer( rolling_window=500, # Smaller window for faster demo forecast_horizon=5, n_iterations=15, # Fewer iterations for speed data_period="2y" # Less data for faster download ) # Optimize parameters optimization_results = optimizer.optimize_portfolio(demo_tickers, max_workers=2) optimization_time = time.time() - start_time print(f"\n ⏱️ Portfolio optimization completed in {optimization_time:.1f} seconds") # Save results with portfolio name timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") results_filename = f"arima_optimization_{selected_portfolio}_{timestamp}.json" results_file = optimizer.save_optimization_results(optimization_results, results_filename) optimizer.print_optimization_summary(optimization_results) # Step 3: Test the optimized parameters print(f"\n🔬 Step 3: Testing {selected_portfolio} Portfolio Performance") portfolio_comparison = [] for ticker in demo_tickers: if ticker in optimization_results: print(f"\n Testing {ticker}...") try: # Test with optimized parameters forecast_opt, residuals_opt, df_opt = get_arima_forecast(ticker, use_optimized_params=True) # Test with default parameters forecast_def, residuals_def, df_def = get_arima_forecast(ticker, use_optimized_params=False) # Calculate improvement metrics opt_rmse = np.sqrt(np.mean(residuals_opt**2)) def_rmse = np.sqrt(np.mean(residuals_def**2)) rmse_improvement = (def_rmse - opt_rmse) / def_rmse * 100 opt_order = optimization_results[ticker]['best_order'] portfolio_comparison.append({ 'ticker': ticker, 'portfolio': selected_portfolio, 'optimized_order': opt_order, 'optimized_forecast': forecast_opt, 'default_forecast': forecast_def, 'rmse_improvement': rmse_improvement }) print(f" Optimized ARIMA{opt_order}: ${forecast_opt:.2f} (RMSE: {opt_rmse:.4f})") print(f" Default ARIMA(1,1,1): ${forecast_def:.2f} (RMSE: {def_rmse:.4f})") if rmse_improvement > 0: print(f" ✅ Improved by {rmse_improvement:.1f}%") else: print(f" ⚠️ Worse by {abs(rmse_improvement):.1f}%") except Exception as e: print(f" ❌ Error testing {ticker}: {e}") # Portfolio summary if portfolio_comparison: print(f"\n📊 {selected_portfolio.upper()} PORTFOLIO SUMMARY") print("=" * 60) avg_improvement = np.mean([x['rmse_improvement'] for x in portfolio_comparison]) improved_count = len([x for x in portfolio_comparison if x['rmse_improvement'] > 0]) print(f"📈 Average RMSE Improvement: {avg_improvement:.1f}%") print(f"📊 Success Rate: {improved_count}/{len(portfolio_comparison)} stocks improved") # Show best and worst best = max(portfolio_comparison, key=lambda x: x['rmse_improvement']) worst = min(portfolio_comparison, key=lambda x: x['rmse_improvement']) print(f"🏆 Best: {best['ticker']} (+{best['rmse_improvement']:.1f}%)") print(f"⚠️ Worst: {worst['ticker']} ({worst['rmse_improvement']:+.1f}%)") print(f"\n✅ {selected_portfolio} Portfolio Optimization Complete!") print(f"📁 Results saved to: {results_file}") return { 'portfolio_name': selected_portfolio, 'optimization_results': optimization_results, 'comparison_results': portfolio_comparison, 'results_file': results_file } def run_complete_workflow(): """Run the complete ARIMA optimization workflow.""" print("🚀 COMPLETE ARIMA OPTIMIZATION WORKFLOW") print("=" * 60) # Step 1: Load portfolio configuration and show options print("\n📋 Step 1: Portfolio Selection") config = load_portfolio_config() available_portfolios = list(config.get("tickers", {}).keys()) print(f" 📊 Available portfolios: {len(available_portfolios)}") for i, portfolio in enumerate(available_portfolios, 1): ticker_count = len(config["tickers"][portfolio]) print(f" {i:2d}. {portfolio:<20} ({ticker_count:2d} stocks)") # For demo, use first portfolio demo_portfolio = available_portfolios[0] print(f"\n 🎯 Demo using: {demo_portfolio}") return run_portfolio_optimization_demo(demo_portfolio) def quick_demo(): """Quick demonstration with just one stock.""" print("🚀 QUICK ARIMA OPTIMIZATION DEMO") print("=" * 40) # Demo with AAPL ticker = "AAPL" print(f"📊 Optimizing parameters for {ticker}...") # Quick optimization optimizer = ARIMAParameterOptimizer( rolling_window=300, forecast_horizon=5, n_iterations=10, data_period="1y" ) result = optimizer.optimize_single_stock(ticker) if result: print(f"✅ Best parameters: ARIMA{result['best_order']}") print(f"📊 Average RMSE: {result['best_avg_rmse']:.4f}") # Save and test optimizer.save_optimization_results({ticker: result}) # Test the forecast forecast, residuals, df = get_arima_forecast(ticker) print(f"🎯 Forecast for {ticker}: ${forecast:.2f}") return result else: print(f"❌ Optimization failed for {ticker}") return None if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="ARIMA Optimization Workflow Example") parser.add_argument('--quick', action='store_true', help='Run quick demo with one stock') parser.add_argument('--full', action='store_true', help='Run full workflow with sample stocks') args = parser.parse_args() if args.quick: quick_demo() elif args.full: run_complete_workflow() else: # No arguments provided - show help and available portfolios print("🚀 ARIMA OPTIMIZATION WORKFLOW EXAMPLES") print("=" * 50) # Load and show available portfolios config = load_portfolio_config() portfolios = list(config.get("tickers", {}).keys()) print(f"📊 Available portfolios in your config:") for i, portfolio_name in enumerate(portfolios, 1): ticker_count = len(config["tickers"][portfolio_name]) print(f" {i:2d}. {portfolio_name:<20} ({ticker_count:2d} stocks)") print(f"\n🎯 Choose a demo option:") print(f" --quick : Quick demo with AAPL (2-3 minutes)") print(f" --portfolio PORTFOLIO_NAME : Demo with specific portfolio (5-10 minutes)") print(f" --full : Full workflow with first portfolio (5-10 minutes)") print(f" --list-portfolios : Show available portfolios") print(f"\n📝 Examples:") print(f" python example_arima_workflow.py --quick") if portfolios: print(f" python example_arima_workflow.py --portfolio {portfolios[0]}") print(f" python example_arima_workflow.py --full")