迅投实盘 1：一个简单的下单程序，跑通 python 实盘全流程
环境说明
迅投将 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 loguru
import 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') # 在终端中找到自己的股票账号
- 配置交易环境 代码中下单等操作均为异步（迅投擅长高频交易，所以建议从一开始就有异步下单的习惯）。
class MyXtQuantTraderCallback(XtQuantTraderCallback):
def on_disconnected(self):
""" 连接断开 :return: """
print(datetime.now(), '连接断开回调')
def on_stock_order(self, order):
""" 委托回报推送 :param order: XtOrder 对象 :return: """
print(datetime.now(), '委托回调 投资备注', order.order_remark)
def on_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}")
def on_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}")
def on_cancel_error(self, cancel_error):
""" 撤单失败推送 :param cancel_error: XtCancelError 对象 :return: """
print(datetime.now(), sys._getframe().f_code.co_name)
def on_order_stock_async_response(self, response):
""" 异步下单回报推送 :param response: XtOrderResponse 对象 :return: """
print(f"异步委托回调 投资备注：{response.order_remark}")
def on_cancel_order_stock_async_response(self, response):
""" :param response: XtCancelOrderResponse 对象 :return: """
print(datetime.now(), sys._getframe().f_code.co_name)
def on_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, '测试下单函数', "测试下单函数")
这个下单函数适用于股票期货期权等，实际上这是一个委托，详细说明如下：
def order_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
class MyXtQuantTraderCallback(XtQuantTraderCallback):
def on_disconnected(self):
""" 连接断开 :return: """
print(datetime.now(), '连接断开回调')
def on_stock_order(self, order):
""" 委托回报推送 :param order: XtOrder 对象 :return: """
print(datetime.now(), '委托回调 投资备注', order.order_remark)
def on_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}")
def on_order_error(self, order_error):
""" 委托失败推送 :param order_error:XtOrderError 对象 :return: """
print(f"委托报错回调 {order_error.order_remark}{order_error.error_msg}")
def on_cancel_error(self, cancel_error):
""" 撤单失败推送 :param cancel_error: XtCancelError 对象 :return: """
(datetime.now(), sys._getframe().f_code.co_name)
():
()
():
(datetime.now(), sys._getframe().f_code.co_name)
():
(datetime.now(), sys._getframe().f_code.co_name)
path =
session_id = (time.time())
xt_trader = XtQuantTrader(path, session_id)
account = StockAccount(, )
trade_call_back = MyXtQuantTraderCallback()
xt_trader.register_callback(trade_call_back)
xt_trader.start()
connect_result = xt_trader.connect()
subscribe_result = xt_trader.subscribe(account)
logger.debug()
xt_trader.order_stock_async(account, , xtconstant.STOCK_BUY, , xtconstant.LATEST_PRICE, -, , )
logger.success()
执行完成后，就可以在终端的【交易】->【股票交易】中看到结果了，其中：
- 所有的委托都会在'任务列表'中展示，这是详细的说明委托的结果与当前成交进度。
- 所有委托也会在'委托'这个界面中展示。
- 成交的结果会显示在'成交'界面。
- 当前的持仓会在'持仓'界面（有时候持仓界面来不及及时刷新，等一会即可）。
