matitos
9c3bee87ac
FastAPI backend with a single-strategy async engine, paper and Binance brokers, pure-python indicators, a backtester that shares the engine's portfolio math, and a Vue dashboard with a live WebSocket feed. Adds the Gitea Actions CI/CD workflow that deploys to Traefik via docker_default.
121 lines
3.5 KiB
Python
121 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Sequence
|
|
|
|
from app.schemas import Candle
|
|
|
|
Number = float | None
|
|
|
|
|
|
def sma(values: Sequence[float], period: int) -> list[Number]:
|
|
if period <= 0:
|
|
raise ValueError("period must be positive")
|
|
out: list[Number] = [None] * len(values)
|
|
window_sum = 0.0
|
|
for i, value in enumerate(values):
|
|
window_sum += value
|
|
if i >= period:
|
|
window_sum -= values[i - period]
|
|
if i >= period - 1:
|
|
out[i] = window_sum / period
|
|
return out
|
|
|
|
|
|
def ema(values: Sequence[float], period: int) -> list[Number]:
|
|
if period <= 0:
|
|
raise ValueError("period must be positive")
|
|
out: list[Number] = [None] * len(values)
|
|
if len(values) < period:
|
|
return out
|
|
multiplier = 2 / (period + 1)
|
|
seed = sum(values[:period]) / period
|
|
out[period - 1] = seed
|
|
prev = seed
|
|
for i in range(period, len(values)):
|
|
prev = (values[i] - prev) * multiplier + prev
|
|
out[i] = prev
|
|
return out
|
|
|
|
|
|
def rsi(values: Sequence[float], period: int = 14) -> list[Number]:
|
|
if period <= 0:
|
|
raise ValueError("period must be positive")
|
|
out: list[Number] = [None] * len(values)
|
|
if len(values) <= period:
|
|
return out
|
|
gains = 0.0
|
|
losses = 0.0
|
|
for i in range(1, period + 1):
|
|
change = values[i] - values[i - 1]
|
|
gains += max(change, 0.0)
|
|
losses += max(-change, 0.0)
|
|
avg_gain = gains / period
|
|
avg_loss = losses / period
|
|
out[period] = _rsi_value(avg_gain, avg_loss)
|
|
for i in range(period + 1, len(values)):
|
|
change = values[i] - values[i - 1]
|
|
gain = max(change, 0.0)
|
|
loss = max(-change, 0.0)
|
|
avg_gain = (avg_gain * (period - 1) + gain) / period
|
|
avg_loss = (avg_loss * (period - 1) + loss) / period
|
|
out[i] = _rsi_value(avg_gain, avg_loss)
|
|
return out
|
|
|
|
|
|
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 true_range(candles: Sequence[Candle]) -> list[float]:
|
|
out: list[float] = []
|
|
for i, candle in enumerate(candles):
|
|
if i == 0:
|
|
out.append(candle.high - candle.low)
|
|
continue
|
|
prev_close = candles[i - 1].close
|
|
out.append(
|
|
max(
|
|
candle.high - candle.low,
|
|
abs(candle.high - prev_close),
|
|
abs(candle.low - prev_close),
|
|
)
|
|
)
|
|
return out
|
|
|
|
|
|
def atr(candles: Sequence[Candle], period: int = 14) -> list[Number]:
|
|
if period <= 0:
|
|
raise ValueError("period must be positive")
|
|
tr = true_range(candles)
|
|
out: list[Number] = [None] * len(tr)
|
|
if len(tr) < period:
|
|
return out
|
|
prev = sum(tr[:period]) / period
|
|
out[period - 1] = prev
|
|
for i in range(period, len(tr)):
|
|
prev = (prev * (period - 1) + tr[i]) / period
|
|
out[i] = prev
|
|
return out
|
|
|
|
|
|
def rolling_std(values: Sequence[float], period: int) -> list[Number]:
|
|
if period <= 0:
|
|
raise ValueError("period must be positive")
|
|
out: list[Number] = [None] * len(values)
|
|
for i in range(period - 1, len(values)):
|
|
window = values[i - period + 1 : i + 1]
|
|
mean = sum(window) / period
|
|
variance = sum((x - mean) ** 2 for x in window) / period
|
|
out[i] = variance**0.5
|
|
return out
|
|
|
|
|
|
def last(series: Sequence[Number]) -> Number:
|
|
for value in reversed(series):
|
|
if value is not None:
|
|
return value
|
|
return None
|