Tags: healthtech, iot, python, caregiving
Cardiovascular disease accounts for 32% of deaths among Canadians aged 65+. Yet most digital health tooling is built for hospitals, not homes. This guide explores how to think about building practical, data-driven heart health monitoring systems for aging adults — whether you're a developer working in healthtech, a family caregiver with a technical background, or a product team building senior wellness tools.
Note: This article draws from a broader guide on cardiovascular care for Montreal seniors published by Signature Care, a bilingual home care provider. The clinical context here is real — we're translating it into technical architecture.
Why Senior Cardiovascular Monitoring Is a Hard Problem
Aging changes the cardiovascular system in ways that make symptom detection genuinely difficult:
- Arteries lose elasticity → blood pressure readings become more variable
- Heart muscle thickening → standard HRV baselines shift
- Frailty compounds risk → a 4x increase in one-year mortality for frail patients with existing heart conditions
- Symptoms present atypically → seniors often show fatigue or confusion rather than classic chest pain
This means naive threshold alerting ("heart rate > 100 = alert") produces enormous false positive rates. Useful systems need personalized baselines, trend detection, and contextual awareness.
System Architecture Overview
A practical home-based cardiovascular monitoring stack typically involves four layers:
┌─────────────────────────────────────────────────┐
│ Caregiver/Family Dashboard │
├─────────────────────────────────────────────────┤
│ Alert Engine + Trend Analysis │
├─────────────────────────────────────────────────┤
│ Data Normalization + Baseline Model │
├─────────────────────────────────────────────────┤
│ Sensor Layer (Wearables, BP Cuffs, Scales) │
└─────────────────────────────────────────────────┘
Each layer has distinct failure modes worth designing around.
Layer 1: Sensor Selection and Data Ingestion
Recommended Consumer Devices with APIs
| Device Type | Example | Protocol | Notes |
|---|---|---|---|
| Wrist wearable | Fitbit, Apple Watch | REST / HealthKit | HRV, resting HR, SpO2 |
| Blood pressure cuff | Omron Connect | Bluetooth LE / REST | Clinically validated |
| Smart scale | Withings Body+ | REST API | Weight, BMI, body water |
| Pulse oximeter | Nonin / Wellue | BLE | SpO2, respiratory rate |
Basic Data Ingestion Pattern (Python)
import requests
from datetime import datetime, timedelta
import json
class CardioDataCollector:
"""
Aggregates cardiovascular metrics from multiple device APIs.
Designed for daily batch collection with optional real-time hooks.
"""
def __init__(self, config: dict):
self.fitbit_token = config["fitbit_token"]
self.omron_token = config["omron_token"]
self.withings_token = config["withings_token"]
self.patient_id = config["patient_id"]
def fetch_resting_heart_rate(self, date: str) -> dict:
"""Fetch resting HR from Fitbit API."""
url = f"https://api.fitbit.com/1/user/-/activities/heart/date/{date}/1d.json"
headers = {"Authorization": f"Bearer {self.fitbit_token}"}
response = requests.get(url, headers=headers)
response.raise_for_status()
data = response.json()
resting_hr = data["activities-heart"][0]["value"].get("restingHeartRate")
return {
"metric": "resting_heart_rate",
"value": resting_hr,
"date": date,
"source": "fitbit",
"patient_id": self.patient_id
}
def fetch_blood_pressure(self, start_date: str, end_date: str) -> list:
"""Fetch BP readings from Omron Connect."""
url = "https://openapi.ocp.omronhealthcare-ap.com/measurements/bloodpressure"
headers = {"Authorization": f"Bearer {self.omron_token}"}
params = {"startDate": start_date, "endDate": end_date}
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
readings = []
for entry in response.json().get("measurements", []):
readings.append({
"metric": "blood_pressure",
"systolic": entry["systolic"],
"diastolic": entry["diastolic"],
"pulse": entry["pulse"],
"timestamp": entry["dateTime"],
"source": "omron",
"patient_id": self.patient_id
})
return readings
Key design decision: Collect everything into a normalized schema early. Downstream models shouldn't care which vendor produced a heart rate reading.
Layer 2: Baseline Modeling (The Hard Part)
Raw thresholds fail seniors because individual cardiovascular baselines vary enormously. A resting HR of 58 bpm is normal for one person and a clinical anomaly for another.
Building a Rolling Personal Baseline
import numpy as np
import pandas as pd
from scipy import stats
class CardioBaselineModel:
"""
Maintains a rolling personal baseline for cardiovascular metrics.
Uses a 30-day window with exponential weighting to account for
seasonal and medication-driven shifts.
"""
METRICS = ["resting_heart_rate", "systolic_bp", "diastolic_bp", "weight_kg"]
WINDOW_DAYS = 30
def __init__(self, patient_id: str, db_connection):
self.patient_id = patient_id
self.db = db_connection
def compute_baseline(self, metric: str) -> dict:
"""
Returns mean, std, and alert thresholds for a given metric.
Thresholds are set at 2 standard deviations from personal mean,
not population norms.
"""
df = self._load_recent_readings(metric, days=self.WINDOW_DAYS)
if len(df) < 7:
# Insufficient data — fall back to clinical population norms
return self._population_fallback(metric)
# Exponential weighting — more recent readings count more
weights = np.exp(np.linspace(0, 1, len(df)))
weighted_mean = np.average(df["value"], weights=weights)
weighted_std = np.sqrt(
np.average((df["value"] - weighted_mean) ** 2, weights=weights)
)
return {
"metric": metric,
"personal_mean": round(weighted_mean, 2),
"personal_std": round(weighted_std, 2),
"alert_low": round(weighted_mean - (2 * weighted_std), 2),
"alert_high": round(weighted_mean + (2 * weighted_std), 2),
"data_points": len(df),
"window_days": self.WINDOW_DAYS,
"baseline_type": "personal"
}
def _population_fallback(self, metric: str) -> dict:
"""
Clinical reference ranges for seniors 65+.
Source: Canadian Cardiovascular Society guidelines.
"""
fallbacks = {
"resting_heart_rate": {"mean": 70, "std": 10},
"systolic_bp": {"mean": 130, "std": 15},
"diastolic_bp": {"mean": 80, "std": 10},
"weight_kg": {"mean": None, "std": None} # Too variable, skip
}
params = fallbacks.get(metric, {})
if not params["mean"]:
return {"baseline_type": "unavailable", "metric": metric}
return {
"metric": metric,
"personal_mean": params["mean"],
"personal_std": params["std"],
"alert_low": params["mean"] - (2 * params["std"]),
"alert_high": params["mean"] + (2 * params["std"]),
"data_points": 0,
"baseline_type": "population_fallback"
}
def detect_trend(self, metric: str, window_days: int = 7) -> dict:
"""
Runs linear regression over recent readings to catch gradual drift
before it crosses an alert threshold.
Gradual BP creep over 5 days is clinically significant even if
no single reading triggers an alert.
"""
df = self._load_recent_readings(metric, days=window_days)
if len(df) < 3:
return {"trend": "insufficient_data"}
x = np.arange(len(df))
slope, intercept, r_value, p_value, std_err = stats.linregress(x, df["value"])
return {
"metric": metric,
"slope_per_day": round(slope, 4),
"r_squared": round(r_value ** 2, 4),
"p_value": round(p_value, 4),
"trend_direction": "increasing" if slope > 0 else "decreasing",
"statistically_significant": p_value < 0.05
}
def _load_recent_readings(self, metric: str, days: int) -> pd.DataFrame:
query = """
SELECT value, recorded_at
FROM health_readings
WHERE patient_id = %s
AND metric = %s
AND recorded_at >= NOW() - INTERVAL '%s days'
ORDER BY recorded_at ASC
"""
return pd.read_sql(query, self.db, params=[self.patient_id, metric, days])
Layer 3: Alert Engine
Good alert design for senior care has a different optimization target than typical monitoring: false negatives are catastrophic, but alert fatigue leads to ignored notifications — which is also catastrophic.
python
from enum import Enum
from dataclasses import dataclass
from typing import Optional
import logging
class AlertSeverity(Enum):
INFORMATIONAL = "info" # Log only, no notification
CAREGIVER = "caregiver" # Notify family/home care team
URGENT = "urgent" # Notify + recommend calling 811
EMERGENCY = "emergency" # Notify + recommend calling 911
@dataclass
class CardioAlert:
patient_id: str
metric: str
current_value: float
baseline_mean: float
deviation_sigmas: float
severity: AlertSeverity
message: str
trend_context: Optional[dict] = None
class AlertEngine:
"""
Multi-factor alert system combining threshold deviation,
trend analysis, and symptom context.
Clinical warning signs translated to code:
- Chest discomfort → not detectable via wearable, captured via symptom log
- Unusual fatigue → activity level drop vs. baseline
- Dizziness → accelerometer event patterns + HR spike
- Swelling → weight trend increase
- Irregular heartbeat → HRV anomaly detection
"""
# Absolute clinical thresholds (override personal baseline)
HARD_LIMITS = {
"systolic_bp": {"critical_high": 180, "critical_low": 90},
"diastolic_bp": {"critical_high": 120, "critical_low": 60},
"resting_heart_rate": {"critical_high": 120, "critical_low": 40},
"spo2": {"critical_low": 90}
}
def evaluate(
self,
reading: dict,
baseline: dict,
trend: dict,
recent_symptoms: list
) -> Optional[CardioAlert]:
metric = reading["metric"]
value = reading["value"]
# 1. Check hard clinical limits first
hard_alert = self._check_hard_limits(metric, value)
if hard_alert:
return hard_alert
# 2. Check personal baseline deviation
if baseline["baseline_type"] != "unavailable":
sigma_deviation = (
(value - baseline["personal_mean"]) / baseline["personal_std"]
if baseline["personal_std"] > 0 else 0
)
severity = self._sigma_to_severity(sigma_deviation)
# 3. Escalate if trend is also significant
if (trend.get("statistically_significant") and
trend.get("slope_per_day") and
severity == AlertSeverity.INFORMATIONAL):
severity = AlertSeverity.CAREGIVER
# 4. Escalate if symptoms co-occur
if recent_symptoms and severity == AlertSeverity.CAREGIVER:
severity = AlertSeverity.URGENT
if severity != AlertSeverity.INFORMATIONAL:
return CardioAlert(
patient_id=reading["patient_id"],
metric=metric,
current_value=value,
baseline_mean=baseline["personal_mean"],
deviation_sigmas=round(sigma_deviation, 2),
severity=severity,
message=self._generate_message(metric, value, severity),
trend_context=trend
)
return None # No alert needed
def _sigma_to_severity(self, sigma: float) -> AlertSeverity:
abs_sigma = abs(sigma)
if abs_sigma < 1.5:
return AlertSeverity.INFORMATIONAL
elif abs_sigma < 2.0:
return AlertSeverity.CAREGIVER
elif abs_sigma < 3.0:
return AlertSeverity.URGENT
else:
return AlertSeverity.EMERGENCY
def _check_hard_limits(self, metric: str, value: float) -> Optional[CardioAlert]:
limits = self.HARD_LIMITS.get(metric, {})
if limits.get("critical_high") and value >= limits["critical_high"]:
return CardioAlert(
patient_id="", # To be filled by caller
metric=metric,
current_value=value,
baseline_mean=0,
deviation_sigmas=999,
severity=AlertSeverity.EMERGENCY,
message=f"{metric} reading of {value} exceeds critical limit. "
f"Call 911 if patient shows symptoms."
)
if limits.get("critical_low") and value <= limits["critical_low"]:
return CardioAlert(
patient_id="",
metric=metric,
current_value=value,
baseline_mean=0,
deviation_sigmas=-999,
severity=AlertSeverity.EMERGENCY,
message=f"{metric} reading of {value} below critical minimum. "
f"Seek immediate medical attention."
)
return None
def _generate_message(self, metric: str, value: float, severity: AlertSeverity) -> str:
templates = {
AlertSeverity.CAREGIVER: (
f"{metric.replace('_', ' ').title()} reading of {value} "
f"is outside normal personal range. Monitor closely."
),
AlertSeverity.URGENT: (
f"{metric.replace('_', ' ').title()} reading of {value} "
f"is significantly abnormal. Consider calling Info-Santé 811."
),
AlertSeverity.
Top comments (0)