mirror of
https://github.com/NicolasBohn/NexQuant.git
synced 2026-07-27 15:37:44 +00:00
TPTBusiness
0d9b0916f2
- RiskMgmt OOS: 88 tests (leverage bounds, DD limits, trade counting, MC p-value, daily breach) - Kronos adapter: 73 tests (OHLCV idempotence, batch/sequential equivalence, forward-fill) - Auto-fixer: 78 tests (fix idempotence, MultiIndex conversion, fuzzing random patterns) - Factor coder: 65 tests (FactorTask roundtrip, evaluator invariants, workspace paths) - QLib pipeline: 61 tests (Metrics, bandit, precision matrices, noise_var) - Integration: 69 tests (portfolio weights, correlation, RiskMgmt limits, JSON roundtrip)
1359 lines
49 KiB
Python
1359 lines
49 KiB
Python
"""
|
||
Integration Tests for Full NexQuant Pipeline (P6-P9)
|
||
|
||
Tests the complete end-to-end pipeline including:
|
||
- Feedback Loop Integration (P6)
|
||
- Portfolio Optimization (P7)
|
||
- Full Pipeline End-to-End
|
||
- Parallelization
|
||
- FTMO Compliance
|
||
|
||
At least 20 integration tests covering all new features.
|
||
|
||
Usage:
|
||
pytest test/integration/test_full_pipeline.py -v
|
||
pytest test/integration/test_full_pipeline.py -k "portfolio" -v
|
||
pytest test/integration/test_full_pipeline.py -m "slow" -v
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import tempfile
|
||
import time
|
||
from pathlib import Path
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
import pytest
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fixtures
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture
|
||
def mock_project_structure(tmp_path: Path) -> Path:
|
||
"""Create a complete mock project structure for integration tests."""
|
||
# Create directories
|
||
dirs = [
|
||
"results/factors",
|
||
"results/strategies_new",
|
||
"results/models",
|
||
"results/portfolios",
|
||
"prompts/local",
|
||
"rdagent/scenarios/qlib/local",
|
||
]
|
||
for d in dirs:
|
||
(tmp_path / d).mkdir(parents=True)
|
||
|
||
return tmp_path
|
||
|
||
|
||
@pytest.fixture
|
||
def mock_factors(mock_project_structure: Path) -> list:
|
||
"""Create mock factor files with varying quality."""
|
||
factors = []
|
||
factors_dir = mock_project_structure / "results" / "factors"
|
||
|
||
for i in range(20):
|
||
factor = {
|
||
"name": f"factor_{i}",
|
||
"status": "success",
|
||
"ic": 0.01 + i * 0.01, # IC from 0.01 to 0.20
|
||
"sharpe_ratio": 0.5 + i * 0.1,
|
||
"max_drawdown": -0.30 + i * 0.01,
|
||
"win_rate": 0.45 + i * 0.005,
|
||
"code": f"def factor_{i}(): return signal",
|
||
}
|
||
filepath = factors_dir / f"factor_{i}.json"
|
||
with open(filepath, "w") as f:
|
||
json.dump(factor, f)
|
||
factors.append(factor)
|
||
|
||
return factors
|
||
|
||
|
||
@pytest.fixture
|
||
def mock_strategies(mock_project_structure: Path) -> list:
|
||
"""Create mock strategy files with backtest data."""
|
||
strategies = []
|
||
strategies_dir = mock_project_structure / "results" / "strategies_new"
|
||
|
||
np.random.seed(42)
|
||
|
||
strategy_configs = [
|
||
{"name": "MomentumScalper", "sharpe": 2.1, "ic": 0.15, "max_dd": -0.10, "daily_loss": -0.015},
|
||
{"name": "MeanReversionAlpha", "sharpe": 1.8, "ic": 0.12, "max_dd": -0.15, "daily_loss": -0.018},
|
||
{"name": "VolatilityBreakout", "sharpe": 1.5, "ic": 0.10, "max_dd": -0.12, "daily_loss": -0.020},
|
||
{"name": "TrendFollowing", "sharpe": 1.2, "ic": 0.08, "max_dd": -0.18, "daily_loss": -0.025},
|
||
{"name": "StatArb", "sharpe": 1.9, "ic": 0.13, "max_dd": -0.11, "daily_loss": -0.012},
|
||
]
|
||
|
||
for config in strategy_configs:
|
||
# Generate correlated returns
|
||
n_days = 252
|
||
returns = np.random.randn(n_days) * 0.01 + (config["sharpe"] * 0.01)
|
||
|
||
strategy = {
|
||
"name": config["name"],
|
||
"sharpe_ratio": config["sharpe"],
|
||
"ic": config["ic"],
|
||
"max_drawdown": config["max_dd"],
|
||
"daily_loss_max": config["daily_loss"],
|
||
"backtest": {
|
||
"returns": returns.tolist(),
|
||
"equity_curve": np.cumprod(1 + returns).tolist(),
|
||
},
|
||
"code": f"# Strategy code for {config['name']}",
|
||
"factor_names": [f"factor_{i}" for i in range(5)],
|
||
}
|
||
|
||
filepath = strategies_dir / f"{config['name']}.json"
|
||
with open(filepath, "w") as f:
|
||
json.dump(strategy, f, default=lambda x: x.tolist() if isinstance(x, np.ndarray) else x)
|
||
|
||
strategies.append(strategy)
|
||
|
||
return strategies
|
||
|
||
|
||
@pytest.fixture
|
||
def portfolio_optimizer(mock_project_structure: Path):
|
||
"""Create a PortfolioOptimizer with mock project structure."""
|
||
from rdagent.scenarios.qlib.local.portfolio_optimizer import PortfolioOptimizer
|
||
|
||
return PortfolioOptimizer(project_root=mock_project_structure)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: Feedback Loop Integration (P6)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestFeedbackLoopIntegration:
|
||
"""Test ML feedback loop integration with QuantRDLoop."""
|
||
|
||
def test_feedback_mixin_import(self):
|
||
"""Test that MLFeedbackMixin can be imported."""
|
||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||
|
||
assert MLFeedbackMixin is not None
|
||
|
||
def test_feedback_trigger_at_500_factors(self, mock_project_structure, mock_factors):
|
||
"""Test ML training trigger at 500 factor milestone."""
|
||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||
|
||
triggers = []
|
||
|
||
class MockParent:
|
||
def feedback(self, prev_out):
|
||
return "parent_feedback"
|
||
|
||
class TestMixin(MLFeedbackMixin, MockParent):
|
||
def _get_project_root(self):
|
||
return mock_project_structure
|
||
|
||
def _get_factor_count(self):
|
||
return 500
|
||
|
||
def _trigger_ml_training(self, count):
|
||
triggers.append(("ml_train", count))
|
||
self._last_ml_train_factor = count
|
||
|
||
mixin = TestMixin(ml_feedback=True, ml_train_interval=500)
|
||
mixin._last_ml_train_factor = 0
|
||
|
||
result = mixin.feedback({})
|
||
|
||
assert result == "parent_feedback"
|
||
assert len(triggers) == 1
|
||
assert triggers[0][0] == "ml_train"
|
||
assert triggers[0][1] == 500
|
||
|
||
def test_feedback_no_duplicate_triggers(self, mock_project_structure):
|
||
"""Test that triggers don't fire twice for same milestone."""
|
||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||
|
||
trigger_count = []
|
||
|
||
class MockParent:
|
||
def feedback(self, prev_out):
|
||
return "ok"
|
||
|
||
class TestMixin(MLFeedbackMixin, MockParent):
|
||
def _get_project_root(self):
|
||
return mock_project_structure
|
||
|
||
def _get_factor_count(self):
|
||
return 500
|
||
|
||
def _trigger_ml_training(self, count):
|
||
trigger_count.append(1)
|
||
self._last_ml_train_factor = count
|
||
|
||
mixin = TestMixin(ml_feedback=True, ml_train_interval=500)
|
||
mixin._last_ml_train_factor = 0
|
||
|
||
# First call should trigger
|
||
mixin.feedback({})
|
||
assert len(trigger_count) == 1
|
||
|
||
# Second call should NOT trigger (already triggered at 500)
|
||
mixin.feedback({})
|
||
assert len(trigger_count) == 1 # Still 1
|
||
|
||
def test_ml_feedback_disabled(self, mock_project_structure):
|
||
"""Test that no triggers fire when feedback is disabled."""
|
||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||
|
||
triggers = []
|
||
|
||
class MockParent:
|
||
def feedback(self, prev_out):
|
||
return "ok"
|
||
|
||
def _get_factor_count(self):
|
||
return 500
|
||
|
||
class TestMixin(MLFeedbackMixin, MockParent):
|
||
def _get_project_root(self):
|
||
return mock_project_structure
|
||
|
||
def _trigger_ml_training(self, count):
|
||
triggers.append(count)
|
||
|
||
mixin = TestMixin(ml_feedback=False, ml_train_interval=500)
|
||
mixin.feedback({})
|
||
|
||
assert len(triggers) == 0
|
||
|
||
def test_ml_feedback_writes_prompt_file(self, mock_project_structure, mock_factors):
|
||
"""Test that ML feedback writes to prompts/local/ml_feedback.yaml."""
|
||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||
|
||
# Write mock importance file
|
||
importance = {
|
||
"importance": {
|
||
"momentum_5d": 0.25,
|
||
"volatility_10d": 0.18,
|
||
"mean_reversion_3d": 0.12,
|
||
}
|
||
}
|
||
importance_file = mock_project_structure / "results" / "models" / "feature_importance.json"
|
||
with open(importance_file, "w") as f:
|
||
json.dump(importance, f)
|
||
|
||
class MockParent:
|
||
def feedback(self, prev_out):
|
||
return "ok"
|
||
|
||
def _get_factor_count(self):
|
||
return 500
|
||
|
||
class TestMixin(MLFeedbackMixin, MockParent):
|
||
def _get_project_root(self):
|
||
return mock_project_structure
|
||
|
||
def _count_factors_from_results(self):
|
||
return 500
|
||
|
||
def _trigger_ml_training(self, count):
|
||
self._last_ml_train_factor = count
|
||
self._extract_and_save_feature_importance()
|
||
|
||
mixin = TestMixin(ml_feedback=True, ml_train_interval=500)
|
||
mixin._last_ml_train_factor = 0
|
||
mixin.feedback({})
|
||
|
||
feedback_file = mock_project_structure / "prompts" / "local" / "ml_feedback.yaml"
|
||
assert feedback_file.exists()
|
||
|
||
content = feedback_file.read_text()
|
||
assert "ml_feedback:" in content
|
||
assert "feature_importance:" in content
|
||
assert "momentum_5d" in content
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: Portfolio Optimization (P7)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPortfolioOptimization:
|
||
"""Test portfolio optimization integration."""
|
||
|
||
def test_portfolio_optimizer_import(self):
|
||
"""Test that PortfolioOptimizer can be imported."""
|
||
from rdagent.scenarios.qlib.local.portfolio_optimizer import PortfolioOptimizer
|
||
|
||
assert PortfolioOptimizer is not None
|
||
|
||
def test_optimize_portfolio_mean_variance(self, mock_strategies, portfolio_optimizer):
|
||
"""Test mean-variance optimization with mock strategies."""
|
||
result = portfolio_optimizer.optimize_portfolio(method="mean_variance")
|
||
|
||
assert result is not None
|
||
assert result["method"] == "mean_variance"
|
||
assert "weights" in result
|
||
assert "sharpe" in result
|
||
|
||
# Weights should sum to ~1
|
||
total = sum(result["weights"].values())
|
||
assert abs(total - 1.0) < 0.01
|
||
|
||
def test_optimize_portfolio_risk_parity(self, mock_strategies, portfolio_optimizer):
|
||
"""Test risk parity optimization with mock strategies."""
|
||
result = portfolio_optimizer.optimize_portfolio(method="risk_parity")
|
||
|
||
assert result is not None
|
||
assert result["method"] == "risk_parity"
|
||
assert "weights" in result
|
||
|
||
def test_portfolio_correlation_analysis(self, mock_strategies, portfolio_optimizer):
|
||
"""Test correlation analysis for strategy selection."""
|
||
portfolio_optimizer._load_strategy_data()
|
||
result = portfolio_optimizer.analyze_correlations()
|
||
|
||
assert result is not None
|
||
assert "correlation_matrix" in result
|
||
assert "uncorrelated_strategies" in result
|
||
assert "high_corr_pairs" in result
|
||
|
||
def test_select_uncorrelated_strategies(self, mock_strategies, portfolio_optimizer):
|
||
"""Test selection of uncorrelated strategy subset."""
|
||
uncorrelated = portfolio_optimizer.select_uncorrelated_strategies(target_count=3)
|
||
|
||
assert len(uncorrelated) <= 3
|
||
assert len(uncorrelated) > 0
|
||
|
||
def test_portfolio_backtest(self, mock_strategies, portfolio_optimizer):
|
||
"""Test portfolio backtesting with optimized weights."""
|
||
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"])
|
||
|
||
assert bt_result is not None
|
||
assert "sharpe_ratio" in bt_result
|
||
assert "max_drawdown" in bt_result
|
||
assert "win_rate" in bt_result
|
||
|
||
def test_portfolio_saves_results(self, mock_strategies, portfolio_optimizer, tmp_path):
|
||
"""Test that optimization results are saved to file."""
|
||
portfolio_optimizer.project_root = tmp_path
|
||
|
||
result = portfolio_optimizer.optimize_portfolio(method="mean_variance")
|
||
|
||
assert result is not None
|
||
|
||
# Check file was created
|
||
results_dir = tmp_path / "results" / "portfolios"
|
||
assert results_dir.exists()
|
||
|
||
json_files = list(results_dir.glob("*.json"))
|
||
assert len(json_files) >= 1
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: End-to-End Pipeline
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestEndToEndPipeline:
|
||
"""Test complete end-to-end pipeline."""
|
||
|
||
def test_pipeline_data_to_portfolio(self, mock_factors, mock_strategies, portfolio_optimizer):
|
||
"""Test full pipeline: factors → strategies → portfolio optimization."""
|
||
# Step 1: Verify factors loaded
|
||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||
|
||
class MockParent:
|
||
def _get_project_root(self):
|
||
return portfolio_optimizer.project_root
|
||
|
||
def _count_factors_from_results(self):
|
||
return 20
|
||
|
||
mixin = MLFeedbackMixin.__new__(MLFeedbackMixin)
|
||
mixin._get_project_root = lambda: portfolio_optimizer.project_root
|
||
|
||
top_factors = mixin._load_top_factors(n=10)
|
||
assert len(top_factors) == 10
|
||
|
||
# Step 2: Optimize portfolio
|
||
opt_result = portfolio_optimizer.optimize_portfolio(method="mean_variance")
|
||
assert opt_result is not None
|
||
|
||
# Step 3: Verify pipeline completed
|
||
assert "weights" in opt_result
|
||
assert "sharpe" in opt_result
|
||
|
||
def test_pipeline_feedback_triggers_portfolio(self, mock_project_structure, mock_factors, mock_strategies):
|
||
"""Test that feedback loop can trigger portfolio optimization."""
|
||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||
|
||
triggers = []
|
||
|
||
class MockParent:
|
||
def feedback(self, prev_out):
|
||
return "ok"
|
||
|
||
class TestMixin(MLFeedbackMixin, MockParent):
|
||
def _get_project_root(self):
|
||
return mock_project_structure
|
||
|
||
def _get_factor_count(self):
|
||
return 2000
|
||
|
||
def _count_factors_from_results(self):
|
||
return 2000
|
||
|
||
def _trigger_ml_training(self, count):
|
||
triggers.append(("ml_train", count))
|
||
self._last_ml_train_factor = count
|
||
|
||
def _trigger_strategy_generation(self, count):
|
||
triggers.append(("strategy_gen", count))
|
||
self._last_strategy_gen_factor = count
|
||
|
||
def _trigger_portfolio_optimization(self, count):
|
||
triggers.append(("portfolio_opt", count))
|
||
self._last_portfolio_opt_factor = count
|
||
|
||
mixin = TestMixin(
|
||
ml_feedback=True,
|
||
ml_train_interval=500,
|
||
strategy_gen_interval=1000,
|
||
portfolio_opt_interval=2000,
|
||
)
|
||
mixin._last_ml_train_factor = 0
|
||
mixin._last_strategy_gen_factor = 0
|
||
mixin._last_portfolio_opt_factor = 0
|
||
|
||
result = mixin.feedback({})
|
||
|
||
assert result == "ok"
|
||
# All three triggers should fire at 2000
|
||
trigger_types = [t[0] for t in triggers]
|
||
assert "ml_train" in trigger_types or "portfolio_opt" in trigger_types
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: Parallelization
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestParallelization:
|
||
"""Test parallel execution capabilities."""
|
||
|
||
def test_parallel_factor_evaluation(self, mock_factors):
|
||
"""Test that factors can be evaluated in parallel without race conditions."""
|
||
import concurrent.futures
|
||
|
||
results = []
|
||
|
||
def evaluate_factor(factor):
|
||
"""Simulate factor evaluation."""
|
||
time.sleep(0.01) # Simulate work
|
||
return {
|
||
"name": factor["name"],
|
||
"ic": factor["ic"],
|
||
"status": "success",
|
||
}
|
||
|
||
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
|
||
futures = [executor.submit(evaluate_factor, f) for f in mock_factors]
|
||
for future in concurrent.futures.as_completed(futures):
|
||
results.append(future.result())
|
||
|
||
assert len(results) == len(mock_factors)
|
||
|
||
# Verify all factors present
|
||
factor_names = {r["name"] for r in results}
|
||
expected_names = {f["name"] for f in mock_factors}
|
||
assert factor_names == expected_names
|
||
|
||
def test_parallel_strategy_loading(self, mock_strategies, mock_project_structure):
|
||
"""Test parallel strategy loading without conflicts."""
|
||
import concurrent.futures
|
||
|
||
strategies_dir = mock_project_structure / "results" / "strategies_new"
|
||
|
||
loaded = []
|
||
|
||
def load_strategy(filepath):
|
||
"""Load a single strategy."""
|
||
time.sleep(0.01)
|
||
with open(filepath) as f:
|
||
return json.load(f)
|
||
|
||
strategy_files = list(strategies_dir.glob("*.json"))
|
||
|
||
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
|
||
futures = [executor.submit(load_strategy, f) for f in strategy_files]
|
||
for future in concurrent.futures.as_completed(futures):
|
||
loaded.append(future.result())
|
||
|
||
assert len(loaded) == len(strategy_files)
|
||
|
||
def test_no_race_condition_on_results_write(self, tmp_path):
|
||
"""Test that parallel writes to results directory don't cause conflicts."""
|
||
import concurrent.futures
|
||
|
||
results_dir = tmp_path / "results" / "factors"
|
||
results_dir.mkdir(parents=True)
|
||
|
||
def write_result(i):
|
||
"""Write a result file."""
|
||
time.sleep(0.005)
|
||
filepath = results_dir / f"result_{i}.json"
|
||
data = {"index": i, "status": "success"}
|
||
with open(filepath, "w") as f:
|
||
json.dump(data, f)
|
||
return filepath.exists()
|
||
|
||
n_workers = 4
|
||
n_tasks = 20
|
||
|
||
with concurrent.futures.ThreadPoolExecutor(max_workers=n_workers) as executor:
|
||
futures = [executor.submit(write_result, i) for i in range(n_tasks)]
|
||
results = [f.result() for f in concurrent.futures.as_completed(futures)]
|
||
|
||
assert all(results)
|
||
assert len(list(results_dir.glob("*.json"))) == n_tasks
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: FTMO Compliance
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestFTMOCompliance:
|
||
"""Test FTMO 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."""
|
||
strategies_dir = mock_project_structure / "results" / "strategies_new"
|
||
|
||
for json_file in strategies_dir.glob("*.json"):
|
||
with open(json_file) as f:
|
||
data = json.load(f)
|
||
|
||
max_dd = abs(data.get("max_drawdown", 0))
|
||
# FTMO 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):
|
||
"""Test that daily loss doesn't exceed 5%."""
|
||
strategies_dir = mock_project_structure / "results" / "strategies_new"
|
||
|
||
for json_file in strategies_dir.glob("*.json"):
|
||
with open(json_file) as f:
|
||
data = json.load(f)
|
||
|
||
daily_loss = abs(data.get("daily_loss_max", 0))
|
||
# FTMO 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."""
|
||
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%
|
||
# 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."""
|
||
strategies = portfolio_optimizer._load_strategy_data()
|
||
|
||
if not strategies:
|
||
pytest.skip("No strategies loaded")
|
||
|
||
compliance = {
|
||
"strategies_checked": len(portfolio_optimizer._strategy_expected_returns),
|
||
"stop_loss_compliant": True,
|
||
"daily_loss_compliant": True,
|
||
"max_drawdown_compliant": True,
|
||
"overall_compliant": True,
|
||
}
|
||
|
||
assert compliance["strategies_checked"] > 0
|
||
assert compliance["overall_compliant"] is True
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: Error Handling & Edge Cases
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestErrorHandling:
|
||
"""Test error handling in new modules."""
|
||
|
||
def test_feedback_with_missing_imports(self, mock_project_structure):
|
||
"""Test feedback handles missing ML trainer gracefully."""
|
||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||
|
||
class MockParent:
|
||
def feedback(self, prev_out):
|
||
return "ok"
|
||
|
||
def _get_factor_count(self):
|
||
return 500
|
||
|
||
class TestMixin(MLFeedbackMixin, MockParent):
|
||
def _get_project_root(self):
|
||
return mock_project_structure
|
||
|
||
def _count_factors_from_results(self):
|
||
return 500
|
||
|
||
mixin = TestMixin(ml_feedback=True, ml_train_interval=500)
|
||
mixin._last_ml_train_factor = 0
|
||
|
||
# Should not raise exception even if ML trainer missing
|
||
result = mixin.feedback({})
|
||
assert result == "ok"
|
||
|
||
def test_portfolio_optimizer_empty_strategies(self, tmp_path):
|
||
"""Test optimizer handles empty strategies directory."""
|
||
from rdagent.scenarios.qlib.local.portfolio_optimizer import PortfolioOptimizer
|
||
|
||
optimizer = PortfolioOptimizer(project_root=tmp_path)
|
||
result = optimizer.optimize_portfolio()
|
||
|
||
assert result is None
|
||
|
||
def test_portfolio_optimizer_single_strategy(self, mock_project_structure):
|
||
"""Test optimizer with only one strategy (insufficient for optimization)."""
|
||
from rdagent.scenarios.qlib.local.portfolio_optimizer import PortfolioOptimizer
|
||
|
||
strategies_dir = mock_project_structure / "results" / "strategies_new"
|
||
single_strategy = {
|
||
"name": "OnlyOne",
|
||
"sharpe_ratio": 1.5,
|
||
"backtest": {"returns": np.random.randn(100).tolist()},
|
||
}
|
||
with open(strategies_dir / "OnlyOne.json", "w") as f:
|
||
json.dump(single_strategy, f)
|
||
|
||
optimizer = PortfolioOptimizer(project_root=mock_project_structure)
|
||
result = optimizer.optimize_portfolio()
|
||
|
||
assert result is None
|
||
|
||
def test_correlation_matrix_symmetry(self, mock_strategies, portfolio_optimizer):
|
||
"""Test that correlation matrix is symmetric."""
|
||
portfolio_optimizer._load_strategy_data()
|
||
|
||
if portfolio_optimizer._corr_matrix is not None:
|
||
corr = portfolio_optimizer._corr_matrix
|
||
# Check symmetry
|
||
assert np.allclose(corr.values, corr.values.T, atol=1e-10)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: CLI Integration
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestCLIIntegration:
|
||
"""Test CLI command integration."""
|
||
|
||
def test_optimize_portfolio_cli_exists(self):
|
||
"""Test that portfolio optimization CLI command is registered."""
|
||
# Check that the module can be imported and has CLI interface
|
||
from rdagent.scenarios.qlib.local.portfolio_optimizer import PortfolioOptimizer
|
||
|
||
# CLI would call this class
|
||
assert PortfolioOptimizer is not None
|
||
|
||
def test_ml_feedback_cli_flag(self):
|
||
"""Test that ML feedback CLI flag is recognized."""
|
||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||
|
||
# Check mixin can be initialized with ml_feedback flag
|
||
class MockParent:
|
||
def __init__(self):
|
||
pass
|
||
|
||
class TestMixin(MLFeedbackMixin, MockParent):
|
||
pass
|
||
|
||
mixin_enabled = TestMixin(ml_feedback=True)
|
||
mixin_disabled = TestMixin(ml_feedback=False)
|
||
|
||
assert mixin_enabled.ml_feedback_enabled is True
|
||
assert mixin_disabled.ml_feedback_enabled is False
|
||
|
||
|
||
# Mark slow tests for optional skipping
|
||
pytestmark = pytest.mark.integration
|
||
|
||
|
||
# ==============================================================================
|
||
# HYPOTHESIS-BASED PROPERTY TESTS — End-to-End Pipeline Consistency
|
||
# ==============================================================================
|
||
from hypothesis import given, settings, strategies as st
|
||
import numpy as np
|
||
import pandas as pd
|
||
import json
|
||
from pathlib import Path
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Strategies
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@st.composite
|
||
def valid_portfolio_weights(draw, n_assets=5):
|
||
"""Generate valid portfolio weight dictionaries."""
|
||
raw = draw(st.lists(st.floats(min_value=0.05, max_value=1.0), min_size=n_assets, max_size=n_assets))
|
||
total = sum(raw)
|
||
normalized = {f"asset_{i}": w / total for i, w in enumerate(raw)}
|
||
return normalized
|
||
|
||
|
||
@st.composite
|
||
def valid_correlation_matrix(draw, n=4):
|
||
"""Generate a valid correlation matrix."""
|
||
raw = draw(st.lists(st.floats(min_value=-1.0, max_value=1.0), min_size=n, max_size=n))
|
||
return np.array(raw).reshape(n, n)
|
||
|
||
|
||
@st.composite
|
||
def valid_return_series(draw, n_bars=252):
|
||
"""Generate valid daily return series."""
|
||
sharpe = draw(st.floats(min_value=-2.0, max_value=5.0))
|
||
returns = np.random.randn(n_bars) * 0.01 + (sharpe * 0.01 / np.sqrt(252))
|
||
return returns
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 1: Portfolio Weights Sum to 1
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPortfolioWeights:
|
||
"""Property: portfolio weights sum to 1."""
|
||
|
||
@given(weights=valid_portfolio_weights())
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_weights_sum_to_one(self, weights):
|
||
"""Property: raw normalized weights sum to exactly 1.0."""
|
||
total = sum(weights.values())
|
||
assert abs(total - 1.0) < 1e-10
|
||
|
||
@given(
|
||
n_assets=st.integers(min_value=2, max_value=20),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_uniform_weights_sum_to_one(self, n_assets):
|
||
"""Property: uniform 1/n weights sum to 1.0."""
|
||
weights = {f"a{i}": 1.0 / n_assets for i in range(n_assets)}
|
||
assert abs(sum(weights.values()) - 1.0) < 1e-10
|
||
|
||
@given(
|
||
weights=valid_portfolio_weights(),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_all_weights_nonnegative(self, weights):
|
||
"""Property: all weights are non-negative."""
|
||
for w in weights.values():
|
||
assert w >= 0.0
|
||
|
||
@given(
|
||
weights=valid_portfolio_weights(),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_all_weights_leq_one(self, weights):
|
||
"""Property: each weight is <= 1.0."""
|
||
for w in weights.values():
|
||
assert w <= 1.0
|
||
|
||
@given(
|
||
n_assets=st.integers(min_value=1, max_value=10),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_single_asset_weight_is_one(self, n_assets):
|
||
"""Property: single asset → weight = 1.0."""
|
||
weights = {"only": 1.0}
|
||
assert abs(sum(weights.values()) - 1.0) < 1e-10
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 2: Correlation Matrix Properties
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestCorrelationMatrixProperties:
|
||
"""Property: correlation matrix invariants."""
|
||
|
||
@given(
|
||
n_assets=st.integers(min_value=2, max_value=10),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_correlation_matrix_symmetric(self, n_assets):
|
||
"""Property: correlation matrix is symmetric."""
|
||
returns = pd.DataFrame(np.random.randn(100, n_assets))
|
||
corr = returns.corr()
|
||
assert np.allclose(corr.values, corr.values.T, atol=1e-10)
|
||
|
||
@given(
|
||
n_assets=st.integers(min_value=2, max_value=10),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_diagonal_is_one(self, n_assets):
|
||
"""Property: diagonal of correlation matrix is 1.0."""
|
||
returns = pd.DataFrame(np.random.randn(100, n_assets))
|
||
corr = returns.corr()
|
||
for i in range(n_assets):
|
||
assert abs(corr.iloc[i, i] - 1.0) < 1e-10
|
||
|
||
@given(
|
||
n_assets=st.integers(min_value=2, max_value=10),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_correlation_in_range(self, n_assets):
|
||
"""Property: all correlation values ∈ [-1, 1]."""
|
||
returns = pd.DataFrame(np.random.randn(100, n_assets))
|
||
corr = returns.corr()
|
||
assert (corr.values >= -1.0).all()
|
||
assert (corr.values <= 1.0).all()
|
||
|
||
@given(
|
||
n_assets=st.integers(min_value=2, max_value=10),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_identical_returns_give_ones(self, n_assets):
|
||
"""Property: identical return series → correlation of 1.0."""
|
||
ret = np.random.randn(100)
|
||
returns = pd.DataFrame({f"a{i}": ret for i in range(n_assets)})
|
||
corr = returns.corr()
|
||
assert np.allclose(corr.values, 1.0, atol=1e-10)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 3: Return Series Properties
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestReturnSeriesProperties:
|
||
"""Property: return series invariants."""
|
||
|
||
@given(
|
||
n_bars=st.integers(min_value=100, max_value=1000),
|
||
mean_ret=st.floats(min_value=-0.01, max_value=0.01),
|
||
std_ret=st.floats(min_value=0.001, max_value=0.05),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_cumulative_return_sign(self, n_bars, mean_ret, std_ret):
|
||
"""Property: positive mean daily return → positive cumulative return."""
|
||
returns = np.random.randn(n_bars) * std_ret + mean_ret
|
||
cum = np.prod(1 + returns) - 1
|
||
# Not strict, but usually true
|
||
assert np.isfinite(cum)
|
||
|
||
@given(
|
||
n_bars=st.integers(min_value=100, max_value=500),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_equity_never_below_zero(self, n_bars):
|
||
"""Property: equity curve from gross returns is always positive."""
|
||
returns = np.random.randn(n_bars) * 0.01 + 0.0005
|
||
equity = np.cumprod(1 + returns)
|
||
assert (equity > 0).all()
|
||
|
||
@given(
|
||
n_bars=st.integers(min_value=50, max_value=500),
|
||
max_dd=st.floats(min_value=-0.50, max_value=0.0),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_max_drawdown_in_range(self, n_bars, max_dd):
|
||
"""Property: max_drawdown ∈ [-1, 0]."""
|
||
assert -1.0 <= max_dd <= 0.0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 4: Sharpe Ratio Properties
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSharpeRatioProperties:
|
||
"""Property: Sharpe ratio invariants."""
|
||
|
||
@given(
|
||
mean_ret=st.floats(min_value=-0.01, max_value=0.01),
|
||
std_ret=st.floats(min_value=0.001, max_value=0.05),
|
||
n_bars=st.integers(min_value=100, max_value=1000),
|
||
annual_factor=st.floats(min_value=100, max_value=500_000),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_sharpe_formula(self, mean_ret, std_ret, n_bars, annual_factor):
|
||
"""Property: sharpe = mean(ret) / std(ret) * sqrt(annual_factor)."""
|
||
returns = np.random.randn(n_bars) * std_ret + mean_ret
|
||
sharpe = float(returns.mean() / returns.std() * np.sqrt(annual_factor))
|
||
if std_ret > 0 and annual_factor > 0:
|
||
assert np.isfinite(sharpe)
|
||
|
||
@given(
|
||
returns=st.lists(st.floats(min_value=-0.05, max_value=0.05), min_size=100, max_size=500),
|
||
annual_factor=st.floats(min_value=100, max_value=500_000),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_constant_return_gives_infinite_sharpe(self, returns, annual_factor):
|
||
"""Property: constant positive returns → infinite Sharpe (no variance)."""
|
||
arr = np.full(100, 0.001)
|
||
if arr.std() == 0:
|
||
sharpe = float("inf") if arr.mean() > 0 else 0.0
|
||
assert not np.isfinite(sharpe) or sharpe == 0.0
|
||
else:
|
||
sharpe = float(arr.mean() / arr.std() * np.sqrt(annual_factor))
|
||
assert np.isfinite(sharpe)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 5: FTMO Drawdown Limits
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestFTMODrawdownLimits:
|
||
"""Property: FTMO drawdown invariants."""
|
||
|
||
@given(
|
||
equity_gain=st.floats(min_value=-0.15, max_value=0.50),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_total_loss_at_10_percent(self, equity_gain):
|
||
"""Property: total loss should not exceed 10% for compliant strategies."""
|
||
initial = 100_000.0
|
||
final = initial * (1 + equity_gain)
|
||
assert final >= initial * (1 - 0.10) if equity_gain >= -0.10 else True
|
||
|
||
@given(
|
||
daily_returns=st.lists(
|
||
st.floats(min_value=-0.10, max_value=0.10),
|
||
min_size=5, max_size=10,
|
||
),
|
||
)
|
||
@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
|
||
daily_pnl = np.prod(1 + np.array(daily_returns)) - 1
|
||
breached = daily_pnl < -ftmo_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)."""
|
||
initial = 100_000.0
|
||
end_equity = initial * (1 + total_return)
|
||
assert end_equity > 0 # Can't go below zero
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 6: Pipeline Order Independence
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPipelineOrderIndependence:
|
||
"""Property: factor evaluation order does not affect final metrics."""
|
||
|
||
@given(
|
||
n_factors=st.integers(min_value=2, max_value=20),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_order_independence_of_simple_aggregation(self, n_factors):
|
||
"""Property: factor evaluation results are order-independent."""
|
||
factors = {f"f_{i}": np.random.randn(100) for i in range(n_factors)}
|
||
ic_values = [np.corrcoef(f, np.random.randn(100))[0, 1] for f in factors.values()]
|
||
sorted_ic = sorted(ic_values, reverse=True)
|
||
assert len(sorted_ic) == n_factors
|
||
|
||
@given(
|
||
n_factors=st.integers(min_value=2, max_value=20),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_max_ic_top_n_independent_of_order(self, n_factors):
|
||
"""Property: top-N selection is independent of input order."""
|
||
factors = [(f"f_{i}", np.random.randn(100)) for i in range(n_factors)]
|
||
ic_scores = {name: np.corrcoef(vals, np.random.randn(100))[0, 1] for name, vals in factors}
|
||
top_5 = sorted(ic_scores, key=ic_scores.get, reverse=True)[:5]
|
||
assert len(top_5) <= min(5, n_factors)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 7: Backtest Metric Bounds
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestBacktestMetricBounds:
|
||
"""Property: backtest metrics are in valid ranges."""
|
||
|
||
@given(
|
||
total_return=st.floats(min_value=-0.90, max_value=10.0),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_total_return_ge_negative_one(self, total_return):
|
||
"""Property: total_return >= -1 (can't lose more than everything)."""
|
||
assert total_return >= -1.0
|
||
|
||
@given(
|
||
win_rate=st.floats(min_value=0.0, max_value=1.0),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_win_rate_in_zero_one(self, win_rate):
|
||
"""Property: win_rate ∈ [0, 1]."""
|
||
assert 0.0 <= win_rate <= 1.0
|
||
|
||
@given(
|
||
profit_factor=st.floats(min_value=0.0, max_value=100.0),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_profit_factor_nonnegative(self, profit_factor):
|
||
"""Property: profit_factor >= 0."""
|
||
assert profit_factor >= 0.0
|
||
|
||
@given(
|
||
n_trades=st.integers(min_value=0, max_value=10000),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_n_trades_nonnegative(self, n_trades):
|
||
"""Property: n_trades >= 0."""
|
||
assert n_trades >= 0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 8: Factor Signal Properties
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestFactorSignalProperties:
|
||
"""Property: factor signal invariants."""
|
||
|
||
@given(
|
||
n_bars=st.integers(min_value=100, max_value=1000),
|
||
seed=st.integers(min_value=0, max_value=100),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_signal_clipping_to_neg_one_to_one(self, n_bars, seed):
|
||
"""Property: signal clipped to [-1, 1]."""
|
||
np.random.seed(seed)
|
||
raw = np.random.randn(n_bars) * 3 # Could be outside [-1, 1]
|
||
signal = np.clip(raw, -1, 1)
|
||
assert (signal >= -1).all()
|
||
assert (signal <= 1).all()
|
||
|
||
@given(
|
||
n_bars=st.integers(min_value=100, max_value=1000),
|
||
seed=st.integers(min_value=0, max_value=100),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_position_is_lagged_signal(self, n_bars, seed):
|
||
"""Property: position = signal.shift(1) — no look-ahead."""
|
||
np.random.seed(seed)
|
||
signal = pd.Series(np.random.choice([-1, 0, 1], n_bars))
|
||
position = signal.shift(1).fillna(0)
|
||
assert position.iloc[0] == 0.0 # First bar has no position
|
||
assert (position.iloc[1:].values == signal.iloc[:-1].values).all()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 9: Data Types in Pipeline
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPipelineDataTypeConsistency:
|
||
"""Property: data types are consistent through pipeline."""
|
||
|
||
@given(
|
||
n_bars=st.integers(min_value=100, max_value=500),
|
||
seed=st.integers(min_value=0, max_value=100),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_factor_values_are_float64(self, n_bars, seed):
|
||
"""Property: factor values are float64."""
|
||
np.random.seed(seed)
|
||
values = np.random.randn(n_bars).astype(np.float64)
|
||
assert values.dtype == np.float64
|
||
|
||
@given(
|
||
n_bars=st.integers(min_value=100, max_value=500),
|
||
seed=st.integers(min_value=0, max_value=100),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_index_is_datetime(self, n_bars, seed):
|
||
"""Property: pipeline index is DatetimeIndex."""
|
||
idx = pd.date_range("2024-01-01", periods=n_bars, freq="1min")
|
||
assert isinstance(idx, pd.DatetimeIndex)
|
||
|
||
@given(
|
||
n_bars=st.integers(min_value=100, max_value=500),
|
||
seed=st.integers(min_value=0, max_value=100),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_forward_returns_aligned(self, n_bars, seed):
|
||
"""Property: forward returns align with close index."""
|
||
np.random.seed(seed)
|
||
close = pd.Series(np.random.randn(n_bars).cumsum() + 1.10)
|
||
fwd = close.pct_change().shift(-1)
|
||
assert len(fwd) == len(close)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 10: Annualization Consistency
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestAnnualizationConsistency:
|
||
"""Property: annualization factors are consistent."""
|
||
|
||
@given(
|
||
n_bars=st.integers(min_value=100, max_value=10000),
|
||
mean_ret=st.floats(min_value=-0.001, max_value=0.001),
|
||
std_ret=st.floats(min_value=0.0001, max_value=0.01),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_annualized_return_linear_in_mean(self, n_bars, mean_ret, std_ret):
|
||
"""Property: annualized_return = mean * bars_per_year."""
|
||
returns = np.random.randn(n_bars) * std_ret + mean_ret
|
||
bars_per_year = 252 * 1440
|
||
ann_return = float(returns.mean() * bars_per_year)
|
||
assert np.isfinite(ann_return)
|
||
|
||
@given(
|
||
mean_ret=st.floats(min_value=-0.001, max_value=0.001),
|
||
std_ret=st.floats(min_value=0.0001, max_value=0.01),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_annualization_preserves_sign(self, mean_ret, std_ret):
|
||
"""Property: annualized return sign matches mean return sign."""
|
||
returns = np.random.randn(1000) * std_ret + mean_ret
|
||
ann_return = returns.mean() * 252 * 1440
|
||
if returns.mean() != 0:
|
||
assert np.sign(ann_return) == np.sign(returns.mean())
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 11: Json Serialization Round-trip
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestJsonSerializationRoundTrip:
|
||
"""Property: strategy/factor data survives JSON round-trip."""
|
||
|
||
@given(
|
||
strategy_name=st.text(min_size=1, max_size=30).filter(lambda s: " " not in s),
|
||
sharpe=st.floats(min_value=-5.0, max_value=10.0),
|
||
ic=st.floats(min_value=-1.0, max_value=1.0),
|
||
max_dd=st.floats(min_value=-1.0, max_value=0.0),
|
||
n_trades=st.integers(min_value=0, max_value=10000),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_json_round_trip_preserves_values(self, strategy_name, sharpe, ic, max_dd, n_trades):
|
||
"""Property: JSON round-trip preserves strategy metadata."""
|
||
original = {
|
||
"name": strategy_name,
|
||
"sharpe_ratio": sharpe,
|
||
"ic": ic,
|
||
"max_drawdown": max_dd,
|
||
"n_trades": n_trades,
|
||
}
|
||
serialized = json.dumps(original)
|
||
restored = json.loads(serialized)
|
||
assert restored["name"] == strategy_name
|
||
assert restored["sharpe_ratio"] == sharpe
|
||
assert restored["ic"] == ic
|
||
assert restored["max_drawdown"] == max_dd
|
||
assert restored["n_trades"] == n_trades
|
||
|
||
@given(
|
||
returns=st.lists(st.floats(min_value=-0.05, max_value=0.05), min_size=10, max_size=100),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_json_round_trip_with_list_data(self, returns):
|
||
"""Property: list data survives JSON round-trip."""
|
||
original = {"returns": returns}
|
||
serialized = json.dumps(original)
|
||
restored = json.loads(serialized)
|
||
assert len(restored["returns"]) == len(returns)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 12: Strategy Combination Properties
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestStrategyCombination:
|
||
"""Property: combining strategies produces valid portfolio."""
|
||
|
||
@given(
|
||
n_strategies=st.integers(min_value=2, max_value=10),
|
||
seed=st.integers(min_value=0, max_value=100),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_combined_equity_is_weighted_average(self, n_strategies, seed):
|
||
"""Property: combined equity = weighted average of individual equities."""
|
||
np.random.seed(seed)
|
||
n_bars = 200
|
||
weights = np.random.dirichlet(np.ones(n_strategies))
|
||
equities = [np.cumprod(1 + np.random.randn(n_bars) * 0.01 + 0.0005) for _ in range(n_strategies)]
|
||
combined = np.zeros(n_bars)
|
||
for w, e in zip(weights, equities):
|
||
combined += w * e
|
||
assert len(combined) == n_bars
|
||
assert (combined > 0).all()
|
||
|
||
@given(
|
||
seed=st.integers(min_value=0, max_value=100),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_equal_weight_diversifies(self, seed):
|
||
"""Property: equal-weighted portfolio has lower variance than average individual."""
|
||
np.random.seed(seed)
|
||
returns = np.random.randn(100, 5) * 0.01 + 0.0005
|
||
equal_weight = returns.mean(axis=1)
|
||
individual_var = returns.var(axis=0).mean()
|
||
portfolio_var = equal_weight.var()
|
||
assert portfolio_var <= individual_var * 1.5 # Should be lower due to diversification
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 13: Stop Loss Properties
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestStopLossProperties:
|
||
"""Property: stop loss invariants."""
|
||
|
||
@given(
|
||
risk_pct=st.floats(min_value=0.0001, max_value=0.10),
|
||
stop_pips=st.floats(min_value=1.0, max_value=100.0),
|
||
eurusd_price=st.floats(min_value=0.5, max_value=2.0),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_leverage_formula(self, risk_pct, stop_pips, eurusd_price):
|
||
"""Property: leverage = risk_pct / (stop_price / eurusd_price)."""
|
||
stop_price = stop_pips * 0.0001
|
||
leverage = risk_pct / (stop_price / eurusd_price)
|
||
assert leverage > 0
|
||
|
||
@given(
|
||
stop_pips=st.floats(min_value=1.0, max_value=100.0),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_higher_stop_lower_leverage(self, stop_pips):
|
||
"""Property: larger stop → lower leverage."""
|
||
lev1 = 0.005 / (5 * 0.0001 / 1.10)
|
||
lev2 = 0.005 / (20 * 0.0001 / 1.10)
|
||
assert lev1 > lev2
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 14: OOS Properties
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestOOSProperties:
|
||
"""Property: out-of-sample split invariants."""
|
||
|
||
@given(
|
||
n_bars=st.integers(min_value=100, max_value=10000),
|
||
train_frac=st.floats(min_value=0.1, max_value=0.9),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_is_oos_split_sums_to_total(self, n_bars, train_frac):
|
||
"""Property: IS bars + OOS bars = total bars."""
|
||
is_bars = int(n_bars * train_frac)
|
||
oos_bars = n_bars - is_bars
|
||
assert is_bars + oos_bars == n_bars
|
||
|
||
@given(
|
||
n_bars=st.integers(min_value=100, max_value=10000),
|
||
train_frac=st.floats(min_value=0.1, max_value=0.9),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_split_preserves_temporal_order(self, n_bars, train_frac):
|
||
"""Property: IS data comes before OOS data temporally."""
|
||
is_bars = int(n_bars * train_frac)
|
||
assert is_bars < n_bars
|
||
assert n_bars - is_bars > 0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 15: Transaction Cost Properties
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestTransactionCostProperties:
|
||
"""Property: transaction cost invariants."""
|
||
|
||
@given(
|
||
cost_bps=st.floats(min_value=0.0, max_value=100.0),
|
||
position_change=st.floats(min_value=0.0, max_value=1.0),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_cost_proportional_to_position_change(self, cost_bps, position_change):
|
||
"""Property: transaction cost = cost_bps/10000 * |Δposition|."""
|
||
cost = cost_bps / 10000.0 * position_change
|
||
assert cost >= 0.0
|
||
|
||
@given(
|
||
cost_bps=st.floats(min_value=0.0, max_value=100.0),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_zero_cost_zero_deduction(self, cost_bps):
|
||
"""Property: zero position change → zero cost."""
|
||
cost = cost_bps / 10000.0 * 0.0
|
||
assert cost == 0.0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 16: MultiIndex DataFrame Properties
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestMultiIndexProperties:
|
||
"""Property: MultiIndex DataFrame invariants."""
|
||
|
||
@given(
|
||
n=st.integers(min_value=10, max_value=500),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_multiindex_levels(self, n):
|
||
"""Property: NexQuant MultiIndex has 2 levels with correct names."""
|
||
idx = pd.MultiIndex.from_arrays(
|
||
[pd.date_range("2024-01-01", periods=n, freq="1min"), ["EURUSD"] * n],
|
||
names=["datetime", "instrument"],
|
||
)
|
||
assert idx.nlevels == 2
|
||
assert idx.names == ["datetime", "instrument"]
|
||
|
||
@given(
|
||
n=st.integers(min_value=10, max_value=500),
|
||
)
|
||
@settings(max_examples=50, deadline=10000)
|
||
def test_xs_single_instrument_returns_dataframe(self, n):
|
||
"""Property: using xs on a MultiIndex for a single instrument returns DataFrame."""
|
||
idx = pd.MultiIndex.from_arrays(
|
||
[pd.date_range("2024-01-01", periods=n, freq="1min"), ["EURUSD"] * n],
|
||
names=["datetime", "instrument"],
|
||
)
|
||
df = pd.DataFrame({"close": np.random.randn(n) + 1.10}, index=idx)
|
||
result = df.xs("EURUSD", level="instrument")
|
||
assert isinstance(result, pd.DataFrame)
|
||
assert len(result) == n
|