Customer Analytics in Python: Cohort Analysis, RFM Segmentation & LTV

import pandas as pd
import numpy as np
import datetime as dt
import matplotlib.pyplot as plt
import seaborn as sns
 
# Visual formatting configuration
pd.set_option('display.max_columns', None)
pd.set_option('display.float_format', lambda x: f'{x:.2f}')
sns.set_theme(style='whitegrid')
 
# Generate synthetic transaction log
np.random.seed(42)
n_transactions = 6000
n_customers = 900
 
start_date = dt.datetime(2025, 1, 1)
date_list = [start_date + dt.timedelta(days=int(np.random.randint(0, 365))) for _ in range(n_transactions)]
customer_ids = np.random.randint(1000, 1000 + n_customers, size=n_transactions)
quantities = np.random.randint(1, 8, size=n_transactions)
unit_prices = np.round(np.random.uniform(15.0, 120.0, size=n_transactions), 2)
 
df = pd.DataFrame({
    'CustomerID': customer_ids,
    'InvoiceDate': date_list,
    'Quantity': quantities,
    'UnitPrice': unit_prices
})
 
# Calculate total transaction revenue
df['TotalSpend'] = df['Quantity'] * df['UnitPrice']
 
print(f"Transaction Log: {df.shape[0]:,} rows across {df['CustomerID'].nunique():,} unique customers")
df.sort_values('InvoiceDate').head()

# 1. Establish reference date (representing 'today')
snapshot_date = df['InvoiceDate'].max() + dt.timedelta(days=1)
print(f"Analysis Snapshot Date: {snapshot_date.strftime('%Y-%m-%d')}")
 
# 2. Aggregate Recency, Frequency, and Monetary spend by CustomerID
rfm = df.groupby('CustomerID').agg({
    'InvoiceDate': lambda x: (snapshot_date - x.max()).days,  # Recency (days)
    'CustomerID': 'count',                                   # Frequency (order count)
    'TotalSpend': 'sum'                                      # Monetary (total revenue)
}).rename(columns={
    'InvoiceDate': 'Recency',
    'CustomerID': 'Frequency',
    'TotalSpend': 'Monetary'
})
 
rfm.head()

# Define quartile labels
r_labels = [4, 3, 2, 1]  # 4 = Most Recent (lowest days)
fm_labels = [1, 2, 3, 4] # 4 = Highest Frequency / Spend
 
# Score Recency (lower days = higher score)
rfm['R_Score'] = pd.qcut(rfm['Recency'], q=4, labels=r_labels).astype(int)
 
# Score Frequency (use rank to resolve tied frequencies)
rfm['F_Score'] = pd.qcut(rfm['Frequency'].rank(method='first'), q=4, labels=fm_labels).astype(int)
 
# Score Monetary
rfm['M_Score'] = pd.qcut(rfm['Monetary'], q=4, labels=fm_labels).astype(int)
 
# Combine into a 3-digit RFM string (e.g., '444')
rfm['RFM_Segment'] = (
    rfm['R_Score'].astype(str) +
    rfm['F_Score'].astype(str) +
    rfm['M_Score'].astype(str)
)
 
# Combined RFM Score sum (3 to 12)
rfm['RFM_Score_Sum'] = rfm['R_Score'] + rfm['F_Score'] + rfm['M_Score']
 
rfm.head()

def assign_segment(row) -> str:
    r = row['R_Score']
    f = row['F_Score']
    
    if r >= 4 and f >= 4:
        return 'Champions'
    elif r >= 3 and f >= 3:
        return 'Loyal Customers'
    elif r >= 3 and f <= 2:
        return 'Potential Loyalists'
    elif r <= 2 and f >= 3:
        return 'At Risk'
    elif r <= 1 and f <= 1:
        return 'Lost Customers'
    else:
        return 'Need Attention'
 
rfm['Segment'] = rfm.apply(assign_segment, axis=1)
 
# Segment breakdown summary
segment_summary = rfm.groupby('Segment').agg(
    Customer_Count=('Recency', 'count'),
    Avg_Recency=('Recency', 'mean'),
    Avg_Frequency=('Frequency', 'mean'),
    Total_Revenue=('Monetary', 'sum'),
    Avg_Revenue=('Monetary', 'mean')
).sort_values(by='Total_Revenue', ascending=False)
 
segment_summary['Revenue_Share_Pct'] = (segment_summary['Total_Revenue'] / segment_summary['Total_Revenue'].sum()) * 100
segment_summary

plt.figure(figsize=(10, 5))
order = rfm['Segment'].value_counts().index
palette = sns.color_palette("mako", len(order))
 
sns.countplot(
    data=rfm,
    y='Segment',
    order=order,
    palette=palette
)
plt.title("Customer Distribution Across RFM Segments", fontsize=14, fontweight='bold', pad=12)
plt.xlabel("Number of Customers")
plt.ylabel("Segment")
plt.tight_layout()
plt.show()

cohort_df = df.copy()
 
# 1. Truncate transaction dates to month start
def get_month(date_val):
    return dt.datetime(date_val.year, date_val.month, 1)
 
cohort_df['InvoiceMonth'] = cohort_df['InvoiceDate'].apply(get_month)
 
# 2. Identify the first purchase month for each customer (Acquisition Cohort)
cohort_df['CohortMonth'] = cohort_df.groupby('CustomerID')['InvoiceMonth'].transform('min')
 
# 3. Calculate CohortIndex (Months elapsed since acquisition)
def get_month_offset(d1, d2):
    return (d1.year - d2.year) * 12 + (d1.month - d2.month)
 
cohort_df['CohortIndex'] = cohort_df.apply(
    lambda row: get_month_offset(row['InvoiceMonth'], row['CohortMonth']),
    axis=1
)
 
cohort_df[['CustomerID', 'InvoiceDate', 'CohortMonth', 'InvoiceMonth', 'CohortIndex']].head()

# 1. Count distinct customers per CohortMonth and CohortIndex
cohort_counts = cohort_df.groupby(['CohortMonth', 'CohortIndex'])['CustomerID'].nunique().reset_index()
 
# 2. Pivot into a 2D matrix
cohort_pivot = cohort_counts.pivot(index='CohortMonth', columns='CohortIndex', values='CustomerID')
 
# 3. Base cohort size (Month 0 active customers)
cohort_size = cohort_pivot.iloc[:, 0]
 
# 4. Divide across columns to compute Retention Rate percentages
retention_matrix = cohort_pivot.divide(cohort_size, axis=0) * 100
retention_matrix.index = retention_matrix.index.strftime('%Y-%m')
 
print("Cohort Retention Percentage Matrix (%):")
print(retention_matrix.iloc[:, :7])

plt.figure(figsize=(14, 8))
sns.heatmap(
    retention_matrix,
    annot=True,
    fmt='.1f',
    cmap='YlGnBu',
    vmin=0.0,
    vmax=60.0,
    linewidths=0.5,
    cbar_kws={'label': 'Retention Rate (%)'}
)
plt.title("Monthly Customer Retention Heatmap by Cohort", fontsize=15, fontweight='bold', pad=15)
plt.xlabel("Months Since Acquisition (Cohort Index)")
plt.ylabel("Acquisition Cohort Month")
plt.tight_layout()
plt.show()

# Calculate business-wide baseline metrics
avg_order_val = df['TotalSpend'].mean()
avg_orders_per_customer = df.shape[0] / df['CustomerID'].nunique()
gross_margin_pct = 0.65  # 65% gross margin
 
# Average monthly retention rate after month 1
avg_retention_m1 = retention_matrix.iloc[:, 1].mean() / 100
monthly_churn_rate = 1.0 - avg_retention_m1
 
# Calculate projected LTV
predicted_ltv = (avg_order_val * avg_orders_per_customer * gross_margin_pct) / monthly_churn_rate
 
print("=" * 45)
print(f"AVERAGE ORDER VALUE:        ${avg_order_val:.2f}")
print(f"AVG PURCHASE FREQUENCY:     {avg_orders_per_customer:.2f} orders/year")
print(f"ESTIMATED MONTHLY CHURN:    {monthly_churn_rate * 100:.1f}%")
print(f"PREDICTED CUSTOMER LTV:     ${predicted_ltv:.2f}")
print("=" * 45)

import pandas as pd
import datetime as dt
 
def run_customer_analytics_pipeline(df: pd.DataFrame):
    """
    Takes transaction DataFrame [CustomerID, InvoiceDate, TotalSpend]
    and computes full RFM segments + Cohort Retention Pivot.
    """
    df['InvoiceDate'] = pd.to_datetime(df['InvoiceDate'])
    snapshot = df['InvoiceDate'].max() + dt.timedelta(days=1)
    
    # 1. RFM Table
    rfm = df.groupby('CustomerID').agg({
        'InvoiceDate': lambda x: (snapshot - x.max()).days,
        'CustomerID': 'count',
        'TotalSpend': 'sum'
    }).rename(columns={'InvoiceDate': 'Recency', 'CustomerID': 'Frequency', 'TotalSpend': 'Monetary'})
    
    # 2. RFM Scores
    rfm['R'] = pd.qcut(rfm['Recency'], q=4, labels=[4, 3, 2, 1]).astype(int)
    rfm['F'] = pd.qcut(rfm['Frequency'].rank(method='first'), q=4, labels=[1, 2, 3, 4]).astype(int)
    rfm['M'] = pd.qcut(rfm['Monetary'], q=4, labels=[1, 2, 3, 4]).astype(int)
    
    # 3. Segments
    def segment(r, f):
        if r >= 4 and f >= 4: return 'Champions'
        if r >= 3 and f >= 3: return 'Loyal'
        if r <= 2 and f >= 3: return 'At Risk'
        if r <= 1 and f <= 1: return 'Lost'
        return 'Potential'
    
    rfm['Segment'] = [segment(r, f) for r, f in zip(rfm['R'], rfm['F'])]
    return rfm