257 lines
9.2 KiB
Python
257 lines
9.2 KiB
Python
|
"""
|
|||
|
|
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),
|
|||
|
|
},
|
|||
|
|
}
|