From 91dace2e7ccd99a8798f4bfe1125e35d5aad1865 Mon Sep 17 00:00:00 2001 From: Chris Farhood Date: Mon, 20 Jul 2026 16:27:36 -0400 Subject: [PATCH] fix(readiness): calendar-anchored windows, disjoint RHR baseline, SWC floor 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 Claude-Session: https://claude.ai/code/session_01NGzHtDvJur9U7ysgRKRUTN --- src/intervals_mcp_server/tools/wellness.py | 17 +- src/intervals_mcp_server/utils/readiness.py | 233 +++++++++++++++----- tests/test_readiness.py | 46 +++- 3 files changed, 233 insertions(+), 63 deletions(-) diff --git a/src/intervals_mcp_server/tools/wellness.py b/src/intervals_mcp_server/tools/wellness.py index 9d815f6..9c80b65 100644 --- a/src/intervals_mcp_server/tools/wellness.py +++ b/src/intervals_mcp_server/tools/wellness.py @@ -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"))) diff --git a/src/intervals_mcp_server/utils/readiness.py b/src/intervals_mcp_server/utils/readiness.py index 4f66398..dbc1264 100644 --- a/src/intervals_mcp_server/utils/readiness.py +++ b/src/intervals_mcp_server/utils/readiness.py @@ -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 = { diff --git a/tests/test_readiness.py b/tests/test_readiness.py index c218e2c..a2f97d1 100644 --- a/tests/test_readiness.py +++ b/tests/test_readiness.py @@ -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] = []