""" 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")