通过Python连接Trading View Webhook与Interactive Brokers时遇500错误

错误核心原因 #

终端报错AttributeError: 'Sanic' object has no attribute 'ib'，是因为代码中混用了app.ib和app.ctx.ib——初始化和启动时用app.ctx.ib存储IB连接对象，但在重连函数和下单逻辑里错误调用了不存在的app.ib属性。此外代码还有几处拼写、逻辑和语法问题，共同导致了500错误。

完整修正代码 #

# imports
from datetime import datetime
from sanic import Sanic
from sanic import response
from ib_insync import *

# Create Sanic object
app = Sanic(__name__)
app.ctx.ib = None

# Create root / homepage
@app.route('/')
async def root(request):
    return response.text('online')

# Listen for signals and submit BUY orders (webhook)
@app.route('/buy', methods=['POST']) 
async def buy(request):
    if request.method == 'POST':
        # Check if we need to reconnect
        await checkIfReconnect()
        # Parse signal data
        try:
            data = request.json
        except:
            return response.json({"status": "error", "msg": "Invalid JSON data"}, status=400)
        # Create order and contract
        order = MarketOrder('BUY', 0.00015)
        contract = Crypto('BTC', 'PAXOS', 'USD')
        # Place order
        trade = app.ctx.ib.placeOrder(contract, order)
        # Wait for the order to be processed and check status
        app.ctx.ib.sleep(1)
        print(f"Buy trade executed: {trade}")
        return response.json({"status": "success", "trade": str(trade)}, status=200)

# Listen for signals and submit SELL orders (webhook)
@app.route('/sell', methods=['POST']) 
async def sell(request):
    if request.method == 'POST':
        # Check if we need to reconnect
        await checkIfReconnect()
        # Parse signal data
        try:
            data = request.json
        except:
            return response.json({"status": "error", "msg": "Invalid JSON data"}, status=400)
        # Create order and contract
        order = MarketOrder('SELL', 0.00015)
        contract = Crypto('BTC', 'PAXOS', 'USD')
        # Place order
        trade = app.ctx.ib.placeOrder(contract, order)
        # Wait for the order to be processed and check status
        app.ctx.ib.sleep(1)
        print(f"Sell trade executed: {trade}")
        return response.json({"status": "success", "trade": str(trade)}, status=200)

# Reconnect to IB if connection is lost
async def checkIfReconnect():
    # 判断连接是否断开，断开则重连
    if not app.ctx.ib.isConnected():
        try:
            app.ctx.ib.disconnect()
        except:
            pass
        app.ctx.ib = IB()
        app.ctx.ib.connect('127.0.0.1', 7496, clientId=1)

# Run app
if __name__ == '__main__':
    # Connect to IB
    app.ctx.ib = IB()
    app.ctx.ib.connect('127.0.0.1', 7496, clientId=1)
    app.run(port=8080, debug=True)

关键修正点说明 #

1. 统一IB对象引用：全程使用app.ctx.ib存储和调用IB连接对象，避免属性不存在的错误
2. 修复JSON解析错误：将request.jason改为正确的request.json，并增加异常处理避免无效JSON导致崩溃
3. 修正重连逻辑：原逻辑判断混乱，改为仅当连接断开时才执行重连操作，同时增加异常处理避免断开已关闭的连接时报错
4. 修正特殊变量名：_name_改为Python标准的__name__双下划线变量
5. 添加响应返回：给每个路由函数添加明确的JSON响应，避免Sanic因无返回值抛出异常
6. 规范缩进：修正原代码中缩进不正确的代码块，确保语法合法

额外注意事项 #

• 确保IB TWS已开启API访问：在TWS设置中勾选"Enable Active X and Socket Clients"，并允许本地IP访问
• 端口匹配：7496是TWS模拟账户的默认端口，实盘账户请使用7497
• ClientID唯一性：确保clientId=1没有被其他IB连接占用，否则会导致连接失败
• 本地测试优先：先用Postman发送POST请求到http://localhost:8080/buy验证服务正常，再通过Ngrok转发到公网

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