Add Supertrend and VWAP strategies plus a short-side example

AGENTS.md

@@ -92,7 +92,7 @@ class MyStrategy(Strategy):
 | --- | --- |
 | Price / candles | `ctx.price`, `ctx.candles`, `ctx.closes`, `ctx.high(n)`, `ctx.low(n)` |
 | Portfolio | `ctx.is_flat`, `ctx.side`, `ctx.position`, `ctx.equity`, `ctx.cash` |
-| Indicators | `ctx.sma(p)`, `ctx.ema(p)`, `ctx.rsi(p)`, `ctx.atr(p)`, `ctx.std(p)` |
+| Indicators | `ctx.sma(p)`, `ctx.ema(p)`, `ctx.rsi(p)`, `ctx.atr(p)`, `ctx.std(p)`, `ctx.macd()`, `ctx.roc(p)`, `ctx.vwap(p)`, `ctx.supertrend(p, mult)` |
 | Orders | `ctx.enter_long(size, stop_loss, take_profit, trailing_stop_pct, reason)`, `ctx.enter_short(...)`, `ctx.close(reason)` |
 
 Notes:

README.md

@@ -207,7 +207,8 @@ appears in the UI immediately. See **[AGENTS.md](AGENTS.md)** for the full
 template and the `StrategyContext` API.
 
 Bundled strategies: `sma_cross`, `ema_cross`, `macd_trend`,
-`bollinger_reversion`, `donchian_breakout`, `momentum`, `rsi_reversion`.
+`bollinger_reversion`, `donchian_breakout`, `momentum`, `rsi_reversion`,
+`supertrend`, `vwap_reversion`, `ema_cross_short` (short-side example).
 
 ## API
 

backend/app/engine/indicators.py

@@ -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:

backend/app/engine/strategies/base.py

@@ -5,7 +5,17 @@ from collections.abc import Sequence
 from dataclasses import dataclass
 from typing import Any, ClassVar, Literal, Protocol
 
-from app.engine.indicators import atr, ema, macd, roc, rolling_std, rsi, sma
+from app.engine.indicators import (
+    atr,
+    ema,
+    macd,
+    roc,
+    rolling_std,
+    rsi,
+    sma,
+    supertrend,
+    vwap,
+)
 from app.models import PositionSide
 from app.schemas import Candle, ParamSpec, StrategyOut
 
@@ -116,6 +126,15 @@ class StrategyContext:
     def roc(self, period: int = 10) -> float | None:
         return roc(self.closes, period)[-1]
 
+    def vwap(self, period: int | None = None) -> float | None:
+        return vwap(self._candles, period)[-1]
+
+    def supertrend(
+        self, period: int = 10, multiplier: float = 3.0
+    ) -> tuple[float | None, bool | None]:
+        line, up = supertrend(self._candles, period, multiplier)
+        return line[-1], up[-1]
+
     # --- order intents ------------------------------------------------
     def enter_long(
         self,

backend/app/engine/strategies/ema_cross_short.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+from typing import ClassVar
+
+from app.engine.indicators import atr, ema
+from app.engine.strategies.base import Strategy, StrategyContext, register
+from app.models import PositionSide
+
+
+@register
+class EmaCrossShort(Strategy):
+    """Short-side example: sell EMA cross-downs, cover on cross-up."""
+
+    key: ClassVar[str] = "ema_cross_short"
+    name: ClassVar[str] = "EMA Crossover (short)"
+    description: ClassVar[str] = (
+        "Open a short when the fast EMA crosses below the slow EMA and cover when "
+        "it crosses back above. Demonstrates short trades; optional ATR stop."
+    )
+    default_params: ClassVar[dict] = {
+        "fast": 9,
+        "slow": 21,
+        "atr_period": 14,
+        "atr_stop_mult": 2.0,
+        "size": 1.0,
+    }
+    param_help: ClassVar[dict] = {
+        "fast": "Fast EMA period",
+        "slow": "Slow EMA period",
+        "atr_period": "ATR lookback for the stop",
+        "atr_stop_mult": "Stop distance in ATRs (0 disables)",
+        "size": "Fraction of equity per short",
+    }
+    param_bounds: ClassVar[dict] = {
+        "fast": (2, 100),
+        "slow": (3, 300),
+        "atr_stop_mult": (0.0, 10.0),
+        "size": (0.01, 1.0),
+    }
+
+    def on_candle(self, ctx: StrategyContext) -> None:
+        fast_period = int(self.params["fast"])
+        slow_period = int(self.params["slow"])
+        if len(ctx.closes) < slow_period + 2:
+            return
+
+        fast = ema(ctx.closes, fast_period)
+        slow = ema(ctx.closes, slow_period)
+        f_prev, f_now = fast[-2], fast[-1]
+        s_prev, s_now = slow[-2], slow[-1]
+        if None in (f_prev, f_now, s_prev, s_now):
+            return
+
+        crossed_down = f_prev >= s_prev and f_now < s_now
+        crossed_up = f_prev <= s_prev and f_now > s_now
+
+        if ctx.is_flat and crossed_down:
+            stop = None
+            mult = float(self.params["atr_stop_mult"])
+            if mult > 0:
+                atr_value = atr(ctx.candles, int(self.params["atr_period"]))[-1]
+                if atr_value:
+                    stop = ctx.price + mult * atr_value
+            ctx.enter_short(
+                size=float(self.params["size"]),
+                stop_loss=stop,
+                reason="EMA cross down",
+            )
+        elif ctx.side == PositionSide.SHORT and crossed_up:
+            ctx.close(reason="EMA cross up")

backend/app/engine/strategies/supertrend.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from typing import ClassVar
+
+from app.engine.indicators import supertrend as supertrend_series
+from app.engine.strategies.base import Strategy, StrategyContext, register
+from app.models import PositionSide
+
+
+@register
+class Supertrend(Strategy):
+    """Trend following on the Supertrend indicator, long or short."""
+
+    key: ClassVar[str] = "supertrend"
+    name: ClassVar[str] = "Supertrend"
+    description: ClassVar[str] = (
+        "Go long when Supertrend flips to an uptrend and short when it flips to "
+        "a downtrend. Uses the Supertrend line as a stop."
+    )
+    default_params: ClassVar[dict] = {
+        "period": 10,
+        "multiplier": 3.0,
+        "allow_shorts": False,
+        "size": 1.0,
+    }
+    param_help: ClassVar[dict] = {
+        "period": "ATR period for the Supertrend bands",
+        "multiplier": "ATR multiplier (band width)",
+        "allow_shorts": "Also take short trades in downtrends",
+        "size": "Fraction of equity per trade",
+    }
+    param_bounds: ClassVar[dict] = {
+        "period": (2, 100),
+        "multiplier": (0.5, 10.0),
+        "size": (0.01, 1.0),
+    }
+
+    def on_candle(self, ctx: StrategyContext) -> None:
+        period = int(self.params["period"])
+        multiplier = float(self.params["multiplier"])
+        if len(ctx.candles) < period + 2:
+            return
+
+        line_series, up_series = supertrend_series(ctx.candles, period, multiplier)
+        line, up = line_series[-1], up_series[-1]
+        prev_up = up_series[-2]
+        if line is None or up is None or prev_up is None:
+            return
+
+        allow_shorts = bool(self.params.get("allow_shorts", False))
+        size = float(self.params["size"])
+
+        if ctx.side == PositionSide.LONG and not up:
+            ctx.close(reason="Supertrend turned down")
+            return
+        if ctx.side == PositionSide.SHORT and up:
+            ctx.close(reason="Supertrend turned up")
+            return
+
+        if ctx.is_flat and up and not prev_up:
+            ctx.enter_long(size=size, stop_loss=line, reason="Supertrend flipped up")
+        elif ctx.is_flat and allow_shorts and not up and prev_up:
+            ctx.enter_short(size=size, stop_loss=line, reason="Supertrend flipped down")

backend/app/engine/strategies/vwap_reversion.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from typing import ClassVar
+
+from app.engine.strategies.base import Strategy, StrategyContext, register
+from app.models import PositionSide
+
+
+@register
+class VwapReversion(Strategy):
+    """Mean reversion around VWAP, long-only."""
+
+    key: ClassVar[str] = "vwap_reversion"
+    name: ClassVar[str] = "VWAP Reversion"
+    description: ClassVar[str] = (
+        "Open long when price falls this far below a rolling VWAP and exit when "
+        "it returns to VWAP. Optional ATR stop."
+    )
+    default_params: ClassVar[dict] = {
+        "period": 20,
+        "deviation_pct": 0.005,
+        "atr_period": 14,
+        "atr_stop_mult": 2.0,
+        "size": 1.0,
+    }
+    param_help: ClassVar[dict] = {
+        "period": "Rolling VWAP window in bars",
+        "deviation_pct": "Entry distance below VWAP (0.005 = 0.5%)",
+        "atr_period": "ATR lookback for the stop",
+        "atr_stop_mult": "Stop distance in ATRs (0 disables)",
+        "size": "Fraction of equity per trade",
+    }
+    param_bounds: ClassVar[dict] = {
+        "period": (2, 200),
+        "deviation_pct": (0.0, 0.2),
+        "atr_stop_mult": (0.0, 10.0),
+        "size": (0.01, 1.0),
+    }
+
+    def on_candle(self, ctx: StrategyContext) -> None:
+        period = int(self.params["period"])
+        if len(ctx.candles) < period + 1:
+            return
+        reference = ctx.vwap(period)
+        if reference is None or reference <= 0:
+            return
+
+        threshold = reference * (1 - float(self.params["deviation_pct"]))
+
+        if ctx.is_flat and ctx.price < threshold:
+            stop = None
+            mult = float(self.params["atr_stop_mult"])
+            if mult > 0:
+                atr_value = ctx.atr(int(self.params["atr_period"]))
+                if atr_value:
+                    stop = ctx.price - mult * atr_value
+            ctx.enter_long(
+                size=float(self.params["size"]),
+                stop_loss=stop,
+                reason="price below VWAP band",
+            )
+        elif ctx.side == PositionSide.LONG and ctx.price >= reference:
+            ctx.close(reason="price returned to VWAP")

backend/tests/test_new_indicators.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+import pytest
+
+from app.engine.indicators import supertrend, vwap
+from app.engine.strategies import get_strategy
+from app.engine.strategies.base import StrategyContext
+from app.models import PositionSide
+from app.schemas import Candle
+
+
+def candles(closes: list[float], volume: float = 1.0) -> list[Candle]:
+    return [
+        Candle(ts=i, open=c, high=c * 1.001, low=c * 0.999, close=c, volume=volume)
+        for i, c in enumerate(closes)
+    ]
+
+
+def ctx(closes: list[float], volume: float = 1.0, position=None) -> StrategyContext:
+    cs = candles(closes, volume)
+    return StrategyContext(candles=cs, position=position, equity=1000.0, cash=1000.0)
+
+
+# --- indicators -----------------------------------------------------------
+def test_vwap_equals_typical_price_when_flat() -> None:
+    cs = candles([100.0, 100.0, 100.0])
+    series = vwap(cs)
+    # Typical price equals the close for symmetric high/low here only if equal;
+    # with high/low spread the VWAP is the volume-weighted typical price.
+    assert series[-1] is not None
+    assert series[-1] == pytest.approx(100.0, abs=0.5)
+
+
+def test_vwap_weights_by_volume() -> None:
+    cs = [
+        Candle(ts=0, open=100, high=100, low=100, close=100, volume=1),
+        Candle(ts=1, open=200, high=200, low=200, close=200, volume=3),
+    ]
+    series = vwap(cs)
+    assert series[-1] == pytest.approx((100 * 1 + 200 * 3) / 4)
+
+
+def test_vwap_rolling_window() -> None:
+    cs = [
+        Candle(ts=0, open=100, high=100, low=100, close=100, volume=1),
+        Candle(ts=1, open=100, high=100, low=100, close=100, volume=1),
+        Candle(ts=2, open=200, high=200, low=200, close=200, volume=1),
+    ]
+    series = vwap(cs, period=2)
+    assert series[0] is None
+    assert series[1] == pytest.approx(100.0)
+    assert series[2] == pytest.approx(150.0)
+
+
+def test_supertrend_direction_flips() -> None:
+    rising = [float(i) for i in range(1, 60)]
+    _, up_rising = supertrend(candles(rising), 10, 3.0)
+    assert up_rising[-1] is True
+
+    falling = [float(i) for i in range(60, 1, -1)]
+    _, up_falling = supertrend(candles(falling), 10, 3.0)
+    assert up_falling[-1] is False
+
+
+def test_supertrend_line_trails_price() -> None:
+    rising = [float(i) for i in range(1, 60)]
+    line, up = supertrend(candles(rising), 10, 3.0)
+    assert line[-1] is not None and up[-1] is True
+    assert line[-1] < rising[-1]  # the stop sits below price in an uptrend
+
+
+# --- strategies -----------------------------------------------------------
+def test_new_strategies_registered() -> None:
+    from app.engine.strategies import list_strategies
+
+    keys = {c.key for c in list_strategies()}
+    assert {"supertrend", "vwap_reversion", "ema_cross_short"} <= keys
+
+
+def test_ema_cross_short_opens_a_short() -> None:
+    strategy = get_strategy("ema_cross_short")({"fast": 2, "slow": 4, "atr_stop_mult": 0})
+    # Fast EMA above the slow then crossing below: mirror of the long case.
+    c = ctx([10, 10, 10, 10, 10, 10, 10, 10, 13, 11, 8])
+    strategy.on_candle(c)
+    intents = c.take_intents()
+    assert intents and intents[0].action == "enter_short"
+    assert intents[0].stop_loss is None
+
+
+def test_ema_cross_short_covers_on_cross_up() -> None:
+    from app.models import Position
+
+    strategy = get_strategy("ema_cross_short")({"fast": 2, "slow": 4, "atr_stop_mult": 0})
+    position = Position(
+        strategy_key="ema_cross_short",
+        symbol="BTC/USDT",
+        side=PositionSide.SHORT,
+        qty=1.0,
+        entry_price=100.0,
+    )
+    c = ctx([10, 10, 10, 10, 10, 10, 10, 10, 7, 9, 12], position=position)
+    strategy.on_candle(c)
+    intents = c.take_intents()
+    assert intents and intents[0].action == "close"
+
+
+def test_vwap_reversion_runs_without_error() -> None:
+    strategy = get_strategy("vwap_reversion")({"period": 5, "deviation_pct": 0.01})
+    c = ctx([100, 100, 100, 100, 100, 90])
+    strategy.on_candle(c)
+    intents = c.take_intents()
+    assert intents and intents[0].action == "enter_long"
+
+
+def test_supertrend_runs_without_error() -> None:
+    strategy = get_strategy("supertrend")({})
+    c = ctx([100 + i for i in range(40)])
+    strategy.on_candle(c)
+    # No assertion on intents (depends on the flip); just must not raise.
+    assert isinstance(c.take_intents(), list)