mirror of
https://ghfast.top/https://github.com/aeroxw/easy_tdx_max.git
synced 2026-09-12 16:54:20 +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。
96 lines
3.1 KiB
Python
96 lines
3.1 KiB
Python
"""Test portfolio optimizers."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
import pytest
|
||
|
||
from easy_tdx.portfolio.optimizer import (
|
||
EqualWeightOptimizer,
|
||
FactorWeightedOptimizer,
|
||
RiskParityOptimizer,
|
||
get_optimizer,
|
||
)
|
||
|
||
|
||
def _make_scores(n: int = 20, seed: int = 42) -> pd.DataFrame:
|
||
rng = np.random.default_rng(seed)
|
||
return pd.DataFrame(
|
||
{
|
||
"code": [f"{i:06d}" for i in range(n)],
|
||
"score": rng.normal(0.02, 0.05, n),
|
||
}
|
||
)
|
||
|
||
|
||
class TestEqualWeight:
|
||
def test_weights_sum_to_one(self):
|
||
w = EqualWeightOptimizer().optimize(_make_scores(), n_stocks=10)
|
||
assert abs(sum(w.values()) - 1.0) < 1e-8
|
||
|
||
def test_n_stocks_selected(self):
|
||
w = EqualWeightOptimizer().optimize(_make_scores(), n_stocks=5)
|
||
assert len(w) == 5
|
||
|
||
def test_all_equal(self):
|
||
w = EqualWeightOptimizer().optimize(_make_scores(), n_stocks=10)
|
||
vals = list(w.values())
|
||
assert all(abs(v - vals[0]) < 1e-8 for v in vals)
|
||
|
||
def test_empty_input(self):
|
||
w = EqualWeightOptimizer().optimize(pd.DataFrame(columns=["code", "score"]), n_stocks=5)
|
||
assert len(w) == 0
|
||
|
||
|
||
class TestFactorWeighted:
|
||
def test_weights_sum_to_one(self):
|
||
w = FactorWeightedOptimizer().optimize(_make_scores(), n_stocks=10)
|
||
assert abs(sum(w.values()) - 1.0) < 1e-6
|
||
|
||
def test_higher_score_higher_weight(self):
|
||
scores = pd.DataFrame({"code": ["A", "B", "C"], "score": [3.0, 2.0, 1.0]})
|
||
w = FactorWeightedOptimizer().optimize(scores, n_stocks=3)
|
||
assert w["A"] > w["C"]
|
||
|
||
def test_no_weight_collapse_small_n(self):
|
||
"""issue #25: n_stocks=2 且得分接近时，权重不应坍缩到接近 0。
|
||
|
||
修复前 scores=[0.5, 0.34] 经"减最小值"后权重变成 ~1.0 / ~6e-8，
|
||
等于单股满仓、n_stocks=2 被忽略。修复后每只标的都有实质权重。
|
||
"""
|
||
scores = pd.DataFrame({"code": ["A", "B"], "score": [0.50, 0.34]})
|
||
w = FactorWeightedOptimizer().optimize(scores, n_stocks=2)
|
||
assert len(w) == 2
|
||
assert abs(sum(w.values()) - 1.0) < 1e-6
|
||
# 两只都应有实质权重（≥ 0.05），低分股不再被压到 ~0
|
||
assert min(w.values()) >= 0.05
|
||
# 高分股权重仍更高
|
||
assert w["A"] > w["B"]
|
||
|
||
|
||
class TestRiskParity:
|
||
def test_weights_sum_to_one(self):
|
||
w = RiskParityOptimizer().optimize(_make_scores(), n_stocks=10)
|
||
assert abs(sum(w.values()) - 1.0) < 1e-6
|
||
|
||
def test_with_volatility_column(self):
|
||
scores = pd.DataFrame(
|
||
{
|
||
"code": ["A", "B", "C"],
|
||
"score": [1.0, 1.0, 1.0],
|
||
"volatility": [0.1, 0.2, 0.4],
|
||
}
|
||
)
|
||
w = RiskParityOptimizer().optimize(scores, n_stocks=3)
|
||
assert w["A"] > w["C"]
|
||
|
||
|
||
class TestRegistry:
|
||
def test_get_optimizer(self):
|
||
assert isinstance(get_optimizer("equal"), EqualWeightOptimizer)
|
||
|
||
def test_unknown_raises(self):
|
||
with pytest.raises(ValueError, match="未知优化器"):
|
||
get_optimizer("nonexistent")
|