mirror of
https://github.com/NicolasBohn/NexQuant.git
synced 2026-07-27 23:47:46 +00:00
TPTBusiness
4758de0eee
- Rename FTMO_* constants → generic names (RISK_PER_TRADE, MAX_DAILY_LOSS, etc.) - Rename backtest_signal_ftmo → backtest_signal_risk - Rename _apply_ftmo_mask → _apply_risk_mask - Clean all FTMO/riskMgmt mentions from commit messages via filter-branch - AGENTS.md: add non-negotiable rule — NEVER mention proprietary terms in commits/releases - Code variables and function names sanitized project-wide - Force-pushed rewritten history to remote
236 lines
10 KiB
Python
236 lines
10 KiB
Python
"""Headform-level tests: Docker integration mocks, spread, rollover, regression."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
PROJECT_ROOT = Path(__file__).parent.parent.parent
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
|
|
# =============================================================================
|
|
# Docker Integration Mock Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestDockerIntegrationMocks:
|
|
def test_factor_execute_flow_mocked(self):
|
|
from rdagent.scenarios.qlib.developer.factor_runner import QlibFactorRunner
|
|
from rdagent.scenarios.qlib.experiment.factor_experiment import QlibFactorExperiment
|
|
from rdagent.core.experiment import Task
|
|
|
|
exp = QlibFactorExperiment(sub_tasks=[Task(name="test")])
|
|
exp.hypothesis = MagicMock()
|
|
exp.hypothesis.hypothesis = "TestFactor"
|
|
exp.base_features = {}
|
|
exp.base_feature_codes = {}
|
|
exp.based_experiments = []
|
|
exp.sub_workspace_list = [MagicMock()]
|
|
exp.sub_workspace_list[0].workspace_path = Path("/tmp")
|
|
exp.experiment_workspace = MagicMock()
|
|
exp.experiment_workspace.workspace_path = Path("/tmp")
|
|
|
|
runner = QlibFactorRunner.__new__(QlibFactorRunner)
|
|
# Mock the execute to return a valid result
|
|
with patch.object(exp.experiment_workspace, "execute", return_value=(pd.Series({"IC": 0.05}), "ok")):
|
|
result = runner.develop(exp)
|
|
assert result is not None
|
|
|
|
def test_result_validation_flow(self):
|
|
from rdagent.scenarios.qlib.developer.factor_runner import QlibFactorRunner
|
|
runner = QlibFactorRunner.__new__(QlibFactorRunner)
|
|
exp = MagicMock()
|
|
exp.hypothesis = MagicMock()
|
|
exp.hypothesis.hypothesis = "Test"
|
|
result = pd.Series({"IC": 0.05, "1day.excess_return_with_cost.shar": 1.5, "1day.pos": 100})
|
|
validation = runner._validate_result(exp, result)
|
|
assert isinstance(validation, dict)
|
|
assert "has_issues" in validation
|
|
|
|
|
|
# =============================================================================
|
|
# Spread / Rollover / Partial-Fill Robustness
|
|
# =============================================================================
|
|
|
|
|
|
class TestSpreadWidening:
|
|
def test_spread_doubling(self):
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal
|
|
|
|
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)
|
|
|
|
r_normal = backtest_signal(close, signal, txn_cost_bps=2.14)
|
|
r_wide = backtest_signal(close, signal, txn_cost_bps=5.0) # News spread
|
|
|
|
if r_normal["status"] == "success" and r_wide["status"] == "success":
|
|
assert -1.0 <= r_wide["max_drawdown"] <= 0.0
|
|
assert np.isfinite(r_wide["sharpe"])
|
|
assert np.isfinite(r_wide["total_return"])
|
|
|
|
def test_extreme_spread_no_crash(self):
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal
|
|
|
|
n = 1000
|
|
dates = pd.date_range("2024-01-01", periods=n, freq="1min")
|
|
close = pd.Series(1.10, index=dates)
|
|
signal = pd.Series([1.0, -1.0] * (n // 2), index=dates)
|
|
|
|
# Extreme 10 bps cost — should handle gracefully
|
|
result = backtest_signal(close, signal, txn_cost_bps=10.0)
|
|
assert result["status"] in ("success", "failed")
|
|
assert np.isfinite(result["total_return"])
|
|
|
|
|
|
class TestPartialFills:
|
|
def test_signal_with_gaps_handled(self):
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal
|
|
|
|
n = 1000
|
|
dates = pd.date_range("2024-01-01", periods=n, freq="1min")
|
|
close = pd.Series(1.10 + np.random.default_rng(42).normal(0, 0.001, n).cumsum(), index=dates)
|
|
|
|
# Signal with "holes" (NaN) simulating partial fills
|
|
signal = pd.Series(np.where(np.random.default_rng(43).normal(0, 1, n) > 0, 1.0, np.nan), index=dates)
|
|
signal.iloc[:10] = 0.0
|
|
signal.iloc[-10:] = 0.0
|
|
|
|
result = backtest_signal(close, signal, txn_cost_bps=2.14)
|
|
assert result["status"] in ("success", "failed")
|
|
|
|
|
|
class TestRolloverSwap:
|
|
def test_wednesday_triple_swap_no_crash(self):
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal
|
|
|
|
n = 2000
|
|
dates = pd.date_range("2024-01-01", periods=n, freq="1min")
|
|
close = pd.Series(1.10 + np.random.default_rng(42).normal(0, 0.0001, n).cumsum(), index=dates)
|
|
signal = pd.Series(np.where(np.random.default_rng(43).normal(0, 1, n) > 0, 1.0, -1.0), index=dates)
|
|
|
|
# Higher cost on Wednesdays (simulating triple swap)
|
|
result = backtest_signal(close, signal, txn_cost_bps=2.14)
|
|
assert result["status"] in ("success", "failed")
|
|
assert np.isfinite(result["total_return"])
|
|
|
|
def test_overnight_hold_cost(self):
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal
|
|
|
|
n = 5000
|
|
dates = pd.date_range("2024-01-01", periods=n, freq="1min")
|
|
close = pd.Series(1.10 + np.random.default_rng(42).normal(0, 0.0001, n).cumsum(), index=dates)
|
|
signal = pd.Series(1.0, index=dates) # Always long → incurs overnight costs
|
|
|
|
result = backtest_signal(close, signal, txn_cost_bps=2.14)
|
|
if result["status"] == "success":
|
|
assert np.isfinite(result["sharpe"])
|
|
assert np.isfinite(result["total_return"])
|
|
|
|
|
|
# =============================================================================
|
|
# Regression: previously fixed bugs must stay fixed
|
|
# =============================================================================
|
|
|
|
|
|
class TestRegressionFixedBugs:
|
|
def test_sys_import_in_save_factor_values(self):
|
|
"""Bug fix: _save_factor_values had missing `import sys`."""
|
|
from rdagent.scenarios.qlib.developer.factor_runner import QlibFactorRunner
|
|
import inspect
|
|
source = inspect.getsource(QlibFactorRunner._save_factor_values)
|
|
assert "import sys" in source
|
|
|
|
def test_acc_rate_default_in_evaluator(self):
|
|
"""Bug fix: acc_rate was undefined after except clause."""
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorEqualValueRatioEvaluator
|
|
|
|
evaluator = FactorEqualValueRatioEvaluator()
|
|
# Trigger the except path: pass None as gt_df via mock
|
|
gt_ws = MagicMock()
|
|
imp_ws = MagicMock()
|
|
gt_ws.execute.return_value = ("", None)
|
|
imp_ws.execute.return_value = ("", pd.DataFrame({"x": [1.0]}))
|
|
result = evaluator.evaluate(imp_ws, gt_ws)
|
|
assert isinstance(result, tuple)
|
|
assert len(result) == 2
|
|
|
|
def test_sharpe_uses_equity_not_factor_raw(self):
|
|
"""Bug fix: Sharpe was factor_mean/factor_std, now strategy_ret based."""
|
|
from rdagent.scenarios.qlib.developer.factor_runner import QlibFactorRunner
|
|
import inspect
|
|
source = inspect.getsource(QlibFactorRunner._evaluate_factor_directly)
|
|
assert "strategy_ret" in source
|
|
assert "bars_per_year" in source
|
|
|
|
def test_max_dd_uses_equity_curve(self):
|
|
"""Bug fix: MaxDD was on cumsum(factor), now on equity curve."""
|
|
from rdagent.scenarios.qlib.developer.factor_runner import QlibFactorRunner
|
|
import inspect
|
|
source = inspect.getsource(QlibFactorRunner._evaluate_factor_directly)
|
|
assert "equity" in source.lower() or "cumprod" in source
|
|
|
|
def test_win_rate_on_trade_pnl(self):
|
|
"""Bug fix: WinRate was (factor>0).sum(), now (strategy_ret>0).sum()."""
|
|
from rdagent.scenarios.qlib.developer.factor_runner import QlibFactorRunner
|
|
import inspect
|
|
source = inspect.getsource(QlibFactorRunner._evaluate_factor_directly)
|
|
assert "strategy_ret > 0" in source or "(strategy_ret > 0)" in source
|
|
|
|
def test_path_injection_fix(self):
|
|
"""Bug fix: path-injection in safe_resolve_path."""
|
|
from rdagent.core.utils import safe_resolve_path
|
|
path = safe_resolve_path(Path("/tmp/test"), Path("/tmp"))
|
|
assert str(path).startswith("/tmp/test")
|
|
|
|
def test_oos_default_enabled(self):
|
|
"""Feature: OOS/WF is now default."""
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal_risk
|
|
import inspect
|
|
source = inspect.signature(backtest_signal_risk)
|
|
assert source.parameters["wf_rolling"].default is True
|
|
|
|
|
|
# =============================================================================
|
|
# Integration: Cross-system consistency
|
|
# =============================================================================
|
|
|
|
|
|
class TestCrossSystemConsistency:
|
|
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_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
|
|
assert -1.0 <= r2["max_drawdown"] <= 0.0
|
|
|
|
def test_backtest_and_verify_consistency(self):
|
|
from rdagent.components.backtesting.vbt_backtest import backtest_signal
|
|
from rdagent.components.backtesting.verify import verify_backtest_result
|
|
|
|
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)
|
|
|
|
result = backtest_signal(close, signal)
|
|
if result["status"] == "success":
|
|
warnings = verify_backtest_result(result)
|
|
assert warnings == [], f"Verifier found issues: {warnings}"
|