Health Score Calculator

def calculate_health_score(customer_data):
    # Normalize metrics to 0-100 scale
    usage_score = normalize_usage_metrics(customer_data['usage'])
    engagement_score = normalize_engagement_metrics(customer_data['engagement'])
    support_score = normalize_support_metrics(customer_data['support'])
    outcome_score = normalize_outcome_metrics(customer_data['outcomes'])

    # Apply weights (should sum to 1.0)
    weights = {
        'usage': 0.35,
        'engagement': 0.25,
        'support': 0.20,
        'outcomes': 0.20
    }

    health_score = (
        usage_score * weights['usage'] +
        engagement_score * weights['engagement'] +
        support_score * weights['support'] +
        outcome_score * weights['outcomes']
    )

    return min(100, max(0, health_score))

def normalize_usage_metrics(usage_data):
    # Example: Daily Active Usage scoring
    days_active = usage_data.get('days_active_last_30', 0)
    feature_adoption = usage_data.get('core_features_used', 0) / usage_data.get('total_features', 1)
    session_depth = min(usage_data.get('avg_session_duration', 0) / 1800, 1)  # Cap at 30 min

    usage_score = (
        (days_active / 30) * 40 +  # 40 points for daily usage
        feature_adoption * 35 +    # 35 points for feature adoption
        session_depth * 25         # 25 points for session depth
    )

    return usage_score * 100

import numpy as np
from datetime import datetime, timedelta

def calculate_time_weighted_score(metrics_history, decay_factor=0.1):
    """
    Apply exponential decay to historical metrics
    Recent data weighted more heavily than older data
    """
    current_date = datetime.now()
    weighted_scores = []

    for entry in metrics_history:
        days_ago = (current_date - entry['date']).days
        weight = np.exp(-decay_factor * days_ago)
        weighted_score = entry['score'] * weight
        weighted_scores.append(weighted_score)

    return np.mean(weighted_scores) if weighted_scores else 0

def calculate_trend_momentum(scores_30_days):
    """
    Calculate momentum factor based on recent trends
    """
    if len(scores_30_days) < 7:
        return 1.0  # Neutral if insufficient data

    recent_avg = np.mean(scores_30_days[-7:])
    prior_avg = np.mean(scores_30_days[-14:-7])

    if prior_avg == 0:
        return 1.0

    momentum = recent_avg / prior_avg
    return min(1.5, max(0.5, momentum))  # Cap momentum effect

-- Example SQL for calculating adoption metrics
WITH user_adoption AS (
    SELECT 
        customer_id,
        COUNT(DISTINCT feature_name) as features_used,
        COUNT(DISTINCT DATE(event_timestamp)) as active_days_30,
        AVG(session_duration_minutes) as avg_session_duration,
        SUM(CASE WHEN feature_name IN ('core_feature_1', 'core_feature_2') 
            THEN 1 ELSE 0 END) as core_feature_usage
    FROM product_events 
    WHERE event_timestamp >= CURRENT_DATE - INTERVAL '30 days'
    GROUP BY customer_id
),
adoption_scores AS (
    SELECT 
        customer_id,
        -- Normalize to 0-100 scale using percentiles
        PERCENT_RANK() OVER (ORDER BY features_used) * 100 as feature_breadth_score,
        PERCENT_RANK() OVER (ORDER BY active_days_30) * 100 as engagement_frequency_score,
        PERCENT_RANK() OVER (ORDER BY avg_session_duration) * 100 as engagement_depth_score,
        CASE WHEN core_feature_usage > 0 THEN 100 ELSE 0 END as core_adoption_score
    FROM user_adoption
)
SELECT * FROM adoption_scores;

def calculate_support_health_score(support_data):
    # Support ticket analysis
    ticket_volume = support_data.get('tickets_30_days', 0)
    avg_resolution_time = support_data.get('avg_resolution_hours', 24)
    satisfaction_score = support_data.get('csat_average', 3)
    escalation_rate = support_data.get('escalation_rate', 0)

    # Volume score (fewer tickets = better, but account for size)
    volume_score = max(0, 100 - (ticket_volume * 10))

    # Resolution time score
    resolution_score = max(0, 100 - (avg_resolution_time / 24 * 20))

    # Satisfaction score (1-5 scale converted to 0-100)
    csat_score = ((satisfaction_score - 1) / 4) * 100

    # Escalation penalty
    escalation_penalty = escalation_rate * 50

    support_score = (
        volume_score * 0.3 +
        resolution_score * 0.2 +
        csat_score * 0.4 +
        max(0, 100 - escalation_penalty) * 0.1
    )

    return support_score

def categorize_health_score(score, account_value=None):
    """
    Segment customers based on health score and account value
    """
    if score >= 80:
        category = "Healthy"
        risk_level = "Low"
    elif score >= 60:
        category = "At Risk"
        risk_level = "Medium" 
    elif score >= 40:
        category = "High Risk"
        risk_level = "High"
    else:
        category = "Critical"
        risk_level = "Critical"

    # Adjust priority based on account value
    priority = "Standard"
    if account_value and account_value > 50000:
        if risk_level in ["High", "Critical"]:
            priority = "Enterprise Alert"
        elif risk_level == "Medium":
            priority = "Enterprise Watch"

    return {
        'category': category,
        'risk_level': risk_level,
        'priority': priority,
        'recommended_action': get_recommended_action(category, priority)
    }

def get_recommended_action(category, priority):
    actions = {
        "Healthy": "Monitor for expansion opportunities",
        "At Risk": "Schedule check-in call within 5 days",
        "High Risk": "Immediate CSM outreach required",
        "Critical": "Executive escalation and retention plan"
    }
    return actions.get(category, "Review account status")

def validate_health_score_accuracy(predictions, actual_outcomes, time_period=90):
    """
    Measure health score predictive accuracy
    """
    from sklearn.metrics import accuracy_score, precision_recall_curve

    # Convert health scores to risk categories for validation
    predicted_risk = [1 if score < 60 else 0 for score in predictions]

    accuracy = accuracy_score(actual_outcomes, predicted_risk)

    # Calculate precision/recall for different thresholds
    precision, recall, thresholds = precision_recall_curve(
        actual_outcomes, 
        [100-score for score in predictions]  # Invert for risk scoring
    )

    return {
        'accuracy': accuracy,
        'precision_recall_data': list(zip(precision, recall, thresholds)),
        'optimal_threshold': thresholds[np.argmax(precision * recall)]
    }