"""D-4c 模式 B reconcile 集成测试（bridge 回报校正 account 账本）。 不依赖 Windows——mock BridgeClient.get_account/get_positions，验证： - mode_b=false / enabled=false → 跳过（不创建 BridgeClient） - mode_b=true + bridge ok → cash/positions 校正 + code 转换 + 持久化 - bridge 失败（get_account/get_positions 返回 None）→ 不阻断、降级（不持久化） - 无 BRIDGE_TOKEN → 跳过 参考 tests/trader/test_shadow_orders.py 的 mock 风格。 """ from types import SimpleNamespace from unittest.mock import patch from sanguo_trader.account import Account from sanguo_trader.bridge_client import from_bridge_code from sanguo_trader.live_orchestrator import reconcile_from_bridge from sanguo_trader.persistence import ( init_db, load_last_balance, load_positions, ) _DATE = "2026-07-11" def _mk_db(tmp_path) -> str: db = str(tmp_path / "t.db") init_db(db) return db def _cfg(mode_b: bool, enabled: bool = True) -> SimpleNamespace: return SimpleNamespace(live={ "enabled": enabled, "bridge_url": "http://b.test", "shadow": True, "mode_b": mode_b, }) class TestFromBridgeCode: def test_sh_prefix(self): assert from_bridge_code("sh600000") == "600000" def test_sz_prefix(self): assert from_bridge_code("sz000001") == "000001" def test_already_plain(self): assert from_bridge_code("600000") == "600000" class TestReconcile: def test_mode_b_false_skips(self, tmp_path): """mode_b=false → 不创建 BridgeClient。""" db = _mk_db(tmp_path) acc = Account(100000.0) with patch("sanguo_trader.bridge_client.BridgeClient") as mbc: reconcile_from_bridge(db, 1, _DATE, acc, _cfg(False)) mbc.assert_not_called() def test_enabled_false_skips(self, tmp_path): """enabled=false → 跳过（即使 mode_b=true）。""" db = _mk_db(tmp_path) acc = Account(100000.0) with patch("sanguo_trader.bridge_client.BridgeClient") as mbc: reconcile_from_bridge(db, 1, _DATE, acc, _cfg(True, enabled=False)) mbc.assert_not_called() def test_no_token_skips(self, tmp_path): """mode_b=true 但无 BRIDGE_TOKEN → 跳过。""" db = _mk_db(tmp_path) acc = Account(100000.0) with patch.dict("os.environ", {}, clear=True), \ patch("sanguo_trader.bridge_client.BridgeClient") as mbc: reconcile_from_bridge(db, 1, _DATE, acc, _cfg(True)) mbc.assert_not_called() def test_reconciles_account_and_positions(self, tmp_path): """mode_b=true + bridge ok → cash/positions 校正 + code 转换 + 持久化。""" db = _mk_db(tmp_path) acc = Account(100000.0) acc.cash = 50000.0 # 模拟账本现金（将被 bridge 覆盖） with patch.dict("os.environ", {"BRIDGE_TOKEN": "tok"}), \ patch("sanguo_trader.bridge_client.BridgeClient") as mbc: mbc.return_value.get_account.return_value = { "ok": True, "cash": 80000.0, "frozen": 5000.0, "market_value": 120000.0, "total": 200000.0, } mbc.return_value.get_positions.return_value = [ {"code": "sh600000", "volume": 1000, "can_use": 800, "avg_price": 10.5}, {"code": "sz000001", "volume": 500, "can_use": 500, "avg_price": 15.0}, ] reconcile_from_bridge(db, 1, _DATE, acc, _cfg(True)) # account 资金校正（bridge 为准） assert acc.cash == 80000.0 assert acc.market_value == 120000.0 # positions 重建（bridge sh/sz 前缀 → 纯数字 key） assert set(acc.positions.keys()) == {"600000", "000001"} assert acc.positions["600000"].volume == 1000 assert acc.positions["600000"].frozen == 200 # volume - can_use = 1000 - 800 assert acc.positions["600000"].avg_price == 10.5 assert acc.positions["000001"].frozen == 0 # 500 - 500 # 持久化（DB 反映 bridge 校正后状态） pos = load_positions(db, 1, "account") assert "600000" in pos and pos["600000"]["volume"] == 1000 bal = load_last_balance(db, 1) assert bal["cash"] == 80000.0 assert bal["market_value"] == 120000.0 assert bal["total_equity"] == 200000.0 def test_get_account_failure_degrades(self, tmp_path): """get_account 返回 None → 不阻断、不覆盖、不持久化（降级模拟账本）。""" db = _mk_db(tmp_path) acc = Account(100000.0) original_cash = acc.cash with patch.dict("os.environ", {"BRIDGE_TOKEN": "tok"}), \ patch("sanguo_trader.bridge_client.BridgeClient") as mbc: mbc.return_value.get_account.return_value = None reconcile_from_bridge(db, 1, _DATE, acc, _cfg(True)) # 不抛 assert acc.cash == original_cash # 未被覆盖 assert load_last_balance(db, 1) is None # 未持久化 def test_get_positions_failure_degrades(self, tmp_path): """get_positions 返回 None → 不阻断、降级（不持久化 reconcile）。""" db = _mk_db(tmp_path) acc = Account(100000.0) with patch.dict("os.environ", {"BRIDGE_TOKEN": "tok"}), \ patch("sanguo_trader.bridge_client.BridgeClient") as mbc: mbc.return_value.get_account.return_value = { "ok": True, "cash": 90000.0, "frozen": 0, "market_value": 50000.0, "total": 140000.0, } mbc.return_value.get_positions.return_value = None reconcile_from_bridge(db, 1, _DATE, acc, _cfg(True)) # 不抛 # 未持久化（降级，DB 保持 live_step step6 的 simulation 状态） assert load_last_balance(db, 1) is None def test_exception_does_not_block(self, tmp_path): """bridge 调用抛异常 → try/except 兜底，不阻断。""" db = _mk_db(tmp_path) acc = Account(100000.0) with patch.dict("os.environ", {"BRIDGE_TOKEN": "tok"}), \ patch("sanguo_trader.bridge_client.BridgeClient") as mbc: mbc.return_value.get_account.side_effect = RuntimeError("boom") reconcile_from_bridge(db, 1, _DATE, acc, _cfg(True)) # 不抛 class TestReconcileTokenSource: """bridge_token 优先级：config > env > skip（reconcile 模式 B）。""" @staticmethod def _ok_account_resp(): return {"ok": True, "cash": 80000.0, "frozen": 0, "market_value": 0.0, "total": 80000.0} def test_config_token_used(self, tmp_path): """(a) config 有 bridge_token → 用 config token（env 未设）。""" db = _mk_db(tmp_path) acc = Account(100000.0) cfg = SimpleNamespace(live={ "enabled": True, "bridge_url": "http://b.test", "shadow": True, "mode_b": True, "bridge_token": "cfg-token", }) with patch.dict("os.environ", {}, clear=True), \ patch("sanguo_trader.bridge_client.BridgeClient") as mbc: mbc.return_value.get_account.return_value = self._ok_account_resp() mbc.return_value.get_positions.return_value = [] reconcile_from_bridge(db, 1, _DATE, acc, cfg) assert mbc.call_args[0][1] == "cfg-token" # BridgeClient(url, token) def test_env_fallback_when_config_missing(self, tmp_path): """(b) config 无 bridge_token → fallback 到 BRIDGE_TOKEN env。""" db = _mk_db(tmp_path) acc = Account(100000.0) cfg = SimpleNamespace(live={ "enabled": True, "bridge_url": "http://b.test", "shadow": True, "mode_b": True, }) with patch.dict("os.environ", {"BRIDGE_TOKEN": "env-token"}), \ patch("sanguo_trader.bridge_client.BridgeClient") as mbc: mbc.return_value.get_account.return_value = self._ok_account_resp() mbc.return_value.get_positions.return_value = [] reconcile_from_bridge(db, 1, _DATE, acc, cfg) assert mbc.call_args[0][1] == "env-token" def test_no_token_skips(self, tmp_path): """(c) config 和 env 都无 bridge_token → 跳过、warning。""" db = _mk_db(tmp_path) acc = Account(100000.0) cfg = SimpleNamespace(live={ "enabled": True, "bridge_url": "http://b.test", "shadow": True, "mode_b": True, }) with patch.dict("os.environ", {}, clear=True), \ patch("sanguo_trader.bridge_client.BridgeClient") as mbc: reconcile_from_bridge(db, 1, _DATE, acc, cfg) mbc.assert_not_called()