From f0ac999dbec752fbae4fd5a13928289383cf29a2 Mon Sep 17 00:00:00 2001 From: TPTBusiness Date: Fri, 8 May 2026 18:32:38 +0200 Subject: [PATCH] Revert "feat: prioritize Kronos foundation model factors in strategy selection" This reverts commit b58fc5622dd08ab81ba890db1896f06f2266fe29. --- .../qlib/local/strategy_orchestrator.py | 1552 ----------------- 1 file changed, 1552 deletions(-) delete mode 100644 rdagent/scenarios/qlib/local/strategy_orchestrator.py diff --git a/rdagent/scenarios/qlib/local/strategy_orchestrator.py b/rdagent/scenarios/qlib/local/strategy_orchestrator.py deleted file mode 100644 index 5b1883fe..00000000 --- a/rdagent/scenarios/qlib/local/strategy_orchestrator.py +++ /dev/null @@ -1,1552 +0,0 @@ -""" -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=0.3, - max_drawdown=-0.30, - ) - 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 - -import numpy as np -import pandas as pd -from rdagent.scenarios.qlib.local.optuna_optimizer import OptunaOptimizer -from rdagent.components.prompt_loader import load_prompt - -# OHLCV data path -OHLCV_PATH = Path(os.getenv( - "PREDIX_OHLCV_PATH", - "/home/nico/Predix/git_ignore_folder/factor_implementation_source_data/intraday_pv.h5", -)) -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.30, - min_win_rate: float = 0.40, - results_dir: str | None = None, - use_optuna: bool = True, - optuna_trials: int = 20, - continuous_optimization: bool = True, - ): - """ - 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 - use_optuna : bool - Enable Optuna hyperparameter optimization - optuna_trials : int - Number of Optuna trials per strategy - continuous_optimization : bool - If True, optimize ALL strategies (including rejected ones) - Optuna can often rescue strategies with bad initial parameters - """ - 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 - self.use_optuna = use_optuna - self.optuna_trials = optuna_trials - self.continuous_optimization = continuous_optimization - - if results_dir is None: - project_root = Path(__file__).parent.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" - - # Cache OHLCV close — loaded once per orchestrator instance - self._ohlcv_close = None - - # Track generated combos to avoid duplicates - self._generated_combos: set[tuple] = set() - # Track rejected strategies for feedback loop - self._rejected_history: list[dict] = [] - - # 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}", - ) - - @property - def ohlcv_close(self) -> pd.Series | None: - """Lazy-load and cache OHLCV close prices.""" - if self._ohlcv_close is None: - self._ohlcv_close = self._load_ohlcv_close() - return self._ohlcv_close - - def _load_ohlcv_close(self) -> pd.Series | None: - """Load OHLCV close prices from HDF5 file (internal, use ohlcv_close property).""" - if not OHLCV_PATH.exists(): - logger.warning(f"OHLCV data not found: {OHLCV_PATH}") - return None - - try: - ohlcv = pd.read_hdf(str(OHLCV_PATH), key="data") - if "$close" in ohlcv.columns: - close = ohlcv["$close"].dropna() - elif "close" in ohlcv.columns: - close = ohlcv["close"].dropna() - else: - close = ohlcv.select_dtypes(include=[np.number]).iloc[:, 0].dropna() - - # Handle MultiIndex - if isinstance(close.index, pd.MultiIndex): - try: - close = close.xs("EURUSD", level="instrument") - except KeyError: - idx = close.index.get_level_values("instrument") == "EURUSD" - close = close[idx] - close.index = close.index.droplevel("instrument") - - return close - except Exception as e: - logger.warning(f"Failed to load OHLCV data: {e}") - return None - - 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") - - # Select diverse factor TYPES, not just top IC - # This ensures we get momentum, volatility, session, volume, etc. - type_keywords = { - "kronos": [], "momentum": [], "trend": [], "volatility": [], - "volume": [], "session": [], "london": [], "range": [], - "vwap": [], "return": [], "ofi": [], "spread": [], - "close": [], "divergence": [], "other": [], - } - - for f in factors_with_files: - name = f.get("factor_name", "").lower() - matched = False - if "kronos" in name: - type_keywords["kronos"].append(f) - matched = True - else: - for kw in type_keywords: - if kw == "kronos": - continue - if kw in name: - type_keywords[kw].append(f) - matched = True - break - if not matched: - type_keywords["other"].append(f) - - # Select best from each type (ensures diversity) - selected = [] - already_names = set() - - # Priority order: kronos first (foundation model), then momentum, divergence, etc. - priority_types = ["kronos", "momentum", "divergence", "volatility", "session", - "london", "range", "vwap", "volume", "ofi", "spread", - "return", "trend", "close", "other"] - - per_type = max(2, self.top_factors // len(priority_types)) - - for kw in priority_types: - for f in sorted(type_keywords[kw], key=lambda x: abs(x.get("ic", 0)), reverse=True): - if f["factor_name"] not in already_names: - selected.append(f) - already_names.add(f["factor_name"]) - if len([s for s in selected if s["factor_name"] in [x["factor_name"] for x in type_keywords[kw]]]) >= per_type: - break - - # Fill remaining with highest IC not yet selected - if len(selected) < self.top_factors: - remaining = [f for f in factors_with_files if f["factor_name"] not in already_names] - remaining.sort(key=lambda x: abs(x.get("ic", 0)), reverse=True) - selected.extend(remaining[:self.top_factors - len(selected)]) - - # Log diversity - type_counts = {} - for f in selected: - name = f.get("factor_name", "").lower() - matched = False - for kw in type_keywords: - if kw in name: - type_counts[kw] = type_counts.get(kw, 0) + 1 - matched = True - break - if not matched: - type_counts["other"] = type_counts.get("other", 0) + 1 - - logger.info(f"Selected {len(selected)} diverse factors: {type_counts}") - - return selected[:self.top_factors] - - def load_factor_values(self, factor_name: str) -> pd.Series | None: - """ - 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 empty DataFrame - if df.empty or len(df.columns) == 0: - logger.warning(f"Empty parquet file: {parquet_path}") - return None - - # 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, - max_retries: int = 3, - ) -> str | None: - """ - 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 - max_retries : int - Maximum number of retry attempts with feedback (default: 3) - - 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 with proper template variable replacement - if isinstance(self.strategy_prompt, dict): - user_prompt_raw = self.strategy_prompt.get("user", "") - # Replace ALL template variables - user_prompt = user_prompt_raw - user_prompt = user_prompt.replace("{{ factors }}", str(factor_ics)) - user_prompt = user_prompt.replace("{{ ic_values }}", str(factor_ics)) # IC values for each factor - user_prompt = user_prompt.replace("{{ additional_context }}", f"Strategy name: {strategy_name}") - user_prompt = user_prompt.replace("{{ trading_style }}", self.trading_style) - user_prompt = user_prompt.replace("{{ min_sharpe }}", str(self.min_sharpe)) - - if "{{" in user_prompt: - unreplaced = [w for w in user_prompt.split() if "{{" in w] - logger.warning( - f"Unreplaced template variables in prompt for '{strategy_name}': {unreplaced}" - ) - user_prompt = user_prompt.replace("{{ max_drawdown }}", str(self.max_drawdown)) - - # === FEEDBACK LOOP: Inject insights from previously rejected strategies === - if self._rejected_history: - recent_rejects = self._rejected_history[-5:] - feedback_lines = ["\n--- Lessons from previous failed strategies ---"] - for r in recent_rejects: - feedback_lines.append(f"- '{r['name']}' rejected: {r['reason']} (Sharpe={r.get('sharpe', 0):.2f})") - feedback_lines.append("Avoid repeating these mistakes. Generate DIFFERENT code.\n") - user_prompt += "\n".join(feedback_lines) - - system_prompt = self.strategy_prompt.get("system", "") - else: - user_prompt = "" - system_prompt = "" - - 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": system_prompt, - "user_prompt": user_prompt, - } - - # Try LLM first with retry logic - if self.strategy_prompt is not None: - last_error = None - for attempt in range(1, max_retries + 1): - try: - feedback = last_error if attempt > 1 else None - code = self._generate_with_llm(context, attempt=attempt, feedback=feedback) - if code: - logger.info(f"LLM generated valid code on attempt {attempt} ({len(code)} chars)") - return code - last_error = f"Attempt {attempt}: LLM returned empty or invalid code" - logger.warning(f"LLM attempt {attempt}/{max_retries} failed: {last_error}") - except Exception as e: - last_error = f"Attempt {attempt}: {e!s}" - logger.warning(f"LLM attempt {attempt}/{max_retries} failed with exception: {e}") - - logger.warning( - f"LLM strategy generation failed after {max_retries} attempts. " - f"Last error: {last_error}", - ) - - # Fallback: generate template code programmatically - logger.info("Using fallback code generation") - return self._generate_fallback_code(context) - - def _generate_with_llm( - self, - context: dict[str, Any], - attempt: int = 1, - feedback: str | None = None, - ) -> str | None: - """ - Generate strategy code using LLM with APIBackend (same as Factor Coder). - - Parameters - ---------- - context : Dict[str, Any] - Prompt context including system and user prompts - attempt : int - Current retry attempt number (1-based) - feedback : str, optional - Feedback message from previous failed attempts - - Returns - ------- - str or None - Validated Python strategy code, or None if invalid - """ - - # Build user message with optional feedback - user_content = context.get("user_prompt", "") - if feedback: - user_content += f"\n\nPREVIOUS ATTEMPT FAILED: {feedback}\n\nPlease output ONLY valid JSON with a 'code' field." - - system_content = context.get("system_prompt", "") - - # Use APIBackend factory function (same as Factor Coder) - from rdagent.oai.llm_utils import APIBackend - - try: - response = APIBackend().build_messages_and_create_chat_completion( - user_prompt=user_content, - system_prompt=system_content, - json_mode=True, - ) - except Exception as e: - logger.warning(f"APIBackend call failed (attempt {attempt}): {e}") - return None - - if not response: - logger.warning(f"APIBackend returned empty response (attempt {attempt})") - return None - - # Log response for debugging - logger.info(f"[DEBUG] APIBackend response (attempt {attempt}): {len(response)} chars, preview: {response[:100]!r}") - - # === STEP 1: Try JSON extraction FIRST === - content = response.strip() - json_data = self._extract_json(content) - - if json_data is not None: - logger.info(f"[DEBUG] JSON extracted successfully, keys: {list(json_data.keys())}") - code = self._extract_code_from_json(json_data) - if code: - if self._validate_python_code(code): - logger.info(f"[DEBUG] Valid Python code extracted ({len(code)} chars)") - return code - logger.warning(f"JSON 'code' field contains invalid Python (attempt {attempt}). Preview: {code[:200]}") - else: - logger.warning(f"JSON parsed but no valid 'code' field found (attempt {attempt}). Keys: {list(json_data.keys())}") - - # === STEP 2: Fallback - Extract Python code block directly (like Factor Coder) === - import re - code_block_match = re.search(r"```python\s*\n(.*?)\n```", content, re.DOTALL) - if code_block_match: - code = code_block_match.group(1).strip() - if code and self._validate_python_code(code): - logger.info(f"[DEBUG] Python code block extracted ({len(code)} chars)") - return code - - logger.warning(f"All extraction methods failed (attempt {attempt}). Response preview: {response[:200]}") - return None - - def _extract_json(self, content: str) -> dict[str, Any] | None: - """ - Extract JSON object from LLM response content. - - Tries multiple strategies: - 1. Direct parse if content starts with { - 2. Find ```json ... ``` blocks - 3. Find ```python ... ``` blocks (Qwen often wraps JSON in python blocks) - 4. Find first { to last } in content - 5. Find ``` ... ``` blocks (any language tag) - - Returns - ------- - Dict or None - Parsed JSON data, or None if no valid JSON found - """ - import json as json_module - import re - - # Strategy 1: Direct parse - stripped = content.strip() - if stripped.startswith("{") and stripped.endswith("}"): - try: - return json_module.loads(stripped) - except json_module.JSONDecodeError: - pass - - # Strategy 2: Find ```json ... ``` blocks - json_block_match = re.search(r"```json\s*\n(.*?)\n```", content, re.DOTALL) - if json_block_match: - try: - return json_module.loads(json_block_match.group(1)) - except json_module.JSONDecodeError: - pass - - # Strategy 3: Find ```python ... ``` blocks (Qwen often puts JSON in python blocks) - python_block_match = re.search(r"```python\s*\n(.*?)\n```", content, re.DOTALL) - if python_block_match: - block = python_block_match.group(1).strip() - if block.startswith("{") and block.endswith("}"): - try: - return json_module.loads(block) - except json_module.JSONDecodeError: - pass - - # Strategy 4: Find first { to last } - first_brace = content.find("{") - last_brace = content.rfind("}") - if first_brace != -1 and last_brace > first_brace: - json_str = content[first_brace : last_brace + 1] - try: - return json_module.loads(json_str) - except json_module.JSONDecodeError: - # Try to fix common JSON issues (trailing commas, unescaped newlines) - try: - # Remove trailing commas before } or ] - json_str_fixed = re.sub(r",\s*([}\]])", r"\1", json_str) - return json_module.loads(json_str_fixed) - except json_module.JSONDecodeError: - pass - - # Strategy 5: Find ``` ... ``` blocks (any language tag) - code_block_match = re.search(r"```\w*\s*\n(.*?)\n```", content, re.DOTALL) - if code_block_match: - block = code_block_match.group(1).strip() - if block.startswith("{") and block.endswith("}"): - try: - return json_module.loads(block) - except json_module.JSONDecodeError: - pass - - return None - - def _extract_code_from_json(self, json_data: dict[str, Any]) -> str | None: - """ - Extract Python code from parsed JSON data. - - Checks multiple possible field names: 'code', 'strategy_code', 'python_code' - Also handles nested JSON where code might have literal \n characters. - - Returns - ------- - str or None - Extracted Python code - """ - # Try known field names in priority order - for key in ("code", "strategy_code", "python_code", "strategy"): - if key in json_data: - code_value = json_data[key] - if isinstance(code_value, str) and code_value.strip(): - # Handle JSON-escaped newlines - code = code_value.replace("\\n", "\n") - return code.strip() - - return None - - def _extract_code_from_raw(self, content: str) -> str | None: - """ - Extract Python code from raw (non-JSON) LLM response. - - Tries: - 1. Extract from ```python ... ``` blocks - 2. Extract from ``` ... ``` blocks - 3. Use entire content as code - - Returns - ------- - str or None - Extracted Python code - """ - import re - from collections import Counter - - code = content.strip() - - # Try to find ```python blocks - python_match = re.search(r"```python\s*\n(.*?)\n```", code, re.DOTALL) - if python_match: - code = python_match.group(1) - else: - # Try generic ``` blocks - block_match = re.search(r"```\s*\n(.*?)\n```", code, re.DOTALL) - if block_match: - code = block_match.group(1) - - if not code or not code.strip(): - return None - - # Remove markdown code fences if still present - code = code.replace("```python", "").replace("```", "").strip() - - # Smart dedent: find most common non-zero indent - lines = code.split("\n") - indents = Counter() - for line in lines: - if line.strip(): # Non-empty line - indent = len(line) - len(line.lstrip()) - indents[indent] += 1 - - # Remove empty lines from consideration - 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 commentary lines - final_lines = [] - for line in code.split("\n"): - stripped = line.strip() - if not stripped: - continue - # Stop at commentary (bullets, conversational text) - commentary_patterns = ["*", "\u2022", "Wait", "Also", "One more", "Note:", "Here", "This strategy"] - if any(stripped.startswith(p) for p in commentary_patterns): - break - final_lines.append(line) - - code = "\n".join(final_lines).strip() - - # Remove non-ASCII characters (emojis, etc.) - code = code.encode("ascii", "ignore").decode("ascii").strip() - - return code or None - - def _validate_python_code(self, code: str) -> bool: - """ - Validate that a string is valid Python code. - - Parameters - ---------- - code : str - Code string to validate - - Returns - ------- - bool - True if code compiles successfully - """ - if not code or len(code) < 10: - logger.debug("Code too short to be valid") - return False - - try: - compile(code, "", "exec") - return True - except SyntaxError as e: - logger.debug(f"Python syntax error: {e}") - return False - - 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"], - ) - ics = context["factor_ics"] - ic_weights = [str(abs(ics.get(n, 0.01))) for n in factor_names if n != "timestamp"] - - lines = [] - lines.append(f'"""{context["strategy_name"]} - {style_config.title()} Strategy') - lines.append(f'Auto-generated by Predix Strategy Orchestrator') - lines.append(f'Factors: {", ".join(factor_names)}') - lines.append(f'"""') - lines.append(f'import numpy as np') - lines.append(f'import pandas as pd') - lines.append(f'') - lines.append(f'STRATEGY_NAME = "{context["strategy_name"]}"') - lines.append(f'TRADING_STYLE = "{style_config}"') - lines.append(f'FACTOR_NAMES = {json.dumps(factor_names)}') - lines.append(f'') - lines.append(f'factor_data = pd.DataFrame({{') - lines.append(f' {factor_assignments}') - lines.append(f'}})') - lines.append(f'') - lines.append(f'factor_norm = (factor_data - factor_data.mean()) / factor_data.std()') - lines.append(f'') - lines.append(f'weights = np.array([{", ".join(ic_weights)}])') - lines.append(f'weights = weights / weights.sum()') - lines.append(f'') - lines.append(f'combined_signal = (factor_norm * weights).sum(axis=1)') - lines.append(f'') - lines.append(f'entry_threshold = 0.5') - lines.append(f'exit_threshold = 0.2') - lines.append(f'') - lines.append(f'signal = pd.Series(0, index=combined_signal.index)') - lines.append(f'signal[combined_signal > entry_threshold] = 1') - lines.append(f'signal[combined_signal < -entry_threshold] = -1') - lines.append(f'signal[abs(combined_signal) < exit_threshold] = 0') - lines.append(f'') - lines.append(f'signal = signal.rolling(window=3, min_periods=1).mean().round().astype(int)') - return "\n".join(lines) - - def _preflight_check(self, code: str) -> str | None: - """Fast pre-backtest validation. Returns error string or None if OK.""" - # 1) Syntax check - try: - compile(code, "", "exec") - except SyntaxError as e: - return f"Syntax error: {e}" - - # 2) Quick sandbox: run with tiny fake data, check signal exists and varies - try: - import numpy as np - import pandas as pd - - fake_data = pd.DataFrame({ - "a": np.random.default_rng(0).normal(0, 1, 200), - "b": np.random.default_rng(1).normal(0, 1, 200), - }) - fake_close = pd.Series(1.0 + np.cumsum(np.random.default_rng(2).normal(0, 0.001, 200))) - local_vars = {"factors": fake_data, "close": fake_close} - exec(code, {"np": np, "pd": pd, "numpy": np}, local_vars) - - if "signal" not in local_vars: - return "No 'signal' variable produced" - signal = local_vars["signal"] - if not isinstance(signal, (pd.Series, np.ndarray)): - return f"signal is {type(signal).__name__}, not Series/ndarray" - if isinstance(signal, pd.Series) and len(signal) < 2: - return "signal has < 2 elements" - # Check signal varies (not constant) - if isinstance(signal, pd.Series): - unique = signal.dropna().nunique() - else: - unique = len(set(signal[~np.isnan(signal)])) if len(signal) > 0 else 0 - if unique <= 1: - return "signal is constant (no variation)" - except SystemExit: - pass # OK, some templates call exit(1) intentionally - except Exception as e: - return f"Sandbox error: {type(e).__name__}: {e}" - - return None # All clear - - def _load_benchmark_strategies(self, n: int = 3) -> list[dict]: - """Load top-N existing strategies as benchmarks.""" - benchmarks = [] - if self.strategies_dir.exists(): - for f in sorted(self.strategies_dir.glob("*.json")): - try: - data = json.loads(f.read_text()) - s = data.get("sharpe_ratio") or data.get("summary", {}).get("sharpe") - if s and s > 0: - benchmarks.append({"name": data.get("strategy_name", f.stem)[:40], "sharpe": s}) - except Exception: - pass - benchmarks.sort(key=lambda x: x["sharpe"], reverse=True) - return benchmarks[:n] - - def evaluate_strategy( - self, strategy_code: str, strategy_name: str, factors: list[dict[str, Any]], - *, override_code: str | None = None, - ) -> dict[str, Any]: - """ - Evaluate a strategy by executing its code and calculating metrics. - - Parameters - ---------- - strategy_code : str - Python strategy code to execute (original) - strategy_name : str - Name of the strategy - factors : List[Dict[str, Any]] - List of factor info dicts used by this strategy - override_code : str, optional - If provided, use this code instead of strategy_code (for Optuna patching) - """ - code_to_run = override_code or strategy_code - try: - # === PRE-FLIGHT VALIDATION === - preflight_issue = self._preflight_check(strategy_code) - if preflight_issue: - return { - "strategy_name": strategy_name, - "status": "rejected", - "reason": f"Pre-flight failed: {preflight_issue}", - "factors_used": [f["factor_name"] for f in factors], - } - - # 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 - # 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") - - # Forward-fill daily factors to match OHLCV 1-min index - # Many factors are daily (1 value per day), need to ffill to 1-min - # FIX 6: Track ffill ratio for data quality monitoring - close = self.ohlcv_close - if close is not None: - original_len = len(df_factors) - df_factors = df_factors.reindex(close.index).ffill() - - # Log how much was ffill'd - ffill_ratio = 1.0 - (original_len / len(df_factors)) if len(df_factors) > original_len else 0.0 - logger.info( - f"[DEBUG] {strategy_name}: data quality: " - f"original_rows={original_len}, " - f"ffill_rows={len(df_factors) - original_len}, " - f"ffill_ratio={ffill_ratio:.2%}", - ) - - df_factors = df_factors.dropna() - - if len(df_factors) < 1000: - return { - "strategy_name": strategy_name, - "status": "rejected", - "reason": f"Insufficient numeric data after conversion ({len(df_factors)} rows)", - "factors_used": factor_names, - } - - # close is already loaded above for ffill, reuse it - # Reindex close to match factor index - if close is not None: - close = close.reindex(df_factors.index) - - # Execute strategy code with factor data and close prices - local_vars = {"factors": df_factors} - if close is not None: - local_vars["close"] = close - - try: - exec(code_to_run, {"np": np, "pd": pd, "numpy": np}, local_vars) - except Exception as e: - import traceback - logger.error( - f"Strategy code execution failed for '{strategy_name}': {e}\n" - f"{traceback.format_exc()[-2000:]}" - ) - return { - "strategy_name": strategy_name, - "status": "rejected", - "reason": f"Code execution error: {e!s}", - "factors_used": factor_names, - } - - signal = local_vars.get("signal") - if signal is None or (isinstance(signal, pd.Series) and signal.empty): - return { - "strategy_name": strategy_name, - "status": "rejected", - "reason": "Strategy did not produce valid 'signal' variable", - "factors_used": factor_names, - } - - logger.info( - f"[DEBUG] {strategy_name}: signal stats: " - f"len={len(signal)}, " - f"long={int((signal > 0).sum())}, " - f"short={int((signal < 0).sum())}, " - f"flat={int((signal == 0).sum())}, " - f"unique={signal.nunique()}", - ) - - # Delegate all metric computation to the single source of truth. - # Same formulas as every other backtest path in the repo. - from rdagent.components.backtesting.vbt_backtest import ( - DEFAULT_TXN_COST_BPS, - backtest_signal_ftmo, - ) - - # Reuse the already-loaded close from above; create a synthetic proxy if unavailable - if close is None: - logger.warning("OHLCV data unavailable, using factor-mean proxy") - proxy = df_factors.mean(axis=1).astype(float) - synthetic_close = (1 + proxy.pct_change().fillna(0) * 0.0001).cumprod() * 100.0 - close_for_bt = synthetic_close - else: - close_for_bt = close.reindex(signal.index).ffill() - - txn_cost_bps = float(os.getenv("TXN_COST_BPS", DEFAULT_TXN_COST_BPS)) - bt = backtest_signal_ftmo( - close=close_for_bt, - signal=signal, - txn_cost_bps=txn_cost_bps, - wf_rolling=True, - mc_n_permutations=200, - ) - - if bt.get("status") != "success": - return { - "strategy_name": strategy_name, - "status": "rejected", - "reason": bt.get("reason", "backtest failed"), - "factors_used": factor_names, - } - - sharpe = bt["sharpe"] - max_dd = bt["max_drawdown"] - win_rate = bt["win_rate"] - num_real_trades = bt["n_trades"] - - # === WALK-FORWARD OOS METRICS === - oos_sharpe = bt.get("wf_oos_sharpe_mean", 0) - oos_consistency = bt.get("wf_oos_consistency", 0) - oos_monthly = bt.get("wf_oos_monthly_return_mean", 0) - wf_n_windows = bt.get("wf_n_windows", 0) - - # === MULTI-TIMEFRAME CHECK === - mtf_result = self._check_multi_timeframe(signal, close_for_bt, strategy_name) - - # === STABILITY CHECK === - stability = self._check_stability(signal, close_for_bt, strategy_name) - - logger.info( - f"[DEBUG] {strategy_name}: bt stats: " - f"sharpe={sharpe:.4f} dd={max_dd:.4f} wr={win_rate:.4f} " - f"trades={num_real_trades} OOS={oos_sharpe:.2f} " - f"MTF={mtf_result['passed']} stability={stability['passed']}", - ) - - accepted = self._check_acceptance(sharpe, max_dd, win_rate, oos_sharpe, stability, mtf_result) - - metrics = { - "strategy_name": strategy_name, - "status": "accepted" if accepted else "rejected", - "sharpe_ratio": round(sharpe, 4), - "annualized_return": round(bt["annualized_return"], 6), - "annual_return_cagr": round(bt["annual_return_cagr"], 6), - "max_drawdown": round(max_dd, 6), - "win_rate": round(win_rate, 4), - "volatility": round(bt["volatility"], 6), - "total_return": round(bt["total_return"], 6), - "num_periods": bt["n_bars"], - "num_real_trades": num_real_trades, - "profit_factor": round(bt["profit_factor"], 4) if np.isfinite(bt["profit_factor"]) else None, - "sortino": round(bt["sortino"], 4), - "calmar": round(bt["calmar"], 4), - "factors_used": factor_names, - "trading_style": self.trading_style, - "generated_at": datetime.now().isoformat(), - # Walk-Forward - "oos_sharpe": round(oos_sharpe, 4), - "oos_consistency": round(oos_consistency, 4), - "oos_monthly_return": round(oos_monthly, 6), - "wf_n_windows": wf_n_windows, - # Multi-Тimeframe - "mtf_passed": mtf_result["passed"], - "mtf_timeframes": mtf_result["timeframes"], - "mtf_sharpes": mtf_result["sharpes"], - # Stability - "stability_passed": stability["passed"], - "stability_worst_sharpe": round(stability["worst_sharpe"], 4), - } - if "data_quality_flag" in bt: - metrics["data_quality_flag"] = bt["data_quality_flag"] - - if metrics["status"] == "rejected": - logger.info( - f"[DEBUG] {strategy_name}: rejection breakdown: " - f"sharpe={sharpe:.4f} (need>={self.min_sharpe}), " - f"dd={max_dd:.4f} (need>={self.max_drawdown}), " - f"wr={win_rate:.4f} (need>={self.min_win_rate}), " - f"OOS={oos_sharpe:.2f}, MTF={mtf_result['passed']}, Stab={stability['passed']}", - ) - metrics["reason"] = self._get_rejection_reason(sharpe, max_dd, win_rate, oos_sharpe, stability, mtf_result) - - 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: {e!s}", - "factors_used": [], - } - - def _check_acceptance(self, sharpe: float, max_dd: float, win_rate: float, - oos_sharpe: float = 0, stability: dict | None = None, - mtf_result: dict | None = None) -> bool: - """Check if strategy meets ALL acceptance criteria.""" - if not (sharpe >= self.min_sharpe and max_dd >= self.max_drawdown and win_rate >= self.min_win_rate): - return False - if oos_sharpe <= 0: - return False - if stability and not stability["passed"]: - return False - if mtf_result and not mtf_result["passed"]: - return False - return True - - def _get_rejection_reason(self, sharpe: float, max_dd: float, win_rate: float, - oos_sharpe: float = 0, stability: dict | None = None, - mtf_result: dict | None = None) -> 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"MaxDD {max_dd:.2%} < {self.max_drawdown:.2%}") - if win_rate < self.min_win_rate: - reasons.append(f"WinRate {win_rate:.1%} < {self.min_win_rate}") - if oos_sharpe <= 0: - reasons.append(f"OOS-Sharpe {oos_sharpe:.2f} <= 0 (overfit)") - if stability and not stability["passed"]: - reasons.append(f"Stability failed (worst 12M={stability['worst_sharpe']:.2f})") - if mtf_result and not mtf_result["passed"]: - reasons.append(f"MTF {mtf_result['passed_frames']}/{len(mtf_result['timeframes'])} timeframes") - return "; ".join(reasons) if reasons else "Unknown" - - def _check_multi_timeframe(self, signal: pd.Series, close: pd.Series, name: str) -> dict: - """Test strategy on multiple timeframes (1min, 5min, 15min, 1h).""" - from rdagent.components.backtesting.vbt_backtest import backtest_signal - timeframes = {"1min": "1min", "5min": "5min", "15min": "15min", "1h": "1h"} - sharpes = {} - passed = 0 - for tf_label, tf_code in timeframes.items(): - try: - if tf_code == "1min": - sig_tf, close_tf = signal, close - else: - sig_tf = signal.resample(tf_code).last().ffill() - close_tf = close.resample(tf_code).last().ffill() - common = sig_tf.index.intersection(close_tf.index) - if len(common) < 100: - sharpes[tf_label] = None - continue - r = backtest_signal(close_tf.loc[common], sig_tf.loc[common], txn_cost_bps=2.14) - sharpes[tf_label] = round(r["sharpe"], 3) - if r["sharpe"] > 0: - passed += 1 - except Exception: - sharpes[tf_label] = None - return {"passed": passed >= 2, "timeframes": list(sharpes.keys()), - "sharpes": sharpes, "passed_frames": passed} - - def _check_stability(self, signal: pd.Series, close: pd.Series, name: str) -> dict: - """Rolling 12-month Sharpe — must never go negative.""" - from rdagent.components.backtesting.vbt_backtest import backtest_signal - n = len(close) - bars_per_year = 252 * 1440 - window = bars_per_year - step = bars_per_year // 4 - if n < window * 2: - return {"passed": True, "worst_sharpe": 0, "windows": 0} - worst = float("inf") - windows = 0 - for start in range(0, n - window, step): - try: - r = backtest_signal(close.iloc[start:start+window], signal.iloc[start:start+window], txn_cost_bps=2.14) - if r["status"] == "success": - worst = min(worst, r["sharpe"]) - windows += 1 - except Exception: - pass - return {"passed": worst >= 0 and windows >= 3, "worst_sharpe": round(worst, 4), "windows": windows} - - def build_ensemble(self, strategies: list[dict], close: pd.Series | None = None) -> dict | None: - """Build Sharpe-weighted ensemble from top-N accepted strategies.""" - accepted = [s for s in strategies if s.get("status") == "accepted" and s.get("sharpe_ratio", 0) > 0] - if len(accepted) < 2: - return None - accepted.sort(key=lambda s: s.get("sharpe_ratio", 0), reverse=True) - top5 = accepted[:5] - signals = {} - for s in top5: - sig = self._load_strategy_signal(s) - if sig is not None: - signals[s["strategy_name"]] = sig - if len(signals) < 2: - return None - df_signals = pd.DataFrame(signals).ffill().dropna() - if len(df_signals) < 1000: - return None - weights = np.array([s.get("sharpe_ratio", 1) for s in top5 if s["strategy_name"] in signals]) - weights = weights / weights.sum() - ensemble_signal = np.sign((df_signals * weights).sum(axis=1)) - if close is None: - close = self.ohlcv_close - if close is None: - return None - common = ensemble_signal.index.intersection(close.index) - from rdagent.components.backtesting.vbt_backtest import backtest_signal_ftmo - bt = backtest_signal_ftmo(close=close.loc[common], signal=ensemble_signal.loc[common], - txn_cost_bps=2.14, wf_rolling=True) - return {"status": "success", "strategy_name": "Ensemble_Top5", - "sharpe_ratio": round(bt["sharpe"], 4), "max_drawdown": round(bt["max_drawdown"], 4), - "win_rate": round(bt["win_rate"], 4), "n_trades": bt["n_trades"], - "oos_sharpe": round(bt.get("wf_oos_sharpe_mean", 0), 4), - "members": list(signals.keys())} - - def _load_strategy_signal(self, strategy: dict) -> pd.Series | None: - """Re-execute strategy code to get its signal series.""" - code = strategy.get("code", "") - factor_names = strategy.get("factors_used") or strategy.get("factor_names", []) - if not code or len(factor_names) < 2: - return None - factor_values = {} - for fname in factor_names: - series = self.load_factor_values(fname) - if series is not None: - factor_values[fname] = series - if len(factor_values) < 2: - return None - df = pd.DataFrame(factor_values).sort_index() - if isinstance(df.index, pd.MultiIndex): - df = df.droplevel(-1) - close = self.ohlcv_close - if close is not None: - df = df.reindex(close.index).ffill().dropna() - if len(df) < 1000: - return None - local_vars = {"factors": df} - try: - exec(code, {"np": np, "pd": pd}, local_vars) - except Exception: - return None - signal = local_vars.get("signal") - if signal is None or not isinstance(signal, pd.Series): - return None - return signal.reindex(close.index).ffill().fillna(0) if close is not None else signal - - 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 or [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 = 2, # Reduced from 4 to 2 to avoid LLM server overload - 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: - # Also save rejected strategies for debugging - self._save_strategy(result) - logger.warning( - f"Strategy REJECTED: {result['strategy_name']} - {result.get('reason', 'unknown')} | " - f"Sharpe={result.get('sharpe_ratio', 'N/A')} | " - f"DD={result.get('max_drawdown', 'N/A')}", - ) - - 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) - - # === DUPLICATE DETECTION === - factor_names_tuple = tuple(sorted(f["factor_name"] for f in factors)) - if factor_names_tuple in self._generated_combos: - return { - "strategy_name": strategy_name, - "status": "rejected", - "reason": f"Duplicate factor combination: {factor_names_tuple}", - } - self._generated_combos.add(factor_names_tuple) - - # Generate code - code = self.generate_strategy_code(factors, strategy_name) - if not code: - self._rejected_history.append({"name": strategy_name, "reason": "Code generation failed", "factors": factor_names_tuple}) - return { - "strategy_name": strategy_name, - "status": "rejected", - "reason": "Code generation failed", - } - - # Evaluate - result = self.evaluate_strategy(code, strategy_name, factors) - result["code"] = code - - # === BENCHMARK COMPARISON === - if result.get("status") == "accepted": - benchmarks = self._load_benchmark_strategies(3) - new_sharpe = result.get("sharpe_ratio", 0) - for b in benchmarks: - if new_sharpe > b["sharpe"]: - result["beats_benchmark"] = b["name"] - result["benchmark_delta"] = round(new_sharpe - b["sharpe"], 2) - break - - # Optimize with Optuna if enabled - if self.use_optuna: - initial_status = result.get("status", "rejected") - initial_sharpe = result.get("sharpe_ratio", float("-inf")) - logger.info(f"Running Optuna optimization for {strategy_name} (initial: {initial_status}, Sharpe={initial_sharpe:.4f})...") - optimizer = OptunaOptimizer(n_trials=self.optuna_trials) - - # Prepare factor values for optimization - factor_values = self._prepare_factor_values(factors) - - if factor_values is not None: - optimized = optimizer.optimize_strategy(result, factor_values) - optimized_sharpe = optimized.get("sharpe_ratio", float("-inf")) - optimized_status = optimized.get("status", "rejected") - best_params = optimized.get("best_params", {}) - - # Check if Optuna improved the strategy - if optimized_sharpe > initial_sharpe: - improvement = optimized_sharpe - initial_sharpe - logger.info( - f"Optuna {'RESCUED' if optimized_status == 'accepted' and initial_status == 'rejected' else 'improved'} " - f"{strategy_name}: Sharpe {initial_sharpe:.4f} → {optimized_sharpe:.4f} (+{improvement:.4f})", - ) - - # Re-evaluate with best parameters to get comparable metrics - if best_params: - patched_code = self._patch_strategy_code(code, best_params) - re_eval = self._evaluate_with_patched_code(patched_code, strategy_name, factors) - if re_eval.get("sharpe_ratio", float("-inf")) > initial_sharpe: - result.update(re_eval) - result["code"] = patched_code - result["best_params"] = best_params - # Clear old rejection reason if now accepted - if result.get("status") == "accepted": - result.pop("reason", None) - logger.info( - f"Re-evaluated {strategy_name} with best params: " - f"Sharpe {initial_sharpe:.4f} → {re_eval.get('sharpe_ratio', 0):.4f}", - ) - else: - result.update(optimized) - result["best_params"] = best_params - # Clear old rejection reason if now accepted - if result.get("status") == "accepted": - result.pop("reason", None) - else: - result.update(optimized) - # Clear old rejection reason if now accepted - if result.get("status") == "accepted": - result.pop("reason", None) - else: - logger.debug(f"Optuna did not improve {strategy_name}: {initial_sharpe:.4f} vs {optimized_sharpe:.4f}") - else: - logger.warning(f"No factor values available for Optuna optimization of {strategy_name}") - - # === FEEDBACK LOOP === - if result.get("status") != "accepted": - self._rejected_history.append({ - "name": strategy_name, - "reason": result.get("reason", "unknown"), - "factors": factor_names_tuple, - "sharpe": result.get("sharpe_ratio", 0), - }) - # Keep last 20 rejected for prompt context - if len(self._rejected_history) > 20: - self._rejected_history = self._rejected_history[-20:] - - return result - - def _prepare_factor_values(self, factors: list[dict]) -> pd.DataFrame | None: - """Prepare factor values DataFrame for Optuna optimization.""" - factor_values = {} - for f in factors: - fname = f.get("factor_name", "") - if fname: - series = self.load_factor_values(fname) - if series is not None: - factor_values[fname] = series - - if factor_values: - df = pd.DataFrame(factor_values) - # Forward-fill to OHLCV index - close = self.ohlcv_close - if close is not None: - df = df.reindex(close.index).ffill() - return df.dropna() - return None - - def _patch_strategy_code(self, code: str, params: dict[str, Any]) -> str: - """Patch strategy code with Optuna's best parameters.""" - import re - patched = code - - entry_thresh = params.get("entry_threshold", 0.8) - exit_thresh = params.get("exit_threshold", 0.3) - zscore_window = params.get("zscore_window", 50) - signal_window = params.get("signal_window", 3) - - param_patterns = [ - (r"entry_thresh\s*=\s*[\d.]+", f"entry_thresh = {entry_thresh}"), - (r"exit_thresh\s*=\s*[\d.]+", f"exit_thresh = {exit_thresh}"), - (r"window\s*=\s*\d+", f"window = {zscore_window}"), - (r"signal_window\s*=\s*\d+", f"signal_window = {signal_window}"), - ] - for pattern, replacement in param_patterns: - patched = re.sub(pattern, replacement, patched) - - # Patch .rolling(N) calls for common window sizes - rolling_pattern = r"\.rolling\((\d+)\)" - def replace_rolling(match): - val = int(match.group(1)) - if val in (20, 30, 50, 100, 200): - return f".rolling({zscore_window})" - return match.group(0) - patched = re.sub(rolling_pattern, replace_rolling, patched) - - return patched - - def _evaluate_with_patched_code(self, patched_code: str, strategy_name: str, factors: list[dict]) -> dict[str, Any]: - """Re-evaluate strategy with patched parameters (delegates to evaluate_strategy).""" - return self.evaluate_strategy(patched_code, strategy_name, factors, override_code=patched_code) - - 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