如何在 PTrade 中实现定时执行特定任务（如每日开盘、盘中轮询）？

在PTrade中，实现定时任务（如每天9:30执行）和间隔任务（如每5分钟执行）主要有以下几种方式。

根据您的需求，我为您提供了两种场景的实现方案：

1. 每天固定时间执行：使用 run_daily 函数。
2. 每隔5分钟执行：
   • 通用方式（回测+实盘）：在 handle_data 中判断分钟数。
   • 实盘专用方式：使用 run_interval 函数。

以下是具体的代码实现示例：

完整策略代码示例

def initialize(context):
    """
    初始化函数，设置策略参数和定时任务
    """
    # 设置股票池（示例）
    g.security = '600570.SS'
    set_universe(g.security)
    
    # 1. 每天早上 9:30 执行一次指定函数
    # 注意：run_daily 只能在 initialize 中调用
    run_daily(context, morning_task, time='9:30')
    
    # 2. (实盘专用) 如果是实盘交易，可以使用 run_interval 每隔300秒(5分钟)执行一次
    # 注意：run_interval 仅在交易模块可用，回测中无效
    if is_trade():
        run_interval(context, interval_task_trade, seconds=300)

def morning_task(context):
    """
    每天 9:30 触发的任务
    """
    log.info("【定时任务】现在是9:30，开始执行早盘任务")
    # 在这里编写您的早盘逻辑，例如获取开盘价、判断高开低开等
    # snapshot = get_snapshot(g.security)
    # log.info(snapshot)

def interval_task_trade(context):
    """
    (实盘专用) 每隔5分钟触发的任务
    """
    log.info("【间隔任务-实盘】每隔5分钟检查一次行情")
    # 实盘逻辑...

def handle_data(context, data):
    """
    盘中运行函数
    注意：为了实现分钟级的判断，策略运行频率必须选择【分钟级别】
    """
    # 3. (通用方式) 在 handle_data 中通过取模运算实现每5分钟执行一次
    # 获取当前时间的分钟数
    current_minute = context.blotter.current_dt.minute
    
    # 判断当前分钟是否能被5整除，且不重复执行（handle_data每分钟运行一次）
    # 这里的逻辑是：9:35, 9:40, 9:45... 会触发
    if current_minute % 5 == 0:
        check_market_condition(context)

def check_market_condition(context):
    """
    每5分钟检查行情的具体逻辑
    """
    current_time = context.blotter.current_dt
    log.info("【间隔任务-通用】当前时间: %s，执行5分钟行情检查" % current_time)
    
    # 示例：获取当前价格
    price = data[g.security]['close']
    log.info("当前 %s 价格: %s" % (g.security, price))

关键点说明

1. run_daily(context, func, time='9:30'):

   • 用途: 用于每天固定时间执行任务。
   • 参数: time 格式为 'HH:MM'。
   • 注意: 在回测模式下，run_daily 的时间触发机制可能受限于回测频率，但在实盘中会严格按照时间触发。

2. handle_data 中的 % 5 == 0 判断:

   • 用途: 这是最通用的方法，适用于回测和实盘。
   • 前提: 策略的运行频率（Frequency）必须设置为 “分钟级别”。如果设置为“日线级别”，handle_data 每天只运行一次，无法实现5分钟间隔。
   • 逻辑: context.blotter.current_dt 获取当前策略时间，通过 minute % 5 == 0 来捕捉 5, 10, 15... 分的时间点。

3. run_interval(context, func, seconds=300):

   • 用途: 专门用于实盘交易的定时轮询。
   • 优势: 独立于 K 线生成机制，基于系统时间间隔触发，非常适合高频监控或非整点触发的逻辑。
   • 限制: 回测模式下不可用（回测中调用会被忽略或报错，建议配合 is_trade() 判断使用）。

您可以根据您是处于回测阶段还是实盘阶段，选择最适合您的方式。如果您正在进行回测，建议使用 handle_data 配合取模运算的方式。