QMT 运维进阶：在策略中集成 SQLite3 实现本地交易流水永久持久化

在量化交易的实际运行中，交易流水的永久保存是实盘运维的核心需求之一。虽然 QMT 提供了 get_trade_detail_data 接口来获取当日的委托和成交，但这些数据通常只保留在客户端的临时缓存或柜台内存中。一旦客户端重启、换机运行或进行历史回测，历史的详细交易流水就很难便捷地追溯。

Python 内置了轻量级关系型数据库 SQLite3，无需额外安装任何第三方库即可直接使用。本文将教你如何在 QMT 策略中集成 SQLite3，实现委托（Order）和成交（Deal）数据的本地永久持久化。

一、 核心设计思路

1. 初始化数据库：在 init(ContextInfo) 中，创建或连接本地 SQLite 数据库文件，并初始化 orders（委托表）和 deals（成交表）。
2. 利用回调实时写入：利用 QMT 提供的交易回报实时主推函数 order_callback 和 deal_callback。当账号的委托或成交状态发生变化时，QMT 会自动触发这两个回调，我们在此处将数据实时 Insert 或 Replace 到 SQLite 数据库中。
3. 绑定资金账号：必须在 init 中调用 ContextInfo.set_account(account) 绑定账号，否则回调函数不会生效。

二、 完整策略源码示例

以下是一个完整的 QMT 策略模板，展示了如何创建数据库、建表以及在收到交易回报时自动持久化数据：

#encoding:gbk
import sqlite3
import os

# 定义数据库保存路径（默认保存在QMT bin.x64目录下，也可指定绝对路径）
DB_PATH = "qmt_trade_flow.db"

def init(ContextInfo):
    # 1. 设定交易账号（请替换为您的实际账号）
    ContextInfo.accid = "6000000248"
    ContextInfo.set_account(ContextInfo.accid)
    
    # 2. 初始化数据库与数据表
    init_db()
    print("SQLite 数据库初始化成功，等待接收交易回报...")

def handlebar(ContextInfo):
    # 策略核心逻辑写在这里
    pass

# ------------------ SQLite 数据库操作函数 ------------------

def init_db():
    """初始化数据库，创建委托表和成交表"""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    
    # 创建委托表 (orders)
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS orders (
            order_sys_id TEXT PRIMARY KEY, 
            account_id TEXT,
            instrument_id TEXT,
            instrument_name TEXT,
            direction INTEGER,
            offset_flag INTEGER,
            limit_price REAL,
            volume_original INTEGER,
            volume_traded INTEGER,
            order_status INTEGER,
            insert_date TEXT,
            insert_time TEXT,
            remark TEXT
        )
    ''')
    
    # 创建成交表 (deals)
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS deals (
            trade_id TEXT PRIMARY KEY,
            order_sys_id TEXT,
            account_id TEXT,
            instrument_id TEXT,
            instrument_name TEXT,
            direction INTEGER,
            price REAL,
            volume INTEGER,
            trade_date TEXT,
            trade_time TEXT,
            trade_amount REAL,
            remark TEXT
        )
    ''')
    
    conn.commit()
    conn.close()

def save_order_to_db(order):
    """将委托对象保存或更新到数据库"""
    try:
        conn = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()
        # 使用 REPLACE INTO，如果委托号已存在则更新状态，不存在则插入
        cursor.execute('''
            REPLACE INTO orders (
                order_sys_id, account_id, instrument_id, instrument_name, 
                direction, offset_flag, limit_price, volume_original, 
                volume_traded, order_status, insert_date, insert_time, remark
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        ''', (
            order.m_strOrderSysID, order.m_strAccountID, order.m_strInstrumentID, order.m_strInstrumentName,
            order.m_nDirection, order.m_nOffsetFlag, order.m_dLimitPrice, order.m_nVolumeTotalOriginal,
            order.m_nVolumeTraded, order.m_nOrderStatus, order.m_strInsertDate, order.m_strInsertTime, order.m_strRemark
        ))
        conn.commit()
        conn.close()
        print(f"[DB] 委托更新成功: {order.m_strInstrumentID} 状态:{order.m_nOrderStatus}")
    except Exception as e:
        print(f"[DB Error] 保存委托失败: {e}")

def save_deal_to_db(deal):
    """将成交对象保存到数据库"""
    try:
        conn = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()
        cursor.execute('''
            REPLACE INTO deals (
                trade_id, order_sys_id, account_id, instrument_id, instrument_name, 
                direction, price, volume, trade_date, trade_time, trade_amount, remark
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        ''', (
            deal.m_strTradeID, deal.m_strOrderSysID, deal.m_strAccountID, deal.m_strInstrumentID, deal.m_strInstrumentName,
            deal.m_nDirection, deal.m_dPrice, deal.m_nVolume, deal.m_strTradeDate, deal.m_strTradeTime, deal.m_dTradeAmount, deal.m_strRemark
        ))
        conn.commit()
        conn.close()
        print(f"[DB] 新增成交记录: {deal.m_strInstrumentID} 价格:{deal.m_dPrice} 数量:{deal.m_nVolume}")
    except Exception as e:
        print(f"[DB Error] 保存成交失败: {e}")

# ------------------ QMT 交易回报实时主推回调 ------------------

def order_callback(ContextInfo, orderInfo):
    """账号委托状态变化主推"""
    # 过滤非本策略关注的账号（可选）
    if orderInfo.m_strAccountID == ContextInfo.accid:
        save_order_to_db(orderInfo)

def deal_callback(ContextInfo, dealInfo):
    """账号成交状态变化主推"""
    if dealInfo.m_strAccountID == ContextInfo.accid:
        save_deal_to_db(dealInfo)

三、 运维进阶与注意事项

1. 数据库锁与多线程安全：
   SQLite 默认在写入时会锁定整张表。如果你的 QMT 策略中使用了多线程（如 ContextInfo.run_time 定时器或自定义线程），请确保不要在多个线程中同时打开同一个 SQLite 连接。建议采用上述示例中“即开即闭”的连接管理方式，或者在连接时设置超时时间：sqlite3.connect(DB_PATH, timeout=10)。

2. 数据路径规划：
   默认相对路径 qmt_trade_flow.db 会保存在 QMT 客户端的运行目录下。为了防止客户端升级或卸载时数据被误删，建议将其指定为绝对路径，例如 D:/QMT_Data/qmt_trade_flow.db，并提前创建好文件夹。

3. 历史数据补录：
   如果由于网络断线或客户端异常关闭导致部分回调丢失，可以在 init 函数中调用 get_trade_detail_data 获取当日的所有委托和成交，进行一次全量比对和补录，确保本地数据库的完整性。