matitos
d1f66c95c4
- 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.
121 lines
4.2 KiB
Python
121 lines
4.2 KiB
Python
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)
|