Customer Churn Prediction for Telecommunications: CDR Feature Engineering, Survival Analysis, and Retention Campaigns
A comprehensive guide to building production churn prediction systems for telco — CDR feature engineering, survival analysis, uplift modeling, and retention campaign design.
Customer Churn Prediction for Telecommunications: CDR Feature Engineering, Survival Analysis, and Retention Campaigns
Telecommunications churn prediction is one of the classic problems in applied data science, and after building churn models for multiple telco operators across Southeast Asia, I can tell you that the problem is far more nuanced than the Kaggle tutorials suggest. The technical challenge of building a churn model is moderate. The business challenge of turning churn predictions into retained customers is where most projects fail.
A typical telco operator loses 1-3% of its subscriber base per month. For an operator with 50 million subscribers, that’s 500,000 to 1.5 million subscribers leaving every month. At an average acquisition cost of $30-50 per subscriber, the financial incentive for effective churn prediction is enormous.
But here’s the thing most data scientists miss: predicting churn is not the same as preventing churn. You can build a model with 95% AUC that is completely useless if you don’t know which customers to intervene with, what intervention to use, and when to apply it. This post covers the full pipeline — from raw Call Detail Records to retention campaigns that actually reduce churn.
CDR Feature Engineering
Understanding Call Detail Records
Call Detail Records (CDRs) are the raw material of telco analytics. Every call, text, and data session generates a record with timestamp, duration, cell tower, and the parties involved. A mid-sized operator generates billions of CDRs per month.
The feature engineering challenge is distilling this raw behavioral data into predictive signals:
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
class CDRFeatureEngine:
"""Feature engineering from Call Detail Records for churn prediction."""
def __init__(self, reference_date: str):
self.reference_date = pd.Timestamp(reference_date)
def engineer_features(
self,
cdr_data: pd.DataFrame,
subscriber_data: pd.DataFrame,
lookback_windows: list[int] = [7, 14, 30, 60, 90],
) -> pd.DataFrame:
"""Engineer churn prediction features from CDR data."""
features = subscriber_data[["subscriber_id"]].copy()
for window_days in lookback_windows:
window_start = self.reference_date - timedelta(days=window_days)
window_cdr = cdr_data[
(cdr_data["timestamp"] >= window_start)
& (cdr_data["timestamp"] < self.reference_date)
]
# Voice features
voice_features = self._voice_features(window_cdr, window_days)
voice_features = voice_features.rename(
columns={col: f"{col}_{window_days}d" for col in voice_features.columns if col != "subscriber_id"}
)
features = features.merge(voice_features, on="subscriber_id", how="left")
# Data usage features
data_features = self._data_features(window_cdr, window_days)
data_features = data_features.rename(
columns={col: f"{col}_{window_days}d" for col in data_features.columns if col != "subscriber_id"}
)
features = features.merge(data_features, on="subscriber_id", how="left")
# Trend features (comparing windows)
features = self._add_trend_features(features, lookback_windows)
# Static features from subscriber data
features = features.merge(
self._subscriber_features(subscriber_data),
on="subscriber_id",
how="left",
)
return features
def _voice_features(self, cdr: pd.DataFrame, window_days: int) -> pd.DataFrame:
"""Extract voice-related features."""
voice = cdr[cdr["service_type"] == "voice"]
agg = voice.groupby("subscriber_id").agg(
voice_call_count=("duration_sec", "count"),
voice_total_duration=("duration_sec", "sum"),
voice_avg_duration=("duration_sec", "mean"),
voice_std_duration=("duration_sec", "std"),
voice_max_duration=("duration_sec", "max"),
voice_unique_contacts=("callee_id", "nunique"),
voice_incoming_count=("direction", lambda x: (x == "incoming").sum()),
voice_outgoing_count=("direction", lambda x: (x == "outgoing").sum()),
).reset_index()
# Call pattern features
if len(voice) > 0:
# Peak hour ratio (9 AM - 6 PM)
voice["hour"] = voice["timestamp"].dt.hour
peak_ratio = voice.groupby("subscriber_id").apply(
lambda x: ((x["hour"] >= 9) & (x["hour"] <= 18)).mean()
).reset_index(name="voice_peak_hour_ratio")
agg = agg.merge(peak_ratio, on="subscriber_id", how="left")
# Weekend ratio
voice["is_weekend"] = voice["timestamp"].dt.dayofweek >= 5
weekend_ratio = voice.groupby("subscriber_id")["is_weekend"].mean().reset_index(name="voice_weekend_ratio")
agg = agg.merge(weekend_ratio, on="subscriber_id", how="left")
# Normalize by window length
agg["voice_calls_per_day"] = agg["voice_call_count"] / window_days
agg["voice_minutes_per_day"] = agg["voice_total_duration"] / 60 / window_days
return agg
def _data_features(self, cdr: pd.DataFrame, window_days: int) -> pd.DataFrame:
"""Extract data usage features."""
data = cdr[cdr["service_type"] == "data"]
agg = data.groupby("subscriber_id").agg(
data_session_count=("data_volume_mb", "count"),
data_total_volume_mb=("data_volume_mb", "sum"),
data_avg_session_mb=("data_volume_mb", "mean"),
data_max_session_mb=("data_volume_mb", "max"),
).reset_index()
agg["data_mb_per_day"] = agg["data_total_volume_mb"] / window_days
return agg
def _add_trend_features(self, features: pd.DataFrame, windows: list[int]) -> pd.DataFrame:
"""Add trend features comparing recent vs historical behavior."""
# Compare 7-day to 30-day metrics
for metric in ["voice_call_count", "voice_total_duration", "data_total_volume_mb"]:
short = f"{metric}_{windows[0]}d"
long = f"{metric}_{windows[2]}d" # 30-day
if short in features.columns and long in features.columns:
# Ratio: <1 means declining, >1 means increasing
features[f"{metric}_trend"] = (
features[short] / features[long].clip(lower=0.01)
)
return features
def _subscriber_features(self, subscriber_data: pd.DataFrame) -> pd.DataFrame:
"""Extract static subscriber features."""
features = subscriber_data[[
"subscriber_id", "tenure_months", "plan_type",
"monthly_charge", "contract_type", "payment_method",
"has_internet", "has_tv", "num_lines",
]].copy()
# Encode categorical features
features["is_postpaid"] = (features["contract_type"] != "prepaid").astype(int)
features["is_auto_pay"] = (features["payment_method"].isin(["auto_bank", "auto_card"])).astype(int)
return featuresThe most predictive features in telco churn are behavioral trends, not absolute values. A customer whose data usage dropped 50% in the last 7 days compared to the previous 30 days is far more likely to churn than a customer with consistently low usage. The trend features — ratios of recent to historical behavior — consistently rank highest in feature importance.
The “Silent Churner” Problem
Not all churners give notice. Many simply stop using the service gradually — fewer calls, less data, eventually the SIM goes dormant. These “silent churners” are the hardest to detect because their behavior changes subtly:
def detect_silent_churners(
features: pd.DataFrame,
activity_thresholds: dict = None,
) -> pd.DataFrame:
"""Detect subscribers showing silent churn patterns."""
if activity_thresholds is None:
activity_thresholds = {
"voice_decline_pct": 0.5, # 50% decline in calls
"data_decline_pct": 0.5, # 50% decline in data
"contact_diversity_drop": 0.3, # 30% fewer unique contacts
}
# Calculate decline metrics
features["voice_decline"] = 1 - (
features["voice_call_count_7d"]
/ features["voice_call_count_30d"].clip(lower=0.01)
)
features["data_decline"] = 1 - (
features["data_total_volume_mb_7d"]
/ features["data_total_volume_mb_30d"].clip(lower=0.01)
)
# Flag silent churners
features["silent_churn_signal"] = (
(features["voice_decline"] > activity_thresholds["voice_decline_pct"])
| (features["data_decline"] > activity_thresholds["data_decline_pct"])
)
return featuresSurvival Analysis for Time-to-Churn
Traditional classification models predict whether a customer will churn within a fixed window. Survival analysis predicts when they’ll churn, which is far more actionable:
from lifelines import CoxPHFitter, KaplanMeierFitter
class ChurnSurvivalModel:
"""Survival analysis for customer churn prediction."""
def __init__(self):
self.cox_model = CoxPHFitter()
self.kmf = KaplanMeierFitter()
def fit(self, features: pd.DataFrame, durations: pd.Series, events: pd.Series):
"""
Fit survival model.
Args:
features: Feature matrix
durations: Time until churn or censoring (days)
events: 1 if churned, 0 if still active (censored)
"""
# Kaplan-Meier for overall survival curve
self.kmf.fit(durations, event_observed=events)
self.survival_curve = self.kmf.survival_function_
# Cox proportional hazards for individual predictions
cox_data = features.copy()
cox_data["duration"] = durations
cox_data["event"] = events
self.cox_model.fit(cox_data, duration_col="duration", event_col="event")
def predict_survival_probability(self, features: pd.DataFrame, time_horizon: int = 30) -> pd.Series:
"""Predict probability of surviving past time_horizon days."""
survival_functions = self.cox_model.predict_survival_function(features)
# Get survival probability at time_horizon
time_idx = (survival_functions.index <= time_horizon).sum() - 1
return survival_functions.iloc[time_idx]
def predict_median_survival(self, features: pd.DataFrame) -> pd.Series:
"""Predict median time until churn."""
return self.cox_model.predict_median(features)
def get_hazard_ratios(self) -> pd.DataFrame:
"""Get feature hazard ratios for interpretability."""
summary = self.cox_model.summary
summary["hazard_ratio"] = np.exp(summary["coef"])
summary["hr_lower"] = np.exp(summary["coef lower 95%"])
summary["hr_upper"] = np.exp(summary["coef upper 95%"])
return summary.sort_values("hazard_ratio", ascending=False)Survival analysis gives you the time dimension that classification models lack. Knowing that a customer has a 70% probability of churning within 30 days versus 90 days allows you to prioritize interventions. The hazard ratios from the Cox model are also inherently interpretable — a hazard ratio of 2.0 for a feature means that feature doubles the instantaneous risk of churning.
The Proportional Hazards Assumption
The Cox model assumes that the effect of each feature is constant over time (proportional hazards). In telco churn, this is often violated:
def check_proportional_hazards(model, data, duration_col, event_col):
"""Test proportional hazards assumption."""
# Schoenfeld residuals test
model.check_assumptions(data, p_value_threshold=0.05)
# If violated, consider time-varying coefficients
# Split data into time intervals and fit piecewise model
time_splits = [30, 60, 90, 180]
models = {}
for i, split in enumerate(time_splits[:-1]):
interval_data = data[
(data[duration_col] >= time_splits[i])
& (data[duration_col] < time_splits[i + 1])
]
if len(interval_data) > 100:
models[f"interval_{split}"] = CoxPHFitter()
models[f"interval_{split}"].fit(interval_data, duration_col, event_col)
return modelsContract type is the most common violator of the proportional hazards assumption. Postpaid customers with contracts have near-zero churn risk until the contract expires, then risk spikes. This pattern is fundamentally non-proportional and requires time-varying modeling.
Uplift Modeling: Who to Retain
The critical insight in retention campaigns: not all predicted churners should be targeted. Some customers will churn regardless of intervention. Others will stay regardless. Intervening with these customers wastes money and can even backfire (reminding a customer to cancel a service they’d forgotten about).
Uplift modeling (also called incremental modeling or treatment effect estimation) predicts the causal effect of an intervention:
from causalml.inference.meta import BaseSClassifier, BaseTClassifier
from sklearn.ensemble import GradientBoostingClassifier
class RetentionUpliftModel:
"""Uplift model for retention campaign targeting."""
def __init__(self):
# T-learner: separate models for treatment and control
self.t_learner = BaseTClassifier(
learner=GradientBoostingClassifier(
n_estimators=200,
max_depth=5,
learning_rate=0.05,
)
)
def fit(self, X: pd.DataFrame, treatment: pd.Series, y: pd.Series):
"""
Fit uplift model from historical campaign data.
Args:
X: Features
treatment: 1 if received retention offer, 0 if control
y: 1 if churned, 0 if retained
"""
self.t_learner.fit(X, treatment, y)
def predict_uplift(self, X: pd.DataFrame) -> np.ndarray:
"""Predict uplift: P(retain | treatment) - P(retain | control)."""
return self.t_learner.predict(X)
def segment_customers(self, X: pd.DataFrame) -> pd.DataFrame:
"""Segment customers into treatment strategy groups."""
uplift = self.predict_uplift(X)
churn_prob_control = self.t_learner.models["control"].predict_proba(X)[:, 1]
segments = pd.DataFrame({
"uplift": uplift,
"churn_prob": churn_prob_control,
})
# Four-quadrant segmentation
segments["segment"] = "persuadable" # Default
segments.loc[
(segments["uplift"] < 0.05) & (segments["churn_prob"] > 0.7),
"segment"
] = "lost_cause" # Will churn regardless
segments.loc[
(segments["uplift"] < 0.05) & (segments["churn_prob"] < 0.3),
"segment"
] = "sure_thing" # Will stay regardless
segments.loc[
(segments["uplift"] < -0.02),
"segment"
] = "sleeping_dog" # Intervention makes things worse
return segmentsThe “sleeping dog” segment is the most important to identify. These are customers who are more likely to churn because of the intervention. A retention call reminds them they’re unhappy. A discount offer makes them realize they’re overpaying. Identifying and avoiding these customers is as valuable as identifying persuadable ones.
Retention Campaign Design
Matching Intervention to Customer
Not all retention interventions are equal. The right intervention depends on the churn reason:
def recommend_intervention(
customer_features: dict,
churn_drivers: dict,
intervention_catalog: list[dict],
) -> dict:
"""Recommend the best retention intervention for a customer."""
# Map churn drivers to intervention types
intervention_mapping = {
"price_sensitivity": ["discount", "plan_downgrade", "loyalty_bonus"],
"service_quality": ["network_upgrade", "priority_support", "service_credit"],
"competitor_offer": ["match_offer", "exclusive_features", "contract_bonus"],
"usage_decline": ["usage_bonus", "content_bundle", "family_plan"],
"billing_issues": ["billing_flexibility", "payment_plan", "fee_waiver"],
}
# Get primary churn driver
primary_driver = max(churn_drivers, key=churn_drivers.get)
candidate_types = intervention_mapping.get(primary_driver, ["discount"])
# Score each intervention based on customer profile
scored_interventions = []
for intervention in intervention_catalog:
if intervention["type"] not in candidate_types:
continue
score = calculate_intervention_fit(customer_features, intervention)
scored_interventions.append({
**intervention,
"fit_score": score,
})
scored_interventions.sort(key=lambda x: x["fit_score"], reverse=True)
return {
"recommended": scored_interventions[0] if scored_interventions else None,
"alternatives": scored_interventions[1:3],
"churn_driver": primary_driver,
}Campaign Measurement
Measuring retention campaign effectiveness requires proper experimental design:
def measure_campaign_impact(
treatment_group: pd.DataFrame,
control_group: pd.DataFrame,
follow_up_days: int = 90,
) -> dict:
"""Measure retention campaign impact with proper controls."""
# Churn rates
treatment_churn = treatment_group["churned"].mean()
control_churn = control_group["churned"].mean()
# Lift
absolute_lift = control_churn - treatment_churn
relative_lift = absolute_lift / control_churn if control_churn > 0 else 0
# Statistical significance
from scipy.stats import chi2_contingency
contingency = pd.crosstab(
pd.concat([treatment_group["churned"], control_group["churned"]]),
["treatment"] * len(treatment_group) + ["control"] * len(control_group),
)
chi2, p_value, _, _ = chi2_contingency(contingency)
# ROI
cost_per_contact = 5.00 # USD
revenue_per_retained = treatment_group["monthly_revenue"].mean() * 12 # Annual
campaign_cost = len(treatment_group) * cost_per_contact
revenue_saved = absolute_lift * len(treatment_group) * revenue_per_retained
roi = (revenue_saved - campaign_cost) / campaign_cost
return {
"treatment_churn_rate": treatment_churn,
"control_churn_rate": control_churn,
"absolute_lift": absolute_lift,
"relative_lift": relative_lift,
"p_value": p_value,
"roi": roi,
"customers_retained": int(absolute_lift * len(treatment_group)),
}Always use a holdout control group. Without a control, you can’t distinguish between “customers who would have been retained anyway” and “customers retained because of the campaign.” The control group is the counterfactual that makes causal inference possible.
Production Deployment
Scoring Pipeline
Churn models need to score the entire subscriber base regularly — typically weekly:
def weekly_churn_scoring(
cdr_data: pd.DataFrame,
subscriber_data: pd.DataFrame,
model,
uplift_model,
reference_date: str,
) -> pd.DataFrame:
"""Weekly churn scoring pipeline."""
# Feature engineering
engine = CDRFeatureEngine(reference_date)
features = engine.engineer_features(cdr_data, subscriber_data)
# Predict churn probability
X = features.drop(columns=["subscriber_id"])
features["churn_probability"] = model.predict_proba(X)[:, 1]
# Predict uplift for retention targeting
features["uplift_score"] = uplift_model.predict_uplift(X)
# Segment
features["retention_segment"] = uplift_model.segment_customers(X)["segment"]
# Priority score: high churn prob × high uplift = highest priority
features["priority_score"] = (
features["churn_probability"]
* features["uplift_score"]
* features["monthly_revenue"]
)
# Rank and select for campaign
features = features.sort_values("priority_score", ascending=False)
features["campaign_rank"] = range(1, len(features) + 1)
return featuresMonitoring Churn Model Performance
def monitor_churn_model(
predictions: pd.DataFrame,
actuals: pd.DataFrame,
lookback_months: int = 3,
) -> dict:
"""Monitor churn model performance over time."""
from sklearn.metrics import roc_auc_score, precision_score, recall_score
merged = predictions.merge(actuals, on="subscriber_id")
monthly_metrics = []
for month in merged["prediction_month"].unique():
month_data = merged[merged["prediction_month"] == month]
metrics = {
"month": month,
"auc": roc_auc_score(month_data["actual_churn"], month_data["churn_probability"]),
"precision_at_10pct": precision_at_k(month_data, k=0.10),
"recall_at_10pct": recall_at_k(month_data, k=0.10),
"predicted_churn_rate": month_data["churn_probability"].mean(),
"actual_churn_rate": month_data["actual_churn"].mean(),
}
monthly_metrics.append(metrics)
return pd.DataFrame(monthly_metrics)Lessons from the Field
-
CDR data is massive but sparse. Most subscribers generate a few records per day. Feature engineering must aggregate efficiently, and most features will be zero for most customers.
-
Seasonality matters. Churn patterns follow academic calendars, contract renewal cycles, and competitor promotions. A model trained on Q1 data will underperform in Q4.
-
The uplift model is more valuable than the churn model. Knowing who will churn is interesting. Knowing who can be saved is actionable.
-
Campaign fatigue is real. If you contact the same customer every month with retention offers, they learn to expect it — and may delay renewal to get a better deal.
-
The best retention is preventing the reason to churn. Network quality improvements, transparent billing, and good customer service reduce churn more cost-effectively than any retention campaign.
Churn prediction is one of those problems where the data science is the easy part. The hard part is building the organizational muscle to act on predictions, measure results, and iterate. The models are a means to an end — and the end is keeping customers who are worth keeping.