feat(portfolio): B2定寸虚拟化InstancePortfolio代理——策略决策层脱离全账户污染(spec§B2+A2清单issue#29)——①live_portfolio_proxy:InstanceContextProxy(portfolio→账本视图,其余属性透传真context,property+__getattr__实现)+InstancePortfolio(每次访问现算)②覆盖A2清单全集:available_cash/cash=账本现金(168万/只定寸污染根治点,.cash fallback同源);positions=账本jq对象(total_amount/amount/closeable_amount(T+1)/avg_cost(台账口径=成交价)/cost_basis/security)+price/last_sale_price从真portfolio同名标的透传(市场数据非所有权,缺→None止损跳过);value/market_value/total_value=现价×量(缺价→0偏保守不误卖);total_value/positions_value=账本equity(现价优先成本兜底);locked_cash=0③清单外属性AttributeError fail-fast(真值=全账户数字,静默透传=污染复发,宁可崩=A1同款house style)④wrap_scheduler包装facade的run_daily/run_monthly(发生在策略initialize注册任务之前,策略代码零改动);functools.wraps保任务名;无账本原样返回=回测/影子零改动⑤live_strategy._setup接线(_ledger None→wrap直通);引擎内部撮合/风控仍看真实账户;+13测试(定寸现金/T+1/现价透传/缺价保守/equity兜底/fail-fast/调度代理注入/无账本零改动/_setup接线两态);portfolio 373绿 [vps]

2026-08-19 22:37:16 +08:00
parent ed15c24db8
commit 4a9208db9d
3 changed files with 412 additions and 0 deletions
@@ -0,0 +1,213 @@
+"""Tests for B2 定寸虚拟化(spec §B2 + A2 属性清单 issue#29)。
+
+核心断言:策略决策层看到的 portfolio = 实例账本视图(现金/持仓/定寸不再被
+全账户污染);现价从真 portfolio 透传(市场数据非所有权);未覆盖属性
+fail-fast;无账本(回测/影子)零改动。
+"""
+from types import SimpleNamespace
+
+import pytest
+
+from sanguo_portfolio.live_instance_ledger import (
+    LiveInstanceLedger, set_active,
+)
+from sanguo_portfolio.live_portfolio_proxy import (
+    InstanceContextProxy, InstancePortfolio, make_proxy_context,
+    wrap_scheduler,
+)
+
+
+class _RealPos:
+    def __init__(self, security, price, amount=0):
+        self.security = security
+        self.price = price
+        self.last_sale_price = price
+        self.total_amount = amount
+
+
+@pytest.fixture()
+def ledger():
+    led = LiveInstanceLedger(initial_cash=1_000_000)
+    # 买入 510300@4.74×4200(费用=est) → cash 减,持仓进账本
+    led.apply_trade(is_buy=True, symbol="510300.XSHG", price=4.74,
+                    volume=4200, trade_id="t1", trade_date="2026-08-19")
+    return led
+
+
+def _real_portfolio(positions):
+    return SimpleNamespace(positions=positions)
+
+
+def _portfolio(ledger, real_positions=None):
+    return InstancePortfolio(
+        ledger, lambda: _real_portfolio(real_positions or {}))
+
+
+# ---------------- 现金/定寸(A2 核心:污染点) ----------------
+
+def test_available_cash_is_ledger_not_account(ledger):
+    """定寸读到的现金=账本现金,不是共享账户 995 万。"""
+    p = _portfolio(ledger)
+    # 账本现金 = 100万 − 4200×4.74 − 佣金 max(19908×0.0003,5)=5.9724 → 980086.0276
+    assert p.available_cash == pytest.approx(
+        1_000_000 - 19_908 - 5.9724, rel=1e-9)
+    assert p.cash == p.available_cash  # _available_cash 的 .cash fallback 同源
+
+
+def test_positions_from_ledger_with_t1(ledger):
+    """持仓=账本视图:当日买入 T+1 → closeable=0,amount=总量,成本=账本均价。"""
+    p = _portfolio(ledger, real_positions={
+        "510300.XSHG": _RealPos("510300.XSHG", 4.80)})
+    pos = p.positions["510300.XSHG"]
+    assert pos.security == "510300.XSHG"
+    assert pos.total_amount == 4200
+    assert pos.amount == 4200
+    assert pos.closeable_amount == 0  # 今日买入,T+1 锁定
+    assert pos.avg_cost == pytest.approx(4.74)  # 台账口径=成交价(不含费)
+    assert pos.cost_basis == pos.avg_cost
+    # 现价从真 portfolio 透传(市场数据非所有权)
+    assert pos.price == 4.80
+    assert pos.last_sale_price == 4.80
+    assert pos.value == pytest.approx(4.80 * 4200)
+
+
+def test_price_missing_value_zero(ledger):
+    """真 portfolio 无同名标的 → price None,value 0(策略当 0 处理,偏保守)。"""
+    p = _portfolio(ledger, real_positions={})
+    pos = p.positions["510300.XSHG"]
+    assert pos.price is None
+    assert pos.value == 0.0
+    assert pos.market_value == 0.0
+
+
+def test_positions_empty_ledger(ledger):
+    """账本无持仓 → 空 dict(轮换退化为只买不卖=正确行为)。"""
+    empty = LiveInstanceLedger(initial_cash=100_000)
+    p = _portfolio(empty)
+    assert p.positions == {}
+
+
+def test_real_positions_list_form_indexed(ledger):
+    """真 positions 为 list(bullet_trade 兼容路径)→ 按 .security 索引。"""
+    p = _portfolio(ledger, real_positions=[
+        _RealPos("600036.XSHG", 38.0), _RealPos("510300.XSHG", 4.75)])
+    assert p.positions["510300.XSHG"].price == 4.75
+
+
+# ---------------- equity / 未覆盖属性 ----------------
+
+def test_total_value_is_ledger_equity(ledger):
+    """total_value=账本 equity(现价优先,缺价成本兜底),非全账户。"""
+    p = _portfolio(ledger, real_positions={
+        "510300.XSHG": _RealPos("510300.XSHG", 4.80)})
+    cash = ledger.cash
+    assert p.total_value == pytest.approx(cash + 4.80 * 4200)
+    assert p.positions_value == pytest.approx(4.80 * 4200)
+
+
+def test_total_value_cost_fallback(ledger):
+    """缺现价 → 账本 equity 的成本价兜底。"""
+    p = _portfolio(ledger)
+    _c, _mv, total = ledger.equity({})
+    assert p.total_value == pytest.approx(total)
+
+
+def test_locked_cash_zero_and_uncovered_fail_fast(ledger):
+    """locked_cash 安全值 0;清单外属性 AttributeError(拒绝静默透传全账户数字)。"""
+    p = _portfolio(ledger)
+    assert p.locked_cash == 0.0
+    with pytest.raises(AttributeError, match="inout"):
+        _ = p.inout
+
+
+# ---------------- context 代理 ----------------
+
+def test_context_proxy_passthrough_and_portfolio(ledger):
+    real = SimpleNamespace(current_dt="2026-08-19 09:35:00",
+                           previous_date="2026-08-18",
+                           portfolio=_real_portfolio({}))
+    proxy = InstanceContextProxy(real, ledger)
+    assert proxy.current_dt == "2026-08-19 09:35:00"
+    assert proxy.previous_date == "2026-08-18"
+    assert isinstance(proxy.portfolio, InstancePortfolio)
+    assert proxy.portfolio.cash == ledger.cash
+
+
+def test_make_proxy_context_none_ledger_identity():
+    ctx = SimpleNamespace(x=1)
+    assert make_proxy_context(ctx, None) is ctx
+
+
+# ---------------- 调度器包装 ----------------
+
+def test_wrap_scheduler_callbacks_get_proxy(ledger):
+    """经包装注册的回调收到代理 context(portfolio=账本视图)。"""
+    captured = {}
+
+    def bt_sched(func, time, **kw):
+        captured["func"] = func
+        captured["time"] = time
+
+    sched = wrap_scheduler(bt_sched, ledger)
+    seen_ctx = {}
+
+    def strategy_task(context):
+        seen_ctx["cash"] = context.portfolio.available_cash
+        seen_ctx["dt"] = context.current_dt
+
+    strategy_task.__name__ = "strategy_task"
+    sched(strategy_task, "09:35")
+    # 调度器侧拿到的 wrapped 保留原函数名(任务命名不受影响)
+    assert captured["func"].__name__ == "strategy_task"
+    real_ctx = SimpleNamespace(
+        current_dt="2026-08-19 09:35:00",
+        portfolio=_real_portfolio(
+            {"510300.XSHG": _RealPos("510300.XSHG", 4.80)}))
+    captured["func"](real_ctx)
+    assert seen_ctx["cash"] == pytest.approx(ledger.cash)
+    assert seen_ctx["dt"] == "2026-08-19 09:35:00"
+
+
+def test_wrap_scheduler_none_ledger_passthrough():
+    """无账本 → 原调度器原样返回(回测/影子零改动)。"""
+    def bt_sched(func, *a, **kw):
+        return "registered"
+
+    assert wrap_scheduler(bt_sched, None) is bt_sched
+
+
+def test_setup_wraps_facade_schedulers(monkeypatch):
+    """_setup 接线:有账本 → facade.run_daily 被换包装;无账本 → 原样。"""
+    from sanguo_portfolio import live_strategy
+
+    def _patch(monkeypatch):
+        import bullet_trade.core as bt_core
+        import bullet_trade.data.api as bt_data_api
+
+        def fake_run_daily(func, t, **kw):
+            return None
+
+        monkeypatch.setattr(bt_core, "run_daily", fake_run_daily)
+        monkeypatch.setattr(bt_core, "run_monthly",
+                            lambda f, d, t, **kw: None)
+        monkeypatch.setattr(bt_data_api, "get_data_provider",
+                            lambda: SimpleNamespace())
+        return bt_core
+
+    bt_core = _patch(monkeypatch)
+    monkeypatch.setenv("SANGUO_LIVE_STRATEGY", "channel_test")
+    try:
+        live_strategy._STATE.update(strategy=None, wired=False)
+        set_active(LiveInstanceLedger(initial_cash=100_000))
+        live_strategy._setup(SimpleNamespace())
+        assert live_strategy._STATE["strategy"].broker.run_daily \
+            is not bt_core.run_daily  # 已包装
+
+        live_strategy._STATE.update(strategy=None, wired=False)
+        set_active(None)
+        live_strategy._setup(SimpleNamespace())
+        assert live_strategy._STATE["strategy"].broker.run_daily \
+            is bt_core.run_daily  # 原样
+    finally:
+        live_strategy._STATE.update(strategy=None, wired=False)
+        set_active(None)