第 12 篇：RSI：超买超卖的度量

import numpy as np                     # 数值计算
import pandas as pd                    # 表格与时间序列
import matplotlib.pyplot as plt        # 画图

def get_data() -> pd.DataFrame:
    """拉真实行情；网络不通时降级为本地模拟数据，保证代码永远能跑。"""
    try:
        import akshare as ak
        df = ak.stock_zh_a_hist(symbol="000001", period="daily",
                                start_date="20200101", end_date="20241231",
                                adjust="qfq")                    # 前复权日线
        df = df.rename(columns={"日期": "date", "收盘": "close"})
        df["date"] = pd.to_datetime(df["date"])
        print(f"已从 akshare 拉取真实行情 {len(df)} 行")
    except Exception as e:                                       # 数据源不可用
        print(f"akshare 不可用（{e}），改用本地模拟数据演示")
        rng = np.random.default_rng(42)                          # 固定种子可复现
        n = 1215                                                 # 约 5 年交易日数
        trend = np.linspace(0, 4, n)                             # 长期缓慢上行
        wave = 1.5 * np.sin(np.arange(n) / 30)                   # 周期项造波段
        noise = rng.normal(0, 0.35, n)                           # 日间随机扰动
        close = 10 + trend + wave + noise
        df = pd.DataFrame({"date": pd.bdate_range("2020-01-01", periods=n),
                           "close": close.round(2)})
    return df.reset_index(drop=True)

df = get_data()

def rsi_wilder_loop(close: pd.Series, n: int = 14) -> pd.Series:
    """手写 Wilder 递推版 RSI：与看盘软件同一套平均方式。"""
    diff = close.diff()                        # 今日收盘 - 昨日收盘
    gain = diff.clip(lower=0)                  # 只记涨：跌的日子记 0
    loss = (-diff).clip(lower=0)               # 只记跌：取绝对值，涨的日子记 0
    rsi = pd.Series(np.nan, index=close.index)
    g = gain.iloc[1 : n + 1].mean()            # 前 n 天简单平均，做递推"种子"
    l = loss.iloc[1 : n + 1].mean()
    rsi.iloc[n] = 100.0 if l == 0 else 100 - 100 / (1 + g / l)
    for i in range(n + 1, len(close)):         # 从第 n+1 行开始递推
        g = (g * (n - 1) + gain.iloc[i]) / n   # 旧平均占 13/14，今天占 1/14
        l = (l * (n - 1) + loss.iloc[i]) / n
        rsi.iloc[i] = 100.0 if l == 0 else 100 - 100 / (1 + g / l)
    return rsi

df["rsi_loop"] = rsi_wilder_loop(df["close"])

def rsi_rolling(close: pd.Series, n: int = 14) -> pd.Series:
    """rolling 简化版：直接对最近 n 天的涨/跌取简单平均。"""
    diff = close.diff()
    gain = diff.clip(lower=0)
    loss = (-diff).clip(lower=0)
    avg_gain = gain.rolling(n).mean()          # 最近 n 天平均涨幅
    avg_loss = loss.rolling(n).mean()          # 最近 n 天平均跌幅
    rs = avg_gain / avg_loss
    rsi = 100 - 100 / (1 + rs)
    # 防除零：区间内只涨不跌时 avg_loss=0，按惯例 RSI 记 100
    return rsi.mask((avg_loss == 0) & (avg_gain > 0), 100.0)

df["rsi_roll"] = rsi_rolling(df["close"])
gap = (df["rsi_loop"] - df["rsi_roll"]).abs()  # 两种实现的逐行差异
print(f"两种实现最大差异 {gap.max():.1f} 点，平均差异 {gap.mean():.1f} 点")
print(df[["close", "rsi_loop", "rsi_roll"]].tail(3).round(1))

overbought = df["rsi_roll"] > 70               # 超买布尔序列
share = overbought.mean()                      # 超买天数占比
stretch = overbought.rolling(10).sum().eq(10)  # 连续 10 天都超买
starts = stretch.astype(int).diff().eq(1).sum()  # "赖着不走"的区间数
print(f"RSI>70 天数占比 {share:.1%}，连续 10 天超买的区间数 {starts}")

fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 7), sharex=True,
                               gridspec_kw={"height_ratios": [2, 1]})
ax1.plot(df["date"], df["close"], lw=1, label="close")       # 上图：价格
ax2.plot(df["date"], df["rsi_loop"], lw=1, color="purple",
         label="RSI(14)")                                    # 下图：递推版 RSI
ax2.axhline(70, color="red", ls="--", lw=0.8)                # 超买参考线
ax2.axhline(30, color="green", ls="--", lw=0.8)              # 超卖参考线
ax2.set_ylim(0, 100)
ax1.legend(loc="upper left")
ax1.set_title("Price vs RSI(14)")
plt.tight_layout()
plt.savefig("rsi_demo.png", dpi=120)           # 存图，方便在无界面环境查看
print("图已保存为 rsi_demo.png")

已从 akshare 拉取真实行情 1215 行

两种实现最大差异 6.4 点，平均差异 1.9 点
      close  rsi_loop  rsi_roll
1212  11.38      58.3      57.1
1213  11.52      62.7      61.2
1214  11.41      60.1      58.8

RSI>70 天数占比 11.2%，连续 10 天超买的区间数 3

图已保存为 rsi_demo.png