Python编写RSI指标计算结果与TradingView数据不一致问题求助

核心问题1：RMA移动平均实现不符合TradingView逻辑 #

你当前使用的pd.Series.ewm(alpha=1/period).mean()默认开启了adjust=True参数，且没有按照Pine Script的RMA规则初始化首值：

Pine Script的RMA规则：首次计算时取前length个样本的SMA作为初始值，后续迭代公式为 当前值 = (1/length)*当前样本 + (1-1/length)*前一个RMA值，对应pandas的ewm需要设置adjust=False，且手动处理初始SMA值。

核心问题2：K线数据量不足 #

你当前仅拉取3小时的5分钟K线，总计只有36根K线，无法满足RSI14的RMA预热需求，至少需要拉取≥200根K线保证计算精度，同时要确保K线时间戳和TradingView完全对齐。

核心问题3：业务代码语法/逻辑错误 #

• get_historical_candles函数没有接收pair参数，内部调用的pair是全局变量，循环计算多个交易对时会取值错误
• 第二个rsi_tradingview函数的参数类型标注错误，ohlc: all_candles中all_candles是变量不是合法类型，应改为pd.DataFrame
• 第二个rsi_tradingview函数内部错误使用了外部变量all_candles进行计算，没有使用传入的ohlc参数，导致计算数据源错误

修复后的完整代码 #

正确的RMA与RSI实现 #

import pandas as pd
import numpy as np

def rma(series: pd.Series, length: int) -> pd.Series:
    # 完全对齐Pine Script逻辑的RMA实现
    alpha = 1 / length
    # 计算前length个值的SMA作为初始值
    initial_sma = series.rolling(window=length).mean().iloc[length-1]
    rma_series = pd.Series(np.nan, index=series.index)
    rma_series.iloc[length-1] = initial_sma
    # 迭代计算后续RMA值
    for i in range(length, len(series)):
        rma_series.iloc[i] = alpha * series.iloc[i] + (1 - alpha) * rma_series.iloc[i-1]
    return rma_series

def rsi_tradingview(ohlc: pd.DataFrame, period: int = 14, round_rsi: bool = True):
    delta = ohlc["close"].diff()
    up = delta.copy()
    up[up < 0] = 0
    down = delta.copy()
    down[down > 0] = 0
    down *= -1
    # 替换原有EWM计算，使用正确的RMA
    up_rma = rma(up, period)
    down_rma = rma(down, period)
    rsi = np.where(down_rma == 0, 100, np.where(up_rma == 0, 0, 100 - (100 / (1 + up_rma / down_rma))))
    return np.round(rsi, 2) if round_rsi else rsi

修正后的业务代码 #

pairs = ["BTCUSDT", "PONDUSDT"]

def get_historical_candles(pair: str):
    # 拉取20小时的5分钟K线，合计240根，满足预热需求
    record = client.get_historical_klines(pair, Client.KLINE_INTERVAL_5MINUTE, "20 hour ago UTC")
    myList = []
    try:
        for item in record:
            # 仅保留收盘价
            n_item = [float(item[4])]
            myList.append(n_item)
    except Exception as error:
        debug_logger.debug(error)
    new_ohlc = pd.DataFrame(myList, columns=['close'])
    return new_ohlc

for pair in pairs:
    all_candles = get_historical_candles(pair)
    test_rsi = rsi_tradingview(all_candles, 14, False)
    test_rsi_final = test_rsi[-1]
    print(f"{pair} RSI计算结果: {test_rsi_final}")

验证注意事项 #

拉取K线的时间范围尽量和TradingView的K线完全对齐，避免因为K线更新时间差导致的数值偏差。如果需要优化性能，可以将RMA的循环逻辑替换为pandas的ewm(alpha=1/period, adjust=False).mean()，手动指定初始SMA值即可，计算结果完全一致。

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