57 lines
2.2 KiB
Python
57 lines
2.2 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()
|