使用ib_async的Interactive Broker API：回调函数错误捕获疑问

问题 #

我正在使用Python及ib_async库开发基于Interactive Broker API的交易机器人，目前已实现连接、获取数据、下单功能，但遇到回调函数的错误处理问题。我的下单代码如下：

import ib_async as ib_api

def order_filled(trade, fill):
    """
    Called when order has been placed
    """

    logging.config.fileConfig(Path(cwd, "logging.ini"), disable_existing_loggers=False)    # commenting this line raised the error
    logger_ibkr = logging.getLogger("IBKR")
    
    for i in range(10):
       print(i)    # it prints 0
       print(notdefined)    # error is not raised
   
ib = ib_api.IB()
ib.connect("127.0.0.1", port=4002)

contract = ib_api.Contract(secType="FUT",
                           symbol="MNQ",
                           currency="USD",
                           exchange="CME",
                           localSymbol="MNQM5"
                           )

order = ib_api.MarketOrder("BUY", 1)
trade = ib.placeOrder(contract, order)
trade.fillEvent += order_filled
ib.run()

我在回调函数中故意使用了未定义的变量notdefined，但代码运行时并未捕获到该错误。我怀疑这与asyncio有关，但未找到合适的解答，希望得到合理的解释。另外，是否存在不使用try/except块就能捕获此类错误的方法？

解答 #

错误未被捕获的原因 #

ib_async底层基于asyncio，回调函数（比如fillEvent绑定的order_filled）是在asyncio事件循环中被调用的。asyncio默认会静默处理回调中的未处理异常，不会主动抛出——这是因为事件循环需要持续运行，单个回调的异常不能直接中断整个程序，因此默认策略是吞掉异常。

你提到注释掉logging.config.fileConfig后错误会触发，这是因为日志配置操作可能修改了Python全局异常处理的上下文，属于偶然触发的情况，并非可靠的错误捕获方式。

不直接在回调内写try/except的捕获方案 #

1. 配置asyncio全局异常处理

在调用ib.run()之前，给事件循环设置全局异常处理函数，统一捕获所有未处理的回调异常：

import asyncio

def global_exception_handler(loop, context):
    exception = context.get("exception")
    if exception:
        print(f"回调未处理异常: {type(exception).__name__}: {str(exception)}")
        # 这里可以添加日志记录、告警推送等逻辑
    else:
        print(f"回调异常信息: {context.get('message')}")

loop = asyncio.get_event_loop()
loop.set_exception_handler(global_exception_handler)

2. 用装饰器批量处理回调异常

自定义一个装饰器，给所有需要的回调函数统一添加异常捕获逻辑，避免在每个回调内部重复写try/except：

def callback_error_catcher(func):
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            print(f"回调函数 {func.__name__} 执行出错: {str(e)}")
            # 可扩展日志上报、异常统计等操作
    return wrapper

# 给目标回调函数添加装饰器
@callback_error_catcher
def order_filled(trade, fill):
    # 原函数逻辑
    logging.config.fileConfig(Path(cwd, "logging.ini"), disable_existing_loggers=False)
    logger_ibkr = logging.getLogger("IBKR")
    
    for i in range(10):
       print(i)
       print(notdefined)

内容的提问来源于stack exchange，提问作者beni