338 lines
15 KiB
Python
338 lines
15 KiB
Python
"""组合策略实盘入口(VPS Windows / miniQMT 直连)——bullet_trade 0.9.2 LiveEngine。
|
||
|
||
supervisor(``sanguo_live.runner.run_supervisor``) 对 strategy_type='portfolio' 的
|
||
live_accounts 行以**子进程**方式拉起本模块,env 传参:
|
||
|
||
SANGUO_QMT_ACCOUNT / SANGUO_QMT_PATH miniQMT 交易账号 / userdata_mini 路径
|
||
SANGUO_LIVE_STRATEGY / _MAX_POOL / _BENCHMARK 组合策略配置
|
||
SANGUO_LIVE_CASH 初始资金(engine NAV 基准)
|
||
SANGUO_LIVE_DB / SANGUO_LIVE_ACCOUNT_ID 快照落库目标(缺省不落)
|
||
|
||
手动用法(交易日 + miniQMT 连接下):
|
||
set SANGUO_QMT_ACCOUNT=66639661
|
||
python -m sanguo_portfolio.runner_live
|
||
|
||
不在 Mac 跑(Mac 无 xtquant/miniQMT 客户端)。
|
||
|
||
历史注记:0.2 之前的 bullet_trade LiveEngine 接受 ``initialize=/broker=`` 直传,
|
||
0.9.x 改为 strategy_file + broker_factory——本模块即按新 API 装配,策略逻辑在
|
||
``sanguo_portfolio/live_strategy.py``(适配文件)。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
# ENV GUARD 必须早于任何 bullet_trade import
|
||
import os
|
||
os.environ.setdefault("DEFAULT_DATA_PROVIDER", "miniqmt")
|
||
|
||
import logging
|
||
import threading
|
||
import time
|
||
from pathlib import Path
|
||
from typing import Any, Dict
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
ADAPTER_FILE = Path(__file__).resolve().parent / "live_strategy.py"
|
||
|
||
# connect 退避重试(2026-08-30):vps-deploy 收尾滚动重启杀全树后立刻重拉,旧引擎
|
||
# 的 QMT 会话尚未在 QMT 端释放,6 引擎并发重连时个别撞上回收窗口 connect()=-1
|
||
# → 整引擎退出(08-30 实录 live_17/20,静默缺引擎直到有人看)。QmtBroker.connect
|
||
# 失败路径自带本次 trader 清理,重试安全;会话回收是秒级,退避重试即免疫。
|
||
CONNECT_RETRY_ATTEMPTS = 5
|
||
CONNECT_RETRY_WAIT_SEC = 10.0
|
||
|
||
|
||
def with_connect_retry(broker: Any,
|
||
attempts: int = CONNECT_RETRY_ATTEMPTS,
|
||
wait_sec: float = CONNECT_RETRY_WAIT_SEC,
|
||
_sleep=time.sleep) -> Any:
|
||
"""包一层 broker.connect 退避重试(仅实盘 QmtBroker 装配处使用)。
|
||
|
||
``_sleep`` 注入仅为单测提速;真跑 ``time.sleep``。
|
||
"""
|
||
orig_connect = broker.connect
|
||
|
||
def _connect() -> bool:
|
||
for i in range(1, attempts + 1):
|
||
try:
|
||
return orig_connect()
|
||
except Exception as e: # noqa: BLE001 - connect 瞬时失败皆可重试
|
||
if i == attempts:
|
||
raise
|
||
logger.warning("[connect-retry] 第 %d/%d 次失败: %s(%.0fs 后重试)",
|
||
i, attempts, e, wait_sec)
|
||
_sleep(wait_sec)
|
||
return False # pragma: no cover - 循环内必 return/raise
|
||
|
||
broker.connect = _connect # type: ignore[method-assign]
|
||
return broker
|
||
|
||
|
||
def build_provider(provider_config: Dict[str, Any] | None = None) -> Any:
|
||
"""构造 live 模式的 SanguoMiniQmtProvider。"""
|
||
from .providers import SanguoMiniQmtProvider
|
||
|
||
cfg = dict(provider_config or {})
|
||
cfg.setdefault("mode", "live")
|
||
cfg.setdefault("auto_download", True)
|
||
return SanguoMiniQmtProvider(cfg)
|
||
|
||
|
||
def live_env() -> Dict[str, str]:
|
||
"""解析 env 实盘配置(带默认值)。独立出来便于单测。"""
|
||
return {
|
||
"strategy": os.environ.get("SANGUO_LIVE_STRATEGY", "all_weather"),
|
||
"max_pool": os.environ.get("SANGUO_LIVE_MAX_POOL", "0"),
|
||
"benchmark": os.environ.get("SANGUO_LIVE_BENCHMARK", "000300.XSHG"),
|
||
"cash": os.environ.get("SANGUO_LIVE_CASH", "1000000"),
|
||
"account": os.environ.get("SANGUO_QMT_ACCOUNT", ""),
|
||
"mini_path": (os.environ.get("SANGUO_QMT_PATH")
|
||
or r"C:\国金QMT交易端模拟\userdata_mini"),
|
||
"db": os.environ.get("SANGUO_LIVE_DB", ""),
|
||
"account_id": os.environ.get("SANGUO_LIVE_ACCOUNT_ID", ""),
|
||
}
|
||
|
||
|
||
def _effective_trade_time(trade: Any) -> Any:
|
||
"""成交时间守卫(2026-08-20 事故):QMT 原始成交时间经引擎 pd.to_datetime 的
|
||
失败形态会落成 1970-01-01 00:00:01 的 datetime——当日 9 笔 traded_at=1970
|
||
落库,前端"今日成交"按日期过滤全空 + 账本 trade_date 失真(T+1 视图错)。
|
||
年份<2000 一律视为无效,回退当前时刻:归因轮询间隔 ≤60s,日期误差只剩
|
||
跨日 60s 窗口,可忽略。"""
|
||
from datetime import datetime
|
||
|
||
v = getattr(trade, "time", None)
|
||
if isinstance(v, datetime) and v.year >= 2000:
|
||
return v
|
||
return datetime.now()
|
||
|
||
|
||
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() 的部分。
|
||
|
||
共享 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 = _effective_trade_time(t)
|
||
date_str = t_time.strftime("%Y-%m-%d")
|
||
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
|
||
|
||
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)
|
||
|
||
# 现价:全账户快照里有(本实例持仓必是其子集);取不到退加权成本
|
||
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=mv, total=total,
|
||
)
|
||
|
||
|
||
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 读)。
|
||
|
||
归因轮询每 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:
|
||
_sync_instance_trades(engine, ledger, db, account_id, strategy_name)
|
||
# 迟到成交对账:名单单直查 QMT(a 窄修)+ 15:05 EOD 全量回填(b 宽修)
|
||
from .live_reconcile import maybe_eod_reconcile, reconcile_pending
|
||
reconcile_pending(engine, ledger, db, account_id, strategy_name)
|
||
maybe_eod_reconcile(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)
|
||
|
||
|
||
def _instance_adapter(account_id: str) -> Path:
|
||
"""按账户复制一份策略适配文件。
|
||
|
||
bullet_trade 实例锁判重键=主机+strategy_path+broker_type+account_identity;
|
||
多实盘共用同一 QMT 账号(合法场景:同账号跑多策略)时 strategy_path 相同会被
|
||
误判"重复实例"拒启 → 每账户一份副本(内容同、路径异)即视为不同逻辑实例。
|
||
"""
|
||
if not account_id:
|
||
return ADAPTER_FILE
|
||
dst = (Path(__file__).resolve().parent.parent / "runtime"
|
||
/ f"live_{account_id}" / ADAPTER_FILE.name)
|
||
dst.parent.mkdir(parents=True, exist_ok=True)
|
||
if not dst.exists() or dst.read_text(encoding="utf-8") != \
|
||
ADAPTER_FILE.read_text(encoding="utf-8"):
|
||
import shutil
|
||
shutil.copyfile(ADAPTER_FILE, dst)
|
||
return dst
|
||
|
||
|
||
def run_live(provider_config: Dict[str, Any] | None = None) -> None:
|
||
"""装配 LiveEngine(strategy_file=适配文件 + QmtBroker)并 run(阻塞)。"""
|
||
from bullet_trade.core.live_engine import LiveEngine # type: ignore
|
||
from bullet_trade.data.api import set_data_provider # type: ignore
|
||
from bullet_trade.broker.qmt import QmtBroker # type: ignore
|
||
|
||
cfg = live_env()
|
||
if not cfg["account"]:
|
||
raise RuntimeError(
|
||
"缺 SANGUO_QMT_ACCOUNT(miniQMT 交易账号),实盘无法启动。"
|
||
"设 set SANGUO_QMT_ACCOUNT=66639661 后重试。"
|
||
)
|
||
|
||
provider = build_provider(provider_config)
|
||
set_data_provider(provider)
|
||
|
||
broker = with_connect_retry(
|
||
QmtBroker(account_id=cfg["account"], data_path=cfg["mini_path"]))
|
||
logger.info("QmtBroker 装配 account=%s data_path=%s(connect 失败退避重试 %d×%.0fs)",
|
||
cfg["account"], cfg["mini_path"],
|
||
CONNECT_RETRY_ATTEMPTS, CONNECT_RETRY_WAIT_SEC)
|
||
|
||
# 实例虚拟账本(共享 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,
|
||
# bullet_trade 每实例锁 runtime 目录(单实例设计);多实盘并行须各用独立目录
|
||
# strategy_name=实例唯一 remark label(2026-09-01 002646 事故:六实例适配
|
||
# 文件同名 live_strategy.py → 订单 remark label 全为 "live_strateg" 跨实
|
||
# 例不可分;EOD 对账重启后按 remark=bt:live_{id}: 重建本实例订单集,
|
||
# 引擎订单表是进程态,只有 remark 跨重启可归因)。仅对注入后新单生效。
|
||
live_config={
|
||
"runtime_dir": str(
|
||
Path(__file__).resolve().parent.parent / "runtime"
|
||
/ f"live_{cfg['account_id'] or 'solo'}"),
|
||
**({"strategy_name": f"live_{cfg['account_id']}"}
|
||
if cfg["account_id"] else {}),
|
||
},
|
||
)
|
||
logger.info(
|
||
"组合 live engine 启动: strategy=%s max_pool=%s benchmark=%s cash=%s",
|
||
cfg["strategy"], cfg["max_pool"], cfg["benchmark"], cfg["cash"],
|
||
)
|
||
|
||
# 快照落库(supervisor 注入 db+account_id 时才开)
|
||
if cfg["db"] and cfg["account_id"]:
|
||
# B 修法(2026-08-25 卖后买现金窗口):下单返回后即时归因——台账 cash
|
||
# 秒级新鲜,同轮「全卖→马上全买」的调仓立刻见到卖出回款;60s 归因
|
||
# 轮询(_snapshot_loop)保留兜底(部分成交后续/异步路径)。
|
||
ledger.on_order_done = lambda: _sync_instance_trades(
|
||
engine, ledger, cfg["db"], int(cfg["account_id"]), cfg["strategy"])
|
||
t = threading.Thread(
|
||
target=_snapshot_loop,
|
||
args=(engine, cfg["db"], int(cfg["account_id"]), ledger),
|
||
daemon=True, name="live-snapshot",
|
||
)
|
||
t.start()
|
||
|
||
engine.run()
|
||
|
||
|
||
def main() -> None:
|
||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
|
||
run_live()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|