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.
64 lines
2.1 KiB
Python
64 lines
2.1 KiB
Python
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")
|