diff --git a/AGENTS.md b/AGENTS.md index 0591946..2d9568c 100644 --- a/AGENTS.md +++ b/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: diff --git a/README.md b/README.md index dbf5196..ee79d5f 100644 --- a/README.md +++ b/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 diff --git a/backend/app/engine/indicators.py b/backend/app/engine/indicators.py index 9c03274..1a5722c 100644 --- a/backend/app/engine/indicators.py +++ b/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: diff --git a/backend/app/engine/strategies/base.py b/backend/app/engine/strategies/base.py index 3c69264..fcab39b 100644 --- a/backend/app/engine/strategies/base.py +++ b/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, diff --git a/backend/app/engine/strategies/ema_cross_short.py b/backend/app/engine/strategies/ema_cross_short.py new file mode 100644 index 0000000..f231623 --- /dev/null +++ b/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") diff --git a/backend/app/engine/strategies/supertrend.py b/backend/app/engine/strategies/supertrend.py new file mode 100644 index 0000000..b97830d --- /dev/null +++ b/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") diff --git a/backend/app/engine/strategies/vwap_reversion.py b/backend/app/engine/strategies/vwap_reversion.py new file mode 100644 index 0000000..aff3479 --- /dev/null +++ b/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") diff --git a/backend/tests/test_new_indicators.py b/backend/tests/test_new_indicators.py new file mode 100644 index 0000000..60aad1c --- /dev/null +++ b/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)