86 lines
3.5 KiB
Python
86 lines
3.5 KiB
Python
"""技术指标（单一事实源）。
|
||
|
||
MVP 用纯 pandas/numpy 实现，避免 Windows 上 TA-Lib C 库的安装痛点。
|
||
算法正确（MACD = 快慢 EMA 之差），接口稳定；阶段1 在 Linux/Docker 上可换 TA-Lib，
|
||
只需保持函数签名（输入 close Series，输出指标），上层无感。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
|
||
def ema(series: pd.Series, span: int) -> pd.Series:
|
||
"""指数移动平均（adjust=False，与 TA-Lib 默认一致）。"""
|
||
return series.ewm(span=span, adjust=False).mean()
|
||
|
||
|
||
def macd(close: pd.Series, fast: int = 12, slow: int = 26, signal: int = 9) -> pd.DataFrame:
|
||
"""MACD：返回 DataFrame[DIF, DEA, HIST]。"""
|
||
dif = ema(close, fast) - ema(close, slow)
|
||
dea = ema(dif, signal)
|
||
hist = (dif - dea) * 2 # A股惯例 MACD 柱 = 2*(DIF-DEA)
|
||
return pd.DataFrame({"macd": dif, "signal": dea, "hist": hist})
|
||
|
||
|
||
def rsi(close: pd.Series, period: int = 14) -> pd.Series:
|
||
"""RSI（Wilder 平滑）。"""
|
||
delta = close.diff()
|
||
gain = delta.clip(lower=0.0)
|
||
loss = -delta.clip(upper=0.0)
|
||
avg_gain = gain.ewm(alpha=1 / period, adjust=False).mean()
|
||
avg_loss = loss.ewm(alpha=1 / period, adjust=False).mean()
|
||
rs = avg_gain / avg_loss.replace(0, np.nan)
|
||
return 100 - (100 / (1 + rs))
|
||
|
||
|
||
def kdj(high: pd.Series, low: pd.Series, close: pd.Series,
|
||
n: int = 9, m1: int = 3, m2: int = 3) -> pd.DataFrame:
|
||
"""KDJ（A股常用：RSV -> K -> D -> J）。"""
|
||
low_n = low.rolling(n, min_periods=1).min()
|
||
high_n = high.rolling(n, min_periods=1).max()
|
||
rsv = (close - low_n) / (high_n - low_n).replace(0, np.nan) * 100
|
||
k = rsv.ewm(alpha=1 / m1, adjust=False).mean()
|
||
d = k.ewm(alpha=1 / m2, adjust=False).mean()
|
||
j = 3 * k - 2 * d
|
||
return pd.DataFrame({"k": k, "d": d, "j": j})
|
||
|
||
|
||
def bollinger(close: pd.Series, period: int = 20, std: float = 2.0) -> pd.DataFrame:
|
||
ma = close.rolling(period, min_periods=1).mean()
|
||
sd = close.rolling(period, min_periods=1).std(ddof=0)
|
||
return pd.DataFrame({"mid": ma, "upper": ma + std * sd, "lower": ma - std * sd})
|
||
|
||
|
||
def ma(close: pd.Series, period: int) -> pd.Series:
|
||
return close.rolling(period, min_periods=1).mean()
|
||
|
||
|
||
def vol_ratio(volume: pd.Series, period: int = 5) -> pd.Series:
|
||
"""量比（通达信口径）：当日成交量 / 前 N 日均量（不含当日，即 REF(MA(VOL,N),1)）。
|
||
|
||
>2 约=倍量、<0.7 约=显著缩量；首根无前值、均量为 0（长期停牌）时为 NaN。
|
||
"""
|
||
base = volume.rolling(period, min_periods=1).mean().shift(1)
|
||
return volume / base.replace(0, np.nan)
|
||
|
||
|
||
def ema2(close: pd.Series, span: int = 10) -> pd.Series:
|
||
"""知行短期趋势线：EMA(EMA(C, span), span)。"""
|
||
return ema(ema(close, span), span)
|
||
|
||
|
||
def avg_ma(close: pd.Series, periods: tuple[int, ...] = (10, 20, 30, 60)) -> pd.Series:
|
||
"""知行多空线：(MA(M1)+MA(M2)+MA(M3)+MA(M4))/4。"""
|
||
return sum(ma(close, p) for p in periods) / len(periods)
|
||
|
||
|
||
def zhixing_dkx(close: pd.Series, m1: int = 14, m2: int = 28, m3: int = 57, m4: int = 114) -> pd.Series:
|
||
"""知行多空线（通达信口径）：(MA(M1)+MA(M2)+MA(M3)+MA(M4))/4，M1~M4 缺省 14/28/57/114。
|
||
|
||
MA 为严格窗口（窗口不满为 NaN，任一为空则多空线为空），与前端 zhixing.ts /
|
||
通达信原公式一致（前 m4-1 根无值）；区别于 avg_ma 的 min_periods=1 渐进口径。
|
||
"""
|
||
mas = [close.rolling(p, min_periods=p).mean() for p in (m1, m2, m3, m4)]
|
||
return sum(mas) / len(mas)
|