Spaces:
Build error
Build error
File size: 5,520 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 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 |
"""
Dependency Injection para FastAPI.
Provee instancias de servicios, repositorios y casos de uso
usando el sistema de DI de FastAPI.
"""
from functools import lru_cache
from fastapi import Depends
# Infrastructure
from app.infrastructure.ml.model_factory import ModelFactory
from app.infrastructure.config.settings import get_settings
# Domain
from app.domain.interfaces.forecast_model import IForecastModel
from app.domain.interfaces.data_transformer import IDataTransformer
from app.domain.services.forecast_service import ForecastService
from app.domain.services.anomaly_service import AnomalyService
# Application
from app.application.use_cases.forecast_use_case import (
ForecastUnivariateUseCase,
ForecastMultiSeriesUseCase
)
from app.application.use_cases.anomaly_use_case import DetectAnomaliesUseCase
from app.application.use_cases.backtest_use_case import BacktestUseCase
# Utils
from app.utils.dataframe_builder import DataFrameBuilder
from app.utils.logger import setup_logger
# Get settings instance
settings = get_settings()
logger = setup_logger(__name__)
# ============================================================================
# Infrastructure Layer Dependencies
# ============================================================================
# Singleton para el modelo de forecasting
_model_instance: IForecastModel = None
def get_forecast_model() -> IForecastModel:
"""
Dependency: Modelo de forecasting (Singleton).
Usa Chronos-2 por defecto. El modelo se carga una sola vez
y se reutiliza en todas las requests.
Returns:
IForecastModel: Instancia del modelo
"""
global _model_instance
if _model_instance is None:
logger.info("Initializing forecast model (first time)")
_model_instance = ModelFactory.create(
model_type="chronos2",
model_id=settings.model_id,
device_map=settings.device_map
)
logger.info(f"Model loaded: {_model_instance.get_model_info()}")
return _model_instance
def get_data_transformer() -> IDataTransformer:
"""
Dependency: Transformador de datos.
Returns:
IDataTransformer: Instancia del transformador
"""
return DataFrameBuilder()
# ============================================================================
# Domain Layer Dependencies
# ============================================================================
def get_forecast_service(
model: IForecastModel = Depends(get_forecast_model),
transformer: IDataTransformer = Depends(get_data_transformer)
) -> ForecastService:
"""
Dependency: Servicio de dominio para forecasting.
Args:
model: Modelo de forecasting
transformer: Transformador de datos
Returns:
ForecastService: Servicio de forecasting
"""
return ForecastService(model=model, transformer=transformer)
def get_anomaly_service(
model: IForecastModel = Depends(get_forecast_model),
transformer: IDataTransformer = Depends(get_data_transformer)
) -> AnomalyService:
"""
Dependency: Servicio de dominio para detecci贸n de anomal铆as.
Args:
model: Modelo de forecasting
transformer: Transformador de datos
Returns:
AnomalyService: Servicio de anomal铆as
"""
return AnomalyService(model=model, transformer=transformer)
# ============================================================================
# Application Layer Dependencies (Use Cases)
# ============================================================================
def get_forecast_univariate_use_case(
service: ForecastService = Depends(get_forecast_service)
) -> ForecastUnivariateUseCase:
"""
Dependency: Caso de uso de pron贸stico univariado.
Args:
service: Servicio de forecasting
Returns:
ForecastUnivariateUseCase: Caso de uso
"""
return ForecastUnivariateUseCase(forecast_service=service)
def get_forecast_multi_series_use_case(
service: ForecastService = Depends(get_forecast_service)
) -> ForecastMultiSeriesUseCase:
"""
Dependency: Caso de uso de pron贸stico multi-series.
Args:
service: Servicio de forecasting
Returns:
ForecastMultiSeriesUseCase: Caso de uso
"""
return ForecastMultiSeriesUseCase(forecast_service=service)
def get_detect_anomalies_use_case(
service: AnomalyService = Depends(get_anomaly_service)
) -> DetectAnomaliesUseCase:
"""
Dependency: Caso de uso de detecci贸n de anomal铆as.
Args:
service: Servicio de anomal铆as
Returns:
DetectAnomaliesUseCase: Caso de uso
"""
return DetectAnomaliesUseCase(anomaly_service=service)
def get_backtest_use_case(
service: ForecastService = Depends(get_forecast_service)
) -> BacktestUseCase:
"""
Dependency: Caso de uso de backtesting.
Args:
service: Servicio de forecasting
Returns:
BacktestUseCase: Caso de uso
"""
return BacktestUseCase(forecast_service=service)
# ============================================================================
# Utility Functions
# ============================================================================
def reset_model():
"""
Resetea el modelo (煤til para testing).
ADVERTENCIA: Solo usar en tests, no en producci贸n.
"""
global _model_instance
_model_instance = None
logger.warning("Model instance reset")
|