ttzzs's picture
Deploy Chronos2 Forecasting API v3.0.0 with new SOLID architecture
c40c447 verified
"""
Chronos-2 Forecasting API - Clean Architecture Version 3.0
Este es el punto de entrada de la aplicación, refactorizado siguiendo
Clean Architecture y principios SOLID.
Características:
- Arquitectura en capas (Presentation, Application, Domain, Infrastructure)
- Dependency Injection completa
- Separación de responsabilidades
- Código mantenible y testeable
"""
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse
from fastapi.middleware.cors import CORSMiddleware
import os
from app.infrastructure.config.settings import get_settings
from app.utils.logger import setup_logger
# Import routers
from app.api.routes import (
health_router,
forecast_router,
anomaly_router,
backtest_router
)
logger = setup_logger(__name__)
settings = get_settings()
# ============================================================================
# Create FastAPI App
# ============================================================================
app = FastAPI(
title=settings.api_title,
version=settings.api_version,
description=settings.api_description,
docs_url="/docs",
redoc_url="/redoc",
openapi_url="/openapi.json"
)
# ============================================================================
# Middleware
# ============================================================================
# CORS Middleware
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# ============================================================================
# API Routes
# ============================================================================
# Health check endpoint (temporal, será movido a routes/health.py)
@app.get("/health", tags=["Health"])
async def health_check():
"""Check if the API is running and model is loaded."""
from app.api.dependencies import get_forecast_model
try:
model = get_forecast_model()
model_info = model.get_model_info()
return {
"status": "ok",
"version": settings.api_version,
"model": model_info
}
except Exception as e:
logger.error(f"Health check failed: {e}")
return {
"status": "error",
"version": settings.api_version,
"error": str(e)
}
# Include routers
app.include_router(health_router)
app.include_router(forecast_router)
app.include_router(anomaly_router)
app.include_router(backtest_router)
# ============================================================================
# Static Files (Excel Add-in)
# ============================================================================
if os.path.exists(settings.static_dir):
logger.info(f"Mounting static files from: {settings.static_dir}")
# Mount subdirectories
for subdir in ["assets", "taskpane", "commands"]:
path = os.path.join(settings.static_dir, subdir)
if os.path.exists(path):
app.mount(f"/{subdir}", StaticFiles(directory=path), name=subdir)
logger.info(f"Mounted /{subdir}")
# Manifest file
manifest_path = os.path.join(settings.static_dir, "manifest.xml")
if os.path.exists(manifest_path):
@app.get("/manifest.xml")
async def get_manifest():
"""Serve Excel Add-in manifest."""
return FileResponse(manifest_path, media_type="application/xml")
logger.info("Manifest endpoint registered")
else:
logger.warning(f"Static directory not found: {settings.static_dir}")
# ============================================================================
# Startup/Shutdown Events
# ============================================================================
@app.on_event("startup")
async def startup_event():
"""Initialize resources on startup."""
logger.info("=" * 60)
logger.info(f"🚀 {settings.api_title} v{settings.api_version}")
logger.info("=" * 60)
logger.info("Architecture: Clean Architecture (4 layers)")
logger.info("Principles: SOLID")
logger.info(f"Model: {settings.model_id}")
logger.info(f"Device: {settings.device_map}")
logger.info("=" * 60)
# Pre-load model
try:
from app.api.dependencies import get_forecast_model
logger.info("Pre-loading forecast model...")
model = get_forecast_model()
logger.info(f"✅ Model loaded: {model.get_model_info()}")
except Exception as e:
logger.error(f"❌ Failed to load model: {e}")
logger.error("API will start but forecasting will fail until model loads")
@app.on_event("shutdown")
async def shutdown_event():
"""Cleanup resources on shutdown."""
logger.info("=" * 60)
logger.info("Shutting down Chronos-2 API...")
logger.info("=" * 60)
# ============================================================================
# Root Endpoint
# ============================================================================
@app.get("/", tags=["Info"])
async def root():
"""API information and documentation links."""
return {
"name": settings.api_title,
"version": settings.api_version,
"description": settings.api_description,
"docs": "/docs",
"health": "/health",
"architecture": "Clean Architecture with SOLID principles",
"layers": {
"presentation": "FastAPI (app/api/)",
"application": "Use Cases (app/application/)",
"domain": "Business Logic (app/domain/)",
"infrastructure": "External Services (app/infrastructure/)"
}
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"app.main_v3:app",
host="0.0.0.0",
port=settings.api_port,
reload=True,
log_level=settings.log_level.lower()
)