From e1d7670971a45c5a8db339729c31b8b8a985d213 Mon Sep 17 00:00:00 2001 From: Lars Date: Sat, 28 Mar 2026 18:45:24 +0100 Subject: [PATCH] feat: Phase 0c - nutrition_metrics.py module complete Data Layer: - get_nutrition_average_data() - all macros in one call - get_nutrition_days_data() - coverage tracking - get_protein_targets_data() - 1.6g/kg and 2.2g/kg targets - get_energy_balance_data() - deficit/surplus/maintenance - get_protein_adequacy_data() - 0-100 score - get_macro_consistency_data() - 0-100 score Placeholder Layer: - get_nutrition_avg() - refactored to use data layer - get_nutrition_days() - refactored to use data layer - get_protein_ziel_low() - refactored to use data layer - get_protein_ziel_high() - refactored to use data layer All 6 nutrition data functions + 4 placeholder refactors complete. Co-Authored-By: Claude Opus 4.6 --- backend/data_layer/__init__.py | 10 +- backend/data_layer/nutrition_metrics.py | 482 ++++++++++++++++++++++++ backend/placeholder_resolver.py | 113 +++--- 3 files changed, 548 insertions(+), 57 deletions(-) create mode 100644 backend/data_layer/nutrition_metrics.py diff --git a/backend/data_layer/__init__.py b/backend/data_layer/__init__.py index d2a7051..b75b517 100644 --- a/backend/data_layer/__init__.py +++ b/backend/data_layer/__init__.py @@ -30,9 +30,9 @@ from .utils import * # Metric modules from .body_metrics import * +from .nutrition_metrics import * # Future imports (will be added as modules are created): -# from .nutrition_metrics import * # from .activity_metrics import * # from .recovery_metrics import * # from .health_metrics import * @@ -48,4 +48,12 @@ __all__ = [ 'get_weight_trend_data', 'get_body_composition_data', 'get_circumference_summary_data', + + # Nutrition Metrics + 'get_nutrition_average_data', + 'get_nutrition_days_data', + 'get_protein_targets_data', + 'get_energy_balance_data', + 'get_protein_adequacy_data', + 'get_macro_consistency_data', ] diff --git a/backend/data_layer/nutrition_metrics.py b/backend/data_layer/nutrition_metrics.py new file mode 100644 index 0000000..492a95c --- /dev/null +++ b/backend/data_layer/nutrition_metrics.py @@ -0,0 +1,482 @@ +""" +Nutrition Metrics Data Layer + +Provides structured data for nutrition tracking and analysis. + +Functions: + - get_nutrition_average_data(): Average calor + +ies, protein, carbs, fat + - get_nutrition_days_data(): Number of days with nutrition data + - get_protein_targets_data(): Protein targets based on weight + - get_energy_balance_data(): Energy balance calculation + - get_protein_adequacy_data(): Protein adequacy score + - get_macro_consistency_data(): Macro consistency analysis + +All functions return structured data (dict) without formatting. +Use placeholder_resolver.py for formatted strings for AI. + +Phase 0c: Multi-Layer Architecture +Version: 1.0 +""" + +from typing import Dict, List, Optional +from datetime import datetime, timedelta, date +from db import get_db, get_cursor, r2d +from data_layer.utils import calculate_confidence, safe_float, safe_int + + +def get_nutrition_average_data( + profile_id: str, + days: int = 30 +) -> Dict: + """ + Get average nutrition values for all macros. + + Args: + profile_id: User profile ID + days: Analysis window (default 30) + + Returns: + { + "kcal_avg": float, + "protein_avg": float, + "carbs_avg": float, + "fat_avg": float, + "data_points": int, + "confidence": str, + "days_analyzed": int + } + + Migration from Phase 0b: + OLD: get_nutrition_avg(pid, field, days) per field + NEW: All macros in one call + """ + with get_db() as conn: + cur = get_cursor(conn) + cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') + + cur.execute( + """SELECT + AVG(kcal) as kcal_avg, + AVG(protein_g) as protein_avg, + AVG(carbs_g) as carbs_avg, + AVG(fat_g) as fat_avg, + COUNT(*) as data_points + FROM nutrition_log + WHERE profile_id=%s AND date >= %s""", + (profile_id, cutoff) + ) + row = cur.fetchone() + + if not row or row['data_points'] == 0: + return { + "kcal_avg": 0.0, + "protein_avg": 0.0, + "carbs_avg": 0.0, + "fat_avg": 0.0, + "data_points": 0, + "confidence": "insufficient", + "days_analyzed": days + } + + data_points = row['data_points'] + confidence = calculate_confidence(data_points, days, "general") + + return { + "kcal_avg": safe_float(row['kcal_avg']), + "protein_avg": safe_float(row['protein_avg']), + "carbs_avg": safe_float(row['carbs_avg']), + "fat_avg": safe_float(row['fat_avg']), + "data_points": data_points, + "confidence": confidence, + "days_analyzed": days + } + + +def get_nutrition_days_data( + profile_id: str, + days: int = 30 +) -> Dict: + """ + Count days with nutrition data. + + Args: + profile_id: User profile ID + days: Analysis window (default 30) + + Returns: + { + "days_with_data": int, + "days_analyzed": int, + "coverage_pct": float, + "confidence": str + } + """ + with get_db() as conn: + cur = get_cursor(conn) + cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') + + cur.execute( + """SELECT COUNT(DISTINCT date) as days + FROM nutrition_log + WHERE profile_id=%s AND date >= %s""", + (profile_id, cutoff) + ) + row = cur.fetchone() + days_with_data = row['days'] if row else 0 + + coverage_pct = (days_with_data / days * 100) if days > 0 else 0 + confidence = calculate_confidence(days_with_data, days, "general") + + return { + "days_with_data": days_with_data, + "days_analyzed": days, + "coverage_pct": coverage_pct, + "confidence": confidence + } + + +def get_protein_targets_data( + profile_id: str +) -> Dict: + """ + Calculate protein targets based on current weight. + + Targets: + - Low: 1.6 g/kg (maintenance) + - High: 2.2 g/kg (muscle building) + + Args: + profile_id: User profile ID + + Returns: + { + "current_weight": float, + "protein_target_low": float, # 1.6 g/kg + "protein_target_high": float, # 2.2 g/kg + "confidence": str + } + """ + with get_db() as conn: + cur = get_cursor(conn) + cur.execute( + """SELECT weight FROM weight_log + WHERE profile_id=%s ORDER BY date DESC LIMIT 1""", + (profile_id,) + ) + row = cur.fetchone() + + if not row: + return { + "current_weight": 0.0, + "protein_target_low": 0.0, + "protein_target_high": 0.0, + "confidence": "insufficient" + } + + weight = safe_float(row['weight']) + + return { + "current_weight": weight, + "protein_target_low": weight * 1.6, + "protein_target_high": weight * 2.2, + "confidence": "high" + } + + +def get_energy_balance_data( + profile_id: str, + days: int = 7 +) -> Dict: + """ + Calculate energy balance (intake - estimated expenditure). + + Note: This is a simplified calculation. + For accurate TDEE, use profile-based calculations. + + Args: + profile_id: User profile ID + days: Analysis window (default 7) + + Returns: + { + "energy_balance": float, # kcal/day (negative = deficit) + "avg_intake": float, + "estimated_tdee": float, + "status": str, # "deficit" | "surplus" | "maintenance" + "confidence": str, + "days_analyzed": int, + "data_points": int + } + """ + with get_db() as conn: + cur = get_cursor(conn) + cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') + + # Get average intake + cur.execute( + """SELECT AVG(kcal) as avg_kcal, COUNT(*) as cnt + FROM nutrition_log + WHERE profile_id=%s AND date >= %s AND kcal IS NOT NULL""", + (profile_id, cutoff) + ) + row = cur.fetchone() + + if not row or row['cnt'] == 0: + return { + "energy_balance": 0.0, + "avg_intake": 0.0, + "estimated_tdee": 0.0, + "status": "unknown", + "confidence": "insufficient", + "days_analyzed": days, + "data_points": 0 + } + + avg_intake = safe_float(row['avg_kcal']) + data_points = row['cnt'] + + # Simple TDEE estimation (this should be improved with profile data) + # For now, use a rough estimate: 2500 kcal for average adult + estimated_tdee = 2500.0 # TODO: Calculate from profile (weight, height, age, activity) + + energy_balance = avg_intake - estimated_tdee + + # Determine status + if energy_balance < -200: + status = "deficit" + elif energy_balance > 200: + status = "surplus" + else: + status = "maintenance" + + confidence = calculate_confidence(data_points, days, "general") + + return { + "energy_balance": energy_balance, + "avg_intake": avg_intake, + "estimated_tdee": estimated_tdee, + "status": status, + "confidence": confidence, + "days_analyzed": days, + "data_points": data_points + } + + +def get_protein_adequacy_data( + profile_id: str, + days: int = 28 +) -> Dict: + """ + Calculate protein adequacy score (0-100). + + Score based on: + - Daily protein intake vs. target (1.6-2.2 g/kg) + - Consistency across days + + Args: + profile_id: User profile ID + days: Analysis window (default 28) + + Returns: + { + "adequacy_score": int, # 0-100 + "avg_protein_g": float, + "target_protein_low": float, + "target_protein_high": float, + "protein_g_per_kg": float, + "days_in_target": int, + "days_with_data": int, + "confidence": str + } + """ + # Get protein targets + targets = get_protein_targets_data(profile_id) + + with get_db() as conn: + cur = get_cursor(conn) + cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') + + cur.execute( + """SELECT + AVG(protein_g) as avg_protein, + COUNT(*) as cnt, + SUM(CASE WHEN protein_g >= %s AND protein_g <= %s THEN 1 ELSE 0 END) as days_in_target + FROM nutrition_log + WHERE profile_id=%s AND date >= %s AND protein_g IS NOT NULL""", + (targets['protein_target_low'], targets['protein_target_high'], profile_id, cutoff) + ) + row = cur.fetchone() + + if not row or row['cnt'] == 0: + return { + "adequacy_score": 0, + "avg_protein_g": 0.0, + "target_protein_low": targets['protein_target_low'], + "target_protein_high": targets['protein_target_high'], + "protein_g_per_kg": 0.0, + "days_in_target": 0, + "days_with_data": 0, + "confidence": "insufficient" + } + + avg_protein = safe_float(row['avg_protein']) + days_with_data = row['cnt'] + days_in_target = row['days_in_target'] + + protein_g_per_kg = avg_protein / targets['current_weight'] if targets['current_weight'] > 0 else 0.0 + + # Calculate adequacy score + # 100 = always in target range + # Scale based on percentage of days in target + average relative to target + target_pct = (days_in_target / days_with_data * 100) if days_with_data > 0 else 0 + + # Bonus/penalty for average protein level + target_mid = (targets['protein_target_low'] + targets['protein_target_high']) / 2 + avg_vs_target = (avg_protein / target_mid) if target_mid > 0 else 0 + + # Weighted score: 70% target days, 30% average level + adequacy_score = int(target_pct * 0.7 + min(avg_vs_target * 100, 100) * 0.3) + adequacy_score = max(0, min(100, adequacy_score)) # Clamp to 0-100 + + confidence = calculate_confidence(days_with_data, days, "general") + + return { + "adequacy_score": adequacy_score, + "avg_protein_g": avg_protein, + "target_protein_low": targets['protein_target_low'], + "target_protein_high": targets['protein_target_high'], + "protein_g_per_kg": protein_g_per_kg, + "days_in_target": days_in_target, + "days_with_data": days_with_data, + "confidence": confidence + } + + +def get_macro_consistency_data( + profile_id: str, + days: int = 28 +) -> Dict: + """ + Calculate macro consistency score (0-100). + + Measures how consistent macronutrient ratios are across days. + High consistency = predictable nutrition, easier to track progress. + + Args: + profile_id: User profile ID + days: Analysis window (default 28) + + Returns: + { + "consistency_score": int, # 0-100 (100 = very consistent) + "avg_protein_pct": float, + "avg_carbs_pct": float, + "avg_fat_pct": float, + "std_dev_protein": float, # Standard deviation + "std_dev_carbs": float, + "std_dev_fat": float, + "confidence": str, + "data_points": int + } + """ + with get_db() as conn: + cur = get_cursor(conn) + cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') + + cur.execute( + """SELECT + protein_g, carbs_g, fat_g, kcal + FROM nutrition_log + WHERE profile_id=%s + AND date >= %s + AND protein_g IS NOT NULL + AND carbs_g IS NOT NULL + AND fat_g IS NOT NULL + AND kcal > 0 + ORDER BY date""", + (profile_id, cutoff) + ) + rows = cur.fetchall() + + if len(rows) < 3: + return { + "consistency_score": 0, + "avg_protein_pct": 0.0, + "avg_carbs_pct": 0.0, + "avg_fat_pct": 0.0, + "std_dev_protein": 0.0, + "std_dev_carbs": 0.0, + "std_dev_fat": 0.0, + "confidence": "insufficient", + "data_points": len(rows) + } + + # Calculate macro percentages for each day + import statistics + + protein_pcts = [] + carbs_pcts = [] + fat_pcts = [] + + for row in rows: + total_kcal = safe_float(row['kcal']) + if total_kcal == 0: + continue + + # Convert grams to kcal (protein=4, carbs=4, fat=9) + protein_kcal = safe_float(row['protein_g']) * 4 + carbs_kcal = safe_float(row['carbs_g']) * 4 + fat_kcal = safe_float(row['fat_g']) * 9 + + macro_kcal_total = protein_kcal + carbs_kcal + fat_kcal + + if macro_kcal_total > 0: + protein_pcts.append(protein_kcal / macro_kcal_total * 100) + carbs_pcts.append(carbs_kcal / macro_kcal_total * 100) + fat_pcts.append(fat_kcal / macro_kcal_total * 100) + + if len(protein_pcts) < 3: + return { + "consistency_score": 0, + "avg_protein_pct": 0.0, + "avg_carbs_pct": 0.0, + "avg_fat_pct": 0.0, + "std_dev_protein": 0.0, + "std_dev_carbs": 0.0, + "std_dev_fat": 0.0, + "confidence": "insufficient", + "data_points": len(protein_pcts) + } + + # Calculate averages and standard deviations + avg_protein_pct = statistics.mean(protein_pcts) + avg_carbs_pct = statistics.mean(carbs_pcts) + avg_fat_pct = statistics.mean(fat_pcts) + + std_protein = statistics.stdev(protein_pcts) if len(protein_pcts) > 1 else 0.0 + std_carbs = statistics.stdev(carbs_pcts) if len(carbs_pcts) > 1 else 0.0 + std_fat = statistics.stdev(fat_pcts) if len(fat_pcts) > 1 else 0.0 + + # Consistency score: inverse of average standard deviation + # Lower std_dev = higher consistency + avg_std = (std_protein + std_carbs + std_fat) / 3 + + # Score: 100 - (avg_std * scale_factor) + # avg_std of 5% = score 75, avg_std of 10% = score 50, avg_std of 20% = score 0 + consistency_score = max(0, min(100, int(100 - (avg_std * 5)))) + + confidence = calculate_confidence(len(protein_pcts), days, "general") + + return { + "consistency_score": consistency_score, + "avg_protein_pct": avg_protein_pct, + "avg_carbs_pct": avg_carbs_pct, + "avg_fat_pct": avg_fat_pct, + "std_dev_protein": std_protein, + "std_dev_carbs": std_carbs, + "std_dev_fat": std_fat, + "confidence": confidence, + "data_points": len(protein_pcts) + } diff --git a/backend/placeholder_resolver.py b/backend/placeholder_resolver.py index 9b127ef..3264c3e 100644 --- a/backend/placeholder_resolver.py +++ b/backend/placeholder_resolver.py @@ -18,6 +18,11 @@ from data_layer.body_metrics import ( get_body_composition_data, get_circumference_summary_data ) +from data_layer.nutrition_metrics import ( + get_nutrition_average_data, + get_nutrition_days_data, + get_protein_targets_data +) # ── Helper Functions ────────────────────────────────────────────────────────── @@ -81,33 +86,32 @@ def get_latest_bf(profile_id: str) -> Optional[str]: def get_nutrition_avg(profile_id: str, field: str, days: int = 30) -> str: - """Calculate average nutrition value.""" - with get_db() as conn: - cur = get_cursor(conn) - cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') + """ + Calculate average nutrition value. - # Map field names to actual column names - field_map = { - 'protein': 'protein_g', - 'fat': 'fat_g', - 'carb': 'carbs_g', - 'kcal': 'kcal' - } - db_field = field_map.get(field, field) + Phase 0c: Refactored to use data_layer.nutrition_metrics.get_nutrition_average_data() + This function now only FORMATS the data for AI consumption. + """ + data = get_nutrition_average_data(profile_id, days) - cur.execute( - f"""SELECT AVG({db_field}) as avg FROM nutrition_log - WHERE profile_id=%s AND date >= %s AND {db_field} IS NOT NULL""", - (profile_id, cutoff) - ) - row = cur.fetchone() - if row and row['avg']: - if field == 'kcal': - return f"{int(row['avg'])} kcal/Tag (Ø {days} Tage)" - else: - return f"{int(row['avg'])}g/Tag (Ø {days} Tage)" + if data['confidence'] == 'insufficient': return "nicht verfügbar" + # Map field names to data keys + field_map = { + 'protein': 'protein_avg', + 'fat': 'fat_avg', + 'carb': 'carbs_avg', + 'kcal': 'kcal_avg' + } + data_key = field_map.get(field, f'{field}_avg') + value = data.get(data_key, 0) + + if field == 'kcal': + return f"{int(value)} kcal/Tag (Ø {days} Tage)" + else: + return f"{int(value)}g/Tag (Ø {days} Tage)" + def get_caliper_summary(profile_id: str) -> str: """Get latest caliper measurements summary.""" @@ -178,48 +182,45 @@ def get_goal_bf_pct(profile_id: str) -> str: def get_nutrition_days(profile_id: str, days: int = 30) -> str: - """Get number of days with nutrition data.""" - with get_db() as conn: - cur = get_cursor(conn) - cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') - cur.execute( - """SELECT COUNT(DISTINCT date) as days FROM nutrition_log - WHERE profile_id=%s AND date >= %s""", - (profile_id, cutoff) - ) - row = cur.fetchone() - return str(row['days']) if row else "0" + """ + Get number of days with nutrition data. + + Phase 0c: Refactored to use data_layer.nutrition_metrics.get_nutrition_days_data() + This function now only FORMATS the data for AI consumption. + """ + data = get_nutrition_days_data(profile_id, days) + return str(data['days_with_data']) def get_protein_ziel_low(profile_id: str) -> str: - """Calculate lower protein target based on current weight (1.6g/kg).""" - with get_db() as conn: - cur = get_cursor(conn) - cur.execute( - """SELECT weight FROM weight_log - WHERE profile_id=%s ORDER BY date DESC LIMIT 1""", - (profile_id,) - ) - row = cur.fetchone() - if row: - return f"{int(float(row['weight']) * 1.6)}" + """ + Calculate lower protein target based on current weight (1.6g/kg). + + Phase 0c: Refactored to use data_layer.nutrition_metrics.get_protein_targets_data() + This function now only FORMATS the data for AI consumption. + """ + data = get_protein_targets_data(profile_id) + + if data['confidence'] == 'insufficient': return "nicht verfügbar" + return f"{int(data['protein_target_low'])}" + def get_protein_ziel_high(profile_id: str) -> str: - """Calculate upper protein target based on current weight (2.2g/kg).""" - with get_db() as conn: - cur = get_cursor(conn) - cur.execute( - """SELECT weight FROM weight_log - WHERE profile_id=%s ORDER BY date DESC LIMIT 1""", - (profile_id,) - ) - row = cur.fetchone() - if row: - return f"{int(float(row['weight']) * 2.2)}" + """ + Calculate upper protein target based on current weight (2.2g/kg). + + Phase 0c: Refactored to use data_layer.nutrition_metrics.get_protein_targets_data() + This function now only FORMATS the data for AI consumption. + """ + data = get_protein_targets_data(profile_id) + + if data['confidence'] == 'insufficient': return "nicht verfügbar" + return f"{int(data['protein_target_high'])}" + def get_activity_summary(profile_id: str, days: int = 14) -> str: """Get activity summary for recent period."""