"""定寸总资产口径(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