""" Feature engineering for the AI coach. Everything here is arithmetic over stored health data — no model calls. The split is deliberate: the numbers a briefing quotes (z-scores, baselines, trend slopes) must be reproducible and testable, and an LLM asked to compute them from a raw CSV gets them wrong often enough to matter. The model's job is to interpret figures that were already computed here, not to derive them. Two windows are used throughout: * **baseline** (default 28 days) — what "normal for this person, lately" means. Short enough to track a training block, long enough for a standard deviation to be worth quoting. * **trend** (default 395 days ≈ 13 months) — the long arc the product spec asks about, and long enough to contain a full season. """ import datetime import statistics from services import health from services import settings as settings_svc from services import fitness_age BASELINE_DAYS = 28 TREND_DAYS = 395 # Sleep targets are personal, but Garmin's own coaching and the ACSM/AASM # adult guidance both land on 7 hours as the floor; the deep/REM shares are # the conventional adult reference bands. SLEEP_TARGET_HOURS = 7.0 REM_REFERENCE_PCT = (20.0, 25.0) DEEP_REFERENCE_PCT = (13.0, 23.0) # |z| beyond this counts as a departure from the personal baseline rather # than day-to-day noise. 1.0 rather than the textbook 2.0: with a 28-day # window a 2-sigma day is roughly a once-a-month event, which is too rare to # drive a daily briefing. Z_NOTABLE = 1.0 def _flatten(day): """One day as a flat metric -> value mapping. `get_summary` nests sleep and omits missing metrics entirely; both are inconvenient for statistics, so sleep is lifted to the top level and derived shares (deep/REM percent) are computed once here. """ flat = {k: v for k, v in day.items() if k != "sleep"} sleep = day.get("sleep") or {} duration = sleep.get("duration") or day.get("sleepDuration") if duration: flat["sleepDuration"] = duration seconds = duration * 3600.0 for src, dest in ( ("deepSeconds", "sleepDeepPct"), ("remSeconds", "sleepRemPct"), ("lightSeconds", "sleepLightPct"), ("awakeSeconds", "sleepAwakePct"), ): value = sleep.get(src) if value is not None and seconds > 0: flat[dest] = round(value / seconds * 100, 1) if sleep.get("quality") is not None: flat["sleepQuality"] = sleep["quality"] sedentary = day.get("sedentarySeconds") if sedentary is not None: flat["sedentaryHours"] = round(sedentary / 3600.0, 1) return flat def flatten_days(rows): """`get_summary` output as flat metric maps, with derived shares filled in. Public counterpart of `_flatten` for the scope builders, which all need the same normalisation before they can compute anything. """ return [_flatten(r) for r in rows] # Metrics the briefing reasons about. `higher_better` drives the plain-language # verdict; None means the direction is not meaningful on its own (steps on a # rest day are not a failure). METRICS = { "sleepDuration": ("睡眠时长", "小时", True), "sleepQuality": ("睡眠评分", "分", True), "sleepDeepPct": ("深睡占比", "%", True), "sleepRemPct": ("REM 占比", "%", True), "heartRate": ("静息心率", "bpm", False), "heartRateVariability": ("HRV", "ms", True), "stress": ("压力均值", "", False), "bodyBatteryHigh": ("身体电量峰值", "", True), "bodyBatteryLow": ("身体电量谷值", "", True), "trainingReadiness": ("训练准备度", "分", True), "enduranceScore": ("耐力分", "分", True), "vo2max": ("最大摄氧量", "ml/kg/min", True), "steps": ("步数", "步", None), "intensityMinutes": ("强度分钟", "分钟", None), "respirationAvg": ("呼吸频率", "次/分", None), "spo2Avg": ("血氧", "%", True), } def _series(rows, metric): """(date, value) pairs where the metric was actually recorded.""" return [(r["date"], r[metric]) for r in rows if r.get(metric) is not None] def _stats(values): if not values: return None mean = statistics.fmean(values) # pstdev, not stdev: these are all the observations in the window, not a # sample drawn from it, and stdev raises on a single point. sd = statistics.pstdev(values) if len(values) > 1 else 0.0 return {"mean": mean, "sd": sd, "n": len(values)} def _verdict(z, higher_better): if higher_better is None or abs(z) < Z_NOTABLE: return "正常" if (z > 0) == bool(higher_better): return "偏好" return "偏差" def deviations(rows, today, baseline_days=BASELINE_DAYS): """How far each of today's metrics sits from its own recent baseline. The baseline deliberately excludes today: comparing a value against a mean it helped produce shrinks its own z-score, and with a 28-day window that bias is large enough to hide a genuine outlier. """ history = [r for r in rows if r["date"] < today.get("date", "")] window = history[-baseline_days:] out = [] for metric, (label, unit, higher_better) in METRICS.items(): value = today.get(metric) if value is None: continue values = [v for _, v in _series(window, metric)] stats = _stats(values) if not stats or stats["n"] < 5: # Too little history for a standard deviation to mean anything. out.append({ "metric": metric, "label": label, "unit": unit, "value": round(float(value), 2), "baselineMean": None, "sd": None, "z": None, "verdict": "基线不足", }) continue sd = stats["sd"] if sd > 0: z = round((float(value) - stats["mean"]) / sd, 2) verdict = _verdict(z, higher_better) else: # A baseline with no spread cannot scale a departure. Reporting # z = 0 here would label a value that differs from every single # observation as perfectly typical, which is the opposite of true. z = None verdict = "正常" if float(value) == stats["mean"] else "基线无波动" out.append({ "metric": metric, "label": label, "unit": unit, "value": round(float(value), 2), "baselineMean": round(stats["mean"], 2), "sd": round(sd, 2), "baselineDays": stats["n"], "z": z, "verdict": verdict, }) # Biggest departures first: that ordering is what the prompt relies on to # keep the interesting metrics inside the model's attention span. out.sort(key=lambda d: abs(d["z"]) if d["z"] is not None else -1, reverse=True) return out def _slope_per_30d(points): """Least-squares slope in units per 30 days. Ordinal dates rather than array indices: gaps in the record (a watch left on the charger for a week) would otherwise compress the x-axis and inflate the slope. """ if len(points) < 3: return None xs = [datetime.date.fromisoformat(d).toordinal() for d, _ in points] ys = [float(v) for _, v in points] mx, my = statistics.fmean(xs), statistics.fmean(ys) denom = sum((x - mx) ** 2 for x in xs) if denom == 0: return None slope = sum((x - mx) * (y - my) for x, y in zip(xs, ys)) / denom return round(slope * 30, 3) def trends(rows, window_days=TREND_DAYS, edge=30): """Long-arc movement per metric: endpoint means plus a fitted slope. Endpoint means (first `edge` days vs last `edge` days) answer "where did this end up"; the slope answers "was it a trend or two different plateaus". Reporting only one of them has misled us before — a metric can finish higher after months of decline if it spikes in the final week. """ if not rows: return [] cutoff = ( datetime.date.fromisoformat(rows[-1]["date"]) - datetime.timedelta(days=window_days) ).isoformat() window = [r for r in rows if r["date"] >= cutoff] out = [] for metric, (label, unit, higher_better) in METRICS.items(): points = _series(window, metric) if len(points) < 10: continue # With fewer than two edges' worth of points the two windows would # overlap and both converge on the overall mean — reporting delta 0 # for a series that visibly moved. Split it in half instead. span = min(edge, len(points) // 2) head = [v for _, v in points[:span]] tail = [v for _, v in points[-span:]] first, last = statistics.fmean(head), statistics.fmean(tail) delta = last - first entry = { "metric": metric, "label": label, "unit": unit, "days": ( datetime.date.fromisoformat(points[-1][0]) - datetime.date.fromisoformat(points[0][0]) ).days, "samples": len(points), "firstMean": round(first, 2), "lastMean": round(last, 2), "delta": round(delta, 2), "slopePer30d": _slope_per_30d(points), } if higher_better is not None and abs(delta) > 0: entry["direction"] = "改善" if (delta > 0) == bool(higher_better) else "退步" out.append(entry) return out def activity_shift(rows, recent=7, prior=30): """Recent activity volume against the weeks before it. Separate from `deviations` because the question is different: not "is today unusual" but "has the last week as a whole dropped off" — the drop that a single quiet day cannot show. """ out = {} for metric in ("steps", "intensityMinutes", "sleepDuration", "bodyBatteryHigh"): points = _series(rows, metric) if len(points) < recent + 5: continue recent_values = [v for _, v in points[-recent:]] prior_values = [v for _, v in points[-(recent + prior):-recent]] if not prior_values: continue r_mean, p_mean = statistics.fmean(recent_values), statistics.fmean(prior_values) out[metric] = { "label": METRICS[metric][0], "recentMean": round(r_mean, 2), "priorMean": round(p_mean, 2), "changePct": round((r_mean - p_mean) / p_mean * 100, 1) if p_mean else None, } return out def _sleep_block(today): duration = today.get("sleepDuration") if duration is None: return None block = { "durationHours": round(float(duration), 2), "targetHours": SLEEP_TARGET_HOURS, "score": today.get("sleepQuality"), "deepPercent": today.get("sleepDeepPct"), "remPercent": today.get("sleepRemPct"), "lightPercent": today.get("sleepLightPct"), "awakePercent": today.get("sleepAwakePct"), "remReference": list(REM_REFERENCE_PCT), "deepReference": list(DEEP_REFERENCE_PCT), } return block def latest_of(rows, metric, within=180): """Most recent recorded value, for metrics that only refresh occasionally. VO2max and endurance score update after a qualifying outdoor session, so reading them off "today" yields None on any indoor or rest day even though the last measured value is still the current one. """ for row in reversed(rows[-within:] if within else rows): if row.get(metric) is not None: return row[metric] return None def build_context(user_id, date=None, rows=None): """The structured payload every coach prompt is assembled from. `date` selects the snapshot day; the default is the newest day on record rather than the calendar date, because a sync may not have run yet today and an empty snapshot produces a briefing about nothing. """ rows = rows if rows is not None else health.get_summary(user_id) if not rows: return None flat = [_flatten(r) for r in rows] if date: matches = [r for r in flat if r["date"] == date] if not matches: return None today = matches[0] history = [r for r in flat if r["date"] <= date] else: today = flat[-1] history = flat profile = settings_svc.get_raw(user_id) age = settings_svc.age_from(profile["birth_date"]) bmi = settings_svc.bmi_from(profile["height_cm"], profile["weight_kg"]) vo2max = latest_of(history, "vo2max") body_age = fitness_age.estimate( age=age, sex=profile["sex"], vo2max=vo2max, resting_hr=latest_of(history, "heartRate", within=30), bmi=bmi, ) start = ( datetime.date.fromisoformat(today["date"]) - datetime.timedelta(days=14) ).isoformat() recent_activities = health.get_activities(user_id, start, today["date"]) sedentary = today.get("sedentaryHours") return { "snapshotDate": today["date"], "userProfile": { "age": age, "sex": profile["sex"], # `estimate` returns {"value": None, "missing": [...]} when the # profile is incomplete, so this is None rather than a number # until a birth date, sex and a VO2max reading all exist. "fitnessAge": (body_age or {}).get("value"), "vo2max": vo2max, "enduranceScore": latest_of(history, "enduranceScore"), "heightCm": profile["height_cm"], "weightKg": profile["weight_kg"], "bmi": bmi, }, "todayMetrics": { "sleep": _sleep_block(today), "autonomicNervous": { "restingHr": today.get("heartRate"), "hrvMs": today.get("heartRateVariability"), "stressAvg": today.get("stress"), "stressMax": today.get("stressMax"), "respirationAvg": today.get("respirationAvg"), "spo2Avg": today.get("spo2Avg"), }, "recovery": { "bodyBatteryPeak": today.get("bodyBatteryHigh"), "bodyBatteryLow": today.get("bodyBatteryLow"), "bodyBatteryCharged": today.get("bodyBatteryCharged"), "bodyBatteryDrained": today.get("bodyBatteryDrained"), "trainingReadiness": today.get("trainingReadiness"), }, "activityToday": { "steps": today.get("steps"), "stepGoal": today.get("stepGoal"), "intensityMinutes": today.get("intensityMinutes"), "sedentaryHours": sedentary, "floorsAscended": today.get("floorsAscended"), "caloriesBurned": today.get("caloriesBurned"), "activeCalories": today.get("activeCalories"), }, }, "deviations": deviations(history, today), "trends": trends(history), "activityShift": activity_shift(history), "recentActivities": [ { "date": a.get("start_time"), "sport": a.get("activity_type"), "durationMin": round((a.get("duration") or 0) / 60) or None, "distanceKm": ( round(a["distance"] / 1000, 2) if a.get("distance") else None ), "calories": a.get("calories"), "avgHr": a.get("heart_rate_average"), "maxHr": a.get("heart_rate_max"), } for a in recent_activities[-15:] ], "dataQuality": { "totalDays": len(history), "firstDate": history[0]["date"], "lastDate": history[-1]["date"], "staleDays": ( datetime.date.today() - datetime.date.fromisoformat(history[-1]["date"]) ).days, }, } def window_context(user_id, metric, start, end, rows=None): """Context for one metric over a user-selected span (chart brush). Narrower than `build_context` on purpose: the question being answered is "what happened to this line here", so the payload carries the selected series plus whatever else moved alongside it in the same window. """ # An unknown metric would otherwise produce a well-formed window with an # empty series, and the model would dutifully write an attribution for a # line that does not exist. if metric not in METRICS: return None rows = rows if rows is not None else health.get_summary(user_id) flat = [_flatten(r) for r in rows] window = [r for r in flat if start <= r["date"] <= end] if not window: return None label, unit, _ = METRICS[metric] points = _series(window, metric) before = [r for r in flat if r["date"] < start][-BASELINE_DAYS:] baseline = _stats([v for _, v in _series(before, metric)]) companions = {} for other in METRICS: if other == metric: continue values = [v for _, v in _series(window, other)] stats = _stats(values) if stats and stats["n"] >= 3: companions[other] = { "label": METRICS[other][0], "mean": round(stats["mean"], 2), "n": stats["n"], } return { "metric": metric, "label": label, "unit": unit, "start": start, "end": end, "points": [{"date": d, "value": v} for d, v in points], "mean": round(statistics.fmean([v for _, v in points]), 2) if points else None, "min": min((v for _, v in points), default=None), "max": max((v for _, v in points), default=None), "slopePer30d": _slope_per_30d(points), "baselineBefore": ( {"mean": round(baseline["mean"], 2), "days": baseline["n"]} if baseline else None ), "companions": companions, "activities": [ { "date": a.get("start_time"), "sport": a.get("activity_type"), "durationMin": round((a.get("duration") or 0) / 60) or None, "calories": a.get("calories"), "avgHr": a.get("heart_rate_average"), } for a in health.get_activities(user_id, start, end)[:40] ], } # Reference bands, kept beside the UI's RANGES table (client/src/lib/ranges.ts). # Each entry says where that metric's band edges came from — the honest answer # for several of them is "general-population orientation figures", and it says # so. Lives here rather than in routes/: the 评分依据 screen serves it, but the # coach also reasons against it, and services must not import from routes. BAND_SOURCES = [ {"metric": "步数", "bands": "<5k 偏低 · 5–8k 一般 · 8–12k 达标 · >12k 优秀", "source": "步数与死亡率的队列研究（约 8000 步起获益明显，12000 步后趋平）"}, {"metric": "静息心率", "bands": "<50 很低 · 50–65 正常 · 65–75 偏高 · >75 较高", "source": "健康成人静息心率 60–100 bpm 为正常范围，规律运动者常低于 60"}, {"metric": "心率变异性", "bands": "<25 偏低 · 25–40 一般 · 40–70 良好 · >70 很好", "source": "夜间 RMSSD 的一般人群分布；个体差异极大，趋势比绝对值更有意义"}, {"metric": "睡眠时长", "bands": "<6h 不足 · 6–7h 偏少 · 7–9h 充足 · >9h 偏多", "source": "美国睡眠医学会 / 睡眠研究会成人 7–9 小时建议"}, {"metric": "压力", "bands": "0–25 休息 · 26–50 偏低 · 51–75 中等 · >75 偏高", "source": "Garmin 官方压力分级，与手表显示一致"}, {"metric": "身体电量", "bands": "0–25 很低 · 26–50 偏低 · 51–75 良好 · >75 充足", "source": "Garmin 官方 Body Battery 分级"}, {"metric": "血氧", "bands": "<90 偏低 · 90–94 略低 · ≥95 正常", "source": "静息血氧饱和度常用临床参考；腕表光学测量误差较大，仅供趋势参考"}, {"metric": "呼吸频率", "bands": "<12 偏低 · 12–20 正常 · >20 偏高", "source": "成人静息呼吸频率 12–20 次/分"}, {"metric": "强度分钟", "bands": "<10 偏少 · 10–21 一般 · ≥21 达标", "source": "WHO 每周 150 分钟中等强度活动，折合每天约 21 分钟"}, {"metric": "训练准备度", "bands": "0–25 很低 · 26–50 偏低 · 51–75 就绪 · >75 很好", "source": "Garmin 官方 Training Readiness 分级"}, {"metric": "爬楼", "bands": "<5 偏少 · 5–10 达标 · >10 优秀", "source": "一般性活动量参考，无权威标准"}, ]