Python实盘交易代码因Broker SDK无响应冻结，求解决方案

核心问题分析 #

第三方SDK提供的api.placeOrder函数在Broker无响应时会永久阻塞，既不返回结果也不抛出异常，且无法修改该函数，直接导致交易主循环停滞，甚至无法通过Ctrl+C终止程序。

可行解决方案 #

1. 线程池+超时机制（推荐） #

利用concurrent.futures.ThreadPoolExecutor将API调用放到独立线程中，并设置超时时间，超时则终止任务，避免阻塞主程序。同时通过锁保证全局状态变量的线程安全。

修改后的完整代码：

import time
import api
import threading
from concurrent.futures import ThreadPoolExecutor, TimeoutError

# 全局状态变量及锁，保证多线程下状态同步
is_in_position = 0
state_lock = threading.Lock()

def isExit():
    # 实现你的退出判断逻辑，返回1需要退出，0不需要
    return 0

def safe_place_order(qty, action, timeout=10):
    """带超时的安全下单封装函数"""
    with ThreadPoolExecutor(max_workers=1) as executor:
        future = executor.submit(api.placeOrder, qty=qty, action=action)
        try:
            return future.result(timeout=timeout)
        except TimeoutError:
            print(f"下单请求超时（{action}），Broker未响应")
            return None
        except Exception as e:
            print(f"下单请求异常（{action}）: {str(e)}")
            return None

def createPosition():
    global is_in_position
    result = safe_place_order(qty=100, action='buy', timeout=10)
    with state_lock:
        if result and result.get('status') == 'OK':
            is_in_position = 1
    
def checkPositionsForChanges():
    global is_in_position
    is_exit = isExit()
    if is_exit:
        result = safe_place_order(qty=100, action='sell', timeout=10)
        with state_lock:
            if result and result.get('status') == 'OK':
                is_in_position = 2

# 主循环
while True:
    with state_lock:
        current_state = is_in_position
    if current_state == 0:
        createPosition()
    elif current_state == 1:
        checkPositionsForChanges()
    elif current_state == 2:
        break
    time.sleep(1)

2. 进程隔离（应对线程无法中断的极端情况） #

如果第三方API基于C扩展实现阻塞，线程超时机制可能无法中断它，此时可以用multiprocessing创建子进程执行API调用，超时则强制终止子进程。

核心封装函数示例：

import multiprocessing

def place_order_worker(qty, action, queue):
    """子进程执行下单任务，结果存入队列"""
    try:
        result = api.placeOrder(qty=qty, action=action)
        queue.put(result)
    except Exception as e:
        queue.put(f"Error: {str(e)}")

def safe_place_order_process(qty, action, timeout=10):
    queue = multiprocessing.Queue()
    process = multiprocessing.Process(target=place_order_worker, args=(qty, action, queue))
    process.start()
    process.join(timeout=timeout)
    
    if process.is_alive():
        # 超时后强制终止子进程
        process.terminate()
        process.join()
        print(f"下单进程超时（{action}），已强制终止")
        return None
    else:
        return queue.get() if not queue.empty() else None

将线程版中的safe_place_order替换为该函数即可，主逻辑保持一致。

额外注意事项 #

• 超时时间设置：根据Broker正常响应时间合理设置（如10-30秒），避免误判正常网络延迟。
• 状态同步：全局状态变量必须通过锁（线程）或进程间通信机制（进程）保证安全，防止多线程/进程下的状态混乱。
• 错误处理：超时后可添加重试逻辑，但需结合指数退避策略，避免频繁请求触发Broker限流。
• 日志优化：建议用专业日志库（如logging）替代print，方便后续问题排查。

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