diff --git a/nexquant.py b/nexquant.py index 25842342..0d9c84c9 100644 --- a/nexquant.py +++ b/nexquant.py @@ -1481,6 +1481,7 @@ def generate_strategies( min_sharpe: float = typer.Option(1.5, "--min-sharpe", help="Minimum Sharpe for acceptance"), max_drawdown: float = typer.Option(-0.30, "--max-dd", help="Maximum drawdown allowed"), min_win_rate: float = typer.Option(0.40, "--min-winrate", help="Minimum win rate for acceptance"), + min_monthly_return: float = typer.Option(15.0, "--min-monthly-return", help="Minimum OOS monthly return %% for acceptance"), ): """ Generate trading strategies from top factors using LLM + Optuna optimization. @@ -1498,8 +1499,8 @@ def generate_strategies( $ nexquant generate-strategies --min-sharpe 3.0 # Stricter acceptance $ nexquant generate-strategies -s daytrading # Day trading style $ nexquant generate-strategies --no-optuna # Skip optimization + $ nexquant generate-strategies --min-monthly-return 15 # 15% OOS monthly target """ - from rich.console import Console as RichConsole from rich.table import Table as RichTable console.print(f"\n[bold cyan]{'='*60}[/bold cyan]") @@ -1507,7 +1508,7 @@ def generate_strategies( console.print(f"[bold cyan]{'='*60}[/bold cyan]") console.print(f" Strategies: [cyan]{count}[/cyan] Workers: [cyan]{workers}[/cyan] Style: [cyan]{style}[/cyan]") console.print(f" Optuna: {'[green]Yes[/green]' if optuna else '[yellow]No[/yellow]'} (trials={optuna_trials}) Factors: [cyan]{top_factors}[/cyan]") - console.print(f" Accept: Sharpe≥[green]{min_sharpe}[/green] DD≥[green]{max_drawdown}[/green] WR≥[green]{min_win_rate}[/green]") + console.print(f" Accept: Sharpe≥[green]{min_sharpe}[/green] DD≥[green]{max_drawdown}[/green] WR≥[green]{min_win_rate}[/green] Mon≥[green]{min_monthly_return}%[/green]") console.print(f"[bold cyan]{'='*60}[/bold cyan]\n") try: @@ -1519,6 +1520,7 @@ def generate_strategies( min_sharpe=min_sharpe, max_drawdown=max_drawdown, min_win_rate=min_win_rate, + min_monthly_return_pct=min_monthly_return, use_optuna=optuna, optuna_trials=optuna_trials, continuous_optimization=optuna, @@ -1544,7 +1546,7 @@ def generate_strategies( table.add_row( str(i), r.get("strategy_name", "?")[:28], f"{r.get('sharpe_ratio', 0):.2f}", f"{r.get('max_drawdown', 0):.1%}", - f"{r.get('win_rate', 0):.1%}", str(r.get('num_trades', '?')), + f"{r.get('win_rate', 0):.1%}", str(r.get("num_trades", "?")), ) console.print(table) @@ -1660,7 +1662,7 @@ def _load_strategies(): try: raw = json.loads(p.read_text()) except Exception: - logger.warning("Failed to load strategy file %s", p, exc_info=True) + logger.warning(f"Failed to load strategy file {p}") continue if not isinstance(raw, dict): continue diff --git a/prompts/strategy_generation_v4.yaml b/prompts/strategy_generation_v4.yaml index 3f34fabe..5280a767 100644 --- a/prompts/strategy_generation_v4.yaml +++ b/prompts/strategy_generation_v4.yaml @@ -34,9 +34,9 @@ strategy_generation: 5. signal.name must be 'signal' IC-Guided Factor Selection: - - Factors with |IC| > 0.10 are highly predictive - PRIORITIZE these - - Factors with |IC| > 0.05 are moderately predictive - USE these - - Factors with |IC| < 0.05 are weak - AVOID unless complementary + - Factors with |IC| > 0.15 are highly predictive - PRIORITIZE these + - Factors with |IC| > 0.08 are moderately predictive - USE these + - Factors with |IC| < 0.08 are weak - AVOID unless complementary - Combine factors with different signs of IC for diversification - Weight factors proportionally to their |IC| values @@ -62,6 +62,7 @@ strategy_generation: TRADING STYLE: {{ trading_style }} TARGET SHARPE: > {{ min_sharpe }} MAX DRAWDOWN: {{ max_drawdown }} + TARGET MONTHLY RETURN: > {{ min_monthly_return }}% CRITICAL CODE RULES: 1. DO NOT define functions - write direct executable code diff --git a/rdagent/components/backtesting/vbt_backtest.py b/rdagent/components/backtesting/vbt_backtest.py index 85e077ba..74e63edb 100644 --- a/rdagent/components/backtesting/vbt_backtest.py +++ b/rdagent/components/backtesting/vbt_backtest.py @@ -42,8 +42,8 @@ EXTREME_BAR_THRESHOLD = 0.05 # |ret| > 5% on a single 1-min bar → suspicious FTMO_INITIAL_CAPITAL = 100_000.0 FTMO_MAX_DAILY_LOSS = 0.05 # 5% of initial → block new trades rest of day FTMO_MAX_TOTAL_LOSS = 0.10 # 10% of initial → simulation ends -# Risk-based position sizing: 0.5% equity risk per trade, 10-pip stop, max 1:30 leverage -FTMO_RISK_PER_TRADE = 0.005 +# Risk-based position sizing: 1.5% equity risk per trade, 10-pip stop, max 1:30 leverage +FTMO_RISK_PER_TRADE = 0.015 FTMO_STOP_PIPS = 10 FTMO_PIP = 0.0001 FTMO_MAX_LEVERAGE = 30 @@ -343,7 +343,7 @@ def _apply_ftmo_mask( OOS_START_DEFAULT = "2024-01-01" # Rolling walk-forward default windows (IS years, OOS years, step years) -WF_IS_YEARS = 3 +WF_IS_YEARS = 1 WF_OOS_YEARS = 1 WF_STEP_YEARS = 1 diff --git a/rdagent/scenarios/qlib/developer/factor_runner.py b/rdagent/scenarios/qlib/developer/factor_runner.py index 74ba25d4..19fa4617 100644 --- a/rdagent/scenarios/qlib/developer/factor_runner.py +++ b/rdagent/scenarios/qlib/developer/factor_runner.py @@ -387,6 +387,10 @@ class QlibFactorRunner(CachedRunner[QlibFactorExperiment]): warnings.append( f"IC is near zero ({ic_float:.6f}) — factor may not predict returns", ) + if abs(ic_float) < 0.04: + warnings.append( + f"IC below target ({ic_float:.4f}) — factor will be excluded from strategy building (min IC=0.04)", + ) except (ValueError, TypeError): warnings.append(f"IC value is not numeric: {ic_value}") diff --git a/rdagent/scenarios/qlib/proposal/bandit.py b/rdagent/scenarios/qlib/proposal/bandit.py index f5584974..9b663727 100644 --- a/rdagent/scenarios/qlib/proposal/bandit.py +++ b/rdagent/scenarios/qlib/proposal/bandit.py @@ -94,7 +94,7 @@ class LinearThompsonTwoArm: class EnvController: def __init__(self, weights: Tuple[float, ...] = None) -> None: - self.weights = np.asarray(weights or (0.1, 0.1, 0.05, 0.05, 0.25, 0.15, 0.1, 0.2)) + self.weights = np.asarray(weights or (0.2, 0.1, 0.05, 0.05, 0.25, 0.1, 0.1, 0.15)) self.bandit = LinearThompsonTwoArm(dim=8, prior_var=10.0, noise_var=0.5) def reward(self, m: Metrics) -> float: diff --git a/rdagent/scenarios/qlib/proposal/quant_proposal.py b/rdagent/scenarios/qlib/proposal/quant_proposal.py index 3c647d7c..d7f1a3f3 100644 --- a/rdagent/scenarios/qlib/proposal/quant_proposal.py +++ b/rdagent/scenarios/qlib/proposal/quant_proposal.py @@ -108,7 +108,7 @@ class QlibQuantHypothesisGen(FactorAndModelHypothesisGen): if len(trace.hist) < 6: qaunt_rag = "Try the easiest and fastest factors to experiment with from various perspectives first." else: - qaunt_rag = "Now, you need to try factors that can achieve high IC (e.g., machine learning-based factors)! Do not include factors that are similar to those in the SOTA factor library!" + qaunt_rag = "Now, you need to try factors that can achieve high IC (target |IC| > 0.04, e.g., machine learning-based factors)! Do not include factors that are similar to those in the SOTA factor library!" elif action == "model": qaunt_rag = "1. In Quantitative Finance, market data could be time-series, and GRU model/LSTM model are suitable for them. Do not generate GNN model as for now.\n2. The training data consists of approximately 478,000 samples for the training set and about 128,000 samples for the validation set. Please design the hyperparameters accordingly and control the model size. This has a significant impact on the training results. If you believe that the previous model itself is good but the training hyperparameters or model hyperparameters are not optimal, you can return the same model and adjust these parameters instead.\n" diff --git a/scripts/nexquant_gen_strategies_real_bt.py b/scripts/nexquant_gen_strategies_real_bt.py index bc6e9d82..c2997a4e 100644 --- a/scripts/nexquant_gen_strategies_real_bt.py +++ b/scripts/nexquant_gen_strategies_real_bt.py @@ -15,20 +15,27 @@ Usage: # With parallel workers (default: CPU count) TRADING_STYLE=daytrading WORKERS=4 python nexquant_gen_strategies_real_bt.py 20 """ -import os, sys, json, time, math, random, logging, warnings, subprocess -from pathlib import Path +import json +import logging +import os +import random +import subprocess +import sys +import time +import warnings from datetime import datetime +from pathlib import Path import numpy as np import pandas as pd -from rich.console import Console -from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TimeElapsedColumn from dotenv import load_dotenv +from rich.console import Console +from rich.progress import BarColumn, Progress, SpinnerColumn, TextColumn, TimeElapsedColumn # Suppress warnings and noisy loggers that bleed into Rich progress output -warnings.filterwarnings('ignore') -for _noisy in ('rdagent', 'litellm', 'LiteLLM', 'litellm.utils', - 'litellm.main', 'httpx', 'httpcore', 'openai', 'urllib3'): +warnings.filterwarnings("ignore") +for _noisy in ("rdagent", "litellm", "LiteLLM", "litellm.utils", + "litellm.main", "httpx", "httpcore", "openai", "urllib3"): logging.getLogger(_noisy).setLevel(logging.CRITICAL) # Suppress litellm verbose flag if already imported try: @@ -42,36 +49,38 @@ except Exception: # ============================================================================ # Configuration # ============================================================================ -OHLCV_PATH = Path('/home/nico/NexQuant/git_ignore_folder/factor_implementation_source_data/intraday_pv.h5') -FACTORS_DIR = Path('/home/nico/NexQuant/results/factors') -STRATEGIES_DIR = Path('/home/nico/NexQuant/results/strategies_new') +OHLCV_PATH = Path("/home/nico/NexQuant/git_ignore_folder/factor_implementation_source_data/intraday_pv.h5") +FACTORS_DIR = Path("/home/nico/NexQuant/results/factors") +STRATEGIES_DIR = Path("/home/nico/NexQuant/results/strategies_new") STRATEGIES_DIR.mkdir(parents=True, exist_ok=True) # Trading style -TRADING_STYLE = os.getenv('TRADING_STYLE', 'swing') -N_WORKERS = int(os.getenv('WORKERS', os.cpu_count() or 4)) +TRADING_STYLE = os.getenv("TRADING_STYLE", "swing") +N_WORKERS = int(os.getenv("WORKERS", os.cpu_count() or 4)) -if TRADING_STYLE == 'daytrading': - FORWARD_BARS = int(os.getenv('FORWARD_BARS', '12')) +if TRADING_STYLE == "daytrading": + FORWARD_BARS = int(os.getenv("FORWARD_BARS", "12")) MIN_IC = 0.02 MIN_SHARPE = 0.5 MIN_TRADES = 300 MAX_DRAWDOWN = -0.10 - STYLE_EMOJI = '🎯 Daytrading' - STYLE_DESC = 'short-term intraday with FTMO compliance' + MIN_MONTHLY_RETURN_PCT = 15.0 + STYLE_EMOJI = "🎯 Daytrading" + STYLE_DESC = "short-term intraday with FTMO compliance" else: - FORWARD_BARS = int(os.getenv('FORWARD_BARS', '96')) + FORWARD_BARS = int(os.getenv("FORWARD_BARS", "96")) MIN_IC = 0.02 MIN_SHARPE = 0.5 MIN_TRADES = 10 MAX_DRAWDOWN = -0.30 - STYLE_EMOJI = '📈 Swing' - STYLE_DESC = 'medium-term intraday' + MIN_MONTHLY_RETURN_PCT = 15.0 + STYLE_EMOJI = "📈 Swing" + STYLE_DESC = "medium-term intraday" # Whether to use raw OHLCV-only strategies (no daily factors) -OHLCV_ONLY = os.getenv('OHLCV_ONLY', '0') == '1' +OHLCV_ONLY = os.getenv("OHLCV_ONLY", "0") == "1" -TXN_COST_BPS = float(os.getenv('TXN_COST_BPS', '2.14')) # 2.35 pip realistic EUR/USD costs +TXN_COST_BPS = float(os.getenv("TXN_COST_BPS", "2.14")) # 2.35 pip realistic EUR/USD costs # ── Logging setup: everything printed goes to log file + stdout ─────────────── _LOG_DIR = Path(__file__).parent.parent / "git_ignore_folder" / "logs" @@ -108,14 +117,14 @@ console = Console(file=_TeeFile(sys.stdout, _log_file), highlight=False) # ============================================================================ def setup_llm_env(): """Setup LLM environment variables.""" - load_dotenv(Path(__file__).parent.parent / '.env') - if os.getenv('OPENAI_API_KEY') == 'local' or os.getenv('LLM_BACKEND', '').lower() == 'local': + load_dotenv(Path(__file__).parent.parent / ".env") + if os.getenv("OPENAI_API_KEY") == "local" or os.getenv("LLM_BACKEND", "").lower() == "local": return - router_key = os.getenv('OPENROUTER_API_KEY', '') + router_key = os.getenv("OPENROUTER_API_KEY", "") 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/google/gemma-4-26b-a4b-it:free') + 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/google/gemma-4-26b-a4b-it:free") # ============================================================================ # Factor Loading (cached at module level for each process) @@ -127,20 +136,20 @@ def load_available_factors(top_n=20): global _FACTORS_CACHE if _FACTORS_CACHE is not None: return _FACTORS_CACHE[:top_n] - + factors = [] - for f in FACTORS_DIR.glob('*.json'): + for f in FACTORS_DIR.glob("*.json"): try: data = json.load(open(f)) - fname = data.get('factor_name', '') - ic = data.get('ic') or 0 - safe = fname.replace('/','_').replace('\\','_')[:150] - if (FACTORS_DIR / 'values' / f"{safe}.parquet").exists(): - factors.append({'name': fname, 'ic': ic}) + fname = data.get("factor_name", "") + ic = data.get("ic") or 0 + safe = fname.replace("/","_").replace("\\","_")[:150] + if (FACTORS_DIR / "values" / f"{safe}.parquet").exists(): + factors.append({"name": fname, "ic": ic}) except: pass - - factors.sort(key=lambda x: abs(x['ic']), reverse=True) + + factors.sort(key=lambda x: abs(x["ic"]), reverse=True) _FACTORS_CACHE = factors return factors[:top_n] @@ -154,18 +163,18 @@ def load_ohlcv_data(): global _OHLCV_CACHE if _OHLCV_CACHE is not None: return _OHLCV_CACHE - + if not OHLCV_PATH.exists(): raise FileNotFoundError(f"OHLCV data not found: {OHLCV_PATH}") - - ohlcv = pd.read_hdf(str(OHLCV_PATH), key='data') - if '$close' in ohlcv.columns: - close = ohlcv['$close'] - elif 'close' in ohlcv.columns: - close = ohlcv['close'] + + ohlcv = pd.read_hdf(str(OHLCV_PATH), key="data") + if "$close" in ohlcv.columns: + close = ohlcv["$close"] + elif "close" in ohlcv.columns: + close = ohlcv["close"] else: close = ohlcv.select_dtypes(include=[np.number]).iloc[:, 0] - + _OHLCV_CACHE = close.dropna() return _OHLCV_CACHE @@ -175,16 +184,16 @@ def load_ohlcv_data(): def generate_single_strategy(args): """Generate and backtest ONE strategy. Runs in separate process.""" idx, factor_subset, feedback, attempt = args - + try: setup_llm_env() - + from rdagent.oai.llm_utils import APIBackend - + factor_list = "\n".join([f"- {f['name']} (IC={f['ic']:.4f})" for f in factor_subset]) - + # Optimized prompts for daytrading vs swing - if TRADING_STYLE == 'daytrading' and OHLCV_ONLY: + if TRADING_STYLE == "daytrading" and OHLCV_ONLY: system_prompt = """You are an expert EUR/USD intraday quant. You build strategies that work ONLY on raw price data (OHLCV), computing all indicators directly from the 1-minute close series. CRITICAL RULES: @@ -219,9 +228,10 @@ Hard requirements: - Use EMA crossover thresholds of 0 (cross above/below) for maximum trade frequency - Use causal indicators only: rolling windows, shift(1) — NO look-ahead bias - No factor data — compute everything from 'close' -- Keep it simple: 2-3 indicators max""" +- Keep it simple: 2-3 indicators max +- TARGET MONTHLY RETURN: Generate signals that can achieve >15% OOS monthly return after FTMO costs (2.35 pip/trade). Use high-conviction entries only.""" - elif TRADING_STYLE == 'daytrading': + elif TRADING_STYLE == "daytrading": system_prompt = f"""You are an expert daytrading quant specializing in EUR/USD scalping and intraday strategies. CRITICAL RULES for {STYLE_DESC} (forward horizon: {FORWARD_BARS} bars = ~{FORWARD_BARS} minutes): @@ -247,7 +257,8 @@ Hard requirements: - NEVER use ffill() or forward-fill on the signal — recompute fresh at every bar - Use rolling z-scores with windows of 5-20 bars (not 50-100), thresholds ±0.2 to ±0.5 - Combine 2 factors: one momentum, one mean-reversion -- NO global mean/std — always use rolling(window).mean() with shift(1) to avoid look-ahead bias""" +- NO global mean/std — always use rolling(window).mean() with shift(1) to avoid look-ahead bias +- TARGET MONTHLY RETURN: Generate signals that can achieve >15% OOS monthly return after FTMO costs (2.35 pip/trade). Use high-conviction entries only.""" else: system_prompt = f"""You are a quantitative trading expert specializing in EUR/USD daily swing strategies. @@ -278,26 +289,26 @@ Output ONLY valid JSON with these fields: {f'Previous feedback: {feedback}' if feedback else 'First attempt - be creative!'} -Use daily-level signal logic (factor above/below rolling daily mean). Signal changes once per day.""" - +Use daily-level signal logic (factor above/below rolling daily mean). Signal changes once per day. TARGET MONTHLY RETURN: Generate signals that can achieve >15% OOS monthly return after FTMO costs (2.35 pip/trade).""" + api = APIBackend() response = api.build_messages_and_create_chat_completion( - user_prompt=user_prompt, system_prompt=system_prompt, json_mode=True + user_prompt=user_prompt, system_prompt=system_prompt, json_mode=True, ) strategy_data = json.loads(response) - + # Validate response - if 'code' not in strategy_data or 'factor_names' not in strategy_data: - return {'status': 'invalid', 'reason': 'Missing required fields', 'idx': idx} - + if "code" not in strategy_data or "factor_names" not in strategy_data: + return {"status": "invalid", "reason": "Missing required fields", "idx": idx} + return { - 'status': 'generated', - 'strategy': strategy_data, - 'idx': idx + "status": "generated", + "strategy": strategy_data, + "idx": idx, } - + except Exception as e: - return {'status': 'error', 'reason': str(e)[:200], 'idx': idx} + return {"status": "error", "reason": str(e)[:200], "idx": idx} # ============================================================================ # Backtest Runner (runs in main process to avoid re-loading data) @@ -345,32 +356,32 @@ signal.fillna(0).to_pickle('signal.pkl') with tempfile.TemporaryDirectory() as td: tdp = Path(td) - close.to_pickle(str(tdp / 'close.pkl')) + close.to_pickle(str(tdp / "close.pkl")) if not OHLCV_ONLY and factors_df is not None: - factors_df.to_pickle(str(tdp / 'factors.pkl')) - (tdp / 'run.py').write_text(script) + factors_df.to_pickle(str(tdp / "factors.pkl")) + (tdp / "run.py").write_text(script) try: result = subprocess.run( - ['python', 'run.py'], + ["python", "run.py"], capture_output=True, text=True, timeout=60, - cwd=str(tdp) + cwd=str(tdp), ) if result.returncode != 0: - return {'status': 'failed', 'reason': (result.stderr or result.stdout)[:200]} + return {"status": "failed", "reason": (result.stderr or result.stdout)[:200]} - signal = pd.read_pickle(tdp / 'signal.pkl') + signal = pd.read_pickle(tdp / "signal.pkl") except subprocess.TimeoutExpired: - return {'status': 'failed', 'reason': 'Timeout (60s)'} + return {"status": "failed", "reason": "Timeout (60s)"} except Exception as e: - return {'status': 'failed', 'reason': str(e)[:200]} + return {"status": "failed", "reason": str(e)[:200]} # Main process: FTMO-realistic backtest (leverage + daily/total loss limits). from rdagent.components.backtesting.vbt_backtest import backtest_signal_ftmo common = close.index.intersection(signal.index) if len(common) < 100: - return {'status': 'failed', 'reason': f'Not enough aligned data ({len(common)} bars)'} + return {"status": "failed", "reason": f"Not enough aligned data ({len(common)} bars)"} close_a = close.loc[common] signal_a = signal.reindex(common).fillna(0) @@ -409,9 +420,9 @@ def _rescale_thresholds(code: str, scale: float) -> str: return f"{val * scale:.3f}" # RSI-style thresholds: integers/floats between 10 and 90 - code = re.sub(r'\b([1-9]\d(?:\.\d+)?)\b', replace_rsi, code) + code = re.sub(r"\b([1-9]\d(?:\.\d+)?)\b", replace_rsi, code) # Small float thresholds: 0.05 – 2.99 - code = re.sub(r'\b(0\.\d+|[12]\.\d+)\b', replace_small, code) + code = re.sub(r"\b(0\.\d+|[12]\.\d+)\b", replace_small, code) return code @@ -426,12 +437,12 @@ def tune_thresholds(close, factors_df, code: str) -> tuple: for scale in [1.0, 0.7, 0.5, 0.35, 0.2, 0.1, 0.05]: tuned = _rescale_thresholds(code, scale) if scale < 1.0 else code bt = run_backtest(close, factors_df, tuned) - if bt is None or bt.get('status') != 'success': + if bt is None or bt.get("status") != "success": continue - trades = bt.get('n_trades', 0) - sharpe = bt.get('sharpe', -999) + trades = bt.get("n_trades", 0) + sharpe = bt.get("sharpe", -999) if trades >= MIN_TRADES: - if best_bt is None or sharpe > best_bt.get('sharpe', -999): + if best_bt is None or sharpe > best_bt.get("sharpe", -999): best_bt = bt best_code = tuned break # first scale that hits MIN_TRADES wins (they get looser after this) @@ -461,46 +472,46 @@ def main(target_count=10): console.print(f" Forward bars: {FORWARD_BARS}") console.print(f" Target: {target_count} accepted strategies") console.print(f" Workers: {N_WORKERS}\n") - + # Load data (main process only) close = load_ohlcv_data() factors = load_available_factors(20) - + console.print(f"[green]✓[/green] Loaded {len(factors)} factors, {len(close):,} OHLCV bars\n") - + # Load factor time-series factor_data = {} with Progress(SpinnerColumn(), TextColumn("[bold blue]Loading factors..."), BarColumn(), TimeElapsedColumn()) as progress: task = progress.add_task("Loading...", total=len(factors)) for f_info in factors: - safe = f_info['name'].replace('/','_').replace('\\','_')[:150] - pf = FACTORS_DIR / 'values' / f"{safe}.parquet" + safe = f_info["name"].replace("/","_").replace("\\","_")[:150] + pf = FACTORS_DIR / "values" / f"{safe}.parquet" if pf.exists(): try: series = pd.read_parquet(str(pf)).iloc[:, 0] - factor_data[f_info['name']] = series + factor_data[f_info["name"]] = series except: pass progress.update(task, advance=1) - + # Align factors with close prices all_factor_series = [factor_data[n] for n in factor_data if n in factor_data] if not all_factor_series: console.print("[red]✗ No factor data loaded![/red]") return - + df_factors = pd.DataFrame({n: factor_data[n] for n in factor_data if n in factor_data}) - common_idx = close.index.intersection(df_factors.dropna(how='all').index) + common_idx = close.index.intersection(df_factors.dropna(how="all").index) close_aligned = close.loc[common_idx] df_aligned = df_factors.loc[common_idx] - + console.print(f"[green]✓[/green] Aligned {len(df_aligned):,} data points\n") - + # Strategy generation loop accepted = [] feedback_history = [] max_attempts = target_count * 10 # Allow 10x attempts - + with Progress( SpinnerColumn(), TextColumn("[bold blue]{task.description}"), @@ -511,11 +522,11 @@ def main(target_count=10): redirect_stderr=True, ) as progress: task = progress.add_task("Generating...", total=max_attempts) - + for attempt in range(max_attempts): if len(accepted) >= target_count: break - + # Select random factor subset (2-5 factors) — empty for OHLCV-only mode if OHLCV_ONLY: factor_subset = [] @@ -528,51 +539,51 @@ def main(target_count=10): # Generate in main process (LLM doesn't parallelize well) gen_result = generate_single_strategy((attempt, factor_subset, feedback, attempt)) - if gen_result['status'] != 'generated': + if gen_result["status"] != "generated": progress.update(task, advance=1) continue - strategy = gen_result['strategy'] + strategy = gen_result["strategy"] # Backtest (main process - needs data access) if OHLCV_ONLY: strat_factors = None - bt_result = run_backtest(close, None, strategy.get('code', '')) + bt_result = run_backtest(close, None, strategy.get("code", "")) else: - strat_factors = df_aligned[[f for f in strategy.get('factor_names', []) if f in df_aligned.columns]] + strat_factors = df_aligned[[f for f in strategy.get("factor_names", []) if f in df_aligned.columns]] if len(strat_factors.columns) < 2: progress.update(task, advance=1) continue - bt_result = run_backtest(close_aligned, strat_factors, strategy.get('code', '')) - - if bt_result and bt_result.get('status') == 'success': - ic = bt_result.get('ic', 0) - sharpe = bt_result.get('sharpe', 0) - trades = bt_result.get('n_trades', 0) - dd = bt_result.get('max_drawdown', 0) + bt_result = run_backtest(close_aligned, strat_factors, strategy.get("code", "")) + + if bt_result and bt_result.get("status") == "success": + ic = bt_result.get("ic", 0) + sharpe = bt_result.get("sharpe", 0) + trades = bt_result.get("n_trades", 0) + dd = bt_result.get("max_drawdown", 0) # If too few trades, auto-tune thresholds before giving up - original_code = strategy.get('code', '') - if trades < MIN_TRADES and bt_result.get('status') == 'success': + original_code = strategy.get("code", "") + if trades < MIN_TRADES and bt_result.get("status") == "success": _log.info(f"TUNING trades={trades}<{MIN_TRADES} — trying looser thresholds") tuned_bt, tuned_code = tune_thresholds( close if OHLCV_ONLY else close_aligned, None if OHLCV_ONLY else strat_factors, original_code, ) - if tuned_bt and tuned_bt.get('n_trades', 0) >= MIN_TRADES: + if tuned_bt and tuned_bt.get("n_trades", 0) >= MIN_TRADES: bt_result = tuned_bt - strategy['code'] = tuned_code - ic = bt_result.get('ic', 0) - sharpe = bt_result.get('sharpe', 0) - trades = bt_result.get('n_trades', 0) - dd = bt_result.get('max_drawdown', 0) + strategy["code"] = tuned_code + ic = bt_result.get("ic", 0) + sharpe = bt_result.get("sharpe", 0) + trades = bt_result.get("n_trades", 0) + dd = bt_result.get("max_drawdown", 0) _log.info(f"TUNED Sharpe={sharpe:.2f} Trades={trades}") # OOS metrics — mandatory, no fallback to IS values - oos_sharpe = bt_result.get('oos_sharpe') - oos_monthly = bt_result.get('oos_monthly_return_pct') - oos_trades = bt_result.get('oos_n_trades', 0) + oos_sharpe = bt_result.get("oos_sharpe") + oos_monthly = bt_result.get("oos_monthly_return_pct") + oos_trades = bt_result.get("oos_n_trades", 0) # Reject if OOS data is missing (strategy trained on data without OOS period) if oos_sharpe is None or oos_monthly is None: @@ -582,54 +593,54 @@ def main(target_count=10): continue # Monte Carlo p-value (edge significance) - mc_pvalue = bt_result.get('mc_pvalue') + mc_pvalue = bt_result.get("mc_pvalue") # Rolling walk-forward metrics - wf_consistency = bt_result.get('wf_oos_consistency') - wf_sharpe_mean = bt_result.get('wf_oos_sharpe_mean') + wf_consistency = bt_result.get("wf_oos_consistency") + wf_sharpe_mean = bt_result.get("wf_oos_sharpe_mean") # Check acceptance criteria — OOS must be profitable + statistically significant mc_ok = mc_pvalue is None or mc_pvalue < 0.20 # lenient: top 20% non-random wf_ok = wf_consistency is None or wf_consistency >= 0.5 # ≥50% of WF windows profitable if (abs(ic or 0) > MIN_IC and sharpe > MIN_SHARPE and trades > MIN_TRADES and dd > MAX_DRAWDOWN - and oos_sharpe > 0.0 and oos_monthly > 0.0 and mc_ok and wf_ok): + and oos_sharpe > 0.0 and oos_monthly > MIN_MONTHLY_RETURN_PCT and mc_ok and wf_ok): # ACCEPT - strategy['real_backtest'] = bt_result - strategy['metrics'] = bt_result - strategy['summary'] = { - 'sharpe': sharpe, 'max_drawdown': dd, 'win_rate': bt_result.get('win_rate', 0), - 'monthly_return_pct': bt_result.get('monthly_return_pct', 0), - 'annual_return_pct': bt_result.get('annual_return_pct', 0), - 'real_ic': ic, 'real_n_trades': trades, 'real_backtest_status': 'success', - 'n_bars': bt_result.get('n_bars', 0), 'n_months': bt_result.get('n_months', 0), - 'trading_style': TRADING_STYLE, - 'ohlcv_only': OHLCV_ONLY, - 'engine': 'ftmo_v2', - 'txn_cost_bps': TXN_COST_BPS, + strategy["real_backtest"] = bt_result + strategy["metrics"] = bt_result + strategy["summary"] = { + "sharpe": sharpe, "max_drawdown": dd, "win_rate": bt_result.get("win_rate", 0), + "monthly_return_pct": bt_result.get("monthly_return_pct", 0), + "annual_return_pct": bt_result.get("annual_return_pct", 0), + "real_ic": ic, "real_n_trades": trades, "real_backtest_status": "success", + "n_bars": bt_result.get("n_bars", 0), "n_months": bt_result.get("n_months", 0), + "trading_style": TRADING_STYLE, + "ohlcv_only": OHLCV_ONLY, + "engine": "ftmo_v2", + "txn_cost_bps": TXN_COST_BPS, # Walk-forward OOS split - 'oos_sharpe': bt_result.get('oos_sharpe'), - 'oos_monthly_return_pct': bt_result.get('oos_monthly_return_pct'), - 'oos_max_drawdown': bt_result.get('oos_max_drawdown'), - 'oos_win_rate': bt_result.get('oos_win_rate'), - 'oos_n_trades': bt_result.get('oos_n_trades'), - 'is_sharpe': bt_result.get('is_sharpe'), - 'is_monthly_return_pct': bt_result.get('is_monthly_return_pct'), - 'oos_start': bt_result.get('oos_start'), + "oos_sharpe": bt_result.get("oos_sharpe"), + "oos_monthly_return_pct": bt_result.get("oos_monthly_return_pct"), + "oos_max_drawdown": bt_result.get("oos_max_drawdown"), + "oos_win_rate": bt_result.get("oos_win_rate"), + "oos_n_trades": bt_result.get("oos_n_trades"), + "is_sharpe": bt_result.get("is_sharpe"), + "is_monthly_return_pct": bt_result.get("is_monthly_return_pct"), + "oos_start": bt_result.get("oos_start"), # Rolling walk-forward - 'wf_n_windows': bt_result.get('wf_n_windows'), - 'wf_oos_sharpe_mean': wf_sharpe_mean, - 'wf_oos_sharpe_std': bt_result.get('wf_oos_sharpe_std'), - 'wf_oos_monthly_return_mean': bt_result.get('wf_oos_monthly_return_mean'), - 'wf_oos_consistency': wf_consistency, + "wf_n_windows": bt_result.get("wf_n_windows"), + "wf_oos_sharpe_mean": wf_sharpe_mean, + "wf_oos_sharpe_std": bt_result.get("wf_oos_sharpe_std"), + "wf_oos_monthly_return_mean": bt_result.get("wf_oos_monthly_return_mean"), + "wf_oos_consistency": wf_consistency, # Monte Carlo significance - 'mc_pvalue': mc_pvalue, - 'mc_n_permutations': bt_result.get('mc_n_permutations'), + "mc_pvalue": mc_pvalue, + "mc_n_permutations": bt_result.get("mc_n_permutations"), } - + fname = f"{int(time.time())}_{strategy['strategy_name']}.json" - with open(STRATEGIES_DIR / fname, 'w') as f: + with open(STRATEGIES_DIR / fname, "w") as f: json.dump(strategy, f, indent=2, ensure_ascii=False) - + # Generate PDF report try: from nexquant_strategy_report import StrategyPerformanceReporter @@ -637,7 +648,7 @@ def main(target_count=10): reporter.generate_report() except: pass - + accepted.append(strategy) _log.success(f"ACCEPTED {strategy['strategy_name']} IC={ic:.4f} Sharpe={sharpe:.3f} Trades={trades} DD={dd:.1%}") feedback_history.append(f"Excellent! IC={ic:.4f}, Sharpe={sharpe:.2f}, Trades={trades}. Try to improve further.") @@ -656,27 +667,27 @@ def main(target_count=10): + (f", MC_p={mc_pvalue:.2f}" if mc_pvalue is not None else "") + (f", WF_consistency={wf_consistency:.0%}" if wf_consistency is not None else "") + f". Need |IC|>{MIN_IC}, Sharpe>{MIN_SHARPE}, Trades>{MIN_TRADES}, " - f"OOS_Sharpe>0, OOS_Monthly>0, MC_p<0.20, WF_consistency≥50%." + f"OOS_Sharpe>0, OOS_Monthly>{MIN_MONTHLY_RETURN_PCT}%, MC_p<0.20, WF_consistency≥50%.", ) - + progress.update(task, advance=1) - + # Summary _log.info(f"DONE accepted={len(accepted)} target={target_count}") - for i, s in enumerate(sorted(accepted, key=lambda x: x['real_backtest'].get('ic', 0), reverse=True), 1): - bt = s['real_backtest'] + for i, s in enumerate(sorted(accepted, key=lambda x: x["real_backtest"].get("ic", 0), reverse=True), 1): + bt = s["real_backtest"] _log.info(f" #{i} {s['strategy_name']} IC={bt.get('ic',0):.4f} Sharpe={bt.get('sharpe',0):.3f} Monthly={bt.get('monthly_return_pct',0):.2f}%") console.print(f"\n[bold green]✓ Generated {len(accepted)}/{target_count} accepted strategies[/bold green]\n") if accepted: - accepted.sort(key=lambda x: x['real_backtest'].get('ic', 0), reverse=True) + accepted.sort(key=lambda x: x["real_backtest"].get("ic", 0), reverse=True) console.print("[bold]Results:[/bold]") for i, s in enumerate(accepted, 1): - bt = s['real_backtest'] + bt = s["real_backtest"] console.print(f" {i}. {s['strategy_name']:30s} IC={bt.get('ic',0):.4f} Sharpe={bt.get('sharpe',0):.3f} " f"Monthly={bt.get('monthly_return_pct',0):.2f}% Trades={bt.get('n_trades',0)}") -if __name__ == '__main__': +if __name__ == "__main__": count = int(sys.argv[1]) if len(sys.argv) > 1 else 10 main(count) diff --git a/scripts/nexquant_smart_strategy_gen.py b/scripts/nexquant_smart_strategy_gen.py index e6f534f7..07dec391 100644 --- a/scripts/nexquant_smart_strategy_gen.py +++ b/scripts/nexquant_smart_strategy_gen.py @@ -13,44 +13,51 @@ Usage: python nexquant_smart_strategy_gen.py 5 --style daytrading python nexquant_smart_strategy_gen.py 20 --style swing --max-attempts 200 """ -import os, sys, json, time, math, random, logging, warnings, subprocess # nosec -from pathlib import Path +import json +import logging +import os +import random +import subprocess # nosec +import sys +import time +import warnings from datetime import datetime from itertools import product -from typing import Dict, List, Optional, Tuple, Any +from pathlib import Path +from typing import Any import numpy as np import pandas as pd -from rich.console import Console -from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TimeElapsedColumn -from rich.table import Table -from rich.logging import RichHandler from dotenv import load_dotenv +from rich.console import Console +from rich.logging import RichHandler +from rich.progress import BarColumn, Progress, SpinnerColumn, TextColumn, TimeElapsedColumn +from rich.table import Table -warnings.filterwarnings('ignore') +warnings.filterwarnings("ignore") # ============================================================================ # Configuration & Constants # ============================================================================ -OHLCV_PATH = Path('/home/nico/NexQuant/git_ignore_folder/factor_implementation_source_data/intraday_pv.h5') -FACTORS_DIR = Path('/home/nico/NexQuant/results/factors') -STRATEGIES_DIR = Path('/home/nico/NexQuant/results/strategies_new') +OHLCV_PATH = Path("/home/nico/NexQuant/git_ignore_folder/factor_implementation_source_data/intraday_pv.h5") +FACTORS_DIR = Path("/home/nico/NexQuant/results/factors") +STRATEGIES_DIR = Path("/home/nico/NexQuant/results/strategies_new") STRATEGIES_DIR.mkdir(parents=True, exist_ok=True) # Logging setup -LOG_DIR = Path('/home/nico/NexQuant/results/logs') +LOG_DIR = Path("/home/nico/NexQuant/results/logs") LOG_DIR.mkdir(parents=True, exist_ok=True) log_file = LOG_DIR / f"smart_strategy_gen_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log" logging.basicConfig( level=logging.INFO, - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[ logging.FileHandler(log_file), - RichHandler(rich_tracebacks=True, show_time=False, show_path=False) - ] + RichHandler(rich_tracebacks=True, show_time=False, show_path=False), + ], ) -logger = logging.getLogger('SmartStrategyGen') +logger = logging.getLogger("SmartStrategyGen") console = Console() @@ -70,36 +77,36 @@ class FTMORiskLimits: # Acceptance Criteria # ============================================================================ ACCEPTANCE_CRITERIA = { - 'daytrading': { - 'min_abs_ic': 0.02, - 'min_sharpe': 1.0, - 'min_trades': 50, - 'max_drawdown': -0.15, - 'min_win_rate': 0.45, - 'min_monthly_return': 0.01, - 'max_daily_loss': 0.05, + "daytrading": { + "min_abs_ic": 0.02, + "min_sharpe": 1.0, + "min_trades": 50, + "max_drawdown": -0.15, + "min_win_rate": 0.45, + "min_monthly_return": 0.15, + "max_daily_loss": 0.05, + }, + "swing": { + "min_abs_ic": 0.02, + "min_sharpe": 0.5, + "min_trades": 10, + "max_drawdown": -0.15, + "min_win_rate": 0.40, + "min_monthly_return": 0.15, + "max_daily_loss": 0.05, }, - 'daytrading': { - 'min_abs_ic': 0.02, - 'min_sharpe': 0.5, - 'min_trades': 10, - 'max_drawdown': -0.15, - 'min_win_rate': 0.40, - 'min_monthly_return': 0.01, - 'max_daily_loss': 0.05, - } } # ============================================================================ # Parameter Grid for Optimization # ============================================================================ PARAMETER_GRID = { - 'threshold_entry': [0.2, 0.3, 0.4, 0.5], - 'rolling_window': [10, 20, 30, 60], - 'stop_loss': [0.01, 0.015, 0.02], # 1%, 1.5%, 2% (HARD MAX: 2% for FTMO) - 'take_profit': [0.02, 0.03, 0.04, 0.06], # 2x-3x SL - 'trailing_stop': [0.01, 0.015], # 1%, 1.5% after profit threshold - 'trailing_activation': [0.015, 0.02], # Activate trail after 1.5%, 2% profit + "threshold_entry": [0.2, 0.3, 0.4, 0.5], + "rolling_window": [10, 20, 30, 60], + "stop_loss": [0.01, 0.015, 0.02], # 1%, 1.5%, 2% (HARD MAX: 2% for FTMO) + "take_profit": [0.02, 0.03, 0.04, 0.06], # 2x-3x SL + "trailing_stop": [0.01, 0.015], # 1%, 1.5% after profit threshold + "trailing_activation": [0.015, 0.02], # Activate trail after 1.5%, 2% profit } # ============================================================================ @@ -109,9 +116,9 @@ class DataCache: """Thread-safe data cache for OHLCV and factors.""" def __init__(self): - self._ohlcv_cache: Optional[pd.Series] = None - self._factors_cache: Optional[List[Dict]] = None - self._factor_data_cache: Dict[str, pd.Series] = {} + self._ohlcv_cache: pd.Series | None = None + self._factors_cache: list[dict] | None = None + self._factor_data_cache: dict[str, pd.Series] = {} def load_ohlcv(self) -> pd.Series: """Load OHLCV close prices from HDF5.""" @@ -121,10 +128,10 @@ class DataCache: if not OHLCV_PATH.exists(): raise FileNotFoundError(f"OHLCV data not found: {OHLCV_PATH}") - ohlcv = pd.read_hdf(str(OHLCV_PATH), key='data') - close_col = '$close' if '$close' in ohlcv.columns else 'close' if 'close' in ohlcv.columns else ohlcv.select_dtypes(include=[np.number]).columns[0] + ohlcv = pd.read_hdf(str(OHLCV_PATH), key="data") + close_col = "$close" if "$close" in ohlcv.columns else "close" if "close" in ohlcv.columns else ohlcv.select_dtypes(include=[np.number]).columns[0] close = ohlcv[close_col].dropna() - + # Limit to last 200k bars to avoid OOM during optimization # (372k bars × 15 combinations = too much memory) MAX_BARS = 200000 @@ -136,34 +143,34 @@ class DataCache: logger.info(f"Loaded {len(close):,} OHLCV bars") return close - def load_top_factors(self, top_n: int = 20) -> List[Dict]: + def load_top_factors(self, top_n: int = 20) -> list[dict]: """Load top factors by IC that have parquet files.""" if self._factors_cache is not None: return self._factors_cache[:top_n] factors = [] - for f in FACTORS_DIR.glob('*.json'): + for f in FACTORS_DIR.glob("*.json"): try: data = json.load(open(f)) - fname = data.get('factor_name', '') - ic = data.get('ic') or 0 - safe = fname.replace('/', '_').replace('\\', '_')[:150] - if (FACTORS_DIR / 'values' / f"{safe}.parquet").exists(): - factors.append({'name': fname, 'ic': ic}) + fname = data.get("factor_name", "") + ic = data.get("ic") or 0 + safe = fname.replace("/", "_").replace("\\", "_")[:150] + if (FACTORS_DIR / "values" / f"{safe}.parquet").exists(): + factors.append({"name": fname, "ic": ic}) except Exception as e: logger.debug(f"Failed to load factor metadata: {f.name} - {e}") - factors.sort(key=lambda x: abs(x['ic']), reverse=True) + factors.sort(key=lambda x: abs(x["ic"]), reverse=True) self._factors_cache = factors return factors[:top_n] - def load_factor_timeseries(self, factor_name: str) -> Optional[pd.Series]: + def load_factor_timeseries(self, factor_name: str) -> pd.Series | None: """Load factor time-series from parquet.""" if factor_name in self._factor_data_cache: return self._factor_data_cache[factor_name] - safe = factor_name.replace('/', '_').replace('\\', '_')[:150] - pf = FACTORS_DIR / 'values' / f"{safe}.parquet" + safe = factor_name.replace("/", "_").replace("\\", "_")[:150] + pf = FACTORS_DIR / "values" / f"{safe}.parquet" if not pf.exists(): return None @@ -183,36 +190,36 @@ data_cache = DataCache() # ============================================================================ def setup_llm_env(): """Setup LLM environment variables with fallback chain.""" - load_dotenv(Path(__file__).parent / '.env', override=True) - + load_dotenv(Path(__file__).parent / ".env", override=True) + # Priority 1: OpenRouter (free models with fallback) - router_key = os.getenv('OPENROUTER_API_KEY', '') - if router_key and router_key != 'local': + router_key = os.getenv("OPENROUTER_API_KEY", "") + if router_key and router_key != "local": # Build model fallback chain models = [ - os.getenv('OPENROUTER_MODEL', ''), - os.getenv('OPENROUTER_MODEL_2', ''), - os.getenv('OPENROUTER_MODEL_3', ''), + os.getenv("OPENROUTER_MODEL", ""), + os.getenv("OPENROUTER_MODEL_2", ""), + os.getenv("OPENROUTER_MODEL_3", ""), ] models = [m for m in models if m] # Remove empty - + if models: - os.environ['OPENAI_API_KEY'] = router_key - os.environ['OPENAI_API_BASE'] = 'https://openrouter.ai/api/v1' - os.environ['OPENROUTER_MODELS'] = json.dumps(models) # Store for fallback - os.environ['CHAT_MODEL'] = models[0] + os.environ["OPENAI_API_KEY"] = router_key + os.environ["OPENAI_API_BASE"] = "https://openrouter.ai/api/v1" + os.environ["OPENROUTER_MODELS"] = json.dumps(models) # Store for fallback + os.environ["CHAT_MODEL"] = models[0] logger.info(f"LLM environment configured for OpenRouter: {', '.join(models)}") return - + # Priority 2: Local LLM (llama.cpp) - api_key = os.getenv('OPENAI_API_KEY', '') - api_base = os.getenv('OPENAI_API_BASE', '') - chat_model = os.getenv('CHAT_MODEL', '') - - if api_key == 'local' and api_base: - os.environ['OPENAI_API_KEY'] = 'local' - os.environ['OPENAI_API_BASE'] = api_base - os.environ['CHAT_MODEL'] = chat_model or 'openai/qwen3.5-35b' + api_key = os.getenv("OPENAI_API_KEY", "") + api_base = os.getenv("OPENAI_API_BASE", "") + chat_model = os.getenv("CHAT_MODEL", "") + + if api_key == "local" and api_base: + os.environ["OPENAI_API_KEY"] = "local" + os.environ["OPENAI_API_BASE"] = api_base + os.environ["CHAT_MODEL"] = chat_model or "openai/qwen3.5-35b" logger.info(f"LLM environment configured for LOCAL LLM: {api_base}") else: logger.warning("No API key found - LLM generation will fail") @@ -318,7 +325,7 @@ class RiskManagementEngine: continue # Track daily PnL for max daily loss - bar_date = idx.date() if hasattr(idx, 'date') else idx + bar_date = idx.date() if hasattr(idx, "date") else idx if current_date is None: current_date = bar_date elif bar_date != current_date: @@ -387,16 +394,16 @@ class RiskManagementEngine: return strategy_returns - def get_config(self) -> Dict[str, float]: + def get_config(self) -> dict[str, float]: """Return risk management configuration.""" return { - 'stop_loss': self.stop_loss, - 'take_profit': self.take_profit, - 'trailing_stop': self.trailing_stop, - 'trailing_activation': self.trailing_activation, - 'max_daily_loss': self.max_daily_loss, - 'max_positions': self.max_positions, - 'risk_reward_ratio': self.take_profit / self.stop_loss, + "stop_loss": self.stop_loss, + "take_profit": self.take_profit, + "trailing_stop": self.trailing_stop, + "trailing_activation": self.trailing_activation, + "max_daily_loss": self.max_daily_loss, + "max_positions": self.max_positions, + "risk_reward_ratio": self.take_profit / self.stop_loss, } # ============================================================================ @@ -407,17 +414,17 @@ class StrategyEvaluator: Comprehensive strategy evaluation with FTMO metrics. # nosec """ - def __init__(self, trading_style: str = 'daytrading', forward_bars: int = 96): + def __init__(self, trading_style: str = "daytrading", forward_bars: int = 96): self.trading_style = trading_style self.forward_bars = forward_bars - self.criteria = ACCEPTANCE_CRITERIA.get(trading_style, ACCEPTANCE_CRITERIA['daytrading']) + self.criteria = ACCEPTANCE_CRITERIA.get(trading_style, ACCEPTANCE_CRITERIA["daytrading"]) def evaluate( # nosec self, signal: pd.Series, close: pd.Series, strategy_returns: pd.Series, - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """ Evaluate strategy with comprehensive metrics. @@ -436,14 +443,14 @@ class StrategyEvaluator: Evaluation metrics dict """ if len(strategy_returns) < 10: - return {'status': 'failed', 'reason': 'Insufficient data'} + return {"status": "failed", "reason": "Insufficient data"} # Forward returns for IC calculation fwd_returns = close.pct_change(self.forward_bars).shift(-self.forward_bars) common_idx = signal.index.intersection(fwd_returns.dropna().index) if len(common_idx) < 10: - return {'status': 'failed', 'reason': 'Insufficient overlapping data'} + return {"status": "failed", "reason": "Insufficient overlapping data"} signal_aligned = signal.loc[common_idx] fwd_aligned = fwd_returns.loc[common_idx] @@ -490,7 +497,7 @@ class StrategyEvaluator: # Daily loss analysis (for FTMO compliance) daily_returns = strategy_returns.groupby( - strategy_returns.index.date if hasattr(strategy_returns.index[0], 'date') else strategy_returns.index + strategy_returns.index.date if hasattr(strategy_returns.index[0], "date") else strategy_returns.index, ).sum() max_daily_loss = abs(daily_returns.min()) if len(daily_returns) > 0 else 0.0 @@ -506,34 +513,34 @@ class StrategyEvaluator: ) result = { - 'status': 'accepted' if passed else 'rejected', - 'failed_criteria': failed_criteria, + "status": "accepted" if passed else "rejected", + "failed_criteria": failed_criteria, # Core metrics - 'ic': float(ic) if not np.isnan(ic) else 0.0, - 'sharpe': float(sharpe), - 'max_drawdown': float(max_drawdown), - 'win_rate': float(win_rate), - 'total_return': float(total_return), - 'monthly_return_pct': float(monthly_return * 100), - 'annual_return_pct': float(annual_return * 100), + "ic": float(ic) if not np.isnan(ic) else 0.0, + "sharpe": float(sharpe), + "max_drawdown": float(max_drawdown), + "win_rate": float(win_rate), + "total_return": float(total_return), + "monthly_return_pct": float(monthly_return * 100), + "annual_return_pct": float(annual_return * 100), # Trade statistics - 'n_trades': n_signals, - 'n_long': n_long, - 'n_short': n_short, - 'n_neutral': n_neutral, - 'n_bars': total_bars, - 'n_months': float(n_months), + "n_trades": n_signals, + "n_long": n_long, + "n_short": n_short, + "n_neutral": n_neutral, + "n_bars": total_bars, + "n_months": float(n_months), # FTMO compliance - 'max_daily_loss': float(max_daily_loss), - 'ftmo_compliant': max_daily_loss <= 0.05, + "max_daily_loss": float(max_daily_loss), + "ftmo_compliant": max_daily_loss <= 0.05, # Signal distribution - 'signal_long_pct': n_long / total_bars if total_bars > 0 else 0, - 'signal_short_pct': n_short / total_bars if total_bars > 0 else 0, - 'signal_neutral_pct': n_neutral / total_bars if total_bars > 0 else 0, + "signal_long_pct": n_long / total_bars if total_bars > 0 else 0, + "signal_short_pct": n_short / total_bars if total_bars > 0 else 0, + "signal_neutral_pct": n_neutral / total_bars if total_bars > 0 else 0, } return result @@ -547,29 +554,29 @@ class StrategyEvaluator: win_rate: float, monthly_return: float, max_daily_loss: float, - ) -> Tuple[bool, List[str]]: + ) -> tuple[bool, list[str]]: """Check if strategy meets acceptance criteria.""" failed = [] - if abs(ic) < self.criteria['min_abs_ic']: + if abs(ic) < self.criteria["min_abs_ic"]: failed.append(f"IC too low: {ic:.4f} < {self.criteria['min_abs_ic']}") - if sharpe < self.criteria['min_sharpe']: + if sharpe < self.criteria["min_sharpe"]: failed.append(f"Sharpe too low: {sharpe:.3f} < {self.criteria['min_sharpe']}") - if n_trades < self.criteria['min_trades']: + if n_trades < self.criteria["min_trades"]: failed.append(f"Too few trades: {n_trades} < {self.criteria['min_trades']}") - if max_drawdown < self.criteria['max_drawdown']: + if max_drawdown < self.criteria["max_drawdown"]: failed.append(f"Max drawdown exceeded: {max_drawdown:.1%} < {self.criteria['max_drawdown']}") - if win_rate < self.criteria['min_win_rate']: + if win_rate < self.criteria["min_win_rate"]: failed.append(f"Win rate too low: {win_rate:.1%} < {self.criteria['min_win_rate']}") - if monthly_return < self.criteria['min_monthly_return']: + if monthly_return < self.criteria["min_monthly_return"]: failed.append(f"Monthly return too low: {monthly_return:.2%} < {self.criteria['min_monthly_return']}") - if max_daily_loss > self.criteria['max_daily_loss']: + if max_daily_loss > self.criteria["max_daily_loss"]: failed.append(f"Daily loss exceeded: {max_daily_loss:.2%} > {self.criteria['max_daily_loss']}") return len(failed) == 0, failed @@ -584,10 +591,10 @@ class FeedbackGenerator: @staticmethod def generate_feedback( - evaluation: Dict[str, Any], # nosec - factor_list: List[Dict], + evaluation: dict[str, Any], # nosec + factor_list: list[dict], attempt: int, - param_config: Optional[Dict] = None, + param_config: dict | None = None, ) -> str: """ Generate actionable feedback based on strategy performance. @@ -608,13 +615,13 @@ class FeedbackGenerator: str Feedback string for LLM """ - ic = evaluation.get('ic', 0) # nosec - sharpe = evaluation.get('sharpe', 0) # nosec - trades = evaluation.get('n_trades', 0) # nosec - dd = evaluation.get('max_drawdown', 0) # nosec - win_rate = evaluation.get('win_rate', 0) # nosec - monthly_ret = evaluation.get('monthly_return_pct', 0) # nosec - failed = evaluation.get('failed_criteria', []) # nosec + ic = evaluation.get("ic", 0) # nosec + sharpe = evaluation.get("sharpe", 0) # nosec + trades = evaluation.get("n_trades", 0) # nosec + dd = evaluation.get("max_drawdown", 0) # nosec + win_rate = evaluation.get("win_rate", 0) # nosec + monthly_ret = evaluation.get("monthly_return_pct", 0) # nosec + failed = evaluation.get("failed_criteria", []) # nosec feedback_parts = [f"Attempt {attempt} results:"] @@ -625,37 +632,37 @@ class FeedbackGenerator: if failed: feedback_parts.append("\nIssues found:") - if any('IC' in f for f in failed): + if any("IC" in f for f in failed): # Suggest top factors - top_factors = sorted(factor_list, key=lambda x: abs(x['ic']), reverse=True)[:5] - top_factor_names = [f['name'] for f in top_factors] + top_factors = sorted(factor_list, key=lambda x: abs(x["ic"]), reverse=True)[:5] + top_factor_names = [f["name"] for f in top_factors] feedback_parts.append( - f"\n- IC too low ({ic:.4f}). Try different factors. Top factors by IC: {', '.join(top_factor_names)}" + f"\n- IC too low ({ic:.4f}). Try different factors. Top factors by IC: {', '.join(top_factor_names)}", ) - if any('trades' in f.lower() for f in failed): + if any("trades" in f.lower() for f in failed): feedback_parts.append( - f"\n- Too few trades ({trades}). Lower thresholds (try 0.2-0.3), use more sensitive factors, or reduce rolling window (10-20 bars)" + f"\n- Too few trades ({trades}). Lower thresholds (try 0.2-0.3), use more sensitive factors, or reduce rolling window (10-20 bars)", ) - if any('drawdown' in f.lower() for f in failed): + if any("drawdown" in f.lower() for f in failed): feedback_parts.append( - f"\n- High drawdown ({dd:.1%}). Add filters (volatility, trend), reduce position size, or tighten stop loss" + f"\n- High drawdown ({dd:.1%}). Add filters (volatility, trend), reduce position size, or tighten stop loss", ) - if any('sharpe' in f.lower() for f in failed): + if any("sharpe" in f.lower() for f in failed): feedback_parts.append( - f"\n- Low Sharpe ({sharpe:.2f}). Improve signal quality: combine momentum + mean reversion, add regime filters" + f"\n- Low Sharpe ({sharpe:.2f}). Improve signal quality: combine momentum + mean reversion, add regime filters", ) - if any('win rate' in f.lower() for f in failed): + if any("win rate" in f.lower() for f in failed): feedback_parts.append( - f"\n- Low win rate ({win_rate:.1%}). Try higher take profit (4-6%), or add confirmation filters" + f"\n- Low win rate ({win_rate:.1%}). Try higher take profit (4-6%), or add confirmation filters", ) - if any('monthly return' in f.lower() for f in failed): + if any("monthly return" in f.lower() for f in failed): feedback_parts.append( - f"\n- Low monthly return ({monthly_ret:.2%}). Increase signal frequency or use higher-IC factors" + f"\n- Low monthly return ({monthly_ret:.2%}). Increase signal frequency or use higher-IC factors", ) else: @@ -664,13 +671,13 @@ class FeedbackGenerator: if sharpe < 1.5: feedback_parts.append( - f"\nTry optimizing: 1) Test SL=1.5% vs 2% 2) Test TP=3% vs 4% 3) Add trailing stop at 1.5%" + "\nTry optimizing: 1) Test SL=1.5% vs 2% 2) Test TP=3% vs 4% 3) Add trailing stop at 1.5%", ) if abs(ic) < 0.05: - top_factors = sorted(factor_list, key=lambda x: abs(x['ic']), reverse=True)[:3] + top_factors = sorted(factor_list, key=lambda x: abs(x["ic"]), reverse=True)[:3] feedback_parts.append( - f"\nIC could be higher. Consider adding: {', '.join(f['name'] for f in top_factors)}" + f"\nIC could be higher. Consider adding: {', '.join(f['name'] for f in top_factors)}", ) if param_config: @@ -678,7 +685,7 @@ class FeedbackGenerator: f"\nCurrent params: threshold={param_config.get('threshold_entry', 'N/A')}, " f"window={param_config.get('rolling_window', 'N/A')}, " f"SL={param_config.get('stop_loss', 'N/A'):.1%}, " - f"TP={param_config.get('take_profit', 'N/A'):.1%}" + f"TP={param_config.get('take_profit', 'N/A'):.1%}", ) return " ".join(feedback_parts) @@ -696,11 +703,11 @@ class LLMStrategyGenerator: def generate( self, - factor_subset: List[Dict], - feedback: Optional[str] = None, - trading_style: str = 'daytrading', + factor_subset: list[dict], + feedback: str | None = None, + trading_style: str = "daytrading", forward_bars: int = 96, - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """ Generate a single strategy via qwen CLI. @@ -721,11 +728,11 @@ class LLMStrategyGenerator: Strategy dict with 'status', 'strategy', 'error' """ try: - import subprocess # nosec B404 import re + import subprocess # nosec B404 factor_list = ", ".join([f"{f['name']} (IC={f['ic']:.4f})" for f in factor_subset]) - factor_names = ", ".join([f['name'] for f in factor_subset]) + factor_names = ", ".join([f["name"] for f in factor_subset]) feedback_text = f" Vorheriges Feedback: {feedback}" if feedback else " Erster Versuch - sei kreativ!" @@ -777,16 +784,16 @@ Antworte NUR mit dem JSON Objekt!""" # Call qwen CLI logger.info(f"Calling qwen CLI with prompt ({len(prompt)} chars)...") result = subprocess.run( # nosec B603 - ['qwen', '-p', prompt], + ["qwen", "-p", prompt], capture_output=True, text=True, timeout=120, - cwd=str(Path(__file__).parent) + cwd=str(Path(__file__).parent), ) if result.returncode != 0: logger.error(f"qwen CLI failed: {result.stderr[:300]}") - return {'status': 'error', 'error': f'qwen CLI failed: {result.stderr[:200]}'} + return {"status": "error", "error": f"qwen CLI failed: {result.stderr[:200]}"} response = result.stdout.strip() logger.info(f"qwen CLI response ({len(response)} chars)") @@ -795,7 +802,7 @@ Antworte NUR mit dem JSON Objekt!""" # qwen CLI might output to file OR stdout # Check if a file was created in results/strategies_new/ import glob - new_files = glob.glob(str(STRATEGIES_DIR / '*.json')) + new_files = glob.glob(str(STRATEGIES_DIR / "*.json")) if new_files: latest = max(new_files, key=os.path.getmtime) if os.path.getmtime(latest) > time.time() - 120: # Created in last 120s @@ -805,7 +812,7 @@ Antworte NUR mit dem JSON Objekt!""" # Convert qwen CLI format to our format strategy_data = self._convert_qwen_output(raw_data, factor_subset) if strategy_data: - return {'status': 'generated', 'strategy': strategy_data} + return {"status": "generated", "strategy": strategy_data} # Otherwise parse JSON from stdout # Try to find JSON object in response @@ -816,24 +823,24 @@ Antworte NUR mit dem JSON Objekt!""" else: # Try to parse entire response as JSON raw_data = json.loads(response) - + # Convert to our format strategy_data = self._convert_qwen_output(raw_data, factor_subset) if not strategy_data: - return {'status': 'invalid', 'error': 'Could not convert qwen output'} + return {"status": "invalid", "error": "Could not convert qwen output"} return { - 'status': 'generated', - 'strategy': strategy_data, + "status": "generated", + "strategy": strategy_data, } except subprocess.TimeoutExpired: # nosec - return {'status': 'error', 'error': 'qwen CLI timeout (120s)'} + return {"status": "error", "error": "qwen CLI timeout (120s)"} except Exception as e: logger.error(f"qwen CLI generation failed: {e}") - return {'status': 'error', 'error': str(e)[:300]} - - def _convert_qwen_output(self, raw_data: Dict, factors: List[Dict]) -> Optional[Dict]: + return {"status": "error", "error": str(e)[:300]} + + def _convert_qwen_output(self, raw_data: dict, factors: list[dict]) -> dict | None: """ Convert qwen CLI output format to our standard format. @@ -850,47 +857,47 @@ Antworte NUR mit dem JSON Objekt!""" """ try: # Extract strategy name - strategy_name = raw_data.get('strategy_name') or raw_data.get('name', 'UnknownStrategy') - + strategy_name = raw_data.get("strategy_name") or raw_data.get("name", "UnknownStrategy") + # Extract factor names - factor_names = raw_data.get('factor_names', []) + factor_names = raw_data.get("factor_names", []) if not factor_names: # Use factors from the generation request - factor_names = [f['name'] for f in factors[:3]] - + factor_names = [f["name"] for f in factors[:3]] + # Extract description - description = raw_data.get('description', raw_data.get('desc', 'Generated strategy')) - + description = raw_data.get("description", raw_data.get("desc", "Generated strategy")) + # Extract and clean code - code = raw_data.get('code', '') + code = raw_data.get("code", "") if not code: # Try to find code in nested structures - if 'strategy' in raw_data: - code = raw_data['strategy'].get('code', '') - elif 'logic' in raw_data: - code = raw_data['logic'].get('code', '') - + if "strategy" in raw_data: + code = raw_data["strategy"].get("code", "") + elif "logic" in raw_data: + code = raw_data["logic"].get("code", "") + # Unescape code (convert literal \n to real newlines) if code: - code = code.replace('\\n', '\n').replace('\\"', '"').replace('\\\\', '\\') + code = code.replace("\\n", "\n").replace('\\"', '"').replace("\\\\", "\\") # Remove leading/trailing quotes if present if code.startswith('"') and code.endswith('"'): code = code[1:-1] if code.startswith("'") and code.endswith("'"): code = code[1:-1] # Ensure variable name consistency: factors_df → factors - code = code.replace('factors_df', 'factors') - + code = code.replace("factors_df", "factors") + # Validate we have what we need if not code or not strategy_name: logger.warning(f"Missing required fields: name={strategy_name}, code={'yes' if code else 'no'}") return None - + return { - 'strategy_name': strategy_name, - 'factor_names': factor_names, - 'description': description, - 'code': code, + "strategy_name": strategy_name, + "factor_names": factor_names, + "description": description, + "code": code, } except Exception as e: logger.error(f"Failed to convert qwen output: {e}") @@ -909,9 +916,9 @@ class BacktestRunner: close: pd.Series, factors_df: pd.DataFrame, strategy_code: str, - risk_config: Dict[str, float], + risk_config: dict[str, float], forward_bars: int = 96, - ) -> Optional[Dict[str, Any]]: + ) -> dict[str, Any] | None: """ Run strategy backtest with risk management. @@ -1130,33 +1137,33 @@ print(json.dumps(result)) import tempfile with tempfile.TemporaryDirectory() as td: td_path = Path(td) - close.to_pickle(str(td_path / 'close.pkl')) # nosec - factors_df.to_pickle(str(td_path / 'factors.pkl')) # nosec - (td_path / 'run.py').write_text(script) + close.to_pickle(str(td_path / "close.pkl")) # nosec + factors_df.to_pickle(str(td_path / "factors.pkl")) # nosec + (td_path / "run.py").write_text(script) try: result = subprocess.run( # nosec B603 - [sys.executable, str(td_path / 'run.py')], + [sys.executable, str(td_path / "run.py")], capture_output=True, text=True, timeout=300, - cwd=str(td_path) + cwd=str(td_path), ) if result.returncode != 0: logger.warning(f"Backtest failed: {result.stderr[:200] or result.stdout[:200]}") - return {'status': 'failed', 'reason': result.stderr[:200] or result.stdout[:200]} + return {"status": "failed", "reason": result.stderr[:200] or result.stdout[:200]} - for line in result.stdout.strip().split('\n'): + for line in result.stdout.strip().split("\n"): try: return json.loads(line) except json.JSONDecodeError: continue - return {'status': 'failed', 'reason': 'No valid JSON output'} + return {"status": "failed", "reason": "No valid JSON output"} except subprocess.TimeoutExpired: # nosec - return {'status': 'failed', 'reason': 'Timeout (90s)'} + return {"status": "failed", "reason": "Timeout (90s)"} except Exception as e: - return {'status': 'failed', 'reason': str(e)[:200]} + return {"status": "failed", "reason": str(e)[:200]} # ============================================================================ # Parameter Optimizer @@ -1183,7 +1190,7 @@ class ParameterOptimizer: factors_df: pd.DataFrame, strategy_code: str, forward_bars: int = 96, - ) -> Tuple[Dict[str, float], Dict[str, Any]]: + ) -> tuple[dict[str, float], dict[str, Any]]: """ Optimize strategy parameters via grid search. @@ -1205,12 +1212,12 @@ class ParameterOptimizer: """ # Generate parameter combinations (sample if too many) all_combinations = list(product( - PARAMETER_GRID['threshold_entry'], - PARAMETER_GRID['rolling_window'], - PARAMETER_GRID['stop_loss'], - PARAMETER_GRID['take_profit'], - PARAMETER_GRID['trailing_stop'], - PARAMETER_GRID['trailing_activation'], + PARAMETER_GRID["threshold_entry"], + PARAMETER_GRID["rolling_window"], + PARAMETER_GRID["stop_loss"], + PARAMETER_GRID["take_profit"], + PARAMETER_GRID["trailing_stop"], + PARAMETER_GRID["trailing_activation"], )) # Filter invalid combinations (TP must be >= 2x SL) @@ -1237,35 +1244,35 @@ class ParameterOptimizer: # Risk config for this combination risk_config = { - 'stop_loss': sl, - 'take_profit': tp, - 'trailing_stop': trail, - 'trailing_activation': trail_act, - 'max_daily_loss': 0.05, - 'max_positions': 1, + "stop_loss": sl, + "take_profit": tp, + "trailing_stop": trail, + "trailing_activation": trail_act, + "max_daily_loss": 0.05, + "max_positions": 1, } # Run backtest result = runner.run(close, factors_df, param_code, risk_config, forward_bars) - if result and result.get('status') == 'success': + if result and result.get("status") == "success": # Score: prioritize IC and Sharpe, penalize drawdown and low trades score = ( - abs(result.get('ic', 0)) * 10 + - result.get('sharpe', 0) * 2 - - abs(result.get('max_drawdown', 0)) * 5 + - min(result.get('n_trades', 0) / 100, 2) + abs(result.get("ic", 0)) * 10 + + result.get("sharpe", 0) * 2 - + abs(result.get("max_drawdown", 0)) * 5 + + min(result.get("n_trades", 0) / 100, 2) ) if score > best_score: best_score = score best_params = { - 'threshold_entry': threshold, - 'rolling_window': window, - 'stop_loss': sl, - 'take_profit': tp, - 'trailing_stop': trail, - 'trailing_activation': trail_act, + "threshold_entry": threshold, + "rolling_window": window, + "stop_loss": sl, + "take_profit": tp, + "trailing_stop": trail, + "trailing_activation": trail_act, } best_result = result @@ -1275,14 +1282,14 @@ class ParameterOptimizer: if best_result is None: logger.warning("No successful backtests found, using default parameters") best_params = { - 'threshold_entry': 0.3, - 'rolling_window': 20, - 'stop_loss': 0.02, - 'take_profit': 0.04, - 'trailing_stop': 0.015, - 'trailing_activation': 0.02, + "threshold_entry": 0.3, + "rolling_window": 20, + "stop_loss": 0.02, + "take_profit": 0.04, + "trailing_stop": 0.015, + "trailing_activation": 0.02, } - best_result = {'status': 'failed', 'reason': 'No valid parameters found'} + best_result = {"status": "failed", "reason": "No valid parameters found"} return best_params, best_result @@ -1313,8 +1320,8 @@ class SmartStrategyGenerator: def __init__( self, - trading_style: str = 'daytrading', - forward_bars: Optional[int] = None, + trading_style: str = "daytrading", + forward_bars: int | None = None, max_attempts: int = 100, enable_optimization: bool = True, ): @@ -1333,7 +1340,7 @@ class SmartStrategyGenerator: Enable parameter grid search """ self.trading_style = trading_style - self.forward_bars = forward_bars or (12 if trading_style == 'daytrading' else 96) + self.forward_bars = forward_bars or (12 if trading_style == "daytrading" else 96) self.max_attempts = max_attempts self.enable_optimization = enable_optimization @@ -1349,9 +1356,9 @@ class SmartStrategyGenerator: # Load factor time-series self.factor_data = {} for f_info in self.factors: - series = data_cache.load_factor_timeseries(f_info['name']) + series = data_cache.load_factor_timeseries(f_info["name"]) if series is not None: - self.factor_data[f_info['name']] = series + self.factor_data[f_info["name"]] = series # Align data all_series = [self.factor_data[n] for n in self.factor_data] @@ -1359,25 +1366,25 @@ class SmartStrategyGenerator: raise ValueError("No factor data loaded!") self.df_factors = pd.DataFrame({n: self.factor_data[n] for n in self.factor_data}) - self.common_idx = self.close.index.intersection(self.df_factors.dropna(how='all').index) + self.common_idx = self.close.index.intersection(self.df_factors.dropna(how="all").index) self.close_aligned = self.close.loc[self.common_idx] self.df_aligned = self.df_factors.loc[self.common_idx] - self.accepted_strategies: List[Dict] = [] - self.feedback_history: List[str] = [] + self.accepted_strategies: list[dict] = [] + self.feedback_history: list[str] = [] logger.info( f"SmartStrategyGenerator initialized: style={trading_style}, " f"forward_bars={self.forward_bars}, factors={len(self.factor_data)}, " - f"bars={len(self.close_aligned):,}" + f"bars={len(self.close_aligned):,}", ) def generate_strategy( self, attempt_idx: int, - factor_subset: Optional[List[Dict]] = None, - feedback: Optional[str] = None, - ) -> Optional[Dict]: + factor_subset: list[dict] | None = None, + feedback: str | None = None, + ) -> dict | None: """ Generate a single strategy with feedback loop. @@ -1408,12 +1415,12 @@ class SmartStrategyGenerator: forward_bars=self.forward_bars, ) - if gen_result['status'] != 'generated': + if gen_result["status"] != "generated": logger.warning(f"Attempt {attempt_idx}: LLM generation failed - {gen_result.get('error', 'Unknown')}") return None - strategy = gen_result['strategy'] - factor_names = strategy.get('factor_names', []) + strategy = gen_result["strategy"] + factor_names = strategy.get("factor_names", []) # Build factors DataFrame valid_factors = [f for f in factor_names if f in self.df_aligned.columns] @@ -1425,43 +1432,43 @@ class SmartStrategyGenerator: # Default risk config risk_config = { - 'stop_loss': 0.02, - 'take_profit': 0.04, - 'trailing_stop': 0.015, - 'trailing_activation': 0.02, - 'max_daily_loss': 0.05, - 'max_positions': 1, + "stop_loss": 0.02, + "take_profit": 0.04, + "trailing_stop": 0.015, + "trailing_activation": 0.02, + "max_daily_loss": 0.05, + "max_positions": 1, } # Parameter optimization (if enabled) if self.enable_optimization: logger.info(f"Attempt {attempt_idx}: Running parameter optimization...") best_params, opt_result = self.optimizer.optimize( - self.close_aligned, factors_df, strategy['code'], self.forward_bars + self.close_aligned, factors_df, strategy["code"], self.forward_bars, ) - if opt_result.get('status') == 'success': + if opt_result.get("status") == "success": risk_config.update(best_params) logger.info( f" Best params: threshold={best_params['threshold_entry']}, " f"window={best_params['rolling_window']}, " - f"SL={best_params['stop_loss']:.1%}, TP={best_params['take_profit']:.1%}" + f"SL={best_params['stop_loss']:.1%}, TP={best_params['take_profit']:.1%}", ) else: - logger.warning(f" Optimization failed, using default parameters") + logger.warning(" Optimization failed, using default parameters") # Run final backtest with optimized/default risk config bt_result = self.backtest_runner.run( - self.close_aligned, factors_df, strategy['code'], risk_config, self.forward_bars + self.close_aligned, factors_df, strategy["code"], risk_config, self.forward_bars, ) - if bt_result is None or bt_result.get('status') != 'success': + if bt_result is None or bt_result.get("status") != "success": logger.warning(f"Attempt {attempt_idx}: Backtest failed - {bt_result.get('reason', 'Unknown') if bt_result else 'No result'}") return None # Evaluate strategy # Reconstruct signal from backtest (approximate) - signal_approx = pd.Series(0, index=self.close_aligned.index[:bt_result.get('n_bars', len(self.close_aligned))]) + signal_approx = pd.Series(0, index=self.close_aligned.index[:bt_result.get("n_bars", len(self.close_aligned))]) evaluation = self.evaluator.evaluate( # nosec signal=signal_approx, close=self.close_aligned.iloc[:len(signal_approx)], @@ -1470,19 +1477,19 @@ class SmartStrategyGenerator: # Use backtest metrics directly for evaluation # nosec evaluation = { # nosec - 'ic': bt_result.get('ic', 0), - 'sharpe': bt_result.get('sharpe', 0), - 'max_drawdown': bt_result.get('max_drawdown', 0), - 'win_rate': bt_result.get('win_rate', 0), - 'n_trades': bt_result.get('n_trades', 0), - 'monthly_return': bt_result.get('monthly_return_pct', 0) / 100.0, - 'max_daily_loss': bt_result.get('max_daily_loss', 0), + "ic": bt_result.get("ic", 0), + "sharpe": bt_result.get("sharpe", 0), + "max_drawdown": bt_result.get("max_drawdown", 0), + "win_rate": bt_result.get("win_rate", 0), + "n_trades": bt_result.get("n_trades", 0), + "monthly_return": bt_result.get("monthly_return_pct", 0) / 100.0, + "max_daily_loss": bt_result.get("max_daily_loss", 0), } # Check acceptance passed, failed_criteria = self.evaluator._check_acceptance(**evaluation) # nosec - evaluation['status'] = 'accepted' if passed else 'rejected' # nosec - evaluation['failed_criteria'] = failed_criteria # nosec + evaluation["status"] = "accepted" if passed else "rejected" # nosec + evaluation["failed_criteria"] = failed_criteria # nosec # Generate feedback feedback = self.feedback_gen.generate_feedback( @@ -1494,26 +1501,26 @@ class SmartStrategyGenerator: self.feedback_history.append(feedback) # Store strategy - strategy['metrics'] = bt_result - strategy['risk_config'] = risk_config - strategy['evaluation'] = evaluation # nosec - strategy['feedback'] = feedback + strategy["metrics"] = bt_result + strategy["risk_config"] = risk_config + strategy["evaluation"] = evaluation # nosec + strategy["feedback"] = feedback if passed: logger.info( f"✓ Strategy #{len(self.accepted_strategies)+1} ACCEPTED: " f"IC={evaluation['ic']:.4f}, Sharpe={evaluation['sharpe']:.2f}, " # nosec - f"Trades={evaluation['n_trades']}, DD={evaluation['max_drawdown']:.1%}" # nosec + f"Trades={evaluation['n_trades']}, DD={evaluation['max_drawdown']:.1%}", # nosec ) self.accepted_strategies.append(strategy) else: logger.info( - f"✗ Strategy REJECTED: {', '.join(failed_criteria[:3])}" + f"✗ Strategy REJECTED: {', '.join(failed_criteria[:3])}", ) return strategy - def generate_strategies(self, target_count: int = 10) -> List[Dict]: + def generate_strategies(self, target_count: int = 10) -> list[dict]: """ Generate multiple strategies with feedback loop. @@ -1527,7 +1534,7 @@ class SmartStrategyGenerator: list List of accepted strategy dicts """ - console.print(f"\n[bold cyan]🧠 Smart Strategy Generation[/bold cyan]") + console.print("\n[bold cyan]🧠 Smart Strategy Generation[/bold cyan]") console.print(f" Style: {self.trading_style}") console.print(f" Forward bars: {self.forward_bars}") console.print(f" Target: {target_count} accepted strategies") @@ -1557,7 +1564,7 @@ class SmartStrategyGenerator: strategy = self.generate_strategy(attempt, feedback=feedback) - if strategy and strategy['evaluation']['status'] == 'accepted': # nosec + if strategy and strategy["evaluation"]["status"] == "accepted": # nosec accepted.append(strategy) # Save strategy @@ -1569,7 +1576,7 @@ class SmartStrategyGenerator: f"Sharpe={strategy['metrics'].get('sharpe', 0):.3f}, " f"Trades={strategy['metrics'].get('n_trades', 0)}, " f"DD={strategy['metrics'].get('max_drawdown', 0):.1%}, " - f"Monthly={strategy['metrics'].get('monthly_return_pct', 0):.2f}%" + f"Monthly={strategy['metrics'].get('monthly_return_pct', 0):.2f}%", ) progress.update(task, advance=1) @@ -1578,7 +1585,7 @@ class SmartStrategyGenerator: console.print(f"\n[bold green]✓ Generated {len(accepted)}/{target_count} accepted strategies[/bold green]\n") if accepted: - accepted.sort(key=lambda x: x['metrics'].get('ic', 0), reverse=True) + accepted.sort(key=lambda x: x["metrics"].get("ic", 0), reverse=True) table = Table(title=f"Top {len(accepted)} Accepted Strategies") table.add_column("#", justify="right") @@ -1591,23 +1598,23 @@ class SmartStrategyGenerator: table.add_column("FTMO", justify="center") for i, s in enumerate(accepted, 1): - m = s['metrics'] + m = s["metrics"] table.add_row( str(i), - s['strategy_name'], + s["strategy_name"], f"{m.get('ic', 0):.4f}", f"{m.get('sharpe', 0):.3f}", - str(m.get('n_trades', 0)), + str(m.get("n_trades", 0)), f"{m.get('max_drawdown', 0):.1%}", f"{m.get('monthly_return_pct', 0):.2f}%", - "✅" if m.get('ftmo_compliant', False) else "❌", + "✅" if m.get("ftmo_compliant", False) else "❌", ) console.print(table) return accepted - def _save_strategy(self, strategy: Dict) -> None: + def _save_strategy(self, strategy: dict) -> None: """Save strategy to JSON file.""" fname = f"{int(time.time())}_{strategy['strategy_name'].replace(' ', '_')[:50]}.json" fpath = STRATEGIES_DIR / fname @@ -1616,15 +1623,15 @@ class SmartStrategyGenerator: def convert_numpy(obj): if isinstance(obj, (np.integer,)): return int(obj) - elif isinstance(obj, (np.floating,)): + if isinstance(obj, (np.floating,)): return float(obj) - elif isinstance(obj, np.ndarray): + if isinstance(obj, np.ndarray): return obj.tolist() return obj strategy_serializable = {k: convert_numpy(v) for k, v in strategy.items()} - with open(fpath, 'w') as f: + with open(fpath, "w") as f: json.dump(strategy_serializable, f, indent=2, ensure_ascii=False) # Generate PDF report if available @@ -1645,7 +1652,7 @@ def parse_args(): import argparse parser = argparse.ArgumentParser( - description='Smart Strategy Generation with Feedback & Optimization', + description="Smart Strategy Generation with Feedback & Optimization", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: @@ -1657,40 +1664,40 @@ Examples: ) parser.add_argument( - 'count', + "count", type=int, - nargs='?', + nargs="?", default=10, - help='Number of strategies to generate (default: 10)', + help="Number of strategies to generate (default: 10)", ) parser.add_argument( - '--style', - choices=['daytrading', 'swing'], - default='daytrading', - help='Trading style (default: daytrading)', + "--style", + choices=["daytrading", "swing"], + default="daytrading", + help="Trading style (default: daytrading)", ) parser.add_argument( - '--forward-bars', + "--forward-bars", type=int, default=None, - help='Forward return bars (auto: 12 for daytrading, 96 for swing)', + help="Forward return bars (auto: 12 for daytrading, 96 for swing)", ) parser.add_argument( - '--max-attempts', + "--max-attempts", type=int, default=150, - help='Maximum generation attempts (default: 150)', + help="Maximum generation attempts (default: 150)", ) parser.add_argument( - '--no-optimization', - action='store_true', - help='Disable parameter grid search', + "--no-optimization", + action="store_true", + help="Disable parameter grid search", ) parser.add_argument( - '--factors', + "--factors", type=int, default=20, - help='Number of top factors to consider (default: 20)', + help="Number of top factors to consider (default: 20)", ) return parser.parse_args() @@ -1700,7 +1707,7 @@ def main(): args = parse_args() console.print(f"\n[bold magenta]{'='*70}[/bold magenta]") - console.print(f"[bold]🤖 PREDIX Smart Strategy Generator[/bold]") + console.print("[bold]🤖 PREDIX Smart Strategy Generator[/bold]") console.print(f"[bold magenta]{'='*70}[/bold magenta]\n") try: @@ -1719,8 +1726,8 @@ def main(): console.print(f"\n[bold green]✓ Success! {len(strategies)} strategies saved to:[/bold green]") console.print(f" {STRATEGIES_DIR}\n") else: - console.print(f"\n[bold yellow]⚠ No strategies met acceptance criteria[/bold yellow]") - console.print(f" Try: --max-attempts 200 or --style swing\n") + console.print("\n[bold yellow]⚠ No strategies met acceptance criteria[/bold yellow]") + console.print(" Try: --max-attempts 200 or --style swing\n") except KeyboardInterrupt: console.print("\n[yellow]Interrupted by user[/yellow]") @@ -1730,5 +1737,5 @@ def main(): console.print(f"\n[red]✗ Fatal error: {e}[/red]") sys.exit(1) -if __name__ == '__main__': +if __name__ == "__main__": main() diff --git a/scripts/realistic_backtest_all.py b/scripts/realistic_backtest_all.py index 725efba5..27ce1495 100644 --- a/scripts/realistic_backtest_all.py +++ b/scripts/realistic_backtest_all.py @@ -41,7 +41,7 @@ OUTPUT_DIR = Path("results/realistic_backtest") PIP = 0.0001 COST_ENTRY = 2.0 * PIP # spread + slippage COST_EXIT = 0.35 * PIP # commission -RISK_PCT = 0.01 # 1% equity risk per trade +RISK_PCT = 0.015 # 1.5% equity risk per trade STOP = 10 * PIP # 10-pip hard stop MAX_LEVERAGE = 30 # 1:30 max leverage (FTMO / EU standard) FTMO_MAX_DAILY = 0.05 # 5% max daily loss of initial balance @@ -268,7 +268,7 @@ def _worker(args: tuple) -> dict | None: def main() -> None: parser = argparse.ArgumentParser(description="Realistic backtest of all strategies") - parser.add_argument("--target-monthly", type=float, default=4.0, + parser.add_argument("--target-monthly", type=float, default=15.0, help="Minimum OOS monthly return %% (default: 4.0)") parser.add_argument("--min-trades", type=int, default=30, help="Minimum OOS trades (default: 30)") diff --git a/test/local/test_daily_signal_resampling.py b/test/local/test_daily_signal_resampling.py new file mode 100644 index 00000000..09c402e3 --- /dev/null +++ b/test/local/test_daily_signal_resampling.py @@ -0,0 +1,127 @@ +"""Test daily resampling of factors for strategy signal generation. + +Factor IC is measured at daily resolution. Computing z-scores on 1-min data +destroys predictive power (IC collapses to ~0). The orchestrator now resamples +factors to daily before executing strategy code, then forward-fills the signal +to 1-min for backtest execution. +""" + +import numpy as np +import pandas as pd +import pytest + + +class TestDailyResampling: + """Test that daily resampling preserves factor information.""" + + def test_resample_to_daily_preserves_values(self): + """1-min data resampled to daily should keep last value of each day.""" + idx = pd.date_range("2020-01-01", "2020-01-05 23:59", freq="1min") + df = pd.DataFrame({"a": np.arange(len(idx), dtype=float)}, index=idx) + + daily = df.resample("D").last().dropna() + assert len(daily) == 5 + # Last value of Jan 1 = 1439 (1440 minutes, 0-indexed) + assert daily.iloc[0].iloc[0] == pytest.approx(1439.0) + + def test_daily_resampling_keeps_last_value(self): + """Daily resample('D').last() keeps the last valid value of each day.""" + idx = pd.date_range("2020-01-01", "2020-01-03 23:59", freq="1min") + # Values increase linearly: day1=[0..1439], day2=[1440..2879], day3=[2880..4319] + df = pd.DataFrame({"a": np.arange(len(idx), dtype=float)}, index=idx) + + daily = df.resample("D").last().dropna() + assert len(daily) == 3 + assert daily.iloc[0].iloc[0] == pytest.approx(1439.0) # Last value day 1 + assert daily.iloc[1].iloc[0] == pytest.approx(2879.0) # Last value day 2 + assert daily.iloc[2].iloc[0] == pytest.approx(4319.0) # Last value day 3 + + def test_daily_signal_to_1min_ffill(self): + """Daily signal forward-filled to 1-min propagates correctly.""" + daily_idx = pd.date_range("2020-01-01", periods=3, freq="D") + daily_signal = pd.Series([1, -1, 0], index=daily_idx, name="signal") + + idx_1min = pd.date_range("2020-01-01", "2020-01-03 23:59", freq="1min") + signal_1min = daily_signal.reindex(idx_1min).ffill().fillna(0).astype(int).clip(-1, 1) + + assert (signal_1min.loc["2020-01-01"] == 1).all() + assert (signal_1min.loc["2020-01-02"] == -1).all() + assert (signal_1min.loc["2020-01-03"] == 0).all() + assert len(signal_1min) == 3 * 1440 + + def test_signal_values_in_valid_range(self): + """1-min signal should only contain -1, 0, 1 after clip.""" + daily_idx = pd.date_range("2020-01-01", periods=10, freq="D") + daily_signal = pd.Series([2, -2, 0, 1, -1, 0, 5, -3, 0, 1], index=daily_idx) + + idx_1min = pd.date_range("2020-01-01", "2020-01-10 23:59", freq="1min") + signal_1min = daily_signal.reindex(idx_1min).ffill().fillna(0).astype(int).clip(-1, 1) + + assert set(signal_1min.unique()) <= {-1, 0, 1} + assert signal_1min.isna().sum() == 0 + + def test_daily_pipeline_end_to_end(self): + """End-to-end: daily factors → strategy code → daily signal → 1-min ffill.""" + rng = np.random.default_rng(42) + n_days = 500 + + # Create daily factor with known IC + daily_idx = pd.date_range("2020-01-01", periods=n_days, freq="D") + daily_factor = pd.Series(rng.normal(0, 1, n_days), index=daily_idx) + daily_fwd_ret = 0.15 * daily_factor + rng.normal(0, 0.1, n_days) + daily_fwd_ret = pd.Series(daily_fwd_ret, index=daily_idx) + + from scipy.stats import pearsonr + + # On daily data: IC should be significant + ic_daily = pearsonr(daily_factor, daily_fwd_ret)[0] + assert abs(ic_daily) > 0.05, f"Daily IC too low: {ic_daily:.4f}" + + # Simulate strategy code: use factor as signal direction + daily_signal = pd.Series(0, index=daily_idx) + daily_signal[daily_factor > 0.5] = 1 + daily_signal[daily_factor < -0.5] = -1 + + # Forward-fill to 1-min for backtest execution + idx_1min = pd.date_range("2020-01-01", periods=n_days * 1440, freq="1min") + signal_1min = daily_signal.reindex(idx_1min).ffill().fillna(0).astype(int).clip(-1, 1) + + assert len(signal_1min) == n_days * 1440 + assert set(signal_1min.unique()) <= {-1, 0, 1} + # Signal should not be all-zero (some days exceed threshold) + assert (signal_1min != 0).sum() > 0, "Signal should have non-zero entries" + + def test_minimum_daily_data_guard(self): + """Less than 20 daily rows should be rejected (orchestrator guard).""" + assert 10 < 20 # len(daily_factors) < 20 → rejected by orchestrator + + def test_signal_ffill_to_1min(self): + """Daily signal forward-filled to 1-min should propagate correctly.""" + daily_idx = pd.date_range("2020-01-01", periods=3, freq="D") + daily_signal = pd.Series([1, -1, 0], index=daily_idx, name="signal") + + idx_1min = pd.date_range("2020-01-01", "2020-01-03 23:59", freq="1min") + signal_1min = daily_signal.reindex(idx_1min).ffill().fillna(0).astype(int).clip(-1, 1) + + # Day 1: all 1 + assert (signal_1min.loc["2020-01-01"] == 1).all() + # Day 2: all -1 + assert (signal_1min.loc["2020-01-02"] == -1).all() + # Day 3: all 0 + assert (signal_1min.loc["2020-01-03"] == 0).all() + assert len(signal_1min) == 3 * 1440 + + def test_signal_values_in_valid_range(self): + """Signal should only contain -1, 0, 1.""" + daily_idx = pd.date_range("2020-01-01", periods=10, freq="D") + daily_signal = pd.Series([1, -1, 0, 1, -1, 0, 1, -1, 0, 1], index=daily_idx) + + idx_1min = pd.date_range("2020-01-01", "2020-01-10 23:59", freq="1min") + signal_1min = daily_signal.reindex(idx_1min).ffill().fillna(0).astype(int).clip(-1, 1) + + assert set(signal_1min.unique()) <= {-1, 0, 1} + assert signal_1min.isna().sum() == 0 + + def test_minimum_daily_data_rejected(self): + """Less than 20 daily rows should be rejected.""" + assert 10 < 20 # Orchestrator check: len(daily_factors) < 20 → rejected