Proyecto: predicción y API con FastAPI

Un modelo que solo vive en un notebook no le sirve a nadie más. Para que un equipo comercial consulte un precio estimado, el modelo debe estar detrás de algo que hable HTTP: una API. Aquí recorres ese camino, desde un CSV de catorce filas hasta un JSON que puedes pedir con curl.

¿Qué es un proyecto de Machine Learning con API?

Es un sistema con tres piezas encadenadas: un conjunto de datos histórico, un artefacto entrenado —el pipeline con sus transformaciones y coeficientes— y un servicio que recibe datos por la red, los valida y devuelve el resultado en un formato acordado.

La palabra clave es acordado. Una API no es solo código que funciona en tu máquina: es un contrato. Si promete un precio_usd numérico y un día devuelve "mil doscientos", el cliente se rompe aunque el modelo sea exactamente el mismo.

• No es un notebook: un notebook ejecuta celdas en un orden que solo tú conoces; una API ejecuta una función por petición.
• No es solo un archivo .pkl: el archivo no valida, no distingue un distrito nuevo de una errata y no declara su versión.
• Sí es un producto pequeño: entrada, salida, errores documentados, versión y pruebas.

¿Para qué sirve?

Una API convierte un modelo en algo que otros equipos usan sin instalar scikit-learn ni conocer tus features. Sirve para cuatro cosas:

• Integrar: el frontend o el CRM llaman a POST /predict y reciben un número.
• Centralizar el preprocesado: si el escalado vive en el pipeline, ningún cliente puede olvidarse de estandarizar.
• Controlar la entrada: habitaciones: -3 no llega al modelo; Pydantic lo detiene antes con un mensaje concreto.
• Medir: cada petición registrada dice qué se consulta, cuánto tarda y cuándo el modelo envejece.

El caso de la lección: una inmobiliaria quiere estimar el alquiler mensual de un departamento en Lima. Hoy un analista revisa cada consulta a mano; con la API, la tasación inicial sale en el momento.

¿Cómo funciona por dentro?

El cliente envía un JSON. Uvicorn —el servidor ASGI— acepta la conexión y entrega el cuerpo a FastAPI, que busca POST /predict y lo pasa por el modelo Pydantic Departamento. Si un campo falta, sobra o incumple una restricción, la petición muere con un 422 y tu función ni se ejecuta.

Si la validación pasa, tu función construye un DataFrame de una fila con los nombres de columna que el pipeline aprendió. El ColumnTransformer aplica one-hot a distrito y escalado a las numéricas; el regresor multiplica esa fila por sus coeficientes y devuelve un float de NumPy.

Dos decisiones sostienen todo lo anterior. Empaquetar preprocesado y modelo en un Pipeline hace imposible entrenar con una transformación y servir con otra. Cargar el artefacto una sola vez evita leer disco y reconstruir objetos en cada consulta.

Estructura del proyecto

Decide dónde va cada cosa antes de escribir lógica: lo que entrena no es lo que sirve. El script puede tardar minutos y usar toda la CPU; el servicio debe arrancar en segundos y no sabe entrenar.

1234567891011121314151617

precios-api/
├── api/
│   ├── __init__.py
│   ├── esquemas.py      # contrato de entrada y salida
│   ├── modelo.py        # carga del artefacto
│   └── main.py          # rutas y ciclo de vida
├── data/
│   └── alquileres.csv
├── models/
│   └── precio_v1.joblib
├── scripts/
│   ├── datos.py         # dataset pequeño de trabajo
│   └── entrenar.py      # pipeline + métricas + joblib
├── tests/
│   └── test_api.py
├── requirements.txt
└── Dockerfile

api/__init__.py puede estar vacío, pero convierte la carpeta en un paquete importable. Los datos se leen desde scripts/: si el servicio leyera el CSV, un cambio en el dataset alteraría las predicciones sin reentrenar.

Sintaxis: esquemas, rutas y validación

Un modelo Pydantic declara la forma de los datos. Cada anotación es una comprobación en tiempo de ejecución y, a la vez, documentación: FastAPI genera un esquema OpenAPI visible en /docs. Esa es la pieza que separa una API de un print por HTTP.

1234567891011121314151617

from pydantic import BaseModel, ConfigDict, Field


class Departamento(BaseModel):
    """Datos que envía el cliente. `extra="forbid"` rechaza campos desconocidos."""

    model_config = ConfigDict(extra="forbid")

    distrito: str = Field(min_length=2, max_length=40, examples=["Miraflores"])
    area_m2: float = Field(gt=10, le=1000, examples=[85.0])
    habitaciones: int = Field(ge=0, le=10, examples=[2])
    antiguedad: int = Field(ge=0, le=120, examples=[12])


class Prediccion(BaseModel):
    precio_usd: float
    modelo_version: str

Field(gt=10, le=1000) significa «mayor que 10 y hasta 1000». Cada violación produce un 422 con la ruta del campo (body, area_m2) y el motivo. extra="forbid" es deliberado: si un cliente envía piscina: true, mejor un error que ignorarlo.

Ejemplo básico: mirar el dataset antes de modelar

Con catorce filas no vas a ganar un concurso, y no hace falta. Necesitas un dataset con la forma del real: una categórica, tres numéricas y un objetivo continuo. El análisis mínimo mira filas, tipos, nulos y rangos.

12345678910111213141516171819202122232425

import pandas as pd

FILAS = [
    {"distrito": "Miraflores", "area_m2": 85.0, "habitaciones": 2, "antiguedad": 12, "precio_usd": 1250.0},
    {"distrito": "Miraflores", "area_m2": 120.0, "habitaciones": 3, "antiguedad": 5, "precio_usd": 1900.0},
    {"distrito": "Surco", "area_m2": 95.0, "habitaciones": 2, "antiguedad": 20, "precio_usd": 1100.0},
    {"distrito": "Surco", "area_m2": 140.0, "habitaciones": 3, "antiguedad": 8, "precio_usd": 1650.0},
    {"distrito": "Surco", "area_m2": 70.0, "habitaciones": 1, "antiguedad": 30, "precio_usd": 780.0},
    {"distrito": "San Isidro", "area_m2": 100.0, "habitaciones": 2, "antiguedad": 15, "precio_usd": 1500.0},
    {"distrito": "San Isidro", "area_m2": 130.0, "habitaciones": 3, "antiguedad": 6, "precio_usd": 2100.0},
    {"distrito": "Jesus Maria", "area_m2": 78.0, "habitaciones": 2, "antiguedad": 25, "precio_usd": 900.0},
    {"distrito": "Jesus Maria", "area_m2": 60.0, "habitaciones": 1, "antiguedad": 40, "precio_usd": 650.0},
    {"distrito": "Lince", "area_m2": 68.0, "habitaciones": 2, "antiguedad": 35, "precio_usd": 720.0},
    {"distrito": "Lince", "area_m2": 55.0, "habitaciones": 1, "antiguedad": 45, "precio_usd": 560.0},
    {"distrito": "Barranco", "area_m2": 90.0, "habitaciones": 2, "antiguedad": 18, "precio_usd": 1150.0},
    {"distrito": "Barranco", "area_m2": 110.0, "habitaciones": 3, "antiguedad": 10, "precio_usd": 1480.0},
    {"distrito": "La Molina", "area_m2": 150.0, "habitaciones": 4, "antiguedad": 7, "precio_usd": 2000.0},
]

df = pd.DataFrame(FILAS)
print(df.shape)                       # (14, 5)
print(df.dtypes.to_string())
print(df.isna().sum().sum(), "valores nulos")
print(df["precio_usd"].describe().round(0).to_string())
print(df["distrito"].value_counts().to_string())

La salida confirma 14 filas, ningún nulo y seis distritos con una a tres observaciones. Fíjate en Jesus Maria sin tilde: es deliberado. Si escribes Jesús María al entrenar y Jesus Maria al predecir, el one-hot crea dos columnas y el segundo caso cae como distrito desconocido.

describe() es tu red de seguridad. Un area_m2 con mínimo 0 o máximo 40 000 no es una curiosidad: es un error de captura que arrastra la escala de todo el modelo. Revisa también el objetivo: un alquiler de 12 dólares indica moneda equivocada.

Ejemplo intermedio: pipeline reproducible y serialización

El ColumnTransformer aplica transformaciones distintas a grupos de columnas mediante tuplas (nombre, transformador, columnas). El Pipeline encadena ese preprocesado con el estimador: un solo objeto con fit y predict, y eso es exactamente lo que se guarda en disco.

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758

from pathlib import Path

import joblib
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.linear_model import Ridge
from sklearn.metrics import mean_absolute_error, r2_score
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler

from datos import FILAS

CATEGORICAS = ["distrito"]
NUMERICAS = ["area_m2", "habitaciones", "antiguedad"]
OBJETIVO = "precio_usd"


def construir_pipeline() -> Pipeline:
    preprocesado = ColumnTransformer(
        transformers=[
            ("cat", OneHotEncoder(handle_unknown="ignore"), CATEGORICAS),
            ("num", StandardScaler(), NUMERICAS),
        ]
    )
    return Pipeline(
        steps=[
            ("prep", preprocesado),
            ("reg", Ridge(alpha=1.0)),
        ]
    )


def main() -> None:
    df = pd.DataFrame(FILAS)
    X = df[CATEGORICAS + NUMERICAS]
    y = df[OBJETIVO]

    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.25, random_state=42
    )

    modelo = construir_pipeline()
    modelo.fit(X_train, y_train)

    predicho = modelo.predict(X_test)
    print(f"MAE: {mean_absolute_error(y_test, predicho):.0f} USD")
    print(f"R2:  {r2_score(y_test, predicho):.3f}")

    destino = Path("models")
    destino.mkdir(exist_ok=True)
    ruta = destino / "precio_v1.joblib"
    joblib.dump(modelo, ruta)
    print("Modelo guardado en", ruta)


if __name__ == "__main__":
    main()

Ejecuta python scripts/entrenar.py desde la raíz: obtendrás dos métricas y un archivo de pocos kilobytes. Con once filas de entrenamiento el R2 es ruidoso y puede salir negativo; sirve para comprobar el flujo, no para presumir de precisión.

handle_unknown="ignore" merece una explicación. Sin esa opción, predecir para Pueblo Libre lanza ValueError: Found unknown categories. Con ella, las columnas de distrito quedan a cero y el modelo responde con las numéricas: peor predicción, pero respuesta.

Caso práctico: el endpoint /predict

Primero, un módulo que localice y cargue el artefacto. La ruta se calcula con __file__, no con el directorio de trabajo: uvicorn puede arrancar desde cualquier carpeta. Es el primer fallo que aparece al contenedorizar una API.

12345678910111213141516

from pathlib import Path
from typing import Any

import joblib

RAIZ = Path(__file__).resolve().parent.parent
RUTA_MODELO = RAIZ / "models" / "precio_v1.joblib"


def cargar_modelo(ruta: Path = RUTA_MODELO) -> Any:
    """Devuelve el pipeline serializado. Falla pronto si el archivo no existe."""
    if not ruta.exists():
        raise FileNotFoundError(
            f"No se encontró el modelo en {ruta}. Ejecuta scripts/entrenar.py primero."
        )
    return joblib.load(ruta)

El lifespan es una función asíncrona que FastAPI ejecuta al arrancar y al apagar: lo que está antes del yield ocurre una vez, cuando el proceso sube. Es el sitio correcto para el modelo; cargarlo dentro de predecir leería el disco en cada petición.

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748

from contextlib import asynccontextmanager
from typing import Any, AsyncIterator

import pandas as pd
from fastapi import FastAPI, HTTPException

from api.esquemas import Departamento, Prediccion
from api.modelo import cargar_modelo

VERSION = "1.0.0"
modelo: Any | None = None


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    global modelo
    modelo = cargar_modelo()          # una sola vez, al arrancar
    yield
    modelo = None                     # liberar al apagar


app = FastAPI(
    title="API de precios de alquiler",
    version=VERSION,
    lifespan=lifespan,
)


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok" if modelo is not None else "sin_modelo"}


@app.post("/predict", response_model=Prediccion)
def predecir(departamento: Departamento) -> Prediccion:
    if modelo is None:
        raise HTTPException(status_code=503, detail="El modelo no está cargado")

    entrada = pd.DataFrame([departamento.model_dump()])

    try:
        precio = float(modelo.predict(entrada)[0])
    except Exception as error:  # el detalle va al log, no al cliente
        raise HTTPException(
            status_code=500, detail="No se pudo calcular la predicción"
        ) from error

    return Prediccion(precio_usd=round(precio, 2), modelo_version=VERSION)

Tres detalles que no son menores. float(...) es obligatorio porque predict devuelve numpy.float64. round(..., 2) evita respuestas como 1243.7100000000002. Y el except no devuelve str(error) al cliente: los mensajes internos revelan rutas y versiones.

Probar la API en local

Instala las dependencias en un entorno virtual y arranca con recarga automática mientras desarrollas. --reload reinicia el proceso al guardar un archivo y vuelve a ejecutar el lifespan, así que recarga el modelo cuando reentrenas.

12345678

python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install "fastapi[standard]" scikit-learn pandas joblib pytest httpx
python scripts/entrenar.py
uvicorn api.main:app --reload --port 8000

# INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
# INFO:     Application startup complete.

Abre http://127.0.0.1:8000/docs: verás los endpoints y un botón Try it out. Ese formulario se genera a partir de tus modelos Pydantic. Desde la terminal, el mismo contrato se prueba así:

123456789101112

curl -s http://127.0.0.1:8000/health
# {"status":"ok"}

curl -s -X POST http://127.0.0.1:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"distrito":"Miraflores","area_m2":85,"habitaciones":2,"antiguedad":12}'
# {"precio_usd":1243.71,"modelo_version":"1.0.0"}

curl -s -o /dev/null -w "%{http_code}\n" -X POST http://127.0.0.1:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"distrito":"Miraflores","area_m2":-5,"habitaciones":2,"antiguedad":12}'
# 422

El número exacto depende de tu entrenamiento y de la versión de scikit-learn; lo que debe coincidir es la forma: un objeto con precio_usd y modelo_version. Si recibes {"detail":[...]}, la validación rechazó algo y el arreglo está en esa lista.

Visualización: de un JSON a una matriz de inferencia

Esta es la parte que más cuesta imaginar. El cliente envía cuatro campos; el ColumnTransformer los convierte en una fila con una columna por distrito conocido más tres numéricas. El modelo nunca ve Miraflores: ve un 1 en su columna y ceros en las demás.

La matriz tiene una fila porque la API atiende una consulta a la vez. Un endpoint de lote produciría cincuenta: el pipeline no cambia, solo el número de filas.

—

Tests: comprobar el contrato sin servidor

TestClient ejecuta la aplicación en el mismo proceso, sin abrir un puerto. Úsalo como context manager: dentro del with, FastAPI ejecuta el lifespan y el modelo queda cargado. Sin with, las peticiones reciben 503 porque modelo sigue en None.

12345678910111213141516171819202122232425262728293031323334353637383940414243

import pytest
from fastapi.testclient import TestClient

from api.main import app

VALIDO = {
    "distrito": "Miraflores",
    "area_m2": 85.0,
    "habitaciones": 2,
    "antiguedad": 12,
}


@pytest.fixture
def cliente():
    # El context manager dispara el lifespan: el modelo se carga al entrar.
    with TestClient(app) as c:
        yield c


def test_health(cliente):
    respuesta = cliente.get("/health")
    assert respuesta.status_code == 200
    assert respuesta.json() == {"status": "ok"}


def test_prediccion(cliente):
    respuesta = cliente.post("/predict", json=VALIDO)
    assert respuesta.status_code == 200
    cuerpo = respuesta.json()
    assert cuerpo["precio_usd"] > 0
    assert cuerpo["modelo_version"] == "1.0.0"


def test_area_fuera_de_rango(cliente):
    respuesta = cliente.post("/predict", json={**VALIDO, "area_m2": -5})
    assert respuesta.status_code == 422
    assert respuesta.json()["detail"][0]["loc"] == ["body", "area_m2"]


def test_campo_desconocido(cliente):
    respuesta = cliente.post("/predict", json={**VALIDO, "piscina": True})
    assert respuesta.status_code == 422

Ejecuta pytest -q: los cuatro tests pasan en menos de un segundo. No comprueban el valor exacto de la predicción —eso ataría las pruebas a un entrenamiento concreto— sino el contrato: códigos de estado, campos, tipos y errores.

Variantes del proyecto: clasificador de textos y recomendador

La arquitectura no cambia cuando cambia el problema: pipeline reproducible, artefacto versionado, esquema de entrada, endpoint y tests de contrato. Lo que cambia es el modelo y la forma de la respuesta.

• Clasificador de textos: cambia el ColumnTransformer por TfidfVectorizer y Ridge por LogisticRegression. El endpoint recibe {"texto": "..."} y devuelve etiqueta y probabilidad.
• Recomendador: factorización de matrices y GET /recomendar/{usuario_id}?k=5. La métrica deja de ser MAE y pasa a precisión@k.
• Red neuronal: mismo contrato HTTP, pero el state_dict de PyTorch y el vocabulario se serializan por separado. Olvidar el preprocesamiento produce tensores distintos a los del entrenamiento.

Errores comunes

1234

@app.post("/predict")
def predecir(departamento: Departamento):
    modelo = joblib.load("models/precio_v1.joblib")  # cada petición
    return modelo.predict(pd.DataFrame([departamento.model_dump()]))[0]

123456

@asynccontextmanager
async def lifespan(app: FastAPI):
    global modelo
    modelo = cargar_modelo()
    yield
    modelo = None

Buenas prácticas

• Separa entrenamiento de servicio: scripts/ produce artefactos; api/ solo los consume.
• Un pipeline, un artefacto, una versión: el nombre incluye la versión y la respuesta la declara.
• Valida en la frontera: un error de entrada debe ser un 422, no un 500.
• Falla al arrancar, no al servir: si el modelo no carga, el proceso no debe quedarse vivo.
• No filtres mensajes internos: registra el detalle, devuelve un texto genérico.
• Prueba el contrato, no los decimales: las métricas van en el entrenamiento.

Ejercicios

@app.("/predict", response_model=Prediccion)
def predecir(departamento: Departamento) -> Prediccion:
    if modelo is :        
        raise HTTPException(status_code=503, detail="El modelo no está cargado")

    entrada = pd.DataFrame([departamento.()])
    precio = float(modelo.predict(entrada)[0])
    return Prediccion(precio_usd=round(precio, 2), modelo_version=VERSION)

Soluciones explicadas

En el endpoint de lote, la decisión importante es dónde se valida el tamaño. Un if len(lote.items) > 50 dentro de la función devolvería un 400 con un texto inventado. Al declararlo en el campo, el error es un 422 con la misma estructura que el resto y aparece en /docs.

En la línea base, DummyRegressor también necesita un fit: no aprende nada, pero exige la matriz para conocer las filas. Y cuidado con los tests: precio_usd > 0 deja pasar un disparate de nueve millones; añade una banda de cordura del EDA, como 300 < precio < 5000.

Resumen

• Un proyecto con API separa datos, entrenamiento, artefacto, servicio y pruebas.
• El Pipeline con ColumnTransformer garantiza la misma transformación en entrenamiento e inferencia.
• La ruta del modelo se calcula con Path(__file__), nunca de forma relativa.
• El modelo se carga una sola vez en el lifespan; hacerlo por petición multiplica la latencia.
• Pydantic convierte un dato inválido en un 422 y evita que llegue a tu función.
• TestClient como context manager ejecuta el lifespan y prueba el contrato sin servidor.

Evaluación