带你详细了解miniQMT的xttrader模块（第一篇）全案例分享，最全最用心

QUANT 2025-12-09 16:45:30 58 举报

①XtTraderCallback: 事件的实时接收函数

XtTraderCallback 是一个事件回调函数，用于实时接收交易系统的各类事件通知，例如委托成交、资金变动等。在关键事件发生时自动触发，从而让我们能够及时响应和处理。

from xtquant.xttrader import XtQuantTraderCallback

class MyCallback(XtTraderCallback):
    def on_stock_order(self, order):
        print(f"收到了一个委托回报: {order.order_id}")

②StockAccount: 我们的账户函数
这个对象代表了我们的证券资金账户

在进行任何与账户相关的操作时（如下单、查询资金），我们需要告诉 XtTrader 我们要操作的是哪个账户例如：

from xtquant.xttype import StockAccount
account_id = '0000111' # 填资金账号
acc = StockAccount(account_id)

③xtconstant: 买卖指令
在交易中，比如“买入”、“卖出”、“限价单”等xtconstant 模块就定义了这些术语的“代号”。使用这些代号，可以确保我们的指令被使用例如：

# STOCK_BUY 代表“证券买入”
# FIX_PRICE 代表“限价单”

基于上述内容整理了一套完整的代码，大家可以自行测试学习

import time
from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback
from xtquant.xttype import StockAccount

# --- 1. 定义回调类函数 ---
class MyCallback(XtQuantTraderCallback):
    def on_disconnected(self):
        print("连接断开")

    def on_stock_asset(self, asset):
        print("--- 资金发生变化 ---")
        print(f"总资产: {asset.total_asset}")
        print(f"可用资金: {asset.cash}")

# --- 2. 主程序入口 ---
def main():
    print("程序启动")

    # --- 配置路径和账户 ---
    path = 'D:/XXXXQMT交易端模拟/userdata_mini/' # 非常重要修改为路径
    account_id = '111110000011' # 【重要】修改为你的客户号
    session_id = int(time.time()) # 用当前时间戳作为 session_id

    # --- 创建核心对象 ---
    xt_trader = XtQuantTrader(path, session_id)
    acc = StockAccount(account_id)
    callback = MyCallback()

    # --- 注册回调 ---
    xt_trader.register_callback(callback)

    # --- 启动、连接、订阅 ---
    xt_trader.start()
    connect_result = xt_trader.connect()
    if connect_result != 0:
        print(f"连接失败，错误码: {connect_result}")
        return

    print("连接成功")
    subscribe_result = xt_trader.subscribe(acc)
    if subscribe_result != 0:
        print(f"订阅失败，错误码: {subscribe_result}")
        return

    print("订阅成功")

    # --- 查询一次资金 ---
    asset = xt_trader.query_stock_asset(acc)
    if asset:
        print("--- 首次查询资金 ---")
        print(f"总资产: {asset.total_asset}")
        print(f"可用资金: {asset.cash}")

    # --- 阻塞主线程，等待事件推送 ---
    print("程序正在运行，按 Ctrl+C 停止")
    try:
        xt_trader.run_forever()
    except KeyboardInterrupt:
        print("程序停止")
        xt_trader.stop()

if __name__ == "__main__":
    main()

代码日志如下：