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
301 lines
11 KiB
Python
301 lines
11 KiB
Python
#!/usr/bin/env python
|
|
"""
|
|
NexQuant Systematic Strategy Generator — kein LLM, nur Mathematik.
|
|
|
|
Grid-searched threshold strategies with IC-weighted z-score composites.
|
|
Optionally trains LightGBM directional classifier.
|
|
|
|
Approaches:
|
|
A) IC-weighted z-score composite (always used as base)
|
|
B) Grid-search entry/exit thresholds (primary)
|
|
C) LightGBM directional classifier (optional, if factors ≥ 5)
|
|
D) Factor-ranking top/bottom deciles (fast baseline)
|
|
|
|
Output: Best strategy by OOS Walk-Forward Sharpe, saved to results/strategies_systematic/
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import time
|
|
from datetime import datetime
|
|
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))
|
|
|
|
DATA_PATH = Path("git_ignore_folder/factor_implementation_source_data/intraday_pv.h5")
|
|
FACTORS_DIR = Path("results/factors")
|
|
OUT_DIR = Path("results/strategies_systematic")
|
|
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
TXN_COST_BPS = 2.14
|
|
OOS_START = "2024-01-01"
|
|
WF_WINDOWS = 4
|
|
|
|
|
|
def load_data() -> tuple:
|
|
"""Load OHLCV close prices and top factors."""
|
|
ohlcv = pd.read_hdf(DATA_PATH, key="data")
|
|
close = ohlcv["$close"]
|
|
if isinstance(close.index, pd.MultiIndex):
|
|
close = close.droplevel(-1)
|
|
close = close.sort_index().dropna()
|
|
|
|
factors = []
|
|
for f in sorted(FACTORS_DIR.glob("*.json")):
|
|
try:
|
|
d = json.loads(f.read_text())
|
|
except Exception:
|
|
continue
|
|
if d.get("status") != "success" or d.get("ic") is None:
|
|
continue
|
|
name = d.get("factor_name", f.stem)
|
|
safe = name.replace("/", "_").replace("\\", "_")[:150]
|
|
pf = FACTORS_DIR / "values" / f"{safe}.parquet"
|
|
if pf.exists():
|
|
factors.append({"name": name, "ic": d["ic"]})
|
|
|
|
factors.sort(key=lambda x: abs(x["ic"]), reverse=True)
|
|
return close, factors
|
|
|
|
|
|
def load_factor_values(factor_names: list, close: pd.Series) -> pd.DataFrame:
|
|
"""Load and align factor time series."""
|
|
data = {}
|
|
for name in factor_names:
|
|
safe = name.replace("/", "_").replace("\\", "_")[:150]
|
|
pf = FACTORS_DIR / "values" / f"{safe}.parquet"
|
|
if not pf.exists():
|
|
continue
|
|
series = pd.read_parquet(pf).iloc[:, 0]
|
|
if isinstance(series.index, pd.MultiIndex):
|
|
series = series.droplevel(-1)
|
|
data[name] = series
|
|
|
|
df = pd.DataFrame(data)
|
|
common = close.index.intersection(df.dropna(how="all").index)
|
|
return df.loc[common].ffill(), close.loc[common]
|
|
|
|
|
|
def compute_ic_weighted_composite(factors_df: pd.DataFrame, ics: dict[str, float]) -> pd.Series:
|
|
"""Compute z-score normalized, IC-weighted composite signal."""
|
|
composite = pd.Series(0.0, index=factors_df.index)
|
|
total_abs_ic = 0.0
|
|
|
|
for col in factors_df.columns:
|
|
if col not in ics:
|
|
continue
|
|
ic = ics[col]
|
|
if abs(ic) < 0.001:
|
|
continue
|
|
z = (factors_df[col] - factors_df[col].rolling(20).mean()) / (
|
|
factors_df[col].rolling(20).std() + 1e-8
|
|
)
|
|
weight = ic # Keep sign: if IC < 0, invert factor
|
|
composite += weight * z
|
|
total_abs_ic += abs(ic)
|
|
|
|
if total_abs_ic > 0:
|
|
composite /= total_abs_ic
|
|
return composite
|
|
|
|
|
|
def generate_signal_threshold(composite: pd.Series, entry: float, exit_thresh: float) -> pd.Series:
|
|
"""Generate signal from composite with entry/exit thresholds (vectorized)."""
|
|
signal = pd.Series(0, index=composite.index, dtype=float)
|
|
signal[composite > entry] = 1
|
|
signal[composite < -entry] = -1
|
|
# Simple: no hysteresis for speed. Entry = exit.
|
|
return signal
|
|
|
|
|
|
def generate_signal_ranking(factors_df: pd.DataFrame, ics: dict, top_pct: float = 0.10) -> pd.Series:
|
|
"""Factor-ranking: top/bottom deciles = long/short, daily rebalanced."""
|
|
composite = compute_ic_weighted_composite(factors_df, ics)
|
|
signal = pd.Series(0, index=composite.index)
|
|
|
|
for date, group in composite.groupby(composite.index.normalize()):
|
|
n = len(group)
|
|
k = max(1, int(n * top_pct))
|
|
ranked = group.abs().sort_values(ascending=False)
|
|
top_idx = ranked.index[:k]
|
|
bot_idx = ranked.index[-k:]
|
|
signal.loc[top_idx] = np.sign(composite.loc[top_idx])
|
|
signal.loc[bot_idx] = np.sign(composite.loc[bot_idx]) * -1
|
|
|
|
return signal
|
|
|
|
|
|
def grid_search(close: pd.Series, composite: pd.Series, style: str = "swing") -> dict:
|
|
"""Grid-search optimal entry thresholds."""
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal_risk
|
|
|
|
best = None
|
|
best_sharpe = -999
|
|
|
|
entries = np.arange(0.3, 2.1, 0.3)
|
|
|
|
for entry in entries:
|
|
sig = generate_signal_threshold(composite, entry, 0.0)
|
|
r = backtest_signal_risk(close, sig, txn_cost_bps=TXN_COST_BPS, wf_rolling=True)
|
|
|
|
wf_sharpe = r.get("wf_oos_sharpe_mean", -999) or -999
|
|
if wf_sharpe > best_sharpe:
|
|
best_sharpe = wf_sharpe
|
|
best = {
|
|
"entry": entry,
|
|
"wf_sharpe": wf_sharpe,
|
|
"oos_sharpe": r.get("oos_sharpe", -999),
|
|
"oos_monthly": r.get("oos_monthly_return_pct", 0),
|
|
"oos_dd": r.get("oos_max_drawdown", 0),
|
|
"oos_trades": r.get("oos_n_trades", 0),
|
|
"oos_wr": r.get("oos_win_rate", 0),
|
|
"is_sharpe": r.get("is_sharpe", -999),
|
|
"consistency": r.get("wf_oos_consistency", 0),
|
|
"mc_pvalue": r.get("mc_pvalue", 1),
|
|
"full_result": r,
|
|
}
|
|
print(f" entry={entry:.1f} → WF={wf_sharpe:.3f} OOS_S={r.get('oos_sharpe',0):.3f} OOS_M={r.get('oos_monthly_return_pct',0):.2f}%")
|
|
|
|
return best
|
|
|
|
|
|
def train_lightgbm(factors_df: pd.DataFrame, close: pd.Series, forward_bars: int = 96) -> Optional[dict]:
|
|
"""Train LightGBM directional classifier (approach C)."""
|
|
try:
|
|
import lightgbm as lgb
|
|
except ImportError:
|
|
print(" LightGBM not available — skipping")
|
|
return None
|
|
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal_risk
|
|
|
|
print(" Training LightGBM directional classifier...")
|
|
fwd_ret = close.pct_change(forward_bars).shift(-forward_bars)
|
|
common = factors_df.index.intersection(fwd_ret.dropna().index)
|
|
X = factors_df.loc[common].ffill().values
|
|
y = np.sign(fwd_ret.loc[common].values)
|
|
|
|
split = int(len(X) * 0.7)
|
|
X_train, X_test = X[:split], X[split:]
|
|
y_train, y_test = y[:split], y[split:]
|
|
|
|
model = lgb.LGBMClassifier(n_estimators=200, max_depth=6, num_leaves=31,
|
|
learning_rate=0.05, random_state=42, verbose=-1)
|
|
model.fit(X_train, y_train)
|
|
preds = model.predict(X_test)
|
|
signal = pd.Series(preds, index=common[split:])
|
|
|
|
r = backtest_signal_risk(close.loc[common[split:]], signal,
|
|
txn_cost_bps=TXN_COST_BPS, wf_rolling=True)
|
|
wf = r.get("wf_oos_sharpe_mean", -999) or -999
|
|
print(f" LightGBM: WF_Sharpe={wf:.3f}")
|
|
return {
|
|
"method": "LightGBM",
|
|
"wf_sharpe": wf,
|
|
"oos_sharpe": r.get("oos_sharpe", -999),
|
|
"oos_monthly": r.get("oos_monthly_return_pct", 0),
|
|
"oos_dd": r.get("oos_max_drawdown", 0),
|
|
"oos_trades": r.get("oos_n_trades", 0),
|
|
"full_result": r,
|
|
}
|
|
|
|
|
|
def main():
|
|
print(f"\n{'='*60}")
|
|
print(" NexQuant Systematic Strategy Generator")
|
|
print(f" Cost: {TXN_COST_BPS} bps | OOS: {OOS_START} | WF: {WF_WINDOWS} windows")
|
|
print(f"{'='*60}\n")
|
|
|
|
close, factors = load_data()
|
|
print(f"Loaded: {len(close):,} bars, {len(factors)} factors")
|
|
|
|
# Take top-10 diverse factors
|
|
top_names = [f["name"] for f in factors[:10]]
|
|
ics = {f["name"]: f["ic"] for f in factors[:10]}
|
|
factors_df, close_a = load_factor_values(top_names, close)
|
|
print(f"Aligned: {len(factors_df.columns)} factors, {len(close_a):,} bars\n")
|
|
|
|
results = []
|
|
|
|
# ---- Approach A+B: IC-weighted z-score + grid-search thresholds ----
|
|
print("=== A+B: IC-Weighted Z-Score + Grid-Search Thresholds ===")
|
|
t0 = time.time()
|
|
composite = compute_ic_weighted_composite(factors_df, ics)
|
|
best_thresh = grid_search(close_a, composite)
|
|
if best_thresh:
|
|
best_thresh["method"] = "IC-weighted + thresholds"
|
|
best_thresh["composite_style"] = "zscore"
|
|
best_thresh["factors_used"] = top_names[:5]
|
|
results.append(best_thresh)
|
|
print(f" Best: entry={best_thresh['entry']:.1f} exit={best_thresh['exit']:.1f} "
|
|
f"WF_Sharpe={best_thresh['wf_sharpe']:.3f} ({time.time()-t0:.0f}s)\n")
|
|
|
|
# ---- Approach D: Factor-Ranking Top/Bottom ----
|
|
print("=== D: Factor-Ranking Top/Bottom Deciles ===")
|
|
t0 = time.time()
|
|
sig_rank = generate_signal_ranking(factors_df, ics, top_pct=0.10)
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal_risk
|
|
r_rank = backtest_signal_risk(close_a, sig_rank, txn_cost_bps=TXN_COST_BPS, wf_rolling=True)
|
|
wf_rank = r_rank.get("wf_oos_sharpe_mean", -999) or -999
|
|
results.append({
|
|
"method": "Factor-Ranking D",
|
|
"wf_sharpe": wf_rank,
|
|
"oos_sharpe": r_rank.get("oos_sharpe", -999),
|
|
"oos_monthly": r_rank.get("oos_monthly_return_pct", 0),
|
|
"oos_dd": r_rank.get("oos_max_drawdown", 0),
|
|
"oos_trades": r_rank.get("oos_n_trades", 0),
|
|
"full_result": r_rank,
|
|
})
|
|
print(f" Factor-Ranking: WF_Sharpe={wf_rank:.3f} ({time.time()-t0:.0f}s)\n")
|
|
|
|
# ---- Approach C: LightGBM (if enough factors) ----
|
|
if len(factors_df.columns) >= 5:
|
|
print("=== C: LightGBM Directional Classifier ===")
|
|
t0 = time.time()
|
|
lgb_result = train_lightgbm(factors_df, close_a)
|
|
if lgb_result:
|
|
lgb_result["factors_used"] = top_names[:10]
|
|
results.append(lgb_result)
|
|
print(f" ({time.time()-t0:.0f}s)\n")
|
|
|
|
# ---- Report ----
|
|
results.sort(key=lambda x: x.get("wf_sharpe", -999) or -999, reverse=True)
|
|
|
|
print(f"\n{'='*60}")
|
|
print(" RESULTS (sorted by Walk-Forward OOS Sharpe)")
|
|
print(f"{'='*60}")
|
|
print(f"{'Method':<30} {'WF Sharpe':>10} {'OOS Sharpe':>10} {'OOS Mon%':>8} {'OOS DD%':>8}")
|
|
print("-" * 70)
|
|
|
|
for r in results:
|
|
wf = r.get("wf_sharpe", -999) or -999
|
|
oos_s = r.get("oos_sharpe", -999)
|
|
oos_m = (r.get("oos_monthly", 0) or 0)
|
|
oos_d = (r.get("oos_dd", 0) or 0) * 100
|
|
print(f"{r['method']:<30} {wf:>10.3f} {oos_s:>10.3f} {oos_m:>8.2f}% {oos_d:>7.1f}%")
|
|
|
|
# Save best result
|
|
if results:
|
|
best = results[0]
|
|
best["generated_at"] = datetime.now().isoformat()
|
|
best["n_factors"] = len(factors_df.columns)
|
|
best["n_bars"] = len(close_a)
|
|
best["cost_bps"] = TXN_COST_BPS
|
|
|
|
fname = f"systematic_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{best['method'].replace(' ','_')[:40]}.json"
|
|
with open(OUT_DIR / fname, "w") as f:
|
|
json.dump({k: v for k, v in best.items() if k != "full_result"}, f, indent=2, default=str)
|
|
print(f"\nBest strategy saved: {fname}")
|
|
|
|
print()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|