132 lines
5.5 KiB
Python
132 lines
5.5 KiB
Python
"""通路测试策略(ChannelTestStrategy)单元测试。
|
|
|
|
验证跨类型轮换 + 每日多场景调度(主调仓/部分加减/卖后买/T+1探针),
|
|
用 mock broker 记录下单调用。坑:BrokerFacade 是 dataclass,子类方法重写会被
|
|
父类 __init__ 写入的实例属性遮蔽 → mock 必须在 super().__init__() 后实例注入。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from sanguo_portfolio.strategies import ChannelTestConfig, ChannelTestStrategy
|
|
from sanguo_portfolio.strategies.all_weather import BrokerFacade
|
|
from sanguo_portfolio.strategies.channel_test import UNIVERSE_BY_TYPE
|
|
|
|
|
|
class _MockBroker(BrokerFacade):
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[str, str, float]] = [] # (method, code, value)
|
|
super().__init__()
|
|
self.order_target_value = self._record("otv")
|
|
self.order_value = self._record("ov")
|
|
|
|
def _record(self, method: str):
|
|
def rec(code: str, value: float):
|
|
self.calls.append((method, code, value))
|
|
return None
|
|
return rec
|
|
|
|
|
|
class _Pos:
|
|
def __init__(self, value: float) -> None:
|
|
self.value = value
|
|
|
|
|
|
class _Ctx:
|
|
def __init__(self, positions: dict, cash: float) -> None:
|
|
self.portfolio = type("P", (), {"positions": positions,
|
|
"available_cash": cash,
|
|
"total_value": cash + sum(p.value for p in positions.values())})()
|
|
|
|
|
|
def test_universe_no_star_bj_stocks():
|
|
"""实盘铁律:个股只允许主板+创业板(无 688/920/8 开头)。"""
|
|
for codes in UNIVERSE_BY_TYPE.values():
|
|
for c in codes:
|
|
num = c.split(".")[0]
|
|
if c.endswith(".XSHE") and len(num) == 6 and not num.startswith(("15", "16")):
|
|
assert not num.startswith(("30", "00", "002")) or num.startswith(("30", "00")), c
|
|
# 个股(非 ETF:51/15/56/58 开头是基金)不允许 688/689/920/8 开头
|
|
is_etf = num.startswith(("51", "15", "56", "58"))
|
|
if not is_etf:
|
|
assert not num.startswith(("688", "689", "92", "4", "8")), f"非主板/创业板个股: {c}"
|
|
|
|
|
|
def test_universe_covers_all_types_with_multiple():
|
|
"""每类至少 3 只、类型覆盖宽基/行业/跨境/主板/创业板。"""
|
|
for grp, codes in UNIVERSE_BY_TYPE.items():
|
|
assert len(codes) >= 3, grp
|
|
assert len(UNIVERSE_BY_TYPE) >= 5
|
|
|
|
|
|
def test_target_set_spans_types():
|
|
s = ChannelTestStrategy(provider=None, config=ChannelTestConfig(hold_n=6))
|
|
s._day = 1
|
|
target = s._target_set()
|
|
assert len(target) == 6
|
|
# 跨类型:6 只来自不同类型组
|
|
code2grp = {c: g for g, cs in UNIVERSE_BY_TYPE.items() for c in cs}
|
|
groups = {code2grp[c] for c in target}
|
|
assert len(groups) == 6 # hold_n=6 且组数≥6 → 每组一只
|
|
|
|
|
|
def test_rotate_sells_non_target_and_buys_target():
|
|
broker = _MockBroker()
|
|
s = ChannelTestStrategy(provider=None, broker=broker,
|
|
config=ChannelTestConfig(hold_n=2, period=1, probe_t1=False,
|
|
intraday_partial=False, intraday_swap=False))
|
|
ctx = _Ctx({"510300.XSHG": _Pos(1000), "600519.XSHG": _Pos(1000)}, cash=8000)
|
|
s.rotate(ctx)
|
|
zero_calls = {c for m, c, v in broker.calls if m == "otv" and v == 0}
|
|
assert zero_calls # 有全量卖出(旧持仓不在新目标)
|
|
buys = [c for m, c, v in broker.calls if m == "otv" and v > 0]
|
|
assert len(buys) == 2 # 等权买入 hold_n 只
|
|
|
|
|
|
def test_partial_adjust_buys_more_and_sells_half():
|
|
broker = _MockBroker()
|
|
s = ChannelTestStrategy(provider=None, broker=broker,
|
|
config=ChannelTestConfig(intraday_partial=True))
|
|
ctx = _Ctx({"A": _Pos(1000), "B": _Pos(1000), "C": _Pos(1000)}, cash=1000)
|
|
s.partial_adjust(ctx)
|
|
ups = [(c, v) for m, c, v in broker.calls if v > 1000] # 加仓 >原值
|
|
downs = [(c, v) for m, c, v in broker.calls if 0 < v < 1000] # 部分减仓
|
|
assert ups and all(v >= 1000 for _, v in ups)
|
|
assert downs # 部分卖出(非清仓)
|
|
|
|
|
|
def test_swap_one_sells_then_buys():
|
|
broker = _MockBroker()
|
|
s = ChannelTestStrategy(provider=None, broker=broker,
|
|
config=ChannelTestConfig(intraday_swap=True))
|
|
s._day = 1
|
|
ctx = _Ctx({"510300.XSHG": _Pos(5000), "600519.XSHG": _Pos(5000)}, cash=3000)
|
|
s.swap_one(ctx)
|
|
sells = [c for m, c, v in broker.calls if v == 0]
|
|
buys = [(c, v) for m, c, v in broker.calls if v > 0]
|
|
assert sells == ["510300.XSHG"] # 卖第一只
|
|
assert buys and buys[0][0] not in ("510300.XSHG", "600519.XSHG") # 买未持有的
|
|
|
|
|
|
def test_t1_probe_fires():
|
|
broker = _MockBroker()
|
|
s = ChannelTestStrategy(provider=None, broker=broker,
|
|
config=ChannelTestConfig(probe_t1=True))
|
|
ctx = _Ctx({"510300.XSHG": _Pos(5000)}, cash=0)
|
|
s.t1_probe(ctx)
|
|
assert broker.calls == [("otv", "510300.XSHG", 0)]
|
|
|
|
|
|
def test_initialize_registers_intraday_schedules():
|
|
registered: list[tuple[str, str]] = []
|
|
|
|
class _RegBroker(_MockBroker):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.run_daily = lambda fn, t: registered.append(
|
|
(getattr(fn, "__name__", str(fn)), t))
|
|
|
|
s = ChannelTestStrategy(provider=None, broker=_RegBroker(),
|
|
config=ChannelTestConfig())
|
|
s.initialize(object())
|
|
times = {t for _, t in registered}
|
|
assert {"9:35", "10:45", "13:45", "14:30"} <= times # 四个盘中时点全注册
|