问题描述
解决方案
这是一个非常经典的量化交易策略,通常被称为布林带突破策略(Bollinger Band Breakout)。
策略逻辑
- 计算指标:基于过去
N天的收盘价计算移动平均线(Mean)和标准差(StdDev)。 - 确定通道:
- 上轨(Upper Band)= 均线 +
K倍标准差 - 下轨(Lower Band)= 均线 -
K倍标准差
- 上轨(Upper Band)= 均线 +
- 交易信号:
- 买入:当当前价格突破上轨(Price > Upper Band)时,视为波动率放大且趋势向上,全仓买入。
- 卖出:当当前价格跌破下轨(Price < Lower Band)时,或者跌破中轨(均线)时,视为趋势反转或止损,清仓卖出。(本示例采用跌破下轨卖出)。
PTrade 策略代码实现
以下是完整的策略代码,您可以直接复制到 PTrade 的策略编辑器中运行。
import numpy as np
import pandas as pd
def initialize(context):
"""
初始化函数,设置股票池、参数和手续费
"""
# 1. 设置要操作的股票,这里以贵州茅台为例
g.security = ['600519.SS']
set_universe(g.security)
# 2. 设置策略参数
g.N = 20 # 计算均线和标准差的周期(如20日)
g.k = 2.0 # 标准差的倍数(通常为2倍)
# 3. 设置回测费用(可选,为了回测更真实)
# 设置佣金为万分之三,最低5元
set_commission(commission_ratio=0.0003, min_commission=5.0)
# 设置滑点为0.1%
set_slippage(slippage=0.001)
def before_trading_start(context, data):
"""
盘前处理,每天开盘前运行一次
"""
# 可以在这里过滤停牌、ST股票等,本示例略过
pass
def handle_data(context, data):
"""
盘中运行函数,日线级别每天运行一次
"""
# 遍历股票池中的每一只股票
for stock in g.security:
# 1. 获取历史数据
# 获取过去 g.N 天的收盘价,include=False 表示不包含当前这根K线(避免未来函数)
# 我们需要用过去的数据来构建通道,然后看今天的价格是否突破
hist_data = get_history(g.N, '1d', 'close', stock, include=False)
# 容错处理:如果数据不足 N 天,跳过计算
if len(hist_data) < g.N:
log.info("股票 %s 数据不足,跳过" % stock)
continue
# 2. 计算统计指标
# 提取收盘价序列
close_prices = hist_data['close']
# 计算均值 (中轨)
mean_value = close_prices.mean()
# 计算标准差
std_value = close_prices.std()
# 计算上轨和下轨
upper_band = mean_value + g.k * std_value
lower_band = mean_value - g.k * std_value
# 3. 获取当前价格
# data[stock] 获取的是当前周期的快照数据
current_price = data[stock]['close']
# 获取当前该股票的持仓数量
position = get_position(stock)
current_amount = position.amount
# 4. 交易逻辑判断
# 信号1:价格突破上轨,且当前无持仓 -> 买入
if current_price > upper_band and current_amount == 0:
# 全仓买入(使用当前所有可用现金)
cash = context.portfolio.cash
if cash > 0:
order_value(stock, cash)
log.info("买入信号触发: %s, 现价: %.2f, 上轨: %.2f" % (stock, current_price, upper_band))
# 信号2:价格跌破下轨,且当前有持仓 -> 卖出
# 注意:也可以修改为跌破中轨(mean_value)就止盈止损,取决于策略激进程度
elif current_price < lower_band and current_amount > 0:
# 清仓卖出
order_target(stock, 0)
log.info("卖出信号触发: %s, 现价: %.2f, 下轨: %.2f" % (stock, current_price, lower_band))
def after_trading_end(context, data):
"""
盘后处理
"""
pass
代码关键点解析
get_history(..., include=False):- 这是最关键的一步。在计算布林带通道时,我们通常使用过去 N天的数据。
- 设置
include=False确保我们获取的是截止到昨天收盘的数据。如果设置为True,均线和标准差会包含今天的价格,导致通道随着今天的价格变动而剧烈变动,这通常不是突破策略想要的。
data[stock]['close']:- 这是获取当前回测时间点(或实盘当前时刻)的最新价格,用于和计算出的通道进行比较。
order_value与order_target:order_value(stock, cash):按金额下单,这里示例是用所有现金买入。order_target(stock, 0):调整持仓到目标数量,这里设为0即代表清仓。
- Python 版本兼容性:
- 代码中使用了
%格式化字符串(如"%s" % stock),这是为了兼容 Python 3.5 及更早版本,确保在所有 PTrade 环境中都能运行。
- 代码中使用了
常见问题 (Q&A)
Q: 如何修改策略为“跌破中轨(均线)”就卖出?
A: 将卖出逻辑部分的 lower_band 替换为 mean_value 即可:
elif current_price < mean_value and current_amount > 0:
order_target(stock, 0)
这通常被称为“布林带回归”或“均值回归”止盈,比跌破下轨卖出反应更快,能保住更多利润,但可能会在震荡市中频繁止损。
Q: 这个策略适合什么频率?
A: 上述代码默认是 1d(日线)级别。如果要在分钟级别运行,需要将 get_history 的频率参数改为 '1m' 或 '60m' 等,并在 initialize 中确认策略运行频率设置。
Q: 为什么有时候没有交易?
A: 波动率突破策略依赖于趋势。如果市场处于窄幅震荡(价格始终在上下轨之间波动),策略将不会触发任何交易信号。此外,请检查 g.N 和 g.k 的设置,g.k 设置得越大,通道越宽,触发交易的频率越低。