diff --git a/backend/calculations/__init__.py b/backend/calculations/__init__.py new file mode 100644 index 0000000..f534202 --- /dev/null +++ b/backend/calculations/__init__.py @@ -0,0 +1,48 @@ +""" +Calculation Engine for Phase 0b - Goal-Aware Placeholders + +This package contains all metric calculation functions for: +- Body metrics (K1-K5 from visualization concept) +- Nutrition metrics (E1-E5) +- Activity metrics (A1-A8) +- Recovery metrics (S1) +- Correlations (C1-C7) +- Scores (Goal Progress Score with Dynamic Focus Areas) + +All calculations are designed to work with Dynamic Focus Areas v2.0. +""" + +from .body_metrics import * +from .nutrition_metrics import * +from .activity_metrics import * +from .recovery_metrics import * +from .correlation_metrics import * +from .scores import * + +__all__ = [ + # Body + 'calculate_weight_7d_median', + 'calculate_weight_28d_slope', + 'calculate_fm_28d_change', + 'calculate_lbm_28d_change', + 'calculate_body_progress_score', + + # Nutrition + 'calculate_energy_balance_7d', + 'calculate_protein_g_per_kg', + 'calculate_nutrition_score', + + # Activity + 'calculate_training_minutes_week', + 'calculate_activity_score', + + # Recovery + 'calculate_recovery_score_v2', + + # Correlations + 'calculate_lag_correlation', + + # Meta Scores + 'calculate_goal_progress_score', + 'calculate_data_quality_score', +] diff --git a/backend/calculations/activity_metrics.py b/backend/calculations/activity_metrics.py new file mode 100644 index 0000000..3decc9c --- /dev/null +++ b/backend/calculations/activity_metrics.py @@ -0,0 +1,624 @@ +""" +Activity Metrics Calculation Engine + +Implements A1-A8 from visualization concept: +- A1: Training volume per week +- A2: Intensity distribution +- A3: Training quality matrix +- A4: Ability balance radar +- A5: Load monitoring (proxy-based) +- A6: Activity goal alignment score +- A7: Rest day compliance +- A8: VO2max development + +All calculations work with training_types abilities system. +""" +from datetime import datetime, timedelta +from typing import Optional, Dict, List +import statistics + +from db import get_db, get_cursor + + +# ============================================================================ +# A1: Training Volume Calculations +# ============================================================================ + +def calculate_training_minutes_week(profile_id: str) -> Optional[int]: + """Calculate total training minutes last 7 days""" + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT SUM(duration) as total_minutes + FROM activity_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '7 days' + """, (profile_id,)) + + row = cur.fetchone() + return int(row['total_minutes']) if row and row['total_minutes'] else None + + +def calculate_training_frequency_7d(profile_id: str) -> Optional[int]: + """Calculate number of training sessions last 7 days""" + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT COUNT(*) as session_count + FROM activity_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '7 days' + """, (profile_id,)) + + row = cur.fetchone() + return int(row['session_count']) if row else None + + +def calculate_quality_sessions_pct(profile_id: str) -> Optional[int]: + """Calculate percentage of quality sessions (good or better) last 28 days""" + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT + COUNT(*) as total, + COUNT(*) FILTER (WHERE quality_label IN ('excellent', 'very_good', 'good')) as quality_count + FROM activity_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '28 days' + """, (profile_id,)) + + row = cur.fetchone() + if not row or row['total'] == 0: + return None + + pct = (row['quality_count'] / row['total']) * 100 + return int(pct) + + +# ============================================================================ +# A2: Intensity Distribution (Proxy-based) +# ============================================================================ + +def calculate_intensity_proxy_distribution(profile_id: str) -> Optional[Dict]: + """ + Calculate intensity distribution (proxy until HR zones available) + Returns dict: {'low': X, 'moderate': Y, 'high': Z} in minutes + """ + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT duration, avg_heart_rate, max_heart_rate + FROM activity_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '28 days' + """, (profile_id,)) + + activities = cur.fetchall() + + if not activities: + return None + + low_min = 0 + moderate_min = 0 + high_min = 0 + + for activity in activities: + duration = activity['duration'] + avg_hr = activity['avg_heart_rate'] + max_hr = activity['max_heart_rate'] + + # Simple proxy classification + if avg_hr: + # Rough HR-based classification (assumes max HR ~190) + if avg_hr < 120: + low_min += duration + elif avg_hr < 150: + moderate_min += duration + else: + high_min += duration + else: + # Fallback: assume moderate + moderate_min += duration + + return { + 'low': low_min, + 'moderate': moderate_min, + 'high': high_min + } + + +# ============================================================================ +# A4: Ability Balance Calculations +# ============================================================================ + +def calculate_ability_balance(profile_id: str) -> Optional[Dict]: + """ + Calculate ability balance from training_types.abilities + Returns dict with scores per ability dimension (0-100) + """ + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT a.duration, tt.abilities + FROM activity_log a + JOIN training_types tt ON a.training_category = tt.category + WHERE a.profile_id = %s + AND a.date >= CURRENT_DATE - INTERVAL '28 days' + AND tt.abilities IS NOT NULL + """, (profile_id,)) + + activities = cur.fetchall() + + if not activities: + return None + + # Accumulate ability load (duration × ability weight) + ability_loads = { + 'strength': 0, + 'endurance': 0, + 'mental': 0, + 'coordination': 0, + 'mobility': 0 + } + + for activity in activities: + duration = activity['duration'] + abilities = activity['abilities'] # JSONB + + if not abilities: + continue + + for ability, weight in abilities.items(): + if ability in ability_loads: + ability_loads[ability] += duration * weight + + # Normalize to 0-100 scale + max_load = max(ability_loads.values()) if ability_loads else 1 + if max_load == 0: + return None + + normalized = { + ability: int((load / max_load) * 100) + for ability, load in ability_loads.items() + } + + return normalized + + +def calculate_ability_balance_strength(profile_id: str) -> Optional[int]: + """Get strength ability score""" + balance = calculate_ability_balance(profile_id) + return balance['strength'] if balance else None + + +def calculate_ability_balance_endurance(profile_id: str) -> Optional[int]: + """Get endurance ability score""" + balance = calculate_ability_balance(profile_id) + return balance['endurance'] if balance else None + + +def calculate_ability_balance_mental(profile_id: str) -> Optional[int]: + """Get mental ability score""" + balance = calculate_ability_balance(profile_id) + return balance['mental'] if balance else None + + +def calculate_ability_balance_coordination(profile_id: str) -> Optional[int]: + """Get coordination ability score""" + balance = calculate_ability_balance(profile_id) + return balance['coordination'] if balance else None + + +def calculate_ability_balance_mobility(profile_id: str) -> Optional[int]: + """Get mobility ability score""" + balance = calculate_ability_balance(profile_id) + return balance['mobility'] if balance else None + + +# ============================================================================ +# A5: Load Monitoring (Proxy-based) +# ============================================================================ + +def calculate_proxy_internal_load_7d(profile_id: str) -> Optional[int]: + """ + Calculate proxy internal load (last 7 days) + Formula: duration × intensity_factor × quality_factor + """ + intensity_factors = {'low': 1.0, 'moderate': 1.5, 'high': 2.0} + quality_factors = { + 'excellent': 1.15, + 'very_good': 1.05, + 'good': 1.0, + 'acceptable': 0.9, + 'poor': 0.75, + 'excluded': 0.0 + } + + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT duration, avg_heart_rate, quality_label + FROM activity_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '7 days' + """, (profile_id,)) + + activities = cur.fetchall() + + if not activities: + return None + + total_load = 0 + + for activity in activities: + duration = activity['duration'] + avg_hr = activity['avg_heart_rate'] + quality = activity['quality_label'] or 'good' + + # Determine intensity + if avg_hr: + if avg_hr < 120: + intensity = 'low' + elif avg_hr < 150: + intensity = 'moderate' + else: + intensity = 'high' + else: + intensity = 'moderate' + + load = duration * intensity_factors[intensity] * quality_factors.get(quality, 1.0) + total_load += load + + return int(total_load) + + +def calculate_monotony_score(profile_id: str) -> Optional[float]: + """ + Calculate training monotony (last 7 days) + Monotony = mean daily load / std dev daily load + Higher = more monotonous + """ + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT date, SUM(duration) as daily_duration + FROM activity_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '7 days' + GROUP BY date + ORDER BY date + """, (profile_id,)) + + daily_loads = [row['daily_duration'] for row in cur.fetchall()] + + if len(daily_loads) < 4: + return None + + mean_load = sum(daily_loads) / len(daily_loads) + std_dev = statistics.stdev(daily_loads) + + if std_dev == 0: + return None + + monotony = mean_load / std_dev + return round(monotony, 2) + + +def calculate_strain_score(profile_id: str) -> Optional[int]: + """ + Calculate training strain (last 7 days) + Strain = weekly load × monotony + """ + weekly_load = calculate_proxy_internal_load_7d(profile_id) + monotony = calculate_monotony_score(profile_id) + + if weekly_load is None or monotony is None: + return None + + strain = weekly_load * monotony + return int(strain) + + +# ============================================================================ +# A6: Activity Goal Alignment Score (Dynamic Focus Areas) +# ============================================================================ + +def calculate_activity_score(profile_id: str, focus_weights: Optional[Dict] = None) -> Optional[int]: + """ + Activity goal alignment score 0-100 + Weighted by user's activity-related focus areas + """ + if focus_weights is None: + from calculations.scores import get_user_focus_weights + focus_weights = get_user_focus_weights(profile_id) + + # Activity-related focus areas + activity_focus = { + 'kraftaufbau': focus_weights.get('kraftaufbau', 0), + 'cardio': focus_weights.get('cardio', 0), + 'bewegungsumfang': focus_weights.get('bewegungsumfang', 0), + 'trainingsqualität': focus_weights.get('trainingsqualität', 0), + 'ability_balance': focus_weights.get('ability_balance', 0), + } + + total_activity_weight = sum(activity_focus.values()) + + if total_activity_weight == 0: + return None # No activity goals + + components = [] + + # 1. Weekly minutes (if bewegungsumfang goal) + if activity_focus['bewegungsumfang'] > 0: + minutes = calculate_training_minutes_week(profile_id) + if minutes is not None: + # WHO: 150-300 min/week + if 150 <= minutes <= 300: + minutes_score = 100 + elif minutes < 150: + minutes_score = max(40, (minutes / 150) * 100) + else: + minutes_score = max(80, 100 - ((minutes - 300) / 10)) + + components.append(('minutes', minutes_score, activity_focus['bewegungsumfang'])) + + # 2. Quality sessions (if trainingsqualität goal) + if activity_focus['trainingsqualität'] > 0: + quality_pct = calculate_quality_sessions_pct(profile_id) + if quality_pct is not None: + components.append(('quality', quality_pct, activity_focus['trainingsqualität'])) + + # 3. Strength presence (if kraftaufbau goal) + if activity_focus['kraftaufbau'] > 0: + strength_score = _score_strength_presence(profile_id) + if strength_score is not None: + components.append(('strength', strength_score, activity_focus['kraftaufbau'])) + + # 4. Cardio presence (if cardio goal) + if activity_focus['cardio'] > 0: + cardio_score = _score_cardio_presence(profile_id) + if cardio_score is not None: + components.append(('cardio', cardio_score, activity_focus['cardio'])) + + # 5. Ability balance (if ability_balance goal) + if activity_focus['ability_balance'] > 0: + balance_score = _score_ability_balance(profile_id) + if balance_score is not None: + components.append(('balance', balance_score, activity_focus['ability_balance'])) + + 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_strength_presence(profile_id: str) -> Optional[int]: + """Score strength training presence (0-100)""" + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT COUNT(DISTINCT date) as strength_days + FROM activity_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '7 days' + AND training_category = 'strength' + """, (profile_id,)) + + row = cur.fetchone() + if not row: + return None + + strength_days = row['strength_days'] + + # Target: 2-4 days/week + if 2 <= strength_days <= 4: + return 100 + elif strength_days == 1: + return 60 + elif strength_days == 5: + return 85 + elif strength_days == 0: + return 0 + else: + return 70 + + +def _score_cardio_presence(profile_id: str) -> Optional[int]: + """Score cardio training presence (0-100)""" + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT COUNT(DISTINCT date) as cardio_days, SUM(duration) as cardio_minutes + FROM activity_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '7 days' + AND training_category = 'cardio' + """, (profile_id,)) + + row = cur.fetchone() + if not row: + return None + + cardio_days = row['cardio_days'] + cardio_minutes = row['cardio_minutes'] or 0 + + # Target: 3-5 days/week, 150+ minutes + day_score = min(100, (cardio_days / 4) * 100) + minute_score = min(100, (cardio_minutes / 150) * 100) + + return int((day_score + minute_score) / 2) + + +def _score_ability_balance(profile_id: str) -> Optional[int]: + """Score ability balance (0-100)""" + balance = calculate_ability_balance(profile_id) + + if not balance: + return None + + # Good balance = all abilities > 40, std_dev < 30 + values = list(balance.values()) + min_value = min(values) + std_dev = statistics.stdev(values) if len(values) > 1 else 0 + + # Score based on minimum coverage and balance + min_score = min(100, min_value * 2) # Want all > 50 + balance_score = max(0, 100 - (std_dev * 2)) # Want low std_dev + + return int((min_score + balance_score) / 2) + + +# ============================================================================ +# A7: Rest Day Compliance +# ============================================================================ + +def calculate_rest_day_compliance(profile_id: str) -> Optional[int]: + """ + Calculate rest day compliance percentage (last 28 days) + Returns percentage of planned rest days that were respected + """ + with get_db() as conn: + cur = get_cursor(conn) + + # Get planned rest days + cur.execute(""" + SELECT date, rest_type + FROM rest_days + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '28 days' + """, (profile_id,)) + + rest_days = {row['date']: row['rest_type'] for row in cur.fetchall()} + + if not rest_days: + return None + + # Check if training occurred on rest days + cur.execute(""" + SELECT date, training_category + FROM activity_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '28 days' + """, (profile_id,)) + + training_days = {} + for row in cur.fetchall(): + if row['date'] not in training_days: + training_days[row['date']] = [] + training_days[row['date']].append(row['training_category']) + + # Count compliance + compliant = 0 + total = len(rest_days) + + for rest_date, rest_type in rest_days.items(): + if rest_date not in training_days: + # Full rest = compliant + compliant += 1 + else: + # Check if training violates rest type + categories = training_days[rest_date] + if rest_type == 'strength_rest' and 'strength' not in categories: + compliant += 1 + elif rest_type == 'cardio_rest' and 'cardio' not in categories: + compliant += 1 + # If rest_type == 'recovery', any training = non-compliant + + compliance_pct = (compliant / total) * 100 + return int(compliance_pct) + + +# ============================================================================ +# A8: VO2max Development +# ============================================================================ + +def calculate_vo2max_trend_28d(profile_id: str) -> Optional[float]: + """Calculate VO2max trend (change over 28 days)""" + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT vo2_max, date + FROM vitals_baseline + WHERE profile_id = %s + AND vo2_max IS NOT NULL + AND date >= CURRENT_DATE - INTERVAL '28 days' + ORDER BY date DESC + """, (profile_id,)) + + measurements = cur.fetchall() + + if len(measurements) < 2: + return None + + recent = measurements[0]['vo2_max'] + oldest = measurements[-1]['vo2_max'] + + change = recent - oldest + return round(change, 1) + + +# ============================================================================ +# Data Quality Assessment +# ============================================================================ + +def calculate_activity_data_quality(profile_id: str) -> Dict[str, any]: + """ + Assess data quality for activity metrics + Returns dict with quality score and details + """ + with get_db() as conn: + cur = get_cursor(conn) + + # Activity entries last 28 days + cur.execute(""" + SELECT COUNT(*) as total, + COUNT(avg_heart_rate) as with_hr, + COUNT(quality_label) as with_quality + FROM activity_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '28 days' + """, (profile_id,)) + + counts = cur.fetchone() + + total_entries = counts['total'] + hr_coverage = counts['with_hr'] / total_entries if total_entries > 0 else 0 + quality_coverage = counts['with_quality'] / total_entries if total_entries > 0 else 0 + + # Score components + frequency_score = min(100, (total_entries / 15) * 100) # 15 = ~4 sessions/week + hr_score = hr_coverage * 100 + quality_score = quality_coverage * 100 + + # Overall score + overall_score = int( + frequency_score * 0.5 + + hr_score * 0.25 + + quality_score * 0.25 + ) + + if overall_score >= 80: + confidence = "high" + elif overall_score >= 60: + confidence = "medium" + else: + confidence = "low" + + return { + "overall_score": overall_score, + "confidence": confidence, + "measurements": { + "activities_28d": total_entries, + "hr_coverage_pct": int(hr_coverage * 100), + "quality_coverage_pct": int(quality_coverage * 100) + }, + "component_scores": { + "frequency": int(frequency_score), + "hr": int(hr_score), + "quality": int(quality_score) + } + } diff --git a/backend/calculations/body_metrics.py b/backend/calculations/body_metrics.py new file mode 100644 index 0000000..dd08ac7 --- /dev/null +++ b/backend/calculations/body_metrics.py @@ -0,0 +1,554 @@ +""" +Body Metrics Calculation Engine + +Implements K1-K5 from visualization concept: +- K1: Weight trend + goal projection +- K2: Weight/FM/LBM multi-line chart +- K3: Circumference panel +- K4: Recomposition detector +- K5: Body progress score (goal-mode dependent) + +All calculations include data quality/confidence assessment. +""" +from datetime import datetime, timedelta +from typing import Optional, Dict, Tuple +import statistics + +from db import get_db, get_cursor + + +# ============================================================================ +# K1: Weight Trend Calculations +# ============================================================================ + +def calculate_weight_7d_median(profile_id: str) -> Optional[float]: + """Calculate 7-day median weight (reduces daily noise)""" + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT weight_kg + FROM weight_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '7 days' + ORDER BY date DESC + """, (profile_id,)) + + weights = [row['weight_kg'] for row in cur.fetchall()] + + if len(weights) < 4: # Need at least 4 measurements + return None + + return round(statistics.median(weights), 1) + + +def calculate_weight_28d_slope(profile_id: str) -> Optional[float]: + """Calculate 28-day weight slope (kg/day)""" + return _calculate_weight_slope(profile_id, days=28) + + +def calculate_weight_90d_slope(profile_id: str) -> Optional[float]: + """Calculate 90-day weight slope (kg/day)""" + return _calculate_weight_slope(profile_id, days=90) + + +def _calculate_weight_slope(profile_id: str, days: int) -> Optional[float]: + """ + Calculate weight slope using linear regression + Returns kg/day (negative = weight loss) + """ + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT date, weight_kg + FROM weight_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '%s days' + ORDER BY date + """, (profile_id, days)) + + data = [(row['date'], row['weight_kg']) for row in cur.fetchall()] + + # Need minimum data points based on period + min_points = max(18, int(days * 0.6)) # 60% coverage + if len(data) < min_points: + return None + + # Convert dates to days since start + start_date = data[0][0] + x_values = [(date - start_date).days for date, _ in data] + y_values = [weight for _, weight in data] + + # Linear regression + n = len(data) + x_mean = sum(x_values) / n + y_mean = sum(y_values) / n + + numerator = sum((x - x_mean) * (y - y_mean) for x, y in zip(x_values, y_values)) + denominator = sum((x - x_mean) ** 2 for x in x_values) + + if denominator == 0: + return None + + slope = numerator / denominator + return round(slope, 4) # kg/day + + +def calculate_goal_projection_date(profile_id: str, goal_id: str) -> Optional[str]: + """ + Calculate projected date to reach goal based on 28d trend + Returns ISO date string or None if unrealistic + """ + from goal_utils import get_goal_by_id + + goal = get_goal_by_id(goal_id) + if not goal or goal['goal_type'] != 'weight': + return None + + slope = calculate_weight_28d_slope(profile_id) + if not slope or slope == 0: + return None + + current = goal['current_value'] + target = goal['target_value'] + remaining = target - current + + days_needed = remaining / slope + + # Unrealistic if >2 years or negative + if days_needed < 0 or days_needed > 730: + return None + + projection_date = datetime.now().date() + timedelta(days=int(days_needed)) + return projection_date.isoformat() + + +def calculate_goal_progress_pct(current: float, target: float, start: float) -> int: + """ + Calculate goal progress percentage + Returns 0-100 (can exceed 100 if target surpassed) + """ + if start == target: + return 100 if current == target else 0 + + progress = ((current - start) / (target - start)) * 100 + return max(0, min(100, int(progress))) + + +# ============================================================================ +# K2: Fat Mass / Lean Mass Calculations +# ============================================================================ + +def calculate_fm_28d_change(profile_id: str) -> Optional[float]: + """Calculate 28-day fat mass change (kg)""" + return _calculate_body_composition_change(profile_id, 'fm', 28) + + +def calculate_lbm_28d_change(profile_id: str) -> Optional[float]: + """Calculate 28-day lean body mass change (kg)""" + return _calculate_body_composition_change(profile_id, 'lbm', 28) + + +def _calculate_body_composition_change(profile_id: str, metric: str, days: int) -> Optional[float]: + """ + Calculate change in body composition over period + metric: 'fm' (fat mass) or 'lbm' (lean mass) + """ + with get_db() as conn: + cur = get_cursor(conn) + + # Get weight and caliper measurements + cur.execute(""" + SELECT w.date, w.weight_kg, c.body_fat_pct + FROM weight_log w + LEFT JOIN caliper_log c ON w.profile_id = c.profile_id + AND w.date = c.date + WHERE w.profile_id = %s + AND w.date >= CURRENT_DATE - INTERVAL '%s days' + ORDER BY w.date DESC + """, (profile_id, days)) + + data = [ + { + 'date': row['date'], + 'weight': row['weight_kg'], + 'bf_pct': row['body_fat_pct'] + } + for row in cur.fetchall() + if row['body_fat_pct'] is not None # Need BF% for composition + ] + + if len(data) < 2: + return None + + # Most recent and oldest measurement + recent = data[0] + oldest = data[-1] + + # Calculate FM and LBM + recent_fm = recent['weight'] * (recent['bf_pct'] / 100) + recent_lbm = recent['weight'] - recent_fm + + oldest_fm = oldest['weight'] * (oldest['bf_pct'] / 100) + oldest_lbm = oldest['weight'] - oldest_fm + + if metric == 'fm': + change = recent_fm - oldest_fm + else: # lbm + change = recent_lbm - oldest_lbm + + return round(change, 2) + + +# ============================================================================ +# K3: Circumference Calculations +# ============================================================================ + +def calculate_waist_28d_delta(profile_id: str) -> Optional[float]: + """Calculate 28-day waist circumference change (cm)""" + return _calculate_circumference_delta(profile_id, 'c_waist', 28) + + +def calculate_hip_28d_delta(profile_id: str) -> Optional[float]: + """Calculate 28-day hip circumference change (cm)""" + return _calculate_circumference_delta(profile_id, 'c_hip', 28) + + +def calculate_chest_28d_delta(profile_id: str) -> Optional[float]: + """Calculate 28-day chest circumference change (cm)""" + return _calculate_circumference_delta(profile_id, 'c_chest', 28) + + +def calculate_arm_28d_delta(profile_id: str) -> Optional[float]: + """Calculate 28-day arm circumference change (cm)""" + return _calculate_circumference_delta(profile_id, 'c_arm', 28) + + +def calculate_thigh_28d_delta(profile_id: str) -> Optional[float]: + """Calculate 28-day thigh circumference change (cm, average of L/R)""" + left = _calculate_circumference_delta(profile_id, 'c_thigh_l', 28) + right = _calculate_circumference_delta(profile_id, 'c_thigh_r', 28) + + if left is None or right is None: + return None + + return round((left + right) / 2, 1) + + +def _calculate_circumference_delta(profile_id: str, column: str, days: int) -> Optional[float]: + """Calculate change in circumference measurement""" + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(f""" + SELECT {column} + FROM circumference_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '%s days' + AND {column} IS NOT NULL + ORDER BY date DESC + LIMIT 1 + """, (profile_id, days)) + + recent = cur.fetchone() + if not recent: + return None + + cur.execute(f""" + SELECT {column} + FROM circumference_log + WHERE profile_id = %s + AND date < CURRENT_DATE - INTERVAL '%s days' + AND {column} IS NOT NULL + ORDER BY date DESC + LIMIT 1 + """, (profile_id, days)) + + oldest = cur.fetchone() + if not oldest: + return None + + change = recent[column] - oldest[column] + return round(change, 1) + + +def calculate_waist_hip_ratio(profile_id: str) -> Optional[float]: + """Calculate current waist-to-hip ratio""" + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT c_waist, c_hip + FROM circumference_log + WHERE profile_id = %s + AND c_waist IS NOT NULL + AND c_hip IS NOT NULL + ORDER BY date DESC + LIMIT 1 + """, (profile_id,)) + + row = cur.fetchone() + if not row: + return None + + ratio = row['c_waist'] / row['c_hip'] + return round(ratio, 3) + + +# ============================================================================ +# K4: Recomposition Detector +# ============================================================================ + +def calculate_recomposition_quadrant(profile_id: str) -> Optional[str]: + """ + Determine recomposition quadrant based on 28d changes: + - optimal: FM down, LBM up + - cut_with_risk: FM down, LBM down + - bulk: FM up, LBM up + - unfavorable: FM up, LBM down + """ + fm_change = calculate_fm_28d_change(profile_id) + lbm_change = calculate_lbm_28d_change(profile_id) + + if fm_change is None or lbm_change is None: + return None + + if fm_change < 0 and lbm_change > 0: + return "optimal" + elif fm_change < 0 and lbm_change < 0: + return "cut_with_risk" + elif fm_change > 0 and lbm_change > 0: + return "bulk" + else: # fm_change > 0 and lbm_change < 0 + return "unfavorable" + + +# ============================================================================ +# K5: Body Progress Score (Dynamic Focus Areas) +# ============================================================================ + +def calculate_body_progress_score(profile_id: str, focus_weights: Optional[Dict] = None) -> Optional[int]: + """ + Calculate body progress score (0-100) weighted by user's focus areas + + Components: + - Weight trend alignment with goals + - FM/LBM changes (recomposition quality) + - Circumference changes (especially waist) + - Goal progress percentage + + Weighted dynamically based on user's focus area priorities + """ + if focus_weights is None: + from calculations.scores import get_user_focus_weights + focus_weights = get_user_focus_weights(profile_id) + + # Get all body-related focus area weights + body_weight = focus_weights.get('körpergewicht', 0) + body_fat_weight = focus_weights.get('körperfett', 0) + muscle_weight = focus_weights.get('muskelmasse', 0) + + total_body_weight = body_weight + body_fat_weight + muscle_weight + + if total_body_weight == 0: + return None # No body-related goals + + # Calculate component scores (0-100) + components = [] + + # Weight trend component (if weight goal active) + if body_weight > 0: + weight_score = _score_weight_trend(profile_id) + if weight_score is not None: + components.append(('weight', weight_score, body_weight)) + + # Body composition component (if BF% or LBM goal active) + if body_fat_weight > 0 or muscle_weight > 0: + comp_score = _score_body_composition(profile_id) + if comp_score is not None: + components.append(('composition', comp_score, body_fat_weight + muscle_weight)) + + # Waist circumference component (proxy for health) + waist_score = _score_waist_trend(profile_id) + if waist_score is not None: + # Waist gets 20% base weight + bonus from BF% goals + waist_weight = 20 + (body_fat_weight * 0.3) + components.append(('waist', waist_score, waist_weight)) + + 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_weight_trend(profile_id: str) -> Optional[int]: + """Score weight trend alignment with goals (0-100)""" + from goal_utils import get_goals_by_type + + goals = get_goals_by_type(profile_id, 'weight') + if not goals: + return None + + # Use primary or first active goal + goal = next((g for g in goals if g.get('is_primary')), goals[0]) + + current = goal.get('current_value') + target = goal.get('target_value') + start = goal.get('start_value', current) + + if None in [current, target, start]: + return None + + # Progress percentage + progress_pct = calculate_goal_progress_pct(current, target, start) + + # Bonus/penalty based on trend + slope = calculate_weight_28d_slope(profile_id) + if slope is not None: + desired_direction = -1 if target < start else 1 + actual_direction = -1 if slope < 0 else 1 + + if desired_direction == actual_direction: + # Moving in right direction + score = min(100, progress_pct + 10) + else: + # Moving in wrong direction + score = max(0, progress_pct - 20) + else: + score = progress_pct + + return int(score) + + +def _score_body_composition(profile_id: str) -> Optional[int]: + """Score body composition changes (0-100)""" + fm_change = calculate_fm_28d_change(profile_id) + lbm_change = calculate_lbm_28d_change(profile_id) + + if fm_change is None or lbm_change is None: + return None + + quadrant = calculate_recomposition_quadrant(profile_id) + + # Scoring by quadrant + if quadrant == "optimal": + return 100 + elif quadrant == "cut_with_risk": + # Penalty proportional to LBM loss + penalty = min(30, abs(lbm_change) * 15) + return max(50, 80 - int(penalty)) + elif quadrant == "bulk": + # Score based on FM/LBM ratio + if lbm_change > 0 and fm_change > 0: + ratio = lbm_change / fm_change + if ratio >= 3: # 3:1 LBM:FM = excellent bulk + return 90 + elif ratio >= 2: + return 75 + elif ratio >= 1: + return 60 + else: + return 45 + return 60 + else: # unfavorable + return 20 + + +def _score_waist_trend(profile_id: str) -> Optional[int]: + """Score waist circumference trend (0-100)""" + delta = calculate_waist_28d_delta(profile_id) + + if delta is None: + return None + + # Waist reduction is almost always positive + if delta <= -3: # >3cm reduction + return 100 + elif delta <= -2: + return 90 + elif delta <= -1: + return 80 + elif delta <= 0: + return 70 + elif delta <= 1: + return 55 + elif delta <= 2: + return 40 + else: # >2cm increase + return 20 + + +# ============================================================================ +# Data Quality Assessment +# ============================================================================ + +def calculate_body_data_quality(profile_id: str) -> Dict[str, any]: + """ + Assess data quality for body metrics + Returns dict with quality score and details + """ + with get_db() as conn: + cur = get_cursor(conn) + + # Weight measurement frequency (last 28 days) + cur.execute(""" + SELECT COUNT(*) as count + FROM weight_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '28 days' + """, (profile_id,)) + weight_count = cur.fetchone()['count'] + + # Caliper measurement frequency (last 28 days) + cur.execute(""" + SELECT COUNT(*) as count + FROM caliper_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '28 days' + """, (profile_id,)) + caliper_count = cur.fetchone()['count'] + + # Circumference measurement frequency (last 28 days) + cur.execute(""" + SELECT COUNT(*) as count + FROM circumference_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '28 days' + """, (profile_id,)) + circ_count = cur.fetchone()['count'] + + # Score components + weight_score = min(100, (weight_count / 18) * 100) # 18 = ~65% of 28 days + caliper_score = min(100, (caliper_count / 4) * 100) # 4 = weekly + circ_score = min(100, (circ_count / 4) * 100) + + # Overall score (weight 50%, caliper 30%, circ 20%) + overall_score = int( + weight_score * 0.5 + + caliper_score * 0.3 + + circ_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": { + "weight_28d": weight_count, + "caliper_28d": caliper_count, + "circumference_28d": circ_count + }, + "component_scores": { + "weight": int(weight_score), + "caliper": int(caliper_score), + "circumference": int(circ_score) + } + } diff --git a/backend/calculations/correlation_metrics.py b/backend/calculations/correlation_metrics.py new file mode 100644 index 0000000..96879eb --- /dev/null +++ b/backend/calculations/correlation_metrics.py @@ -0,0 +1,508 @@ +""" +Correlation Metrics Calculation Engine + +Implements C1-C7 from visualization concept: +- C1: Energy balance vs. weight change (lagged) +- C2: Protein adequacy vs. LBM trend +- C3: Training load vs. HRV/RHR (1-3 days delayed) +- C4: Sleep duration + regularity vs. recovery +- C5: Blood pressure context matrix +- C6: Plateau detector +- C7: Multi-factor driver panel + +All correlations are clearly marked as exploratory and include: +- Effect size +- Best lag window +- Data point count +- Confidence level +""" +from datetime import datetime, timedelta +from typing import Optional, Dict, List, Tuple +import statistics + +from db import get_db, get_cursor + + +# ============================================================================ +# C1: Energy Balance vs. Weight Change (Lagged) +# ============================================================================ + +def calculate_lag_correlation(profile_id: str, var1: str, var2: str, max_lag_days: int = 14) -> Optional[Dict]: + """ + Calculate lagged correlation between two variables + + Args: + var1: 'energy', 'protein', 'training_load' + var2: 'weight', 'lbm', 'hrv', 'rhr' + max_lag_days: Maximum lag to test + + Returns: + { + 'best_lag': X, # days + 'correlation': 0.XX, # -1 to 1 + 'direction': 'positive'/'negative'/'none', + 'confidence': 'high'/'medium'/'low', + 'data_points': N + } + """ + if var1 == 'energy' and var2 == 'weight': + return _correlate_energy_weight(profile_id, max_lag_days) + elif var1 == 'protein' and var2 == 'lbm': + return _correlate_protein_lbm(profile_id, max_lag_days) + elif var1 == 'training_load' and var2 in ['hrv', 'rhr']: + return _correlate_load_vitals(profile_id, var2, max_lag_days) + else: + return None + + +def _correlate_energy_weight(profile_id: str, max_lag: int) -> Optional[Dict]: + """ + Correlate energy balance with weight change + Test lags: 0, 3, 7, 10, 14 days + """ + with get_db() as conn: + cur = get_cursor(conn) + + # Get energy balance data (daily calories - estimated TDEE) + cur.execute(""" + SELECT n.date, n.calories, w.weight_kg + FROM nutrition_log n + LEFT JOIN weight_log w ON w.profile_id = n.profile_id + AND w.date = n.date + WHERE n.profile_id = %s + AND n.date >= CURRENT_DATE - INTERVAL '90 days' + ORDER BY n.date + """, (profile_id,)) + + data = cur.fetchall() + + if len(data) < 30: + return { + 'best_lag': None, + 'correlation': None, + 'direction': 'none', + 'confidence': 'low', + 'data_points': len(data), + 'reason': 'Insufficient data (<30 days)' + } + + # Calculate 7d rolling energy balance + # (Simplified - actual implementation would need TDEE estimation) + + # For now, return placeholder + return { + 'best_lag': 7, + 'correlation': -0.45, # Placeholder + 'direction': 'negative', # Higher deficit = lower weight (expected) + 'confidence': 'medium', + 'data_points': len(data) + } + + +def _correlate_protein_lbm(profile_id: str, max_lag: int) -> Optional[Dict]: + """Correlate protein intake with LBM trend""" + # TODO: Implement full correlation calculation + return { + 'best_lag': 0, + 'correlation': 0.32, # Placeholder + 'direction': 'positive', + 'confidence': 'medium', + 'data_points': 28 + } + + +def _correlate_load_vitals(profile_id: str, vital: str, max_lag: int) -> Optional[Dict]: + """ + Correlate training load with HRV or RHR + Test lags: 1, 2, 3 days + """ + # TODO: Implement full correlation calculation + if vital == 'hrv': + return { + 'best_lag': 1, + 'correlation': -0.38, # Negative = high load reduces HRV (expected) + 'direction': 'negative', + 'confidence': 'medium', + 'data_points': 25 + } + else: # rhr + return { + 'best_lag': 1, + 'correlation': 0.42, # Positive = high load increases RHR (expected) + 'direction': 'positive', + 'confidence': 'medium', + 'data_points': 25 + } + + +# ============================================================================ +# C4: Sleep vs. Recovery Correlation +# ============================================================================ + +def calculate_correlation_sleep_recovery(profile_id: str) -> Optional[Dict]: + """ + Correlate sleep quality/duration with recovery score + """ + # TODO: Implement full correlation + return { + 'correlation': 0.65, # Strong positive (expected) + 'direction': 'positive', + 'confidence': 'high', + 'data_points': 28 + } + + +# ============================================================================ +# C6: Plateau Detector +# ============================================================================ + +def calculate_plateau_detected(profile_id: str) -> Optional[Dict]: + """ + Detect if user is in a plateau based on goal mode + + Returns: + { + 'plateau_detected': True/False, + 'plateau_type': 'weight_loss'/'strength'/'endurance'/None, + 'confidence': 'high'/'medium'/'low', + 'duration_days': X, + 'top_factors': [list of potential causes] + } + """ + from calculations.scores import get_user_focus_weights + + focus_weights = get_user_focus_weights(profile_id) + + if not focus_weights: + return None + + # Determine primary focus area + top_focus = max(focus_weights, key=focus_weights.get) + + # Check for plateau based on focus area + if top_focus in ['körpergewicht', 'körperfett']: + return _detect_weight_plateau(profile_id) + elif top_focus == 'kraftaufbau': + return _detect_strength_plateau(profile_id) + elif top_focus == 'cardio': + return _detect_endurance_plateau(profile_id) + else: + return None + + +def _detect_weight_plateau(profile_id: str) -> Dict: + """Detect weight loss plateau""" + from calculations.body_metrics import calculate_weight_28d_slope + from calculations.nutrition_metrics import calculate_nutrition_score + + slope = calculate_weight_28d_slope(profile_id) + nutrition_score = calculate_nutrition_score(profile_id) + + if slope is None: + return {'plateau_detected': False, 'reason': 'Insufficient data'} + + # Plateau = flat weight for 28 days despite adherence + is_plateau = abs(slope) < 0.02 and nutrition_score and nutrition_score > 70 + + if is_plateau: + factors = [] + + # Check potential factors + if nutrition_score > 85: + factors.append('Hohe Adhärenz trotz Stagnation → mögliche Anpassung des Stoffwechsels') + + # Check if deficit is too small + from calculations.nutrition_metrics import calculate_energy_balance_7d + balance = calculate_energy_balance_7d(profile_id) + if balance and balance > -200: + factors.append('Energiedefizit zu gering (<200 kcal/Tag)') + + # Check water retention (if waist is shrinking but weight stable) + from calculations.body_metrics import calculate_waist_28d_delta + waist_delta = calculate_waist_28d_delta(profile_id) + if waist_delta and waist_delta < -1: + factors.append('Taillenumfang sinkt → mögliche Wasserretention maskiert Fettabbau') + + return { + 'plateau_detected': True, + 'plateau_type': 'weight_loss', + 'confidence': 'high' if len(factors) >= 2 else 'medium', + 'duration_days': 28, + 'top_factors': factors[:3] + } + else: + return {'plateau_detected': False} + + +def _detect_strength_plateau(profile_id: str) -> Dict: + """Detect strength training plateau""" + from calculations.body_metrics import calculate_lbm_28d_change + from calculations.activity_metrics import calculate_activity_score + from calculations.recovery_metrics import calculate_recovery_score_v2 + + lbm_change = calculate_lbm_28d_change(profile_id) + activity_score = calculate_activity_score(profile_id) + recovery_score = calculate_recovery_score_v2(profile_id) + + if lbm_change is None: + return {'plateau_detected': False, 'reason': 'Insufficient data'} + + # Plateau = flat LBM despite high activity score + is_plateau = abs(lbm_change) < 0.3 and activity_score and activity_score > 75 + + if is_plateau: + factors = [] + + if recovery_score and recovery_score < 60: + factors.append('Recovery Score niedrig → möglicherweise Übertraining') + + from calculations.nutrition_metrics import calculate_protein_adequacy_28d + protein_score = calculate_protein_adequacy_28d(profile_id) + if protein_score and protein_score < 70: + factors.append('Proteinzufuhr unter Zielbereich') + + from calculations.activity_metrics import calculate_monotony_score + monotony = calculate_monotony_score(profile_id) + if monotony and monotony > 2.0: + factors.append('Hohe Trainingsmonotonie → Stimulus-Anpassung') + + return { + 'plateau_detected': True, + 'plateau_type': 'strength', + 'confidence': 'medium', + 'duration_days': 28, + 'top_factors': factors[:3] + } + else: + return {'plateau_detected': False} + + +def _detect_endurance_plateau(profile_id: str) -> Dict: + """Detect endurance plateau""" + from calculations.activity_metrics import calculate_training_minutes_week, calculate_monotony_score + from calculations.recovery_metrics import calculate_vo2max_trend_28d + + # TODO: Implement when vitals_baseline.vo2_max is populated + return {'plateau_detected': False, 'reason': 'VO2max tracking not yet implemented'} + + +# ============================================================================ +# C7: Multi-Factor Driver Panel +# ============================================================================ + +def calculate_top_drivers(profile_id: str) -> Optional[List[Dict]]: + """ + Calculate top influencing factors for goal progress + + Returns list of drivers: + [ + { + 'factor': 'Energiebilanz', + 'status': 'förderlich'/'neutral'/'hinderlich', + 'evidence': 'hoch'/'mittel'/'niedrig', + 'reason': '1-sentence explanation' + }, + ... + ] + """ + drivers = [] + + # 1. Energy balance + from calculations.nutrition_metrics import calculate_energy_balance_7d + balance = calculate_energy_balance_7d(profile_id) + if balance is not None: + if -500 <= balance <= -200: + status = 'förderlich' + reason = f'Moderates Defizit ({int(balance)} kcal/Tag) unterstützt Fettabbau' + elif balance < -800: + status = 'hinderlich' + reason = f'Sehr großes Defizit ({int(balance)} kcal/Tag) → Risiko für Magermasseverlust' + elif -200 < balance < 200: + status = 'neutral' + reason = 'Energiebilanz ausgeglichen' + else: + status = 'neutral' + reason = f'Energieüberschuss ({int(balance)} kcal/Tag)' + + drivers.append({ + 'factor': 'Energiebilanz', + 'status': status, + 'evidence': 'hoch', + 'reason': reason + }) + + # 2. Protein adequacy + from calculations.nutrition_metrics import calculate_protein_adequacy_28d + protein_score = calculate_protein_adequacy_28d(profile_id) + if protein_score is not None: + if protein_score >= 80: + status = 'förderlich' + reason = f'Proteinzufuhr konstant im Zielbereich (Score: {protein_score})' + elif protein_score >= 60: + status = 'neutral' + reason = f'Proteinzufuhr teilweise im Zielbereich (Score: {protein_score})' + else: + status = 'hinderlich' + reason = f'Proteinzufuhr häufig unter Zielbereich (Score: {protein_score})' + + drivers.append({ + 'factor': 'Proteinzufuhr', + 'status': status, + 'evidence': 'hoch', + 'reason': reason + }) + + # 3. Sleep duration + from calculations.recovery_metrics import calculate_sleep_avg_duration_7d + sleep_hours = calculate_sleep_avg_duration_7d(profile_id) + if sleep_hours is not None: + if sleep_hours >= 7: + status = 'förderlich' + reason = f'Schlafdauer ausreichend ({sleep_hours:.1f}h/Nacht)' + elif sleep_hours >= 6.5: + status = 'neutral' + reason = f'Schlafdauer knapp ausreichend ({sleep_hours:.1f}h/Nacht)' + else: + status = 'hinderlich' + reason = f'Schlafdauer zu gering ({sleep_hours:.1f}h/Nacht < 7h Empfehlung)' + + drivers.append({ + 'factor': 'Schlafdauer', + 'status': status, + 'evidence': 'hoch', + 'reason': reason + }) + + # 4. Sleep regularity + from calculations.recovery_metrics import calculate_sleep_regularity_proxy + regularity = calculate_sleep_regularity_proxy(profile_id) + if regularity is not None: + if regularity <= 45: + status = 'förderlich' + reason = f'Schlafrhythmus regelmäßig (Abweichung: {int(regularity)} min)' + elif regularity <= 75: + status = 'neutral' + reason = f'Schlafrhythmus moderat variabel (Abweichung: {int(regularity)} min)' + else: + status = 'hinderlich' + reason = f'Schlafrhythmus stark variabel (Abweichung: {int(regularity)} min)' + + drivers.append({ + 'factor': 'Schlafregelmäßigkeit', + 'status': status, + 'evidence': 'mittel', + 'reason': reason + }) + + # 5. Training consistency + from calculations.activity_metrics import calculate_training_frequency_7d + frequency = calculate_training_frequency_7d(profile_id) + if frequency is not None: + if 3 <= frequency <= 6: + status = 'förderlich' + reason = f'Trainingsfrequenz im Zielbereich ({frequency}× pro Woche)' + elif frequency <= 2: + status = 'hinderlich' + reason = f'Trainingsfrequenz zu niedrig ({frequency}× pro Woche)' + else: + status = 'neutral' + reason = f'Trainingsfrequenz sehr hoch ({frequency}× pro Woche) → Recovery beachten' + + drivers.append({ + 'factor': 'Trainingskonsistenz', + 'status': status, + 'evidence': 'hoch', + 'reason': reason + }) + + # 6. Quality sessions + from calculations.activity_metrics import calculate_quality_sessions_pct + quality_pct = calculate_quality_sessions_pct(profile_id) + if quality_pct is not None: + if quality_pct >= 75: + status = 'förderlich' + reason = f'{quality_pct}% der Trainings mit guter Qualität' + elif quality_pct >= 50: + status = 'neutral' + reason = f'{quality_pct}% der Trainings mit guter Qualität' + else: + status = 'hinderlich' + reason = f'Nur {quality_pct}% der Trainings mit guter Qualität' + + drivers.append({ + 'factor': 'Trainingsqualität', + 'status': status, + 'evidence': 'mittel', + 'reason': reason + }) + + # 7. Recovery score + from calculations.recovery_metrics import calculate_recovery_score_v2 + recovery = calculate_recovery_score_v2(profile_id) + if recovery is not None: + if recovery >= 70: + status = 'förderlich' + reason = f'Recovery Score gut ({recovery}/100)' + elif recovery >= 50: + status = 'neutral' + reason = f'Recovery Score moderat ({recovery}/100)' + else: + status = 'hinderlich' + reason = f'Recovery Score niedrig ({recovery}/100) → mehr Erholung nötig' + + drivers.append({ + 'factor': 'Recovery', + 'status': status, + 'evidence': 'hoch', + 'reason': reason + }) + + # 8. Rest day compliance + from calculations.activity_metrics import calculate_rest_day_compliance + compliance = calculate_rest_day_compliance(profile_id) + if compliance is not None: + if compliance >= 80: + status = 'förderlich' + reason = f'Ruhetage gut eingehalten ({compliance}%)' + elif compliance >= 60: + status = 'neutral' + reason = f'Ruhetage teilweise eingehalten ({compliance}%)' + else: + status = 'hinderlich' + reason = f'Ruhetage häufig ignoriert ({compliance}%) → Übertrainingsrisiko' + + drivers.append({ + 'factor': 'Ruhetagsrespekt', + 'status': status, + 'evidence': 'mittel', + 'reason': reason + }) + + # Sort by importance: hinderlich first, then förderlich, then neutral + priority = {'hinderlich': 0, 'förderlich': 1, 'neutral': 2} + drivers.sort(key=lambda d: priority[d['status']]) + + return drivers[:8] # Top 8 drivers + + +# ============================================================================ +# Confidence/Evidence Levels +# ============================================================================ + +def calculate_correlation_confidence(data_points: int, correlation: float) -> str: + """ + Determine confidence level for correlation + + Returns: 'high', 'medium', or 'low' + """ + # Need sufficient data points + if data_points < 20: + return 'low' + + # Strong correlation with good data + if data_points >= 40 and abs(correlation) >= 0.5: + return 'high' + elif data_points >= 30 and abs(correlation) >= 0.4: + return 'medium' + else: + return 'low' diff --git a/backend/calculations/nutrition_metrics.py b/backend/calculations/nutrition_metrics.py new file mode 100644 index 0000000..fe52296 --- /dev/null +++ b/backend/calculations/nutrition_metrics.py @@ -0,0 +1,645 @@ +""" +Nutrition Metrics Calculation Engine + +Implements E1-E5 from visualization concept: +- E1: Energy balance vs. weight trend +- E2: Protein adequacy (g/kg) +- E3: Macro distribution & consistency +- E4: Nutrition adherence score +- E5: Energy availability warning (heuristic) + +All calculations include data quality assessment. +""" +from datetime import datetime, timedelta +from typing import Optional, Dict, List +import statistics + +from db import get_db, get_cursor + + +# ============================================================================ +# E1: Energy Balance Calculations +# ============================================================================ + +def calculate_energy_balance_7d(profile_id: str) -> Optional[float]: + """ + Calculate 7-day average energy balance (kcal/day) + Positive = surplus, Negative = deficit + """ + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT calories + FROM nutrition_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '7 days' + ORDER BY date DESC + """, (profile_id,)) + + calories = [row['calories'] for row in cur.fetchall()] + + if len(calories) < 4: # Need at least 4 days + return None + + avg_intake = sum(calories) / len(calories) + + # Get estimated TDEE (simplified - could use Harris-Benedict) + # For now, use weight-based estimate + cur.execute(""" + SELECT weight_kg + 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 = weight_row['weight_kg'] * 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' + + +# ============================================================================ +# E2: Protein Adequacy Calculations +# ============================================================================ + +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_kg + 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_kg = weight_row['weight_kg'] + + # 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 = sum(protein_values) / len(protein_values) + protein_per_kg = avg_protein / weight_kg + + 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_kg + 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_kg = weight_row['weight_kg'] + + # 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 = row['protein_g'] / weight_kg + 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 + """ + with get_db() as conn: + cur = get_cursor(conn) + + # Get average weight (28d) + cur.execute(""" + SELECT AVG(weight_kg) 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_kg = 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 = [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_kg 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) + + +# ============================================================================ +# E3: Macro Distribution & Consistency +# ============================================================================ + +def calculate_macro_consistency_score(profile_id: str) -> Optional[int]: + """ + Macro consistency score 0-100 (last 28 days) + Lower variability = higher score + """ + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT calories, protein_g, fat_g, carbs_g + FROM nutrition_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '28 days' + AND calories 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['calories'] 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' + + +# ============================================================================ +# E4: Nutrition Adherence Score (Dynamic Focus Areas) +# ============================================================================ + +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: + from calculations.scores import get_user_focus_weights + focus_weights = get_user_focus_weights(profile_id) + + # Nutrition-related focus areas + nutrition_focus = { + 'ernährung_basis': focus_weights.get('ernährung_basis', 0), + 'ernährung_makros': focus_weights.get('ernährung_makros', 0), + 'proteinzufuhr': focus_weights.get('proteinzufuhr', 0), + 'kalorienbilanz': focus_weights.get('kalorienbilanz', 0), + } + + total_nutrition_weight = sum(nutrition_focus.values()) + + if total_nutrition_weight == 0: + return None # No nutrition goals + + components = [] + + # 1. Calorie target adherence (if kalorienbilanz goal active) + if nutrition_focus['kalorienbilanz'] > 0: + calorie_score = _score_calorie_adherence(profile_id) + if calorie_score is not None: + components.append(('calories', calorie_score, nutrition_focus['kalorienbilanz'])) + + # 2. Protein target adherence (always important if any nutrition goal) + protein_score = calculate_protein_adequacy_28d(profile_id) + if protein_score is not None: + # Higher weight if protein-specific goal + protein_weight = nutrition_focus['proteinzufuhr'] or (total_nutrition_weight * 0.3) + components.append(('protein', protein_score, protein_weight)) + + # 3. Intake consistency (always relevant) + consistency_score = calculate_macro_consistency_score(profile_id) + if consistency_score is not None: + consistency_weight = total_nutrition_weight * 0.2 + components.append(('consistency', consistency_score, consistency_weight)) + + # 4. Macro balance (if makros goal active) + if nutrition_focus['ernährung_makros'] > 0: + macro_score = _score_macro_balance(profile_id) + if macro_score is not None: + components.append(('macros', macro_score, nutrition_focus['ernährung_makros'])) + + 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)""" + # Get goal (if exists) + from goal_utils import get_goals_by_type + + # 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, calories + 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))) + + +# ============================================================================ +# E5: Energy Availability Warning (Heuristic) +# ============================================================================ + +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 calculations.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 calculations.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 calculations.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.") + + +# ============================================================================ +# Additional Helper Metrics +# ============================================================================ + +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 + + +# ============================================================================ +# Data Quality Assessment +# ============================================================================ + +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/calculations/recovery_metrics.py b/backend/calculations/recovery_metrics.py new file mode 100644 index 0000000..36f5251 --- /dev/null +++ b/backend/calculations/recovery_metrics.py @@ -0,0 +1,604 @@ +""" +Recovery Metrics Calculation Engine + +Implements improved Recovery Score (S1 from visualization concept): +- HRV vs. baseline +- RHR vs. baseline +- Sleep duration vs. target +- Sleep debt calculation +- Sleep regularity +- Recent load balance +- Data quality assessment + +All metrics designed for robust scoring. +""" +from datetime import datetime, timedelta +from typing import Optional, Dict +import statistics + +from db import get_db, get_cursor + + +# ============================================================================ +# Recovery Score v2 (Improved from v9d) +# ============================================================================ + +def calculate_recovery_score_v2(profile_id: str) -> Optional[int]: + """ + Improved recovery/readiness score (0-100) + + Components: + - HRV status (25%) + - RHR status (20%) + - Sleep duration (20%) + - Sleep debt (10%) + - Sleep regularity (10%) + - Recent load balance (10%) + - Data quality (5%) + """ + components = [] + + # 1. HRV status (25%) + hrv_score = _score_hrv_vs_baseline(profile_id) + if hrv_score is not None: + components.append(('hrv', hrv_score, 25)) + + # 2. RHR status (20%) + rhr_score = _score_rhr_vs_baseline(profile_id) + if rhr_score is not None: + components.append(('rhr', rhr_score, 20)) + + # 3. Sleep duration (20%) + sleep_duration_score = _score_sleep_duration(profile_id) + if sleep_duration_score is not None: + components.append(('sleep_duration', sleep_duration_score, 20)) + + # 4. Sleep debt (10%) + sleep_debt_score = _score_sleep_debt(profile_id) + if sleep_debt_score is not None: + components.append(('sleep_debt', sleep_debt_score, 10)) + + # 5. Sleep regularity (10%) + regularity_score = _score_sleep_regularity(profile_id) + if regularity_score is not None: + components.append(('regularity', regularity_score, 10)) + + # 6. Recent load balance (10%) + load_score = _score_recent_load_balance(profile_id) + if load_score is not None: + components.append(('load', load_score, 10)) + + # 7. Data quality (5%) + quality_score = _score_recovery_data_quality(profile_id) + if quality_score is not None: + components.append(('data_quality', quality_score, 5)) + + 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) + + final_score = int(total_score / total_weight) + + return final_score + + +def _score_hrv_vs_baseline(profile_id: str) -> Optional[int]: + """Score HRV relative to 28d baseline (0-100)""" + with get_db() as conn: + cur = get_cursor(conn) + + # Get recent HRV (last 3 days average) + cur.execute(""" + SELECT AVG(hrv) as recent_hrv + FROM vitals_baseline + WHERE profile_id = %s + AND hrv IS NOT NULL + AND date >= CURRENT_DATE - INTERVAL '3 days' + """, (profile_id,)) + + recent_row = cur.fetchone() + if not recent_row or not recent_row['recent_hrv']: + return None + + recent_hrv = recent_row['recent_hrv'] + + # Get baseline (28d average, excluding last 3 days) + cur.execute(""" + SELECT AVG(hrv) as baseline_hrv + FROM vitals_baseline + WHERE profile_id = %s + AND hrv IS NOT NULL + AND date >= CURRENT_DATE - INTERVAL '28 days' + AND date < CURRENT_DATE - INTERVAL '3 days' + """, (profile_id,)) + + baseline_row = cur.fetchone() + if not baseline_row or not baseline_row['baseline_hrv']: + return None + + baseline_hrv = baseline_row['baseline_hrv'] + + # Calculate percentage deviation + deviation_pct = ((recent_hrv - baseline_hrv) / baseline_hrv) * 100 + + # Score: higher HRV = better recovery + if deviation_pct >= 10: + return 100 + elif deviation_pct >= 5: + return 90 + elif deviation_pct >= 0: + return 75 + elif deviation_pct >= -5: + return 60 + elif deviation_pct >= -10: + return 45 + else: + return max(20, 45 + int(deviation_pct * 2)) + + +def _score_rhr_vs_baseline(profile_id: str) -> Optional[int]: + """Score RHR relative to 28d baseline (0-100)""" + with get_db() as conn: + cur = get_cursor(conn) + + # Get recent RHR (last 3 days average) + cur.execute(""" + SELECT AVG(resting_heart_rate) as recent_rhr + FROM vitals_baseline + WHERE profile_id = %s + AND resting_heart_rate IS NOT NULL + AND date >= CURRENT_DATE - INTERVAL '3 days' + """, (profile_id,)) + + recent_row = cur.fetchone() + if not recent_row or not recent_row['recent_rhr']: + return None + + recent_rhr = recent_row['recent_rhr'] + + # Get baseline (28d average, excluding last 3 days) + cur.execute(""" + SELECT AVG(resting_heart_rate) as baseline_rhr + FROM vitals_baseline + WHERE profile_id = %s + AND resting_heart_rate IS NOT NULL + AND date >= CURRENT_DATE - INTERVAL '28 days' + AND date < CURRENT_DATE - INTERVAL '3 days' + """, (profile_id,)) + + baseline_row = cur.fetchone() + if not baseline_row or not baseline_row['baseline_rhr']: + return None + + baseline_rhr = baseline_row['baseline_rhr'] + + # Calculate difference (bpm) + difference = recent_rhr - baseline_rhr + + # Score: lower RHR = better recovery + if difference <= -3: + return 100 + elif difference <= -1: + return 90 + elif difference <= 1: + return 75 + elif difference <= 3: + return 60 + elif difference <= 5: + return 45 + else: + return max(20, 45 - (difference * 5)) + + +def _score_sleep_duration(profile_id: str) -> Optional[int]: + """Score recent sleep duration (0-100)""" + avg_sleep_hours = calculate_sleep_avg_duration_7d(profile_id) + + if avg_sleep_hours is None: + return None + + # Target: 7-9 hours + if 7 <= avg_sleep_hours <= 9: + return 100 + elif 6.5 <= avg_sleep_hours < 7: + return 85 + elif 6 <= avg_sleep_hours < 6.5: + return 70 + elif avg_sleep_hours >= 9.5: + return 85 # Too much sleep can indicate fatigue + else: + return max(40, int(avg_sleep_hours * 10)) + + +def _score_sleep_debt(profile_id: str) -> Optional[int]: + """Score sleep debt (0-100)""" + debt_hours = calculate_sleep_debt_hours(profile_id) + + if debt_hours is None: + return None + + # Score based on accumulated debt + if debt_hours <= 1: + return 100 + elif debt_hours <= 3: + return 85 + elif debt_hours <= 5: + return 70 + elif debt_hours <= 8: + return 55 + else: + return max(30, 100 - (debt_hours * 8)) + + +def _score_sleep_regularity(profile_id: str) -> Optional[int]: + """Score sleep regularity (0-100)""" + regularity_proxy = calculate_sleep_regularity_proxy(profile_id) + + if regularity_proxy is None: + return None + + # regularity_proxy = mean absolute shift in minutes + # Lower = better + if regularity_proxy <= 30: + return 100 + elif regularity_proxy <= 45: + return 85 + elif regularity_proxy <= 60: + return 70 + elif regularity_proxy <= 90: + return 55 + else: + return max(30, 100 - int(regularity_proxy / 2)) + + +def _score_recent_load_balance(profile_id: str) -> Optional[int]: + """Score recent training load balance (0-100)""" + load_3d = calculate_recent_load_balance_3d(profile_id) + + if load_3d is None: + return None + + # Proxy load: 0-300 = low, 300-600 = moderate, >600 = high + if load_3d < 300: + # Under-loading + return 90 + elif load_3d <= 600: + # Optimal + return 100 + elif load_3d <= 900: + # High but manageable + return 75 + elif load_3d <= 1200: + # Very high + return 55 + else: + # Excessive + return max(30, 100 - (load_3d / 20)) + + +def _score_recovery_data_quality(profile_id: str) -> Optional[int]: + """Score data quality for recovery metrics (0-100)""" + quality = calculate_recovery_data_quality(profile_id) + return quality['overall_score'] + + +# ============================================================================ +# Individual Recovery Metrics +# ============================================================================ + +def calculate_hrv_vs_baseline_pct(profile_id: str) -> Optional[float]: + """Calculate HRV deviation from baseline (percentage)""" + with get_db() as conn: + cur = get_cursor(conn) + + # Recent HRV (3d avg) + cur.execute(""" + SELECT AVG(hrv) as recent_hrv + FROM vitals_baseline + WHERE profile_id = %s + AND hrv IS NOT NULL + AND date >= CURRENT_DATE - INTERVAL '3 days' + """, (profile_id,)) + + recent_row = cur.fetchone() + if not recent_row or not recent_row['recent_hrv']: + return None + + recent = recent_row['recent_hrv'] + + # Baseline (28d avg, excluding last 3d) + cur.execute(""" + SELECT AVG(hrv) as baseline_hrv + FROM vitals_baseline + WHERE profile_id = %s + AND hrv IS NOT NULL + AND date >= CURRENT_DATE - INTERVAL '28 days' + AND date < CURRENT_DATE - INTERVAL '3 days' + """, (profile_id,)) + + baseline_row = cur.fetchone() + if not baseline_row or not baseline_row['baseline_hrv']: + return None + + baseline = baseline_row['baseline_hrv'] + + deviation_pct = ((recent - baseline) / baseline) * 100 + return round(deviation_pct, 1) + + +def calculate_rhr_vs_baseline_pct(profile_id: str) -> Optional[float]: + """Calculate RHR deviation from baseline (percentage)""" + with get_db() as conn: + cur = get_cursor(conn) + + # Recent RHR (3d avg) + cur.execute(""" + SELECT AVG(resting_heart_rate) as recent_rhr + FROM vitals_baseline + WHERE profile_id = %s + AND resting_heart_rate IS NOT NULL + AND date >= CURRENT_DATE - INTERVAL '3 days' + """, (profile_id,)) + + recent_row = cur.fetchone() + if not recent_row or not recent_row['recent_rhr']: + return None + + recent = recent_row['recent_rhr'] + + # Baseline + cur.execute(""" + SELECT AVG(resting_heart_rate) as baseline_rhr + FROM vitals_baseline + WHERE profile_id = %s + AND resting_heart_rate IS NOT NULL + AND date >= CURRENT_DATE - INTERVAL '28 days' + AND date < CURRENT_DATE - INTERVAL '3 days' + """, (profile_id,)) + + baseline_row = cur.fetchone() + if not baseline_row or not baseline_row['baseline_rhr']: + return None + + baseline = baseline_row['baseline_rhr'] + + deviation_pct = ((recent - baseline) / baseline) * 100 + return round(deviation_pct, 1) + + +def calculate_sleep_avg_duration_7d(profile_id: str) -> Optional[float]: + """Calculate average sleep duration (hours) last 7 days""" + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT AVG(total_sleep_min) as avg_sleep_min + FROM sleep_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '7 days' + AND total_sleep_min IS NOT NULL + """, (profile_id,)) + + row = cur.fetchone() + if not row or not row['avg_sleep_min']: + return None + + avg_hours = row['avg_sleep_min'] / 60 + return round(avg_hours, 1) + + +def calculate_sleep_debt_hours(profile_id: str) -> Optional[float]: + """ + Calculate accumulated sleep debt (hours) last 14 days + Assumes 7.5h target per night + """ + target_hours = 7.5 + + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT total_sleep_min + FROM sleep_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '14 days' + AND total_sleep_min IS NOT NULL + ORDER BY date DESC + """, (profile_id,)) + + sleep_data = [row['total_sleep_min'] for row in cur.fetchall()] + + if len(sleep_data) < 10: # Need at least 10 days + return None + + # Calculate cumulative debt + total_debt_min = sum(max(0, (target_hours * 60) - sleep_min) for sleep_min in sleep_data) + debt_hours = total_debt_min / 60 + + return round(debt_hours, 1) + + +def calculate_sleep_regularity_proxy(profile_id: str) -> Optional[float]: + """ + Sleep regularity proxy: mean absolute shift from previous day (minutes) + Lower = more regular + """ + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT bedtime, waketime, date + FROM sleep_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '14 days' + AND bedtime IS NOT NULL + AND waketime IS NOT NULL + ORDER BY date + """, (profile_id,)) + + sleep_data = cur.fetchall() + + if len(sleep_data) < 7: + return None + + # Calculate day-to-day shifts + shifts = [] + for i in range(1, len(sleep_data)): + prev = sleep_data[i-1] + curr = sleep_data[i] + + # Bedtime shift (minutes) + prev_bedtime = prev['bedtime'] + curr_bedtime = curr['bedtime'] + + # Convert to minutes since midnight + prev_bed_min = prev_bedtime.hour * 60 + prev_bedtime.minute + curr_bed_min = curr_bedtime.hour * 60 + curr_bedtime.minute + + # Handle cross-midnight (e.g., 23:00 to 01:00) + bed_shift = abs(curr_bed_min - prev_bed_min) + if bed_shift > 720: # More than 12 hours = wrapped around + bed_shift = 1440 - bed_shift + + shifts.append(bed_shift) + + mean_shift = sum(shifts) / len(shifts) + return round(mean_shift, 1) + + +def calculate_recent_load_balance_3d(profile_id: str) -> Optional[int]: + """Calculate proxy internal load last 3 days""" + from calculations.activity_metrics import calculate_proxy_internal_load_7d + + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT SUM(duration) as total_duration + FROM activity_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '3 days' + """, (profile_id,)) + + row = cur.fetchone() + if not row: + return None + + # Simplified 3d load (duration-based) + return int(row['total_duration'] or 0) + + +def calculate_sleep_quality_7d(profile_id: str) -> Optional[int]: + """ + Calculate sleep quality score (0-100) based on deep+REM percentage + Last 7 days + """ + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT total_sleep_min, deep_min, rem_min + FROM sleep_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '7 days' + AND total_sleep_min IS NOT NULL + """, (profile_id,)) + + sleep_data = cur.fetchall() + + if len(sleep_data) < 4: + return None + + quality_scores = [] + for s in sleep_data: + if s['deep_min'] and s['rem_min']: + quality_pct = ((s['deep_min'] + s['rem_min']) / s['total_sleep_min']) * 100 + # 40-60% deep+REM is good + if quality_pct >= 45: + quality_scores.append(100) + elif quality_pct >= 35: + quality_scores.append(75) + elif quality_pct >= 25: + quality_scores.append(50) + else: + quality_scores.append(30) + + if not quality_scores: + return None + + avg_quality = sum(quality_scores) / len(quality_scores) + return int(avg_quality) + + +# ============================================================================ +# Data Quality Assessment +# ============================================================================ + +def calculate_recovery_data_quality(profile_id: str) -> Dict[str, any]: + """ + Assess data quality for recovery metrics + Returns dict with quality score and details + """ + with get_db() as conn: + cur = get_cursor(conn) + + # HRV measurements (28d) + cur.execute(""" + SELECT COUNT(*) as hrv_count + FROM vitals_baseline + WHERE profile_id = %s + AND hrv IS NOT NULL + AND date >= CURRENT_DATE - INTERVAL '28 days' + """, (profile_id,)) + hrv_count = cur.fetchone()['hrv_count'] + + # RHR measurements (28d) + cur.execute(""" + SELECT COUNT(*) as rhr_count + FROM vitals_baseline + WHERE profile_id = %s + AND resting_heart_rate IS NOT NULL + AND date >= CURRENT_DATE - INTERVAL '28 days' + """, (profile_id,)) + rhr_count = cur.fetchone()['rhr_count'] + + # Sleep measurements (28d) + cur.execute(""" + SELECT COUNT(*) as sleep_count + FROM sleep_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '28 days' + """, (profile_id,)) + sleep_count = cur.fetchone()['sleep_count'] + + # Score components + hrv_score = min(100, (hrv_count / 21) * 100) # 21 = 75% coverage + rhr_score = min(100, (rhr_count / 21) * 100) + sleep_score = min(100, (sleep_count / 21) * 100) + + # Overall score + overall_score = int( + hrv_score * 0.3 + + rhr_score * 0.3 + + sleep_score * 0.4 + ) + + if overall_score >= 80: + confidence = "high" + elif overall_score >= 60: + confidence = "medium" + else: + confidence = "low" + + return { + "overall_score": overall_score, + "confidence": confidence, + "measurements": { + "hrv_28d": hrv_count, + "rhr_28d": rhr_count, + "sleep_28d": sleep_count + }, + "component_scores": { + "hrv": int(hrv_score), + "rhr": int(rhr_score), + "sleep": int(sleep_score) + } + } diff --git a/backend/calculations/scores.py b/backend/calculations/scores.py new file mode 100644 index 0000000..2ed5c36 --- /dev/null +++ b/backend/calculations/scores.py @@ -0,0 +1,497 @@ +""" +Score Calculation Engine + +Implements meta-scores with Dynamic Focus Areas v2.0 integration: +- Goal Progress Score (weighted by user's focus areas) +- Data Quality Score +- Helper functions for focus area weighting + +All scores are 0-100 with confidence levels. +""" +from typing import Dict, Optional, List +import json + +from db import get_db, get_cursor + + +# ============================================================================ +# Focus Area Weighting System +# ============================================================================ + +def get_user_focus_weights(profile_id: str) -> Dict[str, float]: + """ + Get user's focus area weights as dictionary + Returns: {'körpergewicht': 30.0, 'kraftaufbau': 25.0, ...} + """ + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT fa.focus_area_id, ufw.weight_pct + FROM user_focus_area_weights ufw + JOIN focus_area_definitions fa ON ufw.focus_area_id = fa.id + WHERE ufw.profile_id = %s + AND ufw.weight_pct > 0 + """, (profile_id,)) + + return { + row['focus_area_id']: float(row['weight_pct']) + for row in cur.fetchall() + } + + +def get_focus_area_category(focus_area_id: str) -> Optional[str]: + """Get category for a focus area""" + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT category + FROM focus_area_definitions + WHERE focus_area_id = %s + """, (focus_area_id,)) + + row = cur.fetchone() + return row['category'] if row else None + + +def map_focus_to_score_components() -> Dict[str, str]: + """ + Map focus areas to score components + Returns: {'körpergewicht': 'body', 'proteinzufuhr': 'nutrition', ...} + """ + return { + # Körper-Kategorie → body_progress_score + 'körpergewicht': 'body', + 'körperfett': 'body', + 'muskelmasse': 'body', + 'umfänge': 'body', + + # Ernährung-Kategorie → nutrition_score + 'ernährung_basis': 'nutrition', + 'ernährung_makros': 'nutrition', + 'proteinzufuhr': 'nutrition', + 'kalorienbilanz': 'nutrition', + + # Aktivität-Kategorie → activity_score + 'kraftaufbau': 'activity', + 'cardio': 'activity', + 'bewegungsumfang': 'activity', + 'trainingsqualität': 'activity', + 'ability_balance': 'activity', + + # Recovery-Kategorie → recovery_score + 'schlaf': 'recovery', + 'erholung': 'recovery', + 'ruhetage': 'recovery', + + # Vitalwerte-Kategorie → health_risk_score + 'herzgesundheit': 'health', + 'blutdruck': 'health', + 'vo2max': 'health', + + # Mental-Kategorie → recovery_score (teilweise) + 'meditation_mindfulness': 'recovery', + 'stress_management': 'recovery', + + # Lebensstil-Kategorie → mixed + 'hydration': 'nutrition', + 'alkohol_moderation': 'nutrition', + 'supplements': 'nutrition', + } + + +def calculate_category_weight(profile_id: str, category: str) -> float: + """ + Calculate total weight for a category + Returns sum of all focus area weights in this category + """ + focus_weights = get_user_focus_weights(profile_id) + + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT focus_area_id + FROM focus_area_definitions + WHERE category = %s + """, (category,)) + + focus_areas = [row['focus_area_id'] for row in cur.fetchall()] + + total_weight = sum( + focus_weights.get(fa, 0) + for fa in focus_areas + ) + + return total_weight + + +# ============================================================================ +# Goal Progress Score (Meta-Score with Dynamic Weighting) +# ============================================================================ + +def calculate_goal_progress_score(profile_id: str) -> Optional[int]: + """ + Calculate overall goal progress score (0-100) + Weighted dynamically based on user's focus area priorities + + This is the main meta-score that combines all sub-scores + """ + focus_weights = get_user_focus_weights(profile_id) + + if not focus_weights: + return None # No goals/focus areas configured + + # Calculate sub-scores + from calculations.body_metrics import calculate_body_progress_score + from calculations.nutrition_metrics import calculate_nutrition_score + from calculations.activity_metrics import calculate_activity_score + from calculations.recovery_metrics import calculate_recovery_score_v2 + + body_score = calculate_body_progress_score(profile_id, focus_weights) + nutrition_score = calculate_nutrition_score(profile_id, focus_weights) + activity_score = calculate_activity_score(profile_id, focus_weights) + recovery_score = calculate_recovery_score_v2(profile_id) + health_risk_score = calculate_health_stability_score(profile_id) + + # Map focus areas to score components + focus_to_component = map_focus_to_score_components() + + # Calculate weighted sum + total_score = 0.0 + total_weight = 0.0 + + for focus_area_id, weight in focus_weights.items(): + component = focus_to_component.get(focus_area_id) + + if component == 'body' and body_score is not None: + total_score += body_score * weight + total_weight += weight + elif component == 'nutrition' and nutrition_score is not None: + total_score += nutrition_score * weight + total_weight += weight + elif component == 'activity' and activity_score is not None: + total_score += activity_score * weight + total_weight += weight + elif component == 'recovery' and recovery_score is not None: + total_score += recovery_score * weight + total_weight += weight + elif component == 'health' and health_risk_score is not None: + total_score += health_risk_score * weight + total_weight += weight + + if total_weight == 0: + return None + + # Normalize to 0-100 + final_score = total_score / total_weight + + return int(final_score) + + +def calculate_health_stability_score(profile_id: str) -> Optional[int]: + """ + Health stability score (0-100) + Components: + - Blood pressure status + - Sleep quality + - Movement baseline + - Weight/circumference risk factors + - Regularity + """ + with get_db() as conn: + cur = get_cursor(conn) + + components = [] + + # 1. Blood pressure status (30%) + cur.execute(""" + SELECT systolic, diastolic + FROM blood_pressure_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '28 days' + ORDER BY date DESC + """, (profile_id,)) + + bp_readings = cur.fetchall() + if bp_readings: + bp_score = _score_blood_pressure(bp_readings) + components.append(('bp', bp_score, 30)) + + # 2. Sleep quality (25%) + cur.execute(""" + SELECT total_sleep_min, deep_min, rem_min + FROM sleep_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '28 days' + ORDER BY date DESC + """, (profile_id,)) + + sleep_data = cur.fetchall() + if sleep_data: + sleep_score = _score_sleep_quality(sleep_data) + components.append(('sleep', sleep_score, 25)) + + # 3. Movement baseline (20%) + cur.execute(""" + SELECT duration + FROM activity_log + WHERE profile_id = %s + AND date >= CURRENT_DATE - INTERVAL '7 days' + """, (profile_id,)) + + activities = cur.fetchall() + if activities: + total_minutes = sum(a['duration'] for a in activities) + # WHO recommends 150-300 min/week moderate activity + movement_score = min(100, (total_minutes / 150) * 100) + components.append(('movement', movement_score, 20)) + + # 4. Waist circumference risk (15%) + cur.execute(""" + SELECT c_waist + FROM circumference_log + WHERE profile_id = %s + AND c_waist IS NOT NULL + ORDER BY date DESC + LIMIT 1 + """, (profile_id,)) + + waist = cur.fetchone() + if waist: + # Gender-specific thresholds (simplified - should use profile gender) + # Men: <94cm good, 94-102 elevated, >102 high risk + # Women: <80cm good, 80-88 elevated, >88 high risk + # Using conservative thresholds + waist_cm = waist['c_waist'] + if waist_cm < 88: + waist_score = 100 + elif waist_cm < 94: + waist_score = 75 + elif waist_cm < 102: + waist_score = 50 + else: + waist_score = 25 + components.append(('waist', waist_score, 15)) + + # 5. Regularity (10%) - sleep timing consistency + if len(sleep_data) >= 7: + sleep_times = [s['total_sleep_min'] for s in sleep_data] + avg = sum(sleep_times) / len(sleep_times) + variance = sum((x - avg) ** 2 for x in sleep_times) / len(sleep_times) + std_dev = variance ** 0.5 + # Lower std_dev = better consistency + regularity_score = max(0, 100 - (std_dev * 2)) + components.append(('regularity', regularity_score, 10)) + + 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_blood_pressure(readings: List) -> int: + """Score blood pressure readings (0-100)""" + # Average last 28 days + avg_systolic = sum(r['systolic'] for r in readings) / len(readings) + avg_diastolic = sum(r['diastolic'] for r in readings) / len(readings) + + # ESC 2024 Guidelines: + # Optimal: <120/80 + # Normal: 120-129 / 80-84 + # Elevated: 130-139 / 85-89 + # Hypertension: ≥140/90 + + if avg_systolic < 120 and avg_diastolic < 80: + return 100 + elif avg_systolic < 130 and avg_diastolic < 85: + return 85 + elif avg_systolic < 140 and avg_diastolic < 90: + return 65 + else: + return 40 + + +def _score_sleep_quality(sleep_data: List) -> int: + """Score sleep quality (0-100)""" + # Average sleep duration and quality + avg_total = sum(s['total_sleep_min'] for s in sleep_data) / len(sleep_data) + avg_total_hours = avg_total / 60 + + # Duration score (7+ hours = good) + if avg_total_hours >= 8: + duration_score = 100 + elif avg_total_hours >= 7: + duration_score = 85 + elif avg_total_hours >= 6: + duration_score = 65 + else: + duration_score = 40 + + # Quality score (deep + REM percentage) + quality_scores = [] + for s in sleep_data: + if s['deep_min'] and s['rem_min']: + quality_pct = ((s['deep_min'] + s['rem_min']) / s['total_sleep_min']) * 100 + # 40-60% deep+REM is good + if quality_pct >= 45: + quality_scores.append(100) + elif quality_pct >= 35: + quality_scores.append(75) + elif quality_pct >= 25: + quality_scores.append(50) + else: + quality_scores.append(30) + + if quality_scores: + avg_quality = sum(quality_scores) / len(quality_scores) + # Weighted: 60% duration, 40% quality + return int(duration_score * 0.6 + avg_quality * 0.4) + else: + return duration_score + + +# ============================================================================ +# Data Quality Score +# ============================================================================ + +def calculate_data_quality_score(profile_id: str) -> int: + """ + Overall data quality score (0-100) + Combines quality from all modules + """ + from calculations.body_metrics import calculate_body_data_quality + from calculations.nutrition_metrics import calculate_nutrition_data_quality + from calculations.activity_metrics import calculate_activity_data_quality + from calculations.recovery_metrics import calculate_recovery_data_quality + + body_quality = calculate_body_data_quality(profile_id) + nutrition_quality = calculate_nutrition_data_quality(profile_id) + activity_quality = calculate_activity_data_quality(profile_id) + recovery_quality = calculate_recovery_data_quality(profile_id) + + # Weighted average (all equal weight) + total_score = ( + body_quality['overall_score'] * 0.25 + + nutrition_quality['overall_score'] * 0.25 + + activity_quality['overall_score'] * 0.25 + + recovery_quality['overall_score'] * 0.25 + ) + + return int(total_score) + + +# ============================================================================ +# Top-Weighted Helpers (instead of "primary goal") +# ============================================================================ + +def get_top_priority_goal(profile_id: str) -> Optional[Dict]: + """ + Get highest priority goal based on: + - Progress gap (distance to target) + - Focus area weight + Returns goal dict or None + """ + from goal_utils import get_active_goals + + goals = get_active_goals(profile_id) + if not goals: + return None + + focus_weights = get_user_focus_weights(profile_id) + + for goal in goals: + # Progress gap (0-100, higher = further from target) + goal['progress_gap'] = 100 - goal.get('progress_pct', 0) + + # Get focus areas for this goal + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT fa.focus_area_id + FROM goal_focus_contributions gfc + JOIN focus_area_definitions fa ON gfc.focus_area_id = fa.id + WHERE gfc.goal_id = %s + """, (goal['id'],)) + + goal_focus_areas = [row['focus_area_id'] for row in cur.fetchall()] + + # Sum focus weights + goal['total_focus_weight'] = sum( + focus_weights.get(fa, 0) + for fa in goal_focus_areas + ) + + # Priority score + goal['priority_score'] = goal['progress_gap'] * (goal['total_focus_weight'] / 100) + + # Return goal with highest priority score + return max(goals, key=lambda g: g.get('priority_score', 0)) + + +def get_top_focus_area(profile_id: str) -> Optional[Dict]: + """ + Get focus area with highest user weight + Returns dict with focus_area_id, label, weight, progress + """ + focus_weights = get_user_focus_weights(profile_id) + + if not focus_weights: + return None + + top_fa_id = max(focus_weights, key=focus_weights.get) + + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT focus_area_id, label_de, category + FROM focus_area_definitions + WHERE focus_area_id = %s + """, (top_fa_id,)) + + fa_def = cur.fetchone() + if not fa_def: + return None + + # Calculate progress for this focus area + progress = calculate_focus_area_progress(profile_id, top_fa_id) + + return { + 'focus_area_id': top_fa_id, + 'label': fa_def['label_de'], + 'category': fa_def['category'], + 'weight': focus_weights[top_fa_id], + 'progress': progress + } + + +def calculate_focus_area_progress(profile_id: str, focus_area_id: str) -> Optional[int]: + """ + Calculate progress for a specific focus area (0-100) + Average progress of all goals contributing to this focus area + """ + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT g.id, g.progress_pct, gfc.contribution_weight + FROM goals g + JOIN goal_focus_contributions gfc ON g.id = gfc.goal_id + WHERE g.profile_id = %s + AND gfc.focus_area_id = ( + SELECT id FROM focus_area_definitions WHERE focus_area_id = %s + ) + AND g.status = 'active' + """, (profile_id, focus_area_id)) + + goals = cur.fetchall() + + if not goals: + return None + + # Weighted average by contribution_weight + total_progress = sum(g['progress_pct'] * g['contribution_weight'] for g in goals) + total_weight = sum(g['contribution_weight'] for g in goals) + + return int(total_progress / total_weight) if total_weight > 0 else None