matitos
d92f152f9b
New williams_r indicator (in [-100, 0]) exposed as ctx.williams_r(period), and a long-only mean-reversion strategy: enter when %R is at/below the oversold level, exit when it reaches the overbought level, with an optional ATR stop. 11 strategies total. Adds indicator and strategy tests; smoke-tested on live BTC/USDT 5m (backtest + parameter sweep).
256 lines
8.2 KiB
Python
256 lines
8.2 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 macd(
|
|
values: Sequence[float],
|
|
fast: int = 12,
|
|
slow: int = 26,
|
|
signal: int = 9,
|
|
) -> tuple[list[Number], list[Number], list[Number]]:
|
|
"""Return (macd line, signal line, histogram), each aligned with ``values``."""
|
|
ema_fast = ema(values, fast)
|
|
ema_slow = ema(values, slow)
|
|
macd_line: list[Number] = [
|
|
(f - s) if f is not None and s is not None else None
|
|
for f, s in zip(ema_fast, ema_slow, strict=True)
|
|
]
|
|
signal_line: list[Number] = [None] * len(values)
|
|
start = next((i for i, value in enumerate(macd_line) if value is not None), None)
|
|
if start is not None:
|
|
valid = [value for value in macd_line[start:] if value is not None]
|
|
for offset, value in enumerate(ema(valid, signal)):
|
|
signal_line[start + offset] = value
|
|
histogram: list[Number] = [
|
|
(m - s) if m is not None and s is not None else None
|
|
for m, s in zip(macd_line, signal_line, strict=True)
|
|
]
|
|
return macd_line, signal_line, histogram
|
|
|
|
|
|
def roc(values: Sequence[float], period: int = 10) -> list[Number]:
|
|
"""Rate of change over ``period`` bars, as a fraction (0.02 = +2%)."""
|
|
if period <= 0:
|
|
raise ValueError("period must be positive")
|
|
out: list[Number] = [None] * len(values)
|
|
for i in range(period, len(values)):
|
|
previous = values[i - period]
|
|
if previous:
|
|
out[i] = (values[i] - previous) / previous
|
|
return out
|
|
|
|
|
|
def williams_r(candles: Sequence[Candle], period: int = 14) -> list[Number]:
|
|
"""Williams %R, in [-100, 0]. Below -80 is oversold, above -20 overbought."""
|
|
if period <= 0:
|
|
raise ValueError("period must be positive")
|
|
out: list[Number] = [None] * len(candles)
|
|
for i in range(period - 1, len(candles)):
|
|
window = candles[i - period + 1 : i + 1]
|
|
highest = max(candle.high for candle in window)
|
|
lowest = min(candle.low for candle in window)
|
|
if highest == lowest:
|
|
out[i] = -50.0
|
|
else:
|
|
out[i] = (highest - candles[i].close) / (highest - lowest) * -100
|
|
return out
|
|
|
|
|
|
def typical_price(candle: Candle) -> float:
|
|
return (candle.high + candle.low + candle.close) / 3
|
|
|
|
|
|
def vwap(candles: Sequence[Candle], period: int | None = None) -> list[Number]:
|
|
"""Volume-weighted average price, rolling over ``period`` bars.
|
|
|
|
With ``period=None`` it is cumulative from the first candle.
|
|
"""
|
|
out: list[Number] = [None] * len(candles)
|
|
pv_sum = 0.0
|
|
volume_sum = 0.0
|
|
window: list[tuple[float, float]] = []
|
|
for i, candle in enumerate(candles):
|
|
price = typical_price(candle)
|
|
volume = candle.volume
|
|
pv_sum += price * volume
|
|
volume_sum += volume
|
|
if period is not None:
|
|
window.append((price, volume))
|
|
if len(window) > period:
|
|
old_price, old_volume = window.pop(0)
|
|
pv_sum -= old_price * old_volume
|
|
volume_sum -= old_volume
|
|
if len(window) < period:
|
|
continue
|
|
if volume_sum:
|
|
out[i] = pv_sum / volume_sum
|
|
return out
|
|
|
|
|
|
def supertrend(
|
|
candles: Sequence[Candle],
|
|
period: int = 10,
|
|
multiplier: float = 3.0,
|
|
) -> tuple[list[float | None], list[bool | None]]:
|
|
"""Supertrend line and trend direction (True = uptrend) per candle."""
|
|
atr_series = atr(candles, period)
|
|
line: list[float | None] = [None] * len(candles)
|
|
up: list[bool | None] = [None] * len(candles)
|
|
|
|
final_upper: float | None = None
|
|
final_lower: float | None = None
|
|
prev_close: float | None = None
|
|
direction_up = True
|
|
|
|
for i, candle in enumerate(candles):
|
|
atr_value = atr_series[i]
|
|
if atr_value is None:
|
|
continue
|
|
midpoint = (candle.high + candle.low) / 2
|
|
basic_upper = midpoint + multiplier * atr_value
|
|
basic_lower = midpoint - multiplier * atr_value
|
|
|
|
if final_upper is None or final_lower is None:
|
|
final_upper, final_lower = basic_upper, basic_lower
|
|
direction_up = candle.close >= midpoint
|
|
else:
|
|
assert prev_close is not None
|
|
final_upper = (
|
|
basic_upper
|
|
if basic_upper < final_upper or prev_close > final_upper
|
|
else final_upper
|
|
)
|
|
final_lower = (
|
|
basic_lower
|
|
if basic_lower > final_lower or prev_close < final_lower
|
|
else final_lower
|
|
)
|
|
if direction_up:
|
|
direction_up = candle.close >= final_lower
|
|
else:
|
|
direction_up = candle.close > final_upper
|
|
|
|
prev_close = candle.close
|
|
line[i] = final_lower if direction_up else final_upper
|
|
up[i] = direction_up
|
|
|
|
return line, up
|
|
|
|
|
|
def last(series: Sequence[Number]) -> Number:
|
|
for value in reversed(series):
|
|
if value is not None:
|
|
return value
|
|
return None
|