使用Python websockets库连接IG API的Lightstreamer流服务遇阻

方案可行性确认 #

用Python的websockets库连接Lightstreamer调用IG API流服务是完全可行的——Lightstreamer原生支持WebSocket协议，IG的实时流服务正是基于Lightstreamer搭建的。

问题排查与解决 #

1. URI无效错误的原因 #

websockets库仅支持WebSocket协议的URL（ws://或wss://开头），你最初使用http://开头的URL不符合库的要求，因此触发无效URI错误，改成wss://是正确的修正方向。

2. 连接失败的核心问题及修复 #

改成wss://后仍连接失败，主要排查以下几点：

（1）路径拼写错误

你的代码中URL路径写的是lighstreamer，少了一个字母t，正确路径应为lightstreamer，这是最可能导致连接失败的低级错误。

（2）缺少身份验证

IG的Lightstreamer连接必须携带认证信息，否则服务器会拒绝连接。你需要先通过IG的REST API登录获取会话凭证（CST和X-SECURITY-TOKEN），再在WebSocket连接时通过请求头传递这些凭证。

（3）端点地址核对

确认IG Demo环境的Lightstreamer端点是否正确，官方提供的Demo端点通常为wss://demo-apd.marketdatasystem.com/lightstreamer，如果有变动请以IG官方API文档为准。

修正后的完整代码示例 #

import websockets
import asyncio
import requests

# 替换为你的IG API密钥、用户名、密码
IG_API_KEY = "你的IG API密钥"
IG_USERNAME = "你的IG用户名"
IG_PASSWORD = "你的IG密码"

# 第一步：通过REST API获取会话凭证
def get_ig_session():
    login_url = "https://demo-api.ig.com/gateway/deal/session"
    headers = {
        "X-IG-API-KEY": IG_API_KEY,
        "Content-Type": "application/json"
    }
    payload = {
        "identifier": IG_USERNAME,
        "password": IG_PASSWORD
    }
    response = requests.post(login_url, headers=headers, json=payload)
    response.raise_for_status()
    return {
        "cst": response.headers["CST"],
        "security_token": response.headers["X-SECURITY-TOKEN"]
    }

# 第二步：建立WebSocket连接并订阅流
async def connect_lightstreamer():
    session = get_ig_session()
    # 纠正路径拼写的正确WebSocket地址
    ws_url = "wss://demo-apd.marketdatasystem.com/lightstreamer"
    
    async with websockets.connect(
        ws_url,
        extra_headers={
            "X-IG-API-KEY": IG_API_KEY,
            "CST": session["cst"],
            "X-SECURITY-TOKEN": session["security_token"]
        }
    ) as websocket:
        print("Lightstreamer连接成功！")
        
        # 示例：订阅EURUSD的买卖报价流
        subscribe_msg = {
            "LS_op": "add",
            "LS_table": "MERGE",
            "LS_id": "eurusd_quote",
            "LS_schema": "BID,OFFER",
            "LS_data_adapter": "QUOTE_ADAPTER",
            "LS_item": "EURUSD"
        }
        await websocket.send(str(subscribe_msg))
        
        # 持续接收流数据
        async for message in websocket:
            print("收到流数据：", message)

# 运行异步任务
asyncio.run(connect_lightstreamer())

额外注意事项 #

• 确保你的网络环境允许访问wss://demo-apd.marketdatasystem.com，如果有代理或防火墙，需要配置对应的网络规则。
• 订阅消息的格式必须符合Lightstreamer的协议规范，具体字段（如LS_data_adapter、LS_item）请参考IG官方流服务文档。

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