策略示例与回测
前面几章讲了机制、下单、行情、数据。这一章把它们串起来，给四套抄起来改参数就能跑的完整策略：基于 handlebar 的回测、双均线实盘立即下单、subscribe 订阅下单、run_time 定时任务全市场扫描，最后讲回测参数和常见陷阱。
示例 1：双均线回测（handlebar）
回测用 handlebar 遍历本地历史，subscribe=False 读本地数据、速度快。复制到策略编辑器，主图选要交易的股票，点回测即可：
#coding:gbk
import numpy as np
def init(C):
C.stock = C.stockcode + '.' + C.market
C.fast = 10 # 快线
C.slow = 20 # 慢线
C.accountid = 'testS'
def handlebar(C):
bar_date = timetag_to_datetime(C.get_bar_timetag(C.barpos), '%Y%m%d%H%M%S')
data = C.get_market_data_ex(['close'], [C.stock], end_time=bar_date,
period=C.period, count=C.slow, subscribe=False)
closes = list(data[C.stock].iloc[:, 0])
if len(closes) < C.slow:
return
ma_f = round(np.mean(closes[-C.fast:]), 2)
ma_s = round(np.mean(closes[-C.slow:]), 2)
account = get_trade_detail_data('test', 'stock', 'account')[0]
holdings = {h.m_strInstrumentID + '.' + h.m_strExchangeID: h.m_nVolume
for h in get_trade_detail_data('test', 'stock', 'position')}
hold = holdings.get(C.stock, 0)
if hold == 0 and ma_f > ma_s:
vol = int(account.m_dAvailable / closes[-1] / 100) * 100 # 整百股
passorder(23, 1101, C.accountid, C.stock, 5, -1, vol, C) # 金叉买入
C.draw_text(1, 1, '开')
elif hold > 0 and ma_f < ma_s:
passorder(24, 1101, C.accountid, C.stock, 5, -1, hold, C) # 死叉卖出
C.draw_text(1, 1, '平')回测撮合规则：指定价在当根 K 线高低点内按指定价撮合，超出按收盘价；委托量大于可用量按可用量撮合。
示例 2：双均线实盘（handlebar + 立即下单）
实盘要点：用 is_last_bar() 跳过历史重放、过滤非交易时段、quicktrade=2 立即下单、用普通全局对象 A 存状态、waiting_list 防超单：
#coding:gbk
import numpy as np, datetime
class a(): pass
A = a()
def init(C):
A.stock = C.stockcode + '.' + C.market
A.acct = account # 模型交易界面选的账号（内置变量）
A.acct_type = accountType
A.amount = 10000
A.fast, A.slow = 17, 27
A.waiting = []
A.buy_code = 23 if A.acct_type == 'STOCK' else 33
A.sell_code = 24 if A.acct_type == 'STOCK' else 34
def handlebar(C):
if not C.is_last_bar(): # 只处理最新 K 线
return
now = datetime.datetime.now().strftime('%H%M%S')
if now < '093000' or now > '150000':
return
account = get_trade_detail_data(A.acct, A.acct_type, 'account')
if len(account) == 0:
print('账号未登录')
return
avail = int(account[0].m_dAvailable)
# 核对未成交委托
if A.waiting:
deals = get_trade_detail_data(A.acct, A.acct_type, 'deal')
done = [d.m_strRemark for d in deals if d.m_strRemark in A.waiting]
A.waiting = [w for w in A.waiting if w not in done]
if A.waiting:
print('有未查到委托，暂停报单', A.waiting)
return
holdings = {h.m_strInstrumentID + '.' + h.m_strExchangeID: h.m_nCanUseVolume
for h in get_trade_detail_data(A.acct, A.acct_type, 'position')}
data = C.get_market_data_ex(['close'], [A.stock], period='1d', count=A.slow + 1)
cl = data[A.stock].values
if len(cl) < A.slow + 1:
return
pre_f, pre_s = np.mean(cl[-A.fast-1:-1]), np.mean(cl[-A.slow-1:-1])
cur_f, cur_s = np.mean(cl[-A.fast:]), np.mean(cl[-A.slow:])
vol = int(A.amount / cl[-1] / 100) * 100
if (A.amount < avail and vol >= 100 and A.stock not in holdings
and pre_f < pre_s and cur_f > cur_s):
msg = f'双均线实盘 买入 {vol}股'
passorder(A.buy_code, 1101, A.acct, A.stock, 14, -1, vol,
'双均线实盘', 2, msg, C)
A.waiting.append(msg)
if (A.stock in holdings and holdings[A.stock] > 0
and pre_f > pre_s and cur_f < cur_s):
msg = f'双均线实盘 卖出 {holdings[A.stock]}股'
passorder(A.sell_code, 1101, A.acct, A.stock, 14, -1, holdings[A.stock],
'双均线实盘', 2, msg, C)
A.waiting.append(msg)示例 3：订阅下单（subscribe）
基于实时分笔推送，新分笔到达触发回调。注意下单函数需要 ContextInfo，所以回调里用闭包捕获 C：
#coding:gbk
class a(): pass
A = a()
A.bought = []
def init(C):
def on_tick(data):
for stock in data:
cur = data[stock]['close']
pre = data[stock]['preClose']
ratio = cur / pre - 1
if ratio > 0 and stock not in A.bought:
msg = f'{stock} 涨幅{ratio:.2%} 买入100股'
print(msg)
# passorder(23, 1101, account, stock, 5, -1, 100,
# '订阅下单', 2, msg, C) # 实测时放开
A.bought.append(stock)
for stock in ['600000.SH', '000001.SZ']:
C.subscribe_quote(stock, period='1d', callback=on_tick)订阅下单默认注释掉真实报单，实测时再放开，避免误下单。
示例 4：定时任务全市场扫描（run_time）
用 run_time 固定间隔触发，配合 get_full_tick 一次取全市场快照，扫描涨幅：
#coding:gbk
import time, datetime
class a(): pass
A = a()
def init(C):
A.hsa = C.get_stock_list_in_sector('沪深A股')
A.vol = {s: C.get_last_volume(s) for s in A.hsa}
A.bought = []
C.run_time('scan', '1nSecond', '2026-08-13 13:20:00') # 每秒一次
def scan(C):
t0 = time.time()
now = datetime.datetime.now()
ticks = C.get_full_tick(A.hsa)
total_mv, total_ratio = 0, 0
for s in A.hsa:
ratio = ticks[s]['lastPrice'] / ticks[s]['lastClose'] - 1
if ratio > 0.09 and s not in A.bought:
print(f'{now} {s} {C.get_stock_name(s)} 涨幅{ratio:.2%}')
# passorder(...) 实测时放开
A.bought.append(s)
mv = ticks[s]['lastPrice'] * A.vol[s]
total_ratio += ratio * mv
total_mv += mv
print(f'{now} A股加权涨幅 {total_ratio/total_mv*100:.2f}% '
f'耗时{time.time()-t0:.3f}s')回测操作流程
- 下载数据：客户端「操作 → 数据管理」，选周期、板块、时间范围「全部」。
- 写策略：策略编辑器粘贴代码，确保第一行
#coding:gbk。 - 设基本信息：默认周期、默认主图（在「我的界面」点回测生效）。
- 副图运行：回测必须以副图模式执行，不要选主图/主图叠加。
- 看报告：收益曲线、净值走势、逐笔日志。
回测参数字段
回测右侧的基本信息与回测参数（如默认周期、主图、资金、起止时间等）决定回测行为；在行情 K 线下点回测则以当前 K 线的周期品种为准。具体字段以客户端「回测参数」面板为准。
常见陷阱
| 陷阱 | 说明 | 对策 |
|---|---|---|
没写 #coding:gbk | 中文乱码、报错 | 第一行必写 |
| 缩进混用 | 缩进错误 | 全文统一空格或 Tab |
| 回测选主图模式 | 回测失败 | 必须副图模式 |
| 多品种未下载数据 | 取数返回空 | 先下载对应周期数据 |
实盘在 init 取实时行情 | 行情未就绪、脏数据 | 实时查询放 9:10 后或 handlebar 内 |
实盘立即下单存 ContextInfo | 状态丢失 | quicktrade=2 用普通全局变量 |
| 未防超单 | 重复报单 | waiting_list + 成交回报核对 |
| 价格笼子废单 | 报价超 ±2% | 控制报价在最新价 ±2% 内 |
三种机制的选用
| 策略类型 | 推荐机制 | 理由 |
|---|---|---|
| 回测 | handlebar | 唯一支持回测的机制 |
| 模拟逐 K 线的实盘 | handlebar（quicktrade=0） | 回测代码可直接迁移 |
| 盘中即时反应 | subscribe_quote | 分笔驱动 |
| 全市场轮询/定时调仓 | run_time | 固定间隔，不逐笔 |
下一步
到这里 QMT 的机制、函数、数据、示例都讲完了。需要查具体函数签名可回到对应章节，或回到 迅投 QMT 总览。想用外部 Python 环境跑，看 XtQuant外部接口。
