博主使用迅投投研的程序开始要写实盘的代码了，这个系列的博客就是记录使用迅投的各种方法，一边学习一边记录。
迅投把python分为两个版本：
内置python依赖客户端编辑器且无法调试，因此这个系列博主尽可能的使用本地的python，在Pycharm中进行开发。
迅投官方知识库：https://dict.thinktrader.net/ xtquant官方说明文档：https://dict.thinktrader.net/nativeApi/start_now.html
下单的示例程序
程序逻辑说明
这里我们就执行一个最简单的操作，看是否打通了整个流程：
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
# 步骤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("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)
# 步骤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("下单完成")
执行完成后，就可以在终端的【交易】->【股票交易】中看到结果了，其中：
