288 lines
12 KiB
Markdown
288 lines
12 KiB
Markdown
# Guide 24: Trend Detection (Rate-of-Change Analysis)
|
|
|
|
## What is Trend Detection?
|
|
|
|
Traditional threshold alerts fire when a vital sign crosses a fixed boundary — "heart rate above 130, alert." But what about a heart rate that's at 85, then 95, then 105, then 115 — all within 30 minutes? No single reading crosses the threshold, but the patient is clearly deteriorating rapidly.
|
|
|
|
**Trend detection** (also called rate-of-change analysis) watches the _speed_ at which a vital sign is changing over time. It calculates the **velocity** — how many units per minute the value is rising or falling — and fires an alert when the velocity exceeds a threshold.
|
|
|
|
Think of it like a speedometer vs a position marker. A threshold alert says "you're past the speed limit." A trend alert says "you're accelerating dangerously fast and will hit the speed limit soon."
|
|
|
|
```
|
|
Heart Rate over 30 minutes:
|
|
|
|
130 ┤ ← Threshold (CRITICAL_HEART_RATE)
|
|
│ ╱
|
|
120 ┤ ╱╱
|
|
│ ╱╱
|
|
110 ┤ ╱╱
|
|
│ ╱╱
|
|
100 ┤ ╱╱
|
|
│╱╱
|
|
90 ┤ ← No individual reading crosses 130...
|
|
│ but the RATE (0.8 bpm/min) exceeds the trend threshold (0.5)
|
|
└──────────────────────────────────────
|
|
0 5 10 15 20 25 30 min
|
|
|
|
→ RAPID_DETERIORATION alert fires at ~20 minutes
|
|
```
|
|
|
|
---
|
|
|
|
## How Trend Detection Works
|
|
|
|
### The 5 Tracked Parameters
|
|
|
|
```csharp
|
|
public static class TrendCalculator
|
|
{
|
|
public static readonly IReadOnlyList<string> TrendCodes = new[]
|
|
{
|
|
"HEART_RATE", "RESP_RATE", "SYSTOLIC_BP", "TEMP_C", "SPO2"
|
|
};
|
|
}
|
|
```
|
|
|
|
Each parameter has a configured velocity threshold (from `appsettings.json`):
|
|
|
|
```json
|
|
{
|
|
"TrendDetection": {
|
|
"WindowMinutes": 30,
|
|
"MaxHistoryEntries": 10,
|
|
"HistoryTtlSeconds": 7200,
|
|
"RateThresholdsPerMinute": {
|
|
"HEART_RATE": 0.5,
|
|
"RESP_RATE": 0.3,
|
|
"SYSTOLIC_BP": 1.0,
|
|
"TEMP_C": 0.05,
|
|
"SPO2": 0.2
|
|
}
|
|
}
|
|
}
|
|
```
|
|
|
|
| Parameter | Threshold | Meaning |
|
|
|-----------|-----------|---------|
|
|
| HEART_RATE | 0.5 /min/min | Heart rate rising by 0.5 bpm each minute (15 bpm over 30 min) |
|
|
| RESP_RATE | 0.3 /min/min | Respiratory rate rising by 0.3/min each minute |
|
|
| SYSTOLIC_BP | 1.0 mmHg/min | Blood pressure dropping by 1 mmHg each minute (30 mmHg in 30 min) |
|
|
| TEMP_C | 0.05 °C/min | Temperature rising by 0.05°C each minute (1.5°C in 30 min) |
|
|
| SPO2 | 0.2 %/min | Oxygen saturation dropping by 0.2% each minute |
|
|
|
|
### Direction Matters
|
|
|
|
For HEART_RATE, RESP_RATE, and TEMP_C, rising values are concerning (tachycardia, tachypnea, fever). For SPO2 and SYSTOLIC_BP, falling values are concerning (desaturation, hypotension). The calculator handles this:
|
|
|
|
```csharp
|
|
public static bool ExceedsThreshold(
|
|
string observationCode, decimal ratePerMinute, decimal thresholdPerMinute) =>
|
|
observationCode switch
|
|
{
|
|
"SPO2" or "SYSTOLIC_BP" => ratePerMinute <= -thresholdPerMinute,
|
|
_ => ratePerMinute >= thresholdPerMinute
|
|
};
|
|
```
|
|
|
|
For SPO2: a rate of -0.3 %/min (dropping) exceeds the threshold of 0.2 (because |-0.3| > 0.2). For HEART_RATE: a rate of +0.7 bpm/min (rising) exceeds the threshold of 0.5.
|
|
|
|
---
|
|
|
|
## The Trend Detection Pipeline
|
|
|
|
```
|
|
Observation arrives (Kafka consumer: trend-analyzer)
|
|
│
|
|
▼
|
|
Is it a trend code? (HEART_RATE, RESP_RATE, etc.)
|
|
│ no → return NotTrendCode
|
|
│ yes
|
|
▼
|
|
Read history from Redis: trend:{encounterId}:{code}
|
|
│
|
|
▼
|
|
Append new entry, trim to window (30 min, max 10 entries)
|
|
│
|
|
▼
|
|
Write updated history back to Redis (2-hour TTL)
|
|
│
|
|
▼
|
|
Enough history? (need >= 2 data points)
|
|
│ no → return InsufficientHistory
|
|
│ yes
|
|
▼
|
|
Compute velocity: (newest value - oldest value) / time difference
|
|
│
|
|
▼
|
|
Exceeds threshold?
|
|
│ no → return Stable
|
|
│ yes
|
|
▼
|
|
Create RAPID_DETERIORATION alert (if not already open)
|
|
```
|
|
|
|
### Redis History Storage
|
|
|
|
The trend detector stores a list of recent readings in a single Redis key as a JSON array:
|
|
|
|
```csharp
|
|
var cache = _redis.GetDatabase();
|
|
var key = TrendCalculator.HistoryKey(encounterId, observationCode);
|
|
|
|
// Read existing history
|
|
var historyJson = await cache.StringGetAsync(key);
|
|
var history = historyJson.HasValue
|
|
? JsonSerializer.Deserialize<List<TrendHistoryEntry>>(historyJson!)
|
|
: new List<TrendHistoryEntry>();
|
|
|
|
// Append new entry
|
|
history.Add(new TrendHistoryEntry(value, recordedAt));
|
|
|
|
// Trim: remove entries outside window, keep max N entries
|
|
var cutoff = recordedAt.AddMinutes(-_options.WindowMinutes);
|
|
history = history
|
|
.Where(e => e.RecordedAt >= cutoff)
|
|
.TakeLast(_options.MaxHistoryEntries) // max 10
|
|
.ToList();
|
|
|
|
// Write back with TTL
|
|
await cache.StringSetAsync(
|
|
key, JsonSerializer.Serialize(history),
|
|
TimeSpan.FromSeconds(_options.HistoryTtlSeconds)); // 2 hours
|
|
```
|
|
|
|
**Why a JSON list instead of a Redis list?** Redis lists support push/pop operations, but trimming by time range (remove entries older than 30 minutes) requires scanning the entire list. Storing the entire history as a JSON string allows the application to deserialize, filter, and reserialize in one read-write cycle. At max 10 entries, the overhead is negligible.
|
|
|
|
### Velocity Calculation
|
|
|
|
```csharp
|
|
public static decimal? ComputeRatePerMinute(
|
|
IReadOnlyList<TrendHistoryEntry> entries, int windowMinutes)
|
|
{
|
|
if (entries.Count < 2) return null;
|
|
|
|
var newest = entries[^1]; // last entry
|
|
var oldest = entries[0]; // first entry
|
|
|
|
var deltaMinutes = (newest.RecordedAt - oldest.RecordedAt).TotalMinutes;
|
|
if (deltaMinutes <= 0 || deltaMinutes > windowMinutes) return null;
|
|
|
|
return (newest.Value - oldest.Value) / (decimal)deltaMinutes;
|
|
}
|
|
```
|
|
|
|
**Simple linear velocity**: The rate is `(newest - oldest) / time elapsed`. This is intentionally simple — no weighted averages, no curve fitting. For clinical safety, a simple slope between the oldest and newest readings in the window is sufficient and easy to reason about.
|
|
|
|
**Why require deltaMinutes > 0?** Two readings with identical timestamps would produce division by zero. Why check `> windowMinutes`? If the oldest and newest readings span more than the window (e.g., a stale entry wasn't properly trimmed), the rate would be artificially diluted.
|
|
|
|
---
|
|
|
|
## Alert Creation
|
|
|
|
When the velocity exceeds the threshold, a `RAPID_DETERIORATION` alert is created:
|
|
|
|
```csharp
|
|
var details = TrendCalculator.DescribeTrend(observationCode, rate.Value, value);
|
|
// "Rapid rise: HEART_RATE rising at 0.72/min (current 118)"
|
|
// "Rapid decline: SPO2 falling at 0.31/min (current 91)"
|
|
|
|
var affected = await db.Database.ExecuteSqlInterpolatedAsync($"""
|
|
INSERT INTO clinical_alerts
|
|
(id, encounter_id, patient_id, alert_type, severity, details,
|
|
observation_code, status, triggered_at)
|
|
SELECT {alertId}, {encounterId}, {patientId},
|
|
'RAPID_DETERIORATION', 'WARNING', {fullDetails},
|
|
{observationCode}, 'OPEN', {triggeredAt}
|
|
WHERE NOT EXISTS (
|
|
SELECT 1 FROM clinical_alerts
|
|
WHERE encounter_id = {encounterId}
|
|
AND alert_type = 'RAPID_DETERIORATION'
|
|
AND observation_code = {observationCode}
|
|
AND status IN ('OPEN', 'ESCALATED')
|
|
)
|
|
""", ct);
|
|
```
|
|
|
|
Key design decisions:
|
|
|
|
- **Alert severity is WARNING**, not CRITICAL — trend detection is predictive ("the patient may deteriorate"), not confirmatory ("the patient has a dangerous vital sign"). The actual threshold breach alert (CRITICAL_HEART_RATE, CRITICAL_SPO2) fires separately when the value crosses the absolute threshold.
|
|
- **`observation_code` in the WHERE clause** — a patient can have simultaneous trend alerts for different parameters (rising heart rate AND falling SpO2), but only one per parameter.
|
|
- **Outbox event for downstream consumers** — the alert appears on the dashboard, triggers Kafka consumers (ES indexer, notification publisher), and may eventually escalate through RabbitMQ if unacknowledged.
|
|
|
|
---
|
|
|
|
## The Kafka Consumer: TrendAnalyzerService
|
|
|
|
```csharp
|
|
public class TrendAnalyzerService : BackgroundService
|
|
{
|
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
|
{
|
|
var config = new ConsumerConfig
|
|
{
|
|
BootstrapServers = _kafkaOptions.BootstrapServers,
|
|
GroupId = "trend-analyzer",
|
|
AutoOffsetReset = AutoOffsetReset.Earliest,
|
|
EnableAutoCommit = false
|
|
};
|
|
|
|
using var consumer = new ConsumerBuilder<string, string>(config).Build();
|
|
consumer.Subscribe(_kafkaOptions.Topics.ObservationRecorded);
|
|
|
|
var guard = new PoisonPillGuard("trend-analyzer",
|
|
_kafkaOptions.MaxPoisonRetries, _logger);
|
|
|
|
while (!stoppingToken.IsCancellationRequested)
|
|
{
|
|
var result = consumer.Consume(stoppingToken);
|
|
var evt = JsonSerializer.Deserialize<TrendObservationEvent>(result.Message.Value)!;
|
|
|
|
using var scope = _services.CreateScope();
|
|
var detector = scope.ServiceProvider.GetRequiredService<TrendDetector>();
|
|
var outcome = await detector.ProcessObservationAsync(
|
|
evt.EncounterId, evt.PatientId,
|
|
evt.ObservationCode, evt.Value, evt.RecordedAt, stoppingToken);
|
|
|
|
if (outcome.Outcome == TrendOutcome.RapidDeterioration)
|
|
_logger.LogWarning(
|
|
"RAPID_DETERIORATION for {Code} in encounter {EncounterId}",
|
|
evt.ObservationCode, evt.EncounterId);
|
|
|
|
consumer.Commit(result);
|
|
guard.OnSuccess();
|
|
}
|
|
}
|
|
}
|
|
```
|
|
|
|
This follows the standard Kafka consumer pattern from Guide 12: consume → process → commit → repeat.
|
|
|
|
---
|
|
|
|
## Example Scenario
|
|
|
|
A patient in the ICU has these heart rate readings over 20 minutes:
|
|
|
|
| Time | Heart Rate | History Window | Velocity |
|
|
|------|-----------|---------------|----------|
|
|
| 14:00 | 82 | [82] | — (need >= 2) |
|
|
| 14:05 | 88 | [82, 88] | +1.2/min (exceeds 0.5) |
|
|
| 14:10 | 95 | [82, 88, 95] | +1.3/min |
|
|
| 14:15 | 103 | [82, 88, 95, 103] | +1.4/min |
|
|
| 14:20 | 112 | [82, 88, 95, 103, 112] | +1.5/min |
|
|
|
|
At 14:05, the velocity (1.2/min) already exceeds the threshold (0.5/min). A `RAPID_DETERIORATION` alert fires. The alert details: "Rapid rise: HEART_RATE rising at 1.20/min (current 88) — velocity 1.20/min over 30min window."
|
|
|
|
The individual readings (82, 88, 95...) are all normal — none cross the critical threshold of 130. But the trend detector catches the rapid acceleration before any threshold is breached.
|
|
|
|
---
|
|
|
|
## Key Takeaways
|
|
|
|
- **Trend detection catches deterioration early** — before absolute thresholds are breached, alerting clinicians to accelerating decline
|
|
- **Velocity is simple and interpretable** — (newest - oldest) / time. No complex statistics. Clinicians can understand "heart rate rising at 0.7 bpm/min."
|
|
- **Direction-aware thresholds** — rising heart rate is bad, but rising SpO2 is good. Falling SpO2 is bad, but falling heart rate may be fine. Each parameter knows which direction to watch.
|
|
- **Redis history with JSON lists** — lightweight storage for sliding-window data with automatic TTL expiry
|
|
- **WARNING severity for predictive alerts** — trend alerts warn about future risk; threshold alerts confirm current danger. Both are needed.
|
|
- **One trend alert per parameter per encounter** — deduplication prevents alert storms when a patient is continuously deteriorating
|