File size: 1,831 Bytes
590a501 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 | 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)
|