What’s includedRFM segmentation computes Recency, Frequency, and

Monetary value per customer to support behavior-based segmentation

suitable for marketing and lifetime value strategies.

�Cohort analysis groups customers by the month of first purchase and

tracks retention across periods to reveal repeat purchasing patterns over
time. �

Delivery performance includes on-time delivery rate using the formula [ \

text{OTD} = \frac{\text{On-time Deliveries}}{\text{Total Deliveries}} \
times 100 ], enabling reliability tracking for logistics operations.

�City-wise analysis aggregates orders, revenue, customers, and delivery

KPIs by location to surface regional performance differences and operational
opportunities.

Python code

# E-commerce Data Analysis Toolkit

# Orders • Customers • Delivery Status • City-wise Analysis

# RFM Segmentation • Cohort Retention • On-time Delivery

From __future__ import annotations

Import pandas as pd

Import numpy as np

From pathlib import Path

From dataclasses import dataclass

From typing import Optional, Dict, List, Tuple

From datetime import datetime

Import warnings

[Link](“ignore”)
@dataclass

Class Schema:

# Minimum columns expected in the orders dataset; map to actual column

names if different.

Order_id: str = “order_id”

Order_date: str = “order_date”

Customer_id: str = “customer_id”

City: str = “city”

State: str = “state”

Country: str = “country”

Status: str = “status” # e.g., ‘delivered’, ‘shipped’, ‘canceled’

Promised_date: str = “promised_date” # promised delivery date

Delivered_date: str = “delivered_date” # actual delivery date (nullable)

Item_id: str = “item_id”

Quantity: str = “quantity”

Unit_price: str = “unit_price”

Discount: str = “discount” # absolute discount amount for the line (can
be 0)

Shipping_cost: str = “shipping_cost” # per order or allocated per line

(optional)

# Optional customer attributes table (if separate)

# If provided, must include at least customer_id

Cust_table_customer_id: Optional[str] = None

@dataclass

Class Paths:

Orders_csv: Path

Customers_csv: Optional[Path] = None # optional, enrich customers if

available

Output_dir: Path = Path(“outputs”)

Def _to_datetime_safe(s: [Link]) -> [Link]:

Return pd.to_datetime(s, errors=”coerce”, utc=True).dt.tz_localize(None)

Def load_orders(paths: Paths, schema: Schema) -> [Link]:

Df = pd.read_csv(paths.orders_csv)

# Normalize column names (lowercase, strip)

[Link] = [[Link]() for c in [Link]]

# Parse dates

If schema.order_date in [Link]:

Df[schema.order_date] = _to_datetime_safe(df[schema.order_date])

If schema.promised_date in [Link]:

Df[schema.promised_date] =
_to_datetime_safe(df[schema.promised_date])

If schema.delivered_date in [Link]:

Df[schema.delivered_date] =
_to_datetime_safe(df[schema.delivered_date])

# Numeric safeguards

For col in [[Link], schema.unit_price, [Link]]:

 If col in [Link]:

Df[col] = pd.to_numeric(df[col], errors=”coerce”).fillna(0)

If schema.shipping_cost in [Link]:

Df[schema.shipping_cost] = pd.to_numeric(df[schema.shipping_cost],
errors=”coerce”).fillna(0)

Else:

Df[schema.shipping_cost] = 0.0

# Compute line revenue net of discount (excluding shipping; add later per
order)

Df[“line_revenue”] = df[[Link]].fillna(0) *
df[schema.unit_price].fillna(0) – df[[Link]].fillna(0)

# Order-level shipping allocation: if shipping_cost is recorded per order,

allocate evenly across lines of the order.

# If it’s already per line, remove this allocation logic as needed.

If schema.order_id in [Link] and schema.shipping_cost in [Link]:

Ship_per_order = [Link](schema.order_id)
[schema.shipping_cost].transform(“max”)

Lines_per_order = [Link](schema.order_id)
[schema.item_id].transform(“nunique”).clip(lower=1)

Df[“line_shipping_alloc”] = ship_per_order / lines_per_order

Else:

Df[“line_shipping_alloc”] = 0.0

Df[“line_total”] = df[“line_revenue”] + df[“line_shipping_alloc”]

# Status normalization
 If [Link] in [Link]:

Df[[Link]] = df[[Link]].[Link]().[Link]()

Else:

Df[[Link]] = “unknown”

# City/state/country fill

For col in [[Link], [Link], [Link]]:

If col not in [Link]:

Df[col] = “Unknown”

Df[col] = df[col].fillna(“Unknown”).astype(str).[Link]()

# Filter out canceled or invalid rows for revenue-calculating analyses

Return df

Def enrich_customers(df_orders: [Link], paths: Paths, schema:

Schema) -> [Link]:

If paths.customers_csv is None:

Return df_orders

Cust = pd.read_csv(paths.customers_csv)

[Link] = [[Link]() for c in [Link]]

Key = schema.cust_table_customer_id or schema.customer_id

If key not in [Link]:

Return df_orders

Return df_orders.merge(cust, left_on=schema.customer_id, right_on=key,

how=”left”)
Def compute_order_level(df: [Link], schema: Schema) ->
[Link]:

# Aggregate lines to orders

Agg = [Link](schema.order_id).agg(

Order_date=(schema.order_date, “max”),

Customer_id=(schema.customer_id, “first”),

City=([Link], “first”),

State=([Link], “first”),

Country=([Link], “first”),

Status=([Link], lambda s: [Link]().iat[0] if len([Link]()) else

“unknown”),

Promised_date=(schema.promised_date, “max”),

Delivered_date=(schema.delivered_date, “max”),

Items=(“line_total”, “size”),

Quantity=([Link], “sum”),

Revenue=(“line_revenue”, “sum”),

Shipping=(“line_shipping_alloc”, “sum”),

Total_value=(“line_total”, “sum”),

).reset_index()

# On-time delivery flag (consider only delivered)

Agg[“is_delivered”] = ~agg[“delivered_date”].isna()

Agg[“on_time”] = [Link](

Agg[“is_delivered”] & agg[“promised_date”].notna(),

(agg[“delivered_date”] <= agg[“promised_date”]).astype(int),

[Link], # unknown if not delivered or promise missing

)
 # Calendar helpers

Agg[“order_month”] = agg[“order_date”].dt.to_period(“M”).astype(str)

Return agg

Def customer_first_order(orders: [Link], schema: Schema) ->

[Link]:

Firsts = (

[Link](schema.customer_id)[“order_date”]

.min()

.reset_index()

.rename(columns={“order_date”: “first_order_date”})

Firsts[“cohort_month”] =
firsts[“first_order_date”].dt.to_period(“M”).astype(str)

Return firsts

Def compute_customer_aggregates(orders: [Link], schema: Schema,

as_of: Optional[[Link]] = None) -> [Link]:

Firsts = customer_first_order(orders, schema)

As_of = as_of or orders[“order_date”].max()

Cx = (

[Link](schema.customer_id)

.agg(

Orders=(“order_id”, “nunique”),

Revenue=(“revenue”, “sum”),

Total_value=(“total_value”, “sum”),
 Last_order=(“order_date”, “max”),

.reset_index()

.merge(firsts, on=schema.customer_id, how=”left”)

# RFM

Cx[“recency_days”] = (as_of – cx[“last_order”]).[Link]

Cx[“frequency”] = cx[“orders”].fillna(0)

Cx[“monetary”] = cx[“revenue”].fillna(0.0)

# RFM scoring (quantile-based; reverse for recency: lower days =>

better/higher score)

# Adjust bins if desired; ensure unique edges by rank method

Cx[“r_score”] = [Link](cx[“recency_days”].rank(method=”first”), q=5,

labels=[5, 4, 3, 2, 1]).astype(int)

Cx[“f_score”] = [Link](cx[“frequency”].rank(method=”first”), q=5,

labels=[1, 2, 3, 4, 5]).astype(int)

Cx[“m_score”] = [Link](cx[“monetary”].rank(method=”first”), q=5,

labels=[1, 2, 3, 4, 5]).astype(int)

Cx[“rfm_score”] = cx[“r_score”] + cx[“f_score”] + cx[“m_score”]

# Simple segmentation tiers

Cx[“rfm_segment”] = [Link](cx[“rfm_score”].rank(method=”first”), q=3,

labels=[“Low-Value”, “Mid-Value”, “High-Value”])

Return cx
Def build_cohort_retention(orders: [Link], schema: Schema) ->
Tuple[[Link], [Link]]:

# Build a customer-level table of first purchase month (cohort) and each

order month

Df = orders[[schema.customer_id, “order_id”,
“order_date”]].dropna().drop_duplicates()

Df[“cohort”] = [Link](schema.customer_id)
[“order_date”].transform(“min”).dt.to_period(“M”)

Df[“order_month”] = df[“order_date”].dt.to_period(“M”)

# Count unique active customers per cohort per period index

Cohorts = (

[Link]([“cohort”, “order_month”])

.agg(n_customers=(schema.customer_id, “nunique”))

.reset_index()

Cohorts[“period_index”] = (cohorts[“order_month”] –
cohorts[“cohort”]).apply(lambda p: p.n)

# Pivot to retention matrices

Retention_abs = cohorts.pivot_table(index=”cohort”,
columns=”order_month”, values=”n_customers”, aggfunc=”sum”).fillna(0)

Base_sizes = [Link][cohorts[“period_index”] == 0, [“cohort”,

“n_customers”]].set_index(“cohort”)[“n_customers”]

Retention_rel = (

Cohorts.pivot_table(index=”cohort”, columns=”period_index”,
values=”n_customers”, aggfunc=”sum”)

.div(base_sizes, axis=0)

.fillna(0.0)

)
 # Stringify periods for output readability

Retention_abs.index = retention_abs.[Link](str)

Retention_abs.columns = retention_abs.[Link](str)

Retention_rel.index = retention_rel.[Link](str)

Retention_rel.columns = [str(c) for c in retention_rel.columns]

Return retention_abs, retention_rel

Def delivery_kpis(orders: [Link]) -> [Link]:

# Compute OTD at various grains

Df = [Link]()

# Only consider delivered orders for OTD numerator/denominator

Delivered = df[df[“is_delivered”]].copy()

Delivered[“on_time_flag”] = delivered[“on_time”].fillna(0).astype(int)

Def rate(x: [Link]) -> float:

Denom = [Link]

Num = [Link]()

Return float(num) / float(denom) if denom > 0 else [Link]

# Daily, weekly, monthly OTD rates

Otd_by_day = [Link](delivered[“delivered_date”].[Link])
[“on_time_flag”].apply(rate).rename(“otd_rate”).reset_index()

Otd_by_week =
[Link](delivered[“delivered_date”].dt.to_period(“W”))
[“on_time_flag”].apply(rate).rename(“otd_rate”).reset_index()
 Otd_by_month =
[Link](delivered[“delivered_date”].dt.to_period(“M”))
[“on_time_flag”].apply(rate).rename(“otd_rate”).reset_index()

# Overall OTD

Overall = [Link]({“metric”: [“otd_overall”], “value”:

[rate(delivered[“on_time_flag”])]})

Return {

“otd_by_day”: otd_by_day,

“otd_by_week”: otd_by_week,

“otd_by_month”: otd_by_month,

“otd_overall”: overall,

Def city_level(orders: [Link], schema: Schema) -> [Link]:

# Summarize by city

G = [Link]([Link]).agg(

Orders=(“order_id”, “nunique”),

Customers=(schema.customer_id, “nunique”),

Revenue=(“revenue”, “sum”),

Shipping=(“shipping”, “sum”),

Total_value=(“total_value”, “sum”),

Delivered=(“is_delivered”, “sum”),

Delivered_orders=(“is_delivered”, “sum”),

On_time_deliveries=(“on_time”, lambda s: [Link](0).sum()),

Delivered_with_promise=(“promised_date”, lambda s: [Link]().sum()),

 ).reset_index()

# OTD denominator is delivered_with_promise to handle missing promised

dates

G[“otd_rate”] = [Link](

G[“delivered_with_promise”] > 0,

G[“on_time_deliveries”] / g[“delivered_with_promise”],

[Link],

# Per-order averages

G[“avg_order_value”] = [Link](g[“orders”] > 0, g[“total_value”] /

g[“orders”], [Link])

G[“avg_revenue_per_order”] = [Link](g[“orders”] > 0, g[“revenue”] /

g[“orders”], [Link])

# Sort by revenue descending

G = g.sort_values(“revenue”, ascending=False).reset_index(drop=True)

Return g

Def kpis_overview(orders: [Link]) -> [Link]:

# High-level KPIs table

Total_orders = orders[“order_id”].nunique()

Total_customers = orders[“customer_id”].nunique()

Delivered_orders = orders[“is_delivered”].sum()

Revenue = orders[“revenue”].sum()

Shipping = orders[“shipping”].sum()
Total_value = orders[“total_value”].sum()

Avg_order_value = total_value / total_orders if total_orders > 0 else [Link]

Return [Link](

“metric”: [

“total_orders”,

“total_customers”,

“delivered_orders”,

“revenue”,

“shipping”,

“total_value”,

“avg_order_value”,

],

“value”: [

Total_orders,

Total_customers,

Delivered_orders,

Revenue,

Shipping,

Total_value,

Avg_order_value,

],

)
Def save_frames(frames: Dict[str, [Link]], outdir: Path) -> None:

[Link](parents=True, exist_ok=True)

For name, df in [Link]():

P = outdir / f”{name}.csv”

Df.to_csv(p, index=False)

Def run_analysis(

Orders_path: str,

Customers_path: Optional[str] = None,

Output_dir: str = “outputs”,

Schema: Schema = Schema(),

) -> Dict[str, [Link]]:

Paths = Paths(orders_csv=Path(orders_path),
customers_csv=Path(customers_path) if customers_path else None,
output_dir=Path(output_dir))

# Load

Raw = load_orders(paths, schema)

Enriched = enrich_customers(raw, paths, schema)

# Order-level aggregation

Orders = compute_order_level(enriched, schema)

# KPIs

Kpi = kpis_overview(orders)
# Customer aggregates + RFM

Cx = compute_customer_aggregates(orders, schema)

# Cohorts

Retention_abs, retention_rel = build_cohort_retention(orders, schema)

# Delivery

Delivery = delivery_kpis(orders)

# City-level

City = city_level(orders, schema)

# Exports

Frames = {

“orders_agg”: orders,

“kpis_overview”: kpi,

“customers_rfm”: cx,

“cohort_retention_absolute”: retention_abs.reset_index(),

“cohort_retention_relative”: retention_rel.reset_index(),

“delivery_otd_by_day”: delivery[“otd_by_day”],

“delivery_otd_by_week”: delivery[“otd_by_week”],

“delivery_otd_by_month”: delivery[“otd_by_month”],

“delivery_otd_overall”: delivery[“otd_overall”],

“city_summary”: city,

Save_frames(frames, paths.output_dir)

Return frames
If __name__ == “__main__”:

# Example usage:

# 1) Ensure orders CSV has columns matching Schema or adjust Schema

accordingly.

# 2) Run the script; outputs will be saved as CSVs in ./outputs

Frames = run_analysis(

Orders_path=”data/[Link]”,

Customers_path=None, # e.g., “data/[Link]” if available

Output_dir=”outputs”,

Schema=Schema(

Order_id=”order_id”,

Order_date=”order_date”,

Customer_id=”customer_id”,

City=”city”,

State=”state”,

Country=”country”,

Status=”status”,

Promised_date=”promised_date”,

Delivered_date=”delivered_date”,

Item_id=”item_id”,

Quantity=”quantity”,

Unit_price=”unit_price”,

Discount=”discount”,

Shipping_cost=”shipping_cost”,

),
 )

# After running, inspect ./outputs/*.csv

How to adaptRFM scoring granularity can be changed by adjusting quantile

bins or using fixed binning for recency, frequency, and monetary scores
depending on business thresholds. �

Cohort definitions can shift from first purchase month to acquisition channel
or campaign to align retention tracking with marketing segmentation.

�On-time delivery can be segmented by region, carrier, or product category

to isolate bottlenecks and focus operational improvements on specific lanes.

�Notes on methodsRFM uses three dimensions—Recency, Frequency,

Monetary—to classify customer value and target lifecycle actions. �Cohort
retention matrices compare cohorts at equal age since first purchase to
highlight repeat purchasing trends and seasonality effects.

�On-time delivery rate quantifies fulfillment reliability using [ \text{OTD} = \

frac{\text{On-time Deliveries}}{\text{Total Deliveries}} \times 100 ]

and benefits from routine monitoring and segmentation. �