How to Build Clinical-Grade Wearable Data Pipelines with Real-Time Analytics

Introduction: The Clinical-Grade Data Pipeline Challenge

Consumer wearable devices generate continuous streams of physiological data, but transforming this raw data into clinically actionable insights requires sophisticated data engineering. A clinical-grade wearable data pipeline must handle real-time streaming, perform data quality validation, detect anomalies, run predictive algorithms, generate clinical alerts, and maintain comprehensive audit trails—all while meeting FDA software validation requirements and HIPAA security standards.

This guide walks you through building a production-ready wearable data pipeline capable of processing millions of data points daily with sub-second latency for critical alerts.

Why Build a Real-Time Wearable Pipeline?

Traditional batch processing (syncing data once daily) is insufficient for clinical applications that require immediate intervention:

Use Cases Requiring Real-Time Processing:

Sepsis detection: Predictive algorithms analyzing vital sign trends need 15-minute windows
Cardiac arrest prediction: Early warning scores require continuous heart rate and respiratory rate streaming
Post-surgical monitoring: Immediate alerts for hemorrhage, infection, or respiratory compromise
Chronic disease management: Real-time feedback loops for diabetes, heart failure, hypertension
Clinical trials: Continuous safety monitoring with automated adverse event detection

However, building such a system traditionally requires 9-18 months and a specialized team. JustCopy.ai reduces this to days. With AI-powered code generation, you can clone proven real-time analytics pipelines and customize them to your clinical algorithms.

System Architecture Overview

High-Level Pipeline Architecture

┌──────────────────────────────────────────────────────────────┐
│                      Data Sources                            │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐    │
│  │ Fitbit   │  │  Apple   │  │  Google  │  │  Medical │    │
│  │   API    │  │HealthKit │  │   Fit    │  │  Devices │    │
│  └─────┬────┘  └─────┬────┘  └─────┬────┘  └─────┬────┘    │
└────────┼─────────────┼─────────────┼─────────────┼──────────┘
         │             │             │             │
         └─────────────┴─────────────┴─────────────┘
                         │
         ┌───────────────▼────────────────┐
         │   API Gateway / Load Balancer  │
         │   - Rate limiting              │
         │   - Authentication             │
         │   - Request routing            │
         └───────────────┬────────────────┘
                         │
         ┌───────────────▼────────────────┐
         │   Data Ingestion Service       │
         │   - Validation                 │
         │   - Normalization              │
         │   - Deduplication              │
         └───────────────┬────────────────┘
                         │
         ┌───────────────▼────────────────┐
         │   Message Queue (Kafka/RabbitMQ)│
         │   - Buffering                  │
         │   - Guaranteed delivery        │
         │   - Topic partitioning         │
         └───────────────┬────────────────┘
                         │
         ┌───────────────┴────────────────┐
         │                                │
         ▼                                ▼
┌────────────────┐              ┌──────────────────┐
│ Stream         │              │  Batch           │
│ Processing     │              │  Processing      │
│ (Real-time)    │              │  (Historical)    │
│ - Kafka Streams│              │  - Apache Spark  │
│ - Flink        │              │  - Airflow       │
└────────┬───────┘              └────────┬─────────┘
         │                                │
         ▼                                ▼
┌────────────────┐              ┌──────────────────┐
│ Time-Series DB │              │  Data Warehouse  │
│ (InfluxDB,     │              │  (PostgreSQL,    │
│  TimescaleDB)  │              │   BigQuery)      │
└────────┬───────┘              └────────┬─────────┘
         │                                │
         └────────────────┬───────────────┘
                          │
         ┌────────────────▼────────────────┐
         │   Analytics & ML Engine         │
         │   - Anomaly detection           │
         │   - Predictive models           │
         │   - Risk scoring                │
         └────────────────┬────────────────┘
                          │
         ┌────────────────▼────────────────┐
         │   Clinical Alerting System      │
         │   - Rule engine                 │
         │   - Alert routing               │
         │   - Escalation logic            │
         └────────────────┬────────────────┘
                          │
         ┌────────────────▼────────────────┐
         │   EHR Integration (FHIR)        │
         │   - Epic, Cerner, Custom        │
         └─────────────────────────────────┘

JustCopy.ai provides complete implementations of this entire architecture with all components integrated, tested, and optimized for healthcare workloads.

Step 1: Set Up Streaming Infrastructure

Technology Stack Selection

Message Queue / Event Streaming:

Apache Kafka: Industry standard, handles millions of events/sec, fault-tolerant
RabbitMQ: Simpler setup, good for moderate throughput
AWS Kinesis: Managed service, seamless AWS integration
Google Pub/Sub: Managed, auto-scaling, GCP integration

Stream Processing:

Apache Flink: Complex event processing, exactly-once semantics
Kafka Streams: Lightweight, built on Kafka, stateful processing
Apache Spark Streaming: Batch + streaming, MLlib integration
AWS Lambda + Kinesis: Serverless, auto-scaling

Time-Series Database:

InfluxDB: Purpose-built for time-series, excellent compression
TimescaleDB: PostgreSQL extension, SQL queries on time-series
Prometheus: Monitoring-focused, good for metrics
AWS Timestream: Managed, serverless time-series DB

Apache Kafka Setup

# Docker Compose for Kafka development environment
version: '3.8'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    ports:
      - "8080:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092

  influxdb:
    image: influxdb:2.7
    ports:
      - "8086:8086"
    volumes:
      - influxdb-data:/var/lib/influxdb2
    environment:
      DOCKER_INFLUXDB_INIT_MODE: setup
      DOCKER_INFLUXDB_INIT_USERNAME: admin
      DOCKER_INFLUXDB_INIT_PASSWORD: adminpassword
      DOCKER_INFLUXDB_INIT_ORG: healthcare-org
      DOCKER_INFLUXDB_INIT_BUCKET: wearable-data

volumes:
  influxdb-data:

Or use JustCopy.ai: Skip infrastructure setup entirely. JustCopy.ai’s AI agents automatically provision Kafka clusters, time-series databases, and stream processing infrastructure on AWS, GCP, or Azure with production-grade configurations.

Step 2: Build Data Ingestion Service

The ingestion service receives wearable data from multiple sources, validates it, and publishes to Kafka topics.

Data Ingestion API

// TypeScript: High-performance data ingestion service
import express from 'express';
import { Kafka, Producer } from 'kafkajs';
import { z } from 'zod'; // Runtime validation library

// Kafka producer setup
const kafka = new Kafka({
  clientId: 'wearable-ingestion-service',
  brokers: ['kafka:9092']
});

const producer: Producer = kafka.producer({
  idempotent: true, // Prevent duplicate messages
  maxInFlightRequests: 5,
  transactionalId: 'wearable-ingestion-txn'
});

// Data validation schemas
const HeartRateDataSchema = z.object({
  patient_id: z.string().uuid(),
  device_id: z.string(),
  device_type: z.enum(['apple_watch', 'fitbit', 'google_fit', 'garmin']),
  timestamp: z.string().datetime(),
  heart_rate: z.number().min(30).max(250),
  measurement_context: z.enum(['resting', 'active', 'exercise', 'sleep']).optional(),
  confidence_level: z.number().min(0).max(1).optional()
});

const ActivityDataSchema = z.object({
  patient_id: z.string().uuid(),
  device_id: z.string(),
  timestamp: z.string().datetime(),
  step_count: z.number().int().min(0).max(100000),
  distance_meters: z.number().min(0),
  active_minutes: z.number().int().min(0),
  calories_burned: z.number().min(0)
});

// Express API
const app = express();
app.use(express.json());

/**
 * Ingest heart rate data
 */
app.post('/api/v1/ingest/heart-rate', async (req, res) => {
  try {
    // Validate request body
    const data = HeartRateDataSchema.parse(req.body);

    // Additional business logic validation
    const validation = await validateHeartRateData(data);
    if (!validation.valid) {
      return res.status(400).json({
        error: 'Validation failed',
        details: validation.errors
      });
    }

    // Check for duplicates
    const isDuplicate = await checkDuplicate(
      data.patient_id,
      'heart_rate',
      data.timestamp,
      data.heart_rate
    );

    if (isDuplicate) {
      return res.status(200).json({
        message: 'Duplicate data point ignored',
        deduplicated: true
      });
    }

    // Enrich data with metadata
    const enrichedData = {
      ...data,
      ingestion_timestamp: new Date().toISOString(),
      data_source: 'api',
      schema_version: '1.0'
    };

    // Publish to Kafka topic
    await producer.send({
      topic: 'wearable.heart-rate',
      messages: [{
        key: data.patient_id, // Partition by patient for ordered processing
        value: JSON.stringify(enrichedData),
        headers: {
          'content-type': 'application/json',
          'patient-id': data.patient_id,
          'device-type': data.device_type
        }
      }]
    });

    res.status(202).json({
      message: 'Data accepted for processing',
      timestamp: enrichedData.ingestion_timestamp
    });

  } catch (error) {
    if (error instanceof z.ZodError) {
      return res.status(400).json({
        error: 'Invalid data format',
        details: error.errors
      });
    }

    console.error('Ingestion error:', error);
    res.status(500).json({ error: 'Internal server error' });
  }
});

/**
 * Batch ingest endpoint for bulk data
 */
app.post('/api/v1/ingest/batch', async (req, res) => {
  const { data_points } = req.body;

  if (!Array.isArray(data_points) || data_points.length === 0) {
    return res.status(400).json({ error: 'data_points must be non-empty array' });
  }

  const results = {
    accepted: 0,
    rejected: 0,
    duplicates: 0,
    errors: []
  };

  // Process in batches for performance
  const messages = [];

  for (const point of data_points) {
    try {
      // Validate based on data type
      const dataType = point.data_type;
      let validated;

      switch (dataType) {
        case 'heart_rate':
          validated = HeartRateDataSchema.parse(point);
          break;
        case 'activity':
          validated = ActivityDataSchema.parse(point);
          break;
        default:
          throw new Error(`Unsupported data type: ${dataType}`);
      }

      // Check duplicates
      const isDuplicate = await checkDuplicate(
        validated.patient_id,
        dataType,
        validated.timestamp,
        validated[dataType === 'heart_rate' ? 'heart_rate' : 'step_count']
      );

      if (isDuplicate) {
        results.duplicates++;
        continue;
      }

      // Add to batch
      messages.push({
        topic: `wearable.${dataType}`,
        messages: [{
          key: validated.patient_id,
          value: JSON.stringify({
            ...validated,
            ingestion_timestamp: new Date().toISOString()
          })
        }]
      });

      results.accepted++;

    } catch (error) {
      results.rejected++;
      results.errors.push({
        data: point,
        error: error.message
      });
    }
  }

  // Send batch to Kafka
  if (messages.length > 0) {
    await producer.sendBatch({
      topicMessages: messages
    });
  }

  res.status(200).json(results);
});

// Start server
async function start() {
  await producer.connect();
  app.listen(3000, () => {
    console.log('Wearable ingestion service listening on port 3000');
  });
}

start();

Data Validation Logic

// Advanced validation with clinical rules
interface ValidationResult {
  valid: boolean;
  errors: string[];
  warnings: string[];
  metadata: {
    clinical_significance?: string;
    alert_required?: boolean;
  };
}

async function validateHeartRateData(data: any): Promise<ValidationResult> {
  const result: ValidationResult = {
    valid: true,
    errors: [],
    warnings: [],
    metadata: {}
  };

  // Get patient context for personalized validation
  const patient = await getPatientContext(data.patient_id);

  // Biologically plausible range
  if (data.heart_rate < 30 || data.heart_rate > 250) {
    result.valid = false;
    result.errors.push(`Heart rate ${data.heart_rate} outside plausible range (30-250)`);
    return result;
  }

  // Context-specific validation
  if (data.measurement_context === 'resting') {
    // Bradycardia
    if (data.heart_rate < 50 && !patient.is_athlete) {
      result.warnings.push('Resting bradycardia detected');
      result.metadata.clinical_significance = 'moderate';
    }

    // Tachycardia
    if (data.heart_rate > 100) {
      result.warnings.push('Resting tachycardia detected');
      result.metadata.clinical_significance = 'moderate';
      result.metadata.alert_required = true;
    }

    // Critical values
    if (data.heart_rate < 40 || data.heart_rate > 130) {
      result.metadata.clinical_significance = 'high';
      result.metadata.alert_required = true;
    }
  }

  // Check for sudden changes from baseline
  const recentHR = await getRecentHeartRate(data.patient_id, '1h');
  if (recentHR) {
    const change = Math.abs(data.heart_rate - recentHR.average);
    if (change > 40) {
      result.warnings.push(`Sudden heart rate change: ${change} bpm in 1 hour`);
      result.metadata.alert_required = true;
    }
  }

  return result;
}

Step 3: Implement Stream Processing

Process real-time data streams to calculate rolling statistics, detect anomalies, and trigger alerts.

Kafka Streams Processing

// Java: Kafka Streams for real-time wearable data processing
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.Stores;
import java.time.Duration;

public class WearableStreamProcessor {

    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();

        // Input stream: Raw heart rate data
        KStream<String, HeartRateData> heartRateStream = builder.stream(
            "wearable.heart-rate",
            Consumed.with(Serdes.String(), heartRateDataSerde())
        );

        // Calculate rolling average over 5-minute windows
        KTable<Windowed<String>, Double> rollingAverage = heartRateStream
            .groupByKey()
            .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
            .aggregate(
                HeartRateAggregate::new,
                (patientId, newData, aggregate) -> {
                    aggregate.addDataPoint(newData.getHeartRate());
                    return aggregate;
                },
                Materialized.with(Serdes.String(), heartRateAggregateSerde())
            )
            .mapValues(aggregate -> aggregate.getAverage());

        // Detect anomalies: Heart rate deviation from rolling average
        heartRateStream
            .join(
                rollingAverage,
                (currentHR, avgHR) -> new AnomalyCheck(currentHR, avgHR),
                JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5))
            )
            .filter((key, anomalyCheck) -> anomalyCheck.isAnomaly())
            .to("wearable.heart-rate.anomalies");

        // Calculate heart rate variability (HRV)
        KStream<String, HeartRateVariability> hrvStream = heartRateStream
            .groupByKey()
            .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
            .aggregate(
                HRVCalculator::new,
                (patientId, newData, calculator) -> {
                    calculator.addRRInterval(60000.0 / newData.getHeartRate());
                    return calculator;
                },
                Materialized.with(Serdes.String(), hrvCalculatorSerde())
            )
            .toStream()
            .mapValues(calculator -> calculator.calculateHRV());

        // Publish HRV metrics
        hrvStream.to("wearable.hrv.metrics");

        // Start streams application
        KafkaStreams streams = new KafkaStreams(builder.build(), getStreamsConfig());
        streams.start();

        // Graceful shutdown
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }

    static class HeartRateAggregate {
        private List<Double> dataPoints = new ArrayList<>();

        void addDataPoint(double hr) {
            dataPoints.add(hr);
        }

        double getAverage() {
            return dataPoints.stream()
                .mapToDouble(Double::doubleValue)
                .average()
                .orElse(0.0);
        }

        double getStdDev() {
            double avg = getAverage();
            double variance = dataPoints.stream()
                .mapToDouble(hr -> Math.pow(hr - avg, 2))
                .average()
                .orElse(0.0);
            return Math.sqrt(variance);
        }
    }

    static class AnomalyCheck {
        private HeartRateData currentData;
        private double rollingAverage;

        AnomalyCheck(HeartRateData currentData, double rollingAverage) {
            this.currentData = currentData;
            this.rollingAverage = rollingAverage;
        }

        boolean isAnomaly() {
            double deviation = Math.abs(currentData.getHeartRate() - rollingAverage);
            // Flag if deviation > 30 bpm from 5-min rolling average
            return deviation > 30;
        }
    }
}

Apache Flink Alternative

# Python: Apache Flink for complex event processing
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import FlinkKafkaConsumer, FlinkKafkaProducer
from pyflink.common.serialization import SimpleStringSchema
from pyflink.datastream.window import TumblingEventTimeWindows
from pyflink.common.time import Time
import json

def process_wearable_stream():
    env = StreamExecutionEnvironment.get_execution_environment()
    env.set_parallelism(4)

    # Kafka consumer for heart rate data
    kafka_consumer = FlinkKafkaConsumer(
        topics='wearable.heart-rate',
        deserialization_schema=SimpleStringSchema(),
        properties={
            'bootstrap.servers': 'kafka:9092',
            'group.id': 'wearable-processor'
        }
    )

    # Read stream
    heart_rate_stream = env.add_source(kafka_consumer)

    # Parse JSON and extract fields
    parsed_stream = heart_rate_stream.map(lambda x: json.loads(x))

    # Detect tachycardia patterns (HR > 100 for 5+ minutes)
    tachycardia_alerts = (
        parsed_stream
        .key_by(lambda x: x['patient_id'])
        .window(TumblingEventTimeWindows.of(Time.minutes(5)))
        .process(TachycardiaDetector())
    )

    # Publish alerts to Kafka
    kafka_producer = FlinkKafkaProducer(
        topic='clinical.alerts.tachycardia',
        serialization_schema=SimpleStringSchema(),
        producer_config={
            'bootstrap.servers': 'kafka:9092'
        }
    )

    tachycardia_alerts.add_sink(kafka_producer)

    env.execute('Wearable Stream Processing')

class TachycardiaDetector(ProcessWindowFunction):
    def process(self, key, context, elements):
        hr_readings = [e['heart_rate'] for e in elements]

        # Check if all readings in window > 100 bpm
        if len(hr_readings) >= 5 and all(hr > 100 for hr in hr_readings):
            yield json.dumps({
                'patient_id': key,
                'alert_type': 'tachycardia',
                'severity': 'moderate',
                'duration_minutes': 5,
                'average_hr': sum(hr_readings) / len(hr_readings),
                'timestamp': context.window().end()
            })

JustCopy.ai’s stream processing templates include pre-built Kafka Streams and Flink applications for common clinical algorithms (sepsis detection, cardiac event prediction, medication adherence monitoring, etc.), ready to deploy and customize.

Step 4: Build Predictive Analytics Engine

Implement machine learning models to predict clinical events before they occur.

Sepsis Prediction Model

# Python: Sepsis prediction using real-time wearable data
import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from influxdb_client import InfluxDBClient
import joblib

class SepsisPredictor:
    def __init__(self, model_path='models/sepsis_rf_model.pkl'):
        self.model = joblib.load(model_path)
        self.influx_client = InfluxDBClient(
            url="http://influxdb:8086",
            token="your-token",
            org="healthcare-org"
        )

    def extract_features(self, patient_id: str) -> dict:
        """Extract features from last 6 hours of wearable data"""
        query = f'''
        from(bucket: "wearable-data")
          |> range(start: -6h)
          |> filter(fn: (r) => r["patient_id"] == "{patient_id}")
          |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
        '''

        # Query InfluxDB for patient data
        result = self.influx_client.query_api().query_data_frame(query)

        if result.empty:
            return None

        # Calculate features
        features = {
            # Heart rate features
            'hr_mean_6h': result['heart_rate'].mean(),
            'hr_std_6h': result['heart_rate'].std(),
            'hr_max_6h': result['heart_rate'].max(),
            'hr_min_6h': result['heart_rate'].min(),
            'hr_trend_6h': self._calculate_trend(result['heart_rate']),

            # Respiratory rate features
            'rr_mean_6h': result['respiratory_rate'].mean(),
            'rr_std_6h': result['respiratory_rate'].std(),

            # Temperature features
            'temp_mean_6h': result['temperature'].mean(),
            'temp_max_6h': result['temperature'].max(),
            'temp_trend_6h': self._calculate_trend(result['temperature']),

            # Activity features
            'activity_mean_6h': result['activity_level'].mean(),
            'activity_std_6h': result['activity_level'].std(),

            # SpO2 features
            'spo2_mean_6h': result['spo2'].mean(),
            'spo2_min_6h': result['spo2'].min(),

            # Heart rate variability
            'hrv_mean_6h': result['hrv'].mean(),
            'hrv_std_6h': result['hrv'].std()
        }

        return features

    def predict_sepsis_risk(self, patient_id: str) -> dict:
        """Predict sepsis risk score (0-100)"""
        features = self.extract_features(patient_id)

        if not features:
            return {
                'patient_id': patient_id,
                'risk_score': None,
                'error': 'Insufficient data'
            }

        # Convert to DataFrame for model input
        feature_df = pd.DataFrame([features])

        # Predict probability
        sepsis_probability = self.model.predict_proba(feature_df)[0][1]
        risk_score = int(sepsis_probability * 100)

        # Determine risk category
        if risk_score < 25:
            category = 'low'
        elif risk_score < 50:
            category = 'moderate'
        elif risk_score < 75:
            category = 'high'
        else:
            category = 'critical'

        return {
            'patient_id': patient_id,
            'risk_score': risk_score,
            'category': category,
            'confidence': sepsis_probability,
            'contributing_factors': self._identify_factors(features),
            'timestamp': pd.Timestamp.now().isoformat()
        }

    def _calculate_trend(self, series: pd.Series) -> float:
        """Calculate linear trend coefficient"""
        if len(series) < 2:
            return 0.0
        x = np.arange(len(series))
        coeffs = np.polyfit(x, series, 1)
        return coeffs[0]  # Slope

    def _identify_factors(self, features: dict) -> list:
        """Identify key contributing factors to sepsis risk"""
        factors = []

        if features['hr_mean_6h'] > 110:
            factors.append('Elevated heart rate')
        if features['temp_max_6h'] > 38.3:
            factors.append('Fever detected')
        if features['rr_mean_6h'] > 20:
            factors.append('Elevated respiratory rate')
        if features['spo2_min_6h'] < 92:
            factors.append('Low oxygen saturation')
        if features['hrv_mean_6h'] < 30:
            factors.append('Decreased heart rate variability')

        return factors


# Real-time prediction service
def continuous_sepsis_monitoring():
    predictor = SepsisPredictor()

    # Get all ICU patients with wearable monitoring
    patients = get_monitored_patients()

    for patient in patients:
        prediction = predictor.predict_sepsis_risk(patient['id'])

        if prediction['risk_score'] and prediction['risk_score'] > 50:
            # Trigger clinical alert
            send_sepsis_alert(prediction)

            # Log prediction to database
            store_prediction(prediction)

# Run every 15 minutes
schedule.every(15).minutes.do(continuous_sepsis_monitoring)

Step 5: Implement Clinical Alerting System

Transform predictions into actionable clinical alerts with proper escalation.

Alert Engine

// TypeScript: Clinical alert engine with escalation logic
import { Kafka } from 'kafkajs';
import twilio from 'twilio';
import nodemailer from 'nodemailer';

interface ClinicalAlert {
  alert_id: string;
  patient_id: string;
  alert_type: 'sepsis_risk' | 'cardiac_event' | 'respiratory_decline' | 'medication_adherence';
  severity: 'low' | 'moderate' | 'high' | 'critical';
  risk_score: number;
  contributing_factors: string[];
  timestamp: string;
  requires_immediate_action: boolean;
}

interface AlertRecipient {
  role: 'nurse' | 'physician' | 'rapid_response_team';
  name: string;
  phone: string;
  email: string;
  sms_enabled: boolean;
}

class ClinicalAlertingSystem {
  private kafka: Kafka;
  private twilioClient: twilio.Twilio;
  private emailClient: nodemailer.Transporter;

  constructor() {
    this.kafka = new Kafka({
      clientId: 'alert-engine',
      brokers: ['kafka:9092']
    });

    this.twilioClient = twilio(
      process.env.TWILIO_ACCOUNT_SID,
      process.env.TWILIO_AUTH_TOKEN
    );

    this.emailClient = nodemailer.createTransport({
      host: process.env.SMTP_HOST,
      port: 587,
      auth: {
        user: process.env.SMTP_USER,
        pass: process.env.SMTP_PASS
      }
    });
  }

  async processAlert(alert: ClinicalAlert): Promise<void> {
    console.log(`Processing alert ${alert.alert_id} for patient ${alert.patient_id}`);

    // Determine alert routing based on severity
    const recipients = await this.getAlertRecipients(alert);

    // Send alerts via appropriate channels
    await Promise.all([
      this.sendEHRAlert(alert),
      this.sendMobileNotifications(alert, recipients),
      this.logAlert(alert)
    ]);

    // Critical alerts require escalation
    if (alert.severity === 'critical') {
      await this.escalateAlert(alert, recipients);
    }

    // Track alert acknowledgment
    await this.trackAlertAcknowledgment(alert);
  }

  async getAlertRecipients(alert: ClinicalAlert): Promise<AlertRecipient[]> {
    // Get patient's care team
    const patient = await db.patients.findById(alert.patient_id);
    const careTeam = await db.careTeam.find({ patient_id: alert.patient_id });

    const recipients: AlertRecipient[] = [];

    // Always notify assigned nurse
    if (patient.assigned_nurse_id) {
      const nurse = await db.staff.findById(patient.assigned_nurse_id);
      recipients.push({
        role: 'nurse',
        name: nurse.name,
        phone: nurse.phone,
        email: nurse.email,
        sms_enabled: nurse.preferences.sms_alerts
      });
    }

    // Notify physician for high/critical alerts
    if (alert.severity === 'high' || alert.severity === 'critical') {
      if (patient.attending_physician_id) {
        const physician = await db.staff.findById(patient.attending_physician_id);
        recipients.push({
          role: 'physician',
          name: physician.name,
          phone: physician.phone,
          email: physician.email,
          sms_enabled: physician.preferences.sms_alerts
        });
      }
    }

    // Activate rapid response team for critical alerts
    if (alert.severity === 'critical') {
      const rrt = await db.rapidResponseTeam.getOnCall();
      recipients.push(...rrt);
    }

    return recipients;
  }

  async sendMobileNotifications(
    alert: ClinicalAlert,
    recipients: AlertRecipient[]
  ): Promise<void> {
    const message = this.formatAlertMessage(alert);

    for (const recipient of recipients) {
      // Send SMS if enabled
      if (recipient.sms_enabled) {
        await this.twilioClient.messages.create({
          body: message,
          from: process.env.TWILIO_PHONE_NUMBER,
          to: recipient.phone
        });
      }

      // Send email
      await this.emailClient.sendMail({
        from: 'Clinical Alerts <alerts@hospital.org>',
        to: recipient.email,
        subject: `[${alert.severity.toUpperCase()}] ${alert.alert_type} - Patient ${alert.patient_id}`,
        html: this.formatAlertEmail(alert, recipient)
      });

      // Send push notification to mobile app
      await this.sendPushNotification(recipient, alert);
    }
  }

  async sendEHRAlert(alert: ClinicalAlert): Promise<void> {
    // Post alert to EHR inbox via FHIR Communication resource
    const fhirCommunication = {
      resourceType: 'Communication',
      status: 'in-progress',
      category: [{
        coding: [{
          system: 'http://terminology.hl7.org/CodeSystem/communication-category',
          code: 'alert',
          display: 'Alert'
        }]
      }],
      priority: this.mapSeverityToPriority(alert.severity),
      subject: {
        reference: `Patient/${alert.patient_id}`
      },
      sent: new Date().toISOString(),
      payload: [{
        contentString: this.formatAlertMessage(alert)
      }],
      note: [{
        text: `Risk Score: ${alert.risk_score}/100\nFactors: ${alert.contributing_factors.join(', ')}`
      }]
    };

    await fhirClient.create({
      resourceType: 'Communication',
      body: fhirCommunication
    });
  }

  async escalateAlert(
    alert: ClinicalAlert,
    recipients: AlertRecipient[]
  ): Promise<void> {
    // Wait 5 minutes for acknowledgment
    await this.sleep(5 * 60 * 1000);

    const acknowledged = await this.isAlertAcknowledged(alert.alert_id);

    if (!acknowledged) {
      console.log(`Alert ${alert.alert_id} not acknowledged, escalating...`);

      // Page charge nurse
      const chargeNurse = await db.staff.getChargeNurse();
      await this.sendUrgentPage(chargeNurse, alert);

      // Escalate to medical director after 10 more minutes
      setTimeout(async () => {
        const stillUnacknowledged = await this.isAlertAcknowledged(alert.alert_id);
        if (stillUnacknowledged) {
          const medicalDirector = await db.staff.getMedicalDirector();
          await this.sendUrgentPage(medicalDirector, alert);
        }
      }, 10 * 60 * 1000);
    }
  }

  formatAlertMessage(alert: ClinicalAlert): string {
    return `
🚨 ${alert.severity.toUpperCase()} ALERT

Patient: ${alert.patient_id}
Type: ${alert.alert_type.replace(/_/g, ' ')}
Risk Score: ${alert.risk_score}/100

Contributing Factors:
${alert.contributing_factors.map(f => `• ${f}`).join('\n')}

Action Required: ${alert.requires_immediate_action ? 'IMMEDIATE' : 'Assess within 30 minutes'}
    `.trim();
  }
}

// Kafka consumer for processing alerts
async function startAlertProcessor() {
  const alertSystem = new ClinicalAlertingSystem();
  const kafka = new Kafka({ clientId: 'alert-processor', brokers: ['kafka:9092'] });
  const consumer = kafka.consumer({ groupId: 'clinical-alerts' });

  await consumer.connect();
  await consumer.subscribe({ topic: 'clinical.alerts.*', fromBeginning: false });

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      const alert: ClinicalAlert = JSON.parse(message.value.toString());
      await alertSystem.processAlert(alert);
    }
  });
}

startAlertProcessor();

Step 6: FDA Validation and Clinical Testing

For clinical use, your wearable data pipeline must meet FDA software validation requirements.

Validation Documentation

Required Documentation:

Software Design Specification: Architecture diagrams, data flow, algorithm descriptions
Risk Analysis: FMEA (Failure Mode and Effects Analysis)
Verification Testing: Unit tests, integration tests proving software works as designed
Validation Testing: Clinical testing proving software meets user needs
Traceability Matrix: Requirements → Design → Tests → Results

Example Verification Test Suite

// Jest: Automated testing for clinical algorithms
describe('Sepsis Risk Prediction', () => {
  let predictor: SepsisPredictor;

  beforeAll(() => {
    predictor = new SepsisPredictor();
  });

  test('should return high risk for SIRS criteria', async () => {
    const mockData = {
      patient_id: 'test-123',
      heart_rate_6h_avg: 115,  // Tachycardia
      respiratory_rate_6h_avg: 24,  // Tachypnea
      temperature_6h_avg: 38.5,  // Fever
      wbc_count: 14000  // Leukocytosis
    };

    const result = await predictor.predict(mockData);

    expect(result.risk_score).toBeGreaterThan(75);
    expect(result.category).toBe('high');
  });

  test('should not generate false positives for athletes', async () => {
    const mockData = {
      patient_id: 'athlete-456',
      heart_rate_6h_avg: 45,  // Athletic bradycardia
      respiratory_rate_6h_avg: 12,
      temperature_6h_avg: 37.0,
      patient_context: { is_athlete: true }
    };

    const result = await predictor.predict(mockData);

    expect(result.risk_score).toBeLessThan(25);
  });

  test('should handle missing data gracefully', async () => {
    const result = await predictor.predict({ patient_id: 'no-data' });

    expect(result.error).toBe('Insufficient data');
    expect(result.risk_score).toBeNull();
  });
});

How JustCopy.ai Can Help

Building a clinical-grade wearable data pipeline requires expertise in distributed systems, stream processing, machine learning, healthcare regulations, and months of development time.

JustCopy.ai compresses this timeline from months to days. With specialized AI agents, you can:

✅ Clone production-ready data pipeline templates with Kafka, Flink, InfluxDB, and ML models ✅ Customize clinical algorithms for your specific use case (sepsis, cardiac events, medication adherence, etc.) ✅ Deploy to AWS/GCP/Azure with automated infrastructure provisioning ✅ Validate with auto-generated test suites meeting FDA requirements ✅ Monitor with real-time dashboards, alerting, and performance analytics ✅ Scale automatically to handle millions of data points per day

JustCopy.ai’s wearable analytics templates include:

Complete Kafka/Flink streaming infrastructure
Time-series database optimization for wearable data
Pre-trained ML models for sepsis, cardiac arrest, fall detection
Clinical alerting engine with EHR integration
FDA validation documentation templates
HIPAA-compliant security architecture
Real-time analytics dashboards for clinicians

Ready to build your clinical-grade wearable data pipeline? Start with JustCopy.ai and deploy a production-ready real-time analytics platform in days, not months.