Chapter 4

Real-time Health Analytics

Machine learning for vital sign analysis, anomaly detection algorithms, predictive health scoring, and AI-powered clinical decision support systems that transform raw data into actionable insights.

From Data to Intelligence

The true value of remote patient monitoring lies not in data collection itself but in the intelligence derived from that data. Raw vital sign measurements become clinically meaningful when analytics transform them into insights: identifying concerning trends, predicting deterioration, and supporting clinical decisions. Modern RPM platforms leverage machine learning and artificial intelligence to deliver analytics capabilities that exceed what human reviewers could achieve manually.

The analytics challenge in RPM is substantial. With continuous monitoring, even a modest patient panel generates millions of data points annually. Manual review is impossible; intelligent automation is essential. The WIA-RPM standard specifies analytics requirements ensuring platforms deliver reliable, explainable, and clinically validated insights.

Analytics Architecture

Baseline Establishment

Effective anomaly detection requires understanding what is normal for each individual patient. Population norms provide starting points, but personalized baselines account for individual variation and enable detection of meaningful deviations. The WIA standard specifies baseline establishment methodology ensuring appropriate personalization while avoiding overfitting to noise.

Personalized Baseline Calculation

Personalized baselines adapt to each patient's typical values, enabling detection of changes that would be missed by population-based thresholds. A blood pressure of 150/90 might be alarming for one patient but normal for another with long-standing hypertension.

// WIA Standard: Baseline Calculation Algorithm
class PatientBaseline:
    def __init__(self, patient_id, parameter):
        self.patient_id = patient_id
        self.parameter = parameter
        self.baseline_window_days = 14
        self.minimum_readings = 10
        self.outlier_threshold_std = 2.5

    def calculate_baseline(self, observations):
        """
        Calculate personalized baseline from historical observations.
        Uses robust statistics to handle outliers.
        """
        if len(observations) < self.minimum_readings:
            return self.get_population_baseline()

        # Filter to baseline window
        recent = self.filter_by_date(observations, self.baseline_window_days)

        # Remove outliers using IQR method
        values = [obs.value for obs in recent]
        q1, q3 = np.percentile(values, [25, 75])
        iqr = q3 - q1
        filtered = [v for v in values if q1 - 1.5*iqr <= v <= q3 + 1.5*iqr]

        # Calculate robust statistics
        return {
            'mean': np.mean(filtered),
            'median': np.median(filtered),
            'std': np.std(filtered),
            'percentile_10': np.percentile(filtered, 10),
            'percentile_90': np.percentile(filtered, 90),
            'sample_size': len(filtered),
            'calculated_date': datetime.now()
        }

    def detect_anomaly(self, value, baseline):
        """
        Detect if new value represents significant deviation.
        """
        z_score = (value - baseline['mean']) / baseline['std']
        return {
            'is_anomaly': abs(z_score) > self.outlier_threshold_std,
            'z_score': z_score,
            'direction': 'high' if z_score > 0 else 'low',
            'severity': self.calculate_severity(z_score)
        }

Baseline Update Strategies

Anomaly Detection Algorithms

Anomaly detection identifies observations that deviate significantly from expected patterns. RPM systems employ multiple detection approaches, from simple threshold breaches to sophisticated machine learning models that recognize complex abnormality patterns.

Threshold-Based Detection

The simplest anomaly detection uses fixed thresholds that trigger alerts when values exceed predefined limits. While limited in sophistication, threshold-based detection remains important for identifying clear danger signals requiring immediate attention.

{
  "vital_sign": "systolic_blood_pressure",
  "thresholds": {
    "critical_low": 80,
    "warning_low": 90,
    "warning_high": 160,
    "critical_high": 180
  },
  "units": "mmHg",
  "alert_levels": {
    "critical": "immediate_notification",
    "warning": "queue_for_review"
  }
}

Statistical Anomaly Detection

Statistical methods detect anomalies based on deviation from historical patterns. These approaches account for individual variation and temporal patterns, reducing false positives compared to static thresholds.

Machine Learning Anomaly Detection

Machine learning models can detect complex anomaly patterns that statistical methods miss. These models learn normal patterns from historical data and flag observations that don't fit learned patterns.

// WIA Standard: Isolation Forest Anomaly Detection
from sklearn.ensemble import IsolationForest
import numpy as np

class VitalSignAnomalyDetector:
    def __init__(self, contamination=0.05):
        self.model = IsolationForest(
            n_estimators=100,
            contamination=contamination,
            random_state=42
        )
        self.is_fitted = False

    def prepare_features(self, observations):
        """
        Extract features for anomaly detection.
        Includes value, time of day, day of week, rate of change.
        """
        features = []
        for i, obs in enumerate(observations):
            feature_vector = [
                obs.value,
                obs.timestamp.hour,
                obs.timestamp.weekday(),
                self.calculate_rate_of_change(observations, i),
                self.calculate_recent_variance(observations, i)
            ]
            features.append(feature_vector)
        return np.array(features)

    def fit(self, historical_observations):
        """Train model on historical normal data."""
        features = self.prepare_features(historical_observations)
        self.model.fit(features)
        self.is_fitted = True

    def predict(self, new_observation, recent_context):
        """
        Predict if new observation is anomalous.
        Returns anomaly score and binary classification.
        """
        if not self.is_fitted:
            raise ValueError("Model must be fitted before prediction")

        context_with_new = recent_context + [new_observation]
        features = self.prepare_features(context_with_new)

        # Get anomaly score (-1 for anomaly, 1 for normal)
        prediction = self.model.predict(features[-1:])
        score = self.model.score_samples(features[-1:])

        return {
            'is_anomaly': prediction[0] == -1,
            'anomaly_score': -score[0],  # Higher = more anomalous
            'confidence': self.calculate_confidence(score[0])
        }

Trend Analysis and Prediction

Beyond detecting point anomalies, sophisticated analytics identify concerning trends before values breach thresholds. Trend analysis enables proactive intervention by predicting deterioration trajectory and time to critical thresholds.

Trend Detection Methods

Deterioration Prediction

Predictive models estimate probability of adverse events—hospitalization, emergency visit, or clinical deterioration—within defined time windows. These predictions enable prioritized attention to highest-risk patients and proactive outreach before events occur.

// WIA Standard: Deterioration Risk Model
class DeteriorationPredictor:
    def __init__(self, prediction_window_hours=72):
        self.prediction_window = prediction_window_hours
        self.model = self.load_pretrained_model()

    def calculate_risk_score(self, patient_data):
        """
        Calculate probability of deterioration in prediction window.
        Combines vital sign trends with patient characteristics.
        """
        features = {
            # Vital sign features
            'bp_systolic_mean_24h': self.aggregate_mean(patient_data['bp_systolic'], 24),
            'bp_systolic_trend_48h': self.calculate_trend(patient_data['bp_systolic'], 48),
            'bp_systolic_variability': self.calculate_cv(patient_data['bp_systolic'], 24),

            'hr_mean_24h': self.aggregate_mean(patient_data['heart_rate'], 24),
            'hr_trend_48h': self.calculate_trend(patient_data['heart_rate'], 48),
            'hr_variability': self.calculate_cv(patient_data['heart_rate'], 24),

            'weight_change_7d': self.calculate_change(patient_data['weight'], 168),
            'spo2_min_24h': self.aggregate_min(patient_data['spo2'], 24),

            # Activity features
            'activity_trend_7d': self.calculate_trend(patient_data['steps'], 168),
            'sleep_quality_trend': self.calculate_trend(patient_data['sleep_score'], 168),

            # Engagement features
            'measurement_compliance': self.calculate_compliance(patient_data),
            'days_since_last_reading': self.days_since_last(patient_data),

            # Patient characteristics
            'age': patient_data['demographics']['age'],
            'comorbidity_count': len(patient_data['conditions']),
            'recent_hospitalization': self.check_recent_hospitalization(patient_data, 30)
        }

        # Model inference
        risk_probability = self.model.predict_proba(features)

        return {
            'risk_score': risk_probability,
            'risk_level': self.categorize_risk(risk_probability),
            'contributing_factors': self.explain_prediction(features),
            'recommended_actions': self.get_recommendations(risk_probability)
        }

    def categorize_risk(self, probability):
        if probability >= 0.7:
            return 'HIGH'
        elif probability >= 0.4:
            return 'MODERATE'
        elif probability >= 0.2:
            return 'ELEVATED'
        else:
            return 'LOW'

Health Scoring Systems

Composite health scores aggregate multiple vital signs and metrics into single summary values that facilitate rapid patient assessment and population-level triage. Well-designed scores balance simplicity with predictive power.

Composite Score Design

// WIA Standard: Composite Health Score Calculation
class HealthScoreCalculator:
    def __init__(self):
        self.component_weights = {
            'vital_signs': 0.40,
            'trends': 0.25,
            'engagement': 0.15,
            'symptoms': 0.20
        }

    def calculate_score(self, patient_data):
        """
        Calculate composite health score (0-100).
        Higher scores indicate better health status.
        """
        components = {}

        # Vital signs component
        components['vital_signs'] = self.score_vital_signs({
            'bp_systolic': patient_data['latest_bp_systolic'],
            'bp_diastolic': patient_data['latest_bp_diastolic'],
            'heart_rate': patient_data['latest_hr'],
            'spo2': patient_data['latest_spo2'],
            'weight_vs_baseline': patient_data['weight_deviation']
        })

        # Trends component
        components['trends'] = self.score_trends({
            'bp_trend': patient_data['bp_trend_7d'],
            'weight_trend': patient_data['weight_trend_7d'],
            'activity_trend': patient_data['activity_trend_7d']
        })

        # Engagement component
        components['engagement'] = self.score_engagement({
            'measurement_compliance': patient_data['compliance_rate'],
            'app_usage': patient_data['app_opens_7d'],
            'message_response': patient_data['message_response_rate']
        })

        # Symptoms component
        components['symptoms'] = self.score_symptoms(
            patient_data['reported_symptoms']
        )

        # Weighted combination
        total_score = sum(
            components[c] * self.component_weights[c]
            for c in components
        )

        return {
            'total_score': round(total_score, 1),
            'components': components,
            'percentile': self.calculate_percentile(total_score),
            'change_from_last_week': self.calculate_change(patient_data)
        }

    def score_vital_signs(self, vitals):
        """Score vital signs based on proximity to optimal ranges."""
        scores = []

        # Blood pressure scoring
        bp_score = self.gaussian_score(
            vitals['bp_systolic'],
            optimal=115,
            acceptable_range=20
        )
        scores.append(bp_score)

        # Heart rate scoring
        hr_score = self.gaussian_score(
            vitals['heart_rate'],
            optimal=70,
            acceptable_range=15
        )
        scores.append(hr_score)

        # SpO2 scoring (asymmetric - only penalize low)
        spo2_score = min(100, vitals['spo2'] * 1.05 - 5)
        scores.append(max(0, spo2_score))

        # Weight deviation scoring
        weight_score = 100 - abs(vitals['weight_vs_baseline']) * 10
        scores.append(max(0, weight_score))

        return np.mean(scores)

Clinical Decision Support

Analytics outputs feed clinical decision support (CDS) systems that present actionable recommendations to care teams. Effective CDS augments clinical judgment without replacing it, providing relevant information at the right time in formats that support decision-making.

CDS Integration Points

Explainable AI Requirements

Clinical adoption of AI-powered analytics requires explainability. Clinicians must understand why the system generated specific alerts or recommendations to appropriately incorporate system outputs into clinical decision-making.

// WIA Standard: Explainable Alert Generation
class ExplainableAlert:
    def generate_explanation(self, alert, model_output):
        """
        Generate human-readable explanation for alert.
        """
        explanation = {
            'summary': self.generate_summary(alert),
            'primary_factors': self.get_top_factors(model_output, n=3),
            'historical_context': self.get_context(alert.patient_id),
            'similar_cases': self.find_similar_outcomes(alert),
            'confidence_level': model_output['confidence'],
            'limitations': self.get_model_limitations()
        }

        return explanation

    def generate_summary(self, alert):
        """Generate plain-language alert summary."""
        templates = {
            'bp_high': "Blood pressure elevated to {value} mmHg, "
                      "{deviation} above patient's typical baseline of {baseline} mmHg. "
                      "This represents a {percentile}th percentile reading for this patient.",
            'weight_gain': "Weight increased by {change} lbs over {days} days, "
                          "suggesting possible fluid retention. "
                          "Current weight {value} lbs vs baseline {baseline} lbs.",
            'deterioration_risk': "72-hour deterioration risk elevated to {risk_level} ({probability}%). "
                                 "Key factors: {factors}."
        }

        return templates[alert.type].format(**alert.data)

    def get_top_factors(self, model_output, n=3):
        """Extract most influential factors in prediction."""
        # SHAP values or feature importance
        factors = model_output['feature_importance']
        sorted_factors = sorted(factors.items(), key=lambda x: abs(x[1]), reverse=True)

        return [
            {
                'factor': f[0],
                'impact': f[1],
                'direction': 'increasing' if f[1] > 0 else 'decreasing',
                'description': self.describe_factor(f[0], f[1])
            }
            for f in sorted_factors[:n]
        ]

Real-time Processing Architecture

RPM analytics must process incoming data with minimal latency to enable timely alerts. Stream processing architectures handle continuous data flows while maintaining state for trend analysis and personalized baselines.

Stream Processing Components

Model Governance and Validation

Healthcare AI models require rigorous governance ensuring safety, effectiveness, and fairness. The WIA standard specifies model validation requirements and ongoing monitoring obligations.

Validation Requirements

Performance Monitoring

// WIA Standard: Model Performance Monitoring
class ModelMonitor:
    def __init__(self, model_id, alert_thresholds):
        self.model_id = model_id
        self.thresholds = alert_thresholds

    def track_performance(self, predictions, outcomes):
        """
        Calculate and log model performance metrics.
        """
        metrics = {
            'auc_roc': roc_auc_score(outcomes, predictions),
            'precision': precision_score(outcomes, predictions > 0.5),
            'recall': recall_score(outcomes, predictions > 0.5),
            'calibration_error': self.calculate_calibration_error(predictions, outcomes),
            'alert_rate': np.mean(predictions > self.alert_threshold),
            'ppv': self.calculate_ppv(predictions, outcomes)
        }

        # Check for performance degradation
        alerts = []
        for metric, value in metrics.items():
            if metric in self.thresholds:
                if value < self.thresholds[metric]['min']:
                    alerts.append({
                        'metric': metric,
                        'value': value,
                        'threshold': self.thresholds[metric]['min'],
                        'severity': 'warning'
                    })

        # Log metrics
        self.log_metrics(metrics)

        return {
            'metrics': metrics,
            'alerts': alerts,
            'timestamp': datetime.now()
        }

    def detect_drift(self, current_window, reference_window):
        """
        Detect data drift that may impact model performance.
        """
        drift_results = {}

        for feature in self.monitored_features:
            ks_stat, p_value = ks_2samp(
                reference_window[feature],
                current_window[feature]
            )
            drift_results[feature] = {
                'ks_statistic': ks_stat,
                'p_value': p_value,
                'drift_detected': p_value < 0.01
            }

        return drift_results

Summary

Real-time health analytics transforms raw RPM data into actionable clinical intelligence. Through personalized baselines, sophisticated anomaly detection, trend prediction, and clinical decision support, analytics enable proactive care that intervenes before adverse events occur.

The WIA-RPM standard establishes requirements for analytics capabilities, model validation, and governance ensuring systems deliver reliable, explainable, and fair predictions. As AI technology advances, analytics capabilities will continue expanding while adherence to validation and governance standards ensures patient safety and clinical trust.