PTrade 策略实战：如何通过异常处理避免因行情或网络问题导致策略中断？

在PTrade量化交易平台中，处理网络波动或数据缺失是保证策略实盘稳定运行的关键。如果不对API调用进行保护，一旦遇到网络超时或数据源异常，策略进程可能会直接抛出异常并终止运行。

以下是处理此类情况的几种核心方法及完整的代码示例。

核心处理机制

1. 异常捕获 (try...except)：
   这是防止程序崩溃的最直接手段。将涉及网络请求或数据获取的代码块包裹在 try 语句中。如果发生错误，程序会跳转到 except 块执行错误处理逻辑（如记录日志），而不是直接退出。

2. 返回值有效性检查 (None / 空值判断)：
   PTrade的许多API（如 get_snapshot, get_history）在获取失败时不会报错，而是返回 None 或空字典/空列表。必须在使用数据前判断其是否为空。

3. 重试机制 (Retry Logic)：
   对于 get_fundamentals 等容易因网络拥堵或流控失败的接口，建议引入重试机制。如果第一次失败，休眠几秒后再次尝试。

4. 数据清洗 (NaN 处理)：
   获取到的历史数据可能包含 NaN（空值），直接参与计算会导致指标错误。需要使用 fillna 或 dropna 进行处理。

健壮的策略代码示例

以下代码展示了如何封装一个带有重试机制的数据获取函数，并在 handle_data 中安全地使用它。

import time
import numpy as np
import pandas as pd

def initialize(context):
    # 设置股票池
    g.security = ['600570.SS', '000001.SZ']
    set_universe(g.security)
    
    # 设置重试次数和重试间隔
    g.retry_count = 3
    g.retry_sleep = 1.0

def get_fundamentals_safe(security, table, fields, max_retries=3, sleep_time=1.0):
    """
    带重试机制的财务数据获取函数
    """
    for i in range(max_retries):
        try:
            # 尝试获取数据
            df = get_fundamentals(security, table, fields)
            
            # 检查返回的数据是否有效
            if df is not None and not df.empty:
                return df
            else:
                log.warning("第 %s 次获取财务数据为空，准备重试..." % (i + 1))
        except Exception as e:
            log.error("第 %s 次获取财务数据发生异常: %s" % (i + 1, e))
        
        # 如果不是最后一次尝试，则休眠等待
        if i < max_retries - 1:
            time.sleep(sleep_time)
            
    log.error("获取财务数据失败，已达到最大重试次数。")
    return None

def handle_data(context, data):
    """
    主交易逻辑，包含异常处理和数据检查
    """
    for stock in g.security:
        # 1. 异常捕获：防止单个股票处理出错导致整个循环中断
        try:
            # --- 获取行情快照 (Snapshot) ---
            # get_snapshot 在回测中不可用，这里仅作演示，实盘中需要
            # 在回测中通常使用 data[stock] 获取当前数据
            
            # 模拟数据获取检查
            current_data = data[stock]
            
            # 2. 数据有效性检查：检查是否停牌或数据缺失
            if np.isnan(current_data['close']) or current_data['volume'] == 0:
                log.info("股票 %s 数据缺失或停牌，跳过本轮处理" % stock)
                continue

            # --- 获取历史数据 (History) ---
            # 使用 try-except 包裹 API 调用
            try:
                # 获取过去5天的收盘价
                h_data = get_history(5, '1d', 'close', stock, include=True)
            except Exception as e:
                log.error("获取 %s 历史数据时发生网络异常: %s" % (stock, e))
                continue # 跳过该股票

            # 检查历史数据是否为空
            if h_data is None or len(h_data) < 5:
                log.warning("股票 %s 历史数据不足，跳过" % stock)
                continue
            
            # 3. 数据清洗：处理 NaN 值
            # 如果历史数据中有 NaN，用前值填充
            close_prices = h_data['close'].fillna(method='ffill')
            
            # 再次检查填充后是否仍有 NaN (例如开头就是 NaN)
            if close_prices.isnull().values.any():
                log.warning("股票 %s 历史数据包含无法填充的空值，跳过" % stock)
                continue

            # --- 获取财务数据 (使用自定义的重试函数) ---
            # 假设我们需要获取市盈率
            q = get_fundamentals_safe(stock, 'valuation', 'pe_ttm', max_retries=g.retry_count)
            
            if q is None:
                log.warning("无法获取 %s 的财务数据，使用默认策略或跳过" % stock)
                pe_ttm = 0 # 或者 continue
            else:
                # 注意：get_fundamentals 返回的是 DataFrame，索引是股票代码
                if stock in q.index:
                    pe_ttm = q['pe_ttm'][stock]
                else:
                    pe_ttm = 0

            # --- 策略逻辑 ---
            current_price = current_data['close']
            avg_price = close_prices.mean()

            # 简单的均线策略示例
            if current_price > avg_price and pe_ttm > 0 and pe_ttm < 50:
                # 下单前也可以做一次 try-except，防止柜台连接异常
                try:
                    order(stock, 100)
                    log.info("买入 %s, 价格: %s" % (stock, current_price))
                except Exception as order_e:
                    log.error("下单失败 %s: %s" % (stock, order_e))

        except Exception as e:
            # 捕获该股票处理过程中的所有未预料异常，防止策略崩溃
            log.error("处理股票 %s 时发生未知错误: %s" % (stock, e))
            # 继续处理下一个股票
            continue

代码关键点解析

1. get_fundamentals_safe 函数：

   • 这是一个自定义函数，封装了 get_fundamentals。
   • 它使用 for 循环实现了重试机制。
   • 它同时捕获了 Exception（网络错误）并检查了返回值是否为 None（数据缺失）。
   • 使用了 time.sleep() 来缓解网络拥堵或流控限制。

2. handle_data 中的 try...except：

   • 最外层的 try...except 放在 for stock in g.security 循环内部。这意味着如果处理 600570.SS 时出错，程序会记录错误并继续处理 000001.SZ，而不会直接停止整个策略。

3. 数据校验流程：

   • 检查 None：API 调用后立即检查。
   • 检查长度：确保获取到的 K 线数量满足计算指标的需求（如计算 5 日均线必须有 5 个数据）。
   • 检查 NaN：使用 pandas 的 fillna 处理空缺数据，并二次检查。

4. 日志记录 (log)：

   • 使用 log.error 记录严重错误，方便盘后排查。
   • 使用 log.warning 记录数据缺失等非致命问题。

通过这种分层防御（重试 -> 校验 -> 异常捕获），策略可以在网络不稳定或部分数据源缺失的情况下保持最大程度的可用性。

Q&A: PTrade 策略稳定性常见问题

Q: 为什么 get_snapshot 有时候返回的数据里全是 0？
A: 在盘前阶段（如 9:15 之前）或刚开盘瞬间，交易所可能尚未推送完整的快照数据。此外，如果股票停牌，成交量等数据也会是 0。在使用快照数据前，务必检查 last_px（最新价）是否大于 0，或检查 trade_status 状态。

Q: get_history 获取的数据包含停牌期间的数据吗？
A: PTrade 的 get_history 默认不跳过停牌日期，停牌期间会使用停牌前的数据填充（价格不变，成交量为 0）。如果需要计算技术指标，建议自行过滤成交量为 0 的数据，或者在计算均线时考虑到这一点。

Q: 为什么 get_fundamentals 容易失败？
A: 该接口是 HTTP 在线获取，且有严格的流控限制（每秒不超过 100 次，单次不超过 500 条）。如果在循环中高频调用且不加 sleep，极易触发流控导致返回失败。建议尽量批量查询（传入股票列表），并配合重试机制。

Q: 如何区分是网络错误还是代码逻辑错误？
A: 通过 try...except Exception as e 捕获异常后，打印 e 的内容。如果是 Timeout、ConnectionError 等通常是网络问题；如果是 KeyError、ValueError 等通常是代码逻辑或数据格式处理问题。