Spaces:
Sleeping
Sleeping
File size: 2,819 Bytes
b1f3a3c | 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 | from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import RedirectResponse
from app.database import engine
from app.ml_models import load_models
from app.routes import customers, predictions
from sqlalchemy import text
# load ML models on startup
models_loaded = load_models()
# create fastapi app
app = FastAPI(
title="E-Commerce Customer Intelligence API",
description="""
This API provides customer insights from your e-commerce data.
## Features
* **Customer Segmentation** - Predict customer segments using K-Means
* **CLV Prediction** - Predict Customer Lifetime Value
* **Customer Data** - Access customer information from database
## Models
* K-Means Clustering (4 segments)
* Random Forest CLV Predictor (95% accuracy)
""",
version="1.0.0",
contact={
"name": "Your Name",
"email": "your.email@example.com",
},
)
# Redirect root to Swagger UI
@app.get("/", include_in_schema=False)
async def redirect_to_docs():
"""Redirect root to Swagger UI"""
return RedirectResponse(url="/docs")
# add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# include routers
app.include_router(customers.router)
app.include_router(predictions.router)
@app.get("/health", tags=["Health"])
def health_check():
"""Check if API and database are working"""
db_status = "unknown"
try:
with engine.connect() as conn:
conn.execute(text("SELECT 1"))
db_status = "connected"
except Exception as e:
db_status = f"error: {str(e)}"
return {
"status": "healthy",
"models_loaded": {
"kmeans": models_loaded,
"clv": models_loaded
},
"database": db_status,
"timestamp": "2024-01-01T00:00:00Z"
}
@app.get("/info", tags=["Info"])
def api_info():
"""Get API information and available endpoints"""
return {
"name": "E-Commerce Customer Intelligence API",
"version": "1.0.0",
"endpoints": {
"GET /": "Redirects to Swagger UI",
"GET /health": "Health check",
"GET /info": "This information",
"GET /customers": "List all customers",
"GET /customers/{id}": "Get customer by ID",
"GET /customers/{id}/transactions": "Get customer transactions",
"POST /predict/segment": "Predict customer segment from RFM",
"POST /predict/clv": "Predict Customer Lifetime Value"
},
"models": {
"customer_segmentation": "K-Means (4 clusters)",
"clv_prediction": "Random Forest (95% accuracy)"
}
} |