PTrade 策略教程：如何结合日线、60分钟线和5分钟线进行多时间周期分析？

这是一个基于PTrade平台的多时间周期交易策略（Multi-Timeframe Strategy）。

策略逻辑概述

本策略采用了经典的**“日线定趋势，分钟线找买点”**的共振思路。

1. 日线级别（趋势判断）：
   • 使用 20日均线 (MA20) 判断长期趋势。
   • 如果昨日收盘价 > 昨日MA20，判定为上涨趋势，允许开仓。
   • 如果昨日收盘价 < 昨日MA20，判定为下跌趋势，禁止开仓并清仓。
2. 分钟级别（择时交易）：
   • 在日线为上涨趋势的前提下，在盘中（handle_data）获取 15分钟均线 (MA15)（通过获取过去15根1分钟K线计算）。
   • 买入信号：当前价格突破分钟MA15，且当前无持仓。
   • 卖出信号：当前价格跌破分钟MA15，或日线趋势转坏。

策略代码实现

import numpy as np

def initialize(context):
    """
    初始化函数，设置股票池、参数和回测环境
    """
    # 1. 设置要操作的股票池 (示例：恒生电子)
    g.security = ['600570.SS']
    set_universe(g.security)
    
    # 2. 设置策略参数
    g.daily_ma_window = 20   # 日线均线周期
    g.minute_ma_window = 15  # 分钟线均线周期
    
    # 3. 设置手续费 (股票：万分之三，最低5元)
    set_commission(commission_ratio=0.0003, min_commission=5.0, type='STOCK')
    # 设置滑点 (0.1%)
    set_slippage(slippage=0.001)
    
    # 全局变量，用于存储每只股票的日线趋势状态
    g.trend_status = {} 

def before_trading_start(context, data):
    """
    盘前处理函数：每天开盘前运行一次，用于计算日线级别的趋势
    """
    log.info("=== 盘前趋势检查 ===")
    
    for stock in g.security:
        # 获取过去 N+1 天的日线数据 (为了计算截止到昨天的均线)
        # count取值要足够计算均线，这里取 daily_ma_window + 5 防止停牌等数据缺失
        daily_data = get_history(g.daily_ma_window + 5, frequency='1d', field='close', security_list=stock, fq='pre', include=False)
        
        # 检查数据长度是否足够
        if len(daily_data) < g.daily_ma_window:
            log.warning("股票 %s 数据不足，跳过日线分析" % stock)
            g.trend_status[stock] = False
            continue
            
        # 截取最近的 N 天数据计算均线
        recent_closes = daily_data['close'].values[-g.daily_ma_window:]
        
        # 计算日线 MA20
        daily_ma = np.mean(recent_closes)
        # 获取昨日收盘价
        last_close = recent_closes[-1]
        
        # 判断趋势：昨日收盘价 > 日线均线
        if last_close > daily_ma:
            g.trend_status[stock] = True
            log.info("股票 %s 日线趋势向上 (收盘价: %.2f > MA%d: %.2f)" % (stock, last_close, g.daily_ma_window, daily_ma))
        else:
            g.trend_status[stock] = False
            log.info("股票 %s 日线趋势向下 (收盘价: %.2f < MA%d: %.2f)" % (stock, last_close, g.daily_ma_window, daily_ma))

def handle_data(context, data):
    """
    盘中处理函数：分钟回测模式下，每分钟运行一次
    """
    # 获取当前时间，避免在非交易时间段误操作
    current_time = context.blotter.current_dt.strftime("%H:%M")
    
    # 遍历股票池
    for stock in g.security:
        # 1. 检查日线趋势
        # 如果日线趋势向下，且有持仓，则清仓
        if not g.trend_status.get(stock, False):
            if get_position(stock).amount > 0:
                log.info("日线趋势转弱，清仓股票: %s" % stock)
                order_target(stock, 0)
            continue # 跳过后续买入逻辑
            
        # 2. 获取分钟级别数据
        # 获取过去 M 分钟的收盘价数据
        minute_data = get_history(g.minute_ma_window, frequency='1m', field='close', security_list=stock, fq='pre', include=True)
        
        if len(minute_data) < g.minute_ma_window:
            continue
            
        # 计算分钟均线
        minute_closes = minute_data['close'].values
        minute_ma = np.mean(minute_closes)
        
        # 获取当前最新价格
        current_price = data[stock]['close']
        
        # 获取当前持仓
        position = get_position(stock)
        
        # 3. 交易逻辑
        # 买入条件：日线向上 + 价格高于分钟均线 + 空仓
        if current_price > minute_ma and position.amount == 0:
            # 全仓买入 (实际交易中建议控制仓位)
            cash = context.portfolio.cash
            if cash > 0:
                order_value(stock, cash)
                log.info("分钟级别突破，买入 %s. 现价: %.2f, 分钟MA: %.2f" % (stock, current_price, minute_ma))
        
        # 卖出条件：价格低于分钟均线 + 有持仓 (止盈/止损)
        elif current_price < minute_ma and position.amount > 0:
            order_target(stock, 0)
            log.info("分钟级别跌破，卖出 %s. 现价: %.2f, 分钟MA: %.2f" % (stock, current_price, minute_ma))

def after_trading_end(context, data):
    """
    盘后处理函数
    """
    log.info("今日交易结束，结算资产: %.2f" % context.portfolio.portfolio_value)

代码详细解析

1. initialize (初始化):

   • 我们定义了两个关键参数：g.daily_ma_window (20天) 和 g.minute_ma_window (15分钟)。
   • g.trend_status 字典用于在 before_trading_start 和 handle_data 之间传递日线趋势的判断结果。

2. before_trading_start (日线分析):

   • API使用: get_history(..., frequency='1d', ...)。
   • 逻辑: 每天开盘前（如09:00左右），系统获取过去的历史日线数据。我们计算过去20天的均线。
   • 目的: 确定当天的“大方向”。如果大方向是跌势，当天盘中无论分钟线怎么走，我们都不开新仓，甚至会清仓。这是一种风险控制手段。

3. handle_data (分钟线分析):

   • 运行频率: 此策略需要在回测设置中选择**“分钟线”**频率，这样 handle_data 才会每分钟执行一次。
   • API使用: get_history(..., frequency='1m', ...)。这里我们获取过去15根1分钟K线来模拟一个短周期的均线。
   • 逻辑:
     • 首先检查 g.trend_status。如果是 False (下跌趋势)，直接跳过买入逻辑，并检查是否需要清仓。
     • 如果趋势为 True，则计算当前的分钟均线。
     • 金叉买入: 现价 > 分钟均线。
     • 死叉卖出: 现价 < 分钟均线。

注意事项

1. 回测频率设置: 在PTrade客户端运行回测时，务必将回测频率设置为 “分钟”。如果设置为“日线”，handle_data 每天只会运行一次，分钟级别的逻辑将失效。
2. Python版本: 代码使用了 %s 和 % 进行字符串格式化，完全兼容 Python 3.5 及以上版本（PTrade通常环境）。
3. 数据获取: get_history 返回的是 DataFrame (或 Series)，通过 .values 转换为 numpy 数组进行 mean() 计算效率较高。
4. 未来函数规避: 在 before_trading_start 中获取日线数据时，使用了 include=False，确保只使用昨天及之前的数据，不包含今天尚未走完的数据。

Q&A: PTrade多周期策略常见问题

Q1: 为什么要在 before_trading_start 里计算日线，而不是在 handle_data 里？

A: 效率和逻辑清晰度。日线指标（如20日均线）在一天之内是不会变的（基于昨日收盘价）。在 before_trading_start 计算一次即可，如果在 handle_data 里每分钟都去取日线数据计算，会浪费计算资源且没有必要。

Q2: 如何修改分钟线的周期，比如看30分钟线？

A: 你可以修改 g.minute_ma_window = 30，这样就是取过去30个1分钟K线的均值。或者，你可以使用 get_history(..., frequency='30m', ...) 直接获取30分钟频率的K线数据，但要注意 handle_data 依然是每分钟触发，你需要判断当前时间是否是30分钟的节点（如 10:00, 10:30），否则可能会造成信号闪烁。本策略采用“最近N分钟均线”的方法更加平滑且易于实现。

Q3: 策略中的 set_universe 是必须的吗？

A: 在回测中是必须的，它告诉系统需要准备哪些股票的数据。在实盘交易中，通常不需要 set_universe，而是直接对你账户里的持仓或你指定的代码列表进行操作，但在 initialize 中设置好全局变量 g.security 是个好习惯。