Windows环境Jupyter Lab下如何实现Python模型拟合超时终止？

核心思路是通过ctypes向运行目标函数的线程主动抛出异常，既避开Windows不支持signal的限制，也不需要多进程，完美适配Jupyter交互式环境，可解决你之前线程方案的两个失效问题：既能真正终止后台运行的目标线程，也可以捕获超时异常执行自定义逻辑。

核心实现代码 #

仅依赖Python标准库，无需引入第三方依赖：

import threading
import ctypes
import time

def _async_raise(tid, exctype):
    """向指定线程ID抛出异常，核心终止逻辑"""
    tid = ctypes.c_long(tid)
    if not ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exctype)):
        raise ValueError("无效的线程ID")
    elif ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None) > 1:
        # 出现异常时重置状态
        ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None)
        raise SystemError("抛出异常失败")

class TimeoutException(Exception):
    """自定义超时异常，方便捕获处理"""
    pass

def run_with_timeout(func, timeout, *args, **kwargs):
    """
    带超时的函数执行逻辑
    func: 要执行的目标函数
    timeout: 超时时间（秒）
    args/kwargs: 目标函数的入参
    超时抛出TimeoutException，正常执行返回函数结果
    """
    res = [None]
    exc = [None]

    def wrapper():
        try:
            res[0] = func(*args, **kwargs)
        except Exception as e:
            exc[0] = e

    t = threading.Thread(target=wrapper)
    t.start()
    t.join(timeout)
    if t.is_alive():
        _async_raise(t.ident, TimeoutException)
        t.join()
        raise TimeoutException(f"函数执行超过{timeout}秒，已终止")
    if exc[0] is not None:
        raise exc[0]
    return res[0]

适配模型拟合场景 #

直接调用上述工具函数实现你需要的fit_with_timeout逻辑：

def fit_with_timeout(mod, MAX_TIMEOUT):
    """
    调用mod.fit()方法
    若在MAX_TIMEOUT时限内完成拟合则返回拟合结果，超时则返回None
    """
    try:
        return run_with_timeout(mod.fit, MAX_TIMEOUT)
    except TimeoutException:
        # 此处可添加自定义超时处理逻辑
        print("模型拟合超时，已终止执行")
        return None

测试验证 #

可以用你之前的测试函数验证效果：

# 耗时测试函数
def f(x):
    print(x, 'start')
    time.sleep(x)
    with open(str(x) + '.txt', 'w') as file:
        file.write(str(x))
    return x

try:
    # 设置1秒超时，目标函数执行需要5秒
    res = run_with_timeout(f, 1, 5)
except TimeoutException:
    print("触发自定义超时逻辑")
# 运行后1秒即触发超时，不会生成5.txt文件，证明线程已被成功终止

注意事项 #

• 该方案仅能终止Python层面的代码执行，如果目标函数调用了长时间阻塞的无GIL C扩展，异常抛出会等C代码返回后才生效，statsmodels的fit方法大部分逻辑是Python层面的，可以正常终止
• 完全兼容Windows系统和Jupyter Lab交互式环境，不受if __name__ == "__main__"限制

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