Broker API订单状态轮询优化咨询：每秒轮询是否可行？如何解决订单确认超时导致的状态更新失败问题？

我目前有一个异步函数用于调用Broker API获取订单状态，代码如下：

async def get_order( subs_details: SubscriptionDetails, strike_unique_id: str, order_id: str ) -> tuple[str, float]:
    URL = conn.BALAJI_ORDER_STATUS
    headers = {"Content-Type": "application/json", "Authorization": conn.BALAJI_API_TOKEN}
    order_price = 0.0
    status_code = 400
    order_status = 'NA'
    error = ''
    attempt, max_retries = 0, 3
    try:
        (broker, token, strategy_id, activation_id) = extract_get_order_details(subs_details)
        payload = {"strike_id": strike_unique_id, "brokerName": broker, "token_id": token, "order_id": order_id}
        logger.info(f"Get Order payload: {payload}")
        while attempt < max_retries:
            async with SessionManager.session.post(URL, json=payload, headers=headers) as r:
                status_code = r.status
                res = await r.json()
                logger.info(f"Get order for {order_id} with response: {res}")
                if status_code != 200:
                    if 'invalid order id' in res.get('message', '').lower():
                        await asyncio.sleep(0.5*(attempt+1))
                        attempt += 1
                        continue
                    asyncio.create_task(strategy_status_signal(strategy_id, activation_id, 'ERROR', 'Internal Error'))
                order_log = res.get("order_log", {})
                error = order_log.get("error_reason") or error
                order_price = order_log.get('order_price') or order_price
                order_status = order_log.get("order_status") or order_status
                if order_status == "NA" or f"API rate limit reached {broker}" in error:
                    logger.info(f"While fetching order details error occured: {error}. Retrying...")
                    await asyncio.sleep(0.5*(attempt+1))
                    attempt += 1
                    continue
                if order_status == "Confirmed":
                    logger.info(f"ORDER COMPLETED for {strike_unique_id} broker:{broker} & order_id:{order_id}")
                    break
                elif order_status == "Rejected":
                    asyncio.create_task(strategy_status_signal(strategy_id, activation_id, 'ERROR', error))
                    logger.info(f"ORDER REJECTED for {strike_unique_id} broker:{broker} & order_id:{order_id}")
                    break
            await asyncio.sleep(0.5*(attempt+1))
            attempt += 1
        else:
            logger.info(f"Max retries exceeded while fetching order details for {strike_unique_id}'s order: {order_id}.")
    except Exception as e:
        logger.error(f"Error while fetching order details: {e}")
    return order_status, order_price, error

当前问题 #

部分订单能立即成交并返回状态，但有不少订单的确认耗时长达15-20秒，甚至更久。现有逻辑设置了max_retries=3，当订单确认耗时超过重试覆盖的时间窗口时，会触发重试上限，导致无法及时向用户更新最终的订单状态。

我已考虑的方案及疑问 #

我梳理了几种潜在解决方案，但都存在顾虑：

• 固定超时替代最大重试：设置90秒固定超时后终止轮询，但超时后依然无法向用户推送后续的状态更新
• 指数退避重试（如e^(attempt/3)策略）：前几次轮询间隔短，后续逐渐延长，但如果订单状态在第3次尝试后立即更新，用户会延迟约45秒才能收到通知
• 恒定轮询：担心频繁的轮询会造成日志污染，还可能引发API限流等其他问题
• 混合超时与恒定轮询：先每秒轮询直至超时，再将订单加入队列，由Worker采用指数退避策略轮询处理

另外，目前暂无法使用Webhooks（Broker暂不支持），但有可能通过Websocket订阅订单ID，在收到确认或拒绝通知时处理。

优化建议 #

1. 优先尝试Websocket订阅方案 #

如果Broker支持通过Websocket订阅指定订单ID的状态变更，这绝对是最优解。不需要主动轮询，而是被动接收状态推送，能第一时间获取成交/拒绝通知，完全规避轮询带来的所有问题。

实现思路：

• 维护一个全局的Websocket连接池，复用连接避免频繁建立/断开
• 当调用get_order时，若首次请求未获得明确状态（Confirmed/Rejected），就订阅该订单ID的Websocket通道
• 收到状态更新推送后，立即触发strategy_status_signal通知用户，更新本地订单记录，并取消该订单的订阅

2. 改进轮询策略：阶梯式间隔+总超时窗口 #

如果暂时无法使用Websocket，可以调整现有轮询逻辑，兼顾即时性和资源消耗：

• 设置总超时窗口（比如120秒），而非固定重试次数，确保覆盖绝大多数延迟成交的场景
• 采用阶梯式间隔轮询：前5次用1秒间隔（快速确认即时成交的订单），接下来10次用2秒间隔，之后改用5秒间隔，直到触发总超时。这样既能快速响应即时订单，又不会在延迟订单的情况下频繁轮询造成资源浪费和日志污染
• 优化终止条件：除了总超时，若收到明确状态（Confirmed/Rejected）或"无效订单ID"的错误，立即终止轮询

调整后的核心逻辑示例：

total_timeout = 120  # 最长轮询120秒
start_time = asyncio.get_event_loop().time()
# 定义阶梯间隔规则：(该间隔下的重试次数, 间隔秒数)
interval_rules = [(5, 1), (10, 2), (float('inf'), 5)]
current_rule_idx = 0
count_in_rule = 0

while asyncio.get_event_loop().time() - start_time < total_timeout:
    # 执行API请求及状态解析逻辑（和原有代码一致）
    async with SessionManager.session.post(URL, json=payload, headers=headers) as r:
        status_code = r.status
        res = await r.json()
        # ... 省略状态解析、错误处理逻辑 ...
        
        # 若拿到明确状态或无效订单ID，直接跳出循环
        if order_status in ("Confirmed", "Rejected") or 'invalid order id' in res.get('message', '').lower():
            break
    
    # 计算下一次轮询的间隔
    max_attempts, interval = interval_rules[current_rule_idx]
    count_in_rule += 1
    if count_in_rule >= max_attempts:
        current_rule_idx = min(current_rule_idx + 1, len(interval_rules)-1)
        count_in_rule = 0
    await asyncio.sleep(interval)

3. 异步队列+后台Worker的解耦方案 #

如果延迟成交的订单占比很高，且希望不阻塞主业务流程，可以将未确认的订单剥离到异步队列中，由后台Worker单独处理：

• 主函数get_order：先进行3次快速轮询（间隔1秒），如果仍未获得明确状态，就把订单信息（subs_details、strike_unique_id、order_id）加入持久化队列（比如Redis Queue），返回"Pending"状态给用户
• 后台Worker：从队列中取出订单，采用指数退避或阶梯间隔的方式持续轮询Broker API，直到获取最终状态或达到最终超时（比如5分钟），然后调用strategy_status_signal通知用户
• 优势：主流程不会被长时轮询阻塞，Worker可以集中处理延迟订单，日志也能单独隔离，避免污染主业务日志
• 注意点：确保队列的持久化特性，避免订单丢失；给每个订单添加处理锁，防止重复消费

4. 日志优化细节 #

如果担心轮询造成日志污染，可以调整日志输出策略：

• 仅在首次重试、状态变更、最终超时/失败时记录INFO级别的日志
• 中间的轮询请求记录DEBUG级别的日志，生产环境可以关闭DEBUG日志，只保留关键节点的信息

总结 #

优先尝试Websocket推送方案，这是最优雅、高效的解决方案；如果暂时无法实现，阶梯式轮询+总超时的组合能很好地平衡即时性和资源消耗；如果延迟订单场景较多，异步队列+Worker的方案能有效解耦主流程，提升系统整体的响应性。

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