ttzzs's picture
Deploy Chronos2 Forecasting API v3.0.0 with new SOLID architecture
c40c447 verified
# πŸ›οΈ 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