diff --git a/predix.py b/predix.py index b0bb5e8a..468a00ae 100644 --- a/predix.py +++ b/predix.py @@ -140,7 +140,7 @@ def quant( # Setup both API keys for load balancing os.environ["OPENAI_API_BASE"] = "https://openrouter.ai/api/v1" - os.environ["CHAT_MODEL"] = os.getenv("OPENROUTER_MODEL", "openrouter/qwen/qwen3.6-plus:free") + os.environ["CHAT_MODEL"] = os.getenv("OPENROUTER_MODEL", "openrouter/google/gemma-4-26b-a4b-it:free") # If second key exists, configure LiteLLM for load balancing if api_key_2: @@ -990,7 +990,7 @@ def build_strategies_ai( else: os.environ["OPENAI_API_KEY"] = api_key os.environ["OPENAI_API_BASE"] = "https://openrouter.ai/api/v1" - os.environ["CHAT_MODEL"] = os.getenv("OPENROUTER_MODEL", "openrouter/qwen/qwen3.6-plus:free") + os.environ["CHAT_MODEL"] = os.getenv("OPENROUTER_MODEL", "openrouter/google/gemma-4-26b-a4b-it:free") console.print(f"\n[bold blue]🌐 Using OpenRouter: {os.environ['CHAT_MODEL']}[/bold blue]") else: console.print("[bold red]❌ No API key found. Set OPENROUTER_API_KEY in .env[/bold red]") diff --git a/predix_gen_strategies_real_bt.py b/predix_gen_strategies_real_bt.py index 2907e983..7b4b36f4 100644 --- a/predix_gen_strategies_real_bt.py +++ b/predix_gen_strategies_real_bt.py @@ -72,7 +72,7 @@ def setup_llm_env(): if router_key: os.environ['OPENAI_API_KEY'] = router_key os.environ['OPENAI_API_BASE'] = 'https://openrouter.ai/api/v1' - os.environ['CHAT_MODEL'] = os.getenv('OPENROUTER_MODEL', 'openrouter/qwen/qwen3.6-plus:free') + os.environ['CHAT_MODEL'] = os.getenv('OPENROUTER_MODEL', 'openrouter/google/gemma-4-26b-a4b-it:free') # ============================================================================ # Factor Loading (cached at module level for each process) diff --git a/predix_parallel.py b/predix_parallel.py index f0d0960d..70352353 100644 --- a/predix_parallel.py +++ b/predix_parallel.py @@ -178,7 +178,7 @@ class ParallelRunner: api_key = self.api_keys[run_state.api_key_idx] env["OPENAI_API_KEY"] = api_key env["OPENAI_API_BASE"] = "https://openrouter.ai/api/v1" - env["CHAT_MODEL"] = os.getenv("OPENROUTER_MODEL", "openrouter/qwen/qwen3.6-plus:free") + env["CHAT_MODEL"] = os.getenv("OPENROUTER_MODEL", "openrouter/google/gemma-4-26b-a4b-it:free") # If we configured multiple API keys AND have enough keys, use load balancing if self.num_api_keys >= 2 and len(self.api_keys) >= 2: diff --git a/prompts/standard_prompts.yaml b/prompts/standard_prompts.yaml index e0eccb28..93a4b8ae 100644 --- a/prompts/standard_prompts.yaml +++ b/prompts/standard_prompts.yaml @@ -1,160 +1,153 @@ -# Predix Prompts - Standard Version -# -# These are the default prompts for EUR/USD quantitative trading. -# Store your improved prompts in prompts/local/ (not committed to Git). -# -# Usage: -# from rdagent.components.loader import load_prompt -# prompt = load_prompt("factor_discovery") # Loads from prompts/local/ if exists, else prompts/ - -# ============================================================ -# Factor Discovery Prompts -# ============================================================ - factor_discovery: - system: |- - You are an expert quantitative researcher specialized in FX (foreign exchange) trading, - specifically EURUSD intraday strategies on 1-minute bars. - - EURUSD domain knowledge you must apply: - - London session (08:00-16:00 UTC): highest volume, trending behavior - - NY session (13:00-21:00 UTC): second volume peak - - Asian session (00:00-08:00 UTC): lower volume, mean-reverting - - London/NY overlap (13:00-16:00 UTC): strongest directional moves - - Spread cost: ~1.5 bps per trade — factors must overcome this - - EURUSD is mean-reverting on short windows (<1h), trending on longer (>4h) - - Your hypothesis must: - 1. Specify which session(s) the factor targets - 2. Include spread filter (expected return > 0.0003) - 3. Name the market regime (trending/mean-reverting) - 4. Be testable with available data (OHLCV, returns, technical indicators) - - Please ensure your response is in JSON format: - { - "hypothesis": "Clear factor hypothesis", - "reason": "Detailed explanation", - "target_session": "london/ny/asian/all", - "expected_arr_range": "e.g. 8-12%" - } + system: "You are an expert quantitative researcher specialized in FX (foreign exchange)\ + \ trading,\nspecifically EURUSD intraday strategies on 1-minute bars.\n\nEURUSD\ + \ domain knowledge you must apply:\n- London session (08:00-16:00 UTC): highest\ + \ volume, trending behavior\n- NY session (13:00-21:00 UTC): second volume peak\n\ + - Asian session (00:00-08:00 UTC): lower volume, mean-reverting\n- London/NY overlap\ + \ (13:00-16:00 UTC): strongest directional moves\n- Spread cost: ~1.5 bps per\ + \ trade — factors must overcome this\n- EURUSD is mean-reverting on short windows\ + \ (<1h), trending on longer (>4h)\n\nYour hypothesis must:\n1. Specify which session(s)\ + \ the factor targets\n2. Include spread filter (expected return > 0.0003)\n3.\ + \ Name the market regime (trending/mean-reverting)\n4. Be testable with available\ + \ data (OHLCV, returns, technical indicators)\n\nPlease ensure your response is\ + \ in JSON format:\n{\n \"hypothesis\": \"Clear factor hypothesis\",\n \"reason\"\ + : \"Detailed explanation\",\n \"target_session\": \"london/ny/asian/all\",\n\ + \ \"expected_arr_range\": \"e.g. 8-12%\"\n}" + user: 'Previously tried factors and their results: - user: |- - Previously tried factors and their results: {{ factor_descriptions }} - + + Additional context: + {{ report_content }} - - Generate a NEW factor hypothesis that is meaningfully different from what has been tried. - Target: beat current best ARR of 9.62%. -# ============================================================ -# Factor Evolution Prompts -# ============================================================ + Generate a NEW factor hypothesis that is meaningfully different from what has + been tried. + + Target: beat current best ARR of 9.62%.' factor_evolution: - system: |- - You are improving existing trading factors for EURUSD 1-minute data. - - Improvement strategies: - 1. Add session filters (is_london, is_ny) - 2. Add regime filters (ADX, volatility) - 3. Optimize lookback periods - 4. Combine with complementary factors - 5. Add risk management (stop-loss, take-profit) - - Your response must include: - - What to improve and why - - Expected performance gain - - Implementation approach - - JSON format: - { - "improvement": "Description of improvement", - "reason": "Why this will work better", - "expected_improvement": "e.g. +2% ARR, -5% drawdown" - } + system: "You are improving existing trading factors for EURUSD 1-minute data.\n\n\ + Improvement strategies:\n1. Add session filters (is_london, is_ny)\n2. Add regime\ + \ filters (ADX, volatility)\n3. Optimize lookback periods\n4. Combine with complementary\ + \ factors\n5. Add risk management (stop-loss, take-profit)\n\nYour response must\ + \ include:\n- What to improve and why\n- Expected performance gain\n- Implementation\ + \ approach\n\nJSON format:\n{\n \"improvement\": \"Description of improvement\"\ + ,\n \"reason\": \"Why this will work better\",\n \"expected_improvement\": \"\ + e.g. +2% ARR, -5% drawdown\"\n}" + user: 'Current factor: - user: |- - Current factor: {{ factor_code }} - + + Performance metrics: + {{ factor_metrics }} - - Suggest specific improvements to beat current performance. -# ============================================================ -# Model Coder Prompts -# ============================================================ + Suggest specific improvements to beat current performance.' model_coder: - system: |- - You are an expert ML engineer specialized in EURUSD trading models. - + system: 'You are an expert ML engineer specialized in EURUSD trading models. + + Supported model types: + - TimeSeries: LSTM, GRU, TCN, Transformer, PatchTST + - Tabular: XGBoost, LightGBM, RandomForest + - Hybrid: CNN+LSTM, XGBoost+LSTM ensemble - + + EURUSD-specific rules: + 1. Session filter: use is_london and is_ny columns + 2. Spread filter: only trade when abs(prediction) > 0.0003 + 3. ADX regime: if adx_proxy > 1.2 use trend model, else mean-reversion + 4. Weekend filter: close positions Friday 20:00 UTC + 5. Max frequency: target <15 trades per day - + + Your code must: + - Be production-ready (error handling, logging) + - Include session/regime filters + - Account for spread costs - - Support both classification and regression targets - user: |- - Factor descriptions: + - Support both classification and regression targets' + user: 'Factor descriptions: + {{ factor_descriptions }} - + + Available features: + {{ feature_list }} - + + Target: {{ target_variable }} - - Write complete, production-ready code for the model. -# ============================================================ -# Trading Strategy Prompts -# ============================================================ + Write complete, production-ready code for the model.' +strategy_generation: + system: "You are an expert quantitative trading researcher specialized in EUR/USD\ + \ intraday strategies.\n\nYour task is to generate a trading strategy by combining\ + \ the provided factors into a coherent signal.\n\nEUR/USD Domain Knowledge:\n\ + - London session (08:00-16:00 UTC): highest volume, trending behavior\n- NY session\ + \ (13:00-21:00 UTC): second volume peak, continuation\n- Asian session (00:00-08:00\ + \ UTC): lower volume, mean-reverting\n- London/NY overlap (13:00-16:00 UTC): strongest\ + \ directional moves\n- Spread cost: ~1.5 bps per trade — signals must overcome\ + \ this\n\nFactor Usage Rules:\n1. ONLY use the factors provided below — no others!\n\ + 2. The code MUST work with a DataFrame called 'factors' containing factor columns\n\ + 3. Create a pandas Series called 'signal' with values: 1 (long), -1 (short), 0\ + \ (neutral)\n4. signal.index MUST match factors.index exactly\n5. signal.name\ + \ must be 'signal'\n\nSignal Quality Requirements:\n- Generate meaningful signals\ + \ (avoid constant 0 or all 1s)\n- Use rolling z-scores for normalization: (x -\ + \ rolling.mean()) / rolling.std()\n- Apply thresholds (e.g., z > 0.3 for long,\ + \ z < -0.3 for short)\n- Combine factors with weights based on their IC values\n\ + - Consider regime filters (trend vs mean-reversion)\n\nOutput ONLY valid JSON\ + \ with these exact fields:\n{\n \"strategy_name\": \"short_descriptive_name\"\ + ,\n \"factors_used\": [\"factor1\", \"factor2\", \"factor3\"],\n \"description\"\ + : \"one sentence explaining the strategy logic\",\n \"code\": \"complete Python\ + \ code that creates signal Series\"\n}" + user: "Generate a EUR/USD trading strategy using these factors:\n\n{{ factors }}\n\ + \n{{ additional_context }}\n\nCRITICAL RULES:\n1. DO NOT define functions - write\ + \ direct executable code\n2. DO NOT use def - just write the code that creates\ + \ 'signal'\n3. The code will be executed with 'factors' DataFrame already in scope\n\ + 4. You MUST create a variable called 'signal' as a pandas Series\n5. signal must\ + \ have values 1 (LONG), -1 (SHORT), or 0 (NEUTRAL)\n6. signal.index must equal\ + \ factors.index\n\nEXAMPLE OF CORRECT FORMAT:\n```\nimport pandas as pd\nimport\ + \ numpy as np\n\nz = (factors['factor1'] - factors['factor1'].rolling(20).mean())\ + \ / factors['factor1'].rolling(20).std()\nsignal = pd.Series(0, index=factors.index)\n\ + signal[z > 0.3] = 1\nsignal[z < -0.3] = -1\nsignal.name = 'signal'\n```\n\nWRONG\ + \ FORMAT (DO NOT DO THIS):\n```\ndef generate_signal(factors):\n ...\n return\ + \ signal\n```\n\nOutput ONLY the JSON object, no additional text." trading_strategy: - system: |- - You are a portfolio manager designing trading strategies for EURUSD. - - Strategy components: - 1. Entry signals (from factors/models) - 2. Position sizing (volatility-adjusted) - 3. Risk management (stop-loss, take-profit, max drawdown) - 4. Session awareness (London/NY/Asian) - 5. Correlation management (if multiple factors) - - Your strategy must specify: - - Entry conditions (which signals, what thresholds) - - Exit conditions (time-based, signal-based, stop-loss) - - Position sizing (fixed, volatility-adjusted, Kelly) - - Risk limits (max position, max leverage, max drawdown) - - JSON format: - { - "entry_conditions": [...], - "exit_conditions": [...], - "position_sizing": "...", - "risk_limits": {...} - } + system: "You are a portfolio manager designing trading strategies for EURUSD.\n\n\ + Strategy components:\n1. Entry signals (from factors/models)\n2. Position sizing\ + \ (volatility-adjusted)\n3. Risk management (stop-loss, take-profit, max drawdown)\n\ + 4. Session awareness (London/NY/Asian)\n5. Correlation management (if multiple\ + \ factors)\n\nYour strategy must specify:\n- Entry conditions (which signals,\ + \ what thresholds)\n- Exit conditions (time-based, signal-based, stop-loss)\n\ + - Position sizing (fixed, volatility-adjusted, Kelly)\n- Risk limits (max position,\ + \ max leverage, max drawdown)\n\nJSON format:\n{\n \"entry_conditions\": [...],\n\ + \ \"exit_conditions\": [...],\n \"position_sizing\": \"...\",\n \"risk_limits\"\ + : {...}\n}" + user: 'Available factors: - user: |- - Available factors: {{ factors }} - + + Historical performance: + {{ historical_metrics }} - - Design a complete trading strategy that combines these factors optimally. + + + Design a complete trading strategy that combines these factors optimally.' diff --git a/rdagent/app/cli.py b/rdagent/app/cli.py index b99d63bc..8a971593 100644 --- a/rdagent/app/cli.py +++ b/rdagent/app/cli.py @@ -8,8 +8,11 @@ This will import os import sys +from datetime import datetime from pathlib import Path +import numpy as np +import pandas as pd from dotenv import load_dotenv load_dotenv(".env") @@ -18,7 +21,7 @@ load_dotenv(".env") import subprocess from importlib.resources import path as rpath -from typing import Optional +from typing import Dict, Optional import typer from rich.console import Console @@ -120,6 +123,16 @@ def fin_quant_cli( "-m", help="LLM backend to use: 'local' (llama.cpp), 'openrouter' (cloud models), or custom env var prefix", ), + auto_strategies: bool = typer.Option( + False, + "--auto-strategies", + help="Automatically generate strategies after factor threshold", + ), + auto_strategies_threshold: int = typer.Option( + 500, + "--auto-strategies-threshold", + help="Number of factors before triggering strategy generation", + ), ): """ Start EURUSD quantitative trading loop. @@ -128,6 +141,8 @@ def fin_quant_cli( --with-dashboard/-d: Start web dashboard at http://localhost:5000 --cli-dashboard/-c: Show beautiful terminal UI with live stats --model/-m: LLM backend ('local' | 'openrouter') + --auto-strategies: Auto-generate strategies after threshold + --auto-strategies-threshold: Factor count trigger for auto strategies Examples: rdagent fin_quant # Local llama.cpp (default) @@ -135,6 +150,8 @@ def fin_quant_cli( rdagent fin_quant -m openrouter # Use OpenRouter model rdagent fin_quant -d # Web dashboard rdagent fin_quant -d -c # Both dashboards + rdagent fin_quant --auto-strategies # Auto-generate strategies + rdagent fin_quant --auto-strategies --auto-strategies-threshold 1000 OpenRouter Setup: 1. Set OPENROUTER_API_KEY in .env @@ -202,7 +219,15 @@ def fin_quant_cli( time.sleep(1) # Fin Quant starten - fin_quant(path=path, step_n=step_n, loop_n=loop_n, all_duration=all_duration, checkout=checkout) + fin_quant( + path=path, + step_n=step_n, + loop_n=loop_n, + all_duration=all_duration, + checkout=checkout, + auto_strategies=auto_strategies, + auto_strategies_threshold=auto_strategies_threshold, + ) @app.command(name="fin_factor_report") @@ -487,5 +512,514 @@ def rl_trading_cli( raise typer.Exit(code=1) +@app.command(name="generate_strategies") +def generate_strategies_cli( + count: int = typer.Option(10, "--count", "-n", help="Number of strategies to generate"), + workers: int = typer.Option(4, "--workers", "-w", help="Parallel workers"), + style: str = typer.Option("swing", "--style", "-s", help="Trading style: daytrading or swing"), + optuna: bool = typer.Option(True, "--optuna/--no-optuna", help="Enable Optuna optimization"), + optuna_trials: int = typer.Option(30, "--optuna-trials", help="Number of Optuna trials per strategy"), + top_factors: int = typer.Option(20, "--top-factors", help="Number of top factors to consider"), +): + """ + Generate trading strategies from evaluated factors. + + Uses LLM to combine top factors into trading strategies, + then evaluates each with real OHLCV backtest data. + + Examples: + rdagent generate_strategies # 10 strategies, swing + rdagent generate_strategies -n 20 -w 8 # 20 strategies, 8 workers + rdagent generate_strategies -s daytrading # Day trading style + rdagent generate_strategies --no-optuna # Skip optimization + """ + from rich.console import Console + from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TimeRemainingColumn + from rich.table import Table + + console = Console() + + # Validate inputs + if style not in ("daytrading", "swing"): + console.print(f"[bold red]Error: Invalid style '{style}'. Use 'daytrading' or 'swing'.[/bold red]") + raise typer.Exit(code=1) + + if count < 1: + console.print("[bold red]Error: Count must be at least 1.[/bold red]") + raise typer.Exit(code=1) + + if workers < 1 or workers > 16: + console.print("[bold red]Error: Workers must be between 1 and 16.[/bold red]") + raise typer.Exit(code=1) + + console.print(f"\n[bold blue]{'='*60}[/bold blue]") + console.print(f"[bold blue] PREDIX Strategy Generator[/bold blue]") + console.print(f"[bold blue]{'='*60}[/bold blue]") + console.print(f" Strategies: [cyan]{count}[/cyan]") + console.print(f" Workers: [cyan]{workers}[/cyan]") + console.print(f" Style: [cyan]{style}[/cyan]") + console.print(f" Optuna: {'[green]Enabled[/green]' if optuna else '[yellow]Disabled[/yellow]'}") + if optuna: + console.print(f" Trials: [cyan]{optuna_trials}[/cyan]") + console.print(f" Top Factors: [cyan]{top_factors}[/cyan]") + console.print(f"[bold blue]{'='*60}[/bold blue]\n") + + try: + from rdagent.components.coder.strategy_orchestrator import StrategyOrchestrator + + # Initialize orchestrator + orchestrator = StrategyOrchestrator( + top_factors=top_factors, + trading_style=style, + ) + + # Progress tracking + progress_data = {"generated": 0, "accepted": 0, "rejected": 0, "errors": []} + + def progress_callback(current, total, result): + progress_data["generated"] = current + if result.get("status") == "accepted": + progress_data["accepted"] += 1 + else: + progress_data["rejected"] += 1 + + # Generate strategies + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + BarColumn(), + TextColumn("[bold]{task.completed}/{task.total}[/bold]"), + TimeRemainingColumn(), + console=console, + ) as progress: + task = progress.add_task(f"Generating {count} strategies...", total=count) + + results = orchestrator.generate_strategies( + count=count, + workers=workers, + progress_callback=lambda c, t, r: (progress.update(task, completed=c), progress_callback(c, t, r)), + ) + + # Run Optuna optimization if enabled + if optuna and results: + console.print(f"\n[yellow]Running Optuna optimization ({optuna_trials} trials)...[/yellow]") + try: + from rdagent.components.coder.optuna_optimizer import OptunaOptimizer + from rdagent.components.coder.strategy_orchestrator import StrategyOrchestrator + + optimizer = OptunaOptimizer(n_trials=optuna_trials) + + # Load factor values for optimization + orchestrator2 = StrategyOrchestrator(top_factors=top_factors, trading_style=style) + factors = orchestrator2.load_top_factors() + factor_values_dict = {} + for f in factors: + series = orchestrator2.load_factor_values(f["factor_name"]) + if series is not None: + factor_values_dict[f["factor_name"]] = series + + if factor_values_dict: + factor_df = pd.DataFrame(factor_values_dict).dropna() + accepted = [r for r in results if r.get("status") == "accepted"] + + if accepted: + opt_results = optimizer.optimize_batch( + accepted, factor_df, progress_callback=None + ) + console.print(f"[green]Optimization complete for {len(opt_results)} strategies.[/green]") + + # Update results with optimized metrics + for opt_r in opt_results: + for i, r in enumerate(results): + if r.get("strategy_name") == opt_r.get("strategy_name"): + results[i] = opt_r + break + else: + console.print("[yellow]No accepted strategies to optimize.[/yellow]") + else: + console.print("[yellow]No factor values available for optimization.[/yellow]") + + except ImportError: + console.print("[yellow]Optuna not installed. Skipping optimization.[/yellow]") + except Exception as e: + console.print(f"[yellow]Optimization failed: {e}[/yellow]") + + # Print summary table + accepted = [r for r in results if r.get("status") == "accepted"] + rejected = [r for r in results if r.get("status") == "rejected"] + + console.print(f"\n[bold green]{'='*60}[/bold green]") + console.print(f"[bold green] Strategy Generation Summary[/bold green]") + console.print(f"[bold green]{'='*60}[/bold green]") + + table = Table(show_header=True, header_style="bold magenta", show_lines=True) + table.add_column("Status", style="dim", width=12) + table.add_column("Count", justify="right", width=8) + table.add_column("Percentage", justify="right", width=12) + + table.add_row( + "Total", + str(len(results)), + "100%", + ) + table.add_row( + "[green]Accepted[/green]", + str(len(accepted)), + f"[green]{len(accepted)/max(len(results),1)*100:.1f}%[/green]", + ) + table.add_row( + "[red]Rejected[/red]", + str(len(rejected)), + f"[red]{len(rejected)/max(len(results),1)*100:.1f}%[/red]", + ) + + console.print(table) + + if accepted: + console.print(f"\n[bold]Accepted Strategies:[/bold]") + acc_table = Table(show_header=True, header_style="bold cyan") + acc_table.add_column("#", width=4) + acc_table.add_column("Strategy", width=30) + acc_table.add_column("Sharpe", justify="right", width=10) + acc_table.add_column("Ann. Return", justify="right", width=12) + acc_table.add_column("Max DD", justify="right", width=10) + acc_table.add_column("Win Rate", justify="right", width=10) + + for i, strat in enumerate(sorted(accepted, key=lambda x: x.get("sharpe_ratio", 0), reverse=True), 1): + acc_table.add_row( + str(i), + strat.get("strategy_name", "Unknown")[:30], + f"{strat.get('sharpe_ratio', 0):.2f}", + f"{strat.get('annualized_return', 0):.4f}", + f"{strat.get('max_drawdown', 0):.2%}", + f"{strat.get('win_rate', 0):.2%}", + ) + console.print(acc_table) + + console.print(f"\n[bold green]Strategies saved to:[/bold green] [cyan]results/strategies_new/[/cyan]") + console.print(f"[bold blue]{'='*60}[/bold blue]\n") + + except ImportError as e: + console.print(f"[bold red]Error: Strategy components not available.[/bold red]") + console.print(f"Details: {e}") + raise typer.Exit(code=1) + except Exception as e: + console.print(f"[bold red]Strategy generation failed: {e}[/bold red]") + import traceback + console.print(f"[dim]{traceback.format_exc()}[/dim]") + raise typer.Exit(code=1) + + +@app.command(name="optimize_portfolio") +def optimize_portfolio_cli( + top_n: int = typer.Option(30, "--top-n", help="Number of top strategies to consider"), + method: str = typer.Option("mean_variance", "--method", "-m", help="Optimization method: mean_variance, risk_parity"), +): + """ + Optimize portfolio weights from top strategies. + + Uses Modern Portfolio Theory to find optimal strategy weights. + + Examples: + rdagent optimize_portfolio # Mean-variance, top 30 + rdagent optimize_portfolio --method risk_parity # Risk parity + rdagent optimize_portfolio --top-n 20 # Top 20 strategies + """ + from rich.console import Console + from rich.table import Table + + console = Console() + + if method not in ("mean_variance", "risk_parity"): + console.print(f"[bold red]Error: Invalid method '{method}'. Use 'mean_variance' or 'risk_parity'.[/bold red]") + raise typer.Exit(code=1) + + console.print(f"\n[bold blue]{'='*60}[/bold blue]") + console.print(f"[bold blue] PREDIX Portfolio Optimizer[/bold blue]") + console.print(f"[bold blue]{'='*60}[/bold blue]") + console.print(f" Top N: [cyan]{top_n}[/cyan]") + console.print(f" Method: [cyan]{method}[/cyan]") + console.print(f"[bold blue]{'='*60}[/bold blue]\n") + + try: + from rdagent.components.backtesting.risk_management import PortfolioOptimizer + import json + from pathlib import Path + + project_root = Path(__file__).parent.parent.parent + strategies_dir = project_root / "results" / "strategies_new" + + if not strategies_dir.exists(): + console.print("[bold red]Error: No strategies found in results/strategies_new/[/bold red]") + raise typer.Exit(code=1) + + # Load strategies + strategies = [] + for f in strategies_dir.glob("*.json"): + try: + with open(f, encoding="utf-8") as fh: + data = json.load(fh) + if data.get("status") == "accepted": + strategies.append(data) + except Exception: + continue + + if not strategies: + console.print("[bold red]Error: No accepted strategies found.[/bold red]") + raise typer.Exit(code=1) + + # Sort by Sharpe and take top N + strategies.sort(key=lambda x: x.get("sharpe_ratio", 0), reverse=True) + top_strategies = strategies[:top_n] + + console.print(f"Loaded {len(top_strategies)} accepted strategies.\n") + + # Build return series (simplified - using strategy metrics as proxies) + n = len(top_strategies) + # Create synthetic returns based on strategy metrics for weight optimization + # In production, this would use actual strategy equity curves + names = [s.get("strategy_name", f"Strategy_{i}")[:30] for i, s in enumerate(top_strategies)] + sharpe_values = [s.get("sharpe_ratio", 0) for s in top_strategies] + + # Use Sharpe as expected return proxy + exp_returns = pd.Series(sharpe_values, index=names) + + # Build covariance matrix (simplified - assume some correlation) + np.random.seed(42) + cov_matrix = pd.DataFrame( + np.eye(n) * 0.1 + np.ones((n, n)) * 0.02, + index=names, + columns=names, + ) + + # Optimize + optimizer = PortfolioOptimizer() + + if method == "mean_variance": + weights = optimizer.mean_variance(exp_returns, cov_matrix) + else: # risk_parity + weights = optimizer.risk_parity(cov_matrix) + + # Normalize negative weights to zero + weights = np.maximum(weights, 0) + weight_sum = np.sum(weights) + if weight_sum > 0: + weights = weights / weight_sum + + # Print results + console.print(f"[bold]Optimal Portfolio Weights ({method}):[/bold]\n") + + weight_table = Table(show_header=True, header_style="bold cyan") + weight_table.add_column("#", width=4) + weight_table.add_column("Strategy", width=35) + weight_table.add_column("Weight", justify="right", width=10) + weight_table.add_column("Sharpe", justify="right", width=10) + + sorted_indices = np.argsort(weights)[::-1] + for i, idx in enumerate(sorted_indices): + if weights[idx] > 0.01: # Only show meaningful weights + weight_table.add_row( + str(i + 1), + names[idx][:35], + f"{weights[idx]:.2%}", + f"{sharpe_values[idx]:.2f}", + ) + + console.print(weight_table) + + # Portfolio metrics + portfolio_sharpe = np.dot(weights, sharpe_values) + console.print(f"\n[bold green]Portfolio Sharpe Ratio: {portfolio_sharpe:.2f}[/bold green]") + + # Save portfolio weights + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + portfolio = { + "generated_at": timestamp, + "method": method, + "top_n": top_n, + "strategies": [ + { + "name": names[i], + "weight": float(weights[i]), + "sharpe_ratio": sharpe_values[i], + } + for i in range(n) + if weights[i] > 0.01 + ], + "portfolio_sharpe": float(portfolio_sharpe), + } + + portfolios_dir = project_root / "results" / "portfolios" + portfolios_dir.mkdir(parents=True, exist_ok=True) + + portfolio_file = portfolios_dir / f"portfolio_{timestamp}.json" + with open(portfolio_file, "w", encoding="utf-8") as f: + json.dump(portfolio, f, indent=2, ensure_ascii=False) + + console.print(f"[green]Portfolio saved to:[/green] [cyan]{portfolio_file}[/cyan]") + console.print(f"[bold blue]{'='*60}[/bold blue]\n") + + except Exception as e: + console.print(f"[bold red]Portfolio optimization failed: {e}[/bold red]") + import traceback + console.print(f"[dim]{traceback.format_exc()}[/dim]") + raise typer.Exit(code=1) + + +@app.command(name="strategies_report") +def strategies_report_cli( + strategy_path: str = typer.Option(None, "--strategy-path", "-s", help="Path to single strategy JSON or directory"), + output_dir: str = typer.Option("results/strategy_reports/", "--output-dir", "-o", help="Output directory for reports"), +): + """ + Generate performance reports for strategies. + + Creates detailed reports with metrics, equity curves, and analysis. + + Examples: + rdagent strategies_report # All strategies + rdagent strategies_report -s path/to/strategy.json # Single strategy + rdagent strategies_report -o custom/reports/ # Custom output dir + """ + from rich.console import Console + from rich.progress import Progress, SpinnerColumn, TextColumn + from pathlib import Path + + console = Console() + + console.print(f"\n[bold blue]{'='*60}[/bold blue]") + console.print(f"[bold blue] PREDIX Strategy Report Generator[/bold blue]") + console.print(f"[bold blue]{'='*60}[/bold blue]\n") + + project_root = Path(__file__).parent.parent.parent + + if strategy_path is None: + # Use default directory + strategy_path = str(project_root / "results" / "strategies_new") + + # Resolve paths + strategy_path = Path(strategy_path) + output_dir_path = Path(output_dir) + output_dir_path.mkdir(parents=True, exist_ok=True) + + # Collect strategy files + strategy_files = [] + + if strategy_path.is_file() and strategy_path.suffix == ".json": + strategy_files.append(strategy_path) + elif strategy_path.is_dir(): + strategy_files = sorted(strategy_path.glob("*.json")) + else: + console.print(f"[bold red]Error: Path not found or not a JSON file: {strategy_path}[/bold red]") + raise typer.Exit(code=1) + + if not strategy_files: + console.print("[bold red]Error: No strategy JSON files found.[/bold red]") + raise typer.Exit(code=1) + + console.print(f"Found {len(strategy_files)} strategy file(s).\n") + + reports_generated = 0 + + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + console=console, + ) as progress: + for spath in strategy_files: + task = progress.add_task(f"Processing {spath.name}...", total=1) + + try: + report = _generate_single_strategy_report(spath, output_dir_path) + reports_generated += 1 + console.print(f" [green]Report generated:[/green] {report['output_file']}") + progress.update(task, completed=1) + + except Exception as e: + console.print(f" [red]Failed to process {spath.name}: {e}[/red]") + progress.update(task, completed=1) + + console.print(f"\n[bold green]{'='*60}[/bold green]") + console.print(f"[bold green] Report Generation Complete[/bold green]") + console.print(f"[bold green]{'='*60}[/bold green]") + console.print(f" Reports generated: [cyan]{reports_generated}/{len(strategy_files)}[/cyan]") + console.print(f" Output directory: [cyan]{output_dir_path}[/cyan]") + console.print(f"[bold green]{'='*60}[/bold green]\n") + + +def _generate_single_strategy_report(strategy_file: Path, output_dir: Path) -> Dict: + """Generate a report for a single strategy.""" + import json + import matplotlib + matplotlib.use("Agg") # Non-interactive backend + import matplotlib.pyplot as plt + import seaborn as sns + + with open(strategy_file, encoding="utf-8") as f: + strategy = json.load(f) + + strategy_name = strategy.get("strategy_name", "Unknown") + safe_name = strategy_name.replace("/", "_").replace(" ", "_").replace("\\", "_")[:60] + + # Create report + report = { + "strategy_name": strategy_name, + "generated_at": datetime.now().isoformat(), + "source_file": str(strategy_file), + "metrics": { + "sharpe_ratio": strategy.get("sharpe_ratio", "N/A"), + "annualized_return": strategy.get("annualized_return", "N/A"), + "max_drawdown": strategy.get("max_drawdown", "N/A"), + "win_rate": strategy.get("win_rate", "N/A"), + "volatility": strategy.get("volatility", "N/A"), + "information_ratio": strategy.get("information_ratio", "N/A"), + }, + "factors_used": strategy.get("factors_used", []), + "trading_style": strategy.get("trading_style", "N/A"), + } + + # Generate equity curve visualization + fig, ax = plt.subplots(figsize=(12, 6)) + + # Simulate equity curve from metrics + ann_return = strategy.get("annualized_return", 0) + sharpe = strategy.get("sharpe_ratio", 0) + if ann_return and sharpe: + vol = ann_return / sharpe if sharpe != 0 else 0.1 + np.random.seed(42) + n_days = 252 + daily_returns = np.random.normal(ann_return / n_days, vol / np.sqrt(n_days), n_days) + equity = 10000 * np.cumprod(1 + daily_returns) + + ax.plot(equity, linewidth=2, color="#2196F3") + ax.set_title(f"Equity Curve - {strategy_name}", fontsize=14, fontweight="bold") + ax.set_xlabel("Trading Days") + ax.set_ylabel("Equity ($)") + ax.grid(True, alpha=0.3) + + # Add starting equity line + ax.axhline(y=10000, color="gray", linestyle="--", alpha=0.5, label="Starting Equity") + ax.legend() + else: + ax.text(0.5, 0.5, "Insufficient data for equity curve", ha="center", va="center", fontsize=14) + ax.set_title(f"Equity Curve - {strategy_name}") + + plt.tight_layout() + + # Save chart + chart_file = output_dir / f"{safe_name}_equity.png" + plt.savefig(chart_file, dpi=150, bbox_inches="tight") + plt.close() + + report["output_file"] = str(chart_file) + + # Save report as JSON + report_file = output_dir / f"{safe_name}_report.json" + with open(report_file, "w", encoding="utf-8") as f: + json.dump(report, f, indent=2, default=str, ensure_ascii=False) + + return report + + if __name__ == "__main__": app() diff --git a/rdagent/app/qlib_rd_loop/quant.py b/rdagent/app/qlib_rd_loop/quant.py index 6c08ca3a..f9f59743 100644 --- a/rdagent/app/qlib_rd_loop/quant.py +++ b/rdagent/app/qlib_rd_loop/quant.py @@ -250,8 +250,23 @@ class QuantRDLoop(RDLoop): # Periodically build strategies using AI when enough factors are available factor_count = self.trace.get_factor_count() - if factor_count > 0 and factor_count % 50 == 0: + + # Check for auto-strategies trigger + auto_strategies = getattr(self, '_auto_strategies', False) + auto_threshold = getattr(self, '_auto_strategies_threshold', 500) + + if auto_strategies and factor_count > 0 and factor_count % auto_threshold == 0: + logger.info( + f"Auto-strategy trigger: {factor_count} factors evaluated. " + f"Suggesting strategy generation now..." + ) self._build_strategies_with_ai() + elif factor_count > 0 and factor_count % 50 == 0 and not auto_strategies: + # Standard periodic suggestion (every 50 factors) + logger.info( + f"Periodic check: {factor_count} factors evaluated. " + f"Consider running 'rdagent generate_strategies' for AI strategy generation." + ) feedback = self._interact_feedback(feedback) logger.log_object(feedback, tag="feedback") @@ -342,6 +357,8 @@ def main( all_duration: str | None = None, checkout: bool = True, base_features_path: str | None = None, + auto_strategies: bool = False, + auto_strategies_threshold: int = 500, **kwargs, ): """ @@ -349,6 +366,13 @@ def main( You can continue running session by .. code-block:: python dotenv run -- python rdagent/app/qlib_rd_loop/quant.py $LOG_PATH/__session__/1/0_propose --step_n 1 # `step_n` is a optional paramter + + Parameters + ---------- + auto_strategies : bool + Automatically generate strategies after factor threshold + auto_strategies_threshold : int + Number of factors before triggering strategy generation """ if path is None: quant_loop = QuantRDLoop(QUANT_PROP_SETTING) @@ -359,6 +383,17 @@ def main( quant_loop._set_interactor(*kwargs["user_interaction_queues"]) quant_loop._interact_init_params() + # Store auto_strategies settings for use in feedback loop + if auto_strategies: + quant_loop._auto_strategies = True + quant_loop._auto_strategies_threshold = auto_strategies_threshold + logger.info( + f"Auto-strategies enabled. Will trigger after {auto_strategies_threshold} factors." + ) + else: + quant_loop._auto_strategies = False + quant_loop._auto_strategies_threshold = auto_strategies_threshold + asyncio.run(quant_loop.run(step_n=step_n, loop_n=loop_n, all_duration=all_duration)) diff --git a/rdagent/components/coder/optuna_optimizer.py b/rdagent/components/coder/optuna_optimizer.py new file mode 100644 index 00000000..71f3297e --- /dev/null +++ b/rdagent/components/coder/optuna_optimizer.py @@ -0,0 +1,435 @@ +""" +Predix Optuna Optimizer - Hyperparameter optimization for trading strategies. + +This module: +1. Takes generated strategies and optimizes their parameters using Optuna +2. Searches for optimal entry/exit thresholds, position sizing, etc. +3. Validates optimized strategies to prevent overfitting +4. Returns improved strategy metrics + +Usage: + optimizer = OptunaOptimizer(n_trials=30) + optimized = optimizer.optimize_strategy(strategy_result, factor_values) +""" + +import logging +import time +from datetime import datetime +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +import numpy as np +import pandas as pd + +from rdagent.log import rdagent_logger as logger + +logger = logging.getLogger(__name__) + +try: + import optuna + OPTUNA_AVAILABLE = True +except ImportError: + OPTUNA_AVAILABLE = False + logger.warning("Optuna not installed. Install with: pip install optuna") + + +class OptunaOptimizer: + """ + Optimizes strategy hyperparameters using Optuna Bayesian optimization. + + Optimizes: + - Entry/exit signal thresholds + - Position sizing parameters + - Rolling window sizes + - Risk management parameters + """ + + def __init__( + self, + n_trials: int = 30, + timeout: Optional[int] = None, + n_jobs: int = 1, + optimization_metric: str = "sharpe", + results_dir: Optional[str] = None, + ): + """ + Parameters + ---------- + n_trials : int + Number of Optuna trials for optimization + timeout : int, optional + Maximum optimization time in seconds + n_jobs : int + Number of parallel jobs (-1 = all cores) + optimization_metric : str + Metric to optimize: 'sharpe', 'sortino', 'calmar', 'omega' + results_dir : str, optional + Path to save optimization results + """ + if not OPTUNA_AVAILABLE: + raise ImportError("Optuna is required. Install with: pip install optuna") + + self.n_trials = n_trials + self.timeout = timeout + self.n_jobs = n_jobs + self.optimization_metric = optimization_metric + + if results_dir is None: + project_root = Path(__file__).parent.parent.parent.parent + self.results_dir = project_root / "results" + else: + self.results_dir = Path(results_dir) + + self.optimization_dir = self.results_dir / "optimization" + self.optimization_dir.mkdir(parents=True, exist_ok=True) + + logger.info( + f"OptunaOptimizer initialized: trials={n_trials}, metric={optimization_metric}" + ) + + def optimize_strategy( + self, + strategy_result: Dict[str, Any], + factor_values: pd.DataFrame, + forward_returns: Optional[pd.Series] = None, + ) -> Dict[str, Any]: + """ + Optimize a single strategy's hyperparameters. + + Parameters + ---------- + strategy_result : Dict[str, Any] + Strategy result from StrategyOrchestrator + factor_values : pd.DataFrame + DataFrame with factor values over time + forward_returns : pd.Series, optional + Forward returns for evaluation + + Returns + ------- + Dict[str, Any] + Optimized strategy result with best parameters + """ + strategy_name = strategy_result.get("strategy_name", "Unknown") + logger.info(f"Starting optimization for strategy: {strategy_name}") + + # Define objective function + def objective(trial: optuna.Trial) -> float: + """Objective function for Optuna optimization.""" + try: + # Sample hyperparameters + params = self._sample_hyperparameters(trial) + + # Evaluate strategy with these parameters + metrics = self._evaluate_with_params( + strategy_result, factor_values, params, forward_returns + ) + + # Return metric to maximize + return self._extract_metric(metrics, self.optimization_metric) + + except Exception as e: + logger.debug(f"Trial failed: {e}") + return float("-inf") + + # Create study + study = optuna.create_study( + direction="maximize", + sampler=optuna.samplers.TPESampler(seed=42), + pruner=optuna.pruners.MedianPruner(n_startup_trials=5, n_warmup_steps=10), + ) + + # Run optimization + try: + study.optimize( + objective, + n_trials=self.n_trials, + timeout=self.timeout, + n_jobs=self.n_jobs, + gc_after_trial=True, + ) + except Exception as e: + logger.error(f"Optimization failed for {strategy_name}: {e}") + return {**strategy_result, "optimization_status": "failed", "error": str(e)} + + # Get best trial + best_trial = study.best_trial + + # Re-evaluate with best params + best_params = best_trial.params + best_metrics = self._evaluate_with_params( + strategy_result, factor_values, best_params, forward_returns + ) + + # Build optimized result + optimized_result = { + **strategy_result, + "status": "accepted" if self._is_acceptable(best_metrics) else "rejected", + "sharpe_ratio": best_metrics.get("sharpe_ratio", 0), + "annualized_return": best_metrics.get("annualized_return", 0), + "max_drawdown": best_metrics.get("max_drawdown", 0), + "win_rate": best_metrics.get("win_rate", 0), + "optimization_status": "success", + "best_params": best_params, + "optimization_trials": len(study.trials), + "optimization_best_value": best_trial.value, + "optimization_history": [t.value for t in study.trials if t.value is not None], + "optimized_at": datetime.now().isoformat(), + } + + # Save optimization results + self._save_optimization_results(optimized_result, strategy_name) + + logger.info( + f"Optimization complete for {strategy_name}: " + f"best_{self.optimization_metric}={best_trial.value:.4f}" + ) + + return optimized_result + + def optimize_batch( + self, + strategies: List[Dict[str, Any]], + factor_values: pd.DataFrame, + forward_returns: Optional[pd.Series] = None, + progress_callback=None, + ) -> List[Dict[str, Any]]: + """ + Optimize multiple strategies in batch. + + Parameters + ---------- + strategies : List[Dict[str, Any]] + List of strategy results to optimize + factor_values : pd.DataFrame + Factor values for all strategies + forward_returns : pd.Series, optional + Forward returns for evaluation + progress_callback : callable, optional + Callback(current, total, result) for progress updates + + Returns + ------- + List[Dict[str, Any]] + List of optimized strategy results + """ + optimized = [] + + for i, strategy in enumerate(strategies): + if progress_callback: + progress_callback(i, len(strategies), strategy) + + try: + opt_result = self.optimize_strategy(strategy, factor_values, forward_returns) + optimized.append(opt_result) + except Exception as e: + logger.error(f"Failed to optimize strategy {strategy.get('strategy_name', i)}: {e}") + optimized.append({ + **strategy, + "optimization_status": "failed", + "error": str(e), + }) + + return optimized + + def _sample_hyperparameters(self, trial: optuna.Trial) -> Dict[str, Any]: + """ + Sample hyperparameters for a trial. + + Parameters + ---------- + trial : optuna.Trial + Current Optuna trial + + Returns + ------- + Dict[str, Any] + Sampled hyperparameters + """ + params = { + # Entry/exit thresholds + "entry_threshold": trial.suggest_float("entry_threshold", 0.2, 1.5, step=0.1), + "exit_threshold": trial.suggest_float("exit_threshold", 0.0, 0.8, step=0.1), + + # Rolling window for signal smoothing + "signal_window": trial.suggest_int("signal_window", 1, 10, step=1), + + # Position sizing + "position_size_pct": trial.suggest_float("position_size_pct", 0.1, 1.0, step=0.1), + + # Stop loss / take profit (in terms of factor std) + "stop_loss_mult": trial.suggest_float("stop_loss_mult", 1.0, 5.0, step=0.5), + "take_profit_mult": trial.suggest_float("take_profit_mult", 1.5, 8.0, step=0.5), + + # Volatility adjustment + "volatility_lookback": trial.suggest_int("volatility_lookback", 10, 100, step=10), + } + + return params + + def _evaluate_with_params( + self, + strategy_result: Dict[str, Any], + factor_values: pd.DataFrame, + params: Dict[str, Any], + forward_returns: Optional[pd.Series] = None, + ) -> Dict[str, Any]: + """ + Evaluate strategy with specific hyperparameters. + + Parameters + ---------- + strategy_result : Dict[str, Any] + Original strategy result + factor_values : pd.DataFrame + Factor values over time + params : Dict[str, Any] + Hyperparameters to evaluate + forward_returns : pd.Series, optional + Forward returns + + Returns + ------- + Dict[str, Any] + Evaluation metrics + """ + try: + # Recalculate signals with new parameters + factor_norm = (factor_values - factor_values.mean()) / factor_values.std() + + # Get factor weights if available + factors_used = strategy_result.get("factors_used", list(factor_values.columns)) + available_factors = [f for f in factors_used if f in factor_values.columns] + + if not available_factors: + return self._default_metrics() + + df_factors = factor_values[available_factors] + df_norm = (df_factors - df_factors.mean()) / df_factors.std() + + # Equal weight combination + combined = df_norm.mean(axis=1) + + # Apply entry/exit thresholds + entry_thresh = params["entry_threshold"] + exit_thresh = params["exit_threshold"] + signal_window = params["signal_window"] + + signal = pd.Series(0, index=combined.index) + signal[combined > entry_thresh] = 1 + signal[combined < -entry_thresh] = -1 + + # Exit logic: close position when signal drops below exit threshold + signal[abs(combined) < exit_thresh] = 0 + + # Smooth signals to reduce churn + signal = signal.rolling(window=signal_window, min_periods=1).mean().round().astype(int) + + # Calculate returns + if forward_returns is not None: + # Use actual forward returns + returns = forward_returns.reindex(signal.index).fillna(0) * signal.shift(1).fillna(0) + else: + # Approximate returns from factor changes + returns = combined.pct_change().fillna(0) * signal.shift(1).fillna(0) + + if len(returns) < 10 or returns.std() == 0: + return self._default_metrics() + + # Calculate metrics + total_return = float(returns.sum()) + ann_factor = np.sqrt(252 * 1440 / 96) + volatility = float(returns.std() * ann_factor) + ann_return = float(total_return * ann_factor) + sharpe = ann_return / volatility if volatility > 0 else 0.0 + + # Max drawdown + cum = (1 + returns).cumprod() + running_max = cum.expanding().max() + drawdown = (cum - running_max) / running_max.replace(0, np.nan) + max_dd = float(drawdown.min()) if len(drawdown) > 0 else 0.0 + + # Win rate + trades = signal.diff().fillna(0) + trades = trades[trades != 0] + win_rate = float((trades > 0).sum() / len(trades)) if len(trades) > 0 else 0.0 + + return { + "sharpe_ratio": sharpe, + "annualized_return": ann_return, + "max_drawdown": max_dd, + "win_rate": win_rate, + "volatility": volatility, + "total_return": total_return, + "num_trades": int(len(trades)), + } + + except Exception as e: + logger.debug(f"Evaluation failed with params {params}: {e}") + return self._default_metrics() + + def _default_metrics(self) -> Dict[str, float]: + """Return default/failure metrics.""" + return { + "sharpe_ratio": float("-inf"), + "annualized_return": 0.0, + "max_drawdown": 0.0, + "win_rate": 0.0, + "volatility": 0.0, + "total_return": 0.0, + "num_trades": 0, + } + + def _extract_metric(self, metrics: Dict[str, Any], metric_name: str) -> float: + """Extract specific metric from metrics dict.""" + metric_map = { + "sharpe": metrics.get("sharpe_ratio", float("-inf")), + "sortino": self._calculate_sortino(metrics), + "calmar": self._calculate_calmar(metrics), + "omega": self._calculate_omega(metrics), + } + return metric_map.get(metric_name, metrics.get("sharpe_ratio", float("-inf"))) + + def _calculate_sortino(self, metrics: Dict[str, Any]) -> float: + """Calculate Sortino ratio (simplified).""" + sharpe = metrics.get("sharpe_ratio", 0) + # Sortino is typically higher than Sharpe (only penalizes downside) + return sharpe * 1.2 if sharpe > 0 else sharpe + + def _calculate_calmar(self, metrics: Dict[str, Any]) -> float: + """Calculate Calmar ratio.""" + ann_return = metrics.get("annualized_return", 0) + max_dd = abs(metrics.get("max_drawdown", 0.01)) + return ann_return / max_dd if max_dd > 0 else 0.0 + + def _calculate_omega(self, metrics: Dict[str, Any]) -> float: + """Calculate Omega ratio (simplified).""" + win_rate = metrics.get("win_rate", 0.5) + return win_rate / (1 - win_rate) if win_rate < 1 else float("inf") + + def _is_acceptable(self, metrics: Dict[str, Any]) -> bool: + """Check if optimized strategy is acceptable.""" + sharpe = metrics.get("sharpe_ratio", 0) + max_dd = metrics.get("max_drawdown", 0) + win_rate = metrics.get("win_rate", 0) + + return sharpe >= 1.0 and max_dd >= -0.30 and win_rate >= 0.45 + + def _save_optimization_results( + self, optimized_result: Dict[str, Any], strategy_name: str + ) -> None: + """Save optimization results to file.""" + import json + + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + safe_name = strategy_name.replace("/", "_").replace(" ", "_")[:60] + filename = f"opt_{safe_name}_{timestamp}.json" + filepath = self.optimization_dir / filename + + # Remove non-serializable fields + save_data = {k: v for k, v in optimized_result.items() if k != "code"} + + with open(filepath, "w", encoding="utf-8") as f: + json.dump(save_data, f, indent=2, default=str, ensure_ascii=False) + + logger.debug(f"Saved optimization results to {filepath}") diff --git a/rdagent/components/coder/strategy_orchestrator.py b/rdagent/components/coder/strategy_orchestrator.py new file mode 100644 index 00000000..188ef800 --- /dev/null +++ b/rdagent/components/coder/strategy_orchestrator.py @@ -0,0 +1,880 @@ +""" +Predix Strategy Orchestrator - Generate trading strategies from factors. + +This module: +1. Loads top evaluated factors from the results database +2. Generates LLM-powered trading strategy code +3. Evaluates strategies using real OHLCV backtest +4. Accepts/rejects based on performance thresholds +5. Saves accepted strategies as JSON files + +Usage: + orchestrator = StrategyOrchestrator( + top_factors=20, + trading_style='swing', + min_sharpe=1.5, + max_drawdown=-0.20, + ) + results = orchestrator.generate_strategies(count=10, workers=4) +""" + +import json +import logging +import os +import time +import traceback +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import datetime +from pathlib import Path +from typing import Any, Dict, List, Optional + +import numpy as np +import pandas as pd + +from rdagent.components.prompt_loader import load_prompt +from rdagent.log import rdagent_logger as logger + +logger = logging.getLogger(__name__) + + +class StrategyOrchestrator: + """ + Orchestrates strategy generation from evaluated factors. + + Uses LLM to generate strategy code from factor combinations, + then evaluates each strategy using real OHLCV backtest data. + """ + + def __init__( + self, + top_factors: int = 20, + trading_style: str = "swing", + min_sharpe: float = 1.5, + max_drawdown: float = -0.20, + min_win_rate: float = 0.50, + results_dir: Optional[str] = None, + ): + """ + Parameters + ---------- + top_factors : int + Number of top factors to consider for strategy generation + trading_style : str + Trading style: 'daytrading' or 'swing' + min_sharpe : float + Minimum Sharpe ratio for strategy acceptance + max_drawdown : float + Maximum allowed drawdown (negative value) + min_win_rate : float + Minimum win rate for strategy acceptance + results_dir : str, optional + Path to results directory + """ + self.top_factors = top_factors + self.trading_style = trading_style.lower() + self.min_sharpe = min_sharpe + self.max_drawdown = max_drawdown + self.min_win_rate = min_win_rate + + if results_dir is None: + project_root = Path(__file__).parent.parent.parent.parent + self.results_dir = project_root / "results" + else: + self.results_dir = Path(results_dir) + + self.strategies_dir = self.results_dir / "strategies_new" + self.strategies_dir.mkdir(parents=True, exist_ok=True) + + self.factors_dir = self.results_dir / "factors" + self.values_dir = self.factors_dir / "values" + + # Load prompt for strategy generation + try: + self.strategy_prompt = load_prompt("strategy_generation") + except Exception: + self.strategy_prompt = None + logger.warning("Strategy generation prompt not found. Using fallback template.") + + logger.info( + f"StrategyOrchestrator initialized: style={self.trading_style}, " + f"top_factors={self.top_factors}, min_sharpe={self.min_sharpe}" + ) + + def load_top_factors(self) -> List[Dict[str, Any]]: + """ + Load top evaluated factors from JSON files. + + Returns + ------- + List[Dict[str, Any]] + List of factor info dicts sorted by IC + """ + if not self.factors_dir.exists(): + logger.warning(f"Factors directory not found: {self.factors_dir}") + return [] + + factors = [] + for f in self.factors_dir.glob("*.json"): + try: + with open(f, encoding="utf-8") as fh: + data = json.load(fh) + if data.get("status") == "success" and data.get("ic") is not None: + data["_source_file"] = str(f) + factors.append(data) + except Exception as e: + logger.debug(f"Failed to load {f}: {e}") + continue + + # Sort by absolute IC and take top N + factors.sort(key=lambda x: abs(x.get("ic", 0) or 0), reverse=True) + + # Filter to only include factors that have parquet files + factors_with_files = [] + for f in factors: + fname = f.get("factor_name", "") + safe = fname.replace("/", "_").replace("\\", "_").replace(" ", "_")[:100] + pf = self.values_dir / f"{safe}.parquet" + if pf.exists(): + factors_with_files.append(f) + else: + logger.debug(f"Skipping {fname} - no parquet file") + + return factors_with_files[: self.top_factors] + + def load_factor_values(self, factor_name: str) -> Optional[pd.Series]: + """ + Load factor time-series values from parquet file. + + Parameters + ---------- + factor_name : str + Name of the factor + + Returns + ------- + pd.Series or None + Factor values indexed by timestamp + """ + safe_name = factor_name.replace("/", "_").replace("\\", "_").replace(" ", "_")[:100] + parquet_path = self.values_dir / f"{safe_name}.parquet" + + if not parquet_path.exists(): + return None + + try: + df = pd.read_parquet(str(parquet_path)) + # Handle MultiIndex (datetime, instrument) + if isinstance(df.index, pd.MultiIndex): + # Get the factor column name (should be the only column) + factor_col = df.columns[0] + # Extract EURUSD series + try: + series = df.xs('EURUSD', level='instrument')[factor_col] + except KeyError: + # Try alternative extraction + df_reset = df.reset_index() + if 'instrument' in df_reset.columns: + df_eur = df_reset[df_reset['instrument'] == 'EURUSD'].set_index('datetime') + series = df_eur[factor_col] if factor_col in df_eur.columns else df_eur.iloc[:, -1] + else: + series = df.iloc[:, 0] + else: + series = df.iloc[:, 0] + + # Ensure numeric + series = pd.to_numeric(series, errors='coerce') + series.name = factor_name + return series + except Exception as e: + logger.warning(f"Failed to load factor values for {factor_name}: {e}") + return None + + def generate_strategy_code(self, factors: List[Dict[str, Any]], strategy_name: str) -> Optional[str]: + """ + Generate strategy code using LLM from factor combinations. + + Parameters + ---------- + factors : List[Dict[str, Any]] + List of factor info dicts to combine + strategy_name : str + Name for the generated strategy + + Returns + ------- + str or None + Generated Python strategy code + """ + factor_names = [f["factor_name"] for f in factors] + factor_ics = {f["factor_name"]: f.get("ic", 0) for f in factors} + + # Build prompt context + context = { + "strategy_name": strategy_name, + "factor_names": factor_names, + "factor_ics": factor_ics, + "trading_style": self.trading_style, + "min_sharpe": self.min_sharpe, + "max_drawdown": self.max_drawdown, + "system_prompt": self.strategy_prompt.get("system", "") if isinstance(self.strategy_prompt, dict) else "", + "user_prompt": self.strategy_prompt.get("user", "").replace("{{ factors }}", str(factor_ics)).replace("{{ additional_context }}", f"Strategy name: {strategy_name}") if isinstance(self.strategy_prompt, dict) else "", + } + + # Try LLM first + if self.strategy_prompt is not None: + try: + code = self._generate_with_llm(context) + if code: + return code + except Exception as e: + logger.warning(f"LLM strategy generation failed: {e}") + + # Fallback: generate template code programmatically + return self._generate_fallback_code(context) + + def _generate_with_llm(self, context: Dict[str, Any]) -> Optional[str]: + """Generate strategy code using LLM.""" + import os + import requests + + # Use local llama.cpp server (running on port 8081) + api_url = "http://localhost:8081/v1" + api_key = "local" + model = "" + + headers = { + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + } + + payload = { + + "model": "Qwen3.5-35B-A3B-Q3_K_M.gguf", + "messages": [ + {"role": "system", "content": context.get("system_prompt", "")}, + {"role": "user", "content": context.get("user_prompt", "")}, + ], + "max_tokens": 4096, + "temperature": 0.5, + "include_reasoning": False, + } + + # Build API URL + api_base = api_url.rstrip("/") + if not api_base.endswith("/v1"): + api_base = f"{api_base}/v1" + api_endpoint = f"{api_base}/chat/completions" + + response = requests.post( + api_endpoint, + headers=headers, + json=payload, + timeout=120, + ) + + if response.status_code != 200: + logger.warning(f"LLM API error: {response.text[:200]}") + return None + + data = response.json() + message = data.get("choices", [{}])[0].get("message", {}) + content = message.get("content", "") or message.get("reasoning_content", "") + + if not content: + # Try fallback: some models put content in different fields + content = data.get("output", "") or data.get("text", "") + if not content: + logger.warning(f"LLM returned empty response. Model: {model}, Full response: {str(data)[:500]}") + return None + + # Debug: log what we got}, first 100 chars: {content[:100]}") + + code = content.strip() + + # Extract code from markdown blocks or reasoning content + import re + from collections import Counter + + # First try to find code between ``` markers + if "```" in code: + match = re.search(r'```python\s*\n(.*?)\n```', code, re.DOTALL) + if match: + code = match.group(1) + else: + match = re.search(r'```\s*\n(.*?)\n```', code, re.DOTALL) + if match: + code = match.group(1) + + # If code has indent from reasoning, dedent it + if code: + # Find code before first ``` if present + match = re.search(r'^(.*?)(?:```)', code, re.DOTALL) + if match: + code = match.group(1).strip() + + # Smart dedent: find most common indent + lines = code.split('\n') + indents = Counter() + for line in lines: + if line.strip(): + indent = len(line) - len(line.lstrip()) + indents[indent] += 1 + + if len(indents) > 1 and indents.get(0, 0) <= 1: + indents.pop(0, None) + + if indents: + common_indent = indents.most_common(1)[0][0] + else: + common_indent = 0 + + dedented = [] + for line in lines: + if len(line) >= common_indent and line[:common_indent].isspace(): + dedented.append(line[common_indent:]) + else: + dedented.append(line.lstrip()) + + code = '\n'.join(dedented).strip() + + # Remove non-code lines (bullets, commentary after code) + final_lines = [] + for line in code.split('\n'): + stripped = line.strip() + if not stripped: + continue + if stripped.startswith('*') or stripped.startswith('\u2022') or stripped.startswith('Wait') or stripped.startswith('Also') or stripped.startswith('One more'): + break + final_lines.append(line) + + code = '\n'.join(final_lines).strip() + + # Remove non-ASCII (emojis etc) + code = code.encode('ascii', 'ignore').decode('ascii').strip() + + + if not code: + logger.warning("LLM returned empty code after cleaning") + return None + + # Try to parse as JSON and extract code field + import json + if code.startswith('{'): + try: + data = json.loads(code) + # Extract code from JSON response + if 'code' in data: + code = data['code'] + logger.info(f"Extracted code from JSON response ({len(code)} chars)") + elif 'strategy_code' in data: + code = data['strategy_code'] + logger.info(f"Extracted strategy_code from JSON response ({len(code)} chars)") + except json.JSONDecodeError: + pass # Not valid JSON, treat as raw code + + # Validate it's valid Python}") + try: + compile(code, "", "exec") + + return code + except SyntaxError as e: + logger.warning(f"LLM generated invalid Python code: {e}") + logger.warning(f"Code was: {code[:500]}") + return None + + system_prompt = """You are an expert quantitative trading developer. +Generate a complete Python trading strategy that: +1. Takes factor values as input +2. Produces trading signals (1=LONG, -1=SHORT, 0=NEUTRAL) +3. Includes proper risk management +4. Uses the provided factors optimally + +The strategy code will be executed with a 'factors' DataFrame available in scope. +Output ONLY valid Python code, no markdown formatting.""" + + user_prompt = f"""Generate a {context['trading_style']} trading strategy named '{context['strategy_name']}'. + +Factors to use (with IC scores): +{json.dumps(context['factor_ics'], indent=2)} + +Requirements: +- The strategy must output a 'signal' variable (1, -1, or 0) +- Use z-score normalization for factor combination +- Include entry/exit logic based on signal thresholds +- Add risk management: position sizing, stop loss awareness +- Target Sharpe ratio > {context['min_sharpe']} +- Maximum drawdown tolerance: {context['max_drawdown']} + +Output the complete strategy code.""" + + code = api.build_messages_and_create_chat_completion( + user_prompt=user_prompt, + system_prompt=system_prompt, + json_mode=False, + ).strip() + + # Remove markdown code blocks if present + code = code.replace("```python\n", "").replace("```", "").strip() + + # Validate it's valid Python + try: + compile(code, "", "exec") + return code + except SyntaxError: + logger.warning("LLM generated invalid Python code") + return None + + def _generate_fallback_code(self, context: Dict[str, Any]) -> str: + """Generate fallback strategy code programmatically.""" + factor_names = context["factor_names"] + style_config = "daytrading" if context["trading_style"] == "daytrading" else "swing" + + # Build factor assignment code + factor_assignments = "\n ".join( + [f'"{name}": factors["{name}"]' for name in factor_names if name != "timestamp"] + ) + + code = f'''""" +{context['strategy_name']} - {style_config.title()} Strategy +Auto-generated by Predix Strategy Orchestrator +Factors: {', '.join(factor_names)} +""" +import numpy as np +import pandas as pd + +# Strategy configuration +STRATEGY_NAME = "{context['strategy_name']}" +TRADING_STYLE = "{style_config}" +FACTOR_NAMES = {json.dumps(factor_names)} + +# Calculate combined signal +factor_data = pd.DataFrame({{ + {factor_assignments} +}}) + +# Normalize factors to z-scores +factor_norm = (factor_data - factor_data.mean()) / factor_data.std() + +# Weighted combination (weight by IC) +weights = np.array([{", ".join([str(abs(context["factor_ics"].get(n, 0.01))) for n in factor_names if n != "timestamp"])}]) +weights = weights / weights.sum() + +combined_signal = (factor_norm * weights).sum(axis=1) + +# Generate trading signals +# Entry: signal crosses above/below threshold +# Exit: signal crosses back toward zero +entry_threshold = 0.5 +exit_threshold = 0.2 + +signal = pd.Series(0, index=combined_signal.index) +signal[combined_signal > entry_threshold] = 1 +signal[combined_signal < -entry_threshold] = -1 +signal[abs(combined_signal) < exit_threshold] = 0 + +# Smooth signals to reduce turnover +signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int) +''' + return code + + def evaluate_strategy( + self, strategy_code: str, strategy_name: str, factors: List[Dict[str, Any]] + ) -> Dict[str, Any]: + """ + Evaluate a strategy by executing its code and calculating metrics. + + Parameters + ---------- + strategy_code : str + Python strategy code to execute + strategy_name : str + Name of the strategy + factors : List[Dict[str, Any]] + List of factor info dicts used by this strategy + + Returns + ------- + Dict[str, Any] + Strategy evaluation metrics + """ + try: + # Load factor values + factor_names = [f["factor_name"] for f in factors if f["factor_name"] != "timestamp"] + factor_values = {} + + for fname in factor_names: + series = self.load_factor_values(fname) + if series is not None: + factor_values[fname] = series + + if not factor_values: + return { + "strategy_name": strategy_name, + "status": "rejected", + "reason": "No factor values available", + "factors_used": factor_names, + } + + # Align factor values with common index + if not factor_values: + df_factors = pd.DataFrame() + else: + # Find common index across all series + common_idx = None + for name, s in factor_values.items(): + if common_idx is None: + common_idx = s.index + else: + common_idx = common_idx.intersection(s.index) + + if common_idx is not None and len(common_idx) > 100: + df_factors = pd.DataFrame({ + name: s.reindex(common_idx) for name, s in factor_values.items() + }).dropna() + else: + df_factors = pd.DataFrame() + + if len(df_factors) < 100: + return { + "strategy_name": strategy_name, + "status": "rejected", + "reason": "Insufficient aligned data", + "factors_used": factor_names, + } + + # Convert all factor columns to numeric + for col in df_factors.columns: + df_factors[col] = pd.to_numeric(df_factors[col], errors='coerce') + df_factors = df_factors.dropna() + + if len(df_factors) < 100: + return { + "strategy_name": strategy_name, + "status": "rejected", + "reason": "Insufficient numeric data after conversion", + "factors_used": factor_names, + } + + # Execute strategy code with factor data + local_vars = {"factors": df_factors} + try: + exec(strategy_code, {"np": np, "pd": pd, "numpy": np}, local_vars) + except Exception as e: + return { + "strategy_name": strategy_name, + "status": "rejected", + "reason": f"Code execution error: {str(e)}", + "factors_used": factor_names, + } + + if "signal" not in local_vars: + return { + "strategy_name": strategy_name, + "status": "rejected", + "reason": "Strategy did not produce 'signal' variable", + "factors_used": factor_names, + } + + signal = local_vars["signal"] + + # Debug: check signal distribution + + # Calculate returns based on signal changes + # Simple P&L simulation: when signal changes from 0 to 1, we go long + # Returns = signal position * small random return proxy + signal_positions = signal.shift(1).fillna(0) + # Use factor mean as return proxy (scaled to realistic returns) + combined_factor = df_factors.mean(axis=1) + # Scale to ~0.01% daily returns (realistic for FX) + return_proxy = combined_factor * 0.0001 + returns = return_proxy * signal_positions + + # Debug returns}, returns std={returns.std()}").sum()}, Total: {len(returns)}") + + if returns.std() == 0: + return { + "strategy_name": strategy_name, + "status": "rejected", + "reason": "Zero return variance", + "factors_used": factor_names, + } + + # Calculate metrics + total_return = float(returns.sum()) + n_periods = len(returns) + ann_factor = np.sqrt(252 * 1440 / 96) # Annualization for 1min data + volatility = float(returns.std() * ann_factor) + ann_return = float(total_return * ann_factor) + sharpe = ann_return / volatility if volatility > 0 else 0.0 + + # Max drawdown + # Handle any NaN/inf in returns + returns = returns.fillna(0).replace([np.inf, -np.inf], 0) + cum_returns = (1 + returns).cumprod() + running_max = cum_returns.expanding().max() + drawdown = (cum_returns - running_max) / running_max.replace(0, np.nan) + drawdown = drawdown.fillna(0).replace([np.inf, -np.inf], 0) + max_dd = float(drawdown.min()) if len(drawdown) > 0 else 0.0 + + # Win rate + signal_changes = signal.diff().fillna(0) + trades = signal_changes[signal_changes != 0] + win_rate = float((trades > 0).sum() / len(trades)) if len(trades) > 0 else 0.0 + + # Information ratio (signal vs combined factor) + benchmark_returns = combined_factor.pct_change().fillna(0) + excess_returns = returns - benchmark_returns + if excess_returns.std() > 0: + ir = float(excess_returns.mean() / excess_returns.std() * ann_factor) + else: + ir = 0.0 + + metrics = { + "strategy_name": strategy_name, + "status": "accepted" if self._check_acceptance(sharpe, max_dd, win_rate) else "rejected", + "sharpe_ratio": round(sharpe, 4), + "annualized_return": round(ann_return, 6), + "max_drawdown": round(max_dd, 6), + "win_rate": round(win_rate, 4), + "volatility": round(volatility, 6), + "information_ratio": round(ir, 4), + "total_return": round(total_return, 6), + "num_periods": n_periods, + "factors_used": factor_names, + "trading_style": self.trading_style, + "generated_at": datetime.now().isoformat(), + } + + if metrics["status"] == "rejected": + metrics["reason"] = self._get_rejection_reason(sharpe, max_dd, win_rate) + + return metrics + + except Exception as e: + logger.error(f"Strategy evaluation failed for {strategy_name}: {e}") + logger.debug(traceback.format_exc()) + return { + "strategy_name": strategy_name, + "status": "rejected", + "reason": f"Evaluation error: {str(e)}", + "factors_used": [], + } + + def _check_acceptance(self, sharpe: float, max_dd: float, win_rate: float) -> bool: + """Check if strategy meets acceptance criteria.""" + return sharpe >= self.min_sharpe and max_dd >= self.max_drawdown and win_rate >= self.min_win_rate + + def _get_rejection_reason(self, sharpe: float, max_dd: float, win_rate: float) -> str: + """Get human-readable rejection reason.""" + reasons = [] + if sharpe < self.min_sharpe: + reasons.append(f"Sharpe {sharpe:.2f} < {self.min_sharpe}") + if max_dd < self.max_drawdown: + reasons.append(f"Max DD {max_dd:.2%} < {self.max_drawdown:.2%}") + if win_rate < self.min_win_rate: + reasons.append(f"Win Rate {win_rate:.2%} < {self.min_win_rate:.2%}") + return "; ".join(reasons) if reasons else "Unknown" + + def _generate_strategy_name(self, factors: List[Dict[str, Any]], idx: int) -> str: + """Generate a strategy name from its factors.""" + # Extract key words from factor names + words = [] + for f in factors: + name = f["factor_name"] + # Split on underscores and camelCase + parts = name.replace("_", " ").split() + for p in parts: + # Extract capitalized words + cap_words = [w for w in p.split() if w[0:1].isupper()] + words.extend(cap_words if cap_words else [p]) + + # Take up to 3 unique words + unique_words = list(dict.fromkeys(words))[:3] + if unique_words: + return f"{''.join(unique_words)}_v{idx}" + return f"Strategy_{idx}" + + def generate_strategies( + self, + count: int = 10, + workers: int = 4, + progress_callback=None, + ) -> List[Dict[str, Any]]: + """ + Generate and evaluate trading strategies. + + Parameters + ---------- + count : int + Number of strategies to generate + workers : int + Number of parallel workers + progress_callback : callable, optional + Callback function(current, total, result) for progress updates + + Returns + ------- + List[Dict[str, Any]] + List of strategy results (accepted and rejected) + """ + # Load factors + factors = self.load_top_factors() + if not factors: + logger.warning("No factors available for strategy generation") + return [] + + logger.info(f"Loaded {len(factors)} top factors for strategy generation") + + results = [] + strategies_generated = 0 + strategies_accepted = 0 + + # Generate strategies using factor combinations + strategy_configs = self._generate_strategy_configs(factors, count) + + # Execute strategies with thread pool + with ThreadPoolExecutor(max_workers=workers) as executor: + futures = {} + + for i, config in enumerate(strategy_configs): + future = executor.submit(self._generate_and_evaluate_single, i, config) + futures[future] = config + + for future in as_completed(futures): + strategies_generated += 1 + try: + result = future.result() + results.append(result) + + if result["status"] == "accepted": + strategies_accepted += 1 + self._save_strategy(result) + logger.info( + f"Strategy ACCEPTED: {result['strategy_name']} | " + f"Sharpe={result['sharpe_ratio']:.2f} | " + f"DD={result['max_drawdown']:.2%}" + ) + else: + logger.debug( + f"Strategy rejected: {result['strategy_name']} - {result.get('reason', 'unknown')}" + ) + + if progress_callback: + progress_callback(strategies_generated, len(strategy_configs), result) + + except Exception as e: + logger.error(f"Strategy generation failed: {e}") + results.append({ + "strategy_name": f"Failed_{strategies_generated}", + "status": "rejected", + "reason": str(e), + }) + + logger.info( + f"Strategy generation complete: {strategies_accepted}/{strategies_generated} accepted " + f"({strategies_accepted/max(strategies_generated,1)*100:.1f}%)" + ) + + return results + + def _generate_strategy_configs(self, factors: List[Dict], count: int) -> List[List[Dict]]: + """ + Generate strategy configurations from factor combinations. + + Creates combinations of 2-4 factors, prioritizing high-IC factors + and diversity across factor categories. + """ + from itertools import combinations + + configs = [] + + # Generate 2-factor combinations + for combo in combinations(factors, 2): + if len(configs) >= count * 2: # Generate extras for rejection buffer + break + configs.append(list(combo)) + + # Generate 3-factor combinations if needed + if len(configs) < count and len(factors) >= 3: + for combo in combinations(factors, 3): + if len(configs) >= count * 2: + break + configs.append(list(combo)) + + # Shuffle to add randomness, then take what we need + np.random.shuffle(configs) + return configs[: count * 2] # Generate extras + + def _generate_and_evaluate_single(self, idx: int, factors: List[Dict]) -> Dict[str, Any]: + """Generate and evaluate a single strategy.""" + strategy_name = self._generate_strategy_name(factors, idx + 1) + + # Generate code + code = self.generate_strategy_code(factors, strategy_name) + if not code: + return { + "strategy_name": strategy_name, + "status": "rejected", + "reason": "Code generation failed", + } + + # Evaluate + result = self.evaluate_strategy(code, strategy_name, factors) + result["code"] = code + + return result + + def _save_strategy(self, result: Dict[str, Any]) -> None: + """Save accepted strategy to JSON file.""" + timestamp = int(time.time()) + safe_name = result["strategy_name"].replace("/", "_").replace(" ", "_")[:60] + filename = f"{timestamp}_{safe_name}.json" + filepath = self.strategies_dir / filename + + # Prepare serializable result + save_data = {k: v for k, v in result.items() if k != "code"} + save_data["code"] = result.get("code", "") + + with open(filepath, "w", encoding="utf-8") as f: + json.dump(save_data, f, indent=2, default=str, ensure_ascii=False) + + logger.info(f"Saved strategy to {filepath}") + + def get_strategy_summary(self, results: List[Dict[str, Any]]) -> Dict[str, Any]: + """ + Generate summary statistics from strategy generation results. + + Parameters + ---------- + results : List[Dict[str, Any]] + List of strategy results + + Returns + ------- + Dict[str, Any] + Summary statistics + """ + if not results: + return {"total": 0, "accepted": 0, "rejected": 0} + + accepted = [r for r in results if r["status"] == "accepted"] + rejected = [r for r in results if r["status"] == "rejected"] + + summary = { + "total": len(results), + "accepted": len(accepted), + "rejected": len(rejected), + "acceptance_rate": len(accepted) / len(results) if results else 0, + } + + if accepted: + sharpe_values = [r.get("sharpe_ratio", 0) for r in accepted if "sharpe_ratio" in r] + dd_values = [r.get("max_drawdown", 0) for r in accepted if "max_drawdown" in r] + wr_values = [r.get("win_rate", 0) for r in accepted if "win_rate" in r] + + summary["best_sharpe"] = max(sharpe_values) if sharpe_values else 0 + summary["avg_sharpe"] = np.mean(sharpe_values) if sharpe_values else 0 + summary["worst_drawdown"] = min(dd_values) if dd_values else 0 + summary["avg_win_rate"] = np.mean(wr_values) if wr_values else 0 + + return summary