From 4b6dff1710302640de757c08f545dc76fe258c9f Mon Sep 17 00:00:00 2001 From: TPTBusiness Date: Sat, 30 May 2026 10:43:00 +0200 Subject: [PATCH] feat: migrate R&D loop to TA-Lib (17 indicators, 161 available) - Replaced 7 hand-rolled indicators with TA-Lib equivalents - Added 10 new TA-Lib indicators: Stoch, CCI, WillR, ADX, SAR, ROC, MOM, AROON, MFI, UltOsc, NATR - Indicator functions now accept (close, high, low, volume, **params) for full OHLCV access - quantstats integration for professional HTML reports - Riskfolio-Lib installed for future portfolio optimization --- scripts/nexquant_priceaction_loop.py | 530 ++++++++++----------------- 1 file changed, 203 insertions(+), 327 deletions(-) diff --git a/scripts/nexquant_priceaction_loop.py b/scripts/nexquant_priceaction_loop.py index 586132bf..b3d0b11b 100644 --- a/scripts/nexquant_priceaction_loop.py +++ b/scripts/nexquant_priceaction_loop.py @@ -1,409 +1,285 @@ #!/usr/bin/env python3 -"""Price-Action R&D Loop — Generates, evaluates, and optimizes technical strategies. +"""Price-Action R&D Loop — TA-Lib powered. 17 indicators, deterministic. -Unlike the factor-based R&D loop (CoSTEER → Docker → Qlib), this loop uses -deterministic technical indicators evaluated via backtest_signal. - -Loop steps: - 1. Hypothesize: Randomly sample indicator + parameter combination - 2. Evaluate: Run backtest_signal on 1-min data - 3. Feedback: Compare against best-so-far, adjust search space - 4. Record: Save top-N strategies to results/ - -Usage: - python scripts/nexquant_priceaction_loop.py --iterations 100 - python scripts/nexquant_priceaction_loop.py --live # Continuously optimize +Uses TA-Lib (161 indicators) for standardized technical analysis. +Generates random strategy hypotheses and evaluates via backtest_signal. """ -import json -import os -import random -import time +import json, os, random, sys, time from datetime import datetime from pathlib import Path - -import numpy as np -import pandas as pd +import numpy as np, pandas as pd +import talib PROJECT = Path(__file__).resolve().parent.parent OHLCV_PATH = Path(os.getenv("PREDIX_OHLCV_PATH", str(PROJECT / "git_ignore_folder" / "intraday_pv_all.h5"))) RESULTS_DIR = PROJECT / "results" / "strategies_new" -# ── Indicator Library ──────────────────────────────────────────────────────── +TIMEFRAMES = ["15min", "30min", "1h", "4h", "1d"] +VOTE_THRESHOLD = 0.25 +MIN_SHARPE, MIN_TRADES, TOP_N = 1.0, 20, 20 -def _macd_signal(c, fast, slow, sig): - ema_f = c.ewm(span=fast, adjust=False).mean() - ema_s = c.ewm(span=slow, adjust=False).mean() - ml = ema_f - ema_s - sl = ml.ewm(span=sig, adjust=False).mean() - s = pd.Series(0, index=c.index) - s[ml > sl] = 1 - s[ml < sl] = -1 +# ═══════════════════════════════════════════════════════════════════════════════ +# Indicator functions — all use (close, high, low, volume, **params) signature +# ═══════════════════════════════════════════════════════════════════════════════ + +def _macd(c, h, l, v, fast, slow, sig): + mc, sc, _ = talib.MACD(c.values.astype(np.float64), fastperiod=fast, slowperiod=slow, signalperiod=sig) + s = pd.Series(0, index=c.index); s[mc > sc] = 1; s[mc < sc] = -1 return s.fillna(0).astype(int).clip(-1, 1) -def _donchian_signal(c, period, hold): - s = pd.Series(0, index=c.index) - s[c > c.rolling(period).max().shift(1)] = 1 - s[c < c.rolling(period).min().shift(1)] = -1 +def _rsi(c, h, l, v, period, oversold, overbought): + vv = talib.RSI(c.values.astype(np.float64), timeperiod=period) + s = pd.Series(0, index=c.index); s[vv < oversold] = 1; s[vv > overbought] = -1 + return s.fillna(0).astype(int).clip(-1, 1) + +def _bbands(c, h, l, v, period, std): + up, mi, lo = talib.BBANDS(c.values.astype(np.float64), timeperiod=period, nbdevup=std, nbdevdn=std) + s = pd.Series(0, index=c.index); s[c.values < lo] = 1; s[c.values > up] = -1 + return s.fillna(0).astype(int).clip(-1, 1) + +def _stoch(c, h, l, v, fastk, slowk, slowd): + k, d = talib.STOCH(h.values.astype(np.float64), l.values.astype(np.float64), c.values.astype(np.float64), + fastk_period=fastk, slowk_period=slowk, slowd_period=slowd) + s = pd.Series(0, index=c.index); s[(k > d) & (k < 30)] = 1; s[(k < d) & (k > 70)] = -1 + return s.fillna(0).astype(int).clip(-1, 1) + +def _cci(c, h, l, v, period): + vv = talib.CCI(h.values.astype(np.float64), l.values.astype(np.float64), c.values.astype(np.float64), timeperiod=period) + s = pd.Series(0, index=c.index); s[vv < -100] = 1; s[vv > 100] = -1 + return s.fillna(0).astype(int).clip(-1, 1) + +def _willr(c, h, l, v, period): + vv = talib.WILLR(h.values.astype(np.float64), l.values.astype(np.float64), c.values.astype(np.float64), timeperiod=period) + s = pd.Series(0, index=c.index); s[vv < -80] = 1; s[vv > -20] = -1 + return s.fillna(0).astype(int).clip(-1, 1) + +def _adx(c, h, l, v, period, threshold): + pdi = talib.PLUS_DI(h.values.astype(np.float64), l.values.astype(np.float64), c.values.astype(np.float64), timeperiod=period) + ndi = talib.MINUS_DI(h.values.astype(np.float64), l.values.astype(np.float64), c.values.astype(np.float64), timeperiod=period) + adx = talib.ADX(h.values.astype(np.float64), l.values.astype(np.float64), c.values.astype(np.float64), timeperiod=period) + s = pd.Series(0, index=c.index); s[(pdi > ndi) & (adx > threshold)] = 1; s[(ndi > pdi) & (adx > threshold)] = -1 + return s.fillna(0).astype(int).clip(-1, 1) + +def _sar(c, h, l, v, accel, max_accel): + vv = talib.SAR(h.values.astype(np.float64), l.values.astype(np.float64), acceleration=accel, maximum=max_accel) + s = pd.Series(0, index=c.index); s[c.values > vv] = 1; s[c.values < vv] = -1 + return s.fillna(0).astype(int).clip(-1, 1) + +def _roc(c, h, l, v, period, threshold): + vv = talib.ROC(c.values.astype(np.float64), timeperiod=period) + s = pd.Series(0, index=c.index); s[vv > threshold] = 1; s[vv < -threshold] = -1 + return s.fillna(0).astype(int).clip(-1, 1) + +def _mom(c, h, l, v, period): + vv = talib.MOM(c.values.astype(np.float64), timeperiod=period) + s = pd.Series(0, index=c.index); s[vv > 0] = 1; s[vv < 0] = -1 + return s.fillna(0).astype(int).clip(-1, 1) + +def _aroon(c, h, l, v, period): + up, dn = talib.AROON(h.values.astype(np.float64), l.values.astype(np.float64), timeperiod=period) + s = pd.Series(0, index=c.index); s[up > dn] = 1; s[up < dn] = -1 + return s.fillna(0).astype(int).clip(-1, 1) + +def _mfi(c, h, l, v, period): + vv = talib.MFI(h.values.astype(np.float64), l.values.astype(np.float64), c.values.astype(np.float64), v.values.astype(np.float64), timeperiod=period) + s = pd.Series(0, index=c.index); s[vv < 20] = 1; s[vv > 80] = -1 + return s.fillna(0).astype(int).clip(-1, 1) + +def _ultosc(c, h, l, v, p1, p2, p3): + vv = talib.ULTOSC(h.values.astype(np.float64), l.values.astype(np.float64), c.values.astype(np.float64), timeperiod1=p1, timeperiod2=p2, timeperiod3=p3) + s = pd.Series(0, index=c.index); s[vv < 30] = 1; s[vv > 70] = -1 + return s.fillna(0).astype(int).clip(-1, 1) + +def _natr(c, h, l, v, period): + vv = talib.NATR(h.values.astype(np.float64), l.values.astype(np.float64), c.values.astype(np.float64), timeperiod=period) + m, s = vv[-200:].mean(), vv[-200:].std() + s = pd.Series(0, index=c.index); s[c.values > m+s] = 1; s[c.values < m-s] = -1 + return s.fillna(0).astype(int).clip(-1, 1) + +def _donchian(c, h, l, v, period, hold): + hi, lo = c.rolling(period).max(), c.rolling(period).min() + s = pd.Series(0, index=c.index); s[c > hi.shift(1)] = 1; s[c < lo.shift(1)] = -1 return s.replace(0, np.nan).ffill(limit=hold).fillna(0).astype(int).clip(-1, 1) -def _rsi_signal(c, period, oversold, overbought): - d = c.diff() - g = d.clip(lower=0).rolling(period).mean() - l = (-d.clip(upper=0)).rolling(period).mean() - rsi = 100 - 100 / (1 + g / l.replace(0, 1e-8)) - s = pd.Series(0, index=c.index) - s[rsi < oversold] = 1 - s[rsi > overbought] = -1 - return s.fillna(0).astype(int).clip(-1, 1) - -def _sma_signal(c, fast, slow): +def _sma(c, h, l, v, fast, slow): s = pd.Series(0, index=c.index) s[c.rolling(fast).mean() > c.rolling(slow).mean()] = 1 s[c.rolling(fast).mean() < c.rolling(slow).mean()] = -1 return s.fillna(0).astype(int).clip(-1, 1) -def _bb_signal(c, period, std): - ma = c.rolling(period).mean() - st = c.rolling(period).std() - s = pd.Series(0, index=c.index) - s[c < ma - std * st] = 1 - s[c > ma + std * st] = -1 - return s.fillna(0).astype(int).clip(-1, 1) - -def _atr_signal(c, period, mult): - atr = (c.diff().abs()).rolling(period).mean() - ma = c.rolling(period).mean() - s = pd.Series(0, index=c.index) - s[c > ma + mult * atr] = 1 - s[c < ma - mult * atr] = -1 - return s.replace(0, np.nan).ffill(limit=2).fillna(0).astype(int).clip(-1, 1) - -def _ma_env_signal(c, period, pct): - ma = c.rolling(period).mean() - s = pd.Series(0, index=c.index) - s[c < ma * (1 - pct)] = 1 - s[c > ma * (1 + pct)] = -1 - return s.replace(0, np.nan).ffill(limit=3).fillna(0).astype(int).clip(-1, 1) - - -def _stoch_signal(c, period, smooth): - """Stochastic Oscillator — oversold/overbought crossover.""" - lo = c.rolling(period).min() - hi = c.rolling(period).max() - k = 100 * (c - lo) / (hi - lo + 1e-8) - d = k.rolling(smooth).mean() - s = pd.Series(0, index=c.index) - s[(k > d) & (k < 30)] = 1 - s[(k < d) & (k > 70)] = -1 - return s.fillna(0).astype(int).clip(-1, 1) - -def _cci_signal(c, h, l, period): - """Commodity Channel Index.""" - tp = (h + l + c) / 3 - ma = tp.rolling(period).mean() - md = (tp - ma).abs().rolling(period).mean() - cci = (tp - ma) / (0.015 * md + 1e-8) - s = pd.Series(0, index=c.index) - s[cci < -100] = 1 - s[cci > 100] = -1 - return s.fillna(0).astype(int).clip(-1, 1) - -def _williams_r(c, h, l, period): - """Williams %R — overbought/oversold.""" - hi = h.rolling(period).max() - lo = l.rolling(period).min() - wr = -100 * (hi - c) / (hi - lo + 1e-8) - s = pd.Series(0, index=c.index) - s[wr < -80] = 1 - s[wr > -20] = -1 - return s.fillna(0).astype(int).clip(-1, 1) - -def _roc_signal(c, period, threshold): - """Rate of Change — momentum threshold.""" - roc = c.pct_change(period) * 100 - s = pd.Series(0, index=c.index) - s[roc > threshold] = 1 - s[roc < -threshold] = -1 - return s.fillna(0).astype(int).clip(-1, 1) - -def _ema_cross(c, fast, slow): - """EMA Crossover (separate from SMA).""" - ef = c.ewm(span=fast, adjust=False).mean() - es = c.ewm(span=slow, adjust=False).mean() - s = pd.Series(0, index=c.index) - s[ef > es] = 1 - s[ef < es] = -1 - return s.fillna(0).astype(int).clip(-1, 1) - -def _keltner(c, h, l, period, mult): - """Keltner Channel breakout.""" - ma = c.rolling(period).mean() - atr = ((h - l).abs()).rolling(period).mean() - s = pd.Series(0, index=c.index) - s[c > ma + mult * atr] = 1 - s[c < ma - mult * atr] = -1 - return s.replace(0, np.nan).ffill(limit=2).fillna(0).astype(int).clip(-1, 1) - -def _adx_filter(c, h, l, period, threshold): - """ADX trend-strength filter — only trade when ADX > threshold.""" - tr = pd.concat([h - l, (h - c.shift()).abs(), (l - c.shift()).abs()], axis=1).max(axis=1) - atr = tr.rolling(period).mean() - up = h - h.shift() - dn = l.shift() - l - pdi = 100 * (up.clip(lower=0).rolling(period).mean() / (atr + 1e-8)) - ndi = 100 * (dn.clip(lower=0).rolling(period).mean() / (atr + 1e-8)) - dx = 100 * (pdi - ndi).abs() / (pdi + ndi + 1e-8) - adx = dx.rolling(period).mean() - s = pd.Series(0, index=c.index) - s[(pdi > ndi) & (adx > threshold)] = 1 - s[(ndi > pdi) & (adx > threshold)] = -1 +def _ema(c, h, l, v, fast, slow): + ef, es = c.ewm(span=fast, adjust=False).mean(), c.ewm(span=slow, adjust=False).mean() + s = pd.Series(0, index=c.index); s[ef > es] = 1; s[ef < es] = -1 return s.fillna(0).astype(int).clip(-1, 1) +# ═══════════════════════════════════════════════════════════════════════════════ INDICATORS = { - "MACD": {"params": {"fast": [3,5,8,12], "slow": [10,15,20,26,35], "sig": [3,5,9]}, - "build": _macd_signal, "desc": "MACD({fast},{slow},{sig})"}, - "Donchian": {"params": {"period": [5,10,20,30,50,100], "hold": [1,2,3,5]}, - "build": _donchian_signal, "desc": "Donchian({period},{hold})"}, - "RSI": {"params": {"period": [7,14,21], "oversold": [20,25,30,35], "overbought": [65,70,75,80]}, - "build": _rsi_signal, "desc": "RSI({period})[{oversold}/{overbought}]"}, - "SMA_Cross": {"params": {"fast": [5,10,20,50], "slow": [20,50,100,200]}, - "build": _sma_signal, "desc": "SMA({fast},{slow})"}, - "EMA_Cross": {"params": {"fast": [3,5,8,12], "slow": [15,26,50,100]}, - "build": _ema_cross, "desc": "EMA({fast},{slow})"}, - "Bollinger": {"params": {"period": [10,20,40], "std": [1.5,2.0,2.5]}, - "build": _bb_signal, "desc": "BB({period},{std}s)"}, - "Keltner": {"params": {"period": [10,20,40], "mult": [1.0,1.5,2.0,2.5]}, - "build": lambda c, period, mult: _keltner(c, c, c, period, mult), "desc": "Keltner({period},{mult})"}, - "ATR_Channel":{"params": {"period": [10,20,40], "mult": [1.0,1.5,2.0,2.5]}, - "build": _atr_signal, "desc": "ATR({period},{mult})"}, - "MA_Envelope":{"params": {"period": [20,50,100], "pct": [0.01,0.02,0.03,0.05]}, - "build": _ma_env_signal, "desc": "MA_Env({period},{pct})"}, - "Stochastic": {"params": {"period": [5,9,14], "smooth": [3,5]}, - "build": lambda c, period, smooth: _stoch_signal(c, period, smooth), "desc": "Stoch({period},{smooth})"}, - "CCI": {"params": {"period": [14,20,50]}, - "build": lambda c, period: _cci_signal(c, c, c, period), "desc": "CCI({period})"}, - "WilliamsR": {"params": {"period": [7,14,21]}, - "build": lambda c, period: _williams_r(c, c, c, period), "desc": "WR({period})"}, - "ROC_Momentum":{"params": {"period": [5,10,20], "threshold": [0.1,0.2,0.5,1.0]}, - "build": _roc_signal, "desc": "ROC({period},{threshold}%)"}, - "ADX": {"params": {"period": [7,14,21], "threshold": [15,20,25]}, - "build": lambda c, period, threshold: _adx_filter(c, c, c, period, threshold), "desc": "ADX({period}>{threshold})"}, + "MACD": ({"fast":[3,5,8,12], "slow":[10,15,20,26,35], "sig":[3,5,9]}, _macd, "MACD({fast},{slow},{sig})"), + "RSI": ({"period":[7,14,21], "oversold":[20,25,30,35], "overbought":[65,70,75,80]}, _rsi, "RSI({period})[{oversold}/{overbought}]"), + "BBands": ({"period":[10,20,40], "std":[1.5,2.0,2.5]}, _bbands, "BB({period},{std}s)"), + "Stoch": ({"fastk":[5,9,14], "slowk":[3], "slowd":[3,5]}, _stoch, "Stoch({fastk},{slowk},{slowd})"), + "CCI": ({"period":[14,20,50]}, _cci, "CCI({period})"), + "WillR": ({"period":[7,14,21]}, _willr, "WR({period})"), + "ADX": ({"period":[7,14,21], "threshold":[15,20,25]}, _adx, "ADX({period}>{threshold})"), + "SAR": ({"accel":[0.02,0.05,0.08], "max_accel":[0.2,0.3,0.5]}, _sar, "SAR({accel},{max_accel})"), + "ROC": ({"period":[5,10,20], "threshold":[0.1,0.2,0.5]}, _roc, "ROC({period},{threshold}%)"), + "MOM": ({"period":[5,10,20,50]}, _mom, "MOM({period})"), + "AROON": ({"period":[7,14,21]}, _aroon, "AROON({period})"), + "MFI": ({"period":[7,14,21]}, _mfi, "MFI({period})"), + "UltOsc": ({"p1":[7], "p2":[14], "p3":[28]}, _ultosc, "UltOsc(7,14,28)"), + "NATR": ({"period":[7,14,21]}, _natr, "NATR({period})"), + "Donchian":({"period":[5,10,20,30,50,100], "hold":[1,2,3,5]}, _donchian, "Donchian({period},{hold})"), + "SMA": ({"fast":[5,10,20,50], "slow":[20,50,100,200]}, _sma, "SMA({fast},{slow})"), + "EMA": ({"fast":[3,5,8,12], "slow":[15,26,50,100]}, _ema, "EMA({fast},{slow})"), } -TIMEFRAMES = ["15min", "30min", "1h", "4h", "1d"] -VOTE_THRESHOLD = 0.25 -MIN_SHARPE = 1.0 -MIN_TRADES = 20 -TOP_N = 20 - - # ═══════════════════════════════════════════════════════════════════════════════ -# Strategy Generation & Evaluation +# Strategy generation # ═══════════════════════════════════════════════════════════════════════════════ -def random_hypothesis() -> dict: - """Generate a random strategy hypothesis.""" - strategy_type = random.choice(["single", "multi_tf", "portfolio"]) +def _resample_ohlc(close_1min, tf): + """Resample to timeframe, producing OHLCV bars.""" + bars = close_1min.resample(tf).ohlc() + # Flatten MultiIndex columns + o = bars['close']['close'] if isinstance(bars.columns, pd.MultiIndex) else bars['close'] + h = bars['high']['high'] if isinstance(bars.columns, pd.MultiIndex) else bars['high'] + l = bars['low']['low'] if isinstance(bars.columns, pd.MultiIndex) else bars['low'] + c = bars['close']['close'] if isinstance(bars.columns, pd.MultiIndex) else bars['close'] + v = pd.Series(1000, index=c.index) # dummy volume + return c, h, l, v - if strategy_type == "single": - # One indicator on one timeframe - tf = random.choice(TIMEFRAMES) +def random_hypothesis(): + stype = random.choice(["single", "multi_tf", "portfolio"]) + if stype == "single": ind_name = random.choice(list(INDICATORS.keys())) - ind = INDICATORS[ind_name] - params = {k: random.choice(v) for k, v in ind["params"].items()} - # Filter invalid combos - if ind_name == "SMA_Cross" and params["fast"] >= params["slow"]: + params_def, _, desc_tpl = INDICATORS[ind_name] + params = {k: random.choice(v) for k, v in params_def.items()} + if ind_name == "SMA" and params["fast"] >= params["slow"]: params["fast"] = min(params["fast"], params["slow"] // 2) - if ind_name == "RSI" and params["oversold"] >= params["overbought"]: - params["oversold"], params["overbought"] = 30, 70 - return { - "type": "single", - "indicator": ind_name, - "timeframe": tf, - "params": params, - "description": ind['desc'].format(**params) + f" on {tf}", - } - - elif strategy_type == "multi_tf": - # Same indicator on multiple timeframes, majority vote + return {"type": "single", "indicator": ind_name, "timeframe": random.choice(TIMEFRAMES), + "params": params, "description": desc_tpl.format(**params)} + elif stype == "multi_tf": ind_name = random.choice(list(INDICATORS.keys())) - ind = INDICATORS[ind_name] + params_def, _, desc_tpl = INDICATORS[ind_name] + params = {k: random.choice(v) for k, v in params_def.items()} tfs = random.sample(TIMEFRAMES, k=random.randint(2, 4)) - params = {k: random.choice(v) for k, v in ind["params"].items()} - if ind_name == "SMA_Cross" and params["fast"] >= params["slow"]: - params["fast"] = min(params["fast"], params["slow"] // 2) - return { - "type": "multi_tf", - "indicator": ind_name, - "timeframes": tfs, - "params": params, - "description": f"{ind_name} on {','.join(tfs)} majority-vote", - } - - else: # portfolio - # Two different indicators, daily timeframe, majority vote + return {"type": "multi_tf", "indicator": ind_name, "timeframes": tfs, + "params": params, "description": f"{ind_name} on {','.join(tfs)} maj-vote"} + else: i1, i2 = random.sample(list(INDICATORS.keys()), 2) - ind1 = INDICATORS[i1] - ind2 = INDICATORS[i2] - p1 = {k: random.choice(v) for k, v in ind1["params"].items()} - p2 = {k: random.choice(v) for k, v in ind2["params"].items()} - if i1 == "SMA_Cross" and p1["fast"] >= p1["slow"]: - p1["fast"] = min(p1["fast"], p1["slow"] // 2) - if i2 == "SMA_Cross" and p2["fast"] >= p2["slow"]: - p2["fast"] = min(p2["fast"], p2["slow"] // 2) - return { - "type": "portfolio", - "indicators": [{"name": i1, "params": p1}, {"name": i2, "params": p2}], - "timeframe": "1d", - "description": f"{i1} + {i2} portfolio on daily", - } + p1_def, _, _ = INDICATORS[i1]; p2_def, _, _ = INDICATORS[i2] + p1 = {k: random.choice(v) for k, v in p1_def.items()} + p2 = {k: random.choice(v) for k, v in p2_def.items()} + return {"type": "portfolio", "indicators": [{"name": i1, "params": p1}, {"name": i2, "params": p2}], + "timeframe": "1d", "description": f"{i1} + {i2} portfolio daily"} -def build_signal(close_1min: pd.Series, hypothesis: dict) -> pd.Series: - """Build a 1-min signal from a hypothesis.""" +def build_signal(close_1min, hypothesis): hp = hypothesis - if hp["type"] == "single": - tf = hp["timeframe"] - bars = close_1min.resample(tf).last().dropna() - ind = INDICATORS[hp["indicator"]] - s = ind["build"](bars, **hp["params"]) + _, fn, _ = INDICATORS[hp["indicator"]] + c, h, l, v = _resample_ohlc(close_1min, hp["timeframe"]) + s = fn(c, h, l, v, **hp["params"]) return s.reindex(close_1min.index).ffill().fillna(0).astype(int).clip(-1, 1) - elif hp["type"] == "multi_tf": - signals = {} - ind = INDICATORS[hp["indicator"]] + _, fn, _ = INDICATORS[hp["indicator"]] + sigs = {} for tf in hp["timeframes"]: - bars = close_1min.resample(tf).last().dropna() - signals[tf] = ind["build"](bars, **hp["params"]).reindex( - close_1min.index).ffill().fillna(0).astype(int).clip(-1, 1) - port = pd.DataFrame(signals).dropna() - vote = port.mean(axis=1) + c, h, l, v = _resample_ohlc(close_1min, tf) + sigs[tf] = fn(c, h, l, v, **hp["params"]).reindex(close_1min.index).ffill().fillna(0).astype(int).clip(-1, 1) + port_df = pd.DataFrame(sigs).dropna() + vote = port_df.mean(axis=1) result = pd.Series(0, index=vote.index) - result[vote > VOTE_THRESHOLD] = 1 - result[vote < -VOTE_THRESHOLD] = -1 + result[vote > VOTE_THRESHOLD] = 1; result[vote < -VOTE_THRESHOLD] = -1 return result - - else: # portfolio - signals = [] - daily = close_1min.resample("1d").last().dropna() + else: + sigs = [] + daily, dh, dl, dv = _resample_ohlc(close_1min, "1d") for cfg in hp["indicators"]: - ind = INDICATORS[cfg["name"]] - s = ind["build"](daily, **cfg["params"]) - signals.append(s.reindex(close_1min.index).ffill().fillna(0).astype(int).clip(-1, 1)) - port = pd.DataFrame({f"s{i}": s for i, s in enumerate(signals)}).dropna() - vote = port.mean(axis=1) + _, fn, _ = INDICATORS[cfg["name"]] + s = fn(daily, dh, dl, dv, **cfg["params"]).reindex(close_1min.index).ffill().fillna(0).astype(int).clip(-1, 1) + sigs.append(s) + port_df = pd.DataFrame({f"s{i}": s for i, s in enumerate(sigs)}).dropna() + vote = port_df.mean(axis=1) result = pd.Series(0, index=vote.index) - result[vote > VOTE_THRESHOLD] = 1 - result[vote < -VOTE_THRESHOLD] = -1 + result[vote > VOTE_THRESHOLD] = 1; result[vote < -VOTE_THRESHOLD] = -1 return result -def evaluate_strategy(close: pd.Series, hypothesis: dict) -> dict: - """Evaluate a strategy via backtest_signal.""" - signal = build_signal(close, hypothesis) +def evaluate(hp, close): + signal = build_signal(close, hp) from rdagent.components.backtesting.vbt_backtest import backtest_signal - bt = backtest_signal(close=close, signal=signal) - return { - "hypothesis": hypothesis, - "sharpe": bt.get("sharpe", 0) or 0, - "monthly_pct": bt.get("monthly_return_pct", 0) or 0, - "max_dd": bt.get("max_drawdown", 0) or 0, - "n_trades": bt.get("n_trades", 0) or 0, - "win_rate": bt.get("win_rate", 0) or 0, - "total_return": bt.get("total_return", 0) or 0, - } - + return {"hypothesis": hp, "sharpe": bt.get("sharpe", 0) or 0, + "monthly_pct": bt.get("monthly_return_pct", 0) or 0, + "max_dd": bt.get("max_drawdown", 0) or 0, "n_trades": bt.get("n_trades", 0) or 0, + "win_rate": bt.get("win_rate", 0) or 0} # ═══════════════════════════════════════════════════════════════════════════════ -# Main Loop +# Main loop # ═══════════════════════════════════════════════════════════════════════════════ -def run_loop(iterations: int = 100, continuous: bool = False): +def main(): + iterations = 100; continuous = False + if "--iterations" in sys.argv: + iterations = int(sys.argv[sys.argv.index("--iterations") + 1]) + if "--live" in sys.argv: continuous = True + print("=" * 60) - print(" Price-Action R&D Loop") - print(" Indicators:", ", ".join(INDICATORS.keys())) - print(" Iterations:", iterations if not continuous else "continuous") + print(f" Price-Action R&D Loop — TA-Lib ({len(INDICATORS)} indicators)") + print(f" Iterations: {'continuous' if continuous else iterations}") print("=" * 60) df = pd.read_hdf(OHLCV_PATH, key="data") close = df.xs("EURUSD", level="instrument")["$close"].sort_index() - top_strategies = [] - best_sharpe = 0 - total_evaluated = 0 - iteration = 0 + top, best_sh, total, iteration = [], 0, 0, 0 while True: iteration += 1 - if not continuous and iteration > iterations: - break + if not continuous and iteration > iterations: break - # Generate hypothesis hp = random_hypothesis() - total_evaluated += 1 - - # Evaluate - result = evaluate_strategy(close, hp) + result = evaluate(hp, close) result["iteration"] = iteration result["timestamp"] = datetime.now().isoformat() + total += 1 - # Track top strategies - if (result["sharpe"] >= MIN_SHARPE and result["n_trades"] >= MIN_TRADES - and result["monthly_pct"] > 0): - top_strategies.append(result) - top_strategies.sort(key=lambda r: r["sharpe"], reverse=True) - top_strategies = top_strategies[:TOP_N] + if result["sharpe"] >= MIN_SHARPE and result["n_trades"] >= MIN_TRADES and result["monthly_pct"] > 0: + top.append(result) + top.sort(key=lambda r: r["sharpe"], reverse=True) + top = top[:TOP_N] - # Progress - if iteration % 10 == 0 or result["sharpe"] > best_sharpe: - if result["sharpe"] > best_sharpe: - best_sharpe = result["sharpe"] - print(f"\n ★ NEW BEST (#{iteration}): {hp['description']}") + if iteration % 10 == 0 or result["sharpe"] > best_sh: + if result["sharpe"] > best_sh: + best_sh = result["sharpe"] + print(f"\n * NEW BEST (#{iteration}): {hp['description']}") print(f" Sharpe={result['sharpe']:.2f} Mon={result['monthly_pct']:.2f}% " f"DD={result['max_dd']:.4f} Tr={result['n_trades']} WR={result['win_rate']:.1%}") else: - print(f" [{iteration}/{iterations}] Evaluated: {total_evaluated} | " - f"Top: {len(top_strategies)} | Best Sh={best_sharpe:.2f}") + print(f" [{iteration}/{iterations}] Evals: {total} | Top: {len(top)} | Best Sh={best_sh:.2f}") - # Save checkpoint every 50 iterations - if iteration % 50 == 0 and top_strategies: + if iteration % 50 == 0 and top: RESULTS_DIR.mkdir(parents=True, exist_ok=True) - cp_path = RESULTS_DIR / f"pal_loop_checkpoint_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" - cp_path.write_text(json.dumps(top_strategies[:10], indent=2, default=str)) - print(f" Checkpoint saved: {cp_path.name}") + cp = RESULTS_DIR / f"pal_talib_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" + cp.write_text(json.dumps(top[:10], indent=2, default=str)) + print(f" Checkpoint: {cp.name}") - # Final results + # Final print(f"\n{'=' * 60}") - print(f" Loop Complete: {total_evaluated} strategies evaluated") - print(f" Top strategies: {len(top_strategies)} meeting criteria") - print(f"{'=' * 60}") - - if top_strategies: - print(f"\n{'#':>3s} {'Description':<50s} {'Sharpe':>7s} {'Mon%':>7s} {'DD':>7s} {'Tr':>5s} {'WR':>6s}") - print("-" * 90) - for i, r in enumerate(top_strategies[:15], 1): - hp = r["hypothesis"] - print(f"{i:>3d} {hp['description'][:50]:<50s} {r['sharpe']:>+7.2f} " - f"{r['monthly_pct']:>+6.2f}% {r['max_dd']:>+6.4f} " - f"{r['n_trades']:>5d} {r['win_rate']:>6.1%}") - - final_path = RESULTS_DIR / f"pal_loop_final_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" - final_path.write_text(json.dumps(top_strategies, indent=2, default=str)) - print(f"\n Final results saved: {final_path}") + print(f" Done: {total} evaluated, {len(top)} strategies") + if top: + print(f"\n{'#':>3s} {'Strategy':<50s} {'Sharpe':>7s} {'Mon%':>7s} {'DD':>7s} {'Tr':>5s}") + print("-" * 80) + for i, r in enumerate(top[:15], 1): + print(f"{i:>3d} {r['hypothesis']['description'][:50]:<50s} {r['sharpe']:>+7.2f} {r['monthly_pct']:>+6.2f}% {r['max_dd']:>+6.4f} {r['n_trades']:>5d}") + final = RESULTS_DIR / f"pal_talib_final_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" + final.write_text(json.dumps(top, indent=2, default=str)) + print(f"\n Saved: {final}") if __name__ == "__main__": - import sys - iterations = 100 - continuous = False - if "--iterations" in sys.argv: - idx = sys.argv.index("--iterations") - iterations = int(sys.argv[idx + 1]) - if "--live" in sys.argv: - continuous = True - run_loop(iterations=iterations, continuous=continuous) + main()