140 lines
4.1 KiB
Python
140 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
|
|
def sma(values: list[float], period: int) -> float | None:
|
|
if len(values) < period:
|
|
return None
|
|
return sum(values[-period:]) / period
|
|
|
|
|
|
def rsi(closes: list[float], period: int = 14) -> float | None:
|
|
if len(closes) < period + 1:
|
|
return None
|
|
gains: list[float] = []
|
|
losses: list[float] = []
|
|
for i in range(1, len(closes)):
|
|
delta = closes[i] - closes[i - 1]
|
|
gains.append(max(delta, 0.0))
|
|
losses.append(-min(delta, 0.0))
|
|
avg_gain = sum(gains[:period]) / period
|
|
avg_loss = sum(losses[:period]) / period
|
|
for i in range(period, len(gains)):
|
|
avg_gain = (avg_gain * (period - 1) + gains[i]) / period
|
|
avg_loss = (avg_loss * (period - 1) + losses[i]) / period
|
|
if avg_loss == 0:
|
|
return 100.0
|
|
rs = avg_gain / avg_loss
|
|
return 100.0 - (100.0 / (1.0 + rs))
|
|
|
|
|
|
def _ema_series(values: list[float], period: int) -> list[float]:
|
|
if len(values) < period:
|
|
return []
|
|
k = 2.0 / (period + 1)
|
|
seed = sum(values[:period]) / period
|
|
out = [seed]
|
|
for v in values[period:]:
|
|
out.append(out[-1] + k * (v - out[-1]))
|
|
return out
|
|
|
|
|
|
def macd(
|
|
closes: list[float],
|
|
fast: int = 12,
|
|
slow: int = 26,
|
|
signal: int = 9,
|
|
) -> dict[str, float | None]:
|
|
nothing: dict[str, float | None] = {"macd": None, "signal": None, "hist": None}
|
|
if len(closes) < slow + signal:
|
|
return nothing
|
|
ema_fast = _ema_series(closes, fast)
|
|
ema_slow = _ema_series(closes, slow)
|
|
offset = slow - fast
|
|
aligned_fast = ema_fast[offset:]
|
|
macd_line = [f - s for f, s in zip(aligned_fast, ema_slow, strict=False)]
|
|
if len(macd_line) < signal:
|
|
return nothing
|
|
signal_line = _ema_series(macd_line, signal)
|
|
if not signal_line:
|
|
return nothing
|
|
last_macd = macd_line[-1]
|
|
last_sig = signal_line[-1]
|
|
return {
|
|
"macd": last_macd,
|
|
"signal": last_sig,
|
|
"hist": last_macd - last_sig,
|
|
}
|
|
|
|
|
|
def atr(
|
|
highs: list[float],
|
|
lows: list[float],
|
|
closes: list[float],
|
|
period: int = 14,
|
|
) -> float | None:
|
|
if len(closes) < period + 1:
|
|
return None
|
|
trs: list[float] = []
|
|
for i in range(1, len(closes)):
|
|
tr = max(
|
|
highs[i] - lows[i],
|
|
abs(highs[i] - closes[i - 1]),
|
|
abs(lows[i] - closes[i - 1]),
|
|
)
|
|
trs.append(tr)
|
|
if len(trs) < period:
|
|
return None
|
|
avg = sum(trs[:period]) / period
|
|
for i in range(period, len(trs)):
|
|
avg = (avg * (period - 1) + trs[i]) / period
|
|
return avg
|
|
|
|
|
|
def adx(
|
|
highs: list[float],
|
|
lows: list[float],
|
|
closes: list[float],
|
|
period: int = 14,
|
|
) -> dict[str, float | None]:
|
|
nothing: dict[str, float | None] = {"adx": None, "+di": None, "-di": None}
|
|
if len(closes) < 2 * period + 1:
|
|
return nothing
|
|
trs: list[float] = []
|
|
plus_dms: list[float] = []
|
|
minus_dms: list[float] = []
|
|
for i in range(1, len(closes)):
|
|
tr = max(
|
|
highs[i] - lows[i],
|
|
abs(highs[i] - closes[i - 1]),
|
|
abs(lows[i] - closes[i - 1]),
|
|
)
|
|
up = highs[i] - highs[i - 1]
|
|
dn = lows[i - 1] - lows[i]
|
|
plus_dm = up if (up > dn and up > 0) else 0.0
|
|
minus_dm = dn if (dn > up and dn > 0) else 0.0
|
|
trs.append(tr)
|
|
plus_dms.append(plus_dm)
|
|
minus_dms.append(minus_dm)
|
|
|
|
atr_s = sum(trs[:period])
|
|
pdm_s = sum(plus_dms[:period])
|
|
mdm_s = sum(minus_dms[:period])
|
|
dxs: list[float] = []
|
|
pdi = mdi = 0.0
|
|
for i in range(period, len(trs)):
|
|
atr_s = atr_s - atr_s / period + trs[i]
|
|
pdm_s = pdm_s - pdm_s / period + plus_dms[i]
|
|
mdm_s = mdm_s - mdm_s / period + minus_dms[i]
|
|
pdi = 100.0 * pdm_s / atr_s if atr_s else 0.0
|
|
mdi = 100.0 * mdm_s / atr_s if atr_s else 0.0
|
|
s = pdi + mdi
|
|
dx = 100.0 * abs(pdi - mdi) / s if s else 0.0
|
|
dxs.append(dx)
|
|
|
|
if len(dxs) < period:
|
|
return nothing
|
|
adx_val = sum(dxs[:period]) / period
|
|
for i in range(period, len(dxs)):
|
|
adx_val = (adx_val * (period - 1) + dxs[i]) / period
|
|
return {"adx": adx_val, "+di": pdi, "-di": mdi}
|