From 055dc8be211d463121f998ca671a9e44cf622e9c Mon Sep 17 00:00:00 2001 From: Chris Farhood Date: Mon, 20 Jul 2026 15:57:35 -0400 Subject: [PATCH] feat(readiness): add get_training_readiness synthesizer New pure-compute utils/readiness.py (stdlib only, fully unit-tested) assesses readiness from wellness history: HRV via Plews & Laursen 7-day rolling lnRMSSD vs baseline +/- SWC, resting-HR and sleep trends, and conventional-direction subjective inputs (soft warnings only). The get_training_readiness tool fetches the window, normalizes the date-keyed API response, and renders a banded verdict with the contributing signals. Verdict is withheld (not fabricated) when HRV is sparse and fewer than two other core signals have data. Implements #1. Completes the 0.3.0 coaching-context milestone. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01NGzHtDvJur9U7ysgRKRUTN --- src/intervals_mcp_server/server.py | 2 + src/intervals_mcp_server/tools/__init__.py | 2 + src/intervals_mcp_server/tools/wellness.py | 48 ++++- src/intervals_mcp_server/utils/readiness.py | 219 ++++++++++++++++++++ tests/test_readiness.py | 164 +++++++++++++++ 5 files changed, 434 insertions(+), 1 deletion(-) create mode 100644 src/intervals_mcp_server/utils/readiness.py create mode 100644 tests/test_readiness.py diff --git a/src/intervals_mcp_server/server.py b/src/intervals_mcp_server/server.py index 21c3b92..e8ee07f 100644 --- a/src/intervals_mcp_server/server.py +++ b/src/intervals_mcp_server/server.py @@ -90,6 +90,7 @@ from intervals_mcp_server.tools.events import ( # pylint: disable=wrong-import- ) from intervals_mcp_server.tools.gear import get_gear_list # pylint: disable=wrong-import-position # noqa: E402 from intervals_mcp_server.tools.wellness import ( # pylint: disable=wrong-import-position # noqa: E402 + get_training_readiness, get_wellness_data, update_wellness, update_wellness_bulk, @@ -136,6 +137,7 @@ __all__ = [ "get_wellness_data", "update_wellness", "update_wellness_bulk", + "get_training_readiness", "get_athlete_profile", "get_sport_settings", "get_athlete_summary", diff --git a/src/intervals_mcp_server/tools/__init__.py b/src/intervals_mcp_server/tools/__init__.py index a01d0d7..8eadbfe 100644 --- a/src/intervals_mcp_server/tools/__init__.py +++ b/src/intervals_mcp_server/tools/__init__.py @@ -36,6 +36,7 @@ from intervals_mcp_server.tools.power_curves import ( # noqa: F401 ) from intervals_mcp_server.tools.gear import get_gear_list # noqa: F401 from intervals_mcp_server.tools.wellness import ( # noqa: F401 + get_training_readiness, get_wellness_data, update_wellness, update_wellness_bulk, @@ -89,6 +90,7 @@ __all__ = [ "get_wellness_data", "update_wellness", "update_wellness_bulk", + "get_training_readiness", "get_athlete_profile", "get_sport_settings", "get_athlete_summary", diff --git a/src/intervals_mcp_server/tools/wellness.py b/src/intervals_mcp_server/tools/wellness.py index e27c02b..9d815f6 100644 --- a/src/intervals_mcp_server/tools/wellness.py +++ b/src/intervals_mcp_server/tools/wellness.py @@ -4,13 +4,14 @@ Wellness-related MCP tools for Intervals.icu. This module contains tools for retrieving athlete wellness data. """ -from datetime import datetime +from datetime import datetime, timedelta from typing import Any from intervals_mcp_server import credentials from intervals_mcp_server.api.client import make_intervals_request from intervals_mcp_server.credentials import CredentialError from intervals_mcp_server.utils.formatting import format_wellness_entry +from intervals_mcp_server.utils.readiness import assess_readiness, render_readiness from intervals_mcp_server.utils.validation import resolve_date_params, validate_date # Import mcp instance from shared module for tool registration @@ -280,3 +281,48 @@ async def update_wellness_bulk(entries: list[dict[str, Any]]) -> str: return f"Error updating wellness data: {result.get('message')}" return f"Updated {len(records)} day(s):\n" + "\n".join(summaries) + + +@mcp.tool() +async def get_training_readiness(days: int = 45) -> str: + """Assess training readiness from recent wellness data. + + Synthesizes the athlete's recent wellness history into a readiness read: + HRV-guided (7-day rolling lnRMSSD vs baseline +/- smallest worthwhile change), + resting-HR and sleep trends, and subjective inputs (soreness/fatigue/stress/ + mood/motivation). When there is too little data — notably fewer than ~2 weeks + of HRV — the verdict is withheld rather than guessed, and the report lists which + signals it could and could not use. + + Args: + days: How many days of history to analyze (default 45; minimum 14 is enforced). + """ + try: + athlete_id_to_use, api_key = await credentials.resolve_caller_credentials() + except CredentialError as exc: + return str(exc) + + end = datetime.now() + start = end - timedelta(days=max(days, 14)) + params = {"oldest": start.strftime("%Y-%m-%d"), "newest": end.strftime("%Y-%m-%d")} + + result = await make_intervals_request( + url=f"/athlete/{athlete_id_to_use}/wellness", api_key=api_key, params=params + ) + + if isinstance(result, dict) and "error" in result: + return f"Error fetching wellness data: {result.get('message')}" + + records: list[dict[str, Any]] = [] + if isinstance(result, dict): + for date_str, data in result.items(): + if isinstance(data, dict): + data.setdefault("id", date_str) + records.append(data) + elif isinstance(result, list): + records = [r for r in result if isinstance(r, dict)] + + if not records: + return "No wellness data found to assess readiness." + + return render_readiness(assess_readiness(records)) diff --git a/src/intervals_mcp_server/utils/readiness.py b/src/intervals_mcp_server/utils/readiness.py new file mode 100644 index 0000000..4f66398 --- /dev/null +++ b/src/intervals_mcp_server/utils/readiness.py @@ -0,0 +1,219 @@ +""" +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. + +Nothing is fabricated: a metric with too little data 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 +worse; mood/motivation: higher is better) and only ever contribute a soft +warning, never a hard alert. +""" + +from __future__ import annotations + +import math +import statistics +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 + +_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] = [] + for r in records: + v = r.get(key) + if isinstance(v, (int, float)) and not isinstance(v, bool): + if positive and v <= 0: + continue + out.append(float(v)) + 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 hrv_signal(records: list[dict[str, Any]]) -> 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: + return { + "name": "HRV", + "level": "nodata", + "detail": f"only {len(vals)} day(s) of HRV — need >= {_MIN_HRV_DAYS} for a baseline", + } + 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 recent_mean < base_mean - swc: + return { + "name": "HRV", + "level": "alert", + "detail": "7-day lnHRV below baseline band — parasympathetic suppression", + } + if recent_mean > base_mean + swc: + return { + "name": "HRV", + "level": "warn", + "detail": "7-day lnHRV above baseline band — super-compensation, " + "or saturation if resting HR is also elevated", + } + 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) + if base > 0 and (recent - base) / base > 0.05: + return { + "name": "Resting HR", + "level": "warn", + "detail": f"7-day RHR {recent:.0f} is >5% above baseline {base:.0f}", + } + return { + "name": "Resting HR", + "level": "ok", + "detail": f"7-day RHR {recent:.0f} near baseline {base:.0f}", + } + + +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: + return {"name": "Sleep", "level": "nodata", "detail": "not enough sleep data"} + last = vals[-1] / 3600 + baseline = [v / 3600 for v in vals[:-1][-_BASELINE_DAYS:]] + mean = statistics.mean(baseline) + if mean > 0 and last < 0.85 * mean: + return { + "name": "Sleep", + "level": "warn", + "detail": f"last night {last:.1f}h below baseline {mean:.1f}h", + } + return { + "name": "Sleep", + "level": "ok", + "detail": f"last night {last:.1f}h near baseline {mean:.1f}h", + } + + +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.""" + signals: list[dict[str, Any]] = [] + 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: + 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 + worse = (latest - mean > threshold) if worse_high else (mean - latest > threshold) + if worse: + direction = "elevated" if worse_high else "low" + signals.append( + { + "name": field.capitalize(), + "level": "warn", + "detail": f"{field} {direction} vs baseline ({latest:g} vs {mean:.1f})", + } + ) + return signals + + +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: + return None + latest = records[-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) + + alerts = [s for s in signals if s["level"] == "alert"] + warns = [s for s in signals if s["level"] == "warn"] + core_with_data = [s for s in core if s["level"] != "nodata"] + + if core[0]["level"] == "nodata" and len(core_with_data) < 2: + verdict = "insufficient" + elif alerts or len(warns) >= 3: + verdict = "red" + elif warns: + verdict = "amber" + else: + verdict = "green" + + return {"verdict": verdict, "signals": signals, "form": form_context(records), "days": len(records)} + + +_VERDICT_LABEL = { + "green": "🟢 Ready — signals within normal range", + "amber": "🟡 Caution — one or more signals off baseline", + "red": "🔴 Compromised — strong or multiple negative signals", + "insufficient": "⚪ Verdict withheld — not enough data to judge", +} +_LEVEL_ICON = {"ok": "✓", "warn": "!", "alert": "‼", "nodata": "·"} + + +def render_readiness(assessment: dict[str, Any]) -> str: + """Render a readiness assessment into a plain-language report.""" + lines = [ + "Training Readiness:", + "", + _VERDICT_LABEL.get(assessment["verdict"], assessment["verdict"]), + f"(based on {assessment['days']} day(s) of wellness data)", + "", + "Signals:", + ] + for s in assessment["signals"]: + lines.append(f" {_LEVEL_ICON.get(s['level'], '-')} {s['name']}: {s['detail']}") + + form = assessment["form"] + if form: + lines += ["", f"Form (TSB): {form['form']} (CTL {form['ctl']} / ATL {form['atl']})"] + + if assessment["verdict"] == "insufficient": + lines += [ + "", + "Log daily HRV (and resting HR) for ~2+ weeks to enable a readiness verdict.", + ] + return "\n".join(lines) diff --git a/tests/test_readiness.py b/tests/test_readiness.py new file mode 100644 index 0000000..c218e2c --- /dev/null +++ b/tests/test_readiness.py @@ -0,0 +1,164 @@ +""" +Tests for the training-readiness feature. + +The pure compute layer (utils/readiness.py) is exercised directly on deterministic +fixtures; one integration test drives the get_training_readiness tool with the HTTP +layer stubbed. Fixtures are built so verdicts are unambiguous. +""" + +import asyncio + +from intervals_mcp_server.tools import wellness +from intervals_mcp_server.utils import readiness + + +def _days(specs: list[dict]) -> list[dict]: + """Build wellness records with sequential dates from a list of field dicts.""" + return [{"id": f"2026-06-{i + 1:02d}", **spec} for i, spec in enumerate(specs)] + + +def _stable(n: int, **fields) -> list[dict]: + return _days([dict(fields) for _ in range(n)]) + + +# --------------------------------------------------------------------------- # +# HRV signal +# --------------------------------------------------------------------------- # +def test_hrv_insufficient_data(): + recs = _stable(10, hrv=50) + sig = readiness.hrv_signal(recs) + assert sig["level"] == "nodata" + + +def test_hrv_normal_band(): + # 23 stable baseline days + 7 stable recent days -> within band + recs = _days([{"hrv": 50 + (i % 3)} for i in range(30)]) + assert readiness.hrv_signal(recs)["level"] == "ok" + + +def test_hrv_suppressed_alert(): + baseline = [{"hrv": 50 + (i % 3)} for i in range(23)] + recent = [{"hrv": 34} for _ in range(7)] + assert readiness.hrv_signal(_days(baseline + recent))["level"] == "alert" + + +def test_hrv_elevated_warn(): + baseline = [{"hrv": 50 + (i % 3)} for i in range(23)] + recent = [{"hrv": 75} for _ in range(7)] + assert readiness.hrv_signal(_days(baseline + recent))["level"] == "warn" + + +# --------------------------------------------------------------------------- # +# RHR / sleep signals +# --------------------------------------------------------------------------- # +def test_rhr_elevated_warn(): + recs = _days([{"restingHR": 48} for _ in range(23)] + [{"restingHR": 56} for _ in range(7)]) + assert readiness.rhr_signal(recs)["level"] == "warn" + + +def test_rhr_normal_ok(): + assert readiness.rhr_signal(_stable(20, restingHR=48))["level"] == "ok" + + +def test_sleep_short_warn(): + recs = _days([{"sleepSecs": 28800} for _ in range(10)] + [{"sleepSecs": 18000}]) + assert readiness.sleep_signal(recs)["level"] == "warn" + + +def test_sleep_nodata(): + assert readiness.sleep_signal(_stable(3, sleepSecs=28800))["level"] == "nodata" + + +# --------------------------------------------------------------------------- # +# subjective signals (conventional direction) +# --------------------------------------------------------------------------- # +def test_subjective_fatigue_elevated_warns(): + recs = _days([{"fatigue": 2} for _ in range(10)] + [{"fatigue": 4}]) + sigs = readiness.subjective_signals(recs) + assert any(s["name"] == "Fatigue" and s["level"] == "warn" for s in sigs) + + +def test_subjective_stable_no_warning(): + assert readiness.subjective_signals(_stable(10, fatigue=2, mood=3)) == [] + + +# --------------------------------------------------------------------------- # +# overall verdict +# --------------------------------------------------------------------------- # +def test_verdict_green_all_stable(): + recs = _days( + [{"hrv": 50 + (i % 3), "restingHR": 48, "sleepSecs": 28800} for i in range(30)] + ) + assert readiness.assess_readiness(recs)["verdict"] == "green" + + +def test_verdict_red_on_hrv_suppression(): + recs = _days( + [{"hrv": 50 + (i % 3), "restingHR": 48, "sleepSecs": 28800} for i in range(23)] + + [{"hrv": 33, "restingHR": 57, "sleepSecs": 28800} for _ in range(7)] + ) + assert readiness.assess_readiness(recs)["verdict"] == "red" + + +def test_verdict_insufficient_when_hrv_sparse_and_little_else(): + # Only 3 days total, no HRV baseline and <2 other core signals with data. + recs = _stable(3, restingHR=48) + out = readiness.assess_readiness(recs) + assert out["verdict"] == "insufficient" + + +def test_verdict_uses_rhr_and_sleep_when_hrv_missing(): + # No HRV, but RHR + sleep both have data -> a verdict is still produced (green here). + recs = _days([{"restingHR": 48, "sleepSecs": 28800} for _ in range(20)]) + out = readiness.assess_readiness(recs) + assert out["verdict"] == "green" + assert any(s["name"] == "HRV" and s["level"] == "nodata" for s in out["signals"]) + + +def test_form_context_computed(): + recs = _days([{"ctl": 60, "atl": 70}]) + assert readiness.form_context(recs) == {"form": -10.0, "ctl": 60, "atl": 70} + + +# --------------------------------------------------------------------------- # +# render + tool integration +# --------------------------------------------------------------------------- # +def test_render_insufficient_mentions_logging(): + out = readiness.render_readiness(readiness.assess_readiness(_stable(3, restingHR=48))) + assert "Verdict withheld" in out + assert "Log daily HRV" in out + + +def test_get_training_readiness_tool(monkeypatch): + # Wellness API returns a date-keyed dict; the tool must normalize and assess it. + records = { + f"2026-06-{i + 1:02d}": {"hrv": 50 + (i % 3), "restingHR": 48, "sleepSecs": 28800} + for i in range(30) + } + calls: list[dict] = [] + + async def fake(**kwargs): + calls.append(kwargs) + return records + + monkeypatch.setattr(wellness, "make_intervals_request", fake) + out = asyncio.run(wellness.get_training_readiness(days=45)) + assert calls[0]["url"] == "/athlete/i1/wellness" + assert "Training Readiness:" in out + assert "🟢 Ready" in out + + +def test_get_training_readiness_no_data(monkeypatch): + async def fake(**kwargs): + return {} + + monkeypatch.setattr(wellness, "make_intervals_request", fake) + assert "No wellness data found" in asyncio.run(wellness.get_training_readiness()) + + +def test_get_training_readiness_error(monkeypatch): + async def fake(**kwargs): + return {"error": True, "message": "down"} + + monkeypatch.setattr(wellness, "make_intervals_request", fake) + assert "Error fetching wellness data: down" in asyncio.run(wellness.get_training_readiness())