from __future__ import annotations from typing import ClassVar from app.engine.indicators import atr, ema from app.engine.strategies.base import Strategy, StrategyContext, register from app.models import PositionSide @register class EmaCrossShort(Strategy): """Short-side example: sell EMA cross-downs, cover on cross-up.""" key: ClassVar[str] = "ema_cross_short" name: ClassVar[str] = "EMA Crossover (short)" description: ClassVar[str] = ( "Open a short when the fast EMA crosses below the slow EMA and cover when " "it crosses back above. Demonstrates short trades; optional ATR stop." ) default_params: ClassVar[dict] = { "fast": 9, "slow": 21, "atr_period": 14, "atr_stop_mult": 2.0, "size": 1.0, } param_help: ClassVar[dict] = { "fast": "Fast EMA period", "slow": "Slow EMA period", "atr_period": "ATR lookback for the stop", "atr_stop_mult": "Stop distance in ATRs (0 disables)", "size": "Fraction of equity per short", } param_bounds: ClassVar[dict] = { "fast": (2, 100), "slow": (3, 300), "atr_stop_mult": (0.0, 10.0), "size": (0.01, 1.0), } def on_candle(self, ctx: StrategyContext) -> None: fast_period = int(self.params["fast"]) slow_period = int(self.params["slow"]) if len(ctx.closes) < slow_period + 2: return fast = ema(ctx.closes, fast_period) slow = ema(ctx.closes, slow_period) f_prev, f_now = fast[-2], fast[-1] s_prev, s_now = slow[-2], slow[-1] if None in (f_prev, f_now, s_prev, s_now): return crossed_down = f_prev >= s_prev and f_now < s_now crossed_up = f_prev <= s_prev and f_now > s_now if ctx.is_flat and crossed_down: stop = None mult = float(self.params["atr_stop_mult"]) if mult > 0: atr_value = atr(ctx.candles, int(self.params["atr_period"]))[-1] if atr_value: stop = ctx.price + mult * atr_value ctx.enter_short( size=float(self.params["size"]), stop_loss=stop, reason="EMA cross down", ) elif ctx.side == PositionSide.SHORT and crossed_up: ctx.close(reason="EMA cross up")