MeeJayandClaude Opus 5
93f0689c1e
Tracker de vie auto-hébergé : suivi poids/calories/sport avec planning de pesées, sevrage tabac (vape) avec modèle de coût DIY et économies, et finances personnelles avec import de relevés bancaires. Architecture : FastAPI + SQLAlchemy 2.0 + PostgreSQL 16, React 18 + TS + Vite + Tailwind + ECharts, déploiement Docker Compose. Modules auto-découverts des deux côtés (pkgutil / import.meta.glob) et framework de connecteurs à deux voies (importeurs de fichiers + ingestion JSON) pour brancher de nouvelles sources sans toucher au noyau. Validé : 292 tests pytest, tsc + vite build, contrat API/web vérifié contre le schéma OpenAPI, et déploiement Docker réel sur PostgreSQL 16 (28 tables, SPA servie par nginx, wizard de premier démarrage). Documentation : README.md, docs/GUIDE.md, CONVENTIONS.md, et les documents de conception et de recherche dans docs/. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
497 lines
16 KiB
Python
497 lines
16 KiB
Python
"""Pure health/nutrition calculations (datamodel-health-vape.md §5).
|
|
|
|
Every function here is side-effect free and unit-tested with exact values
|
|
(app/tests/modules/test_health_calculations.py). No SQLAlchemy imports.
|
|
"""
|
|
|
|
import math
|
|
from dataclasses import dataclass, field
|
|
from datetime import date, timedelta
|
|
|
|
# Alias for the annotations of dataclasses that own a field *named* `date`
|
|
# (Projection). In a class body Python binds the default value before it
|
|
# evaluates the annotation, so `date: date | None = None` would read the field
|
|
# itself (None) instead of datetime.date and raise at import time on Python
|
|
# 3.12 — the runtime of docker/api.Dockerfile. Python 3.14 only hides the bug
|
|
# by deferring annotation evaluation (PEP 649).
|
|
DateT = date
|
|
|
|
# --- Central constants (datamodel §10) ---------------------------------------
|
|
|
|
KCAL_PER_KG_FAT = 7700 # 1 kg of body mass ~ 7700 kcal
|
|
EMA_ALPHA = 0.1 # weight trend smoothing (Hacker's Diet)
|
|
MIN_SLOPE_KG_PER_DAY = 0.005 # below this, slope is considered null
|
|
ACTIVITY_FACTORS: dict[str, float] = {
|
|
"sedentary": 1.2,
|
|
"light": 1.375,
|
|
"moderate": 1.55,
|
|
"active": 1.725,
|
|
"very_active": 1.9,
|
|
}
|
|
CALORIE_FLOOR_MALE = 1500
|
|
CALORIE_FLOOR_FEMALE = 1200
|
|
WORKOUT_OVERLAP_THRESHOLD = 0.8 # cross-source workout dedup (§3.6)
|
|
ADAPTIVE_TDEE_MIN_DAYS = 21 # minimal tracked window for §5.7
|
|
TDEE_SMOOTHING_DAYS = 7 # moving average window for the budget
|
|
MEASURED_TOTAL_MIN_RATIO = 0.8 # total_kcal plausibility guard vs BMR
|
|
|
|
# Field-by-field merge priority for activity_daily (§3.5). Unknown sources
|
|
# rank after every listed one.
|
|
ACTIVITY_SOURCE_PRIORITY: list[str] = [
|
|
"manual",
|
|
"health_connect",
|
|
"fitshow",
|
|
"csv_import",
|
|
"api",
|
|
]
|
|
|
|
MERGE_FIELDS: tuple[str, ...] = (
|
|
"steps",
|
|
"active_kcal",
|
|
"total_kcal",
|
|
"distance_m",
|
|
"active_minutes",
|
|
"floors",
|
|
)
|
|
|
|
|
|
def source_priority(source: str) -> int:
|
|
"""Rank of a source in the merge order; unknown sources go last."""
|
|
try:
|
|
return ACTIVITY_SOURCE_PRIORITY.index(source)
|
|
except ValueError:
|
|
return len(ACTIVITY_SOURCE_PRIORITY)
|
|
|
|
|
|
# --- BMR / TDEE / budget ------------------------------------------------------
|
|
|
|
|
|
def age_on(day: date, birthdate: date) -> int:
|
|
"""Completed years on `day`."""
|
|
years = day.year - birthdate.year
|
|
if (day.month, day.day) < (birthdate.month, birthdate.day):
|
|
years -= 1
|
|
return years
|
|
|
|
|
|
def bmr_mifflin(weight_kg: float, height_cm: float, age: int, sex: str) -> float:
|
|
"""Mifflin-St Jeor BMR; sex in {male, female, other} (§5.1)."""
|
|
base = 10 * weight_kg + 6.25 * height_cm - 5 * age
|
|
offset = {"male": 5.0, "female": -161.0, "other": -78.0}[sex]
|
|
return base + offset
|
|
|
|
|
|
@dataclass
|
|
class TdeeResult:
|
|
kcal: float
|
|
method: str # "measured_total" | "bmr_plus_active" | "estimated"
|
|
|
|
|
|
def tdee_effective(
|
|
bmr: float,
|
|
activity_level: str,
|
|
total_kcal: float | None = None,
|
|
active_kcal: float | None = None,
|
|
) -> TdeeResult:
|
|
"""Effective TDEE of one day, 3-tier preference (§5.2):
|
|
1. measured total_kcal, if plausible (> 0.8 x BMR);
|
|
2. BMR + active_kcal;
|
|
3. BMR x activity factor.
|
|
"""
|
|
if total_kcal is not None and total_kcal > MEASURED_TOTAL_MIN_RATIO * bmr:
|
|
return TdeeResult(kcal=float(total_kcal), method="measured_total")
|
|
if active_kcal is not None:
|
|
return TdeeResult(kcal=bmr + float(active_kcal), method="bmr_plus_active")
|
|
return TdeeResult(kcal=bmr * ACTIVITY_FACTORS[activity_level], method="estimated")
|
|
|
|
|
|
def moving_average(
|
|
values: list[float | None], window: int = TDEE_SMOOTHING_DAYS
|
|
) -> list[float | None]:
|
|
"""Trailing moving average over consecutive daily values; None entries are
|
|
skipped (a day stays None only when no value exists in its window)."""
|
|
out: list[float | None] = []
|
|
for i in range(len(values)):
|
|
chunk = [v for v in values[max(0, i - window + 1) : i + 1] if v is not None]
|
|
out.append(sum(chunk) / len(chunk) if chunk else None)
|
|
return out
|
|
|
|
|
|
def clamp(value: float, lo: float, hi: float) -> float:
|
|
return max(lo, min(hi, value))
|
|
|
|
|
|
@dataclass
|
|
class GoalRate:
|
|
weekly_rate_kg: float # positive = loss
|
|
rate_clamped: bool = False
|
|
|
|
|
|
def resolve_goal_rate(
|
|
mode: str,
|
|
day: date,
|
|
trend_now: float,
|
|
target_weight_kg: float,
|
|
weekly_rate_kg: float | None,
|
|
target_date: date | None,
|
|
) -> GoalRate:
|
|
"""Weekly rate implied by the active goal on `day` (§5.3)."""
|
|
if mode == "maintain":
|
|
return GoalRate(0.0)
|
|
if mode == "weekly_rate":
|
|
return GoalRate(float(weekly_rate_kg or 0.0))
|
|
# target_date: rate recomputed daily from the current trend.
|
|
if target_date is None:
|
|
return GoalRate(0.0)
|
|
weeks_left = max((target_date - day).days / 7, 1.0)
|
|
rate = (trend_now - target_weight_kg) / weeks_left
|
|
clamped = clamp(rate, -0.5, 1.0)
|
|
return GoalRate(clamped, rate_clamped=clamped != rate)
|
|
|
|
|
|
@dataclass
|
|
class BudgetResult:
|
|
kcal: float
|
|
deficit_target: float
|
|
floor_applied: bool
|
|
rate_clamped: bool = False
|
|
|
|
|
|
def daily_budget(
|
|
tdee_smoothed_kcal: float,
|
|
rate: GoalRate,
|
|
sex: str,
|
|
calorie_floor_kcal: int | None = None,
|
|
) -> BudgetResult:
|
|
"""Daily calorie budget = TDEE - weekly_rate x 7700 / 7, floored (§5.3)."""
|
|
deficit = rate.weekly_rate_kg * KCAL_PER_KG_FAT / 7 # = rate x 1100
|
|
budget = tdee_smoothed_kcal - deficit
|
|
floor = calorie_floor_kcal or (
|
|
CALORIE_FLOOR_MALE if sex == "male" else CALORIE_FLOOR_FEMALE
|
|
)
|
|
return BudgetResult(
|
|
kcal=max(budget, float(floor)),
|
|
deficit_target=deficit,
|
|
floor_applied=budget < floor,
|
|
rate_clamped=rate.rate_clamped,
|
|
)
|
|
|
|
|
|
# --- Weight trend (EMA), slope, projection ------------------------------------
|
|
|
|
|
|
def weight_trend(
|
|
entries: list[tuple[date, float]], alpha: float = EMA_ALPHA
|
|
) -> list[tuple[date, float]]:
|
|
"""EMA trend with gap correction (§5.5).
|
|
|
|
`entries` = one (day, weight) per local day, ascending — the FIRST
|
|
weigh-in of each day. Missing days are handled through
|
|
alpha_eff = 1 - (1 - alpha) ** gap_days.
|
|
Returned trend values are rounded to 2 decimals; the running trend keeps
|
|
full precision.
|
|
"""
|
|
out: list[tuple[date, float]] = []
|
|
prev_date: date | None = None
|
|
trend: float | None = None
|
|
for d, w in entries:
|
|
if trend is None or prev_date is None:
|
|
trend = w
|
|
else:
|
|
gap = (d - prev_date).days
|
|
alpha_eff = 1 - (1 - alpha) ** gap
|
|
trend = trend + alpha_eff * (w - trend)
|
|
out.append((d, round(trend, 2)))
|
|
prev_date = d
|
|
return out
|
|
|
|
|
|
def trend_at_day(trend: list[tuple[date, float]], day: date) -> float | None:
|
|
"""Trend value carried forward to `day` (last known point <= day)."""
|
|
value: float | None = None
|
|
for point_day, point_value in trend:
|
|
if point_day > day:
|
|
break
|
|
value = point_value
|
|
return value
|
|
|
|
|
|
def regression_slope(points: list[tuple[date, float]]) -> float | None:
|
|
"""OLS slope in kg/day over (day, value) points; None if < 3 points (§5.5)."""
|
|
if len(points) < 3:
|
|
return None
|
|
t0 = points[0][0]
|
|
ts = [float((d - t0).days) for d, _ in points]
|
|
ws = [w for _, w in points]
|
|
t_mean = sum(ts) / len(ts)
|
|
w_mean = sum(ws) / len(ws)
|
|
denom = sum((t - t_mean) ** 2 for t in ts)
|
|
if denom == 0:
|
|
return None
|
|
num = sum((t - t_mean) * (w - w_mean) for t, w in zip(ts, ws, strict=True))
|
|
return num / denom
|
|
|
|
|
|
@dataclass
|
|
class Projection:
|
|
status: str # "reached" | "not_converging" | "ok"
|
|
date: DateT | None = None
|
|
|
|
|
|
def project_target_date(
|
|
trend_now: float,
|
|
target_weight: float,
|
|
slope_kg_day: float | None,
|
|
today: date,
|
|
) -> Projection:
|
|
"""Projected date of reaching the target at the observed slope (§5.6)."""
|
|
delta = trend_now - target_weight # > 0 while weight remains to lose
|
|
if abs(delta) < 0.1:
|
|
return Projection(status="reached")
|
|
losing_needed = delta > 0
|
|
if (
|
|
slope_kg_day is None
|
|
or abs(slope_kg_day) < MIN_SLOPE_KG_PER_DAY
|
|
or (losing_needed and slope_kg_day >= 0)
|
|
or (not losing_needed and slope_kg_day <= 0)
|
|
):
|
|
return Projection(status="not_converging")
|
|
days = abs(delta / slope_kg_day)
|
|
if days > 3650:
|
|
return Projection(status="not_converging")
|
|
return Projection(status="ok", date=today + timedelta(days=round(days)))
|
|
|
|
|
|
# --- Energy balance & TDEE calibration (§5.4 / §5.7) --------------------------
|
|
|
|
|
|
def energy_balance(intake_kcal: float | None, tdee_kcal: float | None) -> float | None:
|
|
"""balance = intake - TDEE (negative = deficit); None when untracked."""
|
|
if intake_kcal is None or tdee_kcal is None:
|
|
return None
|
|
return intake_kcal - tdee_kcal
|
|
|
|
|
|
@dataclass
|
|
class CalibrationResult:
|
|
status: str # "ok" | "insufficient_data"
|
|
tracked_days: int = 0
|
|
expected_change_kg: float | None = None
|
|
actual_change_kg: float | None = None
|
|
gap_kg: float | None = None
|
|
tdee_adaptive_kcal: float | None = None
|
|
tdee_correction_kcal: float | None = None
|
|
|
|
|
|
def tdee_calibration(
|
|
tracked: list[tuple[float, float]],
|
|
trend_start: float | None,
|
|
trend_end: float | None,
|
|
) -> CalibrationResult:
|
|
"""Cumulative deficit vs actual (trend) loss + adaptive TDEE (§5.7).
|
|
|
|
`tracked` = [(intake_kcal, tdee_effective_kcal)] for tracked days only
|
|
(days with at least one food entry). Requires >= 21 tracked days.
|
|
"""
|
|
tracked_days = len(tracked)
|
|
if (
|
|
tracked_days < ADAPTIVE_TDEE_MIN_DAYS
|
|
or trend_start is None
|
|
or trend_end is None
|
|
):
|
|
return CalibrationResult(status="insufficient_data", tracked_days=tracked_days)
|
|
expected = sum(intake - tdee for intake, tdee in tracked) / KCAL_PER_KG_FAT
|
|
actual = trend_end - trend_start
|
|
mean_intake = sum(intake for intake, _ in tracked) / tracked_days
|
|
mean_tdee = sum(tdee for _, tdee in tracked) / tracked_days
|
|
tdee_adaptive = mean_intake - actual * KCAL_PER_KG_FAT / tracked_days
|
|
return CalibrationResult(
|
|
status="ok",
|
|
tracked_days=tracked_days,
|
|
expected_change_kg=expected,
|
|
actual_change_kg=actual,
|
|
gap_kg=actual - expected,
|
|
tdee_adaptive_kcal=tdee_adaptive,
|
|
tdee_correction_kcal=tdee_adaptive - mean_tdee,
|
|
)
|
|
|
|
|
|
# --- Activity merge & workout overlap -----------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class MergedActivity:
|
|
date: date
|
|
steps: int | None = None
|
|
active_kcal: float | None = None
|
|
total_kcal: float | None = None
|
|
distance_m: int | None = None
|
|
active_minutes: int | None = None
|
|
floors: int | None = None
|
|
field_sources: dict[str, str] = field(default_factory=dict)
|
|
|
|
|
|
def merge_activity_day(rows: list) -> MergedActivity:
|
|
"""Field-by-field merge of one day's rows by source priority (§3.5).
|
|
|
|
`rows` = every activity row of the same (user, date), any source order;
|
|
objects only need `.date`, `.source` and the MERGE_FIELDS attributes.
|
|
Never sums two sources (double-count risk).
|
|
"""
|
|
ordered = sorted(rows, key=lambda r: (source_priority(r.source), r.source))
|
|
merged = MergedActivity(date=ordered[0].date)
|
|
for field_name in MERGE_FIELDS:
|
|
for row in ordered:
|
|
value = getattr(row, field_name)
|
|
if value is not None:
|
|
if field_name in ("active_kcal", "total_kcal"):
|
|
value = float(value)
|
|
setattr(merged, field_name, value)
|
|
merged.field_sources[field_name] = row.source
|
|
break
|
|
return merged
|
|
|
|
|
|
def overlap_ratio(a_start, a_end, b_start, b_end) -> float:
|
|
"""Overlap seconds / duration of the SHORTER interval (datetimes)."""
|
|
inter = (min(a_end, b_end) - max(a_start, b_start)).total_seconds()
|
|
if inter <= 0:
|
|
return 0.0
|
|
shorter = min((a_end - a_start).total_seconds(), (b_end - b_start).total_seconds())
|
|
if shorter <= 0:
|
|
return 0.0
|
|
return inter / shorter
|
|
|
|
|
|
# --- Body composition ---------------------------------------------------------
|
|
|
|
|
|
def bmi(weight_kg: float, height_cm: float) -> float | None:
|
|
"""Body mass index = kg / m²."""
|
|
if height_cm <= 0:
|
|
return None
|
|
height_m = height_cm / 100
|
|
return weight_kg / (height_m * height_m)
|
|
|
|
|
|
def navy_body_fat_pct(
|
|
sex: str,
|
|
height_cm: float,
|
|
waist_cm: float | None,
|
|
neck_cm: float | None,
|
|
hips_cm: float | None = None,
|
|
) -> float | None:
|
|
"""US Navy body-fat estimate (§3.3); None when inputs are missing."""
|
|
if not waist_cm or not neck_cm or height_cm <= 0:
|
|
return None
|
|
if sex == "female":
|
|
if not hips_cm:
|
|
return None
|
|
inner = waist_cm + hips_cm - neck_cm
|
|
if inner <= 0:
|
|
return None
|
|
denom = 1.29579 - 0.35004 * math.log10(inner) + 0.22100 * math.log10(height_cm)
|
|
else:
|
|
inner = waist_cm - neck_cm
|
|
if inner <= 0:
|
|
return None
|
|
denom = 1.0324 - 0.19077 * math.log10(inner) + 0.15456 * math.log10(height_cm)
|
|
if denom == 0:
|
|
return None
|
|
return 495 / denom - 450
|
|
|
|
|
|
# --- Planning / adherence (addendum-planning.md) ------------------------------
|
|
|
|
# Habit kinds; "done" is DERIVED from the data tables, there is no check-in
|
|
# table in v1: weigh_in -> WeightEntry, workout -> Workout, food_log -> FoodEntry.
|
|
SCHEDULE_KINDS: tuple[str, ...] = ("weigh_in", "workout", "food_log")
|
|
|
|
|
|
@dataclass
|
|
class DayAdherence:
|
|
day: date
|
|
planned: bool
|
|
done: bool
|
|
|
|
@property
|
|
def missed(self) -> bool:
|
|
return self.planned and not self.done
|
|
|
|
@property
|
|
def status(self) -> str:
|
|
"""Calendar-heatmap status: planned+done / planned+missed / off-plan."""
|
|
if self.planned:
|
|
return "done" if self.done else "missed"
|
|
return "done_unplanned" if self.done else "rest"
|
|
|
|
|
|
def is_planned(weekdays: list[int] | None, day: date, enabled: bool = True) -> bool:
|
|
"""True when `day` falls on a planned weekday (0 = Monday … 6 = Sunday)."""
|
|
if not enabled or not weekdays:
|
|
return False
|
|
return day.weekday() in set(weekdays)
|
|
|
|
|
|
def date_range(start: date, end: date) -> list[date]:
|
|
"""Inclusive list of local days from `start` to `end`."""
|
|
if end < start:
|
|
return []
|
|
return [start + timedelta(days=i) for i in range((end - start).days + 1)]
|
|
|
|
|
|
def build_adherence(
|
|
start: date,
|
|
end: date,
|
|
weekdays: list[int] | None,
|
|
enabled: bool,
|
|
done_days: set[date],
|
|
) -> list[DayAdherence]:
|
|
"""Per-day planned/done status over an inclusive local-day range."""
|
|
return [
|
|
DayAdherence(
|
|
day=day,
|
|
planned=is_planned(weekdays, day, enabled),
|
|
done=day in done_days,
|
|
)
|
|
for day in date_range(start, end)
|
|
]
|
|
|
|
|
|
def adherence_pct(days: list[DayAdherence]) -> float | None:
|
|
"""done ÷ planned, in percent; None when nothing was planned."""
|
|
planned = [d for d in days if d.planned]
|
|
if not planned:
|
|
return None
|
|
return 100.0 * sum(1 for d in planned if d.done) / len(planned)
|
|
|
|
|
|
@dataclass
|
|
class StreakResult:
|
|
current: int = 0
|
|
best: int = 0
|
|
|
|
|
|
def compute_streaks(days: list[DayAdherence], today: date) -> StreakResult:
|
|
"""Streaks over PLANNED days only (addendum-planning.md).
|
|
|
|
`best` = longest run of consecutive planned-and-done days. `current` = run
|
|
ending at the most recent planned day; a planned day equal to `today` that
|
|
is not done yet does not break the streak (the day is not over).
|
|
"""
|
|
planned = sorted((d for d in days if d.planned), key=lambda d: d.day)
|
|
best = run = 0
|
|
for entry in planned:
|
|
run = run + 1 if entry.done else 0
|
|
best = max(best, run)
|
|
current = 0
|
|
for entry in reversed(planned):
|
|
if entry.day > today:
|
|
continue # future planned days are not part of the current streak
|
|
if entry.day == today and not entry.done:
|
|
continue # grace period: today is still open
|
|
if not entry.done:
|
|
break
|
|
current += 1
|
|
return StreakResult(current=current, best=best)
|