如何利用实时数据更新线性回归线（基于Binance WebSocket与DataFrame实现）

我明白你的需求：你想用WebSocket接收ETHUSDT的实时K线数据，然后动态更新之前用历史数据拟合的线性回归线，但目前代码里的更新逻辑不生效。让我帮你梳理问题并给出修改方案。

现有代码的核心问题 #

1. Y_pred没有实时更新：你在get_binance_bars里只拟合了一次历史数据的回归线，之后收到实时消息时没有重新训练模型，所以Y_pred还是旧的历史预测值，自然无法反映实时价格变动。
2. live_price的赋值逻辑错误：df['live_price'] = close会把整个列都设置为当前的实时价格，这不符合你“实时变动的数据存储在df['live_price']”的需求，应该只更新对应行的值。
3. 实时数据未纳入模型训练：你没有把WebSocket传来的实时K线数据加入到拟合回归线的数据集里，模型一直用的是初始的20根历史K线。

修改后的完整代码 #

import websocket, json
import requests
import numpy as np
import pandas as pd
import datetime as dt
from datetime import datetime, date
from sklearn.linear_model import LinearRegression

symbol = "ETHUSDT"
tf = "1m"
now = datetime.now()
today = date.today()
d = int(today.strftime("%d"))
m = int(today.strftime("%m"))
y = int(today.strftime("%Y"))
hr = int(now.strftime("%H"))
mn = int(now.strftime("%M"))

# 全局变量：保存线性回归模型，方便实时更新
linear_regressor = LinearRegression()

def get_binance_bars(ticker, interval, startTime, endTime):
    url = "https://api.binance.com/api/v3/klines"
    startTime = str(int(startTime.timestamp() * 1000))
    endTime = str(int(endTime.timestamp() * 1000))
    limit = '1000'
    req_params = {"symbol" : ticker, 'interval' : interval, 'startTime' : startTime, 'endTime' : endTime, 'limit' : limit}
    df = pd.DataFrame(json.loads(requests.get(url, params = req_params).text))
    if (len(df.index) == 0):
        return None
    df = df.iloc[:, 0:4]
    df.columns = ['time', 'high', 'low', 'close']
    # 转换数据类型
    df[['high', 'low', 'close']] = df[['high', 'low', 'close']].astype(float)
    # 转换时间格式
    df['time'] = [dt.datetime.fromtimestamp(x / 1000.0) for x in df['time']]
    # 移除最后一行，避免和实时K线重复
    df.drop(df.tail(1).index,inplace=True)
    # 添加live_price和Y_pred列
    df['live_price'] = np.nan
    df['Y_pred'] = np.nan
    # 初始拟合模型并计算预测值
    update_linear_regression(df)
    return df

def update_linear_regression(df):
    """专门用于更新线性回归模型和Y_pred的函数"""
    global linear_regressor
    # 取最近20个数据点（包括实时更新的live_price，如果有的话）
    # 优先用live_price，如果没有则用close
    df['used_price'] = df['live_price'].fillna(df['close'])
    # 取最后20行的时间和价格，转换为时间戳方便模型训练
    X = df['time'].iloc[-20:].apply(lambda x: x.timestamp()).values.reshape(-1, 1)
    Y = df['used_price'].iloc[-20:].values.reshape(-1, 1)
    # 重新训练模型
    linear_regressor.fit(X, Y)
    # 计算预测值并更新Y_pred列
    Y_pred = linear_regressor.predict(X)
    df.iloc[-20:, df.columns.get_loc('Y_pred')] = Y_pred
    # 删除临时列
    df.drop('used_price', axis=1, inplace=True)

SOCKET = "wss://stream.binance.com:9443/ws/"+symbol.lower()+"@kline_"+tf
df = get_binance_bars(symbol, tf, dt.datetime(y, m, d, hr-hr, mn), dt.datetime(y, m, d, hr, mn))

def on_open(ws):
    print('opened connection')

def on_close(ws):
    print('closed connection')

def on_message(ws, message):
    global df
    json_message = json.loads(message)
    kline = json_message['k']
    # 获取实时K线数据
    current_time = dt.datetime.fromtimestamp(kline['t']/1000)
    high = float(kline['h'])
    low = float(kline['l'])
    close = float(kline['c'])
    
    # 找到当前实时K线对应的行（匹配时间），如果没有则添加一行
    mask = df['time'] == current_time
    if mask.any():
        # 更新该行的live_price
        df.loc[mask, 'live_price'] = close
        # 可选：同步更新high和low
        df.loc[mask, 'high'] = high
        df.loc[mask, 'low'] = low
    else:
        # 如果当前K线是新的（比如初始历史数据之后的新K线），添加新行
        new_row = pd.DataFrame({
            'time': [current_time],
            'high': [high],
            'low': [low],
            'close': [close],
            'live_price': [close],
            'Y_pred': [np.nan]
        })
        df = pd.concat([df, new_row], ignore_index=True)
        # 保持数据集规模，只保留最近100行（可根据需求调整）
        df = df.iloc[-100:]
    
    # 重新更新线性回归模型和Y_pred
    update_linear_regression(df)
    # 打印最后20行查看实时效果
    print(df.tail(20))

ws = websocket.WebSocketApp(SOCKET, on_open=on_open, on_close=on_close, on_message=on_message)
ws.run_forever()

关键修改点说明 #

1. 新增update_linear_regression函数：把拟合模型、计算预测值的逻辑抽成独立函数，方便每次实时数据到来时调用，避免代码重复。
2. 实时数据整合：收到WebSocket消息时，要么更新对应时间行的live_price（当前正在运行的K线），要么添加新的K线行（当新的K线周期开始时）。
3. 动态更新模型：每次收到实时数据后，调用update_linear_regression，用最新的20个数据点（包含实时价格）重新训练模型，再更新Y_pred列。
4. 数据处理优化：用fillna处理live_price为空的情况（历史数据没有实时价格时用close替代），确保模型训练的数据完整。

这样修改后，每当WebSocket收到新的实时数据，你的线性回归线就会自动更新，完美匹配实时价格的变动。

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