PROJECT 001 · PYTHON SOURCE
Retail customer analytics
Study-specific code. Shared modules, dependency versions, and reproduction instructions are included in all project files.
Download Python file ↓"""Reproducible sales, retention and customer analysis of UCI Online Retail."""
from __future__ import annotations
import json
import math
import sqlite3
from pathlib import Path
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from matplotlib.ticker import FuncFormatter, PercentFormatter
PROJECT = Path(__file__).resolve().parents[1]
OUTPUT = PROJECT / "outputs"
FIGURES = PROJECT / "reports" / "figures"
def load_data() -> pd.DataFrame:
path = PROJECT / "data" / "raw" / "Online Retail.xlsx"
if not path.exists():
raise FileNotFoundError("Run python scripts/download_data.py retail from the repository root.")
return pd.read_excel(path, dtype={"InvoiceNo": "string", "StockCode": "string", "CustomerID": "string"})
def prepare(raw: pd.DataFrame) -> tuple[pd.DataFrame, dict]:
required = {"InvoiceNo", "StockCode", "Quantity", "UnitPrice", "InvoiceDate", "CustomerID", "Country"}
if not required.issubset(raw.columns):
raise ValueError(f"Missing required columns: {required - set(raw.columns)}")
df = raw.copy()
df["InvoiceDate"] = pd.to_datetime(df["InvoiceDate"], errors="raise")
if df[["InvoiceNo", "StockCode", "Quantity", "UnitPrice", "InvoiceDate", "Country"]].isna().any().any():
raise ValueError("A required transaction field is missing; inspect before proceeding.")
if not np.isfinite(df[["Quantity", "UnitPrice"]].to_numpy()).all():
raise ValueError("Non-finite transaction values")
df["is_cancellation"] = df["InvoiceNo"].str.upper().str.startswith("C")
df["line_value_gbp"] = df["Quantity"] * df["UnitPrice"]
df["is_sale"] = (df["Quantity"] > 0) & (df["UnitPrice"] > 0) & ~df["is_cancellation"]
df["is_ledger"] = (df["UnitPrice"] > 0) & (df["Quantity"] != 0)
df["month"] = df["InvoiceDate"].dt.strftime("%Y-%m")
audit = {
"raw_rows": len(raw),
"exact_repeat_rows_retained": int(raw.duplicated().sum()),
"missing_customer_id_rows": int(raw["CustomerID"].isna().sum()),
"missing_description_rows": int(raw["Description"].isna().sum()) if "Description" in raw else 0,
"nonpositive_price_rows": int((df["UnitPrice"] <= 0).sum()),
"negative_quantity_rows": int((df["Quantity"] < 0).sum()),
"cancellation_rows": int(df["is_cancellation"].sum()),
"positive_quantity_cancellation_rows": int((df["is_cancellation"] & (df["Quantity"] > 0)).sum()),
"positive_sales_rows": int(df["is_sale"].sum()),
"first_timestamp": str(df["InvoiceDate"].min()),
"last_timestamp": str(df["InvoiceDate"].max()),
"gross_sales_if_exact_repeats_removed_gbp": float(df.loc[df["is_sale"] & ~raw.duplicated(), "line_value_gbp"].sum()),
}
return df, audit
def customer_segments(sales: pd.DataFrame, snapshot: pd.Timestamp) -> pd.DataFrame:
known = sales.dropna(subset=["CustomerID"])
rfm = known.groupby("CustomerID").agg(
last_purchase=("InvoiceDate", "max"),
frequency=("InvoiceNo", "nunique"),
monetary_gbp=("line_value_gbp", "sum"),
)
rfm["recency_days"] = (snapshot - rfm["last_purchase"].dt.normalize()).dt.days
# Transparent, mutually exclusive planning rules; not a learned churn model.
rfm["segment"] = np.select([
(rfm["recency_days"] <= 30) & (rfm["frequency"] >= 5),
rfm["recency_days"] > 90,
rfm["frequency"] == 1,
], ["Recent frequent buyers", "Inactive over 90 days", "One-time buyers"], default="Other repeat buyers")
return rfm.reset_index()
def cohort_retention(sales: pd.DataFrame) -> tuple[pd.DataFrame, pd.Series]:
# Drop the partial final calendar month; unseen later months remain missing.
last = sales["InvoiceDate"].max()
complete_end = last.to_period("M") - 1
known = sales.dropna(subset=["CustomerID"]).copy()
known["purchase_month"] = known["InvoiceDate"].dt.to_period("M")
known = known[known["purchase_month"] <= complete_end]
first = known.groupby("CustomerID")["purchase_month"].transform("min")
known["cohort"] = first
known["age"] = (known["purchase_month"].dt.year - first.dt.year) * 12 + known["purchase_month"].dt.month - first.dt.month
counts = known.groupby(["cohort", "age"])["CustomerID"].nunique()
cohorts = pd.period_range(known["cohort"].min(), complete_end, freq="M")
matrix = pd.DataFrame(np.nan, index=cohorts, columns=range(len(cohorts)), dtype=float)
sizes = known.groupby("cohort")["CustomerID"].nunique().reindex(cohorts)
for cohort in cohorts:
for age in range(complete_end.ordinal - cohort.ordinal + 1):
matrix.loc[cohort, age] = counts.get((cohort, age), 0) / sizes.loc[cohort]
matrix.index = matrix.index.astype(str)
sizes.index = sizes.index.astype(str)
return matrix, sizes
def summarize(df: pd.DataFrame, audit: dict) -> dict:
sales = df[df["is_sale"]].copy()
ledger = df[df["is_ledger"]]
gross = float(sales["line_value_gbp"].sum())
credits = float(-ledger.loc[ledger["Quantity"] < 0, "line_value_gbp"].sum())
net = float(ledger["line_value_gbp"].sum())
# Cancellations with positive quantity require manual classification.
if audit["positive_quantity_cancellation_rows"]:
raise ValueError("Positive-quantity cancellations found; resolve their business meaning first.")
if not np.isclose(gross - credits, net, atol=1e-6, rtol=0):
raise AssertionError("Sales minus credits does not reconcile to the signed ledger")
snapshot = df["InvoiceDate"].max().normalize() + pd.Timedelta(days=1)
rfm = customer_segments(sales, snapshot)
known_gross = float(rfm["monetary_gbp"].sum())
top_count = math.ceil(len(rfm) * 0.1)
cohort, sizes = cohort_retention(sales)
monthly = sales.groupby("month").agg(gross_sales_gbp=("line_value_gbp", "sum"), orders=("InvoiceNo", "nunique"))
monthly["net_invoiced_gbp"] = ledger.groupby("month")["line_value_gbp"].sum()
monthly["complete_month"] = monthly.index < df["month"].max()
monthly["average_order_value_gbp"] = monthly["gross_sales_gbp"] / monthly["orders"]
segment = rfm.groupby("segment").agg(customers=("CustomerID", "size"), gross_sales_gbp=("monetary_gbp", "sum"))
segment["sales_share"] = segment["gross_sales_gbp"] / known_gross
metrics = {
**audit, "snapshot_date": snapshot.strftime("%Y-%m-%d"),
"gross_invoiced_sales_gbp": gross, "credits_gbp": credits,
"net_invoiced_value_gbp": net, "positive_sale_invoices": int(sales["InvoiceNo"].nunique()),
"identified_customers": len(rfm), "identified_sales_gbp": known_gross,
"identified_sales_share": known_gross / gross,
"average_order_value_gbp": gross / sales["InvoiceNo"].nunique(),
"repeat_buyer_share": float((rfm["frequency"] > 1).mean()),
"top_decile_customer_count": top_count,
"top_decile_sales_share": float(rfm.nlargest(top_count, "monetary_gbp")["monetary_gbp"].sum() / known_gross),
"uk_sales_share": float(sales.loc[sales["Country"] == "United Kingdom", "line_value_gbp"].sum() / gross),
"duplicate_sensitivity_gbp": gross - audit["gross_sales_if_exact_repeats_removed_gbp"],
}
return {"metrics": metrics, "monthly": monthly, "rfm": rfm, "segments": segment, "cohorts": cohort, "cohort_sizes": sizes}
def run_sql(df: pd.DataFrame, result: dict) -> None:
OUTPUT.mkdir(parents=True, exist_ok=True)
sql_df = df[["InvoiceNo", "StockCode", "InvoiceDate", "CustomerID", "Country", "Quantity", "UnitPrice", "line_value_gbp", "month", "is_sale", "is_ledger"]]
with sqlite3.connect(OUTPUT / "retail.sqlite") as conn:
sql_df.to_sql("invoice_lines", conn, index=False, if_exists="replace")
conn.execute("CREATE INDEX IF NOT EXISTS idx_customer ON invoice_lines(CustomerID)")
for query in sorted((PROJECT / "sql").glob("*.sql")):
table = pd.read_sql_query(query.read_text(encoding="utf-8"), conn)
table.to_csv(OUTPUT / f"sql_{query.stem}.csv", index=False)
if query.stem == "01_monthly_sales":
expected = result["monthly"]["gross_sales_gbp"].to_numpy()
np.testing.assert_allclose(table["gross_sales_gbp"], expected, atol=1e-6, rtol=0)
np.testing.assert_array_equal(table["orders"], result["monthly"]["orders"])
if query.stem == "03_customer_segments":
check = table.set_index("segment").sort_index()
expected = result["segments"].sort_index()
np.testing.assert_array_equal(check["customers"], expected["customers"])
np.testing.assert_allclose(check["gross_sales_gbp"], expected["gross_sales_gbp"], atol=1e-6, rtol=0)
def make_figures(result: dict) -> None:
FIGURES.mkdir(parents=True, exist_ok=True)
plt.rcParams.update({"font.size": 11, "axes.spines.top": False, "axes.spines.right": False, "axes.titleweight": "bold"})
monthly = result["monthly"]
fig, ax = plt.subplots(figsize=(11, 5), layout="constrained")
colors = ["#164e99" if complete else "#c6d1e2" for complete in monthly["complete_month"]]
ax.bar(monthly.index, monthly["gross_sales_gbp"], color=colors)
ax.yaxis.set_major_formatter(FuncFormatter(lambda v, _: f"£{v / 1e6:.1f}m"))
ax.tick_params(axis="x", rotation=45)
ax.set(title="Monthly gross invoiced sales", ylabel="GBP", xlabel="December 2011 ends on 9 December; do not compare with full months")
ax.grid(axis="y", alpha=.15)
fig.savefig(FIGURES / "monthly_sales.png", dpi=160)
plt.close(fig)
matrix = result["cohorts"].iloc[:, :7]
fig, ax = plt.subplots(figsize=(10, 6), layout="constrained")
cmap = plt.get_cmap("Blues").copy()
cmap.set_bad("#ededed")
im = ax.imshow(matrix, cmap=cmap, vmin=0, vmax=1, aspect="auto")
ax.set_xticks(range(len(matrix.columns)), [f"M{n}" for n in matrix.columns])
ax.set_yticks(range(len(matrix)), matrix.index)
for i in range(len(matrix)):
for j in range(len(matrix.columns)):
value = matrix.iloc[i, j]
if pd.notna(value):
ax.text(j, i, f"{value:.0%}", ha="center", va="center", color="white" if value > .6 else "#15243b", fontsize=10)
ax.set(title="Observed customer purchase retention", ylabel="First purchase month observed", xlabel="Months since first observed purchase · grey = not yet observable")
fig.colorbar(im, ax=ax, format=PercentFormatter(1), fraction=.035)
fig.savefig(FIGURES / "cohort_retention.png", dpi=160)
plt.close(fig)
segments = result["segments"].sort_values("gross_sales_gbp")
fig, ax = plt.subplots(figsize=(10, 4.5), layout="constrained")
ax.barh(segments.index, segments["sales_share"], color="#164e99")
ax.xaxis.set_major_formatter(PercentFormatter(1))
ax.set(title="Sales by customer segment", xlabel="Share of gross sales with an identified customer")
fig.savefig(FIGURES / "customer_segments.png", dpi=160)
plt.close(fig)
def write_outputs(result: dict) -> None:
OUTPUT.mkdir(parents=True, exist_ok=True)
for key in ["monthly", "segments", "cohorts"]:
result[key].to_csv(OUTPUT / f"{key}.csv")
result["rfm"].to_csv(OUTPUT / "customer_segments.csv", index=False)
result["cohort_sizes"].rename("customers").to_csv(OUTPUT / "cohort_sizes.csv")
(OUTPUT / "metrics.json").write_text(json.dumps(result["metrics"], indent=2) + "\n", encoding="utf-8")
def main() -> dict:
print("Retail: reading source workbook", flush=True)
df, audit = prepare(load_data())
result = summarize(df, audit)
write_outputs(result)
run_sql(df, result)
make_figures(result)
print(json.dumps(result["metrics"], indent=2), flush=True)
return result
if __name__ == "__main__":
main()