from abc import ABC, abstractmethod from datetime import datetime from app.models.schemas import KlineData, OrderRequest, StrategyPerformance class BaseStrategy(ABC): def __init__(self, strategy_id: str, symbol: str, params: dict | None = None): self.strategy_id = strategy_id self.symbol = symbol self.params = params or {} self.signals: list[dict] = [] self._performance = StrategyPerformance(strategy_id=strategy_id) @property @abstractmethod def name(self) -> str: ... @property @abstractmethod def description(self) -> str: ... @property @abstractmethod def default_params(self) -> dict: ... @abstractmethod def calculate_signal(self, klines: list[KlineData]) -> OrderRequest | None: ... def get_param(self, key: str, default=None): return self.params.get(key, self.default_params.get(key, default)) def record_signal(self, signal_type: str, price: float, reason: str): self.signals.append({ "type": signal_type, "price": price, "reason": reason, "timestamp": datetime.utcnow().isoformat(), }) if len(self.signals) > 500: self.signals = self.signals[-300:] @property def performance(self) -> StrategyPerformance: return self._performance def update_performance(self, pnl: float): self._performance.total_trades += 1 self._performance.total_pnl += pnl if pnl > 0: self._performance.winning_trades += 1 elif pnl < 0: self._performance.losing_trades += 1 total = self._performance.total_trades if total > 0: self._performance.win_rate = round(self._performance.winning_trades / total * 100, 2)