71 lines
2.5 KiB
Python
71 lines
2.5 KiB
Python
|
from datetime import date, timedelta
|
||
|
|
from decimal import Decimal
|
||
|
|
|
||
|
|
from app.service.allocation_backtest_service import (
|
||
|
|
AllocationBacktestEngine,
|
||
|
|
AllocationBacktestService,
|
||
|
|
BacktestObservation,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def observation(day: int, cash: str, equity: str, liquid: bool = True) -> BacktestObservation:
|
||
|
|
return BacktestObservation(
|
||
|
|
trade_date=date(2026, 1, 1) + timedelta(days=day),
|
||
|
|
returns_pct={
|
||
|
|
"cash_management_etf": Decimal(cash),
|
||
|
|
"equity_etf": Decimal(equity),
|
||
|
|
},
|
||
|
|
liquidity_observed=liquid,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_backtest_compares_static_and_dynamic_performance() -> None:
|
||
|
|
observations = [observation(0, "0", "10"), observation(1, "0", "-10")]
|
||
|
|
result = AllocationBacktestEngine.run(
|
||
|
|
observations,
|
||
|
|
{"cash_management_etf": 50, "bond_etf": 0, "equity_etf": 50},
|
||
|
|
{
|
||
|
|
observations[1].trade_date: {
|
||
|
|
"cash_management_etf": 100,
|
||
|
|
"bond_etf": 0,
|
||
|
|
"equity_etf": 0,
|
||
|
|
}
|
||
|
|
},
|
||
|
|
)
|
||
|
|
|
||
|
|
assert result.observation_count == 2
|
||
|
|
assert result.static is not None and result.dynamic is not None
|
||
|
|
assert result.static.total_return_pct == Decimal("-0.2500")
|
||
|
|
assert result.dynamic.total_return_pct == Decimal("5.0000")
|
||
|
|
assert result.dynamic_rebalance_count == 1
|
||
|
|
assert result.liquidity_history_coverage_pct == Decimal("100")
|
||
|
|
assert result.status == "insufficient_history"
|
||
|
|
|
||
|
|
|
||
|
|
def test_rolling_metrics_and_adequate_history_are_ready() -> None:
|
||
|
|
observations = [observation(day, "1", "2") for day in range(20)]
|
||
|
|
metrics = AllocationBacktestService._rolling_metrics(observations)
|
||
|
|
|
||
|
|
assert {item.asset_class for item in metrics} == {"cash_management_etf", "equity_etf"}
|
||
|
|
assert all(item.product_count == 20 for item in metrics)
|
||
|
|
result = AllocationBacktestEngine.run(
|
||
|
|
observations,
|
||
|
|
{"cash_management_etf": 50, "bond_etf": 0, "equity_etf": 50},
|
||
|
|
{},
|
||
|
|
)
|
||
|
|
assert result.status == "ready"
|
||
|
|
assert result.limitations == ("历史观察不足 120 个交易日，动态优化无法覆盖完整窗口。",)
|
||
|
|
|
||
|
|
|
||
|
|
def test_backtest_reports_liquidity_coverage_limit() -> None:
|
||
|
|
observations = [observation(day, "0", "0", day < 10) for day in range(20)]
|
||
|
|
result = AllocationBacktestEngine.run(
|
||
|
|
observations,
|
||
|
|
{"cash_management_etf": 100, "bond_etf": 0, "equity_etf": 0},
|
||
|
|
{},
|
||
|
|
)
|
||
|
|
|
||
|
|
assert result.status == "partial"
|
||
|
|
assert result.liquidity_history_coverage_pct == Decimal("50")
|
||
|
|
assert any("80%" in limitation for limitation in result.limitations)
|