From dae56e29aa7ad5d133eade9ba63aabe843bf8e3d Mon Sep 17 00:00:00 2001 From: claude_dev Date: Wed, 19 Aug 2026 18:39:13 +0800 Subject: [PATCH] =?UTF-8?q?fix(live):=20instance=E8=99=9A=E6=8B=9F?= =?UTF-8?q?=E8=B4=A6=E6=9C=AC+=E6=88=90=E4=BA=A4=E5=BD=92=E5=9B=A0+?= =?UTF-8?q?=E5=AE=9E=E7=9B=98=E6=97=A5=E5=BF=97=E9=BB=91=E6=B4=9E=E2=80=94?= =?UTF-8?q?=E2=80=94=E5=85=B1=E4=BA=ABQMT=E8=B4=A6=E6=88=B7=E4=B8=89?= =?UTF-8?q?=E5=AE=B3=E6=A0=B9=E6=B2=BB=E7=AC=AC=E4=B8=80=E6=AD=A5=20[vps]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 2026-08-19盘后QMT实锤:①互卖当日真实发生(518880两个实例各卖183700、 600036两实例各卖42900)②买入sizing被全账户污染(channel_test实盘买 183700股@9.12≈168万=全账户995万/6,影子才16.8万=100万/6)③QMT委托带 remark=bt:live_strateg:实例指纹(归因可交叉验证)。 ①live_instance_ledger(新模块):每live实例一份虚拟子账本(现金=初始−Σ买 −Σ费+Σ卖,持仓+移动加权成本+T+1当日买入锁定,线程安全锁),由**本实例 真实成交**驱动——engine.get_trades()按order_id∈engine.get_orders()归因 (引擎_broker_order_index已映射回本实例id空间),别家实例/手动单不进账; 卖超账面/无账面卖出如实留痕不崩;restore_from_trades重启恢复。 ②runner_live:_sync_trades旧轮询(写不进live_trades的坏件)替换为归因 落库(方向取自订单is_buy);快照落库从context.portfolio全账户改**实例视图** (positions=账本持仓T+1冻结;balance=虚拟现金+市值,现价取全账户快照/退 加权成本)——治8实例同写一份全账户持仓+收益率=全账户/初始资金无意义; balance节流:有成交立即写否则≥5分钟(治1440行/天/实例量偏大遗留)。 ③通道注入:BrokerFacade.get_instance_positions字段+live_strategy._setup 读get_active()注入positions_view(策略session消费,getattr兜底回退 context.portfolio;回测/单测无账本=保持None)。 ④实盘引擎日志黑洞根治:supervisor子进程stdout/stderr落logs/live_{aid}.log (>5MB截断+spawn标记,对齐影子#88同款修法;8-17起实盘委托/成交零留存)。 +15测试(算术/归因过滤/幂等/快照实例视图/通道注入/日志重定向); portfolio+live+shadow 400绿。 --- sanguo_live/runner.py | 26 +- sanguo_portfolio/live_instance_ledger.py | 190 +++++++++++++++ sanguo_portfolio/live_strategy.py | 9 + sanguo_portfolio/runner_live.py | 196 ++++++++------- sanguo_portfolio/strategies/all_weather.py | 4 + tests/api/test_portfolio_live.py | 60 ++++- tests/portfolio/test_live_instance_ledger.py | 236 +++++++++++++++++++ 7 files changed, 629 insertions(+), 92 deletions(-) create mode 100644 sanguo_portfolio/live_instance_ledger.py create mode 100644 tests/portfolio/test_live_instance_ledger.py diff --git a/sanguo_live/runner.py b/sanguo_live/runner.py index 839ac68..65d8fdf 100644 --- a/sanguo_live/runner.py +++ b/sanguo_live/runner.py @@ -375,10 +375,28 @@ def _start_portfolio_subprocess( pass logger.info("[supervisor] 拉起组合实盘子进程 (account=%s strategy=%s)", account_row.get("id"), env.get("SANGUO_LIVE_STRATEGY")) - return subprocess.Popen( - [sys.executable, "-m", "sanguo_portfolio.runner_live"], - env=env, - ) + # 可观测性:子进程 stdout/stderr 落 logs/live_{aid}.log(>5MB 轮转截断)。 + # 原实现继承 schtask 控制台(=黑洞)——影子侧 2026-08-16 已修(#88 同款), + # 实盘侧一直没补:8-17 起实盘引擎的委托/成交/异常零留存,排查只能靠 QMT 端。 + from pathlib import Path as _P + log_dir = _P(__file__).resolve().parents[1] / "logs" + log_dir.mkdir(parents=True, exist_ok=True) + log_path = log_dir / f"live_{account_row.get('id')}.log" + try: + if log_path.exists() and log_path.stat().st_size > 5 * 1024 * 1024: + log_path.write_text("", encoding="utf-8") # 超限重置,避免崩溃循环刷爆盘 + fh = open(log_path, "ab") + fh.write(f"\n==== spawn {time.strftime('%Y-%m-%d %H:%M:%S')} ====\n".encode()) + return subprocess.Popen( + [sys.executable, "-X", "utf8", "-m", "sanguo_portfolio.runner_live"], + env=env, stdout=fh, stderr=subprocess.STDOUT, + ) + except Exception: + logger.warning("[supervisor] 日志重定向失败,退回继承控制台", exc_info=True) + return subprocess.Popen( + [sys.executable, "-m", "sanguo_portfolio.runner_live"], + env=env, + ) def _stop_portfolio_subprocess(proc: subprocess.Popen) -> None: diff --git a/sanguo_portfolio/live_instance_ledger.py b/sanguo_portfolio/live_instance_ledger.py new file mode 100644 index 0000000..b5be2ed --- /dev/null +++ b/sanguo_portfolio/live_instance_ledger.py @@ -0,0 +1,190 @@ +"""共享 QMT 账户下的 per-instance 虚拟子账本(实盘/影子组合引擎通用通道)。 + +背景(2026-08-19 三日体检):8 路组合实盘全打同一 miniQMT 账号,LiveEngine 的 +``context.portfolio`` 是券商同步的**全账户**视图(8 路策略+手动持仓并集)—— +channel_test 轮换会卖掉别家持仓、对账 8 对全 FAIL、前端收益率=全账户/初始资金 +毫无意义。策略 session 拍板:卖出范围应限**本实例持仓**,前后端 session 出通道。 + +本模块即该通道: +- ``LiveInstanceLedger`` 由**本实例的真实成交**(engine.get_trades() 中 order_id + ∈ engine.get_orders() 的部分)驱动的虚拟账本——现金=初始−Σ买−Σ费+Σ卖, + 持仓=成交聚合+移动加权成本,与 ShadowBroker.restore_from_trades 同一套算术。 +- 进程内通道 ``set_active()/get_active()``:runner_live 建好账本后 set,适配层 + ``live_strategy._setup`` 读到即注入 facade.get_instance_positions——策略侧 + ``getattr(broker, "get_instance_positions", None)`` 消费,回测/无台账时回退 + context.portfolio(策略 session 接入,前后端只出通道)。 + +费用口径:QMT 成交快照常无佣金字段,按 live_strategy 的 OrderCost 估算 +(佣金 max(成交额×0.0003, 5)+卖出印花税 0.001);快照带实际费用则用实际。 +虚拟现金与真实账户费用有细微漂移,仅供实例视图/风控,不做资金对账依据。 +""" +from __future__ import annotations + +import logging +import threading +from typing import Any, Dict, Iterable, Optional, Tuple + +logger = logging.getLogger(__name__) + +COMMISSION_RATE = 0.0003 +MIN_COMMISSION = 5.0 +STAMP_TAX = 0.001 + + +def estimate_fee(is_buy: bool, price: float, volume: int) -> float: + """按 live_strategy OrderCost 估算一笔成交的费用。""" + value = price * volume + fee = max(value * COMMISSION_RATE, MIN_COMMISSION) + if not is_buy: + fee += value * STAMP_TAX + return fee + + +class LiveInstanceLedger: + """一个 live 实例的虚拟子账本(共享账户的切片视图)。 + + 只记本实例自己的成交;别家策略/手动持仓不在账内 → 台账空仓时策略 + 不卖任何东西(正是互卖事故要的行为)。 + """ + + def __init__(self, initial_cash: float = 1_000_000.0): + self.initial_cash = float(initial_cash) + self.cash: float = float(initial_cash) + # symbol -> {"volume": int, "avg_cost": float} + self.positions: Dict[str, Dict[str, float]] = {} + # symbol -> (买入日期 str, 当日买入量) —— T+1 可卖视图 + self._today_bought: Dict[str, Tuple[str, int]] = {} + self._seen_trade_ids: set[str] = set() + # poller 线程写 / 策略线程(handle_data)读 —— 实盘视图一致性 + self._lock = threading.Lock() + # 有新成交未落 balance 快照 → 下个快照周期必写(节流档位见 runner_live) + self.dirty = True + + # ------------------ 成交驱动 ------------------ + def apply_trade( + self, + is_buy: bool, + symbol: str, + price: float, + volume: int, + trade_id: str, + trade_date: str, + fee: Optional[float] = None, + ) -> bool: + """应用一笔本实例成交;trade_id 重复返回 False(幂等)。 + + fee=None 时按费率估算;快照带实际佣金/印花税则传实际值。 + """ + if not trade_id or trade_id in self._seen_trade_ids: + return False + if price <= 0 or volume <= 0: + logger.warning("[instance-ledger] 非法成交跳过 %s %s x%s@%s", + trade_id, symbol, volume, price) + return False + with self._lock: + self._seen_trade_ids.add(trade_id) + actual_fee = fee if (fee is not None and fee > 0) else \ + estimate_fee(is_buy, price, volume) + value = price * volume + if is_buy: + self.cash -= value + actual_fee + pos = self.positions.setdefault( + symbol, {"volume": 0, "avg_cost": 0.0}) + total_cost = pos["avg_cost"] * pos["volume"] + value + pos["volume"] += volume + pos["avg_cost"] = total_cost / pos["volume"] if pos["volume"] else 0.0 + date, bought = self._today_bought.get(symbol, ("", 0)) + self._today_bought[symbol] = ( + trade_date, bought + volume if date == trade_date else volume) + else: + self.cash += value - actual_fee + pos = self.positions.get(symbol) + if pos is None: + # 账上无此标的的卖出(如 bootstrap 缺口前的旧仓):现金照收, + # 持仓无账可扣——如实留痕,不崩 + logger.warning( + "[instance-ledger] 卖出无账面持仓 %s x%s@%s(只入现金)", + symbol, volume, price) + else: + if pos["volume"] < volume: + logger.warning( + "[instance-ledger] 卖出超账面 %s: want %s have %s(按账面扣)", + symbol, volume, int(pos["volume"])) + volume = int(pos["volume"]) + pos["volume"] -= volume + if pos["volume"] == 0: + pos["avg_cost"] = 0.0 + del self.positions[symbol] + self.dirty = True + return True + + def restore_from_trades(self, rows: Iterable[Dict[str, Any]]) -> int: + """重启恢复:重放 DB 已归因成交(live_trades 行),返回重放笔数。 + + 行格式 = sanguo_live.persistence.list_trades 的返回: + direction(buy/sell)/symbol/price/volume/traded_at/vt_tradeid。 + """ + count = 0 + for r in rows: + applied = self.apply_trade( + is_buy=str(r.get("direction", "")) == "buy", + symbol=str(r.get("symbol", "")), + price=float(r.get("price") or 0), + volume=int(float(r.get("volume") or 0)), + trade_id=str(r.get("vt_tradeid") or ""), + trade_date=str(r.get("traded_at", ""))[:10], + ) + if applied: + count += 1 + if count: + logger.info("[instance-ledger] 重启恢复 %d 笔成交: cash=%.2f 持仓 %d 只", + count, self.cash, len(self.positions)) + return count + + # ------------------ 视图 ------------------ + def positions_view(self, now_date: str = "") -> Dict[str, Dict[str, Any]]: + """实例持仓视图(引擎快照同构,供策略/落库): + {symbol: {amount, closeable_amount(T+1), avg_cost}}。 + """ + view: Dict[str, Dict[str, Any]] = {} + with self._lock: + items = list(self.positions.items()) + for sym, pos in items: + vol = int(pos["volume"]) + if vol <= 0: + continue + date, bought = self._today_bought.get(sym, ("", 0)) + locked = bought if date and date == now_date else 0 + view[sym] = { + "amount": vol, + "closeable_amount": max(vol - locked, 0), + "avg_cost": float(pos["avg_cost"]), + } + return view + + def equity(self, prices: Dict[str, float]) -> Tuple[float, float, float]: + """(现金, 市值, 总资产)。prices 缺失/<=0 的标的最加权成本兜底。""" + with self._lock: + cash = self.cash + mv = 0.0 + for sym, pos in list(self.positions.items()): + price = prices.get(sym) or 0.0 + if price <= 0: + price = float(pos["avg_cost"]) + mv += price * pos["volume"] + return cash, mv, cash + mv + + +# ------------------ 进程内通道(runner ↔ 适配层) ------------------ +_ACTIVE: Optional[LiveInstanceLedger] = None + + +def set_active(ledger: Optional[LiveInstanceLedger]) -> None: + """runner_live 装配账本后调用;None 清除(测试隔离)。""" + global _ACTIVE + _ACTIVE = ledger + + +def get_active() -> Optional[LiveInstanceLedger]: + """适配层/策略侧取当前实例账本;未装配(回测/单测)返回 None。""" + return _ACTIVE diff --git a/sanguo_portfolio/live_strategy.py b/sanguo_portfolio/live_strategy.py index 23324d9..33249a3 100644 --- a/sanguo_portfolio/live_strategy.py +++ b/sanguo_portfolio/live_strategy.py @@ -113,6 +113,15 @@ def _setup(context): run_daily=bt_run_daily, run_monthly=bt_run_monthly, ) + # 实例持仓通道(2026-08-19 共享QMT账户互卖根治):runner_live 已 set_active + # 时注入,策略侧 getattr(broker,'get_instance_positions',None) 消费; + # 回测/测试无账本 → 保持 None,策略回退 context.portfolio + from . import live_instance_ledger as _lil + _ledger = _lil.get_active() + if _ledger is not None: + strategy.broker.get_instance_positions = _ledger.positions_view + logger.info("instance 台账通道已注入: get_instance_positions (cash=%.2f 持仓 %d 只)", + _ledger.cash, len(_ledger.positions)) # A 股费用 + 滑点(与回测默认一致) set_order_cost( OrderCost( diff --git a/sanguo_portfolio/runner_live.py b/sanguo_portfolio/runner_live.py index 0d9c056..fda5711 100644 --- a/sanguo_portfolio/runner_live.py +++ b/sanguo_portfolio/runner_live.py @@ -60,102 +60,125 @@ def live_env() -> Dict[str, str]: } -def _snapshot_once(engine: Any, db: str, account_id: int) -> None: - """单次快照:portfolio → live_positions/live_balance。 +def _sync_instance_trades( + engine: Any, ledger: Any, db: str, account_id: int, strategy_name: str, +) -> None: + """归因成交:engine.get_trades() 只留 order_id ∈ engine.get_orders() 的部分。 - 现金<=0 视为「broker 账户尚未同步完成」跳过 balance 落库: - QMT 持仓先到、资金后到时 total=持仓市值(无现金),写库会成为前端 - 收益率的基线 → 假收益率 341080%(2026-08-14 实况)。满仓账户的 - cash 本就≈0,此情形少牺牲(balance 少几条,positions 照落)。 + 共享 QMT 账户下 broker 成交是全账户的(8 路策略+手动);引擎 _orders 只登记 + 本进程提交的订单,且 Trade.order_id 已被引擎映射回本实例 id 空间 + (_broker_order_index)——「本实例订单」的判定天然成立。归因后的成交: + ①驱动实例虚拟账本 ②落 live_trades(account_id=本实例,方向取自订单)。 + 跨日:QMT 只查当日成交,历史靠 DB 已存行(重启时 restore_from_trades 重放)。 + """ + from sanguo_live.persistence import save_trade + + try: + orders = engine.get_orders() or {} + own_buy: Dict[str, bool] = { + str(oid): bool(getattr(o, "is_buy", True)) + for oid, o in orders.items() + } + trades = engine.get_trades() or {} + except Exception as e: # noqa: BLE001 + logger.warning("[live-trades] 查订单/成交失败 (account=%s): %s", account_id, e) + return + for tid, t in trades.items(): + oid = str(getattr(t, "order_id", "")) + if oid not in own_buy: + continue # 别家实例/手动单,不归因给本实例 + is_buy = own_buy[oid] + t_time = getattr(t, "time", None) + date_str = t_time.strftime("%Y-%m-%d") if hasattr(t_time, "strftime") else str(t_time or "") + applied = ledger.apply_trade( + is_buy=is_buy, + symbol=str(getattr(t, "security", "")), + price=float(getattr(t, "price", 0) or 0), + volume=int(getattr(t, "amount", 0) or 0), + trade_id=str(tid), + trade_date=date_str[:10], + fee=(float(getattr(t, "commission", 0) or 0) + + float(getattr(t, "tax", 0) or 0)) or None, + ) + if not applied: + continue + save_trade(db, account_id, { + "strategy_name": strategy_name, + "symbol": str(getattr(t, "security", "")), + "direction": "buy" if is_buy else "sell", + "offset": "open" if is_buy else "close", + "price": float(getattr(t, "price", 0) or 0), + "volume": int(getattr(t, "amount", 0) or 0), + "traded_at": (t_time.strftime("%Y-%m-%d %H:%M:%S") + if hasattr(t_time, "strftime") else str(t_time or "")), + "vt_tradeid": str(tid), + }) + logger.info("[live-trades] 本实例成交落库 (account=%s %s %s x%s@%s)", + account_id, "买入" if is_buy else "卖出", + getattr(t, "security", ""), getattr(t, "amount", 0), + getattr(t, "price", 0)) + + +def _snapshot_once(engine: Any, db: str, account_id: int, ledger: Any) -> None: + """单次快照:**实例虚拟账本** → live_positions/live_balance。 + + 2026-08-19 前落的是 context.portfolio(全账户)——8 实例同写一份全账户持仓、 + 收益率=全账户/初始资金(共享 QMT 账号下毫无意义)。改落实例视图: + - positions = 账本持仓(T+1 冻结=当日买入); + - balance = 账本现金 + 持仓市值(价格取全账户快照的现价,取不到用加权成本)。 + 全账户真实数字由 QMT 客户端随时可查,不再经本表透传。 """ from datetime import datetime from sanguo_live.persistence import save_balance, save_positions - portfolio = engine.context.portfolio - positions: Dict[str, Dict[str, Any]] = {} - for sym, pos in (getattr(portfolio, "positions", None) or {}).items(): - vol = int(getattr(pos, "total_amount", 0) or 0) - if vol <= 0: - continue - positions[str(sym)] = { - "volume": float(vol), - "frozen": float(vol - int(getattr(pos, "closeable_amount", vol) or 0)), - "avg_price": float(getattr(pos, "avg_cost", 0) or 0), + now_date = datetime.now().strftime("%Y-%m-%d") + view = ledger.positions_view(now_date) + positions: Dict[str, Dict[str, Any]] = { + sym: { + "volume": float(p["amount"]), + "frozen": float(p["amount"] - p["closeable_amount"]), + "avg_price": float(p["avg_cost"]), } + for sym, p in view.items() + } save_positions(db, account_id, positions) - cash = float(getattr(portfolio, "available_cash", 0) or 0) - total = float(getattr(portfolio, "total_value", 0) or 0) - if cash <= 0: - logger.info("[live-snapshot] cash=%s(账户未同步完成?),跳过 balance " - "(account=%s total=%s)", cash, account_id, total) - return + + # 现价:全账户快照里有(本实例持仓必是其子集);取不到退加权成本 + prices: Dict[str, float] = {} + for sym, pos in (getattr(engine.context.portfolio, "positions", None) + or {}).items(): + price = float(getattr(pos, "price", 0) or 0) \ + or float(getattr(pos, "current_price", 0) or 0) + if price > 0: + prices[str(sym)] = price + cash, mv, total = ledger.equity(prices) save_balance( db, account_id, datetime.now().strftime("%Y-%m-%d %H:%M:%S"), - cash, market_value=max(total - cash, 0.0), total=total, + cash, market_value=mv, total=total, ) -def _sync_trades(engine: Any, db: str, account_id: int) -> None: - """轮询 broker 当日成交 → live_trades(去重 by trade_id)。 +def _snapshot_loop(engine: Any, db: str, account_id: int, ledger: Any, + interval_sec: float = 60.0, + snap_min_interval: float = 300.0) -> None: + """后台线程:归因成交→实例账本→快照落库(供 API 读)。 - bullet_trade BrokerBase 无成交回调,组合实盘此前完全没人写 live_trades - (2026-08-14 用户发现"没有成交记录")。QMT 只查当日成交,跨日靠 DB 已存行; - 方向从 get_orders 的 is_buy 映射,查不到留空。 - """ - from sanguo_live.persistence import list_trades, save_trade - - broker = getattr(engine, "broker", None) - if broker is None: - return - try: - trades = broker.get_trades() or [] - except Exception as e: # noqa: BLE001 - logger.warning("[live-trades] 查成交失败 (account=%s): %s", account_id, e) - return - if not trades: - return - known = {str(t.get("vt_tradeid") or "") for t in list_trades(db, account_id)} - side_map: Dict[str, str] = {} - try: - for o in broker.get_orders() or []: - oid = str(o.get("order_id") or "") - if oid and o.get("is_buy") is not None: - side_map[oid] = "buy" if o["is_buy"] else "sell" - except Exception: # noqa: BLE001 - 方向映射失败不阻断成交落库 - pass - for t in trades: - tid = str(t.get("trade_id") or "") - if not tid or tid in known: - continue - save_trade(db, account_id, { - "strategy_name": t.get("strategy_name") or "", - "symbol": t.get("security") or "", - "direction": side_map.get(str(t.get("order_id") or ""), ""), - "offset": "", - "price": float(t.get("price") or 0), - "volume": int(t.get("amount") or 0), - "traded_at": str(t.get("time") or ""), - "vt_tradeid": tid, - }) - logger.info("[live-trades] 成交落库 (account=%s %s %s x%s@%s)", - account_id, t.get("security"), side_map.get( - str(t.get("order_id") or ""), "?"), - t.get("amount"), t.get("price")) - - -def _snapshot_loop(engine: Any, db: str, account_id: int, - interval_sec: float = 60.0) -> None: - """后台线程:定时把 engine 组合快照落库(供 API 读)。 - - LiveEngine 的账户/持仓由 broker 同步进 context.portfolio(LivePortfolioProxy), - 这里只读转储;任何异常只 warning 不中断(engine 主循环不受影响)。 + 归因轮询每 60s;快照(positions+balance)只在**有新成交或 ≥5 分钟**时写—— + 旧版 60s 无脑写 balance(1440 行/天/实例,2026-08-19 上午 8 账户 4848 行) + 量偏大遗留一并治。任何异常只 warning 不中断(engine 主循环不受影响)。 """ + strategy_name = os.environ.get("SANGUO_LIVE_STRATEGY", "") + last_snap = 0.0 while True: time.sleep(interval_sec) try: - _snapshot_once(engine, db, account_id) - _sync_trades(engine, db, account_id) + _sync_instance_trades(engine, ledger, db, account_id, strategy_name) + now = time.time() + if ledger.dirty or now - last_snap >= snap_min_interval: + _snapshot_once(engine, db, account_id, ledger) + ledger.dirty = False + last_snap = now except Exception as e: # noqa: BLE001 logger.warning("[live-snapshot] 落库失败 (account=%s): %s", account_id, e) @@ -198,6 +221,21 @@ def run_live(provider_config: Dict[str, Any] | None = None) -> None: broker = QmtBroker(account_id=cfg["account"], data_path=cfg["mini_path"]) logger.info("QmtBroker 装配 account=%s data_path=%s", cfg["account"], cfg["mini_path"]) + # 实例虚拟账本(共享 QMT 账户的切片视图,2026-08-19 互卖/对账/收益率三问题同根): + # 先建+恢复再起 engine——适配层 _setup 经 get_active() 注入 facade 通道 + from .live_instance_ledger import LiveInstanceLedger, set_active + ledger = LiveInstanceLedger(initial_cash=float(cfg["cash"] or 1_000_000)) + if cfg["db"] and cfg["account_id"]: + try: + from sanguo_live.persistence import list_trades + n = ledger.restore_from_trades( + list_trades(cfg["db"], int(cfg["account_id"]))) + logger.info("[instance-ledger] 恢复 %d 笔: cash=%.2f 持仓 %d 只", + n, ledger.cash, len(ledger.positions)) + except Exception as e: # noqa: BLE001 - 无库/表未建不阻断启动(空账本起跑) + logger.warning("[instance-ledger] 恢复失败,空账本起跑: %s", e) + set_active(ledger) + engine = LiveEngine( _instance_adapter(cfg["account_id"]), broker_factory=lambda: broker, @@ -215,7 +253,7 @@ def run_live(provider_config: Dict[str, Any] | None = None) -> None: if cfg["db"] and cfg["account_id"]: t = threading.Thread( target=_snapshot_loop, - args=(engine, cfg["db"], int(cfg["account_id"])), + args=(engine, cfg["db"], int(cfg["account_id"]), ledger), daemon=True, name="live-snapshot", ) t.start() diff --git a/sanguo_portfolio/strategies/all_weather.py b/sanguo_portfolio/strategies/all_weather.py index 7eeae6a..40cd2c7 100644 --- a/sanguo_portfolio/strategies/all_weather.py +++ b/sanguo_portfolio/strategies/all_weather.py @@ -50,6 +50,10 @@ class BrokerFacade: set_order_cost: Callable[[Any, str], None] = lambda cost, type: None run_daily: Callable[..., None] = lambda *a, **kw: None run_monthly: Callable[..., None] = lambda *a, **kw: None + # 实例持仓通道(2026-08-19 共享QMT账户互卖根治):live 时 runner_live 注入 + # ledger.positions_view → {symbol:{amount,closeable_amount,avg_cost}}; + # 回测/无台账为 None → 策略回退 context.portfolio(策略 session 消费) + get_instance_positions: Optional[Callable[[], Dict[str, Dict[str, Any]]]] = None # ------------------------ 策略 ------------------------ diff --git a/tests/api/test_portfolio_live.py b/tests/api/test_portfolio_live.py index 69963bd..98b9ab2 100644 --- a/tests/api/test_portfolio_live.py +++ b/tests/api/test_portfolio_live.py @@ -278,32 +278,37 @@ def test_update_live_normalizes_vt_symbol(tmp_path, monkeypatch): assert acc["vt_symbol"] == "300024.SZSE" -def test_snapshot_once_skips_unsynced_cash(): - """cash<=0(账户未同步完成)不落 balance——治假收益率(2026-08-14 实况 341080%)。""" +def test_snapshot_once_broker_snapshot_never_pollutes_balance(): + """balance=实例虚拟账本(2026-08-19):broker 快照 cash=0/乱值不落库—— + 旧版治假收益率(341080%)的守卫换形态:不再读 broker 现金,快照只供现价。""" import os import tempfile import types from sanguo_live.persistence import init_db, list_balance + from sanguo_portfolio.live_instance_ledger import LiveInstanceLedger from sanguo_portfolio.runner_live import _snapshot_once db = os.path.join(tempfile.mkdtemp(), "l.db") init_db(db) + led = LiveInstanceLedger(initial_cash=1_000_000) + led.apply_trade(True, "000001.XSHE", 10.0, 100, "t1", "2026-08-19") def _mk_portfolio(cash, total): p = types.SimpleNamespace( - available_cash=cash, total_value=total, positions={}) + available_cash=cash, total_value=total, + positions={"000001.XSHE": types.SimpleNamespace(price=0.0)}) ctx = types.SimpleNamespace(portfolio=p) return types.SimpleNamespace(context=ctx) - # 未同步完成: cash=0, total=持仓市值 → 不落 balance - _snapshot_once(_mk_portfolio(0, 2931.0), db, 3) - assert list_balance(db, 3) == [] - # 正常: cash>0 → 落库 - _snapshot_once(_mk_portfolio(9_997_077.51, 10_000_008.51), db, 3) + # broker 未同步(cash=0/无现价):balance 仍=账本算术(现金+加权成本市值) + _snapshot_once(_mk_portfolio(0, 2931.0), db, 3, led) rows = list_balance(db, 3) assert len(rows) == 1 - assert rows[0]["total"] == 10_000_008.51 + assert rows[0]["cash"] == 999_000.0 - 5.0 + assert rows[0]["market_value"] == 100 * 10.0 # 现价缺失退加权成本 + # broker 快照的现金/总值完全不被引用(共享全账户数字不进实例表) + assert rows[0]["total"] == rows[0]["cash"] + rows[0]["market_value"] def test_get_live_return_uses_first_snapshot_baseline(live_db): @@ -317,3 +322,40 @@ def test_get_live_return_uses_first_snapshot_baseline(live_db): acc = rl.get_live(aid) assert acc["latest_equity"] == 10_150_000 assert acc["total_return"] == (10_150_000 - 10_000_000) / 10_000_000 + + +def test_start_portfolio_subprocess_redirects_child_output(tmp_path, monkeypatch): + """实盘引擎子进程 stdout/stderr 落 logs/live_{aid}.log(2026-08-19 黑洞根治)。 + + 8-17 起实盘引擎继承 schtask 控制台(=黑洞),委托/成交/异常零留存——影子侧 + 2026-08-16 已修同款,这里对齐:spawn 标记 + >5MB 截断 + Popen 收 stdout。 + """ + # __file__ 指到 tmp,日志目录落在 tmp/logs 不污染仓库 + fake_file = tmp_path / "sanguo_live" / "runner.py" + fake_file.parent.mkdir(parents=True) + fake_file.write_text("# probe", encoding="utf-8") + monkeypatch.setattr(live_runner, "__file__", str(fake_file)) + + captured = {} + + def fake_popen(argv, env=None, stdout=None, stderr=None): + captured["argv"] = argv + captured["stderr"] = stderr + stdout.write("probe-line\n".encode()) + stdout.flush() + stdout.close() + return object() + + monkeypatch.setattr(live_runner.subprocess, "Popen", fake_popen) + live_runner._start_portfolio_subprocess( + {"id": 77, "account": "66639661", "strategy_class": "channel_test", + "setting": "{}"}, str(tmp_path / "l.db")) + + assert captured["argv"][1:] == ["-X", "utf8", "-m", + "sanguo_portfolio.runner_live"] + assert captured["stderr"] is not None # stderr 并入 stdout + log = tmp_path / "logs" / "live_77.log" + assert log.exists() + text = log.read_text(encoding="utf-8") + assert "==== spawn" in text + assert "probe-line" in text diff --git a/tests/portfolio/test_live_instance_ledger.py b/tests/portfolio/test_live_instance_ledger.py new file mode 100644 index 0000000..a98b8a3 --- /dev/null +++ b/tests/portfolio/test_live_instance_ledger.py @@ -0,0 +1,236 @@ +"""实例虚拟账本(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)