diff --git a/backend/data_layer/__init__.py b/backend/data_layer/__init__.py index 0fb8da2..63ec722 100644 --- a/backend/data_layer/__init__.py +++ b/backend/data_layer/__init__.py @@ -34,6 +34,7 @@ from .nutrition_metrics import * from .activity_metrics import * from .recovery_metrics import * from .health_metrics import * +from .scores import * # Future imports (will be added as modules are created): # from .goals import * @@ -134,4 +135,18 @@ __all__ = [ 'get_resting_heart_rate_data', 'get_heart_rate_variability_data', 'get_vo2_max_data', + + # Scoring Metrics + 'get_user_focus_weights', + 'get_focus_area_category', + 'map_focus_to_score_components', + 'map_category_de_to_en', + 'calculate_category_weight', + 'calculate_goal_progress_score', + 'calculate_health_stability_score', + 'calculate_data_quality_score', + 'get_top_priority_goal', + 'get_top_focus_area', + 'calculate_focus_area_progress', + 'calculate_category_progress', ] diff --git a/backend/data_layer/scores.py b/backend/data_layer/scores.py new file mode 100644 index 0000000..c279f2d --- /dev/null +++ b/backend/data_layer/scores.py @@ -0,0 +1,583 @@ +""" +Scoring Metrics Data Layer + +Provides structured scoring and focus weight functions for all metrics. + +Functions: + - get_user_focus_weights(): User focus area weights (from DB) + - get_focus_area_category(): Category for a focus area + - map_focus_to_score_components(): Mapping of focus areas to score components + - map_category_de_to_en(): Category translation DE→EN + - calculate_category_weight(): Weight for a category + - calculate_goal_progress_score(): Goal progress scoring + - calculate_health_stability_score(): Health stability scoring + - calculate_data_quality_score(): Overall data quality + - get_top_priority_goal(): Top goal by weight + - get_top_focus_area(): Top focus area by weight + - calculate_focus_area_progress(): Progress for specific focus area + - calculate_category_progress(): Progress for category + +All functions return structured data (dict) or simple values. +Use placeholder_resolver.py for formatted strings for AI. + +Phase 0c: Multi-Layer Architecture +Version: 1.0 +""" + +from typing import Dict, List, Optional +from datetime import datetime, timedelta, date +from db import get_db, get_cursor, r2d + +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 ufw.focus_area_id, ufw.weight as weight_pct, fa.key + 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 > 0 + """, (profile_id,)) + + return { + row['key']: 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 + Keys match focus_area_definitions.key (English lowercase) + Returns: {'weight_loss': 'body', 'strength': 'activity', ...} + """ + return { + # Body Composition → body_progress_score + 'weight_loss': 'body', + 'muscle_gain': 'body', + 'body_recomposition': 'body', + + # Training - Strength → activity_score + 'strength': 'activity', + 'strength_endurance': 'activity', + 'power': 'activity', + + # Training - Mobility → activity_score + 'flexibility': 'activity', + 'mobility': 'activity', + + # Endurance → activity_score (could also map to health) + 'aerobic_endurance': 'activity', + 'anaerobic_endurance': 'activity', + 'cardiovascular_health': 'health', + + # Coordination → activity_score + 'balance': 'activity', + 'reaction': 'activity', + 'rhythm': 'activity', + 'coordination': 'activity', + + # Mental → recovery_score (mental health is part of recovery) + 'stress_resistance': 'recovery', + 'concentration': 'recovery', + 'willpower': 'recovery', + 'mental_health': 'recovery', + + # Recovery → recovery_score + 'sleep_quality': 'recovery', + 'regeneration': 'recovery', + 'rest': 'recovery', + + # Health → health + 'metabolic_health': 'health', + 'blood_pressure': 'health', + 'hrv': 'health', + 'general_health': 'health', + + # Nutrition → nutrition_score + 'protein_intake': 'nutrition', + 'calorie_balance': 'nutrition', + 'macro_consistency': 'nutrition', + 'meal_timing': 'nutrition', + 'hydration': 'nutrition', + } + + +def map_category_de_to_en(category_de: str) -> str: + """ + Map German category names to English database names + """ + mapping = { + 'körper': 'body_composition', + 'ernährung': 'nutrition', # Note: no nutrition category in DB, returns empty + 'aktivität': 'training', + 'recovery': 'recovery', + 'vitalwerte': 'health', + 'mental': 'mental', + 'lebensstil': 'health', # Maps to general health + } + return mapping.get(category_de, category_de) + + +def calculate_category_weight(profile_id: str, category: str) -> float: + """ + Calculate total weight for a category + Accepts German or English category names + Returns sum of all focus area weights in this category + """ + # Map German to English if needed + category_en = map_category_de_to_en(category) + + focus_weights = get_user_focus_weights(profile_id) + + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT key + FROM focus_area_definitions + WHERE category = %s + """, (category_en,)) + + focus_areas = [row['key'] 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 measured_at >= CURRENT_DATE - INTERVAL '28 days' + ORDER BY measured_at 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 duration_minutes, deep_minutes, rem_minutes + 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_min + 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_min'] 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['duration_minutes'] 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['duration_minutes'] 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_minutes'] and s['rem_minutes']: + quality_pct = ((s['deep_minutes'] + s['rem_minutes']) / s['duration_minutes']) * 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') or 0) + + # Get focus areas for this goal + with get_db() as conn: + cur = get_cursor(conn) + cur.execute(""" + SELECT fa.key as focus_area_key + 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_key'] 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 key, name_de, category + FROM focus_area_definitions + WHERE key = %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['name_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 key = %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 + +def calculate_category_progress(profile_id: str, category: str) -> Optional[int]: + """ + Calculate progress score for a focus area category (0-100). + + Args: + profile_id: User's profile ID + category: Category name ('körper', 'ernährung', 'aktivität', 'recovery', 'vitalwerte', 'mental', 'lebensstil') + + Returns: + Progress score 0-100 or None if no data + """ + # Map category to score calculation functions + category_scores = { + 'körper': 'body_progress_score', + 'ernährung': 'nutrition_score', + 'aktivität': 'activity_score', + 'recovery': 'recovery_score', + 'vitalwerte': 'recovery_score', # Use recovery score as proxy for vitals + 'mental': 'recovery_score', # Use recovery score as proxy for mental (sleep quality) + 'lebensstil': 'data_quality_score', # Use data quality as proxy for lifestyle consistency + } + + score_func_name = category_scores.get(category.lower()) + if not score_func_name: + return None + + # Call the appropriate score function + if score_func_name == 'body_progress_score': + from calculations.body_metrics import calculate_body_progress_score + return calculate_body_progress_score(profile_id) + elif score_func_name == 'nutrition_score': + from calculations.nutrition_metrics import calculate_nutrition_score + return calculate_nutrition_score(profile_id) + elif score_func_name == 'activity_score': + from calculations.activity_metrics import calculate_activity_score + return calculate_activity_score(profile_id) + elif score_func_name == 'recovery_score': + from calculations.recovery_metrics import calculate_recovery_score_v2 + return calculate_recovery_score_v2(profile_id) + elif score_func_name == 'data_quality_score': + return calculate_data_quality_score(profile_id) + + return None diff --git a/backend/placeholder_resolver.py b/backend/placeholder_resolver.py index c6e57f7..187adda 100644 --- a/backend/placeholder_resolver.py +++ b/backend/placeholder_resolver.py @@ -417,8 +417,8 @@ def _safe_int(func_name: str, profile_id: str) -> str: import traceback try: # Import calculations dynamically to avoid circular imports - from calculations import scores, correlation_metrics - from data_layer import body_metrics, nutrition_metrics, activity_metrics, recovery_metrics + from calculations import correlation_metrics + from data_layer import body_metrics, nutrition_metrics, activity_metrics, recovery_metrics, scores # Map function names to actual functions func_map = { @@ -480,8 +480,7 @@ def _safe_float(func_name: str, profile_id: str, decimals: int = 1) -> str: """ import traceback try: - from calculations import scores - from data_layer import body_metrics, nutrition_metrics, activity_metrics, recovery_metrics + from data_layer import body_metrics, nutrition_metrics, activity_metrics, recovery_metrics, scores func_map = { 'weight_7d_median': body_metrics.calculate_weight_7d_median,