Spaces:
Build error
Build error
| # ποΈ Chronos2 Server - Architecture Documentation | |
| **Version**: 3.0.0 | |
| **Date**: 2025-11-09 | |
| **Author**: Claude AI | |
| **Status**: Production Ready | |
| --- | |
| ## π Table of Contents | |
| 1. [Overview](#overview) | |
| 2. [Architecture Principles](#architecture-principles) | |
| 3. [System Architecture](#system-architecture) | |
| 4. [Layer Details](#layer-details) | |
| 5. [Design Patterns](#design-patterns) | |
| 6. [Data Flow](#data-flow) | |
| 7. [Component Diagrams](#component-diagrams) | |
| 8. [SOLID Principles](#solid-principles) | |
| 9. [Testing Strategy](#testing-strategy) | |
| 10. [Deployment Architecture](#deployment-architecture) | |
| --- | |
| ## π― Overview | |
| Chronos2 Server is a **time series forecasting API** powered by Amazon's Chronos-2 model. The system follows **Clean Architecture** principles with strict layer separation and **SOLID** design principles. | |
| ### Key Features | |
| - β **Probabilistic forecasting** with quantile predictions | |
| - β **Anomaly detection** using forecast bounds | |
| - β **Backtesting** for model evaluation | |
| - β **Multi-series forecasting** support | |
| - β **Excel integration** via Office Add-in | |
| - β **REST API** with OpenAPI documentation | |
| ### Technology Stack | |
| **Backend:** | |
| - FastAPI (Python 3.10+) | |
| - Chronos-2 (Amazon ML model) | |
| - Pandas (Data manipulation) | |
| - Pydantic (Data validation) | |
| **Frontend:** | |
| - Office.js (Excel Add-in) | |
| - Vanilla JavaScript (ES6+) | |
| - HTML5/CSS3 | |
| **Testing:** | |
| - pytest (Unit & Integration tests) | |
| - pytest-cov (Coverage reports) | |
| - FastAPI TestClient (API testing) | |
| **Deployment:** | |
| - Docker (Containerization) | |
| - HuggingFace Spaces (Hosting) | |
| --- | |
| ## ποΈ Architecture Principles | |
| ### Clean Architecture | |
| The system follows **Clean Architecture** (Uncle Bob) with 4 distinct layers: | |
| ``` | |
| βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| β Presentation Layer β | |
| β (API Routes, Controllers, Excel UI) β | |
| ββββββββββββββββββββ¬βββββββββββββββββββββββββββββββββββββββ | |
| β Depends on β | |
| ββββββββββββββββββββΌβββββββββββββββββββββββββββββββββββββββ | |
| β Application Layer β | |
| β (Use Cases, DTOs, Mappers) β | |
| ββββββββββββββββββββ¬βββββββββββββββββββββββββββββββββββββββ | |
| β Depends on β | |
| ββββββββββββββββββββΌβββββββββββββββββββββββββββββββββββββββ | |
| β Domain Layer β | |
| β (Business Logic, Models, Services, Interfaces) β | |
| ββββββββββββββββββββ¬βββββββββββββββββββββββββββββββββββββββ | |
| β Depends on β | |
| ββββββββββββββββββββΌβββββββββββββββββββββββββββββββββββββββ | |
| β Infrastructure Layer β | |
| β (External Services, ML Models, Config, DB) β | |
| βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| ``` | |
| **Dependency Rule**: Dependencies point **inward** only. Inner layers know nothing about outer layers. | |
| ### Design Goals | |
| 1. **Maintainability**: Easy to understand and modify | |
| 2. **Testability**: Components can be tested in isolation | |
| 3. **Scalability**: Easy to add new features | |
| 4. **Flexibility**: Easy to swap implementations | |
| 5. **Reliability**: Robust error handling | |
| --- | |
| ## π¨ System Architecture | |
| ### High-Level Architecture | |
| ``` | |
| ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| β CLIENT LAYER β | |
| β ββββββββββββββββββββββ ββββββββββββββββββββββ β | |
| β β Excel Add-in β β REST Clients β β | |
| β β (Office.js) β β (curl, Postman) β β | |
| β ββββββββββ¬ββββββββββββ ββββββββββ¬ββββββββββββ β | |
| βββββββββββββΌβββββββββββββββββββββββββββββββΌββββββββββββββββββββ | |
| β β | |
| ββββββββββββββββ¬ββββββββββββββββ | |
| β HTTP/HTTPS | |
| ββββββββββββββββΌβββββββββββββββ | |
| β FastAPI Server β | |
| β (API Gateway/Router) β | |
| ββββββββββββββββ¬βββββββββββββββ | |
| β | |
| ββββββββββββββββββββ΄βββββββββββββββββββ | |
| β β | |
| βββββββββΌβββββββββ ββββββββββΌβββββββββ | |
| β API Routes β β Static Files β | |
| β /forecast β β (Excel UI) β | |
| β /anomaly β βββββββββββββββββββ | |
| β /backtest β | |
| β /health β | |
| βββββββββ¬βββββββββ | |
| β | |
| β Dependency Injection | |
| β | |
| βββββββββΌβββββββββββββββββββββββββββββββββββββββββ | |
| β APPLICATION LAYER β | |
| β ββββββββββββββββ ββββββββββββββββ β | |
| β β Use Cases β β Mappers β β | |
| β β (Business β β (DTO β β β | |
| β β Workflows) β β Domain) β β | |
| β ββββββββ¬ββββββββ ββββββββββββββββ β | |
| βββββββββββΌβββββββββββββββββββββββββββββββββββββββ | |
| β | |
| βββββββββββΌβββββββββββββββββββββββββββββββββββββββ | |
| β DOMAIN LAYER β | |
| β ββββββββββββββββ ββββββββββββββββ β | |
| β β Services β β Models β β | |
| β β - Forecast β β - TimeSeriesβ β | |
| β β - Anomaly β β - Config β β | |
| β β - Backtest β β - Result β β | |
| β ββββββββ¬ββββββββ ββββββββββββββββ β | |
| β β β | |
| β ββββββββΌββββββββββββββββββββ β | |
| β β Interfaces β β | |
| β β - IForecastModel β β | |
| β β - IDataTransformer β β | |
| β ββββββββββββββββββββββββββββ β | |
| ββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| β | |
| βββββββββββΌβββββββββββββββββββββββββββββββββββββββ | |
| β INFRASTRUCTURE LAYER β | |
| β ββββββββββββββββ ββββββββββββββββ β | |
| β β ML Models β β Config β β | |
| β β - Chronos2 β β - Settings β β | |
| β β - Factory β β - Logger β β | |
| β ββββββββββββββββ ββββββββββββββββ β | |
| β ββββββββββββββββ ββββββββββββββββ β | |
| β β Transformersβ β Generators β β | |
| β β - DataFrame β β - Timestamp β β | |
| β β - Builder β β β β | |
| β ββββββββββββββββ ββββββββββββββββ β | |
| ββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| ``` | |
| --- | |
| ## π¦ Layer Details | |
| ### 1. Presentation Layer (API) | |
| **Location**: `app/api/` | |
| **Responsibilities**: | |
| - HTTP request/response handling | |
| - Input validation (Pydantic) | |
| - Error handling and formatting | |
| - API documentation (OpenAPI) | |
| - CORS middleware | |
| **Components**: | |
| ``` | |
| app/api/ | |
| βββ dependencies.py # Dependency injection setup | |
| βββ routes/ | |
| β βββ health.py # Health check endpoints | |
| β βββ forecast.py # Forecasting endpoints | |
| β βββ anomaly.py # Anomaly detection endpoints | |
| β βββ backtest.py # Backtesting endpoints | |
| βββ middleware/ | |
| βββ cors.py # CORS configuration | |
| ``` | |
| **Example Route**: | |
| ```python | |
| @router.post("/forecast/univariate") | |
| async def forecast_univariate( | |
| request: ForecastUnivariateRequest, | |
| use_case: ForecastUseCase = Depends(get_forecast_use_case) | |
| ) -> ForecastUnivariateResponse: | |
| """Univariate forecasting endpoint""" | |
| # 1. Validate request (Pydantic) | |
| # 2. Execute use case | |
| # 3. Return response | |
| result = use_case.execute(request) | |
| return result | |
| ``` | |
| **Key Principles**: | |
| - β **SRP**: Routes only handle HTTP concerns | |
| - β **DIP**: Depends on use cases (abstractions) | |
| - β No business logic in routes | |
| --- | |
| ### 2. Application Layer | |
| **Location**: `app/application/` | |
| **Responsibilities**: | |
| - Orchestrate business workflows (Use Cases) | |
| - Transform between API and Domain models (Mappers) | |
| - Coordinate multiple domain services | |
| - Transaction boundaries | |
| **Components**: | |
| ``` | |
| app/application/ | |
| βββ dtos/ # Data Transfer Objects | |
| β βββ forecast_dtos.py # Forecast DTOs | |
| β βββ anomaly_dtos.py # Anomaly DTOs | |
| β βββ backtest_dtos.py # Backtest DTOs | |
| βββ use_cases/ # Business workflows | |
| β βββ forecast_use_case.py | |
| β βββ anomaly_use_case.py | |
| β βββ backtest_use_case.py | |
| βββ mappers/ # DTO β Domain mapping | |
| βββ forecast_mapper.py | |
| βββ anomaly_mapper.py | |
| βββ backtest_mapper.py | |
| ``` | |
| **Example Use Case**: | |
| ```python | |
| class ForecastUnivariateUseCase: | |
| """Orchestrates univariate forecasting workflow""" | |
| def __init__(self, forecast_service: ForecastService): | |
| self.forecast_service = forecast_service | |
| def execute(self, input_dto: ForecastInputDTO) -> ForecastOutputDTO: | |
| # 1. Validate DTO | |
| input_dto.validate() | |
| # 2. Map DTO β Domain | |
| series = mapper.to_time_series(input_dto) | |
| config = mapper.to_forecast_config(input_dto) | |
| # 3. Execute domain logic | |
| result = self.forecast_service.forecast_univariate(series, config) | |
| # 4. Map Domain β DTO | |
| output_dto = mapper.to_output_dto(result) | |
| return output_dto | |
| ``` | |
| **Key Principles**: | |
| - β **SRP**: Use cases orchestrate, don't implement logic | |
| - β **OCP**: New use cases without modifying existing | |
| - β **DIP**: Depends on domain interfaces | |
| --- | |
| ### 3. Domain Layer (Core) | |
| **Location**: `app/domain/` | |
| **Responsibilities**: | |
| - Define business rules | |
| - Implement core algorithms | |
| - Define domain models (entities, value objects) | |
| - Define interfaces (ports) | |
| **Components**: | |
| ``` | |
| app/domain/ | |
| βββ models/ # Domain models | |
| β βββ time_series.py # TimeSeries entity | |
| β βββ forecast_config.py # ForecastConfig value object | |
| β βββ forecast_result.py # ForecastResult entity | |
| β βββ anomaly.py # Anomaly models | |
| βββ services/ # Business logic | |
| β βββ forecast_service.py | |
| β βββ anomaly_service.py | |
| β βββ backtest_service.py | |
| βββ interfaces/ # Abstractions (ports) | |
| βββ forecast_model.py # IForecastModel | |
| βββ data_transformer.py # IDataTransformer | |
| ``` | |
| **Example Domain Service**: | |
| ```python | |
| class ForecastService: | |
| """Domain service for forecasting logic""" | |
| def __init__( | |
| self, | |
| model: IForecastModel, # Abstraction (DIP) | |
| transformer: IDataTransformer | |
| ): | |
| self.model = model | |
| self.transformer = transformer | |
| def forecast_univariate( | |
| self, | |
| series: TimeSeries, # Domain model | |
| config: ForecastConfig | |
| ) -> ForecastResult: | |
| # 1. Validate domain rules | |
| if not series.validate(): | |
| raise ValueError("Invalid time series") | |
| # 2. Transform to ML format | |
| context_df = self.transformer.build_context_df( | |
| series.values, series.timestamps | |
| ) | |
| # 3. Call ML model | |
| pred_df = self.model.predict( | |
| context_df, | |
| config.prediction_length, | |
| config.quantile_levels | |
| ) | |
| # 4. Transform back to domain | |
| result = self.transformer.parse_prediction_result(pred_df) | |
| # 5. Return domain model | |
| return ForecastResult(**result) | |
| ``` | |
| **Key Principles**: | |
| - β **SRP**: Each service has one business responsibility | |
| - β **DIP**: Depends on interfaces, not implementations | |
| - β **ISP**: Small, focused interfaces | |
| - β No dependencies on outer layers | |
| --- | |
| ### 4. Infrastructure Layer | |
| **Location**: `app/infrastructure/` | |
| **Responsibilities**: | |
| - Implement domain interfaces (adapters) | |
| - External service integration (ML models, databases) | |
| - Configuration management | |
| - Logging, monitoring | |
| **Components**: | |
| ``` | |
| app/infrastructure/ | |
| βββ ml/ # ML model implementations | |
| β βββ chronos_model.py # Chronos2 adapter | |
| β βββ model_factory.py # Factory pattern | |
| βββ config/ # Configuration | |
| β βββ settings.py # Pydantic settings | |
| βββ persistence/ # Data persistence (future) | |
| βββ cache.py # Caching layer | |
| ``` | |
| **Example Infrastructure**: | |
| ```python | |
| class ChronosModel(IForecastModel): | |
| """Adapter for Chronos-2 model (DIP)""" | |
| def __init__(self, model_id: str, device_map: str): | |
| self.pipeline = Chronos2Pipeline.from_pretrained( | |
| model_id, device_map=device_map | |
| ) | |
| def predict( | |
| self, | |
| context_df: pd.DataFrame, | |
| prediction_length: int, | |
| quantile_levels: List[float] | |
| ) -> pd.DataFrame: | |
| """Implements IForecastModel interface""" | |
| return self.pipeline.predict_df( | |
| context_df, | |
| prediction_length=prediction_length, | |
| quantile_levels=quantile_levels | |
| ) | |
| ``` | |
| **Factory Pattern**: | |
| ```python | |
| class ModelFactory: | |
| """Factory for creating forecast models (OCP)""" | |
| _models = { | |
| "chronos2": ChronosModel, | |
| # Future: "prophet": ProphetModel, | |
| # Future: "arima": ARIMAModel, | |
| } | |
| @classmethod | |
| def create(cls, model_type: str, **kwargs) -> IForecastModel: | |
| """Create model instance""" | |
| if model_type not in cls._models: | |
| raise ValueError(f"Unknown model: {model_type}") | |
| model_class = cls._models[model_type] | |
| return model_class(**kwargs) | |
| @classmethod | |
| def register_model(cls, name: str, model_class: Type[IForecastModel]): | |
| """Register new model (OCP - extension)""" | |
| cls._models[name] = model_class | |
| ``` | |
| **Key Principles**: | |
| - β **DIP**: Implements domain interfaces | |
| - β **OCP**: Factory allows extension without modification | |
| - β **SRP**: Each adapter has one external responsibility | |
| --- | |
| ## π¨ Design Patterns | |
| ### 1. Dependency Injection (DI) | |
| **Implementation**: FastAPI `Depends()` | |
| ```python | |
| # Define dependencies | |
| def get_forecast_model() -> IForecastModel: | |
| """Singleton model instance""" | |
| return ModelFactory.create("chronos2", model_id=settings.model_id) | |
| def get_forecast_service( | |
| model: IForecastModel = Depends(get_forecast_model), | |
| transformer: IDataTransformer = Depends(get_data_transformer) | |
| ) -> ForecastService: | |
| """Inject dependencies""" | |
| return ForecastService(model=model, transformer=transformer) | |
| # Use in routes | |
| @router.post("/forecast/univariate") | |
| async def forecast( | |
| use_case: ForecastUseCase = Depends(get_forecast_use_case) | |
| ): | |
| return use_case.execute(...) | |
| ``` | |
| **Benefits**: | |
| - β Loose coupling | |
| - β Easy testing (mock dependencies) | |
| - β Configurable at runtime | |
| --- | |
| ### 2. Factory Pattern | |
| **Implementation**: `ModelFactory` | |
| ```python | |
| # Create model | |
| model = ModelFactory.create("chronos2", model_id="amazon/chronos-2") | |
| # Extend without modifying factory | |
| ModelFactory.register_model("custom", CustomModel) | |
| model = ModelFactory.create("custom", ...) | |
| ``` | |
| **Benefits**: | |
| - β OCP compliance (Open for extension) | |
| - β Centralized model creation | |
| - β Easy to add new models | |
| --- | |
| ### 3. Repository Pattern (Implicit) | |
| **Implementation**: `DataFrameBuilder` | |
| ```python | |
| class DataFrameBuilder(IDataTransformer): | |
| """Repository-like interface for data""" | |
| def build_context_df(self, values, timestamps): | |
| """Build context from raw data""" | |
| ... | |
| def parse_prediction_result(self, pred_df): | |
| """Parse model output""" | |
| ... | |
| ``` | |
| **Benefits**: | |
| - β Data access abstraction | |
| - β Easy to swap data sources | |
| - β Testable with mocks | |
| --- | |
| ### 4. Strategy Pattern (Implicit) | |
| **Implementation**: `IForecastModel` interface | |
| ```python | |
| # Different strategies | |
| model1 = ChronosModel(...) | |
| model2 = ProphetModel(...) # Future | |
| # Same interface | |
| service = ForecastService(model=model1) # β | |
| service = ForecastService(model=model2) # β | |
| ``` | |
| **Benefits**: | |
| - β LSP compliance (Liskov Substitution) | |
| - β Interchangeable implementations | |
| - β Easy A/B testing | |
| --- | |
| ## π Data Flow | |
| ### Forecast Request Flow | |
| ``` | |
| 1. Client Request | |
| β | |
| POST /forecast/univariate | |
| Body: {"values": [100, 102, 105], "prediction_length": 3} | |
| 2. API Layer (Route) | |
| β | |
| - Validate request (Pydantic) | |
| - Inject use case | |
| β | |
| ForecastUnivariateUseCase | |
| 3. Application Layer (Use Case) | |
| β | |
| - Validate DTO | |
| - Map DTO β Domain models | |
| β | |
| ForecastService | |
| 4. Domain Layer (Service) | |
| β | |
| - Validate business rules | |
| - Call transformer | |
| β | |
| DataFrameBuilder.build_context_df() | |
| 5. Infrastructure Layer (Transformer) | |
| β | |
| - Build DataFrame | |
| β | |
| Back to Domain (Service) | |
| β | |
| - Call model | |
| β | |
| IForecastModel.predict() | |
| 6. Infrastructure Layer (Model) | |
| β | |
| - Chronos2Pipeline.predict_df() | |
| β | |
| Back to Domain (Service) | |
| β | |
| - Parse result | |
| β | |
| DataFrameBuilder.parse_prediction_result() | |
| 7. Domain Layer (Service) | |
| β | |
| - Create ForecastResult (domain model) | |
| β | |
| Back to Application (Use Case) | |
| 8. Application Layer (Use Case) | |
| β | |
| - Map Domain β DTO | |
| β | |
| Back to API (Route) | |
| 9. API Layer (Route) | |
| β | |
| - Serialize response (Pydantic) | |
| β | |
| 200 OK | |
| Body: {"timestamps": [...], "median": [...], "quantiles": {...}} | |
| 10. Client Response | |
| ``` | |
| **Key Observations**: | |
| - β Clear layer boundaries | |
| - β Each layer has specific responsibility | |
| - β Dependencies point inward | |
| - β Domain models never leak to API | |
| --- | |
| ## π Component Diagrams | |
| ### Forecasting Component Interaction | |
| ``` | |
| βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| β CLIENT β | |
| ββββββββββββββββββββββ¬βββββββββββββββββββββββββββββββββββββ | |
| β HTTP POST | |
| ββββββββββββΌβββββββββββ | |
| β ForecastController β (API Layer) | |
| ββββββββββββ¬βββββββββββ | |
| β execute() | |
| ββββββββββββΌβββββββββββββββ | |
| β ForecastUseCase β (Application Layer) | |
| ββββββββββββ¬βββββββββββββββ | |
| β forecast_univariate() | |
| ββββββββββββΌβββββββββββββββ | |
| β ForecastService β (Domain Layer) | |
| β ββββββββββββββββββββ β | |
| β β TimeSeries β β | |
| β β ForecastConfig β β | |
| β β ForecastResult β β | |
| β ββββββββββββββββββββ β | |
| ββββββββ¬βββββββββββ¬ββββββββ | |
| β β | |
| build_df() β β predict() | |
| β β | |
| ββββββββββββΌβββββ βββΌβββββββββββββββ | |
| β DataFrame β β ChronosModel β (Infrastructure) | |
| β Builder β β (IForecastModel)β | |
| ββββββββββββββββββ ββββββββββββββββββ | |
| β | |
| βββββββΌβββββββ | |
| β Chronos2 β | |
| β Pipeline β | |
| ββββββββββββββ | |
| ``` | |
| --- | |
| ## π― SOLID Principles | |
| ### Single Responsibility Principle (SRP) β | |
| **Each class has ONE reason to change** | |
| ```python | |
| # β Good: Separate responsibilities | |
| class ForecastService: | |
| """Only forecasting logic""" | |
| def forecast_univariate(self, series, config): | |
| ... | |
| class DataFrameBuilder: | |
| """Only data transformation""" | |
| def build_context_df(self, values): | |
| ... | |
| class ChronosModel: | |
| """Only ML inference""" | |
| def predict(self, context_df): | |
| ... | |
| ``` | |
| **Violations to avoid**: | |
| ```python | |
| # β Bad: Multiple responsibilities | |
| class ForecastServiceBad: | |
| def forecast(self, values): | |
| # Data transformation (should be separate) | |
| df = self._build_dataframe(values) | |
| # ML inference (should be separate) | |
| pred = self._call_model(df) | |
| # HTTP response (should be in API layer) | |
| return {"status": 200, "data": pred} | |
| ``` | |
| --- | |
| ### Open/Closed Principle (OCP) β | |
| **Open for extension, closed for modification** | |
| ```python | |
| # β Good: Extension without modification | |
| class ModelFactory: | |
| _models = {"chronos2": ChronosModel} | |
| @classmethod | |
| def register_model(cls, name, model_class): | |
| """Extend by registering new models""" | |
| cls._models[name] = model_class | |
| # Add new model without modifying factory | |
| ModelFactory.register_model("prophet", ProphetModel) | |
| ``` | |
| **Example extension**: | |
| ```python | |
| # New model type (no changes to existing code) | |
| class ProphetModel(IForecastModel): | |
| def predict(self, context_df, prediction_length, quantile_levels): | |
| # Prophet-specific implementation | |
| ... | |
| # Register and use | |
| ModelFactory.register_model("prophet", ProphetModel) | |
| model = ModelFactory.create("prophet") | |
| service = ForecastService(model=model) # Works! | |
| ``` | |
| --- | |
| ### Liskov Substitution Principle (LSP) β | |
| **Subtypes must be substitutable for their base types** | |
| ```python | |
| # β Good: Any IForecastModel works | |
| def forecast_with_any_model(model: IForecastModel): | |
| result = model.predict(df, 7, [0.5]) | |
| # Works with ChronosModel, ProphetModel, etc. | |
| return result | |
| # All implementations honor the contract | |
| model1 = ChronosModel(...) | |
| model2 = ProphetModel(...) # Future | |
| forecast_with_any_model(model1) # β Works | |
| forecast_with_any_model(model2) # β Works | |
| ``` | |
| --- | |
| ### Interface Segregation Principle (ISP) β | |
| **Clients shouldn't depend on methods they don't use** | |
| ```python | |
| # β Good: Small, focused interfaces | |
| class IForecastModel(ABC): | |
| """Only forecasting methods""" | |
| @abstractmethod | |
| def predict(self, context_df, prediction_length, quantile_levels): | |
| pass | |
| @abstractmethod | |
| def get_model_info(self): | |
| pass | |
| class IDataTransformer(ABC): | |
| """Only transformation methods""" | |
| @abstractmethod | |
| def build_context_df(self, values): | |
| pass | |
| @abstractmethod | |
| def parse_prediction_result(self, pred_df): | |
| pass | |
| ``` | |
| **Violations to avoid**: | |
| ```python | |
| # β Bad: Fat interface | |
| class IForecastModelBad(ABC): | |
| @abstractmethod | |
| def predict(self, ...): pass | |
| @abstractmethod | |
| def build_dataframe(self, ...): pass # Should be separate | |
| @abstractmethod | |
| def validate_input(self, ...): pass # Should be separate | |
| @abstractmethod | |
| def format_response(self, ...): pass # Should be separate | |
| ``` | |
| --- | |
| ### Dependency Inversion Principle (DIP) β | |
| **Depend on abstractions, not concretions** | |
| ```python | |
| # β Good: Depends on abstraction | |
| class ForecastService: | |
| def __init__( | |
| self, | |
| model: IForecastModel, # Abstract interface | |
| transformer: IDataTransformer # Abstract interface | |
| ): | |
| self.model = model | |
| self.transformer = transformer | |
| ``` | |
| **Violations to avoid**: | |
| ```python | |
| # β Bad: Depends on concrete implementation | |
| class ForecastServiceBad: | |
| def __init__(self): | |
| # Coupled to Chronos2Pipeline directly | |
| self.model = Chronos2Pipeline.from_pretrained(...) | |
| # Coupled to DataFrameBuilder directly | |
| self.transformer = DataFrameBuilder() | |
| ``` | |
| --- | |
| ## π§ͺ Testing Strategy | |
| ### Test Pyramid | |
| ``` | |
| β±β² | |
| β± β² | |
| β± E2E β² 10 tests (Integration) | |
| β±βββββββββ² | |
| β± β² | |
| β± Integrationβ² 25 tests (API, Services) | |
| β±βββββββββββββββ² | |
| β± β² | |
| β± Unit Tests β² 45 tests (Fast, Isolated) | |
| β±βββββββββββββββββββββ² | |
| ``` | |
| ### Unit Tests (45+) | |
| **Focus**: Individual components in isolation | |
| **Tools**: pytest, unittest.mock | |
| **Example**: | |
| ```python | |
| def test_forecast_service(mock_model, mock_transformer): | |
| """Test service logic with mocks""" | |
| service = ForecastService(mock_model, mock_transformer) | |
| series = TimeSeries(values=[100, 102, 105]) | |
| config = ForecastConfig(prediction_length=3) | |
| result = service.forecast_univariate(series, config) | |
| assert len(result.timestamps) == 3 | |
| mock_model.predict.assert_called_once() | |
| ``` | |
| ### Integration Tests (25+) | |
| **Focus**: Multiple components working together | |
| **Tools**: FastAPI TestClient | |
| **Example**: | |
| ```python | |
| @patch('app.infrastructure.ml.chronos_model.Chronos2Pipeline') | |
| def test_forecast_endpoint_e2e(mock_pipeline): | |
| """Test complete API flow""" | |
| mock_pipeline.predict_df.return_value = sample_df | |
| response = client.post("/forecast/univariate", json={ | |
| "values": [100, 102, 105], | |
| "prediction_length": 3 | |
| }) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert "timestamps" in data | |
| ``` | |
| ### Test Coverage | |
| - **Domain Layer**: 80% | |
| - **Application Layer**: 70% | |
| - **Infrastructure Layer**: 85% | |
| - **API Layer**: 90% | |
| - **Overall**: ~80% | |
| --- | |
| ## π Deployment Architecture | |
| ### Container Architecture | |
| ``` | |
| βββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| β Docker Container β | |
| β βββββββββββββββββββββββββββββββββββββββββββββ β | |
| β β FastAPI Application β β | |
| β β βββββββββββββββ ββββββββββββββββ β β | |
| β β β API Server β β Static Filesβ β β | |
| β β β (Port 8000)β β (Excel UI) β β β | |
| β β βββββββββββββββ ββββββββββββββββ β β | |
| β βββββββββββββββββββββββββββββββββββββββββββββ β | |
| β βββββββββββββββββββββββββββββββββββββββββββββ β | |
| β β Chronos-2 Model β β | |
| β β (amazon/chronos-2) β β | |
| β βββββββββββββββββββββββββββββββββββββββββββββ β | |
| βββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| β | |
| βββββββββββΌββββββββββ | |
| β HuggingFace β | |
| β Spaces β | |
| β (Public URL) β | |
| βββββββββββββββββββββ | |
| ``` | |
| ### Environment Configuration | |
| **Production**: | |
| - HuggingFace Spaces | |
| - CPU inference (free tier) | |
| - Public HTTPS endpoint | |
| **Development**: | |
| - Local Docker | |
| - Hot reload | |
| - Debug mode | |
| **Testing**: | |
| - CI/CD pipeline | |
| - Automated tests | |
| - Coverage reports | |
| --- | |
| ## π Performance Considerations | |
| ### Model Loading | |
| ```python | |
| # Singleton pattern for model (loaded once) | |
| _model_instance = None | |
| def get_forecast_model(): | |
| global _model_instance | |
| if _model_instance is None: | |
| _model_instance = ModelFactory.create("chronos2") | |
| return _model_instance | |
| ``` | |
| ### Caching Strategy (Future) | |
| ```python | |
| # Redis cache for repeated forecasts | |
| @cache(ttl=3600) | |
| def forecast_univariate(values, prediction_length): | |
| ... | |
| ``` | |
| ### Async Processing (Future) | |
| ```python | |
| # Background tasks for long forecasts | |
| @router.post("/forecast/async") | |
| async def forecast_async(background_tasks: BackgroundTasks): | |
| background_tasks.add_task(long_forecast) | |
| return {"task_id": "..."} | |
| ``` | |
| --- | |
| ## π Security Considerations | |
| ### Input Validation | |
| - β Pydantic validation at API layer | |
| - β Domain validation in services | |
| - β Type hints throughout | |
| ### Error Handling | |
| - β Structured error responses | |
| - β No sensitive data in errors | |
| - β Logging for debugging | |
| ### CORS Configuration | |
| ```python | |
| # Configurable CORS | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=settings.cors_origins, | |
| allow_methods=["GET", "POST"], | |
| allow_headers=["*"] | |
| ) | |
| ``` | |
| --- | |
| ## π References | |
| ### Architecture Patterns | |
| - **Clean Architecture**: Robert C. Martin | |
| - **Domain-Driven Design**: Eric Evans | |
| - **SOLID Principles**: Robert C. Martin | |
| ### Frameworks & Libraries | |
| - **FastAPI**: https://fastapi.tiangolo.com/ | |
| - **Chronos**: https://github.com/amazon-science/chronos-forecasting | |
| - **Pydantic**: https://docs.pydantic.dev/ | |
| --- | |
| ## π Learning Resources | |
| ### For New Developers | |
| 1. Read `DEVELOPMENT.md` for setup instructions | |
| 2. Review `API.md` for endpoint documentation | |
| 3. Study test examples in `tests/` | |
| 4. Start with simple features (add endpoint) | |
| ### Architecture Books | |
| - "Clean Architecture" by Robert C. Martin | |
| - "Domain-Driven Design" by Eric Evans | |
| - "Patterns of Enterprise Application Architecture" by Martin Fowler | |
| --- | |
| **Last Updated**: 2025-11-09 | |
| **Version**: 3.0.0 | |
| **Maintainer**: Claude AI | |