Spaces:
Build error
Build error
File size: 5,767 Bytes
c40c447 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 |
"""
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
|