"""Reproducible descriptive N-of-1 phase evaluation for Sprint 6F-C.""" from __future__ import annotations from datetime import date, timedelta import hashlib import json import math import sqlite3 from statistics import median, quantiles from typing import Any, Callable from dashboard_v5.association_engine import EVENT_BY_ID, event_days from dashboard_v5.metric_catalog_v2 import BY_ID_V2 from dashboard_v5.sprint6f_c_schema import METHOD_VERSION from dashboard_v5.comparison_contract import INFLUENCES as COMPARISON_INFLUENCES from dashboard_v5.comparison_contract import _influence as comparison_influence from dashboard_v5.observation_contract import PUBLIC_CONTRACT_VERSION DISCLAIMER = ( "Zeitliche persönliche Beobachtung – daraus lässt sich keine Ursache ableiten." ) def _json(raw: Any) -> Any: return json.loads(str(raw)) def _label(identifier: str) -> str: if identifier in EVENT_BY_ID: return EVENT_BY_ID[identifier].label item = BY_ID_V2.get(identifier) return item.label if item else "Nicht freigegebener Wert" def _public_question(row: sqlite3.Row) -> dict[str, Any]: return { "contract_version": PUBLIC_CONTRACT_VERSION, "id": str(row["id"]), "title": str(row["title"]), "question": str(row["question"]), "influences": _json(row["influences_json"]), "outcomes": _json(row["outcomes_json"]), "lag_min": int(row["lag_min"]), "lag_max": int(row["lag_max"]), "start_date": str(row["start_date"]), "end_date": str(row["end_date"]), "status": str(row["status"]), "note": str(row["note"]), "include_doctor": bool(row["include_doctor"]), "method_version": str(row["method_version"]), "created_at": str(row["created_at"]), "updated_at": str(row["updated_at"]), } def _public_phase(row: sqlite3.Row) -> dict[str, Any]: return { "id": str(row["id"]), "observation_id": str(row["observation_id"]), "phase_type": str(row["phase_type"]), "name": str(row["name"]), "start_date": str(row["start_date"]), "end_date": str(row["end_date"]), "behavior_goal": str(row["behavior_goal"]), "metrics": _json(row["metrics_json"]), "events": _json(row["events_json"]), "lag_min": int(row["lag_min"]), "lag_max": int(row["lag_max"]), "adherence": str(row["adherence"]), "notes": str(row["notes"]), "created_at": str(row["created_at"]), "updated_at": str(row["updated_at"]), } def list_observations( connection: sqlite3.Connection, statuses: set[str] | None = None ) -> list[dict[str, Any]]: connection.row_factory = sqlite3.Row rows = connection.execute( "SELECT * FROM personal_observations ORDER BY CASE status WHEN 'active' THEN 0 WHEN 'draft' THEN 1 WHEN 'paused' THEN 2 WHEN 'completed' THEN 3 ELSE 4 END,updated_at DESC LIMIT 200" ).fetchall() return [ _public_question(row) for row in rows if not statuses or str(row["status"]) in statuses ] def observation_detail( connection: sqlite3.Connection, observation_id: str ) -> dict[str, Any]: connection.row_factory = sqlite3.Row row = connection.execute( "SELECT * FROM personal_observations WHERE id=?", (observation_id,) ).fetchone() if row is None: raise KeyError("observation_not_found") question = _public_question(row) phases = [ _public_phase(item) for item in connection.execute( "SELECT * FROM personal_observation_phases WHERE observation_id=? ORDER BY start_date,id", (observation_id,), ) ] checkins = [ dict(item) for item in connection.execute( "SELECT day,phase_id,adherence,stress,sleep_disruption,infection,unusual_activity,travel,medication_change_fact,supplement_change_fact,note,updated_at FROM personal_observation_checkins WHERE observation_id=? ORDER BY day", (observation_id,), ) ] results = [ dict(item) for item in connection.execute( "SELECT id,result_version,method_version,configuration_hash,created_at FROM personal_observation_results WHERE observation_id=? ORDER BY result_version DESC", (observation_id,), ) ] return { **question, "phases": phases, "checkins": checkins, "results": results, "metric_labels": { item: _label(item) for item in set( question["influences"] + question["outcomes"] + [metric for phase in phases for metric in phase["metrics"]] ) }, "event_labels": { item: _label(item) for item in set( [event for phase in phases for event in phase["events"]] + [ item for item in question["influences"] + question["outcomes"] if item in EVENT_BY_ID ] ) }, "medical_statement": DISCLAIMER, } def _quartiles(values: list[float]) -> tuple[float, float, float]: middle = median(values) if len(values) == 1: return values[0], middle, values[0] q1, _q2, q3 = quantiles(values, n=4, method="inclusive") return q1, middle, q3 def _metric_summary( points: list[dict[str, Any]], start: date, end: date ) -> dict[str, Any]: valid = [] for point in points: try: day = date.fromisoformat(str(point.get("date", ""))[:10]) value = float(point.get("value")) except (ValueError, TypeError): continue if start <= day <= end and math.isfinite(value): valid.append((day.isoformat(), value)) valid.sort() expected = (end - start).days + 1 observed = len({item[0] for item in valid}) status = ( "insufficient" if observed < 7 else "descriptive" if observed < 14 else "exploratory" ) result: dict[str, Any] = { "documented_days": observed, "missing_days": max(0, expected - observed), "expected_days": expected, "coverage": round(observed / expected, 4), "status": status, "points": [{"date": day, "value": value} for day, value in valid], } if observed >= 7: q1, med, q3 = _quartiles([item[1] for item in valid]) result.update( { "median": round(med, 4), "q1": round(q1, 4), "q3": round(q3, 4), "iqr": round(q3 - q1, 4), } ) return result def evaluate_observation( connection: sqlite3.Connection, observation_id: str, series_loader: Callable[[str, date, date], dict[str, Any]], ) -> dict[str, Any]: detail = observation_detail(connection, observation_id) phases = detail["phases"] metric_ids = list( dict.fromkeys( detail["influences"] + detail["outcomes"] + [item for phase in phases for item in phase["metrics"]] ) ) if len(metric_ids) > 8: raise ValueError("too_many_metrics") result_phases = [] baseline: dict[str, float] = {} for phase in phases: start, end = ( date.fromisoformat(phase["start_date"]), date.fromisoformat(phase["end_date"]), ) summaries = {} for metric_id in metric_ids: if metric_id in EVENT_BY_ID: continue payload = series_loader(metric_id, start, end) summary = _metric_summary(list(payload.get("points", [])), start, end) summary.update( { "metric_id": metric_id, "label": _label(metric_id), "unit": str(payload.get("unit", "")), "source": str(payload.get("source", {}).get("type", "documented")), } ) if phase["phase_type"] == "baseline" and summary.get("median") is not None: baseline[metric_id] = float(summary["median"]) elif ( summary.get("median") is not None and metric_id in baseline and summary["status"] == "exploratory" ): summary["change_from_personal_baseline"] = round( float(summary["median"]) - baseline[metric_id], 4 ) summaries[metric_id] = summary event_ids = list( dict.fromkeys( [ item for item in detail["influences"] + detail["outcomes"] if item in EVENT_BY_ID ] + phase["events"] ) ) events = [] for event_id in event_ids: for day in event_days(connection, event_id, start, end): events.append( {"date": day, "event_id": event_id, "label": _label(event_id)} ) events.sort(key=lambda item: (item["date"], item["event_id"])) checkins = [ item for item in detail["checkins"] if phase["start_date"] <= item["day"] <= phase["end_date"] ] adherence = { key: sum(item["adherence"] == key for item in checkins) for key in ("yes", "partial", "no", "unknown") } factors = [] for item in checkins: for key in ("sleep_disruption", "infection", "unusual_activity", "travel"): if item[key] == "yes": factors.append({"date": item["day"], "type": key}) if item["stress"] in {"moderate", "high"}: factors.append({"date": item["day"], "type": "stress"}) if item["medication_change_fact"]: factors.append({"date": item["day"], "type": "medication_change_fact"}) if item["supplement_change_fact"]: factors.append({"date": item["day"], "type": "supplement_change_fact"}) expected = (end - start).days + 1 phase_status = ( "insufficient" if max( [item["documented_days"] for item in summaries.values()] or [len(checkins)] ) < 7 else "descriptive" if max( [item["documented_days"] for item in summaries.values()] or [len(checkins)] ) < 14 else "exploratory" ) result_phases.append( { **phase, "expected_days": expected, "checkin_days": len({item["day"] for item in checkins}), "missing_checkin_days": max( 0, expected - len({item["day"] for item in checkins}) ), "adherence": adherence, "metrics": summaries, "events": events, "event_count": len(events), "known_factors": factors, "summary_level": phase_status, "summary_label": "Datenbasis noch nicht ausreichend" if phase_status == "insufficient" else "Rein deskriptive Zusammenfassung" if phase_status == "descriptive" else "Explorativer Phasenvergleich", } ) config = { key: detail[key] for key in ( "id", "title", "question", "influences", "outcomes", "lag_min", "lag_max", "start_date", "end_date", "method_version", ) } hash_payload = json.dumps( {"configuration": config, "phases": phases}, ensure_ascii=True, sort_keys=True, separators=(",", ":"), ) return { "version": 1, "configuration": config, "phases": result_phases, "configuration_hash": hashlib.sha256(hash_payload.encode()).hexdigest(), "method_version": METHOD_VERSION, "medical_statement": DISCLAIMER, } def snapshot_payload(analysis: dict[str, Any]) -> dict[str, Any]: if analysis.get("contract_version") == PUBLIC_CONTRACT_VERSION: return { "configuration": analysis["plan"], "phases": [], "summary": { "contract_version": PUBLIC_CONTRACT_VERSION, "event_count": analysis["event_count"], "events_with_followup": analysis["events_with_followup"], "result_value_count": analysis["result_value_count"], "coverage": analysis["coverage"], "coverage_by_metric": analysis["coverage_by_metric"], "evaluable": analysis["evaluable"], "summaries": analysis["summaries"], "limitations": analysis["limitations"], "medical_statement": analysis["medical_statement"], }, "configuration_hash": analysis["configuration_hash"], } compact = [] for phase in analysis["phases"]: compact.append( { key: value for key, value in phase.items() if key not in {"events", "known_factors"} } | { "metrics": { metric: { key: value for key, value in summary.items() if key != "points" } for metric, summary in phase["metrics"].items() }, "event_count": phase["event_count"], "known_factor_count": len(phase["known_factors"]), } ) return { "configuration": analysis["configuration"], "phases": compact, "summary": { "phase_count": len(compact), "method_version": analysis["method_version"], "medical_statement": DISCLAIMER, }, "configuration_hash": analysis["configuration_hash"], } def _first_point( points: list[dict[str, Any]], first: date, last: date ) -> dict[str, Any] | None: candidates = [] for point in points: try: point_day = date.fromisoformat(str(point.get("date", ""))[:10]) raw_value = point.get("value") if raw_value is None: continue value = float(raw_value) except (TypeError, ValueError): continue if first <= point_day <= last and math.isfinite(value): candidates.append((point_day, value, point)) if not candidates: return None candidates.sort(key=lambda item: item[0]) point_day, value, source = candidates[0] return { "date": point_day.isoformat(), "value": value, "quality": source.get("quality"), } def _plan_status(persisted: str, evaluable: bool, ended: bool) -> str: if persisted == "draft": return "prepared" if persisted == "paused": return "paused" if persisted == "completed": return "completed" if persisted == "archived": return "archived" if evaluable: return "evaluable" return "insufficient_data" if ended else "collecting" def evaluate_plan( connection: sqlite3.Connection, observation_id: str, series_loader: Callable[[str, date, date], dict[str, Any]], *, today: date | None = None, ) -> dict[str, Any]: """Build the Sprint 7C-C event-centric, descriptive read model.""" detail = observation_detail(connection, observation_id) start = date.fromisoformat(detail["start_date"]) end = date.fromisoformat(detail["end_date"]) today_value = today or date.today() follow_end = min( end + timedelta(days=max(1, int(detail["lag_max"]))), today_value, ) influence_id = detail["influences"][0] allowed_influences = {item[0] for item in COMPARISON_INFLUENCES} if influence_id not in allowed_influences or len(detail["outcomes"]) > 2: raise ValueError("observation_plan_not_supported") events, _components = comparison_influence(connection, influence_id, start, end) events.sort(key=lambda item: (item["date"], item.get("time") or "", item["id"])) series: dict[str, dict[str, Any]] = {} for metric_id in detail["outcomes"]: payload = ( series_loader(metric_id, start, follow_end) if start <= follow_end else {"points": [], "unit": "", "source": {"type": "not_documented"}} ) series[metric_id] = { "label": payload.get("label") or _label(metric_id), "unit": str(payload.get("unit") or ""), "source": str(payload.get("source", {}).get("type") or "documented"), "points": list(payload.get("points") or []), } context_events: list[dict[str, Any]] = [] for context_id, _label_text, _kind in COMPARISON_INFLUENCES: try: loaded, _ = comparison_influence(connection, context_id, start, end) except Exception: continue context_events.extend(loaded) context_events.sort(key=lambda item: (item["date"], item.get("time") or "", item["id"])) observations = [] follow_values: dict[str, list[float]] = {metric: [] for metric in series} counted_measurements: dict[str, set[str]] = {metric: set() for metric in series} covered_by_metric: dict[str, int] = {metric: 0 for metric in series} followed_events = 0 for event in events: event_day = date.fromisoformat(event["date"]) outcomes = [] event_has_unique_followup = False for metric_id, metric in series.items(): same_day = influence_id in {"nutrition.profile", "nutrition.histamine"} and metric_id.startswith("symptom.") window_start = event_day if same_day else event_day + timedelta(days=max(1, detail["lag_min"])) window_end = event_day + timedelta(days=max(1, detail["lag_max"])) if window_end < window_start: window_end = window_start point = _first_point(metric["points"], window_start, window_end) measurement_key = point["date"] if point else None used_for_summary = bool( point is not None and measurement_key not in counted_measurements[metric_id] ) if point is not None: if used_for_summary: counted_measurements[metric_id].add(str(measurement_key)) follow_values[metric_id].append(float(point["value"])) covered_by_metric[metric_id] += 1 event_has_unique_followup = True outcomes.append( { "metric_id": metric_id, "label": metric["label"], "unit": metric["unit"], "source": metric["source"], "value": point["value"] if point else None, "date": point["date"] if point else None, "status": "documented" if point else "missing", "used_for_summary": used_for_summary, "duplicate_assignment": bool(point is not None and not used_for_summary), } ) context_end = max( [date.fromisoformat(item["date"]) for item in outcomes if item["date"]] or [event_day + timedelta(days=max(1, detail["lag_max"]))] ) companions = [ { "category": item["category"], "date": item["date"], "time": item.get("time"), "title": item["title"], "source": item["source"], } for item in context_events if item["id"] != event["id"] and event_day <= date.fromisoformat(item["date"]) <= context_end ] observations.append( { "event": event, "outcomes": outcomes, "missing_outcomes": [item["metric_id"] for item in outcomes if item["status"] == "missing"], "companion_events": companions, "detail_date": event["date"], } ) followed_events += int(event_has_unique_followup) result_value_count = sum(len(values) for values in follow_values.values()) event_count = len(events) coverage = followed_events / event_count if event_count else 0.0 coverage_by_metric = { metric_id: round(count / event_count, 4) if event_count else 0.0 for metric_id, count in covered_by_metric.items() } mapped_entries = sum(int(item.get("mapped_entries") or 0) for item in events) documented_entries = sum(int(item.get("documented_entries") or 0) for item in events) histamine_mapping_coverage = ( mapped_entries / documented_entries if documented_entries else None ) histamine_review_open = influence_id == "nutrition.histamine" and any( item.get("review_status") != "complete_mapping" for item in events ) eligible = ( event_count >= 5 and coverage >= 0.7 and result_value_count >= 5 and all(count >= 5 for count in covered_by_metric.values()) and all(value >= 0.7 for value in coverage_by_metric.values()) and not histamine_review_open ) summaries = [] for metric_id, metric in series.items(): follow = follow_values[metric_id] period = [] for point in metric["points"]: try: point_day = date.fromisoformat(str(point.get("date", ""))[:10]) value = float(point.get("value")) except (TypeError, ValueError): continue if start <= point_day <= end and math.isfinite(value): period.append(value) summary = { "metric_id": metric_id, "label": metric["label"], "unit": metric["unit"], "follow_value_count": len(follow), "period_value_count": len(period), } if eligible and follow and period: follow_median = float(median(follow)) period_median = float(median(period)) summary.update( { "follow_median": round(follow_median, 4), "period_median": round(period_median, 4), "numeric_difference": round(follow_median - period_median, 4), "range": [round(min(follow), 4), round(max(follow), 4)], } ) summaries.append(summary) limitations = ["Fehlende Werte bleiben unbekannt; es wird nicht interpoliert."] if event_count < 5: limitations.append("Weniger als fünf dokumentierte Einflussereignisse.") if coverage < 0.7: limitations.append("Bei weniger als 70 Prozent der Ereignisse liegt eine Folgemessung vor.") if result_value_count < 5: limitations.append("Im Vergleichszeitraum liegen weniger als fünf Ergebniswerte vor.") if any(count < 5 for count in covered_by_metric.values()): limitations.append("Für mindestens einen ausgewählten Beobachtungswert liegen weniger als fünf eigenständige Folgemessungen vor.") if any(value < 0.7 for value in coverage_by_metric.values()): limitations.append("Für mindestens einen ausgewählten Beobachtungswert liegt die Abdeckung unter 70 Prozent.") if histamine_review_open: limitations.append("Mindestens eine Histaminzuordnung ist noch offen; es wird keine Gruppenzusammenfassung gezeigt.") config = { key: detail[key] for key in ( "id", "title", "question", "influences", "outcomes", "lag_min", "lag_max", "start_date", "end_date", "status", "phases", "created_at", "updated_at" ) } digest = hashlib.sha256( json.dumps(config, ensure_ascii=True, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() return { "contract_version": PUBLIC_CONTRACT_VERSION, "timezone": "Europe/Zurich", "plan": config, "user_status": _plan_status(detail["status"], eligible, today_value > end), "expected_window": { "from_day_offset": detail["lag_min"], "to_day_offset": detail["lag_max"], "sleep_rule": "Die Schlafmessung wird der tatsächlich folgenden, auf ihrem lokalen Endtag geführten Schlafperiode zugeordnet.", "daily_rule": "Tageswerte werden dem nächsten lokal definierten Tag im Beobachtungsfenster zugeordnet.", }, "event_count": event_count, "events_with_followup": followed_events, "result_value_count": result_value_count, "coverage": round(coverage, 4), "coverage_by_metric": coverage_by_metric, "minimum_requirements": {"events": 5, "coverage": 0.7, "result_values": 5}, "evaluable": eligible, "summary_label": "Deskriptive Zusammenfassung" if eligible else "Noch nicht genügend Beobachtungen für eine Zusammenfassung.", "summaries": summaries, "observations": observations, "histamine": { "mapped_entries": mapped_entries, "documented_entries": documented_entries, "mapping_coverage": round(histamine_mapping_coverage, 4) if histamine_mapping_coverage is not None else None, "review_open": histamine_review_open, "statement": "Ernährungstage sind teilweise dokumentiert; es gibt keine Gruppe ‚histaminfrei‘.", } if influence_id == "nutrition.histamine" else None, "limitations": limitations, "comparison": { "metrics": detail["outcomes"], "influences": detail["influences"], "from": detail["start_date"], "to": follow_end.isoformat(), }, "configuration_hash": digest, "medical_statement": "Dies ist eine persönliche, deskriptive Beobachtung. Sie zeigt keine Ursache oder medizinisch bestätigte Wirkung.", }