EA11 Python/QMT-miniQMT 策略

五、第一行代码：连接交易的完整流程

连接交易固定五步，一步都不能少：

from xtquant import xttrader
from xtquant.xttype import StockAccount

# 1. 创建交易对象：路径 + 会话ID
session_id = 12345 # 随便起一个唯一数字
xt_trader = xttrader.XtQuantTrader(r'D:\QMT\userdata_mini', session_id)

# 2. 启动交易线程
xt_trader.start()

# 3. 连接：返回 0 表示成功
connect_result = xt_trader.connect()
print('连接结果:', connect_result) # 0 = 成功，非 0 = 失败

# 4. 创建资金账号对象并订阅
account = StockAccount('你的资金账号')
subscribe_result = xt_trader.subscribe(account)
print('订阅结果:', subscribe_result) # 0 = 成功，-1 = 失败

# 5.（可选）让脚本一直活着，等回调
xt_trader.run_forever() # 阻塞当前线程，直到 stop() 被调用

1、三个容易踩的坑

• 路径必须是 userdata_mini：在你的 QMT 安装目录下找，不是 userdata。miniQMT 模式的数据目录专门叫这个名字。
• session_id 不能撞：同一台机器同时跑多个脚本，每个脚本的 session_id 必须不同，否则互相挤掉线。
• connect 是一次性连接：断线后不会自动重连，必须再次主动调用 connect()。稳妥的做法是配合回调 on_disconnected 做重连（见第十二章）。

白话注解：这五步就像打电话——创建话筒、开机、拨号、报身份、保持通话。少一步都是"喂喂喂"听不见。

六、查资金、持仓、委托、成交（四大查询）

# 资金
asset = xt_trader.query_stock_asset(account)
if asset: # 查询失败返回 None，先判空！
    print(f'总资产: {asset.total_asset}')
    print(f'可用资金: {asset.cash}')
    print(f'冻结资金: {asset.frozen_cash}')
    print(f'持仓市值: {asset.market_value}')

# 持仓列表
positions = xt_trader.query_stock_positions(account)
for pos in (positions or []):
    print(f'{pos.stock_code} 持仓 {pos.volume} 可用 {pos.can_use_volume} '
          f'成本 {pos.open_price} 昨仓 {pos.yesterday_volume}')

# 当日委托（第二个参数 True 表示只查可撤的委托）
orders = xt_trader.query_stock_orders(account, cancelable_only=False)

# 当日成交
trades = xt_trader.query_stock_trades(account)

返回 None 的陷阱：这四个查询在"失败"或"当天确实没有记录"时返回 None，不是空列表。直接 for pos in positions 会抛 TypeError。养成 positions or [] 的习惯。

白话注解：volume 是总持仓（含冻结），can_use_volume 才是此刻真正能卖的量。今天刚买入的股票 T+1，volume 有数、can_use_volume 是 0。

七、下单第一单：order_stock 八个参数逐个讲

下单接口签名（本地实测）：

order_stock(account, stock_code, order_type, order_volume,
            price_type, price, strategy_name='', order_remark='')

import xtquant.xtconstant as c

# 买入 100 股，限价 10.50
order_id = xt_trader.order_stock(
    account,
    '600000.SH',
c.STOCK_BUY, # 买入
100, # 100 股（A股一手 = 100 股）
c.FIX_PRICE, # 限价
    10.50,
'dual_ma', # 策略名
'' # 备注
)
print('订单编号:', order_id) # > 0 成功；-1 委托失败

# 卖出：注意用 can_use_volume 判断，不是 volume
sell_qty = 100
xt_trader.order_stock(account, '600000.SH', c.STOCK_SELL,
                      sell_qty, c.FIX_PRICE, 11.20, 'dual_ma', '')

• 返回值是系统生成的订单编号 order_id，大于 0 成功，-1 表示委托失败（比如资金不足、代码写错）。
• A股买入必须是 100 股的整数倍（科创板最低 200 股）；卖出可以卖零股（如分红送的碎股）。
• 代码后缀：.SH 沪市、.SZ 深市、.BJ 北交所。

白话注解：strategy_name 就像快递单上的"寄件人"字段，填了它，你查委托记录时一眼就知道这单是哪个策略干的。

八、拿最新价：get_full_tick（get_last_price 不存在）

警告：xtdata.get_last_price() 不存在。网上大量旧代码写了这个函数，本地实测 xtquant_250807 中没有此接口，直接调用会报 AttributeError。拿最新价的正确姿势是 get_full_tick：

from xtquant import xtdata

tick = xtdata.get_full_tick(['600000.SH'])
print(tick['600000.SH']['lastPrice']) # 最新价
print(tick['600000.SH']['askPrice']) # 卖一价（列表）
print(tick['600000.SH']['bidPrice']) # 买一价（列表）

要点：

• 参数是代码列表，可以一次取多只，也支持传市场名如 ['SH', 'SZ'] 拉全市场；
• 返回 {代码: 行情字典}，字典里的 lastPrice 就是最新价；
• 它取的是"全推快照"，收盘后取到的是收盘时点的快照。

白话注解：把"取最新价"想成看路口的实时监控画面——get_full_tick 就是"啪"给你截一帧。要连续看，就得订阅（下一章）。

九、订阅实时行情：subscribe_quote / subscribe_whole_quote

警告：xtdata.register_callback() 不存在。旧教程用 xtdata.register_callback(on_data) 全局注册回调，本地实测该接口在新版 xtquant 中已不存在。正确做法是把回调函数直接作为参数传给订阅接口。

1、单股订阅 subscribe_quote

from xtquant import xtdata

def on_data(datas):
    # datas 格式：{ 代码: [数据, ...] }
    for code, bars in datas.items():
        for bar in bars:
            print(code, bar['lastPrice'])

# 回调通过 callback 参数传入
seq = xtdata.subscribe_quote('600000.SH', period='tick', count=0,
                             callback=on_data)
print('订阅号:', seq) # > 0 成功，-1 失败

# 不想收了就反订阅
xtdata.unsubscribe_quote(seq)

# 脚本末尾需要阻塞住，否则主线程退出回调就没了
xtdata.run() # 持续循环等待，断线会抛异常

2、全推订阅 subscribe_whole_quote

def on_tick(datas):
    # datas 格式：{ 代码: 行情字典 }，注意和单股订阅不同，没有列表
    for code, tick in datas.items():
        print(code, tick['lastPrice'])

# 支持传市场名订阅全市场，或传代码列表只订几只
xtdata.subscribe_whole_quote(['SH', 'SZ'], callback=on_tick)
xtdata.run()

• subscribe_quote：订阅一只，K线或 tick 都行（period 决定），单股订阅建议不超过 50 只；
• subscribe_whole_quote：全推分笔快照，订阅后立刻推送一次当前最新数据，适合扫全市场或几十只以上；
• 两者返回的都是订阅号 seq，成功大于 0，失败 -1。

白话注解：单股订阅像"点外卖"，一单一单送到门口；全推像"自助餐"，所有菜（全市场行情）一起端上来，你自己挑。

十、委托生命周期与回调：XtQuantTraderCallback

主动查询是"你去问"，回调是"它主动报"。做实盘必须写回调，否则委托被拒了你都不知道。继承回调类、重写需要的方法：

from xtquant.xttrader import XtQuantTraderCallback

class MyCallback(XtQuantTraderCallback):
    def on_connected(self):
        print('已连接')

    def on_disconnected(self):
        print('断线了！触发重连逻辑')

    def on_stock_order(self, order):
        # 委托回报推送：每次委托状态变化都推一次
        print(f"委托回报 {order.stock_code} 状态 {order.order_status}")

    def on_stock_trade(self, trade):
        # 成交回报推送：真金白银成交了
        print(f"成交 {trade.stock_code} {trade.traded_volume}股 "
              f"@{trade.traded_price}")

    def on_order_error(self, order_error):
        # 委托失败推送：这里能看到 order_stock 返回 -1 的真正原因
        print(f"下单失败 {order_error.error_id}: {order_error.error_msg}")

    def on_cancel_error(self, cancel_error):
        print(f"撤单失败 {cancel_error.error_msg}")

    def on_order_stock_async_response(self, response):
        # 异步下单的正式回报（对应 seq）
        print(f"异步下单回报 seq={response.seq} order_id={response.order_id}")

    def on_account_status(self, status):
        print('账号状态变化', status.account_id)

# 注册回调后，再 start/connect
xt_trader.register_callback(MyCallback())
xt_trader.start()
xt_trader.connect()

本地实测回调类全部可用方法：on_connected / on_disconnected / on_stock_order / on_stock_trade / on_stock_asset / on_stock_position / on_order_error / on_cancel_error / on_order_stock_async_response / on_cancel_order_stock_async_response / on_account_status 及银行转账、约券等异步回报。重写你需要的即可。

白话注解：query 是你主动打电话问"我的单子咋样了"；回调是券商主动给你发短信"您的委托已成交"。两个都用，才不漏单。

十一、数据对象字段表：XtAsset / XtPosition / XtOrder / XtTrade

四个查询返回的对象字段如下（本地从包源码逐个提取，保证无虚构）：

1、XtAsset（query_stock_asset 返回）

2、XtPosition（query_stock_positions 返回）

3、XtOrder（query_stock_orders 返回）

4、XtTrade（query_stock_trades 返回）

十二、同步 vs 异步下单：order_stock_async 与撤单

1、同步下单 order_stock

• 调用后阻塞等柜台回执，返回订单编号 order_id（>0 成功 / -1 失败）；
• 简单可靠，但等待期间你的脚本干不了别的；扫几十只股票轮询下单时会很慢。

2、异步下单 order_stock_async

• 参数完全一样（本地实测签名一致），立刻返回一个请求序号 seq（>0 提交成功 / -1 提交失败）；
• seq ≠ order_id！真正的订单编号通过 on_order_stock_async_response 回调送达（response.seq 和 response.order_id 对得上号）；
• 失败信息从 on_order_error 回调来。

# 异步下单
seq = xt_trader.order_stock_async(account, '600000.SH',
                                  c.STOCK_BUY, 100, c.FIX_PRICE,
                                  10.50, 'dual_ma', '')
print('请求序号:', seq) # 等回调拿真正的 order_id

3、撤单

# 同步撤单：0 = 成功发出，-1 = 失败
xt_trader.cancel_order_stock(account, order_id)

# 异步撤单：返回请求序号，结果走 on_cancel_order_stock_async_response
xt_trader.cancel_order_stock_async(account, order_id)

# 只查还能撤的委托
cancelable = xt_trader.query_stock_orders(account, cancelable_only=True)
for od in (cancelable or []):
    xt_trader.cancel_order_stock(account, od.order_id)

白话注解：同步像窗口排队，办完才走；异步像取号机，号（seq）到手就走，叫号（回调）再回来处理。批量下单用异步，单笔下单用同步。

十三、期货交易专题：开平仓方向与平今平昨

期货下单也是 order_stock，但方向常量换成 FUTURE_*，数量单位是手：

import xtquant.xtconstant as c
from xtquant.xttype import StockAccount

future_acc = StockAccount('你的期货账号', 'FUTURE') # 关键：第二参数

# 开多 1 手，最新价（价格参数填 0）
oid = xt_trader.order_stock(
    future_acc,
'rb2605.SHFE', # 期货代码：品种+月份.交易所
c.FUTURE_OPEN_LONG, # 开多
1, # 1 手
c.LATEST_PRICE, # 最新价
    0,
    'dual_ma', ''
)

# 平今多（上海/能源中心区分平今平昨，平错方向会被拒单）
xt_trader.order_stock(future_acc, 'rb2605.SHFE',
                      c.FUTURE_CLOSE_LONG_TODAY, 1,
                      c.LATEST_PRICE, 0, 'dual_ma', '')

# 平昨多
xt_trader.order_stock(future_acc, 'rb2605.SHFE',
                      c.FUTURE_CLOSE_LONG_HISTORY, 1,
                      c.LATEST_PRICE, 0, 'dual_ma', '')

期货代码后缀（本地实测 xtconstant 市场常量）：.SHFE 上期所、.DCE 大商所、.CZCE 郑商所、.INE 能源中心、.CFFEX 中金所、.GFEX 广期所。

1、平今平昨规则（重要）

• 上期所（SHFE）/ 能源中心（INE）：严格区分平今/平昨，平今手续费常更高（个别品种平今免费），方向用错直接拒单；
• 郑商所 / 大商所 / 中金所：默认先平昨，不强制区分——但 xtquant 常量仍用 _TODAY/_HISTORY 传参，柜台自动处理；
• 先查 query_stock_positions 返回的 yesterday_volume（昨仓）决定该平今还是平昨。

再强调：期货账户用 StockAccount(账号, 'FUTURE') 创建，没有 FutureAccount 类。期货自带杠杆，教学代码之外，实盘务必先在模拟环境跑通。

十四、get_market_data 三兄弟怎么选

三者参数签名一致（本地实测）：(field_list=[], stock_list=[], period='1d', start_time='', end_time='', count=-1, dividend_type='none', fill_data=True)。

1、tick 数据的特别之处

period='tick' 时返回的不是 DataFrame，而是 {代码: numpy数组}（按时间戳升序），每个元素是分笔快照字典。做逐笔回测或盘口研究时注意格式差异。

2、fill_data 参数

fill_data=True（默认）会把停牌等造成的空缺用前一交易日的数据向后填充；做严格逐笔研究时设为 False，自己处理缺失。

白话注解：ex 版是"一人一张成绩单"，matrix 版是"全班一张大表"。分析一只股票拿 ex，横向对比几十只拿 matrix。

十五、板块与合约信息：查代码、涨跌停、最小变动价

1、板块成分股

from xtquant import xtdata

xtdata.download_sector_data() # 先下载板块分类信息
stocks = xtdata.get_stock_list_in_sector('沪深A股')
print(len(stocks), stocks[:5])

2、合约信息（还能验代码对不对）

info = xtdata.get_instrument_detail('600000.SH')
if info is None:
print('代码不存在！') # 查不到合约返回 None
else:
    print(info['InstrumentName']) # 浦发银行
print(info['UpStopPrice']) # 今日涨停价
print(info['DownStopPrice']) # 今日跌停价
print(info['PriceTick']) # 最小变动价位
    print(info['VolumeMultiple']) # 合约乘数（期货用）
print(info['IsTrading']) # 当前是否在交易

• get_instrument_detail(代码, iscomplete=False)：默认返回常用字段；iscomplete=True 返回手续费率、期权类型等全量字段；
• 合约信息（涨跌停价等）需要行情连接，实盘下单前用它做限价合法性校验，避免被柜台拒单。

白话注解：get_instrument_detail 就是"查户口"——这个代码存不存在、今天能交易吗、涨停跌停多少钱，一次问清。下单前查一下，能省一半拒单烦恼。

十六、财务数据与除权：get_financial_data / get_divid_factors

1、财务数据

xtdata.download_financial_data(['600000.SH'], table_list=['Income', 'Balance'])

data = xtdata.get_financial_data(['600000.SH'],
                                 table_list=['Balance', 'Income', 'CashFlow',
                                             'Capital', 'Holdernum',
                                             'Top10holder', 'Pershareindex'])
df = data['600000.SH']['Income'] # 利润表 DataFrame

可用表：Balance 资产负债表 / Income 利润表 / CashFlow 现金流量表 / Capital 股本表 / Holdernum 股东数 / Top10holder 十大股东 / Top10flowholder 十大流通股东 / Pershareindex 每股指标。

2、除权除息

df = xtdata.get_divid_factors('600000.SH') # 除权数据 DataFrame

K线复权直接在 get_market_data 的 dividend_type 里指定（front/back），不必手动算。注意：不同复权方式下算出的均线、收益率不同，回测前后用同一种复权，别混。

3、交易日历

dates = xtdata.get_trading_dates('SH', start_time='20240101',
                                 end_time='20241231')
holidays = xtdata.get_holidays() # 当年节假日（先 download_holiday_data）

十七、实盘部署：客户端在线与 session_id 规则

1、客户端在线是硬前提

miniQMT 模式下，QMT 客户端必须保持登录运行（行情连接、柜台通道都靠它转发）。客户端一关，所有接口立刻失效。

2、部署清单

1. 选 Windows 云主机/VPS（xtquant 和 QMT 客户端是 Windows 程序），网络离券商柜台越近越好；
2. 装 QMT 客户端，登录并切到 miniQMT 模式，设置开机自启；
3. pip install xtquant，脚本路径写死服务器上的 userdata_mini 绝对路径；
4. 脚本用 run_forever() 或主循环保活，注册 on_disconnected 重连；
5. 留意客户端定期强制更新和重新登录的机制，最好配套"每天收盘前人工确认在线"的流程。

3、session_id 规则

• 同一台机器上，每个脚本进程用不同的 session_id；
• 重复的 session_id 会导致两个脚本互相顶掉连接；
• 常见做法：以时间戳生成 int(time.time()) 或写配置文件固定分配。

白话注解：VPS 是"永远不关机的远程电脑"。策略放家里电脑，一停电一断网就全瞎；放 VPS 上 7×24 在线，客户端和脚本各占一个进程，各用各的 session_id。

十八、本课小结

• xtquant = xtdata（行情）+ xttrader（交易）+ xttype（对象）+ xtconstant（常量）；
• 连接五步走：建对象 → start → connect → subscribe → run_forever；
• order_stock 八参数下单，返回 order_id；async 版返回 seq；
• 查询四件套返回 None 而非空列表，先判空；
• 回调传给订阅函数（register_callback 不存在）；最新价用 get_full_tick（get_last_price 不存在）；
• 期货账户 StockAccount(账号, 'FUTURE')（FutureAccount 不存在），平仓只有平今/平昨；
• 历史数据先 download 再 get；实盘要防重复下单、查涨跌停、处理断线重连；
• xtquant 无回测引擎：回测用 Backtrader / vn.py，xtquant 接实盘。

白话注解：QMT 就是"券商给你的 Python 遥控器"。遥控器好用，但电视（客户端）必须开着；信号（回测）再好，也要记住：实盘是另一场比赛。

本站固定免责声明：本教程仅讲解软件编程与操作方法，不构成投资建议或策略推荐。回测业绩不代表未来收益，EA自动交易风险大，请谨慎决策。