提升 PTrade 策略稳定性：如何在代码中应用 try-except 进行异常处理？

在PTrade量化交易平台中，异常处理（Exception Handling）是保证策略在实盘或模拟盘中长期稳定运行的关键。由于网络波动、数据缺失或逻辑边界条件（如除以零）等原因，策略可能会遇到意料之外的错误。如果不进行处理，这些错误会导致策略直接停止运行（Crash），从而错过交易机会或无法及时止损。

以下是在PTrade策略中实施异常处理的具体方法和最佳实践。

核心机制：try...except 结构

Python 提供了 try...except 语句块来捕获和处理异常。在PTrade中，我们通常结合 log 函数将错误信息记录下来，同时保证策略主循环（如 handle_data）不中断。

基本语法结构

def handle_data(context, data):
    try:
        # 这里放置主要的策略逻辑
        # 例如：获取数据、计算指标、下单交易
        pass
        
    except Exception as e:
        # 当 try 块中的代码报错时，会执行这里的代码
        # 使用 log.error 记录错误堆栈或提示信息
        log.error("策略运行出现异常: %s" % str(e))
        
        # 可选：执行一些清理工作或风控操作

常见场景与代码示例

以下是三个最常见的需要异常处理的场景：

1. 数据获取异常处理

行情接口（如 get_history, get_snapshot）在某些极端情况下可能返回 None 或空数据。直接对空数据进行索引或计算会导致 TypeError 或 IndexError。

def handle_data(context, data):
    try:
        security = g.security
        # 获取历史数据
        history = get_history(10, '1d', 'close', security)
        
        # 检查数据是否获取成功
        if history is None or len(history) < 10:
            log.warning("数据获取不足或为空，跳过本次计算")
            return
            
        # 计算均线
        ma5 = history['close'][-5:].mean()
        
        # 正常的交易逻辑...
        
    except Exception as e:
        # 捕获所有未预料到的错误
        log.error("handle_data 数据处理部分发生错误: %s" % str(e))

2. 交易下单异常处理

下单函数（如 order, order_target）虽然内部有一定容错，但在计算下单数量时（例如资金分配）容易出现数学错误。

def place_order_safe(security, cash_available):
    try:
        current_price = get_snapshot(security)[security]['last_px']
        
        # 避免价格为0导致除零错误（虽然罕见，但可能发生在停牌或数据异常时）
        if current_price <= 0:
            log.info("当前价格异常(<=0)，放弃下单")
            return

        # 计算买入数量，向下取整到100股
        amount = int(cash_available / current_price / 100) * 100
        
        if amount > 0:
            order(security, amount)
            log.info("成功下单: %s, 数量: %s" % (security, amount))
        else:
            log.info("资金不足以买入一手")
            
    except ZeroDivisionError:
        log.error("发生除零错误，可能是价格为0")
    except Exception as e:
        log.error("下单逻辑出错: %s" % str(e))

3. 完整的策略框架示例

这是一个集成了异常处理的完整策略结构，建议在实盘中使用类似的结构。

def initialize(context):
    g.security = '600570.SS'
    set_universe(g.security)

def handle_data(context, data):
    # 使用 try-except 包裹整个 handle_data 逻辑
    # 这样即使某一次循环出错，也不会导致整个策略进程终止，下个周期仍会继续运行
    try:
        security = g.security
        
        # 1. 获取数据
        close_data = get_history(5, '1d', 'close', security)
        
        # 保护性判断：防止数据获取失败
        if close_data is None or len(close_data) == 0:
            log.warning("未获取到历史数据")
            return

        # 2. 计算逻辑
        current_price = data[security]['close']
        avg_price = close_data['close'].mean()
        
        # 3. 交易逻辑
        if current_price > avg_price:
            # 尝试下单
            _safe_order(security, 100)
            
    except Exception as e:
        # 记录详细的错误信息，方便盘后排查
        log.error("handle_data 主循环发生未捕获异常: %s" % str(e))
        # 可以在这里添加发送邮件通知的代码（如果券商支持）

def _safe_order(security, amount):
    """
    封装一个安全的下单函数
    """
    try:
        order(security, amount)
    except Exception as e:
        log.error("下单失败 [%s]: %s" % (security, str(e)))

最佳实践建议

1. 不要过度使用空的 except:：
   尽量使用 except Exception as e: 并打印 e，这样你才能知道具体出了什么错。如果只写 except: 且不打印日志，会导致错误被“吞掉”，难以调试。

2. 区分致命错误与非致命错误：

   • 非致命错误（如某分钟数据缺失）：记录日志，return 跳过本次循环即可。
   • 致命错误（如逻辑严重缺陷）：在 except 块中可以考虑发送通知（如 set_email_info 配置邮件报警），甚至在极端情况下手动停止策略。

3. 关键变量检查：
   在进行除法运算前，务必检查分母是否为 0；在调用对象属性前，检查对象是否为 None。

4. 利用 finally 块（可选）：
   如果你有一些必须执行的代码（例如无论是否报错都要记录一条“运行结束”的日志），可以使用 finally。

   try:
       # 逻辑
       pass
   except Exception as e:
       log.error(str(e))
   finally:
       # 无论是否出错都会执行
       pass 
   

Q&A

Q: 为什么我的策略在回测没问题，实盘却报错停止了？
A: 回测环境的数据通常是清洗过的，比较完整。实盘环境更复杂，可能会遇到网络抖动导致数据返回 None、交易所临时停牌、或者价格瞬间异常等情况。因此，实盘策略必须加入健壮的异常处理机制。

Q: log.error 会停止策略运行吗？
A: 不会。log.error 只是将一条日志标记为“错误”级别并打印出来，方便你在日志中筛选查看。真正防止策略停止的是 try...except 结构。

Q: 如何在报错时收到通知？
A: PTrade 提供了 set_email_info 接口（仅交易模块可用）。你可以在 except 块中结合逻辑判断，如果错误严重，可以触发邮件发送逻辑（需券商环境支持）。