带你详细了解miniQMT的xttrader模块 (第六篇) 我下单了，怎么知道是否成交呢？

QUANT 2025-12-12 16:13:28 38 举报

这个问题的本质在于：

我们在 MiniQMT 上进行了"买入"指令，返回显示"委托成功"这只是说我们的订单被交易所收到了，并不代表已经成交

成交需要时间，可能是 1 秒，也可能是 10 秒钟，可能是 1分钟甚至可能永远成交不了（因为价格不合适等各种问题）

所以在mini的编程中，我们需要一个机制来实时监听这个订单的状态变化。这就是 on_stock_order 回调的作用

https://www.lianghuaba.net/thread-214.htm

关键点 1：on_stock_order 什么时候被调用？

当你的订单状态发生任何变化时，这个方法都会被自动调用。比如：
•刚下单 → 状态变为"已报"(50) → 调用一次
•下单有部分成交 → 状态变为"部成"(55) → 再调用一次
•全部成交 → 状态变为"已成"(56) → 再调用一次
所以一个订单可能会触发多次 on_stock_order 回调

关键点 2：订单状态码的含义如下：

关键点 3：on_stock_order 和 on_stock_trade 的区别

•on_stock_order：订单的状态变化（如从"已报"变成"部成"）
•on_stock_trade：实际发生了成交（每一笔成交都会触发）
一个订单可能会有多个成交（比如分批成交），所以 on_stock_trade 可能被调用多次。

我们来看完整的实战代码：

import time
import logging
from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback
from xtquant.xttype import StockAccount
from xtquant.xtconstant import STOCK_BUY, FIX_PRICE

# 配置日志，这样你能看到所有发生的事
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)


# ============================================================================
# 第一步：定义一个追踪订单的回调类
# ============================================================================

class OrderTracker(XtQuantTraderCallback):
    """这个类的唯一目的：追踪你的订单状态"""

    def __init__(self):
        super().__init__()
        # 用一个字典存储所有订单的信息
        # key: order_id, value: 订单的详细信息
        self.orders = {}

    def on_stock_order(self, order):
        """
        每当订单状态有任何变化，这个方法就会被自动调用。
        比如：下单成功、部分成交、全部成交、被撤销等。
        """
        order_id = order.order_id
        stock_code = order.stock_code
        status = order.order_status
        traded_volume = order.traded_volume
        order_volume = order.order_volume

        # 把订单信息存起来
        self.orders[order_id] = {
            'stock_code': stock_code,
            'status': status,
            'traded_volume': traded_volume,
            'order_volume': order_volume,
            'price': order.price
        }

        # 根据不同的状态打印不同的信息
        if status == 50:  # 已报
            logging.info(f"✓ 订单 {order_id} 已报到交易所 ({stock_code})")

        elif status == 55:  # 部成（部分成交）
            logging.info(
                f"⚡ 订单 {order_id} 部分成交！"
                f"已成交 {traded_volume} 股，还剩 {order_volume - traded_volume} 股"
            )

        elif status == 56:  # 已成（全部成交）
            logging.info(
                f"✓✓ 订单 {order_id} 全部成交！"
                f"成交数量：{traded_volume} 股，成交价格：{order.price}"
            )

        elif status == 54:  # 已撤
            logging.info(
                f"✗ 订单 {order_id} 已被撤销"
                f"已成交 {traded_volume} 股，未成交 {order_volume - traded_volume} 股"
            )

    def on_stock_trade(self, trade):
        """
        每当有成交发生，这个方法就会被调用。
        注意：on_stock_order 和 on_stock_trade 的区别：
        - on_stock_order：订单状态变化（可能多次）
        - on_stock_trade：实际成交（每笔成交都会触发）
        """
        logging.info(
            f"💰 成交回报：{trade.stock_code} "
            f"成交价 {trade.traded_price} × {trade.traded_volume} 股 = "
            f"{trade.traded_amount} 元"
        )

    def on_order_error(self, order_error):
        """如果下单失败，这个方法会被调用"""
        logging.error(
            f"❌ 订单 {order_error.order_id} 下单失败！"
            f"错误原因：{order_error.error_msg}"
        )

    def get_order_status(self, order_id):
        """查询某个订单的当前状态"""
        if order_id in self.orders:
            return self.orders[order_id]
        else:
            return None


# ============================================================================
# 第二步：主程序
# ============================================================================

def main():
    # 配置你的账户信息
    path = 'D:/XXXXQMT交易端模拟/userdata_mini/'  # 【修改】为你的路径
    account_id = '111100011'  # 【修改】为你的客户号

    # 初始化交易对象
    session_id = int(time.time())
    xt_trader = XtQuantTrader(path, session_id)
    acc = StockAccount(account_id)

    # 创建并注册我们的订单追踪器
    tracker = OrderTracker()
    xt_trader.register_callback(tracker)

    # 启动交易引擎
    xt_trader.start()

    # 连接到 MiniQMT
    if xt_trader.connect() != 0:
        logging.error("连接失败！请检查 MiniQMT 是否已启动并登录")
        return

    logging.info("连接成功！")

    # 订阅账户
    if xt_trader.subscribe(acc) != 0:
        logging.error("订阅失败！")
        return

    logging.info("订阅成功！")

    # 等待一下，确保订阅信息处理完毕
    time.sleep(1)

    # ========================================================================
    # 现在开始下单！
    # ========================================================================

    stock_code = '600000.SH'  # 浦发银行
    price = 8.0  # 设置一个相对较低的价格，这样不会立即成交
    volume = 100  # 买 100 股

    logging.info(f"准备下单：买入 {stock_code} {volume} 股，价格 {price} 元")

    # 下单
    order_id = xt_trader.order_stock(
        acc,
        stock_code,
        STOCK_BUY,
        volume,
        FIX_PRICE,
        price,
        'test_strategy',  # 策略名称
        'test_order'  # 订单备注
    )

    if order_id <= 0:
        logging.error(f"下单失败，错误码：{order_id}")
        xt_trader.stop()
        return

    logging.info(f"下单成功！订单编号：{order_id}")

    # ========================================================================
    # 等待订单的各种状态变化
    # ========================================================================

    logging.info("现在程序会持续监听订单状态，按 Ctrl+C 停止")
    logging.info("=" * 60)

    try:
        # 这个方法会一直运行，直到你按 Ctrl+C 或调用 xt_trader.stop()
        xt_trader.run_forever()
    except KeyboardInterrupt:
        logging.info("\n收到停止信号，准备清理...")

        # 如果订单还在途中，可以选择撤销
        order_status = tracker.get_order_status(order_id)
        if order_status and order_status['status'] in [49, 50]:  # 待报或已报
            logging.info(f"订单还在途中，准备撤销...")
            cancel_result = xt_trader.cancel_order_stock(acc, order_id)
            if cancel_result == 0:
                logging.info("撤单请求已发送")
            else:
                logging.error(f"撤单失败，错误码：{cancel_result}")

        # 等待一下，让撤单完成
        time.sleep(2)

        # 停止交易引擎
        xt_trader.stop()
        logging.info("程序已停止")


if __name__ == "__main__":
    main()

代码日志：