第四节：自定义异常类——创建业务专属异常，提升代码可读性

说实话，刚入行那会儿，我对异常处理的理解就是「try-except 一把梭」。不管什么错误，统统捕获，然后打印个日志完事。直到有一次，我在实盘跑市商策略，半夜三点被报警电话吵醒——系统崩了，但日志里全是清一色的 Exception，我根本分不清是行情数据断了，还是订单簿更新慢了，还是交易所接口超时了。

嗯，从那天起，我养成了一个习惯：为业务逻辑创建专属的异常类。说白了，就是让异常自己「说话」。

为什么需要自定义异常？

Python 内置的异常类型其实不少，比如 ValueError、KeyError、TimeoutError。但在量化交易这种复杂场景下，这些通用异常根本不够用。你想想看：

  • 行情数据缺失，抛一个 KeyError？
  • 订单簿深度不够，抛一个 ValueError？
  • 交易所限频触发，抛一个 RuntimeError？

这些异常混在一起，你根本没法快速定位问题。更麻烦的是，不同模块的开发者可能用不同的方式处理同一个业务错误，代码维护起来简直是一场噩梦。

核心观点：自定义异常类，本质上是给错误打上「业务标签」。让异常类型本身就能传达足够的信息，而不是靠你去翻日志里的字符串。

如何设计市商策略的异常体系？

我个人习惯，先画一个异常继承树。根节点是 MarketMakingError，继承自 Exception。然后往下分：

class MarketMakingError(Exception):
    """市商策略基础异常"""
    def __init__(self, message, error_code=None):
        self.error_code = error_code
        super().__init__(message)

class DataError(MarketMakingError):
    """数据相关异常"""
    pass

class OrderError(MarketMakingError):
    """订单相关异常"""
    pass

class RiskError(MarketMakingError):
    """风控相关异常"""
    pass

class ExchangeError(MarketMakingError):
    """交易所接口异常"""
    pass

你看，这样一分类，异常的含义就清晰多了。我在项目中遇到过好几次，运维同事半夜排查问题，看到 RiskError 就知道是风控触发了，根本不需要翻代码。

实战：给异常加上「上下文」

光有分类还不够。异常信息里最好带上业务上下文。比如订单簿深度不足，你不仅要告诉系统「深度不够」，还要说清楚是哪个交易对、当前深度是多少、阈值是多少。

class InsufficientDepthError(DataError):
    """订单簿深度不足异常"""
    def __init__(self, symbol, current_depth, required_depth):
        self.symbol = symbol
        self.current_depth = current_depth
        self.required_depth = required_depth
        message = (
            f"交易对 {symbol} 深度不足: "
            f"当前深度 {current_depth:.2f}, "
            f"需要 {required_depth:.2f}"
        )
        super().__init__(message, error_code="DEPTH_001")

这样写的好处是什么？你捕获到这个异常后，可以直接从异常对象里拿到 symbol、current_depth 这些字段，做后续的自动化处理。比如自动切换交易对，或者调整报价宽度。

小技巧：给每个异常类分配一个唯一的 error_code。这样在日志系统里，你可以直接按 error_code 做聚合统计，快速发现高频错误。

避坑指南：别滥用自定义异常

我曾经见过一个项目，自定义了上百个异常类，每个函数都抛不同的异常。结果呢？调用方要写几十个 except 分支，代码比业务逻辑还长。

我的建议是：异常粒度要适中。一般来说，一个模块有 3-5 个自定义异常就够了。比如市商策略，我通常只定义：

异常类 触发场景 示例
DataError 行情数据缺失、延迟、格式错误 K线数据断流超过 5 秒
OrderError 下单失败、撤单超时、订单状态异常 挂单后 10 秒未成交
RiskError 风控阈值触发、仓位超限、资金不足 单边持仓超过总资产 30%
ExchangeError 交易所接口限频、网络超时、返回异常 API 调用返回 429 状态码

你看，四个异常类，覆盖了市商策略 90% 的异常场景。够用，又不冗余。

异常处理的最佳实践

有了自定义异常，接下来就是怎么用。我总结了几条经验：

  1. 在业务边界处捕获：比如在行情数据接入层、订单执行层、风控检查层这些关键节点，把底层异常转换成业务异常。
  2. 保留原始异常链：转换异常时，记得把原始异常传进去，方便调试。
try:
    depth = exchange.get_orderbook(symbol)
except TimeoutError as e:
    raise ExchangeError(
        f"获取 {symbol} 订单簿超时",
        error_code="EXCH_002"
    ) from e  # 保留原始异常
  1. 统一异常处理入口：在策略主循环里，用一个统一的异常处理器来捕获所有自定义异常，做统一的告警和恢复逻辑。
def strategy_loop():
    while True:
        try:
            # 核心策略逻辑
            update_market_data()
            check_risk()
            place_orders()
        except RiskError as e:
            # 风控异常：立即平仓，停止策略
            emergency_shutdown(e)
        except DataError as e:
            # 数据异常：等待恢复，不操作
            wait_and_retry(e)
        except ExchangeError as e:
            # 交易所异常：切换备用接口
            switch_exchange(e)
        except Exception as e:
            # 兜底：记录日志，人工介入
            logger.critical(f"未知异常: {e}", exc_info=True)

警告：千万不要在 except 块里直接 pass！我曾经见过一个线上事故，就是因为异常被静默吞掉，导致策略在错误状态下运行了整整两个小时，亏损惨重。至少也要打一条日志。

总结

自定义异常类，说白了就是给你的代码穿上「防弹衣」。它让错误信息变得有结构、有层次、有业务含义。你想想看，当你的运维同事半夜被叫醒，看到日志里写着 RiskError: 交易对 BTC-USDT 单边仓位超限，他是不是能立刻知道该做什么？

嗯，这就是自定义异常的价值。它不只是代码层面的优化，更是团队协作和系统稳定性的基石。

← 上一章 📖 返回目录 下一章 →