diff --git a/AGENTS.md b/AGENTS.md index b8c972b..6ed02b3 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)`, `ctx.macd()`, `ctx.roc(p)`, `ctx.vwap(p)`, `ctx.supertrend(p, mult)` | +| 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)`, `ctx.williams_r(p)` | | 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 e5bca00..b118bd4 100644 --- a/README.md +++ b/README.md @@ -208,7 +208,8 @@ template and the `StrategyContext` API. Bundled strategies: `sma_cross`, `ema_cross`, `macd_trend`, `bollinger_reversion`, `donchian_breakout`, `momentum`, `rsi_reversion`, -`supertrend`, `vwap_reversion`, `ema_cross_short` (short-side example). +`supertrend`, `vwap_reversion`, `ema_cross_short` (short-side example), +`williams_r`. ## API diff --git a/backend/app/engine/indicators.py b/backend/app/engine/indicators.py index 1a5722c..854be7d 100644 --- a/backend/app/engine/indicators.py +++ b/backend/app/engine/indicators.py @@ -151,6 +151,22 @@ def roc(values: Sequence[float], period: int = 10) -> list[Number]: 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 diff --git a/backend/app/engine/strategies/base.py b/backend/app/engine/strategies/base.py index fcab39b..b6d35c6 100644 --- a/backend/app/engine/strategies/base.py +++ b/backend/app/engine/strategies/base.py @@ -15,6 +15,7 @@ from app.engine.indicators import ( sma, supertrend, vwap, + williams_r, ) from app.models import PositionSide from app.schemas import Candle, ParamSpec, StrategyOut @@ -129,6 +130,9 @@ class StrategyContext: def vwap(self, period: int | None = None) -> float | None: return vwap(self._candles, period)[-1] + def williams_r(self, period: int = 14) -> float | None: + return williams_r(self._candles, period)[-1] + def supertrend( self, period: int = 10, multiplier: float = 3.0 ) -> tuple[float | None, bool | None]: diff --git a/backend/app/engine/strategies/williams_r.py b/backend/app/engine/strategies/williams_r.py new file mode 100644 index 0000000..4a89270 --- /dev/null +++ b/backend/app/engine/strategies/williams_r.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +from typing import ClassVar + +from app.engine.strategies.base import Strategy, StrategyContext, register +from app.models import PositionSide + + +@register +class WilliamsR(Strategy): + """Williams %R mean reversion, long-only.""" + + key: ClassVar[str] = "williams_r" + name: ClassVar[str] = "Williams %R" + description: ClassVar[str] = ( + "Enter long when Williams %R is oversold (at or below the oversold " + "level) and exit when it becomes overbought. Optional ATR stop." + ) + default_params: ClassVar[dict] = { + "period": 14, + "oversold": -80.0, + "overbought": -20.0, + "atr_period": 14, + "atr_stop_mult": 2.0, + "size": 1.0, + } + param_help: ClassVar[dict] = { + "period": "Williams %R lookback", + "oversold": "Oversold level (e.g. -80)", + "overbought": "Overbought level (e.g. -20)", + "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, 100), + "oversold": (-100.0, -50.0), + "overbought": (-50.0, 0.0), + "atr_stop_mult": (0.0, 10.0), + "size": (0.01, 1.0), + } + + def on_candle(self, ctx: StrategyContext) -> None: + value = ctx.williams_r(int(self.params["period"])) + if value is None: + return + + if ctx.is_flat and value <= float(self.params["oversold"]): + 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=f"Williams %R {value:.1f} oversold", + ) + elif ctx.side == PositionSide.LONG and value >= float(self.params["overbought"]): + ctx.close(reason=f"Williams %R {value:.1f} overbought") diff --git a/backend/tests/test_williams_r.py b/backend/tests/test_williams_r.py new file mode 100644 index 0000000..eb06124 --- /dev/null +++ b/backend/tests/test_williams_r.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +import pytest + +from app.engine.indicators import williams_r +from app.engine.strategies import get_strategy +from app.engine.strategies.base import StrategyContext +from app.models import Position, PositionSide +from app.schemas import Candle + + +def candles(closes: list[float]) -> list[Candle]: + return [ + Candle(ts=i, open=c, high=c * 1.001, low=c * 0.999, close=c, volume=1.0) + for i, c in enumerate(closes) + ] + + +def ctx(closes: list[float], position=None) -> StrategyContext: + return StrategyContext(candles=candles(closes), position=position, equity=1000.0, cash=1000.0) + + +# --- indicator ------------------------------------------------------------ +def test_williams_r_bounds() -> None: + series = williams_r(candles([100, 102, 98, 101, 99, 103, 97, 100]), 5) + for value in series: + if value is not None: + assert -100.0 <= value <= 0.0 + + +def test_williams_r_rising_is_overbought() -> None: + rising = [float(i) for i in range(1, 30)] + value = williams_r(candles(rising), 14)[-1] + assert value is not None and value > -50 # near the top of the range + + +def test_williams_r_falling_is_oversold() -> None: + falling = [float(i) for i in range(30, 1, -1)] + value = williams_r(candles(falling), 14)[-1] + assert value is not None and value < -50 # near the bottom of the range + + +def test_williams_r_flat_series_is_neutral() -> None: + flat = [100.0] * 20 + assert williams_r(candles(flat), 14)[-1] == pytest.approx(-50.0) + + +# --- strategy ------------------------------------------------------------- +def test_williams_r_registered() -> None: + from app.engine.strategies import list_strategies + + assert "williams_r" in {c.key for c in list_strategies()} + + +def test_enters_long_when_oversold() -> None: + strategy = get_strategy("williams_r")({"period": 5, "oversold": -80, "atr_stop_mult": 0}) + falling = [100 - i for i in range(15)] + c = ctx([float(x) for x in falling]) + strategy.on_candle(c) + intents = c.take_intents() + assert intents and intents[0].action == "enter_long" + + +def test_exits_when_overbought() -> None: + strategy = get_strategy("williams_r")({"period": 5, "overbought": -20, "atr_stop_mult": 0}) + position = Position( + strategy_key="williams_r", + symbol="BTC/USDT", + side=PositionSide.LONG, + qty=1.0, + entry_price=100.0, + ) + rising = [100 + i for i in range(15)] + c = ctx([float(x) for x in rising], position=position) + strategy.on_candle(c) + intents = c.take_intents() + assert intents and intents[0].action == "close" + + +def test_atr_stop_is_set_when_enabled() -> None: + strategy = get_strategy("williams_r")({"period": 5, "oversold": -80, "atr_stop_mult": 2.0}) + falling = [float(100 - i) for i in range(15)] + c = ctx(falling) + strategy.on_candle(c) + intents = c.take_intents() + assert intents and intents[0].action == "enter_long" + assert intents[0].stop_loss is not None