217 lines
9.6 KiB
Python
217 lines
9.6 KiB
Python
"""定寸总资产口径(A 修法 2026-08-25 → 全量统一 2026-08-26)。
|
|
|
|
08-25 9:30 事故:small_cap 同轮「全卖19只→马上全买20只」,台账现金只被
|
|
runner_live 的 60s 归因轮询更新(唯一 cash 更新入口)→ 买入定寸在两轮轮询
|
|
之间评估 → 卖出回款不可见 → 20 笔买入全部目标 0,全天空仓。B 修法
|
|
(191270c,下单返回即时归因入账)已做;A 修法=防御纵深第二层:定寸口径回
|
|
总资产 per_value = 总资产/N(现金+持仓市值,卖出前后不变),即使台账归因
|
|
滞后/异常,定寸也不再依赖未入账的现金。
|
|
|
|
**2026-08-26 用户拍板「都要统一ex的」**:原版(回测)四策略定寸同步换
|
|
_total_value——原 JQ 语义 ``per_value = stock_value/len`` 本就是总资产口径,
|
|
cash 口径是移植时引入的失真;统一后回测↔实盘对齐,原版与 _ex 副本同参
|
|
选股+定寸一致(本文件参数化跑 原版+_ex 两类即验收)。守卫 cash<=0 → total<=0。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Any, List, Tuple
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock
|
|
|
|
from tests.portfolio.conftest import FakeContext, FakePosition
|
|
from sanguo_portfolio.strategies import (
|
|
AllWeatherExStrategy,
|
|
AllWeatherStrategy,
|
|
MomentumTimingExStrategy,
|
|
MomentumTimingStrategy,
|
|
SmallCapExStrategy,
|
|
SmallCapStrategy,
|
|
ValueSelectionExStrategy,
|
|
ValueSelectionStrategy,
|
|
)
|
|
from sanguo_portfolio.strategies.all_weather import _total_value
|
|
|
|
|
|
# =================== 公共装配 ===================
|
|
class _RecordingBroker:
|
|
"""记录 order_target_value / order_value 委托;**不动 context**——精确模拟
|
|
「卖出已成交、台账归因未跑」的现金窗口(事故形态)。"""
|
|
|
|
def __init__(self) -> None:
|
|
self.orders: List[Tuple[str, float]] = []
|
|
self.values: List[Tuple[str, float]] = []
|
|
|
|
def order_target_value(self, code: str, value: float) -> Any:
|
|
self.orders.append((code, float(value)))
|
|
return object() # 非 None = 下单成功
|
|
|
|
def order_value(self, code: str, value: float) -> Any:
|
|
self.values.append((code, float(value)))
|
|
return object()
|
|
|
|
|
|
def _stale_cash_ctx(total: float = 10500.0, stale_cash: float = 1000.0) -> FakeContext:
|
|
"""旧持仓 OLD_A(市值9000)/OLD_B(市值500);现金=回款未入账的旧值 1000。
|
|
|
|
``portfolio.total_value`` 模拟 jq/B2 语义(引擎现算 现金+Σ市值,与归因无关):
|
|
- B2 实盘 InstancePortfolio.total_value = ledger equity(现价,缺价回退成本)
|
|
- 回测 jq portfolio 原生 total_value
|
|
现金口径会把定寸算成 1000/2=500;总资产口径 10500/2=5250——断言锚点。
|
|
"""
|
|
ctx = FakeContext(
|
|
current_dt=datetime(2026, 8, 25, 9, 30),
|
|
positions={
|
|
"OLD_A.XSHG": FakePosition("OLD_A.XSHG", avg_cost=9.0, price=9.0, total=1000),
|
|
"OLD_B.XSHG": FakePosition("OLD_B.XSHG", avg_cost=5.0, price=5.0, total=100),
|
|
},
|
|
cash=stale_cash,
|
|
)
|
|
ctx.portfolio.total_value = total
|
|
return ctx
|
|
|
|
|
|
def _falsy_status(stocks):
|
|
return {s: {"is_limit_up": False, "is_limit_down": False, "is_paused": False}
|
|
for s in stocks}
|
|
|
|
|
|
def _assert_sized_by_total(broker: _RecordingBroker) -> None:
|
|
"""卖出旧仓两只 + 新仓三只各按 总资产/3=3500 定寸(绝非现金/N≈333)。
|
|
|
|
目标 3 只 > 旧仓 2 只 = 事故真实形状(19 旧仓/20 目标):买入守卫
|
|
``len(positions) >= target_num`` 因持仓视图滞后不提前 break,全部买入。
|
|
"""
|
|
sells = [(c, v) for c, v in broker.orders if v == 0]
|
|
buys = [(c, v) for c, v in broker.orders if v > 0]
|
|
assert {c for c, _ in sells} == {"OLD_A.XSHG", "OLD_B.XSHG"}
|
|
assert [c for c, _ in buys] == ["NEW_1.XSHG", "NEW_2.XSHG", "NEW_3.XSHG"]
|
|
for _, v in buys:
|
|
assert v == pytest.approx(10500.0 / 3)
|
|
assert v > 1000.0 # 现金口径(1000/3≈333)必缩水——回归红线
|
|
|
|
|
|
# =================== helper 层 ===================
|
|
class TestTotalValueHelper:
|
|
def test_reads_portfolio_total_value(self):
|
|
"""portfolio 带 total_value(jq 原生/B2)→ 直接读。"""
|
|
ctx = FakeContext(cash=123.0)
|
|
ctx.portfolio.total_value = 4567.0
|
|
assert _total_value(ctx) == pytest.approx(4567.0)
|
|
|
|
def test_fallback_cash_plus_positions_when_attr_missing(self):
|
|
"""缺 total_value 属性 → 现金 + Σ持仓市值(同语义兜底)。"""
|
|
pos = FakePosition("A.XSHG", avg_cost=1.0, price=2.0, total=100)
|
|
pos.value = 900.0
|
|
ctx = FakeContext(positions={"A.XSHG": pos}, cash=1000.0)
|
|
assert _total_value(ctx) == pytest.approx(1900.0)
|
|
|
|
|
|
# =================== 轮动定寸:原版+_ex 同参一致(2026-08-26 统一验收) ===================
|
|
def _run_momentum_rotate(strategy, ctx) -> None:
|
|
"""统一入口:_ex 直接打 _rotate_positions;原版定寸内联 handle_data,
|
|
数据层全打桩走通到调仓尾。"""
|
|
stocks = ["NEW_1.XSHG", "NEW_2.XSHG", "NEW_3.XSHG"]
|
|
if hasattr(strategy, "_rotate_positions"):
|
|
strategy._rotate_positions(ctx, stocks)
|
|
return
|
|
strategy._cal_buy_sign = lambda *a, **k: True
|
|
strategy._stock_pool_cached = lambda idx, d: ["I1.XSHG"]
|
|
strategy._ensure_day_panel = lambda *a, **k: None
|
|
strategy._find_stock_pool = lambda *a, **k: list(stocks)
|
|
strategy._select_stocks = lambda *a, **k: list(stocks)
|
|
strategy._get_limit_status = lambda ss, d: _falsy_status(ss)
|
|
strategy.handle_data(ctx)
|
|
|
|
|
|
@pytest.mark.parametrize("cls", [MomentumTimingStrategy, MomentumTimingExStrategy],
|
|
ids=["原版", "ex"])
|
|
def test_momentum_rotate_sizes_by_total_value(cls):
|
|
"""momentum 调仓:总资产定寸,对未入账卖出免疫(原版与 _ex 同断言)。"""
|
|
broker = _RecordingBroker()
|
|
strat = cls(provider=MagicMock(), broker=broker)
|
|
_run_momentum_rotate(strat, _stale_cash_ctx())
|
|
_assert_sized_by_total(broker)
|
|
|
|
|
|
@pytest.mark.parametrize("cls", [SmallCapStrategy, SmallCapExStrategy],
|
|
ids=["原版", "ex"])
|
|
def test_small_cap_rebalance_sizes_by_total_value(cls):
|
|
"""small_cap _rebalance:同型回归(原版与 _ex 同断言)。"""
|
|
broker = _RecordingBroker()
|
|
strat = cls(provider=MagicMock(), broker=broker)
|
|
strat.in_position_stocks = ["NEW_1.XSHG", "NEW_2.XSHG", "NEW_3.XSHG"]
|
|
strat._rebalance(_stale_cash_ctx())
|
|
_assert_sized_by_total(broker)
|
|
|
|
|
|
@pytest.mark.parametrize("cls", [ValueSelectionStrategy, ValueSelectionExStrategy],
|
|
ids=["原版", "ex"])
|
|
def test_value_monthly_adjustment_sizes_by_total_value(cls):
|
|
"""value monthly_adjustment:同型回归(选股段打桩,只验定寸腿)。"""
|
|
broker = _RecordingBroker()
|
|
strat = cls(provider=MagicMock(), broker=broker)
|
|
strat._stock_pool = lambda *a, **k: ["C1.XSHG", "C2.XSHG"]
|
|
strat._get_stock_list = lambda *a, **k: ["NEW_1.XSHG", "NEW_2.XSHG", "NEW_3.XSHG"]
|
|
strat._get_limit_status = lambda stocks, date: _falsy_status(stocks)
|
|
strat.monthly_adjustment(_stale_cash_ctx())
|
|
_assert_sized_by_total(broker)
|
|
|
|
|
|
# =================== all_weather 两现场(2026-08-26 统一补齐) ===================
|
|
@pytest.mark.parametrize("cls", [AllWeatherStrategy, AllWeatherExStrategy],
|
|
ids=["原版", "ex"])
|
|
def test_all_weather_stop_loss_adds_by_total_value(cls):
|
|
"""stop_loss 补跌加仓:order_value = 总资产/n_pick(非 现金/n_pick)。
|
|
|
|
形态:STOP 触发 -8% 止损(num_sold=1)→ HOLD 补跌加仓 1 只;
|
|
n_pick=1 → 加仓额=总资产 10500,现金口径只会下 1000。
|
|
"""
|
|
broker = _RecordingBroker()
|
|
strat = cls(provider=MagicMock(), broker=broker)
|
|
strat.yesterday_hl_list = []
|
|
strat.config.stop_loss_pct = 0.92
|
|
strat.config.stock_num = 5
|
|
ctx = FakeContext(
|
|
current_dt=datetime(2026, 8, 26, 14, 0),
|
|
positions={
|
|
"STOP.XSHG": FakePosition("STOP.XSHG", avg_cost=10.0, price=8.9, total=100),
|
|
"HOLD.XSHG": FakePosition("HOLD.XSHG", avg_cost=10.0, price=9.7, total=100),
|
|
},
|
|
cash=1000.0,
|
|
)
|
|
ctx.portfolio.total_value = 10500.0
|
|
strat.stop_loss(ctx)
|
|
assert broker.orders == [("STOP.XSHG", 0)] # 止损卖出
|
|
assert broker.values == [("HOLD.XSHG", pytest.approx(10500.0))]
|
|
assert broker.values[0][1] > 1000.0 # 现金口径回归红线
|
|
|
|
|
|
@pytest.mark.parametrize("cls", [AllWeatherStrategy, AllWeatherExStrategy],
|
|
ids=["原版", "ex"])
|
|
def test_all_weather_monthly_spreads_total_over_empty_slots(cls):
|
|
"""monthly_adjustment 买入腿:每股 = 总资产/空余槽位(非 现金/空余槽位)。
|
|
|
|
旧仓 2 只(卖出后视图滞后仍计 2)+ 目标 3 只 → 空余 1 槽 → 每只 10500,
|
|
现金口径只会下 1000。
|
|
"""
|
|
broker = _RecordingBroker()
|
|
strat = cls(provider=MagicMock(), broker=broker)
|
|
strat.yesterday_hl_list = []
|
|
strat.config.trend_threshold = 0.01
|
|
strat._stock_pool = lambda idx, d: ["B1.XSHG"] if "000300" in idx else ["S1.XSHG"]
|
|
strat._market_cap_top = lambda stocks, d, top, n: list(stocks)
|
|
strat._trend_mean = lambda lst, d, w: 1.0 if lst and "B1" in lst[0] else 0.5
|
|
strat._pick_big_universe = lambda b, cur, prev: [
|
|
"NEW_1.XSHG", "NEW_2.XSHG", "NEW_3.XSHG"]
|
|
strat._get_limit_status = lambda stocks, date: _falsy_status(stocks)
|
|
strat.monthly_adjustment(_stale_cash_ctx())
|
|
sells = {c for c, v in broker.orders if v == 0}
|
|
buys = [(c, v) for c, v in broker.orders if v > 0]
|
|
assert sells == {"OLD_A.XSHG", "OLD_B.XSHG"}
|
|
assert [c for c, _ in buys] == ["NEW_1.XSHG", "NEW_2.XSHG", "NEW_3.XSHG"]
|
|
for _, v in buys:
|
|
assert v == pytest.approx(10500.0) # 总资产/(3-2 空余槽)
|
|
assert v > 1000.0
|