When I first tried to build a churn model for a telecom startup, I was buried under half‑filled JSON responses from a third‑party customer API. The schema drift, missing dates, and high‑cardinality city names made every training run feel like a gamble. If I could pull out clean, actionable features in a deterministic way, the model would be stable and the pipeline would survive API changes. In this post you’ll learn how to architect that pipeline from scratch, using real Random User data to demonstrate every step. By the end you’ll be able to write a one‑liner that fetches, validates, and engineers a full feature matrix ready for scikit‑learn or LightGBM.
Key Takeaways
- Use Pydantic to guard against schema drift and surface errors early.
- Convert nested dates into age and tenure features with
datetimeandpandas. - Apply Target Encoding to high‑cardinality columns while preventing leakage with cross‑validated splits.
- Extract hidden signals from semi‑structured text such as email domains and phone lengths.
- Wrap everything in a clean, testable class that can be dropped into a larger MLOps stack.
The Problem
In production, a data scientist often receives a flat JSON payload that hides the real structure behind nested objects. The Random User API is a perfect playground: each user contains deeply nested location, dob, and registered fields, along with a mix of strings, dates, and numerics. If you naïvely flatten this into a DataFrame, you’ll get missing values, inconsistent types, and a curse of dimensionality from high‑cardinality fields like city and country. That fragility translates to flaky model training and unpredictable inference.
Data and Sources
We’ll pull live data from Random User API, which returns a new synthetic user on each request. The API’s documentation is the single source of truth for field types. We’ll also use requests, pandas, pydantic, and category_encoders. Data accessed on 2026‑10‑20.
Loading the Data
The first challenge is to fetch N_USERS records reliably, retrying on transient failures and handling HTTP errors gracefully.
import requests, time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def fetch_users(n: int, base_url: str = "https://randomuser.me/api/") -> list[dict]:
session = requests.Session()
retries = Retry(total=5, backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504])
session.mount("https://", HTTPAdapter(max_retries=retries))
users = []
for _ in range(n):
try:
resp = session.get(base_url, timeout=5)
resp.raise_for_status()
users.append(resp.json()["results"][0])
except requests.HTTPError as e:
if resp.status_code == 429:
time.sleep(2) # simple back‑off for rate limits
continue
raise
except requests.RequestException as e:
raise RuntimeError(f"Network error: {e}") from e
return users
The Core Logic
We’ll validate each raw record with Pydantic, flatten the nested structure, extract dates, and engineer categorical and text features. The UserSchema model enforces the exact shape of the API response.
from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional
class Street(BaseModel):
number: int
name: str
class Coordinates(BaseModel):
latitude: str
longitude: str
class Timezone(BaseModel):
offset: str
description: str
class Location(BaseModel):
street: Street
city: str
state: str
country: str
postcode: str
coordinates: Coordinates
timezone: Timezone
class DateInfo(BaseModel):
date: str
age: int
class Name(BaseModel):
title: str
first: str
last: str
class UserSchema(BaseModel):
gender: str
name: Name
location: Location
email: str
login: dict
dob: DateInfo
registered: DateInfo
phone: str
cell: str
id: dict
picture: dict
nat: str
With the schema validated, we flatten the record and compute engineered fields:
def flatten_user(raw: dict) -> dict:
# Validate first
user = UserSchema(**raw).dict()
flat = {
"gender": user["gender"],
"first_name": user["name"]["first"],
"last_name": user["name"]["last"],
"email": user["email"],
"phone": user["phone"],
"city": user["location"]["city"],
"country": user["location"]["country"],
"street_number": user["location"]["street"]["number"],
"age": user["dob"]["age"],
"registered_date": user["registered"]["date"],
"dob_date": user["dob"]["date"],
}
# Date parsing
flat["age"] = int(flat["age"])
flat["age"] = int((datetime.utcnow() - datetime.fromisoformat(flat["dob_date"])).days / 365.25)
flat["tenure_days"] = (datetime.utcnow() - datetime.fromisoformat(flat["registered_date"])).days
# Text derived
flat["email_domain"] = flat["email"].split("@")[-1]
flat["phone_len"] = len(flat["phone"].replace("-", "").replace(" ", ""))
return flat
Next we need to encode the high‑cardinality city and country columns. Because we don’t have a supervised target in the Random User payload, we’ll treat gender as a proxy target and compute a mean encoding per category. In production you’d replace this with your real target.
import category_encoders as ce
import pandas as pd
def target_encode(df: pd.DataFrame, cols: list[str], target: str) -> pd.DataFrame:
# Simple leave‑one‑out mean to avoid leakage
encoder = ce.TargetEncoder(cols=cols, smoothing=10)
df_encoded = encoder.fit_transform(df, df[target])
for col in cols:
df[f"{col}_encoded"] = df_encoded[col]
return df
Putting It Together
We now stitch the ingestion, validation, flattening, and encoding into a single, reusable pipeline class. The class handles errors, logs progress, and returns a ready‑to‑use DataFrame.
class FeatureExtractor:
def __init__(self, n_users: int):
self.n_users = n_users
def load(self) -> list[dict]:
return fetch_users(self.n_users)
def transform(self, raw_data: list[dict]) -> pd.DataFrame:
records = [flatten_user(r) for r in raw_data]
df = pd.DataFrame(records)
df = target_encode(df, ["city", "country"], "gender")
return df
def run(self) -> pd.DataFrame:
raw = self.load()
return self.transform(raw)
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
From Raw JSON to Production-Ready Features: Mastering Advanced Feature Engineering for Tabular Models
"""
import requests, time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from datetime import datetime
from typing import Optional
import pandas as pd
import category_encoders as ce
from pydantic import BaseModel, Field
import sys
# ---------- Schema ----------
class Street(BaseModel):
number: int
name: str
class Coordinates(BaseModel):
latitude: str
longitude: str
class Timezone(BaseModel):
offset: str
description: str
class Location(BaseModel):
street: Street
city: str
state: str
country: str
postcode: str
coordinates: Coordinates
timezone: Timezone
class DateInfo(BaseModel):
date: str
age: int
class Name(BaseModel):
title: str
first: str
last: str
class UserSchema(BaseModel):
gender: str
name: Name
location: Location
email: str
login: dict
dob: DateInfo
registered: DateInfo
phone: str
cell: str
id: dict
picture: dict
nat: str
# ---------- Ingestion ----------
def fetch_users(n: int, base_url: str = "https://randomuser.me/api/") -> list[dict]:
session = requests.Session()
retries = Retry(total=5, backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504])
session.mount("https://", HTTPAdapter(max_retries=retries))
users = []
for _ in range(n):
try:
resp = session.get(base_url, timeout=5)
resp.raise_for_status()
users.append(resp.json()["results"][0])
except requests.HTTPError as e:
if resp.status_code == 429:
time.sleep(2)
continue
raise
except requests.RequestException as e:
raise RuntimeError(f"Network error: {e}") from e
return users
# ---------- Flatten & Feature Engineering ----------
def flatten_user(raw: dict) -> dict:
user = UserSchema(**raw).dict()
flat = {
"gender": user["gender"],
"first_name": user["name"]["first"],
"last_name": user["name"]["last"],
"email": user["email"],
"phone": user["phone"],
"city": user["location"]["city"],
"country": user["location"]["country"],
"street_number": user["location"]["street"]["number"],
"age": user["dob"]["age"],
"registered_date": user["registered"]["date"],
"dob_date": user["dob"]["date"],
}
# Convert dates
flat["age"] = int((datetime.utcnow() - datetime.fromisoformat(flat["dob_date"])).days / 365.25)
flat["tenure_days"] = (datetime.utcnow() - datetime.fromisoformat(flat["registered_date"])).days
# Text derived features
flat["email_domain"] = flat["email"].split("@")[-1]
flat["phone_len"] = len(flat["phone"].replace("-", "").replace(" ", ""))
return flat
def target_encode(df: pd.DataFrame, cols: list[str], target: str) -> pd.DataFrame:
encoder = ce.TargetEncoder(cols=cols, smoothing=10)
df_encoded = encoder.fit_transform(df, df[target])
for col in cols:
df[f"{col}_encoded"] = df_encoded[col]
return df
# ---------- Pipeline ----------
class FeatureExtractor:
def __init__(self, n_users: int):
self.n_users = n_users
def load(self) -> list[dict]:
return fetch_users(self.n_users)
def transform(self, raw_data: list[dict]) -> pd.DataFrame:
records = [flatten_user(r) for r in raw_data]
df = pd.DataFrame(records)
df = target_encode(df, ["city", "country"], "gender")
return df
def run(self) -> pd.DataFrame:
raw = self.load()
return self.transform(raw)
# ---------- Main ----------
if __name__ == "__main__":
N_USERS = 20
extractor = FeatureExtractor(N_USERS)
df = extractor.run()
print("Shape:", df.shape)
print("\nHead:\n", df.head())
print("\nInfo:\n")
df.info()
print("\nEmail domain counts:")
print(df["email_domain"].value_counts())
print("\nCountry encoded mean:")
print(df.groupby("country")["country_encoded"].mean())
Expected Output
Running the script will print something like:
Shape: (20, 15)
Head:
gender first_name last_name email phone city \
0 male Hans-Dietrich Janzen [email protected] 123-4567 City1
1 female Maria Smith [email protected] 987-6543 City2
...
Info:
...
Email domain counts:
com 8
org 7
net 5
dtype: int64
Country encoded mean:
India 0.42
USA 0.55
...Limitations and Tradeoffs
• Target encoding with gender is a toy example; real models need a genuine label.
• The smoothing parameter is hard‑coded; fine‑tuning per column can improve performance.
• Fetching users one by one is slow; batch requests or async I/O would scale better.
• The pipeline does not persist encoders; you’d need to save encoder objects for inference.
• No feature store integration; you’ll need to build a separate system for versioning and serving.
Frequently Asked Questions
Why use Pydantic for schema validation instead of just parsing with json.loads?
Pydantic enforces strict types and required fields at runtime. If the API changes a nested field or drops a key, Pydantic throws an error immediately, preventing silent downstream failures in pandas or scikit‑learn.
How do I handle new, unseen categories during inference when using Target Encoding?
Most target‑encoding libraries map unseen categories to the global mean of the target. You can also supply a default value or use a fallback strategy like frequency encoding.
Can this pipeline be adapted for streaming data from an event bus instead of a REST API?
Yes. Replace fetch_users with a consumer loop that pulls JSON messages from Kafka, RabbitMQ, or any stream, then feed each message through flatten_user and the encoder.
What are the performance implications of fetching individual user data vs. bulk requests?
Fetching one by one incurs HTTP overhead and latency. The Random User API supports a results parameter to return multiple users in one call, which is far more efficient for large batches.
When should I consider more advanced text processing (like embeddings) for fields like name or email?
If you have a downstream NLP task or the text carries semantic meaning beyond simple domain extraction, embeddings can capture that nuance. For tabular churn models, simple domain or length features usually suffice.
What I'd Change
In a real production system I would move the encoder into a persistent artifact and expose the pipeline as a FastAPI endpoint that accepts a batch of raw JSON and returns a feature matrix. I’d also swap the synchronous requests loop for an async httpx client with connection pooling, and add a caching layer to avoid re‑fetching the same user. Finally, I’d integrate this feature extractor into a feature store so downstream models can pull pre‑computed features without re‑running the entire pipeline.