mirror of
https://github.com/NicolasBohn/NexQuant.git
synced 2026-07-27 15:37:44 +00:00
TPTBusiness
4758de0eee
- Rename FTMO_* constants → generic names (RISK_PER_TRADE, MAX_DAILY_LOSS, etc.) - Rename backtest_signal_ftmo → backtest_signal_risk - Rename _apply_ftmo_mask → _apply_risk_mask - Clean all FTMO/riskMgmt mentions from commit messages via filter-branch - AGENTS.md: add non-negotiable rule — NEVER mention proprietary terms in commits/releases - Code variables and function names sanitized project-wide - Force-pushed rewritten history to remote
1742 lines
64 KiB
Python
1742 lines
64 KiB
Python
#!/usr/bin/env python
|
||
"""
|
||
Smart Strategy Generation with Feedback Loop, Parameter Optimization & RiskMgmt Risk Management.
|
||
|
||
Generates EUR/USD daytrading strategies using LLM with:
|
||
- Adaptive feedback loop (IC, trades, drawdown-based suggestions)
|
||
- Grid search for optimal parameters (thresholds, SL/TP, trailing stops)
|
||
- Mandatory RiskMgmt-compliant risk management layer
|
||
- Comprehensive evaluation metrics # nosec
|
||
|
||
Usage:
|
||
python nexquant_smart_strategy_gen.py 10
|
||
python nexquant_smart_strategy_gen.py 5 --style daytrading
|
||
python nexquant_smart_strategy_gen.py 20 --style swing --max-attempts 200
|
||
"""
|
||
import json
|
||
import logging
|
||
import os
|
||
import random
|
||
import subprocess # nosec
|
||
import sys
|
||
import time
|
||
import warnings
|
||
from datetime import datetime
|
||
from itertools import product
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
from dotenv import load_dotenv
|
||
from rich.console import Console
|
||
from rich.logging import RichHandler
|
||
from rich.progress import BarColumn, Progress, SpinnerColumn, TextColumn, TimeElapsedColumn
|
||
from rich.table import Table
|
||
|
||
warnings.filterwarnings("ignore")
|
||
|
||
# ============================================================================
|
||
# Configuration & Constants
|
||
# ============================================================================
|
||
OHLCV_PATH = Path("/home/nico/NexQuant/git_ignore_folder/factor_implementation_source_data/intraday_pv.h5")
|
||
FACTORS_DIR = Path("/home/nico/NexQuant/results/factors")
|
||
STRATEGIES_DIR = Path("/home/nico/NexQuant/results/strategies_new")
|
||
STRATEGIES_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
# Logging setup
|
||
LOG_DIR = Path("/home/nico/NexQuant/results/logs")
|
||
LOG_DIR.mkdir(parents=True, exist_ok=True)
|
||
log_file = LOG_DIR / f"smart_strategy_gen_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||
handlers=[
|
||
logging.FileHandler(log_file),
|
||
RichHandler(rich_tracebacks=True, show_time=False, show_path=False),
|
||
],
|
||
)
|
||
logger = logging.getLogger("SmartStrategyGen")
|
||
|
||
console = Console()
|
||
|
||
# ============================================================================
|
||
# RiskMgmt Risk Management Constants
|
||
# ============================================================================
|
||
class RiskMgmtRiskLimits:
|
||
"""RiskMgmt-compliant risk management constants."""
|
||
MAX_DAILY_LOSS_PCT = 0.05 # 5% max daily loss (RiskMgmt rule)
|
||
MAX_PER_TRADE_LOSS_PCT = 0.02 # 2% max per trade
|
||
MAX_TOTAL_DRAWDOWN = 0.10 # 10% max overall drawdown
|
||
MAX_POSITIONS = 1 # Only 1 position at a time
|
||
MIN_RISK_REWARD_RATIO = 2.0 # TP must be at least 2x SL
|
||
POSITION_RISK_PCT = 0.01 # 1% risk per trade
|
||
|
||
# ============================================================================
|
||
# Acceptance Criteria
|
||
# ============================================================================
|
||
ACCEPTANCE_CRITERIA = {
|
||
"daytrading": {
|
||
"min_abs_ic": 0.02,
|
||
"min_sharpe": 1.0,
|
||
"min_trades": 50,
|
||
"max_drawdown": -0.15,
|
||
"min_win_rate": 0.45,
|
||
"min_monthly_return": 0.15,
|
||
"max_daily_loss": 0.05,
|
||
},
|
||
"swing": {
|
||
"min_abs_ic": 0.02,
|
||
"min_sharpe": 0.5,
|
||
"min_trades": 10,
|
||
"max_drawdown": -0.15,
|
||
"min_win_rate": 0.40,
|
||
"min_monthly_return": 0.15,
|
||
"max_daily_loss": 0.05,
|
||
},
|
||
}
|
||
|
||
# ============================================================================
|
||
# Parameter Grid for Optimization
|
||
# ============================================================================
|
||
PARAMETER_GRID = {
|
||
"threshold_entry": [0.2, 0.3, 0.4, 0.5],
|
||
"rolling_window": [10, 20, 30, 60],
|
||
"stop_loss": [0.01, 0.015, 0.02], # 1%, 1.5%, 2% (HARD MAX: 2% for RiskMgmt)
|
||
"take_profit": [0.02, 0.03, 0.04, 0.06], # 2x-3x SL
|
||
"trailing_stop": [0.01, 0.015], # 1%, 1.5% after profit threshold
|
||
"trailing_activation": [0.015, 0.02], # Activate trail after 1.5%, 2% profit
|
||
}
|
||
|
||
# ============================================================================
|
||
# Data Loading (Cached)
|
||
# ============================================================================
|
||
class DataCache:
|
||
"""Thread-safe data cache for OHLCV and factors."""
|
||
|
||
def __init__(self):
|
||
self._ohlcv_cache: pd.Series | None = None
|
||
self._factors_cache: list[dict] | None = None
|
||
self._factor_data_cache: dict[str, pd.Series] = {}
|
||
|
||
def load_ohlcv(self) -> pd.Series:
|
||
"""Load OHLCV close prices from HDF5."""
|
||
if self._ohlcv_cache is not None:
|
||
return self._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")
|
||
close_col = "$close" if "$close" in ohlcv.columns else "close" if "close" in ohlcv.columns else ohlcv.select_dtypes(include=[np.number]).columns[0]
|
||
close = ohlcv[close_col].dropna()
|
||
|
||
# Limit to last 200k bars to avoid OOM during optimization
|
||
# (372k bars × 15 combinations = too much memory)
|
||
MAX_BARS = 200000
|
||
if len(close) > MAX_BARS:
|
||
close = close.iloc[-MAX_BARS:]
|
||
logger.info(f"Trimmed OHLCV data to last {MAX_BARS:,} bars (from {len(ohlcv[close_col]):,})")
|
||
|
||
self._ohlcv_cache = close
|
||
logger.info(f"Loaded {len(close):,} OHLCV bars")
|
||
return close
|
||
|
||
def load_top_factors(self, top_n: int = 20) -> list[dict]:
|
||
"""Load top factors by IC that have parquet files."""
|
||
if self._factors_cache is not None:
|
||
return self._factors_cache[:top_n]
|
||
|
||
factors = []
|
||
for f in FACTORS_DIR.glob("*.json"):
|
||
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 Exception as e:
|
||
logger.debug(f"Failed to load factor metadata: {f.name} - {e}")
|
||
|
||
factors.sort(key=lambda x: abs(x["ic"]), reverse=True)
|
||
self._factors_cache = factors
|
||
return factors[:top_n]
|
||
|
||
def load_factor_timeseries(self, factor_name: str) -> pd.Series | None:
|
||
"""Load factor time-series from parquet."""
|
||
if factor_name in self._factor_data_cache:
|
||
return self._factor_data_cache[factor_name]
|
||
|
||
safe = factor_name.replace("/", "_").replace("\\", "_")[:150]
|
||
pf = FACTORS_DIR / "values" / f"{safe}.parquet"
|
||
|
||
if not pf.exists():
|
||
return None
|
||
|
||
try:
|
||
series = pd.read_parquet(str(pf)).iloc[:, 0]
|
||
self._factor_data_cache[factor_name] = series
|
||
return series
|
||
except Exception as e:
|
||
logger.debug(f"Failed to load factor data: {factor_name} - {e}")
|
||
return None
|
||
|
||
data_cache = DataCache()
|
||
|
||
# ============================================================================
|
||
# LLM Setup
|
||
# ============================================================================
|
||
def setup_llm_env():
|
||
"""Setup LLM environment variables with fallback chain."""
|
||
load_dotenv(Path(__file__).parent / ".env", override=True)
|
||
|
||
# Priority 1: OpenRouter (free models with fallback)
|
||
router_key = os.getenv("OPENROUTER_API_KEY", "")
|
||
if router_key and router_key != "local":
|
||
# Build model fallback chain
|
||
models = [
|
||
os.getenv("OPENROUTER_MODEL", ""),
|
||
os.getenv("OPENROUTER_MODEL_2", ""),
|
||
os.getenv("OPENROUTER_MODEL_3", ""),
|
||
]
|
||
models = [m for m in models if m] # Remove empty
|
||
|
||
if models:
|
||
os.environ["OPENAI_API_KEY"] = router_key
|
||
os.environ["OPENAI_API_BASE"] = "https://openrouter.ai/api/v1"
|
||
os.environ["OPENROUTER_MODELS"] = json.dumps(models) # Store for fallback
|
||
os.environ["CHAT_MODEL"] = models[0]
|
||
logger.info(f"LLM environment configured for OpenRouter: {', '.join(models)}")
|
||
return
|
||
|
||
# Priority 2: Local LLM (llama.cpp)
|
||
api_key = os.getenv("OPENAI_API_KEY", "")
|
||
api_base = os.getenv("OPENAI_API_BASE", "")
|
||
chat_model = os.getenv("CHAT_MODEL", "")
|
||
|
||
if api_key == "local" and api_base:
|
||
os.environ["OPENAI_API_KEY"] = "local"
|
||
os.environ["OPENAI_API_BASE"] = api_base
|
||
os.environ["CHAT_MODEL"] = chat_model or "openai/qwen3.5-35b"
|
||
logger.info(f"LLM environment configured for LOCAL LLM: {api_base}")
|
||
else:
|
||
logger.warning("No API key found - LLM generation will fail")
|
||
|
||
# ============================================================================
|
||
# Risk Management Engine
|
||
# ============================================================================
|
||
class RiskManagementEngine:
|
||
"""
|
||
RiskMgmt-compliant risk management layer.
|
||
|
||
Applies stop loss, take profit, trailing stop, and daily loss limits
|
||
to strategy returns.
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
stop_loss: float = 0.02,
|
||
take_profit: float = 0.04,
|
||
trailing_stop: float = 0.015,
|
||
trailing_activation: float = 0.02,
|
||
max_daily_loss: float = 0.05,
|
||
max_positions: int = 1,
|
||
):
|
||
"""
|
||
Initialize risk management parameters.
|
||
|
||
Parameters
|
||
----------
|
||
stop_loss : float
|
||
Stop loss percentage (default 2%)
|
||
take_profit : float
|
||
Take profit percentage (default 4%, 2x SL)
|
||
trailing_stop : float
|
||
Trailing stop distance (default 1.5%)
|
||
trailing_activation : float
|
||
Profit level to activate trailing stop (default 2%)
|
||
max_daily_loss : float
|
||
Maximum daily loss percentage (default 5%)
|
||
max_positions : int
|
||
Maximum concurrent positions (default 1)
|
||
"""
|
||
# Validate RiskMgmt compliance
|
||
if stop_loss > 0.02:
|
||
raise ValueError(f"Stop loss {stop_loss:.2%} exceeds RiskMgmt max of 2%")
|
||
if take_profit < stop_loss * 2:
|
||
raise ValueError(f"Take profit {take_profit:.2%} must be at least 2x SL ({stop_loss*2:.2%})")
|
||
if max_daily_loss > 0.05:
|
||
raise ValueError(f"Daily loss {max_daily_loss:.2%} exceeds RiskMgmt max of 5%")
|
||
|
||
self.stop_loss = stop_loss
|
||
self.take_profit = take_profit
|
||
self.trailing_stop = trailing_stop
|
||
self.trailing_activation = trailing_activation
|
||
self.max_daily_loss = max_daily_loss
|
||
self.max_positions = max_positions
|
||
|
||
@property
|
||
def risk_reward_ratio(self) -> float:
|
||
"""Calculate risk/reward ratio (TP/SL)."""
|
||
return self.take_profit / self.stop_loss if self.stop_loss > 0 else 0.0
|
||
|
||
def apply_risk_management(
|
||
self,
|
||
signal: pd.Series,
|
||
close: pd.Series,
|
||
) -> pd.Series:
|
||
"""
|
||
Apply SL/TP/Trailing stop to signal-based strategy.
|
||
|
||
Parameters
|
||
----------
|
||
signal : pd.Series
|
||
Trading signals (1=LONG, -1=SHORT, 0=NEUTRAL)
|
||
close : pd.Series
|
||
Close prices
|
||
|
||
Returns
|
||
-------
|
||
pd.Series
|
||
Strategy returns after risk management
|
||
"""
|
||
if len(signal) == 0 or len(close) == 0:
|
||
return pd.Series(dtype=float)
|
||
|
||
# Align indices
|
||
common_idx = signal.index.intersection(close.index)
|
||
signal = signal.loc[common_idx].fillna(0)
|
||
close = close.loc[common_idx]
|
||
|
||
# Calculate returns
|
||
returns = close.pct_change().fillna(0)
|
||
strategy_returns = pd.Series(0.0, index=common_idx)
|
||
|
||
position = 0 # 0=neutral, 1=long, -1=short
|
||
entry_price = 0.0
|
||
highest_profit = 0.0
|
||
daily_pnl = 0.0
|
||
current_date = None
|
||
|
||
for i, idx in enumerate(common_idx):
|
||
if i == 0:
|
||
continue
|
||
|
||
# Track daily PnL for max daily loss
|
||
bar_date = idx.date() if hasattr(idx, "date") else idx
|
||
if current_date is None:
|
||
current_date = bar_date
|
||
elif bar_date != current_date:
|
||
daily_pnl = 0.0 # Reset daily PnL
|
||
current_date = bar_date
|
||
|
||
current_price = close.iloc[i]
|
||
prev_price = close.iloc[i - 1]
|
||
current_signal = signal.iloc[i]
|
||
|
||
# Check if we should exit position due to SL/TP/Trailing
|
||
if position != 0:
|
||
pnl_pct = 0.0
|
||
if position == 1: # Long
|
||
pnl_pct = (current_price - entry_price) / entry_price
|
||
elif position == -1: # Short
|
||
pnl_pct = (entry_price - current_price) / entry_price
|
||
|
||
# Stop Loss hit
|
||
if pnl_pct <= -self.stop_loss:
|
||
strategy_returns.iloc[i] = -self.stop_loss * position
|
||
daily_pnl += -self.stop_loss
|
||
position = 0
|
||
highest_profit = 0.0
|
||
continue
|
||
|
||
# Take Profit hit
|
||
if pnl_pct >= self.take_profit:
|
||
strategy_returns.iloc[i] = self.take_profit * position
|
||
daily_pnl += self.take_profit
|
||
position = 0
|
||
highest_profit = 0.0
|
||
continue
|
||
|
||
# Trailing Stop (activate after profit threshold)
|
||
if pnl_pct >= self.trailing_activation:
|
||
highest_profit = max(highest_profit, pnl_pct)
|
||
if (highest_profit - pnl_pct) >= self.trailing_stop:
|
||
strategy_returns.iloc[i] = pnl_pct * position
|
||
daily_pnl += pnl_pct
|
||
position = 0
|
||
highest_profit = 0.0
|
||
continue
|
||
|
||
# Normal position PnL
|
||
if position == 1:
|
||
strategy_returns.iloc[i] = (current_price - prev_price) / prev_price
|
||
elif position == -1:
|
||
strategy_returns.iloc[i] = -(current_price - prev_price) / prev_price
|
||
|
||
# Update daily PnL
|
||
daily_pnl += strategy_returns.iloc[i]
|
||
|
||
# Check max daily loss
|
||
if daily_pnl <= -self.max_daily_loss:
|
||
strategy_returns.iloc[i] = strategy_returns.iloc[i] # Keep the loss
|
||
position = 0 # Stop trading for the day
|
||
highest_profit = 0.0
|
||
continue
|
||
|
||
# Enter new position (only if neutral and max positions not exceeded)
|
||
if position == 0 and current_signal != 0:
|
||
position = int(np.sign(current_signal))
|
||
entry_price = current_price
|
||
highest_profit = 0.0
|
||
|
||
return strategy_returns
|
||
|
||
def get_config(self) -> dict[str, float]:
|
||
"""Return risk management configuration."""
|
||
return {
|
||
"stop_loss": self.stop_loss,
|
||
"take_profit": self.take_profit,
|
||
"trailing_stop": self.trailing_stop,
|
||
"trailing_activation": self.trailing_activation,
|
||
"max_daily_loss": self.max_daily_loss,
|
||
"max_positions": self.max_positions,
|
||
"risk_reward_ratio": self.take_profit / self.stop_loss,
|
||
}
|
||
|
||
# ============================================================================
|
||
# Strategy Evaluator
|
||
# ============================================================================
|
||
class StrategyEvaluator:
|
||
"""
|
||
Comprehensive strategy evaluation with RiskMgmt metrics. # nosec
|
||
"""
|
||
|
||
def __init__(self, trading_style: str = "daytrading", forward_bars: int = 96):
|
||
self.trading_style = trading_style
|
||
self.forward_bars = forward_bars
|
||
self.criteria = ACCEPTANCE_CRITERIA.get(trading_style, ACCEPTANCE_CRITERIA["daytrading"])
|
||
|
||
def evaluate( # nosec
|
||
self,
|
||
signal: pd.Series,
|
||
close: pd.Series,
|
||
strategy_returns: pd.Series,
|
||
) -> dict[str, Any]:
|
||
"""
|
||
Evaluate strategy with comprehensive metrics.
|
||
|
||
Parameters
|
||
----------
|
||
signal : pd.Series
|
||
Trading signals
|
||
close : pd.Series
|
||
Close prices
|
||
strategy_returns : pd.Series
|
||
Strategy returns after risk management
|
||
|
||
Returns
|
||
-------
|
||
dict
|
||
Evaluation metrics dict
|
||
"""
|
||
if len(strategy_returns) < 10:
|
||
return {"status": "failed", "reason": "Insufficient data"}
|
||
|
||
# Forward returns for IC calculation
|
||
fwd_returns = close.pct_change(self.forward_bars).shift(-self.forward_bars)
|
||
common_idx = signal.index.intersection(fwd_returns.dropna().index)
|
||
|
||
if len(common_idx) < 10:
|
||
return {"status": "failed", "reason": "Insufficient overlapping data"}
|
||
|
||
signal_aligned = signal.loc[common_idx]
|
||
fwd_aligned = fwd_returns.loc[common_idx]
|
||
|
||
# IC (Information Coefficient)
|
||
ic = signal_aligned.corr(fwd_aligned) if signal_aligned.std() > 0 else 0.0
|
||
|
||
# Basic metrics
|
||
total_bars = len(strategy_returns)
|
||
n_signals = int((signal != signal.shift(1)).sum())
|
||
n_long = int((signal == 1).sum())
|
||
n_short = int((signal == -1).sum())
|
||
n_neutral = int((signal == 0).sum())
|
||
|
||
# Returns metrics
|
||
cum_returns = (1 + strategy_returns).cumprod()
|
||
total_return = cum_returns.iloc[-1] - 1 if len(cum_returns) > 0 else 0.0
|
||
|
||
# Annualization factor (assuming 252 trading days, 1440 minutes per day)
|
||
bars_per_year = 252 * 1440 / self.forward_bars
|
||
n_months = total_bars / (bars_per_year / 12) if total_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
|
||
|
||
# Sharpe Ratio
|
||
if strategy_returns.std() > 0:
|
||
sharpe = strategy_returns.mean() / strategy_returns.std() * np.sqrt(bars_per_year)
|
||
else:
|
||
sharpe = 0.0
|
||
|
||
# Max Drawdown
|
||
running_max = cum_returns.expanding().max()
|
||
drawdown = (cum_returns - running_max) / running_max.replace(0, np.nan)
|
||
max_drawdown = drawdown.min() if len(drawdown) > 0 else 0.0
|
||
|
||
# Win Rate
|
||
active_returns = strategy_returns[strategy_returns != 0]
|
||
win_rate = (active_returns > 0).sum() / len(active_returns) if len(active_returns) > 0 else 0.0
|
||
|
||
# Daily loss analysis (for RiskMgmt compliance)
|
||
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.0
|
||
|
||
# Acceptance check
|
||
passed, failed_criteria = self._check_acceptance(
|
||
ic=ic if not np.isnan(ic) else 0,
|
||
sharpe=sharpe,
|
||
n_trades=n_signals,
|
||
max_drawdown=max_drawdown,
|
||
win_rate=win_rate,
|
||
monthly_return=monthly_return,
|
||
max_daily_loss=max_daily_loss,
|
||
)
|
||
|
||
result = {
|
||
"status": "accepted" if passed else "rejected",
|
||
"failed_criteria": failed_criteria,
|
||
|
||
# Core metrics
|
||
"ic": float(ic) if not np.isnan(ic) else 0.0,
|
||
"sharpe": float(sharpe),
|
||
"max_drawdown": float(max_drawdown),
|
||
"win_rate": float(win_rate),
|
||
"total_return": float(total_return),
|
||
"monthly_return_pct": float(monthly_return * 100),
|
||
"annual_return_pct": float(annual_return * 100),
|
||
|
||
# Trade statistics
|
||
"n_trades": n_signals,
|
||
"n_long": n_long,
|
||
"n_short": n_short,
|
||
"n_neutral": n_neutral,
|
||
"n_bars": total_bars,
|
||
"n_months": float(n_months),
|
||
|
||
# RiskMgmt compliance
|
||
"max_daily_loss": float(max_daily_loss),
|
||
"riskmgmt_compliant": max_daily_loss <= 0.05,
|
||
|
||
# Signal distribution
|
||
"signal_long_pct": n_long / total_bars if total_bars > 0 else 0,
|
||
"signal_short_pct": n_short / total_bars if total_bars > 0 else 0,
|
||
"signal_neutral_pct": n_neutral / total_bars if total_bars > 0 else 0,
|
||
}
|
||
|
||
return result
|
||
|
||
def _check_acceptance(
|
||
self,
|
||
ic: float,
|
||
sharpe: float,
|
||
n_trades: int,
|
||
max_drawdown: float,
|
||
win_rate: float,
|
||
monthly_return: float,
|
||
max_daily_loss: float,
|
||
) -> tuple[bool, list[str]]:
|
||
"""Check if strategy meets acceptance criteria."""
|
||
failed = []
|
||
|
||
if abs(ic) < self.criteria["min_abs_ic"]:
|
||
failed.append(f"IC too low: {ic:.4f} < {self.criteria['min_abs_ic']}")
|
||
|
||
if sharpe < self.criteria["min_sharpe"]:
|
||
failed.append(f"Sharpe too low: {sharpe:.3f} < {self.criteria['min_sharpe']}")
|
||
|
||
if n_trades < self.criteria["min_trades"]:
|
||
failed.append(f"Too few trades: {n_trades} < {self.criteria['min_trades']}")
|
||
|
||
if max_drawdown < self.criteria["max_drawdown"]:
|
||
failed.append(f"Max drawdown exceeded: {max_drawdown:.1%} < {self.criteria['max_drawdown']}")
|
||
|
||
if win_rate < self.criteria["min_win_rate"]:
|
||
failed.append(f"Win rate too low: {win_rate:.1%} < {self.criteria['min_win_rate']}")
|
||
|
||
if monthly_return < self.criteria["min_monthly_return"]:
|
||
failed.append(f"Monthly return too low: {monthly_return:.2%} < {self.criteria['min_monthly_return']}")
|
||
|
||
if max_daily_loss > self.criteria["max_daily_loss"]:
|
||
failed.append(f"Daily loss exceeded: {max_daily_loss:.2%} > {self.criteria['max_daily_loss']}")
|
||
|
||
return len(failed) == 0, failed
|
||
|
||
# ============================================================================
|
||
# Feedback Generator
|
||
# ============================================================================
|
||
class FeedbackGenerator:
|
||
"""
|
||
Generate intelligent feedback for LLM strategy improvement.
|
||
"""
|
||
|
||
@staticmethod
|
||
def generate_feedback(
|
||
evaluation: dict[str, Any], # nosec
|
||
factor_list: list[dict],
|
||
attempt: int,
|
||
param_config: dict | None = None,
|
||
) -> str:
|
||
"""
|
||
Generate actionable feedback based on strategy performance.
|
||
|
||
Parameters
|
||
----------
|
||
evaluation : dict # nosec
|
||
Strategy evaluation metrics # nosec
|
||
factor_list : list
|
||
Available factors with IC values
|
||
attempt : int
|
||
Current attempt number
|
||
param_config : dict, optional
|
||
Current parameter configuration
|
||
|
||
Returns
|
||
-------
|
||
str
|
||
Feedback string for LLM
|
||
"""
|
||
ic = evaluation.get("ic", 0) # nosec
|
||
sharpe = evaluation.get("sharpe", 0) # nosec
|
||
trades = evaluation.get("n_trades", 0) # nosec
|
||
dd = evaluation.get("max_drawdown", 0) # nosec
|
||
win_rate = evaluation.get("win_rate", 0) # nosec
|
||
monthly_ret = evaluation.get("monthly_return_pct", 0) # nosec
|
||
failed = evaluation.get("failed_criteria", []) # nosec
|
||
|
||
feedback_parts = [f"Attempt {attempt} results:"]
|
||
|
||
# Performance summary
|
||
feedback_parts.append(f"IC={ic:.4f}, Sharpe={sharpe:.2f}, Trades={trades}, DD={dd:.1%}, WinRate={win_rate:.1%}, Monthly={monthly_ret:.2f}%")
|
||
|
||
# Specific suggestions based on failures
|
||
if failed:
|
||
feedback_parts.append("\nIssues found:")
|
||
|
||
if any("IC" in f for f in failed):
|
||
# Suggest top factors
|
||
top_factors = sorted(factor_list, key=lambda x: abs(x["ic"]), reverse=True)[:5]
|
||
top_factor_names = [f["name"] for f in top_factors]
|
||
feedback_parts.append(
|
||
f"\n- IC too low ({ic:.4f}). Try different factors. Top factors by IC: {', '.join(top_factor_names)}",
|
||
)
|
||
|
||
if any("trades" in f.lower() for f in failed):
|
||
feedback_parts.append(
|
||
f"\n- Too few trades ({trades}). Lower thresholds (try 0.2-0.3), use more sensitive factors, or reduce rolling window (10-20 bars)",
|
||
)
|
||
|
||
if any("drawdown" in f.lower() for f in failed):
|
||
feedback_parts.append(
|
||
f"\n- High drawdown ({dd:.1%}). Add filters (volatility, trend), reduce position size, or tighten stop loss",
|
||
)
|
||
|
||
if any("sharpe" in f.lower() for f in failed):
|
||
feedback_parts.append(
|
||
f"\n- Low Sharpe ({sharpe:.2f}). Improve signal quality: combine momentum + mean reversion, add regime filters",
|
||
)
|
||
|
||
if any("win rate" in f.lower() for f in failed):
|
||
feedback_parts.append(
|
||
f"\n- Low win rate ({win_rate:.1%}). Try higher take profit (4-6%), or add confirmation filters",
|
||
)
|
||
|
||
if any("monthly return" in f.lower() for f in failed):
|
||
feedback_parts.append(
|
||
f"\n- Low monthly return ({monthly_ret:.2%}). Increase signal frequency or use higher-IC factors",
|
||
)
|
||
|
||
else:
|
||
# Strategy passed - suggest optimization
|
||
feedback_parts.append("\n✓ Strategy meets all criteria!")
|
||
|
||
if sharpe < 1.5:
|
||
feedback_parts.append(
|
||
"\nTry optimizing: 1) Test SL=1.5% vs 2% 2) Test TP=3% vs 4% 3) Add trailing stop at 1.5%",
|
||
)
|
||
|
||
if abs(ic) < 0.05:
|
||
top_factors = sorted(factor_list, key=lambda x: abs(x["ic"]), reverse=True)[:3]
|
||
feedback_parts.append(
|
||
f"\nIC could be higher. Consider adding: {', '.join(f['name'] for f in top_factors)}",
|
||
)
|
||
|
||
if param_config:
|
||
feedback_parts.append(
|
||
f"\nCurrent params: threshold={param_config.get('threshold_entry', 'N/A')}, "
|
||
f"window={param_config.get('rolling_window', 'N/A')}, "
|
||
f"SL={param_config.get('stop_loss', 'N/A'):.1%}, "
|
||
f"TP={param_config.get('take_profit', 'N/A'):.1%}",
|
||
)
|
||
|
||
return " ".join(feedback_parts)
|
||
|
||
# ============================================================================
|
||
# LLM Strategy Generator
|
||
# ============================================================================
|
||
class LLMStrategyGenerator:
|
||
"""
|
||
Generate trading strategies using LLM with feedback loop.
|
||
"""
|
||
|
||
def __init__(self):
|
||
setup_llm_env()
|
||
|
||
def generate(
|
||
self,
|
||
factor_subset: list[dict],
|
||
feedback: str | None = None,
|
||
trading_style: str = "daytrading",
|
||
forward_bars: int = 96,
|
||
) -> dict[str, Any]:
|
||
"""
|
||
Generate a single strategy via qwen CLI.
|
||
|
||
Parameters
|
||
----------
|
||
factor_subset : list
|
||
List of factor dicts with 'name' and 'ic'
|
||
feedback : str, optional
|
||
Previous feedback for improvement
|
||
trading_style : str
|
||
'daytrading' or 'swing'
|
||
forward_bars : int
|
||
Forward return horizon
|
||
|
||
Returns
|
||
-------
|
||
dict
|
||
Strategy dict with 'status', 'strategy', 'error'
|
||
"""
|
||
try:
|
||
import re
|
||
import subprocess # nosec B404
|
||
|
||
factor_list = ", ".join([f"{f['name']} (IC={f['ic']:.4f})" for f in factor_subset])
|
||
factor_names = ", ".join([f["name"] for f in factor_subset])
|
||
|
||
feedback_text = f" Vorheriges Feedback: {feedback}" if feedback else " Erster Versuch - sei kreativ!"
|
||
|
||
prompt = f"""Du bist ein quantitativer Trading-Experte. Erzeuge eine EUR/USD Daytrading-Strategie als JSON.
|
||
|
||
Faktoren: {factor_list}
|
||
|
||
⚠️ WICHTIG - DU MUSST VIELE SIGNALE GENERIEREN! ⚠️
|
||
Die Strategie MUSS mindestens 50+ Trades über den Datensatz erzeugen.
|
||
Verwende DESHALB diese Regeln:
|
||
1. Schwellenwerte MÜSSEN niedrig sein: 0.1 bis 0.25 (NICHT höher!)
|
||
2. Verwende Z-Score Normalisierung mit FENSTERN VON 10-20 Bars (kurz!)
|
||
3. Erstelle Signale für JEDE Bar wo der Z-Score den Schwellenwert überschreitet
|
||
4. Vermeide zu strenge Filter - die Strategie soll AKTIV traden!
|
||
5. Kombiniere 2-4 Faktoren mit GEWICHTEN für diversifizierte Signale
|
||
|
||
BEISPIEL für gute Signal-Logik:
|
||
```python
|
||
z = (factor - factor.rolling(15).mean()) / factor.rolling(15).std()
|
||
signal = pd.Series(0, index=close.index)
|
||
signal[z > 0.15] = 1 # NIEDRIGER Schwellenwert = VIELE Signale!
|
||
signal[z < -0.15] = -1 # Auch negative Signale für Shorts
|
||
```
|
||
|
||
❌ SCHLECHT: signal[composite > 0.5] = 1 (zu streng, nur 1 Trade!)
|
||
✅ GUT: signal[composite > 0.15] = 1 (niedrig, viele Trades!)
|
||
|
||
Anforderungen:
|
||
- Trading-Stil: Daytrading mit {forward_bars}-Bar Forward Returns
|
||
- ZIEL: 50-200+ Trades gesamt (sehr aktiv!)
|
||
- Schwellenwerte: 0.1-0.25 (sehr niedrig!)
|
||
- Rolling Windows: 10-20 Bars (kurz!)
|
||
- Erstelle signal Series mit Werten 1, -1, 0
|
||
|
||
{feedback_text}
|
||
|
||
WICHTIG: Das JSON MUSS diese Felder haben:
|
||
{{
|
||
"strategy_name": "kurzer_Name",
|
||
"factor_names": ["faktor1", "faktor2"],
|
||
"description": "Ein Satz Beschreibung",
|
||
"code": "Python Code der signal Series erzeugt"
|
||
}}
|
||
|
||
Der Python Code MUSS mit DataFrame 'factors' und Series 'close' arbeiten und eine Series 'signal' erzeugen.
|
||
|
||
Antworte NUR mit dem JSON Objekt!"""
|
||
|
||
# Call qwen CLI
|
||
logger.info(f"Calling qwen CLI with prompt ({len(prompt)} chars)...")
|
||
result = subprocess.run( # nosec B603
|
||
["qwen", "-p", prompt],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=120,
|
||
cwd=str(Path(__file__).parent),
|
||
)
|
||
|
||
if result.returncode != 0:
|
||
logger.error(f"qwen CLI failed: {result.stderr[:300]}")
|
||
return {"status": "error", "error": f"qwen CLI failed: {result.stderr[:200]}"}
|
||
|
||
response = result.stdout.strip()
|
||
logger.info(f"qwen CLI response ({len(response)} chars)")
|
||
|
||
# Extract JSON from response
|
||
# qwen CLI might output to file OR stdout
|
||
# Check if a file was created in results/strategies_new/
|
||
import glob
|
||
new_files = glob.glob(str(STRATEGIES_DIR / "*.json"))
|
||
if new_files:
|
||
latest = max(new_files, key=os.path.getmtime)
|
||
if os.path.getmtime(latest) > time.time() - 120: # Created in last 120s
|
||
logger.info(f"Strategy file found: {latest}")
|
||
with open(latest) as f:
|
||
raw_data = json.load(f)
|
||
# Convert qwen CLI format to our format
|
||
strategy_data = self._convert_qwen_output(raw_data, factor_subset)
|
||
if strategy_data:
|
||
return {"status": "generated", "strategy": strategy_data}
|
||
|
||
# Otherwise parse JSON from stdout
|
||
# Try to find JSON object in response
|
||
json_match = re.search(r'\{[^{}]*"strategy_name"[^{}]*\}', response, re.DOTALL)
|
||
if json_match:
|
||
strategy_str = json_match.group()
|
||
raw_data = json.loads(strategy_str)
|
||
else:
|
||
# Try to parse entire response as JSON
|
||
raw_data = json.loads(response)
|
||
|
||
# Convert to our format
|
||
strategy_data = self._convert_qwen_output(raw_data, factor_subset)
|
||
if not strategy_data:
|
||
return {"status": "invalid", "error": "Could not convert qwen output"}
|
||
|
||
return {
|
||
"status": "generated",
|
||
"strategy": strategy_data,
|
||
}
|
||
|
||
except subprocess.TimeoutExpired: # nosec
|
||
return {"status": "error", "error": "qwen CLI timeout (120s)"}
|
||
except Exception as e:
|
||
logger.error(f"qwen CLI generation failed: {e}")
|
||
return {"status": "error", "error": str(e)[:300]}
|
||
|
||
def _convert_qwen_output(self, raw_data: dict, factors: list[dict]) -> dict | None:
|
||
"""
|
||
Convert qwen CLI output format to our standard format.
|
||
|
||
qwen CLI may output:
|
||
- code as string with literal \n
|
||
- Different field names (name vs strategy_name)
|
||
- Nested structures
|
||
|
||
We need:
|
||
- strategy_name: str
|
||
- factor_names: List[str]
|
||
- description: str
|
||
- code: str (executable Python with real newlines) # nosec
|
||
"""
|
||
try:
|
||
# Extract strategy name
|
||
strategy_name = raw_data.get("strategy_name") or raw_data.get("name", "UnknownStrategy")
|
||
|
||
# Extract factor names
|
||
factor_names = raw_data.get("factor_names", [])
|
||
if not factor_names:
|
||
# Use factors from the generation request
|
||
factor_names = [f["name"] for f in factors[:3]]
|
||
|
||
# Extract description
|
||
description = raw_data.get("description", raw_data.get("desc", "Generated strategy"))
|
||
|
||
# Extract and clean code
|
||
code = raw_data.get("code", "")
|
||
if not code:
|
||
# Try to find code in nested structures
|
||
if "strategy" in raw_data:
|
||
code = raw_data["strategy"].get("code", "")
|
||
elif "logic" in raw_data:
|
||
code = raw_data["logic"].get("code", "")
|
||
|
||
# Unescape code (convert literal \n to real newlines)
|
||
if code:
|
||
code = code.replace("\\n", "\n").replace('\\"', '"').replace("\\\\", "\\")
|
||
# Remove leading/trailing quotes if present
|
||
if code.startswith('"') and code.endswith('"'):
|
||
code = code[1:-1]
|
||
if code.startswith("'") and code.endswith("'"):
|
||
code = code[1:-1]
|
||
# Ensure variable name consistency: factors_df → factors
|
||
code = code.replace("factors_df", "factors")
|
||
|
||
# Validate we have what we need
|
||
if not code or not strategy_name:
|
||
logger.warning(f"Missing required fields: name={strategy_name}, code={'yes' if code else 'no'}")
|
||
return None
|
||
|
||
return {
|
||
"strategy_name": strategy_name,
|
||
"factor_names": factor_names,
|
||
"description": description,
|
||
"code": code,
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"Failed to convert qwen output: {e}")
|
||
return None
|
||
|
||
# ============================================================================
|
||
# Backtest Runner
|
||
# ============================================================================
|
||
class BacktestRunner:
|
||
"""
|
||
Run backtests in isolated subprocess with risk management. # nosec
|
||
"""
|
||
|
||
@staticmethod
|
||
def run(
|
||
close: pd.Series,
|
||
factors_df: pd.DataFrame,
|
||
strategy_code: str,
|
||
risk_config: dict[str, float],
|
||
forward_bars: int = 96,
|
||
) -> dict[str, Any] | None:
|
||
"""
|
||
Run strategy backtest with risk management.
|
||
|
||
Parameters
|
||
----------
|
||
close : pd.Series
|
||
Close prices
|
||
factors_df : pd.DataFrame
|
||
Factor values DataFrame
|
||
strategy_code : str
|
||
Python code string for signal generation
|
||
risk_config : dict
|
||
Risk management configuration (SL, TP, trailing, etc.)
|
||
forward_bars : int
|
||
Forward return horizon
|
||
|
||
Returns
|
||
-------
|
||
dict or None
|
||
Backtest results dict or None on failure
|
||
"""
|
||
# Build backtest script with risk management
|
||
risk_code = f"""
|
||
# Risk Management Configuration
|
||
STOP_LOSS = {risk_config['stop_loss']}
|
||
TAKE_PROFIT = {risk_config['take_profit']}
|
||
TRAILING_STOP = {risk_config['trailing_stop']}
|
||
TRAILING_ACTIVATION = {risk_config['trailing_activation']}
|
||
MAX_DAILY_LOSS = {risk_config['max_daily_loss']}
|
||
MAX_POSITIONS = {risk_config['max_positions']}
|
||
|
||
def apply_risk_management_with_params(signal, close_prices, sl, tp, trailing, trail_activation):
|
||
\"\"\"Apply SL/TP/Trailing stop to signals.\"\"\"
|
||
if len(signal) == 0 or len(close_prices) == 0:
|
||
return pd.Series(0.0, index=signal.index)
|
||
|
||
common_idx = signal.index.intersection(close_prices.index)
|
||
sig = signal.loc[common_idx].fillna(0)
|
||
prices = close_prices.loc[common_idx]
|
||
|
||
strategy_returns = pd.Series(0.0, index=common_idx)
|
||
position = 0
|
||
entry_price = 0.0
|
||
highest_profit = 0.0
|
||
daily_pnl = 0.0
|
||
current_date = None
|
||
|
||
for i, idx in enumerate(common_idx):
|
||
if i == 0:
|
||
continue
|
||
|
||
bar_date = idx.date() if hasattr(idx, 'date') else idx
|
||
if current_date is None:
|
||
current_date = bar_date
|
||
elif bar_date != current_date:
|
||
daily_pnl = 0.0
|
||
current_date = bar_date
|
||
|
||
current_price = prices.iloc[i]
|
||
prev_price = prices.iloc[i - 1]
|
||
current_signal = sig.iloc[i]
|
||
|
||
if position != 0:
|
||
pnl_pct = 0.0
|
||
if position == 1:
|
||
pnl_pct = (current_price - entry_price) / entry_price
|
||
elif position == -1:
|
||
pnl_pct = (entry_price - current_price) / entry_price
|
||
|
||
# Stop Loss
|
||
if pnl_pct <= -sl:
|
||
strategy_returns.iloc[i] = -sl * position
|
||
daily_pnl += -sl
|
||
position = 0
|
||
highest_profit = 0.0
|
||
continue
|
||
|
||
# Take Profit
|
||
if pnl_pct >= tp:
|
||
strategy_returns.iloc[i] = tp * position
|
||
daily_pnl += tp
|
||
position = 0
|
||
highest_profit = 0.0
|
||
continue
|
||
|
||
# Trailing Stop
|
||
if pnl_pct >= trail_activation:
|
||
highest_profit = max(highest_profit, pnl_pct)
|
||
if (highest_profit - pnl_pct) >= trailing:
|
||
strategy_returns.iloc[i] = pnl_pct * position
|
||
daily_pnl += pnl_pct
|
||
position = 0
|
||
highest_profit = 0.0
|
||
continue
|
||
|
||
# Normal PnL
|
||
if position == 1:
|
||
strategy_returns.iloc[i] = (current_price - prev_price) / prev_price
|
||
elif position == -1:
|
||
strategy_returns.iloc[i] = -(current_price - prev_price) / prev_price
|
||
|
||
daily_pnl += strategy_returns.iloc[i]
|
||
|
||
# Max daily loss
|
||
if daily_pnl <= -{risk_config['max_daily_loss']}:
|
||
position = 0
|
||
highest_profit = 0.0
|
||
continue
|
||
|
||
# Enter position
|
||
if position == 0 and current_signal != 0:
|
||
position = int(np.sign(current_signal))
|
||
entry_price = current_price
|
||
highest_profit = 0.0
|
||
|
||
return strategy_returns
|
||
"""
|
||
|
||
script = f"""
|
||
import pandas as pd
|
||
import numpy as np
|
||
import json
|
||
import sys
|
||
|
||
close = pd.read_pickle('close.pkl') # nosec
|
||
factors = pd.read_pickle('factors.pkl') # nosec
|
||
|
||
try:
|
||
{chr(10).join(' ' + line for line in strategy_code.split(chr(10)))}
|
||
except Exception as e:
|
||
print(f"ERROR: Strategy execution failed: {{e}}", file=sys.stderr) # nosec
|
||
sys.exit(1)
|
||
|
||
if 'signal' not in dir():
|
||
print("ERROR: No signal variable created", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
# Apply risk management
|
||
{risk_code}
|
||
|
||
signal = signal.fillna(0)
|
||
strategy_returns = apply_risk_management_with_params(signal, close, STOP_LOSS, TAKE_PROFIT, TRAILING_STOP, TRAILING_ACTIVATION)
|
||
|
||
# Calculate metrics
|
||
common_idx = close.index.intersection(signal.index)
|
||
close_aligned = close.loc[common_idx]
|
||
signal_aligned = signal.loc[common_idx]
|
||
fwd_returns = close_aligned.pct_change({forward_bars}).shift(-{forward_bars})
|
||
|
||
ic = signal_aligned.corr(fwd_returns.dropna()) if signal_aligned.std() > 0 else 0
|
||
total_return = (1 + strategy_returns).prod() - 1
|
||
cum_returns = (1 + strategy_returns).cumprod()
|
||
running_max = cum_returns.expanding().max()
|
||
drawdown = (cum_returns - running_max) / running_max.replace(0, np.nan)
|
||
max_dd = drawdown.min() if len(drawdown) > 0 else 0
|
||
|
||
active_returns = strategy_returns[strategy_returns != 0]
|
||
win_rate = (active_returns > 0).sum() / len(active_returns) if len(active_returns) > 0 else 0
|
||
n_trades = int((signal_aligned != signal_aligned.shift(1)).sum())
|
||
|
||
bars_per_year = 252 * 1440 / {forward_bars}
|
||
if strategy_returns.std() > 0:
|
||
sharpe = strategy_returns.mean() / strategy_returns.std() * np.sqrt(bars_per_year)
|
||
else:
|
||
sharpe = 0
|
||
|
||
n_bars = len(strategy_returns)
|
||
n_months = n_bars / (bars_per_year / 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
|
||
|
||
# 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
|
||
|
||
result = {{
|
||
"status": "success",
|
||
"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),
|
||
"annual_return_pct": float(annual_return * 100),
|
||
"n_bars": int(n_bars),
|
||
"n_months": float(n_months),
|
||
"n_long": int((signal_aligned == 1).sum()),
|
||
"n_short": int((signal_aligned == -1).sum()),
|
||
"n_neutral": int((signal_aligned == 0).sum()),
|
||
"max_daily_loss": float(max_daily_loss),
|
||
"riskmgmt_compliant": max_daily_loss <= 0.05,
|
||
}}
|
||
|
||
def sanitize_val(v):
|
||
if isinstance(v, (np.integer,)): return int(v)
|
||
if isinstance(v, (np.floating,)): return float(v)
|
||
if isinstance(v, np.bool_): return bool(v)
|
||
if isinstance(v, float):
|
||
import math
|
||
if math.isnan(v): return 0.0
|
||
if math.isinf(v): return -999.0 if v < 0 else 999.0
|
||
return v
|
||
|
||
result = {{k: sanitize_val(v) for k, v in result.items()}}
|
||
print(json.dumps(result))
|
||
"""
|
||
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as td:
|
||
td_path = Path(td)
|
||
close.to_pickle(str(td_path / "close.pkl")) # nosec
|
||
factors_df.to_pickle(str(td_path / "factors.pkl")) # nosec
|
||
(td_path / "run.py").write_text(script)
|
||
|
||
try:
|
||
result = subprocess.run( # nosec B603
|
||
[sys.executable, str(td_path / "run.py")],
|
||
capture_output=True, text=True, timeout=300,
|
||
cwd=str(td_path),
|
||
)
|
||
|
||
if result.returncode != 0:
|
||
logger.warning(f"Backtest failed: {result.stderr[:200] or result.stdout[:200]}")
|
||
return {"status": "failed", "reason": result.stderr[:200] or result.stdout[:200]}
|
||
|
||
for line in result.stdout.strip().split("\n"):
|
||
try:
|
||
return json.loads(line)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
|
||
return {"status": "failed", "reason": "No valid JSON output"}
|
||
|
||
except subprocess.TimeoutExpired: # nosec
|
||
return {"status": "failed", "reason": "Timeout (90s)"}
|
||
except Exception as e:
|
||
return {"status": "failed", "reason": str(e)[:200]}
|
||
|
||
# ============================================================================
|
||
# Parameter Optimizer
|
||
# ============================================================================
|
||
class ParameterOptimizer:
|
||
"""
|
||
Grid search for optimal strategy parameters.
|
||
"""
|
||
|
||
def __init__(self, max_combinations: int = 50):
|
||
"""
|
||
Initialize optimizer.
|
||
|
||
Parameters
|
||
----------
|
||
max_combinations : int
|
||
Maximum parameter combinations to test
|
||
"""
|
||
self.max_combinations = max_combinations
|
||
|
||
def optimize(
|
||
self,
|
||
close: pd.Series,
|
||
factors_df: pd.DataFrame,
|
||
strategy_code: str,
|
||
forward_bars: int = 96,
|
||
) -> tuple[dict[str, float], dict[str, Any]]:
|
||
"""
|
||
Optimize strategy parameters via grid search.
|
||
|
||
Parameters
|
||
----------
|
||
close : pd.Series
|
||
Close prices
|
||
factors_df : pd.DataFrame
|
||
Factor values
|
||
strategy_code : str
|
||
Strategy Python code
|
||
forward_bars : int
|
||
Forward return horizon
|
||
|
||
Returns
|
||
-------
|
||
tuple
|
||
(best_params, best_result)
|
||
"""
|
||
# Generate parameter combinations (sample if too many)
|
||
all_combinations = list(product(
|
||
PARAMETER_GRID["threshold_entry"],
|
||
PARAMETER_GRID["rolling_window"],
|
||
PARAMETER_GRID["stop_loss"],
|
||
PARAMETER_GRID["take_profit"],
|
||
PARAMETER_GRID["trailing_stop"],
|
||
PARAMETER_GRID["trailing_activation"],
|
||
))
|
||
|
||
# Filter invalid combinations (TP must be >= 2x SL)
|
||
valid_combinations = [
|
||
c for c in all_combinations
|
||
if c[3] >= c[2] * 2 # take_profit >= 2 * stop_loss
|
||
]
|
||
|
||
# Sample if too many
|
||
if len(valid_combinations) > self.max_combinations:
|
||
valid_combinations = random.sample(valid_combinations, self.max_combinations)
|
||
|
||
logger.info(f"Testing {len(valid_combinations)} parameter combinations...")
|
||
|
||
best_result = None
|
||
best_params = None
|
||
best_score = -np.inf
|
||
|
||
runner = BacktestRunner()
|
||
|
||
for idx, (threshold, window, sl, tp, trail, trail_act) in enumerate(valid_combinations):
|
||
# Modify strategy code with current parameters
|
||
param_code = self._inject_parameters(strategy_code, threshold, window)
|
||
|
||
# Risk config for this combination
|
||
risk_config = {
|
||
"stop_loss": sl,
|
||
"take_profit": tp,
|
||
"trailing_stop": trail,
|
||
"trailing_activation": trail_act,
|
||
"max_daily_loss": 0.05,
|
||
"max_positions": 1,
|
||
}
|
||
|
||
# Run backtest
|
||
result = runner.run(close, factors_df, param_code, risk_config, forward_bars)
|
||
|
||
if result and result.get("status") == "success":
|
||
# Score: prioritize IC and Sharpe, penalize drawdown and low trades
|
||
score = (
|
||
abs(result.get("ic", 0)) * 10 +
|
||
result.get("sharpe", 0) * 2 -
|
||
abs(result.get("max_drawdown", 0)) * 5 +
|
||
min(result.get("n_trades", 0) / 100, 2)
|
||
)
|
||
|
||
if score > best_score:
|
||
best_score = score
|
||
best_params = {
|
||
"threshold_entry": threshold,
|
||
"rolling_window": window,
|
||
"stop_loss": sl,
|
||
"take_profit": tp,
|
||
"trailing_stop": trail,
|
||
"trailing_activation": trail_act,
|
||
}
|
||
best_result = result
|
||
|
||
if (idx + 1) % 10 == 0:
|
||
logger.info(f" Tested {idx + 1}/{len(valid_combinations)} combinations, best score={best_score:.3f}")
|
||
|
||
if best_result is None:
|
||
logger.warning("No successful backtests found, using default parameters")
|
||
best_params = {
|
||
"threshold_entry": 0.3,
|
||
"rolling_window": 20,
|
||
"stop_loss": 0.02,
|
||
"take_profit": 0.04,
|
||
"trailing_stop": 0.015,
|
||
"trailing_activation": 0.02,
|
||
}
|
||
best_result = {"status": "failed", "reason": "No valid parameters found"}
|
||
|
||
return best_params, best_result
|
||
|
||
def _inject_parameters(
|
||
self,
|
||
strategy_code: str,
|
||
threshold: float,
|
||
window: int,
|
||
) -> str:
|
||
"""
|
||
Inject parameters into strategy code - DISABLED for stability.
|
||
qwen CLI generates code with its own thresholds which work better.
|
||
"""
|
||
# Don't modify qwen CLI generated code - it already has good parameters
|
||
return strategy_code
|
||
|
||
# ============================================================================
|
||
# Smart Strategy Generator (Main Class)
|
||
# ============================================================================
|
||
class SmartStrategyGenerator:
|
||
"""
|
||
Main strategy generator with feedback loop, optimization, and risk management.
|
||
|
||
Usage:
|
||
generator = SmartStrategyGenerator(trading_style='daytrading')
|
||
strategies = generator.generate_strategies(target_count=10)
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
trading_style: str = "daytrading",
|
||
forward_bars: int | None = None,
|
||
max_attempts: int = 100,
|
||
enable_optimization: bool = True,
|
||
):
|
||
"""
|
||
Initialize strategy generator.
|
||
|
||
Parameters
|
||
----------
|
||
trading_style : str
|
||
'daytrading' or 'swing'
|
||
forward_bars : int, optional
|
||
Forward return horizon (auto-detected from style)
|
||
max_attempts : int
|
||
Maximum generation attempts
|
||
enable_optimization : bool
|
||
Enable parameter grid search
|
||
"""
|
||
self.trading_style = trading_style
|
||
self.forward_bars = forward_bars or (12 if trading_style == "daytrading" else 96)
|
||
self.max_attempts = max_attempts
|
||
self.enable_optimization = enable_optimization
|
||
|
||
self.llm_generator = LLMStrategyGenerator()
|
||
self.evaluator = StrategyEvaluator(trading_style, self.forward_bars) # nosec
|
||
self.feedback_gen = FeedbackGenerator()
|
||
self.optimizer = ParameterOptimizer(max_combinations=15)
|
||
self.backtest_runner = BacktestRunner()
|
||
|
||
self.factors = data_cache.load_top_factors(20)
|
||
self.close = data_cache.load_ohlcv()
|
||
|
||
# Load factor time-series
|
||
self.factor_data = {}
|
||
for f_info in self.factors:
|
||
series = data_cache.load_factor_timeseries(f_info["name"])
|
||
if series is not None:
|
||
self.factor_data[f_info["name"]] = series
|
||
|
||
# Align data
|
||
all_series = [self.factor_data[n] for n in self.factor_data]
|
||
if not all_series:
|
||
raise ValueError("No factor data loaded!")
|
||
|
||
self.df_factors = pd.DataFrame({n: self.factor_data[n] for n in self.factor_data})
|
||
self.common_idx = self.close.index.intersection(self.df_factors.dropna(how="all").index)
|
||
self.close_aligned = self.close.loc[self.common_idx]
|
||
self.df_aligned = self.df_factors.loc[self.common_idx]
|
||
|
||
self.accepted_strategies: list[dict] = []
|
||
self.feedback_history: list[str] = []
|
||
|
||
logger.info(
|
||
f"SmartStrategyGenerator initialized: style={trading_style}, "
|
||
f"forward_bars={self.forward_bars}, factors={len(self.factor_data)}, "
|
||
f"bars={len(self.close_aligned):,}",
|
||
)
|
||
|
||
def generate_strategy(
|
||
self,
|
||
attempt_idx: int,
|
||
factor_subset: list[dict] | None = None,
|
||
feedback: str | None = None,
|
||
) -> dict | None:
|
||
"""
|
||
Generate a single strategy with feedback loop.
|
||
|
||
Parameters
|
||
----------
|
||
attempt_idx : int
|
||
Attempt number (for logging)
|
||
factor_subset : list, optional
|
||
Subset of factors to use (random if None)
|
||
feedback : str, optional
|
||
Previous feedback
|
||
|
||
Returns
|
||
-------
|
||
dict or None
|
||
Strategy dict or None if failed
|
||
"""
|
||
# Select factor subset
|
||
if factor_subset is None:
|
||
n_factors = random.randint(2, min(5, len(self.factors)))
|
||
factor_subset = random.sample(self.factors, n_factors)
|
||
|
||
# Generate strategy via LLM
|
||
gen_result = self.llm_generator.generate(
|
||
factor_subset=factor_subset,
|
||
feedback=feedback,
|
||
trading_style=self.trading_style,
|
||
forward_bars=self.forward_bars,
|
||
)
|
||
|
||
if gen_result["status"] != "generated":
|
||
logger.warning(f"Attempt {attempt_idx}: LLM generation failed - {gen_result.get('error', 'Unknown')}")
|
||
return None
|
||
|
||
strategy = gen_result["strategy"]
|
||
factor_names = strategy.get("factor_names", [])
|
||
|
||
# Build factors DataFrame
|
||
valid_factors = [f for f in factor_names if f in self.df_aligned.columns]
|
||
if len(valid_factors) < 2:
|
||
logger.warning(f"Attempt {attempt_idx}: Insufficient valid factors ({len(valid_factors)})")
|
||
return None
|
||
|
||
factors_df = self.df_aligned[valid_factors]
|
||
|
||
# Default risk config
|
||
risk_config = {
|
||
"stop_loss": 0.02,
|
||
"take_profit": 0.04,
|
||
"trailing_stop": 0.015,
|
||
"trailing_activation": 0.02,
|
||
"max_daily_loss": 0.05,
|
||
"max_positions": 1,
|
||
}
|
||
|
||
# Parameter optimization (if enabled)
|
||
if self.enable_optimization:
|
||
logger.info(f"Attempt {attempt_idx}: Running parameter optimization...")
|
||
best_params, opt_result = self.optimizer.optimize(
|
||
self.close_aligned, factors_df, strategy["code"], self.forward_bars,
|
||
)
|
||
|
||
if opt_result.get("status") == "success":
|
||
risk_config.update(best_params)
|
||
logger.info(
|
||
f" Best params: threshold={best_params['threshold_entry']}, "
|
||
f"window={best_params['rolling_window']}, "
|
||
f"SL={best_params['stop_loss']:.1%}, TP={best_params['take_profit']:.1%}",
|
||
)
|
||
else:
|
||
logger.warning(" Optimization failed, using default parameters")
|
||
|
||
# Run final backtest with optimized/default risk config
|
||
bt_result = self.backtest_runner.run(
|
||
self.close_aligned, factors_df, strategy["code"], risk_config, self.forward_bars,
|
||
)
|
||
|
||
if bt_result is None or bt_result.get("status") != "success":
|
||
logger.warning(f"Attempt {attempt_idx}: Backtest failed - {bt_result.get('reason', 'Unknown') if bt_result else 'No result'}")
|
||
return None
|
||
|
||
# Evaluate strategy
|
||
# Reconstruct signal from backtest (approximate)
|
||
signal_approx = pd.Series(0, index=self.close_aligned.index[:bt_result.get("n_bars", len(self.close_aligned))])
|
||
evaluation = self.evaluator.evaluate( # nosec
|
||
signal=signal_approx,
|
||
close=self.close_aligned.iloc[:len(signal_approx)],
|
||
strategy_returns=pd.Series(dtype=float), # Already computed in backtest
|
||
)
|
||
|
||
# Use backtest metrics directly for evaluation # nosec
|
||
evaluation = { # nosec
|
||
"ic": bt_result.get("ic", 0),
|
||
"sharpe": bt_result.get("sharpe", 0),
|
||
"max_drawdown": bt_result.get("max_drawdown", 0),
|
||
"win_rate": bt_result.get("win_rate", 0),
|
||
"n_trades": bt_result.get("n_trades", 0),
|
||
"monthly_return": bt_result.get("monthly_return_pct", 0) / 100.0,
|
||
"max_daily_loss": bt_result.get("max_daily_loss", 0),
|
||
}
|
||
|
||
# Check acceptance
|
||
passed, failed_criteria = self.evaluator._check_acceptance(**evaluation) # nosec
|
||
evaluation["status"] = "accepted" if passed else "rejected" # nosec
|
||
evaluation["failed_criteria"] = failed_criteria # nosec
|
||
|
||
# Generate feedback
|
||
feedback = self.feedback_gen.generate_feedback(
|
||
evaluation=evaluation, # nosec
|
||
factor_list=self.factors,
|
||
attempt=attempt_idx,
|
||
param_config=risk_config,
|
||
)
|
||
self.feedback_history.append(feedback)
|
||
|
||
# Store strategy
|
||
strategy["metrics"] = bt_result
|
||
strategy["risk_config"] = risk_config
|
||
strategy["evaluation"] = evaluation # nosec
|
||
strategy["feedback"] = feedback
|
||
|
||
if passed:
|
||
logger.info(
|
||
f"✓ Strategy #{len(self.accepted_strategies)+1} ACCEPTED: "
|
||
f"IC={evaluation['ic']:.4f}, Sharpe={evaluation['sharpe']:.2f}, " # nosec
|
||
f"Trades={evaluation['n_trades']}, DD={evaluation['max_drawdown']:.1%}", # nosec
|
||
)
|
||
self.accepted_strategies.append(strategy)
|
||
else:
|
||
logger.info(
|
||
f"✗ Strategy REJECTED: {', '.join(failed_criteria[:3])}",
|
||
)
|
||
|
||
return strategy
|
||
|
||
def generate_strategies(self, target_count: int = 10) -> list[dict]:
|
||
"""
|
||
Generate multiple strategies with feedback loop.
|
||
|
||
Parameters
|
||
----------
|
||
target_count : int
|
||
Number of accepted strategies to generate
|
||
|
||
Returns
|
||
-------
|
||
list
|
||
List of accepted strategy dicts
|
||
"""
|
||
console.print("\n[bold cyan]🧠 Smart Strategy Generation[/bold cyan]")
|
||
console.print(f" Style: {self.trading_style}")
|
||
console.print(f" Forward bars: {self.forward_bars}")
|
||
console.print(f" Target: {target_count} accepted strategies")
|
||
console.print(f" Factors: {len(self.factor_data)}")
|
||
console.print(f" Data points: {len(self.close_aligned):,}\n")
|
||
|
||
max_attempts = min(self.max_attempts, target_count * 15)
|
||
accepted = []
|
||
|
||
with Progress(
|
||
SpinnerColumn(),
|
||
TextColumn("[bold blue]{task.description}"),
|
||
BarColumn(),
|
||
TextColumn("[bold green]{task.completed}/{task.total}"),
|
||
TimeElapsedColumn(),
|
||
) as progress:
|
||
task = progress.add_task(f"Generating {self.trading_style} strategies...", total=max_attempts)
|
||
|
||
for attempt in range(max_attempts):
|
||
if len(accepted) >= target_count:
|
||
break
|
||
|
||
progress.update(task, description=f"Attempt {attempt+1}...")
|
||
|
||
# Get feedback from last attempt
|
||
feedback = self.feedback_history[-1] if self.feedback_history and random.random() < 0.7 else None
|
||
|
||
strategy = self.generate_strategy(attempt, feedback=feedback)
|
||
|
||
if strategy and strategy["evaluation"]["status"] == "accepted": # nosec
|
||
accepted.append(strategy)
|
||
|
||
# Save strategy
|
||
self._save_strategy(strategy)
|
||
|
||
console.print(
|
||
f"[green]✓ Strategy #{len(accepted)}:[/green] {strategy['strategy_name']} "
|
||
f"IC={strategy['metrics'].get('ic', 0):.4f}, "
|
||
f"Sharpe={strategy['metrics'].get('sharpe', 0):.3f}, "
|
||
f"Trades={strategy['metrics'].get('n_trades', 0)}, "
|
||
f"DD={strategy['metrics'].get('max_drawdown', 0):.1%}, "
|
||
f"Monthly={strategy['metrics'].get('monthly_return_pct', 0):.2f}%",
|
||
)
|
||
|
||
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["metrics"].get("ic", 0), reverse=True)
|
||
|
||
table = Table(title=f"Top {len(accepted)} Accepted Strategies")
|
||
table.add_column("#", justify="right")
|
||
table.add_column("Name")
|
||
table.add_column("IC", justify="right")
|
||
table.add_column("Sharpe", justify="right")
|
||
table.add_column("Trades", justify="right")
|
||
table.add_column("Max DD", justify="right")
|
||
table.add_column("Monthly %", justify="right")
|
||
table.add_column("RiskMgmt", justify="center")
|
||
|
||
for i, s in enumerate(accepted, 1):
|
||
m = s["metrics"]
|
||
table.add_row(
|
||
str(i),
|
||
s["strategy_name"],
|
||
f"{m.get('ic', 0):.4f}",
|
||
f"{m.get('sharpe', 0):.3f}",
|
||
str(m.get("n_trades", 0)),
|
||
f"{m.get('max_drawdown', 0):.1%}",
|
||
f"{m.get('monthly_return_pct', 0):.2f}%",
|
||
"✅" if m.get("riskmgmt_compliant", False) else "❌",
|
||
)
|
||
|
||
console.print(table)
|
||
|
||
return accepted
|
||
|
||
def _save_strategy(self, strategy: dict) -> None:
|
||
"""Save strategy to JSON file."""
|
||
fname = f"{int(time.time())}_{strategy['strategy_name'].replace(' ', '_')[:50]}.json"
|
||
fpath = STRATEGIES_DIR / fname
|
||
|
||
# Convert numpy types for JSON serialization
|
||
def convert_numpy(obj):
|
||
if isinstance(obj, (np.integer,)):
|
||
return int(obj)
|
||
if isinstance(obj, (np.floating,)):
|
||
return float(obj)
|
||
if isinstance(obj, np.ndarray):
|
||
return obj.tolist()
|
||
return obj
|
||
|
||
strategy_serializable = {k: convert_numpy(v) for k, v in strategy.items()}
|
||
|
||
with open(fpath, "w") as f:
|
||
json.dump(strategy_serializable, f, indent=2, ensure_ascii=False)
|
||
|
||
# Generate PDF report if available
|
||
try:
|
||
from nexquant_strategy_report import StrategyPerformanceReporter
|
||
reporter = StrategyPerformanceReporter(strategy)
|
||
reporter.generate_report()
|
||
except Exception as e:
|
||
logger.debug(f"Failed to generate report: {e}")
|
||
|
||
logger.info(f"Saved strategy: {fpath}")
|
||
|
||
# ============================================================================
|
||
# CLI Interface
|
||
# ============================================================================
|
||
def parse_args():
|
||
"""Parse command line arguments."""
|
||
import argparse
|
||
|
||
parser = argparse.ArgumentParser(
|
||
description="Smart Strategy Generation with Feedback & Optimization",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog="""
|
||
Examples:
|
||
python nexquant_smart_strategy_gen.py 10
|
||
python nexquant_smart_strategy_gen.py 5 --style daytrading
|
||
python nexquant_smart_strategy_gen.py 20 --style swing --max-attempts 200
|
||
python nexquant_smart_strategy_gen.py 10 --no-optimization
|
||
""",
|
||
)
|
||
|
||
parser.add_argument(
|
||
"count",
|
||
type=int,
|
||
nargs="?",
|
||
default=10,
|
||
help="Number of strategies to generate (default: 10)",
|
||
)
|
||
parser.add_argument(
|
||
"--style",
|
||
choices=["daytrading", "swing"],
|
||
default="daytrading",
|
||
help="Trading style (default: daytrading)",
|
||
)
|
||
parser.add_argument(
|
||
"--forward-bars",
|
||
type=int,
|
||
default=None,
|
||
help="Forward return bars (auto: 12 for daytrading, 96 for swing)",
|
||
)
|
||
parser.add_argument(
|
||
"--max-attempts",
|
||
type=int,
|
||
default=150,
|
||
help="Maximum generation attempts (default: 150)",
|
||
)
|
||
parser.add_argument(
|
||
"--no-optimization",
|
||
action="store_true",
|
||
help="Disable parameter grid search",
|
||
)
|
||
parser.add_argument(
|
||
"--factors",
|
||
type=int,
|
||
default=20,
|
||
help="Number of top factors to consider (default: 20)",
|
||
)
|
||
|
||
return parser.parse_args()
|
||
|
||
def main():
|
||
"""Main entry point."""
|
||
args = parse_args()
|
||
|
||
console.print(f"\n[bold magenta]{'='*70}[/bold magenta]")
|
||
console.print("[bold]🤖 PREDIX Smart Strategy Generator[/bold]")
|
||
console.print(f"[bold magenta]{'='*70}[/bold magenta]\n")
|
||
|
||
try:
|
||
# Initialize generator
|
||
generator = SmartStrategyGenerator(
|
||
trading_style=args.style,
|
||
forward_bars=args.forward_bars,
|
||
max_attempts=args.max_attempts,
|
||
enable_optimization=not args.no_optimization,
|
||
)
|
||
|
||
# Generate strategies
|
||
strategies = generator.generate_strategies(target_count=args.count)
|
||
|
||
if strategies:
|
||
console.print(f"\n[bold green]✓ Success! {len(strategies)} strategies saved to:[/bold green]")
|
||
console.print(f" {STRATEGIES_DIR}\n")
|
||
else:
|
||
console.print("\n[bold yellow]⚠ No strategies met acceptance criteria[/bold yellow]")
|
||
console.print(" Try: --max-attempts 200 or --style swing\n")
|
||
|
||
except KeyboardInterrupt:
|
||
console.print("\n[yellow]Interrupted by user[/yellow]")
|
||
sys.exit(0)
|
||
except Exception as e:
|
||
logger.exception(f"Fatal error: {e}")
|
||
console.print(f"\n[red]✗ Fatal error: {e}[/red]")
|
||
sys.exit(1)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|