Code Examples

Ready-to-use code snippets and complete examples

Basic Prediction

import pandas as pd
import xgboost as xgb
import shap

# Load pre-trained model
model = xgb.XGBClassifier()
model.load_model('models/churn_xgboost_v2.json')

# Prepare customer data
customer = pd.DataFrame([{
    'tenure': 12,
    'monthly_charges': 79.50,
    'total_charges': 954.00,
    'contract_encoded': 0,  # month-to-month
    'payment_encoded': 1,   # electronic check
    'internet_encoded': 2,  # fiber optic
    'support_tickets': 3
}])

# Make prediction
churn_prob = model.predict_proba(customer)[0, 1]
print(f"Churn Probability: {churn_prob:.2%}")

# Get SHAP explanation
explainer = shap.TreeExplainer(model)
shap_values = explainer.shap_values(customer)
print(f"Top factor: {customer.columns[abs(shap_values[0]).argmax()]}")

Batch Processing

import pandas as pd
from churn_predictor import ChurnPredictor

# Initialize predictor
predictor = ChurnPredictor(model_path='models/churn_xgboost_v2.json')

# Load customer dataset
customers = pd.read_csv('data/customers.csv')

# Batch predict with parallel processing
results = predictor.predict_batch(
    customers,
    include_explanations=True,
    n_jobs=4
)

# Filter high-risk customers
high_risk = results[results['churn_probability'] > 0.7]
print(f"High risk customers: {len(high_risk)}")

# Export for retention campaign
high_risk.to_csv('output/high_risk_customers.csv', index=False)

Complete Training Pipeline

import pandas as pd
import xgboost as xgb
from sklearn.model_selection import train_test_split
from sklearn.metrics import roc_auc_score, classification_report
import optuna

# Load and prepare data
df = pd.read_csv('data/telco_churn.csv')
X = df.drop('Churn', axis=1)
y = df['Churn'].map({'Yes': 1, 'No': 0})

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, stratify=y, random_state=42
)

# Optuna hyperparameter optimization
def objective(trial):
    params = {
        'n_estimators': trial.suggest_int('n_estimators', 100, 500),
        'max_depth': trial.suggest_int('max_depth', 3, 10),
        'learning_rate': trial.suggest_float('learning_rate', 0.01, 0.3),
        'scale_pos_weight': trial.suggest_float('scale_pos_weight', 1, 10),
        'subsample': trial.suggest_float('subsample', 0.5, 1.0),
        'colsample_bytree': trial.suggest_float('colsample_bytree', 0.5, 1.0),
    }

    model = xgb.XGBClassifier(**params, random_state=42)
    model.fit(X_train, y_train, eval_set=[(X_test, y_test)],
              early_stopping_rounds=50, verbose=False)

    return roc_auc_score(y_test, model.predict_proba(X_test)[:, 1])

study = optuna.create_study(direction='maximize')
study.optimize(objective, n_trials=100)

# Train final model
best_model = xgb.XGBClassifier(**study.best_params, random_state=42)
best_model.fit(X_train, y_train)

# Evaluate
y_pred = best_model.predict(X_test)
y_prob = best_model.predict_proba(X_test)[:, 1]
print(f"AUC-ROC: {roc_auc_score(y_test, y_prob):.4f}")
print(classification_report(y_test, y_pred))

# Save model
best_model.save_model('models/churn_xgboost_optimized.json')

SHAP Analysis

import shap
import matplotlib.pyplot as plt

# Create SHAP explainer
explainer = shap.TreeExplainer(model)
shap_values = explainer.shap_values(X_test)

# Global feature importance
shap.summary_plot(shap_values, X_test, show=False)
plt.savefig('output/shap_summary.png', dpi=150, bbox_inches='tight')

# Single customer explanation
customer_idx = 0
shap.waterfall_plot(
    shap.Explanation(
        values=shap_values[customer_idx],
        base_values=explainer.expected_value,
        data=X_test.iloc[customer_idx],
        feature_names=X_test.columns.tolist()
    )
)
plt.savefig('output/shap_waterfall.png', dpi=150, bbox_inches='tight')

# Dependence plot for key feature
shap.dependence_plot('tenure', shap_values, X_test)
plt.savefig('output/shap_dependence_tenure.png', dpi=150, bbox_inches='tight')

API Integration

import requests

API_URL = "https://api.churnpredict.io/v1"
API_KEY = "your-api-key"

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

# Single prediction
customer_data = {
    "customer_id": "CUST-12345",
    "tenure": 12,
    "monthly_charges": 79.50,
    "contract_type": "month-to-month",
    "payment_method": "electronic_check",
    "support_tickets": 3
}

response = requests.post(
    f"{API_URL}/predict",
    headers=headers,
    json=customer_data
)

result = response.json()
print(f"Customer: {result['customer_id']}")
print(f"Churn Probability: {result['churn_probability']:.2%}")
print(f"Risk Level: {result['risk_level']}")
print(f"Recommended Actions: {result['recommended_actions']}")

Retention Campaign Automation

from churn_predictor import ChurnPredictor
from retention_engine import RetentionEngine

# Initialize
predictor = ChurnPredictor('models/churn_xgboost_v2.json')
retention = RetentionEngine(config='config/retention_rules.yaml')

# Score all customers
customers = load_customers()
predictions = predictor.predict_batch(customers, include_explanations=True)

# Segment by risk
segments = {
    'critical': predictions[predictions['churn_probability'] > 0.8],
    'high': predictions[(predictions['churn_probability'] > 0.6) &
                        (predictions['churn_probability'] <= 0.8)],
    'medium': predictions[(predictions['churn_probability'] > 0.4) &
                          (predictions['churn_probability'] <= 0.6)]
}

# Generate personalized actions
for segment_name, segment_df in segments.items():
    for _, customer in segment_df.iterrows():
        actions = retention.get_recommended_actions(
            customer_id=customer['customer_id'],
            risk_factors=customer['top_risk_factors'],
            customer_value=customer['ltv']
        )

        # Execute highest ROI action
        best_action = max(actions, key=lambda x: x['expected_roi'])
        retention.execute_action(
            customer_id=customer['customer_id'],
            action=best_action
        )

print(f"Retention campaign launched for {len(predictions)} customers")