Initial TradeBot implementation

crypto_spot_bot/indicators.py

@@ -0,0 +1,106 @@
+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)
+    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.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