feat(live): D-4c模式B reconcile—bridge回报驱动账本(真桥验证通过)
- bridge_client: from_bridge_code(sh/sz→纯数字码, to_bridge_code逆函数) - live_orchestrator: reconcile_from_bridge 读bridge /account /positions校正account现金+持仓+持久化, 默认mode_b=false - live_step step8: 影子后调reconcile(mode_b=true生效, mode_b=false跳过) - config: live.mode_b开关(默认false模式A) - test_reconcile: 10例(cash/positions校正+code转换+失败降级+mode_b跳过) - NAS环境15 passed(reconcile10+shadow5无回归) - 真桥集成: live_step mode_b=true → reconcile读bridge → account校正(1000万/空仓=bridge真实账本)+持久化 安全: mode_b默认关+bridge失败降级不阻断+token走env
This commit is contained in:
@@ -46,3 +46,4 @@ live:
|
||||
enabled: false # 总开关（false=影子分支整个跳过，live_step 行为不变）
|
||||
bridge_url: https://bridge.mysanguo.top
|
||||
shadow: true # 模式A影子下单（模拟撮合为准，信号同步POST bridge影子）
|
||||
mode_b: false # D-4c 模式B: bridge回报校正账本（默认关，切实盘再开）
|
||||
|
||||
@@ -41,6 +41,18 @@ def to_bridge_code(symbol: str) -> str:
|
||||
return f"sh{symbol}"
|
||||
|
||||
|
||||
def from_bridge_code(code: str) -> str:
|
||||
"""bridge code（sh600000/sz000001）→ sanguo 纯数字码（to_bridge_code 逆函数）。
|
||||
|
||||
D-4c 模式 B reconcile：bridge /positions 返回 sh/sz 前缀码，
|
||||
account.positions 以纯数字码为 key（与 PaperEngine trade.symbol 一致）。
|
||||
已是纯数字则原样返回。
|
||||
"""
|
||||
if code[:2].lower() in ("sh", "sz"):
|
||||
return code[2:]
|
||||
return code
|
||||
|
||||
|
||||
class BridgeClient:
|
||||
"""QMT bridge HTTP 客户端（影子下单旁路）。
|
||||
|
||||
|
||||
@@ -23,6 +23,7 @@ from .models import AccountConfig, MatchSession, OrderSide, PaperOrder
|
||||
from .position_ledger import PositionLedger
|
||||
from .persistence import (
|
||||
load_last_balance, load_positions, save_positions,
|
||||
save_daily_balance,
|
||||
load_pending_orders, save_pending_orders,
|
||||
save_shadow_order, is_trade_shadowed,
|
||||
)
|
||||
@@ -170,6 +171,10 @@ def live_step(db_path: str, account_id: int, data_source, cfg, today: str | None
|
||||
# enabled=false 时 _shadow_trades_to_bridge 立即 return，live_step 行为完全不变
|
||||
_shadow_trades_to_bridge(db_path, account_id, today, cfg)
|
||||
|
||||
# 8. 模式 B reconcile（D-4c，spec §5.3 模式 B）：bridge 回报校正 account 账本
|
||||
# mode_b=false 时 reconcile_from_bridge 立即 return，live_step 行为不变
|
||||
reconcile_from_bridge(db_path, account_id, today, account, cfg)
|
||||
|
||||
|
||||
def _shadow_trades_to_bridge(db_path: str, account_id: int, today: str, cfg) -> None:
|
||||
"""当日成交影子下单到 bridge（D-3，spec §5 模式 A）。
|
||||
@@ -228,6 +233,77 @@ def _shadow_trades_to_bridge(db_path: str, account_id: int, today: str, cfg) ->
|
||||
logger.warning("live_step %s: 影子下单异常（不阻断）: %s", account_id, e)
|
||||
|
||||
|
||||
def reconcile_from_bridge(db_path: str, account_id: int, today: str,
|
||||
account: Account, cfg) -> None:
|
||||
"""模式 B: bridge 真实回报校正 account 账本（spec §5.2/§5.3 模式 B）。
|
||||
|
||||
bridge /account + /positions 为准，覆盖 account.cash/market_value/positions，
|
||||
纠模拟撮合漂移（实盘成交价/分红/拆股等导致的账本偏差）。
|
||||
默认关闭（cfg.live.mode_b != True）；任何失败仅记日志，不阻断 live_step。
|
||||
bridge 失败 → warning return，降级用模拟账本（live_step step 6 已存的 simulation 状态）。
|
||||
"""
|
||||
try:
|
||||
live_cfg = getattr(cfg, "live", None) or {}
|
||||
if not live_cfg.get("enabled"):
|
||||
return
|
||||
if not live_cfg.get("mode_b"):
|
||||
return
|
||||
token = os.environ.get("BRIDGE_TOKEN")
|
||||
if not token:
|
||||
logger.warning("reconcile %s: mode_b 启用但 BRIDGE_TOKEN 未设，跳过", account_id)
|
||||
return
|
||||
url = live_cfg.get("bridge_url")
|
||||
if not url:
|
||||
logger.warning("reconcile %s: mode_b 启用但 bridge_url 未配，跳过", account_id)
|
||||
return
|
||||
|
||||
from .bridge_client import BridgeClient, from_bridge_code
|
||||
|
||||
client = BridgeClient(url, token)
|
||||
|
||||
# 1. 校正资金（bridge /account 为准）
|
||||
acc_resp = client.get_account()
|
||||
if acc_resp is None:
|
||||
logger.warning("reconcile %s: get_account 失败，降级模拟账本", account_id)
|
||||
return
|
||||
account.cash = float(acc_resp.get("cash", account.cash))
|
||||
account.market_value = float(acc_resp.get("market_value", account.market_value))
|
||||
total = acc_resp.get("total")
|
||||
if total is None:
|
||||
total = account.equity
|
||||
|
||||
# 2. 重建持仓（bridge /positions 为准；bridge code → 纯数字 key）
|
||||
pos_resp = client.get_positions()
|
||||
if pos_resp is None:
|
||||
logger.warning("reconcile %s: get_positions 失败，降级模拟账本", account_id)
|
||||
return
|
||||
new_positions: dict[str, PositionLedger] = {}
|
||||
for p in pos_resp:
|
||||
code = from_bridge_code(p.get("code", ""))
|
||||
vol = int(p.get("volume", 0))
|
||||
if vol <= 0:
|
||||
continue
|
||||
can_use = int(p.get("can_use", vol))
|
||||
new_positions[code] = PositionLedger(
|
||||
code, volume=vol,
|
||||
frozen=max(vol - can_use, 0),
|
||||
avg_price=float(p.get("avg_price", 0.0)),
|
||||
)
|
||||
account.positions = new_positions
|
||||
|
||||
# 3. 持久化校正后账本（account scope + daily balance）
|
||||
save_positions(db_path, account_id, "account", {
|
||||
sym: {"volume": p.volume, "frozen": p.frozen, "avg_price": p.avg_price}
|
||||
for sym, p in account.positions.items()}, today)
|
||||
save_daily_balance(db_path, account_id, today,
|
||||
account.cash, account.market_value, total)
|
||||
logger.info("reconcile %s @%s 完成: cash=%.2f mv=%.2f positions=%d",
|
||||
account_id, today, account.cash, account.market_value,
|
||||
len(account.positions))
|
||||
except Exception as e: # noqa: BLE001 reconcile 绝不阻断 live_step
|
||||
logger.warning("reconcile %s: 异常（不阻断）: %s", account_id, e)
|
||||
|
||||
|
||||
def list_live_accounts(db_path: str) -> list[int]:
|
||||
"""所有 mode=live & status=running 的 account_id（live_runner 遍历用）。"""
|
||||
with sqlite3.connect(db_path) as conn:
|
||||
|
||||
@@ -0,0 +1,144 @@
|
||||
"""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)) # 不抛
|
||||
Reference in New Issue
Block a user