平替miniQMT的聚宽跟单方案【附源码】

# encoding:gbk"""QMT 轮询监控脚本 — 从 stock_db.db 读取 pending 信号并执行下单运行方式：  在 QMT 交易端中运行此脚本（通过 QMT 的 Python 策略运行功能）流程：  1. 轮询 stock_db.db 获取 trade_status='pending' 的信号  2. 提交限价委托，状态改为 'submitted'（已报待成）  3. xtquant 回调跟踪委托状态，成交后改为 'completed'     废单/撤单改为 'failed'依赖：  - xtquant（QMT 内置）  - stock_db.db（由 jq_follower_monitor 或测试买入写入）QMT 入口: init(ContextInfo)"""import sqlite3import osimport timefrom datetime import datetime# ========== 配置 ==========# exe 运行时默认路径：exe同级目录下的 config/stock_db.dbDB_PATH = r"stock_db.db数据库路径"#DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config", "stock_db.db")POLL_INTERVAL = 1      # 轮询间隔（秒）STOCK_ACCOUNT = "实盘资金账号"       # QMT 资金账号，留空从 ContextInfo 获取# xtquant 导入（QMT 环境可用）# 回调类（依赖 xtquant）class _DummyCallback:    passtry:    from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback    from xtquant.xttype import StockAccount    from xtquant import xtconstant    class TradeStatusCallback(XtQuantTraderCallback):        def on_stock_order(self, order):            order_id = order.order_sysid            if order_id not in _pending_orders:                return            status_map = {                xtconstant.ORDER_SUCCEEDED: "completed",                xtconstant.ORDER_PART_SUCC: "completed",                xtconstant.ORDER_JUNK: "failed",                xtconstant.ORDER_CANCELED: "cancelled",                xtconstant.ORDER_PART_CANCEL: "cancelled",            }            new_status = status_map.get(order.order_status)            if new_status:                _update_status_by_order_id(order_id, new_status)        def on_order_error(self, order_error):            order_id = order_error.order_id            if order_id in _pending_orders:                _update_status_by_order_id(order_id, "failed")        def on_disconnected(self):            print(f"[qmt_sql] 回调: 连接断开")        def on_account_status(self, status):            pass        def on_stock_trade(self, trade):            pass        def on_cancel_error(self, cancel_error):            pass        def on_order_stock_async_response(self, response):            pass    _xtquant_ok = Trueexcept ImportError:    _xtquant_ok = False    class TradeStatusCallback(_DummyCallback):        def on_stock_order(self, order): pass        def on_order_error(self, order_error): pass        def on_disconnected(self): pass        def on_account_status(self, status): pass        def on_stock_trade(self, trade): pass        def on_cancel_error(self, cancel_error): pass        def on_order_stock_async_response(self, response): pass# 待成交订单映射表: {order_id: record_id}_pending_orders = {}# ========== 数据库操作 ==========def _connect_db():    """连接 stock_db.db"""    if not os.path.exists(DB_PATH):        print(f"[qmt_sql] 数据库不存在: {DB_PATH}")        return None, None    try:        conn = sqlite3.connect(DB_PATH, timeout=10)        cursor = conn.cursor()        return conn, cursor    except sqlite3.Error as e:        print(f"[qmt_sql] 数据库连接失败: {e}")        return None, Nonedef _save_pending_orders():    """保存待成交映射到文件（Crash 恢复用）"""    import pickle    try:        pk_file = DB_PATH + ".pending.pk"        with open(pk_file, "wb") as f:            pickle.dump(_pending_orders, f)    except Exception:        passdef _query_pending_signals(cursor):    """查询所有 pending 状态的信号"""    try:        cursor.execute(            "SELECT id, code, price, amount, side, strategy_id, strategy_name "            "FROM order_daily WHERE trade_status='pending' AND created_at >= datetime('now', '-3 seconds') ORDER BY id ASC;"        )        return cursor.fetchall()    except sqlite3.Error as e:        print(f"[qmt_sql] 查询失败: {e}")        return []def _update_status(conn, cursor, record_id, status):    """更新信号状态"""    try:        cursor.execute(            "UPDATE order_daily SET trade_status=? WHERE id=?;",            (status, record_id)        )        conn.commit()        print(f"[qmt_sql] 信号 id={record_id} 状态 -> {status}")    except sqlite3.Error as e:        print(f"[qmt_sql] 更新状态失败: {e}")def _update_status_by_order_id(order_id, status):    """根据 order_id 从映射表找到 record_id，更新状态"""    record_id = _pending_orders.get(order_id)    if record_id is None:        return    conn, cursor = _connect_db()    if conn and cursor:        _update_status(conn, cursor, record_id, status)        conn.close()    # 终态后移出映射表    if status in ("completed", "failed", "cancelled"):        _pending_orders.pop(order_id, None)# ========== 股票代码转换 ==========def _to_xt_code(stock_code):    """sh600036 -> 600036.SH, sz000001 -> 000001.SZ"""    if len(stock_code) < 8:        return stock_code    prefix = stock_code[:2].lower()    code = stock_code[2:]    return {        "sh": f"{code}.SH",        "sz": f"{code}.SZ",        "bj": f"{code}.BJ",    }.get(prefix, stock_code)# ========== QMT 下单 ==========def _place_order(ctx_info, stock_code, action, price, amount, record_id):    """通过 passorder 下单（QMT Native API）"""    xt_code = _to_xt_code(stock_code)    order_code = 23 if action == "buy" else 24    try:        print(f"[qmt_sql]   passorder 参数: order_code={order_code}, account={account_id}, stock={xt_code}, price={price}, amount={amount}")        order_id = passorder(            order_code, 1101, account_id, xt_code, 14,            price, amount, "qmt_sql", 1, "QMT", ctx_info        )        if order_id and order_id > 0:            _pending_orders[order_id] = record_id            _save_pending_orders()            print(f"[qmt_sql]    委托提交, order_id={order_id}")        return order_id    except Exception as e:        print(f"[qmt_sql]    passorder 异常: {e}")        return -1    #字数限制，完整版加我联系方式领取