""" Chart.js-kompatible Payloads für Lag-Korrelationen C1–C3 und Treiber C4. Gemeinsame Quelle für GET /charts/* und history_overview_viz.chart_payloads (Issue 53). """ from __future__ import annotations from typing import Any, Dict from data_layer.correlations import calculate_lag_correlation, calculate_top_drivers def build_weight_energy_correlation_chart_payload(profile_id: str, max_lag: int) -> Dict[str, Any]: corr_data = calculate_lag_correlation(profile_id, "energy_balance", "weight", max_lag) if not corr_data or corr_data.get("correlation") is None: msg = "Nicht genug Daten für Korrelationsanalyse" if isinstance(corr_data, dict): msg = str(corr_data.get("interpretation") or corr_data.get("reason") or msg) return { "chart_type": "scatter", "data": {"labels": [], "datasets": []}, "metadata": { "confidence": "insufficient", "data_points": corr_data.get("data_points", 0) if isinstance(corr_data, dict) else 0, "message": msg, "lag_details": corr_data.get("lag_details") if isinstance(corr_data, dict) else None, "tdee_kcal_used": corr_data.get("tdee_kcal_used") if isinstance(corr_data, dict) else None, }, } best_lag = corr_data.get("best_lag_days", corr_data.get("best_lag", 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(float(correlation), 3), "best_lag_days": best_lag, "interpretation": corr_data.get("interpretation", ""), "data_points": corr_data.get("data_points", 0), "lag_details": corr_data.get("lag_details"), "tdee_kcal_used": corr_data.get("tdee_kcal_used"), "layer_1": "correlations._correlate_energy_weight", }, } def build_lbm_protein_correlation_chart_payload(profile_id: str, max_lag: int) -> Dict[str, Any]: corr_data = calculate_lag_correlation(profile_id, "protein", "lbm", max_lag) if not corr_data or corr_data.get("correlation") is None: msg = "Nicht genug Daten für LBM-Protein Korrelation" if isinstance(corr_data, dict): msg = str(corr_data.get("interpretation") or corr_data.get("reason") or msg) return { "chart_type": "scatter", "data": {"labels": [], "datasets": []}, "metadata": { "confidence": "insufficient", "data_points": corr_data.get("data_points", 0) if isinstance(corr_data, dict) else 0, "message": msg, "lag_details": corr_data.get("lag_details") if isinstance(corr_data, dict) else None, }, } best_lag = corr_data.get("best_lag_days", corr_data.get("best_lag", 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(float(correlation), 3), "best_lag_days": best_lag, "interpretation": corr_data.get("interpretation", ""), "data_points": corr_data.get("data_points", 0), "lag_details": corr_data.get("lag_details"), "layer_1": "correlations._correlate_protein_lbm", }, } def build_load_vitals_correlation_chart_payload(profile_id: str, max_lag: int) -> Dict[str, Any]: corr_hrv = calculate_lag_correlation(profile_id, "load", "hrv", max_lag) corr_rhr = calculate_lag_correlation(profile_id, "load", "rhr", max_lag) def _abs_corr(c: Any) -> float: if not c or c.get("correlation") is None: return -1.0 try: return abs(float(c["correlation"])) except (TypeError, ValueError): return -1.0 if _abs_corr(corr_hrv) < 0 and _abs_corr(corr_rhr) < 0: msg = "Nicht genug Daten für Load-Vitals Korrelation" h_msg = corr_hrv.get("interpretation") if isinstance(corr_hrv, dict) else None r_msg = corr_rhr.get("interpretation") if isinstance(corr_rhr, dict) else None if h_msg or r_msg: msg = f"HRV: {h_msg or '—'} · RHR: {r_msg or '—'}" return { "chart_type": "scatter", "data": {"labels": [], "datasets": []}, "metadata": { "confidence": "insufficient", "data_points": 0, "message": msg, "lag_details_hrv": corr_hrv.get("lag_details") if isinstance(corr_hrv, dict) else None, "lag_details_rhr": corr_rhr.get("lag_details") if isinstance(corr_rhr, dict) else None, }, } if _abs_corr(corr_hrv) >= _abs_corr(corr_rhr): corr_data = corr_hrv metric_name = "HRV" else: corr_data = corr_rhr metric_name = "RHR" 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": str(corr_data.get("interpretation") or "Nicht genug Daten für Load-Vitals Korrelation"), }, } best_lag = corr_data.get("best_lag_days", corr_data.get("best_lag", 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(float(correlation), 3), "best_lag_days": best_lag, "metric": metric_name, "interpretation": corr_data.get("interpretation", ""), "data_points": corr_data.get("data_points", 0), "lag_details": corr_data.get("lag_details"), "layer_1": "correlations._correlate_load_vitals", }, } def build_recovery_performance_chart_payload(profile_id: str) -> Dict[str, Any]: 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", }, } hindering = [d for d in drivers if d.get("impact", "") == "hindering"] helpful = [d for d in drivers if d.get("impact", "") == "helpful"] 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))) colors.append("#EF4444") for d in top_helpful: labels.append(f"✅ {d.get('factor', '')}") values.append(abs(d.get("score", 0))) 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), }, }