问题描述
解决方案
这是一个基于**均值回归(Mean Reversion)**原理的日内交易策略。
策略逻辑
该策略的核心思想是:价格总是围绕均线波动。当价格大幅偏离均线(突破布林带上轨或下轨)时,认为价格被高估或低估,大概率会回归均值。
- 指标选择:使用布林带(Bollinger Bands)。
- 中轨:N分钟的移动平均线(MA)。
- 上轨:中轨 + K倍标准差。
- 下轨:中轨 - K倍标准差。
- 买入信号:当价格跌破下轨时,视为超卖,进行买入。
- 卖出信号:当价格突破上轨时,视为超买,卖出持仓。
- 日内风控:在收盘前(如14:55)强制平仓,不持仓过夜,规避隔夜风险。
适用标的
由于是日内交易(当天买当天卖),该策略最适合T+0交易品种,如:
- 可转债(代码通常以11、12开头)
- ETF(部分跨境ETF或债券ETF支持T+0,或者利用底仓进行股票T+0操作)
- 期货
*注意:如果是普通A股股票,由于是T+1制度,当日买入无法当日卖出,需要配合底仓(T+0工厂模式)才能实现日内回转。下方的代码以**可转债(T+0)*为例编写。
策略代码
import numpy as np
def initialize(context):
"""
策略初始化函数
"""
# 设定要交易的标的,这里以一个可转债为例(支持T+0)
# 如果是股票回测,请确保理解T+1限制
g.security = '128041.SZ'
# 设置股票池
set_universe(g.security)
# 策略参数设置
g.n = 20 # 均线周期(使用过去20根分钟K线)
g.k = 2.0 # 标准差倍数(布林带宽度)
# 设置手续费(可选,模拟真实交易成本)
# 股票/转债通常设置为万分之三左右
set_commission(commission_ratio=0.0003, min_commission=5.0)
def handle_data(context, data):
"""
盘中运行函数,分钟级别回测每分钟执行一次
"""
security = g.security
# ---------------------------------------------------
# 1. 时间控制:收盘前强制平仓(日内策略核心)
# ---------------------------------------------------
current_dt = context.blotter.current_dt
current_time = current_dt.strftime('%H:%M')
# 每天14:55之后强制平仓,不再开新仓
if current_time >= '14:55':
position = get_position(security)
if position.amount > 0:
log.info("收盘前强制平仓: %s" % security)
order_target(security, 0)
return
# ---------------------------------------------------
# 2. 获取数据
# ---------------------------------------------------
# 获取过去 N 个分钟的收盘价数据
# include=True 表示包含当前这一分钟的即时数据(回测中)
hist = get_history(g.n, '1m', 'close', security, fq=None, include=True)
# 如果数据长度不足,不进行计算
if len(hist['close']) < g.n:
return
# 将数据转换为numpy数组方便计算
close_prices = hist['close'].values
# ---------------------------------------------------
# 3. 计算指标 (布林带)
# ---------------------------------------------------
# 计算均值 (中轨)
mid_line = np.mean(close_prices)
# 计算标准差
std_dev = np.std(close_prices)
# 计算上轨和下轨
upper_band = mid_line + g.k * std_dev
lower_band = mid_line - g.k * std_dev
# 获取当前最新价格
current_price = data[security]['close']
# 获取当前持仓数量
position = get_position(security)
curr_amount = position.amount
# 获取当前可用现金
cash = context.portfolio.cash
# ---------------------------------------------------
# 4. 交易逻辑
# ---------------------------------------------------
# 信号1:价格跌破下轨 -> 买入 (做多)
# 限制:当前无持仓时才买入,避免连续加仓
if current_price < lower_band and curr_amount == 0:
# 全仓买入 (也可以设置为固定金额或固定数量)
if cash > 0:
order_value(security, cash)
log.info("价格(%.2f)跌破下轨(%.2f),买入 %s" % (current_price, lower_band, security))
# 信号2:价格突破上轨 -> 卖出 (平仓)
# 限制:当前有持仓时才卖出
elif current_price > upper_band and curr_amount > 0:
order_target(security, 0)
log.info("价格(%.2f)突破上轨(%.2f),卖出平仓 %s" % (current_price, upper_band, security))
def after_trading_end(context, data):
"""
盘后处理函数
"""
log.info("今日交易结束,资产净值: %s" % context.portfolio.portfolio_value)
代码详解
-
initialize:- 定义了全局变量
g.n和g.k,方便后续调整参数。 set_universe设置了标的。建议选择波动性较大且支持T+0的品种。
- 定义了全局变量
-
handle_data:- 强制平仓逻辑:这是日内策略最重要的一环。代码检测时间是否到达
14:55,如果是,则调用order_target(security, 0)卖出所有持仓,确保不持仓过夜。 - 数据获取:使用
get_history获取过去20分钟的收盘价。include=True确保计算用到的是最新数据。 - 指标计算:利用
numpy库计算均值(MA)和标准差(STD),进而得出布林带的上轨和下轨。 - 交易信号:
- 买入:
current_price < lower_band。这里加了curr_amount == 0的判断,意味着策略是“开仓-平仓”模式,而不是金字塔式加仓。 - 卖出:
current_price > upper_band。触发后清空仓位。
- 买入:
- 强制平仓逻辑:这是日内策略最重要的一环。代码检测时间是否到达
注意事项
- Python版本兼容性:代码使用了
%s格式化字符串,完全兼容 Python 3.5 及以上版本,未使用 f-string。 - API调用:所有API(如
get_history,order_value)均直接调用,符合PTrade规范。 - 回测频率:请在回测界面选择 “分钟线” 级别回测。
- 实盘与回测差异:
- 在实盘中,
handle_data每分钟运行一次。 - 如果需要更灵敏的反应(如Tick级别),需要使用
tick_data函数,但逻辑会更复杂(需要处理高频数据的噪声)。上述代码是基于分钟K线的,稳定性更好。
- 在实盘中,
常见问题 (Q&A)
Q: 为什么我回测股票时,下午卖出失败?
A: 普通A股是T+1制度。如果你上午买入,下午触发卖出信号是无法卖出的。如果你想做股票的日内交易,需要在 initialize 中使用 set_yesterday_position 设置底仓,或者交易可转债/ETF。
Q: 如何调整策略的灵敏度?
A: 修改 g.n 和 g.k:
- 减小
g.n(如改为10) 或 减小g.k(如改为1.5),策略会更敏感,交易次数变多,但假信号也会增加。 - 增大
g.n或g.k,策略会更稳健,交易次数减少。
Q: 可以做空吗?
A: PTrade支持融券做空(需开通两融权限)。如果需要做空,可以将买入逻辑改为 marginsec_open(融券卖出),卖出逻辑改为 marginsec_close(买券还券)。但上述代码仅实现了“低买高卖”的做多逻辑。