问题描述
如何编写一个策略，让它在日线级别判断趋势方向，然后在分钟线级别寻找买卖点？
解决方案
这是一个非常经典的量化交易思路，通常被称为**“多周期共振”**策略。
在 PTrade 中实现这个逻辑，最佳的架构如下：
- 日线趋势判断：放在
before_trading_start函数中执行。因为日线数据在开盘前就是固定的，不需要在盘中每分钟重复计算，这样可以提高回测和实盘的效率。 - 分钟线买卖点：放在
handle_data函数中执行。该函数在回测频率设为“分钟”时，会每分钟触发一次，适合捕捉盘中的瞬时信号。
策略逻辑示例
为了演示，我将编写一个具体的策略：
- 日线趋势（大方向）：使用 5日均线 > 10日均线 定义为上涨趋势。
- 分钟买卖（小波段）：在上涨趋势确立的前提下，使用 15分钟级别的 MACD 金叉 作为买入信号，死叉 作为卖出信号。
PTrade 策略代码
def initialize(context):
"""
初始化函数，设置股票池和全局变量
"""
# 设定要操作的股票，这里以恒生电子为例
g.security = '600570.SS'
set_universe(g.security)
# 策略参数设置
g.daily_short_window = 5 # 日线短周期均线
g.daily_long_window = 10 # 日线长周期均线
# 分钟级别MACD参数
g.minute_freq = '15m' # 使用15分钟K线寻找买卖点
# 趋势状态标识
g.is_uptrend = False
def before_trading_start(context, data):
"""
盘前处理函数：每天开盘前运行一次
用于判断日线级别的大趋势
"""
# 获取足够长的日线历史数据来计算均线
# 多取几天以防停牌或计算边界问题
count = g.daily_long_window + 5
history_data = get_history(count, '1d', 'close', g.security, fq='pre', include=False)
if len(history_data) < g.daily_long_window:
log.info("历史数据不足，跳过今日趋势判断")
g.is_uptrend = False
return
# 提取收盘价序列
close_prices = history_data['close'].values
# 计算短期和长期均线
# 注意：values[-N:] 取最后N个数据
ma_short = close_prices[-g.daily_short_window:].mean()
ma_long = close_prices[-g.daily_long_window:].mean()
# 判断趋势：短均线 > 长均线 视为上涨趋势
if ma_short > ma_long:
g.is_uptrend = True
log.info("今日日线趋势：看多 (MA5=%.2f > MA10=%.2f)" % (ma_short, ma_long))
else:
g.is_uptrend = False
log.info("今日日线趋势：看空/震荡 (MA5=%.2f <= MA10=%.2f)" % (ma_short, ma_long))
def handle_data(context, data):
"""
盘中处理函数：回测模式下每分钟运行一次
用于判断分钟级别的买卖点
"""
# 如果日线不是上涨趋势，且当前无持仓，则直接跳过，不进行买入操作
# 如果有持仓，即使日线转空，也需要在下面逻辑中寻找卖点平仓
position = get_position(g.security).amount
if not g.is_uptrend and position == 0:
return
# 获取分钟级别的历史数据用于计算MACD
# 获取过去100根15分钟K线
minute_history = get_history(100, g.minute_freq, 'close', g.security, fq='pre', include=True)
if len(minute_history) < 50:
return
close_prices = minute_history['close'].values
# 使用PTrade内置函数计算MACD
# get_MACD 返回三个数组: dif, dea, macd
dif, dea, macd_bar = get_MACD(close_prices, 12, 26, 9)
# 获取当前和上一周期的DIF、DEA值
curr_dif = dif[-1]
curr_dea = dea[-1]
prev_dif = dif[-2]
prev_dea = dea[-2]
# 资金情况
cash = context.portfolio.cash
current_price = data[g.security]['close']
# --- 交易逻辑 ---
# 买入信号：日线趋势向上 + 分钟线MACD金叉 (DIF上穿DEA)
# 金叉判定：上一个时刻 DIF < DEA，当前时刻 DIF > DEA
if g.is_uptrend and position == 0:
if prev_dif < prev_dea and curr_dif > curr_dea:
# 全仓买入
order_value(g.security, cash)
log.info("分钟级别MACD金叉，且日线趋势向上，买入 %s" % g.security)
# 卖出信号：分钟线MACD死叉 (DIF下穿DEA) 或者 日线趋势已经转坏(在before_trading_start中判定)
# 死叉判定：上一个时刻 DIF > DEA，当前时刻 DIF < DEA
elif position > 0:
# 触发死叉卖出
if prev_dif > prev_dea and curr_dif < curr_dea:
order_target(g.security, 0)
log.info("分钟级别MACD死叉，卖出止盈/止损 %s" % g.security)
# 补充逻辑：如果日线已经转为空头（隔日），强制清仓
elif not g.is_uptrend:
order_target(g.security, 0)
log.info("日线趋势转空，强制清仓 %s" % g.security)
代码关键点解析
-
before_trading_start的作用：- 这个函数在每天 9:30 开盘前运行。
- 我们在这里调用
get_history(..., '1d', ...)获取日线数据。 - 计算结果存入全局变量
g.is_uptrend。这相当于给当天的交易定了一个“基调”：今天只准买，还是只准卖（或观望）。
-
handle_data的作用：- 回测时选择分钟级别回测，这个函数会每分钟执行。
- 我们在这里调用
get_history(..., '15m', ...)。注意，虽然函数是每分钟运行，但我们获取的是过去 N 根 15分钟 的 K 线数据。 - MACD 计算：使用了 PTrade 内置的
get_MACD函数，它直接返回指标序列，非常方便。 - 金叉/死叉判断：通过比较当前时刻和上一时刻 DIF 与 DEA 的位置关系来判断交叉。
-
Python 版本兼容性：
- 代码使用了
%s和%f进行字符串格式化，完全兼容 Python 3.5（PTrade 的常用环境），没有使用高版本 Python 的 f-string。
- 代码使用了
如何在 PTrade 中运行
- 新建一个策略。
- 将上述代码复制粘贴到代码编辑区。
- 重要设置：
- 在右侧或下方的回测参数设置中，回测频率必须选择 “分钟”。如果选择“日线”，
handle_data每天只会运行一次，无法捕捉分钟级别的信号。
- 在右侧或下方的回测参数设置中，回测频率必须选择 “分钟”。如果选择“日线”，
- 点击“运行回测”。
常见问题 Q&A
Q: 为什么不在 handle_data 里同时计算日线和分钟线？
A: 虽然可以，但效率极低。日线指标在一天之内是不会变的（除非你用实时日线，但通常趋势判断用的是昨日收盘后的数据）。在 handle_data 里每分钟重复计算几百次相同的日线均线是浪费计算资源。
Q: get_history 在分钟回测里获取 '15m' 数据，返回的是什么？
A: 它返回的是截止到当前分钟为止，合成好的 15 分钟 K 线数据。PTrade 系统会自动处理数据的合成。
Q: 如何更换指标？
A: 你可以把 before_trading_start 里的均线逻辑换成 MACD 或布林带判断趋势；也可以把 handle_data 里的 MACD 换成 KDJ 或 RSI 寻找买卖点。结构保持不变即可。