TCPDashboard/strategies/base.py

"""
Base classes and interfaces for trading strategies.

This module provides the foundation for all trading strategies
with common functionality and type definitions.
"""

from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional
import pandas as pd
from utils.logger import get_logger

from .data_types import StrategyResult
from data.common.data_types import OHLCVCandle


class BaseStrategy(ABC):
    """
    Abstract base class for all trading strategies.
    
    Provides common functionality and enforces consistent interface
    across all strategy implementations.
    """
    
    def __init__(self, strategy_name: str, logger=None):
        """
        Initialize base strategy.
        
        Args:
            strategy_name: The name of the strategy
            logger: Optional logger instance
        """
        if logger is None:
            self.logger = get_logger(__name__)
        else:
            self.logger = logger
        self.strategy_name = strategy_name

    def prepare_dataframe(self, candles: List[OHLCVCandle]) -> pd.DataFrame:
        """
        Convert OHLCV candles to pandas DataFrame for calculations.
        
        Args:
            candles: List of OHLCV candles (can be sparse)
            
        Returns:
            DataFrame with OHLCV data, sorted by timestamp
        """
        if not candles:
            return pd.DataFrame()
        
        # Convert to DataFrame
        data = []
        for candle in candles:
            data.append({
                'timestamp': candle.end_time,  # Right-aligned timestamp
                'symbol': candle.symbol,
                'timeframe': candle.timeframe,
                'open': float(candle.open),
                'high': float(candle.high),
                'low': float(candle.low),
                'close': float(candle.close),
                'volume': float(candle.volume),
                'trade_count': candle.trade_count
            })
        
        df = pd.DataFrame(data)
        
        # Sort by timestamp to ensure proper order
        df = df.sort_values('timestamp').reset_index(drop=True)
        
        # Set timestamp as index for time-series operations
        df['timestamp'] = pd.to_datetime(df['timestamp'])
        
        # Set as index, but keep as column
        df.set_index('timestamp', inplace=True)
        
        # Ensure it's datetime
        df['timestamp'] = df.index
        
        return df

    @abstractmethod
    def calculate(self, df: pd.DataFrame, indicators_data: Dict[str, pd.DataFrame], **kwargs) -> List[StrategyResult]:
        """
        Calculate the strategy signals.
        
        Args:
            df: DataFrame with OHLCV data
            indicators_data: Dictionary of pre-calculated indicator DataFrames
            **kwargs: Additional parameters specific to each strategy
            
        Returns:
            List of strategy results with signals
        """
        pass

    @abstractmethod
    def get_required_indicators(self) -> List[Dict[str, Any]]:
        """
        Get list of indicators required by this strategy.
        
        Returns:
            List of indicator configurations needed for strategy calculation
            Format: [{'type': 'sma', 'period': 20}, {'type': 'ema', 'period': 12}]
        """
        pass

    def validate_dataframe(self, df: pd.DataFrame, min_periods: int) -> bool:
        """
        Validate that DataFrame has sufficient data for calculation.
        
        Args:
            df: DataFrame to validate
            min_periods: Minimum number of periods required
            
        Returns:
            True if DataFrame is valid, False otherwise
        """
        if df.empty or len(df) < min_periods:
            if self.logger:
                self.logger.warning(
                    f"Insufficient data: got {len(df)} periods, need {min_periods}"
                )
            return False
        return True

    def validate_indicators_data(self, indicators_data: Dict[str, pd.DataFrame], 
                               required_indicators: List[Dict[str, Any]]) -> bool:
        """
        Validate that all required indicators are present and have sufficient data.
        
        Args:
            indicators_data: Dictionary of indicator DataFrames
            required_indicators: List of required indicator configurations
            
        Returns:
            True if all required indicators are available, False otherwise
        """
        for indicator_config in required_indicators:
            indicator_key = f"{indicator_config['type']}_{indicator_config.get('period', 'default')}"
            
            if indicator_key not in indicators_data:
                if self.logger:
                    self.logger.error(f"Missing required indicator data for key: {indicator_key}")
                raise ValueError(f"Missing required indicator data for key: {indicator_key}")
            
            if indicators_data[indicator_key].empty:
                if self.logger:
                    self.logger.warning(f"Empty data for indicator: {indicator_key}")
                return False
            
            if indicators_data[indicator_key].isnull().values.any():
                if self.logger:
                    self.logger.warning(f"NaN values found in indicator data for key: {indicator_key}")
                return False
        
        return True