#!/usr/bin/env python3 """ Kitzu — Daily Health Brief Engine Reads the unified profile and generates a concise morning health summary. This is the first piece of the inference engine — it cross-references available data sources to produce actionable insights. Usage: python3 daily_brief.py # Generate today's brief python3 daily_brief.py --json # Output as JSON (for dashboard) python3 daily_brief.py --markdown # Output as Markdown file Reads from: kitzu/data/unified/profile.json + kitzu/data/oura/*.json Writes to: kitzu/data/unified/briefs/{date}.json (optional) """ import sys import json import argparse from datetime import date, datetime, timedelta from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from schema import load_profile, load_oura_history, UNIFIED_DIR, OURA_DIR BRIEFS_DIR = UNIFIED_DIR / "briefs" def generate_brief(target_date: date = None) -> dict: """Generate a daily health brief from available data.""" if target_date is None: target_date = date.today() profile = load_profile() oura_history = load_oura_history(14) brief = { "date": target_date.isoformat(), "generated_at": datetime.now().isoformat(), "priority_actions": [], # Immediate, direct, do-it-yourself actions "supplement_schedule": [], # Daily supplement list with timing + priority "sections": [], "alerts": [], "cross_references": [], "data_freshness": {}, } # ── Data Freshness ─────────────────────────────── for source in ["oura", "blood", "genetics", "microbiome", "vitals", "cgm"]: section = profile.get(source, {}) last_sync = section.get("last_sync") brief["data_freshness"][source] = { "last_sync": last_sync, "stale": _is_stale(last_sync, source), } # ── Sleep Section ──────────────────────────────── sleep = profile.get("oura", {}).get("sleep") if sleep: sleep_section = { "title": "Sleep", "icon": "moon", "metrics": { "hours": sleep.get("hours"), "deep_min": sleep.get("deep_min"), "rem_min": sleep.get("rem_min"), "avg_hrv": sleep.get("avg_hrv"), "resting_hr": sleep.get("resting_hr"), "bedtime": sleep.get("bedtime"), "wake_time": sleep.get("wake_time"), }, "assessment": _assess_sleep(sleep, oura_history), "recommendations": [], } # Sleep recommendations based on data hours = sleep.get("hours", 0) deep = sleep.get("deep_min", 0) if hours < 7: sleep_section["recommendations"].append( "Below 7h target. Consider earlier bedtime or reducing evening screen time." ) if deep and deep < 60: sleep_section["recommendations"].append( "Deep sleep under 60 min. Evening red light therapy or magnesium may help." ) # Cross-reference: sleep + cortisol cortisol = profile.get("blood", {}).get("markers", {}).get("Cortisol", {}) if cortisol and cortisol.get("value", 0) > 12: brief["cross_references"].append({ "sources": ["oura", "blood"], "finding": f"Cortisol elevated ({cortisol['value']} µg/dL) + sleep pattern may indicate stress-recovery imbalance.", "action": "Monitor HRV trend. If declining alongside cortisol rise, consider reducing training load.", }) brief["sections"].append(sleep_section) # ── Readiness Section ──────────────────────────── readiness = profile.get("oura", {}).get("readiness") if readiness: score = readiness.get("score", 0) readiness_section = { "title": "Readiness", "icon": "battery", "metrics": {"score": score}, "assessment": _assess_readiness(score), "recommendations": [], } if score < 70: readiness_section["recommendations"].append( "Low readiness. Recovery day — light movement only. Full recovery stack: Sauna (25 min) \u2192 Cold Plunge (3 min) \u2192 Red Light (20 min)." ) brief["alerts"].append({ "level": "warning", "message": f"Readiness score {score} — recovery day recommended.", }) elif score >= 85: readiness_section["recommendations"].append( "High readiness. Good day for intense training or a challenging workout." ) brief["sections"].append(readiness_section) # ── Activity Section ───────────────────────────── steps = profile.get("oura", {}).get("steps") exercise = profile.get("oura", {}).get("exercise") if steps or exercise: activity_section = { "title": "Activity", "icon": "activity", "metrics": {}, "assessment": "", "recommendations": [], } if steps: activity_section["metrics"]["steps"] = steps.get("count") activity_section["metrics"]["active_calories"] = steps.get("active_calories") if exercise: activity_section["metrics"]["sessions"] = exercise.get("sessions", []) brief["sections"].append(activity_section) # ── HRV Trend ──────────────────────────────────── hrv_trend = profile.get("oura", {}).get("hrv_trend", []) if len(hrv_trend) >= 3: hrv_section = { "title": "HRV Trend", "icon": "heart", "metrics": { "current": hrv_trend[0].get("hrv") if hrv_trend else None, "avg_7d": _avg([h["hrv"] for h in hrv_trend[:7] if h.get("hrv")]), "avg_14d": _avg([h["hrv"] for h in hrv_trend[:14] if h.get("hrv")]), "trend": _trend_direction([h["hrv"] for h in hrv_trend[:7] if h.get("hrv")]), }, "assessment": "", "recommendations": [], } trend_dir = hrv_section["metrics"]["trend"] if trend_dir == "declining": hrv_section["assessment"] = "HRV trending down over the past week. Watch for overtraining or stress accumulation." brief["alerts"].append({ "level": "info", "message": "HRV declining — consider lighter training or active recovery.", }) # Cross-reference: HRV declining + ferritin elevated ferritin = profile.get("blood", {}).get("markers", {}).get("Ferritin", {}) if ferritin and ferritin.get("value", 0) > 400: brief["cross_references"].append({ "sources": ["oura", "blood"], "finding": f"HRV declining while ferritin remains elevated ({ferritin['value']} ng/mL). Iron overload can affect cardiac autonomic function.", "action": "Discuss with doctor — therapeutic phlebotomy may improve both ferritin and HRV.", }) elif trend_dir == "improving": hrv_section["assessment"] = "HRV trending up — recovery and adaptation are going well." brief["sections"].append(hrv_section) # ── Blood Markers Watch List ───────────────────── blood = profile.get("blood", {}) if blood.get("markers"): critical = {k: v for k, v in blood["markers"].items() if v.get("status") == "critical"} at_risk = {k: v for k, v in blood["markers"].items() if v.get("status") == "at_risk"} if critical or at_risk: blood_section = { "title": "Biomarker Watch", "icon": "droplet", "metrics": { "critical_count": len(critical), "at_risk_count": len(at_risk), "inner_age": blood.get("inner_age"), "next_test": None, }, "critical_markers": {k: {"value": v["value"], "unit": v["unit"], "range": v["range"]} for k, v in critical.items()}, "at_risk_markers": {k: {"value": v["value"], "unit": v["unit"], "range": v["range"]} for k, v in at_risk.items()}, "assessment": f"{len(critical)} critical, {len(at_risk)} at-risk markers from latest blood test.", "recommendations": [], } # Specific cross-references based on marker combinations ldl = blood["markers"].get("LDL Cholesterol", {}) apob = blood["markers"].get("ApoB", {}) if ldl.get("value", 0) > 130 and apob.get("value", 0) > 100: brief["cross_references"].append({ "sources": ["blood"], "finding": f"LDL ({ldl['value']}) and ApoB ({apob['value']}) both elevated — concordant cardiovascular risk.", "action": "Priority doctor discussion. Lifestyle + potential medication needed.", }) # Glucose + weight cross-reference glucose = blood["markers"].get("Glucose", {}) weight = profile.get("vitals", {}).get("weight", {}) if glucose.get("value", 0) > 99 and weight and weight.get("bmi", 0) > 25: brief["cross_references"].append({ "sources": ["blood", "vitals"], "finding": f"Fasting glucose ({glucose['value']}) borderline + BMI {weight.get('bmi', '?')} — insulin resistance pattern.", "action": "Prioritize weight loss, strength training, and fiber intake. CGM would reveal post-meal spikes.", }) brief["sections"].append(blood_section) # ── Weight / Body Comp ─────────────────────────── weight = profile.get("vitals", {}).get("weight") if weight: weight_section = { "title": "Body Composition", "icon": "scale", "metrics": { "weight_lb": weight.get("weight_lb") or weight.get("value_lb"), "body_fat_pct": weight.get("body_fat_pct"), "muscle_mass": weight.get("muscle_mass"), "bmi": weight.get("bmi"), }, "assessment": "", "recommendations": [], } brief["sections"].append(weight_section) # ── Priority Actions (sticky, actionable, direct) ── brief["priority_actions"] = _generate_priority_actions(profile, oura_history, brief) # ── Supplement Schedule ────────────────────────── brief["supplement_schedule"] = _generate_supplement_schedule(profile) return brief def _generate_supplement_schedule(profile: dict) -> list: """ Generate a structured daily supplement schedule with timing and priority. Priorities are data-driven from blood markers and sleep. 1 = Must-have (red) — directly targets critical markers 2 = High value (orange) — strong supporting role 3 = Good to have (green) — maintenance / longevity Tags: "split" | "added" | "flag" | None """ blood = profile.get("blood", {}).get("markers", {}) sleep = profile.get("oura", {}).get("sleep", {}) ldl = blood.get("LDL Cholesterol", {}).get("value", 0) apob = blood.get("ApoB", {}).get("value", 0) glucose = blood.get("Glucose", {}).get("value", 0) vitd = blood.get("Vitamin D", {}).get("value", 30) testo = blood.get("Testosterone", {}).get("value", 500) cortisol = blood.get("Cortisol", {}).get("value", 0) deep_min = sleep.get("deep_min", 60) or 60 def entry(time_block, time, name, dose, priority, reason, tag=None): return {"time_block": time_block, "time": time, "name": name, "dose": dose, "priority": priority, "reason": reason, "tag": tag} s = [] # ── 30 min before breakfast ────────────────────── s.append(entry( "30 min before breakfast", "7:30 AM", "Fiber (Psyllium Husk)", "5g in water", 1 if ldl > 130 else 2, f"LDL {ldl} — binds bile cholesterol before food hits · must precede breakfast supplements", "split" )) # ── With breakfast ─────────────────────────────── s.append(entry( "With breakfast", "8:00 AM", "Omega-3 (Fish Oil)", "3 gelcaps (3,000 mg EPA+DHA)", 1 if (ldl > 130 or apob > 100) else 2, f"LDL {ldl}, ApoB {apob} — strongest lifestyle lever for lipid panel · verify 1g EPA+DHA per cap" )) s.append(entry( "With breakfast", "8:00 AM", "Berberine", "500 mg", 1 if (glucose > 99 or ldl > 130) else 2, f"Glucose {glucose}, LDL {ldl} — natural metformin analog · 2nd dose at dinner doubles efficacy", "split" )) s.append(entry( "With breakfast", "8:00 AM", "Vitamin K2 (MK-7)", "200 mcg", 1 if (ldl > 130 or apob > 100) else 2, "Non-negotiable with D3 at 5,000 IU — directs calcium to bones, not arteries", "added" )) s.append(entry( "With breakfast", "8:00 AM", "Vitamin D3", "5,000 IU", 1 if vitd < 30 else 2, f"Tested {vitd} ng/mL — fat-soluble, must take with food" )) s.append(entry( "With breakfast", "8:00 AM", "CoQ10", "200 mg", 2, "Heart health · essential if starting statin · statins deplete CoQ10 · fat-soluble" )) s.append(entry( "With breakfast", "8:00 AM", "Zinc Glycinate", "1 cap", 2 if testo < 500 else 3, "Testosterone support · not with calcium · pair with copper below", "flag" )) s.append(entry( "With breakfast", "8:00 AM", "Copper", "2 mg", 2, "Balances long-term zinc use · zinc-induced copper depletion worsens cardiovascular markers", "added" )) s.append(entry( "With breakfast", "8:00 AM", "Alpha Lipoic Acid", "250 mg", 2 if glucose > 99 else 3, f"Antioxidant + glucose support · pairs with berberine for glucose {glucose}" )) s.append(entry( "With breakfast", "8:00 AM", "B-Complex", "1 cap (slow release)", 2, "Energy + methylation · take early only — B vitamins disrupt sleep if taken late" )) s.append(entry( "With breakfast", "8:00 AM", "Resveratrol", "200 mg + 25 mg grape seed", 3, "Cardiovascular + longevity · fat-soluble, requires food for absorption" )) # ── Mid-morning ────────────────────────────────── s.append(entry( "Mid-morning (2+ hrs after breakfast)", "10:00 AM", "Probiotic", "1 cap", 3, "Gut health · must be 2+ hrs from berberine — berberine kills probiotic bacteria if too close", "flag" )) # ── 30 min before dinner ───────────────────────── s.append(entry( "30 min before dinner", "6:00 PM", "Fiber (Psyllium Husk)", "5g in water", 1 if ldl > 130 else 2, f"2nd dose · 10g/day total is therapeutic for LDL {ldl} · slows post-meal glucose spike", "split" )) # ── With dinner ────────────────────────────────── s.append(entry( "With dinner", "6:30 PM", "Berberine", "500 mg", 1 if (glucose > 99 or ldl > 130) else 2, "2nd dose — split dosing across meals doubles efficacy vs single morning dose", "split" )) # ── Before bed ─────────────────────────────────── s.append(entry( "Before bed", "10:30 PM", "Magnesium Glycinate", "330 mg", 1 if deep_min < 60 else 2, f"Deep sleep {deep_min} min — target 60+ · strongest sleep lever in the stack" )) s.append(entry( "Before bed", "10:30 PM", "Ashwagandha (KSM-66)", "600 mg", 2 if (cortisol > 12 or testo < 500) else 3, f"Cortisol {cortisol} µg/dL + testosterone support · KSM-66 is the clinically studied extract" )) s.append(entry( "Before bed", "10:30 PM", "L-Theanine", "200 mg (slow release)", 2 if (deep_min < 60 or cortisol > 12) else 3, "Pairs with magnesium for sleep · cortisol + calm focus support" )) return s def _generate_priority_actions(profile: dict, oura_history: list, brief: dict) -> list: """ Generate immediate, direct, do-it-yourself priority actions. These are things the user can act on TODAY — no doctor needed. Each action has: category, action (direct imperative), reason (short why). Sorted by priority. Max 6 actions to keep it focused. """ actions = [] sleep = profile.get("oura", {}).get("sleep", {}) blood = profile.get("blood", {}).get("markers", {}) vitals = profile.get("vitals", {}) weight_data = vitals.get("weight", {}) hrv_trend = profile.get("oura", {}).get("hrv_trend", []) # ── SLEEP PROTOCOL ─────────────────────────────── hours = sleep.get("hours", 0) deep = sleep.get("deep_min", 0) bedtime = sleep.get("bedtime", "") # Parse bedtime hour bedtime_late = False if bedtime: try: bt_hour = int(bedtime.split(":")[0]) bt_min = int(bedtime.split(":")[1]) if bt_hour >= 23 and bt_min > 15: bedtime_late = True elif bt_hour >= 0 and bt_hour < 6: bedtime_late = True except (ValueError, IndexError): pass if hours < 7 or deep < 60 or bedtime_late: action = "Lights out by 11:00 PM. Screens off by 10:00." if deep < 60: action += " Take Magnesium Glycinate (400mg) + L-Theanine (200mg) 30 min before bed." if bedtime_late: reason = f"Last night: bed at {bedtime}, only {hours}h sleep, {deep}m deep." else: reason = f"Last night: {hours}h sleep, {deep}m deep — both below target." actions.append({ "category": "sleep", "icon": "🌙", "action": action, "reason": reason, "priority": 1, }) # ── EXERCISE LOAD + HRV / RECOVERY ─────────────── # Three-tier intensity detection (best available signal wins): # 1. Training Load (HR zone-weighted minutes) — gold standard # 2. Calories — good proxy when HR data unavailable # 3. Duration — last resort fallback # # Training Load thresholds (zone-weighted minutes): # Hockey 65 min ≈ 180-250 load (lots of Z3/Z4) # Hiking 60 min ≈ 80-120 load (mostly Z1/Z2) # Heavy day: ≥150 training load HEAVY_LOAD_THRESHOLD = 150 # Training load score to flag as "heavy" HEAVY_CAL_THRESHOLD = 800 # Calorie fallback HEAVY_MIN_FALLBACK = 90 # Duration last resort CUMULATIVE_3D_LOAD = 400 # 3-day training load that signals accumulated stress exercise_feed = profile.get("oura", {}).get("exercise_feed", []) yesterday = (date.today() - timedelta(days=1)).isoformat() yesterday_sessions = [e for e in exercise_feed if e.get("date") == yesterday] yesterday_total_min = sum(e.get("duration_min", 0) for e in yesterday_sessions) yesterday_total_cal = sum(e.get("calories", 0) for e in yesterday_sessions) yesterday_total_load = sum(e.get("training_load", 0) for e in yesterday_sessions) yesterday_types = list(set(e.get("type", "Workout") for e in yesterday_sessions)) yesterday_has_zones = any(e.get("training_load") for e in yesterday_sessions) def _is_heavy_day(sessions): """Determine if a day was heavy — training load first, then cal, then duration.""" total_load = sum(e.get("training_load", 0) for e in sessions) total_cal = sum(e.get("calories", 0) for e in sessions) total_min = sum(e.get("duration_min", 0) for e in sessions) has_zones = any(e.get("training_load") for e in sessions) if has_zones: return total_load >= HEAVY_LOAD_THRESHOLD if total_cal > 0: return total_cal >= HEAVY_CAL_THRESHOLD return total_min >= HEAVY_MIN_FALLBACK def _day_load(sessions): """Get the best available load metric for a day.""" total_load = sum(e.get("training_load", 0) for e in sessions) if total_load > 0: return total_load # Estimate load from calories (rough: 1 cal ≈ 0.2 load units) total_cal = sum(e.get("calories", 0) for e in sessions) if total_cal > 0: return total_cal * 0.2 return 0 # Check last 3 days for consecutive heavy load recent_3d_heavy = [] recent_3d_load = [] for d_offset in range(1, 4): d_str = (date.today() - timedelta(days=d_offset)).isoformat() d_sessions = [e for e in exercise_feed if e.get("date") == d_str] recent_3d_heavy.append(_is_heavy_day(d_sessions)) recent_3d_load.append(_day_load(d_sessions)) heavy_yesterday = _is_heavy_day(yesterday_sessions) consecutive_heavy = sum(1 for h in recent_3d_heavy if h) >= 2 cumulative_3d_load = sum(recent_3d_load) hrvs = [h["hrv"] for h in hrv_trend[:7] if h.get("hrv")] if hrvs: avg_hrv = sum(hrvs) / len(hrvs) trend_dir = None for s in brief.get("sections", []): if s.get("title") == "HRV Trend": trend_dir = s.get("metrics", {}).get("trend") readiness = profile.get("oura", {}).get("readiness", {}).get("score", 0) # Heavy exercise yesterday — factor into recovery recommendation if heavy_yesterday and (trend_dir == "declining" or avg_hrv < 30 or readiness < 75): type_str = " + ".join(yesterday_types) action = f"Recovery day. You logged {yesterday_total_min} min of {type_str} yesterday ({yesterday_total_cal} cal). Walk or yoga only — no intense training. Full recovery stack: Sauna (25 min) \u2192 Cold Plunge (3 min) \u2192 Red Light (20 min)." reason = f"Heavy session yesterday + HRV {avg_hrv:.0f}ms avg (trending {'down' if trend_dir == 'declining' else 'low'}). Your body needs to recover." if consecutive_heavy or cumulative_3d_load >= CUMULATIVE_3D_LOAD: action = f"Mandatory rest. Training load {cumulative_3d_load:.0f} over 3 days — your body is accumulating stress. Walk or stretch only. Full recovery stack: Sauna (25 min) \u2192 Cold Plunge (3 min) \u2192 Red Light (20 min)." reason = f"{sum(1 for h in recent_3d_heavy if h)} heavy days in last 3 + HRV trending {'down' if trend_dir == 'declining' else 'low'}. Overtraining risk is real." actions.append({ "category": "recovery", "icon": "💚", "action": action, "reason": reason, "priority": 2, }) elif heavy_yesterday: # Heavy exercise but good HRV/readiness — acknowledge and suggest active recovery type_str = " + ".join(yesterday_types) if readiness >= 85: actions.append({ "category": "recovery", "icon": "💚", "action": f"Good recovery after {yesterday_total_min} min of {type_str}. Readiness {readiness} — you can train today. Include proper warm-up.", "reason": f"Body handled yesterday's load well. HRV {avg_hrv:.0f}ms, readiness {readiness}. Green light for training.", "priority": 6, }) else: actions.append({ "category": "recovery", "icon": "💚", "action": f"Easy day. {yesterday_total_min} min of {type_str} yesterday — readiness only {readiness}. Light training or active recovery.", "reason": f"Moderate recovery after heavy session. HRV {avg_hrv:.0f}ms. Don't stack another hard day.", "priority": 3, }) elif trend_dir == "declining" or avg_hrv < 30: if readiness < 75: action = "Recovery day. Walk or yoga only — no intense training. Full recovery stack: Sauna (25 min) \u2192 Cold Plunge (3 min) \u2192 Red Light (20 min)." else: action = "Go easy today. Cap training at moderate intensity. Post-workout: Sauna (25 min) \u2192 Cold Plunge (3 min) \u2192 Red Light (20 min)." actions.append({ "category": "recovery", "icon": "💚", "action": action, "reason": f"HRV 7-day average: {avg_hrv:.0f}ms (trending {'down' if trend_dir == 'declining' else 'low'}). Your nervous system needs a break.", "priority": 2, }) # ── CORTISOL / STRESS ──────────────────────────── cortisol = blood.get("Cortisol", {}) if cortisol.get("value", 0) > 12: actions.append({ "category": "stress", "icon": "🧘", "action": "20 min morning sunlight walk. No phone. 5 min box breathing before bed (4-4-4-4).", "reason": f"Cortisol at {cortisol['value']} µg/dL — elevated. Daylight + breathwork are the fastest cortisol reset.", "priority": 3, }) # ── GLUCOSE / NUTRITION ────────────────────────── glucose = blood.get("Glucose", {}) if glucose.get("value", 0) > 99: actions.append({ "category": "nutrition", "icon": "🥗", "action": "No refined carbs today. Start meals with protein + fiber. 15 min walk after lunch and dinner.", "reason": f"Fasting glucose {glucose['value']} mg/dL — borderline. Post-meal walks cut glucose spikes by 30%.", "priority": 4, }) # ── CHOLESTEROL / HEART ────────────────────────── ldl = blood.get("LDL Cholesterol", {}) if ldl.get("value", 0) > 130: actions.append({ "category": "heart", "icon": "❤️", "action": "Hit 35g fiber today. Oatmeal + ground flax at breakfast. Omega-3 (3g). Zero saturated fat.", "reason": f"LDL at {ldl['value']} mg/dL. Soluble fiber + omega-3 are the strongest lifestyle levers.", "priority": 5, }) # ── VITAMIN D ──────────────────────────────────── vitd = blood.get("Vitamin D", {}) if vitd.get("value", 0) < 30: actions.append({ "category": "supplement", "icon": "☀️", "action": "Take Vitamin D3 5000 IU + K2 (100mcg) with your fattiest meal today.", "reason": f"Vitamin D at {vitd['value']} ng/mL — below optimal. Take with fat for absorption. K2 directs calcium properly.", "priority": 6, }) # ── TESTOSTERONE ───────────────────────────────── testosterone = blood.get("Testosterone", {}) if testosterone.get("value", 0) < 450: actions.append({ "category": "hormones", "icon": "⚡", "action": "Zinc (30mg) + Ashwagandha (600mg KSM-66) with dinner. Strength training if readiness allows.", "reason": f"Testosterone at {testosterone['value']} ng/dL — suboptimal. Zinc, ashwagandha, and compound lifts are proven boosters.", "priority": 7, }) # ── MOVEMENT ───────────────────────────────────── steps = profile.get("oura", {}).get("steps", {}) yesterday_steps = steps.get("count", 0) # Check if yesterday was a low-movement day if yesterday_steps < 5000: actions.append({ "category": "movement", "icon": "🚶", "action": "Target 10,000 steps today. Set hourly reminders to move. Take calls walking.", "reason": f"Yesterday: only {yesterday_steps:,} steps. Sedentary days compound — get moving today.", "priority": 8, }) # ── HYDRATION (always relevant) ────────────────── body_water = weight_data.get("body_water") if body_water and body_water < 55: actions.append({ "category": "hydration", "icon": "💧", "action": "3L water today minimum. Start with 500mL before coffee. Add electrolytes to one bottle.", "reason": f"Body water at {body_water}% — on the low side. Hydration affects HRV, recovery, and cognitive function.", "priority": 9, }) # Sort by priority and cap at 6 actions.sort(key=lambda a: a["priority"]) return actions[:6] # ── Assessment Helpers ─────────────────────────────────────── def _assess_sleep(sleep: dict, history: list) -> str: """Generate sleep assessment text.""" hours = sleep.get("hours", 0) deep = sleep.get("deep_min", 0) hrv = sleep.get("avg_hrv") parts = [] if hours >= 7.5: parts.append(f"Good duration ({hours}h)") elif hours >= 7: parts.append(f"Adequate duration ({hours}h)") else: parts.append(f"Short sleep ({hours}h — target 7-8h)") if deep >= 90: parts.append(f"excellent deep sleep ({deep} min)") elif deep >= 60: parts.append(f"decent deep sleep ({deep} min)") elif deep > 0: parts.append(f"low deep sleep ({deep} min — target 60+ min)") if hrv: # Compare to recent average recent_hrvs = [r.get("avg_hrv") for r in history[:7] if r.get("avg_hrv")] if recent_hrvs: avg = sum(recent_hrvs) / len(recent_hrvs) if hrv > avg * 1.1: parts.append(f"HRV above average ({hrv:.0f} vs {avg:.0f} avg)") elif hrv < avg * 0.9: parts.append(f"HRV below average ({hrv:.0f} vs {avg:.0f} avg)") return ". ".join(parts) + "." if parts else "" def _assess_readiness(score: int) -> str: if score >= 85: return f"Excellent readiness ({score}). Body is well-recovered." elif score >= 70: return f"Good readiness ({score}). Normal training day." elif score >= 60: return f"Moderate readiness ({score}). Consider lighter intensity." else: return f"Low readiness ({score}). Recovery day recommended." def _avg(values: list) -> float: if not values: return None return round(sum(values) / len(values), 1) def _trend_direction(values: list) -> str: """Determine if values are trending up, down, or stable.""" if len(values) < 3: return "insufficient_data" # Compare first half avg vs second half avg mid = len(values) // 2 recent = sum(values[:mid]) / mid older = sum(values[mid:]) / (len(values) - mid) pct_change = (recent - older) / older * 100 if older else 0 if pct_change > 5: return "improving" elif pct_change < -5: return "declining" return "stable" def _is_stale(last_sync: str, source: str) -> bool: """Check if a data source is stale.""" if not last_sync: return True try: sync_date = datetime.fromisoformat(last_sync).date() except (ValueError, TypeError): return True # Different staleness thresholds by source thresholds = { "oura": 1, # Should sync daily "blood": 90, # Quarterly tests "genetics": 3650, # Basically never stale "microbiome": 365, # Annual "vitals": 7, # Weekly "cgm": 1, # Daily when active } max_days = thresholds.get(source, 30) return (date.today() - sync_date).days > max_days # ── Output Formatters ──────────────────────────────────────── def format_text(brief: dict) -> str: """Format brief as human-readable text.""" lines = [] lines.append(f"{'='*50}") lines.append(f" KITZU DAILY BRIEF — {brief['date']}") lines.append(f"{'='*50}") lines.append("") # Alerts first for alert in brief.get("alerts", []): icon = "⚠️" if alert["level"] == "warning" else "ℹ️" lines.append(f" {icon} {alert['message']}") if brief.get("alerts"): lines.append("") # Sections for section in brief.get("sections", []): lines.append(f" {section['title'].upper()}") lines.append(f" {'-'*30}") # Key metrics for key, value in section.get("metrics", {}).items(): if value is not None and key != "sessions": label = key.replace("_", " ").title() if isinstance(value, float): lines.append(f" {label}: {value:.1f}") else: lines.append(f" {label}: {value}") # Assessment if section.get("assessment"): lines.append(f" → {section['assessment']}") # Recommendations for rec in section.get("recommendations", []): lines.append(f" • {rec}") lines.append("") # Cross-references (the magic) xrefs = brief.get("cross_references", []) if xrefs: lines.append(f" CROSS-REFERENCE INSIGHTS") lines.append(f" {'-'*30}") for xr in xrefs: sources = " + ".join(xr["sources"]) lines.append(f" [{sources}]") lines.append(f" {xr['finding']}") lines.append(f" → {xr['action']}") lines.append("") # Data freshness lines.append(f" DATA STATUS") lines.append(f" {'-'*30}") for source, info in brief.get("data_freshness", {}).items(): status = "✓ Fresh" if not info["stale"] else "⚠ Stale" if info["last_sync"] else "✗ No data" lines.append(f" {source:<14} {status}") lines.append("") lines.append(f" Generated: {brief['generated_at'][:19]}") lines.append(f"{'='*50}") return "\n".join(lines) def save_brief(brief: dict): """Save brief to the data lake.""" BRIEFS_DIR.mkdir(parents=True, exist_ok=True) path = BRIEFS_DIR / f"{brief['date']}.json" path.write_text(json.dumps(brief, indent=2, default=str)) # ── Main ───────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser(description="Kitzu Daily Health Brief") parser.add_argument("--json", action="store_true", help="Output as JSON") parser.add_argument("--markdown", action="store_true", help="Save as Markdown file") parser.add_argument("--save", action="store_true", help="Save brief to data lake") args = parser.parse_args() brief = generate_brief() if args.json: print(json.dumps(brief, indent=2, default=str)) else: print(format_text(brief)) if args.save or args.markdown: save_brief(brief) brief_date = brief["date"] print(f"\nBrief saved to {BRIEFS_DIR / f'{brief_date}.json'}") if __name__ == "__main__": main()