如何用Python实现实时数据逐行输入KNN交易模型？

1. 实时数据输入模型的实现思路 #

当然可以实现实时数据输入模型的Python逻辑，核心是维护滚动历史数据窗口，每次新数据到来时：

• 更新历史窗口（保留指标计算所需的最小历史长度）
• 基于更新后的窗口计算技术指标
• 将指标输入预训练好的KNN模型生成交易信号

2. 修复Backtest框架内的指标动态更新问题 #

你的代码核心问题是每次计算指标都用了全量历史数据data，而非到当前bar的动态历史切片，且没有利用Backtest框架的内置指标计算机制。以下是修改后的代码：

修改后的KNNStrategy代码 #

import pandas as pd
import numpy as np
from backtesting import Strategy, Backtest
from sklearn.neighbors import KNeighborsClassifier
import yfinance as yf

# 实现支持滚动计算的指标函数
def calculate_macd(close_series, fast_period=12, slow_period=26, signal_period=9):
    ema_fast = close_series.ewm(span=fast_period, adjust=False).mean()
    ema_slow = close_series.ewm(span=slow_period, adjust=False).mean()
    macd = ema_fast - ema_slow
    return macd.iloc[-1]

def calculate_rsi(close_series, period=14):
    delta = close_series.diff()
    gain = delta.where(delta > 0, 0)
    loss = -delta.where(delta < 0, 0)
    avg_gain = gain.rolling(window=period).mean()
    avg_loss = loss.rolling(window=period).mean()
    rs = avg_gain / avg_loss
    rsi = 100 - (100 / (1 + rs))
    return rsi.iloc[-1]

def calculate_bollinger_bands(close_series, period=20, std_dev=2):
    ma = close_series.rolling(window=period).mean()
    std = close_series.rolling(window=period).std()
    ub = ma + std * std_dev
    lb = ma - std * std_dev
    return ub.iloc[-1], lb.iloc[-1]

def calculate_pivots_support_resistance(ohlc_data):
    # 枢轴点基于前一日高低收，1小时线取前24根数据
    if len(ohlc_data) < 24:
        return {k: np.nan for k in ['pivot', 'r1', 's1', 'r2', 's2', 'r3', 's3']}
    prev_day = ohlc_data.iloc[-24:-1]
    high = prev_day.High.max()
    low = prev_day.Low.min()
    close = prev_day.Close.iloc[-1]
    pivot = (high + low + close) / 3
    r1 = 2 * pivot - low
    s1 = 2 * pivot - high
    r2 = pivot + (high - low)
    s2 = pivot - (high - low)
    r3 = high + 2 * (pivot - low)
    s3 = low - 2 * (high - pivot)
    return {'pivot': pivot, 'r1': r1, 's1': s1, 'r2': r2, 's2': s2, 'r3': r3, 's3': s3}

class KNNStrategy(Strategy):
    price_delta = 0.05
    # 指标计算所需最小历史长度
    min_history = 35  # 取MACD最长周期26+RSI周期14的最大值

    def init(self):
        self.last_high = 0
        self.last_low = 0
        # 假设x_train和y_train已提前准备
        self.knn = KNeighborsClassifier(n_neighbors=2)
        self.knn.fit(x_train, y_train)
        self.forecasts = self.I(lambda: np.repeat(np.nan, len(self.data)), name='forecast')
        self.bar_count = 0

    def next(self):
        self.bar_count += 1
        # 等待足够历史数据再启动计算
        if self.bar_count < self.min_history:
            return

        # 取到当前bar为止的动态历史数据
        current_history = self.data.df.iloc[:self.bar_count]
        close = current_history.Close
        high = current_history.High
        low = current_history.Low

        # 动态计算指标
        current_macd = calculate_macd(close)
        latest_rsi = calculate_rsi(close)
        current_UB, current_LB = calculate_bollinger_bands(close)
        pivots = calculate_pivots_support_resistance(current_history)

        # 构造特征矩阵
        X = pd.DataFrame({
            'MACD': [current_macd],
            'RSI': [latest_rsi],
            'UB': [current_UB],
            'LB': [current_LB],
            'Pivot': [pivots['pivot']],
            'R1': [pivots['r1']],
            'S1': [pivots['s1']],
            'R2': [pivots['r2']],
            'S2': [pivots['s2']],
            'R3': [pivots['r3']],
            'S3': [pivots['s3']]
        })

        # 生成交易信号
        forecast = self.knn.predict(X)[0]
        self.forecasts[-1] = forecast

        # 交易逻辑
        current_close = close.iloc[-1]
        if forecast == 0 and self.position.size == 0:
            self.buy(size=.2, sl=low.iloc[-1])
        elif forecast == 1 and self.position.is_long:
            self.position.close()
        elif forecast == 1 and self.position.size == 0:
            self.sell(size=.2, sl=high.iloc[-1])
        elif forecast == 0 and self.position.is_short:
            self.position.close()

# 准备数据与训练集
data = yf.download('EURUSD=X', start='2022-03-30', interval='1h')
# 生成训练用特征与标签（示例逻辑）
train_data = data.iloc[:1000]
def generate_features(df):
    features = []
    for i in range(35, len(df)):
        hist = df.iloc[:i+1]
        macd = calculate_macd(hist.Close)
        rsi = calculate_rsi(hist.Close)
        ub, lb = calculate_bollinger_bands(hist.Close)
        pivots = calculate_pivots_support_resistance(hist)
        features.append([macd, rsi, ub, lb, pivots['pivot'], pivots['r1'], pivots['s1'], pivots['r2'], pivots['s2'], pivots['r3'], pivots['s3']])
    return pd.DataFrame(features)

x_train = generate_features(train_data)
# 示例标签：下一根K线是否上涨（1涨/0跌）
y_train = np.where(train_data.Close.shift(-1) > train_data.Close, 1, 0)[35:1000]

# 运行回测
bt = Backtest(data, KNNStrategy, cash=10000)
results = bt.run()
print(results)

3. 实盘实时数据输入的实现要点 #

如果对接实盘实时数据（如交易所websocket、MT4/MT5接口），核心逻辑如下：

• 用滚动DataFrame或队列维护历史数据，只保留指标计算所需的最小窗口长度
• 新数据加入后，调用指标函数计算当前值
• 将指标输入KNN模型得到信号，对接实盘API执行交易

示例伪代码：

import websocket
import json

# 初始化滚动历史窗口
history_window = pd.DataFrame(columns=['Open', 'High', 'Low', 'Close'])
max_window_size = 35  # 指标所需最大历史长度

def on_message(ws, message):
    global history_window
    # 解析实时数据（格式根据数据源调整）
    realtime_data = json.loads(message)
    new_bar = pd.DataFrame({
        'Open': [realtime_data['open']],
        'High': [realtime_data['high']],
        'Low': [realtime_data['low']],
        'Close': [realtime_data['close']]
    }, index=[pd.to_datetime(realtime_data['timestamp'])])
    
    # 更新滚动窗口
    history_window = pd.concat([history_window, new_bar]).tail(max_window_size)
    
    # 窗口足够时计算信号
    if len(history_window) >= max_window_size:
        current_macd = calculate_macd(history_window.Close)
        latest_rsi = calculate_rsi(history_window.Close)
        current_UB, current_LB = calculate_bollinger_bands(history_window.Close)
        pivots = calculate_pivots_support_resistance(history_window)
        
        X = pd.DataFrame({
            'MACD': [current_macd],
            'RSI': [latest_rsi],
            'UB': [current_UB],
            'LB': [current_LB],
            'Pivot': [pivots['pivot']],
            'R1': [pivots['r1']],
            'S1': [pivots['s1']],
            'R2': [pivots['r2']],
            'S2': [pivots['s2']],
            'R3': [pivots['r3']],
            'S3': [pivots['s3']]
        })
        
        forecast = knn.predict(X)[0]
        # 对接实盘API执行交易
        if forecast == 0 and not has_open_position():
            submit_buy_order(size=0.2, stop_loss=history_window.Low.iloc[-1])
        # 其他交易逻辑...

# 连接实时数据websocket
ws = websocket.WebSocketApp("wss://your-realtime-data-source.com", on_message=on_message)
ws.run_forever()

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