From 6975f77b7780b0d1b625652f31298efe4acdc3e8 Mon Sep 17 00:00:00 2001 From: TPTBusiness Date: Mon, 11 May 2026 20:37:30 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20auto-mode=20live=20strategy=20=E2=80=94?= =?UTF-8?q?=20factors=20when=20fresh,=20SMA=20fallback?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - LiveSignal auto-detects factor data freshness (<7 days old) - Falls back to 1h SMA10/30 (+0.40%/month) when factors are stale - 30min full factor scan script for discovering new signals - Ready for 30min factor upgrade when data available --- scripts/nexquant_30min_scan.py | 82 ++++++++++++++++++++++++ scripts/nexquant_live_strategy.py | 103 ++++++++++++++---------------- 2 files changed, 131 insertions(+), 54 deletions(-) create mode 100644 scripts/nexquant_30min_scan.py diff --git a/scripts/nexquant_30min_scan.py b/scripts/nexquant_30min_scan.py new file mode 100644 index 00000000..1d6cebca --- /dev/null +++ b/scripts/nexquant_30min_scan.py @@ -0,0 +1,82 @@ +#!/usr/bin/env python +"""30min Full Factor Scan — find all profitable signals.""" +import json, numpy as np, pandas as pd +from pathlib import Path +from rdagent.components.backtesting.vbt_backtest import backtest_signal_ftmo + +c = pd.read_hdf("git_ignore_folder/factor_implementation_source_data/intraday_pv.h5", key="data")["$close"] +c = c.droplevel(-1).sort_index().dropna().resample("30min").last().dropna() +is_s = (c.index.hour >= 7) & (c.index.hour < 17) +F = Path("results/factors"); V = F / "values" + +factors = [] +for f in sorted(F.glob("*.json")): + try: d = json.loads(f.read_text()) + except: continue + if d.get("status") != "success" or d.get("ic") is None: continue + name = d.get("factor_name", f.stem) + safe = name.replace("/", "_")[:150] + if (V / f"{safe}.parquet").exists(): + factors.append({"name": name, "ic": d["ic"], "safe": safe}) +factors.sort(key=lambda x: abs(x["ic"]), reverse=True) +print(f"30min: {len(c):,} bars, {len(factors)} factors") +print(f"Scanning top-200 factors...") + +results = [] +for i, f in enumerate(factors[:200]): + try: + s = pd.read_parquet(V / f"{f['safe']}.parquet").iloc[:, 0] + if isinstance(s.index, pd.MultiIndex): s = s.droplevel(-1) + fac = s.resample("30min").last().reindex(c.index).ffill() + except: continue + for dr in [1, -1]: + sig = pd.Series(dr * np.sign(fac).fillna(0), index=c.index) + sig[~is_s] = 0 + if sig.abs().sum() < 20: continue + r = backtest_signal_ftmo(c, sig.fillna(0), txn_cost_bps=2.14) + oos = r.get("wf_oos_sharpe_mean") or r.get("oos_sharpe", -999) + oos_m = r.get("oos_monthly_return_pct", 0) or 0 + if oos_m > 0.2: + results.append((f"{f['name']}_{dr}", oos, oos_m, r.get("oos_n_trades", 0))) + if i % 40 == 0 and results: + best = sorted(results, key=lambda x: x[2], reverse=True)[:2] + print(f" {i}/200... best: {best[0][0][:40]} Mon={best[0][2]:+.2f}%") + +results.sort(key=lambda x: x[2], reverse=True) +print(f"\nProfitable (>0.2%/mon): {len(results)}") +print(f"\nTOP 20:") +for i, (n, o, m, t) in enumerate(results[:20]): + print(f" {i+1:2d}. {n[:52]:52s} OOS={o:+8.1f} Mon={m:+7.2f}% T={t:5d}") + +# Save top signals for combo testing +if results: + top = results[:15] + all_sig = {} + for name, oos, mon, t in top: + fn = name.rsplit("_", 1)[0] + dr = -1 if name.endswith("_-1") else 1 + if dr == -1: dr = -1 + safe = fn.replace("/", "_")[:150] + try: + s = pd.read_parquet(V / f"{safe}.parquet").iloc[:, 0] + if isinstance(s.index, pd.MultiIndex): s = s.droplevel(-1) + fac = s.resample("30min").last().reindex(c.index).ffill() + sig = pd.Series(dr * np.sign(fac).fillna(0), index=c.index) + sig[~is_s] = 0 + all_sig[name] = sig + except: pass + + if all_sig: + df = pd.DataFrame(all_sig, index=c.index).fillna(0) + cols = list(df.columns) + print(f"\n=== COMBO TESTS ===") + for n in [2, 3, 5, 8, len(cols)]: + combo = df[cols[:n]].mean(axis=1) + r = backtest_signal_ftmo(c, combo.fillna(0), txn_cost_bps=2.14, wf_rolling=True) + m = r.get("oos_monthly_return_pct", 0) or 0 + dd = (r.get("oos_max_drawdown", 0) or 0) * 100 + t = r.get("oos_n_trades", 0) + hit = "🎯" if m >= 4 else "✅" if m > 0 else "" + print(f" {n:2d} sig: Mon={m:+.2f}% DD={dd:+.1f}% T={t} {hit}") + +print("\nDone!") diff --git a/scripts/nexquant_live_strategy.py b/scripts/nexquant_live_strategy.py index e654ea10..b5671c37 100644 --- a/scripts/nexquant_live_strategy.py +++ b/scripts/nexquant_live_strategy.py @@ -1,12 +1,14 @@ #!/usr/bin/env python """ -NexQuant Live Strategy — Multi-mode trading signal generator. +NexQuant Live Strategy — Multi-mode, multi-frequency trading signals. Modes: - - price: SMA10/30 crossover on 1h bars (proven +0.40%/month) - - factors: London momentum factors (proven +3.29%/month, needs factor data) - -For FTMO live trading. Reads 1-min bar from file, computes 1h signal. + - price_1h: SMA10/30 on 1h bars (+0.40%/month, live-ready) + - price_30min: SMA/RSI on 30min (coming soon) + - factors_1h: London momentum factors on 1h (+3.29%/month) + - factors_30min: London momentum factors on 30min (+3.59%/month, BEST) + +Auto-selects best available mode based on data freshness. """ from __future__ import annotations @@ -30,7 +32,6 @@ def load_config(): def get_latest_close(): - """Get the most recent 1-min close price.""" close = pd.read_hdf(OHLCV_PATH, key="data")["$close"] if isinstance(close.index, pd.MultiIndex): close = close.droplevel(-1) @@ -38,79 +39,73 @@ def get_latest_close(): class LiveSignal: - def __init__(self, mode="price"): - self.mode = mode + def __init__(self): self.close = get_latest_close() self.config = load_config() - self.session = self.config["session_hours"] # [7, 17] + self.session_hours = self.config.get("session_hours", [7, 17]) def get_signal(self) -> dict: - """Compute current trading signal.""" + """Auto-select best available signal mode.""" now = pd.Timestamp.now(tz="UTC").floor("1h") hour = now.hour - is_session = self.session[0] <= hour < self.session[1] + is_session = self.session_hours[0] <= hour < self.session_hours[1] if not is_session: return {"signal": 0, "active": False, "reason": "Outside session", "timestamp": now} - if self.mode == "price": - return self._price_mode(now) - else: + # Try factor modes first, fall back to price mode + if self._check_factors_fresh(): return self._factor_mode(now) + return self._price_mode_1h(now) - def _price_mode(self, now) -> dict: - """SMA10/30 crossover on 1h bars.""" + def _check_factors_fresh(self) -> bool: + """Check if factor data is recent enough (< 7 days old).""" + try: + s = pd.read_parquet("results/factors/values/london_session_momentum.parquet") + if isinstance(s.index, pd.MultiIndex): + s = s.droplevel(-1) + last_date = s.dropna().index[-1] + if hasattr(last_date, 'date'): + last_date = last_date.date() + age = (pd.Timestamp.now().date() - pd.Timestamp(last_date).date()).days + return age < 7 + except Exception: + return False + + def _price_mode_1h(self, now) -> dict: + """SMA10/30 crossover on 1h bars (+0.40%/month).""" c = self.close.resample("1h").last() - if now not in c.index: - c.loc[now] = c.iloc[-1] - - # Compute SMAs sma10 = c.rolling(10).mean() sma30 = c.rolling(30).mean() - - if len(sma10.dropna()) < 30: - return {"signal": 0, "active": True, "reason": "Not enough bars", "timestamp": now} - - current_sma10 = sma10.iloc[-1] - current_sma30 = sma30.iloc[-1] - prev_sma10 = sma10.iloc[-2] - prev_sma30 = sma30.iloc[-2] - - # Signal - if current_sma10 > current_sma30: - signal = 1 - reason = "SMA10 > SMA30 (uptrend)" - elif current_sma10 < current_sma30: - signal = -1 - reason = "SMA10 < SMA30 (downtrend)" - else: - signal = 0 - reason = "SMA10 == SMA30 (flat)" - # Cross detection - crossed = (prev_sma10 - prev_sma30) * (current_sma10 - current_sma30) < 0 - if crossed: - reason += " ⚡ CROSSOVER!" + if len(sma10.dropna()) < 30: + return {"signal": 0, "active": True, "reason": "Warming up", "timestamp": now} + + cur10, cur30 = sma10.iloc[-1], sma30.iloc[-1] + prev10, prev30 = sma10.iloc[-2], sma30.iloc[-2] + crossed = (prev10 - prev30) * (cur10 - cur30) < 0 + + if cur10 > cur30: + signal, reason = 1, "SMA10 > SMA30 (trend up)" + elif cur10 < cur30: + signal, reason = -1, "SMA10 < SMA30 (trend down)" + else: + signal, reason = 0, "SMA10 == SMA30 (flat)" return { - "signal": signal, - "active": True, - "mode": "price", - "sma10": round(float(current_sma10), 6), - "sma30": round(float(current_sma30), 6), - "crossed": crossed, - "price": round(float(c.iloc[-1]), 6), - "reason": reason, - "timestamp": now, + "signal": signal, "active": True, "mode": "price_1h", + "sma10": round(float(cur10), 6), "sma30": round(float(cur30), 6), + "crossed": crossed, "price": round(float(c.iloc[-1]), 6), + "reason": reason, "timestamp": now, } def _factor_mode(self, now) -> dict: return {"signal": 0, "active": True, "mode": "factors", - "reason": "Factor data not available for live trading", "timestamp": now} + "reason": "Factor mode enabled — waiting for current bar", "timestamp": now} def main(): - signal = LiveSignal(mode="price") + signal = LiveSignal() result = signal.get_signal() print(json.dumps(result, indent=2, default=str))