| 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) |
|
|
