如何在Python的Interactive Brokers API脚本中捕获KeyboardInterrupt并优雅断连

针对你遇到的KeyboardInterrupt无法捕获、信号处理函数无法传递app实例的问题，这里有几个实用的解决方案：

1. 用全局变量持有app实例（简单直接） #

把IB API的实例设为全局变量，信号处理函数直接调用全局变量的disconnect()方法，无需额外传参：

from ibapi.client import EClient
from ibapi.wrapper import EWrapper
import signal

class IBApi(EWrapper, EClient):
    def __init__(self):
        EClient.__init__(self, self)

# 全局存储app实例
app = None

def handle_interrupt(signum, frame):
    if app:
        print("正在断开IB连接...")
        app.disconnect()
        exit(0)

def main():
    global app
    app = IBApi()
    app.connect("127.0.0.1", 7497, clientId=1)
    
    # 注册SIGINT信号（对应Ctrl+C）的处理函数
    signal.signal(signal.SIGINT, handle_interrupt)
    
    # 启动IB消息循环
    app.run()

if __name__ == "__main__":
    main()

2. 类内封装信号处理（更优雅，避免全局变量） #

把信号注册和断连逻辑封装到API类内部，直接用实例方法处理信号：

from ibapi.client import EClient
from ibapi.wrapper import EWrapper
import signal
import threading

class IBApi(EWrapper, EClient):
    def __init__(self):
        EClient.__init__(self, self)
        # 在初始化时注册信号处理
        signal.signal(signal.SIGINT, self._handle_interrupt)
    
    def _handle_interrupt(self, signum, frame):
        print("收到中断信号，开始优雅断连")
        self.disconnect()
        # 如果用线程运行消息循环，等待线程结束
        if hasattr(self, "run_thread"):
            self.run_thread.join(timeout=2)
        exit(0)

def main():
    app = IBApi()
    app.connect("127.0.0.1", 7497, clientId=1)
    
    # 将app.run()放到线程中，避免主线程被阻塞
    app.run_thread = threading.Thread(target=app.run, daemon=True)
    app.run_thread.start()
    
    # 主线程保持活跃，等待中断信号
    app.run_thread.join()

if __name__ == "__main__":
    main()

3. 用functools.partial绑定实例到信号处理函数 #

如果不想用全局变量或类内处理，可通过functools.partial把app实例绑定到信号处理函数上：

from ibapi.client import EClient
from ibapi.wrapper import EWrapper
import signal
import functools

class IBApi(EWrapper, EClient):
    def __init__(self):
        EClient.__init__(self, self)

def handle_interrupt(app, signum, frame):
    print("执行优雅断连")
    app.disconnect()
    exit(0)

def main():
    app = IBApi()
    app.connect("127.0.0.1", 7497, clientId=1)
    
    # 绑定app实例到处理函数
    bound_handler = functools.partial(handle_interrupt, app)
    signal.signal(signal.SIGINT, bound_handler)
    
    app.run()

if __name__ == "__main__":
    main()

补充说明 #

你之前无法捕获KeyboardInterrupt的核心原因是app.run()是阻塞式方法，它会占用主线程持续处理IB的消息循环，导致Python解释器无法及时响应中断信号。将app.run()放到独立线程中运行，主线程就能正常接收并处理中断信号了。

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