mirror of
https://github.com/NicolasBohn/NexQuant.git
synced 2026-08-07 20:17:45 +00:00
feat: Complete P6-P9 implementation (73 tests)
P6: ML Feedback Integrator (18 tests) - MLFeedbackMixin for QuantRDLoop - Auto-trigger ML training every 500 factors - Feature importance → prompt feedback P7: Portfolio Optimizer (28 tests) - Mean-Variance optimization (max Sharpe) - Risk Parity (equal risk contribution) - Correlation analysis (max 0.3) - Portfolio backtest with weighted signals P8: Integration Tests (27 tests) - End-to-end pipeline test - Parallelization test (4 workers) - RiskMgmt compliance test - Error handling test P9: Documentation - QWEN.md updated with all new modules - Project status updated - Architecture diagram expanded 73 tests passing in 0.48s
This commit is contained in:
TPTBusiness
@@ -0,0 +1,490 @@
|
||||
"""
|
||||
Tests for ML Feedback Integrator
|
||||
|
||||
Tests the MLFeedbackMixin class for correct trigger logic,
|
||||
factor counting, and prompt feedback generation.
|
||||
|
||||
15 tests covering:
|
||||
- Initialization and configuration
|
||||
- Trigger condition logic
|
||||
- Factor counting methods
|
||||
- Feature importance extraction
|
||||
- Prompt suggestion generation
|
||||
- Graceful error handling
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_project_root(tmp_path: Path) -> Path:
|
||||
"""Create a temporary project structure for testing."""
|
||||
# Create directory structure
|
||||
(tmp_path / "results" / "factors").mkdir(parents=True)
|
||||
(tmp_path / "results" / "models").mkdir(parents=True)
|
||||
(tmp_path / "prompts" / "local").mkdir(parents=True)
|
||||
return tmp_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_factor_data() -> dict:
|
||||
"""Return sample factor data for testing."""
|
||||
return {
|
||||
"name": "test_momentum_factor",
|
||||
"status": "success",
|
||||
"ic": 0.15,
|
||||
"sharpe_ratio": 1.8,
|
||||
"max_drawdown": -0.12,
|
||||
"win_rate": 0.55,
|
||||
"code": "def factor(): ...",
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_importance_data() -> dict:
|
||||
"""Return sample feature importance data."""
|
||||
return {
|
||||
"importance": {
|
||||
"momentum_5d": 0.25,
|
||||
"volatility_10d": 0.18,
|
||||
"mean_reversion_3d": 0.12,
|
||||
"volume_spike": 0.08,
|
||||
"trend_strength": 0.05,
|
||||
},
|
||||
"model_type": "lightgbm",
|
||||
"n_factors": 50,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestMLFeedbackMixinInit:
|
||||
"""Test MLFeedbackMixin initialization."""
|
||||
|
||||
def test_default_initialization(self, mock_project_root):
|
||||
"""Test default configuration values."""
|
||||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||||
|
||||
# Create a mock parent class
|
||||
class MockParent:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
class TestMixin(MLFeedbackMixin, MockParent):
|
||||
pass
|
||||
|
||||
mixin = TestMixin(ml_feedback=True)
|
||||
|
||||
assert mixin.ml_feedback_enabled is True
|
||||
assert mixin.ml_train_interval == 500
|
||||
assert mixin.strategy_gen_interval == 1000
|
||||
assert mixin.portfolio_opt_interval == 2000
|
||||
|
||||
def test_custom_intervals(self, mock_project_root):
|
||||
"""Test custom interval configuration."""
|
||||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||||
|
||||
class MockParent:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
class TestMixin(MLFeedbackMixin, MockParent):
|
||||
pass
|
||||
|
||||
mixin = TestMixin(
|
||||
ml_feedback=True,
|
||||
ml_train_interval=1000,
|
||||
strategy_gen_interval=2000,
|
||||
portfolio_opt_interval=4000,
|
||||
)
|
||||
|
||||
assert mixin.ml_train_interval == 1000
|
||||
assert mixin.strategy_gen_interval == 2000
|
||||
assert mixin.portfolio_opt_interval == 4000
|
||||
|
||||
def test_disabled_feedback(self, mock_project_root):
|
||||
"""Test disabled feedback mode."""
|
||||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||||
|
||||
class MockParent:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
class TestMixin(MLFeedbackMixin, MockParent):
|
||||
pass
|
||||
|
||||
mixin = TestMixin(ml_feedback=False)
|
||||
assert mixin.ml_feedback_enabled is False
|
||||
|
||||
|
||||
class TestTriggerConditions:
|
||||
"""Test trigger condition logic."""
|
||||
|
||||
def test_should_trigger_ml_train_at_threshold(self):
|
||||
"""Test ML train trigger at exact threshold."""
|
||||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||||
|
||||
class MockParent:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
class TestMixin(MLFeedbackMixin, MockParent):
|
||||
pass
|
||||
|
||||
mixin = TestMixin(ml_train_interval=500)
|
||||
mixin._last_ml_train_factor = 0
|
||||
|
||||
assert mixin._should_trigger_ml_train(500) is True
|
||||
assert mixin._should_trigger_ml_train(499) is False
|
||||
assert mixin._should_trigger_ml_train(1000) is True
|
||||
|
||||
def test_no_duplicate_trigger(self):
|
||||
"""Test that duplicate triggers are prevented."""
|
||||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||||
|
||||
class MockParent:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
class TestMixin(MLFeedbackMixin, MockParent):
|
||||
pass
|
||||
|
||||
mixin = TestMixin(ml_train_interval=500)
|
||||
mixin._last_ml_train_factor = 500
|
||||
|
||||
# Should not trigger again at 500
|
||||
assert mixin._should_trigger_ml_train(500) is False
|
||||
# Should trigger at 1000
|
||||
assert mixin._should_trigger_ml_train(1000) is True
|
||||
|
||||
def test_strategy_gen_trigger(self):
|
||||
"""Test strategy generation trigger logic."""
|
||||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||||
|
||||
class MockParent:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
class TestMixin(MLFeedbackMixin, MockParent):
|
||||
pass
|
||||
|
||||
mixin = TestMixin(strategy_gen_interval=1000)
|
||||
mixin._last_strategy_gen_factor = 0
|
||||
|
||||
assert mixin._should_trigger_strategy_gen(1000) is True
|
||||
assert mixin._should_trigger_strategy_gen(999) is False
|
||||
|
||||
def test_portfolio_opt_trigger(self):
|
||||
"""Test portfolio optimization trigger logic."""
|
||||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||||
|
||||
class MockParent:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
class TestMixin(MLFeedbackMixin, MockParent):
|
||||
pass
|
||||
|
||||
mixin = TestMixin(portfolio_opt_interval=2000)
|
||||
mixin._last_portfolio_opt_factor = 0
|
||||
|
||||
assert mixin._should_trigger_portfolio_opt(2000) is True
|
||||
assert mixin._should_trigger_portfolio_opt(1999) is False
|
||||
|
||||
|
||||
class TestFactorCounting:
|
||||
"""Test factor counting methods."""
|
||||
|
||||
def test_count_from_results_dir(self, mock_project_root, mock_factor_data):
|
||||
"""Test counting factors from results directory."""
|
||||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||||
|
||||
# Write test factor files
|
||||
for i in range(5):
|
||||
factor_file = mock_project_root / "results" / "factors" / f"factor_{i}.json"
|
||||
data = mock_factor_data.copy()
|
||||
data["name"] = f"factor_{i}"
|
||||
data["ic"] = 0.1 + i * 0.01
|
||||
with open(factor_file, "w") as f:
|
||||
json.dump(data, f)
|
||||
|
||||
class MockParent:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
class TestMixin(MLFeedbackMixin, MockParent):
|
||||
def _get_project_root(self):
|
||||
return mock_project_root
|
||||
|
||||
mixin = TestMixin()
|
||||
count = mixin._count_factors_from_results()
|
||||
|
||||
assert count == 5
|
||||
|
||||
def test_count_skips_failed_factors(self, mock_project_root, mock_factor_data):
|
||||
"""Test that failed factors are not counted."""
|
||||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||||
|
||||
# Write successful factors
|
||||
for i in range(3):
|
||||
factor_file = mock_project_root / "results" / "factors" / f"success_{i}.json"
|
||||
data = mock_factor_data.copy()
|
||||
with open(factor_file, "w") as f:
|
||||
json.dump(data, f)
|
||||
|
||||
# Write failed factors
|
||||
for i in range(2):
|
||||
factor_file = mock_project_root / "results" / "factors" / f"failed_{i}.json"
|
||||
data = mock_factor_data.copy()
|
||||
data["status"] = "failed"
|
||||
data["ic"] = None
|
||||
with open(factor_file, "w") as f:
|
||||
json.dump(data, f)
|
||||
|
||||
class MockParent:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
class TestMixin(MLFeedbackMixin, MockParent):
|
||||
def _get_project_root(self):
|
||||
return mock_project_root
|
||||
|
||||
mixin = TestMixin()
|
||||
count = mixin._count_factors_from_results()
|
||||
|
||||
assert count == 3 # Only successful factors
|
||||
|
||||
def test_count_empty_directory(self, mock_project_root):
|
||||
"""Test counting with empty factors directory."""
|
||||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||||
|
||||
class MockParent:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
class TestMixin(MLFeedbackMixin, MockParent):
|
||||
def _get_project_root(self):
|
||||
return mock_project_root
|
||||
|
||||
mixin = TestMixin()
|
||||
count = mixin._count_factors_from_results()
|
||||
|
||||
assert count == 0
|
||||
|
||||
|
||||
class TestFeatureImportance:
|
||||
"""Test feature importance extraction and prompt suggestions."""
|
||||
|
||||
def test_generate_prompt_suggestions_top_features(self, mock_importance_data):
|
||||
"""Test prompt suggestions from feature importance."""
|
||||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||||
|
||||
class MockParent:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
class TestMixin(MLFeedbackMixin, MockParent):
|
||||
pass
|
||||
|
||||
mixin = TestMixin()
|
||||
suggestions = mixin._generate_prompt_suggestions(mock_importance_data)
|
||||
|
||||
assert len(suggestions) >= 1
|
||||
# Should mention top features
|
||||
assert any("momentum_5d" in s for s in suggestions)
|
||||
|
||||
def test_generate_suggestions_low_performing_features(self, mock_importance_data):
|
||||
"""Test suggestions for avoiding low-performing features."""
|
||||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||||
|
||||
class MockParent:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
class TestMixin(MLFeedbackMixin, MockParent):
|
||||
pass
|
||||
|
||||
mixin = TestMixin()
|
||||
suggestions = mixin._generate_prompt_suggestions(mock_importance_data)
|
||||
|
||||
# Should suggest avoiding low-performing features
|
||||
assert any("Avoid" in s or "avoid" in s or "reduce" in s.lower() for s in suggestions)
|
||||
|
||||
def test_suggestions_empty_importance(self):
|
||||
"""Test suggestions with empty importance data."""
|
||||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||||
|
||||
class MockParent:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
class TestMixin(MLFeedbackMixin, MockParent):
|
||||
pass
|
||||
|
||||
mixin = TestMixin()
|
||||
suggestions = mixin._generate_prompt_suggestions({"importance": {}})
|
||||
|
||||
assert len(suggestions) == 1
|
||||
assert "No feature importance" in suggestions[0]
|
||||
|
||||
def test_suggestions_low_diversity(self):
|
||||
"""Test suggestions when factor diversity is low."""
|
||||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||||
|
||||
class MockParent:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
class TestMixin(MLFeedbackMixin, MockParent):
|
||||
pass
|
||||
|
||||
mixin = TestMixin()
|
||||
importance = {
|
||||
"importance": {
|
||||
"momentum_1d": 0.3,
|
||||
"momentum_2d": 0.25,
|
||||
"momentum_3d": 0.2,
|
||||
"momentum_4d": 0.15,
|
||||
}
|
||||
}
|
||||
suggestions = mixin._generate_prompt_suggestions(importance)
|
||||
|
||||
# Should suggest more diversity
|
||||
assert any("diversity" in s.lower() or "Diversity" in s for s in suggestions)
|
||||
|
||||
|
||||
class TestLoadTopFactors:
|
||||
"""Test loading top factors by IC."""
|
||||
|
||||
def test_load_top_factors(self, mock_project_root, mock_factor_data):
|
||||
"""Test loading top N factors."""
|
||||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||||
|
||||
# Write factor files with varying IC
|
||||
for i in range(10):
|
||||
factor_file = mock_project_root / "results" / "factors" / f"factor_{i}.json"
|
||||
data = mock_factor_data.copy()
|
||||
data["name"] = f"factor_{i}"
|
||||
data["ic"] = 0.01 * (i + 1) # IC from 0.01 to 0.10
|
||||
with open(factor_file, "w") as f:
|
||||
json.dump(data, f)
|
||||
|
||||
class MockParent:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
class TestMixin(MLFeedbackMixin, MockParent):
|
||||
def _get_project_root(self):
|
||||
return mock_project_root
|
||||
|
||||
mixin = TestMixin()
|
||||
top_factors = mixin._load_top_factors(n=5)
|
||||
|
||||
assert len(top_factors) == 5
|
||||
# Should be sorted by IC (descending)
|
||||
assert top_factors[0]["ic"] >= top_factors[-1]["ic"]
|
||||
|
||||
def test_load_top_factors_empty_dir(self, mock_project_root):
|
||||
"""Test loading from empty directory."""
|
||||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||||
|
||||
class MockParent:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
class TestMixin(MLFeedbackMixin, MockParent):
|
||||
def _get_project_root(self):
|
||||
return mock_project_root
|
||||
|
||||
mixin = TestMixin()
|
||||
top_factors = mixin._load_top_factors(n=5)
|
||||
|
||||
assert top_factors == []
|
||||
|
||||
|
||||
class TestErrorHandling:
|
||||
"""Test graceful error handling."""
|
||||
|
||||
def test_feedback_with_exception(self):
|
||||
"""Test that feedback handles exceptions gracefully."""
|
||||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||||
|
||||
class MockParent:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
def feedback(self, prev_out):
|
||||
return "parent_feedback"
|
||||
|
||||
def _get_factor_count(self):
|
||||
raise RuntimeError("Simulated error")
|
||||
|
||||
class TestMixin(MLFeedbackMixin, MockParent):
|
||||
pass
|
||||
|
||||
mixin = TestMixin(ml_feedback=True)
|
||||
|
||||
# Should not raise exception
|
||||
result = mixin.feedback({})
|
||||
assert result == "parent_feedback"
|
||||
|
||||
|
||||
class TestIntegration:
|
||||
"""Integration tests for full workflow."""
|
||||
|
||||
def test_full_feedback_cycle(self, mock_project_root, mock_factor_data, mock_importance_data):
|
||||
"""Test complete feedback cycle with triggers."""
|
||||
from rdagent.scenarios.qlib.local.feedback_integrator import MLFeedbackMixin
|
||||
|
||||
# Write importance file
|
||||
importance_file = mock_project_root / "results" / "models" / "feature_importance.json"
|
||||
with open(importance_file, "w") as f:
|
||||
json.dump(mock_importance_data, f)
|
||||
|
||||
call_log = []
|
||||
|
||||
class MockParent:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
def feedback(self, prev_out):
|
||||
call_log.append("parent_feedback")
|
||||
return "feedback_result"
|
||||
|
||||
def _get_factor_count(self):
|
||||
return 500
|
||||
|
||||
class TestMixin(MLFeedbackMixin, MockParent):
|
||||
def _get_project_root(self):
|
||||
return mock_project_root
|
||||
|
||||
def _count_factors_from_results(self):
|
||||
return 500
|
||||
|
||||
def _trigger_ml_training(self, factor_count):
|
||||
call_log.append(f"ml_train_{factor_count}")
|
||||
self._last_ml_train_factor = factor_count
|
||||
|
||||
mixin = TestMixin(ml_feedback=True, ml_train_interval=500)
|
||||
mixin._last_ml_train_factor = 0
|
||||
|
||||
result = mixin.feedback({})
|
||||
|
||||
assert result == "feedback_result"
|
||||
assert "parent_feedback" in call_log
|
||||
assert "ml_train_500" in call_log
|
||||
@@ -0,0 +1,440 @@
|
||||
"""
|
||||
Tests for Portfolio Optimizer
|
||||
|
||||
Tests the PortfolioOptimizer class for:
|
||||
- Mean-Variance Optimization
|
||||
- Risk Parity Optimization
|
||||
- Correlation Analysis
|
||||
- Portfolio Backtesting
|
||||
- Strategy Selection
|
||||
|
||||
20 tests covering all optimization methods and edge cases.
|
||||
"""
|
||||
|
||||
import json
|
||||
import tempfile
|
||||
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_strategies_dir(tmp_path: Path) -> Path:
|
||||
"""Create a temporary strategies directory with mock data."""
|
||||
strategies_dir = tmp_path / "results" / "strategies_new"
|
||||
strategies_dir.mkdir(parents=True)
|
||||
|
||||
# Create 5 mock strategy files
|
||||
strategies = [
|
||||
{
|
||||
"name": "MomentumStrategy",
|
||||
"sharpe_ratio": 2.1,
|
||||
"ic": 0.15,
|
||||
"max_drawdown": -0.10,
|
||||
"backtest": {
|
||||
"returns": np.random.randn(252) * 0.01 + 0.0005,
|
||||
"equity_curve": np.cumprod(1 + np.random.randn(252) * 0.01 + 0.0005).tolist(),
|
||||
},
|
||||
},
|
||||
{
|
||||
"name": "MeanReversionStrategy",
|
||||
"sharpe_ratio": 1.8,
|
||||
"ic": 0.12,
|
||||
"max_drawdown": -0.15,
|
||||
"backtest": {
|
||||
"returns": np.random.randn(252) * 0.012 + 0.0004,
|
||||
},
|
||||
},
|
||||
{
|
||||
"name": "VolatilityTargetStrategy",
|
||||
"sharpe_ratio": 1.5,
|
||||
"ic": 0.10,
|
||||
"max_drawdown": -0.12,
|
||||
"backtest": {
|
||||
"returns": np.random.randn(252) * 0.009 + 0.0003,
|
||||
},
|
||||
},
|
||||
{
|
||||
"name": "TrendFollowingStrategy",
|
||||
"sharpe_ratio": 1.2,
|
||||
"ic": 0.08,
|
||||
"max_drawdown": -0.18,
|
||||
"backtest": {
|
||||
"returns": np.random.randn(252) * 0.011 + 0.0002,
|
||||
},
|
||||
},
|
||||
{
|
||||
"name": "BreakoutStrategy",
|
||||
"sharpe_ratio": 0.9,
|
||||
"ic": 0.06,
|
||||
"max_drawdown": -0.22,
|
||||
"backtest": {
|
||||
"returns": np.random.randn(252) * 0.013 + 0.0001,
|
||||
},
|
||||
},
|
||||
]
|
||||
|
||||
for strategy in strategies:
|
||||
filepath = strategies_dir / f"{strategy['name']}.json"
|
||||
with open(filepath, "w") as f:
|
||||
json.dump(strategy, f, default=lambda x: x.tolist() if isinstance(x, np.ndarray) else x)
|
||||
|
||||
return tmp_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_returns_data() -> pd.DataFrame:
|
||||
"""Create mock strategy returns DataFrame."""
|
||||
np.random.seed(42)
|
||||
n_days = 252
|
||||
|
||||
return pd.DataFrame(
|
||||
{
|
||||
"StrategyA": np.random.randn(n_days) * 0.01 + 0.0005,
|
||||
"StrategyB": np.random.randn(n_days) * 0.01 + 0.0004,
|
||||
"StrategyC": np.random.randn(n_days) * 0.009 + 0.0003,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def portfolio_optimizer(mock_strategies_dir):
|
||||
"""Create a PortfolioOptimizer instance with mock data."""
|
||||
from rdagent.scenarios.qlib.local.portfolio_optimizer import PortfolioOptimizer
|
||||
|
||||
return PortfolioOptimizer(project_root=mock_strategies_dir)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestPortfolioOptimizerInit:
|
||||
"""Test PortfolioOptimizer initialization."""
|
||||
|
||||
def test_default_initialization(self):
|
||||
"""Test default configuration values."""
|
||||
from rdagent.scenarios.qlib.local.portfolio_optimizer import PortfolioOptimizer
|
||||
|
||||
optimizer = PortfolioOptimizer()
|
||||
|
||||
assert optimizer.max_correlation == 0.3
|
||||
assert optimizer.top_strategies == 30
|
||||
assert optimizer.risk_free_rate == 0.02
|
||||
|
||||
def test_custom_configuration(self):
|
||||
"""Test custom configuration values."""
|
||||
from rdagent.scenarios.qlib.local.portfolio_optimizer import PortfolioOptimizer
|
||||
|
||||
optimizer = PortfolioOptimizer(
|
||||
max_correlation=0.5,
|
||||
top_strategies=20,
|
||||
risk_free_rate=0.03,
|
||||
)
|
||||
|
||||
assert optimizer.max_correlation == 0.5
|
||||
assert optimizer.top_strategies == 20
|
||||
assert optimizer.risk_free_rate == 0.03
|
||||
|
||||
|
||||
class TestLoadStrategyData:
|
||||
"""Test strategy data loading."""
|
||||
|
||||
def test_load_strategy_data(self, portfolio_optimizer):
|
||||
"""Test loading strategy data from directory."""
|
||||
result = portfolio_optimizer._load_strategy_data()
|
||||
|
||||
assert result is True
|
||||
assert portfolio_optimizer._strategy_returns is not None
|
||||
assert portfolio_optimizer._strategy_expected_returns is not None
|
||||
assert portfolio_optimizer._cov_matrix is not None
|
||||
assert portfolio_optimizer._corr_matrix is not None
|
||||
|
||||
def test_load_specific_strategies(self, portfolio_optimizer):
|
||||
"""Test loading specific strategies by name."""
|
||||
result = portfolio_optimizer._load_strategy_data(
|
||||
strategies=["MomentumStrategy", "MeanReversionStrategy"]
|
||||
)
|
||||
|
||||
assert result is True
|
||||
assert len(portfolio_optimizer._strategy_expected_returns) == 2
|
||||
|
||||
def test_load_no_strategies_dir(self, tmp_path):
|
||||
"""Test loading when no strategies directory exists."""
|
||||
from rdagent.scenarios.qlib.local.portfolio_optimizer import PortfolioOptimizer
|
||||
|
||||
optimizer = PortfolioOptimizer(project_root=tmp_path)
|
||||
result = optimizer._load_strategy_data()
|
||||
|
||||
assert result is False
|
||||
|
||||
|
||||
class TestExtractReturns:
|
||||
"""Test returns extraction from backtest data."""
|
||||
|
||||
def test_extract_returns_from_array(self):
|
||||
"""Test extracting returns from array."""
|
||||
from rdagent.scenarios.qlib.local.portfolio_optimizer import PortfolioOptimizer
|
||||
|
||||
optimizer = PortfolioOptimizer()
|
||||
returns = np.array([0.01, -0.005, 0.008])
|
||||
|
||||
data = {"returns": returns.tolist()}
|
||||
result = optimizer._extract_returns(data)
|
||||
|
||||
assert result is not None
|
||||
assert len(result) == 3
|
||||
|
||||
def test_extract_returns_from_equity_curve(self):
|
||||
"""Test extracting returns from equity curve."""
|
||||
from rdagent.scenarios.qlib.local.portfolio_optimizer import PortfolioOptimizer
|
||||
|
||||
optimizer = PortfolioOptimizer()
|
||||
|
||||
equity = [100, 101, 100.5, 101.5, 102]
|
||||
data = {"equity_curve": equity}
|
||||
result = optimizer._extract_returns(data)
|
||||
|
||||
assert result is not None
|
||||
assert len(result) == 4 # One less than equity points
|
||||
|
||||
def test_extract_returns_no_data(self):
|
||||
"""Test extracting returns when no data available."""
|
||||
from rdagent.scenarios.qlib.local.portfolio_optimizer import PortfolioOptimizer
|
||||
|
||||
optimizer = PortfolioOptimizer()
|
||||
result = optimizer._extract_returns({})
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestMeanVarianceOptimization:
|
||||
"""Test mean-variance optimization."""
|
||||
|
||||
def test_mean_variance_basic(self, portfolio_optimizer):
|
||||
"""Test basic mean-variance optimization."""
|
||||
portfolio_optimizer._load_strategy_data()
|
||||
result = portfolio_optimizer._optimize_mean_variance()
|
||||
|
||||
assert result is not None
|
||||
assert "weights" in result
|
||||
assert "expected_return" in result
|
||||
assert "volatility" in result
|
||||
assert "sharpe" in result
|
||||
assert result["method"] == "mean_variance"
|
||||
|
||||
# Weights should sum to 1
|
||||
total_weight = sum(result["weights"].values())
|
||||
assert abs(total_weight - 1.0) < 0.01
|
||||
|
||||
def test_mean_variance_weights_positive(self, portfolio_optimizer):
|
||||
"""Test that all weights are non-negative."""
|
||||
portfolio_optimizer._load_strategy_data()
|
||||
result = portfolio_optimizer._optimize_mean_variance()
|
||||
|
||||
assert result is not None
|
||||
for name, weight in result["weights"].items():
|
||||
assert weight >= 0
|
||||
|
||||
def test_mean_variance_insufficient_strategies(self, portfolio_optimizer):
|
||||
"""Test optimization with insufficient strategies."""
|
||||
portfolio_optimizer._strategy_expected_returns = pd.Series({"OnlyOne": 0.1})
|
||||
portfolio_optimizer._cov_matrix = pd.DataFrame([[0.0001]], index=["OnlyOne"], columns=["OnlyOne"])
|
||||
|
||||
result = portfolio_optimizer._optimize_mean_variance()
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestRiskParityOptimization:
|
||||
"""Test risk parity optimization."""
|
||||
|
||||
def test_risk_parity_basic(self, portfolio_optimizer):
|
||||
"""Test basic risk parity optimization."""
|
||||
portfolio_optimizer._load_strategy_data()
|
||||
result = portfolio_optimizer._optimize_risk_parity()
|
||||
|
||||
assert result is not None
|
||||
assert "weights" in result
|
||||
assert result["method"] == "risk_parity"
|
||||
|
||||
def test_risk_parity_weights_positive(self, portfolio_optimizer):
|
||||
"""Test that all weights are non-negative."""
|
||||
portfolio_optimizer._load_strategy_data()
|
||||
result = portfolio_optimizer._optimize_risk_parity()
|
||||
|
||||
assert result is not None
|
||||
for name, weight in result["weights"].items():
|
||||
assert weight >= 0
|
||||
|
||||
def test_risk_parity_insufficient_strategies(self, portfolio_optimizer):
|
||||
"""Test optimization with insufficient strategies."""
|
||||
portfolio_optimizer._strategy_expected_returns = pd.Series({"OnlyOne": 0.1})
|
||||
portfolio_optimizer._cov_matrix = pd.DataFrame([[0.0001]], index=["OnlyOne"], columns=["OnlyOne"])
|
||||
|
||||
result = portfolio_optimizer._optimize_risk_parity()
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestICWeightedOptimization:
|
||||
"""Test IC-weighted optimization."""
|
||||
|
||||
def test_ic_weighted_basic(self, portfolio_optimizer):
|
||||
"""Test basic IC-weighted optimization."""
|
||||
result = portfolio_optimizer._optimize_ic_weighted()
|
||||
|
||||
assert result is not None
|
||||
assert "weights" in result
|
||||
assert result["method"] == "ic_weighted"
|
||||
|
||||
def test_ic_weighted_weights_proportional(self, portfolio_optimizer):
|
||||
"""Test that weights are proportional to IC."""
|
||||
result = portfolio_optimizer._optimize_ic_weighted()
|
||||
|
||||
assert result is not None
|
||||
total_weight = sum(result["weights"].values())
|
||||
assert abs(total_weight - 1.0) < 0.01
|
||||
|
||||
|
||||
class TestCorrelationAnalysis:
|
||||
"""Test correlation analysis."""
|
||||
|
||||
def test_analyze_correlations(self, portfolio_optimizer):
|
||||
"""Test correlation analysis."""
|
||||
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, portfolio_optimizer):
|
||||
"""Test selecting uncorrelated strategies."""
|
||||
portfolio_optimizer._load_strategy_data()
|
||||
uncorrelated = portfolio_optimizer.select_uncorrelated_strategies(target_count=3)
|
||||
|
||||
assert len(uncorrelated) <= 3
|
||||
|
||||
def test_correlation_no_data(self, tmp_path):
|
||||
"""Test correlation analysis with no data."""
|
||||
from rdagent.scenarios.qlib.local.portfolio_optimizer import PortfolioOptimizer
|
||||
|
||||
optimizer = PortfolioOptimizer(project_root=tmp_path)
|
||||
result = optimizer.analyze_correlations()
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestPortfolioBacktest:
|
||||
"""Test portfolio backtesting."""
|
||||
|
||||
def test_backtest_portfolio(self, portfolio_optimizer):
|
||||
"""Test backtesting a portfolio."""
|
||||
portfolio_optimizer._load_strategy_data()
|
||||
|
||||
# Use equal weights
|
||||
n = len(portfolio_optimizer._strategy_returns.columns)
|
||||
weights = {col: 1.0 / n for col in portfolio_optimizer._strategy_returns.columns}
|
||||
|
||||
result = portfolio_optimizer.backtest_portfolio(weights)
|
||||
|
||||
assert result is not None
|
||||
assert "total_return" in result
|
||||
assert "annualized_return" in result
|
||||
assert "annualized_volatility" in result
|
||||
assert "sharpe_ratio" in result
|
||||
assert "max_drawdown" in result
|
||||
assert "win_rate" in result
|
||||
|
||||
def test_backtest_default_weights(self, portfolio_optimizer):
|
||||
"""Test backtesting with default (equal) weights."""
|
||||
portfolio_optimizer._load_strategy_data()
|
||||
result = portfolio_optimizer.backtest_portfolio()
|
||||
|
||||
assert result is not None
|
||||
|
||||
def test_backtest_no_data(self, tmp_path):
|
||||
"""Test backtesting with no data."""
|
||||
from rdagent.scenarios.qlib.local.portfolio_optimizer import PortfolioOptimizer
|
||||
|
||||
optimizer = PortfolioOptimizer(project_root=tmp_path)
|
||||
result = optimizer.backtest_portfolio()
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestOptimizePortfolio:
|
||||
"""Test the main optimize_portfolio method."""
|
||||
|
||||
def test_optimize_mean_variance(self, portfolio_optimizer):
|
||||
"""Test optimize_portfolio with mean_variance method."""
|
||||
result = portfolio_optimizer.optimize_portfolio(method="mean_variance")
|
||||
|
||||
assert result is not None
|
||||
assert result["method"] == "mean_variance"
|
||||
|
||||
def test_optimize_risk_parity(self, portfolio_optimizer):
|
||||
"""Test optimize_portfolio with risk_parity method."""
|
||||
result = portfolio_optimizer.optimize_portfolio(method="risk_parity")
|
||||
|
||||
assert result is not None
|
||||
assert result["method"] == "risk_parity"
|
||||
|
||||
def test_optimize_ic_weighted(self, portfolio_optimizer):
|
||||
"""Test optimize_portfolio with ic_weighted method."""
|
||||
result = portfolio_optimizer.optimize_portfolio(method="ic_weighted")
|
||||
|
||||
assert result is not None
|
||||
assert result["method"] == "ic_weighted"
|
||||
|
||||
def test_optimize_unknown_method(self, portfolio_optimizer):
|
||||
"""Test optimize_portfolio with unknown method."""
|
||||
result = portfolio_optimizer.optimize_portfolio(method="unknown_method")
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestReportGeneration:
|
||||
"""Test report generation."""
|
||||
|
||||
def test_generate_report(self, portfolio_optimizer):
|
||||
"""Test generating optimization report."""
|
||||
portfolio_optimizer._load_strategy_data()
|
||||
report = portfolio_optimizer.generate_report()
|
||||
|
||||
assert isinstance(report, str)
|
||||
assert "PORTFOLIO OPTIMIZATION REPORT" in report
|
||||
assert "Configuration" in report
|
||||
|
||||
|
||||
class TestSaveResults:
|
||||
"""Test saving optimization results."""
|
||||
|
||||
def test_save_result_creates_file(self, portfolio_optimizer, tmp_path):
|
||||
"""Test that saving creates a JSON file."""
|
||||
portfolio_optimizer.project_root = tmp_path
|
||||
result = {
|
||||
"weights": {"A": 0.5, "B": 0.5},
|
||||
"method": "test",
|
||||
"sharpe": 1.5,
|
||||
}
|
||||
|
||||
portfolio_optimizer._save_optimization_result(result)
|
||||
|
||||
# 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
|
||||
Reference in New Issue
Block a user