第 16 篇：指标库实战：TA-Lib 与 pandas-ta

import pandas_ta as ta
sma20 = ta.sma(close, length=20)      # 返回 Series，名字叫 SMA_20
rsi14 = ta.rsi(close, length=14)      # 返回 Series，名字叫 RSI_14

import numpy as np
import pandas as pd


def load_data():
    """优先用 akshare 拉真实行情，失败则退回本地模拟数据，保证代码一定能跑。"""
    try:
        import akshare as ak
        raw = ak.stock_zh_a_hist(                      # A 股日线行情接口
            symbol="000001", period="daily",
            start_date="20200101", end_date="20241231", adjust="qfq",
        )
        raw = raw.rename(columns={
            "日期": "date", "开盘": "open", "收盘": "close",
            "最高": "high", "最低": "low", "成交量": "volume",
        })
        df = raw[["date", "open", "high", "low", "close", "volume"]].copy()
        df["date"] = pd.to_datetime(df["date"])
        df = df.set_index("date")
        print(f"已加载 akshare 真实数据：平安银行，共 {len(df)} 个交易日")
    except Exception as e:                             # 网络不通、接口变动时走这里
        print(f"akshare 不可用（{e}），改用本地模拟数据")
        rng = np.random.default_rng(42)                # 固定种子，结果可复现
        dates = pd.bdate_range("2020-01-01", "2024-12-31")
        n = len(dates)
        ret = rng.normal(0.0002, 0.018, n)             # 日收益：微小漂移 + 波动
        close = 10.0 * np.cumprod(1 + ret)
        open_ = close * (1 + rng.normal(0, 0.003, n))
        high = np.maximum(open_, close) * (1 + np.abs(rng.normal(0, 0.006, n)))
        low = np.minimum(open_, close) * (1 - np.abs(rng.normal(0, 0.006, n)))
        volume = rng.integers(5_000_000, 200_000_000, n)
        df = pd.DataFrame({"open": open_, "high": high, "low": low,
                           "close": close, "volume": volume}, index=dates)
        df.index.name = "date"
        print(f"已生成模拟数据：{dates[0].date()} ~ {dates[-1].date()}，共 {n} 个交易日")
    return df


df = load_data()

import numpy as np
import pandas as pd

# —— 第一步：探测当前环境里哪个库能用，保证三种环境都能跑 ——
lib_name = None
try:
    import talib                    # TA-Lib 的 Python 封装，快但安装门槛高
    lib_name = "TA-Lib"
except ImportError:
    try:
        import pandas_ta as ta      # 纯 Python 实现，pip 一步到位
        lib_name = "pandas-ta"
    except ImportError:
        lib_name = None             # 两个都没有，稍后只展示手写版
print("当前使用的指标库：", lib_name or "都没有，只展示手写版")


def lib_indicators(df):
    """用检测到的指标库算五个常用指标，统一返回 {名字: Series} 字典。"""
    close = df["close"].to_numpy(dtype="float64")   # TA-Lib 只认 float64 数组
    high = df["high"].to_numpy(dtype="float64")
    low = df["low"].to_numpy(dtype="float64")
    out = {}
    if lib_name == "TA-Lib":
        out["SMA20"] = pd.Series(talib.SMA(close, timeperiod=20), index=df.index)
        out["EMA20"] = pd.Series(talib.EMA(close, timeperiod=20), index=df.index)
        macd, _, _ = talib.MACD(close)              # 默认参数 12/26/9，这里只用 MACD 线
        out["MACD"] = pd.Series(macd, index=df.index)
        out["RSI14"] = pd.Series(talib.RSI(close, timeperiod=14), index=df.index)
        out["ATR14"] = pd.Series(talib.ATR(high, low, close, timeperiod=14), index=df.index)
    elif lib_name == "pandas-ta":
        out["SMA20"] = ta.sma(df["close"], length=20)
        out["EMA20"] = ta.ema(df["close"], length=20)
        out["MACD"] = ta.macd(df["close"]).iloc[:, 0]   # 三列里的第一列是 MACD 线
        out["RSI14"] = ta.rsi(df["close"], length=14)
        out["ATR14"] = ta.atr(df["high"], df["low"], df["close"], length=14)
    return out

import numpy as np
import pandas as pd


def hand_indicators(df):
    """用 pandas 手写同样的五个指标（第 10~14 篇的做法收拢成一个函数）。"""
    close, high, low = df["close"], df["high"], df["low"]
    out = {}
    out["SMA20"] = close.rolling(20).mean()                   # 简单均线
    out["EMA20"] = close.ewm(span=20, adjust=False).mean()    # 指数均线
    ema_fast = close.ewm(span=12, adjust=False).mean()
    ema_slow = close.ewm(span=26, adjust=False).mean()
    out["MACD"] = ema_fast - ema_slow                         # MACD 线 = 快慢 EMA 之差
    delta = close.diff()
    gain = delta.clip(lower=0).ewm(alpha=1 / 14, adjust=False).mean()
    loss = (-delta.clip(upper=0)).ewm(alpha=1 / 14, adjust=False).mean()
    out["RSI14"] = 100 - 100 / (1 + gain / loss)              # Wilder 平滑版 RSI
    prev_close = close.shift(1)
    tr = pd.concat([high - low, (high - prev_close).abs(),
                    (low - prev_close).abs()], axis=1).max(axis=1)   # 真实波幅
    out["ATR14"] = tr.ewm(alpha=1 / 14, adjust=False).mean()  # Wilder 平滑版 ATR
    return out


lib = lib_indicators(df)
hand = hand_indicators(df)

# —— 一致性核对：跳过前 250 根预热期，再比每个指标的最大绝对差 ——
print("\n一致性核对（从第 250 根 bar 开始比较）：")
for name in ["SMA20", "EMA20", "MACD", "RSI14", "ATR14"]:
    if name not in lib:
        continue
    diff = (lib[name] - hand[name]).iloc[250:].abs().max()
    verdict = "一致" if diff < 1e-4 else "有差异，先查口径"
    print(f"  {name:>6}: 最大差异 = {diff:.6f} -> {verdict}")

print("\n最新一根 bar 的指标值：")
for name, s in (lib or hand).items():   # 库没装就打印手写版
    print(f"  {name:>6} = {s.iloc[-1]:.3f}")

当前使用的指标库： TA-Lib
已加载 akshare 真实数据：平安银行，共 1215 个交易日

一致性核对（从第 250 根 bar 开始比较）：
  SMA20: 最大差异 = 0.000000 -> 一致
  EMA20: 最大差异 = 0.000001 -> 一致
  MACD : 最大差异 = 0.000003 -> 一致
  RSI14: 最大差异 = 0.000018 -> 一致
  ATR14: 最大差异 = 0.000037 -> 一致

最新一根 bar 的指标值：
  SMA20 = 10.852
  EMA20 = 10.837
  MACD  = 0.026
  RSI14 = 54.318
  ATR14 = 0.194