farhoodlabs/intervalsicu-mcp
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

fix(readiness): calendar-anchored windows, disjoint RHR baseline, SWC floor

Browse Source 
Address three confirmed review findings:
- Windows were sample-count based, so "7-day" and "last night" claims could be
  built from weeks-old data. All signals now use calendar windows anchored on a
  reference date (the tool passes today); stale metrics report "no recent data"
  and the verdict is withheld instead of presenting old samples as current.
- rhr_signal's `or vals[:-1]` fallback compared the recent week against itself at
  the sample minimum, reading a uniformly-ill week as "ok". The baseline is now
  disjoint by construction and insufficient baselines return nodata.
- The HRV SWC band had no floor, so a near-constant baseline flagged trivial
  fluctuations (50->49) as red "parasympathetic suppression". SWC now floors at
  0.05 ln units (~5% rMSSD, on the order of normal day-to-day variation).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NGzHtDvJur9U7ysgRKRUTN
This commit is contained in:
Chris Farhood
2026-07-20 16:27:36 -04:00
parent 95d7681b03
commit 91dace2e7c
3 changed files with 233 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/intervals_mcp_server/tools/wellness.py
+16 -1
View File
@@ -250,6 +250,12 @@ async def update_wellness_bulk(entries: list[dict[str, Any]]) -> str:
if len(entries) > 92:
return f"Too many entries ({len(entries)}). Limit a bulk update to 92 days."
# Recognized entry keys: the shared snake_case field names plus date/sleep_hours.
# Anything else (e.g. API-style camelCase like "restingHR") is rejected rather
# than silently dropped — otherwise values the caller asked to record would be
# lost behind a success message.
allowed_keys = {snake for snake, _ in _WELLNESS_FIELD_MAP} | {"date", "sleep_hours"}
records: list[dict[str, Any]] = []
summaries: list[str] = []
for i, entry in enumerate(entries):
@@ -263,6 +269,13 @@ async def update_wellness_bulk(entries: list[dict[str, Any]]) -> str:
except ValueError as exc:
return f"Error in entry {i}: {exc}"
unknown = sorted(set(entry) - allowed_keys)
if unknown:
return (
f"Error: entry {i} ({date}) has unrecognized field(s): {', '.join(unknown)}. "
f"Valid fields: {', '.join(sorted(allowed_keys - {'date'}))}."
)
payload = _wellness_payload(entry)
if not payload:
return f"Error: entry {i} ({date}) has no wellness fields to update."
@@ -325,4 +338,6 @@ async def get_training_readiness(days: int = 45) -> str:
if not records:
return "No wellness data found to assess readiness."
return render_readiness(assess_readiness(records))
# Anchor the calendar windows on today so weeks-old data reads as "no recent
# data" rather than being presented as the athlete's current state.
return render_readiness(assess_readiness(records, reference_date=end.strftime("%Y-%m-%d")))
src/intervals_mcp_server/utils/readiness.py
+172 -61
View File
@@ -2,12 +2,16 @@
Training-readiness computation for Intervals.icu wellness data.
Pure functions (no I/O) so they can be unit-tested on fixtures. The HRV method
follows Plews & Laursen: a 7-day rolling mean of ``ln(rMSSD)`` compared to a
rolling baseline, with a "normal" band of baseline mean +/- the smallest
worthwhile change (SWC = 0.5 x baseline SD). Resting HR, sleep and subjective
inputs are each compared to their own recent baseline.
follows Plews & Laursen: a rolling mean of ``ln(rMSSD)`` over the last 7 calendar
days compared to a baseline from the preceding ~30 days, with a "normal" band of
baseline mean +/- the smallest worthwhile change (SWC = 0.5 x baseline SD, with a
floor so a near-constant baseline can't produce a zero-width band). Resting HR,
sleep and subjective inputs are each compared to their own recent baseline.
Nothing is fabricated: a metric with too little data reports "no data" rather
All windows are **calendar-based**, anchored on ``reference_date`` (callers should
pass today): a metric whose samples are older than the window reports "no recent
data" instead of silently treating stale samples as current. Nothing is
fabricated: a metric with too little data in its window reports "no data" rather
than defaulting, and the overall verdict is withheld (not guessed) when the
objective signals are too sparse to be meaningful. Subjective fields use the
conventional Intervals.icu direction (soreness/fatigue/stress/injury: higher is
@@ -19,55 +23,102 @@ from __future__ import annotations
import math
import statistics
from datetime import date, timedelta
from typing import Any
_MIN_HRV_DAYS = 14 # rolling-baseline HRV method needs at least this many samples
_MIN_RHR_DAYS = 7
_MIN_SLEEP_DAYS = 5
_MIN_SUBJ_DAYS = 5
_RECENT_DAYS = 7
_BASELINE_DAYS = 30
_MIN_HRV_RECENT = 4 # samples needed inside the 7-day window
_MIN_HRV_BASELINE = 7
_MIN_RHR_RECENT = 4
_MIN_RHR_BASELINE = 5
_MIN_SLEEP_BASELINE = 5
_MIN_SUBJ_BASELINE = 5
_SUBJ_LATEST_MAX_AGE = 3 # days; older subjective entries aren't "current" feelings
# Floor for the HRV smallest-worthwhile-change band, in ln(rMSSD) units. A
# near-constant baseline (coarsely-rounded device output, very steady athlete)
# would otherwise give SWC ~= 0 and flag trivial fluctuations as alerts. 0.05 ln
# units is ~5% in rMSSD — on the order of normal day-to-day variation.
_SWC_FLOOR = 0.05
_SUBJ_WORSE_HIGH = ("soreness", "fatigue", "stress", "injury")
_SUBJ_WORSE_LOW = ("motivation", "mood")
def _numeric_series(records: list[dict[str, Any]], key: str, positive: bool = False) -> list[float]:
"""Ordered numeric values for ``key`` (records assumed oldest->newest)."""
out: list[float] = []
def _parse_date(value: Any) -> date | None:
try:
return date.fromisoformat(str(value)[:10])
except (ValueError, TypeError):
return None
def _dated_series(
records: list[dict[str, Any]], key: str, positive: bool = False
) -> list[tuple[date, float]]:
"""Date-sorted ``(date, value)`` pairs for ``key``; undated/non-numeric skipped."""
out: list[tuple[date, float]] = []
for r in records:
if not isinstance(r, dict):
continue
d = _parse_date(r.get("id") or r.get("date"))
v = r.get(key)
if isinstance(v, (int, float)) and not isinstance(v, bool):
if positive and v <= 0:
continue
out.append(float(v))
if d is None or not isinstance(v, (int, float)) or isinstance(v, bool):
continue
if positive and v <= 0:
continue
out.append((d, float(v)))
out.sort(key=lambda p: p[0])
return out
def _sorted_by_date(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
return sorted(
[r for r in records if isinstance(r, dict)],
key=lambda r: str(r.get("id") or r.get("date") or ""),
)
def _windows(
pairs: list[tuple[date, float]], ref: date
) -> tuple[list[float], list[float]]:
"""Split values into recent (last 7 calendar days) and baseline (30 before that)."""
recent_start = ref - timedelta(days=_RECENT_DAYS)
baseline_start = recent_start - timedelta(days=_BASELINE_DAYS)
recent = [v for d, v in pairs if recent_start < d <= ref]
baseline = [v for d, v in pairs if baseline_start < d <= recent_start]
return recent, baseline
def hrv_signal(records: list[dict[str, Any]]) -> dict[str, Any]:
def _newest_date(records: list[dict[str, Any]]) -> date | None:
dates = [
d
for d in (_parse_date(r.get("id") or r.get("date")) for r in records if isinstance(r, dict))
if d is not None
]
return max(dates) if dates else None
def _resolve_ref(records: list[dict[str, Any]], reference_date: str | None) -> date | None:
return _parse_date(reference_date) if reference_date else _newest_date(records)
def hrv_signal(records: list[dict[str, Any]], reference_date: str | None = None) -> dict[str, Any]:
"""HRV readiness via 7-day rolling lnRMSSD vs baseline band (mean +/- SWC)."""
vals = _numeric_series(records, "hrv", positive=True)
if len(vals) < _MIN_HRV_DAYS:
pairs = _dated_series(records, "hrv", positive=True)
ref = _resolve_ref(records, reference_date)
if ref is None or not pairs:
return {"name": "HRV", "level": "nodata", "detail": "no HRV data"}
recent_vals, baseline_vals = _windows(pairs, ref)
if len(recent_vals) < _MIN_HRV_RECENT:
return {
"name": "HRV",
"level": "nodata",
"detail": f"only {len(vals)} day(s) of HRV — need >= {_MIN_HRV_DAYS} for a baseline",
"detail": f"only {len(recent_vals)} HRV sample(s) in the last {_RECENT_DAYS} days",
}
ln = [math.log(v) for v in vals]
recent = ln[-_RECENT_DAYS:]
baseline = ln[:-_RECENT_DAYS][-_BASELINE_DAYS:]
if len(baseline) < 2:
return {"name": "HRV", "level": "nodata", "detail": "not enough baseline days"}
recent_mean = statistics.mean(recent)
base_mean = statistics.mean(baseline)
swc = 0.5 * statistics.pstdev(baseline)
if len(baseline_vals) < _MIN_HRV_BASELINE:
return {
"name": "HRV",
"level": "nodata",
"detail": f"only {len(baseline_vals)} baseline day(s) — need >= {_MIN_HRV_BASELINE}",
}
recent_mean = statistics.mean(math.log(v) for v in recent_vals)
ln_base = [math.log(v) for v in baseline_vals]
base_mean = statistics.mean(ln_base)
swc = max(0.5 * statistics.pstdev(ln_base), _SWC_FLOOR)
if recent_mean < base_mean - swc:
return {
"name": "HRV",
@@ -84,14 +135,27 @@ def hrv_signal(records: list[dict[str, Any]]) -> dict[str, Any]:
return {"name": "HRV", "level": "ok", "detail": "7-day lnHRV within normal band"}
def rhr_signal(records: list[dict[str, Any]]) -> dict[str, Any]:
"""Resting-HR readiness: 7-day mean vs baseline, flag if >5% above."""
vals = _numeric_series(records, "restingHR", positive=True)
if len(vals) < _MIN_RHR_DAYS:
return {"name": "Resting HR", "level": "nodata", "detail": f"only {len(vals)} day(s) of RHR"}
recent = statistics.mean(vals[-_RECENT_DAYS:])
baseline = vals[:-_RECENT_DAYS][-_BASELINE_DAYS:] or vals[:-1]
base = statistics.mean(baseline)
def rhr_signal(records: list[dict[str, Any]], reference_date: str | None = None) -> dict[str, Any]:
"""Resting-HR readiness: 7-day mean vs a disjoint 30-day baseline, flag if >5% above."""
pairs = _dated_series(records, "restingHR", positive=True)
ref = _resolve_ref(records, reference_date)
if ref is None or not pairs:
return {"name": "Resting HR", "level": "nodata", "detail": "no resting-HR data"}
recent_vals, baseline_vals = _windows(pairs, ref)
if len(recent_vals) < _MIN_RHR_RECENT:
return {
"name": "Resting HR",
"level": "nodata",
"detail": f"only {len(recent_vals)} RHR sample(s) in the last {_RECENT_DAYS} days",
}
if len(baseline_vals) < _MIN_RHR_BASELINE:
return {
"name": "Resting HR",
"level": "nodata",
"detail": f"only {len(baseline_vals)} baseline day(s) of RHR — need >= {_MIN_RHR_BASELINE}",
}
recent = statistics.mean(recent_vals)
base = statistics.mean(baseline_vals)
if base > 0 and (recent - base) / base > 0.05:
return {
"name": "Resting HR",
@@ -105,13 +169,27 @@ def rhr_signal(records: list[dict[str, Any]]) -> dict[str, Any]:
}
def sleep_signal(records: list[dict[str, Any]]) -> dict[str, Any]:
"""Sleep readiness: last night vs baseline mean, flag if <85%."""
vals = _numeric_series(records, "sleepSecs", positive=True)
if len(vals) < _MIN_SLEEP_DAYS:
def sleep_signal(records: list[dict[str, Any]], reference_date: str | None = None) -> dict[str, Any]:
"""Sleep readiness: last night (dated within a day of reference) vs baseline mean."""
pairs = _dated_series(records, "sleepSecs", positive=True)
ref = _resolve_ref(records, reference_date)
if ref is None or not pairs:
return {"name": "Sleep", "level": "nodata", "detail": "no sleep data"}
last_date, last_secs = pairs[-1]
if (ref - last_date).days > 1:
return {
"name": "Sleep",
"level": "nodata",
"detail": f"no sleep logged since {last_date.isoformat()}",
}
baseline = [
v / 3600
for d, v in pairs
if d != last_date and ref - timedelta(days=_BASELINE_DAYS) < d <= ref
]
if len(baseline) < _MIN_SLEEP_BASELINE:
return {"name": "Sleep", "level": "nodata", "detail": "not enough sleep data"}
last = vals[-1] / 3600
baseline = [v / 3600 for v in vals[:-1][-_BASELINE_DAYS:]]
last = last_secs / 3600
mean = statistics.mean(baseline)
if mean > 0 and last < 0.85 * mean:
return {
@@ -126,16 +204,29 @@ def sleep_signal(records: list[dict[str, Any]]) -> dict[str, Any]:
}
def subjective_signals(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Soft warnings when a subjective field has moved off baseline in the worse direction."""
def subjective_signals(
records: list[dict[str, Any]], reference_date: str | None = None
) -> list[dict[str, Any]]:
"""Soft warnings when a *current* subjective field has moved off baseline for the worse."""
signals: list[dict[str, Any]] = []
ref = _resolve_ref(records, reference_date)
if ref is None:
return signals
fields = [(f, True) for f in _SUBJ_WORSE_HIGH] + [(f, False) for f in _SUBJ_WORSE_LOW]
for field, worse_high in fields:
vals = _numeric_series(records, field)
if len(vals) < _MIN_SUBJ_DAYS:
pairs = _dated_series(records, field)
if not pairs:
continue
latest_date, latest = pairs[-1]
if (ref - latest_date).days > _SUBJ_LATEST_MAX_AGE:
continue # stale entries aren't current feelings
baseline = [
v
for d, v in pairs
if d != latest_date and ref - timedelta(days=_BASELINE_DAYS) < d <= ref
]
if len(baseline) < _MIN_SUBJ_BASELINE:
continue
latest = vals[-1]
baseline = vals[:-1][-_BASELINE_DAYS:]
mean = statistics.mean(baseline)
sd = statistics.pstdev(baseline) if len(baseline) > 1 else 0.0
threshold = max(sd, 0.5) # require a meaningful move, not noise
@@ -154,20 +245,35 @@ def subjective_signals(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
def form_context(records: list[dict[str, Any]]) -> dict[str, Any] | None:
"""Latest Form (TSB = CTL - ATL) if both components are present."""
if not records:
dated = sorted(
[r for r in records if isinstance(r, dict)],
key=lambda r: str(r.get("id") or r.get("date") or ""),
)
if not dated:
return None
latest = records[-1]
latest = dated[-1]
ctl, atl = latest.get("ctl"), latest.get("atl")
if isinstance(ctl, (int, float)) and isinstance(atl, (int, float)):
return {"form": round(ctl - atl, 1), "ctl": ctl, "atl": atl}
return None
def assess_readiness(records: list[dict[str, Any]]) -> dict[str, Any]:
"""Produce a structured readiness assessment from wellness records."""
records = _sorted_by_date(records)
core = [hrv_signal(records), rhr_signal(records), sleep_signal(records)]
signals = core + subjective_signals(records)
def assess_readiness(
records: list[dict[str, Any]], reference_date: str | None = None
) -> dict[str, Any]:
"""Produce a structured readiness assessment from wellness records.
``reference_date`` (YYYY-MM-DD) anchors the calendar windows — pass today so
stale data reads as "no recent data" instead of masquerading as current. If
omitted, the newest record's date is used (fixture-friendly, but blind to
how old that record is).
"""
core = [
hrv_signal(records, reference_date),
rhr_signal(records, reference_date),
sleep_signal(records, reference_date),
]
signals = core + subjective_signals(records, reference_date)
alerts = [s for s in signals if s["level"] == "alert"]
warns = [s for s in signals if s["level"] == "warn"]
@@ -182,7 +288,12 @@ def assess_readiness(records: list[dict[str, Any]]) -> dict[str, Any]:
else:
verdict = "green"
return {"verdict": verdict, "signals": signals, "form": form_context(records), "days": len(records)}
return {
"verdict": verdict,
"signals": signals,
"form": form_context(records),
"days": len(records),
}
_VERDICT_LABEL = {
tests/test_readiness.py
+45 -1
View File
@@ -7,6 +7,7 @@ layer stubbed. Fixtures are built so verdicts are unambiguous.
"""
import asyncio
from datetime import date, timedelta
from intervals_mcp_server.tools import wellness
from intervals_mcp_server.utils import readiness
@@ -17,6 +18,14 @@ def _days(specs: list[dict]) -> list[dict]:
return [{"id": f"2026-06-{i + 1:02d}", **spec} for i, spec in enumerate(specs)]
def _days_ending_today(specs: list[dict]) -> list[dict]:
"""Like _days, but the last record is dated today (for tool-level tests)."""
start = date.today() - timedelta(days=len(specs) - 1)
return [
{"id": (start + timedelta(days=i)).isoformat(), **spec} for i, spec in enumerate(specs)
]
def _stable(n: int, **fields) -> list[dict]:
return _days([dict(fields) for _ in range(n)])
@@ -48,6 +57,13 @@ def test_hrv_elevated_warn():
assert readiness.hrv_signal(_days(baseline + recent))["level"] == "warn"
def test_hrv_constant_baseline_small_dip_is_not_alert():
# A near-constant baseline gives SWC ~ 0; the floor must keep a trivial
# 50 -> 49 fluctuation from producing a false "Compromised" alert.
recs = _days([{"hrv": 50} for _ in range(23)] + [{"hrv": 49} for _ in range(7)])
assert readiness.hrv_signal(recs)["level"] == "ok"
# --------------------------------------------------------------------------- #
# RHR / sleep signals
# --------------------------------------------------------------------------- #
@@ -60,6 +76,13 @@ def test_rhr_normal_ok():
assert readiness.rhr_signal(_stable(20, restingHR=48))["level"] == "ok"
def test_rhr_minimum_days_is_nodata_not_self_baseline():
# With only 7 samples there is no disjoint baseline; a uniformly-elevated
# (ill) week must NOT read "ok" from being compared against itself.
sig = readiness.rhr_signal(_stable(7, restingHR=58))
assert sig["level"] == "nodata"
def test_sleep_short_warn():
recs = _days([{"sleepSecs": 28800} for _ in range(10)] + [{"sleepSecs": 18000}])
assert readiness.sleep_signal(recs)["level"] == "warn"
@@ -123,6 +146,22 @@ def test_form_context_computed():
# --------------------------------------------------------------------------- #
# render + tool integration
# --------------------------------------------------------------------------- #
def test_stale_data_withholds_verdict():
# Daily logging that STOPPED 3 weeks ago must not produce a current verdict:
# with today as the reference date every calendar window is empty.
old = _days([{"hrv": 50 + (i % 3), "restingHR": 48, "sleepSecs": 28800} for i in range(30)])
out = readiness.assess_readiness(old, reference_date=date.today().isoformat())
assert out["verdict"] == "insufficient"
assert all(s["level"] == "nodata" for s in out["signals"])
def test_sleep_not_logged_recently_is_nodata():
recs = _days([{"sleepSecs": 28800} for _ in range(10)])
sig = readiness.sleep_signal(recs, reference_date="2026-07-01") # 3 weeks later
assert sig["level"] == "nodata"
assert "no sleep logged since" in sig["detail"]
def test_render_insufficient_mentions_logging():
out = readiness.render_readiness(readiness.assess_readiness(_stable(3, restingHR=48)))
assert "Verdict withheld" in out
@@ -131,8 +170,13 @@ def test_render_insufficient_mentions_logging():
def test_get_training_readiness_tool(monkeypatch):
# Wellness API returns a date-keyed dict; the tool must normalize and assess it.
start = date.today() - timedelta(days=29)
records = {
f"2026-06-{i + 1:02d}": {"hrv": 50 + (i % 3), "restingHR": 48, "sleepSecs": 28800}
(start + timedelta(days=i)).isoformat(): {
"hrv": 50 + (i % 3),
"restingHR": 48,
"sleepSecs": 28800,
}
for i in range(30)
}
calls: list[dict] = []