Hospital Readmission Risk Pipeline
Build a FHIR ETL pipeline that scores patient readmission risk with LACE.
Introduction
30 Second Summary
Every year, hospitals face financial penalties when patients are readmitted within 30 days of discharge. Identifying which patients are most likely to come back is a data problem that health systems solve with automated pipelines.
In this project, you will build a Python pipeline that extracts live patient data from a healthcare API, scores each patient's readmission risk using a clinically validated index, and outputs a report identifying who needs the most attention.
What You'll Build
A complete healthcare data pipeline that connects to a real FHIR R4 server, transforms patient encounter data into an analytics database, and computes LACE index readmission risk scores.
By the end of this project, you'll have:
- A Python extraction script that pulls Patient, Encounter, and Condition resources from a live FHIR R4 API.
- A SQLite analytics database with three normalized tables storing transformed healthcare data.
- A LACE index scoring module that computes readmission risk per patient and categorizes them into Low, Moderate, and High severity.
- Secret Mission: Extend the pipeline to flag patients with high-risk comorbidity combinations and generate a care-coordination priority list.
Are there any prerequisites?
This project assumes familiarity with Python and SQL. You do not need any prior healthcare or FHIR experience.
Before We Start
Before We Start...
Set Up Your Project Environment
Before you can extract data from a healthcare server, you need a clean Python environment with the right dependencies.
In this step, you will create a project folder, install the two packages your pipeline needs (requests for HTTP calls and pandas for data manipulation), and confirm that the FHIR server is reachable.
In this step, get ready to:
- Create your project directory and Python virtual environment.
- Verify connectivity to the HAPI FHIR R4 public test server.
- Create the file structure for your pipeline modules.
Create project directory and virtual environment
A virtual environment isolates your project's packages from the rest of your system.
This means the specific versions you install here won't conflict with other Python projects on your machine.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir readmission-risk-pipeline
cd readmission-risk-pipeline- Confirm the folder exists by running:
lsYou should see an empty folder (no output) since you just created it.
- Create your virtual environment by running:
macOS
python3 -m venv venvWindows
python -m venv venv- Activate the virtual environment by running:
macOS
source venv/bin/activateWindows
venv\Scripts\activateYou should see (venv) appear at the beginning of your terminal prompt.
What does (venv) mean?
The (venv) prefix tells you that any pip install commands will install packages inside this project only.
If you close and reopen your terminal later, you will need to re-activate the environment before running the pipeline.
- Install the two project dependencies by running:
pip install requests==2.34.2 pandas==3.0.3What are these packages?
- requests handles all HTTP communication with the FHIR server.
- pandas gives you powerful data manipulation tools for building the cohort report.
- SQLite is already built into Python, so no extra install is needed for the database.
You should see output ending with Successfully installed followed by the package names.
Seeing a permission error or install failure?
- Make sure you see (venv) in your terminal prompt. If not, re-run the activation command above.
- If you get a network timeout, check your internet connection and try the install command again.
Why is pip failing to install my packages?
Verify FHIR server connectivity
The HAPI FHIR public test server is a free, read-only server containing synthetic patient data.
Before writing pipeline code, you need to confirm your machine can reach the server and receive a valid response.
What is a CapabilityStatement?
Every FHIR server exposes a /metadata endpoint that describes what the server supports.
The response is a CapabilityStatement resource listing available resource types, search parameters, and FHIR version. It is the standard way to verify a FHIR server is online and healthy.
- Test your connection to the FHIR server by running this command:
macOS
python3 -c "import requests; r = requests.get('http://hapi.fhir.org/baseR4/metadata', headers={'Accept': 'application/fhir+json'}); print(r.json()['resourceType'])"Windows
python -c "import requests; r = requests.get('http://hapi.fhir.org/baseR4/metadata', headers={'Accept': 'application/fhir+json'}); print(r.json()['resourceType'])"You should see CapabilityStatement printed in your terminal.
What just happened?
- Your script sent an HTTP GET request to http://hapi.fhir.org/baseR4/metadata.
- The server responded with a JSON document whose resourceType field is CapabilityStatement.
- This confirms network access, server availability, and that your requests package is installed correctly.
Not seeing CapabilityStatement?
- If you see ModuleNotFoundError: No module named 'requests', your virtual environment is not active. Re-run the activation command from the previous substep.
- If you see a connection timeout, the HAPI server may be temporarily down. Wait a minute and try again.
- If you see a JSONDecodeError, the server may be returning HTML instead of JSON. Make sure the URL matches exactly.
Help me debug my FHIR server connectivity test.
Create project file structure
Your pipeline will have four Python modules, each handling one stage of the workflow.
You will also need a data/ directory where the pipeline will store raw JSON extracts, the SQLite database, and the final CSV report.
- Create the four pipeline modules and the data directory by running these commands:
macOS
echo "# empty placeholder" > extract.py
echo "# empty placeholder" > transform.py
echo "# empty placeholder" > score.py
echo "# empty placeholder" > main.py
mkdir dataWindows
"# empty placeholder" | Out-File extract.py
"# empty placeholder" | Out-File transform.py
"# empty placeholder" | Out-File score.py
"# empty placeholder" | Out-File main.py
mkdir dataWhat does each file do?
- extract.py will pull raw Patient, Encounter, and Condition resources from the FHIR API.
- transform.py will flatten the nested JSON and load it into SQLite.
- score.py will compute the LACE readmission risk index for each patient.
- main.py will orchestrate the full pipeline and generate the final report.
- Confirm the files and folder were created by running:
lsYou should see data, extract.py, main.py, score.py, transform.py, and venv in the output.
Missing a file or folder?
- Double-check you are inside the readmission-risk-pipeline folder. Run pwd (macOS) or Get-Location (Windows) to confirm.
- If a file is missing, re-run the corresponding echo command from the code block above.
Help me fix my project file structure.
- Open the readmission-risk-pipeline folder in Cursor so you can edit your files in the next steps.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Cursor and press Enter to open it.
- Click File in the top menu bar.
- Click Open Folder.
- Navigate to your Desktop, select the readmission-risk-pipeline folder, then click Open.
You should see all four .py files and the data folder listed in the Explorer sidebar on the left.
Your environment is ready. Next up, you will build the extraction module that pulls live patient data from the FHIR API.
Extract Patient Data from the FHIR API
Your environment is ready and your FHIR server is confirmed reachable. Now it's time to pull real healthcare data from that server into your pipeline.
In this step, you'll write the extraction layer that queries the HAPI FHIR server for Patient, Encounter, and Condition resources. You'll handle pagination, respect the Minimum Necessary principle, and save raw JSON locally for transformation later.
In this step, get ready to:
- Document HIPAA awareness and regulatory context in your extraction script.
- Extract inpatient Encounter resources with FHIR Bundle pagination.
- Fetch related Patient and Condition resources for each encounter.
Understand HIPAA context and why we use synthetic data
Before writing any extraction code, you need to understand the regulatory landscape surrounding healthcare data.
What is HIPAA?
HIPAA (Health Insurance Portability and Accountability Act) protects electronic Protected Health Information (ePHI). Any individually identifiable health information is considered PHI.
Three key principles apply to data pipelines: the Minimum Necessary rule (only request data fields you actually need), required safeguards (administrative, physical, and technical protections), and Business Associate Agreements with data sources.
The HAPI FHIR public test server contains only synthetic data. Its homepage explicitly states: "Do not store any information here that contains personal health information." This makes it safe for learning, but your code should still demonstrate production-aware patterns.
- In Cursor, switch to the extract.py file that currently contains just a placeholder comment.
- Replace the placeholder with the following module docstring, imports, constants, and helper functions:
"""
FHIR Data Extraction Module
Extracts Patient, Encounter, and Condition resources from HAPI FHIR R4 server.
HIPAA AWARENESS NOTE:
This pipeline uses the HAPI FHIR PUBLIC TEST SERVER which contains ONLY synthetic data.
In a production environment, this code would require:
- Business Associate Agreement (BAA) with the data source
- Encryption in transit (HTTPS) and at rest
- Access controls and role-based permissions
- Audit logging of all data access
- Minimum necessary principle: only extract fields needed for analysis
Reference: HHS.gov HIPAA Security Rule (45 CFR Part 164)
"""
import json
import os
import requests
FHIR_BASE_URL = "http://hapi.fhir.org/baseR4"
DATA_DIR = "data"
MAX_ENCOUNTERS = 200
def ensure_data_dir():
"""Create data directory if it doesn't exist."""
os.makedirs(DATA_DIR, exist_ok=True)
def test_server_connectivity():
"""Verify the FHIR server is reachable by fetching its CapabilityStatement."""
url = f"{FHIR_BASE_URL}/metadata"
response = requests.get(url, headers={"Accept": "application/fhir+json"})
response.raise_for_status()
data = response.json()
assert data["resourceType"] == "CapabilityStatement", "Unexpected response"
print(f"Connected to: {data.get('implementation', {}).get('description', 'HAPI FHIR Server')}")
print(f"FHIR Version: {data.get('fhirVersion', 'unknown')}")
return TrueWhat does this code do?
- The module docstring documents exactly what safeguards a production pipeline would need. This is a pattern real health tech teams follow to show auditors that HIPAA was considered.
- FHIR_BASE_URL points to the HAPI FHIR R4 public test server. MAX_ENCOUNTERS caps extraction at 200 to keep the project manageable.
- test_server_connectivity() fetches the server's CapabilityStatement (a standard FHIR endpoint at /metadata) to confirm the server is responding before making data requests.
Extract Encounter resources with pagination
FHIR servers return search results as Bundle resources. When results span multiple pages, the Bundle includes a link entry with relation next pointing to the next page. Your extraction function needs to follow these links until it collects enough data.
- Add the encounter extraction function below test_server_connectivity() by copying the following code:
def extract_encounters():
"""
Extract finished inpatient encounters from the FHIR server.
Uses pagination to collect up to MAX_ENCOUNTERS resources.
Applies minimum necessary principle: only requesting inpatient, finished encounters.
"""
encounters = []
url = f"{FHIR_BASE_URL}/Encounter?class=IMP&status=finished&_count=50"
while url and len(encounters) < MAX_ENCOUNTERS:
print(f"Fetching encounters... ({len(encounters)} so far)")
response = requests.get(url, headers={"Accept": "application/fhir+json"})
response.raise_for_status()
bundle = response.json()
if "entry" in bundle:
for entry in bundle["entry"]:
if len(encounters) >= MAX_ENCOUNTERS:
break
encounters.append(entry["resource"])
# Follow pagination link
url = None
for link in bundle.get("link", []):
if link["relation"] == "next":
url = link["url"]
break
print(f"Extracted {len(encounters)} encounters")
return encountersWhat does this code do?
- The search query Encounter?class=IMP&status=finished&_count=50 applies the Minimum Necessary principle. It only requests inpatient (IMP) encounters that are finished, 50 at a time.
- The while loop follows FHIR Bundle pagination by looking for a link with relation == "next". When no next link exists, the loop ends.
- The inner loop breaks early once MAX_ENCOUNTERS is reached, preventing the script from pulling the entire server's dataset.
Why IMP for inpatient?
FHIR uses the v3-ActCode system for encounter classes. IMP means inpatient encounter. Other values include EMER (emergency), AMB (ambulatory), and VR (virtual). The LACE index specifically scores inpatient admissions, so we filter here.
Extract related Patient and Condition resources
Each Encounter references a Patient through its subject field. You need each patient's demographics and their full list of diagnosed conditions for the LACE scoring later.
- Add the patient extraction function below extract_encounters():
def extract_patients(patient_ids):
"""
Fetch Patient resources for a list of patient IDs.
Only retrieves demographics needed for the analysis (minimum necessary).
"""
patients = []
for i, patient_id in enumerate(patient_ids):
if i % 10 == 0:
print(f"Fetching patients... ({i}/{len(patient_ids)})")
url = f"{FHIR_BASE_URL}/Patient/{patient_id}"
try:
response = requests.get(url, headers={"Accept": "application/fhir+json"})
response.raise_for_status()
patients.append(response.json())
except requests.exceptions.HTTPError:
print(f" Warning: Patient {patient_id} not found, skipping")
continue
print(f"Extracted {len(patients)} patients")
return patientsWhat does this code do?
- For each patient ID, it sends a direct GET /Patient/{id} request. This is a FHIR read operation (fetching a single known resource).
- The try/except block gracefully handles missing patients. The HAPI test server data can have dangling references, so skipping 404s keeps the pipeline running.
- Progress printing every 10 patients gives you feedback during longer extraction runs.
- Add the condition extraction function below extract_patients():
def extract_conditions(patient_ids):
"""
Fetch Condition resources for each patient.
These map to comorbidities needed for the Charlson/LACE scoring.
"""
all_conditions = []
for i, patient_id in enumerate(patient_ids):
if i % 10 == 0:
print(f"Fetching conditions... ({i}/{len(patient_ids)})")
url = f"{FHIR_BASE_URL}/Condition?subject=Patient/{patient_id}&_count=100"
try:
response = requests.get(url, headers={"Accept": "application/fhir+json"})
response.raise_for_status()
bundle = response.json()
if "entry" in bundle:
for entry in bundle["entry"]:
all_conditions.append(entry["resource"])
except requests.exceptions.HTTPError:
continue
print(f"Extracted {len(all_conditions)} conditions")
return all_conditionsWhat does this code do?
- The search Condition?subject=Patient/{id}&_count=100 fetches all conditions (diagnoses) linked to a specific patient. These conditions feed into the Charlson Comorbidity Index later.
- Unlike patient fetches, conditions come back as a Bundle (search result) rather than a single resource. The code iterates through bundle["entry"] to collect each Condition resource.
- Add the orchestration functions and entry point below extract_conditions() to tie everything together:
def get_patient_ids_from_encounters(encounters):
"""Extract unique patient IDs from encounter resources."""
patient_ids = set()
for enc in encounters:
subject = enc.get("subject", {}).get("reference", "")
if subject.startswith("Patient/"):
patient_ids.add(subject.replace("Patient/", ""))
return list(patient_ids)
def run_extraction():
"""Main extraction pipeline."""
ensure_data_dir()
print("=" * 60)
print("FHIR DATA EXTRACTION")
print("=" * 60)
print("\n[1/4] Testing server connectivity...")
test_server_connectivity()
print("\n[2/4] Extracting encounters...")
encounters = extract_encounters()
print("\n[3/4] Extracting patients...")
patient_ids = get_patient_ids_from_encounters(encounters)
print(f"Found {len(patient_ids)} unique patients")
patients = extract_patients(patient_ids)
print("\n[4/4] Extracting conditions...")
conditions = extract_conditions(patient_ids)
# Save raw data to JSON files
with open(os.path.join(DATA_DIR, "raw_encounters.json"), "w") as f:
json.dump(encounters, f, indent=2)
with open(os.path.join(DATA_DIR, "raw_patients.json"), "w") as f:
json.dump(patients, f, indent=2)
with open(os.path.join(DATA_DIR, "raw_conditions.json"), "w") as f:
json.dump(conditions, f, indent=2)
print(f"\nRaw data saved to {DATA_DIR}/")
return encounters, patients, conditions
if __name__ == "__main__":
run_extraction()What does this code do?
- get_patient_ids_from_encounters() parses the subject.reference field from each Encounter to build a deduplicated list of patient IDs.
- run_extraction() orchestrates the full pipeline in four numbered stages: connectivity test, encounter extraction, patient extraction, and condition extraction.
- All raw data is saved as JSON files in the data/ directory. These serve as an intermediate cache so you don't need to re-query the server during later transform and scoring steps.
- Save extract.py.
- Run the extraction pipeline by executing this command in your terminal:
python extract.pyThe script will take 1-2 minutes as it fetches data from the live FHIR server. You should see numbered progress messages for each stage, ending with a summary of how many resources were saved.
Seeing a connection error?
- Make sure your virtual environment is active (you should see (venv) in your terminal prompt). If not, run source venv/bin/activate again.
- The HAPI FHIR server is public and occasionally goes down for maintenance. If you get a 500 or timeout error, wait a few minutes and try again.
- Check that the URL in FHIR_BASE_URL is exactly http://hapi.fhir.org/baseR4 with no trailing slash.
Help me debug my FHIR extraction connection error.
- Confirm the JSON files were created by running:
ls data/You should see raw_encounters.json, raw_patients.json, and raw_conditions.json listed in the output.
✔️ Awesome, I've got everything!
Great. Double check you've saved extract.py and that all three JSON files exist in your data/ directory.
ⓧ I'd like to double check the full code
Here is the complete extract.py file for comparison:
"""
FHIR Data Extraction Module
Extracts Patient, Encounter, and Condition resources from HAPI FHIR R4 server.
HIPAA AWARENESS NOTE:
This pipeline uses the HAPI FHIR PUBLIC TEST SERVER which contains ONLY synthetic data.
In a production environment, this code would require:
- Business Associate Agreement (BAA) with the data source
- Encryption in transit (HTTPS) and at rest
- Access controls and role-based permissions
- Audit logging of all data access
- Minimum necessary principle: only extract fields needed for analysis
Reference: HHS.gov HIPAA Security Rule (45 CFR Part 164)
"""
import json
import os
import requests
FHIR_BASE_URL = "http://hapi.fhir.org/baseR4"
DATA_DIR = "data"
MAX_ENCOUNTERS = 200
def ensure_data_dir():
"""Create data directory if it doesn't exist."""
os.makedirs(DATA_DIR, exist_ok=True)
def test_server_connectivity():
"""Verify the FHIR server is reachable by fetching its CapabilityStatement."""
url = f"{FHIR_BASE_URL}/metadata"
response = requests.get(url, headers={"Accept": "application/fhir+json"})
response.raise_for_status()
data = response.json()
assert data["resourceType"] == "CapabilityStatement", "Unexpected response"
print(f"Connected to: {data.get('implementation', {}).get('description', 'HAPI FHIR Server')}")
print(f"FHIR Version: {data.get('fhirVersion', 'unknown')}")
return True
def extract_encounters():
"""
Extract finished inpatient encounters from the FHIR server.
Uses pagination to collect up to MAX_ENCOUNTERS resources.
Applies minimum necessary principle: only requesting inpatient, finished encounters.
"""
encounters = []
url = f"{FHIR_BASE_URL}/Encounter?class=IMP&status=finished&_count=50"
while url and len(encounters) < MAX_ENCOUNTERS:
print(f"Fetching encounters... ({len(encounters)} so far)")
response = requests.get(url, headers={"Accept": "application/fhir+json"})
response.raise_for_status()
bundle = response.json()
if "entry" in bundle:
for entry in bundle["entry"]:
if len(encounters) >= MAX_ENCOUNTERS:
break
encounters.append(entry["resource"])
# Follow pagination link
url = None
for link in bundle.get("link", []):
if link["relation"] == "next":
url = link["url"]
break
print(f"Extracted {len(encounters)} encounters")
return encounters
def extract_patients(patient_ids):
"""
Fetch Patient resources for a list of patient IDs.
Only retrieves demographics needed for the analysis (minimum necessary).
"""
patients = []
for i, patient_id in enumerate(patient_ids):
if i % 10 == 0:
print(f"Fetching patients... ({i}/{len(patient_ids)})")
url = f"{FHIR_BASE_URL}/Patient/{patient_id}"
try:
response = requests.get(url, headers={"Accept": "application/fhir+json"})
response.raise_for_status()
patients.append(response.json())
except requests.exceptions.HTTPError:
print(f" Warning: Patient {patient_id} not found, skipping")
continue
print(f"Extracted {len(patients)} patients")
return patients
def extract_conditions(patient_ids):
"""
Fetch Condition resources for each patient.
These map to comorbidities needed for the Charlson/LACE scoring.
"""
all_conditions = []
for i, patient_id in enumerate(patient_ids):
if i % 10 == 0:
print(f"Fetching conditions... ({i}/{len(patient_ids)})")
url = f"{FHIR_BASE_URL}/Condition?subject=Patient/{patient_id}&_count=100"
try:
response = requests.get(url, headers={"Accept": "application/fhir+json"})
response.raise_for_status()
bundle = response.json()
if "entry" in bundle:
for entry in bundle["entry"]:
all_conditions.append(entry["resource"])
except requests.exceptions.HTTPError:
continue
print(f"Extracted {len(all_conditions)} conditions")
return all_conditions
def get_patient_ids_from_encounters(encounters):
"""Extract unique patient IDs from encounter resources."""
patient_ids = set()
for enc in encounters:
subject = enc.get("subject", {}).get("reference", "")
if subject.startswith("Patient/"):
patient_ids.add(subject.replace("Patient/", ""))
return list(patient_ids)
def run_extraction():
"""Main extraction pipeline."""
ensure_data_dir()
print("=" * 60)
print("FHIR DATA EXTRACTION")
print("=" * 60)
print("\n[1/4] Testing server connectivity...")
test_server_connectivity()
print("\n[2/4] Extracting encounters...")
encounters = extract_encounters()
print("\n[3/4] Extracting patients...")
patient_ids = get_patient_ids_from_encounters(encounters)
print(f"Found {len(patient_ids)} unique patients")
patients = extract_patients(patient_ids)
print("\n[4/4] Extracting conditions...")
conditions = extract_conditions(patient_ids)
# Save raw data to JSON files
with open(os.path.join(DATA_DIR, "raw_encounters.json"), "w") as f:
json.dump(encounters, f, indent=2)
with open(os.path.join(DATA_DIR, "raw_patients.json"), "w") as f:
json.dump(patients, f, indent=2)
with open(os.path.join(DATA_DIR, "raw_conditions.json"), "w") as f:
json.dump(conditions, f, indent=2)
print(f"\nRaw data saved to {DATA_DIR}/")
return encounters, patients, conditions
if __name__ == "__main__":
run_extraction()Your raw FHIR data is extracted and saved locally. Next up, you'll transform these nested JSON resources into clean SQLite tables ready for analysis.
Transform and Load into SQLite
You now have raw FHIR data sitting in your data/ directory as JSON files. These files contain deeply nested structures that are great for interoperability but terrible for analytics queries.
In this step, you'll build the transformation layer that flattens FHIR resources into tabular rows and loads them into a SQLite database. Once loaded, you can run SQL queries against your healthcare data.
In this step, get ready to:
- Flatten nested FHIR JSON into flat dictionaries suitable for database rows.
- Create a SQLite database with normalized tables for patients, encounters, and conditions.
- Run data quality checks to confirm everything loaded correctly.
Flatten FHIR resources into tabular format
FHIR resources are deeply nested JSON objects. An Encounter contains a subject.reference path to find the patient, and a period object holding start and end timestamps. Your parse functions will extract exactly the fields needed for analytics.
- Open transform.py in Cursor. Replace the placeholder comment with the module setup and all three parse functions by pasting the following code:
"""\nFHIR Data Transformation and Loading Module\nTransforms raw FHIR JSON into normalized SQLite tables.\n"""
import json
import os
import sqlite3
from datetime import datetime
DATA_DIR = "data"
DB_PATH = os.path.join(DATA_DIR, "readmission.db")
def parse_encounter(resource):
"""Flatten a FHIR Encounter resource into a dictionary."""
# Extract patient reference from the encounter
patient_ref = resource.get("subject", {}).get("reference", "")
patient_id = patient_ref.replace("Patient/", "") if patient_ref else None
# Get encounter classification code (IMP, EMER, AMB, etc.)
class_code = resource.get("class", {}).get("code", "unknown")
# Extract period and calculate length of stay in days
period = resource.get("period", {})
start_date = period.get("start")
end_date = period.get("end")
length_of_stay = None
if start_date and end_date:
try:
start = datetime.fromisoformat(start_date.replace("Z", "+00:00"))
end = datetime.fromisoformat(end_date.replace("Z", "+00:00"))
length_of_stay = max(1, (end - start).days)
except (ValueError, TypeError):
length_of_stay = None
return {
"id": resource.get("id"),
"patient_id": patient_id,
"class_code": class_code,
"status": resource.get("status"),
"start_date": start_date,
"end_date": end_date,
"length_of_stay_days": length_of_stay,
}
def parse_patient(resource):
"""Flatten a FHIR Patient resource into a dictionary."""
return {
"id": resource.get("id"),
"birth_date": resource.get("birthDate"),
"gender": resource.get("gender"),
}
def parse_condition(resource):
"""Flatten a FHIR Condition resource into a dictionary."""
# Extract patient reference
patient_ref = resource.get("subject", {}).get("reference", "")
patient_id = patient_ref.replace("Patient/", "") if patient_ref else None
# Pull the first coding entry for code, display, and system
coding = resource.get("code", {}).get("coding", [])
code = coding[0].get("code") if coding else None
display = coding[0].get("display") if coding else None
system = coding[0].get("system") if coding else None
# Extract clinical status (active, resolved, etc.)
clinical_status = None
cs_coding = resource.get("clinicalStatus", {}).get("coding", [])
if cs_coding:
clinical_status = cs_coding[0].get("code")
return {
"id": resource.get("id"),
"patient_id": patient_id,
"code": code,
"system": system,
"display": display,
"clinical_status": clinical_status,
}What does this code do?
Each parse function takes a raw FHIR JSON resource and returns a flat dictionary with only the fields you need for analytics.
- parse_encounter() extracts the patient reference, encounter class code, dates, and calculates length of stay in days. It uses max(1, ...) to ensure same-day encounters count as at least 1 day.
- parse_patient() pulls just the demographics needed for the LACE score: ID, birth date, and gender.
- parse_condition() extracts the diagnosis coding (SNOMED or ICD) plus the clinical status. These map to Charlson comorbidity categories later.
- Save transform.py.
- You won't see a change yet. The next chunk adds the database creation logic that uses these parse functions.
Why flatten instead of querying JSON directly?
FHIR JSON is optimized for data exchange between systems, not for analytics. A single Encounter resource nests the patient reference three levels deep and stores dates inside a period sub-object. Flattening into rows lets you write simple SQL joins across patients, encounters, and conditions.
Create SQLite schema and load data
Now you need a database to hold the flattened data. Python's built-in sqlite3 module gives you a full relational database with zero setup. No server, no installation, no configuration.
- Add the database creation and loading functions below your parse_condition() function in transform.py:
def create_database():
"""Create SQLite database with the analytics schema."""
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
# Drop existing tables for a clean reload
cursor.execute("DROP TABLE IF EXISTS patients")
cursor.execute("DROP TABLE IF EXISTS encounters")
cursor.execute("DROP TABLE IF EXISTS conditions")
cursor.execute("DROP TABLE IF EXISTS lace_scores")
# Patients table: demographics only
cursor.execute("""
CREATE TABLE patients (
id TEXT PRIMARY KEY,
birth_date TEXT,
gender TEXT
)
""")
# Encounters table: each hospital visit
cursor.execute("""
CREATE TABLE encounters (
id TEXT PRIMARY KEY,
patient_id TEXT,
class_code TEXT,
status TEXT,
start_date TEXT,
end_date TEXT,
length_of_stay_days INTEGER,
FOREIGN KEY (patient_id) REFERENCES patients(id)
)
""")
# Conditions table: diagnoses per patient
cursor.execute("""
CREATE TABLE conditions (
id TEXT PRIMARY KEY,
patient_id TEXT,
code TEXT,
system TEXT,
display TEXT,
clinical_status TEXT,
FOREIGN KEY (patient_id) REFERENCES patients(id)
)
""")
conn.commit()
return conn
def load_data(conn, encounters_raw, patients_raw, conditions_raw):
"""Transform and load raw FHIR data into SQLite."""
cursor = conn.cursor()
# Load patients
patients_loaded = 0
for resource in patients_raw:
parsed = parse_patient(resource)
if parsed["id"]:
cursor.execute(
"INSERT OR IGNORE INTO patients (id, birth_date, gender) VALUES (?, ?, ?)",
(parsed["id"], parsed["birth_date"], parsed["gender"]),
)
patients_loaded += 1
# Load encounters
encounters_loaded = 0
for resource in encounters_raw:
parsed = parse_encounter(resource)
if parsed["id"] and parsed["patient_id"]:
cursor.execute(
"""INSERT OR IGNORE INTO encounters
(id, patient_id, class_code, status, start_date, end_date, length_of_stay_days)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(
parsed["id"],
parsed["patient_id"],
parsed["class_code"],
parsed["status"],
parsed["start_date"],
parsed["end_date"],
parsed["length_of_stay_days"],
),
)
encounters_loaded += 1
# Load conditions
conditions_loaded = 0
for resource in conditions_raw:
parsed = parse_condition(resource)
if parsed["id"] and parsed["patient_id"]:
cursor.execute(
"""INSERT OR IGNORE INTO conditions
(id, patient_id, code, system, display, clinical_status)
VALUES (?, ?, ?, ?, ?, ?)""",
(
parsed["id"],
parsed["patient_id"],
parsed["code"],
parsed["system"],
parsed["display"],
parsed["clinical_status"],
),
)
conditions_loaded += 1
conn.commit()
return patients_loaded, encounters_loaded, conditions_loadedWhat does this code do?
create_database() sets up the SQLite schema with three normalized tables. It drops existing tables first so you can re-run the pipeline cleanly. The lace_scores table is also dropped preemptively for the scoring step later.
load_data() loops through each raw resource, parses it into a flat dictionary, and inserts it. INSERT OR IGNORE skips duplicates gracefully. Records missing an ID or patient reference are silently skipped to avoid loading corrupt data.
- Save transform.py.
- You won't see a change yet. The next chunk adds the validation and orchestration functions that tie everything together.
Validate data quality
A pipeline that loads data without checking it is a pipeline you cannot trust. You'll add validation that counts records, checks for null references, and verifies that length-of-stay calculations are positive.
- Add the validation and main orchestration functions below your load_data() function in transform.py:
def validate_data(conn):
"""Run data quality checks and print summary."""
cursor = conn.cursor()
print("\n--- Data Quality Report ---")
# Count records in each table
cursor.execute("SELECT COUNT(*) FROM patients")
patient_count = cursor.fetchone()[0]
print(f"Patients loaded: {patient_count}")
cursor.execute("SELECT COUNT(*) FROM encounters")
encounter_count = cursor.fetchone()[0]
print(f"Encounters loaded: {encounter_count}")
cursor.execute("SELECT COUNT(*) FROM conditions")
condition_count = cursor.fetchone()[0]
print(f"Conditions loaded: {condition_count}")
# Check for null patient references
cursor.execute("SELECT COUNT(*) FROM encounters WHERE patient_id IS NULL")
null_refs = cursor.fetchone()[0]
print(f"Encounters with null patient refs: {null_refs}")
# Verify length of stay calculations
cursor.execute("SELECT COUNT(*) FROM encounters WHERE length_of_stay_days IS NULL")
null_los = cursor.fetchone()[0]
print(f"Encounters with null LOS: {null_los}")
cursor.execute(
"SELECT COUNT(*) FROM encounters WHERE length_of_stay_days IS NOT NULL AND length_of_stay_days > 0"
)
valid_los = cursor.fetchone()[0]
print(f"Encounters with valid LOS: {valid_los}")
print("--- End Report ---\n")
return patient_count, encounter_count, condition_count
def run_transform():
"""Main transformation pipeline."""
print("=" * 60)
print("FHIR DATA TRANSFORMATION & LOADING")
print("=" * 60)
# Load raw JSON files from disk
with open(os.path.join(DATA_DIR, "raw_encounters.json")) as f:
encounters_raw = json.load(f)
with open(os.path.join(DATA_DIR, "raw_patients.json")) as f:
patients_raw = json.load(f)
with open(os.path.join(DATA_DIR, "raw_conditions.json")) as f:
conditions_raw = json.load(f)
print(f"\nRaw data: {len(encounters_raw)} encounters, {len(patients_raw)} patients, {len(conditions_raw)} conditions")
# Create database and load transformed data
conn = create_database()
patients_loaded, encounters_loaded, conditions_loaded = load_data(
conn, encounters_raw, patients_raw, conditions_raw
)
print(f"Loaded: {patients_loaded} patients, {encounters_loaded} encounters, {conditions_loaded} conditions")
# Run quality checks
validate_data(conn)
conn.close()
print(f"Database saved to: {DB_PATH}")
if __name__ == "__main__":
run_transform()What does this code do?
validate_data() queries each table and checks for data quality issues. It reports null patient references (which would break joins) and missing length-of-stay values (which would break LACE scoring).
run_transform() is the orchestrator. It reads the raw JSON files from disk, creates the database, loads the data, runs validation, and prints a summary. The if __name__ guard lets you run this module standalone or import it from main.py later.
- Save transform.py.
- Run the transformation pipeline by running this command in your terminal:
python transform.py- You should see output showing the number of records loaded per table, followed by the Data Quality Report. The "Encounters loaded" count should be 50 or higher.
Seeing a FileNotFoundError?
- Make sure you ran extract.py in the previous step and that data/raw_encounters.json, data/raw_patients.json, and data/raw_conditions.json exist in your data/ directory.
- Confirm you're running the command from inside the readmission-risk-pipeline folder, not a parent directory.
- Check that your virtual environment is still active (you should see (venv) in your terminal prompt).
Why can't my transform script find the raw data files?
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on.
ⓧ I'd like to double check the full code
Here is the complete transform.py file for comparison:
"""
FHIR Data Transformation and Loading Module
Transforms raw FHIR JSON into normalized SQLite tables.
"""
import json
import os
import sqlite3
from datetime import datetime
DATA_DIR = "data"
DB_PATH = os.path.join(DATA_DIR, "readmission.db")
def parse_encounter(resource):
"""Flatten a FHIR Encounter resource into a dictionary."""
patient_ref = resource.get("subject", {}).get("reference", "")
patient_id = patient_ref.replace("Patient/", "") if patient_ref else None
class_code = resource.get("class", {}).get("code", "unknown")
period = resource.get("period", {})
start_date = period.get("start")
end_date = period.get("end")
length_of_stay = None
if start_date and end_date:
try:
start = datetime.fromisoformat(start_date.replace("Z", "+00:00"))
end = datetime.fromisoformat(end_date.replace("Z", "+00:00"))
length_of_stay = max(1, (end - start).days)
except (ValueError, TypeError):
length_of_stay = None
return {
"id": resource.get("id"),
"patient_id": patient_id,
"class_code": class_code,
"status": resource.get("status"),
"start_date": start_date,
"end_date": end_date,
"length_of_stay_days": length_of_stay,
}
def parse_patient(resource):
"""Flatten a FHIR Patient resource into a dictionary."""
return {
"id": resource.get("id"),
"birth_date": resource.get("birthDate"),
"gender": resource.get("gender"),
}
def parse_condition(resource):
"""Flatten a FHIR Condition resource into a dictionary."""
patient_ref = resource.get("subject", {}).get("reference", "")
patient_id = patient_ref.replace("Patient/", "") if patient_ref else None
coding = resource.get("code", {}).get("coding", [])
code = coding[0].get("code") if coding else None
display = coding[0].get("display") if coding else None
system = coding[0].get("system") if coding else None
clinical_status = None
cs_coding = resource.get("clinicalStatus", {}).get("coding", [])
if cs_coding:
clinical_status = cs_coding[0].get("code")
return {
"id": resource.get("id"),
"patient_id": patient_id,
"code": code,
"system": system,
"display": display,
"clinical_status": clinical_status,
}
def create_database():
"""Create SQLite database with the analytics schema."""
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("DROP TABLE IF EXISTS patients")
cursor.execute("DROP TABLE IF EXISTS encounters")
cursor.execute("DROP TABLE IF EXISTS conditions")
cursor.execute("DROP TABLE IF EXISTS lace_scores")
cursor.execute("""
CREATE TABLE patients (
id TEXT PRIMARY KEY,
birth_date TEXT,
gender TEXT
)
""")
cursor.execute("""
CREATE TABLE encounters (
id TEXT PRIMARY KEY,
patient_id TEXT,
class_code TEXT,
status TEXT,
start_date TEXT,
end_date TEXT,
length_of_stay_days INTEGER,
FOREIGN KEY (patient_id) REFERENCES patients(id)
)
""")
cursor.execute("""
CREATE TABLE conditions (
id TEXT PRIMARY KEY,
patient_id TEXT,
code TEXT,
system TEXT,
display TEXT,
clinical_status TEXT,
FOREIGN KEY (patient_id) REFERENCES patients(id)
)
""")
conn.commit()
return conn
def load_data(conn, encounters_raw, patients_raw, conditions_raw):
"""Transform and load raw FHIR data into SQLite."""
cursor = conn.cursor()
# Load patients
patients_loaded = 0
for resource in patients_raw:
parsed = parse_patient(resource)
if parsed["id"]:
cursor.execute(
"INSERT OR IGNORE INTO patients (id, birth_date, gender) VALUES (?, ?, ?)",
(parsed["id"], parsed["birth_date"], parsed["gender"]),
)
patients_loaded += 1
# Load encounters
encounters_loaded = 0
for resource in encounters_raw:
parsed = parse_encounter(resource)
if parsed["id"] and parsed["patient_id"]:
cursor.execute(
"""INSERT OR IGNORE INTO encounters
(id, patient_id, class_code, status, start_date, end_date, length_of_stay_days)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(
parsed["id"],
parsed["patient_id"],
parsed["class_code"],
parsed["status"],
parsed["start_date"],
parsed["end_date"],
parsed["length_of_stay_days"],
),
)
encounters_loaded += 1
# Load conditions
conditions_loaded = 0
for resource in conditions_raw:
parsed = parse_condition(resource)
if parsed["id"] and parsed["patient_id"]:
cursor.execute(
"""INSERT OR IGNORE INTO conditions
(id, patient_id, code, system, display, clinical_status)
VALUES (?, ?, ?, ?, ?, ?)""",
(
parsed["id"],
parsed["patient_id"],
parsed["code"],
parsed["system"],
parsed["display"],
parsed["clinical_status"],
),
)
conditions_loaded += 1
conn.commit()
return patients_loaded, encounters_loaded, conditions_loaded
def validate_data(conn):
"""Run data quality checks and print summary."""
cursor = conn.cursor()
print("\n--- Data Quality Report ---")
cursor.execute("SELECT COUNT(*) FROM patients")
patient_count = cursor.fetchone()[0]
print(f"Patients loaded: {patient_count}")
cursor.execute("SELECT COUNT(*) FROM encounters")
encounter_count = cursor.fetchone()[0]
print(f"Encounters loaded: {encounter_count}")
cursor.execute("SELECT COUNT(*) FROM conditions")
condition_count = cursor.fetchone()[0]
print(f"Conditions loaded: {condition_count}")
cursor.execute("SELECT COUNT(*) FROM encounters WHERE patient_id IS NULL")
null_refs = cursor.fetchone()[0]
print(f"Encounters with null patient refs: {null_refs}")
cursor.execute("SELECT COUNT(*) FROM encounters WHERE length_of_stay_days IS NULL")
null_los = cursor.fetchone()[0]
print(f"Encounters with null LOS: {null_los}")
cursor.execute(
"SELECT COUNT(*) FROM encounters WHERE length_of_stay_days IS NOT NULL AND length_of_stay_days > 0"
)
valid_los = cursor.fetchone()[0]
print(f"Encounters with valid LOS: {valid_los}")
print("--- End Report ---\n")
return patient_count, encounter_count, condition_count
def run_transform():
"""Main transformation pipeline."""
print("=" * 60)
print("FHIR DATA TRANSFORMATION & LOADING")
print("=" * 60)
with open(os.path.join(DATA_DIR, "raw_encounters.json")) as f:
encounters_raw = json.load(f)
with open(os.path.join(DATA_DIR, "raw_patients.json")) as f:
patients_raw = json.load(f)
with open(os.path.join(DATA_DIR, "raw_conditions.json")) as f:
conditions_raw = json.load(f)
print(f"\nRaw data: {len(encounters_raw)} encounters, {len(patients_raw)} patients, {len(conditions_raw)} conditions")
conn = create_database()
patients_loaded, encounters_loaded, conditions_loaded = load_data(
conn, encounters_raw, patients_raw, conditions_raw
)
print(f"Loaded: {patients_loaded} patients, {encounters_loaded} encounters, {conditions_loaded} conditions")
validate_data(conn)
conn.close()
print(f"Database saved to: {DB_PATH}")
if __name__ == "__main__":
run_transform()Your analytics database is ready. Next up, you'll implement the LACE readmission risk scoring algorithm to turn this raw data into actionable clinical insights.
Compute LACE Readmission Risk Scores
Your SQLite database is loaded with patient encounters and conditions. Now you need a way to identify which patients are most likely to be readmitted within 30 days.
The LACE index is a clinically validated scoring tool (van Walraven et al., CMAJ 2010) that hospitals use to predict readmission risk. It combines four factors: Length of stay, Acuity of admission, Comorbidity burden, and Emergency department visits. In this step, you will implement the full LACE scoring algorithm and compute risk scores for every patient in your database.
In this step, get ready to:
- Implement the four LACE scoring functions (L, A, C, E).
- Calculate composite LACE scores for each patient.
- Handle edge cases and generate a risk distribution report.
Implement the LACE scoring functions
Each letter in LACE maps to a clinical risk factor with a specific points table. Your scoring module will translate raw encounter data into these point values.
- Open score.py in Cursor. Replace the placeholder comment with the module header and comorbidity keyword mapping by pasting this code:
"""
LACE Index Scoring Module
Implements the LACE readmission risk index (van Walraven et al., CMAJ 2010).
LACE Score Components:
- L: Length of stay (0-7 points)
- A: Acuity of admission (0 or 3 points)
- C: Comorbidity via Charlson index (0-5 points)
- E: Emergency department visits in prior 6 months (0-4 points)
Total range: 0-19
Risk categories: Low (0-4), Moderate (5-9), High (10+)
"""
import os
import sqlite3
from datetime import datetime, timedelta
DATA_DIR = "data"
DB_PATH = os.path.join(DATA_DIR, "readmission.db")
# Charlson comorbidity keyword mapping
# Maps condition display text keywords to Charlson weights
CHARLSON_KEYWORDS = {
"myocardial infarction": 1,
"heart attack": 1,
"cerebrovascular": 1,
"stroke": 1,
"transient ischemic": 1,
"peripheral vascular": 1,
"diabetes": 1,
"congestive heart failure": 2,
"heart failure": 2,
"chronic obstructive pulmonary": 2,
"copd": 2,
"liver disease": 2,
"cirrhosis": 2,
"cancer": 2,
"malignant neoplasm": 2,
"carcinoma": 2,
"lymphoma": 2,
"leukemia": 2,
"dementia": 3,
"alzheimer": 3,
"connective tissue": 3,
"lupus": 3,
"rheumatoid": 3,
"hiv": 4,
"aids": 4,
"metastatic": 6,
}What does this code set up?
The module docstring documents the LACE score components and their point ranges. The total possible score is 0 to 19.
The CHARLSON_KEYWORDS dictionary maps condition display text (from the FHIR Condition resources) to Charlson Comorbidity Index weights. Higher weights indicate conditions with greater mortality risk. Metastatic cancer carries the heaviest weight at 6 points.
- Save score.py. You will not see output yet. This mapping is the foundation that the scoring functions will use.
- Add the Length of stay and Acuity scoring functions below the CHARLSON_KEYWORDS dictionary:
def score_length_of_stay(days):
"""L component: Length of stay scoring."""
if days is None or days < 1:
return 0
elif days == 1:
return 1
elif days == 2:
return 2
elif days == 3:
return 3
elif 4 <= days <= 6:
return 4
elif 7 <= days <= 13:
return 5
else: # 14+
return 7
def score_acuity(class_code):
"""A component: Acuity of admission. Emergency/acute = 3, else = 0."""
acute_codes = ["IMP", "EMER", "emergency", "inpatient"]
if class_code and class_code.lower() in [c.lower() for c in acute_codes]:
return 3
return 0What do these functions do?
The L score converts length of stay into a 0-7 point scale. Longer stays indicate higher risk. A patient staying 14+ days receives the maximum 7 points.
The A score is binary. If the encounter was an acute or emergency admission (class codes IMP or EMER), the patient receives 3 points. Elective admissions receive 0.
- Save score.py and verify the L function works by running this quick test:
python3 -c "from score import score_length_of_stay; print(score_length_of_stay(5))"- You should see 4 printed. A 5-day stay falls in the 4-6 day bracket, earning 4 points.
Seeing a ModuleNotFoundError?
Make sure your terminal is in the readmission-risk-pipeline directory and your virtual environment is activated. The from score import statement needs to find score.py in the current directory.
Also verify you saved the file. Unsaved changes in Cursor will not be picked up by Python.
help me debug a ModuleNotFoundError when importing score.py
- Add the Comorbidity and Emergency visit scoring functions below score_acuity:
def score_comorbidity(conditions):
"""
C component: Charlson Comorbidity Index mapped to LACE points.
Sums Charlson weights from patient conditions, then applies LACE mapping:
CCI 0 = 0pt, 1 = 1pt, 2 = 2pt, 3 = 3pt, 4+ = 5pt
"""
charlson_score = 0
matched_conditions = []
for condition_display in conditions:
if not condition_display:
continue
display_lower = condition_display.lower()
for keyword, weight in CHARLSON_KEYWORDS.items():
if keyword in display_lower:
charlson_score += weight
matched_conditions.append((condition_display, weight))
break # Only count each condition once
# Map Charlson score to LACE C points
if charlson_score == 0:
return 0, matched_conditions
elif charlson_score == 1:
return 1, matched_conditions
elif charlson_score == 2:
return 2, matched_conditions
elif charlson_score == 3:
return 3, matched_conditions
else: # 4+
return 5, matched_conditions
def score_emergency_visits(conn, patient_id, index_admission_date):
"""
E component: Count ED/emergency encounters in 6 months before index admission.
Excludes the index admission itself.
"""
if not index_admission_date:
return 0
try:
index_date = datetime.fromisoformat(index_admission_date.replace("Z", "+00:00"))
six_months_prior = index_date - timedelta(days=180)
except (ValueError, TypeError):
return 0
cursor = conn.cursor()
cursor.execute(
"""
SELECT COUNT(*) FROM encounters
WHERE patient_id = ?
AND class_code IN ('EMER', 'emergency')
AND start_date >= ?
AND start_date < ?
""",
(patient_id, six_months_prior.isoformat(), index_date.isoformat()),
)
count = cursor.fetchone()[0]
# Cap at 4 points
return min(count, 4)What do the C and E functions do?
The C function loops through a patient's conditions, matches them against the CHARLSON_KEYWORDS dictionary, and sums the weights. The raw Charlson score is then mapped to LACE points (capped at 5 for a Charlson score of 4+). It returns both the LACE C score and the list of matched conditions for reporting.
The E function queries the encounters table for emergency visits within 180 days before the patient's index admission. The count is capped at 4 points per the published LACE rules.
- Save score.py. The next chunk will use all four functions together.
Calculate scores for each patient
With the four scoring components ready, you need an orchestrator function that loops through every patient with a valid encounter, computes their composite LACE score, and stores the results in a new database table.
- Add the calculate_lace_scores function below score_emergency_visits:
def calculate_lace_scores(conn):
"""Calculate LACE scores for all patients with valid encounters."""
cursor = conn.cursor()
# Create lace_scores table
cursor.execute("DROP TABLE IF EXISTS lace_scores")
cursor.execute("""
CREATE TABLE lace_scores (
patient_id TEXT PRIMARY KEY,
l_score INTEGER,
a_score INTEGER,
c_score INTEGER,
e_score INTEGER,
total_score INTEGER,
risk_category TEXT,
top_comorbidities TEXT
)
""")
# Get most recent inpatient encounter per patient
cursor.execute("""
SELECT patient_id, id, class_code, start_date, end_date, length_of_stay_days
FROM encounters
WHERE length_of_stay_days IS NOT NULL
AND length_of_stay_days > 0
ORDER BY start_date DESC
""")
encounters = cursor.fetchall()
# Track which patients we've already scored (use most recent encounter)
scored_patients = set()
scores_computed = 0
skipped = 0
for patient_id, enc_id, class_code, start_date, end_date, los_days in encounters:
if patient_id in scored_patients:
continue
scored_patients.add(patient_id)
# L: Length of stay
l_score = score_length_of_stay(los_days)
# A: Acuity
a_score = score_acuity(class_code)
# C: Comorbidity
cursor.execute(
"SELECT display FROM conditions WHERE patient_id = ?", (patient_id,)
)
condition_displays = [row[0] for row in cursor.fetchall()]
c_score, matched_conditions = score_comorbidity(condition_displays)
top_comorbidities = "; ".join(
[f"{name} (+{w})" for name, w in matched_conditions[:5]]
)
# E: Emergency visits
e_score = score_emergency_visits(conn, patient_id, start_date)
# Total
total_score = l_score + a_score + c_score + e_score
# Risk category
if total_score <= 4:
risk_category = "Low"
elif total_score <= 9:
risk_category = "Moderate"
else:
risk_category = "High"
cursor.execute(
"""INSERT INTO lace_scores
(patient_id, l_score, a_score, c_score, e_score, total_score, risk_category, top_comorbidities)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
(patient_id, l_score, a_score, c_score, e_score, total_score, risk_category, top_comorbidities),
)
scores_computed += 1
conn.commit()
print(f"LACE scores computed: {scores_computed} patients")
print(f"Skipped (data quality): {skipped} patients")
return scores_computedWhat does this function do?
This function creates a lace_scores table in your SQLite database and loops through all encounters that have a valid length of stay. It orders encounters by start_date DESC and uses a scored_patients set to ensure each patient is scored only once (using their most recent admission).
For each patient, it calls all four scoring functions, sums the results, assigns a risk category (Low 0-4, Moderate 5-9, High 10+), and inserts the row into the database.
- Save score.py. The next chunk adds the run function so you can execute the scoring.
Handle edge cases and run the scoring pipeline
The scoring logic already handles several edge cases gracefully. Patients with missing encounter dates are filtered out by the SQL WHERE clause. Patients with no conditions receive a C score of 0. Patients with only one encounter receive an E score of 0.
- Add the run_scoring function and entry point below calculate_lace_scores:
def run_scoring():
"""Main scoring pipeline."""
print("=" * 60)
print("LACE INDEX SCORING")
print("=" * 60)
conn = sqlite3.connect(DB_PATH)
scores_computed = calculate_lace_scores(conn)
# Print distribution
cursor = conn.cursor()
cursor.execute("SELECT risk_category, COUNT(*) FROM lace_scores GROUP BY risk_category")
print("\nRisk Distribution:")
for category, count in cursor.fetchall():
print(f" {category}: {count} patients")
cursor.execute("SELECT AVG(total_score), MIN(total_score), MAX(total_score) FROM lace_scores")
avg, min_s, max_s = cursor.fetchone()
print(f"\nScore Statistics: avg={avg:.1f}, min={min_s}, max={max_s}")
conn.close()
if __name__ == "__main__":
run_scoring()What does run_scoring do?
This function connects to the SQLite database, runs the scoring pipeline, and prints a summary. The risk distribution shows how many patients fall into each category. The score statistics give you the average, minimum, and maximum LACE scores across your cohort.
- Save score.py and run the scoring module:
python3 score.py- You should see output showing the number of patients scored, a risk distribution (Low, Moderate, and/or High), and score statistics with average, minimum, and maximum values.
Seeing 0 patients scored?
Make sure you ran Steps 2 and 3 first (extract.py then transform.py). The scoring module reads from the encounters and conditions tables in data/readmission.db. If those tables are empty, there is nothing to score.
If you see a NameError, check that all four scoring functions (score_length_of_stay, score_acuity, score_comorbidity, score_emergency_visits) are defined above calculate_lace_scores in the file.
help me debug my LACE scoring script that is returning unexpected results
Why do some patients show a score of 0?
A LACE score of 0 is valid. It means the patient had a very short stay (under 1 day mapped to 0), a non-acute admission (A=0), no matched comorbidities (C=0), and no prior ED visits (E=0). The HAPI FHIR test data is synthetic, so not every patient will have rich encounter histories.
✔️ Awesome, I've got everything!
Great. Double check you have saved score.py and the output shows a risk distribution.
ⓧ I'd like to double check the full code
Here is the complete score.py file for comparison:
"""
LACE Index Scoring Module
Implements the LACE readmission risk index (van Walraven et al., CMAJ 2010).
LACE Score Components:
- L: Length of stay (0-7 points)
- A: Acuity of admission (0 or 3 points)
- C: Comorbidity via Charlson index (0-5 points)
- E: Emergency department visits in prior 6 months (0-4 points)
Total range: 0-19
Risk categories: Low (0-4), Moderate (5-9), High (10+)
"""
import os
import sqlite3
from datetime import datetime, timedelta
DATA_DIR = "data"
DB_PATH = os.path.join(DATA_DIR, "readmission.db")
# Charlson comorbidity keyword mapping
# Maps condition display text keywords to Charlson weights
CHARLSON_KEYWORDS = {
"myocardial infarction": 1,
"heart attack": 1,
"cerebrovascular": 1,
"stroke": 1,
"transient ischemic": 1,
"peripheral vascular": 1,
"diabetes": 1,
"congestive heart failure": 2,
"heart failure": 2,
"chronic obstructive pulmonary": 2,
"copd": 2,
"liver disease": 2,
"cirrhosis": 2,
"cancer": 2,
"malignant neoplasm": 2,
"carcinoma": 2,
"lymphoma": 2,
"leukemia": 2,
"dementia": 3,
"alzheimer": 3,
"connective tissue": 3,
"lupus": 3,
"rheumatoid": 3,
"hiv": 4,
"aids": 4,
"metastatic": 6,
}
def score_length_of_stay(days):
"""L component: Length of stay scoring."""
if days is None or days < 1:
return 0
elif days == 1:
return 1
elif days == 2:
return 2
elif days == 3:
return 3
elif 4 <= days <= 6:
return 4
elif 7 <= days <= 13:
return 5
else: # 14+
return 7
def score_acuity(class_code):
"""A component: Acuity of admission. Emergency/acute = 3, else = 0."""
acute_codes = ["IMP", "EMER", "emergency", "inpatient"]
if class_code and class_code.lower() in [c.lower() for c in acute_codes]:
return 3
return 0
def score_comorbidity(conditions):
"""
C component: Charlson Comorbidity Index mapped to LACE points.
Sums Charlson weights from patient conditions, then applies LACE mapping:
CCI 0 = 0pt, 1 = 1pt, 2 = 2pt, 3 = 3pt, 4+ = 5pt
"""
charlson_score = 0
matched_conditions = []
for condition_display in conditions:
if not condition_display:
continue
display_lower = condition_display.lower()
for keyword, weight in CHARLSON_KEYWORDS.items():
if keyword in display_lower:
charlson_score += weight
matched_conditions.append((condition_display, weight))
break # Only count each condition once
# Map Charlson score to LACE C points
if charlson_score == 0:
return 0, matched_conditions
elif charlson_score == 1:
return 1, matched_conditions
elif charlson_score == 2:
return 2, matched_conditions
elif charlson_score == 3:
return 3, matched_conditions
else: # 4+
return 5, matched_conditions
def score_emergency_visits(conn, patient_id, index_admission_date):
"""
E component: Count ED/emergency encounters in 6 months before index admission.
Excludes the index admission itself.
"""
if not index_admission_date:
return 0
try:
index_date = datetime.fromisoformat(index_admission_date.replace("Z", "+00:00"))
six_months_prior = index_date - timedelta(days=180)
except (ValueError, TypeError):
return 0
cursor = conn.cursor()
cursor.execute(
"""
SELECT COUNT(*) FROM encounters
WHERE patient_id = ?
AND class_code IN ('EMER', 'emergency')
AND start_date >= ?
AND start_date < ?
""",
(patient_id, six_months_prior.isoformat(), index_date.isoformat()),
)
count = cursor.fetchone()[0]
# Cap at 4 points
return min(count, 4)
def calculate_lace_scores(conn):
"""Calculate LACE scores for all patients with valid encounters."""
cursor = conn.cursor()
# Create lace_scores table
cursor.execute("DROP TABLE IF EXISTS lace_scores")
cursor.execute("""
CREATE TABLE lace_scores (
patient_id TEXT PRIMARY KEY,
l_score INTEGER,
a_score INTEGER,
c_score INTEGER,
e_score INTEGER,
total_score INTEGER,
risk_category TEXT,
top_comorbidities TEXT
)
""")
# Get most recent inpatient encounter per patient
cursor.execute("""
SELECT patient_id, id, class_code, start_date, end_date, length_of_stay_days
FROM encounters
WHERE length_of_stay_days IS NOT NULL
AND length_of_stay_days > 0
ORDER BY start_date DESC
""")
encounters = cursor.fetchall()
# Track which patients we've already scored (use most recent encounter)
scored_patients = set()
scores_computed = 0
skipped = 0
for patient_id, enc_id, class_code, start_date, end_date, los_days in encounters:
if patient_id in scored_patients:
continue
scored_patients.add(patient_id)
# L: Length of stay
l_score = score_length_of_stay(los_days)
# A: Acuity
a_score = score_acuity(class_code)
# C: Comorbidity
cursor.execute(
"SELECT display FROM conditions WHERE patient_id = ?", (patient_id,)
)
condition_displays = [row[0] for row in cursor.fetchall()]
c_score, matched_conditions = score_comorbidity(condition_displays)
top_comorbidities = "; ".join(
[f"{name} (+{w})" for name, w in matched_conditions[:5]]
)
# E: Emergency visits
e_score = score_emergency_visits(conn, patient_id, start_date)
# Total
total_score = l_score + a_score + c_score + e_score
# Risk category
if total_score <= 4:
risk_category = "Low"
elif total_score <= 9:
risk_category = "Moderate"
else:
risk_category = "High"
cursor.execute(
"""INSERT INTO lace_scores
(patient_id, l_score, a_score, c_score, e_score, total_score, risk_category, top_comorbidities)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
(patient_id, l_score, a_score, c_score, e_score, total_score, risk_category, top_comorbidities),
)
scores_computed += 1
conn.commit()
print(f"LACE scores computed: {scores_computed} patients")
print(f"Skipped (data quality): {skipped} patients")
return scores_computed
def run_scoring():
"""Main scoring pipeline."""
print("=" * 60)
print("LACE INDEX SCORING")
print("=" * 60)
conn = sqlite3.connect(DB_PATH)
scores_computed = calculate_lace_scores(conn)
# Print distribution
cursor = conn.cursor()
cursor.execute("SELECT risk_category, COUNT(*) FROM lace_scores GROUP BY risk_category")
print("\nRisk Distribution:")
for category, count in cursor.fetchall():
print(f" {category}: {count} patients")
cursor.execute("SELECT AVG(total_score), MIN(total_score), MAX(total_score) FROM lace_scores")
avg, min_s, max_s = cursor.fetchone()
print(f"\nScore Statistics: avg={avg:.1f}, min={min_s}, max={max_s}")
conn.close()
if __name__ == "__main__":
run_scoring()Your pipeline now computes clinically validated readmission risk scores for every patient. Next up, you will generate a comprehensive cohort report that joins these scores with patient demographics and exports it as a CSV for clinical review.
Generate Risk-Stratified Cohort Report
Your LACE scores are computed and sitting in the database. Every patient now has a risk category. But raw scores in a table are not what a care coordination team needs.
In this step, you will build the reporting layer that joins scores with patient demographics, exports an actionable CSV, and prints a clinical summary identifying the highest-risk patients. This is the deliverable a hospital would hand to its readmission prevention team.
In this step, get ready to:
- Build a SQL query that joins LACE scores with patient demographics and encounter data.
- Export a risk-stratified cohort report to CSV and print summary statistics.
- Add clinical context annotations and wire up the full pipeline orchestrator.
Build the cohort report query
A useful clinical report does not just list scores. It joins the scores with demographics, encounter history, and comorbidity details so a care coordinator can see everything in one row per patient.
- Open main.py in Cursor. This file is currently empty.
- Add the module docstring, imports, constants, and the beginning of the generate_report() function by copying the following code:
"""
Hospital Readmission Risk Pipeline
Main orchestrator that runs the full ETL + scoring + reporting pipeline.
"""
import os
import sqlite3
import pandas as pd
from extract import run_extraction
from transform import run_transform, DB_PATH
from score import run_scoring
DATA_DIR = "data"
REPORT_PATH = os.path.join(DATA_DIR, "readmission_risk_report.csv")
def generate_report():
"""Generate the final risk-stratified cohort report."""
print("=" * 60)
print("COHORT REPORT GENERATION")
print("=" * 60)
conn = sqlite3.connect(DB_PATH)
# Build comprehensive report query
query = """
SELECT
ls.patient_id,
p.gender,
p.birth_date,
e.start_date AS last_admission_date,
e.length_of_stay_days,
ls.l_score,
ls.a_score,
ls.c_score,
ls.e_score,
ls.total_score,
ls.risk_category,
ls.top_comorbidities
FROM lace_scores ls
JOIN patients p ON ls.patient_id = p.id
LEFT JOIN encounters e ON ls.patient_id = e.patient_id
AND e.length_of_stay_days IS NOT NULL
GROUP BY ls.patient_id
HAVING e.start_date = MAX(e.start_date)
ORDER BY ls.total_score DESC
"""
df = pd.read_sql_query(query, conn)What does this query do?
This SQL query joins three tables to assemble a complete patient view for the report.
- JOIN patients p ON ls.patient_id = p.id links each score to the patient's gender and birth date.
- LEFT JOIN encounters e pulls in the most recent admission date and length of stay.
- GROUP BY ls.patient_id with HAVING e.start_date = MAX(e.start_date) ensures only the most recent encounter per patient is included.
- ORDER BY ls.total_score DESC puts the highest-risk patients at the top of the report.
Why import DB_PATH from transform?
The database path is defined once in transform.py and reused here. This keeps all modules pointing at the same file without hardcoding the path in multiple places.
Export to CSV and print summary statistics
The raw query results need two additions before export: a calculated age column (derived from birth date) and summary statistics that give a quick overview of the cohort's risk distribution.
- Add the following code directly below the df = pd.read_sql_query(query, conn) line, still inside the generate_report() function:
# Calculate age from birth_date
def calculate_age(birth_date):
if not birth_date:
return None
try:
born = pd.to_datetime(birth_date)
today = pd.Timestamp.now()
return int((today - born).days / 365.25)
except (ValueError, TypeError):
return None
df["age"] = df["birth_date"].apply(calculate_age)
# Export full report
df.to_csv(REPORT_PATH, index=False)
print(f"\nFull report exported to: {REPORT_PATH}")
print(f"Total patients scored: {len(df)}")
# Summary statistics
print("\n--- SUMMARY STATISTICS ---")
print(f"\nRisk Distribution:")
risk_counts = df["risk_category"].value_counts()
for category in ["High", "Moderate", "Low"]:
count = risk_counts.get(category, 0)
pct = (count / len(df) * 100) if len(df) > 0 else 0
print(f" {category}: {count} patients ({pct:.1f}%)")
print(f"\nAverage LACE Score: {df['total_score'].mean():.1f}")
print(f"Median LACE Score: {df['total_score'].median():.1f}")What does this code do?
- calculate_age() converts birth dates to approximate years using the difference between today and the birth timestamp.
- df.to_csv(REPORT_PATH, index=False) exports the full DataFrame to a CSV file without pandas row indices.
- The summary loop prints the count and percentage of patients in each risk category (High, Moderate, Low) so you can see the distribution at a glance.
Add clinical context annotations and wire up the orchestrator
A care coordinator scanning this report needs to quickly see which comorbidities are driving each high-risk patient's score. You will also add a HIPAA reminder that documents what access controls a production version of this report would require.
- Add the following code directly below the Median LACE Score print line, still inside generate_report():
# Top 5 highest risk patients
print("\n--- TOP 5 HIGHEST RISK PATIENTS ---")
top5 = df.head(5)
for _, row in top5.iterrows():
print(f" Patient {row['patient_id'][:8]}... | "
f"Score: {row['total_score']} ({row['risk_category']}) | "
f"L={row['l_score']} A={row['a_score']} C={row['c_score']} E={row['e_score']} | "
f"Age: {row['age']} | LOS: {row['length_of_stay_days']}d")
if row["top_comorbidities"]:
print(f" Comorbidities: {row['top_comorbidities']}")
# HIPAA reminder
print("\n" + "=" * 60)
print("HIPAA REMINDER: In production, this report would require:")
print("- Role-based access controls (only authorized care coordinators)")
print("- Audit logging of who viewed this data and when")
print("- Encryption at rest for the database and CSV files")
print("- Data retention policies per organizational requirements")
print("=" * 60)
conn.close()What does this code do?
- df.head(5) grabs the top 5 rows. Since the data is already sorted by total score descending, these are the highest-risk patients.
- Each patient's LACE component breakdown (L, A, C, E) is printed so a reviewer can see which factor drives the risk.
- The HIPAA reminder documents the safeguards that would be required before deploying this report in a real clinical environment.
Now you need the main() function that orchestrates the entire pipeline from extraction through reporting.
- Add the main() orchestrator function at the bottom of main.py, after the closing of generate_report():
def main():
"""Run the complete pipeline."""
print("\n" + "=" * 60)
print("HOSPITAL READMISSION RISK PIPELINE")
print("Using LACE Index (van Walraven et al., CMAJ 2010)")
print("Data source: HAPI FHIR R4 Public Test Server (synthetic data)")
print("=" * 60 + "\n")
# Step 1: Extract
run_extraction()
# Step 2: Transform & Load
print()
run_transform()
# Step 3: Score
print()
run_scoring()
# Step 4: Report
print()
generate_report()
print("\nPipeline complete!")
if __name__ == "__main__":
main()What does the orchestrator do?
The main() function calls each pipeline stage in sequence: extract from the FHIR API, transform into SQLite, compute LACE scores, and generate the report. Running main.py executes the entire pipeline end-to-end.
- Save main.py.
- Run the full pipeline by executing this command in your terminal:
python main.pyThe pipeline will take a minute or two as it fetches data from the FHIR server. You should see progress messages for each stage (extraction, transformation, scoring, report generation).
At the end, you should see a summary with risk distribution percentages, average/median LACE scores, and the top 5 highest-risk patients with their score breakdowns.
Seeing a ModuleNotFoundError?
- If you see ModuleNotFoundError: No module named 'extract', make sure you are running the command from inside the readmission-risk-pipeline directory where all your .py files live.
- If you see ModuleNotFoundError: No module named 'pandas', your virtual environment may not be active. Run source venv/bin/activate and try again.
- If the FHIR server returns connection errors, it may be temporarily overloaded. Wait 30 seconds and try again.
Help me debug my pipeline run
- Confirm the CSV file was created by running:
ls data/readmission_risk_report.csvYou should see data/readmission_risk_report.csv listed in the output.
✔️ Awesome, I've got everything!
Great. Double check you have saved main.py and that your terminal output shows the risk distribution summary and top 5 patients.
ⓧ I'd like to double check the full code
Here is the complete main.py file. Compare yours against it:
"""
Hospital Readmission Risk Pipeline
Main orchestrator that runs the full ETL + scoring + reporting pipeline.
"""
import os
import sqlite3
import pandas as pd
from extract import run_extraction
from transform import run_transform, DB_PATH
from score import run_scoring
DATA_DIR = "data"
REPORT_PATH = os.path.join(DATA_DIR, "readmission_risk_report.csv")
def generate_report():
"""Generate the final risk-stratified cohort report."""
print("=" * 60)
print("COHORT REPORT GENERATION")
print("=" * 60)
conn = sqlite3.connect(DB_PATH)
# Build comprehensive report query
query = """
SELECT
ls.patient_id,
p.gender,
p.birth_date,
e.start_date AS last_admission_date,
e.length_of_stay_days,
ls.l_score,
ls.a_score,
ls.c_score,
ls.e_score,
ls.total_score,
ls.risk_category,
ls.top_comorbidities
FROM lace_scores ls
JOIN patients p ON ls.patient_id = p.id
LEFT JOIN encounters e ON ls.patient_id = e.patient_id
AND e.length_of_stay_days IS NOT NULL
GROUP BY ls.patient_id
HAVING e.start_date = MAX(e.start_date)
ORDER BY ls.total_score DESC
"""
df = pd.read_sql_query(query, conn)
# Calculate age from birth_date
def calculate_age(birth_date):
if not birth_date:
return None
try:
born = pd.to_datetime(birth_date)
today = pd.Timestamp.now()
return int((today - born).days / 365.25)
except (ValueError, TypeError):
return None
df["age"] = df["birth_date"].apply(calculate_age)
# Export full report
df.to_csv(REPORT_PATH, index=False)
print(f"\nFull report exported to: {REPORT_PATH}")
print(f"Total patients scored: {len(df)}")
# Summary statistics
print("\n--- SUMMARY STATISTICS ---")
print(f"\nRisk Distribution:")
risk_counts = df["risk_category"].value_counts()
for category in ["High", "Moderate", "Low"]:
count = risk_counts.get(category, 0)
pct = (count / len(df) * 100) if len(df) > 0 else 0
print(f" {category}: {count} patients ({pct:.1f}%)")
print(f"\nAverage LACE Score: {df['total_score'].mean():.1f}")
print(f"Median LACE Score: {df['total_score'].median():.1f}")
# Top 5 highest risk patients
print("\n--- TOP 5 HIGHEST RISK PATIENTS ---")
top5 = df.head(5)
for _, row in top5.iterrows():
print(f" Patient {row['patient_id'][:8]}... | "
f"Score: {row['total_score']} ({row['risk_category']}) | "
f"L={row['l_score']} A={row['a_score']} C={row['c_score']} E={row['e_score']} | "
f"Age: {row['age']} | LOS: {row['length_of_stay_days']}d")
if row["top_comorbidities"]:
print(f" Comorbidities: {row['top_comorbidities']}")
# HIPAA reminder
print("\n" + "=" * 60)
print("HIPAA REMINDER: In production, this report would require:")
print("- Role-based access controls (only authorized care coordinators)")
print("- Audit logging of who viewed this data and when")
print("- Encryption at rest for the database and CSV files")
print("- Data retention policies per organizational requirements")
print("=" * 60)
conn.close()
def main():
"""Run the complete pipeline."""
print("\n" + "=" * 60)
print("HOSPITAL READMISSION RISK PIPELINE")
print("Using LACE Index (van Walraven et al., CMAJ 2010)")
print("Data source: HAPI FHIR R4 Public Test Server (synthetic data)")
print("=" * 60 + "\n")
# Step 1: Extract
run_extraction()
# Step 2: Transform & Load
print()
run_transform()
# Step 3: Score
print()
run_scoring()
# Step 4: Report
print()
generate_report()
print("\nPipeline complete!")
if __name__ == "__main__":
main()Secret mission
Build a Comorbidity Deep-Dive with Care Prioritization
Your LACE scoring works, but a care coordinator needs more than a single number. They need to know which specific comorbidities are driving each patient's risk and who to call first. In this extension, you'll map SNOMED codes to Charlson categories and build a multi-factor prioritization list that ranks the highest-risk patients for immediate outreach.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally, so there are no ongoing costs or cloud resources to worry about.
Decide whether to keep your project files for future reference or delete them to free up disk space.
Resources you used:
- Local project directory (readmission-risk-pipeline/) containing Python scripts, virtual environment, and pip packages.
- SQLite database file (data/readmission.db) with patients, encounters, conditions, and LACE scores tables.
- CSV reports (data/readmission_risk_report.csv and data/care_priority_list.csv).
- Raw FHIR JSON files (data/raw_encounters.json, data/raw_patients.json, data/raw_conditions.json).
No cleanup of external resources is needed. The HAPI FHIR server is public and read-only, so nothing was created on their end.
Keep everything running
No action needed. Choose this if you want to keep experimenting with the pipeline or query the SQLite database further.
- Your project directory contains everything. All scripts, data, and the virtual environment stay on your machine.
- You can query the SQLite database anytime with the sqlite3 CLI or any database browser tool.
- To reactivate the virtual environment later, navigate to the readmission-risk-pipeline directory and run source venv/bin/activate.
- Re-running python main.py will fetch fresh data from the HAPI FHIR server and regenerate all reports.
Pause - I'll come back to this later
Since everything is local, there is nothing actively consuming resources while idle. The only action is to deactivate the virtual environment if it is still active in your terminal.
- Deactivate the virtual environment by running:
deactivate- Your files, database, and reports remain untouched on disk until you return.
Delete - I don't want to use this again
Remove the entire project directory to clear all scripts, the virtual environment, the database, and all generated reports.
- Deactivate the virtual environment first (if still active):
deactivate- Navigate to the parent directory and remove the project folder:
macOS
cd ~/Desktop
rm -rf readmission-risk-pipelineWindows
cd ~\Desktop
rmdir /s /q readmission-risk-pipeline- This removes all Python scripts, the virtual environment (including installed packages), the SQLite database, raw JSON files, and all CSV reports.
- No external accounts or cloud resources need deletion. The HAPI FHIR server is public and you did not create anything on it.
Nice Work!
Nice Work!
You just built a complete healthcare data engineering pipeline from scratch. Starting with a live FHIR R4 server, you extracted real clinical resources, transformed them into an analytics-ready schema, and computed clinically validated readmission risk scores.
You've learned how to:
- Build a Python ETL pipeline that extracts live healthcare data from a FHIR R4 API, including Bundle pagination and RESTful search patterns.
- Implement the clinically validated LACE index for 30-day readmission risk prediction, applying HIPAA awareness principles throughout.
- Transform nested FHIR JSON into a normalized SQLite analytics schema and generate a risk-stratified patient cohort report ready for clinical review.
- Secret Mission: Deep-dived into Charlson comorbidity mapping with SNOMED code lookups and built a care prioritization list ranking high-risk patients by multiple risk factors.
Ready to quiz yourself?