内置Python行情与数据函数
策略要能下单,先得能取行情。QMT 内置 Python 的行情能力分两半:历史数据(落本地、回测和盘后分析用)和实时数据(盘中订阅推送)。理解「先下载、再取数」的流程,以及哪些周期是合成的,是用对这一组函数的关键。
先下载数据:download_history_data
QMT 的历史数据存在本地,取数前要先把对应周期的数据下载下来。下载函数:
download_history_data(stockcode, period, startTime, endTime)
| 参数 | 类型 | 说明 |
|---|---|---|
stockcode | string | 品种代码,'600000.SH' 形式 |
period | string | 周期:'tick'/'1m'/'5m'/'1d' 等基础周期 |
startTime | string | 起始时间,'20200101' 或 '20200101093000',可空 |
endTime | string | 结束时间,格式同上,可空 |
incrementally | bool | 可选,从本地最后一条往后增量下载 |
def init(C):
download_history_data('000001.SZ', '1d', '20230101', '') # 全量
download_history_data('000001.SZ', '1d', '20230101', '', incrementally=True) # 增量
也可以在客户端「操作 → 数据管理」批量下载,并在右下角「行情 → 批量下载」勾「定时下载」做每日自动更新。
周期与合成规则
QMT 的周期分基础周期(实际存储)和合成周期(由基础周期合成):
| 分类 | 周期 |
|---|---|
| 基础周期 | tick、1m、5m、1d |
| 分钟合成 | 3m(←1m);10m/15m/30m/60m/2h/3h/4h(←5m) |
| 日以上合成 | 2d/3d/5d/1w/1mon/1q/1hy/1y(←1d) |
取数规则:
- 取历史合成周期:要先下载基础周期。如取 15m 历史要先下 5m。
- 取实时:可直接订阅原始周期(如直接订阅 15m)。
- 同时用基础和合成周期(如 5m 和 15m),只需下一次 5m。
取历史行情:get_market_data_ex
取历史 K 线的主力函数,挂在 ContextInfo 上:
data = C.get_market_data_ex(
fields=[], # 字段列表,如 ['close','open','high','low','volume']
stock_code=[], # 品种列表,如 ['600000.SH']
period='follow', # 周期,'follow'=跟随主图
start_time='', # 起始时间
end_time='', # 结束时间
count=-1, # 取最近 N 根,-1=不限
dividend_type='follow', # 复权类型
fill_data=True, # 是否补齐缺失
subscribe=False # 是否订阅实时(回测填 False,读本地)
)
返回 {品种代码: DataFrame} 字典,DataFrame 列即请求的字段:
data = C.get_market_data_ex(['close'], [C.stock], end_time=bar_date,
period=C.period, count=20, subscribe=False)
closes = list(data[C.stock].iloc[:, 0]) # 最近 20 根收盘价
⚠️ 该函数不建议在
init里调用——init阶段只能取到本地已有数据,行情未必就绪。需要预加载数据,放在第一次handlebar或用定时任务。
该函数除行情外,还能取特色数据(资金流向、订单流等),获取方式见 数据字典。
订阅实时推送:subscribe_quote
盘中想随分笔行情即时反应,用 subscribe_quote 在 init 里订阅,指定回调函数:
def init(C):
def on_data(data):
for stock in data:
cur = data[stock]['close']
pre = data[stock]['preClose']
ratio = cur / pre - 1
print(stock, '涨幅', ratio)
C.subscribe_quote('600000.SH', period='1d', callback=on_data)
订阅全市场 K 线(K 线全推,需 VIP 权限):
def init(C):
stocks = C.get_stock_list_in_sector('沪深A股')
for s in stocks:
C.subscribe_quote(stock_code=s, period='1m',
dividend_type='none', result_type='dict',
callback=on_bar)
全市场快照:get_full_tick
要一次性拿一大批品种的最新快照(如全市场扫描涨幅),用 get_full_tick,比逐个订阅快得多:
def job(C):
ticks = C.get_full_tick(A.hsa) # A.hsa 是品种列表
for stock in A.hsa:
ratio = ticks[stock]['lastPrice'] / ticks[stock]['lastClose'] - 1
if ratio > 0.09:
print(stock, '涨停附近', ratio)
返回的每个品种含 lastPrice、lastClose、open、high、low、volume 等字段。
品种与板块
| 函数 | 作用 | 示例 |
|---|---|---|
C.get_stock_list_in_sector(name) | 取板块成分代码列表 | C.get_stock_list_in_sector('沪深A股') |
C.get_stock_name(code) | 取品种中文名 | C.get_stock_name('600000.SH') |
C.get_instrument_detail(code) | 取合约基础信息(上市日、流通股本、涨跌停价等) | 见下 |
C.get_trading_dates(...) | 取交易日列表 | — |
C.get_open_date(code) | 取上市日 | — |
get_instrument_detail 返回字典,常用字段:
info = C.get_instrument_detail('600000.SH')
info['InstrumentName'] # 名称
info['OpenDate'] # 上市日期
info['UpStopPrice'] # 涨停价
info['DownStopPrice'] # 跌停价
info['FloatVolume'] # 流通股本
info['TotalVolume'] # 总股本
info['ProductType'] # 合约类型
旧版本客户端里该函数名为
ContextInfo.get_instrumentdetail,新版本统一为get_instrument_detail。
时间转换:timetag_to_datetime
K 线时间标签和可读时间互转:
bar_date = timetag_to_datetime(C.get_bar_timetag(C.barpos), '%Y%m%d%H%M%S')
# 如 '20260813150000'
行情函数选择速查
| 需求 | 函数 | subscribe |
|---|---|---|
| 回测取历史 K 线 | get_market_data_ex | False |
| 实盘取最近 N 根 K 线 | get_market_data_ex | True 或默认 |
| 盘中随分笔反应 | subscribe_quote | — |
| 全市场一次性快照 | get_full_tick | — |
| 取板块成分 | get_stock_list_in_sector | — |
| 取合约信息 | get_instrument_detail | — |
下一步
行情和下单都会了。如果想脱离 QMT 客户端、用自己的 Python 环境跑,看 XtQuant外部接口。