Lead Classification API 🚀

Production-ready MLOps pipeline and REST API for real-time sales lead classification. Designed under Clean Architecture principles and optimized to process high-throughput workloads (simulated ingestion rate of 50,000 leads per month).

---

📋 Project Overview

The objective is to classify sales leads into "likely to convert" or "unlikely to convert" to optimize marketing and sales team resources.

• Model: A Scikit-Learn RandomForestClassifier pipeline trained with realistic synthetic categorical and numerical features.
• Serving Layer: A FastAPI service optimized with lifespan hooks for model residency in RAM and non-blocking endpoint execution.
• Packaging: Containerized via a multi-stage Dockerfile running as a non-privileged user with a native python-based health check.

---

🏛️ Architecture & Flow

This project strictly adheres to Clean Architecture patterns, decoupling the core domain models, the business prediction logic, the application services, and the presentation layer (FastAPI).

Key Design Decisions

1. Lifespan Model Loading: The .joblib pipeline is loaded once during container startup and stored in the application state. It resides in memory to achieve sub-millisecond predictions, preventing redundant disk I/O on incoming requests.
2. Non-blocking Event Loop: The endpoint is defined as a synchronous function (def instead of async def), prompting FastAPI to run the CPU-bound inference in an external thread pool. This prevents the single-threaded async event loop from starving.
3. Zero Training-Serving Skew: The entire Scikit-Learn Pipeline (imputers, StandardScaler, OneHotEncoder, and estimator) is serialized together. The API receives raw data and feeds it directly to the pipeline, guaranteeing mathematical consistency.

Inference Execution Flow

Below is the request-response sequence flow diagram detailing how a lead prediction is executed:

sequenceDiagram
    autonumber
    actor Client as Client HTTP
    participant API as FastAPI (main.py)
    participant Schema as LeadInput (Pydantic)
    participant Pool as Starlette Threadpool
    participant Service as LeadPredictorService
    participant ML as Scikit-Learn Pipeline (joblib)
    participant Output as LeadPrediction (Pydantic)

    Client->>API: HTTP POST /api/v1/leads/predict (JSON Payload)
    activate API
    
    API->>Schema: Validate schema and constraints
    activate Schema
    alt Validation Fails
        Schema-->>API: ValidationError (Invalid inputs)
        API-->>Client: HTTP 422 Unprocessable Entity
    else Validation Succeeds
        Schema-->>API: LeadInput Object Instance
    end
    deactivate Schema

    Note over API, Pool: Handler defined with synchronous "def"<br/>FastAPI runs it in an external thread pool
    
    API->>Pool: Offload execution of predict_lead()
    activate Pool
    
    Pool->>Service: predict(lead_input)
    activate Service
    
    Service->>Service: Convert LeadInput to 1-row Pandas DataFrame
    
    Service->>ML: pipeline.predict_proba(df)
    activate ML
    ML-->>Service: Return probabilities [[prob_class_0, prob_class_1]]
    deactivate ML
    
    Service->>ML: pipeline.predict(df)
    activate ML
    ML-->>Service: Return class label [[class_label]]
    deactivate ML
    
    Service->>Output: Instantiate LeadPrediction
    activate Output
    Output-->>Service: Validated prediction object
    deactivate Output
    
    Service-->>Pool: Return LeadPrediction
    deactivate Service
    
    Pool-->>API: Resolve task and return prediction object
    deactivate Pool
    
    API-->>Client: HTTP 200 OK (JSON Payload Response)
    deactivate API

Loading

---

📁 Repository Structure

leads_predictor/
├── docs/                         # System architecture documentation
│   └── architecture.md
├── config/                       # Static configuration properties
├── data/                         # Ignored local raw/processed datasets
├── models/                       # Serialized model registries (.joblib)
├── notebooks/                    # Jupyter Notebooks for EDA and prototyping
├── src/                          # Packaged application source code
│   ├── core/                     # Logging, configurations, exceptions
│   │   ├── config.py             # Pydantic Settings env loader
│   │   └── exceptions.py         # Domain custom exceptions
│   ├── domain/                   # Domain schemas (Pydantic models)
│   │   └── leads.py              # Strict schemas for Leads
│   ├── services/                 # Prediction service implementations
│   │   └── predictor.py          # LeadPredictorService loaded with pipeline
│   ├── api/                      # Presentation layer (FastAPI)
│   │   ├── v1/
│   │   │   ├── endpoints/
│   │   │   │   └── leads.py      # Inference REST endpoint
│   │   │   └── router.py         # Router aggregates
│   │   └── main.py               # API FastAPI initialization & lifespan
│   └── scripts/                  # Offline scripts
│       └── train.py              # Synthetic generation and training script
├── tests/                        # Full test suite
│   ├── unit/                     # Unit testing (training, services)
│   └── integration/              # Integration testing (FastAPI endpoints)
├── .dockerignore                 # Exclusions context for Docker build
├── .gitignore                    # Local environment and data gitignore
├── Dockerfile                    # Production secure multi-stage Dockerfile
├── pyproject.toml                # Project configurations & dependency locking
└── README.md                     # Project master documentation

---

🚀 Quick Start (Dockerized Serving)

The application is fully containerized using a secure, lightweight multi-stage build.

1. Build the Production Docker Image

Run the command below in the project root:

docker build -t lead-classification-api:latest .

2. Run the Container

Start the container mapping port 8000:

docker run -d --name lead-api -p 8000:8000 lead-classification-api:latest

3. Verify Container Status

Check that the server initialized and loaded the model cleanly:

docker logs lead-api

You can query the native healthcheck to confirm it is healthy:

curl -i http://localhost:8000/health

4. Make a Prediction Request

Test real-time inference using curl with a sample lead JSON payload:

curl -i -X POST http://localhost:8000/api/v1/leads/predict \
  -H "Content-Type: application/json" \
  -d '{
    "traffic_source": "email",
    "time_on_site_seconds": 600,
    "pages_viewed": 5,
    "industry": "insurance",
    "device_type": "desktop",
    "lead_score": 90.0
  }'

5. Stop and Clean Up

docker stop lead-api && docker rm lead-api

---

🛠️ Development & Testing

If you wish to modify the code locally, set up the development environment as follows:

# Create and activate virtual environment
python3 -m venv .venv
source .venv/bin/activate

# Install dependencies in editable mode with development tools
pip install -e .[dev]

# Run offline training script to generate the model artifact
python src/scripts/train.py

# Run static quality check and formatters (Ruff and Mypy)
ruff check .
ruff format .
mypy src/

# Run the complete test suite (unit and integration tests)
pytest

---

👤 Author

Francisco Javier Gómez Pulido
AI Lead at Aapex. AI Lead & MLOps Architect specialized in building scalable architectures, data science pipelines, and generating real business impact through production-grade artificial intelligence. Holds a Double Degree in Mathematics and Computer Science and a Master's degree in Artificial Intelligence.

📫 Let's connect:

• LinkedIn: linkedin.com/in/frangomezpulido
• GitHub: github.com/fragompul
• Email: frangomezpulido2002@gmail.com