233 lines
9.2 KiB
Python
233 lines
9.2 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"
|
|
|
|
|
|
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", "30"),
|
|
"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 _snapshot_once(engine: Any, db: str, account_id: int) -> None:
|
|
"""单次快照:portfolio → live_positions/live_balance。
|
|
|
|
现金<=0 视为「broker 账户尚未同步完成」跳过 balance 落库:
|
|
QMT 持仓先到、资金后到时 total=持仓市值(无现金),写库会成为前端
|
|
收益率的基线 → 假收益率 341080%(2026-08-14 实况)。满仓账户的
|
|
cash 本就≈0,此情形少牺牲(balance 少几条,positions 照落)。
|
|
"""
|
|
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),
|
|
}
|
|
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
|
|
save_balance(
|
|
db, account_id, datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
cash, market_value=max(total - cash, 0.0), total=total,
|
|
)
|
|
|
|
|
|
def _sync_trades(engine: Any, db: str, account_id: int) -> None:
|
|
"""轮询 broker 当日成交 → live_trades(去重 by trade_id)。
|
|
|
|
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 主循环不受影响)。
|
|
"""
|
|
while True:
|
|
time.sleep(interval_sec)
|
|
try:
|
|
_snapshot_once(engine, db, account_id)
|
|
_sync_trades(engine, db, account_id)
|
|
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 = QmtBroker(account_id=cfg["account"], data_path=cfg["mini_path"])
|
|
logger.info("QmtBroker 装配 account=%s data_path=%s", cfg["account"], cfg["mini_path"])
|
|
|
|
engine = LiveEngine(
|
|
_instance_adapter(cfg["account_id"]),
|
|
broker_factory=lambda: broker,
|
|
# bullet_trade 每实例锁 runtime 目录(单实例设计);多实盘并行须各用独立目录
|
|
live_config={"runtime_dir": str(
|
|
Path(__file__).resolve().parent.parent / "runtime"
|
|
/ f"live_{cfg['account_id'] or 'solo'}")},
|
|
)
|
|
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"]:
|
|
t = threading.Thread(
|
|
target=_snapshot_loop,
|
|
args=(engine, cfg["db"], int(cfg["account_id"])),
|
|
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()
|