如何维持TradingView WebSocket永久连接？遇1000正常关闭错误

问题背景 #

从TradingView流式获取多加密货币行情数据并写入对应Pandas DataFrame，期望连接持续运行至手动关闭，但连接约1分20秒后自动断开，报错信息：ConnectionClosedOK: received 1000 (OK); then sent 1000 (OK)。尝试过Ping、更换WebSocket库等方法未解决，用户不熟悉WebSocket与异步编程，怀疑async with导致连接关闭。

核心原因 #

1. 缺少心跳维持：TradingView的WebSocket服务器会在一段时间无交互后主动断开连接，原代码未实现心跳包发送机制。
2. 阻塞异步事件循环：使用同步的time.sleep(0.05)会阻塞整个异步事件循环，导致无法及时处理服务器的Ping响应或发送心跳。
3. 初始化逻辑顺序错误：原代码先接收数据再发送初始化消息，可能导致服务器未正确识别会话状态。

修复步骤 #

1. 添加定时心跳任务：发送TradingView约定格式的心跳包，保持连接活跃。
2. 替换同步sleep为异步延迟：用asyncio.sleep()替代time.sleep()，避免阻塞事件循环。
3. 调整初始化顺序：先发送所有初始化消息，再进入数据接收循环。
4. 优化异常处理：捕获WebSocket连接异常，支持重连逻辑。

完整修复代码 #

import json 
import websockets
import pandas as pd
import urllib
import asyncio
import string
import random
import re
from datetime import datetime
import nest_asyncio
nest_asyncio.apply()

wsParams = {"from": "symbols/BINANCE-ETHUSDT/"}
websocketUri = f"wss://data.tradingview.com/socket.io/websocket?{urllib.parse.urlencode(wsParams)}"

def generateChartSession():
    stringLength=12
    letters = string.ascii_lowercase
    random_string= ''.join(random.choice(letters) for i in range(stringLength))
    return "cs_" + random_string
    
initMessages = [
    {"m": "set_data_quality", "p": ["low"]},
    {"m": "set_auth_token", "p": ["unauthorized_user_token"]}]

symbols = ['BINANCE:LRCUSDT', 'BINANCE:ADAUSDT', 'BINANCE:ETHUSDT', 'BINANCE:IMXUSDT', 'BINANCE:BTCUSDT']
dfs={} # 存储各币种的行情DataFrame
symbols_session_link = {}

i=0
for symbol in symbols:
    i+=1
    session = generateChartSession()

    initMessages.append({"m":"chart_create_session","p":[session ,""]})
    initMessages.append({"m":"switch_timezone","p":[session ,"Etc/UTC"]})

    lastPart = '={"symbol":"' + symbol+  '","adjustment":"splits","session":"extended"}'
    initMessages.append({"m":"resolve_symbol","p":[session ,f"symbol_{i}",lastPart]})
    initMessages.append({"m":"create_series","p":[session ,"s1","s1",f"symbol_{i}","1",2]}) # 订阅1分钟K线，获取最近2根
    
    symbolName = symbol.split(":")[1]
    symbols_session_link[session] = symbolName
    # 读取本地CSV初始化DataFrame，文件不存在则创建空DF
    try:
        df = pd.read_csv(f'G:\\My Drive\\Python Daytrading\\stock data\\test\\{symbolName}.csv')
    except FileNotFoundError:
        df = pd.DataFrame(columns=['datetime', 'symbol', 'open', 'high', 'low', 'close', 'volume'])
    dfs[symbolName] = df

def strip(text):
    noDataReg = re.match('~m~\d+~m~~h~\d+', text, re.MULTILINE)
    if not noDataReg:
        dataReg = re.split('~m~\d+~m~', text)
        return [json.loads(t) for t in dataReg if t]
    return []

def unstrip(text):
    return f"~m~{len(json.dumps(text))}~m~{json.dumps(text)}"

async def init(websocket):
    for m in initMessages:
        await websocket.send(unstrip(m))
        await asyncio.sleep(0.1)  # 给服务器留处理时间

async def send_heartbeat(websocket):
    """定时发送心跳包，维持连接"""
    while True:
        await websocket.send('~h~1')  # TradingView约定的心跳格式
        await asyncio.sleep(30)  # 每30秒发送一次

async def startReceiving(websocket):
    # 先完成会话初始化
    await init(websocket)
    # 启动心跳任务
    asyncio.create_task(send_heartbeat(websocket))

    while True:
        try:
            data = await websocket.recv()
            payloads = strip(data)
            for p in payloads:
                try:
                    # 仅处理行情更新消息
                    if p["m"] == "timescale_update" and "s1" in p["p"][1]:
                        last_tick = [[t["v"][0], t["v"][1], t["v"][2], t["v"][3], t["v"][4], t["v"][5]] 
                                    for t in p["p"][1]["s1"]["s"]]
                        
                        candleMinuteTime = datetime.fromtimestamp(p["p"][1]["s1"]["lbs"]['bar_close_time'])
                        candleMinuteTime = candleMinuteTime.replace(minute=candleMinuteTime.minute-1)
                        candleMinuteTime_str = candleMinuteTime.strftime('%Y-%m-%d %H:%M:%S')
                        
                        symbol_name = symbols_session_link[p['p'][0]]
                        df = dfs[symbol_name]
                        
                        # 更新或添加K线数据
                        if len(df) == 0 or df.iloc[-1]['datetime'] != candleMinuteTime_str:
                            row = {
                                'datetime': candleMinuteTime_str,
                                'symbol': symbol_name,
                                'open': last_tick[0][1],
                                'high': last_tick[0][2],
                                'low': last_tick[0][3],
                                'close': last_tick[0][4],
                                'volume': last_tick[0][5]
                            }
                            dfs[symbol_name] = pd.concat([df, pd.DataFrame([row])], ignore_index=True)
                        else:
                            df.at[df.index[-1], 'open'] = last_tick[0][1]
                            df.at[df.index[-1], 'high'] = last_tick[0][2]
                            df.at[df.index[-1], 'low'] = last_tick[0][3]
                            df.at[df.index[-1], 'close'] = last_tick[0][4]
                            df.at[df.index[-1], 'volume'] = last_tick[0][5]
                        
                        print(f"{symbol_name} | {candleMinuteTime_str} | 收盘价: {last_tick[0][4]}")
                        # 可选：定期保存到CSV，避免数据丢失
                        # dfs[symbol_name].to_csv(f'G:\\My Drive\\Python Daytrading\\stock data\\test\\{symbol_name}.csv', index=False)
                        
                    await asyncio.sleep(0.05)
                except Exception as e:
                    # 可打开注释查看具体错误
                    # print(f"处理数据出错: {e}")
                    pass
        except websockets.exceptions.ConnectionClosed:
            print("连接已断开，尝试重连...")
            break

async def websocketConnect():
    while True:
        try:
            async with websockets.client.connect(
                websocketUri, 
                extra_headers={"Origin": "https://www.tradingview.com"},
                ping_interval=20,  # 启用内置Ping机制，20秒一次
                ping_timeout=60
            ) as websocket:
                print("WebSocket连接成功")
                await startReceiving(websocket)
        except Exception as e:
            print(f"连接出错: {e}")
            await asyncio.sleep(10)  # 重连间隔10秒

if __name__ == '__main__':
    asyncio.run(websocketConnect())

关键说明 #

• 双重心跳保障：自定义send_heartbeat任务发送TradingView约定的心跳包，同时启用websockets库内置的ping_interval和ping_timeout，确保连接稳定。
• 异步非阻塞：用asyncio.sleep()替代同步sleep，保证事件循环能及时处理服务器响应和心跳。
• 会话初始化优化：先发送所有初始化消息再进入接收循环，确保服务器正确识别会话状态。
• 重连支持：捕获连接关闭异常后自动尝试重连，提升程序健壮性。

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