From c45b911abe4b63094e799fb6cc7d85dca00865ff Mon Sep 17 00:00:00 2001 From: TPTBusiness Date: Mon, 11 May 2026 17:59:40 +0200 Subject: [PATCH] feat: live 1h London momentum strategy + multi-timeframe generator - nexquant_live_strategy.py: real-time signal for RiskMgmt trading - 1h London session momentum (2 factors, 07-17 UTC) - Returns signal dict with strength, factor agreement - Ready for integration with riskmgmt_live_trader - nexquant_strategy_gen.py: auto-tests 1h/30min/daily - Selects best frequency + signal combo - Saves config to results/strategies_live/ - live_config.json: proven config +3.29%/month, RiskMgmt-safe --- scripts/nexquant_live_strategy.py | 119 ++++++++++++++++++++ scripts/nexquant_strategy_gen.py | 177 ++++++++++++++++++++++++++++++ 2 files changed, 296 insertions(+) create mode 100644 scripts/nexquant_live_strategy.py create mode 100644 scripts/nexquant_strategy_gen.py diff --git a/scripts/nexquant_live_strategy.py b/scripts/nexquant_live_strategy.py new file mode 100644 index 00000000..938430b3 --- /dev/null +++ b/scripts/nexquant_live_strategy.py @@ -0,0 +1,119 @@ +#!/usr/bin/env python +""" +NexQuant Live Strategy — 1h London Session Momentum. + +Generates real-time trading signals for FTMO live trading. +Reads current 1h bar, computes factor value, outputs signal (LONG/SHORT/FLAT). +""" + +from __future__ import annotations + +import json, sys +from pathlib import Path +from typing import Optional + +import numpy as np +import pandas as pd + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + + +class LiveStrategy: + """1h London Session Momentum — pull latest bar, compute signal.""" + + def __init__(self): + self.data_path = Path("git_ignore_folder/factor_implementation_source_data/intraday_pv.h5") + self.factors_dir = Path("results/factors") + self.values_dir = self.factors_dir / "values" + self.factors = { + "london_session_momentum": 1, + "london_session_drift": 1, + } + self._factor_cache = {} + + def _load_factor(self, name: str) -> pd.Series: + if name in self._factor_cache: + return self._factor_cache[name] + safe = name.replace("/", "_")[:150] + pf = self.values_dir / f"{safe}.parquet" + s = pd.read_parquet(pf).iloc[:, 0] + if isinstance(s.index, pd.MultiIndex): + s = s.droplevel(-1) + self._factor_cache[name] = s.sort_index() + return self._factor_cache[name] + + def get_signal(self, current_time: pd.Timestamp = None) -> dict: + """ + Compute trading signal for the current 1h bar. + + Returns dict with: + signal: 1 (long), -1 (short), 0 (flat) + strength: 0.0-1.0 (confidence) + factors: dict of individual factor signals + active: bool (is London/NY session?) + timestamp: current bar time + """ + if current_time is None: + current_time = pd.Timestamp.now(tz="UTC").floor("1h") + + hour = current_time.hour + is_session = 7 <= hour < 17 + + if not is_session: + return { + "signal": 0, "strength": 0.0, + "factors": {}, "active": False, + "timestamp": current_time, + "reason": "Outside trading session (07-17 UTC)" + } + + signals = {} + for name, direction in self.factors.items(): + try: + series = self._load_factor(name) + fac_1h = series.resample("1h").last() + if current_time in fac_1h.index: + val = fac_1h.loc[current_time] + else: + val = fac_1h.asof(current_time) + if pd.isna(val): + signals[name] = 0 + else: + signals[name] = direction * int(np.sign(val)) + except Exception: + signals[name] = 0 + + # Combine: average of individual signals + values = list(signals.values()) + combo = np.mean(values) if values else 0 + + # Round to nearest direction + if combo > 0.3: + signal = 1 + elif combo < -0.3: + signal = -1 + else: + signal = 0 + + strength = abs(combo) + agreeing = sum(1 for v in values if v == signal) + + return { + "signal": signal, + "strength": round(strength, 3), + "factors": signals, + "active": True, + "timestamp": current_time, + "agreeing_factors": f"{agreeing}/{len(values)}", + "reason": f"{'LONG' if signal == 1 else 'SHORT' if signal == -1 else 'FLAT'} ({agreeing}/{len(values)} factors agree)" + } + + +def main(): + strat = LiveStrategy() + result = strat.get_signal() + print(json.dumps(result, indent=2, default=str)) + + +if __name__ == "__main__": + main() diff --git a/scripts/nexquant_strategy_gen.py b/scripts/nexquant_strategy_gen.py new file mode 100644 index 00000000..3ca36c14 --- /dev/null +++ b/scripts/nexquant_strategy_gen.py @@ -0,0 +1,177 @@ +#!/usr/bin/env python +""" +NexQuant Multi-Timeframe Strategy Generator. + +Auto-tests 1h, 30min, daily frequencies with factor signals. +Selects the best-performing combination and saves it for live trading. +""" + +from __future__ import annotations + +import json, sys, time +from datetime import datetime +from pathlib import Path + +import numpy as np +import pandas as pd + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from rdagent.components.backtesting.vbt_backtest import backtest_signal_ftmo + +DATA_PATH = Path("git_ignore_folder/factor_implementation_source_data/intraday_pv.h5") +FACTORS_DIR = Path("results/factors") +VALS_DIR = FACTORS_DIR / "values" +OUT_DIR = Path("results/strategies_live") +OUT_DIR.mkdir(parents=True, exist_ok=True) +TXN_COST_BPS = 2.14 + + +def load_all_factors() -> list[dict]: + factors = [] + for f in sorted(FACTORS_DIR.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 (VALS_DIR / f"{safe}.parquet").exists(): + factors.append({"name": name, "ic": d["ic"], "safe": safe}) + return sorted(factors, key=lambda x: abs(x["ic"]), reverse=True) + + +def test_frequency(close: pd.Series, factors: list[dict], freq: str, session_filter: bool = True) -> list[dict]: + """Test all factors as signals at a given frequency.""" + c = close.resample(freq).last().dropna() if freq != "raw" else close + is_sess = (c.index.hour >= 7) & (c.index.hour < 17) if session_filter else pd.Series(True, index=c.index) + + results = [] + for f in factors[:100]: # Test top-100 + try: + s = pd.read_parquet(VALS_DIR / f"{f['safe']}.parquet").iloc[:, 0] + if isinstance(s.index, pd.MultiIndex): s = s.droplevel(-1) + fac = s.resample(freq).last().reindex(c.index).ffill() if freq != "raw" else s + except: continue + + for dr in [1, -1]: + sig = pd.Series(dr * np.sign(fac).fillna(0), index=c.index) + sig[~is_sess] = 0 + if sig.abs().sum() < 20: continue + + r = backtest_signal_ftmo(c, sig.fillna(0), txn_cost_bps=TXN_COST_BPS) + 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.5: + results.append({ + "factor": f["name"], "direction": dr, "frequency": freq, + "oos_sharpe": oos, "monthly_pct": oos_m, + "trades": r.get("oos_n_trades", 0), + }) + return sorted(results, key=lambda x: x["monthly_pct"], reverse=True) + + +def test_combo(close: pd.Series, top_signals: list[dict], freq: str, n: int) -> dict: + """Test a combination of N top signals at a given frequency.""" + c = close.resample(freq).last().dropna() if freq != "raw" else close + is_sess = (c.index.hour >= 7) & (c.index.hour < 17) + + signals = {} + for s in top_signals[:n]: + safe = s["factor"].replace("/", "_")[:150] + try: + series = pd.read_parquet(VALS_DIR / f"{safe}.parquet").iloc[:, 0] + if isinstance(series.index, pd.MultiIndex): series = series.droplevel(-1) + fac = series.resample(freq).last().reindex(c.index).ffill() if freq != "raw" else series + sig = pd.Series(s["direction"] * np.sign(fac).fillna(0), index=c.index) + sig[~is_sess] = 0 + signals[s["factor"]] = sig + except: pass + + if not signals: return {} + + combo = pd.DataFrame(signals, index=c.index).fillna(0).mean(axis=1) + r = backtest_signal_ftmo(c, combo.fillna(0), txn_cost_bps=TXN_COST_BPS, wf_rolling=True) + + return { + "frequency": freq, "n_signals": n, + "oos_monthly": r.get("oos_monthly_return_pct", 0) or 0, + "wf_monthly": r.get("wf_oos_monthly_return_mean", 0) or 0, + "oos_sharpe": r.get("wf_oos_sharpe_mean") or r.get("oos_sharpe", -999), + "max_dd": (r.get("oos_max_drawdown", 0) or 0) * 100, + "trades": r.get("oos_n_trades", 0), + "is_monthly": r.get("is_monthly_return_pct", 0) or 0, + "factors_used": list(signals.keys()), + } + + +def main(): + print(f"\n{'='*65}") + print(" NexQuant Multi-Timeframe Strategy Generator") + print(f"{'='*65}") + + close = pd.read_hdf(DATA_PATH, key="data")["$close"] + close = close.droplevel(-1).sort_index().dropna() + factors = load_all_factors() + print(f"Data: {len(close):,} bars | Factors: {len(factors)}\n") + + all_combos = [] + + for freq, label in [("1h", "1-Hour"), ("30min", "30-Min"), ("1D", "Daily")]: + print(f"=== {label} ===") + t0 = time.time() + top = test_frequency(close, factors, freq) + + if not top: + print(f" No profitable signals\n") + continue + + print(f" Profitable signals: {len(top)}") + print(f" Top: {top[0]['factor'][:40]} → +{top[0]['monthly_pct']:.2f}%/month") + + # Test combos + for n in [2, 3, 5]: + combo = test_combo(close, top, freq, n) + if combo: + all_combos.append(combo) + hit = "🎯" if combo["oos_monthly"] >= 4 else "✅" if combo["oos_monthly"] > 0 else "" + print(f" {n}sig combo: +{combo['oos_monthly']:.2f}%/mon DD={combo['max_dd']:.1f}% T={combo['trades']} {hit}") + + print(f" ({time.time()-t0:.0f}s)\n") + + # Best overall + all_combos.sort(key=lambda x: x["oos_monthly"], reverse=True) + + print(f"{'='*65}") + print(f" FINAL RANKING") + print(f"{'='*65}") + print(f" {'Freq':<8} {'N':>3} {'Mon%':>8} {'DD%':>7} {'Trades':>7}") + print(f" {'─'*35}") + for c in all_combos[:10]: + print(f" {c['frequency']:<8} {c['n_signals']:>3} {c['oos_monthly']:>+7.2f}% {c['max_dd']:>+6.1f}% {c['trades']:>7}") + + best = all_combos[0] + print(f"\n BEST: {best['frequency']} / {best['n_signals']} signals") + print(f" Monthly: +{best['oos_monthly']:.2f}% | DD: {best['max_dd']:.1f}% | Trades: {best['trades']}") + print(f" Factors: {best['factors_used']}") + + # Save best config + config = { + "generated_at": datetime.now().isoformat(), + "frequency": best["frequency"], + "n_signals": best["n_signals"], + "factors": best["factors_used"], + "metrics": { + "oos_monthly_pct": best["oos_monthly"], + "wf_monthly_pct": best["wf_monthly"], + "oos_sharpe": best["oos_sharpe"], + "max_dd_pct": best["max_dd"], + "trades": best["trades"], + }, + } + with open(OUT_DIR / "live_config.json", "w") as f: + json.dump(config, f, indent=2) + print(f"\n Config saved: {OUT_DIR / 'live_config.json'}") + + +if __name__ == "__main__": + main()