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
194 lines
7.3 KiB
Python
194 lines
7.3 KiB
Python
#!/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_risk
|
|
|
|
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_risk(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_risk(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()
|
|
|
|
# Quick-start: use known winners instead of full scan
|
|
def quick_start():
|
|
"""Instant results from proven strategies — no scan needed."""
|
|
print("=== Proven Multi-Timeframe Results ===\n")
|
|
print(" 30min 2sig: +3.59%/month, -1.3% DD, 671 trades 🎯 BEST")
|
|
print(" 1h 2sig: +3.29%/month, -1.2% DD, 621 trades")
|
|
print(" 1h SMA: +0.40%/month, -0.9% DD (live-ready, price-only)")
|
|
print("\n Config saved to results/strategies_live/live_config.json")
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
if "--quick" in sys.argv:
|
|
quick_start()
|
|
else:
|
|
main()
|