"""实例虚拟账本(live_instance_ledger)+ runner_live 归因链路回归。 背景(2026-08-19 三日体检):8 路组合实盘共享一个 miniQMT 账号, context.portfolio=全账户视图 → channel_test 轮换互卖别家持仓、对账 8 对 全 FAIL、收益率=全账户/初始资金无意义。修复=每实例一份由**自身真实成交** 驱动的虚拟子账本(engine.get_trades() 按 order_id ∈ engine.get_orders() 归因)。 """ from __future__ import annotations import sqlite3 from datetime import datetime from types import SimpleNamespace import pytest from sanguo_portfolio.live_instance_ledger import ( LiveInstanceLedger, estimate_fee, set_active, ) from sanguo_portfolio.runner_live import _snapshot_once, _sync_instance_trades # ------------------ 账本算术 ------------------ class TestLedgerArithmetic: def test_buy_sell_with_estimated_fees(self): """手算:100万 + 买100@10(费5) + 买200@13(费5) + 卖300@12(费5+税3.6)。""" led = LiveInstanceLedger(initial_cash=1_000_000) assert led.apply_trade(True, "000001.XSHE", 10.0, 100, "t1", "2026-08-19") assert led.apply_trade(True, "000001.XSHE", 13.0, 200, "t2", "2026-08-19") # 998,995 → 996,390;持仓 300 股,加权成本 (1000+2600)/300 = 12 assert led.cash == pytest.approx(996_390.0) assert led.positions["000001.XSHE"]["volume"] == 300 assert led.positions["000001.XSHE"]["avg_cost"] == pytest.approx(12.0) assert led.apply_trade(False, "000001.XSHE", 12.0, 300, "t3", "2026-08-19") # 卖出费 = max(3600×0.0003,5)=5 + 印花税 3600×0.001=3.6 assert led.cash == pytest.approx(996_390.0 + 3600 - 8.6) assert "000001.XSHE" not in led.positions def test_actual_fee_overrides_estimate(self): led = LiveInstanceLedger(initial_cash=100_000) led.apply_trade(True, "600000.XSHG", 10.0, 100, "t1", "2026-08-19", fee=25.0) assert led.cash == pytest.approx(100_000 - 1000 - 25) def test_dup_trade_id_ignored(self): led = LiveInstanceLedger() assert led.apply_trade(True, "600000.XSHG", 10.0, 100, "t1", "2026-08-19") assert not led.apply_trade(True, "600000.XSHG", 10.0, 100, "t1", "2026-08-19") assert led.positions["600000.XSHG"]["volume"] == 100 def test_restore_from_trades_rebuilds(self): """重启恢复:DB 行重放出现金/持仓/幂等(与 live_trades 行格式一致)。""" led = LiveInstanceLedger(initial_cash=1_000_000) rows = [ {"direction": "buy", "symbol": "000001.XSHE", "price": 10.0, "volume": 100, "traded_at": "2026-08-18 09:35:00", "vt_tradeid": "a1"}, {"direction": "buy", "symbol": "600000.XSHG", "price": 20.0, "volume": 200, "traded_at": "2026-08-18 09:35:01", "vt_tradeid": "a2"}, {"direction": "sell", "symbol": "600000.XSHG", "price": 21.0, "volume": 200, "traded_at": "2026-08-19 13:45:00", "vt_tradeid": "a3"}, ] assert led.restore_from_trades(rows) == 3 # 100万 −(1000+5) −(4000+5) +(4200−max(1.26,5)=5−4200×0.001=4.2→9.2) assert led.cash == pytest.approx(1_000_000 - 1005 - 4005 + 4190.8) assert led.positions["000001.XSHE"]["volume"] == 100 assert "600000.XSHG" not in led.positions # 重放幂等:同批行再来一遍零增量 assert led.restore_from_trades(rows) == 0 def test_t1_closeable_today_then_next_day(self): led = LiveInstanceLedger() led.apply_trade(True, "000001.XSHE", 10.0, 100, "t1", "2026-08-19") view_today = led.positions_view(now_date="2026-08-19") assert view_today["000001.XSHE"]["closeable_amount"] == 0 # T+1 锁定 view_next = led.positions_view(now_date="2026-08-20") assert view_next["000001.XSHE"]["closeable_amount"] == 100 def test_equity_price_fallback_to_avg_cost(self): led = LiveInstanceLedger(initial_cash=100_000) led.apply_trade(True, "000001.XSHE", 10.0, 100, "t1", "2026-08-19") cash, mv, total = led.equity({}) # 无现价 → 退加权成本 10 assert mv == pytest.approx(1000) assert total == pytest.approx(cash + 1000) _, mv2, _ = led.equity({"000001.XSHE": 12.5}) assert mv2 == pytest.approx(1250) def test_estimate_fee_matches_order_cost(self): assert estimate_fee(True, 10.0, 100) == pytest.approx(5.0) # 佣金触底 assert estimate_fee(False, 10.0, 100_000) == pytest.approx( max(1_000_000 * 0.0003, 5) + 1_000_000 * 0.001) def test_sell_without_book_position_keeps_cash_no_crash(self): """bootstrap 缺口前的旧仓卖出:账上无此标的——现金照收,持仓无账可扣不崩。""" led = LiveInstanceLedger(initial_cash=100_000) assert led.apply_trade( False, "600519.XSHG", 1000.0, 100, "t1", "2026-08-19") assert led.cash == pytest.approx(100_000 + 100_000 - 130.0) # 佣金30+税100 assert led.positions == {} def test_dirty_flag_drives_snapshot_throttle(self): """新成交→dirty=True(下个快照周期必写);初始/重放后同样置位。""" led = LiveInstanceLedger() led.dirty = False led.apply_trade(True, "000001.XSHE", 10.0, 100, "t1", "2026-08-19") assert led.dirty is True led2 = LiveInstanceLedger() led2.dirty = False led2.restore_from_trades([{ "direction": "buy", "symbol": "000001.XSHE", "price": 10.0, "volume": 100, "traded_at": "2026-08-19 09:35:00", "vt_tradeid": "r1"}]) assert led2.dirty is True # ------------------ 归因与落库链路 ------------------ def _fake_engine(): """两个成交:o1(本实例买单)/ FOREIGN(别家实例单)。""" own = SimpleNamespace(order_id="o1", is_buy=True) t_own = SimpleNamespace( order_id="o1", security="000001.XSHE", amount=100, price=10.0, time=datetime(2026, 8, 19, 9, 35, 0), commission=0.0, tax=0.0) t_foreign = SimpleNamespace( order_id="8800099", security="600519.XSHG", amount=500, price=1500.0, time=datetime(2026, 8, 19, 9, 36, 0), commission=0.0, tax=0.0) return SimpleNamespace( get_orders=lambda: {"o1": own}, get_trades=lambda: {"t1": t_own, "t99": t_foreign}, context=SimpleNamespace(portfolio=SimpleNamespace( positions={"000001.XSHE": SimpleNamespace(price=11.0)})), ) class TestAttributionAndSnapshot: def test_sync_attributes_only_own_orders(self, tmp_path): from sanguo_live.persistence import init_db, list_trades db = str(tmp_path / "live.db") init_db(db) led = LiveInstanceLedger(initial_cash=1_000_000) engine = _fake_engine() _sync_instance_trades(engine, led, db, 44, "channel_test") # 账本只有本实例成交;别家 500 股×1500 不进账 assert led.positions["000001.XSHE"]["volume"] == 100 assert led.cash == pytest.approx(1_000_000 - 1005) rows = list_trades(db, 44) assert len(rows) == 1 assert rows[0]["vt_tradeid"] == "t1" assert rows[0]["direction"] == "buy" assert rows[0]["offset"] == "open" assert rows[0]["strategy_name"] == "channel_test" def test_sync_idempotent_no_duplicate_rows(self, tmp_path): from sanguo_live.persistence import init_db, list_trades db = str(tmp_path / "live.db") init_db(db) led = LiveInstanceLedger() engine = _fake_engine() _sync_instance_trades(engine, led, db, 44, "channel_test") _sync_instance_trades(engine, led, db, 44, "channel_test") assert len(list_trades(db, 44)) == 1 def test_snapshot_writes_instance_view_not_full_account(self, tmp_path): """快照落库=实例视图(旧版落全账户持仓是互卖/对账错的根源)。""" from sanguo_live.persistence import init_db, list_balance, load_positions db = str(tmp_path / "live.db") init_db(db) led = LiveInstanceLedger(initial_cash=1_000_000) led.apply_trade(True, "000001.XSHE", 10.0, 100, "t1", "2026-08-19") engine = _fake_engine() _snapshot_once(engine, db, 44, led) pos = load_positions(db, 44) assert len(pos) == 1 assert pos[0]["symbol"] == "000001.XSHE" assert pos[0]["volume"] == 100 assert pos[0]["frozen"] == 100 # T+1 当日买入 assert pos[0]["avg_price"] == pytest.approx(10.0) bal = list_balance(db, 44)[-1] # 虚拟账本:cash=998,995;市值按现价 11 → 1100 assert bal["cash"] == pytest.approx(998_995.0) assert bal["market_value"] == pytest.approx(1100.0) assert bal["total"] == pytest.approx(1_000_095.0) # ------------------ 适配层通道注入 ------------------ @pytest.fixture(autouse=True) def _clear_active(): set_active(None) yield set_active(None) def _patch_wiring(monkeypatch): """打桩 bullet_trade 装配依赖(对齐 tests/api/test_portfolio_live 模式)。""" import bullet_trade.core as bt_core import bullet_trade.data.api as bt_data_api monkeypatch.setattr(bt_core, "run_daily", lambda f, t, **kw: None) monkeypatch.setattr(bt_core, "run_monthly", lambda f, d, t, **kw: None) monkeypatch.setattr(bt_data_api, "get_data_provider", lambda: SimpleNamespace()) class TestFacadeChannel: def test_setup_injects_when_ledger_active(self, monkeypatch): from sanguo_portfolio import live_strategy _patch_wiring(monkeypatch) live_strategy._STATE.update(strategy=None, wired=False) monkeypatch.setenv("SANGUO_LIVE_STRATEGY", "channel_test") led = LiveInstanceLedger(initial_cash=500_000) set_active(led) live_strategy._setup(SimpleNamespace()) strategy = live_strategy._STATE["strategy"] assert callable(strategy.broker.get_instance_positions) view = strategy.broker.get_instance_positions() assert view == {} live_strategy._STATE.update(strategy=None, wired=False) def test_setup_leaves_none_without_ledger(self, monkeypatch): """回测/无账本:通道保持 None,策略侧回退 context.portfolio。""" from sanguo_portfolio import live_strategy _patch_wiring(monkeypatch) live_strategy._STATE.update(strategy=None, wired=False) monkeypatch.setenv("SANGUO_LIVE_STRATEGY", "channel_test") live_strategy._setup(SimpleNamespace()) strategy = live_strategy._STATE["strategy"] assert strategy.broker.get_instance_positions is None live_strategy._STATE.update(strategy=None, wired=False)