547 lines
18 KiB
Python
547 lines
18 KiB
Python
"""
|
|
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 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',
|
|
}
|
|
|
|
|
|
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 key
|
|
FROM focus_area_definitions
|
|
WHERE category = %s
|
|
""", (category,))
|
|
|
|
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
|