mitai-jinkendo/backend/routers/charts.py
Lars 176be3233e
All checks were successful
Deploy Development / deploy (push) Successful in 46s
Build Test / lint-backend (push) Successful in 0s
Build Test / build-frontend (push) Successful in 14s
fix: add missing prefix to charts router
Charts router had no prefix, causing 404 errors.

Fixed:
- Added prefix="/api/charts" to APIRouter()
- Changed all endpoint paths from "/charts/..." to "/..."
  (prefix already includes /api/charts)

Now endpoints resolve correctly:
/api/charts/energy-balance
/api/charts/recovery-score
etc.

All 23 chart endpoints now accessible.
2026-03-29 07:08:05 +02:00

2247 lines
67 KiB
Python

"""
Charts Router - Chart.js-compatible Data Endpoints
Provides structured data for frontend charts/diagrams.
All endpoints return Chart.js-compatible JSON format:
{
"chart_type": "line" | "bar" | "scatter" | "pie",
"data": {
"labels": [...],
"datasets": [...]
},
"metadata": {
"confidence": "high" | "medium" | "low" | "insufficient",
"data_points": int,
...
}
}
Phase 0c: Multi-Layer Architecture
Version: 1.0
"""
from fastapi import APIRouter, Depends, HTTPException, Query
from typing import Dict, List, Optional
from datetime import datetime, timedelta
from auth import require_auth
from data_layer.body_metrics import (
get_weight_trend_data,
get_body_composition_data,
get_circumference_summary_data
)
from data_layer.nutrition_metrics import (
get_nutrition_average_data,
get_protein_targets_data,
get_protein_adequacy_data,
get_macro_consistency_data
)
from data_layer.activity_metrics import (
get_activity_summary_data,
get_training_type_distribution_data,
calculate_training_minutes_week,
calculate_quality_sessions_pct,
calculate_proxy_internal_load_7d,
calculate_monotony_score,
calculate_strain_score,
calculate_ability_balance
)
from data_layer.recovery_metrics import (
get_sleep_duration_data,
get_sleep_quality_data,
calculate_recovery_score_v2,
calculate_hrv_vs_baseline_pct,
calculate_rhr_vs_baseline_pct,
calculate_sleep_debt_hours
)
from data_layer.correlations import (
calculate_lag_correlation,
calculate_correlation_sleep_recovery,
calculate_top_drivers
)
from data_layer.utils import serialize_dates, safe_float, calculate_confidence
router = APIRouter(prefix="/api/charts", tags=["charts"])
# ── Body Charts ─────────────────────────────────────────────────────────────
@router.get("/weight-trend")
def get_weight_trend_chart(
days: int = Query(default=90, ge=7, le=365, description="Analysis window in days"),
session: dict = Depends(require_auth)
) -> Dict:
"""
Weight trend chart data.
Returns Chart.js-compatible line chart with:
- Raw weight values
- Trend line (if enough data)
- Confidence indicator
Args:
days: Analysis window (7-365 days, default 90)
session: Auth session (injected)
Returns:
{
"chart_type": "line",
"data": {
"labels": ["2026-01-01", ...],
"datasets": [{
"label": "Gewicht",
"data": [85.0, 84.5, ...],
"borderColor": "#1D9E75",
"backgroundColor": "rgba(29, 158, 117, 0.1)",
"tension": 0.4
}]
},
"metadata": {
"confidence": "high",
"data_points": 60,
"first_value": 92.0,
"last_value": 85.0,
"delta": -7.0,
"direction": "decreasing"
}
}
Metadata fields:
- confidence: Data quality ("high", "medium", "low", "insufficient")
- data_points: Number of weight entries in period
- first_value: First weight in period (kg)
- last_value: Latest weight (kg)
- delta: Weight change (kg, negative = loss)
- direction: "increasing" | "decreasing" | "stable"
"""
profile_id = session['profile_id']
# Get structured data from data layer
trend_data = get_weight_trend_data(profile_id, days)
# Early return if insufficient data
if trend_data['confidence'] == 'insufficient':
return {
"chart_type": "line",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Nicht genug Daten für Trend-Analyse"
}
}
# Get raw data points for chart
from db import get_db, get_cursor
with get_db() as conn:
cur = get_cursor(conn)
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
cur.execute(
"""SELECT date, weight FROM weight_log
WHERE profile_id=%s AND date >= %s
ORDER BY date""",
(profile_id, cutoff)
)
rows = cur.fetchall()
# Format for Chart.js
labels = [row['date'].isoformat() for row in rows]
values = [float(row['weight']) for row in rows]
return {
"chart_type": "line",
"data": {
"labels": labels,
"datasets": [
{
"label": "Gewicht",
"data": values,
"borderColor": "#1D9E75",
"backgroundColor": "rgba(29, 158, 117, 0.1)",
"borderWidth": 2,
"tension": 0.4,
"fill": True,
"pointRadius": 3,
"pointHoverRadius": 5
}
]
},
"metadata": serialize_dates({
"confidence": trend_data['confidence'],
"data_points": trend_data['data_points'],
"first_value": trend_data['first_value'],
"last_value": trend_data['last_value'],
"delta": trend_data['delta'],
"direction": trend_data['direction'],
"first_date": trend_data['first_date'],
"last_date": trend_data['last_date'],
"days_analyzed": trend_data['days_analyzed']
})
}
@router.get("/body-composition")
def get_body_composition_chart(
days: int = Query(default=90, ge=7, le=365),
session: dict = Depends(require_auth)
) -> Dict:
"""
Body composition chart (body fat percentage, lean mass trend).
Returns Chart.js-compatible multi-line chart.
Args:
days: Analysis window (7-365 days, default 90)
session: Auth session (injected)
Returns:
Chart.js format with datasets for:
- Body fat percentage (%)
- Lean mass (kg, if weight available)
"""
profile_id = session['profile_id']
# Get latest body composition
comp_data = get_body_composition_data(profile_id, days)
if comp_data['confidence'] == 'insufficient':
return {
"chart_type": "line",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Körperfett-Messungen vorhanden"
}
}
# For PoC: Return single data point
# TODO in bulk migration: Fetch historical caliper entries
return {
"chart_type": "line",
"data": {
"labels": [comp_data['date'].isoformat() if comp_data['date'] else ""],
"datasets": [
{
"label": "Körperfett %",
"data": [comp_data['body_fat_pct']],
"borderColor": "#D85A30",
"backgroundColor": "rgba(216, 90, 48, 0.1)",
"borderWidth": 2
}
]
},
"metadata": serialize_dates({
"confidence": comp_data['confidence'],
"data_points": comp_data['data_points'],
"body_fat_pct": comp_data['body_fat_pct'],
"method": comp_data['method'],
"date": comp_data['date']
})
}
@router.get("/circumferences")
def get_circumferences_chart(
max_age_days: int = Query(default=90, ge=7, le=365),
session: dict = Depends(require_auth)
) -> Dict:
"""
Latest circumference measurements as bar chart.
Shows most recent measurement for each body point.
Args:
max_age_days: Maximum age of measurements (default 90)
session: Auth session (injected)
Returns:
Chart.js bar chart with all circumference points
"""
profile_id = session['profile_id']
circ_data = get_circumference_summary_data(profile_id, max_age_days)
if circ_data['confidence'] == 'insufficient':
return {
"chart_type": "bar",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Umfangsmessungen vorhanden"
}
}
# Sort by value (descending) for better visualization
measurements = sorted(
circ_data['measurements'],
key=lambda m: m['value'],
reverse=True
)
labels = [m['point'] for m in measurements]
values = [m['value'] for m in measurements]
return {
"chart_type": "bar",
"data": {
"labels": labels,
"datasets": [
{
"label": "Umfang (cm)",
"data": values,
"backgroundColor": "#1D9E75",
"borderColor": "#085041",
"borderWidth": 1
}
]
},
"metadata": serialize_dates({
"confidence": circ_data['confidence'],
"data_points": circ_data['data_points'],
"newest_date": circ_data['newest_date'],
"oldest_date": circ_data['oldest_date'],
"measurements": circ_data['measurements'] # Full details
})
}
# ── Nutrition Charts ────────────────────────────────────────────────────────
@router.get("/energy-balance")
def get_energy_balance_chart(
days: int = Query(default=28, ge=7, le=90),
session: dict = Depends(require_auth)
) -> Dict:
"""
Energy balance timeline (E1).
Shows daily calorie intake over time with optional TDEE reference line.
Args:
days: Analysis window (7-90 days, default 28)
session: Auth session (injected)
Returns:
Chart.js line chart with daily kcal intake
"""
profile_id = session['profile_id']
from db import get_db, get_cursor
with get_db() as conn:
cur = get_cursor(conn)
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
cur.execute(
"""SELECT date, kcal
FROM nutrition_log
WHERE profile_id=%s AND date >= %s AND kcal IS NOT NULL
ORDER BY date""",
(profile_id, cutoff)
)
rows = cur.fetchall()
if not rows:
return {
"chart_type": "line",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Ernährungsdaten vorhanden"
}
}
labels = [row['date'].isoformat() for row in rows]
values = [safe_float(row['kcal']) for row in rows]
# Calculate average for metadata
avg_kcal = sum(values) / len(values) if values else 0
datasets = [
{
"label": "Kalorien",
"data": values,
"borderColor": "#1D9E75",
"backgroundColor": "rgba(29, 158, 117, 0.1)",
"borderWidth": 2,
"tension": 0.3,
"fill": True
}
]
# Add TDEE reference line (estimated)
# TODO: Get actual TDEE from profile calculation
estimated_tdee = 2500.0
datasets.append({
"label": "TDEE (geschätzt)",
"data": [estimated_tdee] * len(labels),
"borderColor": "#888",
"borderWidth": 1,
"borderDash": [5, 5],
"fill": False,
"pointRadius": 0
})
from data_layer.utils import calculate_confidence
confidence = calculate_confidence(len(rows), days, "general")
return {
"chart_type": "line",
"data": {
"labels": labels,
"datasets": datasets
},
"metadata": serialize_dates({
"confidence": confidence,
"data_points": len(rows),
"avg_kcal": round(avg_kcal, 1),
"estimated_tdee": estimated_tdee,
"first_date": rows[0]['date'],
"last_date": rows[-1]['date']
})
}
@router.get("/macro-distribution")
def get_macro_distribution_chart(
days: int = Query(default=28, ge=7, le=90),
session: dict = Depends(require_auth)
) -> Dict:
"""
Macronutrient distribution pie chart (E2).
Shows average protein/carbs/fat distribution over period.
Args:
days: Analysis window (7-90 days, default 28)
session: Auth session (injected)
Returns:
Chart.js pie chart with macro percentages
"""
profile_id = session['profile_id']
# Get average macros
macro_data = get_nutrition_average_data(profile_id, days)
if macro_data['confidence'] == 'insufficient':
return {
"chart_type": "pie",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Ernährungsdaten vorhanden"
}
}
# Calculate calories from macros (protein/carbs = 4 kcal/g, fat = 9 kcal/g)
protein_kcal = macro_data['protein_avg'] * 4
carbs_kcal = macro_data['carbs_avg'] * 4
fat_kcal = macro_data['fat_avg'] * 9
total_kcal = protein_kcal + carbs_kcal + fat_kcal
if total_kcal == 0:
return {
"chart_type": "pie",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Makronährstoff-Daten"
}
}
protein_pct = (protein_kcal / total_kcal * 100)
carbs_pct = (carbs_kcal / total_kcal * 100)
fat_pct = (fat_kcal / total_kcal * 100)
return {
"chart_type": "pie",
"data": {
"labels": ["Protein", "Kohlenhydrate", "Fett"],
"datasets": [
{
"data": [round(protein_pct, 1), round(carbs_pct, 1), round(fat_pct, 1)],
"backgroundColor": [
"#1D9E75", # Protein (green)
"#F59E0B", # Carbs (amber)
"#EF4444" # Fat (red)
],
"borderWidth": 2,
"borderColor": "#fff"
}
]
},
"metadata": {
"confidence": macro_data['confidence'],
"data_points": macro_data['data_points'],
"protein_g": round(macro_data['protein_avg'], 1),
"carbs_g": round(macro_data['carbs_avg'], 1),
"fat_g": round(macro_data['fat_avg'], 1),
"protein_pct": round(protein_pct, 1),
"carbs_pct": round(carbs_pct, 1),
"fat_pct": round(fat_pct, 1)
}
}
@router.get("/protein-adequacy")
def get_protein_adequacy_chart(
days: int = Query(default=28, ge=7, le=90),
session: dict = Depends(require_auth)
) -> Dict:
"""
Protein adequacy timeline (E3).
Shows daily protein intake vs. target range.
Args:
days: Analysis window (7-90 days, default 28)
session: Auth session (injected)
Returns:
Chart.js line chart with protein intake + target bands
"""
profile_id = session['profile_id']
# Get protein targets
targets = get_protein_targets_data(profile_id)
from db import get_db, get_cursor
with get_db() as conn:
cur = get_cursor(conn)
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
cur.execute(
"""SELECT date, protein_g
FROM nutrition_log
WHERE profile_id=%s AND date >= %s AND protein_g IS NOT NULL
ORDER BY date""",
(profile_id, cutoff)
)
rows = cur.fetchall()
if not rows:
return {
"chart_type": "line",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Protein-Daten vorhanden"
}
}
labels = [row['date'].isoformat() for row in rows]
values = [safe_float(row['protein_g']) for row in rows]
datasets = [
{
"label": "Protein (g)",
"data": values,
"borderColor": "#1D9E75",
"backgroundColor": "rgba(29, 158, 117, 0.2)",
"borderWidth": 2,
"tension": 0.3,
"fill": False
}
]
# Add target range bands
target_low = targets['protein_target_low']
target_high = targets['protein_target_high']
datasets.append({
"label": "Ziel Min",
"data": [target_low] * len(labels),
"borderColor": "#888",
"borderWidth": 1,
"borderDash": [5, 5],
"fill": False,
"pointRadius": 0
})
datasets.append({
"label": "Ziel Max",
"data": [target_high] * len(labels),
"borderColor": "#888",
"borderWidth": 1,
"borderDash": [5, 5],
"fill": False,
"pointRadius": 0
})
from data_layer.utils import calculate_confidence
confidence = calculate_confidence(len(rows), days, "general")
# Count days in target
days_in_target = sum(1 for v in values if target_low <= v <= target_high)
return {
"chart_type": "line",
"data": {
"labels": labels,
"datasets": datasets
},
"metadata": serialize_dates({
"confidence": confidence,
"data_points": len(rows),
"target_low": round(target_low, 1),
"target_high": round(target_high, 1),
"days_in_target": days_in_target,
"target_compliance_pct": round(days_in_target / len(values) * 100, 1) if values else 0,
"first_date": rows[0]['date'],
"last_date": rows[-1]['date']
})
}
@router.get("/nutrition-consistency")
def get_nutrition_consistency_chart(
days: int = Query(default=28, ge=7, le=90),
session: dict = Depends(require_auth)
) -> Dict:
"""
Nutrition consistency score (E5).
Shows macro consistency score as bar chart.
Args:
days: Analysis window (7-90 days, default 28)
session: Auth session (injected)
Returns:
Chart.js bar chart with consistency metrics
"""
profile_id = session['profile_id']
consistency_data = get_macro_consistency_data(profile_id, days)
if consistency_data['confidence'] == 'insufficient':
return {
"chart_type": "bar",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Nicht genug Daten für Konsistenz-Analyse"
}
}
# Show consistency score + macro averages
labels = [
"Gesamt-Score",
f"Protein ({consistency_data['avg_protein_pct']:.0f}%)",
f"Kohlenhydrate ({consistency_data['avg_carbs_pct']:.0f}%)",
f"Fett ({consistency_data['avg_fat_pct']:.0f}%)"
]
# Score = 100 - std_dev (inverted for display)
# Higher bar = more consistent
protein_consistency = max(0, 100 - consistency_data['std_dev_protein'] * 10)
carbs_consistency = max(0, 100 - consistency_data['std_dev_carbs'] * 10)
fat_consistency = max(0, 100 - consistency_data['std_dev_fat'] * 10)
values = [
consistency_data['consistency_score'],
protein_consistency,
carbs_consistency,
fat_consistency
]
return {
"chart_type": "bar",
"data": {
"labels": labels,
"datasets": [
{
"label": "Konsistenz-Score",
"data": values,
"backgroundColor": ["#1D9E75", "#1D9E75", "#F59E0B", "#EF4444"],
"borderColor": "#085041",
"borderWidth": 1
}
]
},
"metadata": {
"confidence": consistency_data['confidence'],
"data_points": consistency_data['data_points'],
"consistency_score": consistency_data['consistency_score'],
"std_dev_protein": round(consistency_data['std_dev_protein'], 2),
"std_dev_carbs": round(consistency_data['std_dev_carbs'], 2),
"std_dev_fat": round(consistency_data['std_dev_fat'], 2)
}
}
# ── Activity Charts ─────────────────────────────────────────────────────────
@router.get("/training-volume")
def get_training_volume_chart(
weeks: int = Query(default=12, ge=4, le=52),
session: dict = Depends(require_auth)
) -> Dict:
"""
Training volume week-over-week (A1).
Shows weekly training minutes over time.
Args:
weeks: Number of weeks to analyze (4-52, default 12)
session: Auth session (injected)
Returns:
Chart.js bar chart with weekly training minutes
"""
profile_id = session['profile_id']
from db import get_db, get_cursor
with get_db() as conn:
cur = get_cursor(conn)
cutoff = (datetime.now() - timedelta(weeks=weeks)).strftime('%Y-%m-%d')
# Get weekly aggregates
cur.execute(
"""SELECT
DATE_TRUNC('week', date) as week_start,
SUM(duration_min) as total_minutes,
COUNT(*) as session_count
FROM activity_log
WHERE profile_id=%s AND date >= %s
GROUP BY week_start
ORDER BY week_start""",
(profile_id, cutoff)
)
rows = cur.fetchall()
if not rows:
return {
"chart_type": "bar",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Aktivitätsdaten vorhanden"
}
}
labels = [row['week_start'].strftime('KW %V') for row in rows]
values = [safe_float(row['total_minutes']) for row in rows]
confidence = calculate_confidence(len(rows), weeks * 7, "general")
return {
"chart_type": "bar",
"data": {
"labels": labels,
"datasets": [
{
"label": "Trainingsminuten",
"data": values,
"backgroundColor": "#1D9E75",
"borderColor": "#085041",
"borderWidth": 1
}
]
},
"metadata": serialize_dates({
"confidence": confidence,
"data_points": len(rows),
"avg_minutes_week": round(sum(values) / len(values), 1) if values else 0,
"total_sessions": sum(row['session_count'] for row in rows)
})
}
@router.get("/training-type-distribution")
def get_training_type_distribution_chart(
days: int = Query(default=28, ge=7, le=90),
session: dict = Depends(require_auth)
) -> Dict:
"""
Training type distribution (A2).
Shows distribution of training categories as pie chart.
Args:
days: Analysis window (7-90 days, default 28)
session: Auth session (injected)
Returns:
Chart.js pie chart with training categories
"""
profile_id = session['profile_id']
dist_data = get_training_type_distribution_data(profile_id, days)
if dist_data['confidence'] == 'insufficient':
return {
"chart_type": "pie",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Trainingstypen-Daten"
}
}
labels = [item['category'] for item in dist_data['distribution']]
values = [item['count'] for item in dist_data['distribution']]
# Color palette for training categories
colors = [
"#1D9E75", "#3B82F6", "#F59E0B", "#EF4444",
"#8B5CF6", "#10B981", "#F97316", "#06B6D4"
]
return {
"chart_type": "pie",
"data": {
"labels": labels,
"datasets": [
{
"data": values,
"backgroundColor": colors[:len(values)],
"borderWidth": 2,
"borderColor": "#fff"
}
]
},
"metadata": {
"confidence": dist_data['confidence'],
"total_sessions": dist_data['total_sessions'],
"categorized_sessions": dist_data['categorized_sessions'],
"uncategorized_sessions": dist_data['uncategorized_sessions']
}
}
@router.get("/quality-sessions")
def get_quality_sessions_chart(
days: int = Query(default=28, ge=7, le=90),
session: dict = Depends(require_auth)
) -> Dict:
"""
Quality session rate (A3).
Shows percentage of quality sessions (RPE >= 7 or duration >= 60min).
Args:
days: Analysis window (7-90 days, default 28)
session: Auth session (injected)
Returns:
Chart.js bar chart with quality metrics
"""
profile_id = session['profile_id']
# Calculate quality session percentage
quality_pct = calculate_quality_sessions_pct(profile_id, days)
from db import get_db, get_cursor
with get_db() as conn:
cur = get_cursor(conn)
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
cur.execute(
"""SELECT COUNT(*) as total
FROM activity_log
WHERE profile_id=%s AND date >= %s""",
(profile_id, cutoff)
)
row = cur.fetchone()
total_sessions = row['total'] if row else 0
if total_sessions == 0:
return {
"chart_type": "bar",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Aktivitätsdaten"
}
}
quality_count = int(quality_pct / 100 * total_sessions)
regular_count = total_sessions - quality_count
return {
"chart_type": "bar",
"data": {
"labels": ["Qualitäts-Sessions", "Reguläre Sessions"],
"datasets": [
{
"label": "Anzahl",
"data": [quality_count, regular_count],
"backgroundColor": ["#1D9E75", "#888"],
"borderColor": "#085041",
"borderWidth": 1
}
]
},
"metadata": {
"confidence": calculate_confidence(total_sessions, days, "general"),
"data_points": total_sessions,
"quality_pct": round(quality_pct, 1),
"quality_count": quality_count,
"regular_count": regular_count
}
}
@router.get("/load-monitoring")
def get_load_monitoring_chart(
days: int = Query(default=28, ge=14, le=90),
session: dict = Depends(require_auth)
) -> Dict:
"""
Load monitoring (A4).
Shows acute load (7d) vs chronic load (28d) and ACWR.
Args:
days: Analysis window (14-90 days, default 28)
session: Auth session (injected)
Returns:
Chart.js line chart with load metrics
"""
profile_id = session['profile_id']
# Calculate loads
acute_load = calculate_proxy_internal_load_7d(profile_id)
chronic_load = calculate_proxy_internal_load_7d(profile_id, days=28)
# ACWR (Acute:Chronic Workload Ratio)
acwr = acute_load / chronic_load if chronic_load > 0 else 0
# Fetch daily loads for timeline
from db import get_db, get_cursor
with get_db() as conn:
cur = get_cursor(conn)
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
cur.execute(
"""SELECT
date,
SUM(duration_min * COALESCE(rpe, 5)) as daily_load
FROM activity_log
WHERE profile_id=%s AND date >= %s
GROUP BY date
ORDER BY date""",
(profile_id, cutoff)
)
rows = cur.fetchall()
if not rows:
return {
"chart_type": "line",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Load-Daten"
}
}
labels = [row['date'].isoformat() for row in rows]
values = [safe_float(row['daily_load']) for row in rows]
return {
"chart_type": "line",
"data": {
"labels": labels,
"datasets": [
{
"label": "Tages-Load",
"data": values,
"borderColor": "#1D9E75",
"backgroundColor": "rgba(29, 158, 117, 0.1)",
"borderWidth": 2,
"tension": 0.3,
"fill": True
}
]
},
"metadata": serialize_dates({
"confidence": calculate_confidence(len(rows), days, "general"),
"data_points": len(rows),
"acute_load_7d": round(acute_load, 1),
"chronic_load_28d": round(chronic_load, 1),
"acwr": round(acwr, 2),
"acwr_status": "optimal" if 0.8 <= acwr <= 1.3 else "suboptimal"
})
}
@router.get("/monotony-strain")
def get_monotony_strain_chart(
days: int = Query(default=7, ge=7, le=28),
session: dict = Depends(require_auth)
) -> Dict:
"""
Monotony & Strain (A5).
Shows training monotony and strain scores.
Args:
days: Analysis window (7-28 days, default 7)
session: Auth session (injected)
Returns:
Chart.js bar chart with monotony and strain
"""
profile_id = session['profile_id']
monotony = calculate_monotony_score(profile_id, days)
strain = calculate_strain_score(profile_id, days)
if monotony == 0 and strain == 0:
return {
"chart_type": "bar",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Nicht genug Daten für Monotonie-Analyse"
}
}
return {
"chart_type": "bar",
"data": {
"labels": ["Monotonie", "Strain"],
"datasets": [
{
"label": "Score",
"data": [round(monotony, 2), round(strain, 1)],
"backgroundColor": ["#F59E0B", "#EF4444"],
"borderColor": "#085041",
"borderWidth": 1
}
]
},
"metadata": {
"confidence": "medium", # Fixed for monotony calculations
"monotony_score": round(monotony, 2),
"strain_score": round(strain, 1),
"monotony_status": "high" if monotony > 2.0 else "normal",
"strain_status": "high" if strain > 10000 else "normal"
}
}
@router.get("/ability-balance")
def get_ability_balance_chart(
days: int = Query(default=28, ge=7, le=90),
session: dict = Depends(require_auth)
) -> Dict:
"""
Ability balance radar chart (A6).
Shows training distribution across 5 abilities.
Args:
days: Analysis window (7-90 days, default 28)
session: Auth session (injected)
Returns:
Chart.js radar chart with ability balance
"""
profile_id = session['profile_id']
balance_data = calculate_ability_balance(profile_id, days)
if balance_data['total_minutes'] == 0:
return {
"chart_type": "radar",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Aktivitätsdaten"
}
}
labels = ["Kraft", "Ausdauer", "Beweglichkeit", "Gleichgewicht", "Geist"]
values = [
balance_data['strength_pct'],
balance_data['endurance_pct'],
balance_data['flexibility_pct'],
balance_data['balance_pct'],
balance_data['mind_pct']
]
return {
"chart_type": "radar",
"data": {
"labels": labels,
"datasets": [
{
"label": "Fähigkeiten-Balance (%)",
"data": values,
"borderColor": "#1D9E75",
"backgroundColor": "rgba(29, 158, 117, 0.2)",
"borderWidth": 2,
"pointBackgroundColor": "#1D9E75",
"pointBorderColor": "#fff",
"pointHoverBackgroundColor": "#fff",
"pointHoverBorderColor": "#1D9E75"
}
]
},
"metadata": {
"confidence": balance_data['confidence'],
"total_minutes": balance_data['total_minutes'],
"strength_pct": round(balance_data['strength_pct'], 1),
"endurance_pct": round(balance_data['endurance_pct'], 1),
"flexibility_pct": round(balance_data['flexibility_pct'], 1),
"balance_pct": round(balance_data['balance_pct'], 1),
"mind_pct": round(balance_data['mind_pct'], 1)
}
}
@router.get("/volume-by-ability")
def get_volume_by_ability_chart(
days: int = Query(default=28, ge=7, le=90),
session: dict = Depends(require_auth)
) -> Dict:
"""
Training volume by ability (A8).
Shows absolute minutes per ability category.
Args:
days: Analysis window (7-90 days, default 28)
session: Auth session (injected)
Returns:
Chart.js bar chart with volume per ability
"""
profile_id = session['profile_id']
from db import get_db, get_cursor
with get_db() as conn:
cur = get_cursor(conn)
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
cur.execute(
"""SELECT
COALESCE(ability, 'unknown') as ability,
SUM(duration_min) as total_minutes
FROM activity_log
WHERE profile_id=%s AND date >= %s
GROUP BY ability
ORDER BY total_minutes DESC""",
(profile_id, cutoff)
)
rows = cur.fetchall()
if not rows:
return {
"chart_type": "bar",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Ability-Daten"
}
}
# Map ability names to German
ability_map = {
"strength": "Kraft",
"endurance": "Ausdauer",
"flexibility": "Beweglichkeit",
"balance": "Gleichgewicht",
"mind": "Geist",
"unknown": "Nicht zugeordnet"
}
labels = [ability_map.get(row['ability'], row['ability']) for row in rows]
values = [safe_float(row['total_minutes']) for row in rows]
total_minutes = sum(values)
return {
"chart_type": "bar",
"data": {
"labels": labels,
"datasets": [
{
"label": "Trainingsminuten",
"data": values,
"backgroundColor": "#1D9E75",
"borderColor": "#085041",
"borderWidth": 1
}
]
},
"metadata": {
"confidence": calculate_confidence(len(rows), days, "general"),
"data_points": len(rows),
"total_minutes": total_minutes
}
}
# ── Recovery Charts ─────────────────────────────────────────────────────────
@router.get("/recovery-score")
def get_recovery_score_chart(
days: int = Query(default=28, ge=7, le=90),
session: dict = Depends(require_auth)
) -> Dict:
"""
Recovery score timeline (R1).
Shows daily recovery scores over time.
Args:
days: Analysis window (7-90 days, default 28)
session: Auth session (injected)
Returns:
Chart.js line chart with recovery scores
"""
profile_id = session['profile_id']
# For PoC: Use current recovery score and create synthetic timeline
# TODO: Store historical recovery scores for true timeline
current_score = calculate_recovery_score_v2(profile_id)
if current_score is None:
return {
"chart_type": "line",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Recovery-Daten vorhanden"
}
}
# Fetch vitals for timeline approximation
from db import get_db, get_cursor
with get_db() as conn:
cur = get_cursor(conn)
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
cur.execute(
"""SELECT date, resting_hr, hrv_ms
FROM vitals_baseline
WHERE profile_id=%s AND date >= %s
ORDER BY date""",
(profile_id, cutoff)
)
rows = cur.fetchall()
if not rows:
return {
"chart_type": "line",
"data": {
"labels": [datetime.now().strftime('%Y-%m-%d')],
"datasets": [
{
"label": "Recovery Score",
"data": [current_score],
"borderColor": "#1D9E75",
"backgroundColor": "rgba(29, 158, 117, 0.1)",
"borderWidth": 2,
"tension": 0.3,
"fill": True
}
]
},
"metadata": {
"confidence": "low",
"data_points": 1,
"current_score": current_score
}
}
# Simple proxy: Use HRV as recovery indicator (higher HRV = better recovery)
# This is a placeholder until we store actual recovery scores
labels = [row['date'].isoformat() for row in rows]
# Normalize HRV to 0-100 scale (assume typical range 20-100ms)
values = [min(100, max(0, safe_float(row['hrv_ms']) if row['hrv_ms'] else 50)) for row in rows]
return {
"chart_type": "line",
"data": {
"labels": labels,
"datasets": [
{
"label": "Recovery Score (proxy)",
"data": values,
"borderColor": "#1D9E75",
"backgroundColor": "rgba(29, 158, 117, 0.1)",
"borderWidth": 2,
"tension": 0.3,
"fill": True
}
]
},
"metadata": serialize_dates({
"confidence": calculate_confidence(len(rows), days, "general"),
"data_points": len(rows),
"current_score": current_score,
"note": "Score based on HRV proxy; true recovery score calculation in development"
})
}
@router.get("/hrv-rhr-baseline")
def get_hrv_rhr_baseline_chart(
days: int = Query(default=28, ge=7, le=90),
session: dict = Depends(require_auth)
) -> Dict:
"""
HRV/RHR vs baseline (R2).
Shows HRV and RHR trends vs. baseline values.
Args:
days: Analysis window (7-90 days, default 28)
session: Auth session (injected)
Returns:
Chart.js multi-line chart with HRV and RHR
"""
profile_id = session['profile_id']
from db import get_db, get_cursor
with get_db() as conn:
cur = get_cursor(conn)
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
cur.execute(
"""SELECT date, resting_hr, hrv_ms
FROM vitals_baseline
WHERE profile_id=%s AND date >= %s
ORDER BY date""",
(profile_id, cutoff)
)
rows = cur.fetchall()
if not rows:
return {
"chart_type": "line",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Vitalwerte vorhanden"
}
}
labels = [row['date'].isoformat() for row in rows]
hrv_values = [safe_float(row['hrv_ms']) if row['hrv_ms'] else None for row in rows]
rhr_values = [safe_float(row['resting_hr']) if row['resting_hr'] else None for row in rows]
# Calculate baselines (28d median)
hrv_baseline = calculate_hrv_vs_baseline_pct(profile_id) # This returns % deviation
rhr_baseline = calculate_rhr_vs_baseline_pct(profile_id) # This returns % deviation
# For chart, we need actual baseline values (approximation)
hrv_filtered = [v for v in hrv_values if v is not None]
rhr_filtered = [v for v in rhr_values if v is not None]
avg_hrv = sum(hrv_filtered) / len(hrv_filtered) if hrv_filtered else 50
avg_rhr = sum(rhr_filtered) / len(rhr_filtered) if rhr_filtered else 60
datasets = [
{
"label": "HRV (ms)",
"data": hrv_values,
"borderColor": "#1D9E75",
"backgroundColor": "rgba(29, 158, 117, 0.1)",
"borderWidth": 2,
"tension": 0.3,
"yAxisID": "y1",
"fill": False
},
{
"label": "RHR (bpm)",
"data": rhr_values,
"borderColor": "#3B82F6",
"backgroundColor": "rgba(59, 130, 246, 0.1)",
"borderWidth": 2,
"tension": 0.3,
"yAxisID": "y2",
"fill": False
}
]
return {
"chart_type": "line",
"data": {
"labels": labels,
"datasets": datasets
},
"metadata": serialize_dates({
"confidence": calculate_confidence(len(rows), days, "general"),
"data_points": len(rows),
"avg_hrv": round(avg_hrv, 1),
"avg_rhr": round(avg_rhr, 1),
"hrv_vs_baseline_pct": hrv_baseline,
"rhr_vs_baseline_pct": rhr_baseline
})
}
@router.get("/sleep-duration-quality")
def get_sleep_duration_quality_chart(
days: int = Query(default=28, ge=7, le=90),
session: dict = Depends(require_auth)
) -> Dict:
"""
Sleep duration + quality (R3).
Shows sleep duration and quality score over time.
Args:
days: Analysis window (7-90 days, default 28)
session: Auth session (injected)
Returns:
Chart.js multi-line chart with sleep metrics
"""
profile_id = session['profile_id']
duration_data = get_sleep_duration_data(profile_id, days)
quality_data = get_sleep_quality_data(profile_id, days)
if duration_data['confidence'] == 'insufficient':
return {
"chart_type": "line",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Schlafdaten vorhanden"
}
}
from db import get_db, get_cursor
with get_db() as conn:
cur = get_cursor(conn)
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
cur.execute(
"""SELECT date, total_sleep_min
FROM sleep_log
WHERE profile_id=%s AND date >= %s
ORDER BY date""",
(profile_id, cutoff)
)
rows = cur.fetchall()
if not rows:
return {
"chart_type": "line",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Schlafdaten"
}
}
labels = [row['date'].isoformat() for row in rows]
duration_hours = [safe_float(row['total_sleep_min']) / 60 if row['total_sleep_min'] else None for row in rows]
# Quality score (simple proxy: % of 8 hours)
quality_scores = [(d / 8 * 100) if d else None for d in duration_hours]
datasets = [
{
"label": "Schlafdauer (h)",
"data": duration_hours,
"borderColor": "#3B82F6",
"backgroundColor": "rgba(59, 130, 246, 0.1)",
"borderWidth": 2,
"tension": 0.3,
"yAxisID": "y1",
"fill": True
},
{
"label": "Qualität (%)",
"data": quality_scores,
"borderColor": "#1D9E75",
"backgroundColor": "rgba(29, 158, 117, 0.1)",
"borderWidth": 2,
"tension": 0.3,
"yAxisID": "y2",
"fill": False
}
]
return {
"chart_type": "line",
"data": {
"labels": labels,
"datasets": datasets
},
"metadata": serialize_dates({
"confidence": duration_data['confidence'],
"data_points": len(rows),
"avg_duration_hours": round(duration_data['avg_duration_hours'], 1),
"sleep_quality_score": quality_data.get('sleep_quality_score', 0)
})
}
@router.get("/sleep-debt")
def get_sleep_debt_chart(
days: int = Query(default=28, ge=7, le=90),
session: dict = Depends(require_auth)
) -> Dict:
"""
Sleep debt accumulation (R4).
Shows cumulative sleep debt over time.
Args:
days: Analysis window (7-90 days, default 28)
session: Auth session (injected)
Returns:
Chart.js line chart with sleep debt
"""
profile_id = session['profile_id']
current_debt = calculate_sleep_debt_hours(profile_id)
if current_debt is None:
return {
"chart_type": "line",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Schlafdaten für Schulden-Berechnung"
}
}
from db import get_db, get_cursor
with get_db() as conn:
cur = get_cursor(conn)
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
cur.execute(
"""SELECT date, total_sleep_min
FROM sleep_log
WHERE profile_id=%s AND date >= %s
ORDER BY date""",
(profile_id, cutoff)
)
rows = cur.fetchall()
if not rows:
return {
"chart_type": "line",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Schlafdaten"
}
}
labels = [row['date'].isoformat() for row in rows]
# Calculate cumulative debt (target 8h/night)
target_hours = 8.0
cumulative_debt = 0
debt_values = []
for row in rows:
actual_hours = safe_float(row['total_sleep_min']) / 60 if row['total_sleep_min'] else 0
daily_deficit = target_hours - actual_hours
cumulative_debt += daily_deficit
debt_values.append(cumulative_debt)
return {
"chart_type": "line",
"data": {
"labels": labels,
"datasets": [
{
"label": "Schlafschuld (Stunden)",
"data": debt_values,
"borderColor": "#EF4444",
"backgroundColor": "rgba(239, 68, 68, 0.1)",
"borderWidth": 2,
"tension": 0.3,
"fill": True
}
]
},
"metadata": serialize_dates({
"confidence": calculate_confidence(len(rows), days, "general"),
"data_points": len(rows),
"current_debt_hours": round(current_debt, 1),
"final_debt_hours": round(cumulative_debt, 1)
})
}
@router.get("/vital-signs-matrix")
def get_vital_signs_matrix_chart(
days: int = Query(default=7, ge=7, le=30),
session: dict = Depends(require_auth)
) -> Dict:
"""
Vital signs matrix (R5).
Shows latest vital signs as horizontal bar chart.
Args:
days: Max age of measurements (7-30 days, default 7)
session: Auth session (injected)
Returns:
Chart.js horizontal bar chart with vital signs
"""
profile_id = session['profile_id']
from db import get_db, get_cursor
with get_db() as conn:
cur = get_cursor(conn)
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
# Get latest vitals
cur.execute(
"""SELECT resting_hr, hrv_ms, vo2_max, spo2, respiratory_rate
FROM vitals_baseline
WHERE profile_id=%s AND date >= %s
ORDER BY date DESC
LIMIT 1""",
(profile_id, cutoff)
)
vitals_row = cur.fetchone()
# Get latest blood pressure
cur.execute(
"""SELECT systolic, diastolic
FROM blood_pressure_log
WHERE profile_id=%s AND date >= %s
ORDER BY date DESC, time DESC
LIMIT 1""",
(profile_id, cutoff)
)
bp_row = cur.fetchone()
if not vitals_row and not bp_row:
return {
"chart_type": "bar",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine aktuellen Vitalwerte"
}
}
labels = []
values = []
if vitals_row:
if vitals_row['resting_hr']:
labels.append("Ruhepuls (bpm)")
values.append(safe_float(vitals_row['resting_hr']))
if vitals_row['hrv_ms']:
labels.append("HRV (ms)")
values.append(safe_float(vitals_row['hrv_ms']))
if vitals_row['vo2_max']:
labels.append("VO2 Max")
values.append(safe_float(vitals_row['vo2_max']))
if vitals_row['spo2']:
labels.append("SpO2 (%)")
values.append(safe_float(vitals_row['spo2']))
if vitals_row['respiratory_rate']:
labels.append("Atemfrequenz")
values.append(safe_float(vitals_row['respiratory_rate']))
if bp_row:
if bp_row['systolic']:
labels.append("Blutdruck sys (mmHg)")
values.append(safe_float(bp_row['systolic']))
if bp_row['diastolic']:
labels.append("Blutdruck dia (mmHg)")
values.append(safe_float(bp_row['diastolic']))
if not labels:
return {
"chart_type": "bar",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Keine Vitalwerte verfügbar"
}
}
return {
"chart_type": "bar",
"data": {
"labels": labels,
"datasets": [
{
"label": "Wert",
"data": values,
"backgroundColor": "#1D9E75",
"borderColor": "#085041",
"borderWidth": 1
}
]
},
"metadata": {
"confidence": "medium",
"data_points": len(values),
"note": "Latest measurements within last " + str(days) + " days"
}
}
# ── Correlation Charts ──────────────────────────────────────────────────────
@router.get("/weight-energy-correlation")
def get_weight_energy_correlation_chart(
max_lag: int = Query(default=14, ge=7, le=28),
session: dict = Depends(require_auth)
) -> Dict:
"""
Weight vs energy balance correlation (C1).
Shows lag correlation between energy intake and weight change.
Args:
max_lag: Maximum lag days to analyze (7-28, default 14)
session: Auth session (injected)
Returns:
Chart.js scatter chart with correlation data
"""
profile_id = session['profile_id']
corr_data = calculate_lag_correlation(profile_id, "energy_balance", "weight", max_lag)
if not corr_data or corr_data.get('correlation') is None:
return {
"chart_type": "scatter",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Nicht genug Daten für Korrelationsanalyse"
}
}
# Create lag vs correlation data for chart
# For simplicity, show best lag point as single data point
best_lag = corr_data.get('best_lag_days', 0)
correlation = corr_data.get('correlation', 0)
return {
"chart_type": "scatter",
"data": {
"labels": [f"Lag {best_lag} Tage"],
"datasets": [
{
"label": "Korrelation",
"data": [{"x": best_lag, "y": correlation}],
"backgroundColor": "#1D9E75",
"borderColor": "#085041",
"borderWidth": 2,
"pointRadius": 8
}
]
},
"metadata": {
"confidence": corr_data.get('confidence', 'low'),
"correlation": round(correlation, 3),
"best_lag_days": best_lag,
"interpretation": corr_data.get('interpretation', ''),
"data_points": corr_data.get('data_points', 0)
}
}
@router.get("/lbm-protein-correlation")
def get_lbm_protein_correlation_chart(
max_lag: int = Query(default=14, ge=7, le=28),
session: dict = Depends(require_auth)
) -> Dict:
"""
Lean mass vs protein intake correlation (C2).
Shows lag correlation between protein intake and lean mass change.
Args:
max_lag: Maximum lag days to analyze (7-28, default 14)
session: Auth session (injected)
Returns:
Chart.js scatter chart with correlation data
"""
profile_id = session['profile_id']
corr_data = calculate_lag_correlation(profile_id, "protein", "lbm", max_lag)
if not corr_data or corr_data.get('correlation') is None:
return {
"chart_type": "scatter",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Nicht genug Daten für LBM-Protein Korrelation"
}
}
best_lag = corr_data.get('best_lag_days', 0)
correlation = corr_data.get('correlation', 0)
return {
"chart_type": "scatter",
"data": {
"labels": [f"Lag {best_lag} Tage"],
"datasets": [
{
"label": "Korrelation",
"data": [{"x": best_lag, "y": correlation}],
"backgroundColor": "#3B82F6",
"borderColor": "#1E40AF",
"borderWidth": 2,
"pointRadius": 8
}
]
},
"metadata": {
"confidence": corr_data.get('confidence', 'low'),
"correlation": round(correlation, 3),
"best_lag_days": best_lag,
"interpretation": corr_data.get('interpretation', ''),
"data_points": corr_data.get('data_points', 0)
}
}
@router.get("/load-vitals-correlation")
def get_load_vitals_correlation_chart(
max_lag: int = Query(default=14, ge=7, le=28),
session: dict = Depends(require_auth)
) -> Dict:
"""
Training load vs vitals correlation (C3).
Shows lag correlation between training load and HRV/RHR.
Args:
max_lag: Maximum lag days to analyze (7-28, default 14)
session: Auth session (injected)
Returns:
Chart.js scatter chart with correlation data
"""
profile_id = session['profile_id']
# Try HRV first
corr_hrv = calculate_lag_correlation(profile_id, "load", "hrv", max_lag)
corr_rhr = calculate_lag_correlation(profile_id, "load", "rhr", max_lag)
# Use whichever has stronger correlation
if corr_hrv and corr_rhr:
corr_data = corr_hrv if abs(corr_hrv.get('correlation', 0)) > abs(corr_rhr.get('correlation', 0)) else corr_rhr
metric_name = "HRV" if corr_data == corr_hrv else "RHR"
elif corr_hrv:
corr_data = corr_hrv
metric_name = "HRV"
elif corr_rhr:
corr_data = corr_rhr
metric_name = "RHR"
else:
return {
"chart_type": "scatter",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Nicht genug Daten für Load-Vitals Korrelation"
}
}
best_lag = corr_data.get('best_lag_days', 0)
correlation = corr_data.get('correlation', 0)
return {
"chart_type": "scatter",
"data": {
"labels": [f"Load → {metric_name} (Lag {best_lag}d)"],
"datasets": [
{
"label": "Korrelation",
"data": [{"x": best_lag, "y": correlation}],
"backgroundColor": "#F59E0B",
"borderColor": "#D97706",
"borderWidth": 2,
"pointRadius": 8
}
]
},
"metadata": {
"confidence": corr_data.get('confidence', 'low'),
"correlation": round(correlation, 3),
"best_lag_days": best_lag,
"metric": metric_name,
"interpretation": corr_data.get('interpretation', ''),
"data_points": corr_data.get('data_points', 0)
}
}
@router.get("/recovery-performance")
def get_recovery_performance_chart(
session: dict = Depends(require_auth)
) -> Dict:
"""
Recovery vs performance correlation (C4).
Shows relationship between recovery metrics and training quality.
Args:
session: Auth session (injected)
Returns:
Chart.js bar chart with top drivers
"""
profile_id = session['profile_id']
# Get top drivers (hindering/helpful factors)
drivers = calculate_top_drivers(profile_id)
if not drivers or len(drivers) == 0:
return {
"chart_type": "bar",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": 0,
"message": "Nicht genug Daten für Driver-Analyse"
}
}
# Separate hindering and helpful
hindering = [d for d in drivers if d.get('impact', '') == 'hindering']
helpful = [d for d in drivers if d.get('impact', '') == 'helpful']
# Take top 3 of each
top_hindering = hindering[:3]
top_helpful = helpful[:3]
labels = []
values = []
colors = []
for d in top_hindering:
labels.append(f"{d.get('factor', '')}")
values.append(-abs(d.get('score', 0))) # Negative for hindering
colors.append("#EF4444")
for d in top_helpful:
labels.append(f"{d.get('factor', '')}")
values.append(abs(d.get('score', 0))) # Positive for helpful
colors.append("#1D9E75")
if not labels:
return {
"chart_type": "bar",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "low",
"data_points": 0,
"message": "Keine signifikanten Treiber gefunden"
}
}
return {
"chart_type": "bar",
"data": {
"labels": labels,
"datasets": [
{
"label": "Impact Score",
"data": values,
"backgroundColor": colors,
"borderColor": "#085041",
"borderWidth": 1
}
]
},
"metadata": {
"confidence": "medium",
"hindering_count": len(top_hindering),
"helpful_count": len(top_helpful),
"total_factors": len(drivers)
}
}
# ── Health Endpoint ──────────────────────────────────────────────────────────
@router.get("/health")
def health_check() -> Dict:
"""
Health check endpoint for charts API.
Returns:
{
"status": "ok",
"version": "1.0",
"available_charts": [...]
}
"""
return {
"status": "ok",
"version": "1.0",
"phase": "0c",
"available_charts": [
{
"category": "body",
"endpoint": "/charts/weight-trend",
"type": "line",
"description": "Weight trend over time"
},
{
"category": "body",
"endpoint": "/charts/body-composition",
"type": "line",
"description": "Body fat % and lean mass"
},
{
"category": "body",
"endpoint": "/charts/circumferences",
"type": "bar",
"description": "Latest circumference measurements"
},
{
"category": "nutrition",
"endpoint": "/charts/energy-balance",
"type": "line",
"description": "Daily calorie intake vs. TDEE"
},
{
"category": "nutrition",
"endpoint": "/charts/macro-distribution",
"type": "pie",
"description": "Protein/Carbs/Fat distribution"
},
{
"category": "nutrition",
"endpoint": "/charts/protein-adequacy",
"type": "line",
"description": "Protein intake vs. target range"
},
{
"category": "nutrition",
"endpoint": "/charts/nutrition-consistency",
"type": "bar",
"description": "Macro consistency score"
},
{
"category": "activity",
"endpoint": "/charts/training-volume",
"type": "bar",
"description": "Weekly training minutes"
},
{
"category": "activity",
"endpoint": "/charts/training-type-distribution",
"type": "pie",
"description": "Training category distribution"
},
{
"category": "activity",
"endpoint": "/charts/quality-sessions",
"type": "bar",
"description": "Quality session rate"
},
{
"category": "activity",
"endpoint": "/charts/load-monitoring",
"type": "line",
"description": "Acute vs chronic load + ACWR"
},
{
"category": "activity",
"endpoint": "/charts/monotony-strain",
"type": "bar",
"description": "Training monotony and strain"
},
{
"category": "activity",
"endpoint": "/charts/ability-balance",
"type": "radar",
"description": "Training balance across 5 abilities"
},
{
"category": "activity",
"endpoint": "/charts/volume-by-ability",
"type": "bar",
"description": "Training volume per ability"
},
{
"category": "recovery",
"endpoint": "/charts/recovery-score",
"type": "line",
"description": "Recovery score timeline"
},
{
"category": "recovery",
"endpoint": "/charts/hrv-rhr-baseline",
"type": "line",
"description": "HRV and RHR vs baseline"
},
{
"category": "recovery",
"endpoint": "/charts/sleep-duration-quality",
"type": "line",
"description": "Sleep duration and quality"
},
{
"category": "recovery",
"endpoint": "/charts/sleep-debt",
"type": "line",
"description": "Cumulative sleep debt"
},
{
"category": "recovery",
"endpoint": "/charts/vital-signs-matrix",
"type": "bar",
"description": "Latest vital signs overview"
},
{
"category": "correlations",
"endpoint": "/charts/weight-energy-correlation",
"type": "scatter",
"description": "Weight vs energy balance (lag correlation)"
},
{
"category": "correlations",
"endpoint": "/charts/lbm-protein-correlation",
"type": "scatter",
"description": "Lean mass vs protein intake (lag correlation)"
},
{
"category": "correlations",
"endpoint": "/charts/load-vitals-correlation",
"type": "scatter",
"description": "Training load vs HRV/RHR (lag correlation)"
},
{
"category": "correlations",
"endpoint": "/charts/recovery-performance",
"type": "bar",
"description": "Top drivers (hindering/helpful factors)"
}
]
}