迅投实盘1：一个简单的下单程序，跑通python实盘全流程

Ne0inhk

16 Mar 2026 — 5 min read

博主使用迅投投研的程序开始要写实盘的代码了，这个系列的博客就是记录使用迅投的各种方法，一边学习一边记录。

迅投把python分为两个版本：

内置python：指的是使用迅投客户端，客户端可以新建python脚本写策略
原生python：使用xtquant包就可以通过本地python操作客户端执行策略

内置python依赖客户端编辑器且无法调试，因此这个系列博主尽可能的使用本地的python，在Pycharm中进行开发。

迅投官方知识库：https://dict.thinktrader.net/
xtquant官方说明文档：https://dict.thinktrader.net/nativeApi/start_now.html

下单的示例程序

程序逻辑说明

这里我们就执行一个最简单的操作，看是否打通了整个流程：

需要先安装好python环境，并且把xtquant的压缩包放到根目录，使用如下代码判断环境是否配置好

import time, datetime, sys from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback from xtquant.xttype import StockAccount from xtquant import xtconstant from loguru import logger # 一个日志模块，安装方式：pip install loguruimport pandas as pd # 常用数据分析工具，安装方式：pip install pandas

首先是一些很套路的写法，用于链接本地python与终端

# 步骤1. 初始化 path =r'E:\迅投极速交易终端睿智融科版\userdata'# 这个填"自己的终端安装路径\userdata" session_id =int(time.time())# 这个可以自定义，确保唯一即可 xt_trader = XtQuantTrader(path, session_id) account = StockAccount('xxxxx','STOCK')# 在终端中找到自己的股票账号

配置交易环境
代码中下单等操作均为异步（迅投很擅长高频交易，所以建议从一开始就有异步下单的习惯）

classMyXtQuantTraderCallback(XtQuantTraderCallback):defon_disconnected(self):""" 连接断开 :return: """print(datetime.now(),'连接断开回调')defon_stock_order(self, order):""" 委托回报推送 :param order: XtOrder对象 :return: """print(datetime.now(),'委托回调 投资备注', order.order_remark)defon_stock_trade(self, trade):""" 成交变动推送 :param trade: XtTrade对象 :return: """print(datetime.now(),'成交回调', trade.order_remark,f"委托方向(48买 49卖) {trade.offset_flag} 成交价格 {trade.traded_price} 成交数量 {trade.traded_volume}")defon_order_error(self, order_error):""" 委托失败推送 :param order_error:XtOrderError 对象 :return: """# print("on order_error callback")# print(order_error.order_id, order_error.error_id, order_error.error_msg)print(f"委托报错回调 {order_error.order_remark}{order_error.error_msg}")defon_cancel_error(self, cancel_error):""" 撤单失败推送 :param cancel_error: XtCancelError 对象 :return: """print(datetime.now(), sys._getframe().f_code.co_name)defon_order_stock_async_response(self, response):""" 异步下单回报推送 :param response: XtOrderResponse 对象 :return: """print(f"异步委托回调 投资备注: {response.order_remark}")defon_cancel_order_stock_async_response(self, response):""" :param response: XtCancelOrderResponse 对象 :return: """print(datetime.now(), sys._getframe().f_code.co_name)defon_account_status(self, status):""" :param response: XtAccountStatus 对象 :return: """print(datetime.now(), sys._getframe().f_code.co_name)# 步骤2.连接交易 trade_call_back = MyXtQuantTraderCallback() xt_trader.register_callback(trade_call_back) xt_trader.start()# 启动交易线程# 步骤3.连接交易主推 connect_result = xt_trader.connect() subscribe_result = xt_trader.subscribe(account)# 获取交易主推

其中MyXtQuantTraderCallback这个是实盘中，出现事件后的回调类，函数名都是固定的，注意不要修改。

更多细节请参考：XtQuant.Xttrade 交易模块

最后一步就是下单操作了

比如我们买入平安银行100股，按照最新价下单：

xt_trader.order_stock_async(account,'000001.SZ', xtconstant.STOCK_BUY,100, xtconstant.LATEST_PRICE,-1,'测试下单函数',"测试下单函数")

这个下单函数适用于股票期货期权等，实际上这是一个委托，详细说明如下：

deforder_stock_async(self, account, stock_code, order_type, order_volume, price_type, price, strategy_name='',order_remark=''):""" :param account: 证券账号 :param stock_code: 证券代码, 例如"600000.SH" :param order_type: 委托类型, 23:买, 24:卖 :param order_volume: 委托数量, 股票以'股'为单位, 债券以'张'为单位 :param price_type: 报价类型, 详见帮助手册 :param price: 报价价格, 如果price_type为指定价, 那price为指定的价格, 否则填0 :param strategy_name: 策略名称 :param order_remark: 委托备注 :return: 返回下单请求序号, 成功委托后的下单请求序号为大于0的正整数, 如果为-1表示委托失败 """

这里的所有字段都可以参考：委托 XtOrder

完整代码

import time, datetime, sys from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback from xtquant.xttype import StockAccount from xtquant import xtconstant from loguru import logger classMyXtQuantTraderCallback(XtQuantTraderCallback):defon_disconnected(self):""" 连接断开 :return: """print(datetime.now(),'连接断开回调')defon_stock_order(self, order):""" 委托回报推送 :param order: XtOrder对象 :return: """print(datetime.now(),'委托回调 投资备注', order.order_remark)defon_stock_trade(self, trade):""" 成交变动推送 :param trade: XtTrade对象 :return: """print(datetime.now(),'成交回调', trade.order_remark,f"委托方向(48买 49卖) {trade.offset_flag} 成交价格 {trade.traded_price} 成交数量 {trade.traded_volume}")defon_order_error(self, order_error):""" 委托失败推送 :param order_error:XtOrderError 对象 :return: """# print("on order_error callback")# print(order_error.order_id, order_error.error_id, order_error.error_msg)print(f"委托报错回调 {order_error.order_remark}{order_error.error_msg}")defon_cancel_error(self, cancel_error):""" 撤单失败推送 :param cancel_error: XtCancelError 对象 :return: """print(datetime.now(), sys._getframe().f_code.co_name)defon_order_stock_async_response(self, response):""" 异步下单回报推送 :param response: XtOrderResponse 对象 :return: """print(f"异步委托回调 投资备注: {response.order_remark}")defon_cancel_order_stock_async_response(self, response):""" :param response: XtCancelOrderResponse 对象 :return: """print(datetime.now(), sys._getframe().f_code.co_name)defon_account_status(self, status):""" :param response: XtAccountStatus 对象 :return: """print(datetime.now(), sys._getframe().f_code.co_name)# 步骤1. 初始化 path =r'E:\迅投极速交易终端睿智融科版\userdata'# 这里写自己的迅投路径 session_id =int(time.time()) xt_trader = XtQuantTrader(path, session_id) account = StockAccount('xxxxxxx','STOCK')# 这里写自己的股票代码# 步骤2.连接交易 trade_call_back = MyXtQuantTraderCallback() xt_trader.register_callback(trade_call_back) xt_trader.start()# 启动交易线程# 步骤3.连接交易主推 connect_result = xt_trader.connect() subscribe_result = xt_trader.subscribe(account)# 获取交易主推# 执行下单！ logger.debug("准备下单") xt_trader.order_stock_async(account,'000001.SZ', xtconstant.STOCK_BUY,100, xtconstant.LATEST_PRICE,-1,'测试下单函数',"测试下单函数") logger.success("下单完成")

执行完成后，就可以在终端的【交易】->【股票交易】中看到结果了，其中：