""" Generador de timestamps para series temporales. Este módulo proporciona utilidades para generar timestamps, aplicando el principio SRP (Single Responsibility Principle). """ from typing import List, Union from datetime import datetime, timedelta import pandas as pd from app.utils.logger import setup_logger logger = setup_logger(__name__) class TimestampGenerator: """ Generador de timestamps para series temporales. Proporciona métodos para generar diferentes tipos de timestamps: - Rangos de fechas (date_range) - Índices enteros (integer_index) - Continuación de series existentes (continue_from) """ @staticmethod def generate_date_range( start: Union[str, datetime], periods: int, freq: str = "D" ) -> List[str]: """ Genera un rango de fechas. Args: start: Fecha de inicio (string ISO o datetime) periods: Número de períodos freq: Frecuencia (D=diario, W=semanal, M=mensual, etc.) Returns: Lista de timestamps como strings ISO Example: >>> gen = TimestampGenerator() >>> gen.generate_date_range("2025-01-01", 5, "D") ['2025-01-01', '2025-01-02', '2025-01-03', '2025-01-04', '2025-01-05'] """ try: dates = pd.date_range( start=pd.to_datetime(start), periods=periods, freq=freq ) result = dates.astype(str).tolist() logger.debug(f"Generated {len(result)} timestamps with freq={freq}") return result except Exception as e: logger.error(f"Error generating date range: {e}") raise ValueError(f"Error generando fechas: {e}") from e @staticmethod def generate_integer_index( periods: int, start: int = 0 ) -> List[int]: """ Genera un índice entero secuencial. Args: periods: Número de períodos start: Valor inicial del índice Returns: Lista de enteros Example: >>> gen = TimestampGenerator() >>> gen.generate_integer_index(5, start=10) [10, 11, 12, 13, 14] """ if periods < 1: raise ValueError("periods debe ser >= 1") result = list(range(start, start + periods)) logger.debug(f"Generated integer index: {start} to {start + periods - 1}") return result @staticmethod def continue_from( last_timestamp: Union[str, int], periods: int, freq: str = "D" ) -> List[str]: """ Continúa una serie temporal desde el último timestamp. Args: last_timestamp: Último timestamp de la serie existente periods: Número de períodos futuros freq: Frecuencia (solo para fechas) Returns: Lista de timestamps futuros Example: >>> gen = TimestampGenerator() >>> gen.continue_from("2025-01-05", 3, "D") ['2025-01-06', '2025-01-07', '2025-01-08'] """ try: # Intentar parsear como fecha if isinstance(last_timestamp, str): try: last_date = pd.to_datetime(last_timestamp) next_date = last_date + pd.Timedelta(1, unit=freq) return TimestampGenerator.generate_date_range( next_date, periods, freq ) except: # Si falla, intentar como entero last_int = int(last_timestamp) return TimestampGenerator.generate_integer_index( periods, start=last_int + 1 ) else: # Entero return TimestampGenerator.generate_integer_index( periods, start=last_timestamp + 1 ) except Exception as e: logger.error(f"Error continuing timestamps: {e}") raise ValueError(f"Error continuando timestamps: {e}") from e @staticmethod def infer_frequency(timestamps: List[str]) -> str: """ Infiere la frecuencia de una lista de timestamps. Args: timestamps: Lista de timestamps (strings ISO) Returns: Código de frecuencia (D, W, M, etc.) Raises: ValueError: Si no se puede inferir la frecuencia """ if len(timestamps) < 2: raise ValueError("Se necesitan al menos 2 timestamps para inferir frecuencia") try: dates = pd.to_datetime(timestamps) freq = pd.infer_freq(dates) if freq is None: # Fallback: calcular diferencia promedio diffs = dates.diff().dropna() avg_diff = diffs.mean() if avg_diff.days == 1: freq = "D" elif avg_diff.days == 7: freq = "W" elif 28 <= avg_diff.days <= 31: freq = "M" else: freq = "D" # Default logger.warning(f"Frecuencia inferida aproximadamente: {freq}") logger.debug(f"Inferred frequency: {freq}") return freq except Exception as e: logger.error(f"Error inferring frequency: {e}") return "D" # Default seguro