问题描述
请帮我编写一个分钟级别的成交量突破日内交易策略。
解决方案
这是一个基于 QMT Python API 编写的 分钟级别成交量突破日内交易策略。
策略逻辑说明
- 核心思想：捕捉日内资金活跃的时刻。当某分钟的成交量剧烈放大（超过过去一段时间平均成交量的 N 倍），且价格上涨时，视为买入信号。
- 运行周期：1分钟线（或5分钟线）。
- 买入条件：
- 当前 K 线成交量 > 过去 20 根 K 线成交量均值的 2 倍（可调整）。
- 当前 K 线收盘价 > 开盘价（阳线）。
- 当前无持仓。
- 卖出条件（日内平仓）：
- 时间达到 14:55（收盘前强制平仓，规避隔夜风险）。
- 或者达到止损/止盈条件（代码中预留了逻辑位置）。
策略代码
# -*- coding: gbk -*-
import pandas as pd
import numpy as np
import time
def init(ContextInfo):
"""
策略初始化函数
"""
# --- 策略参数设置 ---
ContextInfo.stock = '600000.SH' # 交易标的：浦发银行（示例）
ContextInfo.period = '1m' # 周期：1分钟
ContextInfo.account_id = '6000000000' # 请替换为您的实盘/模拟资金账号
# 策略参数
ContextInfo.ma_len = 20 # 成交量均线周期
ContextInfo.vol_multiplier = 2.0 # 突破倍数（当前量 > 均量 * 2.0）
ContextInfo.stop_loss_pct = 0.02 # 止损百分比 (2%)
# 运行时变量
ContextInfo.holding = False # 是否持仓标记
ContextInfo.buy_price = 0.0 # 买入价格记录
# 设定股票池（必须设置，否则无法获取历史数据）
ContextInfo.set_universe([ContextInfo.stock])
print("策略初始化完成，监控标的: {}".format(ContextInfo.stock))
def handlebar(ContextInfo):
"""
K线运行函数，每根K线执行一次
"""
# 获取当前K线的时间
index = ContextInfo.barpos
timetag = ContextInfo.get_bar_timetag(index)
# 将时间戳转换为 HHMMSS 格式字符串，例如 "145500"
current_time_str = timetag_to_datetime(timetag, '%H%M%S')
# --- 1. 尾盘强制平仓逻辑 (14:55) ---
if current_time_str >= '145500' and ContextInfo.holding:
print("尾盘时间 {}，执行强制平仓".format(current_time_str))
order_target_value(ContextInfo.stock, 0, ContextInfo, ContextInfo.account_id)
ContextInfo.holding = False
return
# 如果是回测模式且不是最后一根K线，或者实盘模式下不是最新K线，则不执行后续逻辑
# 注意：实盘中通常只在最新K线走完或最新tick时交易，这里简化为每根bar结束判断
if not ContextInfo.is_last_bar():
return
# --- 2. 获取行情数据 ---
# 获取过去 N+2 根K线的数据，确保有足够数据计算MA
count = ContextInfo.ma_len + 5
data_map = ContextInfo.get_market_data_ex(
['open', 'close', 'volume'],
[ContextInfo.stock],
period=ContextInfo.period,
count=count,
dividend_type='front'
)
if ContextInfo.stock not in data_map:
return
df = data_map[ContextInfo.stock]
# 数据长度不足，无法计算均线，直接返回
if len(df) < ContextInfo.ma_len + 1:
return
# --- 3. 计算指标 ---
# 计算成交量均线 (不包含当前K线，用过去20根的均值作为基准)
# shift(1) 是为了取“前”20根的均值，避免当前K线巨大的成交量拉高均值本身
df['vol_ma'] = df['volume'].shift(1).rolling(window=ContextInfo.ma_len).mean()
# 获取最新一根K线的数据
current_bar = df.iloc[-1]
current_vol = current_bar['volume']
current_close = current_bar['close']
current_open = current_bar['open']
ref_vol_ma = current_bar['vol_ma']
# 防止除0错误或无效数据
if np.isnan(ref_vol_ma) or ref_vol_ma == 0:
return
# --- 4. 交易逻辑 ---
# 4.1 止损逻辑 (如果持仓)
if ContextInfo.holding:
if current_close < ContextInfo.buy_price * (1 - ContextInfo.stop_loss_pct):
print("触发止损，当前价: {}, 买入价: {}".format(current_close, ContextInfo.buy_price))
order_target_value(ContextInfo.stock, 0, ContextInfo, ContextInfo.account_id)
ContextInfo.holding = False
return # 持仓状态下不进行买入判断
# 4.2 买入开仓逻辑
# 条件1: 成交量突破 (当前量 > 均量 * 倍数)
condition_vol = current_vol > (ref_vol_ma * ContextInfo.vol_multiplier)
# 条件2: 价格上涨 (收盘 > 开盘，即阳线)
condition_price = current_close > current_open
# 条件3: 非尾盘 (14:30之后不再开新仓，防止被套无法T+0卖出 - 针对可转债或期货可忽略此条，股票T+1需注意)
# 注意：如果是A股股票T+1，日内策略通常指“底仓T+0”或“可转债/ETF”。
# 这里假设是可转债或有底仓，或者是为了演示日内买入逻辑。
condition_time = current_time_str < '143000'
if condition_vol and condition_price and condition_time:
print("=== 触发买入信号 ===")
print("时间: {}, 现量: {}, 均量: {}, 现价: {}".format(
current_time_str, current_vol, ref_vol_ma, current_close
))
# 下单买入 (示例：买入1手 / 100股)
# 实盘请注意：order_shares 的最后一个参数是账号ID
order_shares(ContextInfo.stock, 100, ContextInfo, ContextInfo.account_id)
ContextInfo.holding = True
ContextInfo.buy_price = current_close
# 辅助函数：封装下单接口 (使用 order_shares 更简单，也可以用 passorder)
def order_shares(stock_code, amount, ContextInfo, account_id):
"""
封装的按股数下单函数
amount > 0 买入
amount < 0 卖出
"""
# 23: 买入, 24: 卖出
opType = 23 if amount > 0 else 24
# 1101: 单股单账号普通交易
orderType = 1101
# 5: 最新价报价
prType = 5
# 调用 QMT 核心下单函数
passorder(opType, orderType, account_id, stock_code, prType, -1, abs(amount), ContextInfo)
print("下单执行: 代码={} 数量={}".format(stock_code, amount))
def order_target_value(stock_code, target_value, ContextInfo, account_id):
"""
封装的目标价值下单函数 (用于平仓，目标价值设为0即为全平)
注意：这里简化处理，实盘中建议使用 get_trade_detail_data 获取精确持仓
"""
# 这里简单演示全平逻辑，直接卖出所有持仓
# 在真实策略中，应该先查询持仓数量，再生成卖单
# 这里为了演示方便，假设我们知道要卖出，且使用 passorder 卖出
# 如果是平仓 (target_value == 0)
if target_value == 0:
# 24: 卖出
# 1123: 单股单账号可用比例下单 (1代表100%卖出)
passorder(24, 1123, account_id, stock_code, 5, -1, 1, ContextInfo)
print("执行全仓卖出: {}".format(stock_code))
代码关键点解析
-
# -*- coding: gbk -*-:- QMT 的 Python 编辑器默认使用 GBK 编码，这行代码必须放在第一行，否则中文注释会乱码或报错。
-
get_market_data_ex:- 这是 QMT 获取行情数据最高效的接口。
period='1m'：指定获取 1 分钟线数据。count=ContextInfo.ma_len + 5：我们只需要计算均线，所以不需要下载全部历史数据，取最近的 N+5 根即可，提高运行速度。dividend_type='front'：使用前复权数据，保证价格连续性。
-
成交量均线计算:
df['volume'].shift(1)：这一点非常重要。在计算突破时，均线的基准应该是不包含当前这根正在剧烈波动的K线的。如果把当前巨大的成交量算进均值里，均值会被拉高，导致突破条件变难。
-
时间控制 (
timetag_to_datetime):- 日内交易最重要的是平仓机制。代码中设置了
14:55强制平仓，防止持仓过夜（对于 T+0 品种如可转债、期货非常关键）。 - 如果是 A 股股票（T+1），当天的买入是无法当天卖出的。此策略如果是用于 A 股，通常需要配合底仓做 T+0，或者仅作为选股买入策略。
- 日内交易最重要的是平仓机制。代码中设置了
-
下单接口 (
passorder):- QMT 的核心下单函数是
passorder。代码中为了方便阅读，封装了order_shares简易函数。 opType=23代表买入，opType=24代表卖出。prType=5代表使用最新价下单。
- QMT 的核心下单函数是
如何在 QMT 中运行
- 打开 QMT 客户端，进入【模型研究】或【策略交易】界面。
- 新建一个 Python 策略。
- 将上述代码完整复制粘贴进去。
- 修改账号：将
ContextInfo.account_id修改为您实际的资金账号。 - 修改标的：将
ContextInfo.stock修改为您想交易的股票或可转债代码（如'110077.SH'）。 - 点击【编译】检查无误后，可以点击【运行】或【回测】。
常见问题 Q&A
Q: 这个策略可以直接用于 A 股股票吗？
A: A 股是 T+1 制度。如果你当天买入，当天是无法卖出的（无法执行 14:55 的平仓逻辑）。
- 如果是做 ETF 或 可转债（T+0 品种），此策略完全适用。
- 如果是做 股票，你需要有底仓（昨仓），并在代码中加入逻辑：先判断是否有昨仓，如果有，则买入后在尾盘卖出同等数量的昨仓（做 T）。
Q: 如何获取真实的持仓数据？
A: 代码中使用了 ContextInfo.holding 变量来模拟持仓状态。在实盘中，建议使用 get_trade_detail_data(account_id, 'stock', 'position') 接口来查询真实的账户持仓，这样更稳健。
Q: 为什么 get_market_data_ex 返回的数据要用 shift(1)？
A: 为了避免“未来函数”效应。我们在判断当前时刻是否突破时，参考的“平均水平”应该是过去的历史水平，不应该包含当前这根还没走完或者刚刚走出来的异常 K 线。