test(strategy): A3预检——B2定寸虚拟化×A1通道三明治集成测试5条(issue#29) [vps]

复刻 live_strategy._setup 实盘接线(A1 broker通道注入+B2 wrap_scheduler代理
context 同开=VPS重建后将运行的真实状态),真实 LiveInstanceLedger+真策略驱动:
①available_cash 过代理=账本现金(99.3万预算口径,非全账户396万)
②channel_test 等权买入 per=(账本现金+Σ自己市值)/N≈50万/只(hold_n=2),
  精确断言 spec §A3 验收口径(非168万=995万/6量级)
③三明治下轮换只卖台账标的(别家600036不动)
④止损链路三层通:账本avg_cost+真账户现价透传(6.0<7.0*0.92触发,只卖自己)
⑤无账本 make_proxy_context 原样返回(模拟盘零改动铁律)
19/19绿;portfolio 367绿(3失败=Mac缺bullet_trade环境性,NAS CI过)

tests/portfolio/test_instance_view_isolation.py

@@ -13,6 +13,8 @@ from __future__ import annotations
 from datetime import datetime
 from unittest.mock import MagicMock
 
+import pytest
+
 from tests.portfolio.conftest import FakeContext, FakePosition
 from sanguo_portfolio.strategies import (
     ChannelTestConfig,
@@ -256,3 +258,118 @@ class TestAllWeatherStopLossIsolation:
             if c.args[1] == 0
         }
         assert sell_codes == {"518880.XSHG"}
+
+
+# =================== A3 预检:B2 定寸虚拟化 × A1 通道 三明治(=VPS 实盘真实状态) ===================
+# live_strategy._setup 的接线:A1 broker 通道注入 + B2 wrap_scheduler 代理 context
+# 同时生效。本组用真实 LiveInstanceLedger + make_proxy_context 驱动真策略,
+# 验证 spec §A3 验收口径(定寸=预算/N,卖出只卖自己)在部署前就成立。
+from sanguo_portfolio.live_instance_ledger import LiveInstanceLedger  # noqa: E402
+from sanguo_portfolio.live_portfolio_proxy import make_proxy_context  # noqa: E402
+from sanguo_portfolio.strategies.all_weather import _available_cash  # noqa: E402
+
+
+def _own_ledger() -> LiveInstanceLedger:
+    """预算 100 万的实例账本,买过 518880×1000@7.0(现金≈99.3万)。"""
+    led = LiveInstanceLedger(initial_cash=1_000_000.0)
+    led.apply_trade(is_buy=True, symbol="518880.XSHG", price=7.0, volume=1000,
+                    trade_id="T-BUY-1", trade_date="2024-10-08")
+    return led
+
+
+def _full_account_ctx() -> FakeContext:
+    """共享 QMT 全账户(前后端 session 8-19 实锤口径):现金 396 万+
+    本实例 518880 + 别家 600036,总账户≈995 万量级。"""
+    return FakeContext(
+        current_dt=datetime(2024, 10, 8, 9, 35),
+        cash=3_961_768.0,
+        positions={
+            "518880.XSHG": FakePosition("518880.XSHG", avg_cost=7.0, price=7.5),
+            "600036.XSHG": FakePosition("600036.XSHG", avg_cost=35.0, price=36.0),
+        },
+    )
+
+
+class TestB2SizingSandwich:
+    """B2(4a9208d)+A1(5a91be2) 同开:定寸走账本现金,卖出走账本标的。"""
+
+    def _sandwich(self, strategy, ledger):
+        """复刻 live_strategy._setup 两层接线(A1 通道 + B2 代理 context)。"""
+        strategy.broker.get_instance_positions = ledger.positions_view
+        return make_proxy_context(_full_account_ctx(), ledger)
+
+    def test_available_cash_is_ledger_cash_not_account(self):
+        """定寸读数=账本现金(≈99.3万),不是全账户 396 万——168万/只污染根治点。"""
+        led = _own_ledger()
+        ctx = make_proxy_context(_full_account_ctx(), led)
+        cash = _available_cash(ctx)
+        assert cash == led.cash
+        assert 990_000 < cash < 1_000_000   # 预算口径
+        assert cash < 3_000_000              # 绝不是全账户
+
+    def test_channel_test_buy_sized_by_budget(self):
+        """spec §A3 验收:channel_test 等权买入 per=(账本现金+Σ自己市值)/N,
+        ≈50万/只(hold_n=2),不再是 168万(=995万/6)量级。"""
+        led = _own_ledger()
+        broker = _RecordingBroker()
+        s = ChannelTestStrategy(
+            provider=None, broker=broker,
+            config=ChannelTestConfig(hold_n=2, period=1, probe_t1=False,
+                                     intraday_partial=False, intraday_swap=False),
+        )
+        ctx = self._sandwich(s, led)
+        s.rotate(ctx)
+        buys = [(c, v) for c, v in broker.calls if v > 0]
+        assert len(buys) == 2
+        own_mv = 7.5 * 1000  # 518880 现价×量(现价从真 portfolio 透传)
+        expect_per = (led.cash + own_mv) / 2
+        for _code, v in buys:
+            assert v == pytest.approx(expect_per, rel=1e-6)
+        assert expect_per < 600_000  # 预算/2 量级,非全账户/2
+
+    def test_sandwich_sells_only_own(self):
+        """三明治下轮换只卖台账里的 518880,别家 600036 不动。"""
+        led = _own_ledger()
+        broker = _RecordingBroker()
+        s = ChannelTestStrategy(
+            provider=None, broker=broker,
+            config=ChannelTestConfig(hold_n=2, period=1, probe_t1=False,
+                                     intraday_partial=False, intraday_swap=False),
+        )
+        ctx = self._sandwich(s, led)
+        s.rotate(ctx)
+        sells = [c for c, v in broker.calls if v == 0]
+        assert sells == ["518880.XSHG"]
+
+    def test_stop_loss_price_flows_through_both_layers(self):
+        """止损链路三层通:账本 avg_cost(7.0) + 真账户现价透传(6.0) → 触发;
+        别家 600036 同跌不动。"""
+        led = _own_ledger()
+        from sanguo_portfolio.strategies import AllWeatherStrategy
+        provider = MagicMock(name="provider")
+        provider.get_limit_status_batch.side_effect = lambda codes, date=None: {
+            c: {"is_limit_up": False, "is_limit_down": False, "is_paused": False}
+            for c in codes
+        }
+        broker = BrokerFacade()
+        broker.order_target_value = MagicMock(return_value=MagicMock(filled=100))
+        s = AllWeatherStrategy(provider=provider, broker=broker)
+        real_ctx = FakeContext(
+            current_dt=datetime(2024, 10, 8, 14, 0),
+            positions={
+                "518880.XSHG": FakePosition("518880.XSHG", avg_cost=7.0, price=6.0),
+                "600036.XSHG": FakePosition("600036.XSHG", avg_cost=40.0, price=30.0),
+            },
+        )
+        s.broker.get_instance_positions = led.positions_view
+        s.stop_loss(make_proxy_context(real_ctx, led))
+        sell_codes = {
+            c.args[0] for c in broker.order_target_value.call_args_list
+            if c.args[1] == 0
+        }
+        assert sell_codes == {"518880.XSHG"}   # 6.0 < 7.0*0.92=6.44 触发,只卖自己
+
+    def test_no_ledger_proxy_is_identity(self):
+        """无账本(回测/影子/模拟盘)→ make_proxy_context 原样返回,零改动铁律。"""
+        ctx = _full_account_ctx()
+        assert make_proxy_context(ctx, None) is ctx