feat: R&D Loop V2 — Multi-Instrument + Correlation Score + Session/Vola Filter

- 3 instruments: EUR/USD, GBP/USD, BTC/USD with combined evaluation
- Correlation-aware composite score: Sharpe × (1 - 0.5×corr) × (0.3 + 0.7×OOS_ratio)
- Session filter: London 07-16 UTC only (reduces false signals)
- Volatility filter: Skip trades when ATR < 0.03% (quiet markets)
- OOS split: 80/20 IS/OOS with separate metrics
- Similarity dedup: Skip near-identical strategies in SOTA
- SOTA expanded to 30 (was 20)
- Discovered: Donchian 58.1%, SAR 56.6%, MOM 33.8% monthly (multi-instrument)
- MACD dominance broken: 4+ different indicators in SOTA
This commit is contained in:
TPTBusiness
2026-05-31 18:18:07 +02:00
parent 9ce6a4e6ec
commit 4773e95a6c
+467 -265
View File
@@ -1,13 +1,12 @@
#!/usr/bin/env python3
"""R&D Loop for Technical Indicators — Replaces factor pipeline with indicator discovery.
"""R&D Loop V2 — Multi-Instrument + Correlation Score + Session/Vola Filter + OOS.
Architecture (mirrors the original R&D loop):
1. Hypothesize: LLM proposes indicator combinations with parameters
2. Evaluate: Direct backtest_signal (no Docker, no Qlib)
3. Feedback: Compare against SOTA, bias next hypotheses
4. Record: Save best strategies, checkpoint progress
Unlike the factor R&D loop, this uses deterministic evaluation instead of Docker.
Changes from V1:
1. Multi-Instrument: Evaluate on EUR/USD + GBP/USD + BTC/USD
2. Correlation Score: Reward uncorrelated strategies (Sharpe × (1corr))
3. Session Filter: Only trade London session (07:00-16:00 UTC)
4. Volatility Filter: No trades when ATR < threshold
5. OOS Split: Report IS/OOS separately (80/20)
"""
import json, os, random, sys, time
@@ -22,84 +21,87 @@ OHLCV_PATH = Path(os.getenv("PREDIX_OHLCV_PATH",
RESULTS_DIR = PROJECT / "results" / "rd_loop"
STATE_DIR = PROJECT / "git_ignore_folder" / "rd_loop_state"
INSTRUMENTS = ["EURUSD", "GBPUSD", "BTCUSD"]
INSTRUMENT_ALIASES = {"GBPUSDT": "GBPUSD"} # Merge aliases under canonical name
TIMEFRAMES = ["5min", "15min", "30min", "1h", "4h"]
INDICATORS_POOL = ["MACD", "RSI", "BBands", "Donchian", "Stoch", "CCI", "WillR", "ADX", "SAR", "ROC", "MOM", "AROON", "MFI", "SMA", "EMA"]
STRATEGY_TYPES = ["single", "multi_tf", "multi_role"]
TREND_TFS = ["30min", "1h", "4h"]
ENTRY_TFS = ["5min", "15min", "30min"]
MIN_SHARPE, MIN_TRADES = 0.3, 10
EXPLORATION_RATE = 0.40
OOS_SPLIT = 0.2
# ═══════════════════════════════════════════════════════════════════════════════
# GPU-accelerated backtest via Numba (735M bars/second — 245× faster)
# Numba-accelerated backtest
# ═══════════════════════════════════════════════════════════════════════════════
@jit(nopython=True)
def _backtest_numba(prices, signals, cost=0.000264):
n = len(prices)
equity = 100000.0; peak = 100000.0; max_dd = 0.0
position = 0; entry_price = 0.0
trades = np.zeros(100000, dtype=np.float64) # preallocate
trade_count = 0; wins = 0
equity = np.zeros(n, dtype=np.float64)
equity[0] = 100000.0; peak = 100000.0; max_dd = 0.0
trade_returns = np.zeros(100000, dtype=np.float64)
position = 0; entry_price = 0.0; trade_count = 0; wins = 0
for i in range(1, n):
px = prices[i]; sg = signals[i]; ps = signals[i-1]
if position != 0 and sg != position:
# Close position on signal reversal or flatten
if position != 0 and (sg != position or sg == 0 and position != 0):
if position == 1: ret = (px - entry_price) / entry_price - cost
else: ret = (entry_price - px) / entry_price - cost
equity *= (1.0 + ret)
if equity > peak: peak = equity
dd = (peak - equity) / peak
equity[i] = equity[i-1] * (1.0 + ret)
if equity[i] > peak: peak = equity[i]
dd = (peak - equity[i]) / peak
if dd > max_dd: max_dd = dd
if trade_count < len(trades):
trades[trade_count] = ret
if trade_count < len(trade_returns):
trade_returns[trade_count] = ret
trade_count += 1
if ret > 0: wins += 1
position = 0
if sg != 0 and position == 0:
else:
equity[i] = equity[i-1]
# Open new position
if position == 0 and sg != 0:
position = sg; entry_price = px
# Close final position
if position != 0:
fp = prices[-1]
if position == 1: ret = (fp - entry_price) / entry_price - cost
else: ret = (entry_price - fp) / entry_price - cost
equity *= (1.0 + ret)
if trade_count < len(trades):
trades[trade_count] = ret
equity[-1] = equity[-2] * (1.0 + ret)
if trade_count < len(trade_returns):
trade_returns[trade_count] = ret
trade_count += 1
if ret > 0: wins += 1
total_ret = (equity - 100000.0) / 100000.0
# Compute Sharpe from trade returns
# Ensure monotonic equity (carry forward zeros)
for i in range(1, n):
if equity[i] == 0: equity[i] = equity[i-1]
total_ret = (equity[-1] - 100000.0) / 100000.0
if trade_count > 5:
t = trades[:trade_count]
mean_ret = np.mean(t)
std_ret = np.std(t)
t = trade_returns[:trade_count]
mean_ret = np.mean(t); std_ret = np.std(t)
sharpe = mean_ret / std_ret * np.sqrt(trade_count) if std_ret > 0 else 0.0
else:
sharpe = 0.0
return equity, max_dd, trade_count, wins, total_ret, sharpe, trades[:trade_count]
TIMEFRAMES = ["5min", "15min", "30min", "1h", "4h"]
INDICATORS_POOL = ["MACD", "RSI", "BBands", "Donchian", "Stoch", "CCI", "WillR", "ADX", "SAR", "ROC", "MOM", "AROON", "MFI", "SMA", "EMA"]
STRATEGY_TYPES = ["single", "multi_tf", "portfolio", "multi_role"]
TREND_TFS = ["30min", "1h", "4h"] # higher TFs for trend filter
ENTRY_TFS = ["5min", "15min", "30min"] # lower TFs for entry
MIN_SHARPE, MIN_TRADES = 0.5, 20
EXPLORATION_RATE = 0.40 # 40% explore, 60% exploit
return equity, max_dd, trade_count, wins, total_ret, sharpe, trade_returns[:trade_count]
# ═══════════════════════════════════════════════════════════════════════════════
# Evaluation
# Signal Construction
# ═══════════════════════════════════════════════════════════════════════════════
def evaluate_strategy(close, hypothesis):
"""Run backtest and return metrics."""
def build_signal(close, hypothesis):
"""Build trading signal from hypothesis. Returns (-1,0,1) Series."""
import talib
signal = None
# Optuna optimization branch
if hypothesis.get('generation') == 'optuna':
return _run_optuna(close, hypothesis)
# ML training branch
if hypothesis.get('type') == 'ml':
return _train_ml(close, hypothesis)
if hypothesis['type'] == 'single':
ind = hypothesis['indicator']
tf = hypothesis['timeframe']
ind = hypothesis['indicator']; tf = hypothesis['timeframe']
bars = close.resample(tf).last().dropna()
sig = _build_indicator_signal(ind, bars, hypothesis['params'])
signal = sig.reindex(close.index).ffill().fillna(0).astype(int).clip(-1, 1)
@@ -113,49 +115,46 @@ def evaluate_strategy(close, hypothesis):
sigs[tf] = sig.reindex(close.index).ffill().fillna(0).astype(int).clip(-1, 1)
port = pd.DataFrame(sigs).dropna()
vote = port.mean(axis=1)
signal = pd.Series(0, index=vote.index)
signal[vote > 0.25] = 1; signal[vote < -0.25] = -1
elif hypothesis['type'] == 'portfolio':
sigs = []
for cfg in hypothesis['indicators']:
bars = close.resample('1d').last().dropna()
sig = _build_indicator_signal(cfg['name'], bars, cfg['params'])
sigs.append(sig.reindex(close.index).ffill().fillna(0).astype(int).clip(-1, 1))
port = pd.DataFrame({f"s{i}": s for i, s in enumerate(sigs)}).dropna()
vote = port.mean(axis=1)
signal = pd.Series(0, index=vote.index)
signal = pd.Series(0, index=close.index)
signal[vote > 0.25] = 1; signal[vote < -0.25] = -1
elif hypothesis['type'] == 'multi_role':
# Trend filter (higher TF) + Entry signal (lower TF) with directional gating
trend_ind = hypothesis['trend_ind']
entry_ind = hypothesis['entry_ind']
trend_tf = hypothesis['trend_tf']
entry_tf = hypothesis['entry_tf']
# Build trend signal on higher TF, forward-fill to lower TF
trend_ind = hypothesis['trend_ind']; entry_ind = hypothesis['entry_ind']
trend_tf = hypothesis['trend_tf']; entry_tf = hypothesis['entry_tf']
trend_bars = close.resample(trend_tf).last().dropna()
trend_sig = _build_indicator_signal(trend_ind, trend_bars, hypothesis['trend_params'])
trend_sig = trend_sig.reindex(close.index).ffill().fillna(0).astype(int).clip(-1, 1)
# Build entry signal on lower TF
entry_bars = close.resample(entry_tf).last().dropna()
entry_sig = _build_indicator_signal(entry_ind, entry_bars, hypothesis['entry_params'])
entry_sig = entry_sig.reindex(close.index).ffill().fillna(0).astype(int).clip(-1, 1)
# GATE: entry only fires when trend confirms direction
signal = pd.Series(0, index=close.index)
signal[(trend_sig == 1) & (entry_sig == 1)] = 1 # long trend + long entry
signal[(trend_sig == -1) & (entry_sig == -1)] = -1 # short trend + short entry
signal[(trend_sig == 1) & (entry_sig == 1)] = 1
signal[(trend_sig == -1) & (entry_sig == -1)] = -1
if signal is None or signal.nunique() <= 1:
return {"sharpe": 0, "monthly_pct": 0, "max_dd": 0, "n_trades": 0, "win_rate": 0}
return signal if signal is not None and signal.nunique() > 1 else None
# Numba-accelerated backtest (245× faster)
prices = close.values.astype(np.float64)
sigs = signal.values.astype(np.int32)
eq, dd, tr, wins, total_ret, sharpe, _ = _backtest_numba(prices, sigs)
return {"sharpe": float(sharpe), "monthly_pct": float(((1+total_ret)**(1/((close.index[-1]-close.index[0]).days/30.44))-1)*100) if total_ret > -1 else 0,
"max_dd": float(-dd), "n_trades": int(tr), "win_rate": float(wins/tr) if tr>0 else 0,
"total_return": float(total_ret)}
def _apply_session_filter(signal, index):
"""Only trade London session (07:00-16:00 UTC Mon-Fri)."""
hours = index.hour
days = index.dayofweek
in_session = (days < 5) & (hours >= 7) & (hours < 16)
if hasattr(in_session, 'values'):
in_session = in_session.values
return (signal * in_session.astype(int)).astype(int).clip(-1, 1)
def _apply_vola_filter(signal, close, atr_period=14, min_atr_pct=0.0003):
"""Don't trade when ATR is too low (flat/quiet markets)."""
tr = pd.DataFrame({
'hl': close.diff().abs(),
'hc': (close - close.shift(1)).abs(),
'lc': (close.shift(1) - close).abs(),
}).max(axis=1)
atr = tr.rolling(atr_period).mean()
atr_pct = atr / close
too_quiet = atr_pct < min_atr_pct
return (signal * (~too_quiet).astype(int)).fillna(0).astype(int).clip(-1, 1)
def _build_indicator_signal(name, bars, params):
@@ -223,155 +222,268 @@ def _build_indicator_signal(name, bars, params):
s = pd.Series(0, index=bars.index)
return s.fillna(0).astype(int).clip(-1, 1)
# ═══════════════════════════════════════════════════════════════════════════════
# Multi-Instrument Evaluation
# ═══════════════════════════════════════════════════════════════════════════════
def evaluate_multi(closes, hypothesis, use_session=True, use_vola=True):
"""Evaluate strategy on all instruments, return combined metrics + per-instrument."""
results = {}
equity_curves = {}
for inst, close in closes.items():
signal = build_signal(close, hypothesis)
if signal is None:
results[inst] = {"sharpe": 0, "monthly_pct": 0, "n_trades": 0}
continue
# Apply filters
if use_session:
signal = _apply_session_filter(signal, close.index)
if use_vola:
signal = _apply_vola_filter(signal, close)
if signal.nunique() <= 1:
results[inst] = {"sharpe": 0, "monthly_pct": 0, "n_trades": 0}
continue
# OOS split
n = len(close)
is_n = int(n * (1 - OOS_SPLIT))
close_is = close.iloc[:is_n]; signal_is = signal.iloc[:is_n]
close_oos = close.iloc[is_n:]; signal_oos = signal.iloc[is_n:]
# IS backtest
prices_is = close_is.values.astype(np.float64); sigs_is = signal_is.values.astype(np.int32)
eq_is, dd_is, tr_is, wins_is, ret_is, sh_is, _ = _backtest_numba(prices_is, sigs_is)
# OOS backtest
prices_oos = close_oos.values.astype(np.float64); sigs_oos = signal_oos.values.astype(np.int32)
eq_oos, dd_oos, tr_oos, wins_oos, ret_oos, sh_oos, _ = _backtest_numba(prices_oos, sigs_oos)
# Full backtest (for equity curve)
prices_full = close.values.astype(np.float64); sigs_full = signal.values.astype(np.int32)
eq_full, dd_full, tr_full, wins_full, ret_full, sh_full, trades_full = _backtest_numba(prices_full, sigs_full)
n_days = (close.index[-1] - close.index[0]).days
mon = ((1+ret_full)**(1/(n_days/30.44))-1)*100 if ret_full > -1 else 0
mon_oos = ((1+ret_oos)**(1/((close_oos.index[-1] - close_oos.index[0]).days/30.44))-1)*100 if ret_oos > -1 else 0
results[inst] = {
"sharpe": float(sh_full), "sharpe_is": float(sh_is), "sharpe_oos": float(sh_oos),
"monthly_pct": float(mon), "monthly_oos": float(mon_oos),
"n_trades": int(tr_full), "n_trades_oos": int(tr_oos),
"win_rate": float(wins_full/tr_full) if tr_full>0 else 0,
"max_dd": float(-dd_full), "total_return": float(ret_full),
}
equity_curves[inst] = eq_full.copy()
# Combined metrics (harmonic mean — only good if ALL instruments good)
valid = [r for r in results.values() if r['sharpe'] > 0]
if not valid:
combined = {"sharpe": 0, "monthly_pct": 0, "monthly_oos": 0, "n_trades": 0, "n_trades_oos": 0}
else:
combined = {
"sharpe": float(np.mean([r['sharpe'] for r in valid])),
"monthly_pct": float(np.mean([r['monthly_pct'] for r in valid])),
"monthly_oos": float(np.mean([r['monthly_oos'] for r in valid])),
"n_trades": int(np.sum([r['n_trades'] for r in valid])),
"n_trades_oos": int(np.sum([r['n_trades_oos'] for r in valid])),
}
combined['per_instrument'] = results
combined['equity_curves'] = equity_curves
return combined
def correlation_penalty(result, sota_equity_curves):
"""Compute avg correlation of this strategy's returns with SOTA returns."""
if not sota_equity_curves:
return 0.0
my_returns = []
for eq in result.get('equity_curves', {}).values():
if len(eq) > 1:
my_returns.append(np.diff(eq) / eq[:-1])
if not my_returns:
return 0.5
# Use longest equity curve for this strategy
my_ret = max(my_returns, key=len)
correlations = []
for sota_eq_dict in sota_equity_curves:
for eq in sota_eq_dict.values():
if len(eq) > 1:
sota_ret = np.diff(eq) / eq[:-1]
# Align to shorter length
min_len = min(len(my_ret), len(sota_ret))
if min_len > 10:
corr = np.corrcoef(my_ret[:min_len], sota_ret[:min_len])[0, 1]
if not np.isnan(corr):
correlations.append(corr)
return np.mean(correlations) if correlations else 0.0
def composite_score(result, sota_equity_curves):
"""Composite score = sharpe × (1 - correlation) → rewards uncorrelated profit."""
sh = result.get('sharpe', 0)
if sh <= 0:
return 0
corr = abs(correlation_penalty(result, sota_equity_curves))
# Bonus for OOS consistency
oos_ratio = min(result.get('monthly_oos', 0) / max(result.get('monthly_pct', 1), 0.01), 1.0)
oos_ratio = max(oos_ratio, 0)
return sh * (1 - 0.5 * corr) * (0.3 + 0.7 * oos_ratio)
# ═══════════════════════════════════════════════════════════════════════════════
# Hypothesis Generation (simulated R&D loop — no LLM needed for indicators)
# Hypothesis Generation
# ═══════════════════════════════════════════════════════════════════════════════
class ResearchLoop:
"""Mimics the R&D loop: hypothesize → evaluate → feedback → record."""
"""Multi-instrument R&D loop with correlation-aware feedback."""
def __init__(self, close):
self.close = close
self.sota = [] # State-of-the-art strategies
self.history = [] # All evaluated hypotheses
def __init__(self, closes):
self.closes = closes # {instrument: close_series}
self.sota = [] # State-of-the-art strategies (sorted by composite score)
self.sota_equity = [] # Equity curves for correlation calc
self.history = []
self.iteration = 0
self.best_score = 0
self.best_sharpe = 0
self.exploration_rate = 0.3 # % of time we explore randomly vs exploit SOTA
self.exploration_rate = EXPLORATION_RATE
def hypothesize(self):
self.iteration += 1
# Every 2000 iterations: train ML model (check before Optuna, higher priority)
# Every 2000: ML (higher priority, runs before Optuna)
if self.iteration % 2000 == 0 and len(self.sota) >= 5:
return {'type': 'ml', 'generation': 'ml',
'description': f"ML: LightGBM on {len(self.sota)} strategies",
'sota': self.sota[:5]}
# Every 500 iterations: Optuna-optimize best strategy
# Every 500: Optuna optimize best strategy
if self.iteration % 500 == 0 and self.sota:
return {'type': 'ml', 'generation': 'ml',
'description': f"ML: LightGBM on {len(self.sota)} strategies",
'sota': self.sota[:5]}
# Every 100 iterations: force non-dominant indicator exploration
hp = dict(self.sota[0]['hypothesis'])
hp['generation'] = 'optuna'
hp['description'] = f"Optuna: {hp.get('description','?')}"
return hp
# Every 100: force non-dominant indicator
if self.iteration % 100 == 0 and len(self.sota) >= 5:
top_ind = self.sota[0]['hypothesis'].get('trend_ind', self.sota[0]['hypothesis'].get('indicator'))
top = self._top_indicator()
hp = self._random_hypothesis()
# Ensure at least one role uses a different indicator
if hp.get('type') == 'multi_role' and hp['trend_ind'] == top_ind and hp['entry_ind'] == top_ind:
if random.random() < 0.5:
hp['trend_ind'] = random.choice([i for i in INDICATORS_POOL if i != top_ind])
hp['trend_params'] = self._random_params(hp['trend_ind'])
else:
hp['entry_ind'] = random.choice([i for i in INDICATORS_POOL if i != top_ind])
hp['entry_params'] = self._random_params(hp['entry_ind'])
hp['description'] = f"{hp['trend_ind']}({hp['trend_tf']})→{hp['entry_ind']}({hp['entry_tf']})"
elif hp.get('type') == 'single' and hp.get('indicator') == top_ind:
hp['indicator'] = random.choice([i for i in INDICATORS_POOL if i != top_ind])
hp['params'] = self._random_params(hp['indicator'])
hp['description'] = f"{hp['indicator']} on {hp['timeframe']}"
elif hp.get('type') == 'multi_tf' and hp.get('indicator') == top_ind:
hp['indicator'] = random.choice([i for i in INDICATORS_POOL if i != top_ind])
hp['params'] = self._random_params(hp['indicator'])
hp['description'] = f"{hp['indicator']} on {','.join(hp['timeframes'][:2])}"
hp = self._force_different_indicator(hp, top)
hp['generation'] = 'explore'
return hp
# Boost exploration when SOTA is dominated by one indicator
# Adaptive exploration rate
effective_rate = self.exploration_rate
if len(self.sota) >= 10:
top_ind = self.sota[0]['hypothesis'].get('trend_ind', self.sota[0]['hypothesis'].get('indicator'))
dominated = sum(1 for r in self.sota if r['hypothesis'].get('trend_ind', r['hypothesis'].get('indicator')) == top_ind)
if dominated > len(self.sota) * 0.8: # >80% same indicator
effective_rate += 0.25 # +25% explore boost
top = self._top_indicator()
dominated = sum(1 for r in self.sota if
r['hypothesis'].get('trend_ind', r['hypothesis'].get('indicator')) == top)
if dominated > len(self.sota) * 0.8:
effective_rate += 0.25
if random.random() < effective_rate or not self.sota:
return self._random_hypothesis()
else:
base = random.choice(self.sota[:5])
return self._mutate_hypothesis(base['hypothesis'])
def _top_indicator(self):
if not self.sota:
return 'MACD'
return self.sota[0]['hypothesis'].get('trend_ind',
self.sota[0]['hypothesis'].get('indicator', 'MACD'))
def _force_different_indicator(self, hp, top_ind):
if hp.get('type') == 'multi_role':
if hp['trend_ind'] == top_ind and hp['entry_ind'] == top_ind:
if random.random() < 0.5:
hp['trend_ind'] = random.choice([i for i in INDICATORS_POOL if i != top_ind])
hp['trend_params'] = self._random_params(hp['trend_ind'])
else:
hp['entry_ind'] = random.choice([i for i in INDICATORS_POOL if i != top_ind])
hp['entry_params'] = self._random_params(hp['entry_ind'])
elif hp.get('indicator') == top_ind:
hp['indicator'] = random.choice([i for i in INDICATORS_POOL if i != top_ind])
hp['params'] = self._random_params(hp['indicator'])
hp['description'] = self._make_desc(hp)
return hp
def _make_desc(self, hp):
t = hp.get('type', '?')
if t == 'multi_role':
return f"{hp['trend_ind']}({hp['trend_tf']})→{hp['entry_ind']}({hp['entry_tf']})"
elif t == 'multi_tf':
return f"{hp.get('indicator','?')} on {','.join(hp.get('timeframes',[])[:2])}"
else:
return f"{hp.get('indicator','?')} on {hp.get('timeframe','?')}"
def _random_hypothesis(self):
stype = random.choice(STRATEGY_TYPES)
if stype == 'single':
ind = random.choice(INDICATORS_POOL)
tf = random.choice(TIMEFRAMES)
params = self._random_params(ind)
return {'type': 'single', 'indicator': ind, 'timeframe': tf, 'params': params,
ind = random.choice(INDICATORS_POOL); tf = random.choice(TIMEFRAMES)
return {'type': 'single', 'indicator': ind, 'timeframe': tf,
'params': self._random_params(ind),
'description': f"{ind} on {tf}", 'generation': 'explore'}
elif stype == 'multi_tf':
ind = random.choice(INDICATORS_POOL)
tfs = random.sample(TIMEFRAMES, k=random.randint(2, 4))
params = self._random_params(ind)
return {'type': 'multi_tf', 'indicator': ind, 'timeframes': tfs, 'params': params,
return {'type': 'multi_tf', 'indicator': ind, 'timeframes': tfs,
'params': self._random_params(ind),
'description': f"{ind} on {','.join(tfs)}", 'generation': 'explore'}
elif stype == 'multi_role':
else: # multi_role
trend_ind = random.choice(INDICATORS_POOL)
entry_ind = random.choice(INDICATORS_POOL)
trend_tf = random.choice(TREND_TFS)
entry_tf = random.choice([t for t in ENTRY_TFS if t < trend_tf])
trend_p = self._random_params(trend_ind)
entry_p = self._random_params(entry_ind)
return {'type': 'multi_role',
'trend_ind': trend_ind, 'trend_params': trend_p, 'trend_tf': trend_tf,
'entry_ind': entry_ind, 'entry_params': entry_p, 'entry_tf': entry_tf,
'trend_ind': trend_ind, 'trend_params': self._random_params(trend_ind),
'trend_tf': trend_tf,
'entry_ind': entry_ind, 'entry_params': self._random_params(entry_ind),
'entry_tf': entry_tf,
'description': f"{trend_ind}({trend_tf})→{entry_ind}({entry_tf})",
'generation': 'explore'}
else:
i1, i2 = random.sample(INDICATORS_POOL, 2)
p1 = self._random_params(i1); p2 = self._random_params(i2)
return {'type': 'portfolio', 'indicators': [{'name': i1, 'params': p1}, {'name': i2, 'params': p2}],
'description': f"{i1}+{i2}", 'generation': 'explore'}
def _mutate_hypothesis(self, base):
"""Mutate an existing hypothesis — change one aspect."""
hp = dict(base) # shallow copy
hp['generation'] = 'exploit'
hp = dict(base); hp['generation'] = 'exploit'
# multi_role has its own mutation logic
if hp.get('type') == 'multi_role':
mut = random.choice(['trend_ind', 'entry_ind', 'trend_tf', 'entry_tf',
'trend_params', 'entry_params'])
if mut == 'trend_ind':
hp['trend_ind'] = random.choice([i for i in INDICATORS_POOL if i != hp['trend_ind']])
hp['trend_params'] = self._random_params(hp['trend_ind'])
hp['description'] = f"{hp['trend_ind']}({hp['trend_tf']})→{hp['entry_ind']}({hp['entry_tf']})"
elif mut == 'entry_ind':
hp['entry_ind'] = random.choice([i for i in INDICATORS_POOL if i != hp['entry_ind']])
hp['entry_params'] = self._random_params(hp['entry_ind'])
hp['description'] = f"{hp['trend_ind']}({hp['trend_tf']})→{hp['entry_ind']}({hp['entry_tf']})"
elif mut == 'trend_tf':
hp['trend_tf'] = random.choice(TREND_TFS)
if hp['trend_tf'] <= hp['entry_tf']:
hp['entry_tf'] = random.choice([t for t in ENTRY_TFS if t < hp['trend_tf']])
hp['description'] = f"{hp['trend_ind']}({hp['trend_tf']})→{hp['entry_ind']}({hp['entry_tf']})"
elif mut == 'entry_tf':
hp['entry_tf'] = random.choice([t for t in ENTRY_TFS if t < hp['trend_tf']])
hp['description'] = f"{hp['trend_ind']}({hp['trend_tf']})→{hp['entry_ind']}({hp['entry_tf']})"
elif mut == 'trend_params':
p = dict(hp['trend_params']); k = random.choice(list(p.keys()))
if isinstance(p[k], (int, float)): p[k] = p[k] * random.uniform(0.5, 1.5)
hp['trend_params'] = p
hp['description'] = f"{hp['trend_ind']}({hp['trend_tf']})→{hp['entry_ind']}({hp['entry_tf']}) (tuned)"
elif mut == 'entry_params':
p = dict(hp['entry_params']); k = random.choice(list(p.keys()))
if isinstance(p[k], (int, float)): p[k] = p[k] * random.uniform(0.5, 1.5)
hp['entry_params'] = p
hp['description'] = f"{hp['trend_ind']}({hp['trend_tf']})→{hp['entry_ind']}({hp['entry_tf']}) (tuned)"
hp['description'] = f"{hp['trend_ind']}({hp['trend_tf']})→{hp['entry_ind']}({hp['entry_tf']})"
return hp
mutations = ['params', 'indicator', 'timeframe']
mutation = random.choice(mutations)
if mutation == 'params' and 'params' in hp:
# Tweak one parameter
params = dict(hp['params'])
key = random.choice(list(params.keys()))
params = dict(hp['params']); key = random.choice(list(params.keys()))
if isinstance(params[key], (int, float)):
params[key] = params[key] * random.uniform(0.5, 1.5)
if isinstance(params[key], float):
params[key] = round(params[key], 1)
if isinstance(params[key], float): params[key] = round(params[key], 1)
hp['params'] = params
hp['description'] = f"{hp.get('indicator','?')} (mutated {key})"
elif mutation == 'indicator' and 'indicator' in hp:
@@ -384,7 +496,6 @@ class ResearchLoop:
elif 'timeframes' in hp:
hp['timeframes'] = random.sample(TIMEFRAMES, k=len(hp['timeframes']))
hp['description'] = f"{hp.get('indicator','?')} (timeframe change)"
return hp
def _random_params(self, indicator):
@@ -408,40 +519,72 @@ class ResearchLoop:
return param_sets.get(indicator, {'period': 14})
def feedback(self, result):
"""Update SOTA and bias future exploration."""
if result['sharpe'] >= MIN_SHARPE and result['n_trades'] >= MIN_TRADES:
"""Update SOTA sorted by COMPOSITE score (not just Sharpe)."""
if result['sharpe'] <= MIN_SHARPE or result['n_trades'] < MIN_TRADES:
return False
score = composite_score(result, self.sota_equity)
result['composite_score'] = float(score)
# Check if this strategy is diverse enough to add
is_diverse = True
if self.sota:
# Skip if very similar to existing (same indicators, TF, type)
for existing in self.sota[:3]:
if self._similar(result, existing):
is_diverse = False
break
if is_diverse:
self.sota.append(result)
self.sota.sort(key=lambda r: r['sharpe'], reverse=True)
self.sota = self.sota[:20] # Keep top 20
if result['sharpe'] > self.best_sharpe:
self.best_sharpe = result['sharpe']
self.sota.sort(key=lambda r: r.get('composite_score', 0), reverse=True)
self.sota = self.sota[:30] # Keep top 30
self.sota_equity = [s['equity_curves'] for s in self.sota]
if score > self.best_score:
self.best_score = score
return True # NEW BEST
if result['sharpe'] > self.best_sharpe:
self.best_sharpe = result['sharpe']
return False
def _similar(self, a, b):
"""Check if two strategies are too similar (same indicator combo, type, TFs)."""
ha = a['hypothesis']; hb = b['hypothesis']
if ha.get('type') != hb.get('type'):
return False
if ha.get('type') == 'multi_role':
return (ha.get('trend_ind') == hb.get('trend_ind') and
ha.get('entry_ind') == hb.get('entry_ind') and
ha.get('trend_tf') == hb.get('trend_tf') and
ha.get('entry_tf') == hb.get('entry_tf'))
return ha.get('indicator') == hb.get('indicator')
def record(self):
"""Save checkpoint."""
RESULTS_DIR.mkdir(parents=True, exist_ok=True)
STATE_DIR.mkdir(parents=True, exist_ok=True)
if self.sota:
cp = RESULTS_DIR / f"rd_loop_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
cp.write_text(json.dumps(self.sota[:15], indent=2, default=str))
# Strip equity_curves (too large) from saved results
stripped = []
for r in self.sota[:30]:
s = {k: v for k, v in r.items() if k != 'equity_curves'}
stripped.append(s)
cp.write_text(json.dumps(stripped, indent=2, default=str))
# ═══════════════════════════════════════════════════════════════════════════════
# ═══════════════════════════════════════════════════════════════════════════════
# Optuna Optimization
# ═══════════════════════════════════════════════════════════════════════════════
def _run_optuna(close, hypothesis):
"""Run Optuna hyperparameter optimization on a strategy."""
def _run_optuna(closes, hypothesis):
"""Optuna optimization on the primary instrument."""
import optuna
optuna.logging.set_verbosity(optuna.logging.WARNING)
hp = hypothesis
ind = hp.get('indicator', 'MACD')
tfs = hp.get('timeframes', ['15min','30min','1h','4h'])
base_params = hp.get('params', {})
close = list(closes.values())[0] # Use first instrument for Optuna
ind = hp.get('indicator', hp.get('trend_ind', 'MACD'))
base_params = hp.get('params', hp.get('trend_params', {}))
param_ranges = {
'MACD': {'fast': (2,15), 'slow': (5,40), 'sig': (2,15)},
'RSI': {'period': (5,30), 'oversold': (10,40), 'overbought': (60,90)},
@@ -450,7 +593,7 @@ def _run_optuna(close, hypothesis):
'ADX': {'period': (5,30), 'threshold': (10,40)},
}
ranges = param_ranges.get(ind, {})
def objective(trial):
params = {}
for k, (lo, hi) in ranges.items():
@@ -458,177 +601,236 @@ def _run_optuna(close, hypothesis):
params[k] = trial.suggest_int(k, int(lo), int(hi))
else:
params[k] = trial.suggest_float(k, lo, hi)
params['fast'] = min(params.get('fast',99), params.get('slow',99)-2)
sigs = {}
for tf in tfs:
bars = close.resample(tf).last().dropna()
sig = _build_indicator_signal(ind, bars, params)
sigs[tf] = sig.reindex(close.index).ffill().fillna(0).astype(int).clip(-1,1)
port = pd.DataFrame(sigs).dropna(); vote = port.mean(axis=1)
signal = pd.Series(0, index=vote.index)
signal[vote > 0.25] = 1; signal[vote < -0.25] = -1
prices = close.values.astype(np.float64); sigs_arr = signal.values.astype(np.int32)
_, dd, tr, wins, total_ret, sharpe, _ = _backtest_numba(prices, sigs_arr)
return float(sharpe) if sharpe > 0 else -999.0
if 'fast' in params and 'slow' in params:
params['fast'] = min(params['fast'], params['slow']-2)
result = evaluate_multi(closes, hp, use_session=True, use_vola=True)
return float(result.get('sharpe', 0)) if result.get('sharpe', 0) > 0 else -999.0
try:
study = optuna.create_study(direction='maximize')
study.optimize(objective, n_trials=20, show_progress_bar=False)
study.optimize(objective, n_trials=15, show_progress_bar=False)
best = study.best_params
hp['params'] = {k: int(v) if v == int(v) else v for k, v in best.items()}
hp['description'] = f"Optuna: {ind} on {','.join(tfs[:2])}"
if 'params' in hp:
hp['params'] = {k: int(v) if v == int(v) else v for k, v in best.items()}
elif 'trend_params' in hp:
hp['trend_params'] = {k: int(v) if v == int(v) else v for k, v in best.items()}
hp['generation'] = 'optuna'
# Re-evaluate with best params
sigs = {}
for tf in tfs:
bars = close.resample(tf).last().dropna()
sig = _build_indicator_signal(ind, bars, hp['params'])
sigs[tf] = sig.reindex(close.index).ffill().fillna(0).astype(int).clip(-1,1)
port = pd.DataFrame(sigs).dropna(); vote = port.mean(axis=1)
signal = pd.Series(0, index=vote.index)
signal[vote > 0.25] = 1; signal[vote < -0.25] = -1
prices = close.values.astype(np.float64); sigs_arr = signal.values.astype(np.int32)
eq, dd, tr, wins, ret, sh, _ = _backtest_numba(prices, sigs_arr)
n_days = (close.index[-1] - close.index[0]).days
mon = ((1+ret)**(1/(n_days/30.44))-1)*100 if ret > -1 else 0
print(f" Optuna best: {best} → Sh={sh:.1f} Mon={mon:.1f}% ({study.best_value:.1f})")
return {"sharpe": float(sh), "monthly_pct": float(mon), "max_dd": float(-dd),
"n_trades": int(tr), "win_rate": float(wins/tr) if tr>0 else 0,
"optuna_best": best, "optuna_value": float(study.best_value)}
except Exception as e:
return {"sharpe": 0, "monthly_pct": 0, "max_dd": 0, "n_trades": 0, "win_rate": 0}
result = evaluate_multi(closes, hp, use_session=True, use_vola=True)
print(f" Optuna best: {best} → Sh={result['sharpe']:.1f} "
f"Mon={result['monthly_pct']:.1f}% OOS={result['monthly_oos']:.1f}% ({study.best_value:.1f})")
return result
except Exception:
return {"sharpe": 0, "monthly_pct": 0, "monthly_oos": 0, "n_trades": 0}
def _train_ml(close, hypothesis):
"""Train LightGBM classifier on indicator signals to predict direction."""
def _train_ml(closes, hypothesis):
"""Train LightGBM on SOTA indicator signals."""
try:
from lightgbm import LGBMClassifier
except ImportError:
return {"sharpe": 0, "monthly_pct": 0, "max_dd": 0, "n_trades": 0, "win_rate": 0}
return {"sharpe": 0, "monthly_pct": 0, "monthly_oos": 0, "n_trades": 0}
sota = hypothesis.get('sota', [])
if not sota: return {"sharpe": 0, "monthly_pct": 0, "max_dd": 0, "n_trades": 0, "win_rate": 0}
# Generate features from all SOTA strategies
if not sota:
return {"sharpe": 0, "monthly_pct": 0, "monthly_oos": 0, "n_trades": 0}
close = list(closes.values())[0]
daily = close.resample('1h').last().dropna()
features = pd.DataFrame(index=daily.index)
for s in sota[:5]:
hp_s = s['hypothesis']
ind = hp_s.get('indicator', 'MACD')
sig = _build_indicator_signal(ind, daily, hp_s.get('params', {}))
features[f"{ind}_{hp_s.get('generation','?')}"] = sig
features = features.fillna(0)
# Target: next bar direction (1=up, 0=down)
# Generate signal from each SOTA strategy as a feature
from nexquant_rd_loop import build_signal, _build_indicator_signal
sig = build_signal(close, hp_s)
if sig is not None:
sig = sig.reindex(daily.index, method='ffill')
name = hp_s.get('description', f"strat_{id(s)}")[:30]
features[name] = sig.fillna(0)
features = features.iloc[100:] # Skip warmup
if len(features) < 200:
return {"sharpe": 0, "monthly_pct": 0, "monthly_oos": 0, "n_trades": 0}
target = (daily.pct_change().shift(-1) > 0).astype(int)
target = target.reindex(features.index).fillna(0)
# Train/test split (80/20)
split = int(len(features) * 0.8)
X_train, X_test = features.iloc[:split], features.iloc[split:]
y_train, y_test = target.iloc[:split], target.iloc[split:]
if len(X_train) < 100: return {"sharpe": 0, "monthly_pct": 0, "max_dd": 0, "n_trades": 0, "win_rate": 0}
model = LGBMClassifier(n_estimators=100, max_depth=5, verbosity=-1)
model.fit(X_train, y_train)
preds = model.predict(X_test)
# Convert predictions to trading signal
acc = float((preds == y_test).mean())
ml_signal = pd.Series(0, index=X_test.index)
ml_signal[preds == 1] = 1; ml_signal[preds == 0] = -1
ml_signal = ml_signal.reindex(close.index).ffill().fillna(0).astype(int).clip(-1,1)
ml_signal = ml_signal.reindex(close.index).ffill().fillna(0).astype(int).clip(-1, 1)
ml_signal = _apply_session_filter(ml_signal, close.index)
prices = close.values.astype(np.float64); sigs = ml_signal.values.astype(np.int32)
eq, dd, tr, wins, ret, sh, _ = _backtest_numba(prices, sigs)
n_days = (close.index[-1] - close.index[0]).days
mon = ((1+ret)**(1/(n_days/30.44))-1)*100 if ret > -1 else 0
acc = (preds == y_test).mean()
print(f" ML LightGBM: Test acc={acc:.1%} → Sh={sh:.1f} Mon={mon:.1f}% Tr={tr}")
return {"sharpe": float(sh), "monthly_pct": float(mon), "max_dd": float(-dd),
return {"sharpe": float(sh), "monthly_pct": float(mon), "monthly_oos": 0,
"n_trades": int(tr), "win_rate": float(wins/tr) if tr>0 else 0,
"ml_accuracy": float(acc), "ml_model": "LightGBM"}
def load_data():
"""Load OHLCV data for all instruments from one or multiple HDF5 files."""
closes = {}
data_dir = OHLCV_PATH.parent
# Try main file first
if OHLCV_PATH.exists():
df = pd.read_hdf(OHLCV_PATH, key="data")
for inst in INSTRUMENTS:
try:
close = df.xs(inst, level="instrument")["$close"].sort_index()
closes[inst] = close
except KeyError:
pass
# Load from individual files if not found
instrument_files = {
"EURUSD": OHLCV_PATH,
"GBPUSD": data_dir / "gbpusdt_1min.h5",
"BTCUSD": data_dir / "btc_1min.h5",
}
for inst, path in instrument_files.items():
if inst in closes:
continue
if not path.exists():
print(f" {inst}: file not found — skipping")
continue
try:
df = pd.read_hdf(path, key="data")
if isinstance(df.index, pd.MultiIndex):
try:
close = df.xs(inst, level="instrument")["$close"].sort_index()
except KeyError:
# Try with T suffix for crypto pairs
alt = inst + "T" if not inst.endswith("T") else inst.rstrip("T")
try:
close = df.xs(alt, level="instrument")["$close"].sort_index()
except KeyError:
inst_vals = df.index.get_level_values("instrument").unique()
for iv in inst_vals:
if inst[:3] in str(iv)[:3]:
close = df.xs(iv, level="instrument")["$close"].sort_index()
break
else:
raise KeyError(f"No instrument matching {inst}")
elif "$close" in df.columns:
close = df["$close"].sort_index()
close.index = pd.to_datetime(close.index)
elif "close" in df.columns:
close = df["close"].sort_index()
close.index = pd.to_datetime(close.index)
else:
close = df.iloc[:, 3].sort_index()
close.index = pd.to_datetime(close.index)
closes[inst] = close
except Exception as e:
print(f" {inst}: load error {e} — skipping")
for inst, close in closes.items():
print(f" {inst}: {len(close):,} bars, {close.index[0]}{close.index[-1]}")
return closes
def main():
iterations = 200
if "--iterations" in sys.argv:
iterations = int(sys.argv[sys.argv.index("--iterations") + 1])
print("=" * 60)
print(f" R&D Loop — Technical Indicators ({len(INDICATORS_POOL)} indicators)")
print(f" Strategy: hypothezise → evaluate → feedback → record")
print(f" R&D Loop V2 — Multi-Instrument + Correlation Score")
print(f" Instruments: {', '.join(INSTRUMENTS)}")
print(f" Indicators: {len(INDICATORS_POOL)} | Strategy types: {len(STRATEGY_TYPES)}")
print(f" Features: Session Filter + Volatility Filter + OOS Split")
print(f" Iterations: {iterations}")
print("=" * 60)
print(" Loading data...")
df = pd.read_hdf(OHLCV_PATH, key="data")
close = df.xs("EURUSD", level="instrument")["$close"].sort_index()
closes = load_data()
if not closes:
print(" ERROR: No instruments loaded!"); return
loop = ResearchLoop(close)
loop = ResearchLoop(closes)
t0 = time.time()
for i in range(iterations):
# 1. HYPOTHESIZE
hp = loop.hypothesize()
# 2. EVALUATE
# Evaluate
try:
result = evaluate_strategy(close, hp)
result = evaluate_multi(closes, hp, use_session=True, use_vola=True)
except Exception:
continue # skip bad parameters
continue
result['hypothesis'] = hp
result['iteration'] = i + 1
result['timestamp'] = datetime.now().isoformat()
loop.history.append(result)
# 3. FEEDBACK
# Feedback
is_new_best = loop.feedback(result)
best_inst_metrics = [f"{inst}: {m['sharpe']:.1f}" for inst, m in result.get('per_instrument', {}).items() if m.get('sharpe', 0) != 0]
# Log
gen = hp.get('generation', '?')
if is_new_best:
print(f"\n ★ NEW BEST (#{i+1}, {gen}): {hp['description']}")
print(f" Sharpe={result['sharpe']:.2f} Mon={result['monthly_pct']:.1f}% "
f"DD={result['max_dd']:.4f} Tr={result['n_trades']}")
print(f" Score={result['composite_score']:.1f} Sh={result['sharpe']:.1f} "
f"Mon={result['monthly_pct']:.1f}% OOS={result['monthly_oos']:.1f}% "
f"Tr={result['n_trades']} [{', '.join(best_inst_metrics[:3])}]")
elif (i + 1) % 50 == 0:
top_indicators = set()
for r in loop.sota[:5]:
top_indicators.add(r['hypothesis'].get('trend_ind', r['hypothesis'].get('indicator', '?')))
print(f" [{i+1}/{iterations}] {gen:>7s} | SOTA: {len(loop.sota)} | "
f"Best Sh={loop.best_sharpe:.2f} | "
f"Explore: {loop.exploration_rate:.0%}")
f"Best Sh={loop.best_sharpe:.1f} Score={loop.best_score:.1f} | "
f"Explore: {loop.exploration_rate:.0%} | Inds: {','.join(sorted(top_indicators)[:4])}")
# 4. RECORD checkpoint
if (i + 1) % 100 == 0:
loop.record()
# Adaptive exploration: higher rate needed for indicator discovery
if len(loop.sota) > 10:
loop.exploration_rate = max(0.15, EXPLORATION_RATE - len(loop.sota) * 0.005)
loop.exploration_rate = max(0.15, EXPLORATION_RATE - len(loop.sota) * 0.003)
# Final
elapsed = time.time() - t0
print(f"\n{'=' * 60}")
print(f" R&D Loop Complete: {iterations} iterations in {elapsed:.0f}s")
print(f" SOTA Strategies: {len(loop.sota)}")
print(f" R&D Loop V2 Complete: {iterations} iterations in {elapsed:.0f}s")
print(f" SOTA Strategies: {len(loop.sota)} | Best Score: {loop.best_score:.1f}")
print(f"{'=' * 60}")
if loop.sota:
print(f"\n TOP DISCOVERIES:")
print(f"\n TOP DISCOVERIES (by composite score):")
for i, r in enumerate(loop.sota[:15], 1):
hp = r['hypothesis']
print(f" {i:>2d}. {hp['description'][:50]:50s} Sh={r['sharpe']:+.2f} Mon={r['monthly_pct']:+.1f}% "
f"Tr={r['n_trades']} ({hp.get('generation','?')})")
per_inst = r.get('per_instrument', {})
insts = ' '.join([f"{k}:{v['sharpe']:.0f}" for k, v in per_inst.items() if v['sharpe'] != 0])
print(f" {i:>2d}. {hp['description'][:45]:45s} "
f"Sc={r['composite_score']:.1f} Sh={r['sharpe']:+.1f} "
f"Mo={r['monthly_pct']:+.1f}% OOS={r['monthly_oos']:+.1f}% "
f"[{insts}]")
final = RESULTS_DIR / f"rd_loop_final_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
final.write_text(json.dumps(loop.sota, indent=2, default=str))
stripped = [{k: v for k, v in r.items() if k != 'equity_curves'} for r in loop.sota]
final.write_text(json.dumps(stripped, indent=2, default=str))
print(f"\n Saved: {final}")
# Learnings summary
exploit_best = [r for r in loop.sota if r['hypothesis'].get('generation') == 'exploit']
explore_best = [r for r in loop.sota if r['hypothesis'].get('generation') == 'explore']
print(f"\n Exploit wins: {len(exploit_best)} (avg Sh={np.mean([r['sharpe'] for r in exploit_best]):.1f})" if exploit_best else "")
print(f" Explore wins: {len(explore_best)} (avg Sh={np.mean([r['sharpe'] for r in explore_best]):.1f})" if explore_best else "")
optuna_best = [r for r in loop.sota if r['hypothesis'].get('generation') == 'optuna']
ml_best = [r for r in loop.sota if r['hypothesis'].get('generation') == 'ml']
print(f" Exploit: {len(exploit_best)} | Explore: {len(explore_best)} | "
f"Optuna: {len(optuna_best)} | ML: {len(ml_best)}")
if __name__ == "__main__":