diff --git a/backend/data_layer/__init__.py b/backend/data_layer/__init__.py index 83dffe0..1352974 100644 --- a/backend/data_layer/__init__.py +++ b/backend/data_layer/__init__.py @@ -68,7 +68,7 @@ __all__ = [ 'calculate_body_progress_score', 'calculate_body_data_quality', - # Nutrition Metrics + # Nutrition Metrics (Basic) 'get_nutrition_average_data', 'get_nutrition_days_data', 'get_protein_targets_data', @@ -76,6 +76,20 @@ __all__ = [ 'get_protein_adequacy_data', 'get_macro_consistency_data', + # Nutrition Metrics (Calculated) + 'calculate_energy_balance_7d', + 'calculate_energy_deficit_surplus', + 'calculate_protein_g_per_kg', + 'calculate_protein_days_in_target', + 'calculate_protein_adequacy_28d', + 'calculate_macro_consistency_score', + 'calculate_intake_volatility', + 'calculate_nutrition_score', + 'calculate_energy_availability_warning', + 'calculate_fiber_avg_7d', + 'calculate_sugar_avg_7d', + 'calculate_nutrition_data_quality', + # Activity Metrics 'get_activity_summary_data', 'get_activity_detail_data', diff --git a/backend/data_layer/nutrition_metrics.py b/backend/data_layer/nutrition_metrics.py index 492a95c..6a695a0 100644 --- a/backend/data_layer/nutrition_metrics.py +++ b/backend/data_layer/nutrition_metrics.py @@ -480,3 +480,614 @@ def get_macro_consistency_data( "confidence": confidence, "data_points": len(protein_pcts) } + + +# ============================================================================ +# Calculated Metrics (migrated from calculations/nutrition_metrics.py) +# ============================================================================ +# These functions return simple values for placeholders. +# Use get_*_data() functions above for structured chart data. + + +def calculate_energy_balance_7d(profile_id: str) -> Optional[float]: + """ + Calculate 7-day average energy balance (kcal/day) + Positive = surplus, Negative = deficit + + Migration from Phase 0b: + Used by placeholders that need single balance value + """ + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT kcal + FROM nutrition_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '7 days' + ORDER BY date DESC + """, (profile_id,)) + + calories = [row['kcal'] for row in cur.fetchall()] + + if len(calories) < 4: # Need at least 4 days + return None + + avg_intake = float(sum(calories) / len(calories)) + + # Get estimated TDEE (simplified - could use Harris-Benedict) + # For now, use weight-based estimate + cur.execute(""" + SELECT weight + FROM weight_log + WHERE profile_id = %s + ORDER BY date DESC + LIMIT 1 + """, (profile_id,)) + + weight_row = cur.fetchone() + if not weight_row: + return None + + # Simple TDEE estimate: bodyweight (kg) × 30-35 + # TODO: Improve with activity level, age, gender + estimated_tdee = float(weight_row['weight']) * 32.5 + + balance = avg_intake - estimated_tdee + + return round(balance, 0) + + +def calculate_energy_deficit_surplus(profile_id: str, days: int = 7) -> Optional[str]: + """ + Classify energy balance as deficit/maintenance/surplus + Returns: 'deficit', 'maintenance', 'surplus', or None + """ + balance = calculate_energy_balance_7d(profile_id) + + if balance is None: + return None + + if balance < -200: + return 'deficit' + elif balance > 200: + return 'surplus' + else: + return 'maintenance' + + +def calculate_protein_g_per_kg(profile_id: str) -> Optional[float]: + """Calculate average protein intake in g/kg bodyweight (last 7 days)""" + with get_db() as conn: + cur = get_cursor(conn) + + # Get recent weight + cur.execute(""" + SELECT weight + FROM weight_log + WHERE profile_id = %s + ORDER BY date DESC + LIMIT 1 + """, (profile_id,)) + + weight_row = cur.fetchone() + if not weight_row: + return None + + weight = float(weight_row['weight']) + + # Get protein intake + cur.execute(""" + SELECT protein_g + FROM nutrition_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '7 days' + AND protein_g IS NOT NULL + ORDER BY date DESC + """, (profile_id,)) + + protein_values = [row['protein_g'] for row in cur.fetchall()] + + if len(protein_values) < 4: + return None + + avg_protein = float(sum(protein_values) / len(protein_values)) + protein_per_kg = avg_protein / weight + + return round(protein_per_kg, 2) + + +def calculate_protein_days_in_target(profile_id: str, target_low: float = 1.6, target_high: float = 2.2) -> Optional[str]: + """ + Calculate how many days in last 7 were within protein target + Returns: "5/7" format or None + """ + with get_db() as conn: + cur = get_cursor(conn) + + # Get recent weight + cur.execute(""" + SELECT weight + FROM weight_log + WHERE profile_id = %s + ORDER BY date DESC + LIMIT 1 + """, (profile_id,)) + + weight_row = cur.fetchone() + if not weight_row: + return None + + weight = float(weight_row['weight']) + + # Get protein intake last 7 days + cur.execute(""" + SELECT protein_g, date + FROM nutrition_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '7 days' + AND protein_g IS NOT NULL + ORDER BY date DESC + """, (profile_id,)) + + protein_data = cur.fetchall() + + if len(protein_data) < 4: + return None + + # Count days in target range + days_in_target = 0 + total_days = len(protein_data) + + for row in protein_data: + protein_per_kg = float(row['protein_g']) / weight + if target_low <= protein_per_kg <= target_high: + days_in_target += 1 + + return f"{days_in_target}/{total_days}" + + +def calculate_protein_adequacy_28d(profile_id: str) -> Optional[int]: + """ + Protein adequacy score 0-100 (last 28 days) + Based on consistency and target achievement + """ + import statistics + + with get_db() as conn: + cur = get_cursor(conn) + + # Get average weight (28d) + cur.execute(""" + SELECT AVG(weight) as avg_weight + FROM weight_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '28 days' + """, (profile_id,)) + + weight_row = cur.fetchone() + if not weight_row or not weight_row['avg_weight']: + return None + + weight = float(weight_row['avg_weight']) + + # Get protein intake (28d) + cur.execute(""" + SELECT protein_g + FROM nutrition_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '28 days' + AND protein_g IS NOT NULL + """, (profile_id,)) + + protein_values = [float(row['protein_g']) for row in cur.fetchall()] + + if len(protein_values) < 18: # 60% coverage + return None + + # Calculate metrics + protein_per_kg_values = [p / weight for p in protein_values] + avg_protein_per_kg = sum(protein_per_kg_values) / len(protein_per_kg_values) + + # Target range: 1.6-2.2 g/kg for active individuals + target_mid = 1.9 + + # Score based on distance from target + if 1.6 <= avg_protein_per_kg <= 2.2: + base_score = 100 + elif avg_protein_per_kg < 1.6: + # Below target + base_score = max(40, 100 - ((1.6 - avg_protein_per_kg) * 40)) + else: + # Above target (less penalty) + base_score = max(80, 100 - ((avg_protein_per_kg - 2.2) * 10)) + + # Consistency bonus/penalty + std_dev = statistics.stdev(protein_per_kg_values) + if std_dev < 0.3: + consistency_bonus = 10 + elif std_dev < 0.5: + consistency_bonus = 0 + else: + consistency_bonus = -10 + + final_score = min(100, max(0, base_score + consistency_bonus)) + + return int(final_score) + + +def calculate_macro_consistency_score(profile_id: str) -> Optional[int]: + """ + Macro consistency score 0-100 (last 28 days) + Lower variability = higher score + """ + import statistics + + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT kcal, protein_g, fat_g, carbs_g + FROM nutrition_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '28 days' + AND kcal IS NOT NULL + ORDER BY date DESC + """, (profile_id,)) + + data = cur.fetchall() + + if len(data) < 18: + return None + + # Calculate coefficient of variation for each macro + def cv(values): + """Coefficient of variation (std_dev / mean)""" + if not values or len(values) < 2: + return None + mean = sum(values) / len(values) + if mean == 0: + return None + std_dev = statistics.stdev(values) + return std_dev / mean + + calories_cv = cv([d['kcal'] for d in data]) + protein_cv = cv([d['protein_g'] for d in data if d['protein_g']]) + fat_cv = cv([d['fat_g'] for d in data if d['fat_g']]) + carbs_cv = cv([d['carbs_g'] for d in data if d['carbs_g']]) + + cv_values = [v for v in [calories_cv, protein_cv, fat_cv, carbs_cv] if v is not None] + + if not cv_values: + return None + + avg_cv = sum(cv_values) / len(cv_values) + + # Score: lower CV = higher score + # CV < 0.2 = excellent consistency + # CV > 0.5 = poor consistency + if avg_cv < 0.2: + score = 100 + elif avg_cv < 0.3: + score = 85 + elif avg_cv < 0.4: + score = 70 + elif avg_cv < 0.5: + score = 55 + else: + score = max(30, 100 - (avg_cv * 100)) + + return int(score) + + +def calculate_intake_volatility(profile_id: str) -> Optional[str]: + """ + Classify intake volatility: 'stable', 'moderate', 'high' + """ + consistency = calculate_macro_consistency_score(profile_id) + + if consistency is None: + return None + + if consistency >= 80: + return 'stable' + elif consistency >= 60: + return 'moderate' + else: + return 'high' + + +def calculate_nutrition_score(profile_id: str, focus_weights: Optional[Dict] = None) -> Optional[int]: + """ + Nutrition adherence score 0-100 + Weighted by user's nutrition-related focus areas + """ + if focus_weights is None: + # Import here to avoid circular dependency + from data_layer.scores import get_user_focus_weights + focus_weights = get_user_focus_weights(profile_id) + + # Nutrition-related focus areas (English keys from DB) + protein_intake = focus_weights.get('protein_intake', 0) + calorie_balance = focus_weights.get('calorie_balance', 0) + macro_consistency = focus_weights.get('macro_consistency', 0) + meal_timing = focus_weights.get('meal_timing', 0) + hydration = focus_weights.get('hydration', 0) + + total_nutrition_weight = protein_intake + calorie_balance + macro_consistency + meal_timing + hydration + + if total_nutrition_weight == 0: + return None # No nutrition goals + + components = [] + + # 1. Calorie target adherence (if calorie_balance goal active) + if calorie_balance > 0: + calorie_score = _score_calorie_adherence(profile_id) + if calorie_score is not None: + components.append(('calories', calorie_score, calorie_balance)) + + # 2. Protein target adherence (if protein_intake goal active) + if protein_intake > 0: + protein_score = calculate_protein_adequacy_28d(profile_id) + if protein_score is not None: + components.append(('protein', protein_score, protein_intake)) + + # 3. Intake consistency (if macro_consistency goal active) + if macro_consistency > 0: + consistency_score = calculate_macro_consistency_score(profile_id) + if consistency_score is not None: + components.append(('consistency', consistency_score, macro_consistency)) + + # 4. Macro balance (always relevant if any nutrition goal) + if total_nutrition_weight > 0: + macro_score = _score_macro_balance(profile_id) + if macro_score is not None: + # Use 20% of total weight for macro balance + components.append(('macros', macro_score, total_nutrition_weight * 0.2)) + + if not components: + return None + + # Weighted average + total_score = sum(score * weight for _, score, weight in components) + total_weight = sum(weight for _, _, weight in components) + + return int(total_score / total_weight) + + +def _score_calorie_adherence(profile_id: str) -> Optional[int]: + """Score calorie target adherence (0-100)""" + # Check for energy balance goal + # For now, use energy balance calculation + balance = calculate_energy_balance_7d(profile_id) + + if balance is None: + return None + + # Score based on whether deficit/surplus aligns with goal + # Simplified: assume weight loss goal = deficit is good + # TODO: Check actual goal type + + abs_balance = abs(balance) + + # Moderate deficit/surplus = good + if 200 <= abs_balance <= 500: + return 100 + elif 100 <= abs_balance <= 700: + return 85 + elif abs_balance <= 900: + return 70 + elif abs_balance <= 1200: + return 55 + else: + return 40 + + +def _score_macro_balance(profile_id: str) -> Optional[int]: + """Score macro balance (0-100)""" + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT protein_g, fat_g, carbs_g, kcal + FROM nutrition_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '28 days' + AND protein_g IS NOT NULL + AND fat_g IS NOT NULL + AND carbs_g IS NOT NULL + ORDER BY date DESC + """, (profile_id,)) + + data = cur.fetchall() + + if len(data) < 18: + return None + + # Calculate average macro percentages + macro_pcts = [] + for row in data: + total_kcal = (row['protein_g'] * 4) + (row['fat_g'] * 9) + (row['carbs_g'] * 4) + if total_kcal == 0: + continue + + protein_pct = (row['protein_g'] * 4 / total_kcal) * 100 + fat_pct = (row['fat_g'] * 9 / total_kcal) * 100 + carbs_pct = (row['carbs_g'] * 4 / total_kcal) * 100 + + macro_pcts.append((protein_pct, fat_pct, carbs_pct)) + + if not macro_pcts: + return None + + avg_protein_pct = sum(p for p, _, _ in macro_pcts) / len(macro_pcts) + avg_fat_pct = sum(f for _, f, _ in macro_pcts) / len(macro_pcts) + avg_carbs_pct = sum(c for _, _, c in macro_pcts) / len(macro_pcts) + + # Reasonable ranges: + # Protein: 20-35% + # Fat: 20-35% + # Carbs: 30-55% + + score = 100 + + # Protein score + if not (20 <= avg_protein_pct <= 35): + if avg_protein_pct < 20: + score -= (20 - avg_protein_pct) * 2 + else: + score -= (avg_protein_pct - 35) * 1 + + # Fat score + if not (20 <= avg_fat_pct <= 35): + if avg_fat_pct < 20: + score -= (20 - avg_fat_pct) * 2 + else: + score -= (avg_fat_pct - 35) * 2 + + # Carbs score + if not (30 <= avg_carbs_pct <= 55): + if avg_carbs_pct < 30: + score -= (30 - avg_carbs_pct) * 1.5 + else: + score -= (avg_carbs_pct - 55) * 1.5 + + return max(40, min(100, int(score))) + + +def calculate_energy_availability_warning(profile_id: str) -> Optional[Dict]: + """ + Heuristic energy availability warning + Returns dict with warning level and reasons + """ + warnings = [] + severity = 'none' # none, low, medium, high + + # 1. Check for sustained large deficit + balance = calculate_energy_balance_7d(profile_id) + if balance and balance < -800: + warnings.append('Anhaltend großes Energiedefizit (>800 kcal/Tag)') + severity = 'medium' + + if balance < -1200: + warnings.append('Sehr großes Energiedefizit (>1200 kcal/Tag)') + severity = 'high' + + # 2. Check recovery score + from data_layer.recovery_metrics import calculate_recovery_score_v2 + recovery = calculate_recovery_score_v2(profile_id) + if recovery and recovery < 50: + warnings.append('Recovery Score niedrig (<50)') + if severity == 'none': + severity = 'low' + elif severity == 'medium': + severity = 'high' + + # 3. Check LBM trend + from data_layer.body_metrics import calculate_lbm_28d_change + lbm_change = calculate_lbm_28d_change(profile_id) + if lbm_change and lbm_change < -1.0: + warnings.append('Magermasse sinkt (>1kg in 28 Tagen)') + if severity == 'none': + severity = 'low' + elif severity in ['low', 'medium']: + severity = 'high' + + # 4. Check sleep quality + from data_layer.recovery_metrics import calculate_sleep_quality_7d + sleep_quality = calculate_sleep_quality_7d(profile_id) + if sleep_quality and sleep_quality < 60: + warnings.append('Schlafqualität verschlechtert') + if severity == 'none': + severity = 'low' + + if not warnings: + return None + + return { + 'severity': severity, + 'warnings': warnings, + 'recommendation': _get_energy_warning_recommendation(severity) + } + + +def _get_energy_warning_recommendation(severity: str) -> str: + """Get recommendation text based on severity""" + if severity == 'high': + return ("Mögliche Unterversorgung erkannt. Erwäge eine Reduktion des Energiedefizits, " + "Erhöhung der Proteinzufuhr und mehr Erholung. Dies ist keine medizinische Diagnose.") + elif severity == 'medium': + return ("Hinweise auf aggressives Defizit. Beobachte Recovery, Schlaf und Magermasse genau.") + else: + return ("Leichte Hinweise auf Belastung. Monitoring empfohlen.") + + +def calculate_fiber_avg_7d(profile_id: str) -> Optional[float]: + """Calculate average fiber intake (g/day) last 7 days""" + # TODO: Implement when fiber column added to nutrition_log + return None + + +def calculate_sugar_avg_7d(profile_id: str) -> Optional[float]: + """Calculate average sugar intake (g/day) last 7 days""" + # TODO: Implement when sugar column added to nutrition_log + return None + + +def calculate_nutrition_data_quality(profile_id: str) -> Dict[str, any]: + """ + Assess data quality for nutrition metrics + Returns dict with quality score and details + """ + with get_db() as conn: + cur = get_cursor(conn) + + # Nutrition entries last 28 days + cur.execute(""" + SELECT COUNT(*) as total, + COUNT(protein_g) as with_protein, + COUNT(fat_g) as with_fat, + COUNT(carbs_g) as with_carbs + FROM nutrition_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '28 days' + """, (profile_id,)) + + counts = cur.fetchone() + + total_entries = counts['total'] + protein_coverage = counts['with_protein'] / total_entries if total_entries > 0 else 0 + macro_coverage = min(counts['with_fat'], counts['with_carbs']) / total_entries if total_entries > 0 else 0 + + # Score components + frequency_score = min(100, (total_entries / 21) * 100) # 21 = 75% of 28 days + protein_score = protein_coverage * 100 + macro_score = macro_coverage * 100 + + # Overall score (frequency 50%, protein 30%, macros 20%) + overall_score = int( + frequency_score * 0.5 + + protein_score * 0.3 + + macro_score * 0.2 + ) + + # Confidence level + if overall_score >= 80: + confidence = "high" + elif overall_score >= 60: + confidence = "medium" + else: + confidence = "low" + + return { + "overall_score": overall_score, + "confidence": confidence, + "measurements": { + "entries_28d": total_entries, + "protein_coverage_pct": int(protein_coverage * 100), + "macro_coverage_pct": int(macro_coverage * 100) + }, + "component_scores": { + "frequency": int(frequency_score), + "protein": int(protein_score), + "macros": int(macro_score) + } + } diff --git a/backend/placeholder_resolver.py b/backend/placeholder_resolver.py index f3c1744..f6e3da8 100644 --- a/backend/placeholder_resolver.py +++ b/backend/placeholder_resolver.py @@ -417,8 +417,8 @@ def _safe_int(func_name: str, profile_id: str) -> str: import traceback try: # Import calculations dynamically to avoid circular imports - from calculations import scores, nutrition_metrics, activity_metrics, recovery_metrics, correlation_metrics - from data_layer import body_metrics + from calculations import scores, activity_metrics, recovery_metrics, correlation_metrics + from data_layer import body_metrics, nutrition_metrics # Map function names to actual functions func_map = { @@ -480,8 +480,8 @@ def _safe_float(func_name: str, profile_id: str, decimals: int = 1) -> str: """ import traceback try: - from calculations import nutrition_metrics, activity_metrics, recovery_metrics, scores - from data_layer import body_metrics + from calculations import activity_metrics, recovery_metrics, scores + from data_layer import body_metrics, nutrition_metrics func_map = { 'weight_7d_median': body_metrics.calculate_weight_7d_median,