Spaces:
Build error
Build error
File size: 5,381 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 |
"""
Caso de uso para Backtesting.
Implementa la l贸gica de aplicaci贸n para evaluar la precisi贸n
de pron贸sticos usando datos hist贸ricos.
"""
import math
from typing import List
from app.domain.services.forecast_service import ForecastService
from app.domain.models.time_series import TimeSeries
from app.domain.models.forecast_config import ForecastConfig
from app.application.dtos.backtest_dtos import (
BacktestInputDTO,
BacktestOutputDTO,
BacktestMetricsDTO
)
from app.utils.logger import setup_logger
logger = setup_logger(__name__)
class BacktestUseCase:
"""
Caso de uso: Backtesting.
Responsabilidad: Evaluar precisi贸n del modelo usando datos hist贸ricos.
Divide los datos en train/test y calcula m茅tricas de error.
"""
def __init__(self, forecast_service: ForecastService):
"""
Inicializa el caso de uso.
Args:
forecast_service: Servicio de dominio para forecasting
"""
self.forecast_service = forecast_service
logger.info("BacktestUseCase initialized")
def execute(self, input_dto: BacktestInputDTO) -> BacktestOutputDTO:
"""
Ejecuta el caso de uso de backtesting.
Args:
input_dto: Datos de entrada con serie completa y tama帽o de test
Returns:
BacktestOutputDTO: Resultados del backtest con m茅tricas
Raises:
ValueError: Si los datos son inv谩lidos
RuntimeError: Si falla el backtest
"""
logger.info(
f"Executing backtest: {len(input_dto.values)} total points, "
f"{input_dto.test_size} test points"
)
# Validar entrada
input_dto.validate()
# Dividir en train/test
train_values = input_dto.values[:-input_dto.test_size]
test_values = input_dto.values[-input_dto.test_size:]
train_timestamps = None
test_timestamps = None
if input_dto.timestamps:
train_timestamps = input_dto.timestamps[:-input_dto.test_size]
test_timestamps = input_dto.timestamps[-input_dto.test_size:]
logger.info(f"Train size: {len(train_values)}, Test size: {len(test_values)}")
# Crear modelos de dominio para train
train_series = TimeSeries(
values=train_values,
timestamps=train_timestamps,
freq=input_dto.freq
)
config = ForecastConfig(
prediction_length=input_dto.test_size,
quantile_levels=input_dto.quantile_levels,
freq=input_dto.freq
)
# Ejecutar pron贸stico sobre datos de train
try:
result = self.forecast_service.forecast_univariate(train_series, config)
logger.info(f"Forecast completed: {len(result.median)} predictions")
except Exception as e:
logger.error(f"Backtest forecast failed: {e}", exc_info=True)
raise RuntimeError(f"Backtest forecast failed: {str(e)}") from e
# Comparar con valores reales
forecast_values = result.median
actual_values = test_values
# Calcular errores
errors = [
actual - forecast
for actual, forecast in zip(actual_values, forecast_values)
]
# Calcular m茅tricas
metrics = self._calculate_metrics(actual_values, forecast_values)
logger.info(
f"Backtest metrics - MAE: {metrics.mae:.2f}, "
f"MAPE: {metrics.mape:.2f}%, RMSE: {metrics.rmse:.2f}"
)
# Preparar timestamps de salida
if test_timestamps:
output_timestamps = test_timestamps
else:
output_timestamps = result.timestamps
# Crear DTO de salida
output_dto = BacktestOutputDTO(
forecast_values=forecast_values,
actual_values=actual_values,
errors=errors,
metrics=metrics,
timestamps=output_timestamps,
quantiles=result.quantiles if result.quantiles else None
)
return output_dto
def _calculate_metrics(
self,
actual: List[float],
forecast: List[float]
) -> BacktestMetricsDTO:
"""
Calcula m茅tricas de error para el backtest.
Args:
actual: Valores reales
forecast: Valores pronosticados
Returns:
BacktestMetricsDTO: M茅tricas calculadas
"""
n = len(actual)
# Mean Absolute Error
mae = sum(abs(a - f) for a, f in zip(actual, forecast)) / n
# Mean Absolute Percentage Error
mape_values = []
for a, f in zip(actual, forecast):
if a != 0:
mape_values.append(abs((a - f) / a))
mape = (sum(mape_values) / len(mape_values) * 100) if mape_values else 0.0
# Mean Squared Error
mse = sum((a - f) ** 2 for a, f in zip(actual, forecast)) / n
# Root Mean Squared Error
rmse = math.sqrt(mse)
return BacktestMetricsDTO(
mae=mae,
mape=mape,
rmse=rmse,
mse=mse
)
|