960 lines
34 KiB
Python
960 lines
34 KiB
Python
"""Unit tests for strategy comparison module."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
from datetime import datetime
|
|
from unittest.mock import Mock, patch
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from openclaw.backtest.analyzer import BacktestResult, TradeRecord
|
|
from openclaw.comparison.comparator import ComparisonResult, StrategyComparator
|
|
from openclaw.comparison.metrics import (
|
|
ComparisonMetrics,
|
|
MetricFilter,
|
|
MultiObjectiveOptimizer,
|
|
RiskLevel,
|
|
)
|
|
from openclaw.comparison.report import ComparisonReport, ReportFormat, generate_quick_summary
|
|
from openclaw.comparison.statistical_tests import StatisticalTests
|
|
|
|
|
|
def create_test_backtest_result(
|
|
initial_capital: float = 100000.0,
|
|
final_capital: float = 110000.0,
|
|
total_trades: int = 10,
|
|
) -> BacktestResult:
|
|
"""Create a test BacktestResult with proper structure."""
|
|
now = datetime.now()
|
|
equity_curve = [initial_capital + (final_capital - initial_capital) * i / total_trades
|
|
for i in range(total_trades + 1)]
|
|
|
|
trades = []
|
|
for i in range(total_trades):
|
|
is_win = i % 2 == 0 # Alternate wins and losses
|
|
trade = TradeRecord(
|
|
entry_time=now,
|
|
exit_time=now,
|
|
side="long",
|
|
entry_price=100.0,
|
|
exit_price=110.0 if is_win else 90.0,
|
|
quantity=10.0,
|
|
pnl=100.0 if is_win else -50.0,
|
|
is_win=is_win,
|
|
)
|
|
trades.append(trade)
|
|
|
|
return BacktestResult(
|
|
initial_capital=initial_capital,
|
|
final_capital=final_capital,
|
|
equity_curve=equity_curve,
|
|
timestamps=[now] * len(equity_curve),
|
|
trades=trades,
|
|
start_time=now,
|
|
end_time=now,
|
|
)
|
|
|
|
|
|
class TestRiskLevel:
|
|
"""Tests for RiskLevel enum."""
|
|
|
|
def test_risk_level_values(self):
|
|
"""Test risk level enum values."""
|
|
assert RiskLevel.CONSERVATIVE.value == "conservative"
|
|
assert RiskLevel.MODERATE.value == "moderate"
|
|
assert RiskLevel.AGGRESSIVE.value == "aggressive"
|
|
assert RiskLevel.SPECULATIVE.value == "speculative"
|
|
|
|
|
|
class TestComparisonMetrics:
|
|
"""Tests for ComparisonMetrics dataclass."""
|
|
|
|
def test_default_values(self):
|
|
"""Test default metric values."""
|
|
metrics = ComparisonMetrics(strategy_name="test")
|
|
assert metrics.strategy_name == "test"
|
|
assert metrics.total_return == 0.0
|
|
assert metrics.sharpe_ratio == 0.0
|
|
assert metrics.max_drawdown == 0.0
|
|
assert metrics.win_rate == 0.0
|
|
|
|
def test_risk_level_conservative(self):
|
|
"""Test conservative risk level classification."""
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
max_drawdown=0.05,
|
|
volatility=0.10,
|
|
)
|
|
assert metrics.risk_level == RiskLevel.CONSERVATIVE
|
|
|
|
def test_risk_level_moderate(self):
|
|
"""Test moderate risk level classification."""
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
max_drawdown=0.10,
|
|
volatility=0.20,
|
|
)
|
|
assert metrics.risk_level == RiskLevel.MODERATE
|
|
|
|
def test_risk_level_aggressive(self):
|
|
"""Test aggressive risk level classification."""
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
max_drawdown=0.20,
|
|
volatility=0.30,
|
|
)
|
|
assert metrics.risk_level == RiskLevel.AGGRESSIVE
|
|
|
|
def test_risk_level_speculative(self):
|
|
"""Test speculative risk level classification."""
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
max_drawdown=0.30,
|
|
volatility=0.50,
|
|
)
|
|
assert metrics.risk_level == RiskLevel.SPECULATIVE
|
|
|
|
def test_risk_score_calculation(self):
|
|
"""Test risk score calculation."""
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
max_drawdown=0.25, # 50 score
|
|
volatility=0.30, # 60 score
|
|
var_95=-0.05, # 50 score
|
|
)
|
|
expected = 0.4 * 50 + 0.4 * 60 + 0.2 * 50 # 54.0
|
|
assert metrics.risk_score == pytest.approx(expected)
|
|
|
|
def test_risk_score_capped_at_100(self):
|
|
"""Test risk score is capped at 100."""
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
max_drawdown=0.60, # Would be 120, capped at 100
|
|
volatility=0.60, # Would be 120, capped at 100
|
|
)
|
|
assert metrics.risk_score <= 100.0
|
|
|
|
def test_return_risk_ratio(self):
|
|
"""Test return to risk ratio calculation."""
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
total_return=0.20,
|
|
max_drawdown=0.10,
|
|
volatility=0.15,
|
|
)
|
|
risk_score = metrics.risk_score
|
|
expected_ratio = 0.20 / risk_score
|
|
assert metrics.return_risk_ratio == pytest.approx(expected_ratio)
|
|
|
|
def test_return_risk_ratio_infinite(self):
|
|
"""Test return/risk ratio when risk is zero."""
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
total_return=0.10,
|
|
max_drawdown=0.0,
|
|
volatility=0.0,
|
|
var_95=0.0,
|
|
)
|
|
assert metrics.return_risk_ratio == float("inf")
|
|
|
|
def test_to_dict(self):
|
|
"""Test conversion to dictionary."""
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
total_return=0.10,
|
|
sharpe_ratio=1.5,
|
|
)
|
|
d = metrics.to_dict()
|
|
assert d["strategy_name"] == "test"
|
|
assert d["total_return"] == 0.10
|
|
assert d["total_return_pct"] == 10.0
|
|
assert d["sharpe_ratio"] == 1.5
|
|
assert "risk_level" in d
|
|
assert "risk_score" in d
|
|
|
|
def test_from_backtest_result(self):
|
|
"""Test creation from BacktestResult."""
|
|
result = create_test_backtest_result(
|
|
initial_capital=100000.0,
|
|
final_capital=115000.0,
|
|
total_trades=10,
|
|
)
|
|
metrics = ComparisonMetrics.from_backtest_result("test_strategy", result)
|
|
assert metrics.strategy_name == "test_strategy"
|
|
assert metrics.total_return == 0.15 # (115000 - 100000) / 100000
|
|
assert metrics.num_trades == 10
|
|
|
|
def test_from_backtest_result_with_trades(self):
|
|
"""Test creation from BacktestResult with trades."""
|
|
result = create_test_backtest_result(
|
|
initial_capital=100000.0,
|
|
final_capital=110000.0,
|
|
total_trades=4,
|
|
)
|
|
metrics = ComparisonMetrics.from_backtest_result("test", result)
|
|
# Trades: [100, -50, 100, -50] -> average = (100 - 50 + 100 - 50) / 4 = 25.0
|
|
assert metrics.avg_trade == 25.0
|
|
assert metrics.num_trades == 4
|
|
|
|
|
|
class TestMetricFilter:
|
|
"""Tests for MetricFilter class."""
|
|
|
|
def test_matches_all_criteria(self):
|
|
"""Test filter matching all criteria."""
|
|
filter_criteria = MetricFilter(
|
|
min_sharpe=1.0,
|
|
min_return=0.10,
|
|
max_drawdown=0.20,
|
|
)
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
sharpe_ratio=1.5,
|
|
total_return=0.15,
|
|
max_drawdown=0.15,
|
|
)
|
|
assert filter_criteria.matches(metrics) is True
|
|
|
|
def test_fails_min_sharpe(self):
|
|
"""Test filter fails on min_sharpe."""
|
|
filter_criteria = MetricFilter(min_sharpe=1.0)
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
sharpe_ratio=0.5,
|
|
)
|
|
assert filter_criteria.matches(metrics) is False
|
|
|
|
def test_fails_min_return(self):
|
|
"""Test filter fails on min_return."""
|
|
filter_criteria = MetricFilter(min_return=0.10)
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
total_return=0.05,
|
|
)
|
|
assert filter_criteria.matches(metrics) is False
|
|
|
|
def test_fails_max_drawdown(self):
|
|
"""Test filter fails on max_drawdown."""
|
|
filter_criteria = MetricFilter(max_drawdown=0.10)
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
max_drawdown=0.20,
|
|
)
|
|
assert filter_criteria.matches(metrics) is False
|
|
|
|
def test_fails_min_win_rate(self):
|
|
"""Test filter fails on min_win_rate."""
|
|
filter_criteria = MetricFilter(min_win_rate=0.50)
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
win_rate=0.40,
|
|
)
|
|
assert filter_criteria.matches(metrics) is False
|
|
|
|
def test_fails_min_profit_factor(self):
|
|
"""Test filter fails on min_profit_factor."""
|
|
filter_criteria = MetricFilter(min_profit_factor=1.5)
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
profit_factor=1.2,
|
|
)
|
|
assert filter_criteria.matches(metrics) is False
|
|
|
|
def test_fails_risk_levels(self):
|
|
"""Test filter fails on risk_levels."""
|
|
filter_criteria = MetricFilter(risk_levels=[RiskLevel.CONSERVATIVE])
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
max_drawdown=0.30, # speculative
|
|
volatility=0.50,
|
|
)
|
|
assert filter_criteria.matches(metrics) is False
|
|
|
|
def test_passes_risk_levels(self):
|
|
"""Test filter passes on risk_levels."""
|
|
filter_criteria = MetricFilter(
|
|
risk_levels=[RiskLevel.CONSERVATIVE, RiskLevel.MODERATE]
|
|
)
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
max_drawdown=0.05,
|
|
volatility=0.10,
|
|
)
|
|
assert filter_criteria.matches(metrics) is True
|
|
|
|
def test_fails_min_trades(self):
|
|
"""Test filter fails on min_trades."""
|
|
filter_criteria = MetricFilter(min_trades=50)
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
num_trades=30,
|
|
)
|
|
assert filter_criteria.matches(metrics) is False
|
|
|
|
def test_empty_filter_matches_all(self):
|
|
"""Test empty filter matches all metrics."""
|
|
filter_criteria = MetricFilter()
|
|
metrics = ComparisonMetrics(strategy_name="test")
|
|
assert filter_criteria.matches(metrics) is True
|
|
|
|
|
|
class TestMultiObjectiveOptimizer:
|
|
"""Tests for MultiObjectiveOptimizer class."""
|
|
|
|
def test_default_weights(self):
|
|
"""Test default optimizer weights."""
|
|
optimizer = MultiObjectiveOptimizer()
|
|
assert "return" in optimizer.weights
|
|
assert "sharpe" in optimizer.weights
|
|
assert "drawdown" in optimizer.weights
|
|
assert abs(sum(optimizer.weights.values()) - 1.0) < 0.01
|
|
|
|
def test_custom_weights(self):
|
|
"""Test custom optimizer weights."""
|
|
custom_weights = {"return": 0.5, "sharpe": 0.5}
|
|
optimizer = MultiObjectiveOptimizer(weights=custom_weights)
|
|
assert optimizer.weights == custom_weights
|
|
|
|
def test_score_calculation(self):
|
|
"""Test score calculation."""
|
|
optimizer = MultiObjectiveOptimizer()
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
total_return=0.25, # 50 score
|
|
sharpe_ratio=1.0, # 50 score
|
|
max_drawdown=0.25, # 50 score
|
|
win_rate=0.35, # 50 score
|
|
profit_factor=1.0, # 50 score
|
|
)
|
|
score = optimizer.score(metrics)
|
|
assert score > 0
|
|
assert score <= 100
|
|
|
|
def test_score_with_all_weights(self):
|
|
"""Test score with all weight types."""
|
|
weights = {
|
|
"return": 0.2,
|
|
"sharpe": 0.2,
|
|
"drawdown": 0.2,
|
|
"win_rate": 0.2,
|
|
"profit_factor": 0.2,
|
|
}
|
|
optimizer = MultiObjectiveOptimizer(weights=weights)
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
total_return=0.5,
|
|
sharpe_ratio=2.0,
|
|
max_drawdown=0.0,
|
|
win_rate=0.7,
|
|
profit_factor=2.0,
|
|
)
|
|
score = optimizer.score(metrics)
|
|
assert score == pytest.approx(100.0, rel=0.01)
|
|
|
|
def test_score_with_calmar(self):
|
|
"""Test score with Calmar weight."""
|
|
weights = {"calmar": 1.0}
|
|
optimizer = MultiObjectiveOptimizer(weights=weights)
|
|
metrics = ComparisonMetrics(
|
|
strategy_name="test",
|
|
calmar_ratio=3.0,
|
|
)
|
|
score = optimizer.score(metrics)
|
|
assert score == pytest.approx(100.0, rel=0.01)
|
|
|
|
def test_rank_strategies(self):
|
|
"""Test ranking strategies."""
|
|
optimizer = MultiObjectiveOptimizer()
|
|
metrics_list = [
|
|
ComparisonMetrics(strategy_name="low", total_return=0.10, sharpe_ratio=0.5),
|
|
ComparisonMetrics(strategy_name="high", total_return=0.30, sharpe_ratio=1.5),
|
|
ComparisonMetrics(strategy_name="mid", total_return=0.20, sharpe_ratio=1.0),
|
|
]
|
|
ranked = optimizer.rank(metrics_list)
|
|
assert len(ranked) == 3
|
|
assert ranked[0][0].strategy_name == "high"
|
|
assert ranked[-1][0].strategy_name == "low"
|
|
|
|
def test_select_best(self):
|
|
"""Test selecting top N strategies."""
|
|
optimizer = MultiObjectiveOptimizer()
|
|
metrics_list = [
|
|
ComparisonMetrics(strategy_name="low", total_return=0.10),
|
|
ComparisonMetrics(strategy_name="high", total_return=0.30),
|
|
ComparisonMetrics(strategy_name="mid", total_return=0.20),
|
|
]
|
|
best = optimizer.select_best(metrics_list, top_n=2)
|
|
assert len(best) == 2
|
|
assert best[0][0].strategy_name == "high"
|
|
|
|
|
|
class TestStatisticalTests:
|
|
"""Tests for StatisticalTests class."""
|
|
|
|
def test_t_test_equal_means(self):
|
|
"""Test t-test with equal means."""
|
|
tests = StatisticalTests()
|
|
returns1 = np.random.normal(0, 0.01, 100)
|
|
returns2 = np.random.normal(0, 0.01, 100)
|
|
t_stat, p_value = tests.t_test(returns1, returns2)
|
|
assert isinstance(t_stat, float)
|
|
assert isinstance(p_value, float)
|
|
assert 0 <= p_value <= 1
|
|
|
|
def test_t_test_different_means(self):
|
|
"""Test t-test with different means."""
|
|
tests = StatisticalTests()
|
|
returns1 = np.random.normal(0.001, 0.01, 100)
|
|
returns2 = np.random.normal(-0.001, 0.01, 100)
|
|
t_stat, p_value = tests.t_test(returns1, returns2)
|
|
# Should detect significant difference
|
|
assert p_value < 0.1 or t_stat > 1.0
|
|
|
|
def test_t_test_empty_arrays(self):
|
|
"""Test t-test with empty arrays."""
|
|
tests = StatisticalTests()
|
|
t_stat, p_value = tests.t_test(np.array([]), np.array([]))
|
|
assert t_stat == 0.0
|
|
assert p_value == 1.0
|
|
|
|
def test_paired_t_test(self):
|
|
"""Test paired t-test."""
|
|
tests = StatisticalTests()
|
|
returns1 = np.random.normal(0.001, 0.01, 100)
|
|
returns2 = returns1 + np.random.normal(0, 0.005, 100)
|
|
t_stat, p_value = tests.paired_t_test(returns1, returns2)
|
|
assert isinstance(t_stat, float)
|
|
assert isinstance(p_value, float)
|
|
|
|
def test_sharpe_difference_test(self):
|
|
"""Test Sharpe difference test."""
|
|
tests = StatisticalTests()
|
|
returns1 = np.random.normal(0.001, 0.01, 100)
|
|
returns2 = np.random.normal(0.0005, 0.01, 100)
|
|
z_stat, p_value = tests.sharpe_difference_test(returns1, returns2)
|
|
assert isinstance(z_stat, float)
|
|
assert isinstance(p_value, float)
|
|
|
|
def test_mann_whitney_u_test(self):
|
|
"""Test Mann-Whitney U test."""
|
|
tests = StatisticalTests()
|
|
returns1 = np.random.normal(0.001, 0.01, 50)
|
|
returns2 = np.random.normal(0, 0.01, 50)
|
|
u_stat, p_value = tests.mann_whitney_u_test(returns1, returns2)
|
|
assert isinstance(u_stat, float)
|
|
assert isinstance(p_value, float)
|
|
|
|
def test_kolmogorov_smirnov_test(self):
|
|
"""Test KS test."""
|
|
tests = StatisticalTests()
|
|
returns1 = np.random.normal(0, 0.01, 100)
|
|
returns2 = np.random.normal(0, 0.02, 100)
|
|
ks_stat, p_value = tests.kolmogorov_smirnov_test(returns1, returns2)
|
|
assert isinstance(ks_stat, float)
|
|
assert isinstance(p_value, float)
|
|
|
|
def test_levene_test(self):
|
|
"""Test Levene test for equal variances."""
|
|
tests = StatisticalTests()
|
|
returns1 = np.random.normal(0, 0.01, 100)
|
|
returns2 = np.random.normal(0, 0.02, 100)
|
|
w_stat, p_value = tests.levene_test(returns1, returns2)
|
|
assert isinstance(w_stat, float)
|
|
assert isinstance(p_value, float)
|
|
|
|
def test_jarque_bera_test_normal(self):
|
|
"""Test Jarque-Bera test with normal distribution."""
|
|
tests = StatisticalTests()
|
|
returns = np.random.normal(0, 0.01, 1000)
|
|
jb_stat, p_value = tests.jarque_bera_test(returns)
|
|
assert isinstance(jb_stat, float)
|
|
assert isinstance(p_value, float)
|
|
# Normal distribution should have high p-value
|
|
assert p_value > 0.01
|
|
|
|
def test_jarque_bera_test_non_normal(self):
|
|
"""Test Jarque-Bera test with non-normal distribution."""
|
|
tests = StatisticalTests()
|
|
returns = np.random.standard_t(3, 1000) * 0.01 # Fat tails
|
|
jb_stat, p_value = tests.jarque_bera_test(returns)
|
|
# Fat-tailed distribution should reject normality
|
|
assert p_value < 0.05 or jb_stat > 10
|
|
|
|
def test_is_normal_distribution(self):
|
|
"""Test normality check."""
|
|
tests = StatisticalTests()
|
|
normal_returns = np.random.normal(0, 0.01, 1000)
|
|
assert tests.is_normal_distribution(normal_returns) is True
|
|
|
|
def test_confidence_interval(self):
|
|
"""Test confidence interval calculation."""
|
|
tests = StatisticalTests()
|
|
returns = np.random.normal(0.001, 0.01, 100)
|
|
lower, upper = tests.calculate_confidence_interval(returns, confidence=0.95)
|
|
assert lower < upper
|
|
assert lower < np.mean(returns) < upper
|
|
|
|
def test_omega_ratio(self):
|
|
"""Test Omega ratio calculation."""
|
|
tests = StatisticalTests()
|
|
returns = np.array([0.01, -0.005, 0.02, -0.01, 0.015])
|
|
omega = tests.omega_ratio(returns, threshold=0)
|
|
assert omega > 0
|
|
|
|
def test_omega_ratio_no_losses(self):
|
|
"""Test Omega ratio with no losses."""
|
|
tests = StatisticalTests()
|
|
returns = np.array([0.01, 0.02, 0.015])
|
|
omega = tests.omega_ratio(returns)
|
|
assert omega == float("inf")
|
|
|
|
def test_calculate_drawdown_statistics(self):
|
|
"""Test drawdown statistics calculation."""
|
|
tests = StatisticalTests()
|
|
equity = np.array([100, 110, 105, 115, 100, 95, 110, 120])
|
|
dd_stats = tests.calculate_drawdown_statistics(equity)
|
|
assert "max_drawdown" in dd_stats
|
|
assert "avg_drawdown" in dd_stats
|
|
assert "max_drawdown_duration" in dd_stats
|
|
assert dd_stats["max_drawdown"] <= 0
|
|
|
|
def test_compare_drawdowns(self):
|
|
"""Test drawdown comparison."""
|
|
tests = StatisticalTests()
|
|
equity1 = np.array([100, 105, 102, 108, 110])
|
|
equity2 = np.array([100, 95, 90, 95, 100])
|
|
comparison = tests.compare_drawdowns(equity1, equity2)
|
|
assert "max_dd_diff" in comparison
|
|
assert "max_dd_ratio" in comparison
|
|
|
|
def test_calculate_information_ratio(self):
|
|
"""Test Information ratio calculation."""
|
|
tests = StatisticalTests()
|
|
returns = np.random.normal(0.001, 0.01, 100)
|
|
benchmark = np.random.normal(0.0005, 0.008, 100)
|
|
ir = tests.calculate_information_ratio(returns, benchmark)
|
|
assert isinstance(ir, float)
|
|
|
|
def test_calculate_beta(self):
|
|
"""Test beta calculation."""
|
|
tests = StatisticalTests()
|
|
market = np.random.normal(0.001, 0.01, 100)
|
|
returns = market * 1.2 + np.random.normal(0, 0.005, 100)
|
|
beta = tests.calculate_beta(returns, market)
|
|
assert beta > 0
|
|
# Beta should be close to 1.2
|
|
assert abs(beta - 1.2) < 0.5
|
|
|
|
def test_calculate_beta_empty(self):
|
|
"""Test beta with empty arrays."""
|
|
tests = StatisticalTests()
|
|
beta = tests.calculate_beta(np.array([]), np.array([]))
|
|
assert beta == 1.0
|
|
|
|
def test_calculate_alpha(self):
|
|
"""Test alpha calculation."""
|
|
tests = StatisticalTests()
|
|
market = np.random.normal(0.001, 0.01, 100)
|
|
returns = market + 0.0005 # Positive alpha
|
|
alpha = tests.calculate_alpha(returns, market)
|
|
assert isinstance(alpha, float)
|
|
|
|
|
|
class TestComparisonResult:
|
|
"""Tests for ComparisonResult class."""
|
|
|
|
def test_default_values(self):
|
|
"""Test default values."""
|
|
result = ComparisonResult()
|
|
assert result.metrics == []
|
|
assert result.best_strategy == ""
|
|
assert result.rankings == {}
|
|
assert result.recommendations == []
|
|
|
|
def test_get_metric_found(self):
|
|
"""Test getting existing metric."""
|
|
metric = ComparisonMetrics(strategy_name="test")
|
|
result = ComparisonResult(metrics=[metric])
|
|
found = result.get_metric("test")
|
|
assert found is not None
|
|
assert found.strategy_name == "test"
|
|
|
|
def test_get_metric_not_found(self):
|
|
"""Test getting non-existent metric."""
|
|
result = ComparisonResult()
|
|
found = result.get_metric("nonexistent")
|
|
assert found is None
|
|
|
|
def test_get_top_strategies(self):
|
|
"""Test getting top N strategies."""
|
|
metrics = [
|
|
ComparisonMetrics(strategy_name="low", total_return=0.10),
|
|
ComparisonMetrics(strategy_name="high", total_return=0.30),
|
|
ComparisonMetrics(strategy_name="mid", total_return=0.20),
|
|
]
|
|
result = ComparisonResult(metrics=metrics)
|
|
top = result.get_top_strategies(n=2)
|
|
assert len(top) == 2
|
|
assert top[0].strategy_name == "high"
|
|
|
|
def test_to_dict(self):
|
|
"""Test conversion to dictionary."""
|
|
metric = ComparisonMetrics(strategy_name="test", total_return=0.15)
|
|
result = ComparisonResult(
|
|
metrics=[metric],
|
|
best_strategy="test",
|
|
recommendations=["Good strategy"],
|
|
)
|
|
d = result.to_dict()
|
|
assert d["best_strategy"] == "test"
|
|
assert len(d["metrics"]) == 1
|
|
assert len(d["recommendations"]) == 1
|
|
|
|
|
|
class TestStrategyComparator:
|
|
"""Tests for StrategyComparator class."""
|
|
|
|
def test_initialization(self):
|
|
"""Test comparator initialization."""
|
|
mock_factory = Mock()
|
|
comparator = StrategyComparator(engine_factory=mock_factory, max_workers=2)
|
|
assert comparator.engine_factory == mock_factory
|
|
assert comparator.max_workers == 2
|
|
|
|
@patch("concurrent.futures.ThreadPoolExecutor")
|
|
def test_compare_strategies(self, mock_executor_class):
|
|
"""Test strategy comparison."""
|
|
# Mock the executor
|
|
mock_executor = Mock()
|
|
mock_executor_class.return_value.__enter__ = Mock(return_value=mock_executor)
|
|
mock_executor_class.return_value.__exit__ = Mock(return_value=False)
|
|
|
|
# Mock future results
|
|
mock_future = Mock()
|
|
mock_future.result.return_value = create_test_backtest_result(
|
|
initial_capital=100000.0,
|
|
final_capital=110000.0,
|
|
total_trades=10,
|
|
)
|
|
mock_executor.submit.return_value = mock_future
|
|
|
|
# Create comparator and run comparison
|
|
mock_engine = Mock()
|
|
mock_factory = Mock(return_value=mock_engine)
|
|
|
|
comparator = StrategyComparator(engine_factory=mock_factory)
|
|
strategies = {
|
|
"strategy1": lambda x: x,
|
|
}
|
|
data = np.array([1, 2, 3, 4])
|
|
|
|
result = comparator.compare(strategies, data)
|
|
|
|
assert isinstance(result, ComparisonResult)
|
|
assert len(result.metrics) == 1
|
|
assert result.metrics[0].strategy_name == "strategy1"
|
|
|
|
def test_calculate_rankings(self):
|
|
"""Test rankings calculation."""
|
|
mock_factory = Mock()
|
|
comparator = StrategyComparator(engine_factory=mock_factory)
|
|
|
|
metrics = [
|
|
ComparisonMetrics(strategy_name="high_return", total_return=0.30, sharpe_ratio=1.0),
|
|
ComparisonMetrics(strategy_name="low_return", total_return=0.10, sharpe_ratio=1.5),
|
|
]
|
|
|
|
rankings = comparator._calculate_rankings(metrics)
|
|
|
|
assert "total_return" in rankings
|
|
assert "sharpe_ratio" in rankings
|
|
assert rankings["total_return"][0] == "high_return"
|
|
assert rankings["sharpe_ratio"][0] == "low_return"
|
|
|
|
def test_select_best_strategy(self):
|
|
"""Test best strategy selection."""
|
|
mock_factory = Mock()
|
|
comparator = StrategyComparator(engine_factory=mock_factory)
|
|
|
|
metrics = [
|
|
ComparisonMetrics(strategy_name="best", total_return=0.30, sharpe_ratio=1.5),
|
|
ComparisonMetrics(strategy_name="worst", total_return=0.10, sharpe_ratio=0.5),
|
|
]
|
|
|
|
best = comparator._select_best_strategy(metrics)
|
|
assert best == "best"
|
|
|
|
def test_select_best_strategy_empty(self):
|
|
"""Test best strategy selection with empty list."""
|
|
mock_factory = Mock()
|
|
comparator = StrategyComparator(engine_factory=mock_factory)
|
|
best = comparator._select_best_strategy([])
|
|
assert best == ""
|
|
|
|
def test_generate_recommendations(self):
|
|
"""Test recommendation generation."""
|
|
mock_factory = Mock()
|
|
comparator = StrategyComparator(engine_factory=mock_factory)
|
|
|
|
metrics = [
|
|
ComparisonMetrics(strategy_name="high_return", total_return=0.30, sharpe_ratio=1.0),
|
|
ComparisonMetrics(strategy_name="low_risk", max_drawdown=0.05, volatility=0.08),
|
|
]
|
|
rankings = {
|
|
"total_return": ["high_return", "low_risk"],
|
|
}
|
|
|
|
recs = comparator._generate_recommendations(metrics, rankings)
|
|
|
|
assert len(recs) > 0
|
|
assert any("high_return" in rec for rec in recs)
|
|
|
|
def test_filter_strategies(self):
|
|
"""Test strategy filtering."""
|
|
mock_factory = Mock()
|
|
comparator = StrategyComparator(engine_factory=mock_factory)
|
|
|
|
metrics = [
|
|
ComparisonMetrics(strategy_name="good", sharpe_ratio=1.5),
|
|
ComparisonMetrics(strategy_name="bad", sharpe_ratio=0.5),
|
|
]
|
|
filter_criteria = MetricFilter(min_sharpe=1.0)
|
|
|
|
filtered = comparator.filter_strategies(metrics, filter_criteria)
|
|
|
|
assert len(filtered) == 1
|
|
assert filtered[0].strategy_name == "good"
|
|
|
|
def test_create_empty_result(self):
|
|
"""Test creating empty result for failed backtest."""
|
|
mock_factory = Mock()
|
|
comparator = StrategyComparator(engine_factory=mock_factory)
|
|
|
|
result = comparator._create_empty_result(initial_capital=100000.0)
|
|
|
|
assert result.initial_capital == 100000.0
|
|
assert result.final_capital == 100000.0
|
|
assert len(result.equity_curve) == 1
|
|
assert result.equity_curve[0] == 100000.0
|
|
assert result.trades == []
|
|
|
|
|
|
class TestComparisonReport:
|
|
"""Tests for ComparisonReport class."""
|
|
|
|
def test_default_initialization(self):
|
|
"""Test default initialization."""
|
|
report = ComparisonReport()
|
|
assert report.title == "Strategy Comparison Report"
|
|
assert report.include_charts is True
|
|
assert report.format == ReportFormat.MARKDOWN
|
|
|
|
def test_generate_markdown(self):
|
|
"""Test Markdown report generation."""
|
|
report = ComparisonReport()
|
|
metric = ComparisonMetrics(
|
|
strategy_name="test",
|
|
total_return=0.15,
|
|
sharpe_ratio=1.2,
|
|
)
|
|
result = ComparisonResult(
|
|
metrics=[metric],
|
|
best_strategy="test",
|
|
rankings={"total_return": ["test"]},
|
|
recommendations=["Good strategy"],
|
|
)
|
|
|
|
content = report.generate(result, format=ReportFormat.MARKDOWN)
|
|
|
|
assert "# Strategy Comparison Report" in content
|
|
assert "test" in content
|
|
assert "15.00%" in content or "0.15" in content
|
|
|
|
def test_generate_json(self):
|
|
"""Test JSON report generation."""
|
|
report = ComparisonReport()
|
|
metric = ComparisonMetrics(strategy_name="test", total_return=0.15)
|
|
result = ComparisonResult(metrics=[metric], best_strategy="test")
|
|
|
|
content = report.generate(result, format=ReportFormat.JSON)
|
|
|
|
data = json.loads(content)
|
|
assert data["title"] == "Strategy Comparison Report"
|
|
assert data["best_strategy"] == "test"
|
|
|
|
def test_generate_html(self):
|
|
"""Test HTML report generation."""
|
|
report = ComparisonReport()
|
|
metric = ComparisonMetrics(strategy_name="test", total_return=0.15)
|
|
result = ComparisonResult(metrics=[metric], best_strategy="test")
|
|
|
|
content = report.generate(result, format=ReportFormat.HTML)
|
|
|
|
assert "<html>" in content.lower()
|
|
assert "test" in content
|
|
|
|
def test_generate_csv(self):
|
|
"""Test CSV report generation."""
|
|
report = ComparisonReport()
|
|
metric = ComparisonMetrics(strategy_name="test", total_return=0.15)
|
|
result = ComparisonResult(metrics=[metric], best_strategy="test")
|
|
|
|
content = report.generate(result, format=ReportFormat.CSV)
|
|
|
|
assert "Strategy," in content
|
|
assert "test" in content
|
|
|
|
def test_save_report(self):
|
|
"""Test saving report to file."""
|
|
report = ComparisonReport()
|
|
metric = ComparisonMetrics(strategy_name="test", total_return=0.15)
|
|
result = ComparisonResult(metrics=[metric], best_strategy="test")
|
|
|
|
with tempfile.NamedTemporaryFile(mode='w', suffix='.md', delete=False) as f:
|
|
filepath = f.name
|
|
|
|
report.save(result, filepath)
|
|
|
|
with open(filepath, 'r') as f:
|
|
content = f.read()
|
|
assert "Strategy Comparison Report" in content
|
|
|
|
def test_infer_format_from_path(self):
|
|
"""Test format inference from file path."""
|
|
report = ComparisonReport()
|
|
|
|
assert report._infer_format_from_path("test.json") == ReportFormat.JSON
|
|
assert report._infer_format_from_path("test.html") == ReportFormat.HTML
|
|
assert report._infer_format_from_path("test.csv") == ReportFormat.CSV
|
|
assert report._infer_format_from_path("test.md") == ReportFormat.MARKDOWN
|
|
assert report._infer_format_from_path("test.txt") == ReportFormat.MARKDOWN
|
|
|
|
|
|
class TestQuickSummary:
|
|
"""Tests for quick summary function."""
|
|
|
|
def test_generate_quick_summary(self):
|
|
"""Test quick summary generation."""
|
|
metrics = [
|
|
ComparisonMetrics(strategy_name="best", total_return=0.30, sharpe_ratio=1.5),
|
|
ComparisonMetrics(strategy_name="worst", total_return=0.10, sharpe_ratio=0.5),
|
|
]
|
|
result = ComparisonResult(
|
|
metrics=metrics,
|
|
best_strategy="best",
|
|
)
|
|
|
|
summary = generate_quick_summary(result)
|
|
|
|
assert "Strategy Comparison Summary" in summary
|
|
assert "Total Strategies: 2" in summary
|
|
assert "Best Strategy: best" in summary
|
|
assert "best" in summary
|
|
|
|
def test_quick_summary_risk_analysis(self):
|
|
"""Test risk analysis in quick summary."""
|
|
metrics = [
|
|
ComparisonMetrics(strategy_name="conservative", max_drawdown=0.05, volatility=0.08),
|
|
ComparisonMetrics(strategy_name="speculative", max_drawdown=0.30, volatility=0.50),
|
|
]
|
|
result = ComparisonResult(metrics=metrics)
|
|
|
|
summary = generate_quick_summary(result)
|
|
|
|
assert "Risk Analysis:" in summary
|
|
assert "Conservative:" in summary
|
|
assert "Speculative:" in summary
|
|
|
|
|
|
class TestIntegration:
|
|
"""Integration tests for the comparison module."""
|
|
|
|
def test_full_comparison_workflow(self):
|
|
"""Test full comparison workflow."""
|
|
# Create comparison result
|
|
metrics = [
|
|
ComparisonMetrics(
|
|
strategy_name="momentum",
|
|
total_return=0.25,
|
|
sharpe_ratio=1.2,
|
|
max_drawdown=0.15,
|
|
win_rate=0.55,
|
|
profit_factor=1.5,
|
|
num_trades=100,
|
|
),
|
|
ComparisonMetrics(
|
|
strategy_name="mean_reversion",
|
|
total_return=0.15,
|
|
sharpe_ratio=1.0,
|
|
max_drawdown=0.10,
|
|
win_rate=0.60,
|
|
profit_factor=1.3,
|
|
num_trades=150,
|
|
),
|
|
]
|
|
|
|
result = ComparisonResult(
|
|
metrics=metrics,
|
|
best_strategy="momentum",
|
|
rankings={
|
|
"total_return": ["momentum", "mean_reversion"],
|
|
"sharpe_ratio": ["momentum", "mean_reversion"],
|
|
},
|
|
recommendations=[
|
|
"Highest return: momentum",
|
|
"Lowest risk: mean_reversion",
|
|
],
|
|
)
|
|
|
|
# Generate report
|
|
report = ComparisonReport()
|
|
markdown = report.generate(result, format=ReportFormat.MARKDOWN)
|
|
json_report = report.generate(result, format=ReportFormat.JSON)
|
|
|
|
assert "momentum" in markdown
|
|
assert "mean_reversion" in markdown
|
|
|
|
data = json.loads(json_report)
|
|
assert len(data["metrics"]) == 2
|
|
|
|
def test_filter_and_optimize_integration(self):
|
|
"""Test filter and optimizer integration."""
|
|
metrics = [
|
|
ComparisonMetrics(strategy_name="good", sharpe_ratio=1.5, total_return=0.20),
|
|
ComparisonMetrics(strategy_name="bad", sharpe_ratio=0.5, total_return=0.10),
|
|
ComparisonMetrics(strategy_name="excellent", sharpe_ratio=2.0, total_return=0.30),
|
|
]
|
|
|
|
# Filter strategies
|
|
filter_criteria = MetricFilter(min_sharpe=1.0)
|
|
filtered = [m for m in metrics if filter_criteria.matches(m)]
|
|
|
|
assert len(filtered) == 2
|
|
|
|
# Rank filtered strategies
|
|
optimizer = MultiObjectiveOptimizer()
|
|
ranked = optimizer.rank(filtered)
|
|
|
|
assert ranked[0][0].strategy_name == "excellent"
|
|
|
|
def test_statistical_tests_integration(self):
|
|
"""Test statistical tests integration."""
|
|
tests = StatisticalTests()
|
|
|
|
# Generate sample returns
|
|
returns1 = np.random.normal(0.001, 0.01, 100)
|
|
returns2 = np.random.normal(0.0005, 0.012, 100)
|
|
|
|
# Run tests
|
|
t_stat, p_value = tests.t_test(returns1, returns2)
|
|
sharpe_diff, sharpe_p = tests.sharpe_difference_test(returns1, returns2)
|
|
lower, upper = tests.calculate_confidence_interval(returns1)
|
|
|
|
# All results should be valid
|
|
assert isinstance(t_stat, float)
|
|
assert isinstance(sharpe_diff, float)
|
|
assert lower < upper
|