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).
88 lines
2.9 KiB
Python
88 lines
2.9 KiB
Python
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
|