第 14 篇：ATR 与 KDJ：波动与摆动

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

def get_data() -> pd.DataFrame:
    """优先拉真实行情；不可用时降级为本地模拟 OHLC，保证代码永远能跑。"""
    try:
        import akshare as ak
        df = ak.stock_zh_a_hist(symbol="000333", period="daily",
                                start_date="20200101", end_date="20241231",
                                adjust="qfq")                    # 前复权日线
        df = df.rename(columns={"日期": "date", "最高": "high",
                                "最低": "low", "收盘": "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(2024)                        # 固定种子可复现
        n = 1215                                                 # 约 5 年交易日
        ret = rng.normal(0.0002, 0.018, n)                       # 日收益：微漂移+波动
        close = 100 * np.exp(np.cumsum(ret))                     # 收盘价路径
        vol = 0.010 + 0.008 * np.abs(np.sin(np.arange(n) / 50))  # 波动率周期起伏
        up = np.abs(rng.normal(0, 1, n)) * vol                   # 当日向上晃动幅度
        dn = np.abs(rng.normal(0, 1, n)) * vol                   # 当日向下晃动幅度
        df = pd.DataFrame({
            "date": pd.bdate_range("2020-01-01", periods=n),
            "high": (close * (1 + up)).round(2),                 # 当日最高
            "low": (close * (1 - dn)).round(2),                  # 当日最低
            "close": close.round(2)})
    return df.reset_index(drop=True)

df = get_data()

def atr(df: pd.DataFrame, n: int = 14) -> pd.DataFrame:
    """ATR：TR 三项取最大，再做 Wilder 平滑；顺带输出百分比版。"""
    prev = df["close"].shift(1)                    # 昨日收盘
    tr = pd.concat([df["high"] - df["low"],         # 情形一：当日高低差
                    (df["high"] - prev).abs(),      # 情形二：最高与昨收的距离
                    (df["low"] - prev).abs()],      # 情形三：最低与昨收的距离
                   axis=1).max(axis=1)              # 三项取最大 = 真实波幅
    # alpha=1/n 的 ewm 正是 Wilder 递推（RSI 篇手写循环的等价一行版）
    df["tr"] = tr
    df["atr"] = tr.ewm(alpha=1 / n, adjust=False).mean()
    df["atr_pct"] = df["atr"] / df["close"] * 100   # 换算成价格的百分比
    return df

df = atr(df)
print(df[["high", "low", "close", "atr", "atr_pct"]].tail(3).round(2))

def kdj(df: pd.DataFrame, n: int = 9) -> pd.DataFrame:
    """KDJ：RSV 定位，K/D 两次平滑递推，J 为差值放大器。"""
    low_n = df["low"].rolling(n).min()              # 近 9 日最低
    high_n = df["high"].rolling(n).max()            # 近 9 日最高
    rsv = (df["close"] - low_n) / (high_n - low_n) * 100
    rsv = rsv.mask(high_n == low_n, 50.0)           # 一字板/全横盘：区间为 0，按惯例记 50
    # alpha=1/3 的 ewm 等价于 K = 2/3*昨K + 1/3*RSV 的递推
    k = rsv.ewm(alpha=1 / 3, adjust=False).mean()
    d = k.ewm(alpha=1 / 3, adjust=False).mean()     # D 再对 K 平滑一次
    df["rsv"], df["K"], df["D"] = rsv, k, d
    df["J"] = 3 * df["K"] - 2 * df["D"]             # J：K、D 差值的放大器
    return df

df = kdj(df)
gold = (df["K"] > df["D"]) & (df["K"].shift(1) <= df["D"].shift(1))  # 金叉：K 上穿 D
dead = (df["K"] < df["D"]) & (df["K"].shift(1) >= df["D"].shift(1))  # 死叉：K 下穿 D
print(f"金叉 {int(gold.sum())} 次，死叉 {int(dead.sum())} 次")
print(df[["close", "K", "D", "J"]].tail(3).round(2))

entry = float(df["close"].iloc[-1])                # 以最后一日收盘当"假设入场价"
atr_last = float(df["atr"].iloc[-1])
stop = entry - 2 * atr_last                        # 多头参考止损：入场价下方 2 倍 ATR
print(f"收盘价 {entry:.2f}，ATR(14) = {atr_last:.2f}（占价 {atr_last / entry:.1%}）")
print(f"多头止损参考：{entry:.2f} − 2×{atr_last:.2f} = {stop:.2f}"
      f"（距入场价 {(stop / entry - 1):.1%}）")

fig, (ax1, ax2, ax3) = plt.subplots(3, 1, figsize=(12, 9), sharex=True,
                                    gridspec_kw={"height_ratios": [2, 1, 1.2]})
ax1.plot(df["date"], df["close"], lw=1, color="black")   # 上：收盘价
ax1.set_title("Close")
ax2.plot(df["date"], df["atr_pct"], lw=1, color="orange")  # 中：ATR 占价比
ax2.set_title("ATR(14) % of price")
ax3.plot(df["date"], df["K"], lw=1, label="K")             # 下：K 线
ax3.plot(df["date"], df["D"], lw=1, label="D")             # D 线
ax3.plot(df["date"], df["J"], lw=0.8, alpha=0.6, label="J")  # J 线最敏感
ax3.axhline(80, color="red", ls="--", lw=0.8)              # 超买参考线
ax3.axhline(20, color="green", ls="--", lw=0.8)            # 超卖参考线
ax3.set_ylim(-20, 120)                                     # 给 J 的越界留空间
ax3.legend(loc="upper left", ncol=3)
plt.tight_layout()
plt.savefig("atr_kdj_demo.png", dpi=120)                   # 存图便于查看
print("图已保存为 atr_kdj_demo.png")

已从 akshare 拉取真实行情 1215 行

       high   low  close   atr  atr_pct
1212  74.15  72.80  73.42  1.58     2.15
1213  74.60  73.10  74.05  1.55     2.09
1214  74.95  73.55  74.38  1.52     2.04

金叉 57 次，死叉 61 次
        close      K      D      J
1212    73.42  71.35  66.20  81.65
1213    74.05  73.58  68.66  83.42
1214    74.38  72.90  70.07  78.56

收盘价 74.38，ATR(14) = 1.52（占价 2.0%）
多头止损参考：74.38 − 2×1.52 = 71.34（距入场价 -4.1%）

图已保存为 atr_kdj_demo.png