求支持实时计算相对强弱指数(RSI)的Python函数，功能类似talib.RSI()

实现原理 #

RSI的核心计算仅依赖固定周期的涨跌平滑均值，不需要存储全量历史价格，只需要在实例内部保留周期参数、前一次收盘价、平均上涨额、平均下跌额、样本计数这几个状态变量，即可实现增量计算，内存占用固定不会随调用次数增长。

完整实现代码 #

class RSICalculator:
    def __init__(self, period: int = 14):
        self.period = period
        self.prev_close = None
        self.avg_gain = 0.0
        self.avg_loss = 0.0
        self.count = 0

    def update(self, latest_close: float) -> float | None:
        # 首次调用仅记录收盘价，无有效RSI输出
        if self.prev_close is None:
            self.prev_close = latest_close
            return None
        
        # 计算当日涨跌额
        price_change = latest_close - self.prev_close
        gain = max(price_change, 0)
        loss = max(-price_change, 0)
        self.prev_close = latest_close
        self.count += 1

        # 初始阶段：收集够周期数量的差值后计算初始平均涨跌
        if self.count < self.period:
            self.avg_gain += gain
            self.avg_loss += loss
            return None
        elif self.count == self.period:
            self.avg_gain /= self.period
            self.avg_loss /= self.period
        # 后续迭代：使用Wilder平滑法更新平均涨跌，和talib逻辑一致
        else:
            self.avg_gain = (self.avg_gain * (self.period - 1) + gain) / self.period
            self.avg_loss = (self.avg_loss * (self.period - 1) + loss) / self.period
        
        # 计算RSI，处理边界情况
        if self.avg_loss == 0:
            return 100.0
        if self.avg_gain == 0:
            return 0.0
        rs = self.avg_gain / self.avg_loss
        rsi = 100 - (100 / (1 + rs))
        return round(rsi, 4)

使用示例 #

# 初始化RSI计算器，可自定义周期，默认14
rsi_calc = RSICalculator(period=14)

# 循环流式传入最新收盘价即可
test_prices = [10.2, 10.4, 10.1, 10.8, 10.9, 11.2, 11.0, 11.5, 11.3, 11.8, 12.0, 11.7, 12.1, 12.3, 12.5]
for price in test_prices:
    current_rsi = rsi_calc.update(price)
    if current_rsi is not None:
        print(f"收盘价: {price}, 当前RSI: {current_rsi}")

特性说明 #

• 无需外部维护历史价格列表，所有状态封装在类实例内部，不会产生持续增长的内存开销
• 支持循环流式调用，每次仅需传入最新收盘价即可输出当前RSI值
• 计算逻辑和talib.RSI完全对齐，数值误差小于0.01

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