sanguo_vnpy_v2/tests/portfolio/test_sizing_total_value.py

"""定寸总资产口径(A 修法,前后端 session 2026-08-25 移交)。

08-25 9:30 事故:small_cap 同轮「全卖19只→马上全买20只」,台账现金只被
runner_live 的 60s 归因轮询更新(唯一 cash 更新入口)→ 买入定寸在两轮轮询
之间评估 → 卖出回款不可见 → 20 笔买入全部目标 0,全天空仓。B 修法
(191270c,下单返回即时归因入账)已做;本组测试钉 **A 修法=防御纵深第二层**:
定寸口径回总资产 per_value = 总资产/N(现金+持仓市值,卖出前后不变),
即使台账归因滞后/异常,定寸也不再依赖未入账的现金。

原 JQ 语义 ``per_value = stock_value/len`` 本就是总资产口径——cash 口径是
移植时引入的失真,本修法=回原语义。守卫同步 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 (
    MomentumTimingExStrategy,
    SmallCapExStrategy,
    ValueSelectionExStrategy,
)
from sanguo_portfolio.strategies.all_weather import _total_value


# =================== 公共装配 ===================
class _RecordingBroker:
    """记录 order_target_value 委托;**不动 context**——精确模拟「卖出已成交、
    台账归因未跑」的现金窗口(事故形态)。"""

    def __init__(self) -> None:
        self.orders: List[Tuple[str, float]] = []

    def order_target_value(self, code: str, value: float) -> Any:
        self.orders.append((code, float(value)))
        return object()  # 非 None = 下单成功


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 _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)


# =================== 策略层:卖出未入账时定寸不缩水 ===================
def test_momentum_rotate_sizes_by_total_value():
    """momentum _rotate_positions:总资产定寸,对未入账卖出免疫。"""
    broker = _RecordingBroker()
    strat = MomentumTimingExStrategy(provider=MagicMock(), broker=broker)
    strat._rotate_positions(_stale_cash_ctx(), ["NEW_1.XSHG", "NEW_2.XSHG", "NEW_3.XSHG"])
    _assert_sized_by_total(broker)


def test_small_cap_rebalance_sizes_by_total_value():
    """small_cap _rebalance:同型回归。"""
    broker = _RecordingBroker()
    strat = SmallCapExStrategy(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)


def test_value_monthly_adjustment_sizes_by_total_value():
    """value monthly_adjustment:同型回归(选股段打桩,只验定寸腿)。"""
    broker = _RecordingBroker()
    strat = ValueSelectionExStrategy(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: {
        s: {"is_limit_up": False, "is_limit_down": False, "is_paused": False}
        for s in stocks
    }
    strat.monthly_adjustment(_stale_cash_ctx())
    _assert_sized_by_total(broker)