chore: Organize utility scripts into scripts/ directory

Moved 13 scripts from root to scripts/:
- create_strategy.py
- debug_backtest.py
- predix_add_risk_management.py
- predix_batch_backtest.py
- predix_full_eval.py
- predix_gen_strategies_real_bt.py
- predix_parallel.py
- predix_quick_daytrading.py
- predix_rebacktest_strategies.py
- predix_simple_eval.py
- predix_smart_strategy_gen.py
- predix_strategy_report.py
- watchdog_generator.sh

Kept in root (intentional):
- predix.py (main entry point)
- start_llama.sh (convenience startup)
- start_strategy_loop.sh (convenience startup)

Root directory: 44 files → 31 files
This commit is contained in:
TPTBusiness
2026-04-09 14:45:49 +02:00
parent d99e065ea1
commit a7d1d97596
11 changed files with 0 additions and 0 deletions
+58
View File
@@ -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)")
+169
View File
@@ -0,0 +1,169 @@
#!/usr/bin/env python
"""Debug backtest logic: check alignment, signal quality, and IC calculation."""
import json
import numpy as np
import pandas as pd
from pathlib import Path
OHLCV_PATH = Path('/home/nico/Predix/git_ignore_folder/factor_implementation_source_data/intraday_pv.h5')
FACTORS_DIR = Path('/home/nico/Predix/results/factors')
VALUES_DIR = FACTORS_DIR / 'values'
print("=" * 70)
print("🔍 BACKTEST DEBUG - Alignment & IC Check")
print("=" * 70)
# 1. Load OHLCV close prices
print("\n1️⃣ Loading OHLCV close prices...")
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]
close = close.dropna()
print(f" Close prices: {len(close):,} bars")
print(f" Date range: {close.index.min()}{close.index.max()}")
print(f" Sample: {close.head(3).values}")
# Calculate forward returns (what we predict)
returns = close.pct_change().dropna()
print(f" Returns: {len(returns):,} values")
print(f" Return stats: mean={returns.mean():.8f}, std={returns.std():.8f}")
# 2. Load factor parquet files
print("\n2️⃣ Loading factor time-series...")
factor_files = sorted(VALUES_DIR.glob('*.parquet'))[:5] # Top 5
for ff in factor_files:
df_factor = pd.read_parquet(str(ff))
if len(df_factor.columns) > 0:
col = df_factor.iloc[:, 0]
# Align with close
common = close.index.intersection(col.dropna().index)
if len(common) > 0:
c = close.loc[common]
f = col.loc[common]
r = returns.loc[common] if len(returns) > 0 else pd.Series()
# Simple test: factor value vs next return
# IC = correlation(factor, forward_return)
fwd_r = r.shift(-1) # Next bar return
common2 = f.dropna().index.intersection(fwd_r.dropna().index)
if len(common2) > 100:
ic = f.loc[common2].corr(fwd_r.loc[common2])
print(f" {ff.stem:40s} len={len(f):>8,} IC_vs_next_return={ic:.6f}")
else:
print(f" {ff.stem:40s} len={len(f):>8,} (not enough common data)")
# 3. Test simple signals
print("\n3️⃣ Testing SIMPLE signals (what should work)...")
# Load metadata for best factors
factors_meta = []
for f in FACTORS_DIR.glob('*.json'):
try:
d = json.load(open(f))
ic = d.get('ic', 0) or 0
fname = d.get('factor_name', '')
if abs(ic) > 0.1:
factors_meta.append({'name': fname, 'ic': ic})
except:
pass
factors_meta.sort(key=lambda x: abs(x['ic']), reverse=True)
top = factors_meta[:3]
print(f" Top factors: {[(f['name'], f['ic']) for f in top]}")
# For each top factor, test simple signal
for fm in top:
fname = fm['name']
safe = fname.replace('/', '_').replace('\\', '_')[:150]
pf = VALUES_DIR / f"{safe}.parquet"
if not pf.exists():
print(f"{fname}: parquet not found")
continue
factor_series = pd.read_parquet(str(pf)).iloc[:, 0]
factor_ic = fm['ic']
# Align
common = close.index.intersection(factor_series.dropna().index)
if len(common) < 1000:
print(f"{fname}: not enough common data ({len(common)})")
continue
f = factor_series.loc[common]
r = returns.loc[common]
# Test A: Raw factor value vs next return
fwd = r.shift(-1)
common2 = f.index.intersection(fwd.dropna().index)
ic_raw = f.loc[common2].corr(fwd.loc[common2])
# Test B: Factor sign as signal (positive → LONG)
signal_a = (f > 0).astype(int).replace(0, -1) # +1 or -1
strat_ret_a = signal_a.shift(1) * r
sharpe_a = strat_ret_a.mean() / strat_ret_a.std() * np.sqrt(252*1440/96) if strat_ret_a.std() > 0 else 0
ic_signal = signal_a.corr(fwd)
# Test C: Factor percentile as signal
pct = f.rank(pct=True)
signal_c = (pct > 0.6).astype(int).replace(0, -1)
strat_ret_c = signal_c.shift(1) * r
sharpe_c = strat_ret_c.mean() / strat_ret_c.std() * np.sqrt(252*1440/96) if strat_ret_c.std() > 0 else 0
# Test D: What the LLM strategies actually compute (factor z-score → threshold)
w = 60
z = (f - f.rolling(w).mean()) / f.rolling(w).std()
z = z.fillna(0)
signal_d = (z > 0.5).astype(int).replace(0, -1)
strat_ret_d = signal_d.shift(1) * r
sharpe_d = strat_ret_d.mean() / strat_ret_d.std() * np.sqrt(252*1440/96) if strat_ret_d.std() > 0 else 0
print(f"\n 📊 {fname} (factor IC={factor_ic:.4f}):")
print(f" Raw factor IC vs fwd return: {ic_raw:.6f}")
print(f" Signal (sign) IC={ic_signal:.6f} Sharpe={sharpe_a:.4f}")
print(f" Signal (percentile) Sharpe={sharpe_c:.4f}")
print(f" Signal (z-score thresh) Sharpe={sharpe_d:.4f}")
# 4. Key finding
print("\n" + "=" * 70)
print("4️⃣ KEY INSIGHT:")
print("=" * 70)
# Check if factor values are already aligned with returns or are predictions
# Load one factor and check timing
if top:
fname = top[0]['name']
safe = fname.replace('/', '_').replace('\\', '_')[:150]
pf = VALUES_DIR / f"{safe}.parquet"
f = pd.read_parquet(str(pf)).iloc[:, 0]
# What does a HIGH factor value mean?
# If factor IC is positive (0.25), high values should predict positive returns
# Let's check: when factor is in top 10%, what's the average NEXT return?
common = close.index.intersection(f.dropna().index)
fv = f.loc[common]
rv = returns.loc[common]
fwd = rv.shift(-1)
common2 = fv.index.intersection(fwd.dropna().index)
fv2 = fv.loc[common2]
fwd2 = fwd.loc[common2]
top_decile = fv2 > fv2.quantile(0.9)
bot_decile = fv2 < fv2.quantile(0.1)
avg_ret_when_high = fwd2[top_decile].mean()
avg_ret_when_low = fwd2[bot_decile].mean()
print(f"\n Factor: {fname} (IC={top[0]['ic']:.4f})")
print(f" Avg NEXT return when factor in TOP 10%: {avg_ret_when_high*100:.6f}%")
print(f" Avg NEXT return when factor in BOT 10%: {avg_ret_when_low*100:.6f}%")
print(f" Difference: {(avg_ret_when_high - avg_ret_when_low)*100:.6f}%")
print(f"\n → If difference > 0, factor has predictive power")
print(f" → If difference ≈ 0, factor has NO predictive power for next-bar returns")
print(f" → If difference < 0, factor is INVERTED (use negative)")
+337
View File
@@ -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()
File diff suppressed because it is too large Load Diff
+658
View File
@@ -0,0 +1,658 @@
"""
Predix Full Data Factor Evaluator - Evaluate factors with FULL 1min data.
Evaluates factors using the complete intraday_pv.h5 dataset (2022-2026, ~2.26M rows)
instead of the debug dataset (2024 only, ~371K rows).
Usage:
python predix_full_eval.py --top 100 # Evaluate top 100 factors with full data
python predix_full_eval.py --all # Evaluate all factors
python predix_full_eval.py --parallel 4 # 4 parallel workers
"""
import json
import os
import sys
import warnings
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import List, Optional
import numpy as np
import pandas as pd
from rich.console import Console
from rich.progress import (
Progress,
SpinnerColumn,
TextColumn,
BarColumn,
TaskProgressColumn,
TimeElapsedColumn,
)
from rich.table import Table
from rich.panel import Panel
console = Console()
# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
PROJECT_ROOT = Path(__file__).resolve().parent
WORKSPACE_DIR = PROJECT_ROOT / "git_ignore_folder" / "RD-Agent_workspace"
# FULL data file (2022-2026, ~72MB)
FULL_DATA_FILE = PROJECT_ROOT / "git_ignore_folder" / "factor_implementation_source_data" / "intraday_pv.h5"
RESULTS_DIR = PROJECT_ROOT / "results"
BACKTESTS_DIR = RESULTS_DIR / "backtests"
DB_DIR = RESULTS_DIR / "db"
DB_PATH = DB_DIR / "backtest_results.db"
EVAL_SUMMARY_PATH = RESULTS_DIR / "eval_summary.json"
# Ensure directories exist
BACKTESTS_DIR.mkdir(parents=True, exist_ok=True)
DB_DIR.mkdir(parents=True, exist_ok=True)
# ---------------------------------------------------------------------------
# Data structures
# ---------------------------------------------------------------------------
@dataclass
class FactorInfo:
"""Factor information."""
workspace_hash: str
factor_name: str
factor_code: str
@dataclass
class EvalResult:
"""Evaluation result for a single factor."""
factor_name: str
workspace_hash: str
factor_code: str = ""
factor_description: str = ""
status: str = "" # success, failed
ic: Optional[float] = None
rank_ic: Optional[float] = None
sharpe: Optional[float] = None
annualized_return: Optional[float] = None
max_drawdown: Optional[float] = None
win_rate: Optional[float] = None
non_null_count: int = 0
total_count: int = 0
error_message: Optional[str] = None
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
def to_dict(self) -> dict:
return {k: v for k, v in self.__dict__.items()}
# ---------------------------------------------------------------------------
# Factor description extractor
# ---------------------------------------------------------------------------
def _extract_factor_description(code: str) -> str:
"""Extract docstring or description from factor code."""
import re
# Try to extract docstring
match = re.search(r'"""(.*?)"""', code, re.DOTALL)
if match:
return match.group(1).strip()[:500]
# Try to extract from comments
lines = code.split('\n')
desc_lines = []
for line in lines[:20]:
if line.strip().startswith('#') and not line.strip().startswith('#!'):
desc_lines.append(line.strip()[1:].strip())
if desc_lines:
return ' '.join(desc_lines)[:500]
return "No description available"
# ---------------------------------------------------------------------------
# Factor scanner
# ---------------------------------------------------------------------------
def scan_factors(workspace_dir: Path, skip_evaluated: bool = True) -> List[FactorInfo]:
"""Scan workspace directories for unique factor codes.
Parameters
----------
workspace_dir : Path
Path to workspace directory
skip_evaluated : bool
If True, skip factors that already have valid results in results/factors/
Returns
-------
List[FactorInfo]
List of factor information
"""
factors = []
seen_names = set()
# Load already evaluated factors (if skip_evaluated is True)
evaluated_factors = set()
if skip_evaluated:
project_root = Path(__file__).parent
factors_dir = project_root / "results" / "factors"
if factors_dir.exists():
import json
import glob
for f in glob.glob(str(factors_dir / "*.json")):
try:
with open(f) as fh:
data = json.load(fh)
if data.get("status") == "success" and data.get("ic") is not None:
evaluated_factors.add(data.get("factor_name"))
except Exception:
pass
print(f" Found {len(evaluated_factors)} already evaluated factors - skipping")
for ws in workspace_dir.iterdir():
if not ws.is_dir():
continue
factor_file = ws / "factor.py"
result_file = ws / "result.h5"
if not factor_file.exists():
continue
# Read factor name from result.h5
factor_name = None
if result_file.exists():
try:
result = pd.read_hdf(str(result_file), key="data")
if result is not None and len(result.columns) > 0:
factor_name = result.columns[0]
except Exception:
pass
if factor_name is None:
# Try to extract from code
code = factor_file.read_text()
import re
match = re.search(r'def calculate_(\w+)', code)
if match:
factor_name = match.group(1)
else:
factor_name = f"factor_{ws.name}"
# Skip duplicates
if factor_name in seen_names:
continue
# Skip already evaluated factors
if skip_evaluated and factor_name in evaluated_factors:
continue
seen_names.add(factor_name)
factors.append(FactorInfo(
workspace_hash=ws.name,
factor_name=factor_name,
factor_code=factor_file.read_text(),
))
return factors
# ---------------------------------------------------------------------------
# Factor evaluator
# ---------------------------------------------------------------------------
def evaluate_factor_full(factor: FactorInfo, full_data: pd.DataFrame,
forward_return_bars: int = 96) -> EvalResult:
"""
Evaluate a factor using the FULL dataset.
Parameters
----------
factor : FactorInfo
Factor information with code
full_data : pd.DataFrame
Full intraday_pv.h5 data
forward_return_bars : int
Number of bars for forward return calculation
Returns
-------
EvalResult
"""
import tempfile
import subprocess
with tempfile.TemporaryDirectory(prefix="predix_full_") as tmp_dir:
ws = Path(tmp_dir)
try:
# Copy full data to temp workspace
import shutil
shutil.copy(str(FULL_DATA_FILE), str(ws / "intraday_pv.h5"))
# Write factor code
(ws / "factor.py").write_text(factor.factor_code, encoding="utf-8")
# Execute factor code
proc = subprocess.run(
[sys.executable, str(ws / "factor.py")],
cwd=str(ws),
capture_output=True,
text=True,
timeout=120,
)
if proc.returncode != 0:
return EvalResult(
factor_name=factor.factor_name,
workspace_hash=factor.workspace_hash,
status="failed",
error_message=f"Execution failed: {proc.stderr[:300]}",
)
# Read result
result_file = ws / "result.h5"
if not result_file.exists():
return EvalResult(
factor_name=factor.factor_name,
workspace_hash=factor.workspace_hash,
status="failed",
error_message="No result.h5 generated",
)
result = pd.read_hdf(str(result_file), key="data")
total_count = len(result)
factor_val = result.iloc[:, 0]
non_null_count = factor_val.notna().sum()
if non_null_count < 1000:
return EvalResult(
factor_name=factor.factor_name,
workspace_hash=factor.workspace_hash,
status="failed",
non_null_count=non_null_count,
total_count=total_count,
error_message=f"Too few valid values: {non_null_count}",
)
# Compute forward returns
col_close = "$close"
if col_close not in full_data.columns:
col_close = next((c for c in full_data.columns if "close" in c.lower()), None)
if col_close is None:
return EvalResult(
factor_name=factor.factor_name,
workspace_hash=factor.workspace_hash,
status="failed",
error_message=f"No close column found",
)
close = full_data[col_close]
forward_ret = close.groupby(level="instrument").shift(-forward_return_bars) / close - 1
# Compute IC
valid_idx = factor_val.dropna().index.intersection(forward_ret.dropna().index)
if len(valid_idx) < 1000:
return EvalResult(
factor_name=factor.factor_name,
workspace_hash=factor.workspace_hash,
status="failed",
non_null_count=non_null_count,
total_count=total_count,
error_message=f"Too little overlap: {len(valid_idx)}",
)
with warnings.catch_warnings():
warnings.simplefilter("ignore")
ic = factor_val.loc[valid_idx].corr(forward_ret.loc[valid_idx])
rank_ic = factor_val.loc[valid_idx].corr(forward_ret.loc[valid_idx], method="spearman")
# Compute Sharpe
factor_mean = factor_val.loc[valid_idx].mean()
factor_std = factor_val.loc[valid_idx].std()
sharpe = factor_mean / factor_std if factor_std > 0 else 0
# Annualized return
ann_factor = np.sqrt(252 * 1440 / forward_return_bars)
annualized_return = float(factor_mean * ann_factor * 100)
# Max drawdown
cum_perf = factor_val.loc[valid_idx].cumsum()
running_max = cum_perf.expanding().max()
drawdown = (cum_perf - running_max) / running_max.replace(0, np.nan)
max_drawdown = float(drawdown.min()) if len(drawdown) > 0 else 0
# Win rate
win_rate = float((factor_val.loc[valid_idx] > 0).sum()) / len(valid_idx)
return EvalResult(
factor_name=factor.factor_name,
workspace_hash=factor.workspace_hash,
factor_code=factor.factor_code,
factor_description=_extract_factor_description(factor.factor_code),
status="success",
ic=float(ic) if ic is not None and not np.isnan(ic) else None,
rank_ic=float(rank_ic) if rank_ic is not None and not np.isnan(rank_ic) else None,
sharpe=float(sharpe) if sharpe is not None and not np.isnan(sharpe) else None,
annualized_return=annualized_return,
max_drawdown=max_drawdown,
win_rate=win_rate,
non_null_count=non_null_count,
total_count=total_count,
)
except subprocess.TimeoutExpired:
return EvalResult(
factor_name=factor.factor_name,
workspace_hash=factor.workspace_hash,
status="failed",
error_message="Execution timeout (120s)",
)
except Exception as e:
return EvalResult(
factor_name=factor.factor_name,
workspace_hash=factor.workspace_hash,
status="failed",
error_message=str(e)[:500],
)
# ---------------------------------------------------------------------------
# Parallel evaluation
# ---------------------------------------------------------------------------
def run_evaluation(
factors: List[FactorInfo],
full_data: pd.DataFrame,
n_workers: int = 4,
) -> List[EvalResult]:
"""Run factor evaluation in parallel using threads."""
results = []
with Progress(
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
BarColumn(),
TaskProgressColumn(),
TimeElapsedColumn(),
console=console,
) as progress:
task = progress.add_task(f"Evaluating {len(factors)} factors with FULL data...", total=len(factors))
with ThreadPoolExecutor(max_workers=n_workers) as executor:
futures = {executor.submit(evaluate_factor_full, f, full_data): f for f in factors}
for future in as_completed(futures):
factor = futures[future]
result = None
try:
result = future.result(timeout=300)
results.append(result)
except Exception as e:
# Handle timeout and other exceptions
fname = str(getattr(factor, 'factor_name', 'unknown'))[:40]
results.append(EvalResult(
factor_name=fname,
workspace_hash=getattr(factor, 'workspace_hash', 'unknown'),
status="failed",
error_message=f"Exception: {str(e)[:300]}",
))
result = results[-1]
n_success = sum(1 for r in results if r.status == "success")
n_fail = sum(1 for r in results if r.status == "failed")
# Save immediately after each factor
if result is not None:
save_single_result(result)
# Update progress with safe string handling
fname = str(getattr(factor, 'factor_name', 'unknown'))[:40]
progress.update(
task,
advance=1,
description=f"Evaluating: {n_success}{n_fail}❌ | {fname}",
)
return results
# ---------------------------------------------------------------------------
# Results storage
# ---------------------------------------------------------------------------
FACTORS_DIR = RESULTS_DIR / "factors"
FACTORS_DIR.mkdir(parents=True, exist_ok=True)
def save_single_result(r: EvalResult) -> None:
"""Save a single factor result to results/factors/."""
if r.status != "success":
return
safe_name = r.factor_name.replace("/", "_").replace("\\", "_").replace(" ", "_")[:100]
json_path = FACTORS_DIR / f"{safe_name}.json"
with open(json_path, "w", encoding="utf-8") as f:
json.dump(r.to_dict(), f, indent=2, default=str)
def save_results(results: List[EvalResult]) -> None:
"""Save evaluation results to JSON and SQLite."""
successful = [r for r in results if r.status == "success"]
failed = [r for r in results if r.status == "failed"]
# Sort by IC
successful.sort(key=lambda r: abs(r.ic) if r.ic is not None else 0, reverse=True)
# Save ALL successful results to results/factors/
for r in successful:
# Safe filename (remove special chars)
safe_name = r.factor_name.replace("/", "_").replace("\\", "_").replace(" ", "_")[:100]
json_path = FACTORS_DIR / f"{safe_name}.json"
with open(json_path, "w", encoding="utf-8") as f:
json.dump(r.to_dict(), f, indent=2, default=str)
# Save summary
valid_ic = [r.ic for r in results if r.ic is not None]
valid_sharpe = [r.sharpe for r in results if r.sharpe is not None]
summary = {
"generated_at": datetime.now().isoformat(),
"total_evaluated": len(results),
"successful": len(successful),
"failed": len(failed),
"success_rate": len(successful) / len(results) if results else 0,
"avg_ic": float(np.mean(valid_ic)) if valid_ic else 0,
"best_ic": float(max(valid_ic, key=abs, default=0)),
"avg_sharpe": float(np.mean(valid_sharpe)) if valid_sharpe else 0,
"best_sharpe": float(max(valid_sharpe, default=0)),
"top_20_by_ic": [r.to_dict() for r in successful[:20]],
"all_results": [r.to_dict() for r in results],
}
with open(EVAL_SUMMARY_PATH, "w", encoding="utf-8") as f:
json.dump(summary, f, indent=2, default=str)
# Save to SQLite
try:
import sqlite3
conn = sqlite3.connect(str(DB_PATH))
c = conn.cursor()
c.execute("""CREATE TABLE IF NOT EXISTS factor_evaluations (
id INTEGER PRIMARY KEY,
factor_name TEXT,
workspace_hash TEXT,
ic REAL,
rank_ic REAL,
sharpe REAL,
annualized_return REAL,
max_drawdown REAL,
win_rate REAL,
non_null_count INTEGER,
total_count INTEGER,
status TEXT,
timestamp TEXT
)""")
for r in results:
c.execute("""INSERT INTO factor_evaluations
(factor_name, workspace_hash, ic, rank_ic, sharpe,
annualized_return, max_drawdown, win_rate,
non_null_count, total_count, status, timestamp)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(r.factor_name, r.workspace_hash, r.ic, r.rank_ic, r.sharpe,
r.annualized_return, r.max_drawdown, r.win_rate,
r.non_null_count, r.total_count, r.status, r.timestamp))
conn.commit()
conn.close()
except Exception as e:
console.print(f"[yellow]SQLite save warning: {e}[/yellow]")
# ---------------------------------------------------------------------------
# Display
# ---------------------------------------------------------------------------
def display_results(results: List[EvalResult]) -> None:
"""Display evaluation results as a table."""
successful = [r for r in results if r.status == "success"]
successful.sort(key=lambda r: abs(r.ic) if r.ic is not None else 0, reverse=True)
table = Table(
title="Factor Evaluation Results (FULL DATA)",
show_header=True,
header_style="bold cyan",
)
table.add_column("#", justify="center", width=4)
table.add_column("Factor", width=40)
table.add_column("IC", justify="right", width=10)
table.add_column("Rank IC", justify="right", width=10)
table.add_column("Sharpe", justify="right", width=10)
table.add_column("Ann. Ret %", justify="right", width=10)
table.add_column("Max DD", justify="right", width=10)
table.add_column("Win Rate", justify="right", width=10)
for i, r in enumerate(successful[:20], 1):
table.add_row(
str(i),
r.factor_name[:38],
f"{r.ic:.6f}" if r.ic is not None else "N/A",
f"{r.rank_ic:.6f}" if r.rank_ic is not None else "N/A",
f"{r.sharpe:.4f}" if r.sharpe is not None else "N/A",
f"{r.annualized_return:.4f}" if r.annualized_return is not None else "N/A",
f"{r.max_drawdown:.4f}" if r.max_drawdown is not None else "N/A",
f"{r.win_rate:.2%}" if r.win_rate is not None else "N/A",
)
console.print()
console.print(table)
# Summary
valid_ic = [r.ic for r in results if r.ic is not None]
valid_sharpe = [r.sharpe for r in results if r.sharpe is not None]
console.print(Panel(
f"[bold]Evaluation Summary (FULL DATA)[/bold]\n"
f"Total evaluated: {len(results)}\n"
f"Successful: {len(successful)}\n"
f"Failed: {len(results) - len(successful)}\n"
f"Avg IC: {np.mean(valid_ic):.6f} (n={len(valid_ic)})\n"
f"Best IC: {max(valid_ic, key=abs, default=0):.6f}\n"
f"Avg Sharpe: {np.mean(valid_sharpe):.4f} (n={len(valid_sharpe)})\n"
f"Best Sharpe: {max(valid_sharpe, default=0):.4f}\n"
f"Saved to: {EVAL_SUMMARY_PATH}\n"
f"Database: {DB_PATH}",
border_style="green",
))
# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
def main(
top: int = 100,
all_factors: bool = False,
parallel: int = 4,
force: bool = False,
) -> None:
"""Main entry point."""
console.print(Panel(
"[bold cyan]Predix Full Data Factor Evaluator[/bold cyan]\n"
f"Using FULL 1min data: {FULL_DATA_FILE}",
border_style="cyan",
))
# Load full data
if not FULL_DATA_FILE.exists():
console.print(f"[red]Full data file not found: {FULL_DATA_FILE}[/red]")
return
console.print(f"\n[dim]Loading full data...[/dim]")
full_data = pd.read_hdf(str(FULL_DATA_FILE), key="data")
console.print(f"[bold green]✓ Loaded {len(full_data):,} rows ({full_data.index.get_level_values('datetime').min()} to {full_data.index.get_level_values('datetime').max()})[/bold green]")
# Scan factors (skip already evaluated by default)
console.print(f"\n[dim]Scanning workspaces...[/dim]")
factors = scan_factors(WORKSPACE_DIR, skip_evaluated=not force)
console.print(f"[bold]Total unique factors found: {len(factors)}[/bold]")
if force:
console.print("[yellow]⚠️ Force mode: Re-evaluating ALL factors[/yellow]")
else:
console.print("[dim]Skipping already evaluated factors[/dim]")
if not factors:
console.print("[red]No factors found![/red]")
return
# Select factors to evaluate
if all_factors:
to_evaluate = factors
else:
to_evaluate = factors[:top]
console.print(f"\n[bold green]Selected {len(to_evaluate)} factors for evaluation[/bold green]")
console.print(f" Using {parallel} parallel workers")
# Run evaluation
results = run_evaluation(to_evaluate, full_data, n_workers=parallel)
# Save results
console.print(f"\n[bold cyan]Saving results...[/bold cyan]")
save_results(results)
# Display
display_results(results)
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(
description="Predix Full Data Factor Evaluator"
)
parser.add_argument(
"--top", "-n",
type=int,
default=100,
help="Number of factors to evaluate (default: 100)",
)
parser.add_argument(
"--all", "-a",
action="store_true",
help="Evaluate all discovered factors",
)
parser.add_argument(
"--parallel", "-p",
type=int,
default=4,
help="Number of parallel workers (default: 4)",
)
parser.add_argument(
"--force", "-f",
action="store_true",
help="Force re-evaluation of ALL factors (even already evaluated)",
)
args = parser.parse_args()
main(
top=args.top,
all_factors=args.all,
parallel=args.parallel,
force=args.force,
)
+471
View File
@@ -0,0 +1,471 @@
#!/usr/bin/env python
"""
Parallel AI Strategy Generation with REAL OHLCV Backtest.
Generates multiple trading strategies in parallel using LLM,
each with real backtesting on OHLCV data.
Usage:
# 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 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 rich.console import Console
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TimeElapsedColumn
from dotenv import load_dotenv
# Suppress warnings
warnings.filterwarnings('ignore')
logging.getLogger('rdagent').setLevel(logging.WARNING)
# ============================================================================
# Configuration
# ============================================================================
OHLCV_PATH = Path('/home/nico/Predix/git_ignore_folder/factor_implementation_source_data/intraday_pv.h5')
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
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'))
MIN_IC = 0.02
MIN_SHARPE = 0.5
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'))
MIN_IC = 0.02
MIN_SHARPE = 0.5
MIN_TRADES = 10
MAX_DRAWDOWN = -1.0
STYLE_EMOJI = '📈 Swing'
STYLE_DESC = 'medium-term intraday'
console = Console()
# ============================================================================
# LLM Configuration (Process-safe)
# ============================================================================
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/google/gemma-4-26b-a4b-it:free')
# ============================================================================
# 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."""
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})
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_ohlcv_data():
"""Load OHLCV close prices."""
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']
else:
close = ohlcv.select_dtypes(include=[np.number]).iloc[:, 0]
_OHLCV_CACHE = close.dropna()
return _OHLCV_CACHE
# ============================================================================
# Strategy Generation (LLM call - runs in separate process)
# ============================================================================
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':
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):
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'
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 trading strategy using these factors:
{factor_list}
{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
)
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:
return {'status': 'error', 'reason': str(e)[:200], 'idx': idx}
# ============================================================================
# Backtest Runner (runs in main process to avoid re-loading data)
# ============================================================================
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
import tempfile
script = f"""
import pandas as pd
import numpy as np
import json
close = pd.read_pickle('close.pkl')
factors = pd.read_pickle('factors.pkl')
try:
{chr(10).join(' ' + l for l in strategy_code.split(chr(10)))}
except:
print("ERROR: Strategy execution failed")
exit(1)
if 'signal' not in dir():
print("ERROR: No signal generated")
exit(1)
signal = signal.fillna(0)
common_idx = close.index.intersection(signal.index)
close = close.loc[common_idx]
signal = signal.loc[common_idx]
FORWARD_BARS = {FORWARD_BARS}
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)
ic = signal_aligned.corr(fwd_returns)
strategy_returns = signal_aligned * fwd_returns
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 / {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
annual_return = (1 + total_return) ** (12 / n_months) - 1
else:
monthly_return = total_return
annual_return = total_return * 12
result = {{
"status": "success",
"sharpe": float(sharpe),
"max_drawdown": float(max_dd) if not np.isnan(max_dd) else -0.20,
"win_rate": float(win_rate),
"ic": float(ic) if not np.isnan(ic) else 0,
"n_trades": n_trades,
"total_return": float(total_return),
"monthly_return_pct": float(monthly_return * 100),
"annual_return_pct": float(annual_return * 100),
"n_bars": int(n_bars),
"n_months": float(n_months),
"signal_long": int((signal_aligned == 1).sum()),
"signal_short": int((signal_aligned == -1).sum()),
"signal_neutral": int((signal_aligned == 0).sum()),
}}
print(json.dumps(result))
"""
with tempfile.TemporaryDirectory() as td:
tdp = Path(td)
close.to_pickle(str(tdp / 'close.pkl'))
factors_df.to_pickle(str(tdp / 'factors.pkl'))
script_path = tdp / 'run.py'
script_path.write_text(script)
try:
result = subprocess.run(
['python', str(script_path)],
capture_output=True, text=True, timeout=60,
cwd=str(tdp)
)
if result.returncode != 0:
return {'status': 'failed', 'reason': result.stderr[:200] or result.stdout[:200]}
for line in result.stdout.strip().split('\n'):
try:
return json.loads(line)
except:
continue
return {'status': 'failed', 'reason': 'No valid JSON output'}
except subprocess.TimeoutExpired:
return {'status': 'failed', 'reason': 'Timeout (60s)'}
except Exception as e:
return {'status': 'failed', 'reason': str(e)[:200]}
# ============================================================================
# Main Parallel Strategy Generation
# ============================================================================
def main(target_count=10):
"""Generate strategies in parallel with real backtesting."""
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
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]
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(accepted) >= target_count:
break
# Select random factor subset (2-5 factors)
n_factors = random.randint(2, min(5, len(factors)))
factor_subset = random.sample(factors, n_factors)
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
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
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)
# 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())}_{strategy['strategy_name']}.json"
with open(STRATEGIES_DIR / fname, 'w') as f:
json.dump(strategy, f, indent=2, ensure_ascii=False)
# Generate PDF report
try:
from predix_strategy_report import StrategyPerformanceReporter
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_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.update(task, advance=1)
# Summary
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)
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__':
count = int(sys.argv[1]) if len(sys.argv) > 1 else 10
main(count)
+554
View File
@@ -0,0 +1,554 @@
"""
Predix Parallel Runner - Run multiple factor experiments concurrently.
Spawns N subprocesses, each running `predix.py quant` with isolated config:
- Separate log files (fin_quant_run1.log, fin_quant_run2.log, etc.)
- Separate result directories (results/runs/run1/, results/runs/run2/, etc.)
- Separate workspace directories
- API key distribution across multiple keys (round-robin)
Usage:
python predix_parallel.py --runs 5 --api-keys 2
python predix_parallel.py --runs 3 --model openrouter
python predix_parallel.py --runs 5 --model local --api-keys 1
"""
import os
import signal
import subprocess
import sys
import time
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional
from dotenv import load_dotenv
from rich.console import Console
from rich.live import Live
from rich.panel import Panel
from rich.table import Table
from rich.markdown import Markdown
from rich.layout import Layout
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TaskProgressColumn
# Load environment variables from .env file
load_dotenv(Path(__file__).parent / ".env")
console = Console()
class RunState:
"""Tracks the state of a single parallel run."""
def __init__(self, run_id: int, api_key_idx: int, model: str):
self.run_id = run_id
self.api_key_idx = api_key_idx
self.model = model
self.process: Optional[subprocess.Popen] = None
self.status: str = "pending" # pending, running, success, failed, stopped
self.start_time: Optional[datetime] = None
self.end_time: Optional[datetime] = None
self.exit_code: Optional[int] = None
self.error_message: Optional[str] = None
self.log_file: str = f"fin_quant_run{run_id}.log"
@property
def elapsed(self) -> str:
"""Get elapsed time as human-readable string."""
if self.start_time is None:
return "--:--:--"
end = self.end_time or datetime.now()
delta = end - self.start_time
total_seconds = int(delta.total_seconds())
hours, remainder = divmod(total_seconds, 3600)
minutes, seconds = divmod(remainder, 60)
return f"{hours:02d}:{minutes:02d}:{seconds:02d}"
@property
def status_icon(self) -> str:
"""Get icon for current status."""
icons = {
"pending": "",
"running": "🔄",
"success": "",
"failed": "",
"stopped": "⏹️",
}
return icons.get(self.status, "")
class ParallelRunner:
"""
Manages multiple concurrent factor experiment runs.
Spawns subprocesses with isolated configurations, monitors progress,
and handles graceful shutdown.
"""
def __init__(
self,
num_runs: int = 5,
num_api_keys: int = 2,
model: str = "openrouter",
):
"""
Initialize parallel runner.
Parameters
----------
num_runs : int
Number of concurrent runs to spawn
num_api_keys : int
Number of API keys to distribute across (1 or 2)
model : str
LLM backend: 'local' (llama.cpp) or 'openrouter' (cloud)
"""
self.num_runs = num_runs
self.num_api_keys = num_api_keys
self.model = model
self.runs: List[RunState] = []
self.project_root = Path(__file__).parent
self._shutdown_requested = False
# Read API keys from environment
self.api_keys = self._load_api_keys()
# Validate we have enough API keys
if self.model == "openrouter" and len(self.api_keys) < num_api_keys:
console.print(
f"[yellow]⚠️ Requested {num_api_keys} API keys, but only {len(self.api_keys)} found in .env[/yellow]"
)
console.print(
f"[dim]Distributing across {len(self.api_keys)} available key(s)[/dim]"
)
self.num_api_keys = len(self.api_keys)
# Initialize run states
for i in range(1, num_runs + 1):
# Round-robin API key assignment
api_key_idx = (i - 1) % max(self.num_api_keys, 1)
run_state = RunState(run_id=i, api_key_idx=api_key_idx, model=model)
self.runs.append(run_state)
def _load_api_keys(self) -> List[str]:
"""Load API keys from environment variables."""
keys = []
if self.model == "openrouter":
key1 = os.getenv("OPENROUTER_API_KEY", "")
key2 = os.getenv("OPENROUTER_API_KEY_2", "")
if key1:
keys.append(key1)
if key2:
keys.append(key2)
else:
# For local mode, we just need the llama.cpp endpoint
keys.append("local")
if not keys or (len(keys) == 1 and keys[0] == "local"):
keys = ["local"]
return keys
def _build_env(self, run_state: RunState) -> Dict[str, str]:
"""
Build isolated environment for a subprocess.
Parameters
----------
run_state : RunState
The run state object containing run configuration
Returns
-------
dict
Environment variables dict for subprocess
"""
# Start with a copy of current environment
env = os.environ.copy()
# Set parallel run ID for isolation
env["PARALLEL_RUN_ID"] = str(run_state.run_id)
# Set workspace isolation
workspace_dir = self.project_root / f"RD-Agent_workspace_run{run_state.run_id}"
env["RD_AGENT_WORKSPACE"] = str(workspace_dir)
# Configure API key for this run
if self.model == "openrouter" and run_state.api_key_idx < len(self.api_keys):
api_key = self.api_keys[run_state.api_key_idx]
env["OPENAI_API_KEY"] = api_key
env["OPENAI_API_BASE"] = "https://openrouter.ai/api/v1"
env["CHAT_MODEL"] = os.getenv("OPENROUTER_MODEL", "openrouter/google/gemma-4-26b-a4b-it:free")
# If we configured multiple API keys AND have enough keys, use load balancing
if self.num_api_keys >= 2 and len(self.api_keys) >= 2:
env["OPENAI_API_KEY"] = f"{self.api_keys[0]},{self.api_keys[1]}"
env["LITELLM_PARALLEL_CALLS"] = "2"
elif self.model == "local":
env["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "local")
env["OPENAI_API_BASE"] = os.getenv("OPENAI_API_BASE", "http://localhost:8081/v1")
env["CHAT_MODEL"] = os.getenv("CHAT_MODEL", "openai/qwen3.5-35b")
return env
def _build_command(self, run_state: RunState) -> List[str]:
"""
Build the subprocess command to run predix quant.
Parameters
----------
run_state : RunState
The run state object containing run configuration
Returns
-------
list
Command list for subprocess.Popen
"""
cmd = [
sys.executable, # Use same Python interpreter
str(self.project_root / "predix.py"),
"quant",
"--model", run_state.model,
"--run-id", str(run_state.run_id),
"--log-file", run_state.log_file,
]
return cmd
def _start_run(self, run_state: RunState) -> None:
"""
Start a single run as a subprocess.
Parameters
----------
run_state : RunState
The run state to start
"""
env = self._build_env(run_state)
cmd = self._build_command(run_state)
# Ensure results directory exists
results_dir = self.project_root / "results" / "runs" / f"run{run_state.run_id}"
results_dir.mkdir(parents=True, exist_ok=True)
# Open log file for appending
log_path = self.project_root / run_state.log_file
log_f = open(log_path, "a", encoding="utf-8")
# Start subprocess
run_state.process = subprocess.Popen(
cmd,
env=env,
cwd=str(self.project_root),
stdout=log_f,
stderr=subprocess.STDOUT,
)
run_state.status = "running"
run_state.start_time = datetime.now()
console.print(
f"[dim] ▶️ Run {run_state.run_id} started (PID: {run_state.process.pid}, "
f"API Key: {run_state.api_key_idx + 1}, Model: {run_state.model})[/dim]"
)
def _check_run(self, run_state: RunState) -> None:
"""
Check if a run is still running and update status.
Parameters
----------
run_state : RunState
The run state to check
"""
if run_state.status != "running" or run_state.process is None:
return
poll_result = run_state.process.poll()
if poll_result is not None:
# Process has finished
run_state.exit_code = poll_result
run_state.end_time = datetime.now()
if poll_result == 0:
run_state.status = "success"
console.print(
f"[bold green] ✅ Run {run_state.run_id} completed "
f"({run_state.elapsed})[/bold green]"
)
else:
run_state.status = "failed"
run_state.error_message = f"Exit code: {poll_result}"
console.print(
f"[bold red] ❌ Run {run_state.run_id} failed "
f"({run_state.elapsed}, exit code: {poll_result})[/bold red]"
)
def _stop_run(self, run_state: RunState) -> None:
"""
Gracefully stop a running subprocess.
Parameters
----------
run_state : RunState
The run state to stop
"""
if run_state.process is None or run_state.status != "running":
return
try:
# Try graceful termination first
run_state.process.terminate()
try:
run_state.process.wait(timeout=10)
except subprocess.TimeoutExpired:
# Force kill if not responding
run_state.process.kill()
run_state.process.wait()
except Exception as e:
console.print(f"[yellow] ⚠️ Error stopping run {run_state.run_id}: {e}[/yellow]")
run_state.status = "stopped"
run_state.end_time = datetime.now()
def _render_dashboard(self) -> Panel:
"""
Render the live dashboard panel showing all run states.
Returns
-------
Panel
Rich Panel object with dashboard content
"""
# Summary stats
pending = sum(1 for r in self.runs if r.status == "pending")
running = sum(1 for r in self.runs if r.status == "running")
success = sum(1 for r in self.runs if r.status == "success")
failed = sum(1 for r in self.runs if r.status == "failed")
stopped = sum(1 for r in self.runs if r.status == "stopped")
# Build summary table
table = Table(
title="🔀 Predix Parallel Run Dashboard",
show_header=True,
header_style="bold cyan",
expand=True,
)
table.add_column("Run", justify="center", width=6)
table.add_column("Status", justify="center", width=10)
table.add_column("Elapsed", justify="center", width=10)
table.add_column("API Key", justify="center", width=8)
table.add_column("Model", justify="center", width=12)
table.add_column("Exit", justify="center", width=6)
table.add_column("Log File", justify="left")
for run in self.runs:
table.add_row(
f"#{run.run_id}",
f"{run.status_icon} {run.status}",
run.elapsed,
str(run.api_key_idx + 1),
run.model,
str(run.exit_code) if run.exit_code is not None else "--",
run.log_file,
)
# Summary panel
total = len(self.runs)
summary_text = (
f"**Summary:** {total} total | "
f"{success} done | "
f"{running} running | "
f"{pending} pending | "
f"{failed} failed"
)
if self._shutdown_requested:
summary_text += "\n⚠️ **Shutdown requested - stopping all runs...**"
from rich.console import Group
return Group(table, Panel(Markdown(summary_text), border_style="blue"))
def _signal_handler(self, signum, frame) -> None:
"""Handle SIGINT/SIGTERM for graceful shutdown."""
if self._shutdown_requested:
# Second Ctrl+C - force kill everything
console.print("\n[bold red]🛑 Force killing all runs![/bold red]")
for run in self.runs:
if run.process and run.status == "running":
run.process.kill()
sys.exit(1)
self._shutdown_requested = True
console.print("\n[yellow]⏹️ Shutdown requested - gracefully stopping all runs...[/yellow]")
console.print("[dim]Press Ctrl+C again to force kill[/dim]")
for run in self.runs:
if run.status == "running":
self._stop_run(run)
def run(self) -> Dict[str, int]:
"""
Execute all parallel runs and show live dashboard.
Returns
-------
dict
Summary with keys: total, success, failed, stopped
"""
# Register signal handlers
signal.signal(signal.SIGINT, self._signal_handler)
signal.signal(signal.SIGTERM, self._signal_handler)
console.print(f"\n[bold cyan]{'=' * 60}[/bold cyan]")
console.print(f"[bold cyan]🔀 Predix Parallel Runner[/bold cyan]")
console.print(f"[bold cyan]{'=' * 60}[/bold cyan]")
console.print(f" Runs: {self.num_runs}")
console.print(f" API Keys: {self.num_api_keys} ({len(self.api_keys)} available)")
console.print(f" Model: {self.model}")
console.print(f" Log pattern: fin_quant_run{{1..{self.num_runs}}}.log")
console.print(f" Results: results/runs/run{{1..{self.num_runs}}}/")
console.print()
# Start all runs
for run in self.runs:
if self._shutdown_requested:
break
self._start_run(run)
# Small delay to prevent overwhelming the system
time.sleep(1)
# Monitor loop with live dashboard
with Live(refresh_per_second=2, screen=True) as live:
live.update(self._render_dashboard())
while True:
if self._shutdown_requested:
# Check if all runs are stopped
all_stopped = all(
r.status in ("success", "failed", "stopped", "pending")
for r in self.runs
)
if all_stopped:
break
# Update all run statuses
for run in self.runs:
self._check_run(run)
# Check if all runs are complete
all_done = all(
r.status in ("success", "failed", "stopped")
for r in self.runs
)
if all_done:
break
live.update(self._render_dashboard())
time.sleep(0.5)
# Final summary
success_count = sum(1 for r in self.runs if r.status == "success")
failed_count = sum(1 for r in self.runs if r.status == "failed")
stopped_count = sum(1 for r in self.runs if r.status == "stopped")
console.print(f"\n[bold cyan]{'=' * 60}[/bold cyan]")
console.print(f"[bold cyan]📊 Parallel Run Summary[/bold cyan]")
console.print(f"[bold cyan]{'=' * 60}[/bold cyan]")
console.print(f" ✅ Success: {success_count}/{self.num_runs}")
console.print(f" ❌ Failed: {failed_count}/{self.num_runs}")
if stopped_count > 0:
console.print(f" ⏹️ Stopped: {stopped_count}/{self.num_runs}")
total_time = None
for run in self.runs:
if run.start_time and run.end_time:
delta = run.end_time - run.start_time
console.print(
f" Run #{run.run_id}: {run.status} ({delta.total_seconds():.0f}s)"
)
return {
"total": self.num_runs,
"success": success_count,
"failed": failed_count,
"stopped": stopped_count,
}
def main(
runs: int = 5,
api_keys: int = 2,
model: str = "openrouter",
) -> Dict[str, int]:
"""
Run multiple factor experiments in parallel.
Parameters
----------
runs : int
Number of concurrent runs to spawn
api_keys : int
Number of API keys to distribute across
model : str
LLM backend: 'local' (llama.cpp) or 'openrouter' (cloud)
Returns
-------
dict
Summary with keys: total, success, failed, stopped
"""
runner = ParallelRunner(num_runs=runs, num_api_keys=api_keys, model=model)
return runner.run()
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(
description="Predix Parallel Runner - Run multiple factor experiments concurrently"
)
parser.add_argument(
"--runs", "-n",
type=int,
default=5,
help="Number of concurrent runs (default: 5, max recommended: 25)",
)
parser.add_argument(
"--api-keys", "-k",
type=int,
default=2,
help="Number of API keys to distribute across (default: 2)",
)
parser.add_argument(
"--model", "-m",
type=str,
default="openrouter",
choices=["local", "openrouter"],
help="LLM backend: 'local' (llama.cpp) or 'openrouter' (cloud)",
)
parser.add_argument(
"--force",
action="store_true",
help="Skip resource warnings (allow >25 runs)",
)
args = parser.parse_args()
# Resource warnings for high run counts
if args.runs > 50 and not args.force:
console.print(f"\n[bold red]⚠️ {args.runs} runs exceeds safe limit (50)[/bold red]")
console.print("[yellow]This will likely cause memory exhaustion and API throttling.[/yellow]")
console.print("[yellow]Use --force to override.[/yellow]")
sys.exit(1)
elif args.runs > 25:
console.print(f"\n[yellow]⚠️ {args.runs} runs - high resource usage expected[/yellow]")
console.print(f" Estimated RAM: ~{args.runs * 0.65:.0f} GB")
console.print(f" Use --force to confirm.\n")
import time
time.sleep(2)
result = main(runs=args.runs, api_keys=args.api_keys, model=args.model)
# Exit with appropriate code
if result["failed"] > 0:
sys.exit(1)
sys.exit(0)
+165
View File
@@ -0,0 +1,165 @@
#!/usr/bin/env python
"""Re-evaluate strategies with real backtests - robust version."""
import json, subprocess, tempfile, re, numpy as np, pandas as pd
from pathlib import Path
from rich.progress import Progress
def load_factors(names, vdir):
"""Load factor time-series."""
dfs = {}
for n in names:
for v in [n, n.replace('/','_').replace('\\','_')[:150], n.replace('.','_')[:150]]:
p = vdir / f"{v}.parquet"
if p.exists():
try:
df = pd.read_parquet(str(p))
if df is not None and len(df.columns) > 0:
dfs[n] = df.iloc[:, 0]
break
except: pass
return dfs
def fix_code(code, available):
"""Fix strategy code to handle missing factors."""
fixed = code
# Fix: df['missing_factor'] → pd.Series(0, index=df.index)
for match in re.finditer(r"df\['([^']+)'\]", code):
fname = match.group(1)
if fname not in available:
fixed = fixed.replace(
f"df['{fname}']",
f"pd.Series(0, index=df.index, name='{fname}')", 1
)
# Fix: df[["f1", "f2"]] → filter to available only
for match in re.finditer(r'df\[\[([^\]]+)\]\]', code):
factors_str = match.group(1)
factors = [f.strip().strip("'\"") for f in factors_str.split(',')]
avail = [f for f in factors if f in available]
if avail and len(avail) < len(factors):
new_list = ", ".join(f"'{f}'" for f in avail)
fixed = fixed.replace(f"df[[{factors_str}]]", f"df[[{new_list}]]", 1)
return fixed
def run_bt(fdfs, code):
"""Run backtest."""
df = pd.DataFrame(fdfs).dropna()
if len(df) < 100 or len(df.columns) < 2:
return None
avail = list(df.columns)
fixed = fix_code(code, avail)
with tempfile.TemporaryDirectory() as td:
tdp = Path(td)
df.to_parquet(str(tdp / "factors.parquet")) # MUST be named factors.parquet
script = tdp / "run.py"
script.write_text(f"""
import pandas as pd, numpy as np
df = pd.read_parquet('factors.parquet')
try:
{chr(10).join(' ' + l for l in fixed.split(chr(10)))}
except Exception as e:
pass
try:
if 'signal' not in dir():
signal = pd.Series(np.where(df.mean(axis=1) > 0, 1, -1), index=df.index)
signal.name = 'signal'
signal.to_pickle('s.pkl')
print("OK")
except Exception as e:
print(f"ERROR: {{e}}")
""")
try:
r = subprocess.run(["python", str(script)], capture_output=True, text=True, timeout=60, cwd=str(tdp))
if r.returncode != 0:
return None
sig = pd.read_pickle(str(tdp / "s.pkl"))
except:
return None
fwd = df.mean(axis=1).shift(-96).dropna()
sig = sig.loc[fwd.index]
if len(sig) < 100: return None
ic = sig.corr(fwd)
rets = sig * fwd
std = rets.std()
sharpe = rets.mean()/std * np.sqrt(252*1440/96) if std > 0 and not np.isnan(std) else 0
sharpe = min(max(sharpe, -5), 5)
cum = (1+rets).cumprod().replace([np.inf,-np.inf], np.nan).fillna(1)
dd = ((cum - cum.cummax())/cum.cummax().replace(0, np.nan)).min()
mdd = min(max(dd if not np.isnan(dd) else -0.20, -1.0), 0.0)
wr = (rets>0).sum()/len(rets)
trades = int((sig != sig.shift(1)).sum())
tot = cum.iloc[-1] - 1
if np.isnan(tot) or np.isinf(tot): tot = 0
tot = max(min(tot, 1.0), -0.5)
nm = len(rets)/(252*1440/96/12)
mon = (1+tot)**(1/nm)-1 if nm > 0 and (1+tot) > 0 else tot
ann = mon * 12
mon = max(min(mon, 0.20), -0.20)
ann = max(min(ann, 2.0), -1.0)
ic = ic if not np.isnan(ic) else 0
return {"status":"success", "sharpe":float(sharpe), "max_drawdown":float(mdd),
"win_rate":float(wr), "ic":float(ic), "n_trades":trades,
"monthly_return_pct":float(mon*100), "annual_return_pct":float(ann*100),
"n_signals":len(sig), "n_long":int((sig==1).sum()),
"n_short":int((sig==-1).sum()), "n_neutral":int((sig==0).sum())}
def main(count=None):
sdir = Path('/home/nico/Predix/results/strategies')
vdir = Path('/home/nico/Predix/results/factors/values')
files = []
for f in sorted(sdir.glob('*.json'), reverse=True):
try:
d = json.load(open(f))
if isinstance(d, dict) and 'strategy_name' in d:
files.append(f)
except: pass
if count: files = files[:count]
print(f"Re-evaluating {len(files)} strategies...\n")
results, updated = [], 0
with Progress() as p:
task = p.add_task("Backtesting...", total=len(files))
for f in files:
try:
data = json.load(open(f))
fdfs = load_factors(data.get('factor_names', []), vdir)
if len(fdfs) >= 3:
bt = run_bt(fdfs, data.get('code', ''))
if bt:
data['metrics']['real_backtest'] = bt
data['summary'] = {"sharpe":bt['sharpe'], "max_drawdown":bt['max_drawdown'],
"win_rate":bt['win_rate'], "monthly_return_pct":bt['monthly_return_pct'],
"annual_return_pct":bt['annual_return_pct'], "real_ic":bt['ic'],
"real_n_trades":bt['n_trades'], "real_backtest_status":"success"}
with open(f, 'w') as out: json.dump(data, out, indent=2, ensure_ascii=False)
updated += 1
results.append({'name':data['strategy_name'], **bt})
except:
pass
p.update(task, advance=1)
print(f"\n✅ Updated {updated}/{len(files)}")
if results:
results.sort(key=lambda x: x['sharpe'], reverse=True)
print(f"\n{'='*75}\n🏆 TOP 10\n{'='*75}")
print(f"{'#':>3} {'Name':<30} {'Sharpe':>7} {'Monat':>8} {'MaxDD':>8} {'IC':>7} {'Trades':>7}")
print("-" * 70)
for i, r in enumerate(results[:10], 1):
print(f"{i:3d} {r['name']:30s} {r['sharpe']:7.3f} {r['monthly_return_pct']:7.2f}% {r['max_drawdown']:7.2%} {r['ic']:7.4f} {r['n_trades']:7d}")
if __name__ == "__main__":
import sys
main(int(sys.argv[1]) if len(sys.argv) > 1 else None)
+496
View File
@@ -0,0 +1,496 @@
"""
Predix Simple Factor Evaluator - Direct IC/Sharpe computation.
Evaluates existing factor results by computing IC and Sharpe directly
from factor values and forward returns, without Qlib infrastructure.
Usage:
python predix_simple_eval.py --top 100 # Evaluate top 100 factors
python predix_simple_eval.py --all # Evaluate all
python predix_simple_eval.py --parallel 4 # 4 parallel workers
"""
import json
import os
import sys
import time
import warnings
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import List, Optional
import numpy as np
import pandas as pd
from rich.console import Console
from rich.progress import (
Progress,
SpinnerColumn,
TextColumn,
BarColumn,
TaskProgressColumn,
TimeElapsedColumn,
)
from rich.table import Table
from rich.panel import Panel
console = Console()
# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
PROJECT_ROOT = Path(__file__).resolve().parent
WORKSPACE_DIR = PROJECT_ROOT / "git_ignore_folder" / "RD-Agent_workspace"
RESULTS_DIR = PROJECT_ROOT / "results"
BACKTESTS_DIR = RESULTS_DIR / "backtests"
DB_DIR = RESULTS_DIR / "db"
DB_PATH = DB_DIR / "backtest_results.db"
EVAL_SUMMARY_PATH = RESULTS_DIR / "eval_summary.json"
# Ensure directories exist
BACKTESTS_DIR.mkdir(parents=True, exist_ok=True)
DB_DIR.mkdir(parents=True, exist_ok=True)
# ---------------------------------------------------------------------------
# Data structures
# ---------------------------------------------------------------------------
@dataclass
class FactorWorkspace:
"""Represents a factor workspace with code and results."""
workspace_hash: str
factor_name: str
workspace_path: Path
result_path: Optional[Path] = None
data_path: Optional[Path] = None
@dataclass
class EvalResult:
"""Evaluation result for a single factor."""
factor_name: str
workspace_hash: str
status: str # success, failed
ic: Optional[float] = None
rank_ic: Optional[float] = None
sharpe: Optional[float] = None
annualized_return: Optional[float] = None
max_drawdown: Optional[float] = None
win_rate: Optional[float] = None
non_null_count: int = 0
total_count: int = 0
error_message: Optional[str] = None
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
def to_dict(self) -> dict:
return {k: v for k, v in self.__dict__.items()}
# ---------------------------------------------------------------------------
# Workspace scanner
# ---------------------------------------------------------------------------
def scan_workspaces(workspace_dir: Path) -> List[FactorWorkspace]:
"""Scan workspace directories for factors with results."""
workspaces = []
for ws in workspace_dir.iterdir():
if not ws.is_dir():
continue
result_file = ws / "result.h5"
data_file = ws / "intraday_pv.h5"
if not result_file.exists() or not data_file.exists():
continue
# Read factor name from result.h5
try:
result = pd.read_hdf(str(result_file), key="data")
if result is not None and len(result.columns) > 0:
factor_name = result.columns[0]
workspaces.append(FactorWorkspace(
workspace_hash=ws.name,
factor_name=factor_name,
workspace_path=ws,
result_path=result_file,
data_path=data_file,
))
except Exception:
continue
return workspaces
# ---------------------------------------------------------------------------
# Factor evaluator
# ---------------------------------------------------------------------------
def evaluate_factor(ws: FactorWorkspace, forward_return_bars: int = 96) -> EvalResult:
"""
Evaluate a factor by computing IC and Sharpe from factor values.
Parameters
----------
ws : FactorWorkspace
Workspace with result.h5 and intraday_pv.h5
forward_return_bars : int
Number of bars for forward return calculation (96 = 96 minutes for 1min data)
Returns
-------
EvalResult
"""
try:
# Load data
df = pd.read_hdf(str(ws.data_path), key="data")
result = pd.read_hdf(str(ws.result_path), key="data")
total_count = len(result)
factor_val = result.iloc[:, 0]
non_null_count = factor_val.notna().sum()
# Skip if too few valid values
if non_null_count < 100:
return EvalResult(
factor_name=ws.factor_name,
workspace_hash=ws.workspace_hash,
status="failed",
non_null_count=non_null_count,
total_count=total_count,
error_message=f"Too few valid values: {non_null_count}",
)
# Compute forward returns
# Handle column name escaping
col_close = "$close"
if col_close not in df.columns:
# Try alternative column name
col_close = next((c for c in df.columns if "close" in c.lower()), None)
if col_close is None:
return EvalResult(
factor_name=ws.factor_name,
workspace_hash=ws.workspace_hash,
status="failed",
non_null_count=non_null_count,
total_count=total_count,
error_message=f"No close column found. Columns: {list(df.columns)}",
)
close = df[col_close]
forward_ret = close.groupby(level="instrument").shift(-forward_return_bars) / close - 1
# Compute IC (Information Coefficient)
valid_idx = factor_val.dropna().index.intersection(forward_ret.dropna().index)
if len(valid_idx) < 100:
return EvalResult(
factor_name=ws.factor_name,
workspace_hash=ws.workspace_hash,
status="failed",
non_null_count=non_null_count,
total_count=total_count,
error_message=f"Too little overlap: {len(valid_idx)}",
)
ic = factor_val.loc[valid_idx].corr(forward_ret.loc[valid_idx])
rank_ic = factor_val.loc[valid_idx].corr(forward_ret.loc[valid_idx], method="spearman")
# Compute factor-level Sharpe (mean/std of factor values)
factor_mean = factor_val.loc[valid_idx].mean()
factor_std = factor_val.loc[valid_idx].std()
sharpe = factor_mean / factor_std if factor_std > 0 else 0
# Annualized return (assuming 252 trading days, 1440 minutes per day)
ann_factor = np.sqrt(252 * 1440 / forward_return_bars)
annualized_return = float(factor_mean * ann_factor * 100) # in percent
# Max drawdown approximation (cumulative factor performance)
cum_perf = factor_val.loc[valid_idx].cumsum()
running_max = cum_perf.expanding().max()
drawdown = (cum_perf - running_max) / running_max.replace(0, np.nan)
max_drawdown = float(drawdown.min()) if len(drawdown) > 0 else 0
# Win rate (percentage of positive factor values)
win_rate = float((factor_val.loc[valid_idx] > 0).sum()) / len(valid_idx)
return EvalResult(
factor_name=ws.factor_name,
workspace_hash=ws.workspace_hash,
status="success",
ic=float(ic) if ic is not None else None,
rank_ic=float(rank_ic) if rank_ic is not None else None,
sharpe=float(sharpe),
annualized_return=annualized_return,
max_drawdown=max_drawdown,
win_rate=win_rate,
non_null_count=non_null_count,
total_count=total_count,
)
except Exception as e:
return EvalResult(
factor_name=ws.factor_name,
workspace_hash=ws.workspace_hash,
status="failed",
error_message=str(e)[:500],
)
# ---------------------------------------------------------------------------
# Parallel evaluation
# ---------------------------------------------------------------------------
def run_evaluation(
workspaces: List[FactorWorkspace],
n_workers: int = 4,
) -> List[EvalResult]:
"""Run factor evaluation in parallel using threads."""
results = []
with Progress(
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
BarColumn(),
TaskProgressColumn(),
TimeElapsedColumn(),
console=console,
) as progress:
task = progress.add_task(f"Evaluating {len(workspaces)} factors...", total=len(workspaces))
with ThreadPoolExecutor(max_workers=n_workers) as executor:
futures = {executor.submit(evaluate_factor, ws): ws for ws in workspaces}
for future in as_completed(futures):
ws = futures[future]
try:
result = future.result(timeout=300)
results.append(result)
except Exception as e:
results.append(EvalResult(
factor_name=ws.factor_name,
workspace_hash=ws.workspace_hash,
status="failed",
error_message=f"Timeout/Exception: {str(e)[:300]}",
))
n_success = sum(1 for r in results if r.status == "success")
n_fail = sum(1 for r in results if r.status == "failed")
progress.update(
task,
advance=1,
description=f"Evaluating: {n_success}{n_fail}❌ | {ws.factor_name[:40]}",
)
return results
# ---------------------------------------------------------------------------
# Results storage
# ---------------------------------------------------------------------------
def save_results(results: List[EvalResult]) -> None:
"""Save evaluation results to JSON and SQLite."""
# Save as JSON
successful = [r for r in results if r.status == "success"]
failed = [r for r in results if r.status == "failed"]
# Sort by IC
successful.sort(key=lambda r: abs(r.ic) if r.ic is not None else 0, reverse=True)
# Save individual results
for r in successful[:50]: # Top 50
json_path = BACKTESTS_DIR / f"{r.factor_name}_{r.workspace_hash}.json"
with open(json_path, "w", encoding="utf-8") as f:
json.dump(r.to_dict(), f, indent=2, default=str)
# Save summary
valid_ic = [r.ic for r in results if r.ic is not None]
valid_sharpe = [r.sharpe for r in results if r.sharpe is not None]
summary = {
"generated_at": datetime.now().isoformat(),
"total_evaluated": len(results),
"successful": len(successful),
"failed": len(failed),
"success_rate": len(successful) / len(results) if results else 0,
"avg_ic": float(np.mean(valid_ic)) if valid_ic else 0,
"best_ic": float(max(valid_ic, key=abs)) if valid_ic else 0,
"avg_sharpe": float(np.mean(valid_sharpe)) if valid_sharpe else 0,
"best_sharpe": float(max(valid_sharpe)) if valid_sharpe else 0,
"top_20_by_ic": [r.to_dict() for r in successful[:20]],
"all_results": [r.to_dict() for r in results],
}
with open(EVAL_SUMMARY_PATH, "w", encoding="utf-8") as f:
json.dump(summary, f, indent=2, default=str)
# Save to SQLite
try:
import sqlite3
conn = sqlite3.connect(str(DB_PATH))
c = conn.cursor()
c.execute("""CREATE TABLE IF NOT EXISTS factor_evaluations (
id INTEGER PRIMARY KEY,
factor_name TEXT,
workspace_hash TEXT,
ic REAL,
rank_ic REAL,
sharpe REAL,
annualized_return REAL,
max_drawdown REAL,
win_rate REAL,
non_null_count INTEGER,
total_count INTEGER,
status TEXT,
timestamp TEXT
)""")
for r in results:
c.execute("""INSERT INTO factor_evaluations
(factor_name, workspace_hash, ic, rank_ic, sharpe,
annualized_return, max_drawdown, win_rate,
non_null_count, total_count, status, timestamp)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(r.factor_name, r.workspace_hash, r.ic, r.rank_ic, r.sharpe,
r.annualized_return, r.max_drawdown, r.win_rate,
r.non_null_count, r.total_count, r.status, r.timestamp))
conn.commit()
conn.close()
except Exception as e:
console.print(f"[yellow]SQLite save warning: {e}[/yellow]")
# ---------------------------------------------------------------------------
# Display
# ---------------------------------------------------------------------------
def display_results(results: List[EvalResult]) -> None:
"""Display evaluation results as a table."""
successful = [r for r in results if r.status == "success"]
successful.sort(key=lambda r: abs(r.ic) if r.ic is not None else 0, reverse=True)
table = Table(
title="Factor Evaluation Results",
show_header=True,
header_style="bold cyan",
)
table.add_column("#", justify="center", width=4)
table.add_column("Factor", width=40)
table.add_column("IC", justify="right", width=10)
table.add_column("Rank IC", justify="right", width=10)
table.add_column("Sharpe", justify="right", width=10)
table.add_column("Ann. Ret %", justify="right", width=10)
table.add_column("Max DD", justify="right", width=10)
table.add_column("Win Rate", justify="right", width=10)
for i, r in enumerate(successful[:20], 1):
table.add_row(
str(i),
r.factor_name[:38],
f"{r.ic:.6f}" if r.ic is not None else "N/A",
f"{r.rank_ic:.6f}" if r.rank_ic is not None else "N/A",
f"{r.sharpe:.4f}" if r.sharpe is not None else "N/A",
f"{r.annualized_return:.4f}" if r.annualized_return is not None else "N/A",
f"{r.max_drawdown:.4f}" if r.max_drawdown is not None else "N/A",
f"{r.win_rate:.2%}" if r.win_rate is not None else "N/A",
)
console.print()
console.print(table)
# Summary
valid_ic = [r.ic for r in results if r.ic is not None]
valid_sharpe = [r.sharpe for r in results if r.sharpe is not None]
console.print(Panel(
f"[bold]Evaluation Summary[/bold]\n"
f"Total evaluated: {len(results)}\n"
f"Successful: {len(successful)}\n"
f"Failed: {len(results) - len(successful)}\n"
f"Avg IC: {np.mean(valid_ic):.6f} (n={len(valid_ic)})\n"
f"Best IC: {max(valid_ic, key=abs, default=0):.6f}\n"
f"Avg Sharpe: {np.mean(valid_sharpe):.4f} (n={len(valid_sharpe)})\n"
f"Best Sharpe: {max(valid_sharpe, default=0):.4f}\n"
f"Saved to: {EVAL_SUMMARY_PATH}\n"
f"Database: {DB_PATH}",
border_style="green",
))
# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
def main(
top: int = 100,
all_factors: bool = False,
parallel: int = 4,
) -> None:
"""Main entry point."""
console.print(Panel(
"[bold cyan]Predix Simple Factor Evaluator[/bold cyan]\n"
f"Scanning workspaces for generated factors...",
border_style="cyan",
))
# Scan workspaces
workspaces = scan_workspaces(WORKSPACE_DIR)
console.print(f"\n[bold]Total workspaces with results: {len(workspaces)}[/bold]")
if not workspaces:
console.print("[red]No factors found![/red]")
return
# Select factors to evaluate
if all_factors:
to_evaluate = workspaces
else:
# Deduplicate by factor name, keep first occurrence
seen = set()
unique = []
for ws in workspaces:
if ws.factor_name not in seen:
seen.add(ws.factor_name)
unique.append(ws)
# Sort by non-null count (prefer factors with more valid values)
to_evaluate = sorted(unique, key=lambda ws: 0, reverse=True)[:top]
console.print(f"[bold green]Selected {len(to_evaluate)} factors for evaluation[/bold green]")
console.print(f" Using {parallel} parallel workers")
# Run evaluation
results = run_evaluation(to_evaluate, n_workers=parallel)
# Save results
console.print(f"\n[bold cyan]Saving results...[/bold cyan]")
save_results(results)
# Display
display_results(results)
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(
description="Predix Simple Factor Evaluator - Direct IC/Sharpe computation"
)
parser.add_argument(
"--top", "-n",
type=int,
default=100,
help="Number of factors to evaluate (default: 100)",
)
parser.add_argument(
"--all", "-a",
action="store_true",
help="Evaluate all discovered factors",
)
parser.add_argument(
"--parallel", "-p",
type=int,
default=4,
help="Number of parallel workers (default: 4)",
)
args = parser.parse_args()
main(
top=args.top,
all_factors=args.all,
parallel=args.parallel,
)
+344
View File
@@ -0,0 +1,344 @@
#!/usr/bin/env python
"""
Strategy Performance Report Generator for Predix.
Generates detailed PDF reports with charts for each accepted strategy.
Features:
- PDF report with all charts embedded
- Equity curve, drawdown, signal distribution, monthly returns
- Factor correlation matrix
- Full metrics table and strategy code
Usage:
python predix_strategy_report.py # All strategies
python predix_strategy_report.py results/strategies_new/123.json # Single strategy
"""
import os, sys, json, warnings
from pathlib import Path
from datetime import datetime
import numpy as np
import pandas as pd
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
from matplotlib.gridspec import GridSpec
import seaborn as sns
from reportlab.lib.pagesizes import A4
from reportlab.platypus import (
SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle,
Image, PageBreak, HRFlowable
)
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib import colors
from reportlab.lib.units import cm
from reportlab.lib.enums import TA_CENTER, TA_LEFT, TA_RIGHT
warnings.filterwarnings('ignore')
# Config
OHLCV_PATH = Path('/home/nico/Predix/git_ignore_folder/factor_implementation_source_data/intraday_pv.h5')
REPORTS_DIR = Path('/home/nico/Predix/results/strategy_reports')
REPORTS_DIR.mkdir(parents=True, exist_ok=True)
# Colors
BG_COLOR = '#1E1E1E'
TEXT_COLOR = '#E0E0E0'
ACCENT_GREEN = '#4CAF50'
ACCENT_RED = '#F44336'
ACCENT_BLUE = '#2196F3'
GRID_COLOR = '#333333'
class StrategyPerformanceReporter:
"""Generate comprehensive PDF + PNG report for a strategy."""
def __init__(self, strategy_data: dict, report_dir: Path = None):
self.strategy = strategy_data
self.name = strategy_data.get('strategy_name', 'unknown')
self.report_dir = report_dir or REPORTS_DIR
self.plots_dir = self.report_dir / 'plots'
self.plots_dir.mkdir(parents=True, exist_ok=True)
self.bt = strategy_data.get('real_backtest', {})
self.summary = strategy_data.get('summary', {})
self.factors = strategy_data.get('factor_names', [])
self.code = strategy_data.get('code', '')
self.description = strategy_data.get('description', '')
plt.style.use('dark_background')
def generate_report(self) -> dict:
"""Generate full report: PNG dashboard + individual charts + PDF + text."""
ts = datetime.now().strftime('%Y%m%d_%H%M%S')
name = f"{ts}_{self.name}"
# PNG charts
fig = self._create_dashboard()
dash_path = self.plots_dir / f"{name}_dashboard.png"
fig.savefig(str(dash_path), dpi=150, bbox_inches='tight', facecolor=BG_COLOR)
plt.close(fig)
self._gen_png(self._plot_equity_curve, f"{self.name}_equity.png", (12, 6))
self._gen_png(self._plot_drawdown, f"{self.name}_drawdown.png", (12, 6))
self._gen_png(self._plot_signal_dist, f"{self.name}_signals.png", (8, 8))
self._gen_png(self._plot_monthly_returns, f"{self.name}_monthly_returns.png", (12, 6))
self._gen_png(self._plot_factor_corr, f"{self.name}_factor_corr.png", (10, 8))
# Text report
txt_path = self.report_dir / f"{name}_report.txt"
self._gen_text_report(txt_path)
# PDF report
pdf_path = self.report_dir / f"{name}_report.pdf"
self._gen_pdf_report(pdf_path)
return {'dashboard': dash_path, 'pdf': pdf_path, 'text': txt_path}
def _gen_png(self, plot_fn, filename, figsize):
fig, ax = plt.subplots(figsize=figsize, facecolor=BG_COLOR)
plot_fn(ax)
p = self.plots_dir / filename
fig.savefig(str(p), dpi=150, bbox_inches='tight', facecolor=BG_COLOR)
plt.close(fig)
# ========== Chart methods ==========
def _create_dashboard(self):
fig = plt.figure(figsize=(20, 24), facecolor=BG_COLOR)
gs = GridSpec(4, 2, figure=fig, hspace=0.35, wspace=0.3)
fig.suptitle(f"Strategy Report: {self.name}", fontsize=20, fontweight='bold', color=TEXT_COLOR, y=0.98)
self._plot_equity_curve(fig.add_subplot(gs[0, 0]))
self._plot_drawdown(fig.add_subplot(gs[0, 1]))
self._plot_signal_dist(fig.add_subplot(gs[1, 0]))
self._plot_monthly_returns(fig.add_subplot(gs[1, 1]))
ax5 = fig.add_subplot(gs[2, 0]); ax5.axis('off')
self._plot_metrics_table(ax5)
ax6 = fig.add_subplot(gs[2, 1]); ax6.axis('off')
self._plot_strategy_code(ax6)
ax7 = fig.add_subplot(gs[3, :]); ax7.axis('off')
self._plot_factors_list(ax7)
return fig
def _plot_equity_curve(self, ax):
n = max(int(self.summary.get('n_months', 12)), 12)
m = self.summary.get('monthly_return_pct', 0) / 100
months = pd.date_range(start='2024-01-01', periods=n, freq='ME')
eq = (1 + m) ** np.arange(n)
ax.fill_between(months, eq, alpha=0.3, color=ACCENT_GREEN)
ax.plot(months, eq, linewidth=2, color=ACCENT_GREEN)
ax.set_title('Equity Curve (Projected)', fontsize=12, color=TEXT_COLOR)
ax.set_ylabel('Equity Multiplier', color=TEXT_COLOR)
ax.grid(True, alpha=0.3, color=GRID_COLOR); ax.tick_params(colors=TEXT_COLOR)
def _plot_drawdown(self, ax):
mdd = abs(self.summary.get('max_drawdown', 0)) or 0.01
n = max(int(self.summary.get('n_months', 12)), 12)
months = pd.date_range(start='2024-01-01', periods=n, freq='ME')
dd = np.concatenate([np.linspace(0, -mdd, n//2), np.linspace(-mdd, 0, n-n//2)])
ax.fill_between(months, dd, alpha=0.5, color=ACCENT_RED)
ax.plot(months, dd, linewidth=1.5, color=ACCENT_RED)
ax.set_title(f'Max Drawdown: {mdd:.2%}', fontsize=12, color=TEXT_COLOR)
ax.set_ylabel('Drawdown', color=TEXT_COLOR)
ax.grid(True, alpha=0.3, color=GRID_COLOR); ax.tick_params(colors=TEXT_COLOR)
ax.axhline(y=0, color=TEXT_COLOR, alpha=0.5, linewidth=0.5)
def _plot_signal_dist(self, ax):
l, s, n = self.bt.get('signal_long', 0), self.bt.get('signal_short', 0), self.bt.get('signal_neutral', 0)
t = l + s + n
if t > 0:
ax.pie([l, s, n], labels=[f'LONG ({l:,})', f'SHORT ({s:,})', f'NEUTRAL ({n:,})'],
colors=[ACCENT_GREEN, ACCENT_RED, '#666'], autopct='%1.1f%%', startangle=90,
textprops={'color': TEXT_COLOR})
ax.set_title('Signal Distribution', fontsize=12, color=TEXT_COLOR)
def _plot_monthly_returns(self, ax):
m = self.summary.get('monthly_return_pct', 0)
n = max(int(self.summary.get('n_months', 12)), 12)
np.random.seed(42)
scale = abs(m) * 0.3 if m != 0 else 1.0
rets = m + np.random.normal(0, scale, n)
cols = [ACCENT_GREEN if r > 0 else ACCENT_RED for r in rets]
ax.bar([f'M{i+1}' for i in range(n)], rets, color=cols, alpha=0.8)
ax.axhline(y=0, color=TEXT_COLOR, alpha=0.5, linewidth=0.5)
ax.set_title(f'Monthly Returns (Avg: {m:.2f}%)', fontsize=12, color=TEXT_COLOR)
ax.set_ylabel('Return %', color=TEXT_COLOR); ax.tick_params(colors=TEXT_COLOR)
ax.grid(True, alpha=0.2, axis='y', color=GRID_COLOR)
def _plot_metrics_table(self, ax):
metrics = [('IC', f"{self.bt.get('ic', 0):.4f}"), ('Sharpe', f"{self.bt.get('sharpe', 0):.3f}"),
('Max DD', f"{self.bt.get('max_drawdown', 0):.2%}"), ('Win Rate', f"{self.bt.get('win_rate', 0):.2%}"),
('Monthly', f"{self.bt.get('monthly_return_pct', 0):.2f}%"), ('Trades', f"{self.bt.get('n_trades', 0):,}")]
y = 0.9
for lab, val in metrics:
c = ACCENT_GREEN if not val.startswith('-') else ACCENT_RED
ax.text(0.1, y, lab, fontsize=11, fontweight='bold', color=TEXT_COLOR, transform=ax.transAxes)
ax.text(0.9, y, val, fontsize=11, fontweight='bold', color=c, transform=ax.transAxes, ha='right')
y -= 0.15
ax.set_title('Key Metrics', fontsize=14, fontweight='bold', color=TEXT_COLOR)
def _plot_strategy_code(self, ax):
code = (self.code or 'No code')[:800] + ('\n...(truncated)' if len(self.code or '') > 800 else '')
ax.text(0.05, 0.95, 'Strategy Code:', fontsize=12, fontweight='bold', color=TEXT_COLOR, transform=ax.transAxes)
ax.text(0.05, 0.88, code, fontsize=8, family='monospace', color='#A5D6A7', transform=ax.transAxes, va='top',
bbox=dict(boxstyle='round,pad=0.5', facecolor='#2C2C2C', alpha=0.8))
def _plot_factors_list(self, ax):
ax.text(0.05, 0.9, f"Factors Used ({len(self.factors)}):", fontsize=14, fontweight='bold', color=TEXT_COLOR, transform=ax.transAxes)
for i, f in enumerate(self.factors[:15]):
ax.text(0.05, 0.75 - i*0.12, f"{f}", fontsize=10, color=ACCENT_BLUE, transform=ax.transAxes)
def _plot_factor_corr(self, ax):
n = len(self.factors)
if n < 2: return
np.random.seed(42)
corr = np.eye(n)
for i in range(n):
for j in range(i+1, n):
v = np.random.uniform(0.1, 0.8); corr[i,j] = corr[j,i] = v
im = ax.imshow(corr, cmap='RdYlGn', aspect='auto', vmin=-1, vmax=1)
ax.set_xticks(range(n)); ax.set_yticks(range(n))
ax.set_xticklabels([f[:20] for f in self.factors], rotation=45, ha='right', color=TEXT_COLOR, fontsize=8)
ax.set_yticklabels([f[:20] for f in self.factors], color=TEXT_COLOR, fontsize=8)
ax.set_title('Factor Correlation', fontsize=14, color=TEXT_COLOR); plt.colorbar(im, ax=ax)
# ========== Report generators ==========
def _gen_text_report(self, path):
with open(path, 'w') as f:
f.write(f"{'='*80}\nSTRATEGY PERFORMANCE REPORT\nGenerated: {datetime.now()}\n{'='*80}\n\n")
f.write(f"Strategy: {self.name}\nDescription: {self.description}\nFactors: {len(self.factors)}\n\n")
f.write(f"{'-'*40}\nPERFORMANCE METRICS\n{'-'*40}\n")
bt = self.bt
f.write(f" {'IC':25s} {bt.get('ic',0):.6f}\n")
f.write(f" {'Sharpe':25s} {bt.get('sharpe',0):.4f}\n")
f.write(f" {'Max Drawdown':25s} {bt.get('max_drawdown',0):.2%}\n")
f.write(f" {'Win Rate':25s} {bt.get('win_rate',0):.2%}\n")
f.write(f" {'Monthly Return':25s} {bt.get('monthly_return_pct',0):.2f}%\n")
f.write(f" {'Annual Return':25s} {bt.get('annual_return_pct',0):.2f}%\n")
f.write(f" {'Total Return':25s} {bt.get('total_return',0):.2%}\n")
f.write(f" {'Trades':25s} {bt.get('n_trades',0):,}\n")
f.write(f" {'Data Points':25s} {bt.get('n_bars',0):,}\n")
f.write(f" {'Period (Months)':25s} {bt.get('n_months',0):.1f}\n\n")
f.write(f"{'-'*40}\nFACTORS\n{'-'*40}\n")
for fac in self.factors: f.write(f"{fac}\n")
f.write(f"\n{'-'*40}\nCODE\n{'-'*40}\n{self.code}\n\n{'='*80}\nEND OF REPORT\n{'='*80}\n")
def _gen_pdf_report(self, pdf_path):
doc = SimpleDocTemplate(str(pdf_path), pagesize=A4,
title=f"Predix: {self.name}", author="Predix AI",
leftMargin=2*cm, rightMargin=2*cm, topMargin=2*cm, bottomMargin=2*cm)
styles = getSampleStyleSheet()
styles.add(ParagraphStyle(name='PTitle', fontName='Helvetica-Bold', fontSize=22, leading=26, alignment=TA_CENTER, textColor=colors.HexColor('#1A237E')))
styles.add(ParagraphStyle(name='PHead', fontName='Helvetica-Bold', fontSize=14, leading=18, spaceBefore=15, spaceAfter=10, textColor=colors.HexColor('#0D47A1')))
styles.add(ParagraphStyle(name='PBody', fontName='Helvetica', fontSize=10, leading=12, spaceAfter=8, textColor=colors.HexColor('#212121')))
styles.add(ParagraphStyle(name='PSmall', fontName='Helvetica', fontSize=8, leading=10, textColor=colors.HexColor('#757575')))
story = []
# Cover
story.append(Spacer(1, 3*cm))
story.append(Paragraph("PREDIX", styles['PTitle']))
story.append(Spacer(1, 0.5*cm))
story.append(HRFlowable(width="80%", thickness=2, color=colors.HexColor('#1A237E'), spaceAfter=20))
story.append(Paragraph(f"Strategy Report: {self.name}", styles['PHead']))
if self.description: story.append(Paragraph(self.description, styles['PBody']))
mc = [["IC", f"{self.bt.get('ic',0):.4f}"],["Sharpe", f"{self.bt.get('sharpe',0):.3f}"],
["Max DD", f"{self.bt.get('max_drawdown',0):.2%}"],["Win Rate", f"{self.bt.get('win_rate',0):.2%}"],
["Monthly", f"{self.bt.get('monthly_return_pct',0):.2f}%"],["Trades", f"{self.bt.get('n_trades',0):,}"]]
t = Table(mc, colWidths=[4*cm,6*cm])
t.setStyle(TableStyle([('FONTNAME',(0,0),(0,-1),'Helvetica-Bold'),('FONTSIZE',(0,0),(-1,-1),12),
('ALIGN',(0,0),(0,-1),'RIGHT'),('ALIGN',(1,0),(1,-1),'LEFT'),('TEXTCOLOR',(0,0),(-1,-1),colors.HexColor('#212121'))]))
story.append(t); story.append(Spacer(1,2*cm))
story.append(Paragraph(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", styles['PSmall']))
story.append(Paragraph(f"Factors: {len(self.factors)}", styles['PSmall']))
story.append(PageBreak())
# Metrics
story.append(Paragraph("1. Performance Metrics", styles['PHead']))
mt = [["Metric","Value"],
["IC", f"{self.bt.get('ic',0):.6f}"],["Sharpe Ratio", f"{self.bt.get('sharpe',0):.4f}"],
["Max Drawdown", f"{self.bt.get('max_drawdown',0):.2%}"],["Win Rate", f"{self.bt.get('win_rate',0):.2%}"],
["Monthly Return", f"{self.bt.get('monthly_return_pct',0):.2f}%"],["Annual Return", f"{self.bt.get('annual_return_pct',0):.2f}%"],
["Total Return", f"{self.bt.get('total_return',0):.2%}"],["Total Trades", f"{self.bt.get('n_trades',0):,}"],
["Data Points", f"{self.bt.get('n_bars',0):,}"],["Long Signals", f"{self.bt.get('signal_long',0):,}"],
["Short Signals", f"{self.bt.get('signal_short',0):,}"],["Neutral Signals", f"{self.bt.get('signal_neutral',0):,}"]]
t = Table(mt, colWidths=[9*cm,7*cm])
t.setStyle(TableStyle([('BACKGROUND',(0,0),(-1,0),colors.HexColor('#1A237E')),('TEXTCOLOR',(0,0),(-1,0),colors.white),
('FONTNAME',(0,0),(-1,0),'Helvetica-Bold'),('FONTSIZE',(0,0),(-1,-1),10),('ALIGN',(0,0),(0,-1),'LEFT'),
('ALIGN',(1,0),(1,-1),'RIGHT'),('GRID',(0,0),(-1,-1),0.5,colors.HexColor('#E0E0E0')),
('BACKGROUND',(0,1),(-1,-1),colors.HexColor('#FAFAFA')),('ROWBACKGROUNDS',(0,1),(-1,-1),[colors.HexColor('#FAFAFA'),colors.white])]))
story.append(t); story.append(PageBreak())
# Charts
story.append(Paragraph("2. Visualizations", styles['PHead']))
# Dashboard
dp = self.plots_dir / f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{self.name}_dashboard.png"
if not dp.exists():
fig = self._create_dashboard()
fig.savefig(str(dp), dpi=150, bbox_inches='tight', facecolor=BG_COLOR); plt.close(fig)
if dp.exists():
story.append(Paragraph("2.1 Strategy Dashboard", styles['PHead']))
story.append(Image(str(dp), width=16*cm, height=19*cm)); story.append(PageBreak())
for label, fname, w, h in [("2.2 Equity Curve","equity",16,8),("2.3 Drawdown","drawdown",16,8),
("2.4 Signals","signals",12,12),("2.5 Monthly Returns","monthly_returns",16,8)]:
fp = self.plots_dir / f"{self.name}_{fname}.png"
if fp.exists():
story.append(Paragraph(label, styles['PHead']))
story.append(Image(str(fp), width=w*cm, height=h*cm)); story.append(Spacer(1,0.5*cm))
story.append(PageBreak())
# Factors
story.append(Paragraph("3. Factors Used", styles['PHead']))
fd = [["#", "Factor"]] + [[str(i+1), f] for i, f in enumerate(self.factors)]
t = Table(fd, colWidths=[2*cm,14*cm])
t.setStyle(TableStyle([('BACKGROUND',(0,0),(-1,0),colors.HexColor('#1A237E')),('TEXTCOLOR',(0,0),(-1,0),colors.white),
('FONTNAME',(0,0),(-1,0),'Helvetica-Bold'),('FONTSIZE',(0,0),(-1,-1),9),
('GRID',(0,0),(-1,-1),0.5,colors.HexColor('#E0E0E0')),('BACKGROUND',(0,1),(-1,-1),colors.HexColor('#FAFAFA'))]))
story.append(t); story.append(Spacer(1,1*cm))
# Code
story.append(Paragraph("4. Strategy Code", styles['PHead']))
for line in (self.code or 'No code').split('\n'):
story.append(Paragraph(f'<font name="Courier" size="8" color="#1B5E20">{line.replace("&","&amp;").replace("<","&lt;").replace(">","&gt;")}</font>', styles['PBody']))
story.append(PageBreak())
# Summary
story.append(Paragraph("5. Summary", styles['PHead']))
story.append(Paragraph(
f"Strategy <b>{self.name}</b> combines {len(self.factors)} factors for EUR/USD. "
f"IC={self.bt.get('ic',0):.4f}, Sharpe={self.bt.get('sharpe',0):.3f}, "
f"Trades={self.bt.get('n_trades',0):,}.", styles['PBody']))
story.append(Spacer(1,1*cm))
story.append(Paragraph("Disclaimer", styles['PHead']))
story.append(Paragraph("Past performance is not indicative of future results. "
"For research purposes only. Trading involves substantial risk.", styles['PSmall']))
doc.build(story)
def generate_report_for_strategy(path: str) -> dict:
with open(path) as f: data = json.load(f)
return StrategyPerformanceReporter(data).generate_report()
def generate_all_reports():
d = Path('/home/nico/Predix/results/strategies_new')
if not d.exists(): print("No strategies."); return
for jf in sorted(d.glob('*.json')):
try:
r = generate_report_for_strategy(str(jf))
print(f"{jf.stem}{r['pdf'].name}")
except Exception as e:
print(f"{jf.stem}: {e}")
if __name__ == '__main__':
if len(sys.argv) > 1:
p = sys.argv[1]
if Path(p).exists():
r = generate_report_for_strategy(p)
print(f"PDF: {r['pdf']}\nDashboard: {r['dashboard']}\nText: {r['text']}")
else: print(f"Not found: {p}")
else: generate_all_reports()
+109
View File
@@ -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: $?"