Codex
141 lines
4.8 KiB
Python
141 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
from statistics import fmean
|
|
|
|
from crypto_spot_bot.models import Candle
|
|
|
|
|
|
def add_indicators(candles: list[Candle]) -> list[Candle]:
|
|
closes = [c.close for c in candles]
|
|
highs = [c.high for c in candles]
|
|
lows = [c.low for c in candles]
|
|
volumes = [c.volume for c in candles]
|
|
ema20 = _ema(closes, 20)
|
|
ema50 = _ema(closes, 50)
|
|
ema200 = _ema(closes, 200)
|
|
rsi14 = _rsi(closes, 14)
|
|
atr14 = _atr(highs, lows, closes, 14)
|
|
macd, macd_signal, macd_hist = _macd(closes)
|
|
volume_ma20 = _sma(volumes, 20)
|
|
for index, candle in enumerate(candles):
|
|
candle.ema_20 = ema20[index]
|
|
candle.ema_50 = ema50[index]
|
|
candle.ema_200 = ema200[index]
|
|
candle.rsi_14 = rsi14[index]
|
|
candle.atr_14 = atr14[index]
|
|
candle.macd = macd[index]
|
|
candle.macd_signal = macd_signal[index]
|
|
candle.macd_hist = macd_hist[index]
|
|
candle.volume_ma_20 = volume_ma20[index]
|
|
return candles
|
|
|
|
|
|
def _ema(values: list[float], period: int) -> list[float | None]:
|
|
if not values:
|
|
return []
|
|
result: list[float | None] = [None] * len(values)
|
|
if len(values) < period:
|
|
return result
|
|
seed = fmean(values[:period])
|
|
result[period - 1] = seed
|
|
multiplier = 2 / (period + 1)
|
|
previous = seed
|
|
for index in range(period, len(values)):
|
|
previous = (values[index] - previous) * multiplier + previous
|
|
result[index] = previous
|
|
return result
|
|
|
|
|
|
def _sma(values: list[float], period: int) -> list[float | None]:
|
|
result: list[float | None] = []
|
|
for index in range(len(values)):
|
|
if index + 1 < period:
|
|
result.append(None)
|
|
else:
|
|
result.append(fmean(values[index + 1 - period : index + 1]))
|
|
return result
|
|
|
|
|
|
def _rsi(closes: list[float], period: int) -> list[float | None]:
|
|
result: list[float | None] = [None] * len(closes)
|
|
if len(closes) <= period:
|
|
return result
|
|
gains: list[float] = []
|
|
losses: list[float] = []
|
|
for index in range(1, period + 1):
|
|
delta = closes[index] - closes[index - 1]
|
|
gains.append(max(delta, 0.0))
|
|
losses.append(abs(min(delta, 0.0)))
|
|
avg_gain = fmean(gains)
|
|
avg_loss = fmean(losses)
|
|
result[period] = _rsi_value(avg_gain, avg_loss)
|
|
for index in range(period + 1, len(closes)):
|
|
delta = closes[index] - closes[index - 1]
|
|
gain = max(delta, 0.0)
|
|
loss = abs(min(delta, 0.0))
|
|
avg_gain = ((avg_gain * (period - 1)) + gain) / period
|
|
avg_loss = ((avg_loss * (period - 1)) + loss) / period
|
|
result[index] = _rsi_value(avg_gain, avg_loss)
|
|
return result
|
|
|
|
|
|
def _rsi_value(avg_gain: float, avg_loss: float) -> float:
|
|
if avg_loss == 0:
|
|
return 100.0
|
|
rs = avg_gain / avg_loss
|
|
return 100 - (100 / (1 + rs))
|
|
|
|
|
|
def _atr(highs: list[float], lows: list[float], closes: list[float], period: int) -> list[float | None]:
|
|
result: list[float | None] = [None] * len(closes)
|
|
true_ranges: list[float] = []
|
|
for index in range(len(closes)):
|
|
if index == 0:
|
|
true_ranges.append(highs[index] - lows[index])
|
|
else:
|
|
true_ranges.append(
|
|
max(
|
|
highs[index] - lows[index],
|
|
abs(highs[index] - closes[index - 1]),
|
|
abs(lows[index] - closes[index - 1]),
|
|
)
|
|
)
|
|
if len(true_ranges) < period:
|
|
return result
|
|
atr = fmean(true_ranges[:period])
|
|
result[period - 1] = atr
|
|
for index in range(period, len(true_ranges)):
|
|
atr = ((atr * (period - 1)) + true_ranges[index]) / period
|
|
result[index] = atr
|
|
return result
|
|
|
|
|
|
def _macd(
|
|
closes: list[float],
|
|
fast_period: int = 12,
|
|
slow_period: int = 26,
|
|
signal_period: int = 9,
|
|
) -> tuple[list[float | None], list[float | None], list[float | None]]:
|
|
ema_fast = _ema(closes, fast_period)
|
|
ema_slow = _ema(closes, slow_period)
|
|
macd_line: list[float | None] = []
|
|
compact_macd: list[float] = []
|
|
compact_indexes: list[int] = []
|
|
for index, (fast, slow) in enumerate(zip(ema_fast, ema_slow)):
|
|
value = None if fast is None or slow is None else fast - slow
|
|
macd_line.append(value)
|
|
if value is not None:
|
|
compact_macd.append(value)
|
|
compact_indexes.append(index)
|
|
|
|
signal_line: list[float | None] = [None] * len(closes)
|
|
hist: list[float | None] = [None] * len(closes)
|
|
compact_signal = _ema(compact_macd, signal_period)
|
|
for compact_index, original_index in enumerate(compact_indexes):
|
|
signal = compact_signal[compact_index]
|
|
macd_value = macd_line[original_index]
|
|
signal_line[original_index] = signal
|
|
if macd_value is not None and signal is not None:
|
|
hist[original_index] = macd_value - signal
|
|
return macd_line, signal_line, hist
|