如何在Python的TradingView-IB桥接程序中检测阻止重复订单

一、核心问题分析 #

你遇到的本质问题是：webhook函数每次处理请求时，局部变量会被重新创建，无法保留上一次的订单信息；同时直接比较MarketOrder对象可能因未实现相等判断逻辑而失效。

二、简洁的解决方案（按优先级推荐） #

方案1：使用应用状态存储订单信息（推荐，适配异步Web框架） #

这个方法比全局变量更规范，适配Starlette/FastAPI这类异步框架，还能避免全局变量的并发风险。

步骤1：初始化应用状态

在创建app实例后，添加需要的状态变量：

from starlette.applications import Starlette
import asyncio

app = Starlette()

# 初始化状态：存储上一次的订单关键信息和并发锁
app.state.last_order_info = {"ticker": "", "action": "", "contracts": 0}
app.state.order_lock = asyncio.Lock()

步骤2：修改webhook函数

@app.route('/webhook', methods=['GET','POST'])
async def webhook(request):
    if request.method == 'POST':
        await checkIfReconnect()
        alert = await request.json()  # Starlette中需await获取JSON数据
        
        # 提取订单关键参数（避免直接比较MarketOrder对象）
        ticker = alert['ticker']
        action = alert['order_action']
        contracts = alert['order_contracts']
        current_order_info = {
            "ticker": ticker,
            "action": action,
            "contracts": contracts
        }
        
        # 加锁避免并发请求的竞态问题
        async with request.app.state.order_lock:
            # 检查是否与上一次订单完全重复
            if current_order_info == request.app.state.last_order_info:
                return HTTPResponse("duplicate order ignored", 200)
            
            # 处理订单逻辑（提交到IB的代码）
            order = MarketOrder(action, contracts, account='aannnnnnn')
            # ... 这里添加提交订单的具体代码（比如调用IB的API）
            
            # 更新应用状态为当前订单信息
            request.app.state.last_order_info = current_order_info
        
        return HTTPResponse("ok", 200)
    return HTTPResponse("", 405)

方案2：添加TradingView警报唯一ID（更可靠的重复检测） #

如果TV重绘会发送完全相同的警报内容，可以在TV脚本中添加唯一标识符，后端记录已处理的ID，彻底避免重复：

步骤1：修改TradingView脚本

在策略中添加唯一ID生成逻辑，并包含在警报消息里：

// 生成唯一警报ID（时间戳+随机数）
var string alertID = str.tostring(timenow) + str.tostring(math.random())

// 警报消息格式（示例）
alertcondition(your_condition, title="Order Signal", message='{"ticker":"{{ticker}}","order_action":"{{strategy.order.action}}","order_contracts":{{strategy.order.contracts}},"alert_id":"' + alertID + '"}')

步骤2：后端修改为检测警报ID

# 初始化应用状态时添加已处理ID集合
app.state.processed_alert_ids = set()

# webhook函数修改
@app.route('/webhook', methods=['GET','POST'])
async def webhook(request):
    if request.method == 'POST':
        await checkIfReconnect()
        alert = await request.json()
        alert_id = alert['alert_id']
        
        async with request.app.state.order_lock:
            if alert_id in request.app.state.processed_alert_ids:
                return HTTPResponse("duplicate alert ignored", 200)
            
            # 处理订单逻辑
            order = MarketOrder(alert['order_action'], alert['order_contracts'], account='aannnnnnn')
            # ... 提交订单的具体代码
            
            # 记录已处理的警报ID
            request.app.state.processed_alert_ids.add(alert_id)
            # 可选：定期清理旧ID，避免集合过大（比如保留最近1小时的）
        
        return HTTPResponse("ok", 200)
    return HTTPResponse("", 405)

三、为什么你之前的尝试失败？ #

• Python中函数内的变量默认是局部变量，每次调用函数都会重新创建，无法保留上一次的值；
• 直接使用global关键字虽然能解决全局变量问题，但异步框架下会有并发竞态风险（多个请求同时修改全局变量导致数据错误）；
• MarketOrder对象默认比较内存地址，而非参数内容，所以直接xorder == order永远不成立，必须提取关键参数做比较。

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