lifetrack/apps/api/app/modules/health/calculations.py

"""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
PROTEIN_G_PER_KG_DEFAULT = 1.6  # default protein target (ux-pages.md §15.3)

# 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 protein_goal_g(
    weight_kg: float | None, g_per_kg: float = PROTEIN_G_PER_KG_DEFAULT
) -> float | None:
    """Daily protein target in grams (ux-pages.md §15.3: g/kg of body weight)."""
    if weight_kg is None or weight_kg <= 0:
        return None
    return round(weight_kg * g_per_kg, 1)


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)