mirror of
https://ghfast.top/https://github.com/aeroxw/easy-tdx.git
synced 2026-09-12 14:34:15 +08:00
GitHub
00b8c6374a
排查发现用户反馈的"回测统计数据缺失/异常"并非服务器连接问题， 而是回测引擎与组合优化器自身的代码缺陷： #23: 首根 bar 访问 close[-1] 崩溃 _SeriesAccessor 负向越界改返回 NaN（不抛 IndexError）； BacktestEngine 新增 warmup_bars 参数跳过指标预热期。 #25-A: FactorWeightedOptimizer 权重坍缩 n_stocks=2 且得分接近时，减最小值把低分标的权重压到 ~6e-8， 等于单股满仓、n_stocks 被无视，进而出现持仓1只/-99.98%回撤。 新增 _apply_weight_floor 权重下限保证入选标的都有实质权重。 #25-B: RebalanceEngine total_trades 统计错误 total_trades = len(equity_curve)（天数）改为 len(trades_df)（真实笔数）。 #22: 绩效别名键 + 数据异常诊断 performance dict 新增 sharpe_ratio/start_cash/end_value 别名键； 资金曲线异常时返回 diagnostic_warning 而非静默全 0，CLI 显示提示。 文档：README 加入回测手册导航；backtest_usage.md 补充 warmup 说明。 测试：新增 6 个回归测试，更新 3 个；932 passed。
89 lines
3.2 KiB
Python
89 lines
3.2 KiB
Python
"""Test RebalanceEngine."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from easy_tdx.portfolio.optimizer import EqualWeightOptimizer, FactorWeightedOptimizer
|
|
from easy_tdx.portfolio.rebalance import RebalanceEngine
|
|
|
|
|
|
def _make_market(n_stocks: int = 10, n_days: int = 120, seed: int = 42) -> dict[str, pd.DataFrame]:
|
|
rng = np.random.default_rng(seed)
|
|
data = {}
|
|
for i in range(n_stocks):
|
|
close = 10.0 + np.cumsum(rng.normal(0.01, 0.5, n_days))
|
|
close = np.maximum(close, 1.0)
|
|
data[f"{i:06d}"] = pd.DataFrame(
|
|
{
|
|
"datetime": pd.date_range("2024-01-01", periods=n_days, freq="D"),
|
|
"open": close,
|
|
"high": close + 0.3,
|
|
"low": close - 0.3,
|
|
"close": close,
|
|
"vol": rng.integers(1e5, 1e7, n_days).astype(float),
|
|
"amount": close * 1e6,
|
|
}
|
|
)
|
|
return data
|
|
|
|
|
|
class TestRebalanceEngine:
|
|
def test_basic_run(self):
|
|
engine = RebalanceEngine(
|
|
optimizer=EqualWeightOptimizer(),
|
|
factor_name="momentum_20d",
|
|
n_stocks=5,
|
|
rebalance_freq="M",
|
|
cash=1_000_000,
|
|
)
|
|
result = engine.run(_make_market(), start_date=20240101, end_date=20240430)
|
|
assert len(result.states) > 0
|
|
assert len(result.rebalance_dates) > 0
|
|
assert len(result.equity_curve) > 0
|
|
assert "total_return" in result.performance
|
|
|
|
def test_with_factor_weighted(self):
|
|
engine = RebalanceEngine(
|
|
optimizer=FactorWeightedOptimizer(),
|
|
factor_name="momentum_20d",
|
|
n_stocks=5,
|
|
)
|
|
result = engine.run(_make_market(), start_date=20240101, end_date=20240430)
|
|
assert len(result.states) > 0
|
|
|
|
def test_empty_data(self):
|
|
result = RebalanceEngine(optimizer=EqualWeightOptimizer()).run({})
|
|
assert result.performance["total_return"] == 0.0
|
|
|
|
def test_equity_curve_dates_sorted(self):
|
|
result = RebalanceEngine(
|
|
optimizer=EqualWeightOptimizer(),
|
|
rebalance_freq="M",
|
|
).run(_make_market(), start_date=20240101, end_date=20240430)
|
|
dates = result.equity_curve["datetime"].tolist()
|
|
assert dates == sorted(dates)
|
|
|
|
def test_trades_recorded(self):
|
|
engine = RebalanceEngine(
|
|
optimizer=EqualWeightOptimizer(),
|
|
n_stocks=3,
|
|
rebalance_freq="M",
|
|
)
|
|
result = engine.run(_make_market(), start_date=20240101, end_date=20240430)
|
|
assert len(result.trades) > 0
|
|
assert "BUY" in result.trades["direction"].values
|
|
|
|
def test_total_trades_matches_trade_rows(self):
|
|
"""issue #25: performance['total_trades'] 应等于真实交易笔数，而非天数。"""
|
|
engine = RebalanceEngine(
|
|
optimizer=EqualWeightOptimizer(),
|
|
n_stocks=3,
|
|
rebalance_freq="M",
|
|
)
|
|
result = engine.run(_make_market(), start_date=20240101, end_date=20240430)
|
|
assert result.performance["total_trades"] == len(result.trades)
|
|
# 修复前 total_trades == len(equity_curve)（天数），明显大于交易笔数
|
|
assert result.performance["total_trades"] != len(result.equity_curve)
|