Recipe 13.6: Python Implementation Example

pip install boto3 requests

import json
import logging
import datetime
from datetime import timezone
from decimal import Decimal
from typing import Optional

import boto3
import requests
from botocore.config import Config

# Structured logging. Never log PHI field values (patient names, IDs in
# combination with conditions, etc.)
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# Neptune SPARQL endpoint. This is the cluster endpoint, accessible only
# from within the VPC. Replace with your actual Neptune cluster endpoint.
NEPTUNE_ENDPOINT = "https://your-neptune-cluster.us-east-1.neptune.amazonaws.com:8182/sparql"

# DynamoDB table for storing care gap results.
RESULTS_TABLE = "care-gap-results"

# Retry config for boto3 calls (DynamoDB, S3).
BOTO3_RETRY_CONFIG = Config(retries={"max_attempts": 3, "mode": "adaptive"})

# Priority scoring thresholds
HIGH_PRIORITY_THRESHOLD = 7.0

# The ontology namespace for our guideline graph.
# All our custom classes and properties live under this prefix.
ONTOLOGY_NS = "http://healthcare-cookbook.example.org/ontology#"

# Prefixes used in SPARQL queries. Defining them once keeps queries readable.
SPARQL_PREFIXES = """
PREFIX ont: <http://healthcare-cookbook.example.org/ontology#>
PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
PREFIX owl: <http://www.w3.org/2002/07/owl#>
PREFIX xsd: <http://www.w3.org/2001/XMLSchema#>
"""

# Quality measure weights for priority scoring.
# Higher weight = more impact on Star Ratings or quality bonuses.
# These are illustrative. Real weights come from CMS measure specifications.
MEASURE_WEIGHTS = {
    "HEDIS-CDC-HbA1c": 1.5,       # Diabetes: HbA1c control (triple-weighted in Stars)
    "HEDIS-CDC-Eye": 1.2,          # Diabetes: eye exam
    "HEDIS-SPC": 1.3,              # Statin therapy for cardiovascular
    "HEDIS-BCS": 1.4,              # Breast cancer screening
    "HEDIS-COL": 1.3,              # Colorectal cancer screening
    "HEDIS-CCS": 1.0,              # Cervical cancer screening
    "HEDIS-CBP": 1.2,              # Controlling blood pressure
}

# Default weight for measures not in the lookup table.
DEFAULT_MEASURE_WEIGHT = 1.0

def load_guideline_ontology(neptune_endpoint: str) -> dict:
    """
    Load a small guideline ontology into Neptune via SPARQL UPDATE.

    This creates the condition hierarchy, recommendation nodes, and the
    relationships between them. In production, you'd use Neptune's bulk
    loader to load a full OWL/RDF file from S3. This inline approach
    demonstrates the structure.

    Returns:
        A dict with status and the number of triples inserted.
    """

    # The ontology defines:
    # 1. Condition class hierarchy (ICD-10 codes -> condition groups)
    # 2. Recommendation nodes with preconditions, actions, and frequencies
    # 3. Action nodes with CPT/LOINC codes that satisfy them

    insert_query = SPARQL_PREFIXES + """
    INSERT DATA {
        # --- Condition Hierarchy ---
        # This is where ontological reasoning pays off. By defining
        # Type2Diabetes as a subclass of DiabetesMellitus, any guideline
        # that applies to DiabetesMellitus automatically applies to
        # Type2Diabetes patients. No explicit rule needed.

        ont:ChronicCondition rdf:type owl:Class .
        ont:DiabetesMellitus rdf:type owl:Class ;
            rdfs:subClassOf ont:ChronicCondition .
        ont:Type2Diabetes rdf:type owl:Class ;
            rdfs:subClassOf ont:DiabetesMellitus ;
            ont:icd10Code "E11" .
        ont:Hypertension rdf:type owl:Class ;
            rdfs:subClassOf ont:ChronicCondition ;
            ont:icd10Code "I10" .
        ont:CardiovascularDisease rdf:type owl:Class ;
            rdfs:subClassOf ont:ChronicCondition .
        ont:CoronaryArteryDisease rdf:type owl:Class ;
            rdfs:subClassOf ont:CardiovascularDisease ;
            ont:icd10Code "I25" .

        # Exclusion conditions
        ont:Hospice rdf:type owl:Class ;
            ont:icd10Code "Z51.5" .
        ont:TerminalIllness rdf:type owl:Class .
        ont:Pregnancy rdf:type owl:Class ;
            ont:icd10Code "Z33" .

        # --- Actions (what should be done) ---
        ont:HbA1cTest rdf:type ont:Action ;
            ont:actionType "lab" ;
            ont:cptCode "83036" ;
            ont:loincCode "4548-4" ;
            ont:description "HbA1c Lab Test" .

        ont:RetinalExam rdf:type ont:Action ;
            ont:actionType "procedure" ;
            ont:cptCode "92004" ;
            ont:description "Dilated Retinal Exam" .

        ont:StatinTherapy rdf:type ont:Action ;
            ont:actionType "medication" ;
            ont:drugClass "statin" ;
            ont:description "Statin Therapy Initiation" .

        ont:BloodPressureCheck rdf:type ont:Action ;
            ont:actionType "procedure" ;
            ont:cptCode "99213" ;
            ont:loincCode "85354-9" ;
            ont:description "Blood Pressure Measurement" .

        # --- Recommendations (the guideline rules) ---
        # Each recommendation links preconditions to an action with a frequency.

        ont:RecDiabetesHbA1c rdf:type ont:Recommendation ;
            ont:recommendationId "rec-diabetes-hba1c" ;
            ont:appliesWhen ont:DiabetesMellitus ;
            ont:recommendedAction ont:HbA1cTest ;
            ont:frequencyMonths 6 ;
            ont:excludedBy ont:Hospice ;
            ont:excludedBy ont:TerminalIllness ;
            ont:measureId "HEDIS-CDC-HbA1c" ;
            ont:priority "high" .

        ont:RecDiabetesRetinal rdf:type ont:Recommendation ;
            ont:recommendationId "rec-diabetes-retinal" ;
            ont:appliesWhen ont:DiabetesMellitus ;
            ont:recommendedAction ont:RetinalExam ;
            ont:frequencyMonths 12 ;
            ont:excludedBy ont:Hospice ;
            ont:measureId "HEDIS-CDC-Eye" ;
            ont:priority "medium" .

        ont:RecASCVDStatin rdf:type ont:Recommendation ;
            ont:recommendationId "rec-ascvd-statin" ;
            ont:appliesWhen ont:DiabetesMellitus ;
            ont:appliesWhen ont:Hypertension ;
            ont:recommendedAction ont:StatinTherapy ;
            ont:frequencyMonths 0 ;
            ont:ageMinimum 40 ;
            ont:excludedBy ont:Pregnancy ;
            ont:measureId "HEDIS-SPC" ;
            ont:priority "high" .

        ont:RecHypertensionBP rdf:type ont:Recommendation ;
            ont:recommendationId "rec-hypertension-bp" ;
            ont:appliesWhen ont:Hypertension ;
            ont:recommendedAction ont:BloodPressureCheck ;
            ont:frequencyMonths 6 ;
            ont:excludedBy ont:Hospice ;
            ont:measureId "HEDIS-CBP" ;
            ont:priority "medium" .
    }
    """

    # Send the SPARQL UPDATE to Neptune.
    response = requests.post(
        neptune_endpoint,
        data={"update": insert_query},
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )

    if response.status_code == 200:
        logger.info("Ontology loaded successfully into Neptune")
        return {"status": "success", "triples_loaded": "~50"}
    else:
        logger.error("Failed to load ontology: %s %s", response.status_code, response.text)
        return {"status": "error", "detail": response.text}

def assemble_patient_facts(patient_id: str) -> dict:
    """
    Assemble a patient's clinical fact set for reasoning.

    In production, this queries multiple data sources:
    - Claims DB for diagnoses and procedures
    - EHR extract for problem lists
    - Lab feeds for recent results
    - Pharmacy data for medications

    Here we return a synthetic patient to demonstrate the pattern.
    The structure is what matters: conditions as ICD-10 codes, services
    with dates and CPT/LOINC codes, medications with fill dates.

    Returns:
        A dict representing the patient's current clinical state.
    """

    # Synthetic patient: 62-year-old male with Type 2 Diabetes and Hypertension.
    # Last HbA1c was 14 months ago. No retinal exam in 18 months.
    # No statin on medication list despite meeting criteria.
    # This patient should trigger 3 care gaps.

    if patient_id == "PAT-2026-00482":
        return {
            "patient_id": "PAT-2026-00482",
            "demographics": {
                "age": 62,
                "sex": "M",
                "enrollment_status": "active",
            },
            "conditions": [
                {"icd10_code": "E11.9", "description": "Type 2 Diabetes without complications"},
                {"icd10_code": "I10", "description": "Essential hypertension"},
            ],
            "recent_services": [
                # HbA1c done 14 months ago (outside the 6-month window)
                {"cpt_code": "83036", "loinc_code": "4548-4", "service_date": "2025-03-10"},
                # Blood pressure check 2 months ago (within 6-month window)
                {"cpt_code": "99213", "loinc_code": "85354-9", "service_date": "2026-03-20"},
                # Retinal exam 18 months ago (outside the 12-month window)
                {"cpt_code": "92004", "loinc_code": None, "service_date": "2024-11-20"},
            ],
            "medications": [
                {"drug_name": "Metformin", "drug_class": "biguanide", "fill_date": "2026-04-15"},
                {"drug_name": "Lisinopril", "drug_class": "ace_inhibitor", "fill_date": "2026-04-15"},
                # Note: no statin in the medication list
            ],
        }

    # Default: empty patient (no data found)
    return {
        "patient_id": patient_id,
        "demographics": {"age": 0, "sex": "unknown", "enrollment_status": "unknown"},
        "conditions": [],
        "recent_services": [],
        "medications": [],
    }

def find_applicable_recommendations(
    patient_facts: dict, neptune_endpoint: str
) -> list:
    """
    Query Neptune to find all guideline recommendations that apply to this patient.

    The SPARQL query uses ontological reasoning: if the patient has a condition
    that is a subclass of a condition referenced by a recommendation, the
    recommendation applies. This is the core value of the knowledge graph
    approach over flat rule tables.

    We also check:
    - Age constraints (if the recommendation has age_minimum/age_maximum)
    - Exclusion conditions (if the patient has any, skip the recommendation)

    Args:
        patient_facts: The patient's assembled clinical state.
        neptune_endpoint: The Neptune SPARQL endpoint URL.

    Returns:
        A list of applicable recommendations with their metadata.
    """

    # Extract the patient's ICD-10 code prefixes. We use prefixes because
    # ICD-10 codes have varying specificity (E11 vs E11.9 vs E11.65).
    # Our ontology maps at the 3-character level (E11 -> Type2Diabetes).
    patient_condition_codes = [
        c["icd10_code"][:3] for c in patient_facts["conditions"]
    ]

    if not patient_condition_codes:
        logger.info("Patient has no conditions. No recommendations apply.")
        return []

    patient_age = patient_facts["demographics"]["age"]

    # Build a SPARQL query that finds recommendations whose preconditions
    # are satisfied by this patient's conditions.
    #
    # The key insight: we use rdfs:subClassOf* (the transitive closure)
    # so that E11 (Type2Diabetes) matches guidelines targeting DiabetesMellitus.
    # Neptune evaluates this SPARQL property path at query time, traversing
    # the subclass hierarchy without requiring a separate reasoning engine.

    # Build a VALUES clause for the patient's condition codes.
    values_clause = " ".join(f'"{code}"' for code in patient_condition_codes)

    query = SPARQL_PREFIXES + f"""
    SELECT ?rec ?recId ?actionDesc ?freqMonths ?measureId ?priority ?condCode ?excludedByCode
    WHERE {{
        # Find recommendations
        ?rec rdf:type ont:Recommendation ;
             ont:recommendationId ?recId ;
             ont:recommendedAction ?action ;
             ont:frequencyMonths ?freqMonths ;
             ont:measureId ?measureId ;
             ont:priority ?priority ;
             ont:appliesWhen ?appliesCondition .

        ?action ont:description ?actionDesc .

        # Match patient conditions using subclass reasoning.
        # ?appliesCondition is the condition class the recommendation targets.
        # ?patientCondClass is the specific condition class matching the patient's code.
        # rdfs:subClassOf* means "is the same class or a subclass of."
        ?patientCondClass rdfs:subClassOf* ?appliesCondition ;
                          ont:icd10Code ?condCode .

        # Filter to only this patient's conditions
        VALUES ?condCode {{ {values_clause} }}

        # Age check (optional properties)
        OPTIONAL {{ ?rec ont:ageMinimum ?ageMin . }}
        FILTER (!BOUND(?ageMin) || {patient_age} >= ?ageMin)

        OPTIONAL {{ ?rec ont:ageMaximum ?ageMax . }}
        FILTER (!BOUND(?ageMax) || {patient_age} <= ?ageMax)

        # Get exclusion conditions (we'll check these in Python)
        OPTIONAL {{
            ?rec ont:excludedBy ?excludedClass .
            ?excludedClass ont:icd10Code ?excludedByCode .
        }}
    }}
    """

    # Execute the SPARQL query against Neptune.
    response = requests.post(
        neptune_endpoint,
        data={"query": query},
        headers={"Accept": "application/sparql-results+json"},
    )

    if response.status_code != 200:
        logger.error("Neptune query failed: %s", response.text)
        return []

    results = response.json().get("results", {}).get("bindings", [])

    # Process results: group by recommendation and check exclusions.
    recommendations = {}
    exclusions_by_rec = {}

    for row in results:
        rec_id = row["recId"]["value"]

        # Collect exclusion codes for each recommendation
        if "excludedByCode" in row and row["excludedByCode"]["value"]:
            exclusions_by_rec.setdefault(rec_id, set()).add(
                row["excludedByCode"]["value"]
            )

        # Store recommendation details (dedup by rec_id)
        if rec_id not in recommendations:
            recommendations[rec_id] = {
                "recommendation_id": rec_id,
                "action_needed": row["actionDesc"]["value"],
                "frequency_months": int(row["freqMonths"]["value"]),
                "measure_id": row["measureId"]["value"],
                "priority": row["priority"]["value"],
                "triggered_by_code": row["condCode"]["value"],
            }

    # Now check exclusions: if the patient has any excluded condition, remove
    # that recommendation from the applicable list.
    patient_code_prefixes = set(patient_condition_codes)

    applicable = []
    for rec_id, rec in recommendations.items():
        excluded_codes = exclusions_by_rec.get(rec_id, set())
        # Check if any of the patient's conditions match an exclusion
        if excluded_codes & patient_code_prefixes:
            logger.info(
                "Recommendation %s excluded: patient has %s",
                rec_id,
                excluded_codes & patient_code_prefixes,
            )
            continue
        applicable.append(rec)

    logger.info("Found %d applicable recommendations for patient", len(applicable))
    return applicable

def identify_gaps(
    applicable_recommendations: list,
    patient_facts: dict,
    evaluation_date: Optional[str] = None,
) -> list:
    """
    For each applicable recommendation, check if the action has been completed
    within the required frequency window. If not, it's a care gap.

    Args:
        applicable_recommendations: Output from find_applicable_recommendations.
        patient_facts: The patient's clinical fact set.
        evaluation_date: ISO date string (YYYY-MM-DD). Defaults to today.

    Returns:
        A list of care gap dicts, each describing a missing action.
    """

    if evaluation_date is None:
        eval_date = datetime.date.today()
    else:
        eval_date = datetime.date.fromisoformat(evaluation_date)

    gaps = []

    for rec in applicable_recommendations:
        freq_months = rec["frequency_months"]

        # frequency_months == 0 means "ongoing" (e.g., should be on a medication).
        # For medication-type actions, we check the medication list instead of services.
        if freq_months == 0:
            # Check if the patient is currently on the recommended medication class.
            # "Currently on" = filled within the last 90 days (approximate).
            action_desc = rec["action_needed"].lower()
            if "statin" in action_desc:
                has_med = any(
                    m["drug_class"] == "statin" for m in patient_facts["medications"]
                )
                if not has_med:
                    gaps.append({
                        "recommendation_id": rec["recommendation_id"],
                        "measure_id": rec["measure_id"],
                        "action_needed": rec["action_needed"],
                        "priority": rec["priority"],
                        "frequency": "ongoing",
                        "last_completed": "never",
                        "days_overdue": None,
                        "justification": [
                            f"Active condition: {rec['triggered_by_code']}"
                        ],
                    })
            continue

        # For time-based recommendations, calculate the cutoff date.
        # Any matching service after this date means the gap is closed.
        cutoff_date = eval_date - datetime.timedelta(days=freq_months * 30)

        # Find services that satisfy this recommendation.
        # Match by CPT code or LOINC code from the recommendation's action.
        # (In production, you'd query the action node in Neptune for its codes.
        # Here we use a simplified lookup.)
        action_codes = _get_action_codes(rec["action_needed"])

        matching_services = [
            s
            for s in patient_facts["recent_services"]
            if (
                s.get("cpt_code") in action_codes.get("cpt", [])
                or s.get("loinc_code") in action_codes.get("loinc", [])
            )
            and datetime.date.fromisoformat(s["service_date"]) >= cutoff_date
        ]

        if not matching_services:
            # Gap found. Find the most recent matching service ever (for context).
            all_matching = [
                s
                for s in patient_facts["recent_services"]
                if (
                    s.get("cpt_code") in action_codes.get("cpt", [])
                    or s.get("loinc_code") in action_codes.get("loinc", [])
                )
            ]

            last_completed = "never"
            days_overdue = None
            if all_matching:
                # Sort by date descending, take the most recent
                most_recent = max(all_matching, key=lambda s: s["service_date"])
                last_completed = most_recent["service_date"]
                last_date = datetime.date.fromisoformat(last_completed)
                # days_overdue = days since last completed minus the frequency window.
                # Example: last HbA1c 431 days ago, frequency 180 days -> 251 days overdue.
                frequency_days = freq_months * 30
                days_overdue = (eval_date - last_date).days - frequency_days

            gaps.append({
                "recommendation_id": rec["recommendation_id"],
                "measure_id": rec["measure_id"],
                "action_needed": rec["action_needed"],
                "priority": rec["priority"],
                "frequency": f"{freq_months} months",
                "last_completed": last_completed,
                "days_overdue": days_overdue,
                "justification": [
                    f"Active condition: {rec['triggered_by_code']}"
                ],
            })

    logger.info("Identified %d care gaps", len(gaps))
    return gaps

def _get_action_codes(action_description: str) -> dict:
    """
    Map an action description to its CPT and LOINC codes.

    In production, this information lives in the Neptune graph on the Action
    nodes. Here we use a simple lookup for the demo actions.
    """
    # This is the kind of thing that belongs in the graph, not in code.
    # We're doing it here to keep the example self-contained.
    lookup = {
        "HbA1c Lab Test": {"cpt": ["83036"], "loinc": ["4548-4"]},
        "Dilated Retinal Exam": {"cpt": ["92004", "92014"], "loinc": []},
        "Blood Pressure Measurement": {"cpt": ["99213"], "loinc": ["85354-9"]},
        "Statin Therapy Initiation": {"cpt": [], "loinc": []},
    }
    return lookup.get(action_description, {"cpt": [], "loinc": []})

def score_gaps(gaps: list, patient_facts: dict) -> list:
    """
    Assign a composite priority score to each gap for outreach prioritization.

    The score combines:
    - Base priority from the guideline (high=3, medium=2, low=1)
    - How overdue the action is (longer overdue = more urgent)
    - Patient comorbidity count (more conditions = higher risk)
    - Quality measure weight (some measures impact Star Ratings more)

    Args:
        gaps: List of care gap dicts from identify_gaps.
        patient_facts: Patient's clinical state (for comorbidity count).

    Returns:
        The same gaps list, sorted by composite_score descending, with
        composite_score added to each gap dict.
    """

    priority_map = {"high": 3, "medium": 2, "low": 1}
    comorbidity_count = len(patient_facts["conditions"])

    for gap in gaps:
        base_score = priority_map.get(gap["priority"], 1)

        # Overdue factor: cap at 2.0 to prevent extreme outliers
        if gap["days_overdue"] is not None and gap["days_overdue"] > 0:
            overdue_factor = min(gap["days_overdue"] / 90.0, 2.0)
        else:
            overdue_factor = 1.0  # "ongoing" gaps (like missing medication)

        # Risk factor based on comorbidity burden
        risk_factor = min(comorbidity_count / 3.0, 2.0)

        # Quality measure weight from our lookup table
        measure_weight = MEASURE_WEIGHTS.get(
            gap["measure_id"], DEFAULT_MEASURE_WEIGHT
        )

        # Composite score formula
        composite = (
            base_score * (1 + overdue_factor) * (1 + risk_factor * 0.3) * measure_weight
        )

        gap["composite_score"] = round(composite, 1)

    # Sort by composite score descending (most urgent first)
    gaps.sort(key=lambda g: g["composite_score"], reverse=True)

    return gaps

# Create DynamoDB resource
dynamodb = boto3.resource("dynamodb", config=BOTO3_RETRY_CONFIG)

def store_gap_results(
    patient_id: str,
    evaluation_date: str,
    scored_gaps: list,
    ontology_version: str = "guidelines-2026-v2.1",
) -> dict:
    """
    Write the care gap evaluation results to DynamoDB.

    The record is keyed by patient_id (partition key) and evaluation_date
    (sort key). This supports both point lookups ("what are patient X's
    current gaps?") and range queries ("show me all evaluations for patient X
    over the last year to track gap closure").

    Args:
        patient_id: The patient identifier.
        evaluation_date: ISO date string for this evaluation.
        scored_gaps: The scored and sorted gap list.
        ontology_version: Which guideline version was used (for audit).

    Returns:
        The full record that was written.
    """

    table = dynamodb.Table(RESULTS_TABLE)

    # Convert floats to Decimal for DynamoDB.
    # DynamoDB's SDK rejects Python floats. You must use Decimal for all
    # numeric values. This is a common gotcha that causes TypeError at runtime.
    dynamodb_gaps = []
    for gap in scored_gaps:
        ddb_gap = {k: v for k, v in gap.items()}
        if ddb_gap.get("composite_score") is not None:
            ddb_gap["composite_score"] = Decimal(str(ddb_gap["composite_score"]))
        if ddb_gap.get("days_overdue") is not None:
            ddb_gap["days_overdue"] = Decimal(str(ddb_gap["days_overdue"]))
        dynamodb_gaps.append(ddb_gap)

    record = {
        "patient_id": patient_id,
        "evaluation_date": evaluation_date,
        "gap_count": len(scored_gaps),
        "gaps": dynamodb_gaps,
        "evaluated_at": datetime.datetime.now(timezone.utc).isoformat(),
        "ontology_version": ontology_version,
    }

    table.put_item(Item=record)

    logger.info(
        "Stored %d gaps for patient %s (eval date: %s)",
        len(scored_gaps),
        patient_id,
        evaluation_date,
    )

    return record

def evaluate_patient(patient_id: str, evaluation_date: Optional[str] = None) -> dict:
    """
    Run the full care gap reasoning pipeline for one patient.

    This is the main entry point. In a batch deployment, Step Functions
    would fan out Lambda invocations, each calling this function for a
    different patient.

    Args:
        patient_id: The patient to evaluate.
        evaluation_date: ISO date string. Defaults to today.

    Returns:
        The stored evaluation record with all identified gaps.
    """

    if evaluation_date is None:
        evaluation_date = datetime.date.today().isoformat()

    # Step 2: Assemble patient facts from data sources.
    print(f"Step 2: Assembling facts for patient {patient_id}")
    patient_facts = assemble_patient_facts(patient_id)
    print(f"  Conditions: {len(patient_facts['conditions'])}")
    print(f"  Recent services: {len(patient_facts['recent_services'])}")
    print(f"  Medications: {len(patient_facts['medications'])}")

    # Step 3: Query Neptune for applicable recommendations.
    print("Step 3: Querying Neptune for applicable recommendations")
    applicable = find_applicable_recommendations(patient_facts, NEPTUNE_ENDPOINT)
    print(f"  Found {len(applicable)} applicable recommendations")

    # Step 4: Identify which recommendations are not satisfied (gaps).
    print("Step 4: Identifying care gaps")
    gaps = identify_gaps(applicable, patient_facts, evaluation_date)
    print(f"  Identified {len(gaps)} care gaps")

    # Step 5: Score and prioritize gaps.
    print("Step 5: Scoring and prioritizing gaps")
    scored_gaps = score_gaps(gaps, patient_facts)
    for gap in scored_gaps:
        print(f"  [{gap['composite_score']}] {gap['action_needed']} ({gap['priority']})")

    # Step 6: Store results in DynamoDB.
    print("Step 6: Storing results in DynamoDB")
    result = store_gap_results(patient_id, evaluation_date, scored_gaps)
    print(f"  Stored {result['gap_count']} gaps")

    return result

# Example: evaluate our synthetic patient.
if __name__ == "__main__":
    # Note: This requires Neptune to be running and the ontology loaded (Step 1).
    # For local testing without Neptune, you can mock find_applicable_recommendations
    # to return a static list.

    result = evaluate_patient(
        patient_id="PAT-2026-00482",
        evaluation_date="2026-05-15",
    )

    print("\n--- Final Result ---")
    print(json.dumps(result, indent=2, default=str))