""" Real-time candle processor for live trade data. This module provides the RealTimeCandleProcessor class for building OHLCV candles from live trade data in real-time. """ from datetime import datetime from typing import Dict, List, Optional, Any, Callable from ..data_types import ( StandardizedTrade, OHLCVCandle, CandleProcessingConfig, ProcessingStats ) from .bucket import TimeframeBucket from .utils import parse_timeframe class RealTimeCandleProcessor: """ Real-time candle processor for live trade data. This class processes trades immediately as they arrive from WebSocket, building candles incrementally and emitting completed candles when time boundaries are crossed. AGGREGATION PROCESS (NO FUTURE LEAKAGE): 1. Trade arrives from WebSocket/API with timestamp T 2. For each configured timeframe (1m, 5m, etc.): a. Calculate which time bucket this trade belongs to b. Get current bucket for this timeframe c. Check if trade timestamp crosses time boundary d. If boundary crossed: complete and emit previous bucket, create new bucket e. Add trade to current bucket (updates OHLCV) 3. Only emit candles when time boundary is definitively crossed 4. Never emit incomplete/future candles during real-time processing TIMESTAMP ALIGNMENT: - Uses RIGHT-ALIGNED timestamps (industry standard) - 1-minute candle covering 09:00:00-09:01:00 gets timestamp 09:01:00 - 5-minute candle covering 09:00:00-09:05:00 gets timestamp 09:05:00 - Candle represents PAST data, never future """ def __init__(self, symbol: str, exchange: str, config: Optional[CandleProcessingConfig] = None, component_name: str = "realtime_candle_processor", logger = None): """ Initialize real-time candle processor. Args: symbol: Trading symbol (e.g., 'BTC-USDT') exchange: Exchange name config: Candle processing configuration component_name: Name for logging/stats logger: Optional logger instance """ self.symbol = symbol self.exchange = exchange self.config = config or CandleProcessingConfig() self.component_name = component_name self.logger = logger # Current buckets for each timeframe self.current_buckets: Dict[str, TimeframeBucket] = {} # Callbacks for completed candles self.candle_callbacks: List[Callable[[OHLCVCandle], None]] = [] # Stats tracking self.stats = ProcessingStats() self.stats.active_timeframes = len(self.config.timeframes) def add_candle_callback(self, callback: Callable[[OHLCVCandle], None]) -> None: """Add callback to be called when candle is completed.""" self.candle_callbacks.append(callback) def process_trade(self, trade: StandardizedTrade) -> List[OHLCVCandle]: """ Process a single trade and return any completed candles. Args: trade: Standardized trade data Returns: List of completed candles (if any time boundaries were crossed) """ self.stats.trades_processed += 1 self.stats.last_trade_time = trade.timestamp completed_candles = [] for timeframe in self.config.timeframes: completed = self._process_trade_for_timeframe(trade, timeframe) if completed: completed_candles.append(completed) self.stats.candles_emitted += 1 self.stats.last_candle_time = completed.end_time return completed_candles def _process_trade_for_timeframe(self, trade: StandardizedTrade, timeframe: str) -> Optional[OHLCVCandle]: """ Process trade for a specific timeframe and return completed candle if boundary crossed. Args: trade: Trade to process timeframe: Timeframe to process for (e.g., '1m', '5m') Returns: Completed candle if time boundary crossed, None otherwise """ # Calculate which bucket this trade belongs to bucket_start = self._get_bucket_start_time(trade.timestamp, timeframe) # Get current bucket for this timeframe current_bucket = self.current_buckets.get(timeframe) completed_candle = None # If we have a current bucket and trade belongs in a new bucket, # complete current bucket and create new one if current_bucket and bucket_start >= current_bucket.end_time: completed_candle = current_bucket.to_candle(is_complete=True) self._emit_candle(completed_candle) current_bucket = None # Create new bucket if needed if not current_bucket: current_bucket = TimeframeBucket( symbol=self.symbol, timeframe=timeframe, start_time=bucket_start, exchange=self.exchange ) self.current_buckets[timeframe] = current_bucket # Add trade to current bucket current_bucket.add_trade(trade) return completed_candle def _get_bucket_start_time(self, timestamp: datetime, timeframe: str) -> datetime: """ IMPORTANT: Uses RIGHT-ALIGNED timestamps Calculate the start time for the bucket that this timestamp belongs to using parsing-based logic. Args: timestamp: Trade timestamp timeframe: Time period (e.g., '1m', '5m', '1h') Returns: Start time for the appropriate bucket Raises: ValueError: If the timeframe is malformed or unsupported """ number, unit = parse_timeframe(timeframe) if unit == 's': seconds = (timestamp.second // number) * number return timestamp.replace(second=seconds, microsecond=0) elif unit == 'm': minutes = (timestamp.minute // number) * number return timestamp.replace(minute=minutes, second=0, microsecond=0) elif unit == 'h': hours = (timestamp.hour // number) * number return timestamp.replace(hour=hours, minute=0, second=0, microsecond=0) elif unit == 'd': # For days, always floor to midnight return timestamp.replace(hour=0, minute=0, second=0, microsecond=0) else: raise ValueError(f"Unsupported timeframe unit: {unit}") def _emit_candle(self, candle: OHLCVCandle) -> None: """Emit completed candle to all registered callbacks.""" for callback in self.candle_callbacks: try: callback(candle) except Exception as e: if self.logger: self.logger.error(f"Error in candle callback: {e}") self.stats.errors_count += 1 def get_current_candles(self, incomplete: bool = True) -> List[OHLCVCandle]: """ Get current (incomplete) candles for all timeframes. Args: incomplete: Whether to mark candles as incomplete (default True) """ return [ bucket.to_candle(is_complete=not incomplete) for bucket in self.current_buckets.values() ] def force_complete_all_candles(self) -> List[OHLCVCandle]: """ Force completion of all current candles (e.g., on connection close). Returns: List of completed candles """ completed = [] for timeframe, bucket in self.current_buckets.items(): candle = bucket.to_candle(is_complete=True) completed.append(candle) self._emit_candle(candle) self.stats.candles_emitted += 1 self.current_buckets.clear() return completed def get_stats(self) -> Dict[str, Any]: """Get processing statistics.""" stats_dict = self.stats.to_dict() stats_dict.update({ 'component': self.component_name, 'symbol': self.symbol, 'exchange': self.exchange, 'active_timeframes': list(self.current_buckets.keys()) }) return stats_dict __all__ = ['RealTimeCandleProcessor']