Add Phase 2 foundation: regime classifier, strategy framework, WebSocket streamer

Phase 1 completion:
- Add DataStreamer for real-time Binance Futures WebSocket data (klines, mark price)
- Add DataValidator for candle validation and gap detection
- Add timeframes module with interval mappings

Phase 2 foundation:
- Add RegimeClassifier with ADX/ATR/Bollinger Band analysis
- Add Regime enum (TRENDING_UP/DOWN, RANGING, HIGH_VOLATILITY, UNCERTAIN)
- Add Strategy ABC defining generate_signal, get_stop_loss, parameters, suitable_regimes
- Add Signal dataclass and SignalType enum for strategy outputs

Testing:
- Add comprehensive test suites for all new modules
- 159 tests passing, 24 skipped (async WebSocket timing)
- 82% code coverage

Dependencies:
- Add pandas-stubs to dev dependencies for mypy compatibility
This commit is contained in:
bnair123
2025-12-27 15:28:28 +04:00
parent 7d63e43b7b
commit eca17b42fe
15 changed files with 2579 additions and 1 deletions

View File

@@ -0,0 +1,6 @@
"""Trading strategies module."""
from tradefinder.strategies.base import Strategy
from tradefinder.strategies.signals import Signal, SignalType
__all__ = ["Signal", "SignalType", "Strategy"]

View File

@@ -0,0 +1,112 @@
"""Base strategy interface for trading strategies."""
from abc import ABC, abstractmethod
from decimal import Decimal
from typing import Any
import structlog
from tradefinder.adapters.types import Candle, Side
from tradefinder.core.regime import Regime
from tradefinder.strategies.signals import Signal
logger = structlog.get_logger(__name__)
class Strategy(ABC):
"""Abstract base class for trading strategies.
All concrete strategies must implement this interface to be compatible
with the trading engine's strategy selection and signal generation.
Example:
class SupertrendStrategy(Strategy):
name = "supertrend"
def generate_signal(self, candles: list[Candle]) -> Signal | None:
# Analyze candles and return signal if conditions met
...
def get_stop_loss(self, entry_price: Decimal, side: Side) -> Decimal:
# Calculate stop loss based on ATR or fixed percentage
...
@property
def parameters(self) -> dict[str, Any]:
return {"period": self._period, "multiplier": self._multiplier}
@property
def suitable_regimes(self) -> list[Regime]:
return [Regime.TRENDING_UP, Regime.TRENDING_DOWN]
"""
name: str # Strategy identifier, must be set by subclasses
@abstractmethod
def generate_signal(self, candles: list[Candle]) -> Signal | None:
"""Analyze candles and generate a trading signal if conditions are met.
Args:
candles: List of OHLCV candles, ordered oldest to newest.
Must contain sufficient history for indicator calculation.
Returns:
Signal object if entry/exit conditions are met, None otherwise.
"""
pass
@abstractmethod
def get_stop_loss(self, entry_price: Decimal, side: Side) -> Decimal:
"""Calculate stop loss price for a given entry.
Args:
entry_price: The planned or actual entry price.
side: Whether this is a BUY (long) or SELL (short) position.
Returns:
Stop loss price. For longs, this should be below entry_price.
For shorts, this should be above entry_price.
"""
pass
@property
@abstractmethod
def parameters(self) -> dict[str, Any]:
"""Return current strategy parameters for display and logging.
Returns:
Dictionary of parameter names to values. Used by UI and logging
to show what settings the strategy is using.
"""
pass
@property
@abstractmethod
def suitable_regimes(self) -> list[Regime]:
"""Return list of regimes where this strategy should be active.
Returns:
List of Regime values. The strategy will only be selected
when the current market regime matches one of these.
"""
pass
def validate_candles(self, candles: list[Candle], min_required: int) -> bool:
"""Check if sufficient candle data is available for signal generation.
Args:
candles: List of candles to validate.
min_required: Minimum number of candles needed.
Returns:
True if enough candles are available, False otherwise.
"""
if len(candles) < min_required:
logger.debug(
"Insufficient candles for strategy",
strategy=self.name,
available=len(candles),
required=min_required,
)
return False
return True

View File

@@ -0,0 +1,87 @@
"""Signal types for trading strategies."""
from dataclasses import dataclass, field
from datetime import datetime
from decimal import Decimal
from enum import Enum
from typing import Any
import structlog
logger = structlog.get_logger(__name__)
class SignalType(str, Enum):
"""Type of trading signal."""
ENTRY_LONG = "entry_long"
ENTRY_SHORT = "entry_short"
EXIT_LONG = "exit_long"
EXIT_SHORT = "exit_short"
@dataclass
class Signal:
"""Trading signal emitted by a strategy.
Attributes:
signal_type: Type of signal (entry/exit, long/short)
symbol: Trading symbol (e.g., "BTCUSDT")
price: Suggested entry/exit price
stop_loss: Stop loss price for risk calculation
take_profit: Optional take profit price
confidence: Signal confidence from 0.0 to 1.0
timestamp: When the signal was generated
strategy_name: Name of the strategy that generated the signal
metadata: Additional signal-specific data
"""
signal_type: SignalType
symbol: str
price: Decimal
stop_loss: Decimal
take_profit: Decimal | None
confidence: float
timestamp: datetime
strategy_name: str
metadata: dict[str, Any] = field(default_factory=dict)
def __post_init__(self) -> None:
"""Validate signal parameters."""
if not 0.0 <= self.confidence <= 1.0:
raise ValueError(f"Confidence must be between 0.0 and 1.0, got {self.confidence}")
if self.price <= Decimal("0"):
raise ValueError(f"Price must be positive, got {self.price}")
if self.stop_loss <= Decimal("0"):
raise ValueError(f"Stop loss must be positive, got {self.stop_loss}")
@property
def is_entry(self) -> bool:
"""Check if this is an entry signal."""
return self.signal_type in (SignalType.ENTRY_LONG, SignalType.ENTRY_SHORT)
@property
def is_exit(self) -> bool:
"""Check if this is an exit signal."""
return self.signal_type in (SignalType.EXIT_LONG, SignalType.EXIT_SHORT)
@property
def is_long(self) -> bool:
"""Check if this is a long-side signal."""
return self.signal_type in (SignalType.ENTRY_LONG, SignalType.EXIT_LONG)
@property
def is_short(self) -> bool:
"""Check if this is a short-side signal."""
return self.signal_type in (SignalType.ENTRY_SHORT, SignalType.EXIT_SHORT)
@property
def risk_reward_ratio(self) -> Decimal | None:
"""Calculate risk/reward ratio if take profit is set."""
if self.take_profit is None:
return None
risk = abs(self.price - self.stop_loss)
reward = abs(self.take_profit - self.price)
if risk == Decimal("0"):
return None
return reward / risk