matitos
9c3bee87ac
FastAPI backend with a single-strategy async engine, paper and Binance brokers, pure-python indicators, a backtester that shares the engine's portfolio math, and a Vue dashboard with a live WebSocket feed. Adds the Gitea Actions CI/CD workflow that deploys to Traefik via docker_default.
66 lines
2.4 KiB
Python
66 lines
2.4 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 RsiReversion(Strategy):
|
|
"""Buy oversold RSI, exit on overbought, with an ATR-based stop."""
|
|
|
|
key: ClassVar[str] = "rsi_reversion"
|
|
name: ClassVar[str] = "RSI Mean Reversion"
|
|
description: ClassVar[str] = (
|
|
"Enter long when RSI drops below the oversold level and exit when it rises "
|
|
"above the overbought level. Places an ATR-based stop loss and take profit."
|
|
)
|
|
default_params: ClassVar[dict] = {
|
|
"period": 14,
|
|
"oversold": 30.0,
|
|
"overbought": 70.0,
|
|
"atr_period": 14,
|
|
"atr_stop_mult": 2.0,
|
|
"atr_target_mult": 3.0,
|
|
"size": 1.0,
|
|
}
|
|
param_help: ClassVar[dict] = {
|
|
"period": "RSI lookback period",
|
|
"oversold": "RSI level below which to buy",
|
|
"overbought": "RSI level above which to exit",
|
|
"atr_period": "ATR lookback for stop sizing",
|
|
"atr_stop_mult": "Stop distance as a multiple of ATR",
|
|
"atr_target_mult": "Take-profit distance as a multiple of ATR",
|
|
"size": "Fraction of equity to allocate per trade",
|
|
}
|
|
param_bounds: ClassVar[dict] = {
|
|
"period": (2, 100),
|
|
"oversold": (1, 50),
|
|
"overbought": (50, 99),
|
|
"atr_stop_mult": (0.5, 10.0),
|
|
"atr_target_mult": (0.5, 20.0),
|
|
"size": (0.01, 1.0),
|
|
}
|
|
|
|
def on_candle(self, ctx: StrategyContext) -> None:
|
|
period = int(self.params["period"])
|
|
current = ctx.rsi(period)
|
|
if current is None:
|
|
return
|
|
|
|
if ctx.is_flat and current < float(self.params["oversold"]):
|
|
stop = target = None
|
|
atr_value = ctx.atr(int(self.params["atr_period"]))
|
|
if atr_value:
|
|
stop = ctx.price - float(self.params["atr_stop_mult"]) * atr_value
|
|
target = ctx.price + float(self.params["atr_target_mult"]) * atr_value
|
|
ctx.enter_long(
|
|
size=float(self.params["size"]),
|
|
stop_loss=stop,
|
|
take_profit=target,
|
|
reason=f"RSI {current:.1f} below oversold",
|
|
)
|
|
elif ctx.side == PositionSide.LONG and current > float(self.params["overbought"]):
|
|
ctx.close(reason=f"RSI {current:.1f} above overbought")
|