Merge pull request 'Platzhalter finalisiert - Option |d und Option |x implementiert' (#77) from develop into main
All checks were successful
Deploy Production / deploy (push) Successful in 1m3s
Build Test / pytest-backend (push) Successful in 4s
Build Test / lint-backend (push) Successful in 0s
Build Test / build-frontend (push) Successful in 16s

Reviewed-on: #77
This commit is contained in:
Lars 2026-04-11 22:10:10 +02:00
commit caeed3fbaa
40 changed files with 3579 additions and 646 deletions

View File

@ -92,16 +92,10 @@ registry = get_registry()
**Package:** `backend/placeholder_registrations/`
**Struktur:**
```
placeholder_registrations/
├── __init__.py # Auto-Import aller Registrations
├── nutrition_part_a.py # Nutrition Basis-Metriken (4 Placeholder)
├── nutrition_part_b.py # Protein-Ziele (5 Placeholder) - TODO
├── body_metrics.py # Körper-Metriken - TODO
├── activity_metrics.py # Aktivitäts-Metriken - TODO
└── ... # Weitere Cluster
```
**Struktur:** Vollständige Cluster-Module (u. a. Ernährung, Körper, Aktivität, Schlaf,
Vitalwerte, Profil/Zeitraum, Phase-0b-Ziele, Korrelationen); siehe `__init__.py` für die
Import-Liste. **Anzahl:** 114 Platzhalter, identisch zu `PLACEHOLDER_MAP` in
`placeholder_resolver.py`.
**Auto-Registration:**
- Import des Package triggert automatische Registrierung aller Placeholder

View File

@ -5,7 +5,9 @@ Folgende Ergebnisse des Tests:
- In der automatischen Zusammenfassung in der Endnode kommt als Überschrift, z.B. Node 10, anstatt den Node-Name auszugeben.
- Alle Änderungen an Nodes scheinen automatisch in den Gesamtflow übernommen zu werden. Diese werden dann nach dem Speichern aktiv. Da muss man sehr vorsichtig sein, bei kurzen Änderungen und dem Ausprobieren.
- Der Testlauf "Execute" sollte auf dem aktuellen Workflowstand ausgeführt werden, auch wenn dieser vom gespeicherten Abweicht. Ich würde natürlich vor dem Speichern den Workflow testen können. Prüfe und bewerte diesen Punkt, setze ihn aber noch nicht um.
- Die Workflows werden aktuell nicht in Analyse und den verfügbaren KI-Asuwertungen angezeigt. ggf. weil wir sie aktuell noch keinem Bereich zuordnen können. Diesen könnten wir ggf. über die Start-Node im Workflow konfigurieren.
- Das löschen von Knoten und Kanten funktioniert aktuell nur über Backspace nicht über entfernen
- Wir sollten auch dafür sorgen, dass jeweils nur eine Start-Node, End-Node in einem Workflow existiert, Prüfe ob mehrere End-Nodes sinnvoll sind, da wir ja auch Logik-Pfade abbilden und ggf. auch eine route beschreiten, die ein anderes Ende hat. (Prüfe, ob das heute schon möglich wäre!)
- Als zukünftige Ausbaustufe sollten wir überlegen, ob wir auch Trigger implementieren, z.B. um Kurzstatements zu generieren, wenn neue Daten hereinkommen und wir diese Bewertungen aktualisieren wollen
- Als zukünftige Ausbaustufe sollten wir überlegen, ob wir auch Trigger implementieren, z.B. um Kurzstatements zu generieren, wenn neue Daten hereinkommen und wir diese Bewertungen aktualisieren wollen
- Exportieren aller KI-Prompts/Templates/Workflows im Admin --> KI-Prompts führt zu einem "internal Server Error", Importieren konnte daraufhin nicht getestet werden
- Das duplizieren von Workflows funktioniert nicht
-

View File

@ -7,7 +7,7 @@
## Gesamt-Übersicht
**Aktuelle Platzhalter:** 116
**Aktuelle Platzhalter:** 114 (PLACEHOLDER_MAP / Registry)
**Nach Phase 0c Migration:**
- ✅ **Bleiben einfach (kein Data Layer):** 8 Platzhalter
- 🔄 **Gehen zu Data Layer:** 108 Platzhalter

View File

@ -105,6 +105,22 @@ frontend/src/
- **Regeln:** Verweise in `.claude/rules/ARCHITECTURE.md` (§3.2 `source`), `.claude/rules/CODING_RULES.md` (§6)
- **Follow-ups:** **Gitea #71** Dry-Run inkl. `import_row_processing`, Nutzer-Mapping-Validierung, Fehler-Hints in der Import-UI ([Issue](http://192.168.2.144:3000/Lars/mitai-jinkendo/issues/71))
### Updates (11.04.2026 - Placeholder Phase A)
- **`main.py`:** `import placeholder_registrations` beim Start, damit die Registry (**114 Keys**, deckungsgleich `PLACEHOLDER_MAP`) und `get_placeholder_catalog()` ohne vorherigen Export-Request konsistent sind.
- **`placeholder_resolver.py`:** `{{top_goal_progress_pct}}` nutzt `_safe_int` statt `_safe_str` (Verdrahtung zu `scores.get_top_priority_goal` korrigiert).
### Updates (11.04.2026 - Gitea #75, nutrition_score Registry)
- **Gitea #75** (offen): Zucker/Ballaststoffe/Lebensmittelqualität, automatisches Lebensmittelprofil, später Mahlzeiten-Timing/Abgleich mit Training — http://192.168.2.144:3000/Lars/mitai-jinkendo/issues/75
- **`nutrition_score`:** Registry in `backend/placeholder_registrations/nutrition_score.py`, Import in `placeholder_registrations/__init__.py`; Legacy-Duplikat unter „Scores“ im Platzhalter-Katalog entfernt.
### Updates (11.04.2026 - Ernährung: TDEE, Bilanz, Kalorien-Score)
- **`data_layer/nutrition_metrics.py`:** TDEE für Bilanz: primär **MifflinSt Jeor BMR × PAL 1,55**, wenn Profil (Größe, Geschlecht, DOB) und Gewicht vorhanden; sonst Fallback **kg × 32,5** (`estimate_tdee_kcal_from_latest_weight`). `get_energy_balance_data` / `calculate_energy_balance_7d` nutzen **tägliche kcal-Summen**. **`_score_calorie_adherence`** (Komponente von `calculate_nutrition_score`) wertet die 7-Tage-Bilanz nach **`profiles.goal_mode`** aus (weight_loss vs. strength/recomposition vs. maintenance/health/endurance).
- **`routers/charts.py`:** `/charts/energy-balance` und Protein-Timeline nutzen dieselbe TDEE-/Tageslogik; ohne `weight_log` liefert Energiebilanz-Chart eine klare Fehlermeldung. Adherence-Endpoint: Kcal-CV über **Tages-Summen**.
- **Doku:** Normative Platzhalter-Zahl **114** (`docs/PLACEHOLDER_*.md`); `placeholder_metadata_complete.py` als **Legacy** gekennzeichnet — maßgeblich `placeholder_registrations/` + `PLACEHOLDER_REGISTRY_FRAMEWORK.md`.
### GUI / Informationsarchitektur (Abnahme dieser Iteration, 2026-04-05)
Admin-Bereich (`AdminShell`, Hub-Routen), Hauptnavigation inkl. **Ziele** (`config/appNav.js`), Einstellungen nur aktives Profil + E-Mail, KI-Analyse Ergebnis in rechter Spalte, **PWA** Bottom-Nav inkl. iOS Safe Area. Zentrale Agent-Doku: **`docs/issues/GUI_IA_ADMIN_NAV_2026-04-05.md`**. Responsive-Epic **Gitea #30:** Phasenplan `docs/issues/PHASE_PLAN_RESPONSIVE_UI.md`**P7 Kern erledigt**, **P8** (Regression/Abnahme) ausstehend; Issue bewusst **nicht** geschlossen.

View File

@ -509,17 +509,24 @@ def calculate_sleep_quality_7d(profile_id: str) -> Optional[int]:
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)
dur = s["duration_minutes"]
if not dur or dur <= 0:
continue
d = s["deep_minutes"]
r = s["rem_minutes"]
if d is None and r is None:
continue
di, ri = (d or 0), (r or 0)
quality_pct = ((di + ri) / dur) * 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 not quality_scores:
return None

View File

@ -51,6 +51,9 @@ __all__ = [
# Body Metrics (Basic)
'get_latest_weight_data',
'get_bmi_data',
'get_profile_goal_weight_data',
'get_profile_goal_bf_pct_data',
'get_weight_trend_data',
'get_body_composition_data',
'get_circumference_summary_data',
@ -99,6 +102,9 @@ __all__ = [
'get_activity_summary_data',
'get_activity_detail_data',
'get_training_type_distribution_data',
'get_training_frequency_by_type_data',
'get_training_inter_session_gap_data',
'get_training_sessions_recent_weeks_data',
# Activity Metrics (Calculated)
'calculate_training_minutes_week',

View File

@ -7,6 +7,9 @@ Functions:
- get_activity_summary_data(): Count, total duration, calories, averages
- get_activity_detail_data(): Detailed activity log entries
- get_training_type_distribution_data(): Training category percentages
- get_training_frequency_by_type_data(): Häufigkeit & Intensität pro activity_type
- get_training_inter_session_gap_data(): Pausen zwischen Einheiten (Stunden)
- get_training_sessions_recent_weeks_data(): Wochen-JSON für KI-Kontext
All functions return structured data (dict) without formatting.
Use placeholder_resolver.py for formatted strings for AI.
@ -15,11 +18,11 @@ Phase 0c: Multi-Layer Architecture
Version: 1.0
"""
from typing import Dict, List, Optional
from datetime import datetime, timedelta, date
from typing import Dict, List, Optional, Any
from datetime import datetime, timedelta, date, time
import statistics
from db import get_db, get_cursor, r2d
from data_layer.utils import calculate_confidence, safe_float, safe_int
from data_layer.utils import calculate_confidence, safe_float, safe_int, serialize_dates
def get_activity_summary_data(
@ -671,9 +674,9 @@ def calculate_activity_score(profile_id: str, focus_weights: Optional[Dict] = No
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)
# Weighted average (float: DB-Aggregate können Decimal sein)
total_score = sum(float(score) * float(weight) for _, score, weight in components)
total_weight = sum(float(weight) for _, _, weight in components)
return int(total_score / total_weight)
@ -725,12 +728,13 @@ def _score_cardio_presence(profile_id: str) -> Optional[int]:
if not row:
return None
cardio_days = row['cardio_days']
cardio_minutes = row['cardio_minutes'] or 0
# psycopg2: SUM() → oft Decimal — vor Mix mit float konvertieren
cardio_days = int(row['cardio_days'] or 0)
cardio_minutes = float(row['cardio_minutes'] or 0)
# Target: 3-5 days/week, 150+ minutes
day_score = min(100, (cardio_days / 4) * 100)
minute_score = min(100, (cardio_minutes / 150) * 100)
day_score = min(100.0, (cardio_days / 4) * 100)
minute_score = min(100.0, (cardio_minutes / 150) * 100)
return int((day_score + minute_score) / 2)
@ -904,3 +908,266 @@ def calculate_activity_data_quality(profile_id: str) -> Dict[str, any]:
"quality": int(quality_score)
}
}
def _session_sort_ts(row: Dict) -> datetime:
"""Einheitlicher Zeitstempel für Sortierung und Pausenberechnung."""
d = row["date"]
if isinstance(d, str):
d = datetime.strptime(d[:10], "%Y-%m-%d").date()
st = row.get("start_time")
if st is None:
t = time(12, 0, 0)
else:
t = st
return datetime.combine(d, t)
def get_training_frequency_by_type_data(
profile_id: str,
days: int = 28,
) -> Dict[str, Any]:
"""
Pro activity_type (Roh-Label aus Import/Anzeige): Häufigkeit & Intensitätskennzahlen.
Returns:
{
"days_analyzed": int,
"confidence": str,
"by_type": [
{
"activity_type": str,
"session_count": int,
"sessions_per_week": float,
"avg_duration_min": float | None,
"avg_kcal_active": float | None,
"avg_hr_avg": float | None,
"avg_hr_max": float | None,
"avg_rpe": float | None,
"avg_kcal_per_min": float | None, # grobe Intensität, wenn kcal & Dauer
},
...
],
}
"""
weeks = max(days / 7.0, 0.01)
with get_db() as conn:
cur = get_cursor(conn)
cutoff = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
cur.execute(
"""
SELECT
activity_type,
COUNT(*)::int AS session_count,
AVG(duration_min)::float AS avg_duration_min,
AVG(kcal_active)::float AS avg_kcal_active,
AVG(hr_avg)::float AS avg_hr_avg,
AVG(hr_max)::float AS avg_hr_max,
AVG(rpe)::float AS avg_rpe,
SUM(COALESCE(duration_min, 0))::float AS sum_duration,
SUM(COALESCE(kcal_active, 0))::float AS sum_kcal
FROM activity_log
WHERE profile_id = %s AND date >= %s
GROUP BY activity_type
ORDER BY session_count DESC
""",
(profile_id, cutoff),
)
rows = [r2d(r) for r in cur.fetchall()]
if not rows:
return {
"days_analyzed": days,
"confidence": "insufficient",
"by_type": [],
}
by_type = []
for r in rows:
sc = int(r["session_count"])
sum_dur = float(r["sum_duration"] or 0)
sum_kcal = float(r["sum_kcal"] or 0)
kcal_per_min = (sum_kcal / sum_dur) if sum_dur > 0 else None
by_type.append(
{
"activity_type": r["activity_type"],
"session_count": sc,
"sessions_per_week": round(sc / weeks, 2),
"avg_duration_min": r["avg_duration_min"],
"avg_kcal_active": r["avg_kcal_active"],
"avg_hr_avg": r["avg_hr_avg"],
"avg_hr_max": r["avg_hr_max"],
"avg_rpe": r["avg_rpe"],
"avg_kcal_per_min": round(kcal_per_min, 2) if kcal_per_min is not None else None,
}
)
total_sessions = sum(x["session_count"] for x in by_type)
confidence = calculate_confidence(total_sessions, days, "general")
return {
"days_analyzed": days,
"confidence": confidence,
"by_type": by_type,
}
def get_training_inter_session_gap_data(
profile_id: str,
days: int = 28,
) -> Dict[str, Any]:
"""
Mittlere/median Pausen zwischen aufeinanderfolgenden Trainingseinheiten (Stunden).
Sortierung: Datum + start_time (fehlend 12:00), dann created.
"""
with get_db() as conn:
cur = get_cursor(conn)
cutoff = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
cur.execute(
"""
SELECT date, start_time, created
FROM activity_log
WHERE profile_id = %s AND date >= %s
ORDER BY date ASC, start_time ASC NULLS LAST, created ASC
""",
(profile_id, cutoff),
)
rows = [r2d(r) for r in cur.fetchall()]
if len(rows) < 2:
return {
"days_analyzed": days,
"confidence": "insufficient",
"gap_hours_median": None,
"gap_hours_mean": None,
"gap_hours_min": None,
"gaps_count": 0,
}
gaps = []
prev_ts = None
for r in rows:
ts = _session_sort_ts(r)
if prev_ts is not None:
gaps.append((ts - prev_ts).total_seconds() / 3600.0)
prev_ts = ts
if not gaps:
return {
"days_analyzed": days,
"confidence": "insufficient",
"gap_hours_median": None,
"gap_hours_mean": None,
"gap_hours_min": None,
"gaps_count": 0,
}
gaps_sorted = sorted(gaps)
mid = len(gaps_sorted) // 2
median = (
gaps_sorted[mid]
if len(gaps_sorted) % 2
else (gaps_sorted[mid - 1] + gaps_sorted[mid]) / 2.0
)
confidence = calculate_confidence(len(rows), days, "general")
return {
"days_analyzed": days,
"confidence": confidence,
"gap_hours_median": round(median, 1),
"gap_hours_mean": round(statistics.mean(gaps), 1),
"gap_hours_min": round(min(gaps), 1),
"gaps_count": len(gaps),
}
def get_training_sessions_recent_weeks_data(
profile_id: str,
weeks: int = 4,
) -> Dict[str, Any]:
"""
Letzte Wochen mit Einzeltrainings für KI-Kontext (Dauer, kcal, HF, Typ).
weeks: Anzahl zurückliegender ISO-Kalenderwochen (Default 4).
"""
days = max(weeks * 7, 7)
with get_db() as conn:
cur = get_cursor(conn)
cutoff = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
cur.execute(
"""
SELECT
a.id,
a.date,
a.start_time,
a.activity_type,
a.training_category,
a.duration_min,
a.kcal_active,
a.hr_avg,
a.hr_max,
a.rpe,
tt.name_de AS training_type_name
FROM activity_log a
LEFT JOIN training_types tt ON tt.id = a.training_type_id
WHERE a.profile_id = %s AND a.date >= %s
ORDER BY a.date ASC, a.start_time ASC NULLS LAST, a.created ASC
""",
(profile_id, cutoff),
)
rows = [r2d(r) for r in cur.fetchall()]
if not rows:
return {
"weeks": [],
"meta": {
"weeks_requested": weeks,
"days_loaded": days,
"session_count": 0,
"confidence": "insufficient",
},
}
by_week: Dict[str, List[Dict]] = {}
for r in rows:
d = r["date"]
if isinstance(d, str):
d = datetime.strptime(d[:10], "%Y-%m-%d").date()
iso = d.isocalendar()
wk = f"{iso.year}-W{iso.week:02d}"
if wk not in by_week:
by_week[wk] = []
dur = r.get("duration_min")
dur_f = float(dur) if dur is not None else None
kcal = r.get("kcal_active")
kcal_f = float(kcal) if kcal is not None else None
hr_a = r.get("hr_avg")
hr_m = r.get("hr_max")
by_week[wk].append(
{
"date": d,
"start_time": str(r["start_time"]) if r.get("start_time") is not None else None,
"activity_type": r.get("activity_type"),
"training_category": r.get("training_category"),
"training_type_name": r.get("training_type_name"),
"duration_min": dur_f,
"kcal_active": kcal_f,
"hr_avg": int(hr_a) if hr_a is not None else None,
"hr_max": int(hr_m) if hr_m is not None else None,
"rpe": int(r["rpe"]) if r.get("rpe") is not None else None,
}
)
week_keys = sorted(by_week.keys())
weeks_out = [{"week_iso": wk, "sessions": by_week[wk]} for wk in week_keys]
confidence = calculate_confidence(len(rows), days, "general")
return serialize_dates(
{
"weeks": weeks_out,
"meta": {
"weeks_requested": weeks,
"days_loaded": days,
"session_count": len(rows),
"confidence": confidence,
},
}
)

View File

@ -5,6 +5,9 @@ Provides structured data for body composition and measurements.
Functions:
- get_latest_weight_data(): Most recent weight entry
- get_bmi_data(): BMI from latest weight + profile height
- get_profile_goal_weight_data(): Zielgewicht (Profilfeld)
- get_profile_goal_bf_pct_data(): Ziel-KFA % (Profilfeld)
- get_weight_trend_data(): Weight trend with slope and direction
- get_body_composition_data(): Body fat percentage and lean mass
- get_circumference_summary_data(): Latest circumference measurements
@ -68,6 +71,105 @@ def get_latest_weight_data(
}
def get_bmi_data(profile_id: str) -> Dict:
"""
BMI from latest weight_log entry and profiles.height (cm).
Returns:
{
"bmi": float | None,
"weight_kg": float | None,
"height_cm": float | None,
"confidence": "high" | "insufficient",
}
"""
with get_db() as conn:
cur = get_cursor(conn)
cur.execute(
"""
SELECT pr.height,
(SELECT wl.weight FROM weight_log wl
WHERE wl.profile_id = pr.id
ORDER BY wl.date DESC
LIMIT 1) AS weight
FROM profiles pr
WHERE pr.id = %s
""",
(profile_id,),
)
row = cur.fetchone()
if not row:
return {
"bmi": None,
"weight_kg": None,
"height_cm": None,
"confidence": "insufficient",
}
height_cm = row["height"]
weight = row["weight"]
if height_cm is None or weight is None:
return {
"bmi": None,
"weight_kg": safe_float(weight) if weight is not None else None,
"height_cm": safe_float(height_cm) if height_cm is not None else None,
"confidence": "insufficient",
}
h = safe_float(height_cm)
w = safe_float(weight)
if h <= 0:
return {
"bmi": None,
"weight_kg": w,
"height_cm": h,
"confidence": "insufficient",
}
height_m = h / 100.0
bmi = w / (height_m ** 2)
return {
"bmi": bmi,
"weight_kg": w,
"height_cm": h,
"confidence": "high",
}
def get_profile_goal_weight_data(profile_id: str) -> Dict:
"""Strategisches Zielgewicht aus profiles.goal_weight (kg), nicht goals-Tabelle."""
with get_db() as conn:
cur = get_cursor(conn)
cur.execute(
"SELECT goal_weight FROM profiles WHERE id=%s",
(profile_id,),
)
row = cur.fetchone()
if not row or row.get("goal_weight") is None:
return {"goal_weight_kg": None, "confidence": "insufficient"}
return {
"goal_weight_kg": safe_float(row["goal_weight"]),
"confidence": "high",
}
def get_profile_goal_bf_pct_data(profile_id: str) -> Dict:
"""Strategisches Ziel-KFA aus profiles.goal_bf_pct (%), nicht goals-Tabelle."""
with get_db() as conn:
cur = get_cursor(conn)
cur.execute(
"SELECT goal_bf_pct FROM profiles WHERE id=%s",
(profile_id,),
)
row = cur.fetchone()
if not row or row.get("goal_bf_pct") is None:
return {"goal_bf_pct": None, "confidence": "insufficient"}
return {
"goal_bf_pct": safe_float(row["goal_bf_pct"]),
"confidence": "high",
}
def get_weight_trend_data(
profile_id: str,
days: int = 28
@ -89,7 +191,8 @@ def get_weight_trend_data(
"confidence": str,
"days_analyzed": int,
"first_date": date,
"last_date": date
"last_date": date,
"series": [{"date": date, "weight": float}, ...], # für Charts ohne zweites Query
}
Confidence Rules:
@ -127,7 +230,8 @@ def get_weight_trend_data(
"delta": 0.0,
"direction": "unknown",
"first_date": None,
"last_date": None
"last_date": None,
"series": [],
}
# Extract values
@ -152,7 +256,11 @@ def get_weight_trend_data(
"confidence": confidence,
"days_analyzed": days,
"first_date": rows[0]['date'],
"last_date": rows[-1]['date']
"last_date": rows[-1]['date'],
"series": [
{"date": r["date"], "weight": safe_float(r["weight"])}
for r in rows
],
}
@ -652,8 +760,8 @@ def calculate_body_progress_score(profile_id: str, focus_weights: Optional[Dict]
if not components:
return None
total_score = sum(score * weight for _, score, weight in components)
total_weight = sum(weight for _, _, weight in components)
total_score = sum(float(score) * float(weight) for _, score, weight in components)
total_weight = sum(float(weight) for _, _, weight in components)
return int(total_score / total_weight)

View File

@ -25,6 +25,88 @@ from datetime import datetime, timedelta, date
from db import get_db, get_cursor, r2d
from data_layer.utils import calculate_confidence, safe_float, safe_int
# Fallback TDEE (kcal/day) when demographics for MifflinSt Jeor are incomplete.
TDEE_KCAL_PER_KG_BODYWEIGHT = 32.5
# PAL applied to MSJ BMR when height, sex, dob and weight are available (moderate activity).
TDEE_PAL_MODERATE = 1.55
def _age_years_from_dob(dob) -> Optional[int]:
if dob is None:
return None
try:
if isinstance(dob, str):
birth = datetime.strptime(dob[:10], "%Y-%m-%d").date()
else:
birth = dob
today = date.today()
return today.year - birth.year - ((today.month, today.day) < (birth.month, birth.day))
except Exception:
return None
def _mifflin_st_jeor_bmr_kcal(
weight_kg: float, height_cm: float, age_years: int, sex_is_male: bool
) -> float:
if sex_is_male:
return 10.0 * weight_kg + 6.25 * height_cm - 5.0 * age_years + 5.0
return 10.0 * weight_kg + 6.25 * height_cm - 5.0 * age_years - 161.0
def estimate_tdee_kcal_from_latest_weight(profile_id: str) -> Optional[float]:
"""
Estimated TDEE (kcal/day).
Primary: MifflinSt Jeor BMR × TDEE_PAL_MODERATE when latest weight plus
profiles.height, profiles.sex, profiles.dob are usable.
Fallback: latest weight (kg) × TDEE_KCAL_PER_KG_BODYWEIGHT (legacy heuristic).
Returns None if no weight on record.
"""
with get_db() as conn:
cur = get_cursor(conn)
cur.execute(
"""SELECT weight FROM weight_log
WHERE profile_id=%s ORDER BY date DESC LIMIT 1""",
(profile_id,),
)
wrow = cur.fetchone()
if not wrow or wrow["weight"] is None:
return None
weight_kg = float(wrow["weight"])
cur.execute(
"SELECT height, sex, dob FROM profiles WHERE id=%s",
(profile_id,),
)
prow = cur.fetchone()
if prow and prow.get("height") and prow.get("sex") is not None and prow.get("dob"):
height_cm = float(prow["height"])
age = _age_years_from_dob(prow["dob"])
if age is not None and 10 < age < 120 and height_cm > 50:
sex_raw = str(prow["sex"]).strip().lower()
sex_is_male = sex_raw in ("m", "male", "männlich", "mann")
bmr = _mifflin_st_jeor_bmr_kcal(weight_kg, height_cm, age, sex_is_male)
if bmr > 400:
return bmr * TDEE_PAL_MODERATE
return weight_kg * TDEE_KCAL_PER_KG_BODYWEIGHT
def _get_profile_goal_mode(profile_id: str) -> str:
"""Strategic goal_mode from profiles (Phase 0a); defaults to health."""
with get_db() as conn:
cur = get_cursor(conn)
cur.execute("SELECT goal_mode FROM profiles WHERE id=%s", (profile_id,))
row = cur.fetchone()
if row and row.get("goal_mode"):
g = str(row["goal_mode"]).strip().lower()
if g:
return g
return "health"
def get_nutrition_average_data(
profile_id: str,
@ -56,20 +138,29 @@ def get_nutrition_average_data(
cur = get_cursor(conn)
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
# Mean over calendar days (per-day sums), not over raw log rows.
cur.execute(
"""SELECT
AVG(kcal) as kcal_avg,
AVG(protein_g) as protein_avg,
AVG(carbs_g) as carbs_avg,
AVG(fat_g) as fat_avg,
COUNT(*) as data_points
FROM nutrition_log
WHERE profile_id=%s AND date >= %s""",
(profile_id, cutoff)
AVG(daily_kcal) AS kcal_avg,
AVG(daily_protein) AS protein_avg,
AVG(daily_carbs) AS carbs_avg,
AVG(daily_fat) AS fat_avg,
COUNT(*)::int AS day_count
FROM (
SELECT date,
COALESCE(SUM(kcal), 0)::float AS daily_kcal,
COALESCE(SUM(protein_g), 0)::float AS daily_protein,
COALESCE(SUM(carbs_g), 0)::float AS daily_carbs,
COALESCE(SUM(fat_g), 0)::float AS daily_fat
FROM nutrition_log
WHERE profile_id=%s AND date >= %s
GROUP BY date
) AS daily""",
(profile_id, cutoff),
)
row = cur.fetchone()
if not row or row['data_points'] == 0:
if not row or row["day_count"] == 0:
return {
"kcal_avg": 0.0,
"protein_avg": 0.0,
@ -80,7 +171,7 @@ def get_nutrition_average_data(
"days_analyzed": days
}
data_points = row['data_points']
data_points = row["day_count"]
confidence = calculate_confidence(data_points, days, "general")
return {
@ -190,79 +281,73 @@ def get_energy_balance_data(
days: int = 7
) -> Dict:
"""
Calculate energy balance (intake - estimated expenditure).
Energy balance (intake - estimated expenditure), kcal/day.
Note: This is a simplified calculation.
For accurate TDEE, use profile-based calculations.
Args:
profile_id: User profile ID
days: Analysis window (default 7)
Returns:
{
"energy_balance": float, # kcal/day (negative = deficit)
"avg_intake": float,
"estimated_tdee": float,
"status": str, # "deficit" | "surplus" | "maintenance"
"confidence": str,
"days_analyzed": int,
"data_points": int
}
Intake: mean of daily total kcal (sum per calendar day).
TDEE: estimate_tdee_kcal_from_latest_weight (MSJ × PAL oder kg-Fallback).
"""
with get_db() as conn:
cur = get_cursor(conn)
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
# Get average intake
cur.execute(
"""SELECT AVG(kcal) as avg_kcal, COUNT(*) as cnt
"""SELECT date, SUM(kcal)::float AS daily_kcal
FROM nutrition_log
WHERE profile_id=%s AND date >= %s AND kcal IS NOT NULL""",
(profile_id, cutoff)
WHERE profile_id=%s AND date >= %s AND kcal IS NOT NULL
GROUP BY date
ORDER BY date""",
(profile_id, cutoff),
)
row = cur.fetchone()
if not row or row['cnt'] == 0:
return {
"energy_balance": 0.0,
"avg_intake": 0.0,
"estimated_tdee": 0.0,
"status": "unknown",
"confidence": "insufficient",
"days_analyzed": days,
"data_points": 0
}
avg_intake = safe_float(row['avg_kcal'])
data_points = row['cnt']
# Simple TDEE estimation (this should be improved with profile data)
# For now, use a rough estimate: 2500 kcal for average adult
estimated_tdee = 2500.0 # TODO: Calculate from profile (weight, height, age, activity)
energy_balance = avg_intake - estimated_tdee
# Determine status
if energy_balance < -200:
status = "deficit"
elif energy_balance > 200:
status = "surplus"
else:
status = "maintenance"
confidence = calculate_confidence(data_points, days, "general")
daily_rows = cur.fetchall()
if not daily_rows:
return {
"energy_balance": energy_balance,
"energy_balance": 0.0,
"avg_intake": 0.0,
"estimated_tdee": 0.0,
"status": "unknown",
"confidence": "insufficient",
"days_analyzed": days,
"data_points": 0,
}
daily_totals = [safe_float(r["daily_kcal"]) for r in daily_rows]
avg_intake = sum(daily_totals) / len(daily_totals)
data_points = len(daily_totals)
estimated_tdee = estimate_tdee_kcal_from_latest_weight(profile_id)
if estimated_tdee is None:
return {
"energy_balance": 0.0,
"avg_intake": avg_intake,
"estimated_tdee": estimated_tdee,
"status": status,
"confidence": confidence,
"estimated_tdee": 0.0,
"status": "unknown",
"confidence": "insufficient",
"days_analyzed": days,
"data_points": data_points
}
energy_balance = avg_intake - estimated_tdee
if energy_balance < -200:
status = "deficit"
elif energy_balance > 200:
status = "surplus"
else:
status = "maintenance"
confidence = calculate_confidence(data_points, days, "general")
return {
"energy_balance": energy_balance,
"avg_intake": avg_intake,
"estimated_tdee": estimated_tdee,
"status": status,
"confidence": confidence,
"days_analyzed": days,
"data_points": data_points
}
def get_protein_adequacy_data(
profile_id: str,
@ -291,7 +376,6 @@ def get_protein_adequacy_data(
"confidence": str
}
"""
# Get protein targets
targets = get_protein_targets_data(profile_id)
with get_db() as conn:
@ -299,60 +383,55 @@ def get_protein_adequacy_data(
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
cur.execute(
"""SELECT
AVG(protein_g) as avg_protein,
COUNT(*) as cnt,
SUM(CASE WHEN protein_g >= %s AND protein_g <= %s THEN 1 ELSE 0 END) as days_in_target
"""SELECT COALESCE(SUM(protein_g), 0)::float AS daily_protein
FROM nutrition_log
WHERE profile_id=%s AND date >= %s AND protein_g IS NOT NULL""",
(targets['protein_target_low'], targets['protein_target_high'], profile_id, cutoff)
WHERE profile_id=%s AND date >= %s
GROUP BY date""",
(profile_id, cutoff),
)
row = cur.fetchone()
if not row or row['cnt'] == 0:
return {
"adequacy_score": 0,
"avg_protein_g": 0.0,
"target_protein_low": targets['protein_target_low'],
"target_protein_high": targets['protein_target_high'],
"protein_g_per_kg": 0.0,
"days_in_target": 0,
"days_with_data": 0,
"confidence": "insufficient"
}
avg_protein = safe_float(row['avg_protein'])
days_with_data = row['cnt']
days_in_target = row['days_in_target']
protein_g_per_kg = avg_protein / targets['current_weight'] if targets['current_weight'] > 0 else 0.0
# Calculate adequacy score
# 100 = always in target range
# Scale based on percentage of days in target + average relative to target
target_pct = (days_in_target / days_with_data * 100) if days_with_data > 0 else 0
# Bonus/penalty for average protein level
target_mid = (targets['protein_target_low'] + targets['protein_target_high']) / 2
avg_vs_target = (avg_protein / target_mid) if target_mid > 0 else 0
# Weighted score: 70% target days, 30% average level
adequacy_score = int(target_pct * 0.7 + min(avg_vs_target * 100, 100) * 0.3)
adequacy_score = max(0, min(100, adequacy_score)) # Clamp to 0-100
confidence = calculate_confidence(days_with_data, days, "general")
rows = cur.fetchall()
if not rows or targets.get("confidence") == "insufficient" or targets["current_weight"] <= 0:
return {
"adequacy_score": adequacy_score,
"avg_protein_g": avg_protein,
"adequacy_score": 0,
"avg_protein_g": 0.0,
"target_protein_low": targets['protein_target_low'],
"target_protein_high": targets['protein_target_high'],
"protein_g_per_kg": protein_g_per_kg,
"days_in_target": days_in_target,
"days_with_data": days_with_data,
"confidence": confidence
"protein_g_per_kg": 0.0,
"days_in_target": 0,
"days_with_data": 0,
"confidence": "insufficient"
}
daily_totals = [safe_float(r["daily_protein"]) for r in rows]
days_with_data = len(daily_totals)
low = targets["protein_target_low"]
high = targets["protein_target_high"]
days_in_target = sum(1 for d in daily_totals if low <= d <= high)
avg_protein = sum(daily_totals) / days_with_data
protein_g_per_kg = avg_protein / targets["current_weight"] if targets["current_weight"] > 0 else 0.0
target_pct = (days_in_target / days_with_data * 100) if days_with_data > 0 else 0
target_mid = (low + high) / 2
avg_vs_target = (avg_protein / target_mid) if target_mid > 0 else 0
adequacy_score = int(target_pct * 0.7 + min(avg_vs_target * 100, 100) * 0.3)
adequacy_score = max(0, min(100, adequacy_score))
confidence = calculate_confidence(days_with_data, days, "general")
return {
"adequacy_score": adequacy_score,
"avg_protein_g": avg_protein,
"target_protein_low": targets['protein_target_low'],
"target_protein_high": targets['protein_target_high'],
"protein_g_per_kg": protein_g_per_kg,
"days_in_target": days_in_target,
"days_with_data": days_with_data,
"confidence": confidence
}
def get_macro_consistency_data(
profile_id: str,
@ -387,16 +466,18 @@ def get_macro_consistency_data(
cur.execute(
"""SELECT
protein_g, carbs_g, fat_g, kcal
COALESCE(SUM(kcal), 0)::float AS kcal,
COALESCE(SUM(protein_g), 0)::float AS protein_g,
COALESCE(SUM(carbs_g), 0)::float AS carbs_g,
COALESCE(SUM(fat_g), 0)::float AS fat_g
FROM nutrition_log
WHERE profile_id=%s
AND date >= %s
AND protein_g IS NOT NULL
AND carbs_g IS NOT NULL
AND fat_g IS NOT NULL
AND kcal > 0
ORDER BY date""",
(profile_id, cutoff)
WHERE profile_id=%s AND date >= %s
GROUP BY date
HAVING COALESCE(SUM(kcal), 0) > 0
AND COALESCE(SUM(protein_g), 0) > 0
AND COALESCE(SUM(carbs_g), 0) > 0
AND COALESCE(SUM(fat_g), 0) > 0""",
(profile_id, cutoff),
)
rows = cur.fetchall()
@ -413,7 +494,6 @@ def get_macro_consistency_data(
"data_points": len(rows)
}
# Calculate macro percentages for each day
import statistics
protein_pcts = []
@ -425,7 +505,6 @@ def get_macro_consistency_data(
if total_kcal == 0:
continue
# Convert grams to kcal (protein=4, carbs=4, fat=9)
protein_kcal = safe_float(row['protein_g']) * 4
carbs_kcal = safe_float(row['carbs_g']) * 4
fat_kcal = safe_float(row['fat_g']) * 9
@ -491,50 +570,15 @@ def get_macro_consistency_data(
def calculate_energy_balance_7d(profile_id: str) -> Optional[float]:
"""
Calculate 7-day average energy balance (kcal/day)
Positive = surplus, Negative = deficit
Migration from Phase 0b:
Used by placeholders that need single balance value
7-day mean energy balance (kcal/day), same rules as get_energy_balance_data(..., 7).
"""
with get_db() as conn:
cur = get_cursor(conn)
cur.execute("""
SELECT kcal
FROM nutrition_log
WHERE profile_id = %s
AND date >= CURRENT_DATE - INTERVAL '7 days'
ORDER BY date DESC
""", (profile_id,))
calories = [row['kcal'] for row in cur.fetchall()]
if len(calories) < 4: # Need at least 4 days
return None
avg_intake = float(sum(calories) / len(calories))
# Get estimated TDEE (simplified - could use Harris-Benedict)
# For now, use weight-based estimate
cur.execute("""
SELECT weight
FROM weight_log
WHERE profile_id = %s
ORDER BY date DESC
LIMIT 1
""", (profile_id,))
weight_row = cur.fetchone()
if not weight_row:
return None
# Simple TDEE estimate: bodyweight (kg) × 30-35
# TODO: Improve with activity level, age, gender
estimated_tdee = float(weight_row['weight']) * 32.5
balance = avg_intake - estimated_tdee
return round(balance, 0)
data = get_energy_balance_data(profile_id, 7)
if data["data_points"] < 4:
return None
tdee = data.get("estimated_tdee") or 0
if tdee <= 0:
return None
return round(float(data["energy_balance"]), 0)
def calculate_energy_deficit_surplus(profile_id: str, days: int = 7) -> Optional[str]:
@ -654,15 +698,14 @@ def calculate_protein_days_in_target(profile_id: str, target_low: float = 1.6, t
def calculate_protein_adequacy_28d(profile_id: str) -> Optional[int]:
"""
Protein adequacy score 0-100 (last 28 days)
Based on consistency and target achievement
Protein adequacy score 0-100 (last 28 days).
Uses per-calendar-day total protein vs. average weight in the window (g/kg per day).
"""
import statistics
with get_db() as conn:
cur = get_cursor(conn)
# Get average weight (28d)
cur.execute("""
SELECT AVG(weight) as avg_weight
FROM weight_log
@ -676,38 +719,29 @@ def calculate_protein_adequacy_28d(profile_id: str) -> Optional[int]:
weight = float(weight_row['avg_weight'])
# Get protein intake (28d)
cur.execute("""
SELECT protein_g
SELECT COALESCE(SUM(protein_g), 0)::float AS daily_protein
FROM nutrition_log
WHERE profile_id = %s
AND date >= CURRENT_DATE - INTERVAL '28 days'
AND protein_g IS NOT NULL
GROUP BY date
""", (profile_id,))
protein_values = [float(row['protein_g']) for row in cur.fetchall()]
daily_totals = [float(row['daily_protein']) for row in cur.fetchall()]
if len(protein_values) < 18: # 60% coverage
if len(daily_totals) < 18:
return None
# Calculate metrics
protein_per_kg_values = [p / weight for p in protein_values]
protein_per_kg_values = [p / weight for p in daily_totals]
avg_protein_per_kg = sum(protein_per_kg_values) / len(protein_per_kg_values)
# Target range: 1.6-2.2 g/kg for active individuals
target_mid = 1.9
# Score based on distance from target
if 1.6 <= avg_protein_per_kg <= 2.2:
base_score = 100
elif avg_protein_per_kg < 1.6:
# Below target
base_score = max(40, 100 - ((1.6 - avg_protein_per_kg) * 40))
else:
# Above target (less penalty)
base_score = max(80, 100 - ((avg_protein_per_kg - 2.2) * 10))
# Consistency bonus/penalty
std_dev = statistics.stdev(protein_per_kg_values)
if std_dev < 0.3:
consistency_bonus = 10
@ -723,20 +757,24 @@ def calculate_protein_adequacy_28d(profile_id: str) -> Optional[int]:
def calculate_macro_consistency_score(profile_id: str) -> Optional[int]:
"""
Macro consistency score 0-100 (last 28 days)
Lower variability = higher score
Macro consistency score 0-100 (last 28 days).
CV of daily totals (kcal and macros), not raw log rows.
"""
import statistics
with get_db() as conn:
cur = get_cursor(conn)
cur.execute("""
SELECT kcal, protein_g, fat_g, carbs_g
SELECT
COALESCE(SUM(kcal), 0)::float AS dk,
COALESCE(SUM(protein_g), 0)::float AS dp,
COALESCE(SUM(fat_g), 0)::float AS df,
COALESCE(SUM(carbs_g), 0)::float AS dc
FROM nutrition_log
WHERE profile_id = %s
AND date >= CURRENT_DATE - INTERVAL '28 days'
AND kcal IS NOT NULL
ORDER BY date DESC
GROUP BY date
HAVING COALESCE(SUM(kcal), 0) > 0
""", (profile_id,))
data = cur.fetchall()
@ -744,9 +782,7 @@ def calculate_macro_consistency_score(profile_id: str) -> Optional[int]:
if len(data) < 18:
return None
# Calculate coefficient of variation for each macro
def cv(values):
"""Coefficient of variation (std_dev / mean)"""
if not values or len(values) < 2:
return None
mean = sum(values) / len(values)
@ -755,10 +791,10 @@ def calculate_macro_consistency_score(profile_id: str) -> Optional[int]:
std_dev = statistics.stdev(values)
return std_dev / mean
calories_cv = cv([d['kcal'] for d in data])
protein_cv = cv([d['protein_g'] for d in data if d['protein_g']])
fat_cv = cv([d['fat_g'] for d in data if d['fat_g']])
carbs_cv = cv([d['carbs_g'] for d in data if d['carbs_g']])
calories_cv = cv([d['dk'] for d in data])
protein_cv = cv([d['dp'] for d in data if d['dp']])
fat_cv = cv([d['df'] for d in data if d['df']])
carbs_cv = cv([d['dc'] for d in data if d['dc']])
cv_values = [v for v in [calories_cv, protein_cv, fat_cv, carbs_cv] if v is not None]
@ -767,9 +803,6 @@ def calculate_macro_consistency_score(profile_id: str) -> Optional[int]:
avg_cv = sum(cv_values) / len(cv_values)
# Score: lower CV = higher score
# CV < 0.2 = excellent consistency
# CV > 0.5 = poor consistency
if avg_cv < 0.2:
score = 100
elif avg_cv < 0.3:
@ -853,40 +886,66 @@ def calculate_nutrition_score(profile_id: str, focus_weights: Optional[Dict] = N
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)
# Weighted average (float: DB-Werte können Decimal sein)
total_score = sum(float(score) * float(weight) for _, score, weight in components)
total_weight = sum(float(weight) for _, _, weight in components)
return int(total_score / total_weight)
def _score_calorie_adherence(profile_id: str) -> Optional[int]:
"""Score calorie target adherence (0-100)"""
# Check for energy balance goal
# For now, use energy balance calculation
"""Score calorie target adherence (0100) using 7d balance vs profiles.goal_mode."""
balance = calculate_energy_balance_7d(profile_id)
if balance is None:
return None
# Score based on whether deficit/surplus aligns with goal
# Simplified: assume weight loss goal = deficit is good
# TODO: Check actual goal type
mode = _get_profile_goal_mode(profile_id)
b = float(balance)
abs_balance = abs(balance)
def _weight_loss(x: float) -> int:
if -550 <= x <= -250:
return 100
if x > 450:
return 38
if -750 <= x < -550 or -250 < x <= 120:
return 82
if x < -1200:
return 52
if -950 <= x < -750 or 120 < x <= 350:
return 68
return 58
# Moderate deficit/surplus = good
if 200 <= abs_balance <= 500:
return 100
elif 100 <= abs_balance <= 700:
return 85
elif abs_balance <= 900:
return 70
elif abs_balance <= 1200:
return 55
else:
def _surplus_friendly(x: float) -> int:
if 80 <= x <= 480:
return 100
if -120 <= x < 80 or 480 < x <= 700:
return 86
if -380 <= x < -120:
return 68
if x > 850:
return 54
if x < -650:
return 44
return 72
def _maintenance(x: float) -> int:
a = abs(x)
if a <= 200:
return 100
if a <= 400:
return 84
if a <= 650:
return 70
if a <= 900:
return 55
return 40
if mode == "weight_loss":
return _weight_loss(b)
if mode in ("strength", "recomposition"):
return _surplus_friendly(b)
return _maintenance(b)
def _score_macro_balance(profile_id: str) -> Optional[int]:
"""Score macro balance (0-100)"""

View File

@ -15,12 +15,50 @@ Phase 0c: Multi-Layer Architecture
Version: 1.0
"""
from typing import Dict, List, Optional
import json
from typing import Dict, List, Optional, Any
from datetime import datetime, timedelta, date
from db import get_db, get_cursor, r2d
from db import get_db, get_cursor
from data_layer.utils import calculate_confidence, safe_float, safe_int
def _parse_sleep_segments(raw: Any) -> Optional[List[dict]]:
"""JSONB kann dict/list/str sein; ungültig → None."""
if raw is None:
return None
if isinstance(raw, str):
try:
raw = json.loads(raw)
except (json.JSONDecodeError, TypeError):
return None
if not isinstance(raw, list):
return None
return raw
def _segment_minutes(seg: Any) -> int:
if not isinstance(seg, dict):
return 0
for key in ("duration_min", "duration_minutes", "minutes"):
v = seg.get(key)
if v is not None:
return max(0, safe_int(v))
return 0
def _normalize_sleep_phase(seg: dict) -> str:
"""Kleinbuchstaben; Apple „Core“-Schlaf wird wie light gewertet."""
if not isinstance(seg, dict):
return ""
p = seg.get("phase")
if p is None:
return ""
s = str(p).strip().lower()
if s in ("core", "asleep"):
return "light"
return s
def get_sleep_duration_data(
profile_id: str,
days: int = 7
@ -51,7 +89,7 @@ def get_sleep_duration_data(
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
cur.execute(
"""SELECT sleep_segments FROM sleep_log
"""SELECT sleep_segments, duration_minutes FROM sleep_log
WHERE profile_id=%s AND date >= %s
ORDER BY date DESC""",
(profile_id, cutoff)
@ -72,12 +110,17 @@ def get_sleep_duration_data(
nights_with_data = 0
for row in rows:
segments = row['sleep_segments']
night_minutes = 0
segments = _parse_sleep_segments(row.get("sleep_segments"))
if segments:
night_minutes = sum(seg.get('duration_min', 0) for seg in segments)
if night_minutes > 0:
total_minutes += night_minutes
nights_with_data += 1
night_minutes = sum(_segment_minutes(seg) for seg in segments)
if night_minutes <= 0:
dm = row.get("duration_minutes")
if dm is not None:
night_minutes = max(0, safe_int(dm))
if night_minutes > 0:
total_minutes += night_minutes
nights_with_data += 1
if nights_with_data == 0:
return {
@ -136,7 +179,9 @@ def get_sleep_quality_data(
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
cur.execute(
"""SELECT sleep_segments FROM sleep_log
"""SELECT sleep_segments, duration_minutes, deep_minutes, rem_minutes,
light_minutes, awake_minutes
FROM sleep_log
WHERE profile_id=%s AND date >= %s
ORDER BY date DESC""",
(profile_id, cutoff)
@ -163,15 +208,29 @@ def get_sleep_quality_data(
count = 0
for row in rows:
segments = row['sleep_segments']
if segments:
# Note: segments use 'phase' key, stored lowercase (deep, rem, light, awake)
deep_rem_min = sum(s.get('duration_min', 0) for s in segments if s.get('phase') in ['deep', 'rem'])
light_min = sum(s.get('duration_min', 0) for s in segments if s.get('phase') == 'light')
awake_min = sum(s.get('duration_min', 0) for s in segments if s.get('phase') == 'awake')
total_min = sum(s.get('duration_min', 0) for s in segments)
deep_rem_min = light_min = awake_min = 0
total_min = 0
used_segments = False
segments = _parse_sleep_segments(row.get("sleep_segments"))
if segments:
total_min = sum(_segment_minutes(s) for s in segments)
if total_min > 0:
deep_rem_min = sum(
_segment_minutes(s)
for s in segments
if _normalize_sleep_phase(s) in ("deep", "rem")
)
light_min = sum(
_segment_minutes(s)
for s in segments
if _normalize_sleep_phase(s) == "light"
)
awake_min = sum(
_segment_minutes(s)
for s in segments
if _normalize_sleep_phase(s) == "awake"
)
quality_pct = (deep_rem_min / total_min) * 100
total_quality += quality_pct
total_deep_rem += deep_rem_min
@ -179,6 +238,28 @@ def get_sleep_quality_data(
total_awake += awake_min
total_all += total_min
count += 1
used_segments = True
if not used_segments:
d, r, l, a = (
row.get("deep_minutes"),
row.get("rem_minutes"),
row.get("light_minutes"),
row.get("awake_minutes"),
)
if d is not None or r is not None or l is not None:
di, ri, li = (d or 0), (r or 0), (l or 0)
phase_sum = di + ri + li
ai = (a or 0) if a is not None else 0
total_min = phase_sum + ai
if total_min > 0 and phase_sum > 0:
quality_pct = ((di + ri) / total_min) * 100
total_quality += quality_pct
total_deep_rem += di + ri
total_light += li
total_awake += ai
total_all += total_min
count += 1
if count == 0:
return {
@ -351,8 +432,8 @@ def calculate_recovery_score_v2(profile_id: str) -> Optional[int]:
return None
# Weighted average
total_score = sum(score * weight for _, score, weight in components)
total_weight = sum(weight for _, _, weight in components)
total_score = sum(float(score) * float(weight) for _, score, weight in components)
total_weight = sum(float(weight) for _, _, weight in components)
final_score = int(total_score / total_weight)
@ -783,17 +864,24 @@ def calculate_sleep_quality_7d(profile_id: str) -> Optional[int]:
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)
dur = s["duration_minutes"]
if not dur or dur <= 0:
continue
d = s["deep_minutes"]
r = s["rem_minutes"]
if d is None and r is None:
continue
di, ri = (d or 0), (r or 0)
quality_pct = ((di + ri) / dur) * 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 not quality_scores:
return None

View File

@ -202,23 +202,24 @@ def calculate_goal_progress_score(profile_id: str) -> Optional[int]:
total_weight = 0.0
for focus_area_id, weight in focus_weights.items():
w = float(weight)
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
total_score += float(body_score) * w
total_weight += w
elif component == 'nutrition' and nutrition_score is not None:
total_score += nutrition_score * weight
total_weight += weight
total_score += float(nutrition_score) * w
total_weight += w
elif component == 'activity' and activity_score is not None:
total_score += activity_score * weight
total_weight += weight
total_score += float(activity_score) * w
total_weight += w
elif component == 'recovery' and recovery_score is not None:
total_score += recovery_score * weight
total_weight += weight
total_score += float(recovery_score) * w
total_weight += w
elif component == 'health' and health_risk_score is not None:
total_score += health_risk_score * weight
total_weight += weight
total_score += float(health_risk_score) * w
total_weight += w
if total_weight == 0:
return None
@ -282,9 +283,9 @@ def calculate_health_stability_score(profile_id: str) -> Optional[int]:
activities = cur.fetchall()
if activities:
total_minutes = sum(a['duration_min'] for a in activities)
total_minutes = float(sum(float(a['duration_min'] or 0) for a in activities))
# WHO recommends 150-300 min/week moderate activity
movement_score = min(100, (total_minutes / 150) * 100)
movement_score = min(100.0, (total_minutes / 150) * 100)
components.append(('movement', movement_score, 20))
# 4. Waist circumference risk (15%)
@ -328,8 +329,8 @@ def calculate_health_stability_score(profile_id: str) -> Optional[int]:
return None
# Weighted average
total_score = sum(score * weight for _, score, weight in components)
total_weight = sum(weight for _, _, weight in components)
total_score = sum(float(score) * float(weight) for _, score, weight in components)
total_weight = sum(float(weight) for _, _, weight in components)
return int(total_score / total_weight)
@ -532,9 +533,11 @@ def calculate_focus_area_progress(profile_id: str, focus_area_id: str) -> Option
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)
# Weighted average by contribution_weight (Numeric → float)
total_progress = sum(
float(g['progress_pct']) * float(g['contribution_weight']) for g in goals
)
total_weight = sum(float(g['contribution_weight']) for g in goals)
return int(total_progress / total_weight) if total_weight > 0 else None

View File

@ -14,6 +14,10 @@ from slowapi.errors import RateLimitExceeded
from db import init_db
# Placeholder registry: load all register_placeholder() side-effects before any request
# so get_placeholder_catalog() and exports see consistent metadata (see Phase A plan).
import placeholder_registrations # noqa: F401
# Import routers
from routers import auth, profiles, weight, circumference, caliper
from routers import activity, nutrition, photos, insights, prompts

View File

@ -1,11 +1,10 @@
"""
Complete Placeholder Metadata Definitions
Complete Placeholder Metadata Definitions (Legacy / Normativ v1)
This module contains manually curated, complete metadata for all 116 placeholders.
It combines automatic extraction with manual annotation to ensure 100% normative compliance.
IMPORTANT: This is the authoritative source for placeholder metadata.
All new placeholders MUST be added here with complete metadata.
Hinweis (2026-04): **Verbindliche Metadaten-Pflege** erfolgt über
`backend/placeholder_registrations/` + `placeholder_registry.py` (114 Keys, deckungsgleich
mit `PLACEHOLDER_MAP`). Dieses Modul bleibt für ältere Generator-/Export-Pfade und
Tests; neue Platzhalter hier nicht mehr duplizieren.
"""
from placeholder_metadata import (
PlaceholderMetadata,
@ -28,7 +27,7 @@ from typing import List
def get_all_placeholder_metadata() -> List[PlaceholderMetadata]:
"""
Returns complete metadata for all 116 placeholders.
Returns complete metadata for all 114 placeholders (Registry ist maßgeblich).
This is the authoritative, manually curated source.
"""
@ -476,7 +475,7 @@ def get_all_placeholder_metadata() -> List[PlaceholderMetadata]:
notes=["Quadrant-Logik basiert auf FM/LBM Delta-Vorzeichen"],
),
# NOTE: Continuing with all 116 placeholders would make this file very long.
# NOTE: Continuing with all 114 placeholders would make this file very long.
# For brevity, I'll create a separate generator that fills all remaining placeholders.
# The pattern is established above - each placeholder gets full metadata.
]

View File

@ -29,14 +29,22 @@ def extract_value_raw(value_display: str, output_type: OutputType, placeholder_t
Returns: (raw_value, success)
"""
if not value_display or value_display in ['nicht verfügbar', 'nicht genug Daten']:
s = (value_display or "").strip()
if (
not s
or s in ['nicht verfügbar', 'nicht genug Daten']
or s.startswith('nicht verfügbar —')
):
# V2 strict mode: missing/unavailable value is not a successful extraction
return None, False
# JSON output type
if output_type == OutputType.JSON:
try:
return json.loads(value_display), True
parsed = json.loads(value_display)
if isinstance(parsed, dict) and parsed.get('_available') is False:
return None, False
return parsed, True
except (json.JSONDecodeError, TypeError):
# Try to find JSON in string
json_match = re.search(r'(\{.*\}|\[.*\])', value_display, re.DOTALL)

View File

@ -8,7 +8,31 @@ Auto-imports all placeholder registrations to populate the global registry.
from . import nutrition_part_a
from . import nutrition_part_b
from . import nutrition_part_c
from . import nutrition_score
from . import body_metrics
from . import body_extras
from . import activity_metrics
from . import activity_session_insights
from . import schlaf_erholung
from . import vitalwerte
from . import profil_zeitraum
from . import phase_0b_meta_scores
from . import phase_0b_ziele_fokus
from . import korrelationen
__all__ = ['nutrition_part_a', 'nutrition_part_b', 'nutrition_part_c', 'body_metrics', 'activity_metrics']
__all__ = [
'nutrition_part_a',
'nutrition_part_b',
'nutrition_part_c',
'nutrition_score',
'body_metrics',
'body_extras',
'activity_metrics',
'activity_session_insights',
'schlaf_erholung',
'vitalwerte',
'profil_zeitraum',
'phase_0b_meta_scores',
'phase_0b_ziele_fokus',
'korrelationen',
]

View File

@ -0,0 +1,19 @@
"""Gemeinsames Evidence-Tagging für Registry-Einträge."""
from placeholder_registry import EvidenceType, PlaceholderMetadata
STANDARD_FIELDS = (
"key", "category", "description", "resolver_module", "resolver_function",
"data_layer_module", "data_layer_function", "source_tables", "semantic_contract",
"unit", "time_window", "output_type", "placeholder_type", "format_hint",
"example_output", "minimum_data_requirements", "confidence_logic",
"missing_value_policy", "layer_1_decision", "layer_2a_decision",
"layer_2b_reuse_possible", "architecture_alignment", "issue_53_alignment",
)
def tag_standard_evidence(meta: PlaceholderMetadata) -> None:
for field in STANDARD_FIELDS:
meta.set_evidence(field, EvidenceType.CODE_DERIVED)
meta.set_evidence("business_meaning", EvidenceType.DRAFT_DERIVED)
meta.set_evidence("known_limitations", EvidenceType.MIXED)

View File

@ -1,7 +1,7 @@
"""
Activity Metrics Placeholder Registrations
Registers all 17 activity-related placeholders in the central placeholder registry.
Registers 17 Aktivitäts-Platzhalter hier; 3 Session-/Erholungs-Keys in activity_session_insights.py (20 gesamt).
Evidence-based metadata with clear tagging of source.
@ -10,6 +10,9 @@ Groups:
- Basic Metrics (7): training_minutes_week, training_frequency_7d, quality_sessions_pct,
proxy_internal_load_7d, monotony_score, strain_score, rest_day_compliance
- Advanced Metrics (7): ability_balance_*, vo2max_trend_28d, activity_score
Resolver: alle Keys gebündelt unter Training / Aktivität in PLACEHOLDER_MAP;
activity_score nicht unter Meta Scores.
"""
from placeholder_registry import (
@ -938,9 +941,9 @@ def register_activity_group_3():
description="VO2 Max Trend über 28 Tage",
category="Aktivität",
resolver_module="backend/placeholder_resolver.py",
resolver_function="get_vo2max_trend_28d",
resolver_function="_safe_float",
data_layer_module="backend/data_layer/activity_metrics.py",
data_layer_function="calculate_vo2max_trend",
data_layer_function="calculate_vo2max_trend_28d",
source_tables=["vitals_baseline"],
time_window="28d",
output_type=OutputType.NUMERIC,
@ -977,8 +980,8 @@ def register_activity_group_3():
"EDGE CASE: Nur 1 Messung → kein Trend → missing_value. "
"EDGE CASE: Große Zeitlücken zwischen Messungen → Trend nicht aussagekräftig."
),
layer_1_decision="Data Layer (activity_metrics.calculate_vo2max_trend) - QUESTIONABLE",
layer_2a_decision="Placeholder Resolver (formatting only)",
layer_1_decision="Data Layer (activity_metrics.calculate_vo2max_trend_28d) — Kategorie diskutierbar",
layer_2a_decision="Placeholder Resolver (_safe_float)",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c Multi-Layer Architecture conform",
issue_53_alignment="Layer separation established"
@ -1020,8 +1023,8 @@ def register_activity_group_3():
description="Gesamtaktivitäts-Score (gewichtet)",
category="Aktivität",
resolver_module="backend/placeholder_resolver.py",
resolver_function="get_activity_score",
data_layer_module="backend/data_layer/scores.py",
resolver_function="_safe_int",
data_layer_module="backend/data_layer/activity_metrics.py",
data_layer_function="calculate_activity_score",
source_tables=["activity_log", "training_types", "rest_days", "vitals_baseline", "user_focus_area_weights"],
time_window="composite (7d, 14d, 28d mixed)",
@ -1065,8 +1068,8 @@ def register_activity_group_3():
"QUESTIONABLE: Vermischt Metriken mit unterschiedlicher Verlässlichkeit "
"(z.B. quality_sessions_pct hat TO_VERIFY Issues)."
),
layer_1_decision="Data Layer (scores.calculate_activity_score)",
layer_2a_decision="Placeholder Resolver (formatting only)",
layer_1_decision="Data Layer (activity_metrics.calculate_activity_score)",
layer_2a_decision="Placeholder Resolver (_safe_int)",
layer_2b_reuse_possible=False,
architecture_alignment="Phase 0c Multi-Layer Architecture conform",
issue_53_alignment="Layer separation established"

View File

@ -0,0 +1,184 @@
"""
Registry: Trainings-Häufigkeit, Pausen zwischen Einheiten, wöchentliche Session-JSON (KI-Rohkontext).
"""
from placeholder_registry import (
PlaceholderMetadata,
MissingValuePolicy,
EvidenceType,
OutputType,
PlaceholderType,
register_placeholder,
)
def _ev(meta: PlaceholderMetadata, field: str, et: EvidenceType = EvidenceType.CODE_DERIVED):
meta.set_evidence(field, et)
def register_activity_session_insights():
md_freq = PlaceholderMetadata(
key="training_frequency_by_type_md",
category="Aktivität",
description=(
"Markdown-Tabelle: pro Trainingsart (activity_type) Sessions, Ø/Woche, "
"Dauer, kcal, HF, RPE, kcal/min (Intensitätsproxy)"
),
resolver_module="backend/placeholder_resolver.py",
resolver_function="get_training_frequency_by_type_md",
data_layer_module="backend/data_layer/activity_metrics.py",
data_layer_function="get_training_frequency_by_type_data",
source_tables=["activity_log"],
semantic_contract=(
"Aggregat über activity_log gruppiert nach activity_type (Roh-Label). "
"sessions_per_week = count / (days/7). avg_kcal_per_min = Summe kcal / Summe min."
),
business_meaning="KI: Häufigkeit & Belastung pro Sportart, Erholungs-/Überlastungs-Kontext",
unit="Markdown",
time_window="default 28 Tage",
output_type=OutputType.TEXT_SUMMARY,
placeholder_type=PlaceholderType.INTERPRETED,
format_hint="GitHub-Flavored Markdown-Tabelle",
example_output="| Art | n | Ø/Woche | … |",
minimum_data_requirements="Mindestens eine Session im Fenster",
quality_filter_policy=None,
confidence_logic="Wie calculate_confidence anhand Session-Anzahl",
missing_value_policy=MissingValuePolicy(
available=False,
value_raw=None,
missing_reason="no_data",
legacy_display="Keine Trainingsdaten",
),
known_limitations=(
"Gruppierung nach activity_type-String (Import-Namen), nicht nur training_type_id. "
"HF/RPE oft NULL je nach Quelle. Pausen-Analyse separater Platzhalter."
),
layer_1_decision="activity_metrics.get_training_frequency_by_type_data",
layer_2a_decision="get_training_frequency_by_type_md",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c",
issue_53_alignment="Layer 1",
evidence={},
)
for f in (
"key", "category", "description", "resolver_module", "resolver_function",
"data_layer_module", "data_layer_function", "source_tables", "semantic_contract",
"unit", "time_window", "output_type", "placeholder_type", "format_hint",
"example_output", "minimum_data_requirements", "confidence_logic",
"missing_value_policy", "layer_1_decision", "layer_2a_decision",
"layer_2b_reuse_possible", "architecture_alignment", "issue_53_alignment",
):
_ev(md_freq, f)
_ev(md_freq, "business_meaning", EvidenceType.DRAFT_DERIVED)
_ev(md_freq, "known_limitations", EvidenceType.MIXED)
register_placeholder(md_freq)
md_gap = PlaceholderMetadata(
key="training_inter_session_gap_md",
category="Aktivität",
description="Median/Mittel/Min der Stunden zwischen aufeinanderfolgenden Trainingseinheiten",
resolver_module="backend/placeholder_resolver.py",
resolver_function="get_training_inter_session_gap_md",
data_layer_module="backend/data_layer/activity_metrics.py",
data_layer_function="get_training_inter_session_gap_data",
source_tables=["activity_log"],
semantic_contract=(
"Sessions chronologisch; Zeitstempel = date + start_time oder 12:00. "
"Lücken in Stunden zwischen aufeinanderfolgenden Starts."
),
business_meaning="KI: ausreichend Erholung zwischen Belastungen? Doppelbelastung?",
unit="Markdown",
time_window="default 28 Tage",
output_type=OutputType.TEXT_SUMMARY,
placeholder_type=PlaceholderType.INTERPRETED,
format_hint="Kurzer Markdown-Fließtext",
example_output="**Pause zwischen Trainings** …",
minimum_data_requirements="Mindestens 2 Sessions",
quality_filter_policy=None,
confidence_logic="calculate_confidence über Session-Anzahl",
missing_value_policy=MissingValuePolicy(
available=False,
value_raw=None,
missing_reason="insufficient_data",
legacy_display="Zu wenige Trainings",
),
known_limitations=(
"Kein Unterscheidung aktiv/passiv außerhalb activity_log. "
"Fehlende Uhrzeit verzerrt Reihenfolge am selben Tag nicht (nur ein künstlicher Mittag)."
),
layer_1_decision="activity_metrics.get_training_inter_session_gap_data",
layer_2a_decision="get_training_inter_session_gap_md",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c",
issue_53_alignment="Layer 1",
evidence={},
)
for f in (
"key", "category", "description", "resolver_module", "resolver_function",
"data_layer_module", "data_layer_function", "source_tables", "semantic_contract",
"unit", "time_window", "output_type", "placeholder_type", "format_hint",
"example_output", "minimum_data_requirements", "confidence_logic",
"missing_value_policy", "layer_1_decision", "layer_2a_decision",
"layer_2b_reuse_possible", "architecture_alignment", "issue_53_alignment",
):
_ev(md_gap, f)
_ev(md_gap, "business_meaning", EvidenceType.DRAFT_DERIVED)
_ev(md_gap, "known_limitations", EvidenceType.MIXED)
register_placeholder(md_gap)
pj = PlaceholderMetadata(
key="training_sessions_recent_json",
category="Aktivität",
description=(
"JSON: letzte ISO-Kalenderwochen mit Einheiten (Datum, Art, Dauer, kcal, HF Ø/max, RPE, Kategorie)"
),
resolver_module="backend/placeholder_resolver.py",
resolver_function="_safe_json",
data_layer_module="backend/data_layer/activity_metrics.py",
data_layer_function="get_training_sessions_recent_weeks_data",
source_tables=["activity_log", "training_types"],
semantic_contract=(
"Struktur weeks[].week_iso, sessions[] mit Feldern für KI-Auswertung. "
"Default 4 ISO-Wochen zurück."
),
business_meaning="Rohkontext für wochenweise Auswertung (Erholung, Intensität) in der KI",
unit="JSON string",
time_window="4 ISO-Wochen (28 Tage Datenfenster)",
output_type=OutputType.JSON,
placeholder_type=PlaceholderType.RAW_DATA,
format_hint="JSON-Objekt als String",
example_output='{"weeks":[...],"meta":{...}}',
minimum_data_requirements="Optional Sessions; meta.confidence bei leer insufficient",
quality_filter_policy=None,
confidence_logic="meta.confidence aus Session-Anzahl",
missing_value_policy=MissingValuePolicy(
available=False,
value_raw=None,
missing_reason="no_data",
legacy_display="{}",
),
known_limitations=(
"Token-Länge bei vielen Sessions beachten. training_type_name nur bei gesetztem training_type_id."
),
layer_1_decision="activity_metrics.get_training_sessions_recent_weeks_data",
layer_2a_decision="_safe_json('training_sessions_recent_json')",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c",
issue_53_alignment="Layer 1",
evidence={},
)
for f in (
"key", "category", "description", "resolver_module", "resolver_function",
"data_layer_module", "data_layer_function", "source_tables", "semantic_contract",
"unit", "time_window", "output_type", "placeholder_type", "format_hint",
"example_output", "minimum_data_requirements", "confidence_logic",
"missing_value_policy", "layer_1_decision", "layer_2a_decision",
"layer_2b_reuse_possible", "architecture_alignment", "issue_53_alignment",
):
_ev(pj, f)
_ev(pj, "business_meaning", EvidenceType.DRAFT_DERIVED)
_ev(pj, "known_limitations", EvidenceType.MIXED)
register_placeholder(pj)
register_activity_session_insights()

View File

@ -0,0 +1,237 @@
"""
Registry: BMI, Profil-Ziele (goal_weight, goal_bf_pct), body_progress_score.
Profilfelder sind unabhängig von der goals-Tabelle; operative Ziele über andere Keys.
"""
from placeholder_registry import (
PlaceholderMetadata,
MissingValuePolicy,
EvidenceType,
OutputType,
PlaceholderType,
register_placeholder,
)
def register_body_extras():
bmi = PlaceholderMetadata(
key="bmi",
category="Körper",
description="Body-Mass-Index aus letztem Gewicht und Profilgröße (cm)",
resolver_module="backend/placeholder_resolver.py",
resolver_function="calculate_bmi",
data_layer_module="backend/data_layer/body_metrics.py",
data_layer_function="get_bmi_data",
source_tables=["profiles", "weight_log"],
semantic_contract=(
"BMI = Gewicht_kg / (Größe_m)² mit Größe_m = profiles.height / 100 "
"und Gewicht = jüngster Eintrag in weight_log."
),
business_meaning="Standard-Körpermaß für Coaching und Risiko-Kontext",
unit="kg/m²",
time_window="latest weight + aktuelle Profilgröße",
output_type=OutputType.NUMERIC,
placeholder_type=PlaceholderType.RAW_DATA,
format_hint="Eine Dezimalstelle, ohne Einheit im String",
example_output="24.3",
minimum_data_requirements="Profil mit height > 0 und mindestens ein weight_log",
quality_filter_policy=None,
confidence_logic="high nur wenn BMI berechenbar; sonst insufficient / Anzeige nicht verfügbar",
missing_value_policy=MissingValuePolicy(
available=False,
value_raw=None,
missing_reason="no_data",
legacy_display="nicht verfügbar",
),
known_limitations=(
"Keine ethnischen Referenzkurven; Profilgröße kann veraltet sein. "
"Unterscheidet nicht Muskelmasse vs. Fett."
),
layer_1_decision="body_metrics.get_bmi_data",
layer_2a_decision="placeholder_resolver.calculate_bmi (Format)",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c",
issue_53_alignment="Layer 1 als Quelle",
evidence={},
)
for field in (
"key", "category", "description", "resolver_module", "resolver_function",
"data_layer_module", "data_layer_function", "source_tables",
"semantic_contract", "business_meaning", "unit", "time_window",
"output_type", "placeholder_type", "format_hint", "example_output",
"minimum_data_requirements", "confidence_logic", "missing_value_policy",
"known_limitations", "layer_1_decision", "layer_2a_decision",
"layer_2b_reuse_possible", "architecture_alignment", "issue_53_alignment",
):
bmi.set_evidence(field, EvidenceType.CODE_DERIVED)
bmi.set_evidence("business_meaning", EvidenceType.DRAFT_DERIVED)
bmi.set_evidence("known_limitations", EvidenceType.MIXED)
register_placeholder(bmi)
gw = PlaceholderMetadata(
key="goal_weight",
category="Körper",
description="Zielgewicht aus Profilfeld profiles.goal_weight (kg)",
resolver_module="backend/placeholder_resolver.py",
resolver_function="get_goal_weight",
data_layer_module="backend/data_layer/body_metrics.py",
data_layer_function="get_profile_goal_weight_data",
source_tables=["profiles"],
semantic_contract=(
"Strategisches Soll-Gewicht im Profil; unabhängig von der goals-Tabelle "
"(dort detaillierte Ziele mit Fortschritt)."
),
business_meaning="Schneller Abgleich Prompt vs. Profil-Default-Zielgewicht",
unit="kg",
time_window="Profil-Snapshot",
output_type=OutputType.NUMERIC,
placeholder_type=PlaceholderType.RAW_DATA,
format_hint="Eine Dezimalstelle oder Text „nicht gesetzt“",
example_output="82.0",
minimum_data_requirements="profiles.goal_weight IS NOT NULL",
quality_filter_policy=None,
confidence_logic="high wenn gesetzt",
missing_value_policy=MissingValuePolicy(
available=False,
value_raw=None,
missing_reason="not_set",
legacy_display="nicht gesetzt",
),
known_limitations="Kann von aktiven goals.weight-Zielen abweichen.",
layer_1_decision="body_metrics.get_profile_goal_weight_data",
layer_2a_decision="placeholder_resolver.get_goal_weight",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c",
issue_53_alignment="Layer 1 als Quelle",
evidence={},
)
for field in (
"key", "category", "description", "resolver_module", "resolver_function",
"data_layer_module", "data_layer_function", "source_tables",
"semantic_contract", "unit", "time_window", "output_type",
"placeholder_type", "format_hint", "example_output",
"minimum_data_requirements", "confidence_logic", "missing_value_policy",
"layer_1_decision", "layer_2a_decision", "layer_2b_reuse_possible",
"architecture_alignment", "issue_53_alignment",
):
gw.set_evidence(field, EvidenceType.CODE_DERIVED)
gw.set_evidence("business_meaning", EvidenceType.DRAFT_DERIVED)
gw.set_evidence("known_limitations", EvidenceType.MIXED)
register_placeholder(gw)
gbf = PlaceholderMetadata(
key="goal_bf_pct",
category="Körper",
description="Ziel-Körperfettanteil aus Profilfeld profiles.goal_bf_pct (%)",
resolver_module="backend/placeholder_resolver.py",
resolver_function="get_goal_bf_pct",
data_layer_module="backend/data_layer/body_metrics.py",
data_layer_function="get_profile_goal_bf_pct_data",
source_tables=["profiles"],
semantic_contract="Strategisches Ziel-KFA im Profil.",
business_meaning="Prompt-Abgleich mit Profil-Ziel-KFA",
unit="%",
time_window="Profil-Snapshot",
output_type=OutputType.NUMERIC,
placeholder_type=PlaceholderType.RAW_DATA,
format_hint="Eine Dezimalstelle oder Text „nicht gesetzt“",
example_output="15.0",
minimum_data_requirements="profiles.goal_bf_pct IS NOT NULL",
quality_filter_policy=None,
confidence_logic="high wenn gesetzt",
missing_value_policy=MissingValuePolicy(
available=False,
value_raw=None,
missing_reason="not_set",
legacy_display="nicht gesetzt",
),
known_limitations="Kann von goals body_fat abweichen.",
layer_1_decision="body_metrics.get_profile_goal_bf_pct_data",
layer_2a_decision="placeholder_resolver.get_goal_bf_pct",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c",
issue_53_alignment="Layer 1 als Quelle",
evidence={},
)
for field in (
"key", "category", "description", "resolver_module", "resolver_function",
"data_layer_module", "data_layer_function", "source_tables",
"semantic_contract", "unit", "time_window", "output_type",
"placeholder_type", "format_hint", "example_output",
"minimum_data_requirements", "confidence_logic", "missing_value_policy",
"layer_1_decision", "layer_2a_decision", "layer_2b_reuse_possible",
"architecture_alignment", "issue_53_alignment",
):
gbf.set_evidence(field, EvidenceType.CODE_DERIVED)
gbf.set_evidence("business_meaning", EvidenceType.DRAFT_DERIVED)
gbf.set_evidence("known_limitations", EvidenceType.MIXED)
register_placeholder(gbf)
bps = PlaceholderMetadata(
key="body_progress_score",
category="Körper",
description="Körper-Fortschritts-Score 0100, gewichtet nach Focus (Abnehmen, Muskelaufbau, Recomp)",
resolver_module="backend/placeholder_resolver.py",
resolver_function="_safe_int",
data_layer_module="backend/data_layer/body_metrics.py",
data_layer_function="calculate_body_progress_score",
source_tables=[
"user_focus_area_weights",
"focus_area_definitions",
"goals",
"weight_log",
"caliper_log",
"circumference_log",
],
semantic_contract=(
"Gewichteter Mittelwert aus bis zu drei Komponenten: Trend vs. Gewichtsziel, "
"Körperzusammensetzung (FM/LBM/Recomp-Quadrant), Taille-Trend. "
"Komponenten nur aktiv, wenn passende Focus-Gewichte > 0."
),
business_meaning="Meta-KPI: passt dokumentierter Körperfortschritt zur gewichteten Körper-Priorität?",
unit="Score (0100)",
time_window="composite (u. a. 28d Deltas, Ziel-Fortschritt)",
output_type=OutputType.NUMERIC,
placeholder_type=PlaceholderType.SCORE,
format_hint="Ganzzahl oder „nicht verfügbar“",
example_output="72",
minimum_data_requirements=(
"Summe der Körper-Focus-Gewichte (weight_loss + muscle_gain + body_recomposition) > 0 "
"und mindestens eine bewertbare Komponente mit Daten."
),
quality_filter_policy=None,
confidence_logic="Kein separates Confidence-Feld; None wenn keine Körper-Gewichtung oder keine Teilscores.",
missing_value_policy=MissingValuePolicy(
available=False,
value_raw=None,
missing_reason="not_applicable",
legacy_display="nicht verfügbar",
),
known_limitations=(
"Abhängig von user_focus_area_weights und aktiven weight-goals für Gewichts-Teilscore. "
"Taille-Score wird mit festem Basisgewicht 20+ eingemischt und kann dominieren."
),
layer_1_decision="body_metrics.calculate_body_progress_score",
layer_2a_decision="placeholder_resolver._safe_int('body_progress_score', …)",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c",
issue_53_alignment="Layer 1 als Quelle",
evidence={},
)
for field in (
"key", "category", "description", "resolver_module", "resolver_function",
"data_layer_module", "data_layer_function", "source_tables",
"semantic_contract", "unit", "time_window", "output_type",
"placeholder_type", "format_hint", "example_output",
"minimum_data_requirements", "confidence_logic", "missing_value_policy",
"layer_1_decision", "layer_2a_decision", "layer_2b_reuse_possible",
"architecture_alignment", "issue_53_alignment",
):
bps.set_evidence(field, EvidenceType.CODE_DERIVED)
bps.set_evidence("business_meaning", EvidenceType.DRAFT_DERIVED)
bps.set_evidence("known_limitations", EvidenceType.MIXED)
register_placeholder(bps)
register_body_extras()

View File

@ -1,7 +1,8 @@
"""
Body Metrics Placeholder Registrations
Registers 17 body composition and measurement placeholders:
Registers 17 Körper-Metriken in diesem Modul; insgesamt 21 Körper-Keys in der Registry
(zusätzlich body_extras.py: bmi, goal_weight, goal_bf_pct, body_progress_score).
Weight & Trends (7):
- weight_aktuell
@ -29,7 +30,7 @@ Summaries (2):
- circ_summary
Evidence-based metadata with comprehensive formula documentation.
Code inspection: backend/data_layer/body_metrics.py (830 lines)
Siehe backend/data_layer/body_metrics.py als Layer-1-Implementierung.
"""
from placeholder_registry import (

View File

@ -0,0 +1,96 @@
"""Registry: Korrelations- und Treiber-Metriken (Data Layer correlations)."""
from placeholder_registry import (
PlaceholderMetadata,
MissingValuePolicy,
OutputType,
PlaceholderType,
register_placeholder,
)
from ._evidence import tag_standard_evidence
CAT = "Korrelationen"
MVP = lambda reason, disp: MissingValuePolicy(
available=False, value_raw=None, missing_reason=reason, legacy_display=disp
)
def register_korrelationen():
for key, dl_fn, desc, tables, sem in [
(
"correlation_energy_weight_lag",
"calculate_lag_correlation",
"JSON: Lag-Korrelation Energiebilanz ↔ Gewicht",
["nutrition_log", "weight_log"],
"correlations.calculate_lag_correlation(pid, 'energy', 'weight')",
),
(
"correlation_protein_lbm",
"calculate_lag_correlation",
"JSON: Lag-Korrelation Protein ↔ Magermasse",
["nutrition_log", "weight_log", "caliper_log"],
"correlations.calculate_lag_correlation(pid, 'protein', 'lbm')",
),
(
"correlation_load_hrv",
"calculate_lag_correlation",
"JSON: Lag-Korrelation Trainingslast ↔ HRV",
["activity_log", "vitals_baseline"],
"correlations.calculate_lag_correlation(pid, 'training_load', 'hrv')",
),
(
"correlation_load_rhr",
"calculate_lag_correlation",
"JSON: Lag-Korrelation Trainingslast ↔ Ruhepuls",
["activity_log", "vitals_baseline"],
"correlations.calculate_lag_correlation(pid, 'training_load', 'rhr')",
),
(
"plateau_detected",
"calculate_plateau_detected",
"JSON: Platten-Erkennung (Gewicht/Körper)",
["weight_log", "caliper_log"],
"correlations.calculate_plateau_detected",
),
(
"top_drivers",
"calculate_top_drivers",
"JSON: Top Treiber für Ziel-/Score-Variablen",
["weight_log", "nutrition_log", "activity_log", "vitals_baseline", "sleep_log"],
"correlations.calculate_top_drivers",
),
]:
m = PlaceholderMetadata(
key=key,
category=CAT,
description=desc,
resolver_module="backend/placeholder_resolver.py",
resolver_function="_safe_json",
data_layer_module="backend/data_layer/correlations.py",
data_layer_function=dl_fn,
source_tables=tables,
semantic_contract=sem,
business_meaning="Strukturierte Korrelationsausgabe für KI",
unit="JSON",
time_window="funktionsintern",
output_type=OutputType.JSON,
placeholder_type=PlaceholderType.RAW_DATA,
format_hint="JSON-String",
example_output="{}",
minimum_data_requirements="Ausreichend gekoppelte Zeitreihen",
quality_filter_policy=None,
confidence_logic="Wie correlations.*",
missing_value_policy=MVP("insufficient_data", "{}"),
known_limitations="Bei wenigen Daten leer oder wenig robust",
layer_1_decision=f"correlations.{dl_fn}",
layer_2a_decision="_safe_json",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c",
issue_53_alignment="Layer 1",
evidence={},
)
tag_standard_evidence(m)
register_placeholder(m)
register_korrelationen()

View File

@ -53,6 +53,13 @@ def register_nutrition_part_a():
"layer_1_decision": "Data Layer (nutrition_metrics.get_nutrition_average_data)",
"layer_2a_decision": "Placeholder Resolver (formatting only)",
"architecture_alignment": "Phase 0c Multi-Layer Architecture conform",
"minimum_data_requirements": (
"Mind. ein Kalendertag mit nutrition_log im Fenster; Mittelwerte aus täglicher Aggregation. "
"Confidence über calculate_confidence(day_count, days) in get_nutrition_average_data."
),
"quality_filter_policy": (
"Kein Outlier-Filter auf Tagesaggregaten; leere Tage fehlen in der Aggregation (kein Imputing)."
),
}
# Common evidence for shared fields
@ -73,8 +80,8 @@ def register_nutrition_part_a():
"layer_2b_reuse_possible": EvidenceType.TO_VERIFY, # not verified in charts
"architecture_alignment": EvidenceType.CODE_DERIVED, # imports from data_layer
"issue_53_alignment": EvidenceType.MIXED, # layer separation visible, issue conformity derived
"minimum_data_requirements": EvidenceType.UNRESOLVED, # not explicit in code
"quality_filter_policy": EvidenceType.UNRESOLVED, # not implemented
"minimum_data_requirements": EvidenceType.CODE_DERIVED,
"quality_filter_policy": EvidenceType.CODE_DERIVED,
}
# ── kcal_avg ──────────────────────────────────────────────────────────────
@ -94,8 +101,6 @@ def register_nutrition_part_a():
known_limitations="nur Intake, kein Bedarf; sagt allein nichts über Zielpassung",
layer_2b_reuse_possible=None, # to_verify - not checked in chart code
issue_53_alignment="Layer separation established",
minimum_data_requirements=None, # unresolved
quality_filter_policy=None, # unresolved
**common_metadata
)
@ -131,8 +136,6 @@ def register_nutrition_part_a():
),
layer_2b_reuse_possible=None,
issue_53_alignment="Layer separation established",
minimum_data_requirements=None,
quality_filter_policy=None,
**common_metadata
)
@ -165,8 +168,6 @@ def register_nutrition_part_a():
),
layer_2b_reuse_possible=None,
issue_53_alignment="Layer separation established",
minimum_data_requirements=None,
quality_filter_policy=None,
**common_metadata
)
@ -196,8 +197,6 @@ def register_nutrition_part_a():
known_limitations="meist im Gesamtkontext der Makroverteilung relevant",
layer_2b_reuse_possible=None,
issue_53_alignment="Layer separation established",
minimum_data_requirements=None,
quality_filter_policy=None,
**common_metadata
)

View File

@ -1,7 +1,7 @@
"""
Placeholder Registrations - Nutrition Part C
Registers 5 nutrition-related placeholders with complete metadata:
Registers 5 nutrition-related placeholders in this file (nutrition_score: siehe nutrition_score.py):
- macro_consistency_score
- energy_balance_7d
- energy_deficit_surplus
@ -113,7 +113,7 @@ energy_balance_metadata = PlaceholderMetadata(
resolver_function="_safe_float('energy_balance_7d', pid, decimals=0)",
data_layer_module="backend/data_layer/nutrition_metrics.py",
data_layer_function="calculate_energy_balance_7d",
source_tables=["nutrition_log", "weight_log"],
source_tables=["nutrition_log", "weight_log", "profiles"],
# Semantic
semantic_contract="Liefert die geschätzte Energiebilanz über 7 Tage als Differenz zwischen durchschnittlicher Energieaufnahme und geschätztem TDEE (Total Daily Energy Expenditure). Positiver Wert = Überschuss, Negativer Wert = Defizit.",
@ -127,11 +127,14 @@ energy_balance_metadata = PlaceholderMetadata(
# Quality
minimum_data_requirements="Mindestens 4 Tage mit Kalorienerfassung in 7-Tage-Fenster. Aktuelles Gewicht aus weight_log erforderlich.",
quality_filter_policy="Unvollständige Intake-Daten und fehlende Gewichtsmessung reduzieren Verlässlichkeit. TDEE-Schätzung ist vereinfacht (weight_kg × 32.5).",
quality_filter_policy=(
"Unvollständige Intake-Daten und fehlende Gewichtsmessung reduzieren Verlässlichkeit. "
"TDEE: MifflinSt Jeor × PAL 1.55 wenn Höhe, Geschlecht, DOB und Gewicht vorhanden, sonst kg×32.5."
),
confidence_logic=(
"Kombiniert Intake-Abdeckung und Robustheit des Verbrauchsmodells. "
"Niedrigere Confidence bei <7 Tagen Daten oder fehlendem Gewicht. "
"TDEE-Modell ist vereinfacht → inherent uncertainty."
"PAL=1.55 ist ein Festwert (moderate Aktivität), kein individuelles Aktivitätslogging."
),
missing_value_policy=MissingValuePolicy(
available=False,
@ -140,11 +143,10 @@ energy_balance_metadata = PlaceholderMetadata(
legacy_display="nicht verfügbar"
),
known_limitations=(
"TDEE-MODELL: Vereinfacht als bodyweight_kg × 32.5 (mittlerer Multiplikator). "
"NICHT berücksichtigt: Aktivitätslevel, Alter, Geschlecht, Stoffwechselanpassungen. "
"TODO in Code: Harris-Benedict oder Mifflin-St Jeor für präzisere TDEE-Schätzung. "
"ACHTUNG: Energiebilanz ist modellbasiert, nicht direkt gemessen. "
"Einheit ist kcal/Tag (daily average), NICHT 7d-Total."
"TDEE: Bei vollständigem Profil (Größe, Geschlecht, DOB, Gewicht) MifflinSt Jeor BMR × 1.55; "
"sonst Fallback kg×32.5. PAL ist nicht nutzerkonfigurierbar. "
"Energiebilanz ist modellbasiert, nicht gemessen. "
"Einheit kcal/Tag (Tagesmittel), nicht 7-Tage-Summe."
),
# Architecture
@ -435,8 +437,9 @@ Part C Registration Complete:
Total Nutrition Cluster:
- Part A: 4 placeholders (kcal_avg, protein_avg, carb_avg, fat_avg)
- Part B: 5 placeholders (protein targets + adequacy)
- Part C: 5 placeholders (consistency + balance + meta)
14 nutrition placeholders total
- Part C: 5 placeholders in dieser Datei (consistency + balance + meta)
- nutrition_score: eigenes Modul nutrition_score.py
15 Ernährungs-Platzhalter gesamt (A+B+C+nutrition_score)
All registrations follow Phase 0c Multi-Layer Architecture:
- Layer 1 (Data Layer): Calculations

View File

@ -0,0 +1,102 @@
"""
Placeholder registration: nutrition_score
Focus-gewichteter Ernährungs-Meta-Score (separates Modul, um nutrition_part_c schlank zu halten).
"""
from placeholder_registry import (
PlaceholderMetadata,
MissingValuePolicy,
EvidenceType,
OutputType,
PlaceholderType,
register_placeholder,
)
nutrition_score_metadata = PlaceholderMetadata(
key="nutrition_score",
category="Ernährung",
description="Ernährungs-Score (0100), gewichtet nach Focus Areas",
resolver_module="backend/placeholder_resolver.py",
resolver_function="_safe_int",
data_layer_module="backend/data_layer/nutrition_metrics.py",
data_layer_function="calculate_nutrition_score",
source_tables=[
"nutrition_log",
"weight_log",
"user_focus_area_weights",
"focus_area_definitions",
],
semantic_contract=(
"Gewichteter Score 0100 aus Komponenten, die nur einfließen, wenn der Nutzer "
"passende Ernährungs-Focus-Gewichte gesetzt hat (z. B. protein_intake, "
"calorie_balance, macro_consistency). Nutzt u. a. Protein-Adequacy, "
"Makro-Konsistenz, Kalorien-Adhärenz (über Energiebilanz) und Makro-Balance."
),
business_meaning=(
"Verdichteter KPI für Prompts: passt die dokumentierte Ernährung zur "
"gewichteten strategischen Priorität des Nutzers?"
),
unit="score (0-100)",
time_window="composite (7d / 28d je Komponente)",
output_type=OutputType.NUMERIC,
placeholder_type=PlaceholderType.SCORE,
format_hint="Ganzzahl; bei fehlender Ernährungs-Gewichtung oft nicht verfügbar",
example_output="72",
minimum_data_requirements=(
"Mindestens eine Ernährungs-Focus-Komponente mit Gewicht > 0; "
"sowie je nach Komponente ausreichende nutrition_log-/weight_log-Abdeckung."
),
quality_filter_policy=None,
confidence_logic=(
"Kein separates Confidence-Feld im Resolver; fehlende Komponenten werden "
"aus der Gewichtung ausgeschlossen. total_nutrition_weight == 0 ergibt keinen Score."
),
missing_value_policy=MissingValuePolicy(
available=False,
value_raw=None,
missing_reason="not_applicable",
legacy_display="nicht verfügbar",
),
known_limitations=(
"Abhängig von user_focus_area_weights; ohne Ernährungs-Fokus liefert die "
"Funktion None. Kalorien-Adhärenz nutzt 7d-Energiebilanz vs. profiles.goal_mode "
"(weight_loss / strength+recomposition / sonst maintenance). "
"_score_macro_balance nutzt zeilenbasierte 28d-Abfrage (langfristig an "
"Tagesaggregation angleichen)."
),
layer_1_decision="Data Layer (nutrition_metrics.calculate_nutrition_score)",
layer_2a_decision="Placeholder Resolver (_safe_int)",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c: Berechnung in nutrition_metrics",
issue_53_alignment="Layer 1 als Quelle; Komponenten nutzen weitere Layer-1-Funktionen",
evidence={},
)
nutrition_score_metadata.set_evidence("key", EvidenceType.CODE_DERIVED)
nutrition_score_metadata.set_evidence("category", EvidenceType.CODE_DERIVED)
nutrition_score_metadata.set_evidence("description", EvidenceType.MIXED)
nutrition_score_metadata.set_evidence("resolver_module", EvidenceType.CODE_DERIVED)
nutrition_score_metadata.set_evidence("resolver_function", EvidenceType.CODE_DERIVED)
nutrition_score_metadata.set_evidence("data_layer_module", EvidenceType.CODE_DERIVED)
nutrition_score_metadata.set_evidence("data_layer_function", EvidenceType.CODE_DERIVED)
nutrition_score_metadata.set_evidence("source_tables", EvidenceType.CODE_DERIVED)
nutrition_score_metadata.set_evidence("semantic_contract", EvidenceType.CODE_DERIVED)
nutrition_score_metadata.set_evidence("business_meaning", EvidenceType.DRAFT_DERIVED)
nutrition_score_metadata.set_evidence("unit", EvidenceType.CODE_DERIVED)
nutrition_score_metadata.set_evidence("time_window", EvidenceType.CODE_DERIVED)
nutrition_score_metadata.set_evidence("output_type", EvidenceType.CODE_DERIVED)
nutrition_score_metadata.set_evidence("placeholder_type", EvidenceType.CODE_DERIVED)
nutrition_score_metadata.set_evidence("format_hint", EvidenceType.CODE_DERIVED)
nutrition_score_metadata.set_evidence("example_output", EvidenceType.CODE_DERIVED)
nutrition_score_metadata.set_evidence("minimum_data_requirements", EvidenceType.MIXED)
nutrition_score_metadata.set_evidence("confidence_logic", EvidenceType.CODE_DERIVED)
nutrition_score_metadata.set_evidence("missing_value_policy", EvidenceType.CODE_DERIVED)
nutrition_score_metadata.set_evidence("known_limitations", EvidenceType.MIXED)
nutrition_score_metadata.set_evidence("layer_1_decision", EvidenceType.CODE_DERIVED)
nutrition_score_metadata.set_evidence("layer_2a_decision", EvidenceType.CODE_DERIVED)
nutrition_score_metadata.set_evidence("layer_2b_reuse_possible", EvidenceType.TO_VERIFY)
nutrition_score_metadata.set_evidence("architecture_alignment", EvidenceType.CODE_DERIVED)
nutrition_score_metadata.set_evidence("issue_53_alignment", EvidenceType.MIXED)
register_placeholder(nutrition_score_metadata)

View File

@ -0,0 +1,66 @@
"""Registry: Meta-Scores (Phase 0b) — Ziel-Fortschritt und Datenqualität."""
from placeholder_registry import (
PlaceholderMetadata,
MissingValuePolicy,
OutputType,
PlaceholderType,
register_placeholder,
)
from ._evidence import tag_standard_evidence
CAT = "Scores (Phase 0b)"
MVP = lambda reason, disp: MissingValuePolicy(
available=False, value_raw=None, missing_reason=reason, legacy_display=disp
)
def register_phase_0b_meta_scores():
for key, dl_fn, desc, unit in [
(
"goal_progress_score",
"calculate_goal_progress_score",
"Aggregierter Ziel-Fortschritt 0100",
"0100",
),
(
"data_quality_score",
"calculate_data_quality_score",
"Datenqualitäts-Score 0100",
"0100",
),
]:
m = PlaceholderMetadata(
key=key,
category=CAT,
description=desc,
resolver_module="backend/placeholder_resolver.py",
resolver_function="_safe_int",
data_layer_module="backend/data_layer/scores.py",
data_layer_function=dl_fn,
source_tables=["goals", "weight_log", "nutrition_log", "activity_log", "profiles"],
semantic_contract=f"scores.{dl_fn} (siehe Data Layer).",
business_meaning="Meta-KPI für Prompt-Gewichtung",
unit=unit,
time_window="composite",
output_type=OutputType.NUMERIC,
placeholder_type=PlaceholderType.SCORE,
format_hint="Ganzzahl als String",
example_output="72",
minimum_data_requirements="Abhängig von Score-Implementierung",
quality_filter_policy=None,
confidence_logic="Wie calculate_* in scores.py",
missing_value_policy=MVP("insufficient_data", "nicht verfügbar"),
known_limitations="Bei dünnen Daten weniger aussagekräftig",
layer_1_decision=f"scores.{dl_fn}",
layer_2a_decision="_safe_int",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0b",
issue_53_alignment="Layer 1",
evidence={},
)
tag_standard_evidence(m)
register_placeholder(m)
register_phase_0b_meta_scores()

View File

@ -0,0 +1,392 @@
"""Registry: Ziele, Fokusbereiche, Kategorie-Scores und formatierte Listen (Phase 0b)."""
from placeholder_registry import (
PlaceholderMetadata,
MissingValuePolicy,
OutputType,
PlaceholderType,
register_placeholder,
)
from ._evidence import tag_standard_evidence
CAT = "Ziele & Fokus (Phase 0b)"
MVP = lambda reason, disp: MissingValuePolicy(
available=False, value_raw=None, missing_reason=reason, legacy_display=disp
)
def register_phase_0b_ziele_fokus():
# Top-Ziel / Top-Fokusbereich
m = PlaceholderMetadata(
key="top_goal_name",
category=CAT,
description="Name/Typ des höchstpriorisierten Ziels",
resolver_module="backend/placeholder_resolver.py",
resolver_function="_safe_str",
data_layer_module="backend/data_layer/scores.py",
data_layer_function="get_top_priority_goal",
source_tables=["goals"],
semantic_contract="Feld name oder goal_type aus get_top_priority_goal",
business_meaning="Priorisierung für KI-Empfehlungen",
unit="text",
time_window="aktuell",
output_type=OutputType.STRING,
placeholder_type=PlaceholderType.INTERPRETED,
format_hint="Kurztext",
example_output="Gewicht 80kg",
minimum_data_requirements="Mindestens ein aktives Ziel",
quality_filter_policy=None,
confidence_logic="scores.get_top_priority_goal",
missing_value_policy=MVP("insufficient_data", "nicht verfügbar"),
known_limitations=None,
layer_1_decision="scores.get_top_priority_goal",
layer_2a_decision="_safe_str('top_goal_name')",
layer_2b_reuse_possible=False,
architecture_alignment="Phase 0b",
issue_53_alignment="Layer 1",
evidence={},
)
tag_standard_evidence(m)
register_placeholder(m)
m = PlaceholderMetadata(
key="top_goal_progress_pct",
category=CAT,
description="Fortschritt Top-Ziel (%)",
resolver_module="backend/placeholder_resolver.py",
resolver_function="_safe_int",
data_layer_module="backend/data_layer/scores.py",
data_layer_function="get_top_priority_goal",
source_tables=["goals"],
semantic_contract="progress_pct aus get_top_priority_goal",
business_meaning="Priorisierung für KI-Empfehlungen",
unit="%",
time_window="aktuell",
output_type=OutputType.NUMERIC,
placeholder_type=PlaceholderType.SCORE,
format_hint="Ganzzahl",
example_output="65",
minimum_data_requirements="Mindestens ein aktives Ziel",
quality_filter_policy=None,
confidence_logic="scores.get_top_priority_goal",
missing_value_policy=MVP("insufficient_data", "nicht verfügbar"),
known_limitations=None,
layer_1_decision="scores.get_top_priority_goal",
layer_2a_decision="_safe_int('top_goal_progress_pct')",
layer_2b_reuse_possible=False,
architecture_alignment="Phase 0b",
issue_53_alignment="Layer 1",
evidence={},
)
tag_standard_evidence(m)
register_placeholder(m)
m = PlaceholderMetadata(
key="top_goal_status",
category=CAT,
description="Status-Label Top-Ziel",
resolver_module="backend/placeholder_resolver.py",
resolver_function="_safe_str",
data_layer_module="backend/data_layer/scores.py",
data_layer_function="get_top_priority_goal",
source_tables=["goals"],
semantic_contract="status aus get_top_priority_goal",
business_meaning="Priorisierung für KI-Empfehlungen",
unit="text",
time_window="aktuell",
output_type=OutputType.STRING,
placeholder_type=PlaceholderType.INTERPRETED,
format_hint="Kurztext",
example_output="active",
minimum_data_requirements="Mindestens ein aktives Ziel",
quality_filter_policy=None,
confidence_logic="scores.get_top_priority_goal",
missing_value_policy=MVP("insufficient_data", "nicht verfügbar"),
known_limitations=None,
layer_1_decision="scores.get_top_priority_goal",
layer_2a_decision="_safe_str('top_goal_status')",
layer_2b_reuse_possible=False,
architecture_alignment="Phase 0b",
issue_53_alignment="Layer 1",
evidence={},
)
tag_standard_evidence(m)
register_placeholder(m)
m = PlaceholderMetadata(
key="top_focus_area_name",
category=CAT,
description="Bezeichnung des gewichtet stärksten Fokusbereichs",
resolver_module="backend/placeholder_resolver.py",
resolver_function="_safe_str",
data_layer_module="backend/data_layer/scores.py",
data_layer_function="get_top_focus_area",
source_tables=["user_focus_area_weights", "focus_area_definitions"],
semantic_contract="label aus get_top_focus_area",
business_meaning="Priorisierung für KI-Empfehlungen",
unit="text",
time_window="aktuell",
output_type=OutputType.STRING,
placeholder_type=PlaceholderType.INTERPRETED,
format_hint="Kurztext",
example_output="Kraft",
minimum_data_requirements="Gewichtete Fokusbereiche",
quality_filter_policy=None,
confidence_logic="scores.get_top_focus_area",
missing_value_policy=MVP("insufficient_data", "nicht verfügbar"),
known_limitations=None,
layer_1_decision="scores.get_top_focus_area",
layer_2a_decision="_safe_str('top_focus_area_name')",
layer_2b_reuse_possible=False,
architecture_alignment="Phase 0b",
issue_53_alignment="Layer 1",
evidence={},
)
tag_standard_evidence(m)
register_placeholder(m)
m = PlaceholderMetadata(
key="top_focus_area_progress",
category=CAT,
description="Fortschritt Top-Fokusbereich (%)",
resolver_module="backend/placeholder_resolver.py",
resolver_function="_safe_int",
data_layer_module="backend/data_layer/scores.py",
data_layer_function="get_top_focus_area",
source_tables=["user_focus_area_weights", "focus_area_definitions", "goals"],
semantic_contract="progress aus get_top_focus_area",
business_meaning="Priorisierung für KI-Empfehlungen",
unit="%",
time_window="aktuell",
output_type=OutputType.NUMERIC,
placeholder_type=PlaceholderType.SCORE,
format_hint="Ganzzahl",
example_output="58",
minimum_data_requirements="Gewichtete Fokusbereiche",
quality_filter_policy=None,
confidence_logic="scores.get_top_focus_area",
missing_value_policy=MVP("insufficient_data", "nicht verfügbar"),
known_limitations=None,
layer_1_decision="scores.get_top_focus_area",
layer_2a_decision="_safe_int('top_focus_area_progress')",
layer_2b_reuse_possible=False,
architecture_alignment="Phase 0b",
issue_53_alignment="Layer 1",
evidence={},
)
tag_standard_evidence(m)
register_placeholder(m)
# Kategorie Progress / Weight (7 Kategorien)
for slug in (
"körper",
"ernährung",
"aktivität",
"recovery",
"vitalwerte",
"mental",
"lebensstil",
):
key_p = f"focus_cat_{slug}_progress"
key_w = f"focus_cat_{slug}_weight"
m_p = PlaceholderMetadata(
key=key_p,
category=CAT,
description=f"Aggregierter Fortschritt Kategorie „{slug}“ (%)",
resolver_module="backend/placeholder_resolver.py",
resolver_function="_safe_int",
data_layer_module="backend/data_layer/scores.py",
data_layer_function="calculate_category_progress",
source_tables=["goals", "focus_area_definitions", "user_focus_area_weights"],
semantic_contract=f"scores.calculate_category_progress(pid, '{slug}')",
business_meaning="Focus-Area-Kategorie-Score",
unit="%",
time_window="aktuell",
output_type=OutputType.NUMERIC,
placeholder_type=PlaceholderType.SCORE,
format_hint="Ganzzahl",
example_output="55",
minimum_data_requirements="Gewichtete Bereiche in Kategorie",
quality_filter_policy=None,
confidence_logic="scores.calculate_category_progress",
missing_value_policy=MVP("insufficient_data", "nicht verfügbar"),
known_limitations=None,
layer_1_decision="scores.calculate_category_progress",
layer_2a_decision="_safe_int",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0b",
issue_53_alignment="Layer 1",
evidence={},
)
tag_standard_evidence(m_p)
register_placeholder(m_p)
m_w = PlaceholderMetadata(
key=key_w,
category=CAT,
description=f"Nutzer-Gewichtung Kategorie „{slug}“ (Anteil 01)",
resolver_module="backend/placeholder_resolver.py",
resolver_function="_safe_float",
data_layer_module="backend/data_layer/scores.py",
data_layer_function="calculate_category_weight",
source_tables=["user_focus_area_weights", "focus_area_definitions"],
semantic_contract=f"scores.calculate_category_weight(pid, '{slug}')",
business_meaning="Kategorie-Gewichtung im Fokusmodell",
unit="01",
time_window="aktuell",
output_type=OutputType.NUMERIC,
placeholder_type=PlaceholderType.INTERPRETED,
format_hint="Dezimal",
example_output="0.25",
minimum_data_requirements="user_focus_area_weights",
quality_filter_policy=None,
confidence_logic="scores.calculate_category_weight",
missing_value_policy=MVP("insufficient_data", "nicht verfügbar"),
known_limitations=None,
layer_1_decision="scores.calculate_category_weight",
layer_2a_decision="_safe_float",
layer_2b_reuse_possible=False,
architecture_alignment="Phase 0b",
issue_53_alignment="Layer 1",
evidence={},
)
tag_standard_evidence(m_w)
register_placeholder(m_w)
# Strukturierte Ziele / Fokus
for key, res_fn, dl_mod, dl_fn, desc, out, ptype in [
(
"active_goals_json",
"_safe_json",
"backend/goal_utils.py",
"get_active_goals",
"Aktive Ziele als JSON",
OutputType.JSON,
PlaceholderType.RAW_DATA,
),
(
"active_goals_md",
"_safe_str",
"backend/placeholder_resolver.py",
"_format_goals_as_markdown",
"Aktive Ziele als Markdown-Tabelle",
OutputType.TEXT_SUMMARY,
PlaceholderType.INTERPRETED,
),
(
"focus_areas_weighted_json",
"_safe_json",
"backend/placeholder_resolver.py",
"_get_focus_areas_weighted_json",
"Gewichtete Fokusbereiche mit Namen (JSON)",
OutputType.JSON,
PlaceholderType.RAW_DATA,
),
(
"focus_areas_weighted_md",
"_safe_str",
"backend/placeholder_resolver.py",
"_format_focus_areas_as_markdown",
"Gewichtete Fokusbereiche als Markdown",
OutputType.TEXT_SUMMARY,
PlaceholderType.INTERPRETED,
),
(
"focus_area_weights_json",
"_safe_json",
"backend/data_layer/scores.py",
"get_user_focus_weights",
"Rohe Gewichtungen key→Anteil (JSON)",
OutputType.JSON,
PlaceholderType.RAW_DATA,
),
]:
m = PlaceholderMetadata(
key=key,
category=CAT,
description=desc,
resolver_module="backend/placeholder_resolver.py",
resolver_function=res_fn,
data_layer_module=dl_mod,
data_layer_function=dl_fn,
source_tables=["goals", "focus_area_definitions", "user_focus_area_weights"],
semantic_contract=f"{dl_fn} (siehe Modul {dl_mod})",
business_meaning="Strukturierte Übersicht für Prompts",
unit="JSON" if out == OutputType.JSON else "markdown",
time_window="aktuell",
output_type=out,
placeholder_type=ptype,
format_hint="String aus Resolver",
example_output="[]" if out == OutputType.JSON else "",
minimum_data_requirements="Ziele bzw. Fokusgewichte",
quality_filter_policy=None,
confidence_logic="Resolver + goal_utils / scores",
missing_value_policy=MVP("insufficient_data", "[]" if out == OutputType.JSON else "nicht verfügbar"),
known_limitations=None,
layer_1_decision=dl_fn,
layer_2a_decision=res_fn,
layer_2b_reuse_possible=False,
architecture_alignment="Phase 0b",
issue_53_alignment="Layer 1",
evidence={},
)
tag_standard_evidence(m)
register_placeholder(m)
for key, res_fn, dl_fn, desc, ex in [
(
"top_3_focus_areas",
"_safe_str",
"_format_top_focus_areas",
"Top-3 Fokusbereiche als formatierter Text",
"1. Kraft …",
),
(
"top_3_goals_behind_schedule",
"_safe_str",
"_format_goals_behind",
"Bis zu drei Ziele hinter Zeitplan",
"",
),
(
"top_3_goals_on_track",
"_safe_str",
"_format_goals_on_track",
"Bis zu drei Ziele im Plan",
"",
),
]:
m = PlaceholderMetadata(
key=key,
category=CAT,
description=desc,
resolver_module="backend/placeholder_resolver.py",
resolver_function=res_fn,
data_layer_module="backend/goal_utils.py",
data_layer_function="get_active_goals",
source_tables=["goals", "focus_area_definitions"],
semantic_contract=f"Resolver {dl_fn}",
business_meaning="Kurzlisten für Coaching-Prompts",
unit="text",
time_window="aktuell",
output_type=OutputType.TEXT_SUMMARY,
placeholder_type=PlaceholderType.INTERPRETED,
format_hint="Freitext / Aufzählung",
example_output=ex,
minimum_data_requirements="Ziele / Fokusdaten",
quality_filter_policy=None,
confidence_logic=dl_fn,
missing_value_policy=MVP("insufficient_data", "nicht verfügbar"),
known_limitations=None,
layer_1_decision="goals + focus aggregation",
layer_2a_decision=dl_fn,
layer_2b_reuse_possible=False,
architecture_alignment="Phase 0b",
issue_53_alignment="Layer 2a",
evidence={},
)
tag_standard_evidence(m)
register_placeholder(m)
register_phase_0b_ziele_fokus()

View File

@ -0,0 +1,139 @@
"""
Registry: Profil-Stammdaten und statische Zeitraum-Labels für Prompts.
"""
from placeholder_registry import (
PlaceholderMetadata,
MissingValuePolicy,
OutputType,
PlaceholderType,
register_placeholder,
)
from ._evidence import tag_standard_evidence
MVP = lambda reason, disp: MissingValuePolicy(
available=False, value_raw=None, missing_reason=reason, legacy_display=disp
)
def register_profil_zeitraum():
cat_profil = "Profil"
for key, desc, res_fn, unit, ptype, out, hint, ex, sem in [
(
"name",
"Anzeigename aus profiles.name",
"get_profile_name",
"text",
PlaceholderType.ATOMIC,
OutputType.STRING,
"Kurzname",
"Max",
"profiles.name, Fallback „Nutzer“.",
),
(
"age",
"Alter in Jahren aus profiles.dob",
"get_profile_age_display",
"Jahre",
PlaceholderType.ATOMIC,
OutputType.STRING,
"Ganzzahl oder unbekannt",
"42",
"Berechnung aus Geburtsdatum; PostgreSQL date oder ISO-String.",
),
(
"height",
"Körpergröße (cm) aus profiles.height",
"get_profile_height_display",
"cm",
PlaceholderType.ATOMIC,
OutputType.STRING,
"Zahl oder unbekannt",
"180",
"profiles.height.",
),
(
"geschlecht",
"Geschlecht (männlich/weiblich) aus profiles.sex",
"get_profile_geschlecht_display",
"Kategorie",
PlaceholderType.ATOMIC,
OutputType.STRING,
"m/w-Mapping",
"männlich",
"sex == 'm' → männlich, sonst weiblich.",
),
]:
m = PlaceholderMetadata(
key=key,
category=cat_profil,
description=desc,
resolver_module="backend/placeholder_resolver.py",
resolver_function=res_fn,
data_layer_module=None,
data_layer_function=None,
source_tables=["profiles"],
semantic_contract=sem,
business_meaning="Profil-Kontext für KI-Prompts",
unit=unit,
time_window="latest profile row",
output_type=out,
placeholder_type=ptype,
format_hint=hint,
example_output=ex,
minimum_data_requirements="Profilzeile",
quality_filter_policy=None,
confidence_logic="Row vorhanden",
missing_value_policy=MVP("no_data", "unbekannt" if key != "name" else "Nutzer"),
known_limitations="Keine diversen Geschlechtsoptionen im Platzhalter",
layer_1_decision="profiles",
layer_2a_decision=res_fn,
layer_2b_reuse_possible=False,
architecture_alignment="Phase 0b",
issue_53_alignment="Resolver",
evidence={},
)
tag_standard_evidence(m)
register_placeholder(m)
cat_zeit = "Zeitraum"
for key, desc, res_fn, sem, ex_out in [
("datum_heute", "Heutiges Datum (lokal)", "get_datum_heute", "datetime.now, Format dd.mm.yyyy", "11.04.2026"),
("zeitraum_7d", "Label „letzte 7 Tage“", "get_zeitraum_label_7d", "Statisches UI/Prompt-Label", "letzte 7 Tage"),
("zeitraum_30d", "Label „letzte 30 Tage“", "get_zeitraum_label_30d", "Statisches UI/Prompt-Label", "letzte 30 Tage"),
("zeitraum_90d", "Label „letzte 90 Tage“", "get_zeitraum_label_90d", "Statisches UI/Prompt-Label", "letzte 90 Tage"),
]:
m = PlaceholderMetadata(
key=key,
category=cat_zeit,
description=desc,
resolver_module="backend/placeholder_resolver.py",
resolver_function=res_fn,
data_layer_module=None,
data_layer_function=None,
source_tables=[],
semantic_contract=sem,
business_meaning="Zeitlicher Bezug im Prompt ohne Datenabfrage",
unit="label",
time_window="n/a",
output_type=OutputType.STRING,
placeholder_type=PlaceholderType.META,
format_hint="Kurzdeutsch",
example_output=ex_out,
minimum_data_requirements=None,
quality_filter_policy=None,
confidence_logic="Immer verfügbar",
missing_value_policy=None,
known_limitations="Kein kalender-basierter Datenfilter allein durch Platzhalter",
layer_1_decision="n/a",
layer_2a_decision=res_fn,
layer_2b_reuse_possible=False,
architecture_alignment="Phase 0b",
issue_53_alignment="Resolver",
evidence={},
)
tag_standard_evidence(m)
register_placeholder(m)
register_profil_zeitraum()

View File

@ -0,0 +1,236 @@
"""
Registry: Schlaf, Ruhetage, Recovery-Score, Schlaf-Metriken, Schlaf-Erholungs-Korrelation.
"""
from placeholder_registry import (
PlaceholderMetadata,
MissingValuePolicy,
EvidenceType,
OutputType,
PlaceholderType,
register_placeholder,
)
CAT = "Schlaf & Erholung"
MVP = lambda reason, disp: MissingValuePolicy(
available=False, value_raw=None, missing_reason=reason, legacy_display=disp
)
def _tag(m: PlaceholderMetadata):
for f in (
"key", "category", "description", "resolver_module", "resolver_function",
"data_layer_module", "data_layer_function", "source_tables", "semantic_contract",
"unit", "time_window", "output_type", "placeholder_type", "format_hint",
"example_output", "minimum_data_requirements", "confidence_logic",
"missing_value_policy", "layer_1_decision", "layer_2a_decision",
"layer_2b_reuse_possible", "architecture_alignment", "issue_53_alignment",
):
m.set_evidence(f, EvidenceType.CODE_DERIVED)
m.set_evidence("business_meaning", EvidenceType.DRAFT_DERIVED)
m.set_evidence("known_limitations", EvidenceType.MIXED)
def register_schlaf_erholung():
# ── formatierte Schlaf-/Ruhetage-Snapshots ───────────────────────────────
m = PlaceholderMetadata(
key="sleep_avg_duration",
category=CAT,
description="Durchschnittliche Schlafdauer (Stunden), formatiert",
resolver_module="backend/placeholder_resolver.py",
resolver_function="get_sleep_avg_duration",
data_layer_module="backend/data_layer/recovery_metrics.py",
data_layer_function="get_sleep_duration_data",
source_tables=["sleep_log"],
semantic_contract="Mittel aus Schlafphasen im Fenster (siehe get_sleep_duration_data).",
business_meaning="KI-Kontext Schlafdauer",
unit="h (Anzeige mit Einheit)",
time_window="7d default im Resolver",
output_type=OutputType.STRING,
placeholder_type=PlaceholderType.INTERPRETED,
format_hint="z. B. 7.2h",
example_output="7.2h",
minimum_data_requirements="sleep_log im Fenster",
quality_filter_policy=None,
confidence_logic="data['confidence'] im Layer1",
missing_value_policy=MVP("no_data", "nicht verfügbar"),
known_limitations="Abhängig von Import/Qualität der Phasen",
layer_1_decision="recovery_metrics.get_sleep_duration_data",
layer_2a_decision="get_sleep_avg_duration",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c",
issue_53_alignment="Layer 1",
evidence={},
)
_tag(m)
register_placeholder(m)
m = PlaceholderMetadata(
key="sleep_avg_quality",
category=CAT,
description="Schlafqualität (Deep+REM %), formatiert",
resolver_module="backend/placeholder_resolver.py",
resolver_function="get_sleep_avg_quality",
data_layer_module="backend/data_layer/recovery_metrics.py",
data_layer_function="get_sleep_quality_data",
source_tables=["sleep_log"],
semantic_contract="Anteil Deep+REM aus Segmenten (siehe get_sleep_quality_data).",
business_meaning="KI-Kontext Schlafqualität",
unit="%",
time_window="7d default",
output_type=OutputType.STRING,
placeholder_type=PlaceholderType.INTERPRETED,
format_hint="Prozent oder nicht verfügbar",
example_output="24%",
minimum_data_requirements="sleep_log mit Phasen",
quality_filter_policy=None,
confidence_logic="Layer-1-Confidence",
missing_value_policy=MVP("no_data", "nicht verfügbar"),
known_limitations="Segment-Schreibweise case-sensitiv normalisiert",
layer_1_decision="recovery_metrics.get_sleep_quality_data",
layer_2a_decision="get_sleep_avg_quality",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c",
issue_53_alignment="Layer 1",
evidence={},
)
_tag(m)
register_placeholder(m)
m = PlaceholderMetadata(
key="rest_days_count",
category=CAT,
description="Anzahl dokumentierter Ruhetage (30d default)",
resolver_module="backend/placeholder_resolver.py",
resolver_function="get_rest_days_count",
data_layer_module="backend/data_layer/recovery_metrics.py",
data_layer_function="get_rest_days_data",
source_tables=["rest_days"],
semantic_contract="Count rest_days im Zeitraum",
business_meaning="Aktive/passive Erholungstags-Übersicht",
unit="count",
time_window="30d default",
output_type=OutputType.STRING,
placeholder_type=PlaceholderType.RAW_DATA,
format_hint="z. B. 2 Ruhetage",
example_output="2 Ruhetage",
minimum_data_requirements="rest_days",
quality_filter_policy=None,
confidence_logic="Immer Zählung, 0 möglich",
missing_value_policy=MVP("no_data", "0 Ruhetage"),
known_limitations="Nur explizit erfasste Ruhetage",
layer_1_decision="recovery_metrics.get_rest_days_data",
layer_2a_decision="get_rest_days_count",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c",
issue_53_alignment="Layer 1",
evidence={},
)
_tag(m)
register_placeholder(m)
m = PlaceholderMetadata(
key="recovery_score",
category=CAT,
description="Recovery-Score 0100 (v2, komposit)",
resolver_module="backend/placeholder_resolver.py",
resolver_function="_safe_int",
data_layer_module="backend/data_layer/recovery_metrics.py",
data_layer_function="calculate_recovery_score_v2",
source_tables=["sleep_log", "vitals_baseline", "activity_log"],
semantic_contract="Gewichteter Score aus Schlaf, Vitaltrends, optional Load (siehe Implementierung).",
business_meaning="Gesamt-Recovery-KPI für Prompts",
unit="0100",
time_window="composite",
output_type=OutputType.NUMERIC,
placeholder_type=PlaceholderType.SCORE,
format_hint="Ganzzahl-String",
example_output="72",
minimum_data_requirements="Teilkomponenten je nach Gewichtung",
quality_filter_policy=None,
confidence_logic="Wie calculate_recovery_score_v2",
missing_value_policy=MVP("insufficient_data", "nicht verfügbar"),
known_limitations="Abhängig von Datenabdeckung HF/HRV/Schlaf",
layer_1_decision="recovery_metrics.calculate_recovery_score_v2",
layer_2a_decision="_safe_int('recovery_score_v2')",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c",
issue_53_alignment="Layer 1",
evidence={},
)
_tag(m)
register_placeholder(m)
for key, dl_fn, desc, unit, tbls, res_fn in [
("sleep_avg_duration_7d", "calculate_sleep_avg_duration_7d", "Durchschnittliche Schlafdauer 7d (h)", "h", ["sleep_log"], "_safe_float"),
("sleep_debt_hours", "calculate_sleep_debt_hours", "Kumulative Schlafschuld (h)", "h", ["sleep_log"], "_safe_float"),
("sleep_regularity_proxy", "calculate_sleep_regularity_proxy", "Schlaf-Regularität (Proxy)", "min", ["sleep_log"], "_safe_float"),
("recent_load_balance_3d", "calculate_recent_load_balance_3d", "Load-Balance 3d (Score)", "score", ["activity_log"], "_safe_int"),
("sleep_quality_7d", "calculate_sleep_quality_7d", "Schlafqualität 7d (0100)", "0-100", ["sleep_log"], "_safe_int"),
]:
m = PlaceholderMetadata(
key=key,
category=CAT,
description=desc,
resolver_module="backend/placeholder_resolver.py",
resolver_function=res_fn,
data_layer_module="backend/data_layer/recovery_metrics.py",
data_layer_function=dl_fn,
source_tables=tbls,
semantic_contract=f"Berechnung {dl_fn} in recovery_metrics.",
business_meaning="Erholungs-Detailmetrik",
unit=unit,
time_window="siehe Funktion",
output_type=OutputType.NUMERIC,
placeholder_type=PlaceholderType.INTERPRETED,
format_hint="numerischer String",
example_output="1.0",
minimum_data_requirements="wie Funktion",
quality_filter_policy=None,
confidence_logic="Funktionsintern",
missing_value_policy=MVP("insufficient_data", "nicht verfügbar"),
known_limitations=None,
layer_1_decision=f"recovery_metrics.{dl_fn}",
layer_2a_decision="Resolver _safe_float/_safe_int",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c",
issue_53_alignment="Layer 1",
evidence={},
)
_tag(m)
register_placeholder(m)
m = PlaceholderMetadata(
key="correlation_sleep_recovery",
category=CAT,
description="JSON: Korrelation Schlaf ↔ Recovery-Indikatoren",
resolver_module="backend/placeholder_resolver.py",
resolver_function="_safe_json",
data_layer_module="backend/data_layer/correlations.py",
data_layer_function="calculate_correlation_sleep_recovery",
source_tables=["sleep_log", "vitals_baseline", "activity_log"],
semantic_contract="Strukturierte Korrelationsauswertung (siehe correlations).",
business_meaning="KI: Zusammenhänge Schlaf und Erholung",
unit="JSON",
time_window="funktionsabhängig",
output_type=OutputType.JSON,
placeholder_type=PlaceholderType.RAW_DATA,
format_hint="JSON-String",
example_output="{}",
minimum_data_requirements="Ausreichend gekoppelte Datenpunkte",
quality_filter_policy=None,
confidence_logic="Wie correlation_metrics",
missing_value_policy=MVP("insufficient_data", "{}"),
known_limitations="Bei wenig Daten leer oder schwach",
layer_1_decision="correlations.calculate_correlation_sleep_recovery",
layer_2a_decision="_safe_json",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c",
issue_53_alignment="Layer 1",
evidence={},
)
_tag(m)
register_placeholder(m)
register_schlaf_erholung()

View File

@ -0,0 +1,180 @@
"""
Registry: Baseline-Vitals (Ruhepuls, HRV, VO2 Max) und Abweichung vs. persönlicher Baseline.
"""
from placeholder_registry import (
PlaceholderMetadata,
MissingValuePolicy,
EvidenceType,
OutputType,
PlaceholderType,
register_placeholder,
)
CAT = "Vitalwerte"
MVP = lambda reason, disp: MissingValuePolicy(
available=False, value_raw=None, missing_reason=reason, legacy_display=disp
)
def _tag(m: PlaceholderMetadata):
for f in (
"key", "category", "description", "resolver_module", "resolver_function",
"data_layer_module", "data_layer_function", "source_tables", "semantic_contract",
"unit", "time_window", "output_type", "placeholder_type", "format_hint",
"example_output", "minimum_data_requirements", "confidence_logic",
"missing_value_policy", "layer_1_decision", "layer_2a_decision",
"layer_2b_reuse_possible", "architecture_alignment", "issue_53_alignment",
):
m.set_evidence(f, EvidenceType.CODE_DERIVED)
m.set_evidence("business_meaning", EvidenceType.DRAFT_DERIVED)
m.set_evidence("known_limitations", EvidenceType.MIXED)
def register_vitalwerte():
m = PlaceholderMetadata(
key="vitals_avg_hr",
category=CAT,
description="Durchschnittlicher Ruhepuls (7d), formatiert",
resolver_module="backend/placeholder_resolver.py",
resolver_function="get_vitals_avg_hr",
data_layer_module="backend/data_layer/health_metrics.py",
data_layer_function="get_resting_heart_rate_data",
source_tables=["vitals_baseline"],
semantic_contract="Mittel RHR aus vitals_baseline im Fenster (siehe health_metrics).",
business_meaning="KI-Kontext kardiovaskuläre Ruhelage",
unit="bpm (Anzeige mit Einheit)",
time_window="7d default im Resolver",
output_type=OutputType.STRING,
placeholder_type=PlaceholderType.INTERPRETED,
format_hint="z. B. 58 bpm",
example_output="58 bpm",
minimum_data_requirements="vitals_baseline im Fenster",
quality_filter_policy=None,
confidence_logic="data['confidence'] im Layer1",
missing_value_policy=MVP("no_data", "nicht verfügbar"),
known_limitations="Nur erfasste Morgen-Baseline-Messungen",
layer_1_decision="health_metrics.get_resting_heart_rate_data",
layer_2a_decision="get_vitals_avg_hr",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c",
issue_53_alignment="Layer 1",
evidence={},
)
_tag(m)
register_placeholder(m)
m = PlaceholderMetadata(
key="vitals_avg_hrv",
category=CAT,
description="Durchschnittliche HRV (7d), formatiert",
resolver_module="backend/placeholder_resolver.py",
resolver_function="get_vitals_avg_hrv",
data_layer_module="backend/data_layer/health_metrics.py",
data_layer_function="get_heart_rate_variability_data",
source_tables=["vitals_baseline"],
semantic_contract="Mittel HRV aus vitals_baseline im Fenster.",
business_meaning="KI-Kontext autonome Regulation / Erholung",
unit="ms (Anzeige mit Einheit)",
time_window="7d default",
output_type=OutputType.STRING,
placeholder_type=PlaceholderType.INTERPRETED,
format_hint="z. B. 45 ms",
example_output="45 ms",
minimum_data_requirements="vitals_baseline mit HRV",
quality_filter_policy=None,
confidence_logic="data['confidence'] im Layer1",
missing_value_policy=MVP("no_data", "nicht verfügbar"),
known_limitations="Geräte-/Messprotokoll kann streuen",
layer_1_decision="health_metrics.get_heart_rate_variability_data",
layer_2a_decision="get_vitals_avg_hrv",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c",
issue_53_alignment="Layer 1",
evidence={},
)
_tag(m)
register_placeholder(m)
m = PlaceholderMetadata(
key="vitals_vo2_max",
category=CAT,
description="Aktueller VO2 Max (letzte Messung), formatiert",
resolver_module="backend/placeholder_resolver.py",
resolver_function="get_vitals_vo2_max",
data_layer_module="backend/data_layer/health_metrics.py",
data_layer_function="get_vo2_max_data",
source_tables=["vitals_baseline"],
semantic_contract="Jüngster vo2_max aus vitals_baseline.",
business_meaning="Ausdauer-/Fitness-Kontext",
unit="ml/kg/min",
time_window="latest",
output_type=OutputType.STRING,
placeholder_type=PlaceholderType.INTERPRETED,
format_hint="eine Dezimalstelle + Einheit",
example_output="42.0 ml/kg/min",
minimum_data_requirements="mindestens eine VO2-Messung",
quality_filter_policy=None,
confidence_logic="data['confidence'] im Layer1",
missing_value_policy=MVP("no_data", "nicht verfügbar"),
known_limitations="Schätzung vs. Labortest je nach Quelle",
layer_1_decision="health_metrics.get_vo2_max_data",
layer_2a_decision="get_vitals_vo2_max",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c",
issue_53_alignment="Layer 1",
evidence={},
)
_tag(m)
register_placeholder(m)
for key, dl_fn, desc, unit, res_fn in [
(
"hrv_vs_baseline_pct",
"calculate_hrv_vs_baseline_pct",
"HRV vs. persönlicher Baseline (%)",
"%",
"_safe_float",
),
(
"rhr_vs_baseline_pct",
"calculate_rhr_vs_baseline_pct",
"Ruhepuls vs. persönlicher Baseline (%)",
"%",
"_safe_float",
),
]:
m = PlaceholderMetadata(
key=key,
category=CAT,
description=desc,
resolver_module="backend/placeholder_resolver.py",
resolver_function=res_fn,
data_layer_module="backend/data_layer/recovery_metrics.py",
data_layer_function=dl_fn,
source_tables=["vitals_baseline"],
semantic_contract=f"Vergleich aktueller Wert zu Baseline (siehe {dl_fn}).",
business_meaning="Erholungs- und Belastungsindikator relativ zur Norm des Nutzers",
unit=unit,
time_window="funktionsintern",
output_type=OutputType.NUMERIC,
placeholder_type=PlaceholderType.INTERPRETED,
format_hint="numerischer Prozent-String",
example_output="5.2",
minimum_data_requirements="Ausreichend Baseline-Historie",
quality_filter_policy=None,
confidence_logic="Funktionsintern",
missing_value_policy=MVP("insufficient_data", "nicht verfügbar"),
known_limitations="Baseline braucht ausreichend Vorlauf",
layer_1_decision=f"recovery_metrics.{dl_fn}",
layer_2a_decision=f"Resolver {res_fn}",
layer_2b_reuse_possible=True,
architecture_alignment="Phase 0c",
issue_53_alignment="Layer 1",
evidence={},
)
_tag(m)
register_placeholder(m)
register_vitalwerte()

View File

@ -258,6 +258,42 @@ class PlaceholderRegistry:
return metadata._resolver_func(profile_id)
def build_ai_placeholder_caption(metadata: PlaceholderMetadata, max_len: int = 400) -> str:
"""
Kurzerklärung / Einordnung für {{key|x}} und Exportfeld ``ai_caption`` (ohne Wert, ohne Einheit).
Inhalt: business_meaning oder gekürzter semantic_contract; bei SCORE-Zeilen die 0100-Skala.
Nicht enthalten: description (die nur bei {{key|d}} angehängt wird) und keine Technischer Bezug: -Zeile.
"""
desc = (metadata.description or "").strip()
bm = (metadata.business_meaning or "").strip()
sc = (metadata.semantic_contract or "").strip()
chunks: List[str] = []
interpret = bm
if not interpret and sc:
interpret = sc if len(sc) <= max_len else sc[: max_len - 1] + ""
if interpret:
il = interpret.lower()
redundant = bool(
desc
and len(desc) >= 10
and desc.lower() in il
)
if not redundant:
chunks.append(interpret)
if metadata.placeholder_type == PlaceholderType.SCORE:
chunks.append("Skala 0100: höher = im Modell günstiger / besser abgestimmt.")
out = " ".join(c for c in chunks if c).strip()
if len(out) > max_len + 120:
out = out[: max_len + 60] + ""
return out
# Global registry instance
_global_registry = PlaceholderRegistry()

File diff suppressed because it is too large Load Diff

View File

@ -12,14 +12,17 @@ import re
from typing import Dict, Any, Optional
from db import get_db, get_cursor, r2d
from fastapi import HTTPException
from placeholder_resolver import get_catalog_row_for_key
def resolve_placeholders(template: str, variables: Dict[str, Any], debug_info: Optional[Dict] = None, catalog: Optional[Dict] = None) -> str:
"""
Replace {{placeholder}} with values from variables dict.
Supports modifiers:
- {{key|d}} - Include description in parentheses (requires catalog)
Modifiers (Katalog aus get_placeholder_catalog empfohlen):
- {{key|d}} Wert description (kurz)
- {{key|x}} nur Erklärung (Katalogfeld ai_caption), ohne Zahlenwert
- {{key|d,x}} Wert description Erklärung
Args:
template: String with {{key}} or {{key|modifiers}} placeholders
@ -40,46 +43,66 @@ def resolve_placeholders(template: str, variables: Dict[str, Any], debug_info: O
parts = full_placeholder.split('|')
key = parts[0].strip()
modifiers = parts[1].strip() if len(parts) > 1 else ''
mods = {x.strip().lower() for x in modifiers.split(",") if x.strip()}
want_d = "d" in mods
want_x = "x" in mods
if key in variables:
value = variables[key]
# Convert dict/list to JSON string
if isinstance(value, (dict, list)):
resolved_value = json.dumps(value, ensure_ascii=False)
else:
resolved_value = str(value)
# Apply modifiers
if 'd' in modifiers:
if catalog:
# Add description from catalog
description = None
for cat_items in catalog.values():
matching = [item for item in cat_items if item['key'] == key]
if matching:
description = matching[0].get('description', '')
break
if description:
resolved_value = f"{resolved_value} ({description})"
else:
# Catalog not available - log warning in debug
if debug_info is not None:
if 'warnings' not in debug_info:
debug_info['warnings'] = []
debug_info['warnings'].append(f"Modifier |d used but catalog not available for {key}")
# Track resolution for debug
def _warn(msg: str):
if debug_info is not None:
resolved[key] = resolved_value[:100] + ('...' if len(resolved_value) > 100 else '')
debug_info.setdefault("warnings", []).append(msg)
return resolved_value
else:
# Keep placeholder if no value found
row = get_catalog_row_for_key(catalog, key) if catalog else None
if want_x and not want_d:
if key not in variables:
if debug_info is not None:
unresolved.append(key)
return match.group(0)
expl = (row.get("ai_caption") or "").strip() if row else ""
if not expl and catalog is None:
_warn(f"Modifier |x für {key}: Katalog fehlt (ai_caption).")
out = expl
if debug_info is not None:
resolved[key] = out[:100] + ("..." if len(out) > 100 else "")
return out
if key not in variables:
if debug_info is not None:
unresolved.append(key)
return match.group(0)
value = variables[key]
if isinstance(value, (dict, list)):
resolved_value = json.dumps(value, ensure_ascii=False)
else:
resolved_value = str(value)
if not want_d and not want_x:
out = resolved_value
if debug_info is not None:
resolved[key] = out[:100] + ("..." if len(out) > 100 else "")
return out
parts = [resolved_value]
if want_d:
if row:
desc = (row.get("description") or "").strip()
if desc:
parts.append(desc)
else:
_warn(f"Modifier |d für {key}: Katalog fehlt (description).")
if want_x:
expl = (row.get("ai_caption") or "").strip() if row else ""
if expl:
parts.append(expl)
elif catalog is not None:
_warn(f"Modifier |x (mit |d) für {key}: ai_caption leer.")
out = "".join(parts)
if debug_info is not None:
resolved[key] = out[:100] + ("..." if len(out) > 100 else "")
return out
result = re.sub(r'\{\{([^}]+)\}\}', replacer, template)
# Store debug info
@ -464,7 +487,7 @@ async def execute_prompt_with_data(
'today': datetime.now().strftime('%Y-%m-%d')
}
# Load placeholder catalog for |d modifier support
# Load placeholder catalog for |d / |x Modifier
try:
catalog = get_placeholder_catalog(profile_id)
except Exception as e:

View File

@ -35,7 +35,8 @@ from data_layer.nutrition_metrics import (
get_nutrition_average_data,
get_protein_targets_data,
get_protein_adequacy_data,
get_macro_consistency_data
get_macro_consistency_data,
get_energy_balance_data,
)
from data_layer.activity_metrics import (
get_activity_summary_data,
@ -118,7 +119,7 @@ def get_weight_trend_chart(
"""
profile_id = session['profile_id']
# Get structured data from data layer
# Get structured data from data layer (includes series — no second weight_log query)
trend_data = get_weight_trend_data(profile_id, days)
# Early return if insufficient data
@ -136,22 +137,12 @@ def get_weight_trend_chart(
}
}
# 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]
series = trend_data.get("series") or []
labels = [
pt["date"].isoformat() if hasattr(pt["date"], "isoformat") else str(pt["date"])
for pt in series
]
values = [pt["weight"] for pt in series]
return {
"chart_type": "line",
@ -346,17 +337,20 @@ def get_energy_balance_chart(
"""
profile_id = session['profile_id']
balance_meta = get_energy_balance_data(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 date, kcal
"""SELECT date, SUM(kcal)::float AS kcal
FROM nutrition_log
WHERE profile_id=%s AND date >= %s AND kcal IS NOT NULL
GROUP BY date
ORDER BY date""",
(profile_id, cutoff)
(profile_id, cutoff),
)
rows = cur.fetchall()
@ -374,7 +368,21 @@ def get_energy_balance_chart(
}
}
# Prepare data
estimated_tdee = balance_meta.get("estimated_tdee") or 0
if estimated_tdee <= 0:
return {
"chart_type": "line",
"data": {
"labels": [],
"datasets": []
},
"metadata": {
"confidence": "insufficient",
"data_points": len(rows),
"message": "Kein Gewicht für TDEE-Schätzung (weight_log erforderlich)"
}
}
labels = []
daily_values = []
avg_7d = []
@ -384,23 +392,19 @@ def get_energy_balance_chart(
labels.append(row['date'].isoformat())
daily_values.append(safe_float(row['kcal']))
# 7d rolling average
start_7d = max(0, i - 6)
window_7d = [safe_float(rows[j]['kcal']) for j in range(start_7d, i + 1)]
avg_7d.append(round(sum(window_7d) / len(window_7d), 1) if window_7d else None)
# 14d rolling average
start_14d = max(0, i - 13)
window_14d = [safe_float(rows[j]['kcal']) for j in range(start_14d, i + 1)]
avg_14d.append(round(sum(window_14d) / len(window_14d), 1) if window_14d else None)
# Calculate TDEE (estimated, should come from profile)
# TODO: Calculate from profile (weight, height, age, activity level)
estimated_tdee = 2500.0
# Calculate deficit/surplus
avg_intake = sum(daily_values) / len(daily_values) if daily_values else 0
energy_balance = avg_intake - estimated_tdee
avg_intake = float(balance_meta.get("avg_intake") or (sum(daily_values) / len(daily_values) if daily_values else 0))
energy_balance = float(balance_meta.get("energy_balance") or (avg_intake - estimated_tdee))
balance_status = balance_meta.get("status") or (
"deficit" if energy_balance < -200 else "surplus" if energy_balance > 200 else "maintenance"
)
datasets = [
{
@ -443,8 +447,7 @@ def get_energy_balance_chart(
}
]
from data_layer.utils import calculate_confidence
confidence = calculate_confidence(len(rows), days, "general")
confidence = balance_meta.get("confidence") or "low"
return {
"chart_type": "line",
@ -458,7 +461,7 @@ def get_energy_balance_chart(
"avg_kcal": round(avg_intake, 1),
"estimated_tdee": estimated_tdee,
"energy_balance": round(energy_balance, 1),
"balance_status": "deficit" if energy_balance < -200 else "surplus" if energy_balance > 200 else "maintenance",
"balance_status": balance_status,
"first_date": rows[0]['date'],
"last_date": rows[-1]['date']
})
@ -587,11 +590,12 @@ def get_protein_adequacy_chart(
cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
cur.execute(
"""SELECT date, protein_g
"""SELECT date, SUM(protein_g)::float AS protein_g
FROM nutrition_log
WHERE profile_id=%s AND date >= %s AND protein_g IS NOT NULL
GROUP BY date
ORDER BY date""",
(profile_id, cutoff)
(profile_id, cutoff),
)
rows = cur.fetchall()
@ -687,7 +691,6 @@ def get_protein_adequacy_chart(
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 daily_values if target_low <= v <= target_high)
return {
@ -978,16 +981,23 @@ def get_nutrition_adherence_score(
# Get nutrition data
cur.execute(
"""SELECT COUNT(*) as cnt,
AVG(kcal) as avg_kcal,
STDDEV(kcal) as std_kcal,
AVG(protein_g) as avg_protein,
AVG(carbs_g) as avg_carbs,
AVG(fat_g) as avg_fat
FROM nutrition_log
WHERE profile_id=%s AND date >= %s
AND kcal IS NOT NULL""",
(profile_id, cutoff)
"""WITH daily AS (
SELECT date,
COALESCE(SUM(kcal), 0)::float AS dk,
COALESCE(SUM(protein_g), 0)::float AS dp,
COALESCE(SUM(carbs_g), 0)::float AS dc,
COALESCE(SUM(fat_g), 0)::float AS df FROM nutrition_log
WHERE profile_id=%s AND date >= %s AND kcal IS NOT NULL
GROUP BY date
)
SELECT COUNT(*)::int AS cnt,
AVG(dk) AS avg_kcal,
STDDEV(dk) AS std_kcal,
AVG(dp) AS avg_protein,
AVG(dc) AS avg_carbs,
AVG(df) AS avg_fat
FROM daily""",
(profile_id, cutoff),
)
stats = cur.fetchone()

View File

@ -17,10 +17,11 @@ from models import (
PromptCreate, PromptUpdate, PromptGenerateRequest,
PipelineConfigCreate, PipelineConfigUpdate
)
from prompt_executor import resolve_placeholders as resolve_prompt_placeholders
from placeholder_resolver import (
resolve_placeholders,
get_unknown_placeholders,
get_placeholder_example_values,
format_value_with_d_modifier,
get_available_placeholders,
get_placeholder_catalog
)
@ -431,7 +432,13 @@ def preview_prompt(data: dict, session: dict=Depends(require_auth)):
template = data.get('template', '')
profile_id = session['profile_id']
resolved = resolve_placeholders(template, profile_id)
catalog = get_placeholder_catalog(profile_id)
processed = get_placeholder_example_values(profile_id)
variables = {
k.replace('{{', '').replace('}}', ''): v
for k, v in processed.items()
}
resolved = resolve_prompt_placeholders(template, variables, None, catalog)
unknown = get_unknown_placeholders(template)
return {
@ -457,8 +464,8 @@ def export_placeholder_values(session: dict = Depends(require_auth)):
"""
Export all available placeholders with their current resolved values.
Returns JSON export suitable for download with all placeholders
resolved for the current user's profile.
Pro Zeile: value = {{key}}, example = Vorschau {{key|d}} (Wert description),
ai_caption = Text für {{key|x}} (Erklärung ohne Wert). JSON für das aktive Profil.
"""
from datetime import datetime
profile_id = session['profile_id']
@ -486,12 +493,16 @@ def export_placeholder_values(session: dict = Depends(require_auth)):
export_data['placeholders_by_category'][category] = []
for item in items:
key = item['key'].replace('{{', '').replace('}}', '')
export_data['placeholders_by_category'][category].append({
raw_val = cleaned_values.get(key, 'nicht verfügbar')
row = {
'key': item['key'],
'description': item['description'],
'value': cleaned_values.get(key, 'nicht verfügbar'),
'example': item.get('example')
})
'value': raw_val,
'example': format_value_with_d_modifier(str(raw_val), item),
}
if item.get('ai_caption'):
row['ai_caption'] = item['ai_caption']
export_data['placeholders_by_category'][category].append(row)
# Also include flat list for easy access
export_data['all_placeholders'] = cleaned_values
@ -583,8 +594,14 @@ def export_placeholder_values_extended(
if 'value_raw' not in metadata.unresolved_fields:
metadata.unresolved_fields.append('value_raw')
# Check availability
if value in ['nicht verfügbar', 'nicht genug Daten', '[Fehler:', '[Nicht']:
# Check availability (Resolver liefert oft „nicht verfügbar — <Grund>“)
sv = str(value)
if (
sv in ['nicht verfügbar', 'nicht genug Daten']
or sv.startswith('nicht verfügbar —')
or sv.startswith('[Fehler:')
or sv.startswith('[Nicht')
):
metadata.available = False
metadata.missing_reason = value
else:
@ -653,11 +670,12 @@ def export_placeholder_values_extended(
export_data['legacy']['placeholders_by_category'][category] = []
for item in items:
key = item['key'].replace('{{', '').replace('}}', '')
raw_val = cleaned_values.get(key, 'nicht verfügbar')
export_data['legacy']['placeholders_by_category'][category].append({
'key': item['key'],
'description': item['description'],
'value': cleaned_values.get(key, 'nicht verfügbar'),
'example': item.get('example')
'value': raw_val,
'example': format_value_with_d_modifier(str(raw_val), item),
})
# Fill metadata flat

65
backend/test_export.py Normal file
View File

@ -0,0 +1,65 @@
#!/usr/bin/env python3
"""Test export-all endpoint"""
import sys
sys.path.insert(0, '/app')
from datetime import datetime
from db import get_db, get_cursor, r2d
print("Testing export-all logic...")
with get_db() as conn:
cur = get_cursor(conn)
cur.execute("SELECT * FROM ai_prompts ORDER BY sort_order, slug")
prompts = [r2d(row) for row in cur.fetchall()]
print(f"Found {len(prompts)} prompts")
# Convert to export format (clean up DB-specific fields)
export_data = []
for idx, p in enumerate(prompts):
print(f"\nProcessing prompt {idx+1}: {p.get('slug')}")
try:
export_item = {
'slug': p['slug'],
'name': p['name'],
'display_name': p.get('display_name'),
'description': p.get('description'),
'type': p.get('type', 'pipeline'),
'category': p.get('category', 'ganzheitlich'),
'template': p.get('template'),
'stages': p.get('stages'),
'output_format': p.get('output_format', 'text'),
'output_schema': p.get('output_schema'),
'question_augmentations': p.get('question_augmentations'),
'graph_data': p.get('graph_data'),
'active': p.get('active', True),
'sort_order': p.get('sort_order', 0)
}
export_data.append(export_item)
print(f" ✓ OK")
except Exception as e:
print(f" ✗ ERROR: {type(e).__name__}: {e}")
import traceback
traceback.print_exc()
break
print(f"\n\nSuccessfully processed {len(export_data)} prompts")
# Try to create the response dict
try:
result = {
'export_date': datetime.now().isoformat(),
'count': len(export_data),
'prompts': export_data
}
print("✓ Result dict created successfully")
# Try JSON serialization
import json
json_str = json.dumps(result)
print(f"✓ JSON serialization OK, length: {len(json_str)}")
except Exception as e:
print(f"✗ ERROR creating result: {type(e).__name__}: {e}")
import traceback
traceback.print_exc()

View File

@ -44,6 +44,15 @@ def test_value_raw_json():
assert not success
assert val is None
# Resolver-Fehlerhülle (kein verwertbares JSON für Charts)
val, success = extract_value_raw(
'{"_available": false, "_reason": "test"}',
OutputType.JSON,
PlaceholderType.RAW_DATA,
)
assert not success
assert val is None
def test_value_raw_number():
"""Numeric outputs must extract numbers without units."""
@ -66,6 +75,13 @@ def test_value_raw_number():
val, success = extract_value_raw('nicht verfügbar', OutputType.NUMBER, PlaceholderType.ATOMIC)
assert not success
val, success = extract_value_raw(
'nicht verfügbar — Keine Messungen (detail)',
OutputType.NUMBER,
PlaceholderType.ATOMIC,
)
assert not success
def test_value_raw_markdown():
"""Markdown outputs keep as string."""

View File

@ -0,0 +1,104 @@
"""Tests für {{key|d}}, {{key|x}}, ai_caption und Unbekannt-Erkennung."""
from placeholder_registry import (
PlaceholderMetadata,
PlaceholderType,
OutputType,
build_ai_placeholder_caption,
)
import placeholder_resolver as pr
from placeholder_resolver import format_value_with_d_modifier
from prompt_executor import resolve_placeholders
def test_build_ai_caption_is_explanation_only():
m = PlaceholderMetadata(
key="test_x",
category="Test",
description="Kurzbeschreibung",
resolver_module="m",
resolver_function="f",
semantic_contract="Lang Vertrag " * 50,
business_meaning="Kernbedeutung für die KI.",
unit="g/day",
placeholder_type=PlaceholderType.INTERPRETED,
output_type=OutputType.NUMERIC,
)
cap = build_ai_placeholder_caption(m)
assert "Kernbedeutung" in cap
assert "Kurzbeschreibung" not in cap
def test_build_ai_caption_protein_avg_no_description_prefix():
m = PlaceholderMetadata(
key="protein_avg",
category="Ernährung",
description="Durchschn. Protein in g (30d)",
resolver_module="m",
resolver_function="f",
business_meaning="Zentraler Placeholder für Muskelerhalt.",
unit="g/day",
placeholder_type=PlaceholderType.INTERPRETED,
output_type=OutputType.NUMERIC,
)
cap = build_ai_placeholder_caption(m)
assert cap.startswith("Zentraler Placeholder")
assert "Durchschn. Protein" not in cap
assert "Technischer Bezug" not in cap
def test_build_ai_caption_score_adds_scale():
m = PlaceholderMetadata(
key="test_score",
category="Test",
description="Score",
resolver_module="m",
resolver_function="f",
business_meaning="Gewichteter Gesamtscore.",
unit="Score (0-100)",
placeholder_type=PlaceholderType.SCORE,
output_type=OutputType.NUMERIC,
)
cap = build_ai_placeholder_caption(m)
assert "0100" in cap or "0-100" in cap
assert "Gewichteter" in cap
def test_placeholder_token_regex_optional_modifier():
m0 = pr._PLACEHOLDER_TOKEN_RE.search("{{fat_avg}}")
assert m0 and m0.group(1) == "fat_avg" and m0.group(2) is None
m1 = pr._PLACEHOLDER_TOKEN_RE.search("{{fat_avg|d}}")
assert m1 and m1.group(1) == "fat_avg" and m1.group(2).strip() == "d"
m2 = pr._PLACEHOLDER_TOKEN_RE.search("{{ protein_avg | d }}")
assert m2 and m2.group(1) == "protein_avg" and m2.group(2).strip() == "d"
m3 = pr._PLACEHOLDER_TOKEN_RE.search("{{k|d,x}}")
assert m3 and m3.group(1) == "k" and m3.group(2).strip() == "d,x"
def test_get_unknown_placeholders_strips_modifier():
unk = pr.get_unknown_placeholders("{{not_a_real_key|d}}")
assert set(unk) == {"not_a_real_key"}
def test_format_value_with_d_modifier_uses_description_only():
row = {
"key": "protein_avg",
"description": "Durchschn. Protein in g (30d)",
"ai_caption": "Nur für |x",
}
out = format_value_with_d_modifier("119g/Tag", row)
assert out == "119g/Tag — Durchschn. Protein in g (30d)"
assert "Nur für |x" not in out
def test_format_value_with_d_modifier_falls_back_to_description():
row = {"description": "Nur Beschreibung", "key": "x"}
assert format_value_with_d_modifier("42", row) == "42 — Nur Beschreibung"
def test_prompt_executor_modifiers_d_x_combined():
catalog = {"E": [{"key": "p", "description": "Desc", "ai_caption": "Expl"}]}
v = {"p": "99"}
assert resolve_placeholders("{{p|d}}", v, None, catalog) == "99 — Desc"
assert resolve_placeholders("{{p|x}}", v, None, catalog) == "Expl"
assert resolve_placeholders("{{p|d,x}}", v, None, catalog) == "99 — Desc — Expl"
assert resolve_placeholders("{{p}}", v, None, catalog) == "99"

View File

@ -18,7 +18,7 @@ This document establishes **mandatory governance rules** for placeholder managem
## 2. Scope
These guidelines apply to:
- All 116 existing placeholders
- All 114 existing placeholders (canonical: `PLACEHOLDER_MAP`)
- All new placeholders
- All modifications to existing placeholders
- All placeholder deprecations

View File

@ -79,7 +79,7 @@ curl -s -H "X-Auth-Token: $TOKEN" \
**Expected response:**
```json
{
"total_placeholders": 116,
"total_placeholders": 114,
"available": 98,
"missing": 18,
"by_type": {

View File

@ -9,12 +9,12 @@
## Executive Summary
This document summarizes the complete implementation of the normative placeholder metadata system for Mitai Jinkendo. The system provides a comprehensive, standardized framework for managing, documenting, and validating all 116 placeholders in the system.
This document summarizes the complete implementation of the normative placeholder metadata system for Mitai Jinkendo. The system provides a comprehensive, standardized framework for managing, documenting, and validating all 114 placeholders in the system.
**Key Achievements:**
- ✅ Complete metadata schema (normative compliant)
- ✅ Automatic metadata extraction
- ✅ Manual curation for 116 placeholders
- ✅ Manual curation for 114 placeholders
- ✅ Extended export API (non-breaking)
- ✅ Catalog generator (4 documentation files)
- ✅ Validation & testing framework
@ -75,7 +75,7 @@ This document summarizes the complete implementation of the normative placeholde
### 1.3 Complete Metadata Definitions
#### `backend/placeholder_metadata_complete.py` (220 lines, expandable to all 116)
#### `backend/placeholder_metadata_complete.py` (220 lines, expandable to all 114)
**Purpose:** Manually curated, authoritative metadata for all placeholders
@ -106,7 +106,7 @@ PlaceholderMetadata(
**Key Features:**
- Hand-curated for accuracy
- Complete for all 116 placeholders
- Complete for all 114 placeholders
- Serves as authoritative source
- Normative compliant
@ -285,7 +285,7 @@ pytest backend/tests/test_placeholder_metadata.py -v
v
┌─────────────────────────────────────────────────────────────┐
│ Complete Registry │
│ (116 placeholders with full metadata) │
│ (114 placeholders with full metadata) │
└──────────┬──────────────────────────────────────────────────┘
├──> Generation Scripts (generate_*.py)
@ -309,7 +309,7 @@ pytest backend/tests/test_placeholder_metadata.py -v
### 3.1 Metadata Extraction Flow
```
1. PLACEHOLDER_MAP (116 entries)
1. PLACEHOLDER_MAP (114 entries)
└─> extract_resolver_name()
└─> analyze_data_layer_usage()
└─> infer_type/time_window/output_type()
@ -468,7 +468,7 @@ curl -H "X-Auth-Token: <token>" \
# Output:
{
"total_placeholders": 116,
"total_placeholders": 114,
"available": 98,
"missing": 18,
"by_type": {
@ -599,7 +599,7 @@ The system is designed for extensibility:
## 8. Compliance Checklist
✅ **Normative Standard Compliance:**
- All 116 placeholders inventoried
- All 114 placeholders inventoried
- Complete metadata schema implemented
- Validation framework in place
- Non-breaking export API