diff --git a/.gitignore b/.gitignore index 3140e8c8..ee173b43 100644 --- a/.gitignore +++ b/.gitignore @@ -103,3 +103,9 @@ test_credentials.py # Closed source local components rdagent/scenarios/qlib/local/ docs/COMPLETE_WORKFLOW.md + +# Local scripts (not for public) +predix_quick_daytrading.py +predix_smart_strategy_gen.py +test/backtesting/test_smart_strategy_gen.py +docs/SMART_STRATEGY_GEN.md diff --git a/create_strategy.py b/create_strategy.py new file mode 100644 index 00000000..4294abca --- /dev/null +++ b/create_strategy.py @@ -0,0 +1,58 @@ +import json +import pandas as pd +import numpy as np + +# Strategy parameters +factors_used = ["daily_ret", "daily_close_return_96", "daily_cc_return", "momentum_1d", "london_mom"] +strategy_name = "ActiveDayMultiFactorScalper" +description = "Daytrading-Strategie mit 5 niedrig-korrelierten Faktoren und niedrigen Schwellenwerten für 50+ Trades" + +# Python code for signal generation +code = '''import numpy as np +import pandas as pd + +# Rolling Z-Scores mit kurzen Fenstern für schnelle Signale +z_daily_ret = (factors["daily_ret"] - factors["daily_ret"].rolling(15).mean()) / factors["daily_ret"].rolling(15).std() +z_close_ret = (factors["daily_close_return_96"] - factors["daily_close_return_96"].rolling(20).mean()) / factors["daily_close_return_96"].rolling(20).std() +z_cc_ret = (factors["daily_cc_return"] - factors["daily_cc_return"].rolling(15).mean()) / factors["daily_cc_return"].rolling(15).std() +z_mom = (factors["momentum_1d"] - factors["momentum_1d"].rolling(25).mean()) / factors["momentum_1d"].rolling(25).std() +z_london = (factors["london_mom"] - factors["london_mom"].rolling(30).mean()) / factors["london_mom"].rolling(30).std() + +# Kombiniere alle Z-Scores mit Gewichtung +composite_signal = ( + 0.25 * z_close_ret + # Höchste IC (0.255) - stärkstes Gewicht + 0.20 * z_london + # Zweithöchste IC (0.1857) + 0.20 * z_daily_ret + # IC 0.1291 + 0.20 * z_cc_ret + # IC 0.1291 + 0.15 * z_mom # IC 0.1291 +) + +# Niedrige Schwellenwerte für häufigere Signale (0.2-0.3) +threshold_long = 0.25 +threshold_short = -0.25 + +# Signal generieren +signal = pd.Series(0, index=close.index, name="signal") +signal[composite_signal > threshold_long] = 1 +signal[composite_signal < threshold_short] = -1 + +# NaN behandeln (am Anfang durch rolling window) +signal = signal.fillna(0).astype(int) +''' + +# Create strategy dict +strategy = { + "strategy_name": strategy_name, + "factor_names": factors_used, + "description": description, + "code": code +} + +# Save to JSON +output_file = f"{strategy_name}_strategy.json" +with open(output_file, "w") as f: + json.dump(strategy, f, indent=2) + +print(f"✅ Strategie gespeichert: {output_file}") +print(f"📊 Faktoren: {', '.join(factors_used)}") +print(f"🎯 Ziel: 50+ Trades mit niedrigen Schwellenwerten (±0.25)") diff --git a/predix_add_risk_management.py b/predix_add_risk_management.py new file mode 100644 index 00000000..f437ad38 --- /dev/null +++ b/predix_add_risk_management.py @@ -0,0 +1,337 @@ +#!/usr/bin/env python +""" +Add FTMO-compliant risk management to existing strategies. + +For each accepted strategy, add: +- Stop Loss: 2% +- Take Profit: 4% (2x SL) +- Trailing Stop: 1.5% after 2% profit +- Re-evaluate with risk management +- Generate Live Trading report + +Usage: + python predix_add_risk_management.py + python predix_add_risk_management.py --live # Mark as live-ready +""" +import os, sys, json, time +from pathlib import Path +from datetime import datetime +import numpy as np +import pandas as pd +from rich.console import Console +from rich.table import Table +from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn + +console = Console() + +STRATEGIES_DIR = Path('results/strategies_new') +OHLCV_PATH = Path('git_ignore_folder/factor_implementation_source_data/intraday_pv.h5') + +# FTMO Risk Parameters +STOP_LOSS = 0.02 # 2% hard stop +TAKE_PROFIT = 0.04 # 4% target (2x SL) +TRAILING_STOP = 0.015 # 1.5% trail after 2% profit +MAX_DAILY_LOSS = 0.05 # 5% FTMO daily limit + +def load_ohlcv(): + """Load OHLCV close prices.""" + 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() + + if isinstance(close.index, pd.MultiIndex): + close_dt_idx = close.index.get_level_values('datetime') + close = pd.Series(close.values, index=close_dt_idx, name='close') + + return close.dropna() + +def apply_risk_management(signal, close, sl=0.02, tp=0.04, trailing=0.015): + """ + Apply Stop Loss, Take Profit, and Trailing Stop to strategy. + + Returns strategy returns after risk management. + """ + FORWARD_BARS = 12 # 12-min forward returns for daytrading + returns_fwd = close.pct_change(FORWARD_BARS).shift(-FORWARD_BARS) + + signal_aligned = signal.loc[returns_fwd.dropna().index] + fwd_returns = returns_fwd.loc[signal_aligned.index] + + if len(signal_aligned) < 100: + return None, None + + strategy_returns = pd.Series(0.0, index=fwd_returns.index) + position = 0 + entry_price = 0 + peak_pnl = 0 + + for i in range(len(fwd_returns)): + sig = signal_aligned.iloc[i] + ret = fwd_returns.iloc[i] + + if position != 0: + # Calculate PnL + pnl = position * ret + + # Check Stop Loss + if pnl <= -sl: + strategy_returns.iloc[i] = -sl + position = 0 + peak_pnl = 0 + continue + + # Check Take Profit + if pnl >= tp: + strategy_returns.iloc[i] = tp + position = 0 + peak_pnl = 0 + continue + + # Check Trailing Stop + if pnl > 0.02: # After 2% profit + peak_pnl = max(peak_pnl, pnl) + if pnl < peak_pnl - trailing: + strategy_returns.iloc[i] = peak_pnl - trailing + position = 0 + peak_pnl = 0 + continue + + strategy_returns.iloc[i] = pnl + peak_pnl = max(peak_pnl, pnl) + + elif sig != 0: + # Enter position + position = sig + entry_price = close.iloc[i] if i < len(close) else 1.0 + + return strategy_returns, signal_aligned + +def evaluate_strategy(strategy_returns, signal_aligned): + """Calculate comprehensive metrics.""" + if strategy_returns is None or len(strategy_returns) < 100: + return None + + ic = signal_aligned.corr(strategy_returns / (strategy_returns.std() + 1e-8)) if strategy_returns.std() > 0 else 0 + + sharpe = strategy_returns.mean() / (strategy_returns.std() + 1e-8) * np.sqrt(252 * 1440 / 12) + + cum = (1 + strategy_returns).cumprod() + running_max = cum.expanding().max() + drawdown = (cum - running_max) / running_max.replace(0, np.nan) + max_dd = drawdown.min() if len(drawdown) > 0 else 0 + + win_rate = (strategy_returns > 0).sum() / len(strategy_returns) + n_trades = int((signal_aligned != signal_aligned.shift(1)).sum()) + total_return = cum.iloc[-1] - 1 + n_bars = len(strategy_returns) + n_months = n_bars / (252 * 1440 / 12 / 12) if n_bars > 0 else 1 + + monthly_return = (1 + total_return) ** (1 / n_months) - 1 if n_months > 0 and (1 + total_return) > 0 else total_return + + # Daily loss check + daily_returns = strategy_returns.groupby(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 + + return { + 'ic': float(ic) if not np.isnan(ic) else 0, + 'sharpe': float(sharpe), + 'max_drawdown': float(max_dd) if not np.isnan(max_dd) else 0, + 'win_rate': float(win_rate), + 'n_trades': n_trades, + 'total_return': float(total_return), + 'monthly_return_pct': float(monthly_return * 100), + 'n_bars': int(n_bars), + 'n_months': float(n_months), + 'max_daily_loss': float(max_daily_loss), + 'ftmo_compliant': max_daily_loss <= MAX_DAILY_LOSS and max_dd > -0.10, + } + +def main(): + console.print("[bold cyan]🔒 Adding FTMO Risk Management to Existing Strategies[/bold cyan]\n") + + # Load OHLCV + console.print("📊 Loading OHLCV data...") + close = load_ohlcv() + console.print(f" ✓ Loaded {len(close):,} bars\n") + + # Load strategies + strategies = [] + for f in sorted(STRATEGIES_DIR.glob('*.json')): + try: + data = json.load(open(f)) + bt = data.get('real_backtest', {}) + if bt.get('status') == 'success': + strategies.append((f, data)) + except: + pass + + console.print(f"📁 Found {len(strategies)} accepted strategies\n") + + # Process each strategy + results = [] + + with Progress( + SpinnerColumn(), + TextColumn("[bold blue]{task.description}"), + BarColumn(), + TextColumn("[bold green]{task.completed}/{task.total}"), + ) as progress: + task = progress.add_task("Processing...", total=len(strategies)) + + for fpath, data in strategies: + name = data.get('strategy_name', 'Unknown') + progress.update(task, description=f"Processing {name}...") + + # Load factors + factor_names = data.get('factor_names', []) + + # Load factor parquet files + factors_data = {} + for fname in factor_names: + safe = fname.replace('/', '_').replace('\\', '_')[:150] + pf = Path('results/factors/values') / f"{safe}.parquet" + if pf.exists(): + try: + df = pd.read_parquet(str(pf)) + if df.index.names == ['datetime', 'instrument']: + df_reset = df.reset_index() + if 'instrument' in df_reset.columns: + df_eur = df_reset[df_reset['instrument'] == 'EURUSD'].copy() + df_eur = df_eur.set_index('datetime') + factors_data[fname] = df_eur.iloc[:, -1] + except: + pass + + if len(factors_data) < 2: + progress.update(task, advance=1) + continue + + # Build factors DataFrame + df_factors = pd.DataFrame(factors_data) + common_idx = close.index.intersection(df_factors.dropna(how='all').index) + close_aligned = close.loc[common_idx] + df_aligned = df_factors.loc[common_idx] + + # Execute strategy code + try: + local_vars = {'factors': df_aligned, 'close': close_aligned} + exec(data.get('code', ''), {}, local_vars) + signal = local_vars.get('signal', pd.Series(0, index=close_aligned.index)) + except: + progress.update(task, advance=1) + continue + + # Apply risk management + strat_returns, sig_aligned = apply_risk_management( + signal, close_aligned, + sl=STOP_LOSS, tp=TAKE_PROFIT, trailing=TRAILING_STOP + ) + + if strat_returns is None: + progress.update(task, advance=1) + continue + + # Evaluate + metrics = evaluate_strategy(strat_returns, sig_aligned) + if metrics is None: + progress.update(task, advance=1) + continue + + # Store result + result = { + 'name': name, + 'file': fpath.name, + 'original_ic': data.get('real_backtest', {}).get('ic', 0), + 'original_sharpe': data.get('real_backtest', {}).get('sharpe', 0), + 'new_ic': metrics['ic'], + 'new_sharpe': metrics['sharpe'], + 'new_max_dd': metrics['max_drawdown'], + 'new_win_rate': metrics['win_rate'], + 'new_trades': metrics['n_trades'], + 'new_monthly_ret': metrics['monthly_return_pct'], + 'max_daily_loss': metrics['max_daily_loss'], + 'ftmo_compliant': bool(metrics['ftmo_compliant']), + } + results.append(result) + + # Update strategy JSON + data['risk_management'] = { + 'stop_loss': STOP_LOSS, + 'take_profit': TAKE_PROFIT, + 'trailing_stop': TRAILING_STOP, + 'trailing_trigger': 0.02, + 'max_daily_loss': MAX_DAILY_LOSS, + 'ftmo_compliant': bool(metrics['ftmo_compliant']), + } + data['evaluated_with_risk_mgmt'] = metrics + data['summary'] = { + 'sharpe': metrics['sharpe'], + 'max_drawdown': metrics['max_drawdown'], + 'win_rate': metrics['win_rate'], + 'monthly_return_pct': metrics['monthly_return_pct'], + 'real_ic': metrics['ic'], + 'real_n_trades': metrics['n_trades'], + 'ftmo_compliant': bool(metrics['ftmo_compliant']), + 'forward_bars': 12, + 'trading_style': 'daytrading', + } + + with open(fpath, 'w') as f: + # Convert numpy types for JSON + def sanitize(obj): + if hasattr(obj, 'item'): return obj.item() + if isinstance(obj, dict): return {k: sanitize(v) for k, v in obj.items()} + if isinstance(obj, list): return [sanitize(v) for v in obj] + if isinstance(obj, (np.bool_, bool)): return bool(obj) + return obj + + json.dump(sanitize(data), f, indent=2, ensure_ascii=False) + + progress.update(task, advance=1) + + # Display results + console.print("\n[bold green]✓ All strategies processed![/bold green]\n") + + table = Table(title="📊 FTMO Risk Management Results") + table.add_column("#", justify="right") + table.add_column("Strategy", style="cyan") + table.add_column("IC", justify="right") + table.add_column("Sharpe", justify="right") + table.add_column("Trades", justify="right") + table.add_column("Monthly %", justify="right") + table.add_column("Max DD", justify="right") + table.add_column("FTMO", justify="center") + + results.sort(key=lambda x: x['new_sharpe'], reverse=True) + for i, r in enumerate(results, 1): + ftmo = "✅" if r['ftmo_compliant'] else "❌" + table.add_row( + str(i), r['name'], + f"{r['new_ic']:.4f}", + f"{r['new_sharpe']:.2f}", + str(r['new_trades']), + f"{r['new_monthly_ret']:.2f}%", + f"{r['new_max_dd']:.1%}", + ftmo + ) + + console.print(table) + + # Summary + ftmo_count = sum(1 for r in results if r['ftmo_compliant']) + console.print(f"\n[bold]FTMO-Compliant:[/bold] {ftmo_count}/{len(results)} strategies") + + if results: + best = results[0] + console.print(f"\n[bold green]🏆 Best Strategy: {best['name']}[/bold green]") + console.print(f" Sharpe: {best['new_sharpe']:.2f}") + console.print(f" Monthly Return: {best['new_monthly_ret']:.2f}%") + console.print(f" Max Drawdown: {best['new_max_dd']:.1%}") + console.print(f" FTMO Compliant: {'✅' if best['ftmo_compliant'] else '❌'}") + +if __name__ == '__main__': + main() diff --git a/predix_gen_strategies_real_bt.py b/predix_gen_strategies_real_bt.py index d17d2e8c..2907e983 100644 --- a/predix_gen_strategies_real_bt.py +++ b/predix_gen_strategies_real_bt.py @@ -1,26 +1,33 @@ #!/usr/bin/env python """ -Generate trading strategies using LLM and backtest with REAL OHLCV data. +Parallel AI Strategy Generation with REAL OHLCV Backtest. -Uses vectorbt (popular backtesting library) for accurate metrics. -Only saves strategies that pass real backtest thresholds. +Generates multiple trading strategies in parallel using LLM, +each with real backtesting on OHLCV data. Usage: - python predix_gen_strategies_real_bt.py # Generate 10 strategies - python predix_gen_strategies_real_bt.py 20 # Generate 20 strategies + # Swing trading (96-bar forward returns) + python predix_gen_strategies_real_bt.py 10 + + # Daytrading with FTMO constraints (12-bar forward returns) + TRADING_STYLE=daytrading python predix_gen_strategies_real_bt.py 5 + + # With parallel workers (default: CPU count) + TRADING_STYLE=daytrading WORKERS=4 python predix_gen_strategies_real_bt.py 20 """ -import json, subprocess, tempfile, os, time, math +import os, sys, json, time, math, random, logging, warnings, subprocess +from pathlib import Path +from datetime import datetime + import numpy as np import pandas as pd -from pathlib import Path from rich.console import Console -from rich.progress import Progress +from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TimeElapsedColumn from dotenv import load_dotenv -# Load .env for API keys -load_dotenv(Path(__file__).parent / ".env") - -console = Console() +# Suppress warnings +warnings.filterwarnings('ignore') +logging.getLogger('rdagent').setLevel(logging.WARNING) # ============================================================================ # Configuration @@ -30,189 +37,186 @@ FACTORS_DIR = Path('/home/nico/Predix/results/factors') STRATEGIES_DIR = Path('/home/nico/Predix/results/strategies_new') STRATEGIES_DIR.mkdir(parents=True, exist_ok=True) -# Trading style: 'swing' (96 bars) or 'daytrading' (12-24 bars) -TRADING_STYLE = os.getenv('TRADING_STYLE', 'swing') # 'swing' or 'daytrading' +# Trading style +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')) # 12 minutes + FORWARD_BARS = int(os.getenv('FORWARD_BARS', '12')) MIN_IC = 0.02 MIN_SHARPE = 0.5 - MIN_TRADES = 50 # More trades for daytrading - MAX_DRAWDOWN = -0.10 # FTMO compliant (10% max) - print("[dim]🎯 Daytrading mode: 12-bar forward returns, FTMO risk limits[/dim]") + MIN_TRADES = 20 + MAX_DRAWDOWN = -0.10 + STYLE_EMOJI = '🎯 Daytrading' + STYLE_DESC = 'short-term intraday with FTMO compliance' else: - FORWARD_BARS = int(os.getenv('FORWARD_BARS', '96')) # 96 minutes (default) + FORWARD_BARS = int(os.getenv('FORWARD_BARS', '96')) MIN_IC = 0.02 MIN_SHARPE = 0.5 MIN_TRADES = 10 - MAX_DRAWDOWN = -1.0 # No limit for swing + MAX_DRAWDOWN = -1.0 + STYLE_EMOJI = '📈 Swing' + STYLE_DESC = 'medium-term intraday' + +console = Console() # ============================================================================ -# OHLCV Data Loading (cached) +# LLM Configuration (Process-safe) # ============================================================================ -_ohlcv_cache = {} - -def load_ohlcv_data() -> pd.DataFrame: - """Load OHLCV data with close prices for backtesting. Returns cached if available.""" - global _ohlcv_cache - if 'close' not in _ohlcv_cache: - if not OHLCV_PATH.exists(): - raise FileNotFoundError(f"OHLCV data not found: {OHLCV_PATH}") - - console.print("[dim]Loading OHLCV data...[/dim]") - df = pd.read_hdf(str(OHLCV_PATH), key='data') - - # Extract close price (handle different column names) - if '$close' in df.columns: - close = df['$close'] - elif 'close' in df.columns: - close = df['close'] - else: - # Try first numeric column - close = df.select_dtypes(include=[np.number]).iloc[:, 0] - - _ohlcv_cache['close'] = close - console.print(f"[green]✓[/green] Loaded {len(close):,} close prices") - - return _ohlcv_cache['close'] - +def setup_llm_env(): + """Setup LLM environment variables.""" + load_dotenv(Path(__file__).parent / '.env') + router_key = os.getenv('OPENROUTER_API_KEY') or os.getenv('OPENAI_API_KEY', '') + if not router_key or router_key == 'local': + 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/qwen/qwen3.6-plus:free') # ============================================================================ -# Factor Loading +# Factor Loading (cached at module level for each process) # ============================================================================ +_FACTORS_CACHE = None + def load_available_factors(top_n=20): """Load top factors that have parquet time-series files.""" - factors = [] + global _FACTORS_CACHE + if _FACTORS_CACHE is not None: + return _FACTORS_CACHE[:top_n] + factors = [] 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, - 'description': data.get('factor_description', '')[:100], - }) + factors.append({'name': fname, 'ic': ic}) except: pass factors.sort(key=lambda x: abs(x['ic']), reverse=True) + _FACTORS_CACHE = factors return factors[:top_n] +# ============================================================================ +# OHLCV Data Loading (cached at module level) +# ============================================================================ +_OHLCV_CACHE = None -def load_factor_time_series(factor_names): - """Load factor time-series and align with OHLCV index.""" - close = load_ohlcv_data() +def load_ohlcv_data(): + """Load OHLCV close prices.""" + global _OHLCV_CACHE + if _OHLCV_CACHE is not None: + return _OHLCV_CACHE - factors = {} - for fname in factor_names: - safe = fname.replace('/','_').replace('\\','_')[:150] - p = FACTORS_DIR / 'values' / f"{safe}.parquet" - if p.exists(): - try: - series = pd.read_parquet(str(p)).iloc[:, 0] - factors[fname] = series - except: - pass + if not OHLCV_PATH.exists(): + raise FileNotFoundError(f"OHLCV data not found: {OHLCV_PATH}") - if not factors: - return None, None + 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] - # Combine and align with close prices - df_factors = pd.DataFrame(factors).dropna() - - # Reindex to match close prices (forward fill factors) - df_factors = df_factors.reindex(close.index).ffill() - - # Remove rows where we don't have close prices - valid = close.dropna().index.intersection(df_factors.dropna(how='all').index) - close = close.loc[valid] - df_factors = df_factors.loc[valid] - - return close, df_factors - + _OHLCV_CACHE = close.dropna() + return _OHLCV_CACHE # ============================================================================ -# LLM Strategy Generation +# Strategy Generation (LLM call - runs in separate process) # ============================================================================ -def generate_strategy_with_llm(factors, previous_feedback=None): - """Generate strategy code using LLM.""" - from rdagent.oai.llm_utils import APIBackend +def generate_single_strategy(args): + """Generate and backtest ONE strategy. Runs in separate process.""" + idx, factor_subset, feedback, attempt = args - # Force OpenRouter - router_key = os.getenv("OPENROUTER_API_KEY") or os.getenv("OPENAI_API_KEY", "") - if not router_key or router_key == "local": - router_key = os.getenv("OPENROUTER_API_KEY", "") - - if not router_key: - console.print("[red]No OPENROUTER_API_KEY found![/red]") - return None - - os.environ["OPENAI_API_KEY"] = router_key - os.environ["OPENAI_API_BASE"] = "https://openrouter.ai/api/v1" - os.environ["CHAT_MODEL"] = os.getenv("OPENROUTER_MODEL", "openrouter/qwen/qwen3.6-plus:free") - - factor_list = "\n".join([f"- {f['name']} (IC={f['ic']:.4f})" for f in factors]) - - system_prompt = """You are a quantitative trading expert. Generate a trading strategy by combining factors. + 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': + system_prompt = f"""You are an expert daytrading quant specializing in EUR/USD scalping and intraday strategies. -CRITICAL RULES: +CRITICAL RULES for {STYLE_DESC} (forward horizon: {FORWARD_BARS} bars = ~{FORWARD_BARS} minutes): +1. ONLY use the factors listed below - no others! +2. The code MUST work with a DataFrame called 'factors' and Series called 'close' +3. Create a pandas Series called 'signal' with values: 1 (long), -1 (short), 0 (neutral) +4. signal.index MUST match close.index +5. signal.name must be 'signal' +6. Optimize for FREQUENT signals (many trades) since the horizon is only {FORWARD_BARS} minutes +7. Use LOWER thresholds (0.2-0.5) to generate more trades for daytrading + +Output ONLY valid JSON with these fields: +{{"strategy_name": "short_name", "factor_names": ["f1", "f2"], "description": "one sentence", "code": "python code"}}""" + + user_prompt = f"""Create a EUR/USD DAYTRADING strategy ({FORWARD_BARS}-minute horizon) using these factors: + +{factor_list} + +{f'Previous feedback: {feedback}' if feedback else 'First attempt - be creative!'} + +Requirements for daytrading: +- Use {FORWARD_BARS}-minute forward returns (not daily) +- Generate frequent signals (aim for 20+ trades in the dataset) +- Use rolling z-scores with short windows (10-30 bars) +- Apply tight thresholds (0.2-0.5) for more trades +- Combine momentum + mean-reversion effectively""" + + else: + system_prompt = f"""You are a quantitative trading expert specializing in EUR/USD intraday strategies. + +CRITICAL RULES for {STYLE_DESC} (forward horizon: {FORWARD_BARS} bars = ~{FORWARD_BARS/60:.1f} hours): 1. ONLY use the factors listed below - no others! 2. The code MUST work with a DataFrame called 'factors' and Series called 'close' 3. Create a pandas Series called 'signal' with values: 1 (long), -1 (short), 0 (neutral) 4. signal.index MUST match close.index 5. signal.name must be 'signal' -The 'close' Series contains EUR/USD close prices. -The 'factors' DataFrame contains factor values aligned with close prices. - Output ONLY valid JSON with these fields: -{ - "strategy_name": "short_name", - "factor_names": ["factor1", "factor2"], - "description": "one sentence", - "code": "python code with \\n for newlines" -}""" - - user_prompt = f"""Generate a EUR/USD trading strategy using these factors: +{{"strategy_name": "short_name", "factor_names": ["f1", "f2"], "description": "one sentence", "code": "python code"}}""" + + user_prompt = f"""Create a EUR/USD trading strategy using these factors: {factor_list} -Previous feedback: {previous_feedback or 'None - first attempt'} - -Create an innovative strategy that combines momentum and mean-reversion signals.""" - - try: +{f'Previous feedback: {feedback}' if feedback else 'First attempt - be creative!'}""" + 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 ) - return json.loads(response) + 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} + + return { + 'status': 'generated', + 'strategy': strategy_data, + 'idx': idx + } + except Exception as e: - console.print(f"[red]LLM Error: {e}[/red]") - return None - + return {'status': 'error', 'reason': str(e)[:200], 'idx': idx} # ============================================================================ -# Real Backtesting with vectorbt +# Backtest Runner (runs in main process to avoid re-loading data) # ============================================================================ -def run_real_backtest(close, df_factors, strategy_code): - """ - Run real backtest using actual OHLCV data. - - FIXED: Uses 96-bar forward returns (matching factor IC evaluation), - not 1-bar returns which are too noisy for 1-min data. - """ - if close is None or df_factors is None or len(df_factors.columns) < 2: +def run_backtest(close, factors_df, strategy_code): + """Run real backtest with actual OHLCV data.""" + if close is None or factors_df is None or len(factors_df.columns) < 2: return None - - # Build test script + + import tempfile + script = f""" import pandas as pd import numpy as np @@ -221,45 +225,49 @@ import json close = pd.read_pickle('close.pkl') factors = pd.read_pickle('factors.pkl') -# Execute strategy code try: {chr(10).join(' ' + l for l in strategy_code.split(chr(10)))} except: print("ERROR: Strategy execution failed") exit(1) -# Validate signal if 'signal' not in dir(): print("ERROR: No signal generated") exit(1) signal = signal.fillna(0) - -# Ensure signal aligns with close common_idx = close.index.intersection(signal.index) close = close.loc[common_idx] signal = signal.loc[common_idx] -# Calculate returns - using 96-bar forward return (matching factor IC horizon) FORWARD_BARS = {FORWARD_BARS} -returns_96 = close.pct_change(FORWARD_BARS).shift(-FORWARD_BARS) -signal_aligned = signal.loc[returns_96.dropna().index] -fwd_returns = returns_96.loc[signal_aligned.index] +returns_fwd = close.pct_change(FORWARD_BARS).shift(-FORWARD_BARS) +signal_aligned = signal.loc[returns_fwd.dropna().index] +fwd_returns = returns_fwd.loc[signal_aligned.index] if len(signal_aligned) < 100 or len(fwd_returns) < 100: print("ERROR: Not enough data after alignment") exit(1) -# Calculate IC: correlation(signal, forward_return) ic = signal_aligned.corr(fwd_returns) - -# Strategy returns strategy_returns = signal_aligned * fwd_returns -# Basic metrics -total_return = (1 + strategy_returns).prod() - 1 +if strategy_returns.std() > 0: + sharpe = strategy_returns.mean() / strategy_returns.std() * np.sqrt(252 * 1440 / {FORWARD_BARS}) +else: + sharpe = 0 + +cum = (1 + strategy_returns).cumprod() +running_max = cum.expanding().max() +drawdown = (cum - running_max) / running_max.replace(0, np.nan) +max_dd = drawdown.min() if len(drawdown) > 0 else 0 + +win_rate = (strategy_returns > 0).sum() / len(strategy_returns) if len(strategy_returns) > 0 else 0 +n_trades = int((signal_aligned != signal_aligned.shift(1)).sum()) + +total_return = cum.iloc[-1] - 1 n_bars = len(strategy_returns) -n_months = n_bars / (252 * 1440 / 96 / 12) if n_bars > 0 else 1 +n_months = n_bars / (252 * 1440 / {FORWARD_BARS} / 12) if n_bars > 0 else 1 if n_months > 0 and (1 + total_return) > 0: monthly_return = (1 + total_return) ** (1 / n_months) - 1 @@ -268,24 +276,6 @@ else: monthly_return = total_return annual_return = total_return * 12 -# Sharpe ratio (annualized for 96-bar horizon) -if strategy_returns.std() > 0: - sharpe = strategy_returns.mean() / strategy_returns.std() * np.sqrt(252 * 1440 / 96) -else: - sharpe = 0 - -# Max Drawdown -cum_returns = (1 + strategy_returns).cumprod() -running_max = cum_returns.expanding().max() -drawdown = (cum_returns - running_max) / running_max.replace(0, np.nan) -max_dd = drawdown.min() if len(drawdown) > 0 else 0 - -# Win rate -win_rate = (strategy_returns > 0).sum() / len(strategy_returns) if len(strategy_returns) > 0 else 0 - -# Trade count (signal changes) -n_trades = int((signal_aligned != signal_aligned.shift(1)).sum()) - result = {{ "status": "success", "sharpe": float(sharpe), @@ -308,156 +298,174 @@ print(json.dumps(result)) with tempfile.TemporaryDirectory() as td: tdp = Path(td) - - # Save close and factors as pickle close.to_pickle(str(tdp / 'close.pkl')) - df_factors.to_pickle(str(tdp / 'factors.pkl')) + factors_df.to_pickle(str(tdp / 'factors.pkl')) - script_path = tdp / "run.py" + script_path = tdp / 'run.py' script_path.write_text(script) try: result = subprocess.run( - ["python", str(script_path)], - capture_output=True, text=True, timeout=120, + ['python', str(script_path)], + capture_output=True, text=True, timeout=60, cwd=str(tdp) ) if result.returncode != 0: - return {"status": "failed", "reason": result.stderr[:300] or result.stdout[:300]} + return {'status': 'failed', 'reason': result.stderr[:200] or result.stdout[:200]} - # Parse JSON output for line in result.stdout.strip().split('\n'): try: return json.loads(line) except: continue - return {"status": "failed", "reason": "No valid output"} - + return {'status': 'failed', 'reason': 'No valid JSON output'} except subprocess.TimeoutExpired: - return {"status": "failed", "reason": "Timeout (120s)"} + return {'status': 'failed', 'reason': 'Timeout (60s)'} except Exception as e: - return {"status": "failed", "reason": str(e)} - + return {'status': 'failed', 'reason': str(e)[:200]} # ============================================================================ -# Main +# Main Parallel Strategy Generation # ============================================================================ -def main(count=10, max_attempts=50): - """Generate and backtest strategies until we have 'count' successful ones.""" - console.print("[bold cyan]🧠 Strategy Generation with REAL Backtest[/bold cyan]") - console.print("[dim]Using vectorbt + real OHLCV data for accurate metrics[/dim]\n") +def main(target_count=10): + """Generate strategies in parallel with real backtesting.""" - try: - factors = load_available_factors(20) - console.print(f"[green]✓[/green] Loaded {len(factors)} factors with time-series\n") - except FileNotFoundError as e: - console.print(f"[red]{e}[/red]") + console.print(f"\n[bold cyan]{STYLE_EMOJI} Parallel Strategy Generation[/bold cyan]") + console.print(f" Style: {STYLE_DESC}") + 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" + if pf.exists(): + try: + series = pd.read_parquet(str(pf)).iloc[:, 0] + 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 - results = [] - feedback = None + 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) + close_aligned = close.loc[common_idx] + df_aligned = df_factors.loc[common_idx] - with Progress() as progress: - task = progress.add_task(f"Generating strategies (target: {count})...", total=max_attempts) + 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}"), + BarColumn(), + TextColumn("[bold green]{task.completed}/{task.total}"), + TimeElapsedColumn(), + ) as progress: + task = progress.add_task("Generating...", total=max_attempts) for attempt in range(max_attempts): - if len(results) >= count: + if len(accepted) >= target_count: break - progress.update(task, description=f"Attempt {attempt+1}/{max_attempts} ({len(results)}/{count} successful)") + # Select random factor subset (2-5 factors) + n_factors = random.randint(2, min(5, len(factors))) + factor_subset = random.sample(factors, n_factors) - # Generate - strat = generate_strategy_with_llm(factors, feedback) - if not strat: - feedback = "LLM failed to generate strategy" - progress.advance(task) + feedback = feedback_history[-1] if feedback_history and random.random() < 0.7 else None + + # Generate in main process (LLM doesn't parallelize well) + gen_result = generate_single_strategy((attempt, factor_subset, feedback, attempt)) + + if gen_result['status'] != 'generated': + progress.update(task, advance=1) continue - # Load real data - try: - close, df_factors = load_factor_time_series(strat.get('factor_names', [])) - except Exception as e: - feedback = f"Data loading error: {e}" - progress.advance(task) + strategy = gen_result['strategy'] + + # Backtest (main process - needs data access) + # Build factors DataFrame for this strategy + 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 - if df_factors is None or len(df_factors.columns) < 2: - feedback = f"Only {len(df_factors.columns) if df_factors is not None else 0} factors available" - progress.advance(task) - continue + bt_result = run_backtest(close_aligned, strat_factors, strategy.get('code', '')) - # Backtest with REAL data - bt = run_real_backtest(close, df_factors, strat.get('code', '')) - - if bt and bt.get('status') == 'success': - ic = bt.get('ic', 0) - sharpe = bt.get('sharpe', 0) - trades = bt.get('n_trades', 0) + 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) - # Acceptance criteria - if abs(ic) > MIN_IC and sharpe > MIN_SHARPE and trades > MIN_TRADES and bt.get("max_drawdown", 0) > MAX_DRAWDOWN: - # SUCCESS - strat['real_backtest'] = bt - strat['metrics'] = bt - strat['summary'] = { - "sharpe": sharpe, - "max_drawdown": bt.get('max_drawdown', 0), - "win_rate": bt.get('win_rate', 0), - "monthly_return_pct": bt.get('monthly_return_pct', 0), - "annual_return_pct": bt.get('annual_return_pct', 0), - "real_ic": ic, - "real_n_trades": trades, - "real_backtest_status": "success", - "n_bars": bt.get('n_bars', 0), - "n_months": bt.get('n_months', 0), + # Check acceptance criteria + if abs(ic) > MIN_IC and sharpe > MIN_SHARPE and trades > MIN_TRADES and dd > MAX_DRAWDOWN: + # 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), } - fname = f"{int(time.time())}_{strat['strategy_name']}.json" + fname = f"{int(time.time())}_{strategy['strategy_name']}.json" with open(STRATEGIES_DIR / fname, 'w') as f: - json.dump(strat, f, indent=2, ensure_ascii=False) - - # Generate performance report automatically + json.dump(strategy, f, indent=2, ensure_ascii=False) + + # Generate PDF report try: from predix_strategy_report import StrategyPerformanceReporter - reporter = StrategyPerformanceReporter(strat) - report_path = reporter.generate_report() - console.print(f" [dim]📊 Report: {report_path.name}[/dim]") - except Exception as e: - console.print(f" [dim]⚠️ Report gen failed: {e}[/dim]") - - results.append(strat) - console.print(f"[green]✓ Strategy #{len(results)}:[/green] {strat['strategy_name']} " - f"IC={ic:.4f}, Sharpe={sharpe:.3f}, Monthly={bt.get('monthly_return_pct', 0):.2f}%, " - f"Trades={trades}") - feedback = f"Good strategy! Sharpe={sharpe:.2f}, IC={ic:.4f}. Try to improve." + reporter = StrategyPerformanceReporter(strategy) + reporter.generate_report() + except: + pass + + accepted.append(strategy) + feedback_history.append(f"Excellent! IC={ic:.4f}, Sharpe={sharpe:.2f}, Trades={trades}. Try to improve further.") + + progress.console.print(f"[green]✓ Strategy #{len(accepted)}:[/green] {strategy['strategy_name']} " + f"IC={ic:.4f}, Sharpe={sharpe:.3f}, Trades={trades}, DD={dd:.1%}") else: - feedback = f"Failed: IC={ic:.4f}, Sharpe={sharpe:.3f}, Trades={trades}. Need |IC|>{MIN_IC}, Sharpe>{MIN_SHARPE}, Trades>{MIN_TRADES}" - else: - feedback = f"Backtest failed: {bt.get('reason', 'Unknown') if bt else 'No result'}" + feedback_history.append(f"Failed: IC={ic:.4f}, Sharpe={sharpe:.2f}, Trades={trades}, DD={dd:.1%}. Need |IC|>{MIN_IC}, Sharpe>{MIN_SHARPE}, Trades>{MIN_TRADES}") - progress.advance(task) - time.sleep(2) + progress.update(task, advance=1) # Summary - console.print(f"\n[bold green]✓ Generated {len(results)} strategies with REAL OHLCV backtests[/bold green]") + console.print(f"\n[bold green]✓ Generated {len(accepted)}/{target_count} accepted strategies[/bold green]\n") - if results: - results.sort(key=lambda x: abs(x['real_backtest']['ic']), reverse=True) - console.print("\n[bold]Results:[/bold]") - console.print(f"{'#':>3} {'Name':<30} {'IC':>7} {'Sharpe':>7} {'Monthly':>9} {'Trades':>7}") - console.print("-" * 70) - for i, r in enumerate(results, 1): - bt = r['real_backtest'] - console.print( - f"{i:3d} {r['strategy_name']:30s} " - f"{bt['ic']:7.4f} {bt['sharpe']:7.3f} " - f"{bt.get('monthly_return_pct', 0):8.2f}% {bt.get('n_trades', 0):7d}" - ) + if accepted: + 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'] + 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__": - import sys +if __name__ == '__main__': count = int(sys.argv[1]) if len(sys.argv) > 1 else 10 main(count) diff --git a/rdagent/oai/llm_conf.py b/rdagent/oai/llm_conf.py index a9a1130e..17264daa 100644 --- a/rdagent/oai/llm_conf.py +++ b/rdagent/oai/llm_conf.py @@ -16,7 +16,7 @@ class LLMSettings(ExtendedBaseSettings): embedding_model: str = "text-embedding-3-small" reasoning_effort: Literal["low", "medium", "high"] | None = None - enable_response_schema: bool = True + enable_response_schema: bool = False # Whether to enable response_schema in chat models. may not work for models that do not support it. # Handling format diff --git a/requirements.txt b/requirements.txt index e802db5a..795e5e2c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -89,6 +89,9 @@ duckduckgo-search pytest pytest-cov +# News & Data (Polymarket, ForexFactory, CryptoPanic) +beautifulsoup4>=4.12.0 + # RL Trading (optional - system works without these) # Install for full RL training: pip install stable-baselines3[extra] gymnasium # Without these, RL trading uses simple momentum fallback diff --git a/start_llama.sh b/start_llama.sh new file mode 100755 index 00000000..2b2dd533 --- /dev/null +++ b/start_llama.sh @@ -0,0 +1,31 @@ +#!/bin/bash +# Start llama.cpp server with Qwen3.5-35B for strategy generation +# Usage: ./start_llama.sh + +MODEL_PATH="$HOME/models/qwen3.5/Qwen3.5-35B-A3B-Q3_K_M.gguf" +PORT=8081 + +# GPU: RTX 5060 Ti (16GB VRAM) +# Modell braucht ~15.7GB bei voller GPU-Nutzung +# Mit 14GB free → reduzieren wir GPU-Layers + Context +GPU_LAYERS=30 # Weniger Layers für VRAM +CTX_SIZE=4096 # 4K Context (reicht für Strategien) + +echo "🚀 Starting llama.cpp server..." +echo " Model: $(basename $MODEL_PATH)" +echo " Port: $PORT" +echo " GPU Layers: $GPU_LAYERS" +echo " Context: $CTX_SIZE" +echo "" + +exec ~/llama.cpp/build/bin/llama-server \ + --model "$MODEL_PATH" \ + --n-gpu-layers $GPU_LAYERS \ + --ctx-size $CTX_SIZE \ + --port $PORT \ + --threads 8 \ + --threads-batch 8 \ + --parallel 1 \ + --flash-attn \ + --jinja \ + --host 0.0.0.0 diff --git a/start_strategy_loop.sh b/start_strategy_loop.sh new file mode 100755 index 00000000..6e419cea --- /dev/null +++ b/start_strategy_loop.sh @@ -0,0 +1,115 @@ +#!/bin/bash +# ============================================================================ +# PREDIX Strategy Generator - Robust Loop +# Restarts automatically on crash, generates strategies continuously. +# ============================================================================ + +SCRIPT_DIR="/home/nico/Predix" +GENERATOR="python ${SCRIPT_DIR}/predix_smart_strategy_gen.py" +TARGET_COUNT=3 +LOGFILE="${SCRIPT_DIR}/results/logs/generator_loop.log" +PIDFILE="/tmp/predix_loop.pid" + +echo $$ > "$PIDFILE" +mkdir -p "${SCRIPT_DIR}/results/logs" + +log() { + echo "$(date '+%Y-%m-%d %H:%M:%S') - $1" | tee -a "$LOGFILE" +} + +cleanup() { + log "Received termination signal. Cleaning up..." + pkill -f "predix_smart_strategy_gen.py" 2>/dev/null + rm -f "$PIDFILE" + log "Cleanup complete. Exiting." + exit 0 +} + +trap cleanup SIGTERM SIGINT + +log "=========================================" +log "🚀 PREDIX Generator Loop Starting" +log "=========================================" +log "Target: ${TARGET_COUNT} strategies per run" +log "Log: ${LOGFILE}" + +ATTEMPT=0 + +while true; do + ATTEMPT=$((ATTEMPT + 1)) + log "" + log "=== Attempt #${ATTEMPT} ===================================" + + # Check disk space + DISK_USAGE=$(df -h ${SCRIPT_DIR} | tail -1 | awk '{print $5}' | sed 's/%//') + if [ "$DISK_USAGE" -gt 90 ]; then + log "⚠️ Disk usage at ${DISK_USAGE}%. Pausing..." + sleep 300 + continue + fi + + # Check if we already have enough strategies + STRAT_COUNT=$(ls ${SCRIPT_DIR}/results/strategies_new/*.json 2>/dev/null | wc -l) + log "📁 Existing strategies: ${STRAT_COUNT}" + + # Kill any stale processes + pkill -9 -f "predix_smart_strategy_gen.py" 2>/dev/null + sleep 2 + + # Start generator + log "🤖 Starting generator..." + cd "$SCRIPT_DIR" + nohup $GENERATOR $TARGET_COUNT > /dev/null 2>&1 & + GEN_PID=$! + log " PID: ${GEN_PID}" + + # Monitor progress + ELAPSED=0 + MAX_WAIT=1800 # 30 minutes max per run + + while kill -0 $GEN_PID 2>/dev/null; do + sleep 30 + ELAPSED=$((ELAPSED + 30)) + + # Check latest log for progress + LATEST_LOG=$(ls -t ${SCRIPT_DIR}/results/logs/smart_strategy_gen_*.log 2>/dev/null | head -1) + if [ -n "$LATEST_LOG" ]; then + LAST_LINE=$(tail -1 "$LATEST_LOG" 2>/dev/null) + if [ $((ELAPSED % 120)) -eq 0 ]; then # Every 2 min + log " ⏱️ ${ELAPSED}s elapsed - ${LAST_LINE:0:80}" + fi + fi + + # Timeout check + if [ $ELAPSED -ge $MAX_WAIT ]; then + log " ⏰ Timeout after ${ELAPSED}s. Killing..." + kill -9 $GEN_PID 2>/dev/null + break + fi + done + + # Check results + wait $GEN_PID 2>/dev/null + EXIT_CODE=$? + + if [ $EXIT_CODE -eq 0 ]; then + log "✅ Generator completed successfully" + elif [ $EXIT_CODE -eq 137 ]; then + log "❌ Generator killed (OOM? Exit 137)" + else + log "⚠️ Generator exited with code ${EXIT_CODE}" + fi + + # Count new strategies + NEW_STRATS=$(ls -t ${SCRIPT_DIR}/results/strategies_new/*.json 2>/dev/null | head -3) + if [ -n "$NEW_STRATS" ]; then + log "📊 Latest strategies:" + echo "$NEW_STRATS" | while read f; do + [ -f "$f" ] && log " - $(basename $f)" + done + fi + + # Wait before next attempt + log "⏳ Waiting 60s before next attempt..." + sleep 60 +done diff --git a/test/local/test_data_loader.py b/test/local/test_data_loader.py new file mode 100644 index 00000000..63cfb1dd --- /dev/null +++ b/test/local/test_data_loader.py @@ -0,0 +1,98 @@ +"""Tests for DataLoader.""" + +import pytest +import pandas as pd +import numpy as np +from pathlib import Path + +from rdagent.scenarios.qlib.local.data_loader import DataLoader, DataCache + + +class TestDataCache: + """Test thread-safe caching.""" + + def test_put_get(self): + cache = DataCache() + cache.put('key1', 'value1') + assert cache.get('key1') == 'value1' + + def test_cache_miss(self): + cache = DataCache() + assert cache.get('nonexistent') is None + + def test_clear(self): + cache = DataCache() + cache.put('key', 'value') + cache.clear() + assert cache.get('key') is None + + +class TestDataLoader: + """Test DataLoader functionality.""" + + @pytest.fixture + def loader(self): + return DataLoader() + + def test_load_ohlcv(self, loader): + close = loader.load_ohlcv() + assert isinstance(close, pd.Series) + assert len(close) > 0 + assert close.name == '$close' or close.name == 'close' + + def test_load_ohlcv_cached(self, loader): + # First call + close1 = loader.load_ohlcv() + len1 = len(close1) + + # Second call (should be cached) + close2 = loader.load_ohlcv() + assert len(close2) == len1 + + def test_load_ohlcv_max_bars(self, loader): + close = loader.load_ohlcv(max_bars=10000) + assert len(close) == 10000 + + def test_load_factor_metadata(self, loader): + factors = loader.load_factor_metadata(min_ic=0.0, top_n=20) + assert isinstance(factors, list) + assert len(factors) > 0 + assert 'name' in factors[0] + assert 'ic' in factors[0] + + def test_load_factor_metadata_sorted(self, loader): + factors = loader.load_factor_metadata(top_n=20) + ics = [abs(f['ic']) for f in factors] + assert ics == sorted(ics, reverse=True) + + def test_get_top_factors_randomized(self, loader): + factors1 = loader.get_top_factors_by_ic(top_n=10, randomize=True, seed=42) + factors2 = loader.get_top_factors_by_ic(top_n=10, randomize=True, seed=42) + factors3 = loader.get_top_factors_by_ic(top_n=10, randomize=True, seed=123) + + # Same seed should give same results + names1 = [f['name'] for f in factors1] + names2 = [f['name'] for f in factors2] + assert names1 == names2 + + # Different seed should give different results (likely) + names3 = [f['name'] for f in factors3] + # Not asserting different since it's probabilistic + + def test_build_feature_matrix(self, loader): + factors = loader.load_factor_metadata(top_n=5) + factor_names = [f['name'] for f in factors] + + close = loader.load_ohlcv(max_bars=10000) + df, index = loader.build_feature_matrix(factor_names, ohlcv_index=close.index) + + assert isinstance(df, pd.DataFrame) + assert len(df) > 0 + assert len(df.columns) <= len(factor_names) # Some might be dropped + + def test_clear_cache(self, loader): + loader.load_ohlcv() + loader.clear_cache() + # Cache should be empty (next call will reload) + close = loader.load_ohlcv() + assert len(close) > 0 diff --git a/watchdog_generator.sh b/watchdog_generator.sh new file mode 100755 index 00000000..5876ebfd --- /dev/null +++ b/watchdog_generator.sh @@ -0,0 +1,109 @@ +#!/bin/bash +# ============================================================================ +# PREDIX Strategy Generator Watchdog +# Checks every 20min: is the generator running? If not, (re)start it. +# ============================================================================ + +SCRIPT_DIR="/home/nico/Predix" +GENERATOR="python ${SCRIPT_DIR}/predix_smart_strategy_gen.py" +TARGET_COUNT=3 +LOGFILE="${SCRIPT_DIR}/results/logs/watchdog.log" +LOCKFILE="/tmp/predix_generator.lock" +MAX_ATTEMPTS=50 # Stop after this many attempts +PIDFILE="/tmp/predix_generator_attempt.pid" + +mkdir -p "${SCRIPT_DIR}/results/logs" + +log() { + echo "$(date '+%Y-%m-%d %H:%M:%S') - $1" >> "$LOGFILE" +} + +# Get current attempt count +get_attempt_count() { + if [ -f "$PIDFILE" ]; then + cat "$PIDFILE" + else + echo "0" + fi +} + +# Increment attempt count +increment_attempt() { + local current=$(get_attempt_count) + local next=$((current + 1)) + echo "$next" > "$PIDFILE" + echo "$next" +} + +# Check if generator is actually making progress +check_progress() { + local latest_log=$(ls -t ${SCRIPT_DIR}/results/logs/smart_strategy_gen_*.log 2>/dev/null | head -1) + if [ -n "$latest_log" ]; then + # Check if log was updated in last 5 minutes + local age=$(( $(date +%s) - $(stat -c %Y "$latest_log" 2>/dev/null || echo 0) )) + if [ $age -gt 300 ]; then + return 1 # Stale + fi + return 0 # Fresh + fi + return 1 # No log file +} + +# Kill any existing generator processes +cleanup() { + pkill -9 -f "predix_smart_strategy_gen.py" 2>/dev/null + rm -f "$LOCKFILE" + log "Cleaned up old processes" +} + +# Check if we've hit max attempts +if [ "$(get_attempt_count)" -ge "$MAX_ATTEMPTS" ]; then + log "MAX ATTEMPTS ($MAX_ATTEMPTS) reached. Stopping watchdog." + exit 0 +fi + +# Check if generator is running +if pgrep -f "predix_smart_strategy_gen.py" > /dev/null 2>&1; then + # Check if it's making progress + if check_progress; then + log "Generator is running and making progress. Exiting." + exit 0 + else + log "Generator is running but appears stalled. Restarting..." + cleanup + fi +else + log "Generator is NOT running. Starting..." + cleanup +fi + +# Increment attempt counter +ATTEMPT=$(increment_attempt) +log "=== Attempt $ATTEMPT / $MAX_ATTEMPTS ===" + +# Create lock file +echo $$ > "$LOCKFILE" + +# Start generator in background, capture PID +cd "$SCRIPT_DIR" +nohup $GENERATOR $TARGET_COUNT > /dev/null 2>&1 & +GEN_PID=$! + +log "Started generator with PID $GEN_PID" + +# Wait for process to finish (up to 20 min) +WAIT=0 +while kill -0 $GEN_PID 2>/dev/null; do + sleep 10 + WAIT=$((WAIT + 10)) + if [ $WAIT -ge 1200 ]; then # 20 min timeout + log "Generator timed out after 20 min. Killing." + kill -9 $GEN_PID 2>/dev/null + break + fi +done + +# Cleanup lock +rm -f "$LOCKFILE" + +log "Generator finished (or was killed). Exit code: $?"