"""Deterministic, synthetic-only SQLite fixture for Dashboard v5 acceptance work. This module deliberately knows nothing about a Dashboard v5 Python/UI API. It only materializes the canonical schema and representative rows at a caller-provided path. Dates are derived from an explicit anchor so freshness tests do not depend on wall-clock time. """ from __future__ import annotations import argparse import importlib.util import json import sqlite3 import tempfile from datetime import date, datetime, time, timedelta, timezone from pathlib import Path from typing import Final ROOT: Final = Path(__file__).resolve().parents[2] DEFAULT_SCHEMA: Final = ROOT / "database" / "schema.sql" SYMPTOM_DIMENSIONS: Final = ( "Aphthen/Mundulzera", "GI/Darm", "Müdigkeit/Fatigue", "Haut", "Augen", "Gelenke", "Vaskulär/Thrombose-Warnzeichen", ) SPRINT4B_LAGS: Final = (0, 1, 2, 3) SPRINT4B_PHASES: Final = ( "baseline_pre_treatment", "early_treatment", "stable_treatment", "unknown", ) SPRINT4B_PREDICTORS: Final = ("apple:heart_rate_variability", "nutrition:histamine_score") SPRINT4B_METHOD: Final = "spearman_7d_block_permutation_complete_case_phase_stratified_bh_v1" SPRINT4B_BASE_FLAGS: Final = [ "complete_case", "phase_stratified", "7d_block_permutation", "no_causality", ] SYNTHETIC_FIXTURE_MARKER: Final = "dashboard-v5-synthetic-fixture-v1" def sprint4b_correlation_fixture() -> dict[str, object]: """Return a deterministic engine-compatible, aggregate-only Sprint-4B grid.""" results: list[dict[str, object]] = [] for predictor in SPRINT4B_PREDICTORS: for lag in SPRINT4B_LAGS: for phase in SPRINT4B_PHASES: status = "computed" expected, eligible, observed = 24, 24 - lag, 18 - lag rho: float | None = round(0.42 - lag * 0.08, 2) p_value: float | None = round(0.01 + lag * 0.02, 2) q_value: float | None = round(0.03 + lag * 0.03, 2) flags = list(SPRINT4B_BASE_FLAGS) if predictor == SPRINT4B_PREDICTORS[0] and lag == 0 and phase == SPRINT4B_PHASES[0]: rho, p_value, q_value = 0.0, 0.0, 0.0 elif predictor == SPRINT4B_PREDICTORS[0] and lag == 3 and phase == "early_treatment": status, observed = "insufficient_n", 3 rho = p_value = q_value = None flags.append("low_n_or_variance") elif predictor == SPRINT4B_PREDICTORS[1] and lag == 2 and phase == "stable_treatment": status, expected, eligible, observed = "insufficient_coverage", 24, 10, 0 rho = p_value = q_value = None flags.append("low_target_coverage") results.append({ "id": {"predictor": predictor, "target": "symptom_total"}, "dimensions": {"lag_days": lag, "medication_phase": phase}, "sample": { "eligible_target_days": eligible, "expected_target_days": expected, "observed_pairs": observed, "missing_pairs": eligible - observed, "target_coverage": round(eligible / expected, 6), }, "statistics": {"rho": rho, "p_value": p_value, "q_value": q_value}, "status": status, "method": SPRINT4B_METHOD, "quality_flags": flags, }) return { "dimensions": { "lag_days": list(SPRINT4B_LAGS), "medication_phases": list(SPRINT4B_PHASES), "lag_direction": "predictor_d_to_target_d_plus_lag", }, "results": results, } def _iso_day(day: date) -> str: return day.isoformat() def _iso_noon(day: date) -> str: return datetime.combine(day, time(12), tzinfo=timezone.utc).isoformat() def build_dashboard_v5_fixture( db_path: str | Path, *, anchor_date: date = date(2026, 6, 15), schema_path: str | Path = DEFAULT_SCHEMA, ) -> dict[str, object]: """Create a new synthetic database and return its deterministic case manifest. ``db_path`` must not already exist. This fail-closed behavior prevents this helper from ever replacing a real database accidentally. """ target = Path(db_path) schema = Path(schema_path) if target.exists(): raise FileExistsError(f"refusing to replace existing database: {target}") if not schema.is_file(): raise FileNotFoundError(f"schema not found: {schema}") target.parent.mkdir(parents=True, exist_ok=True) current = anchor_date - timedelta(days=1) stale = anchor_date - timedelta(days=45) future = anchor_date + timedelta(days=2) complete_symptom_days = [anchor_date - timedelta(days=offset) for offset in range(1, 9)] incomplete_symptom_day = anchor_date - timedelta(days=9) no_data_day = anchor_date - timedelta(days=10) connection = sqlite3.connect(target) connection.row_factory = sqlite3.Row try: connection.executescript(schema.read_text(encoding="utf-8")) connection.execute( "CREATE TABLE dashboard_v5_synthetic_fixture (marker TEXT PRIMARY KEY)" ) connection.execute( "INSERT INTO dashboard_v5_synthetic_fixture(marker) VALUES (?)", (SYNTHETIC_FIXTURE_MARKER,), ) connection.executemany( """INSERT INTO apple_health_records (record_type,metric,start_date,end_date,value,value_text,unit, source_name,source_version,device,file_name,file_hash,record_hash, raw_json,imported_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", [ ("HKQuantityType", "heart_rate", _iso_noon(current), _iso_noon(current), 64.0, None, "count/min", "Synthetic Current Source", "1", "Synthetic Device", "synthetic-current.json", "fixture-current-file", "fixture-current-record", "{}", _iso_noon(current)), ("HKQuantityType", "step_count", _iso_noon(stale), _iso_noon(stale), 1234.0, None, "count", "Synthetic Stale Source", "1", "Synthetic Device", "synthetic-stale.json", "fixture-stale-file", "fixture-stale-record", "{}", _iso_noon(stale)), ("HKQuantityType", "resting_heart_rate", _iso_noon(future), _iso_noon(future), 61.0, None, "count/min", "Synthetic Future Source", "1", "Synthetic Device", "synthetic-future.json", "fixture-future-file", "fixture-future-record", "{}", _iso_noon(future)), ("HKQuantityType", "oxygen_saturation", _iso_noon(current), _iso_noon(current), None, "not measured", "%", "Synthetic Missing Value Source", "1", None, "synthetic-missing.json", "fixture-missing-file", "fixture-missing-record", "{}", _iso_noon(current)), ], ) explorer_rows = [] for offset in range(1, 23): day = anchor_date - timedelta(days=offset) for metric, value, unit in ( ("heart_rate_variability", 42.0 + offset, "ms"), ("resting_heart_rate", 60.0 + offset, "count/min"), ("sleep_analysis", 6.5 + offset / 10, "hr"), ("step_count", 5000.0 + offset * 100, "count"), ): if metric == "sleep_analysis" and offset == 4: continue key = f"fixture-explorer-{metric}-{offset}" explorer_rows.append(( "HKQuantityType", metric, _iso_noon(day), _iso_noon(day), value, None, unit, "Synthetic Explorer Source", "1", "Synthetic Device", f"{key}.json", key, f"{key}-record", "{}", _iso_noon(day), )) connection.executemany( """INSERT INTO apple_health_records (record_type,metric,start_date,end_date,value,value_text,unit, source_name,source_version,device,file_name,file_hash,record_hash, raw_json,imported_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", explorer_rows, ) correlation_results = sprint4b_correlation_fixture()["results"] fixture_timestamp = _iso_noon(anchor_date) assert isinstance(correlation_results, list) for result in correlation_results: assert isinstance(result, dict) connection.execute( """INSERT INTO multimodal_correlation_results (predictor,target,lag_days,medication_phase,n,eligible_target_days, expected_target_days,target_coverage,missing_pairs,rho,p_value,q_value,status, method,quality_flags,interpretation,computed_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( result["id"]["predictor"], result["id"]["target"], result["dimensions"]["lag_days"], result["dimensions"]["medication_phase"], result["sample"]["observed_pairs"], result["sample"]["eligible_target_days"], result["sample"]["expected_target_days"], result["sample"]["target_coverage"], result["sample"]["missing_pairs"], result["statistics"]["rho"], result["statistics"]["p_value"], result["statistics"]["q_value"], result["status"], result["method"], json.dumps(result["quality_flags"]), "Kein belastbares korrigiertes Signal; explorativ und keine Kausalität." if result["status"] == "computed" else ( "Unzureichende Zielabdeckung; keine Korrelation berechnet und keine Kausalität." if result["status"] == "insufficient_coverage" else "Unzureichende vollständige Datenpaare oder Varianz; keine Kausalität." ), fixture_timestamp, ), ) connection.executemany( """INSERT INTO laborwerte (parameter_name,wert,einheit,reference_min,reference_max, abnahme_datum,befund_datum,wert_original,quelle, validierungsstatus,source_type,reference_range_source, verified_against_original,provenance_note,ermittlung_datum) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", [ ( "C-Reaktives Protein (CRP)", value, "mg/L", ref_min, ref_max, day, day, value, "synthetic laboratory fixture", "validiert", "synthetic", "scanned_original", 1, "Synthetic test row; no person or source document.", fixture_timestamp, ) for day, value, ref_min, ref_max in ( ("2024-01-10", "4.2", "0", "5"), ("2024-10-15", "5.1", "0", "6"), ("2025-05-20", "7.4", None, None), ("2025-12-01", "8.8", "0", "5"), (_iso_day(stale), "12.5", "0", "5"), ) ] + [ ( "Faktor VIII", "118", "%", "50", "150", "2025-08-12", "2025-08-12", "118", "synthetic laboratory fixture", "validiert", "synthetic", "scanned_original", 1, "Synthetic single-observation test row; no person or source document.", fixture_timestamp, ) ], ) connection.executemany( "INSERT INTO medikamente(medikament_name,dosierung,anwendungsform,prescription_status,prescription_status_source,prescription_status_provenance,ermittlung_datum) VALUES(?,?,?,?,?,?,?)", [ ("SYNTHETIC_EVENT_ON_MEASUREMENT_GAP", "10 mg", "synthetic", "unknown", "fixture", "synthetic fixture", fixture_timestamp), ("SYNTHETIC_ADMINISTERED_MEDICATION", "10 mg", "synthetic", "active", "fixture", "synthetic fixture", fixture_timestamp), ("SYNTHETIC_PLANNED_MEDICATION_WITH_A_VERY_LONG_LABEL_FOR_RESPONSIVE_LAYOUT", "10 mg", "synthetic", "active", "fixture", "synthetic fixture", fixture_timestamp), ("Hyrimoz / Adalimumab", "40 mg", "subkutan", "unknown", None, None, fixture_timestamp), ("SYNTHETIC_CANCELLED_MEDICATION", "10 mg", "synthetic", "ended", "fixture", "synthetic fixture", fixture_timestamp), ], ) connection.executemany( """INSERT INTO medication_administrations (datum,medication_name,dose,route,event_type,scheduled_next_date,notes,source,created_at) VALUES(?,?,?,?,?,?,?,?,?)""", [ (_iso_day(anchor_date - timedelta(days=4)), "SYNTHETIC_EVENT_ON_MEASUREMENT_GAP", "10 mg", "synthetic", "administered", None, "timeline gap fixture", "fixture", fixture_timestamp), (_iso_day(current), "SYNTHETIC_ADMINISTERED_MEDICATION", "10 mg", "synthetic", "administered", None, "display as administered only", "fixture", fixture_timestamp), (_iso_day(anchor_date + timedelta(days=1)), "SYNTHETIC_PLANNED_MEDICATION_WITH_A_VERY_LONG_LABEL_FOR_RESPONSIVE_LAYOUT", "10 mg", "synthetic", "planned", _iso_day(anchor_date + timedelta(days=1)), "must not be represented as taken", "fixture", fixture_timestamp), (_iso_day(anchor_date + timedelta(days=1)), "SYNTHETIC_CANCELLED_MEDICATION", "10 mg", "synthetic", "cancelled", None, "must not be represented as taken", "fixture", fixture_timestamp), ], ) exposure_days = [anchor_date - timedelta(days=offset) for offset in (10, 9, 8)] meal_specs = ( ("breakfast", "synthetic breakfast component", 2), ("lunch", "synthetic lunch component", 2), ("dinner", "synthetic dinner component", 1), ("snack", "synthetic snack component", 1), ) nutrition_items = [ (1, "fixture", "complete", _iso_day(current), "lunch", "SYNTHETIC_COMPLETE_FOOD_MAPPING", "Synthetic Brand", 100, "g", 0, 0, 0, 0, "fixture-food-complete", fixture_timestamp, fixture_timestamp), (2, "fixture", "incomplete", _iso_day(current), "dinner", "SYNTHETIC_INCOMPLETE_FOOD_MAPPING_WITH_A_VERY_LONG_NAME_THAT_MUST_WRAP", None, None, "g", None, None, None, None, "fixture-food-incomplete", fixture_timestamp, fixture_timestamp), ] histamine_rows = [ (1, "synthetic cucumber", 0, "classified", "fixture_mapping_v1", "high", "synthetic source classification", fixture_timestamp), (2, None, None, "unknown", "fixture_unmapped", "low", "intentionally unmapped fixture row", fixture_timestamp), ] nutrient_rows = [ (1, "nutrient.fiber", 4.2, "g", "4.2"), (1, "nutrient.sugar", 2.0, "g", "2.0"), (1, "nutrient.vitamin_c", 20.0, "unknown", "20"), (2, "nutrient.sodium", 10.0, "mg", "0.01"), ] provenance_rows = [] for day_index, exposure_day in enumerate(exposure_days, start=1): for meal_index, (meal, canonical, score) in enumerate(meal_specs): item_id = 100 + (day_index * 10) + meal_index day_word = ("ONE", "TWO", "THREE")[day_index - 1] product_name = f"SYNTHETIC_{meal.upper()}_EXPOSURE_DAY_{day_word}" kcal = 220 + meal_index * 35 protein = 12 + meal_index carbs = 24 + meal_index * 2 fat = 8 + meal_index nutrition_items.append( (item_id, "fixture", f"exposure-{day_index}-{meal}", _iso_day(exposure_day), meal, product_name, "Synthetic Brand", 100, "g", kcal, protein, carbs, fat, f"fixture-exposure-{day_index}-{meal}", fixture_timestamp, fixture_timestamp) ) histamine_rows.append( (item_id, canonical, score, "classified", "fixture_mapping_v1", "high", "synthetic SIGHi source fixture v1", fixture_timestamp) ) nutrient_rows.extend( [ (item_id, "nutrient.fiber", 2.0 + meal_index, "g", str(2.0 + meal_index)), (item_id, "nutrient.sugar", 3.0 + meal_index, "g", str(3.0 + meal_index)), (item_id, "nutrient.saturated", 1.0 + meal_index / 2, "g", str(1.0 + meal_index / 2)), (item_id, "nutrient.sodium", 100.0 + meal_index * 20, "mg", str(100.0 + meal_index * 20)), ] ) action_id = f"fixture-mapping-{day_index}-{meal}" provenance_rows.append( (action_id, product_name.casefold(), "assign", product_name, canonical, score, "high", "sighi_reference", "SIGHi", "fixture-v1", "synthetic mapping provenance", 0, "unknown", "", "sighi_mapping_load_v2", _iso_day(exposure_day)) ) connection.executemany( """INSERT INTO nutrition_items (id,source,source_item_id,datum,meal,name,brand,amount,amount_unit, kcal,protein_g,carb_g,fat_g,item_hash,imported_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", nutrition_items, ) connection.executemany( """INSERT INTO nutrition_histamine_scores (item_id,canonical_food,sighi_score,traffic_light,tags,confidence,reason,scored_at) VALUES(?,?,?,?,?,?,?,?)""", histamine_rows, ) connection.executemany( """INSERT INTO nutrition_item_nutrients(item_id,nutrient_key,value,unit,raw_value) VALUES(?,?,?,?,?)""", nutrient_rows, ) connection.executemany( """INSERT OR IGNORE INTO histamine_food_rules (canonical_food,sighi_score,category,tags,notes,confidence,source,updated_at) VALUES(?,?,?,?,?,?,?,?)""", [(canonical, score, "synthetic", "fixture", "synthetic source rule", "high", "SIGHi fixture-v1", fixture_timestamp) for _, canonical, score in meal_specs], ) connection.executemany( "INSERT OR IGNORE INTO histamine_food_aliases(alias,canonical_food) VALUES(?,?)", [(f"synthetic {meal} alias", canonical) for meal, canonical, _ in meal_specs], ) connection.executemany( """INSERT INTO nutrition_mapping_provenance (action_id,normalized_name,decision,alias,canonical_food,sighi_score,confidence, mapping_method,source_label,source_version,note,ingredient_review_required, personal_tolerance_status,personal_tolerance_note,definition_version,affected_day) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", provenance_rows, ) connection.execute( """INSERT OR REPLACE INTO histamine_food_rules (canonical_food,sighi_score,category,confidence,source,updated_at) VALUES(?,1,'synthetic','high','synthetic_sighi_fixture','fixture-v1')""", ("synthetic reviewed food",), ) connection.execute( """INSERT INTO nutrition_review_queue (normalized_name,example_name,occurrence_count,first_seen,last_seen,suggested_canonical_food,suggested_score,reason,status,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?)""", ("synthetic incomplete food mapping with a very long name that must wrap", "SYNTHETIC_INCOMPLETE_FOOD_MAPPING_WITH_A_VERY_LONG_NAME_THAT_MUST_WRAP", 1, _iso_day(current), _iso_day(current), "synthetic reviewed food", 1, "synthetic missing mapping", "open", fixture_timestamp), ) worker_spec = importlib.util.spec_from_file_location( "dashboard_v5_fixture_worker", ROOT / "scripts" / "health" / "health_dashboard_action_worker.py", ) assert worker_spec and worker_spec.loader worker_module = importlib.util.module_from_spec(worker_spec) worker_spec.loader.exec_module(worker_module) worker_module.apply_document_schema(connection) worker_module.apply_media_schema(connection) importlib.import_module("dashboard_v5.sprint6i_c_schema").apply_schema(connection) importlib.import_module("dashboard_v5.medication_schema").apply_schema(connection) connection.execute( "UPDATE medikamente SET business_revision=printf('%064x', id) WHERE prescription_status IN ('active','paused','ended')" ) connection.execute( "UPDATE medication_public_identity_key SET key=? WHERE singleton=1", (bytes.fromhex("7cf0" * 16),), ) connection.execute( "UPDATE medication_schema_meta SET installed_at='2026-06-15 12:00:00'" ) recomputed_days = [*exposure_days, current] for nutrition_day in recomputed_days: worker_module.recompute_nutrition_day(connection, _iso_day(nutrition_day)) deterministic_timestamp = "2026-06-15 12:00:00" placeholders = ",".join("?" for _ in recomputed_days) day_values = [_iso_day(day) for day in recomputed_days] connection.execute( "UPDATE nutrition_mapping_provenance SET applied_at=?", (deterministic_timestamp,), ) connection.execute( f"UPDATE nutrition_daily_summary_v2 SET updated_at=? WHERE datum IN ({placeholders})", [deterministic_timestamp, *day_values], ) connection.execute( f"UPDATE nutrition_meal_summary SET updated_at=? WHERE datum IN ({placeholders})", [deterministic_timestamp, *day_values], ) if connection.execute( "SELECT 1 FROM sqlite_master WHERE type='table' AND name='dashboard_schema_migrations'" ).fetchone(): connection.execute( "UPDATE dashboard_schema_migrations SET applied_at=?", (deterministic_timestamp,), ) symptom_rows = [] aphthae_event_day = anchor_date - timedelta(days=6) for day_index, day in enumerate(complete_symptom_days): for dimension_index, dimension in enumerate(SYMPTOM_DIMENSIONS): if dimension == "Aphthen/Mundulzera": score = 2 if day == aphthae_event_day else 0 else: score = (day_index + dimension_index) % 4 symptom_rows.append((_iso_day(day), dimension, str(score), "daily_quick_score", "synthetic complete day", fixture_timestamp)) for dimension in SYMPTOM_DIMENSIONS[:-1]: symptom_rows.append((_iso_day(incomplete_symptom_day), dimension, "1", "daily_quick_score", "synthetic incomplete day", fixture_timestamp)) connection.executemany( "INSERT INTO symptom_log(datum,symptom,schwergrad,kontext,notizen,created_at) VALUES(?,?,?,?,?,?)", symptom_rows, ) connection.commit() fk_errors = connection.execute("PRAGMA foreign_key_check").fetchall() if fk_errors: raise RuntimeError(f"fixture violates foreign keys: {fk_errors}") except Exception: connection.close() target.unlink(missing_ok=True) raise finally: try: connection.close() except Exception: pass target.chmod(0o600) return { "anchor_date": _iso_day(anchor_date), "current_source_date": _iso_day(current), "stale_source_date": _iso_day(stale), "future_source_date": _iso_day(future), "complete_symptom_days": [_iso_day(day) for day in complete_symptom_days], "incomplete_symptom_day": _iso_day(incomplete_symptom_day), "no_data_day": _iso_day(no_data_day), "synthetic_only": True, } def validate_fixture(db_path: str | Path) -> dict[str, object]: """Return structural acceptance checks without importing application code.""" db = Path(db_path) connection = sqlite3.connect(f"file:{db}?mode=ro", uri=True) try: counts = { table: connection.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] for table in ( "apple_health_records", "laborwerte", "medication_administrations", "nutrition_items", "nutrition_histamine_scores", "symptom_log", ) } checks = { "integrity": connection.execute("PRAGMA integrity_check").fetchone()[0], "foreign_key_errors": connection.execute("PRAGMA foreign_key_check").fetchall(), "verified_lab_outliers": connection.execute( """SELECT COUNT(*) FROM laborwerte WHERE verified_against_original=1 AND CAST(wert AS REAL)>CAST(reference_max AS REAL)""" ).fetchone()[0], "complete_symptom_days": connection.execute( """SELECT COUNT(*) FROM (SELECT datum FROM symptom_log WHERE kontext='daily_quick_score' GROUP BY datum HAVING COUNT(DISTINCT symptom)=7)""" ).fetchone()[0], "medication_states": sorted(row[0] for row in connection.execute( "SELECT DISTINCT event_type FROM medication_administrations" )), "food_mapping_states": connection.execute( """SELECT SUM(canonical_food IS NOT NULL AND sighi_score IS NOT NULL), SUM(canonical_food IS NULL OR sighi_score IS NULL) FROM nutrition_histamine_scores""" ).fetchone(), } finally: connection.close() return {"counts": counts, "checks": checks} def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("output", nargs="?", help="new SQLite path; omit for a temporary fixture") parser.add_argument("--anchor-date", type=date.fromisoformat, default=date(2026, 6, 15)) args = parser.parse_args() if args.output: output = Path(args.output) else: output = Path(tempfile.mkdtemp(prefix="dashboard-v5-fixture-")) / "health.db" manifest = build_dashboard_v5_fixture(output, anchor_date=args.anchor_date) print(json.dumps({"path": str(output), "manifest": manifest, "validation": validate_fixture(output)}, indent=2, default=list)) if __name__ == "__main__": main()