Chapter 5 · Pipelines and AI
Preparing ML-Ready Data with SQL and pandas
- Page 18 of 22
- 20 min read
A model learns whatever the training table says — including its mistakes and its accidental hints about the answer. Most of the work in a real ML project is not choosing the algorithm; it is building that table: one row per example, the right features, an honest target, and nothing in it that the model could not have known at prediction time. SQL is where most of that table is built, because the history you need lives in the database.
This page builds a churn dataset end to end: first on the shop, then on a larger generated customer table, and finally a scikit-learn model trained on it. Along the way you will create a leak on purpose and watch it make a model look far better than it is.
What you will learn
- How to frame a prediction problem as a table: entity, cutoff date, features, target and label window
- How to write point-in-time feature queries in SQL (RFM-style history, joined sources) and finish features in pandas
- How missing values, invalid records, categories, scaling and outliers are handled without contaminating the test set
- Why a time-based split, not a random one, is the honest test for a churn model, and how to read results on imbalanced classes
- How to make a dataset reproducible: seeds, versioned snapshots, Parquet files, hashes and pre-training checks
The shape of an ML-ready table
"Will this customer stop buying?" becomes a table only after you pin down five things:
| Decision | Meaning | On this page |
|---|---|---|
| Entity | What one row describes | One customer at one cutoff date |
| Cutoff (as-of date) | The moment you pretend to make the prediction | 2026-04-01 and earlier dates |
| Features | What was known before the cutoff | Orders, spend, recency, support tickets, plan, city |
| Target (label) | What happened after the cutoff | churned = 1 if no order in the next 90 days |
| Population | Which entities get a row at all | Customers who existed (and were active) at the cutoff |
The cutoff splits time in two. Everything left of it may feed features; everything right of it may only feed the target:
features: history before the cutoff target: what happens next
───────────────────────────────────────────────────|─────────────────────────|──▶ time
cutoff cutoff + 90 days
(2026-04-01) (2026-06-30)
A feature that reads anything right of the cutoff is a LEAK.This rule is called point-in-time correctness. In production the model runs on the cutoff date and the future simply does not exist yet; the training table must be built as if that were true.
Point-in-time features on the shop
Here are RFM features — Recency (days since the last order), Frequency (number of orders) and Monetary (money spent) — plus tenure and a churn label, for every shop customer as of 1 April 2026. Cancelled orders do not count:
WITH order_totals AS (
SELECT o.order_id, o.customer_id, o.order_date, o.status,
SUM(oi.quantity * oi.unit_price) AS total
FROM orders AS o
JOIN order_items AS oi ON oi.order_id = o.order_id
GROUP BY o.order_id, o.customer_id, o.order_date, o.status
)
SELECT c.customer_id,
c.city,
CAST(julianday('2026-04-01') - julianday(c.joined_on) AS INTEGER) AS tenure_days,
COUNT(t.order_id) AS frequency,
COALESCE(SUM(t.total), 0) AS monetary,
CAST(julianday('2026-04-01') - julianday(MAX(t.order_date)) AS INTEGER) AS recency_days,
CASE WHEN EXISTS (
SELECT 1 FROM orders AS f
WHERE f.customer_id = c.customer_id
AND f.status <> 'cancelled'
AND f.order_date >= '2026-04-01' AND f.order_date < '2026-07-01'
) THEN 0 ELSE 1 END AS churned
FROM customers AS c
LEFT JOIN order_totals AS t
ON t.customer_id = c.customer_id
AND t.order_date < '2026-04-01' -- the point-in-time filter
AND t.status <> 'cancelled'
WHERE c.joined_on < '2026-04-01' -- the population at the cutoff
GROUP BY c.customer_id, c.city, c.joined_on
ORDER BY c.customer_id;+-------------+------------+-------------+-----------+----------+--------------+---------+
| customer_id | city | tenure_days | frequency | monetary | recency_days | churned |
+-------------+------------+-------------+-----------+----------+--------------+---------+
| 1 | Dhaka | 149 | 2 | 5100 | 57 | 0 |
| 2 | Chattogram | 108 | 2 | 19500 | 24 | 0 |
| 3 | Dhaka | 82 | 2 | 6400 | 4 | 0 |
| 4 | Sylhet | 69 | 0 | 0 | NULL | 1 |
| 5 | NULL | 55 | 1 | 4000 | 17 | 1 |
| 6 | Khulna | 42 | 0 | 0 | NULL | 0 |
| 7 | Dhaka | 2 | 0 | 0 | NULL | 0 |
+-------------+------------+-------------+-----------+----------+--------------+---------+- Customer 8 is missing: they joined on 11 April, after the cutoff. On 1 April nobody could have scored them.
- The filters sit in
ON, notWHERE, so customers with no history keep their row (see LEFT, RIGHT, FULL, CROSS and Self Joins in the beginner tutorial). Customer 4's only order was cancelled; customers 6 and 7 had not ordered yet. - NULL and 0 mean different things. No orders means frequency 0 and monetary 0 — that is a fact. But
recency_daysis NULL: "days since the last order" is undefined when there was none. You decide later how a model should see it. - Features and label use different time windows: features look only before the cutoff, the label only at the three months after it (April to June here; the generated data below uses exactly 90 days).
A bigger, reproducible dataset
Eight customers cannot train a model. The block below generates a realistic one into shop.db: 2,000 customers with a plan, a city and an age (some missing, a few typed wrongly), their orders and their support tickets. Customers leave at random times; before leaving they order less and complain more — the signal a model should find. One seed makes every run identical:
import numpy as np
import pandas as pd
from sqlhelp import con, show
rng = np.random.default_rng(42) # one seed: the same dataset on every run
N = 2000
END = pd.Timestamp("2026-06-30") # the last day of data
signup = pd.Timestamp("2025-01-01") + pd.to_timedelta(rng.integers(0, 450, N), unit="D")
plan = rng.choice(["free", "basic", "pro"], N, p=[0.5, 0.35, 0.15])
city = rng.choice(["Dhaka", "Chattogram", "Sylhet", "Khulna", "Rajshahi", None], N,
p=[0.42, 0.2, 0.12, 0.11, 0.1, 0.05])
age = rng.integers(18, 61, N).astype(float)
age[rng.random(N) < 0.07] = np.nan # never filled in
typo = rng.random(N) < 0.015
age[typo] = rng.choice([0, 7, 230], typo.sum()) # data-entry errors
rate = rng.gamma(2.0, 1.0, N) * np.select([plan == "pro", plan == "basic"], [1.6, 1.2], 1.0)
life = rng.exponential(np.select([plan == "pro", plan == "basic"], [500, 360], 220))
churn_at = signup + pd.to_timedelta(life.round(), unit="D")
customers, orders, tickets = [], [], []
oid = tid = 0
for i in range(N):
stop = min(churn_at[i], END)
days = (stop - signup[i]).days
closed = None
if churn_at[i] <= END and rng.random() < 0.6: # some leavers close the account
c = churn_at[i] + pd.Timedelta(days=int(rng.integers(0, 45)))
closed = c.date().isoformat() if c <= END else None
customers.append((i + 1, signup[i].date().isoformat(), plan[i], city[i],
None if np.isnan(age[i]) else int(age[i]), closed))
if days <= 0:
continue
n = rng.poisson(rate[i] * days / 30)
for d in np.sort(rng.integers(0, days, n)):
left = days - d if churn_at[i] <= END else 999
if rng.random() > min(1.0, left / 75): # interest fades before leaving
continue
oid += 1
amount = round(float(rng.lognormal(6.6 + 0.3 * (plan[i] == "pro"), 0.6)))
orders.append((oid, i + 1, (signup[i] + pd.Timedelta(days=int(d))).date().isoformat(), amount))
m = rng.poisson(0.15 * days / 30 + (2.0 if churn_at[i] <= END else 0))
for d in rng.integers(max(0, days - 60) if churn_at[i] <= END else 0, days, m):
tid += 1 # leavers complain near the end
tickets.append((tid, i + 1, (signup[i] + pd.Timedelta(days=int(d))).date().isoformat()))
con.executescript("""
DROP TABLE IF EXISTS sim_tickets;
DROP TABLE IF EXISTS sim_orders;
DROP TABLE IF EXISTS sim_customers;
CREATE TABLE sim_customers (customer_id INTEGER PRIMARY KEY, signup_date TEXT NOT NULL,
plan TEXT NOT NULL, city TEXT, age INTEGER, closed_on TEXT);
CREATE TABLE sim_orders (order_id INTEGER PRIMARY KEY, customer_id INTEGER NOT NULL,
order_date TEXT NOT NULL, amount INTEGER NOT NULL,
FOREIGN KEY (customer_id) REFERENCES sim_customers (customer_id));
CREATE TABLE sim_tickets (ticket_id INTEGER PRIMARY KEY, customer_id INTEGER NOT NULL,
opened_on TEXT NOT NULL,
FOREIGN KEY (customer_id) REFERENCES sim_customers (customer_id));
""")
with con:
con.executemany("INSERT INTO sim_customers VALUES (?, ?, ?, ?, ?, ?)", customers)
con.executemany("INSERT INTO sim_orders VALUES (?, ?, ?, ?)", orders)
con.executemany("INSERT INTO sim_tickets VALUES (?, ?, ?)", tickets)
show("""SELECT (SELECT COUNT(*) FROM sim_customers) AS customers,
(SELECT COUNT(*) FROM sim_orders) AS orders,
(SELECT COUNT(*) FROM sim_tickets) AS tickets,
(SELECT MAX(order_date) FROM sim_orders) AS last_order""")+-----------+--------+---------+------------+
| customers | orders | tickets | last_order |
+-----------+--------+---------+------------+
| 2000 | 26281 | 4331 | 2026-06-29 |
+-----------+--------+---------+------------+Three tables from two "systems" (sales and support), just like a real company. Note closed_on: the app writes it when a customer closes the account. It will matter later.
Features in SQL: one snapshot per cutoff
The feature query takes the cutoff as a parameter, so the same code builds a snapshot for any date. Each CTE is one source or one window; the population is "customers with at least one order in the 90 days before the cutoff, account still open" — the business wants to know which active customers are about to leave:
FEATURES_SQL = """
WITH active AS (
SELECT DISTINCT customer_id
FROM sim_orders
WHERE order_date >= date(:cutoff, '-90 days') AND order_date < :cutoff
),
history AS ( -- sales: everything BEFORE the cutoff
SELECT customer_id,
COUNT(*) AS orders_all,
SUM(CASE WHEN order_date >= date(:cutoff, '-30 days') THEN 1 ELSE 0 END) AS orders_30d,
SUM(CASE WHEN order_date >= date(:cutoff, '-90 days') THEN 1 ELSE 0 END) AS orders_90d,
SUM(amount) AS spend_all,
CAST(julianday(:cutoff) - julianday(MAX(order_date)) AS INTEGER) AS recency_days
FROM sim_orders
WHERE order_date < :cutoff
GROUP BY customer_id
),
support AS ( -- support: tickets in the last 90 days
SELECT customer_id, COUNT(*) AS tickets_90d
FROM sim_tickets
WHERE opened_on >= date(:cutoff, '-90 days') AND opened_on < :cutoff
GROUP BY customer_id
),
future AS ( -- the label window: AFTER the cutoff
SELECT DISTINCT customer_id
FROM sim_orders
WHERE order_date >= :cutoff AND order_date < date(:cutoff, '+90 days')
)
SELECT :cutoff AS cutoff, c.customer_id, c.plan, c.city, c.age,
CAST(julianday(:cutoff) - julianday(c.signup_date) AS INTEGER) AS tenure_days,
h.orders_all, h.orders_30d, h.orders_90d, h.spend_all, h.recency_days,
COALESCE(s.tickets_90d, 0) AS tickets_90d,
CASE WHEN f.customer_id IS NULL THEN 1 ELSE 0 END AS churned
FROM sim_customers AS c
JOIN active AS a ON a.customer_id = c.customer_id
JOIN history AS h ON h.customer_id = c.customer_id
LEFT JOIN support AS s ON s.customer_id = c.customer_id
LEFT JOIN future AS f ON f.customer_id = c.customer_id
WHERE c.closed_on IS NULL OR c.closed_on >= :cutoff
ORDER BY c.customer_id
"""
def snapshot(cutoff):
return pd.read_sql(FEATURES_SQL, con, params={"cutoff": cutoff})
snaps = {cut: snapshot(cut) for cut in ["2025-10-01", "2026-01-01", "2026-04-01"]}
for cut, df in snaps.items():
print(cut, f"rows={len(df)}", f"churn rate={df['churned'].mean():.1%}")
print(snaps["2026-04-01"].drop(columns="cutoff").head(3).to_string(index=False))2025-10-01 rows=744 churn rate=21.2%
2026-01-01 rows=898 churn rate=20.9%
2026-04-01 rows=992 churn rate=22.3%
customer_id plan city age tenure_days orders_all orders_30d orders_90d spend_all recency_days tickets_90d churned
2 free Chattogram 27.0 107 4 0 2 2909 51 2 1
3 free Dhaka 40.0 161 8 1 5 5718 24 0 0
6 free Khulna 50.0 69 1 0 1 389 59 1 1- The parameter
:cutoffis bound by the driver, never pasted into the string (see Python and Databases: Connections, Parameters and Transactions).pd.read_sqlpassesparamsstraight through. COALESCE(s.tickets_90d, 0)is right here because "no ticket rows" really means zero tickets. NeverCOALESCEan unknown (like age) into a fake number in SQL; keep the NULL and decide in the model pipeline.- About one active customer in five churns. That is class imbalance: a model that always says "stays" is right 78% of the time and useless. You will measure with better metrics below.
- The
agecolumn shows floats (27.0) because it contains NULLs, which pandas stores asNaN.
Cleaning and feature engineering in pandas
Some steps are easier in pandas: rules that need judgment, ratios and transforms. Invalid ages become missing, a missing city becomes its own category, and a skewed amount gets a log:
def prepare(df):
df = df.copy()
df["age"] = df["age"].where(df["age"].between(13, 100)) # 0, 7, 230 -> NaN
df["city"] = df["city"].fillna("unknown") # missing is information too
df["log_spend"] = np.log1p(df["spend_all"]) # tame the long tail
df["trend"] = df["orders_30d"] / (df["orders_90d"] / 3) # last month vs 3-month average
return df
raw = snaps["2026-04-01"]
print("invalid ages:", int((~raw["age"].between(13, 100) & raw["age"].notna()).sum()))
print("missing ages:", int(raw["age"].isna().sum()), "| missing cities:", int(raw["city"].isna().sum()))
print("spend_all: median", raw["spend_all"].median(), "max", raw["spend_all"].max())
print(prepare(raw)[["log_spend", "trend"]].describe().round(2).loc[["mean", "min", "max"]].to_string())invalid ages: 18
missing ages: 73 | missing cities: 46
spend_all: median 9232.0 max 162510
log_spend trend
mean 9.01 1.08
min 5.19 0.00
max 12.00 3.00Where should each step live? A rule of thumb:
| Step | Best place | Why |
|---|---|---|
| Filtering, joins, windows by date, counts and sums over history | SQL | Runs next to the data; ships only one row per entity |
| Ratios, logs, simple validity rules | Either | Choose one place and keep it there |
| Imputing, scaling, one-hot encoding | scikit-learn pipeline | Must be learned from the training rows only |
| Outlier clipping by percentile | Pipeline (or with training-set limits) | The percentile is itself learned from data |
A time-based split and the model
A random split would put rows from April in training and rows from October in the test set — the model would learn from the future it is tested on. A churn model will always be used on a later date than it was trained on, so test it that way: train on the two earlier cutoffs, test on the latest. The training labels (up to 31 March) end before the test cutoff, so nothing from the test period leaks into training.
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import precision_score, recall_score, roc_auc_score
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler
NUMERIC = ["age", "tenure_days", "orders_all", "orders_90d", "log_spend",
"recency_days", "tickets_90d", "trend"]
CATEGORICAL = ["plan", "city"]
def make_model(numeric=NUMERIC):
features = ColumnTransformer([
("num", make_pipeline(SimpleImputer(strategy="median"), StandardScaler()), numeric),
("cat", OneHotEncoder(handle_unknown="ignore"), CATEGORICAL),
])
return make_pipeline(features, LogisticRegression(class_weight="balanced", max_iter=1000))
def evaluate(train, test, numeric=NUMERIC):
model = make_model(numeric).fit(train[numeric + CATEGORICAL], train["churned"])
proba = model.predict_proba(test[numeric + CATEGORICAL])[:, 1]
return model, roc_auc_score(test["churned"], proba), (proba >= 0.5).astype(int)
train = prepare(pd.concat([snaps["2025-10-01"], snaps["2026-01-01"]], ignore_index=True))
test = prepare(snaps["2026-04-01"])
model, auc, pred = evaluate(train, test)
print(f"train rows {len(train)}, test rows {len(test)}")
print(f"always 'stays' accuracy: {1 - test['churned'].mean():.3f}")
print(f"model accuracy: {(pred == test['churned']).mean():.3f}")
print(f"ROC AUC {auc:.3f} | precision {precision_score(test['churned'], pred):.2f}"
f" | recall {recall_score(test['churned'], pred):.2f}")
weights = pd.Series(model[-1].coef_[0], index=model[0].get_feature_names_out())
print(weights.sort_values().round(2).iloc[[0, 1, -2, -1]].to_string())train rows 1642, test rows 992
always 'stays' accuracy: 0.777
model accuracy: 0.806
ROC AUC 0.861 | precision 0.55 | recall 0.72
cat__plan_pro -0.52
num__log_spend -0.51
num__recency_days 0.63
num__tickets_90d 1.62- The imputer and scaler are fitted inside the pipeline, on training rows only. Computing the median or the mean on all rows before splitting is train/test contamination: a small leak, but a real one.
handle_unknown="ignore"keeps the model working when a new city appears in production: the new value simply gets all zeros. The encoder returns a sparse matrix by default (sparse_output=True), which the pipeline handles; setsparse_output=Falseonly when you want to look at the encoded columns yourself.class_weight="balanced"makes each churner count almost four times as much as a customer who stays (the inverse of the class shares, about 21% vs 79%), so the model does not just predict the majority. The accuracy barely beats "always stays", but recall 0.72 says it finds most of the real churners, precision 0.55 says about half of the customers it flags really leave (fine for a cheap retention message, not for an expensive discount), and ROC AUC (0.5 = coin toss, 1.0 = perfect ranking) says it ranks risky customers well.- The weights make sense: more support tickets and a longer time since the last order push towards churn; a pro plan and higher spend push away.
Leakage, measured
Now break it on purpose, twice. Leak 1 is the classic bug: the history CTE forgets its WHERE order_date < :cutoff, so "all orders" includes the label window. Leak 2 is subtle: a column from today's customer table, closed_on, used as a feature. It looks like a fact about the customer, but every account in the population was still open on the cutoff date; the closure happened later.
LEAKY_SQL = FEATURES_SQL.replace(" WHERE order_date < :cutoff\n", "") # forgot the cutoff
leaky = {cut: pd.read_sql(LEAKY_SQL, con, params={"cutoff": cut}) for cut in snaps}
leaky_train = prepare(pd.concat([leaky["2025-10-01"], leaky["2026-01-01"]], ignore_index=True))
leaky_test = prepare(leaky["2026-04-01"])
print("leak 1 recency_days min:", leaky_test["recency_days"].min())
print(f"leak 1 ROC AUC: {evaluate(leaky_train, leaky_test)[1]:.3f}")
closed = pd.read_sql("SELECT customer_id, closed_on IS NOT NULL AS account_closed FROM sim_customers", con)
train2, test2 = train.merge(closed, on="customer_id"), test.merge(closed, on="customer_id")
print(f"leak 2 ROC AUC: {evaluate(train2, test2, NUMERIC + ['account_closed'])[1]:.3f}")
print(f"honest ROC AUC: {auc:.3f}")leak 1 recency_days min: -89
leak 1 ROC AUC: 0.979
leak 2 ROC AUC: 0.880
honest ROC AUC: 0.861Leak 1 gives a negative recency — a "last order" in the future — and a nearly perfect score. Leak 2 is worse in practice: it adds a believable couple of points, nobody gets suspicious, and in production the feature is always 0 for active customers, so the model is quietly worse than its report. Typical sources of leakage in AI work:
- Columns the application overwrites:
status,closed_on,last_login,lifetime_value— they hold today's value, not the value at the cutoff. - Aggregates without a date filter, or windows that end at "now" instead of the cutoff.
- Features computed from the target itself (a "refund requested" flag when predicting refunds).
- The same example, or a near-copy of it, in both train and test: duplicated rows that a random split sends to both sides, the same ticket twice, a near-duplicate document in an LLM eval set. (The same customer at a later cutoff, as in our time-based split, is fine: that is exactly how the model is used.)
Rule of thumb: if a score looks too good, search for the leak before you celebrate.
Reproducible snapshots: checks, Parquet and hashes
A training set you cannot rebuild is a training set you cannot debug. Before training, run cheap checks; then save each snapshot as a versioned file and record a fingerprint of its content:
import hashlib
import json
def check(df, cutoff):
assert not df.duplicated(["customer_id", "cutoff"]).any(), "one row per customer and cutoff"
assert set(df["churned"].unique()) <= {0, 1}, "target must be 0/1"
assert (df["recency_days"] >= 0).all(), "a feature looked past the cutoff"
assert 0.05 < df["churned"].mean() < 0.5, "churn rate outside the expected range"
assert df["cutoff"].eq(cutoff).all()
manifest = []
for cut, df in snaps.items():
check(df, cut)
df.to_parquet(f"churn_features_{cut}.parquet", index=False)
digest = hashlib.sha256(df.to_csv(index=False).encode()).hexdigest()[:16]
manifest.append({"file": f"churn_features_{cut}.parquet", "rows": len(df), "sha256_16": digest})
with open("manifest.json", "w") as f:
json.dump({"seed": 42, "query": "FEATURES_SQL v1", "snapshots": manifest}, f, indent=2)
print(json.dumps(manifest[-1]))
try:
check(leaky["2026-04-01"], "2026-04-01")
except AssertionError as e:
print("leaky snapshot rejected:", e){"file": "churn_features_2026-04-01.parquet", "rows": 992, "sha256_16": "ea1a0912c25b37aa"}
leaky snapshot rejected: a feature looked past the cutoff- Seed every random step (data generation, splits, models). Version the query and the snapshot files; never overwrite last month's training set.
- Hash the content (here a canonical CSV text), not the Parquet bytes, which can change with library versions. Same hash = same data, so two people can prove they trained on the same rows.
- Parquet keeps the dtypes and is small and fast to read (see Importing and Exporting: CSV, JSON, Excel and Parquet). A real feature store does all of this at scale: point-in-time joins, versioned features, the same code for training and serving.
Common mistakes
- Building features from "now".
julianday('now') - julianday(MAX(order_date))works today and is wrong for every training row. Fix: compute every window relative to a:cutoffparameter, and filter every source with< :cutoff. - Using the current value of a mutable column.
customers.statusorclosed_onin a training row describes today. Fix: rebuild the value from dated events (WHERE event_date < :cutoff) or keep a history table (see the slowly changing dimensions in Warehouses, Lakes, Streams and Scale). - Imputing or scaling before the split.
df["age"].fillna(df["age"].median())on all rows lets test data shape training. Fix: putSimpleImputerandStandardScalerin the pipeline and fit on train only. - A random split on time-ordered data. It mixes future and past and over-reports quality. Fix: split by cutoff date; make sure training labels end before the test cutoff.
- Judging an imbalanced problem by accuracy. 78% accuracy is free here. Fix: report ROC AUC, precision and recall, and compare against the "always majority" baseline. Where a random split is right (data with no time order), pass
stratify=ytotrain_test_splitso both sides keep the same share of churners.
Try it yourself
- Easy: Run the shop RFM query with the cutoff
2026-03-01and a label window up to2026-06-01. Which customers drop out of the population, and why? - Medium: Add
avg_order_value(spend divided by orders, before the cutoff) to the snapshot in SQL, then retrain with it as an extra numeric feature and print the test ROC AUC. - Hard: Write
assert_point_in_time(cutoff): it rebuilds the snapshot, then checks with a SQL query that no order or ticket on or after the cutoff could have been counted, by comparingorders_allwith a count of orders strictly before the cutoff for every customer.
Answers
-- 1. Customers 7 and 8 joined after 2026-03-01, so they are not in the population.
WITH order_totals AS (
SELECT o.order_id, o.customer_id, o.order_date, o.status,
SUM(oi.quantity * oi.unit_price) AS total
FROM orders AS o
JOIN order_items AS oi ON oi.order_id = o.order_id
GROUP BY o.order_id, o.customer_id, o.order_date, o.status
)
SELECT c.customer_id,
COUNT(t.order_id) AS frequency,
COALESCE(SUM(t.total), 0) AS monetary,
CAST(julianday('2026-03-01') - julianday(MAX(t.order_date)) AS INTEGER) AS recency_days,
CASE WHEN EXISTS (
SELECT 1 FROM orders AS f
WHERE f.customer_id = c.customer_id
AND f.status <> 'cancelled'
AND f.order_date >= '2026-03-01' AND f.order_date < '2026-06-01'
) THEN 0 ELSE 1 END AS churned
FROM customers AS c
LEFT JOIN order_totals AS t
ON t.customer_id = c.customer_id
AND t.order_date < '2026-03-01'
AND t.status <> 'cancelled'
WHERE c.joined_on < '2026-03-01'
GROUP BY c.customer_id
ORDER BY c.customer_id;# 2. One more column in SQL, one more name in the feature list
AOV_SQL = FEATURES_SQL.replace(
"h.spend_all, h.recency_days,",
"h.spend_all, h.recency_days, ROUND(1.0 * h.spend_all / h.orders_all, 2) AS avg_order_value,")
aov = {cut: pd.read_sql(AOV_SQL, con, params={"cutoff": cut}) for cut in snaps}
aov_train = prepare(pd.concat([aov["2025-10-01"], aov["2026-01-01"]], ignore_index=True))
aov_test = prepare(aov["2026-04-01"])
print(f"ROC AUC with avg_order_value: {evaluate(aov_train, aov_test, NUMERIC + ['avg_order_value'])[1]:.3f}")
# 3. A point-in-time test you can run before every training job
def assert_point_in_time(cutoff):
df = snapshot(cutoff)
truth = pd.read_sql(
"SELECT customer_id, COUNT(*) AS n FROM sim_orders WHERE order_date < :cutoff GROUP BY customer_id",
con, params={"cutoff": cutoff})
merged = df.merge(truth, on="customer_id")
assert len(merged) == len(df), "every customer in the snapshot has history"
assert (merged["orders_all"] == merged["n"]).all(), "orders_all counted orders after the cutoff"
assert (df["recency_days"] >= 0).all(), "a last order after the cutoff"
print(cutoff, "is point-in-time correct")
assert_point_in_time("2026-04-01")Summary
- An ML-ready table has one row per entity per cutoff: features from before the cutoff, a target from the window after it, and a clearly defined population.
- Write feature queries with a
:cutoffparameter and filter every source with it; joins, windows and RFM aggregates belong in SQL. - Keep unknowns as NULL in SQL; impute, scale and encode inside a scikit-learn pipeline fitted on training rows only.
- Split by time, check that training labels end before the test cutoff, and report AUC, precision and recall against the majority baseline.
- Leakage hides in "now", in overwritten columns and in missing date filters; a suspiciously good score is a clue.
- Seed, version, check, save to Parquet and hash every snapshot so any model can be traced to its exact data.
Next: SQL for AI Apps: Embeddings, pgvector and RAG Metadata moves from training data to the data an LLM application produces and uses: conversation logs, token costs, JSON, full-text search and vector search inside PostgreSQL.