Spaces:
Build error
Build error
File size: 5,929 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 |
"""
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()
)
|