mirror of
https://ghfast.top/https://github.com/aeroxw/easy-tdx.git
synced 2026-09-12 22:44:17 +08:00
Justin Gu
5fc398255d
mypy (13 errors → 0): - portfolio/optimizer: register_optimizer 返回类型改为 Callable 装饰器签名 (原标注 type[WeightOptimizer] 导致 4 个子类 Too many arguments) - factor/engine: _datetime_to_int 用 isinstance 收窄替代 object→int 强转 - factor/analysis: 删多余 type:ignore（改由 mypy override 统一处理 scipy） - backtest/orders, execution: np.sqrt 表达式用 float() 包裹消除 no-any-return - MyTT.pyi: MACD 签名删除错误的 LOW/HIGH 参数（与 MyTT.py 实际签名对齐） - pyproject: 新增 scipy mypy override (ignore_missing_imports) ruff format: 8 个 test 文件格式化 验证: 564 passed, mypy 192 文件零错误, ruff check/format 全绿
81 lines
2.4 KiB
Python
81 lines
2.4 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"]
|
|
|
|
|
|
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")
|