586 lines
22 KiB
Python
586 lines
22 KiB
Python
"""Unit tests for backtest performance analyzer.
|
|
|
|
Tests the PerformanceAnalyzer class and its various metric calculations.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from openclaw.backtest.analyzer import (
|
|
BacktestResult,
|
|
PerformanceAnalyzer,
|
|
TradeRecord,
|
|
)
|
|
|
|
|
|
class TestPerformanceAnalyzer:
|
|
"""Test suite for PerformanceAnalyzer."""
|
|
|
|
@pytest.fixture
|
|
def analyzer(self):
|
|
"""Create a PerformanceAnalyzer instance."""
|
|
return PerformanceAnalyzer()
|
|
|
|
@pytest.fixture
|
|
def sample_equity_curve(self):
|
|
"""Create a sample equity curve for testing."""
|
|
# Start with 10000, grow to 15000 with some volatility
|
|
np.random.seed(42)
|
|
returns = np.random.normal(0.0005, 0.02, 252) # Daily returns
|
|
equity = [10000.0]
|
|
for r in returns:
|
|
equity.append(equity[-1] * (1 + r))
|
|
return equity
|
|
|
|
@pytest.fixture
|
|
def sample_timestamps(self, sample_equity_curve):
|
|
"""Create sample timestamps matching equity curve length."""
|
|
start = datetime(2023, 1, 1)
|
|
return [start + timedelta(days=i) for i in range(len(sample_equity_curve))]
|
|
|
|
@pytest.fixture
|
|
def sample_trades(self):
|
|
"""Create sample trades for testing."""
|
|
base_time = datetime(2023, 1, 1)
|
|
return [
|
|
TradeRecord(
|
|
entry_time=base_time + timedelta(days=i),
|
|
exit_time=base_time + timedelta(days=i + 5),
|
|
side="long",
|
|
entry_price=100.0,
|
|
exit_price=110.0 if i % 2 == 0 else 95.0,
|
|
quantity=10.0,
|
|
pnl=100.0 if i % 2 == 0 else -50.0,
|
|
is_win=i % 2 == 0,
|
|
)
|
|
for i in range(20)
|
|
]
|
|
|
|
@pytest.fixture
|
|
def sample_backtest_result(self, sample_equity_curve, sample_timestamps, sample_trades):
|
|
"""Create a complete backtest result."""
|
|
return BacktestResult(
|
|
initial_capital=10000.0,
|
|
final_capital=sample_equity_curve[-1],
|
|
equity_curve=sample_equity_curve,
|
|
timestamps=sample_timestamps,
|
|
trades=sample_trades,
|
|
start_time=sample_timestamps[0],
|
|
end_time=sample_timestamps[-1],
|
|
)
|
|
|
|
class TestReturns:
|
|
"""Tests for return calculations."""
|
|
|
|
def test_calculate_returns_basic(self, analyzer):
|
|
"""Test basic return calculation."""
|
|
equity = [100.0, 110.0, 121.0]
|
|
returns = analyzer.calculate_returns(equity)
|
|
|
|
assert len(returns) == 2
|
|
assert np.isclose(returns[0], 0.10) # 10% return
|
|
assert np.isclose(returns[1], 0.10) # 10% return
|
|
|
|
def test_calculate_returns_empty(self, analyzer):
|
|
"""Test return calculation with empty curve."""
|
|
returns = analyzer.calculate_returns([])
|
|
assert len(returns) == 0
|
|
|
|
def test_calculate_returns_single_point(self, analyzer):
|
|
"""Test return calculation with single point."""
|
|
returns = analyzer.calculate_returns([100.0])
|
|
assert len(returns) == 0
|
|
|
|
def test_calculate_total_return(self, analyzer):
|
|
"""Test total return calculation."""
|
|
equity = [100.0, 110.0, 121.0]
|
|
total_return = analyzer.calculate_total_return(equity)
|
|
|
|
assert np.isclose(total_return, 0.21) # 21% total return
|
|
|
|
def test_calculate_total_return_empty(self, analyzer):
|
|
"""Test total return with empty curve."""
|
|
total_return = analyzer.calculate_total_return([])
|
|
assert total_return == 0.0
|
|
|
|
def test_calculate_annualized_return(self, analyzer):
|
|
"""Test annualized return calculation."""
|
|
start = datetime(2023, 1, 1)
|
|
timestamps = [start + timedelta(days=i) for i in range(366)]
|
|
equity = [100.0, 121.0] + [121.0] * 364 # 21% return over 1 year
|
|
|
|
annualized = analyzer.calculate_annualized_return(equity, timestamps)
|
|
|
|
assert annualized > 0
|
|
assert np.isclose(annualized, 0.21, rtol=0.1) # ~21% annualized
|
|
|
|
def test_calculate_annualized_return_same_day(self, analyzer):
|
|
"""Test annualized return with same start/end day."""
|
|
timestamps = [datetime(2023, 1, 1), datetime(2023, 1, 1)]
|
|
equity = [100.0, 110.0]
|
|
|
|
annualized = analyzer.calculate_annualized_return(equity, timestamps)
|
|
assert annualized == 0.0
|
|
|
|
class TestDrawdown:
|
|
"""Tests for drawdown calculations."""
|
|
|
|
def test_calculate_max_drawdown(self, analyzer):
|
|
"""Test max drawdown calculation."""
|
|
# Peak at 150, drop to 100, recover to 150 (full recovery)
|
|
equity = [100.0, 120.0, 150.0, 140.0, 130.0, 100.0, 110.0, 150.0]
|
|
stats = analyzer.calculate_max_drawdown(equity)
|
|
|
|
assert stats["max_drawdown"] == pytest.approx(0.3333, abs=0.001)
|
|
assert stats["peak"] == 150.0
|
|
assert stats["trough"] == 100.0
|
|
assert stats["recovery_index"] == 7
|
|
|
|
def test_calculate_max_drawdown_no_recovery(self, analyzer):
|
|
"""Test max drawdown without recovery."""
|
|
equity = [100.0, 150.0, 120.0, 100.0] # No recovery
|
|
stats = analyzer.calculate_max_drawdown(equity)
|
|
|
|
assert stats["max_drawdown"] == pytest.approx(0.3333, abs=0.001)
|
|
assert stats["recovery_index"] == -1
|
|
|
|
def test_calculate_max_drawdown_no_drawdown(self, analyzer):
|
|
"""Test with no drawdown (always increasing)."""
|
|
equity = [100.0, 110.0, 120.0, 130.0]
|
|
stats = analyzer.calculate_max_drawdown(equity)
|
|
|
|
assert stats["max_drawdown"] == 0.0
|
|
assert stats["peak"] == 130.0
|
|
assert stats["trough"] == 130.0
|
|
|
|
def test_calculate_max_drawdown_multiple(self, analyzer):
|
|
"""Test with multiple drawdowns."""
|
|
# Two drawdowns: 150->100 (33%) and 180->140 (22%)
|
|
equity = [100.0, 150.0, 120.0, 100.0, 130.0, 180.0, 160.0, 140.0, 170.0]
|
|
stats = analyzer.calculate_max_drawdown(equity)
|
|
|
|
assert stats["max_drawdown"] == pytest.approx(0.3333, abs=0.001)
|
|
assert stats["peak"] == 150.0
|
|
|
|
class TestSharpeRatio:
|
|
"""Tests for Sharpe ratio calculation."""
|
|
|
|
def test_calculate_sharpe_ratio_positive(self, analyzer):
|
|
"""Test Sharpe ratio with positive returns."""
|
|
# Consistent positive returns
|
|
returns = np.array([0.001] * 252) # 0.1% daily
|
|
sharpe = analyzer.calculate_sharpe_ratio(returns, risk_free_rate=0.0)
|
|
|
|
assert sharpe > 10 # Very high Sharpe with consistent returns
|
|
|
|
def test_calculate_sharpe_ratio_zero_volatility(self, analyzer):
|
|
"""Test Sharpe ratio with zero volatility."""
|
|
returns = np.array([0.0] * 10)
|
|
sharpe = analyzer.calculate_sharpe_ratio(returns)
|
|
|
|
assert sharpe == 0.0
|
|
|
|
def test_calculate_sharpe_ratio_empty(self, analyzer):
|
|
"""Test Sharpe ratio with empty returns."""
|
|
returns = np.array([])
|
|
sharpe = analyzer.calculate_sharpe_ratio(returns)
|
|
|
|
assert sharpe == 0.0
|
|
|
|
def test_calculate_sharpe_ratio_with_risk_free_rate(self, analyzer):
|
|
"""Test Sharpe ratio with risk-free rate."""
|
|
# 10% annual return, some volatility
|
|
np.random.seed(42)
|
|
returns = np.random.normal(0.0004, 0.02, 252)
|
|
sharpe = analyzer.calculate_sharpe_ratio(returns, risk_free_rate=0.02)
|
|
|
|
# Should be a reasonable value
|
|
assert isinstance(sharpe, float)
|
|
assert not np.isnan(sharpe)
|
|
|
|
class TestSortinoRatio:
|
|
"""Tests for Sortino ratio calculation."""
|
|
|
|
def test_calculate_sortino_ratio_positive(self, analyzer):
|
|
"""Test Sortino ratio with positive returns."""
|
|
returns = np.array([0.001] * 252)
|
|
sortino = analyzer.calculate_sortino_ratio(returns, risk_free_rate=0.0)
|
|
|
|
assert sortino == float("inf") # No downside
|
|
|
|
def test_calculate_sortino_ratio_with_downside(self, analyzer):
|
|
"""Test Sortino ratio with downside volatility."""
|
|
np.random.seed(42)
|
|
returns = np.random.normal(0.0005, 0.02, 252)
|
|
sortino = analyzer.calculate_sortino_ratio(returns)
|
|
|
|
assert isinstance(sortino, float)
|
|
assert not np.isnan(sortino)
|
|
assert sortino != float("inf")
|
|
|
|
def test_calculate_sortino_ratio_all_negative(self, analyzer):
|
|
"""Test Sortino ratio with all negative returns."""
|
|
returns = np.array([-0.01] * 10)
|
|
sortino = analyzer.calculate_sortino_ratio(returns)
|
|
|
|
assert sortino < 0 # Negative Sortino for negative returns
|
|
|
|
class TestCalmarRatio:
|
|
"""Tests for Calmar ratio calculation."""
|
|
|
|
def test_calculate_calmar_ratio(self, analyzer):
|
|
"""Test Calmar ratio calculation."""
|
|
returns = np.array([0.001] * 252) # ~25% annual return
|
|
max_dd = 0.10 # 10% drawdown
|
|
calmar = analyzer.calculate_calmar_ratio(returns, max_dd)
|
|
|
|
assert calmar > 0
|
|
assert isinstance(calmar, float)
|
|
|
|
def test_calculate_calmar_ratio_zero_drawdown(self, analyzer):
|
|
"""Test Calmar ratio with zero drawdown."""
|
|
returns = np.array([0.001] * 10)
|
|
calmar = analyzer.calculate_calmar_ratio(returns, 0.0)
|
|
|
|
assert calmar == 0.0
|
|
|
|
def test_calculate_calmar_ratio_empty(self, analyzer):
|
|
"""Test Calmar ratio with empty returns."""
|
|
returns = np.array([])
|
|
calmar = analyzer.calculate_calmar_ratio(returns, 0.10)
|
|
|
|
assert calmar == 0.0
|
|
|
|
class TestWinRate:
|
|
"""Tests for win rate calculations."""
|
|
|
|
def test_calculate_win_rate(self, analyzer, sample_trades):
|
|
"""Test win rate calculation."""
|
|
win_rate = analyzer.calculate_win_rate(sample_trades)
|
|
|
|
# 10 wins out of 20 trades
|
|
assert win_rate == 0.5
|
|
|
|
def test_calculate_win_rate_empty(self, analyzer):
|
|
"""Test win rate with no trades."""
|
|
win_rate = analyzer.calculate_win_rate([])
|
|
|
|
assert win_rate == 0.0
|
|
|
|
def test_calculate_win_rate_all_wins(self, analyzer):
|
|
"""Test win rate with all winning trades."""
|
|
base_time = datetime(2023, 1, 1)
|
|
trades = [
|
|
TradeRecord(
|
|
entry_time=base_time,
|
|
exit_time=base_time + timedelta(days=1),
|
|
side="long",
|
|
entry_price=100.0,
|
|
exit_price=110.0,
|
|
quantity=1.0,
|
|
pnl=10.0,
|
|
is_win=True,
|
|
)
|
|
for _ in range(10)
|
|
]
|
|
win_rate = analyzer.calculate_win_rate(trades)
|
|
|
|
assert win_rate == 1.0
|
|
|
|
def test_calculate_loss_rate(self, analyzer, sample_trades):
|
|
"""Test loss rate calculation."""
|
|
loss_rate = analyzer.calculate_loss_rate(sample_trades)
|
|
|
|
assert loss_rate == 0.5
|
|
|
|
class TestProfitFactor:
|
|
"""Tests for profit factor calculation."""
|
|
|
|
def test_calculate_profit_factor(self, analyzer):
|
|
"""Test profit factor calculation."""
|
|
base_time = datetime(2023, 1, 1)
|
|
trades = [
|
|
# Winning trades: +500 total
|
|
TradeRecord(
|
|
entry_time=base_time,
|
|
exit_time=base_time + timedelta(days=1),
|
|
side="long",
|
|
entry_price=100.0,
|
|
exit_price=110.0,
|
|
quantity=10.0,
|
|
pnl=100.0,
|
|
is_win=True,
|
|
),
|
|
TradeRecord(
|
|
entry_time=base_time + timedelta(days=2),
|
|
exit_time=base_time + timedelta(days=3),
|
|
side="long",
|
|
entry_price=100.0,
|
|
exit_price=105.0,
|
|
quantity=10.0,
|
|
pnl=50.0,
|
|
is_win=True,
|
|
),
|
|
# Losing trades: -100 total
|
|
TradeRecord(
|
|
entry_time=base_time + timedelta(days=4),
|
|
exit_time=base_time + timedelta(days=5),
|
|
side="long",
|
|
entry_price=100.0,
|
|
exit_price=95.0,
|
|
quantity=10.0,
|
|
pnl=-50.0,
|
|
is_win=False,
|
|
),
|
|
TradeRecord(
|
|
entry_time=base_time + timedelta(days=6),
|
|
exit_time=base_time + timedelta(days=7),
|
|
side="long",
|
|
entry_price=100.0,
|
|
exit_price=95.0,
|
|
quantity=10.0,
|
|
pnl=-50.0,
|
|
is_win=False,
|
|
),
|
|
]
|
|
pf = analyzer.calculate_profit_factor(trades)
|
|
|
|
assert pf == pytest.approx(1.5, abs=0.01) # 150/100 = 1.5
|
|
|
|
def test_calculate_profit_factor_no_losses(self, analyzer):
|
|
"""Test profit factor with no losing trades."""
|
|
base_time = datetime(2023, 1, 1)
|
|
trades = [
|
|
TradeRecord(
|
|
entry_time=base_time,
|
|
exit_time=base_time + timedelta(days=1),
|
|
side="long",
|
|
entry_price=100.0,
|
|
exit_price=110.0,
|
|
quantity=10.0,
|
|
pnl=100.0,
|
|
is_win=True,
|
|
),
|
|
]
|
|
pf = analyzer.calculate_profit_factor(trades)
|
|
|
|
assert pf == float("inf")
|
|
|
|
def test_calculate_profit_factor_empty(self, analyzer):
|
|
"""Test profit factor with no trades."""
|
|
pf = analyzer.calculate_profit_factor([])
|
|
|
|
assert pf == 0.0
|
|
|
|
class TestAverageTrade:
|
|
"""Tests for average trade calculations."""
|
|
|
|
def test_calculate_avg_trade(self, analyzer, sample_trades):
|
|
"""Test average trade statistics."""
|
|
stats = analyzer.calculate_avg_trade(sample_trades)
|
|
|
|
assert stats["avg_pnl"] > 0
|
|
assert stats["avg_win"] > 0
|
|
assert stats["avg_loss"] > 0
|
|
assert stats["win_loss_ratio"] > 0
|
|
|
|
def test_calculate_avg_trade_empty(self, analyzer):
|
|
"""Test average trade with no trades."""
|
|
stats = analyzer.calculate_avg_trade([])
|
|
|
|
assert stats["avg_pnl"] == 0.0
|
|
assert stats["avg_win"] == 0.0
|
|
assert stats["avg_loss"] == 0.0
|
|
assert stats["win_loss_ratio"] == 0.0
|
|
|
|
class TestVolatility:
|
|
"""Tests for volatility calculations."""
|
|
|
|
def test_calculate_volatility(self, analyzer):
|
|
"""Test volatility calculation."""
|
|
np.random.seed(42)
|
|
returns = np.random.normal(0, 0.02, 252)
|
|
vol = analyzer.calculate_volatility(returns, annualize=True)
|
|
|
|
assert vol > 0
|
|
# Annualized vol should be approximately 0.02 * sqrt(252)
|
|
expected = 0.02 * np.sqrt(252)
|
|
assert np.isclose(vol, expected, rtol=0.2)
|
|
|
|
def test_calculate_volatility_not_annualized(self, analyzer):
|
|
"""Test volatility without annualization."""
|
|
returns = np.array([0.01, -0.01, 0.01, -0.01])
|
|
vol = analyzer.calculate_volatility(returns, annualize=False)
|
|
|
|
assert vol > 0
|
|
|
|
def test_calculate_volatility_empty(self, analyzer):
|
|
"""Test volatility with empty returns."""
|
|
vol = analyzer.calculate_volatility(np.array([]))
|
|
|
|
assert vol == 0.0
|
|
|
|
class TestVaR:
|
|
"""Tests for Value at Risk calculations."""
|
|
|
|
def test_calculate_var(self, analyzer):
|
|
"""Test VaR calculation."""
|
|
np.random.seed(42)
|
|
returns = np.random.normal(0, 0.02, 1000)
|
|
var = analyzer.calculate_var(returns, confidence=0.05)
|
|
|
|
assert var < 0 # VaR should be negative (loss)
|
|
# Approximately 5% of returns should be below VaR
|
|
below_var = np.sum(returns < var)
|
|
assert 30 < below_var < 70 # Allow some tolerance
|
|
|
|
def test_calculate_var_empty(self, analyzer):
|
|
"""Test VaR with empty returns."""
|
|
var = analyzer.calculate_var(np.array([]))
|
|
|
|
assert var == 0.0
|
|
|
|
def test_calculate_cvar(self, analyzer):
|
|
"""Test CVaR calculation."""
|
|
np.random.seed(42)
|
|
returns = np.random.normal(0, 0.02, 1000)
|
|
cvar = analyzer.calculate_cvar(returns, confidence=0.05)
|
|
var = analyzer.calculate_var(returns, confidence=0.05)
|
|
|
|
assert cvar < 0
|
|
assert cvar <= var # CVaR should be worse than VaR
|
|
|
|
class TestConsecutiveStats:
|
|
"""Tests for consecutive trade statistics."""
|
|
|
|
def test_calculate_consecutive_stats(self, analyzer):
|
|
"""Test consecutive stats calculation."""
|
|
base_time = datetime(2023, 1, 1)
|
|
# 3 wins, 2 losses, 4 wins, 1 loss
|
|
trades = [
|
|
TradeRecord(
|
|
entry_time=base_time + timedelta(days=i),
|
|
exit_time=base_time + timedelta(days=i + 1),
|
|
side="long",
|
|
entry_price=100.0,
|
|
exit_price=110.0,
|
|
quantity=1.0,
|
|
pnl=10.0,
|
|
is_win=pattern,
|
|
)
|
|
for i, pattern in enumerate(
|
|
[True, True, True, False, False, True, True, True, True, False]
|
|
)
|
|
]
|
|
stats = analyzer.calculate_consecutive_stats(trades)
|
|
|
|
assert stats["max_consecutive_wins"] == 4
|
|
assert stats["max_consecutive_losses"] == 2
|
|
assert stats["current_streak"] == -1 # Ended with a loss
|
|
|
|
def test_calculate_consecutive_stats_empty(self, analyzer):
|
|
"""Test consecutive stats with no trades."""
|
|
stats = analyzer.calculate_consecutive_stats([])
|
|
|
|
assert stats["max_consecutive_wins"] == 0
|
|
assert stats["max_consecutive_losses"] == 0
|
|
assert stats["current_streak"] == 0
|
|
|
|
class TestGenerateReport:
|
|
"""Tests for report generation."""
|
|
|
|
def test_generate_report_structure(self, analyzer, sample_backtest_result):
|
|
"""Test that report contains all expected keys."""
|
|
report = analyzer.generate_report(sample_backtest_result)
|
|
|
|
expected_keys = [
|
|
"initial_capital",
|
|
"final_capital",
|
|
"total_return",
|
|
"total_return_pct",
|
|
"annualized_return",
|
|
"annualized_return_pct",
|
|
"num_trades",
|
|
"num_winning_trades",
|
|
"num_losing_trades",
|
|
"win_rate",
|
|
"win_rate_pct",
|
|
"loss_rate",
|
|
"profit_factor",
|
|
"avg_pnl",
|
|
"avg_win",
|
|
"avg_loss",
|
|
"win_loss_ratio",
|
|
"max_drawdown",
|
|
"max_drawdown_pct",
|
|
"volatility",
|
|
"sharpe_ratio",
|
|
"sortino_ratio",
|
|
"calmar_ratio",
|
|
"var_5pct",
|
|
"cvar_5pct",
|
|
"max_consecutive_wins",
|
|
"max_consecutive_losses",
|
|
"duration_days",
|
|
"start_time",
|
|
"end_time",
|
|
]
|
|
|
|
for key in expected_keys:
|
|
assert key in report, f"Missing key: {key}"
|
|
|
|
def test_generate_report_values(self, analyzer, sample_backtest_result):
|
|
"""Test that report values are reasonable."""
|
|
report = analyzer.generate_report(sample_backtest_result)
|
|
|
|
assert report["initial_capital"] == 10000.0
|
|
assert report["num_trades"] == 20
|
|
assert 0 <= report["win_rate"] <= 1
|
|
assert report["max_drawdown"] >= 0
|
|
assert report["duration_days"] > 0
|
|
|
|
def test_generate_report_no_trades(self, analyzer, sample_equity_curve, sample_timestamps):
|
|
"""Test report generation with no trades."""
|
|
result = BacktestResult(
|
|
initial_capital=10000.0,
|
|
final_capital=sample_equity_curve[-1],
|
|
equity_curve=sample_equity_curve,
|
|
timestamps=sample_timestamps,
|
|
trades=[],
|
|
start_time=sample_timestamps[0],
|
|
end_time=sample_timestamps[-1],
|
|
)
|
|
report = analyzer.generate_report(result)
|
|
|
|
assert report["num_trades"] == 0
|
|
assert report["win_rate"] == 0.0
|
|
assert report["profit_factor"] == 0.0
|
|
|
|
class TestToDataFrame:
|
|
"""Tests for DataFrame conversion."""
|
|
|
|
def test_to_dataframe(self, analyzer, sample_backtest_result):
|
|
"""Test conversion to DataFrame."""
|
|
df = analyzer.to_dataframe(sample_backtest_result)
|
|
|
|
assert len(df) == len(sample_backtest_result.equity_curve)
|
|
assert "timestamp" in df.columns
|
|
assert "equity" in df.columns
|
|
assert "returns" in df.columns
|
|
assert "drawdown" in df.columns
|
|
|
|
def test_to_dataframe_returns(self, analyzer, sample_backtest_result):
|
|
"""Test that returns column is calculated correctly."""
|
|
df = analyzer.to_dataframe(sample_backtest_result)
|
|
|
|
# First return should be NaN
|
|
assert pd.isna(df["returns"].iloc[0])
|
|
|
|
# Other returns should be calculated
|
|
assert not pd.isna(df["returns"].iloc[1])
|
|
|
|
|
|
import pandas as pd # noqa: E402
|