41 lines
1.8 KiB
Python
41 lines
1.8 KiB
Python
"""
|
|
Pure-function technical indicators. All operate on pandas Series/DataFrame
|
|
and return pandas objects so NaN propagation is handled automatically.
|
|
"""
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
|
|
def ema(series: pd.Series, period: int) -> pd.Series:
|
|
return series.ewm(span=period, min_periods=period, adjust=False).mean()
|
|
|
|
|
|
def rsi(close: pd.Series, period: int = 14) -> pd.Series:
|
|
delta = close.diff()
|
|
gain = delta.clip(lower=0)
|
|
loss = (-delta).clip(lower=0)
|
|
avg_g = gain.ewm(alpha=1 / period, min_periods=period, adjust=False).mean()
|
|
avg_l = loss.ewm(alpha=1 / period, min_periods=period, adjust=False).mean()
|
|
rs = avg_g / avg_l.replace(0, np.inf)
|
|
return 100 - (100 / (1 + rs))
|
|
|
|
|
|
def atr(df: pd.DataFrame, period: int = 14) -> pd.Series:
|
|
hi, lo, pc = df["high"], df["low"], df["close"].shift(1)
|
|
tr = pd.concat([(hi - lo), (hi - pc).abs(), (lo - pc).abs()], axis=1).max(axis=1)
|
|
return tr.ewm(alpha=1 / period, min_periods=period, adjust=False).mean()
|
|
|
|
|
|
def adx(df: pd.DataFrame, period: int = 14) -> pd.Series:
|
|
"""Average Directional Index."""
|
|
hi, lo, pc = df["high"], df["low"], df["close"].shift(1)
|
|
tr = pd.concat([(hi - lo), (hi - pc).abs(), (lo - pc).abs()], axis=1).max(axis=1)
|
|
dm_pos = (hi - hi.shift(1)).clip(lower=0).where((hi - hi.shift(1)) > (lo.shift(1) - lo), 0)
|
|
dm_neg = (lo.shift(1) - lo).clip(lower=0).where((lo.shift(1) - lo) > (hi - hi.shift(1)), 0)
|
|
atr_s = tr.ewm(alpha=1 / period, min_periods=period, adjust=False).mean()
|
|
di_pos = 100 * dm_pos.ewm(alpha=1 / period, min_periods=period, adjust=False).mean() / atr_s
|
|
di_neg = 100 * dm_neg.ewm(alpha=1 / period, min_periods=period, adjust=False).mean() / atr_s
|
|
dx = (100 * (di_pos - di_neg).abs() / (di_pos + di_neg).replace(0, np.nan))
|
|
return dx.ewm(alpha=1 / period, min_periods=period, adjust=False).mean()
|