Add Supertrend and VWAP strategies plus a short-side example

- New indicators: vwap (rolling/cumulative) and supertrend (line + direction),
  exposed on StrategyContext.
- New strategies: supertrend (flip-based, optional shorts), vwap_reversion
  (mean reversion below a rolling VWAP) and ema_cross_short (short-side
  example, mirroring the long EMA cross). 10 strategies total.
- Adds indicator and strategy tests; smoke-tested all three on live BTC/USDT
  5m data.
This commit is contained in:
matitos committed 2026-09-22 14:14:12 +00:00
1 parent c1a7fb9c6b
commit d1f66c95c4
8 files changed
+420 -3

No files matched your search

+81
View File
@@ -151,6 +151,87 @@ def roc(values: Sequence[float], period: int = 10) -> list[Number]:
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: