508 lines
18 KiB
Python
508 lines
18 KiB
Python
"""Weekly fitness chart data and least-squares trend lines."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from collections import defaultdict
|
||
from datetime import date, datetime, timedelta, timezone
|
||
from typing import Any
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.db.models import BodyMetric, FitnessProfile, FoodLog, StepLog, WaterLog, WorkoutLog
|
||
from app.fitness.activity_budget import estimate_workout_active_kcal
|
||
from app.fitness.calculators import (
|
||
EXPECTED_LOOKBACK_DAYS,
|
||
compute_daily_targets,
|
||
compute_expected_targets,
|
||
)
|
||
|
||
METRIC_DEFS: dict[str, dict[str, str]] = {
|
||
"weight_kg": {"label": "Вес", "unit": "кг"},
|
||
"body_fat_pct": {"label": "Жир", "unit": "%"},
|
||
"calories": {"label": "Калории", "unit": "ккал/день"},
|
||
"tdee": {"label": "TDEE факт", "unit": "ккал/день"},
|
||
"tdee_expected": {"label": "TDEE план", "unit": "ккал/день"},
|
||
"protein_g": {"label": "Белок", "unit": "г/день"},
|
||
"water_l": {"label": "Вода", "unit": "л/день"},
|
||
"steps": {"label": "Шаги", "unit": "шаг/день"},
|
||
}
|
||
|
||
|
||
def week_start(day: date) -> date:
|
||
return day - timedelta(days=day.weekday())
|
||
|
||
|
||
def linear_regression(points: list[tuple[float, float]]) -> dict[str, float] | None:
|
||
"""Ordinary least squares y = slope * x + intercept."""
|
||
n = len(points)
|
||
if n < 2:
|
||
return None
|
||
sum_x = sum(x for x, _ in points)
|
||
sum_y = sum(y for _, y in points)
|
||
sum_xx = sum(x * x for x, _ in points)
|
||
sum_xy = sum(x * y for x, y in points)
|
||
denom = n * sum_xx - sum_x * sum_x
|
||
if abs(denom) < 1e-12:
|
||
return None
|
||
slope = (n * sum_xy - sum_x * sum_y) / denom
|
||
intercept = (sum_y - slope * sum_x) / n
|
||
return {"slope": slope, "intercept": intercept}
|
||
|
||
|
||
def _avg(values: list[float]) -> float | None:
|
||
if not values:
|
||
return None
|
||
return sum(values) / len(values)
|
||
|
||
|
||
def _last(values: list[tuple[date, float]]) -> float | None:
|
||
if not values:
|
||
return None
|
||
values.sort(key=lambda item: item[0])
|
||
return values[-1][1]
|
||
|
||
|
||
def _profile_for_charts(row: FitnessProfile | None) -> dict[str, float | int | str | None] | None:
|
||
if row is None:
|
||
return None
|
||
return {
|
||
"sex": row.sex,
|
||
"age": row.age,
|
||
"height_cm": row.height_cm,
|
||
"weight_kg": row.weight_kg,
|
||
"goal": row.goal,
|
||
"neat_base_kcal": row.neat_base_kcal,
|
||
"activity_level": row.activity_level,
|
||
"weekly_workouts": row.weekly_workouts,
|
||
"baseline_steps": row.baseline_steps,
|
||
"baseline_workout_kcal": row.baseline_workout_kcal,
|
||
}
|
||
|
||
|
||
def _load_activity_maps(
|
||
db: Session,
|
||
user_id: int,
|
||
range_start: datetime,
|
||
range_end: datetime,
|
||
weight_kg: float,
|
||
) -> tuple[dict[date, int], dict[date, float]]:
|
||
steps_by_day: dict[date, int] = defaultdict(int)
|
||
workout_kcal_by_day: dict[date, float] = defaultdict(float)
|
||
|
||
steps_rows = db.scalars(
|
||
select(StepLog).where(
|
||
StepLog.user_id == user_id,
|
||
StepLog.logged_at >= range_start,
|
||
StepLog.logged_at <= range_end,
|
||
)
|
||
).all()
|
||
for row in steps_rows:
|
||
steps_by_day[row.logged_at.date()] += row.steps
|
||
|
||
workouts_rows = db.scalars(
|
||
select(WorkoutLog).where(
|
||
WorkoutLog.user_id == user_id,
|
||
WorkoutLog.logged_at >= range_start,
|
||
WorkoutLog.logged_at <= range_end,
|
||
)
|
||
).all()
|
||
for row in workouts_rows:
|
||
d = row.logged_at.date()
|
||
workout_kcal_by_day[d] += estimate_workout_active_kcal(
|
||
{
|
||
"title": row.title,
|
||
"duration_min": row.duration_min,
|
||
"active_calories": row.active_calories,
|
||
},
|
||
weight_kg=weight_kg,
|
||
)
|
||
return steps_by_day, workout_kcal_by_day
|
||
|
||
|
||
def _activity_history_before(
|
||
day: date,
|
||
steps_by_day: dict[date, int],
|
||
workout_kcal_by_day: dict[date, float],
|
||
*,
|
||
days: int = EXPECTED_LOOKBACK_DAYS,
|
||
) -> list[dict[str, float | int]]:
|
||
history: list[dict[str, float | int]] = []
|
||
start = day - timedelta(days=days)
|
||
cursor = start
|
||
while cursor < day:
|
||
history.append(
|
||
{
|
||
"steps": steps_by_day.get(cursor, 0),
|
||
"workout_kcal": workout_kcal_by_day.get(cursor, 0.0),
|
||
}
|
||
)
|
||
cursor += timedelta(days=1)
|
||
return history
|
||
|
||
|
||
def _tdee_actual_for_day(
|
||
profile: dict[str, float | int | str | None],
|
||
steps_by_day: dict[date, int],
|
||
workout_kcal_by_day: dict[date, float],
|
||
day: date,
|
||
) -> float:
|
||
steps = steps_by_day.get(day, 0)
|
||
workout_kcal = workout_kcal_by_day.get(day, 0.0)
|
||
workouts = [{"active_calories": workout_kcal}] if workout_kcal > 0 else []
|
||
return float(compute_daily_targets(profile, steps_total=steps, workouts=workouts)["tdee"])
|
||
|
||
|
||
def _tdee_expected_for_day(
|
||
profile: dict[str, float | int | str | None],
|
||
steps_by_day: dict[date, int],
|
||
workout_kcal_by_day: dict[date, float],
|
||
day: date,
|
||
) -> float:
|
||
history = _activity_history_before(day, steps_by_day, workout_kcal_by_day)
|
||
return float(compute_expected_targets(profile, history=history)["tdee"])
|
||
|
||
|
||
def build_fitness_charts(
|
||
db: Session,
|
||
user_id: int,
|
||
*,
|
||
weeks: int = 52,
|
||
trend: bool = True,
|
||
end_day: date | None = None,
|
||
) -> dict[str, Any]:
|
||
weeks = max(4, min(int(weeks), 52))
|
||
end = end_day or datetime.now(timezone.utc).date()
|
||
last_week_start = week_start(end)
|
||
first_week_start = last_week_start - timedelta(weeks=weeks - 1)
|
||
|
||
range_start = datetime.combine(first_week_start, datetime.min.time(), tzinfo=timezone.utc)
|
||
range_end = datetime.combine(end, datetime.max.time(), tzinfo=timezone.utc)
|
||
|
||
profile_row = db.scalar(
|
||
select(FitnessProfile).where(FitnessProfile.user_id == user_id).limit(1)
|
||
)
|
||
profile = _profile_for_charts(profile_row)
|
||
weight_kg = float(profile["weight_kg"]) if profile else 70.0
|
||
|
||
activity_start = datetime.combine(
|
||
first_week_start - timedelta(days=EXPECTED_LOOKBACK_DAYS),
|
||
datetime.min.time(),
|
||
tzinfo=timezone.utc,
|
||
)
|
||
steps_by_day, workout_kcal_by_day = _load_activity_maps(
|
||
db,
|
||
user_id,
|
||
activity_start,
|
||
range_end,
|
||
weight_kg,
|
||
)
|
||
|
||
daily: dict[date, dict[str, float]] = defaultdict(lambda: {
|
||
"calories": 0.0,
|
||
"protein_g": 0.0,
|
||
"fat_g": 0.0,
|
||
"carbs_g": 0.0,
|
||
"water_ml": 0.0,
|
||
"steps": 0.0,
|
||
})
|
||
daily_flags: dict[date, set[str]] = defaultdict(set)
|
||
|
||
foods = db.scalars(
|
||
select(FoodLog).where(
|
||
FoodLog.user_id == user_id,
|
||
FoodLog.logged_at >= range_start,
|
||
FoodLog.logged_at <= range_end,
|
||
)
|
||
).all()
|
||
for row in foods:
|
||
d = row.logged_at.date()
|
||
daily[d]["calories"] += row.calories
|
||
daily[d]["protein_g"] += row.protein_g
|
||
daily[d]["fat_g"] += row.fat_g
|
||
daily[d]["carbs_g"] += row.carbs_g
|
||
daily_flags[d].add("nutrition")
|
||
|
||
waters = db.scalars(
|
||
select(WaterLog).where(
|
||
WaterLog.user_id == user_id,
|
||
WaterLog.logged_at >= range_start,
|
||
WaterLog.logged_at <= range_end,
|
||
)
|
||
).all()
|
||
for row in waters:
|
||
d = row.logged_at.date()
|
||
daily[d]["water_ml"] += float(row.amount_ml)
|
||
daily_flags[d].add("water")
|
||
|
||
steps_rows = db.scalars(
|
||
select(StepLog).where(
|
||
StepLog.user_id == user_id,
|
||
StepLog.logged_at >= range_start,
|
||
StepLog.logged_at <= range_end,
|
||
)
|
||
).all()
|
||
for row in steps_rows:
|
||
d = row.logged_at.date()
|
||
daily[d]["steps"] += float(row.steps)
|
||
daily_flags[d].add("steps")
|
||
|
||
body_rows = db.scalars(
|
||
select(BodyMetric).where(
|
||
BodyMetric.user_id == user_id,
|
||
BodyMetric.recorded_at >= range_start,
|
||
BodyMetric.recorded_at <= range_end,
|
||
)
|
||
).all()
|
||
body_by_day: dict[date, list[tuple[date, float, float | None]]] = defaultdict(list)
|
||
for row in body_rows:
|
||
d = row.recorded_at.date()
|
||
body_by_day[d].append((d, row.weight_kg, row.body_fat_pct))
|
||
daily_flags[d].add("body")
|
||
|
||
week_slots: list[dict[str, Any]] = []
|
||
cursor = first_week_start
|
||
while cursor <= last_week_start:
|
||
week_slots.append(
|
||
{
|
||
"week_start": cursor.isoformat(),
|
||
"week_end": (cursor + timedelta(days=6)).isoformat(),
|
||
}
|
||
)
|
||
cursor += timedelta(weeks=1)
|
||
|
||
days_with_data = len(daily_flags)
|
||
weeks_with_data = 0
|
||
|
||
def rollup_week(metric: str) -> list[dict[str, Any]]:
|
||
nonlocal weeks_with_data
|
||
points: list[dict[str, Any]] = []
|
||
local_weeks_with_data = 0
|
||
|
||
for idx, slot in enumerate(week_slots):
|
||
ws = date.fromisoformat(slot["week_start"])
|
||
we = date.fromisoformat(slot["week_end"])
|
||
day_cursor = ws
|
||
week_daily_values: list[float] = []
|
||
week_body_weight: list[tuple[date, float]] = []
|
||
week_body_fat: list[tuple[date, float]] = []
|
||
|
||
while day_cursor <= we:
|
||
if day_cursor > end:
|
||
break
|
||
flags = daily_flags.get(day_cursor, set())
|
||
totals = daily.get(day_cursor)
|
||
if metric == "weight_kg":
|
||
for _, w, _ in body_by_day.get(day_cursor, []):
|
||
week_body_weight.append((day_cursor, w))
|
||
elif metric == "body_fat_pct":
|
||
for _, _, bf in body_by_day.get(day_cursor, []):
|
||
if bf is not None:
|
||
week_body_fat.append((day_cursor, bf))
|
||
elif metric == "calories" and totals and "nutrition" in flags:
|
||
week_daily_values.append(totals["calories"])
|
||
elif metric == "protein_g" and totals and "nutrition" in flags:
|
||
week_daily_values.append(totals["protein_g"])
|
||
elif metric == "water_l" and totals and "water" in flags:
|
||
week_daily_values.append(totals["water_ml"] / 1000.0)
|
||
elif metric == "steps" and totals and "steps" in flags:
|
||
week_daily_values.append(totals["steps"])
|
||
elif metric == "tdee" and profile is not None and day_cursor <= end:
|
||
week_daily_values.append(
|
||
_tdee_actual_for_day(profile, steps_by_day, workout_kcal_by_day, day_cursor)
|
||
)
|
||
elif metric == "tdee_expected" and profile is not None and day_cursor <= end:
|
||
week_daily_values.append(
|
||
_tdee_expected_for_day(profile, steps_by_day, workout_kcal_by_day, day_cursor)
|
||
)
|
||
day_cursor += timedelta(days=1)
|
||
|
||
value: float | None
|
||
days_in_week = 0
|
||
if metric == "weight_kg":
|
||
value = _last(week_body_weight)
|
||
days_in_week = len(week_body_weight)
|
||
elif metric == "body_fat_pct":
|
||
value = _last(week_body_fat)
|
||
days_in_week = len(week_body_fat)
|
||
else:
|
||
value = _avg(week_daily_values)
|
||
days_in_week = len(week_daily_values)
|
||
|
||
has_data = value is not None
|
||
if has_data:
|
||
local_weeks_with_data += 1
|
||
|
||
points.append(
|
||
{
|
||
"index": idx,
|
||
"week_start": slot["week_start"],
|
||
"week_end": slot["week_end"],
|
||
"value": round(value, 2) if value is not None else None,
|
||
"days_with_data": days_in_week,
|
||
"has_data": has_data,
|
||
}
|
||
)
|
||
|
||
weeks_with_data = max(weeks_with_data, local_weeks_with_data)
|
||
return points
|
||
|
||
series: dict[str, Any] = {}
|
||
for key, meta in METRIC_DEFS.items():
|
||
points = rollup_week(key)
|
||
reg_points = [(float(p["index"]), float(p["value"])) for p in points if p["has_data"] and p["value"] is not None]
|
||
trend_payload: dict[str, Any] | None = None
|
||
if trend and len(reg_points) >= 2:
|
||
fit = linear_regression(reg_points)
|
||
if fit:
|
||
line = [
|
||
{
|
||
"index": p["index"],
|
||
"week_start": p["week_start"],
|
||
"value": round(fit["slope"] * p["index"] + fit["intercept"], 2),
|
||
}
|
||
for p in points
|
||
]
|
||
trend_payload = {
|
||
"slope_per_week": round(fit["slope"], 4),
|
||
"intercept": round(fit["intercept"], 2),
|
||
"points_with_data": len(reg_points),
|
||
"line": line,
|
||
}
|
||
series[key] = {
|
||
"key": key,
|
||
"label": meta["label"],
|
||
"unit": meta["unit"],
|
||
"points": points,
|
||
"trend": trend_payload,
|
||
"data_points": sum(1 for p in points if p["has_data"]),
|
||
}
|
||
|
||
use_daily = days_with_data > 0 and days_with_data <= 14 and weeks_with_data <= 2
|
||
daily_series: dict[str, Any] | None = None
|
||
if use_daily:
|
||
daily_series = _build_daily_series(
|
||
daily,
|
||
daily_flags,
|
||
body_by_day,
|
||
end,
|
||
trend=trend,
|
||
lookback_days=min(30, max(days_with_data, 7)),
|
||
profile=profile,
|
||
steps_by_day=steps_by_day,
|
||
workout_kcal_by_day=workout_kcal_by_day,
|
||
)
|
||
|
||
return {
|
||
"end_date": end.isoformat(),
|
||
"weeks": weeks,
|
||
"granularity": "day" if use_daily else "week",
|
||
"first_week_start": first_week_start.isoformat(),
|
||
"last_week_start": last_week_start.isoformat(),
|
||
"days_with_data": days_with_data,
|
||
"weeks_with_data": weeks_with_data,
|
||
"series": series,
|
||
"daily_series": daily_series,
|
||
}
|
||
|
||
|
||
def _build_daily_series(
|
||
daily: dict[date, dict[str, float]],
|
||
daily_flags: dict[date, set[str]],
|
||
body_by_day: dict[date, list[tuple[date, float, float | None]]],
|
||
end: date,
|
||
*,
|
||
trend: bool,
|
||
lookback_days: int,
|
||
profile: dict[str, float | int | str | None] | None = None,
|
||
steps_by_day: dict[date, int] | None = None,
|
||
workout_kcal_by_day: dict[date, float] | None = None,
|
||
) -> dict[str, Any]:
|
||
start = end - timedelta(days=lookback_days - 1)
|
||
day_points: list[date] = []
|
||
cursor = start
|
||
while cursor <= end:
|
||
day_points.append(cursor)
|
||
cursor += timedelta(days=1)
|
||
|
||
result: dict[str, Any] = {}
|
||
for key, meta in METRIC_DEFS.items():
|
||
points: list[dict[str, Any]] = []
|
||
for idx, d in enumerate(day_points):
|
||
value: float | None = None
|
||
has_data = False
|
||
if key == "weight_kg":
|
||
body = body_by_day.get(d, [])
|
||
pairs = [(x, w) for x, w, _ in body]
|
||
value = _last(pairs) if pairs else None
|
||
has_data = value is not None
|
||
elif key == "body_fat_pct":
|
||
fat_vals = [(x, bf) for x, _, bf in body_by_day.get(d, []) if bf is not None]
|
||
value = _last(fat_vals) if fat_vals else None
|
||
has_data = value is not None
|
||
else:
|
||
flags = daily_flags.get(d, set())
|
||
totals = daily.get(d)
|
||
if key == "calories" and totals and "nutrition" in flags:
|
||
value = totals["calories"]
|
||
has_data = True
|
||
elif key == "protein_g" and totals and "nutrition" in flags:
|
||
value = totals["protein_g"]
|
||
has_data = True
|
||
elif key == "water_l" and totals and "water" in flags:
|
||
value = totals["water_ml"] / 1000.0
|
||
has_data = True
|
||
elif key == "steps" and totals and "steps" in flags:
|
||
value = totals["steps"]
|
||
has_data = True
|
||
elif key == "tdee" and profile is not None and steps_by_day is not None and workout_kcal_by_day is not None:
|
||
value = _tdee_actual_for_day(profile, steps_by_day, workout_kcal_by_day, d)
|
||
has_data = True
|
||
elif (
|
||
key == "tdee_expected"
|
||
and profile is not None
|
||
and steps_by_day is not None
|
||
and workout_kcal_by_day is not None
|
||
):
|
||
value = _tdee_expected_for_day(profile, steps_by_day, workout_kcal_by_day, d)
|
||
has_data = True
|
||
|
||
points.append(
|
||
{
|
||
"index": idx,
|
||
"date": d.isoformat(),
|
||
"value": round(value, 2) if value is not None else None,
|
||
"has_data": has_data,
|
||
}
|
||
)
|
||
|
||
reg_points = [(float(p["index"]), float(p["value"])) for p in points if p["has_data"] and p["value"] is not None]
|
||
trend_payload: dict[str, Any] | None = None
|
||
if trend and len(reg_points) >= 2:
|
||
fit = linear_regression(reg_points)
|
||
if fit:
|
||
trend_payload = {
|
||
"slope_per_day": round(fit["slope"], 4),
|
||
"intercept": round(fit["intercept"], 2),
|
||
"points_with_data": len(reg_points),
|
||
"line": [
|
||
{
|
||
"index": p["index"],
|
||
"date": p["date"],
|
||
"value": round(fit["slope"] * p["index"] + fit["intercept"], 2),
|
||
}
|
||
for p in points
|
||
],
|
||
}
|
||
|
||
result[key] = {
|
||
"key": key,
|
||
"label": meta["label"],
|
||
"unit": meta["unit"],
|
||
"points": points,
|
||
"trend": trend_payload,
|
||
"data_points": sum(1 for p in points if p["has_data"]),
|
||
}
|
||
|
||
return result
|