287 lines
11 KiB
Python
287 lines
11 KiB
Python
"""
|
|
Technical Indicators Module for OHLCV Data
|
|
|
|
This module provides technical indicator calculations optimized for sparse OHLCV data
|
|
as produced by the TCP Trading Platform's aggregation strategy.
|
|
|
|
IMPORTANT: Handles Sparse Data
|
|
- Missing candles (time gaps) are normal in this system
|
|
- Indicators properly handle gaps without interpolation
|
|
- Uses pandas for efficient vectorized calculations
|
|
- Follows right-aligned timestamp convention
|
|
|
|
Supported Indicators:
|
|
- Simple Moving Average (SMA)
|
|
- Exponential Moving Average (EMA)
|
|
- Relative Strength Index (RSI)
|
|
- Moving Average Convergence Divergence (MACD)
|
|
- Bollinger Bands
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from typing import Dict, List, Optional, Any, Union
|
|
import pandas as pd
|
|
import numpy as np
|
|
|
|
from .result import IndicatorResult
|
|
from ..data_types import OHLCVCandle
|
|
from .base import BaseIndicator
|
|
from .implementations import (
|
|
SMAIndicator,
|
|
EMAIndicator,
|
|
RSIIndicator,
|
|
MACDIndicator,
|
|
BollingerBandsIndicator
|
|
)
|
|
|
|
|
|
class TechnicalIndicators:
|
|
"""
|
|
Technical indicator calculator for OHLCV candle data.
|
|
|
|
This class provides vectorized technical indicator calculations
|
|
designed to handle sparse data efficiently. All calculations use
|
|
pandas for performance and handle missing data appropriately.
|
|
|
|
SPARSE DATA HANDLING:
|
|
- Gaps in timestamps are preserved (no interpolation)
|
|
- Indicators calculate only on available data points
|
|
- Periods with insufficient data return NaN
|
|
- Results maintain original timestamp alignment
|
|
"""
|
|
|
|
def __init__(self, logger=None):
|
|
"""
|
|
Initialize technical indicators calculator.
|
|
|
|
Args:
|
|
logger: Optional logger instance
|
|
"""
|
|
self.logger = logger
|
|
|
|
# Initialize individual indicator calculators
|
|
self._sma = SMAIndicator(logger)
|
|
self._ema = EMAIndicator(logger)
|
|
self._rsi = RSIIndicator(logger)
|
|
self._macd = MACDIndicator(logger)
|
|
self._bollinger = BollingerBandsIndicator(logger)
|
|
|
|
if self.logger:
|
|
self.logger.info("TechnicalIndicators: Initialized indicator calculator")
|
|
|
|
def _prepare_dataframe_from_list(self, candles: List[OHLCVCandle]) -> pd.DataFrame:
|
|
"""
|
|
Convert OHLCV candles to pandas DataFrame for efficient calculations.
|
|
|
|
Args:
|
|
candles: List of OHLCV candles (can be sparse)
|
|
|
|
Returns:
|
|
DataFrame with OHLCV data, sorted by timestamp
|
|
"""
|
|
if not candles:
|
|
return pd.DataFrame()
|
|
|
|
return self._sma.prepare_dataframe(candles)
|
|
|
|
def sma(self, df: pd.DataFrame, period: int,
|
|
price_column: str = 'close') -> List[IndicatorResult]:
|
|
"""
|
|
Calculate Simple Moving Average (SMA).
|
|
|
|
Args:
|
|
df: DataFrame with OHLCV data
|
|
period: Number of periods for moving average
|
|
price_column: Price column to use ('open', 'high', 'low', 'close')
|
|
|
|
Returns:
|
|
List of indicator results with SMA values
|
|
"""
|
|
return self._sma.calculate(df, period=period, price_column=price_column)
|
|
|
|
def ema(self, df: pd.DataFrame, period: int,
|
|
price_column: str = 'close') -> List[IndicatorResult]:
|
|
"""
|
|
Calculate Exponential Moving Average (EMA).
|
|
|
|
Args:
|
|
df: DataFrame with OHLCV data
|
|
period: Number of periods for moving average
|
|
price_column: Price column to use ('open', 'high', 'low', 'close')
|
|
|
|
Returns:
|
|
List of indicator results with EMA values
|
|
"""
|
|
return self._ema.calculate(df, period=period, price_column=price_column)
|
|
|
|
def rsi(self, df: pd.DataFrame, period: int = 14,
|
|
price_column: str = 'close') -> List[IndicatorResult]:
|
|
"""
|
|
Calculate Relative Strength Index (RSI).
|
|
|
|
Args:
|
|
df: DataFrame with OHLCV data
|
|
period: Number of periods for RSI calculation (default 14)
|
|
price_column: Price column to use ('open', 'high', 'low', 'close')
|
|
|
|
Returns:
|
|
List of indicator results with RSI values
|
|
"""
|
|
return self._rsi.calculate(df, period=period, price_column=price_column)
|
|
|
|
def macd(self, df: pd.DataFrame,
|
|
fast_period: int = 12, slow_period: int = 26, signal_period: int = 9,
|
|
price_column: str = 'close') -> List[IndicatorResult]:
|
|
"""
|
|
Calculate Moving Average Convergence Divergence (MACD).
|
|
|
|
Args:
|
|
df: DataFrame with OHLCV data
|
|
fast_period: Fast EMA period (default 12)
|
|
slow_period: Slow EMA period (default 26)
|
|
signal_period: Signal line EMA period (default 9)
|
|
price_column: Price column to use ('open', 'high', 'low', 'close')
|
|
|
|
Returns:
|
|
List of indicator results with MACD, signal, and histogram values
|
|
"""
|
|
return self._macd.calculate(
|
|
df,
|
|
fast_period=fast_period,
|
|
slow_period=slow_period,
|
|
signal_period=signal_period,
|
|
price_column=price_column
|
|
)
|
|
|
|
def bollinger_bands(self, df: pd.DataFrame, period: int = 20,
|
|
std_dev: float = 2.0, price_column: str = 'close') -> List[IndicatorResult]:
|
|
"""
|
|
Calculate Bollinger Bands.
|
|
|
|
Args:
|
|
df: DataFrame with OHLCV data
|
|
period: Number of periods for moving average (default 20)
|
|
std_dev: Number of standard deviations (default 2.0)
|
|
price_column: Price column to use ('open', 'high', 'low', 'close')
|
|
|
|
Returns:
|
|
List of indicator results with upper band, middle band (SMA), and lower band
|
|
"""
|
|
return self._bollinger.calculate(
|
|
df,
|
|
period=period,
|
|
std_dev=std_dev,
|
|
price_column=price_column
|
|
)
|
|
|
|
def calculate_multiple_indicators(self, df: pd.DataFrame,
|
|
indicators_config: Dict[str, Dict[str, Any]]) -> Dict[str, List[IndicatorResult]]:
|
|
"""
|
|
TODO: need make more procedural without hardcoding indicators type and so on
|
|
Calculate multiple indicators at once for efficiency.
|
|
|
|
Args:
|
|
df: DataFrame with OHLCV data
|
|
indicators_config: Configuration for indicators to calculate
|
|
Example: {
|
|
'sma_20': {'type': 'sma', 'period': 20},
|
|
'ema_12': {'type': 'ema', 'period': 12},
|
|
'rsi_14': {'type': 'rsi', 'period': 14},
|
|
'macd': {'type': 'macd'},
|
|
'bb_20': {'type': 'bollinger_bands', 'period': 20}
|
|
}
|
|
|
|
Returns:
|
|
Dictionary mapping indicator names to their results
|
|
"""
|
|
results = {}
|
|
|
|
for indicator_name, config in indicators_config.items():
|
|
indicator_type = config.get('type')
|
|
|
|
try:
|
|
if indicator_type == 'sma':
|
|
period = config.get('period', 20)
|
|
price_column = config.get('price_column', 'close')
|
|
results[indicator_name] = self.sma(df, period, price_column)
|
|
|
|
elif indicator_type == 'ema':
|
|
period = config.get('period', 20)
|
|
price_column = config.get('price_column', 'close')
|
|
results[indicator_name] = self.ema(df, period, price_column)
|
|
|
|
elif indicator_type == 'rsi':
|
|
period = config.get('period', 14)
|
|
price_column = config.get('price_column', 'close')
|
|
results[indicator_name] = self.rsi(df, period, price_column)
|
|
|
|
elif indicator_type == 'macd':
|
|
fast_period = config.get('fast_period', 12)
|
|
slow_period = config.get('slow_period', 26)
|
|
signal_period = config.get('signal_period', 9)
|
|
price_column = config.get('price_column', 'close')
|
|
results[indicator_name] = self.macd(
|
|
df, fast_period, slow_period, signal_period, price_column
|
|
)
|
|
|
|
elif indicator_type == 'bollinger_bands':
|
|
period = config.get('period', 20)
|
|
std_dev = config.get('std_dev', 2.0)
|
|
price_column = config.get('price_column', 'close')
|
|
results[indicator_name] = self.bollinger_bands(
|
|
df, period, std_dev, price_column
|
|
)
|
|
|
|
else:
|
|
if self.logger:
|
|
self.logger.warning(f"Unknown indicator type: {indicator_type}")
|
|
results[indicator_name] = []
|
|
|
|
except Exception as e:
|
|
if self.logger:
|
|
self.logger.error(f"Error calculating {indicator_name}: {e}")
|
|
results[indicator_name] = []
|
|
|
|
return results
|
|
|
|
def calculate(self, indicator_type: str, df: pd.DataFrame, **kwargs) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Calculate a single indicator with dynamic dispatch.
|
|
|
|
Args:
|
|
indicator_type: Name of the indicator (e.g., 'sma', 'ema')
|
|
df: DataFrame with OHLCV data
|
|
**kwargs: Indicator-specific parameters (e.g., period=20)
|
|
|
|
Returns:
|
|
A dictionary containing the indicator results, or None if the type is unknown.
|
|
"""
|
|
# Get the indicator calculation method
|
|
indicator_method = getattr(self, indicator_type, None)
|
|
if not indicator_method:
|
|
if self.logger:
|
|
self.logger.error(f"Unknown indicator type '{indicator_type}'")
|
|
return None
|
|
|
|
try:
|
|
if df.empty:
|
|
return {'data': [], 'metadata': {}}
|
|
|
|
# Call the indicator method
|
|
raw_result = indicator_method(df, **kwargs)
|
|
|
|
# Extract metadata from the first result if available
|
|
metadata = raw_result[0].metadata if raw_result else {}
|
|
|
|
# The methods return List[IndicatorResult], let's package that
|
|
if raw_result:
|
|
return {
|
|
"data": raw_result,
|
|
"metadata": metadata
|
|
}
|
|
return None
|
|
|
|
except Exception as e:
|
|
if self.logger:
|
|
self.logger.error(f"Error calculating {indicator_type}: {e}")
|
|
return None |