From 4758de0eeead0d938d993cfbe4e0b1879fba0463 Mon Sep 17 00:00:00 2001 From: TPTBusiness Date: Fri, 22 May 2026 15:10:36 +0200 Subject: [PATCH] refactor: remove all proprietary terms from codebase and git history MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- rdagent/components/backtesting/__init__.py | 18 +- .../components/backtesting/vbt_backtest.py | 90 ++-- scripts/nexquant_1h_factors.py | 6 +- scripts/nexquant_20tests.py | 4 +- scripts/nexquant_30min_scan.py | 6 +- scripts/nexquant_add_risk_management.py | 30 +- scripts/nexquant_continuous_strategies.py | 4 +- scripts/nexquant_gen_strategies_real_bt.py | 18 +- scripts/nexquant_gridsearch.py | 8 +- scripts/nexquant_infinite_search.py | 4 +- scripts/nexquant_ml_pipeline.py | 6 +- scripts/nexquant_multi_asset.py | 6 +- scripts/nexquant_portfolio.py | 4 +- scripts/nexquant_portfolio_optimizer.py | 6 +- scripts/nexquant_quick_daytrading.py | 6 +- scripts/nexquant_rebacktest_unified.py | 12 +- scripts/nexquant_smart_strategy_gen.py | 36 +- scripts/nexquant_strategy_gen.py | 6 +- scripts/nexquant_systematic.py | 300 ++++++++++++++ scripts/nexquant_unified.py | 166 ++++++++ scripts/realistic_backtest_all.py | 24 +- test/backtesting/test_ftmo_oos.py | 384 +++++++++--------- test/integration/test_full_pipeline.py | 36 +- test/local/test_continuous_strategies.py | 4 +- test/local/test_optuna_optimizer.py | 50 +-- test/local/test_strategy_worker.py | 18 +- test/qlib/test_headform.py | 10 +- test/qlib/test_headform4.py | 8 +- test/qlib/test_open_source_suite.py | 10 +- 29 files changed, 873 insertions(+), 407 deletions(-) create mode 100644 scripts/nexquant_systematic.py create mode 100644 scripts/nexquant_unified.py diff --git a/rdagent/components/backtesting/__init__.py b/rdagent/components/backtesting/__init__.py index a941fd31..4b2c4a42 100644 --- a/rdagent/components/backtesting/__init__.py +++ b/rdagent/components/backtesting/__init__.py @@ -5,18 +5,18 @@ from .risk_management import CorrelationAnalyzer, PortfolioOptimizer, AdvancedRi from .vbt_backtest import ( DEFAULT_BARS_PER_YEAR, DEFAULT_TXN_COST_BPS, - FTMO_INITIAL_CAPITAL, - FTMO_MAX_DAILY_LOSS, - FTMO_MAX_TOTAL_LOSS, - FTMO_MAX_LEVERAGE, - FTMO_RISK_PER_TRADE, + INITIAL_CAPITAL, + MAX_DAILY_LOSS, + MAX_TOTAL_LOSS, + MAX_LEVERAGE, + RISK_PER_TRADE, OOS_START_DEFAULT, WF_IS_YEARS, WF_OOS_YEARS, WF_STEP_YEARS, backtest_from_forward_returns, backtest_signal, - backtest_signal_ftmo, + backtest_signal_risk, monte_carlo_trade_pvalue, walk_forward_rolling, ) @@ -24,10 +24,10 @@ from .vbt_backtest import ( __all__ = [ 'BacktestMetrics', 'FactorBacktester', 'ResultsDatabase', 'CorrelationAnalyzer', 'PortfolioOptimizer', 'AdvancedRiskManager', - 'backtest_signal', 'backtest_signal_ftmo', 'backtest_from_forward_returns', + 'backtest_signal', 'backtest_signal_risk', 'backtest_from_forward_returns', 'monte_carlo_trade_pvalue', 'walk_forward_rolling', 'DEFAULT_BARS_PER_YEAR', 'DEFAULT_TXN_COST_BPS', - 'FTMO_INITIAL_CAPITAL', 'FTMO_MAX_DAILY_LOSS', 'FTMO_MAX_TOTAL_LOSS', - 'FTMO_MAX_LEVERAGE', 'FTMO_RISK_PER_TRADE', 'OOS_START_DEFAULT', + 'INITIAL_CAPITAL', 'MAX_DAILY_LOSS', 'MAX_TOTAL_LOSS', + 'MAX_LEVERAGE', 'RISK_PER_TRADE', 'OOS_START_DEFAULT', 'WF_IS_YEARS', 'WF_OOS_YEARS', 'WF_STEP_YEARS', ] diff --git a/rdagent/components/backtesting/vbt_backtest.py b/rdagent/components/backtesting/vbt_backtest.py index 74e63edb..76433314 100644 --- a/rdagent/components/backtesting/vbt_backtest.py +++ b/rdagent/components/backtesting/vbt_backtest.py @@ -38,15 +38,15 @@ DEFAULT_TXN_COST_BPS = 2.14 DEFAULT_BARS_PER_YEAR = 252 * 1440 # 252 trading days * 1440 min/day = 362,880 EXTREME_BAR_THRESHOLD = 0.05 # |ret| > 5% on a single 1-min bar → suspicious -# FTMO 100k account rules (enforced in backtest_signal when ftmo=True) -FTMO_INITIAL_CAPITAL = 100_000.0 -FTMO_MAX_DAILY_LOSS = 0.05 # 5% of initial → block new trades rest of day -FTMO_MAX_TOTAL_LOSS = 0.10 # 10% of initial → simulation ends +# RiskMgmt 100k account rules (enforced in backtest_signal when riskmgmt=True) +INITIAL_CAPITAL = 100_000.0 +MAX_DAILY_LOSS = 0.05 # 5% of initial → block new trades rest of day +MAX_TOTAL_LOSS = 0.10 # 10% of initial → simulation ends # Risk-based position sizing: 1.5% equity risk per trade, 10-pip stop, max 1:30 leverage -FTMO_RISK_PER_TRADE = 0.015 -FTMO_STOP_PIPS = 10 -FTMO_PIP = 0.0001 -FTMO_MAX_LEVERAGE = 30 +RISK_PER_TRADE = 0.015 +STOP_PIPS = 10 +PIP_SIZE = 0.0001 +MAX_LEVERAGE = 30 def _compute_trade_pnl(position: pd.Series, strategy_returns: pd.Series) -> pd.Series: @@ -274,31 +274,31 @@ def backtest_signal( return result -def _apply_ftmo_mask( +def _apply_risk_mask( signal: pd.Series, close: pd.Series, leverage: float, txn_cost_bps: float, ) -> tuple[pd.Series, dict]: """ - Apply FTMO daily/total loss rules to a signal series. + Apply RiskMgmt daily/total loss rules to a signal series. Returns a masked signal (positions zeroed after each limit breach) and - a dict of FTMO compliance metrics. + a dict of RiskMgmt compliance metrics. """ txn_cost = txn_cost_bps / 10_000.0 position = signal.shift(1).fillna(0) * leverage bar_ret = close.pct_change().fillna(0) - equity = FTMO_INITIAL_CAPITAL - peak_day = FTMO_INITIAL_CAPITAL + equity = INITIAL_CAPITAL + peak_day = INITIAL_CAPITAL masked = signal.copy() daily_breaches = 0 total_breached = False total_breach_ts: pd.Timestamp | None = None current_day = None - day_start_eq = FTMO_INITIAL_CAPITAL + day_start_eq = INITIAL_CAPITAL pos_prev = 0.0 for ts, sig_i in signal.items(): @@ -319,24 +319,24 @@ def _apply_ftmo_mask( masked.at[ts] = 0 continue - daily_loss = (equity - day_start_eq) / FTMO_INITIAL_CAPITAL - total_loss = (equity - FTMO_INITIAL_CAPITAL) / FTMO_INITIAL_CAPITAL + daily_loss = (equity - day_start_eq) / INITIAL_CAPITAL + total_loss = (equity - INITIAL_CAPITAL) / INITIAL_CAPITAL - if daily_loss < -FTMO_MAX_DAILY_LOSS: + if daily_loss < -MAX_DAILY_LOSS: daily_breaches += 1 day_start_eq = -999 # block rest of day masked.at[ts] = 0 - if total_loss < -FTMO_MAX_TOTAL_LOSS: + if total_loss < -MAX_TOTAL_LOSS: total_breached = True total_breach_ts = ts masked.at[ts] = 0 return masked, { - "ftmo_daily_breaches": daily_breaches, - "ftmo_total_breached": total_breached, - "ftmo_total_breach_ts": str(total_breach_ts) if total_breach_ts else None, - "ftmo_compliant": not total_breached and daily_breaches == 0, + "risk_daily_breaches": daily_breaches, + "risk_total_breached": total_breached, + "risk_total_breach_ts": str(total_breach_ts) if total_breach_ts else None, + "risk_compliant": not total_breached and daily_breaches == 0, } @@ -403,7 +403,7 @@ def walk_forward_rolling( """ Rolling walk-forward validation: multiple IS/OOS windows shifted by ``step_years``. - Each window runs an independent FTMO simulation on the IS and OOS slices. + Each window runs an independent RiskMgmt simulation on the IS and OOS slices. Produces aggregate OOS statistics to measure cross-time consistency. Returns @@ -442,7 +442,7 @@ def walk_forward_rolling( for mask, prefix in [(is_mask, "is"), (oos_mask, "oos")]: close_s = close.loc[mask] signal_s = signal.loc[mask] - masked_s, _ = _apply_ftmo_mask(signal_s, close_s, leverage, txn_cost_bps) + masked_s, _ = _apply_risk_mask(signal_s, close_s, leverage, txn_cost_bps) r = backtest_signal(close=close_s, signal=masked_s, txn_cost_bps=txn_cost_bps, bars_per_year=bars_per_year) window[f"{prefix}_sharpe"] = r.get("sharpe", 0.0) @@ -466,14 +466,14 @@ def walk_forward_rolling( } -def backtest_signal_ftmo( +def backtest_signal_risk( close: pd.Series, signal: pd.Series, txn_cost_bps: float = DEFAULT_TXN_COST_BPS, eurusd_price: float = 1.10, - risk_pct: float = FTMO_RISK_PER_TRADE, - stop_pips: float = FTMO_STOP_PIPS, - max_leverage: float = FTMO_MAX_LEVERAGE, + risk_pct: float = RISK_PER_TRADE, + stop_pips: float = STOP_PIPS, + max_leverage: float = MAX_LEVERAGE, bars_per_year: int = DEFAULT_BARS_PER_YEAR, forward_returns: pd.Series | None = None, oos_start: str | None = OOS_START_DEFAULT, @@ -481,15 +481,15 @@ def backtest_signal_ftmo( mc_n_permutations: int = 0, ) -> dict[str, Any]: """ - FTMO-compliant backtest of a strategy signal on EUR/USD. + RiskMgmt-compliant backtest of a strategy signal on EUR/USD. Applies on top of ``backtest_signal``: - Realistic costs: default 2.14 bps (≈ 2.35 pip spread+slippage+commission) - Risk-based position sizing: risk_pct equity per trade, stop_pips hard stop - - Max leverage cap: max_leverage (default 1:30, FTMO standard) - - FTMO daily loss limit (5%): positions zeroed rest of day after breach - - FTMO total loss limit (10%): all positions zeroed after breach - - FTMO-specific metrics added to result dict + - Max leverage cap: max_leverage (default 1:30, RiskMgmt standard) + - RiskMgmt daily loss limit (5%): positions zeroed rest of day after breach + - RiskMgmt total loss limit (10%): all positions zeroed after breach + - RiskMgmt-specific metrics added to result dict - Walk-forward OOS split: IS metrics (before oos_start) + OOS metrics (after) Parameters @@ -507,7 +507,7 @@ def backtest_signal_ftmo( stop_pips : float Hard stop-loss distance in pips (default 10). max_leverage : float - Maximum leverage (default 30 = FTMO 1:30). + Maximum leverage (default 30 = RiskMgmt 1:30). oos_start : str or None Start of out-of-sample period (ISO date). None disables OOS split. wf_rolling : bool @@ -518,11 +518,11 @@ def backtest_signal_ftmo( When > 0, computes ``mc_pvalue``: fraction of permuted sequences whose total return >= real total return. p < 0.05 indicates a genuine edge. """ - stop_price = stop_pips * FTMO_PIP + stop_price = stop_pips * PIP_SIZE leverage_by_risk = risk_pct / (stop_price / eurusd_price) leverage = min(leverage_by_risk, max_leverage) - masked_signal, ftmo_metrics = _apply_ftmo_mask(signal, close, leverage, txn_cost_bps) + masked_signal, risk_metrics = _apply_risk_mask(signal, close, leverage, txn_cost_bps) result = backtest_signal( close=close, @@ -532,14 +532,14 @@ def backtest_signal_ftmo( forward_returns=forward_returns, ) - result.update(ftmo_metrics) - result["ftmo_leverage"] = round(leverage, 2) - result["ftmo_risk_pct"] = risk_pct - result["ftmo_stop_pips"] = stop_pips + result.update(risk_metrics) + result["risk_leverage"] = round(leverage, 2) + result["risk_risk_pct"] = risk_pct + result["risk_stop_pips"] = stop_pips - # Re-scale reported equity metrics to FTMO_INITIAL_CAPITAL - result["ftmo_end_equity"] = FTMO_INITIAL_CAPITAL * (1 + result.get("total_return", 0)) - result["ftmo_monthly_profit"] = FTMO_INITIAL_CAPITAL * result.get("monthly_return", 0) + # Re-scale reported equity metrics to INITIAL_CAPITAL + result["risk_end_equity"] = INITIAL_CAPITAL * (1 + result.get("total_return", 0)) + result["risk_monthly_profit"] = INITIAL_CAPITAL * result.get("monthly_return", 0) # Walk-forward OOS split if oos_start is not None: @@ -551,9 +551,9 @@ def backtest_signal_ftmo( if mask.sum() < 100: return close_s = close.loc[mask] - signal_s = signal.loc[mask] # raw signal, not masked — fresh FTMO sim per period + signal_s = signal.loc[mask] # raw signal, not masked — fresh RiskMgmt sim per period fwd_split = forward_returns.loc[mask] if forward_returns is not None else None - masked_s, _ = _apply_ftmo_mask(signal_s, close_s, leverage, txn_cost_bps) + masked_s, _ = _apply_risk_mask(signal_s, close_s, leverage, txn_cost_bps) split_result = backtest_signal( close=close_s, signal=masked_s, diff --git a/scripts/nexquant_1h_factors.py b/scripts/nexquant_1h_factors.py index 91367fab..c43c9976 100644 --- a/scripts/nexquant_1h_factors.py +++ b/scripts/nexquant_1h_factors.py @@ -1,6 +1,6 @@ import json, numpy as np, pandas as pd from pathlib import Path -from rdagent.components.backtesting.vbt_backtest import backtest_signal_ftmo +from rdagent.components.backtesting.vbt_backtest import backtest_signal_risk close = pd.read_hdf("git_ignore_folder/factor_implementation_source_data/intraday_pv.h5", key="data")["$close"] close = close.droplevel(-1).sort_index().dropna().resample("1h").last().dropna() @@ -34,7 +34,7 @@ for i, f in enumerate(factors[:100]): sig = pd.Series(dr * np.sign(fac).fillna(0), index=close.index) sig[~is_session] = 0 if sig.abs().sum() < 20: continue - r = backtest_signal_ftmo(close, sig.fillna(0), txn_cost_bps=2.14) + r = backtest_signal_risk(close, 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 results.append((f"{f['name']}_{label}", oos, oos_m, r.get("oos_n_trades",0))) @@ -67,7 +67,7 @@ if top: df = pd.DataFrame(all_sig, index=close.index).fillna(0) for n in [3, 5, 8]: combo = df[list(df.columns)[:n]].mean(axis=1) - r = backtest_signal_ftmo(close, combo.fillna(0), txn_cost_bps=2.14, wf_rolling=True) + r = backtest_signal_risk(close, combo.fillna(0), txn_cost_bps=2.14, wf_rolling=True) oos_m = r.get("oos_monthly_return_pct",0) or 0 dd = (r.get("oos_max_drawdown",0) or 0)*100 ann = ((1+oos_m/100)**12-1)*100 diff --git a/scripts/nexquant_20tests.py b/scripts/nexquant_20tests.py index 8887b4e8..bc9e8e1b 100644 --- a/scripts/nexquant_20tests.py +++ b/scripts/nexquant_20tests.py @@ -18,7 +18,7 @@ import pandas as pd sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) -from rdagent.components.backtesting.vbt_backtest import backtest_signal_ftmo +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") @@ -71,7 +71,7 @@ def backtest(signal, close, label="") -> dict: if signal is None or len(signal) < 100: return {"wf_sharpe": -999, "oos_sharpe": -999, "oos_monthly": 0, "oos_dd": 0, "trades": 0} common = close.index.intersection(signal.dropna().index) - r = backtest_signal_ftmo(close.loc[common], signal.reindex(common).fillna(0), + r = backtest_signal_risk(close.loc[common], signal.reindex(common).fillna(0), txn_cost_bps=TXN_COST_BPS, wf_rolling=False) oos = r.get("oos_sharpe", -999) return { diff --git a/scripts/nexquant_30min_scan.py b/scripts/nexquant_30min_scan.py index 1d6cebca..1f73fd1f 100644 --- a/scripts/nexquant_30min_scan.py +++ b/scripts/nexquant_30min_scan.py @@ -2,7 +2,7 @@ """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 +from rdagent.components.backtesting.vbt_backtest import backtest_signal_risk 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() @@ -33,7 +33,7 @@ for i, f in enumerate(factors[:200]): 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) + r = backtest_signal_risk(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: @@ -72,7 +72,7 @@ if results: 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) + r = backtest_signal_risk(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) diff --git a/scripts/nexquant_add_risk_management.py b/scripts/nexquant_add_risk_management.py index 921ad248..ee2b8ced 100644 --- a/scripts/nexquant_add_risk_management.py +++ b/scripts/nexquant_add_risk_management.py @@ -1,6 +1,6 @@ #!/usr/bin/env python """ -Add FTMO-compliant risk management to existing strategies. +Add RiskMgmt-compliant risk management to existing strategies. For each accepted strategy, add: - Stop Loss: 2% @@ -27,11 +27,11 @@ console = Console() STRATEGIES_DIR = Path('results/strategies_new') OHLCV_PATH = Path('git_ignore_folder/factor_implementation_source_data/intraday_pv.h5') -# FTMO Risk Parameters +# RiskMgmt Risk Parameters STOP_LOSS = 0.02 # 2% hard stop TAKE_PROFIT = 0.04 # 4% target (2x SL) TRAILING_STOP = 0.015 # 1.5% trail after 2% profit -MAX_DAILY_LOSS = 0.05 # 5% FTMO daily limit +MAX_DAILY_LOSS = 0.05 # 5% RiskMgmt daily limit def load_ohlcv(): """Load OHLCV close prices.""" @@ -147,11 +147,11 @@ def evaluate_strategy(strategy_returns, signal_aligned): 'n_bars': int(n_bars), 'n_months': float(n_months), 'max_daily_loss': float(max_daily_loss), - 'ftmo_compliant': max_daily_loss <= MAX_DAILY_LOSS and max_dd > -0.10, + 'riskmgmt_compliant': max_daily_loss <= MAX_DAILY_LOSS and max_dd > -0.10, } def main(): - console.print("[bold cyan]🔒 Adding FTMO Risk Management to Existing Strategies[/bold cyan]\n") + console.print("[bold cyan]🔒 Adding RiskMgmt Risk Management to Existing Strategies[/bold cyan]\n") # Load OHLCV console.print("📊 Loading OHLCV data...") @@ -254,7 +254,7 @@ def main(): 'new_trades': metrics['n_trades'], 'new_monthly_ret': metrics['monthly_return_pct'], 'max_daily_loss': metrics['max_daily_loss'], - 'ftmo_compliant': bool(metrics['ftmo_compliant']), + 'riskmgmt_compliant': bool(metrics['riskmgmt_compliant']), } results.append(result) @@ -265,7 +265,7 @@ def main(): 'trailing_stop': TRAILING_STOP, 'trailing_trigger': 0.02, 'max_daily_loss': MAX_DAILY_LOSS, - 'ftmo_compliant': bool(metrics['ftmo_compliant']), + 'riskmgmt_compliant': bool(metrics['riskmgmt_compliant']), } data['evaluated_with_risk_mgmt'] = metrics data['summary'] = { @@ -275,7 +275,7 @@ def main(): 'monthly_return_pct': metrics['monthly_return_pct'], 'real_ic': metrics['ic'], 'real_n_trades': metrics['n_trades'], - 'ftmo_compliant': bool(metrics['ftmo_compliant']), + 'riskmgmt_compliant': bool(metrics['riskmgmt_compliant']), 'forward_bars': 12, 'trading_style': 'daytrading', } @@ -296,7 +296,7 @@ def main(): # Display results console.print("\n[bold green]✓ All strategies processed![/bold green]\n") - table = Table(title="📊 FTMO Risk Management Results") + table = Table(title="📊 RiskMgmt Risk Management Results") table.add_column("#", justify="right") table.add_column("Strategy", style="cyan") table.add_column("IC", justify="right") @@ -304,11 +304,11 @@ def main(): table.add_column("Trades", justify="right") table.add_column("Monthly %", justify="right") table.add_column("Max DD", justify="right") - table.add_column("FTMO", justify="center") + table.add_column("RiskMgmt", justify="center") results.sort(key=lambda x: x['new_sharpe'], reverse=True) for i, r in enumerate(results, 1): - ftmo = "✅" if r['ftmo_compliant'] else "❌" + riskmgmt = "✅" if r['riskmgmt_compliant'] else "❌" table.add_row( str(i), r['name'], f"{r['new_ic']:.4f}", @@ -316,14 +316,14 @@ def main(): str(r['new_trades']), f"{r['new_monthly_ret']:.2f}%", f"{r['new_max_dd']:.1%}", - ftmo + riskmgmt ) console.print(table) # Summary - ftmo_count = sum(1 for r in results if r['ftmo_compliant']) - console.print(f"\n[bold]FTMO-Compliant:[/bold] {ftmo_count}/{len(results)} strategies") + riskmgmt_count = sum(1 for r in results if r['riskmgmt_compliant']) + console.print(f"\n[bold]RiskMgmt-Compliant:[/bold] {riskmgmt_count}/{len(results)} strategies") if results: best = results[0] @@ -331,7 +331,7 @@ def main(): console.print(f" Sharpe: {best['new_sharpe']:.2f}") console.print(f" Monthly Return: {best['new_monthly_ret']:.2f}%") console.print(f" Max Drawdown: {best['new_max_dd']:.1%}") - console.print(f" FTMO Compliant: {'✅' if best['ftmo_compliant'] else '❌'}") + console.print(f" RiskMgmt Compliant: {'✅' if best['riskmgmt_compliant'] else '❌'}") if __name__ == '__main__': main() diff --git a/scripts/nexquant_continuous_strategies.py b/scripts/nexquant_continuous_strategies.py index f5f68652..9352e068 100644 --- a/scripts/nexquant_continuous_strategies.py +++ b/scripts/nexquant_continuous_strategies.py @@ -68,8 +68,8 @@ def build_ml_model(factor_values: pd.DataFrame, close: pd.Series, style: str) -> signal = pd.Series(np.sign(preds), index=common[split:]) # Backtest - from rdagent.components.backtesting.vbt_backtest import backtest_signal_ftmo - bt = backtest_signal_ftmo( + from rdagent.components.backtesting.vbt_backtest import backtest_signal_risk + bt = backtest_signal_risk( close=close_aligned.loc[common[split:]], signal=signal, txn_cost_bps=2.14, diff --git a/scripts/nexquant_gen_strategies_real_bt.py b/scripts/nexquant_gen_strategies_real_bt.py index c2997a4e..420d72be 100644 --- a/scripts/nexquant_gen_strategies_real_bt.py +++ b/scripts/nexquant_gen_strategies_real_bt.py @@ -9,7 +9,7 @@ Usage: # Swing trading (96-bar forward returns) python nexquant_gen_strategies_real_bt.py 10 - # Daytrading with FTMO constraints (12-bar forward returns) + # Daytrading with RiskMgmt constraints (12-bar forward returns) TRADING_STYLE=daytrading python nexquant_gen_strategies_real_bt.py 5 # With parallel workers (default: CPU count) @@ -66,7 +66,7 @@ if TRADING_STYLE == "daytrading": MAX_DRAWDOWN = -0.10 MIN_MONTHLY_RETURN_PCT = 15.0 STYLE_EMOJI = "🎯 Daytrading" - STYLE_DESC = "short-term intraday with FTMO compliance" + STYLE_DESC = "short-term intraday with RiskMgmt compliance" else: FORWARD_BARS = int(os.getenv("FORWARD_BARS", "96")) MIN_IC = 0.02 @@ -229,7 +229,7 @@ Hard requirements: - Use causal indicators only: rolling windows, shift(1) — NO look-ahead bias - No factor data — compute everything from 'close' - Keep it simple: 2-3 indicators max -- TARGET MONTHLY RETURN: Generate signals that can achieve >15% OOS monthly return after FTMO costs (2.35 pip/trade). Use high-conviction entries only.""" +- TARGET MONTHLY RETURN: Generate signals that can achieve >15% OOS monthly return after RiskMgmt costs (2.35 pip/trade). Use high-conviction entries only.""" elif TRADING_STYLE == "daytrading": system_prompt = f"""You are an expert daytrading quant specializing in EUR/USD scalping and intraday strategies. @@ -258,7 +258,7 @@ Hard requirements: - Use rolling z-scores with windows of 5-20 bars (not 50-100), thresholds ±0.2 to ±0.5 - Combine 2 factors: one momentum, one mean-reversion - NO global mean/std — always use rolling(window).mean() with shift(1) to avoid look-ahead bias -- TARGET MONTHLY RETURN: Generate signals that can achieve >15% OOS monthly return after FTMO costs (2.35 pip/trade). Use high-conviction entries only.""" +- TARGET MONTHLY RETURN: Generate signals that can achieve >15% OOS monthly return after RiskMgmt costs (2.35 pip/trade). Use high-conviction entries only.""" else: system_prompt = f"""You are a quantitative trading expert specializing in EUR/USD daily swing strategies. @@ -289,7 +289,7 @@ Output ONLY valid JSON with these fields: {f'Previous feedback: {feedback}' if feedback else 'First attempt - be creative!'} -Use daily-level signal logic (factor above/below rolling daily mean). Signal changes once per day. TARGET MONTHLY RETURN: Generate signals that can achieve >15% OOS monthly return after FTMO costs (2.35 pip/trade).""" +Use daily-level signal logic (factor above/below rolling daily mean). Signal changes once per day. TARGET MONTHLY RETURN: Generate signals that can achieve >15% OOS monthly return after RiskMgmt costs (2.35 pip/trade).""" api = APIBackend() response = api.build_messages_and_create_chat_completion( @@ -376,8 +376,8 @@ signal.fillna(0).to_pickle('signal.pkl') except Exception as e: return {"status": "failed", "reason": str(e)[:200]} - # Main process: FTMO-realistic backtest (leverage + daily/total loss limits). - from rdagent.components.backtesting.vbt_backtest import backtest_signal_ftmo + # Main process: RiskMgmt-realistic backtest (leverage + daily/total loss limits). + from rdagent.components.backtesting.vbt_backtest import backtest_signal_risk common = close.index.intersection(signal.index) if len(common) < 100: @@ -388,7 +388,7 @@ signal.fillna(0).to_pickle('signal.pkl') fwd_returns = close_a.pct_change(FORWARD_BARS).shift(-FORWARD_BARS) from rdagent.components.backtesting.vbt_backtest import OOS_START_DEFAULT - return backtest_signal_ftmo( + return backtest_signal_risk( close=close_a, signal=signal_a, txn_cost_bps=TXN_COST_BPS, @@ -615,7 +615,7 @@ def main(target_count=10): "n_bars": bt_result.get("n_bars", 0), "n_months": bt_result.get("n_months", 0), "trading_style": TRADING_STYLE, "ohlcv_only": OHLCV_ONLY, - "engine": "ftmo_v2", + "engine": "riskmgmt_v2", "txn_cost_bps": TXN_COST_BPS, # Walk-forward OOS split "oos_sharpe": bt_result.get("oos_sharpe"), diff --git a/scripts/nexquant_gridsearch.py b/scripts/nexquant_gridsearch.py index 3a1d1b52..9a241ba6 100644 --- a/scripts/nexquant_gridsearch.py +++ b/scripts/nexquant_gridsearch.py @@ -1,10 +1,10 @@ #!/usr/bin/env python3 -"""Grid-Search Strategy Generator — no LLM, deterministic, FTMO-verified. +"""Grid-Search Strategy Generator — no LLM, deterministic, RiskMgmt-verified. Core idea: Instead of LLM-generated code, use a fixed signal template and grid-search the parameters. Factors are aligned to daily resolution (where they have actual predictive power), signal is forward-filled to 1-min for -FTMO backtest execution. +RiskMgmt backtest execution. Template: z-score → IC-weighted composite → asymmetric thresholds → signal """ @@ -29,7 +29,7 @@ OHLCV_PATH = Path( ) # ── Target ─────────────────────────────────────────────────────────────────── -MIN_MONTHLY_RETURN_PCT = 1.0 # Raw backtest target (FTMO will reduce ~50%) +MIN_MONTHLY_RETURN_PCT = 1.0 # Raw backtest target (RiskMgmt will reduce ~50%) MIN_SHARPE = 0.5 MAX_DRAWDOWN = -0.30 MIN_WIN_RATE = 0.35 @@ -175,7 +175,7 @@ def evaluate_one(args: tuple) -> dict | None: # Forward-fill to 1-min for backtest signal_1min = daily_signal.reindex(close_1min.index).ffill().fillna(0).astype(int).clip(-1, 1) - # Fast backtest (no FTMO mask, no walk-forward — <1s per eval) + # Fast backtest (no RiskMgmt mask, no walk-forward — <1s per eval) from rdagent.components.backtesting.vbt_backtest import backtest_signal bt = backtest_signal( diff --git a/scripts/nexquant_infinite_search.py b/scripts/nexquant_infinite_search.py index ab18b3a0..82e0df32 100644 --- a/scripts/nexquant_infinite_search.py +++ b/scripts/nexquant_infinite_search.py @@ -13,7 +13,7 @@ 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 +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") @@ -63,7 +63,7 @@ def backtest(signal) -> float: common = close.index.intersection(signal.dropna().index) if len(common) < 100: return -999 - r = backtest_signal_ftmo(close.loc[common], signal.reindex(common).fillna(0), + r = backtest_signal_risk(close.loc[common], signal.reindex(common).fillna(0), txn_cost_bps=TXN_COST_BPS, wf_rolling=False) return r.get("oos_sharpe", -999) diff --git a/scripts/nexquant_ml_pipeline.py b/scripts/nexquant_ml_pipeline.py index f2b1205e..1a02e560 100644 --- a/scripts/nexquant_ml_pipeline.py +++ b/scripts/nexquant_ml_pipeline.py @@ -22,7 +22,7 @@ from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier from sklearn.linear_model import LogisticRegression from sklearn.model_selection import TimeSeriesSplit -from rdagent.components.backtesting.vbt_backtest import backtest_signal_ftmo +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") @@ -98,7 +98,7 @@ def make_target(c: pd.Series, horizon: int = 5) -> np.ndarray: def backtest_metric(c, y_pred, split_idx): test_c = c.iloc[split_idx:] sig = pd.Series(y_pred[split_idx:len(test_c)+split_idx], index=test_c.index[:len(y_pred)-split_idx]) - r = backtest_signal_ftmo(test_c.iloc[:len(sig)], sig.astype(float), txn_cost_bps=TXN_COST_BPS) + r = backtest_signal_risk(test_c.iloc[:len(sig)], sig.astype(float), txn_cost_bps=TXN_COST_BPS) return r.get("oos_sharpe", -999) or -999 @@ -190,7 +190,7 @@ def main(): model.fit(X[:split_idx], y_vals[:split_idx]) y_pred = model.predict(X) sig = pd.Series(y_pred[split_idx:len(c)-split_idx+split_idx], index=c.index[split_idx:split_idx+len(y_pred)-split_idx]) - r = backtest_signal_ftmo(c.iloc[split_idx:split_idx+len(sig)], sig.astype(float), txn_cost_bps=TXN_COST_BPS) + r = backtest_signal_risk(c.iloc[split_idx:split_idx+len(sig)], sig.astype(float), txn_cost_bps=TXN_COST_BPS) oos_s = r.get("oos_sharpe", -999) oos_m = (r.get("oos_monthly_return_pct", 0) or 0) diff --git a/scripts/nexquant_multi_asset.py b/scripts/nexquant_multi_asset.py index 857dfb40..b54ae80d 100644 --- a/scripts/nexquant_multi_asset.py +++ b/scripts/nexquant_multi_asset.py @@ -77,7 +77,7 @@ def main(): print(" Quick Daily Strategy Test on Multi-Asset") print(f"{'='*60}") - from rdagent.components.backtesting.vbt_backtest import backtest_signal_ftmo + from rdagent.components.backtesting.vbt_backtest import backtest_signal_risk for asset in df.columns: c = df[asset].dropna() @@ -91,7 +91,7 @@ def main(): sig[f > s] = 1 sig[f < s] = -1 - r = backtest_signal_ftmo(c, sig.fillna(0), txn_cost_bps=2.14, wf_rolling=True) + r = backtest_signal_risk(c, sig.fillna(0), txn_cost_bps=2.14, wf_rolling=True) oos = r.get("wf_oos_sharpe_mean") or r.get("oos_sharpe", -999) oos_m = r.get("oos_monthly_return_pct", 0) or 0 status = "✅" if oos > 0 else " " @@ -106,7 +106,7 @@ def main(): sig = pd.Series(0.0, index=c.index) sig[f > s] = 1 sig[f < s] = -1 - r = backtest_signal_ftmo(c, sig.fillna(0), txn_cost_bps=2.14, wf_rolling=True) + r = backtest_signal_risk(c, sig.fillna(0), txn_cost_bps=2.14, wf_rolling=True) oos = r.get("wf_oos_sharpe_mean") or r.get("oos_sharpe", -999) print(f" SMA10/30 extended: OOS={oos:+8.2f} Mon={r.get('oos_monthly_return_pct',0):+.2f}%") diff --git a/scripts/nexquant_portfolio.py b/scripts/nexquant_portfolio.py index 41fb1a86..f49a8a2b 100644 --- a/scripts/nexquant_portfolio.py +++ b/scripts/nexquant_portfolio.py @@ -14,7 +14,7 @@ import pandas as pd sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) -from rdagent.components.backtesting.vbt_backtest import backtest_signal_ftmo +from rdagent.components.backtesting.vbt_backtest import backtest_signal_risk DATA = Path("git_ignore_folder/factor_implementation_source_data/multi_asset_daily.h5") @@ -85,7 +85,7 @@ def main(): sig_func = STRATEGIES.get(name, lambda c: rsi_signal(c, 21, 25, 75)) sig = sig_func(c).fillna(0) - r = backtest_signal_ftmo(c, sig, txn_cost_bps=2.14, wf_rolling=True) + r = backtest_signal_risk(c, sig, txn_cost_bps=2.14, wf_rolling=True) oos = r.get("wf_oos_sharpe_mean") or r.get("oos_sharpe", -999) oos_m = r.get("oos_monthly_return_pct", 0) or 0 status = "✅" if oos > 0 else " " diff --git a/scripts/nexquant_portfolio_optimizer.py b/scripts/nexquant_portfolio_optimizer.py index 46d1344d..7ade3b64 100644 --- a/scripts/nexquant_portfolio_optimizer.py +++ b/scripts/nexquant_portfolio_optimizer.py @@ -3,7 +3,7 @@ Given N strategies with daily returns, find the optimal combination that: - Maximizes monthly return -- Keeps max drawdown within FTMO limits (10% total, 5% daily) +- Keeps max drawdown within RiskMgmt limits (10% total, 5% daily) - Diversifies across uncorrelated strategies """ @@ -23,8 +23,8 @@ OHLCV_PATH = Path(os.getenv("PREDIX_OHLCV_PATH", str(PROJECT / "git_ignore_folder" / "intraday_pv_all.h5"))) TARGET_MONTHLY = 15.0 -MAX_DD = 0.10 # FTMO: 10% max total drawdown -MAX_DAILY_DD = 0.05 # FTMO: 5% max daily drawdown +MAX_DD = 0.10 # RiskMgmt: 10% max total drawdown +MAX_DAILY_DD = 0.05 # RiskMgmt: 5% max daily drawdown MIN_TRADES = 30 MIN_SHARPE = 0.5 diff --git a/scripts/nexquant_quick_daytrading.py b/scripts/nexquant_quick_daytrading.py index ef1ba70e..8bec5f38 100644 --- a/scripts/nexquant_quick_daytrading.py +++ b/scripts/nexquant_quick_daytrading.py @@ -24,7 +24,7 @@ FACTOR_FILES = Path('results/factors') VALUE_FILES = FACTOR_FILES / 'values' OHLCV_PATH = Path('git_ignore_folder/factor_implementation_source_data/intraday_pv.h5') -# Best daytrading strategies (12-min horizon, optimized for FTMO) +# Best daytrading strategies (12-min horizon, optimized for RiskMgmt) DAYTRADING_COMBOS = [ { 'name': 'MomentumDivergence12min', @@ -236,7 +236,7 @@ def load_factor_series(name): def main(n_strategies=5): console.print("[bold cyan]🎯 Daytrading Strategy Generator (Quick Mode)[/bold cyan]\n") console.print(" Style: 12-minute forward returns") - console.print(" Target: FTMO compliant (IC>0.02, Sharpe>0.5, Trades>20, DD>-10%)\n") + console.print(" Target: RiskMgmt compliant (IC>0.02, Sharpe>0.5, Trades>20, DD>-10%)\n") # Load OHLCV data if not OHLCV_PATH.exists(): @@ -422,7 +422,7 @@ print(json.dumps(result)) trades = result.get('n_trades', 0) dd = result.get('max_drawdown', 0) - # FTMO criteria + # RiskMgmt criteria if abs(ic) > 0.02 and sharpe > 0.5 and trades > 20 and dd > -0.10: strategy = { 'strategy_name': combo['name'], diff --git a/scripts/nexquant_rebacktest_unified.py b/scripts/nexquant_rebacktest_unified.py index 13af6060..d6deeb79 100644 --- a/scripts/nexquant_rebacktest_unified.py +++ b/scripts/nexquant_rebacktest_unified.py @@ -36,7 +36,7 @@ from rich.console import Console from rich.progress import BarColumn, Progress, SpinnerColumn, TextColumn, TimeElapsedColumn sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) -from rdagent.components.backtesting.vbt_backtest import backtest_signal_ftmo # noqa: E402 +from rdagent.components.backtesting.vbt_backtest import backtest_signal_risk # noqa: E402 OHLCV_PATH = Path("/home/nico/NexQuant/git_ignore_folder/factor_implementation_source_data/intraday_pv.h5") FACTORS_VALUES_DIR = Path("/home/nico/NexQuant/results/factors/values") @@ -184,7 +184,7 @@ def rebacktest_one( # Signal can arrive on either the factor index or the close index. signal = signal.reindex(close_a.index).ffill().fillna(0) - result = backtest_signal_ftmo( + result = backtest_signal_risk( close=close_a, signal=signal, txn_cost_bps=txn_cost_bps, @@ -252,10 +252,10 @@ def main() -> None: "real_n_trades": bt.get("n_trades"), "total_return": bt.get("total_return"), "annualized_return": bt.get("annualized_return"), - "ftmo_daily_loss_hit": bt.get("ftmo_daily_loss_hit"), - "ftmo_total_loss_hit": bt.get("ftmo_total_loss_hit"), + "riskmgmt_daily_loss_hit": bt.get("riskmgmt_daily_loss_hit"), + "riskmgmt_total_loss_hit": bt.get("riskmgmt_total_loss_hit"), "trading_style": data.get("summary", {}).get("trading_style"), - "engine": "ftmo_v2", + "engine": "riskmgmt_v2", "txn_cost_bps": args.txn_cost_bps, # Walk-forward OOS "is_sharpe": bt.get("is_sharpe"), @@ -280,7 +280,7 @@ def main() -> None: data["max_drawdown"] = bt.get("max_drawdown") data["win_rate"] = bt.get("win_rate") data["total_return"] = bt.get("total_return") - data["reevaluation_status"] = "ftmo_v2" + data["reevaluation_status"] = "riskmgmt_v2" try: import json as _json f.write_text(_json.dumps(data, indent=2, ensure_ascii=False)) diff --git a/scripts/nexquant_smart_strategy_gen.py b/scripts/nexquant_smart_strategy_gen.py index 07dec391..d1f37f00 100644 --- a/scripts/nexquant_smart_strategy_gen.py +++ b/scripts/nexquant_smart_strategy_gen.py @@ -1,11 +1,11 @@ #!/usr/bin/env python """ -Smart Strategy Generation with Feedback Loop, Parameter Optimization & FTMO Risk Management. +Smart Strategy Generation with Feedback Loop, Parameter Optimization & RiskMgmt Risk Management. Generates EUR/USD daytrading strategies using LLM with: - Adaptive feedback loop (IC, trades, drawdown-based suggestions) - Grid search for optimal parameters (thresholds, SL/TP, trailing stops) -- Mandatory FTMO-compliant risk management layer +- Mandatory RiskMgmt-compliant risk management layer - Comprehensive evaluation metrics # nosec Usage: @@ -62,11 +62,11 @@ logger = logging.getLogger("SmartStrategyGen") console = Console() # ============================================================================ -# FTMO Risk Management Constants +# RiskMgmt Risk Management Constants # ============================================================================ -class FTMORiskLimits: - """FTMO-compliant risk management constants.""" - MAX_DAILY_LOSS_PCT = 0.05 # 5% max daily loss (FTMO rule) +class RiskMgmtRiskLimits: + """RiskMgmt-compliant risk management constants.""" + MAX_DAILY_LOSS_PCT = 0.05 # 5% max daily loss (RiskMgmt rule) MAX_PER_TRADE_LOSS_PCT = 0.02 # 2% max per trade MAX_TOTAL_DRAWDOWN = 0.10 # 10% max overall drawdown MAX_POSITIONS = 1 # Only 1 position at a time @@ -103,7 +103,7 @@ ACCEPTANCE_CRITERIA = { PARAMETER_GRID = { "threshold_entry": [0.2, 0.3, 0.4, 0.5], "rolling_window": [10, 20, 30, 60], - "stop_loss": [0.01, 0.015, 0.02], # 1%, 1.5%, 2% (HARD MAX: 2% for FTMO) + "stop_loss": [0.01, 0.015, 0.02], # 1%, 1.5%, 2% (HARD MAX: 2% for RiskMgmt) "take_profit": [0.02, 0.03, 0.04, 0.06], # 2x-3x SL "trailing_stop": [0.01, 0.015], # 1%, 1.5% after profit threshold "trailing_activation": [0.015, 0.02], # Activate trail after 1.5%, 2% profit @@ -229,7 +229,7 @@ def setup_llm_env(): # ============================================================================ class RiskManagementEngine: """ - FTMO-compliant risk management layer. + RiskMgmt-compliant risk management layer. Applies stop loss, take profit, trailing stop, and daily loss limits to strategy returns. @@ -262,13 +262,13 @@ class RiskManagementEngine: max_positions : int Maximum concurrent positions (default 1) """ - # Validate FTMO compliance + # Validate RiskMgmt compliance if stop_loss > 0.02: - raise ValueError(f"Stop loss {stop_loss:.2%} exceeds FTMO max of 2%") + raise ValueError(f"Stop loss {stop_loss:.2%} exceeds RiskMgmt max of 2%") if take_profit < stop_loss * 2: raise ValueError(f"Take profit {take_profit:.2%} must be at least 2x SL ({stop_loss*2:.2%})") if max_daily_loss > 0.05: - raise ValueError(f"Daily loss {max_daily_loss:.2%} exceeds FTMO max of 5%") + raise ValueError(f"Daily loss {max_daily_loss:.2%} exceeds RiskMgmt max of 5%") self.stop_loss = stop_loss self.take_profit = take_profit @@ -411,7 +411,7 @@ class RiskManagementEngine: # ============================================================================ class StrategyEvaluator: """ - Comprehensive strategy evaluation with FTMO metrics. # nosec + Comprehensive strategy evaluation with RiskMgmt metrics. # nosec """ def __init__(self, trading_style: str = "daytrading", forward_bars: int = 96): @@ -495,7 +495,7 @@ class StrategyEvaluator: active_returns = strategy_returns[strategy_returns != 0] win_rate = (active_returns > 0).sum() / len(active_returns) if len(active_returns) > 0 else 0.0 - # Daily loss analysis (for FTMO compliance) + # Daily loss analysis (for RiskMgmt compliance) daily_returns = strategy_returns.groupby( strategy_returns.index.date if hasattr(strategy_returns.index[0], "date") else strategy_returns.index, ).sum() @@ -533,9 +533,9 @@ class StrategyEvaluator: "n_bars": total_bars, "n_months": float(n_months), - # FTMO compliance + # RiskMgmt compliance "max_daily_loss": float(max_daily_loss), - "ftmo_compliant": max_daily_loss <= 0.05, + "riskmgmt_compliant": max_daily_loss <= 0.05, # Signal distribution "signal_long_pct": n_long / total_bars if total_bars > 0 else 0, @@ -1117,7 +1117,7 @@ result = {{ "n_short": int((signal_aligned == -1).sum()), "n_neutral": int((signal_aligned == 0).sum()), "max_daily_loss": float(max_daily_loss), - "ftmo_compliant": max_daily_loss <= 0.05, + "riskmgmt_compliant": max_daily_loss <= 0.05, }} def sanitize_val(v): @@ -1595,7 +1595,7 @@ class SmartStrategyGenerator: table.add_column("Trades", justify="right") table.add_column("Max DD", justify="right") table.add_column("Monthly %", justify="right") - table.add_column("FTMO", justify="center") + table.add_column("RiskMgmt", justify="center") for i, s in enumerate(accepted, 1): m = s["metrics"] @@ -1607,7 +1607,7 @@ class SmartStrategyGenerator: str(m.get("n_trades", 0)), f"{m.get('max_drawdown', 0):.1%}", f"{m.get('monthly_return_pct', 0):.2f}%", - "✅" if m.get("ftmo_compliant", False) else "❌", + "✅" if m.get("riskmgmt_compliant", False) else "❌", ) console.print(table) diff --git a/scripts/nexquant_strategy_gen.py b/scripts/nexquant_strategy_gen.py index c911535c..e306276d 100644 --- a/scripts/nexquant_strategy_gen.py +++ b/scripts/nexquant_strategy_gen.py @@ -17,7 +17,7 @@ import pandas as pd sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) -from rdagent.components.backtesting.vbt_backtest import backtest_signal_ftmo +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") @@ -58,7 +58,7 @@ def test_frequency(close: pd.Series, factors: list[dict], freq: str, session_fil sig[~is_sess] = 0 if sig.abs().sum() < 20: continue - r = backtest_signal_ftmo(c, sig.fillna(0), txn_cost_bps=TXN_COST_BPS) + 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: @@ -90,7 +90,7 @@ def test_combo(close: pd.Series, top_signals: list[dict], freq: str, n: int) -> 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) + r = backtest_signal_risk(c, combo.fillna(0), txn_cost_bps=TXN_COST_BPS, wf_rolling=True) return { "frequency": freq, "n_signals": n, diff --git a/scripts/nexquant_systematic.py b/scripts/nexquant_systematic.py new file mode 100644 index 00000000..3f1594bd --- /dev/null +++ b/scripts/nexquant_systematic.py @@ -0,0 +1,300 @@ +#!/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() diff --git a/scripts/nexquant_unified.py b/scripts/nexquant_unified.py new file mode 100644 index 00000000..9b358689 --- /dev/null +++ b/scripts/nexquant_unified.py @@ -0,0 +1,166 @@ +#!/usr/bin/env python +""" +NexQuant Unified Loop — fin_quant + autopilot combined. + +Flow: + 1. fin_quant generates a factor → auto-evaluates + 2. New factor tested in quick strategy (1h/30min SMA combo) + 3. Strategy OOS Sharpe feeds back to LLM for better hypotheses + 4. Factors that produce profitable strategies get priority + 5. Single process, no wasted LLM calls on dead-end factors +""" + +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 + +# ── Config ── +DATA_PATH = Path("git_ignore_folder/factor_implementation_source_data/intraday_pv.h5") +TXN_COST_BPS = 2.14 +MIN_MONTHLY_PCT = 0.1 # Minimum monthly return to keep a strategy + + +def load_daily_close(): + close = pd.read_hdf(DATA_PATH, key="data")["$close"] + if isinstance(close.index, pd.MultiIndex): + close = close.droplevel(-1) + return close.sort_index().dropna() + + +def test_factor_as_signal(factor_path: Path, close: pd.Series, freq: str = "1h") -> dict | None: + """Quick-test a factor as a trading signal. Returns metrics or None if unprofitable.""" + try: + series = pd.read_parquet(factor_path).iloc[:, 0] + if isinstance(series.index, pd.MultiIndex): + series = series.droplevel(-1) + fac = series.resample(freq).last().reindex(close.index).ffill() + except Exception: + return None + + is_sess = (close.index.hour >= 7) & (close.index.hour < 17) + + best_result = None + for direction in [1, -1]: + sig = pd.Series(direction * np.sign(fac).fillna(0), index=close.index) + sig[~is_sess] = 0 + if sig.abs().sum() < 20: + continue + + r = backtest_signal_risk(close, sig.fillna(0), txn_cost_bps=TXN_COST_BPS) + oos_m = r.get("oos_monthly_return_pct", 0) or 0 + + if oos_m > (best_result["monthly"] if best_result else MIN_MONTHLY_PCT): + best_result = { + "direction": direction, + "monthly": oos_m, + "oos_sharpe": r.get("oos_sharpe", -999), + "max_dd": r.get("oos_max_drawdown", 0), + "trades": r.get("oos_n_trades", 0), + } + + return best_result + + +def scan_all_factors(): + """Scan ALL factors and rank them by strategy profitability (not IC).""" + close = load_daily_close().resample("1h").last().dropna() + factors_dir = Path("results/factors") + values_dir = factors_dir / "values" + + results = [] + for i, jf in enumerate(sorted(factors_dir.glob("*.json"))): + try: + meta = json.loads(jf.read_text()) + except Exception: + continue + if meta.get("status") != "success": + continue + + name = meta.get("factor_name", jf.stem) + safe = name.replace("/", "_")[:150] + pf = values_dir / f"{safe}.parquet" + if not pf.exists(): + continue + + bt = test_factor_as_signal(pf, close) + if bt: + results.append({ + "factor": name, + "ic": meta.get("ic", 0), + **bt, + }) + + if i % 100 == 0: + profitable = sum(1 for r in results if r.get("monthly", 0) > 0.5) + print(f" Scanned {i}... {profitable} profitable (>0.5%/mon)") + + results.sort(key=lambda x: x.get("monthly", 0), reverse=True) + return results + + +def main(): + print(f"\n{'='*60}") + print(" NexQuant Unified Loop — Factor-to-Strategy Pipeline") + print(f"{'='*60}") + + print("\n=== PHASE 1: Scan all existing factors as strategies ===\n") + t0 = time.time() + ranked = scan_all_factors() + + profitable = [r for r in ranked if r.get("monthly", 0) > 0.5] + print(f"\n Scanned {len(ranked)} factors in {time.time()-t0:.0f}s") + print(f" Profitable (>0.5%/month): {len(profitable)}") + + if profitable: + print(f"\n TOP 10 by Strategy Profitability:") + for i, r in enumerate(profitable[:10]): + print(f" {i+1:2d}. {r['factor'][:45]:45s} Mon={r['monthly']:+.2f}% IC={r['ic']:+.4f} Dir={r['direction']:+d}") + + # Build combo from top signals + print(f"\n=== PHASE 2: Build best combo ===\n") + c = load_daily_close().resample("1h").last().dropna() + is_sess = (c.index.hour >= 7) & (c.index.hour < 17) + + signals = {} + for r in profitable[:10]: + safe = r["factor"].replace("/", "_")[:150] + pf = Path("results/factors/values") / f"{safe}.parquet" + try: + s = pd.read_parquet(pf).iloc[:, 0] + if isinstance(s.index, pd.MultiIndex): + s = s.droplevel(-1) + fac = s.resample("1h").last().reindex(c.index).ffill() + sig = pd.Series(r["direction"] * np.sign(fac).fillna(0), index=c.index) + sig[~is_sess] = 0 + signals[r["factor"]] = sig + except Exception: + pass + + df = pd.DataFrame(signals, index=c.index).fillna(0) + cols = list(df.columns) + for n in [2, 3, 5, len(cols)]: + combo = df[cols[:n]].mean(axis=1) + r = backtest_signal_risk(c, combo.fillna(0), txn_cost_bps=TXN_COST_BPS, 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) + gap = 10 - m + hit = "🎯" if m >= 4 else "" + print(f" {n:2d} sig: Mon={m:+.2f}% DD={dd:+.1f}% T={t} Gap2_10%={gap:+.1f} {hit}") + + print(f"\n Next: feed top factors back to fin_quant LLM for improved hypotheses") + print(f" Run: python scripts/nexquant_unified.py") + return ranked + + +if __name__ == "__main__": + main() diff --git a/scripts/realistic_backtest_all.py b/scripts/realistic_backtest_all.py index 27ce1495..fcd80e74 100644 --- a/scripts/realistic_backtest_all.py +++ b/scripts/realistic_backtest_all.py @@ -4,11 +4,11 @@ Realistic backtest of all strategies in results/strategies_new/. Costs modeled per trade: 1.5 pip spread + 0.5 pip slippage + 0.35 pip commission = 2.35 pip total -FTMO 100k rules enforced: +RiskMgmt 100k rules enforced: - Max daily loss: 5% of initial balance ($5,000) → no trading rest of day if hit - Max total loss: 10% of initial balance ($10,000) → account blown, simulation ends - Position sizing: 1% equity risk per trade, 10-pip stop (no artificial lot cap) - - Max leverage: 1:30 (EU regulation standard, FTMO default) + - Max leverage: 1:30 (EU regulation standard, RiskMgmt default) - Compounding: position size grows with equity each trade Out-of-sample window: 2024-01-01 onwards (never seen during factor research). @@ -43,9 +43,9 @@ COST_ENTRY = 2.0 * PIP # spread + slippage COST_EXIT = 0.35 * PIP # commission RISK_PCT = 0.015 # 1.5% equity risk per trade STOP = 10 * PIP # 10-pip hard stop -MAX_LEVERAGE = 30 # 1:30 max leverage (FTMO / EU standard) -FTMO_MAX_DAILY = 0.05 # 5% max daily loss of initial balance -FTMO_MAX_TOTAL = 0.10 # 10% max total loss of initial balance +MAX_LEVERAGE = 30 # 1:30 max leverage (RiskMgmt / EU standard) +RiskMgmt_MAX_DAILY = 0.05 # 5% max daily loss of initial balance +RiskMgmt_MAX_TOTAL = 0.10 # 10% max total loss of initial balance OOS_START = "2024-01-01" @@ -111,7 +111,7 @@ def _build_signal(factor_names: list[str], full_idx: pd.Index, def _run_engine(sig_arr: np.ndarray, px_arr: np.ndarray, ts_arr: np.ndarray) -> dict: """ - FTMO-compliant backtest engine. + RiskMgmt-compliant backtest engine. Rules enforced: - Daily loss limit: if daily PnL < -5% of initial ($5k), no new trades that day @@ -165,11 +165,11 @@ def _run_engine(sig_arr: np.ndarray, px_arr: np.ndarray, pos = 0 # Check daily loss limit - if (equity - day_start_eq) / INITIAL < -FTMO_MAX_DAILY: + if (equity - day_start_eq) / INITIAL < -RiskMgmt_MAX_DAILY: day_blocked = True # Check total loss limit → account blown - if equity < INITIAL * (1 - FTMO_MAX_TOTAL): + if equity < INITIAL * (1 - RiskMgmt_MAX_TOTAL): blown = True break @@ -361,22 +361,22 @@ def main() -> None: hits.to_csv(out_hits, index=False) print(f"\nFiltered results saved → {out_hits}") - # ── FTMO projection for #1 ──────────────────────────────────────────────── + # ── RiskMgmt projection for #1 ──────────────────────────────────────────────── best_row = (hits if not hits.empty else df.sort_values("oos_monthly_pct", ascending=False)).iloc[0] mon = best_row["oos_monthly_pct"] dd = abs(best_row["oos_dd_pct"]) gross = 100_000 * mon / 100 challenge_m = 10 / max(mon, 0.01) print(f"\n{'='*70}") - print(f" FTMO 100k projection — #{1}: {best_row['name']}") + print(f" RiskMgmt 100k projection — #{1}: {best_row['name']}") print(f"{'='*70}") print(f" OOS monthly return: {mon:+.2f}%") print(f" Monthly gross profit: ${gross:,.0f}") print(f" Trader share (80%): ${gross*0.8:,.0f} / month") print(f" Trader annual (80%): ${gross*0.8*12:,.0f} / year") - print(f" OOS Max Drawdown: {-dd:.2f}% (FTMO limit: 10%)") + print(f" OOS Max Drawdown: {-dd:.2f}% (RiskMgmt limit: 10%)") print(f" Challenge duration: ~{challenge_m:.1f} months to hit +10%") - print(f" FTMO safe? {'YES ✓' if dd < 8 else 'BORDERLINE ⚠' if dd < 10 else 'NO ✗'}") + print(f" RiskMgmt safe? {'YES ✓' if dd < 8 else 'BORDERLINE ⚠' if dd < 10 else 'NO ✗'}") def _print_table(df: pd.DataFrame) -> None: diff --git a/test/backtesting/test_ftmo_oos.py b/test/backtesting/test_ftmo_oos.py index 4ba87fba..0ec27f95 100644 --- a/test/backtesting/test_ftmo_oos.py +++ b/test/backtesting/test_ftmo_oos.py @@ -1,11 +1,11 @@ """ -Tests for backtest_signal_ftmo and walk-forward OOS validation. +Tests for backtest_signal_risk and walk-forward OOS validation. Covers: -- FTMO daily/total loss limits +- RiskMgmt daily/total loss limits - Risk-based leverage calculation - OOS split returns independent IS and OOS metrics -- OOS uses fresh FTMO simulation (not contaminated by IS losses) +- OOS uses fresh RiskMgmt simulation (not contaminated by IS losses) - Monte Carlo permutation test helper """ from __future__ import annotations @@ -16,11 +16,11 @@ import pytest from rdagent.components.backtesting.vbt_backtest import ( OOS_START_DEFAULT, - _apply_ftmo_mask, - backtest_signal_ftmo, - FTMO_INITIAL_CAPITAL, - FTMO_MAX_DAILY_LOSS, - FTMO_MAX_TOTAL_LOSS, + _apply_risk_mask, + backtest_signal_risk, + INITIAL_CAPITAL, + MAX_DAILY_LOSS, + MAX_TOTAL_LOSS, monte_carlo_trade_pvalue, walk_forward_rolling, ) @@ -31,7 +31,7 @@ from rdagent.components.backtesting.vbt_backtest import ( # --------------------------------------------------------------------------- @pytest.fixture def close_2yr() -> pd.Series: - """~3 months of synthetic 1-min EUR/USD (enough bars for all leverage/FTMO tests).""" + """~3 months of synthetic 1-min EUR/USD (enough bars for all leverage/RiskMgmt tests).""" np.random.seed(42) n = 90 * 1440 # 90 days × 1440 min idx = pd.date_range("2022-01-01", periods=n, freq="1min") @@ -59,27 +59,27 @@ def _random_signal(index: pd.Index, seed: int = 0) -> pd.Series: # --------------------------------------------------------------------------- -# FTMO leverage tests +# RiskMgmt leverage tests # --------------------------------------------------------------------------- -def test_ftmo_result_contains_leverage_fields(close_2yr): +def test_riskmgmt_result_contains_leverage_fields(close_2yr): signal = _random_signal(close_2yr.index) - r = backtest_signal_ftmo(close_2yr, signal, oos_start=None) - assert "ftmo_leverage" in r - assert "ftmo_risk_pct" in r - assert "ftmo_stop_pips" in r - assert r["ftmo_leverage"] > 0 + r = backtest_signal_risk(close_2yr, signal, oos_start=None) + assert "riskmgmt_leverage" in r + assert "riskmgmt_risk_pct" in r + assert "riskmgmt_stop_pips" in r + assert r["riskmgmt_leverage"] > 0 -def test_ftmo_leverage_capped_at_max(close_2yr): +def test_riskmgmt_leverage_capped_at_max(close_2yr): signal = _random_signal(close_2yr.index) # With very tight stop (1 pip) risk_pct=0.5% → leverage would be 55x → capped at 30 - r = backtest_signal_ftmo(close_2yr, signal, stop_pips=1, max_leverage=30, oos_start=None) - assert r["ftmo_leverage"] <= 30.0 + r = backtest_signal_risk(close_2yr, signal, stop_pips=1, max_leverage=30, oos_start=None) + assert r["riskmgmt_leverage"] <= 30.0 -def test_ftmo_zero_signal_produces_no_trades(close_2yr): +def test_riskmgmt_zero_signal_produces_no_trades(close_2yr): signal = pd.Series(0.0, index=close_2yr.index) - r = backtest_signal_ftmo(close_2yr, signal, oos_start=None) + r = backtest_signal_risk(close_2yr, signal, oos_start=None) assert r["n_trades"] == 0 assert r["total_return"] == 0.0 @@ -89,7 +89,7 @@ def test_ftmo_zero_signal_produces_no_trades(close_2yr): # --------------------------------------------------------------------------- def test_oos_split_produces_is_and_oos_keys(close_6yr): signal = _random_signal(close_6yr.index) - r = backtest_signal_ftmo(close_6yr, signal, oos_start="2024-01-01") + r = backtest_signal_risk(close_6yr, signal, oos_start="2024-01-01") assert "is_sharpe" in r assert "oos_sharpe" in r @@ -102,19 +102,19 @@ def test_oos_split_produces_is_and_oos_keys(close_6yr): def test_oos_split_bars_sum_to_total(close_6yr): signal = _random_signal(close_6yr.index) - r = backtest_signal_ftmo(close_6yr, signal, oos_start="2024-01-01") + r = backtest_signal_risk(close_6yr, signal, oos_start="2024-01-01") assert r["is_n_bars"] + r["oos_n_bars"] == len(close_6yr) def test_oos_none_disables_split(close_6yr): signal = _random_signal(close_6yr.index) - r = backtest_signal_ftmo(close_6yr, signal, oos_start=None) + r = backtest_signal_risk(close_6yr, signal, oos_start=None) assert "is_sharpe" not in r assert "oos_sharpe" not in r def test_oos_is_independent_of_is_losses(close_6yr): - """OOS must use a fresh FTMO simulation — IS blowup must not zero OOS trades.""" + """OOS must use a fresh RiskMgmt simulation — IS blowup must not zero OOS trades.""" # Force the IS period to blow up immediately with max short on rising market rising = pd.Series( np.linspace(1.0, 2.0, len(close_6yr)), @@ -122,7 +122,7 @@ def test_oos_is_independent_of_is_losses(close_6yr): ) always_short = pd.Series(-1.0, index=close_6yr.index) - r = backtest_signal_ftmo(rising, always_short, oos_start="2024-01-01") + r = backtest_signal_risk(rising, always_short, oos_start="2024-01-01") # IS should be wiped out (total loss limit hit), but OOS must still trade assert r.get("oos_n_trades", 0) is not None @@ -131,7 +131,7 @@ def test_oos_is_independent_of_is_losses(close_6yr): def test_oos_default_start_matches_constant(close_6yr): signal = _random_signal(close_6yr.index) - r = backtest_signal_ftmo(close_6yr, signal) + r = backtest_signal_risk(close_6yr, signal) assert r.get("oos_start") == OOS_START_DEFAULT @@ -143,7 +143,7 @@ def _monte_carlo_pvalue(close: pd.Series, signal: pd.Series, n_permutations: int Estimate p-value: fraction of random permutations that beat the real Sharpe. p < 0.05 → strategy has statistically significant edge. """ - real_r = backtest_signal_ftmo(close, signal, oos_start=None) + real_r = backtest_signal_risk(close, signal, oos_start=None) real_sharpe = real_r.get("sharpe", 0.0) or 0.0 rng = np.random.default_rng(seed) @@ -152,7 +152,7 @@ def _monte_carlo_pvalue(close: pd.Series, signal: pd.Series, n_permutations: int for _ in range(n_permutations): perm = rng.permutation(signal_vals) perm_signal = pd.Series(perm, index=signal.index) - perm_r = backtest_signal_ftmo(close, perm_signal, oos_start=None) + perm_r = backtest_signal_risk(close, perm_signal, oos_start=None) if (perm_r.get("sharpe") or 0.0) >= real_sharpe: beat += 1 return beat / n_permutations @@ -171,7 +171,7 @@ def test_random_signal_has_no_edge(close_2yr): def test_perfect_signal_is_significant(close_2yr): """An oracle signal on hourly bars should beat random permutations significantly. - Per-minute oracle trading is unprofitable due to FTMO transaction costs, so we + Per-minute oracle trading is unprofitable due to RiskMgmt transaction costs, so we use 60-bar held positions (≈1h) where each directional move is large enough to cover the spread. """ @@ -184,14 +184,14 @@ def test_perfect_signal_is_significant(close_2yr): # --------------------------------------------------------------------------- -# FTMO metrics in result dict +# RiskMgmt metrics in result dict # --------------------------------------------------------------------------- -def test_ftmo_result_has_equity_and_profit(close_2yr): +def test_riskmgmt_result_has_equity_and_profit(close_2yr): signal = _random_signal(close_2yr.index) - r = backtest_signal_ftmo(close_2yr, signal, oos_start=None) - assert "ftmo_end_equity" in r - assert "ftmo_monthly_profit" in r - assert r["ftmo_end_equity"] > 0 + r = backtest_signal_risk(close_2yr, signal, oos_start=None) + assert "riskmgmt_end_equity" in r + assert "riskmgmt_monthly_profit" in r + assert r["riskmgmt_end_equity"] > 0 # --------------------------------------------------------------------------- @@ -199,7 +199,7 @@ def test_ftmo_result_has_equity_and_profit(close_2yr): # --------------------------------------------------------------------------- def test_mc_pvalue_in_result(close_2yr): signal = _random_signal(close_2yr.index) - r = backtest_signal_ftmo(close_2yr, signal, oos_start=None, mc_n_permutations=50) + r = backtest_signal_risk(close_2yr, signal, oos_start=None, mc_n_permutations=50) assert "mc_pvalue" in r assert 0.0 <= r["mc_pvalue"] <= 1.0 assert r["mc_n_permutations"] == 50 @@ -207,7 +207,7 @@ def test_mc_pvalue_in_result(close_2yr): def test_mc_pvalue_disabled_by_default(close_2yr): signal = _random_signal(close_2yr.index) - r = backtest_signal_ftmo(close_2yr, signal, oos_start=None) + r = backtest_signal_risk(close_2yr, signal, oos_start=None) assert "mc_pvalue" not in r @@ -222,32 +222,32 @@ def test_mc_zero_trades_returns_one(close_2yr): # --------------------------------------------------------------------------- def test_wf_rolling_keys_in_result(close_6yr): signal = _random_signal(close_6yr.index) - r = backtest_signal_ftmo(close_6yr, signal, oos_start="2024-01-01", wf_rolling=True) + r = backtest_signal_risk(close_6yr, signal, oos_start="2024-01-01", wf_rolling=True) # With only ~150 days of data, windows may be 0 — just check key presence assert "wf_n_windows" in r def test_wf_rolling_enabled_by_default(close_6yr): signal = _random_signal(close_6yr.index) - r = backtest_signal_ftmo(close_6yr, signal, oos_start="2024-01-01") + r = backtest_signal_risk(close_6yr, signal, oos_start="2024-01-01") assert "wf_n_windows" in r def test_wf_consistency_range(close_6yr): """wf_oos_consistency must be in [0, 1] when windows exist.""" signal = _random_signal(close_6yr.index) - r = backtest_signal_ftmo(close_6yr, signal, oos_start="2024-01-01", wf_rolling=True) + r = backtest_signal_risk(close_6yr, signal, oos_start="2024-01-01", wf_rolling=True) c = r.get("wf_oos_consistency") if c is not None: assert 0.0 <= c <= 1.0 # --------------------------------------------------------------------------- -# Direct _apply_ftmo_mask unit tests +# Direct _apply_risk_mask unit tests # --------------------------------------------------------------------------- class TestApplyFtmoMask: - """Direct unit tests for _apply_ftmo_mask — the core FTMO daily/total loss engine.""" + """Direct unit tests for _apply_risk_mask — the core RiskMgmt daily/total loss engine.""" @pytest.fixture def flat_close(self) -> pd.Series: @@ -257,19 +257,19 @@ class TestApplyFtmoMask: def test_returns_compliance_dict(self, flat_close): signal = _random_signal(flat_close.index) - masked, info = _apply_ftmo_mask(signal, flat_close, leverage=1.0, txn_cost_bps=2.14) - assert "ftmo_daily_breaches" in info - assert "ftmo_total_breached" in info - assert "ftmo_total_breach_ts" in info - assert "ftmo_compliant" in info + masked, info = _apply_risk_mask(signal, flat_close, leverage=1.0, txn_cost_bps=2.14) + assert "riskmgmt_daily_breaches" in info + assert "riskmgmt_total_breached" in info + assert "riskmgmt_total_breach_ts" in info + assert "riskmgmt_compliant" in info def test_flat_market_zero_signal_fully_compliant(self, flat_close): """No trades → always compliant.""" signal = pd.Series(0.0, index=flat_close.index) - masked, info = _apply_ftmo_mask(signal, flat_close, leverage=1.0, txn_cost_bps=2.14) - assert info["ftmo_daily_breaches"] == 0 - assert info["ftmo_total_breached"] is False - assert info["ftmo_compliant"] is True + masked, info = _apply_risk_mask(signal, flat_close, leverage=1.0, txn_cost_bps=2.14) + assert info["riskmgmt_daily_breaches"] == 0 + assert info["riskmgmt_total_breached"] is False + assert info["riskmgmt_compliant"] is True # All signals should remain zero assert (masked == 0).all() @@ -282,8 +282,8 @@ class TestApplyFtmoMask: price.iloc[3:20] = 0.00 # crash from 1.10 to 0.00 → massive loss signal = pd.Series(1.0, index=idx) # always long at 30x leverage - masked, info = _apply_ftmo_mask(signal, price, leverage=30.0, txn_cost_bps=0) - assert info["ftmo_daily_breaches"] > 0 + masked, info = _apply_risk_mask(signal, price, leverage=30.0, txn_cost_bps=0) + assert info["riskmgmt_daily_breaches"] > 0 # After breach, signals on same day must be zeroed breach_day = idx[0].date() same_day_late = (idx[-1] if idx[-1].date() == breach_day else idx[20]) @@ -299,9 +299,9 @@ class TestApplyFtmoMask: price.iloc[5:50] = 0.50 # >10% drop with 30x leverage signal = pd.Series(1.0, index=idx) - masked, info = _apply_ftmo_mask(signal, price, leverage=30.0, txn_cost_bps=0) - assert info["ftmo_total_breached"] is True - assert info["ftmo_total_breach_ts"] is not None + masked, info = _apply_risk_mask(signal, price, leverage=30.0, txn_cost_bps=0) + assert info["riskmgmt_total_breached"] is True + assert info["riskmgmt_total_breach_ts"] is not None # After breach, ALL later signals must be zero assert (masked.iloc[100:] == 0).all() @@ -313,9 +313,9 @@ class TestApplyFtmoMask: price.iloc[5:50] = 0.50 signal = pd.Series(1.0, index=idx) - masked, info = _apply_ftmo_mask(signal, price, leverage=30.0, txn_cost_bps=0) + masked, info = _apply_risk_mask(signal, price, leverage=30.0, txn_cost_bps=0) # All signals after breach index must be zero - breach_ts = pd.Timestamp(info["ftmo_total_breach_ts"]) + breach_ts = pd.Timestamp(info["riskmgmt_total_breach_ts"]) assert (masked.loc[masked.index > breach_ts] == 0).all() def test_daily_loss_resets_on_new_day(self): @@ -327,35 +327,35 @@ class TestApplyFtmoMask: price.iloc[5:20] = 1.09 # ~1% drop with 30x → ~30% loss signal = pd.Series(1.0, index=idx) - masked, info = _apply_ftmo_mask(signal, price, leverage=30.0, txn_cost_bps=0) - assert info["ftmo_daily_breaches"] >= 1 + masked, info = _apply_risk_mask(signal, price, leverage=30.0, txn_cost_bps=0) + assert info["riskmgmt_daily_breaches"] >= 1 # Day 2 signals should be active again if not total-breached day2_mask = idx.date > idx[0].date() - if day2_mask.any() and not info["ftmo_total_breached"]: + if day2_mask.any() and not info["riskmgmt_total_breached"]: day2 = idx[day2_mask][0] assert masked.loc[day2] != 0 def test_compliant_flag_false_after_daily_breach(self): - """Even one daily breach makes ftmo_compliant=False.""" + """Even one daily breach makes riskmgmt_compliant=False.""" n = 3000 idx = pd.date_range("2024-01-01", periods=n, freq="1min") price = pd.Series(1.10, index=idx, dtype=float) price.iloc[3:20] = 0.00 signal = pd.Series(1.0, index=idx) - masked, info = _apply_ftmo_mask(signal, price, leverage=30.0, txn_cost_bps=0) - assert info["ftmo_compliant"] is False + masked, info = _apply_risk_mask(signal, price, leverage=30.0, txn_cost_bps=0) + assert info["riskmgmt_compliant"] is False def test_compliant_flag_false_after_total_breach(self): - """Total breach makes ftmo_compliant=False.""" + """Total breach makes riskmgmt_compliant=False.""" n = 5000 idx = pd.date_range("2024-01-01", periods=n, freq="1min") price = pd.Series(1.10, index=idx, dtype=float) price.iloc[5:50] = 0.50 signal = pd.Series(1.0, index=idx) - masked, info = _apply_ftmo_mask(signal, price, leverage=30.0, txn_cost_bps=0) - assert info["ftmo_compliant"] is False + masked, info = _apply_risk_mask(signal, price, leverage=30.0, txn_cost_bps=0) + assert info["riskmgmt_compliant"] is False def test_transaction_costs_reduce_equity(self): """Transaction costs should reduce equity — compliant scenario with fees.""" @@ -365,9 +365,9 @@ class TestApplyFtmoMask: # Alternating signal → lots of position changes → high costs signal = pd.Series([1.0 if i % 2 == 0 else -1.0 for i in range(n)], index=idx) - masked, info = _apply_ftmo_mask(signal, price, leverage=1.0, txn_cost_bps=10.0) + masked, info = _apply_risk_mask(signal, price, leverage=1.0, txn_cost_bps=10.0) # With high costs and flat market, equity should drop - assert "ftmo_daily_breaches" in info + assert "riskmgmt_daily_breaches" in info def test_output_mask_has_same_index(self): n = 2000 @@ -375,13 +375,13 @@ class TestApplyFtmoMask: price = pd.Series(1.10, index=idx) signal = _random_signal(idx, seed=1) - masked, info = _apply_ftmo_mask(signal, price, leverage=1.0, txn_cost_bps=2.14) + masked, info = _apply_risk_mask(signal, price, leverage=1.0, txn_cost_bps=2.14) assert len(masked) == len(signal) assert masked.index.equals(signal.index) # ============================================================================== -# HYPOTHESIS-BASED PROPERTY TESTS — FTMO OOS Metrics, Drawdown Bounds, +# HYPOTHESIS-BASED PROPERTY TESTS — RiskMgmt OOS Metrics, Drawdown Bounds, # Risk Limit Invariants # ============================================================================== from hypothesis import given, settings, strategies as st @@ -390,13 +390,13 @@ import pandas as pd import math from rdagent.components.backtesting.vbt_backtest import ( - _apply_ftmo_mask, + _apply_risk_mask, _compute_trade_pnl, - backtest_signal_ftmo, - FTMO_INITIAL_CAPITAL, - FTMO_MAX_DAILY_LOSS, - FTMO_MAX_TOTAL_LOSS, - FTMO_MAX_LEVERAGE, + backtest_signal_risk, + INITIAL_CAPITAL, + MAX_DAILY_LOSS, + MAX_TOTAL_LOSS, + MAX_LEVERAGE, DEFAULT_TXN_COST_BPS, monte_carlo_trade_pvalue, walk_forward_rolling, @@ -443,7 +443,7 @@ def _make_signal_series( class TestLeverageBounds: - """Property: leverage stays within [0.05, FTMO_MAX_LEVERAGE] for all valid inputs.""" + """Property: leverage stays within [0.05, MAX_LEVERAGE] for all valid inputs.""" @given( risk_pct=st.floats(min_value=0.0001, max_value=0.10), @@ -475,19 +475,19 @@ class TestLeverageBounds: # --------------------------------------------------------------------------- -# Property 2: FTMO Result Dict Shape +# Property 2: RiskMgmt Result Dict Shape # --------------------------------------------------------------------------- class TestFtmoResultDictShape: - """Property: backtest_signal_ftmo returns a consistent dict shape.""" + """Property: backtest_signal_risk returns a consistent dict shape.""" REQUIRED_KEYS = { "status", "sharpe", "max_drawdown", "total_return", "win_rate", "n_trades", "n_bars", "txn_cost_bps", "bars_per_year", - "ftmo_leverage", "ftmo_risk_pct", "ftmo_stop_pips", - "ftmo_daily_breaches", "ftmo_total_breached", "ftmo_compliant", - "ftmo_end_equity", "ftmo_monthly_profit", + "riskmgmt_leverage", "riskmgmt_risk_pct", "riskmgmt_stop_pips", + "riskmgmt_daily_breaches", "riskmgmt_total_breached", "riskmgmt_compliant", + "riskmgmt_end_equity", "riskmgmt_monthly_profit", } @given( @@ -502,7 +502,7 @@ class TestFtmoResultDictShape: """Property: result dict contains all required top-level keys regardless of inputs.""" close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, txn_cost_bps=cost_bps, oos_start=None) + r = backtest_signal_risk(close, signal, txn_cost_bps=cost_bps, oos_start=None) missing = self.REQUIRED_KEYS - set(r.keys()) assert not missing, f"Missing keys: {missing}" @@ -516,7 +516,7 @@ class TestFtmoResultDictShape: """Property: status is 'success' for any valid input.""" close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) assert r["status"] == "success" @@ -541,10 +541,10 @@ class TestSignalSymmetry: close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r1 = backtest_signal_ftmo(close, signal, oos_start=None) - r2 = backtest_signal_ftmo(close, -signal, oos_start=None) + r1 = backtest_signal_risk(close, signal, oos_start=None) + r2 = backtest_signal_risk(close, -signal, oos_start=None) - # Negated signal → total_return should differ (FTMO masking may make both negative) + # Negated signal → total_return should differ (RiskMgmt masking may make both negative) if r1["n_trades"] > 0 and r2["n_trades"] > 0: assert np.isfinite(r1["total_return"]) assert np.isfinite(r2["total_return"]) @@ -562,18 +562,18 @@ class TestSignalSymmetry: close = _make_price_series(n_bars, drift, vol) signal = pd.Series(0.0, index=close.index) - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) assert r["n_trades"] == 0 assert r["total_return"] == 0.0 # --------------------------------------------------------------------------- -# Property 4: FTMO Compliance Invariants +# Property 4: RiskMgmt Compliance Invariants # --------------------------------------------------------------------------- class TestFtmoComplianceInvariants: - """Property: compliance invariants of _apply_ftmo_mask.""" + """Property: compliance invariants of _apply_risk_mask.""" @given( n_bars=st.integers(min_value=100, max_value=3000), @@ -583,14 +583,14 @@ class TestFtmoComplianceInvariants: ) @settings(max_examples=50, deadline=10000) def test_zero_signal_always_compliant(self, n_bars, leverage, cost_bps, seed): - """Property: zero signal → ftmo_compliant=True, daily_breaches=0, total_breached=False.""" + """Property: zero signal → riskmgmt_compliant=True, daily_breaches=0, total_breached=False.""" np.random.seed(seed) price = _make_price_series(n_bars, 0, 0.0001) signal = pd.Series(0.0, index=price.index) - masked, info = _apply_ftmo_mask(signal, price, leverage, cost_bps) - assert info["ftmo_compliant"] is True - assert info["ftmo_daily_breaches"] == 0 - assert info["ftmo_total_breached"] is False + masked, info = _apply_risk_mask(signal, price, leverage, cost_bps) + assert info["riskmgmt_compliant"] is True + assert info["riskmgmt_daily_breaches"] == 0 + assert info["riskmgmt_total_breached"] is False @given( n_bars=st.integers(min_value=100, max_value=3000), @@ -604,7 +604,7 @@ class TestFtmoComplianceInvariants: np.random.seed(seed) price = _make_price_series(n_bars, 0, 0.0001) signal = _make_signal_series(price.index, "ternary") - masked, info = _apply_ftmo_mask(signal, price, leverage, cost_bps) + masked, info = _apply_risk_mask(signal, price, leverage, cost_bps) assert len(masked) == len(signal) assert masked.index.equals(signal.index) # Every element of masked is either 0 or the original signal value @@ -622,7 +622,7 @@ class TestFtmoComplianceInvariants: np.random.seed(seed) price = _make_price_series(n_bars, 0, 0.0001) signal = _make_signal_series(price.index, "continuous") - masked, info = _apply_ftmo_mask(signal, price, leverage, cost_bps) + masked, info = _apply_risk_mask(signal, price, leverage, cost_bps) assert (masked.abs() <= signal.abs()).all() @given( @@ -636,8 +636,8 @@ class TestFtmoComplianceInvariants: idx = pd.date_range("2024-01-01", periods=n_bars, freq="1min") price = pd.Series(1.10, index=idx) signal = _make_signal_series(price.index, "ternary") - _masked, info = _apply_ftmo_mask(signal, price, leverage, 0.0) - assert info["ftmo_total_breached"] is False + _masked, info = _apply_risk_mask(signal, price, leverage, 0.0) + assert info["riskmgmt_total_breached"] is False @given( n_bars=st.integers(min_value=100, max_value=2000), @@ -645,14 +645,14 @@ class TestFtmoComplianceInvariants: ) @settings(max_examples=50, deadline=10000) def test_total_breach_implies_noncompliant(self, n_bars, leverage): - """Property: total_breached=True => ftmo_compliant=False.""" + """Property: total_breached=True => riskmgmt_compliant=False.""" idx = pd.date_range("2024-01-01", periods=n_bars, freq="1min") price = pd.Series(1.10, index=idx) price.iloc[3:50] = 0.50 # Crash to trigger total breach signal = pd.Series(1.0, index=price.index) - masked, info = _apply_ftmo_mask(signal, price, leverage, 0.0) - if info["ftmo_total_breached"]: - assert info["ftmo_compliant"] is False + masked, info = _apply_risk_mask(signal, price, leverage, 0.0) + if info["riskmgmt_total_breached"]: + assert info["riskmgmt_compliant"] is False @given( n_bars=st.integers(min_value=500, max_value=3000), @@ -660,14 +660,14 @@ class TestFtmoComplianceInvariants: ) @settings(max_examples=50, deadline=10000) def test_daily_breach_implies_noncompliant(self, n_bars, leverage): - """Property: daily_breaches > 0 => ftmo_compliant=False.""" + """Property: daily_breaches > 0 => riskmgmt_compliant=False.""" idx = pd.date_range("2024-01-01", periods=n_bars, freq="1min") price = pd.Series(1.10, index=idx) price.iloc[3:20] = 0.00 signal = pd.Series(1.0, index=price.index) - masked, info = _apply_ftmo_mask(signal, price, leverage, 0.0) - if info["ftmo_daily_breaches"] > 0: - assert info["ftmo_compliant"] is False + masked, info = _apply_risk_mask(signal, price, leverage, 0.0) + if info["riskmgmt_daily_breaches"] > 0: + assert info["riskmgmt_compliant"] is False @given( n_bars=st.integers(min_value=100, max_value=3000), @@ -677,13 +677,13 @@ class TestFtmoComplianceInvariants: ) @settings(max_examples=50, deadline=10000) def test_compliant_scenario_has_no_mask_changes(self, n_bars, leverage, cost_bps, seed): - """Property: if ftmo_compliant=True, masked signals equal original signals.""" + """Property: if riskmgmt_compliant=True, masked signals equal original signals.""" np.random.seed(seed) idx = pd.date_range("2024-01-01", periods=n_bars, freq="1min") price = _make_price_series(n_bars, 0.0, 0.00001) signal = _make_signal_series(price.index, "ternary") - masked, info = _apply_ftmo_mask(signal, price, leverage, cost_bps) - if info["ftmo_compliant"]: + masked, info = _apply_risk_mask(signal, price, leverage, cost_bps) + if info["riskmgmt_compliant"]: # In compliant scenarios with very low vol, masked should equal signal pass # This is trivially true since compliance means no breaches @@ -709,10 +709,10 @@ class TestCostMonotonicity: close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r_lo = backtest_signal_ftmo(close, signal, txn_cost_bps=1.0, oos_start=None) - r_hi = backtest_signal_ftmo(close, signal, txn_cost_bps=10.0, oos_start=None) + r_lo = backtest_signal_risk(close, signal, txn_cost_bps=1.0, oos_start=None) + r_hi = backtest_signal_risk(close, signal, txn_cost_bps=10.0, oos_start=None) - # Higher costs should not improve total return (allowing for FTMO mask differences) + # Higher costs should not improve total return (allowing for RiskMgmt mask differences) assert np.isfinite(r_hi["total_return"]) assert np.isfinite(r_lo["total_return"]) @@ -729,8 +729,8 @@ class TestCostMonotonicity: close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r_lo = backtest_signal_ftmo(close, signal, txn_cost_bps=1.0, oos_start=None) - r_hi = backtest_signal_ftmo(close, signal, txn_cost_bps=10.0, oos_start=None) + r_lo = backtest_signal_risk(close, signal, txn_cost_bps=1.0, oos_start=None) + r_hi = backtest_signal_risk(close, signal, txn_cost_bps=10.0, oos_start=None) # Higher costs should not improve annualized return assert np.isfinite(r_hi["annualized_return"]) @@ -757,7 +757,7 @@ class TestDrawdownBounds: np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) dd = r["max_drawdown"] assert -1.0 <= dd <= 0.0 @@ -773,7 +773,7 @@ class TestDrawdownBounds: np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) # total_return >= -1 (can't lose more than everything) assert r["total_return"] >= -1.0 @@ -798,7 +798,7 @@ class TestPositionBounds: np.random.seed(seed) price = _make_price_series(n_bars, 0.0, 0.0001) signal = _make_signal_series(price.index, "continuous") - masked, info = _apply_ftmo_mask(signal, price, leverage, cost_bps) + masked, info = _apply_risk_mask(signal, price, leverage, cost_bps) # Position = masked * leverage, should be in [-leverage, leverage] positions = masked * leverage assert (positions >= -leverage).all() @@ -825,7 +825,7 @@ class TestTradeCounting: np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) assert r["n_trades"] <= r["n_position_changes"] @given( @@ -840,7 +840,7 @@ class TestTradeCounting: np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) assert r["signal_long"] + r["signal_short"] + r["signal_neutral"] == r["n_bars"] @given( @@ -855,19 +855,19 @@ class TestTradeCounting: np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = pd.Series(0.0, index=close.index) - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) assert r["n_trades"] == 0 assert r["win_rate"] == 0.0 assert r["profit_factor"] == 0.0 # --------------------------------------------------------------------------- -# Property 9: FTMO Equity Invariants +# Property 9: RiskMgmt Equity Invariants # --------------------------------------------------------------------------- class TestFtmoEquityInvariants: - """Property: ftmo_end_equity and ftmo_monthly_profit invariants.""" + """Property: riskmgmt_end_equity and riskmgmt_monthly_profit invariants.""" @given( n_bars=st.integers(min_value=200, max_value=1500), @@ -877,13 +877,13 @@ class TestFtmoEquityInvariants: ) @settings(max_examples=50, deadline=10000) def test_end_equity_formula(self, n_bars, drift, vol, seed): - """Property: ftmo_end_equity = FTMO_INITIAL_CAPITAL * (1 + total_return).""" + """Property: riskmgmt_end_equity = INITIAL_CAPITAL * (1 + total_return).""" np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, oos_start=None) - expected_equity = FTMO_INITIAL_CAPITAL * (1 + r["total_return"]) - assert abs(r["ftmo_end_equity"] - expected_equity) < 1.0 + r = backtest_signal_risk(close, signal, oos_start=None) + expected_equity = INITIAL_CAPITAL * (1 + r["total_return"]) + assert abs(r["riskmgmt_end_equity"] - expected_equity) < 1.0 @given( n_bars=st.integers(min_value=200, max_value=1500), @@ -893,12 +893,12 @@ class TestFtmoEquityInvariants: ) @settings(max_examples=50, deadline=10000) def test_end_equity_positive(self, n_bars, drift, vol, seed): - """Property: ftmo_end_equity > 0 always (can't lose more than initial).""" + """Property: riskmgmt_end_equity > 0 always (can't lose more than initial).""" np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, oos_start=None) - assert r["ftmo_end_equity"] > 0 + r = backtest_signal_risk(close, signal, oos_start=None) + assert r["riskmgmt_end_equity"] > 0 @given( n_bars=st.integers(min_value=200, max_value=1500), @@ -908,13 +908,13 @@ class TestFtmoEquityInvariants: ) @settings(max_examples=50, deadline=10000) def test_monthly_profit_sign_matches_monthly_return(self, n_bars, drift, vol, seed): - """Property: sign(ftmo_monthly_profit) = sign(monthly_return).""" + """Property: sign(riskmgmt_monthly_profit) = sign(monthly_return).""" np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) if r["monthly_return"] != 0: - assert np.sign(r["ftmo_monthly_profit"]) == np.sign(r["monthly_return"]) + assert np.sign(r["riskmgmt_monthly_profit"]) == np.sign(r["monthly_return"]) # --------------------------------------------------------------------------- @@ -989,48 +989,48 @@ class TestMonteCarloPValue: # --------------------------------------------------------------------------- -# Property 11: FTMO Loss Limit Invariants +# Property 11: RiskMgmt Loss Limit Invariants # --------------------------------------------------------------------------- class TestFtmoLossLimitInvariants: - """Property: FTMO constants satisfy fundamental ordering.""" + """Property: RiskMgmt constants satisfy fundamental ordering.""" def test_daily_loss_less_than_total_loss(self): - """Property: FTMO_MAX_DAILY_LOSS < FTMO_MAX_TOTAL_LOSS.""" - assert FTMO_MAX_DAILY_LOSS < FTMO_MAX_TOTAL_LOSS + """Property: MAX_DAILY_LOSS < MAX_TOTAL_LOSS.""" + assert MAX_DAILY_LOSS < MAX_TOTAL_LOSS def test_initial_capital_is_100k(self): - """Property: FTMO_INITIAL_CAPITAL = 100_000.""" - assert FTMO_INITIAL_CAPITAL == 100_000.0 + """Property: INITIAL_CAPITAL = 100_000.""" + assert INITIAL_CAPITAL == 100_000.0 def test_max_daily_loss_is_5_percent(self): - """Property: FTMO_MAX_DAILY_LOSS = 0.05 (5%).""" - assert FTMO_MAX_DAILY_LOSS == 0.05 + """Property: MAX_DAILY_LOSS = 0.05 (5%).""" + assert MAX_DAILY_LOSS == 0.05 def test_max_total_loss_is_10_percent(self): - """Property: FTMO_MAX_TOTAL_LOSS = 0.10 (10%).""" - assert FTMO_MAX_TOTAL_LOSS == 0.10 + """Property: MAX_TOTAL_LOSS = 0.10 (10%).""" + assert MAX_TOTAL_LOSS == 0.10 def test_leverage_default_is_30(self): - """Property: FTMO_MAX_LEVERAGE = 30.""" - assert FTMO_MAX_LEVERAGE == 30 + """Property: MAX_LEVERAGE = 30.""" + assert MAX_LEVERAGE == 30 @given( n_bars=st.integers(min_value=100, max_value=2000), - leverage=st.floats(min_value=0.1, max_value=FTMO_MAX_LEVERAGE), + leverage=st.floats(min_value=0.1, max_value=MAX_LEVERAGE), seed=st.integers(min_value=0, max_value=100), ) @settings(max_examples=50, deadline=10000) - def test_total_loss_never_exceeds_ftmo_limit(self, n_bars, leverage, seed): - """Property: _apply_ftmo_mask detects total breach at exactly the FTMO threshold.""" + def test_total_loss_never_exceeds_riskmgmt_limit(self, n_bars, leverage, seed): + """Property: _apply_risk_mask detects total breach at exactly the RiskMgmt threshold.""" np.random.seed(seed) idx = pd.date_range("2024-01-01", periods=n_bars, freq="1min") price = _make_price_series(n_bars, 0.0, 0.00001) signal = _make_signal_series(price.index, "ternary") - _masked, info = _apply_ftmo_mask(signal, price, leverage, 0.0) - assert isinstance(info["ftmo_total_breached"], bool) - assert isinstance(info["ftmo_compliant"], bool) + _masked, info = _apply_risk_mask(signal, price, leverage, 0.0) + assert isinstance(info["riskmgmt_total_breached"], bool) + assert isinstance(info["riskmgmt_compliant"], bool) # --------------------------------------------------------------------------- @@ -1039,7 +1039,7 @@ class TestFtmoLossLimitInvariants: class TestOosIndependence: - """Property: OOS metrics are computed from fresh FTMO simulation.""" + """Property: OOS metrics are computed from fresh RiskMgmt simulation.""" @given( n_bars=st.integers(min_value=300, max_value=2000), @@ -1053,7 +1053,7 @@ class TestOosIndependence: np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) # Without OOS, all bars are in the main result assert "is_n_bars" not in r or r.get("is_n_bars", 0) == 0 assert "oos_n_bars" not in r or r.get("oos_n_bars", 0) == 0 @@ -1073,7 +1073,7 @@ class TestOosIndependence: # Use a date in the middle of the range mid = close.index[len(close) // 2] oos_start_str = mid.strftime("%Y-%m-%d") - r = backtest_signal_ftmo(close, signal, oos_start=oos_start_str) + r = backtest_signal_risk(close, signal, oos_start=oos_start_str) assert r.get("oos_start") == oos_start_str @given( @@ -1090,7 +1090,7 @@ class TestOosIndependence: signal = _make_signal_series(close.index, "ternary") mid = close.index[len(close) // 2] oos_start_str = mid.strftime("%Y-%m-%d") - r = backtest_signal_ftmo(close, signal, oos_start=oos_start_str, wf_rolling=True) + r = backtest_signal_risk(close, signal, oos_start=oos_start_str, wf_rolling=True) c = r.get("wf_oos_consistency") if c is not None: assert 0.0 <= c <= 1.0 @@ -1116,7 +1116,7 @@ class TestSharpeSortinoConsistency: np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) if r["total_return"] > 0: # Sortino is typically >= Sharpe for profitable strategies pass # Not strictly guaranteed but a good sanity check @@ -1133,7 +1133,7 @@ class TestSharpeSortinoConsistency: np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) assert np.isfinite(r["sharpe"]) assert np.isfinite(r["sortino"]) @@ -1213,10 +1213,10 @@ class TestLeverageRiskInvariants: close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r_lo = backtest_signal_ftmo(close, signal, stop_pips=5, oos_start=None) - r_hi = backtest_signal_ftmo(close, signal, stop_pips=20, oos_start=None) + r_lo = backtest_signal_risk(close, signal, stop_pips=5, oos_start=None) + r_hi = backtest_signal_risk(close, signal, stop_pips=20, oos_start=None) - assert r_hi["ftmo_leverage"] <= r_lo["ftmo_leverage"] + assert r_hi["riskmgmt_leverage"] <= r_lo["riskmgmt_leverage"] # --------------------------------------------------------------------------- @@ -1239,7 +1239,7 @@ class TestWalkForwardProperties: np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, wf_rolling=True, oos_start=None) + r = backtest_signal_risk(close, signal, wf_rolling=True, oos_start=None) assert isinstance(r.get("wf_n_windows", 0), int) assert r.get("wf_n_windows", 0) >= 0 @@ -1255,7 +1255,7 @@ class TestWalkForwardProperties: np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, wf_rolling=True, oos_start=None) + r = backtest_signal_risk(close, signal, wf_rolling=True, oos_start=None) assert "wf_n_windows" in r def test_walk_forward_non_datetime_index(self): @@ -1272,7 +1272,7 @@ class TestWalkForwardProperties: class TestSignalClipping: - """Property: backtest_signal_ftmo clips signals to [-1, 1].""" + """Property: backtest_signal_risk clips signals to [-1, 1].""" @given( n_bars=st.integers(min_value=200, max_value=1000), @@ -1285,7 +1285,7 @@ class TestSignalClipping: np.random.seed(seed) close = _make_price_series(n_bars, 0, 0.0001) signal = _make_signal_series(close.index, "continuous") * signal_scale - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) assert r["status"] == "success" @given( @@ -1302,7 +1302,7 @@ class TestSignalClipping: n_nan = int(n_bars * nan_frac) if n_nan > 0: signal.iloc[:n_nan] = np.nan - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) assert r["status"] == "success" @@ -1326,7 +1326,7 @@ class TestMetricRangeInvariants: np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) assert 0.0 <= r["win_rate"] <= 1.0 @given( @@ -1341,7 +1341,7 @@ class TestMetricRangeInvariants: np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) assert r["profit_factor"] >= 0.0 @given( @@ -1356,7 +1356,7 @@ class TestMetricRangeInvariants: np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) assert r["volatility"] >= 0.0 @given( @@ -1371,7 +1371,7 @@ class TestMetricRangeInvariants: np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) assert r["n_trades"] >= 0 @given( @@ -1386,7 +1386,7 @@ class TestMetricRangeInvariants: np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) assert r["n_months"] > 0.0 @@ -1403,14 +1403,14 @@ class TestDeterminism: seed=st.integers(min_value=0, max_value=50), ) @settings(max_examples=50, deadline=10000) - def test_backtest_signal_ftmo_deterministic(self, n_bars, seed): - """Property: calling backtest_signal_ftmo twice with same inputs gives same results.""" + def test_backtest_signal_risk_deterministic(self, n_bars, seed): + """Property: calling backtest_signal_risk twice with same inputs gives same results.""" np.random.seed(seed) close = _make_price_series(n_bars, 0, 0.0001) signal = _make_signal_series(close.index, "ternary") - r1 = backtest_signal_ftmo(close.copy(), signal.copy(), oos_start=None) - r2 = backtest_signal_ftmo(close.copy(), signal.copy(), oos_start=None) + r1 = backtest_signal_risk(close.copy(), signal.copy(), oos_start=None) + r2 = backtest_signal_risk(close.copy(), signal.copy(), oos_start=None) for key in r1: if key in r2: @@ -1421,14 +1421,14 @@ class TestDeterminism: seed=st.integers(min_value=0, max_value=50), ) @settings(max_examples=50, deadline=10000) - def test_apply_ftmo_mask_deterministic(self, n_bars, seed): - """Property: _apply_ftmo_mask is deterministic.""" + def test_apply_risk_mask_deterministic(self, n_bars, seed): + """Property: _apply_risk_mask is deterministic.""" np.random.seed(seed) price = _make_price_series(n_bars, 0, 0.0001) signal = _make_signal_series(price.index, "ternary") - m1, i1 = _apply_ftmo_mask(signal.copy(), price.copy(), leverage=10.0, txn_cost_bps=2.14) - m2, i2 = _apply_ftmo_mask(signal.copy(), price.copy(), leverage=10.0, txn_cost_bps=2.14) + m1, i1 = _apply_risk_mask(signal.copy(), price.copy(), leverage=10.0, txn_cost_bps=2.14) + m2, i2 = _apply_risk_mask(signal.copy(), price.copy(), leverage=10.0, txn_cost_bps=2.14) assert m1.equals(m2) assert i1 == i2 @@ -1456,14 +1456,14 @@ class TestCostSymmetry: # All-long signal long_signal = pd.Series(1.0, index=close.index) - r_long = backtest_signal_ftmo(close, long_signal, txn_cost_bps=2.14, oos_start=None) + r_long = backtest_signal_risk(close, long_signal, txn_cost_bps=2.14, oos_start=None) # All-short signal short_signal = pd.Series(-1.0, index=close.index) - r_short = backtest_signal_ftmo(close, short_signal, txn_cost_bps=2.14, oos_start=None) + r_short = backtest_signal_risk(close, short_signal, txn_cost_bps=2.14, oos_start=None) # With drift near zero, returns should be roughly opposite - # Position change counts may differ due to FTMO masks + # Position change counts may differ due to RiskMgmt masks assert r_long["n_position_changes"] >= 0 assert r_short["n_position_changes"] >= 0 @@ -1488,7 +1488,7 @@ class TestCalmarRatio: np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) assert np.isfinite(r["calmar"]) @@ -1510,7 +1510,7 @@ class TestICProperties: np.random.seed(seed) close = _make_price_series(n_bars, 0, 0.0001) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) assert r["ic"] is None @given( @@ -1524,7 +1524,7 @@ class TestICProperties: close = _make_price_series(n_bars, 0, 0.0001) signal = _make_signal_series(close.index, "ternary") fwd = close.pct_change().shift(-1).fillna(0) - r = backtest_signal_ftmo(close, signal, forward_returns=fwd, oos_start=None) + r = backtest_signal_risk(close, signal, forward_returns=fwd, oos_start=None) if r["ic"] is not None: assert -1.0 <= r["ic"] <= 1.0 @@ -1550,10 +1550,10 @@ class TestExtremeMarketHandling: price = pd.Series(1.10, index=idx, dtype=float) price.iloc[n_bars // 4 : n_bars // 4 + 5] = 1.10 * (1 - crash_magnitude) signal = pd.Series(1.0, index=price.index) - r = backtest_signal_ftmo(price, signal, oos_start=None) + r = backtest_signal_risk(price, signal, oos_start=None) assert r["status"] == "success" # After a large crash, total_breached is expected - assert isinstance(r.get("ftmo_total_breached", False), bool) + assert isinstance(r.get("riskmgmt_total_breached", False), bool) # --------------------------------------------------------------------------- @@ -1571,7 +1571,7 @@ class TestDailyBreachCounting: ) @settings(max_examples=50, deadline=10000) def test_daily_breach_count_never_exceeds_ndays(self, n_days, leverage, seed): - """Property: ftmo_daily_breaches never exceeds number of trading days.""" + """Property: riskmgmt_daily_breaches never exceeds number of trading days.""" np.random.seed(seed) n_bars = n_days * 1440 idx = pd.date_range("2024-01-01", periods=n_bars, freq="1min") @@ -1581,8 +1581,8 @@ class TestDailyBreachCounting: start = d * 1440 + 3 price.iloc[start : start + 20] = 0.50 signal = pd.Series(1.0, index=price.index) - _masked, info = _apply_ftmo_mask(signal, price, leverage, 0.0) - assert info["ftmo_daily_breaches"] <= n_days + _masked, info = _apply_risk_mask(signal, price, leverage, 0.0) + assert info["riskmgmt_daily_breaches"] <= n_days # --------------------------------------------------------------------------- @@ -1613,7 +1613,7 @@ class TestNumericPrecision: np.random.seed(seed) close = _make_price_series(n_bars, drift, vol) signal = _make_signal_series(close.index, "ternary") - r = backtest_signal_ftmo(close, signal, oos_start=None) + r = backtest_signal_risk(close, signal, oos_start=None) for k in self.NUMERIC_KEYS: if k in r: val = r[k] diff --git a/test/integration/test_full_pipeline.py b/test/integration/test_full_pipeline.py index 538cc28f..5fc4f5ba 100644 --- a/test/integration/test_full_pipeline.py +++ b/test/integration/test_full_pipeline.py @@ -6,7 +6,7 @@ Tests the complete end-to-end pipeline including: - Portfolio Optimization (P7) - Full Pipeline End-to-End - Parallelization -- FTMO Compliance +- RiskMgmt Compliance At least 20 integration tests covering all new features. @@ -526,15 +526,15 @@ class TestParallelization: # --------------------------------------------------------------------------- -# Tests: FTMO Compliance +# Tests: RiskMgmt Compliance # --------------------------------------------------------------------------- -class TestFTMOCompliance: - """Test FTMO compliance checks for accepted strategies.""" +class TestRiskMgmtCompliance: + """Test RiskMgmt compliance checks for accepted strategies.""" def test_stop_loss_compliance(self, mock_strategies, mock_project_structure): - """Test that all strategies have max drawdown within FTMO limits.""" + """Test that all strategies have max drawdown within RiskMgmt limits.""" strategies_dir = mock_project_structure / "results" / "strategies_new" for json_file in strategies_dir.glob("*.json"): @@ -542,7 +542,7 @@ class TestFTMOCompliance: data = json.load(f) max_dd = abs(data.get("max_drawdown", 0)) - # FTMO max drawdown limit: 10% + # RiskMgmt max drawdown limit: 10% assert max_dd <= 0.25 or data.get("max_drawdown", 0) < 0 def test_daily_loss_compliance(self, mock_strategies, mock_project_structure): @@ -554,25 +554,25 @@ class TestFTMOCompliance: data = json.load(f) daily_loss = abs(data.get("daily_loss_max", 0)) - # FTMO daily loss limit: 5% + # RiskMgmt daily loss limit: 5% assert daily_loss <= 0.05 or data.get("daily_loss_max", 0) == 0 def test_portfolio_max_drawdown(self, mock_strategies, portfolio_optimizer): - """Test that optimized portfolio respects FTMO drawdown limits.""" + """Test that optimized portfolio respects RiskMgmt drawdown limits.""" opt_result = portfolio_optimizer.optimize_portfolio(method="mean_variance") if opt_result and "weights" in opt_result: bt_result = portfolio_optimizer.backtest_portfolio(opt_result["weights"]) if bt_result: - # FTMO max drawdown: 10% + # RiskMgmt max drawdown: 10% # Portfolio should stay within limits max_dd = abs(bt_result.get("max_drawdown", 0)) # Note: This is a soft check as mock data may vary assert max_dd < 0.50 # Generous threshold for mock data - def test_ftmo_compliance_report(self, mock_strategies, portfolio_optimizer): - """Test generation of FTMO compliance report.""" + def test_riskmgmt_compliance_report(self, mock_strategies, portfolio_optimizer): + """Test generation of RiskMgmt compliance report.""" strategies = portfolio_optimizer._load_strategy_data() if not strategies: @@ -922,12 +922,12 @@ class TestSharpeRatioProperties: # --------------------------------------------------------------------------- -# Property 5: FTMO Drawdown Limits +# Property 5: RiskMgmt Drawdown Limits # --------------------------------------------------------------------------- -class TestFTMODrawdownLimits: - """Property: FTMO drawdown invariants.""" +class TestRiskMgmtDrawdownLimits: + """Property: RiskMgmt drawdown invariants.""" @given( equity_gain=st.floats(min_value=-0.15, max_value=0.50), @@ -948,17 +948,17 @@ class TestFTMODrawdownLimits: @settings(max_examples=50, deadline=10000) def test_daily_loss_at_5_percent(self, daily_returns): """Property: daily P&L breach triggers at −5%.""" - ftmo_daily_max = 0.05 + riskmgmt_daily_max = 0.05 daily_pnl = np.prod(1 + np.array(daily_returns)) - 1 - breached = daily_pnl < -ftmo_daily_max + breached = daily_pnl < -riskmgmt_daily_max assert isinstance(breached, (bool, np.bool_)) @given( total_return=st.floats(min_value=-0.15, max_value=0.50), ) @settings(max_examples=50, deadline=10000) - def test_ftmo_end_equity_formula(self, total_return): - """Property: ftmo_end_equity = initial_capital * (1 + total_return).""" + def test_riskmgmt_end_equity_formula(self, total_return): + """Property: riskmgmt_end_equity = initial_capital * (1 + total_return).""" initial = 100_000.0 end_equity = initial * (1 + total_return) assert end_equity > 0 # Can't go below zero diff --git a/test/local/test_continuous_strategies.py b/test/local/test_continuous_strategies.py index 04103b00..a855bb23 100644 --- a/test/local/test_continuous_strategies.py +++ b/test/local/test_continuous_strategies.py @@ -51,7 +51,7 @@ class TestBuildMLModel: result = build_ml_model(factor_data.iloc[:100], close_data.iloc[:100], "swing") assert result is None - @patch("rdagent.components.backtesting.vbt_backtest.backtest_signal_ftmo") + @patch("rdagent.components.backtesting.vbt_backtest.backtest_signal_risk") def test_sufficient_data_returns_dict(self, mock_bt, factor_data, close_data): mock_bt.return_value = { "sharpe": 1.5, "max_drawdown": -0.1, "win_rate": 0.55, @@ -65,7 +65,7 @@ class TestBuildMLModel: assert result["status"] == "accepted" assert result["type"] == "ml_model" - @patch("rdagent.components.backtesting.vbt_backtest.backtest_signal_ftmo") + @patch("rdagent.components.backtesting.vbt_backtest.backtest_signal_risk") def test_negative_oos_rejected(self, mock_bt, factor_data, close_data): mock_bt.return_value = { "sharpe": 1.5, "max_drawdown": -0.1, "win_rate": 0.55, diff --git a/test/local/test_optuna_optimizer.py b/test/local/test_optuna_optimizer.py index f514c3ad..f960dade 100644 --- a/test/local/test_optuna_optimizer.py +++ b/test/local/test_optuna_optimizer.py @@ -7,7 +7,7 @@ Tests cover: - Parameter space definition and validation - Parameter suggestion mechanisms - Objective function calculation -- FTMO penalty logic +- RiskMgmt penalty logic - Optuna study creation and configuration - Parameter injection into strategy code - Optimization run (mocked, small trial count) @@ -37,11 +37,11 @@ except ImportError: from rdagent.scenarios.qlib.local.optuna_optimizer import ( OptunaOptimizer, PARAMETER_SPACE, - FTMO_MAX_STOP_LOSS, - FTMO_MAX_DRAWDOWN, - FTMO_MAX_DAILY_LOSS, + RiskMgmt_MAX_STOP_LOSS, + RiskMgmt_MAX_DRAWDOWN, + MAX_DAILY_LOSS, PENALTY_MAX_DD, - PENALTY_FTMO_VIOLATION, + PENALTY_RiskMgmt_VIOLATION, OPTUNA_AVAILABLE, ) @@ -205,10 +205,10 @@ class TestParameterSpaceDefinition: assert config['choices'] == [5, 10, 15, 20] def test_parameter_space_stop_loss_config(self): - """Test stop_loss parameter configuration (FTMO compliant).""" + """Test stop_loss parameter configuration (RiskMgmt compliant).""" config = PARAMETER_SPACE['stop_loss'] assert config['type'] == 'categorical' - assert all(c <= FTMO_MAX_STOP_LOSS for c in config['choices']) + assert all(c <= RiskMgmt_MAX_STOP_LOSS for c in config['choices']) def test_parameter_space_take_profit_config(self): """Test take_profit parameter configuration.""" @@ -222,16 +222,16 @@ class TestParameterSpaceDefinition: assert config['type'] == 'categorical' assert config['choices'] == [0.01, 0.015] - def test_ftmo_constants_correct(self): - """Test FTMO compliance constants.""" - assert FTMO_MAX_STOP_LOSS == 0.02 - assert FTMO_MAX_DRAWDOWN == -0.10 - assert FTMO_MAX_DAILY_LOSS == 0.05 + def test_riskmgmt_constants_correct(self): + """Test RiskMgmt compliance constants.""" + assert RiskMgmt_MAX_STOP_LOSS == 0.02 + assert RiskMgmt_MAX_DRAWDOWN == -0.10 + assert MAX_DAILY_LOSS == 0.05 def test_penalty_constants_correct(self): """Test penalty weight constants.""" assert PENALTY_MAX_DD == -10.0 - assert PENALTY_FTMO_VIOLATION == -50.0 + assert PENALTY_RiskMgmt_VIOLATION == -50.0 # ============================================================================= @@ -420,15 +420,15 @@ class TestObjectiveFunction: # ============================================================================= -# FTMO Penalty Tests +# RiskMgmt Penalty Tests # ============================================================================= @pytest.mark.skipif(not OPTUNA_AVAILABLE, reason="Optuna not installed") -class TestFTMOPenalties: - """Test FTMO compliance penalties.""" +class TestRiskMgmtPenalties: + """Test RiskMgmt compliance penalties.""" def test_penalty_max_drawdown_violation(self, optimizer): - """Test penalty when max drawdown exceeds FTMO limit.""" + """Test penalty when max drawdown exceeds RiskMgmt limit.""" study = optuna.create_study(sampler=optuna.samplers.TPESampler(seed=42)) with patch.object(optimizer, '_run_backtest_with_params') as mock_bt: @@ -437,7 +437,7 @@ class TestFTMOPenalties: 'sharpe_ratio': 1.5, 'ic': 0.08, 'total_trades': 25, - 'max_drawdown': -0.12, # Below FTMO_MAX_DRAWDOWN (-0.10) + 'max_drawdown': -0.12, # Below RiskMgmt_MAX_DRAWDOWN (-0.10) } trial = study.ask() @@ -449,10 +449,10 @@ class TestFTMOPenalties: assert history['penalty'] <= PENALTY_MAX_DD def test_penalty_stop_loss_violation(self, optimizer): - """Test penalty when stop loss exceeds FTMO maximum.""" + """Test penalty when stop loss exceeds RiskMgmt maximum.""" study = optuna.create_study(sampler=optuna.samplers.TPESampler(seed=42)) - # Create a custom parameter space that allows FTMO-violating values + # Create a custom parameter space that allows RiskMgmt-violating values violating_space = { **PARAMETER_SPACE, 'stop_loss': {'type': 'categorical', 'choices': [0.01, 0.025, 0.03]}, @@ -475,13 +475,13 @@ class TestFTMOPenalties: value = optimizer.objective(trial) history = optimizer._optimization_history[-1] - assert history['penalty'] <= PENALTY_FTMO_VIOLATION + assert history['penalty'] <= PENALTY_RiskMgmt_VIOLATION # Restore original space optimizer.parameter_space = optimizer.param_space_original def test_no_penalty_compliant_strategy(self, optimizer): - """Test no penalty for FTMO-compliant strategy.""" + """Test no penalty for RiskMgmt-compliant strategy.""" study = optuna.create_study(sampler=optuna.samplers.TPESampler(seed=42)) with patch.object(optimizer, '_run_backtest_with_params') as mock_bt: @@ -490,7 +490,7 @@ class TestFTMOPenalties: 'sharpe_ratio': 1.5, 'ic': 0.08, 'total_trades': 25, - 'max_drawdown': -0.05, # Within FTMO limit + 'max_drawdown': -0.05, # Within RiskMgmt limit } trial = study.ask() @@ -517,7 +517,7 @@ class TestFTMOPenalties: 'sharpe_ratio': 1.5, 'ic': 0.08, 'total_trades': 25, - 'max_drawdown': -0.12, # FTMO violation + 'max_drawdown': -0.12, # RiskMgmt violation } trial = study.ask() @@ -526,7 +526,7 @@ class TestFTMOPenalties: history = optimizer._optimization_history[-1] # Both penalties should apply - expected_penalty = PENALTY_MAX_DD + PENALTY_FTMO_VIOLATION + expected_penalty = PENALTY_MAX_DD + PENALTY_RiskMgmt_VIOLATION assert history['penalty'] == expected_penalty diff --git a/test/local/test_strategy_worker.py b/test/local/test_strategy_worker.py index 22601c04..0a202421 100644 --- a/test/local/test_strategy_worker.py +++ b/test/local/test_strategy_worker.py @@ -452,9 +452,9 @@ class TestAcceptanceGate: assert gate.min_sharpe == 0.5 assert gate.min_trades == 10 assert gate.max_drawdown == -0.15 - assert gate.ftmo_max_sl == 0.02 - assert gate.ftmo_max_daily_loss == 0.05 - assert gate.ftmo_max_dd == 0.10 + assert gate.riskmgmt_max_sl == 0.02 + assert gate.riskmgmt_max_daily_loss == 0.05 + assert gate.riskmgmt_max_dd == 0.10 def test_evaluate_passing_strategy(self, acceptance_gate): """Test evaluation of passing strategy.""" @@ -474,8 +474,8 @@ class TestAcceptanceGate: assert evaluation['checks']['sharpe']['passed'] is True assert evaluation['checks']['trades']['passed'] is True assert evaluation['checks']['max_drawdown']['passed'] is True - assert evaluation['checks']['ftmo_sl']['passed'] is True - assert evaluation['checks']['ftmo_max_dd']['passed'] is True + assert evaluation['checks']['riskmgmt_sl']['passed'] is True + assert evaluation['checks']['riskmgmt_max_dd']['passed'] is True def test_evaluate_failing_ic(self, acceptance_gate): """Test failure due to low IC.""" @@ -540,10 +540,10 @@ class TestAcceptanceGate: assert evaluation['passed'] is False assert any('DD' in r or 'drawdown' in r.lower() for r in evaluation['reasons']) assert evaluation['checks']['max_drawdown']['passed'] is False - assert evaluation['checks']['ftmo_max_dd']['passed'] is False + assert evaluation['checks']['riskmgmt_max_dd']['passed'] is False - def test_evaluate_failing_ftmo_sl(self, acceptance_gate): - """Test FTMO stop loss violation.""" + def test_evaluate_failing_riskmgmt_sl(self, acceptance_gate): + """Test RiskMgmt stop loss violation.""" result = { 'ic': 0.05, 'sharpe_ratio': 1.2, @@ -555,7 +555,7 @@ class TestAcceptanceGate: evaluation = acceptance_gate.evaluate(result) assert evaluation['passed'] is False - assert evaluation['checks']['ftmo_sl']['passed'] is False + assert evaluation['checks']['riskmgmt_sl']['passed'] is False def test_evaluate_ic_none(self, acceptance_gate): """Test when IC is None.""" diff --git a/test/qlib/test_headform.py b/test/qlib/test_headform.py index a27e71e6..1e1a213a 100644 --- a/test/qlib/test_headform.py +++ b/test/qlib/test_headform.py @@ -193,9 +193,9 @@ class TestRegressionFixedBugs: def test_oos_default_enabled(self): """Feature: OOS/WF is now default.""" - from rdagent.components.backtesting.vbt_backtest import backtest_signal_ftmo + from rdagent.components.backtesting.vbt_backtest import backtest_signal_risk import inspect - source = inspect.signature(backtest_signal_ftmo) + source = inspect.signature(backtest_signal_risk) assert source.parameters["wf_rolling"].default is True @@ -205,15 +205,15 @@ class TestRegressionFixedBugs: class TestCrossSystemConsistency: - def test_backtest_signal_ftmo_consistency(self): - from rdagent.components.backtesting.vbt_backtest import backtest_signal, backtest_signal_ftmo + def test_backtest_signal_risk_consistency(self): + from rdagent.components.backtesting.vbt_backtest import backtest_signal, backtest_signal_risk n = 2000 dates = pd.date_range("2024-01-01", periods=n, freq="1min") rng = np.random.default_rng(42) close = pd.Series(1.10 * np.exp(np.cumsum(rng.normal(0, 0.0002, n))), index=dates) signal = pd.Series(np.where(rng.normal(0, 1, n) > 0, 1.0, -1.0), index=dates) r1 = backtest_signal(close, signal, txn_cost_bps=2.14) - r2 = backtest_signal_ftmo(close, signal, txn_cost_bps=2.14, wf_rolling=False) + r2 = backtest_signal_risk(close, signal, txn_cost_bps=2.14, wf_rolling=False) if r1["status"] == "success" and r2.get("status") == "success": assert "sharpe" in r1 and "sharpe" in r2 assert -1.0 <= r1["max_drawdown"] <= 0.0 diff --git a/test/qlib/test_headform4.py b/test/qlib/test_headform4.py index 4e64190c..2bd73732 100644 --- a/test/qlib/test_headform4.py +++ b/test/qlib/test_headform4.py @@ -66,17 +66,17 @@ class TestLiveTraderMock: def test_script_imports(self): import importlib.util spec = importlib.util.spec_from_file_location( - "ftmo_live_trader", - PROJECT_ROOT / "git_ignore_folder/live_trading/ftmo_live_trader.py", + "riskmgmt_live_trader", + PROJECT_ROOT / "git_ignore_folder/live_trading/riskmgmt_live_trader.py", ) assert spec is not None def test_script_has_required_sections(self): - content = (PROJECT_ROOT / "git_ignore_folder/live_trading/ftmo_live_trader.py").read_text() + content = (PROJECT_ROOT / "git_ignore_folder/live_trading/riskmgmt_live_trader.py").read_text() assert "RISK_PCT" in content assert "STOP_PIPS" in content assert "TP_PIPS" in content - assert "FTMO_DAILY_LIMIT" in content + assert "RiskMgmt_DAILY_LIMIT" in content class TestFactorValuesIntegration: diff --git a/test/qlib/test_open_source_suite.py b/test/qlib/test_open_source_suite.py index d7d6a48e..6368fb40 100644 --- a/test/qlib/test_open_source_suite.py +++ b/test/qlib/test_open_source_suite.py @@ -175,22 +175,22 @@ class TestPromptLoader: load_prompt("xyz_nonexistent") -class TestApplyFTMOMask: +class TestApplyRiskMgmtMask: def test_output_same_length(self): - from rdagent.components.backtesting.vbt_backtest import _apply_ftmo_mask + from rdagent.components.backtesting.vbt_backtest import _apply_risk_mask dates = pd.date_range("2024-01-01", periods=100, freq="1min") close = pd.Series(1.10, index=dates) signal = pd.Series(np.where(np.arange(100) % 2 == 0, 1.0, -1.0), index=dates) - masked, metrics = _apply_ftmo_mask(signal, close, leverage=1.0, txn_cost_bps=2.14) + masked, metrics = _apply_risk_mask(signal, close, leverage=1.0, txn_cost_bps=2.14) assert len(masked) == len(signal) assert isinstance(metrics, dict) def test_flat_signal(self): - from rdagent.components.backtesting.vbt_backtest import _apply_ftmo_mask + from rdagent.components.backtesting.vbt_backtest import _apply_risk_mask dates = pd.date_range("2024-01-01", periods=200, freq="1min") close = pd.Series(1.10, index=dates) signal = pd.Series(0.0, index=dates) - masked, metrics = _apply_ftmo_mask(signal, close, leverage=1.0, txn_cost_bps=2.14) + masked, metrics = _apply_risk_mask(signal, close, leverage=1.0, txn_cost_bps=2.14) assert isinstance(metrics, dict)