Patient Readmission Risk Prediction API
Build a healthcare ML microservice with MySQL, scikit-learn, and FastAPI.
Introduction
30 Second Summary
When a patient leaves the hospital, the last thing anyone wants is for them to bounce back within 30 days. Yet 30-day readmissions cost the US healthcare system over $26 billion annually, and hospitals face financial penalties for high readmission rates.
In this project, you will build a patient readmission risk prediction microservice. You'll design a clinical database, train a machine learning model on hospital data, and deploy a prediction API that clinical staff can call to identify at-risk patients before discharge.
What You'll Build
A patient readmission risk prediction microservice built the way a Clinical Analytics team at a large health system would build it, using MySQL, Logistic Regression, and FastAPI.
By the end of this project, you'll have:
- A production-style MySQL database with proper schema design, indexes, foreign keys, and a dedicated service account modeling a hospital's clinical data.
- A trained Logistic Regression model that predicts 30-day patient readmission risk using features extracted via SQL.
- A running FastAPI microservice with environment-based configuration that clinical staff could call to get a readmission risk score.
- Secret Mission: a risk stratification dashboard endpoint that returns all high-risk patients sorted by risk level with batch predictions.
Are there any prerequisites?
You need basic Python knowledge (variables, functions, lists). No prior SQL or machine learning experience is required. This project assumes you already have Python 3.11+ and MySQL installed on your machine.
Set Up Your Project and MySQL Database
Every production service starts with a solid foundation. Before writing any application logic, you need a project folder, the right Python packages, and a running database with proper access controls.
In this step, you'll scaffold your project, install three key dependencies, spin up MySQL as a background service, and create a dedicated database with its own service account.
In this step, get ready to:
- Create your project folder with a virtual environment and install dependencies.
- Start MySQL and create a database with a dedicated service account.
- Configure environment-based credentials for your application.
Create the project and install Python dependencies
Your project needs a dedicated folder, an isolated Python environment, and three packages: a MySQL connector, a machine learning library, and a web framework.
- Press Cmd+Space to open Spotlight Search.
- Type Terminal 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 clinical-analytics
cd clinical-analytics- Confirm the folder exists by running:
lsYou should see the clinical-analytics folder listed in your Desktop output.
- Create a Python virtual environment and activate it by running these commands:
python3 -m venv venv
source venv/bin/activateYour terminal prompt now shows (venv) at the beginning, confirming the virtual environment is active.
What is a virtual environment?
A virtual environment is an isolated Python installation that keeps this project's packages separate from your system Python. Every package you install stays inside the venv folder, so different projects never conflict with each other.
- Install the three project dependencies by running:
pip install mysql-connector-python==9.7.0 scikit-learn==1.8.0 "fastapi[standard]==0.136.3"What are these packages?
mysql-connector-python is Oracle's official Python connector for MySQL databases. scikit-learn provides machine learning algorithms including Logistic Regression. fastapi[standard] is a modern web framework that includes uvicorn for running your API server.
You should see a series of download and install messages ending with a success confirmation.
Seeing install errors?
If scikit-learn fails to install, confirm your Python version is 3.11 or higher by running python3 --version. This version of scikit-learn requires Python 3.11+.
Make sure your virtual environment is active (you see (venv) in your prompt). If not, run source venv/bin/activate again.
Help me fix my pip install errors
- Create the empty project files by running:
touch config.py database.py seed_data.py queries.py train_model.py main.py .env .gitignore- Confirm the files were created by running:
ls -aYou should see all seven Python files plus .env and .gitignore listed in the output.
Start MySQL and create the database and service account
In a real company, the database team never gives application code direct root access. Instead, they create a dedicated service account with only the permissions the app needs. You'll do the same here.
- Start MySQL as a background service by running:
brew services start mysqlYou should see a message confirming MySQL has started successfully.
MySQL already running?
If you see a message saying MySQL is already started, that's fine. It means the service was previously running and you can proceed to the next step.
If you see a bootstrap or permission error, try running brew services restart mysql instead.
Help me start MySQL on macOS
- Connect to MySQL as the root user by running:
mysql -u rootYour prompt changes to mysql> indicating you're inside the MySQL command-line client.
Why no password for root?
Homebrew installs MySQL without a root password by default. This is fine for local development. In production, you would secure root with mysql_secure_installation.
- Create the project database by running this SQL command:
CREATE DATABASE clinical_analytics;- Create a dedicated service account for your application by running:
CREATE USER 'analytics_svc'@'localhost' IDENTIFIED BY 'your-password-here';- Grant the service account only the permissions it needs by running:
GRANT SELECT, INSERT, UPDATE, DELETE ON clinical_analytics.* TO 'analytics_svc'@'localhost';- Apply the permission changes by running:
FLUSH PRIVILEGES;Why a service account instead of root?
Production applications never use root to connect to a database. A service account with scoped permissions means that even if your application code has a bug, it can only SELECT, INSERT, UPDATE, and DELETE. It cannot DROP tables or CREATE new users.
This is the principle of least privilege. Every real company enforces it.
- Verify the database was created by running:
SHOW DATABASES;You should see clinical_analytics in the list of databases.
- Exit the MySQL client by running:
EXIT;Your prompt returns to the normal terminal (with the (venv) prefix).
Configure environment-based credentials
Hardcoding database passwords into source code is a security risk. Production services read credentials from environment variables so secrets never end up in version control.
- Open your project folder in VS Code. Press Cmd+Space to open Spotlight Search.
- Type VS Code and press Enter to open it.
- Click File in the top menu bar.
- Click Open Folder.
- Navigate to Desktop, select the clinical-analytics folder, and click Open.
- Open .env from the Explorer sidebar and paste the following credentials:
DB_HOST=localhost
DB_USER=analytics_svc
DB_PASSWORD=your-password-here
DB_NAME=clinical_analyticsWhat does this file do?
The .env file stores configuration values that your Python code reads at runtime. Each line is a key-value pair. The password here matches the one you set in the CREATE USER statement.
- Open config.py and paste the following code:
import os
DB_CONFIG = {
"host": os.environ.get("DB_HOST", "localhost"),
"user": os.environ.get("DB_USER", "analytics_svc"),
"password": os.environ.get("DB_PASSWORD", "your-password-here"),
"database": os.environ.get("DB_NAME", "clinical_analytics"),
}What does this code do?
os.environ.get() reads a value from the system's environment variables. The second argument is a fallback default. If the environment variable isn't set, Python uses the default value instead.
This pattern means your app works out of the box locally (the defaults match your credentials), but can be reconfigured in production by setting different environment variables.
- Open .gitignore and add a single line to keep your secrets out of version control:
.envWhy add .env to .gitignore?
The .gitignore file tells Git to ignore specified files when committing code. Your .env contains database credentials. If you ever push this project to GitHub, those credentials would be exposed without this safeguard.
- Save all three files (.env, config.py, .gitignore).
✔️ Awesome, I've got everything!
Great. Double check you've saved all files before moving on.
ⓧ I'd like to double check the full code
Here is what each file should look like at this point:
import os
DB_CONFIG = {
"host": os.environ.get("DB_HOST", "localhost"),
"user": os.environ.get("DB_USER", "analytics_svc"),
"password": os.environ.get("DB_PASSWORD", "your-password-here"),
"database": os.environ.get("DB_NAME", "clinical_analytics"),
}DB_HOST=localhost
DB_USER=analytics_svc
DB_PASSWORD=your-password-here
DB_NAME=clinical_analytics.envYour infrastructure is ready. You have a running MySQL database, a service account with scoped permissions, and a Python project configured to connect securely. Next up, you'll design the clinical database schema with five interconnected tables.
Design the Clinical Database Schema
Your project environment is connected to a running MySQL database with a dedicated service account. Now it is time to give that empty database its structure.
In a real hospital system, the database schema defines how clinical data flows between departments. You will create five interconnected tables that model patients, their doctors, hospital visits, diagnoses, and prescriptions. Each table uses typed columns, foreign keys to enforce relationships, and indexes to speed up the queries your ML model will run later.
In this step, get ready to:
- Create the patients and providers tables with proper MySQL types and an index.
- Create the encounters, diagnoses, and prescriptions tables with foreign keys and a stored procedure.
- Run the schema creation script and verify all tables exist in the MySQL CLI.
Create the patients and providers tables
Before writing any CREATE TABLE statements, you need a reusable function that connects Python to your MySQL database. This helper will be imported by every other script in the project.
- In database.py, add the imports and connection helper by pasting the following code:
import mysql.connector
from config import DB_CONFIG
def get_connection():
"""Create and return a MySQL database connection."""
return mysql.connector.connect(**DB_CONFIG)What does this code do?
import mysql.connector brings in Oracle's official MySQL driver for Python. It provides the mysql.connector.connect() function that opens a connection to your database.
The get_connection() function passes your credentials (from DB_CONFIG in config.py) as keyword arguments. Every other script will call this function instead of hardcoding connection details.
Now you will define the schema itself. The patients table stores demographic information. The providers table stores your hospital's clinical staff.
- Below get_connection(), add the create_tables() function with the patients and providers table definitions:
def create_tables():
"""Create the clinical analytics database schema."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS patients (
patient_id INT AUTO_INCREMENT PRIMARY KEY,
first_name VARCHAR(100) NOT NULL,
last_name VARCHAR(100) NOT NULL,
date_of_birth DATE NOT NULL,
gender ENUM('M', 'F', 'O') NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_patient_name (last_name, first_name)
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS providers (
provider_id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(200) NOT NULL,
specialty VARCHAR(100) NOT NULL,
department VARCHAR(100) NOT NULL
)
""")What does this code do?
- INT AUTO_INCREMENT PRIMARY KEY creates a unique, self-incrementing ID for each row. MySQL assigns the next number automatically on insert.
- VARCHAR(100) stores variable-length text up to 100 characters. DATE stores calendar dates. ENUM restricts a column to a fixed set of allowed values.
- INDEX idx_patient_name (last_name, first_name) creates a composite index that speeds up searches by name. Without it, MySQL would scan every row.
- NOT NULL enforces that a value must be provided. Clinical data should never have missing names or birth dates.
Create the encounters, diagnoses, and prescriptions tables
The remaining three tables track what happens during a hospital visit. Each one links back to earlier tables using foreign keys, which guarantee that every encounter belongs to a real patient and a real provider.
- Below the providers CREATE TABLE statement, add the encounters table:
cursor.execute("""
CREATE TABLE IF NOT EXISTS encounters (
encounter_id INT AUTO_INCREMENT PRIMARY KEY,
patient_id INT NOT NULL,
provider_id INT NOT NULL,
admission_date DATE NOT NULL,
discharge_date DATE,
encounter_type ENUM('emergency', 'inpatient', 'outpatient') NOT NULL,
readmitted_30_day TINYINT DEFAULT 0,
FOREIGN KEY (patient_id) REFERENCES patients(patient_id),
FOREIGN KEY (provider_id) REFERENCES providers(provider_id),
INDEX idx_patient_dates (patient_id, admission_date),
INDEX idx_discharge (discharge_date)
)
""")What does this code do?
- FOREIGN KEY (patient_id) REFERENCES patients(patient_id) tells MySQL that every encounter must point to a valid patient. If you try to insert an encounter for a patient that does not exist, MySQL will reject it.
- readmitted_30_day TINYINT DEFAULT 0 is the label your ML model will learn to predict. A value of 1 means the patient was readmitted within 30 days of discharge.
- INDEX idx_patient_dates (patient_id, admission_date) is a composite index that accelerates queries filtering by both patient and date range. Your feature extraction query will rely on this.
The diagnoses and prescriptions tables attach clinical details to each encounter. After those, you will add a stored procedure that automatically flags readmissions.
- Below the encounters statement, add the diagnoses table, prescriptions table, stored procedure, and the function's closing lines:
cursor.execute("""
CREATE TABLE IF NOT EXISTS diagnoses (
diagnosis_id INT AUTO_INCREMENT PRIMARY KEY,
encounter_id INT NOT NULL,
icd_code VARCHAR(10) NOT NULL,
description VARCHAR(255) NOT NULL,
FOREIGN KEY (encounter_id) REFERENCES encounters(encounter_id),
INDEX idx_encounter (encounter_id),
INDEX idx_icd (icd_code)
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS prescriptions (
prescription_id INT AUTO_INCREMENT PRIMARY KEY,
encounter_id INT NOT NULL,
medication VARCHAR(200) NOT NULL,
dosage VARCHAR(100) NOT NULL,
FOREIGN KEY (encounter_id) REFERENCES encounters(encounter_id),
INDEX idx_rx_encounter (encounter_id)
)
""")
# Create stored procedure to flag readmissions
cursor.execute("DROP PROCEDURE IF EXISTS flag_readmissions")
cursor.execute("""
CREATE PROCEDURE flag_readmissions()
BEGIN
UPDATE encounters e1
JOIN encounters e2
ON e1.patient_id = e2.patient_id
AND e1.encounter_id != e2.encounter_id
AND e1.admission_date > e2.discharge_date
AND DATEDIFF(e1.admission_date, e2.discharge_date) BETWEEN 1 AND 30
SET e1.readmitted_30_day = 1;
END
""")
conn.commit()
cursor.close()
conn.close()
print("Schema created successfully.")
if __name__ == "__main__":
create_tables()What does this code do?
- The diagnoses table stores ICD-10 codes (the standard coding system hospitals use for medical conditions). The index on icd_code speeds up queries that filter by diagnosis.
- The flag_readmissions stored procedure uses a self-JOIN on the encounters table. It finds cases where the same patient was admitted within 1 to 30 days of a prior discharge, then sets readmitted_30_day = 1 on those encounters.
- DATEDIFF(e1.admission_date, e2.discharge_date) BETWEEN 1 AND 30 calculates the number of days between two dates. This is how you define the 30-day readmission window.
- conn.commit() saves all schema changes to the database. Without it, MySQL would discard everything when the connection closes.
Why a stored procedure instead of a raw query?
In production hospital systems, business logic like readmission flagging lives in stored procedures that run on a schedule. Wrapping the logic in CALL flag_readmissions() means any team member can trigger it without knowing the underlying SQL.
- Save database.py.
✔️ Awesome, I've got everything!
Great. Double check you have saved the file before running it in the next substep.
ⓧ I'd like to double check the full code
import mysql.connector
from config import DB_CONFIG
def get_connection():
"""Create and return a MySQL database connection."""
return mysql.connector.connect(**DB_CONFIG)
def create_tables():
"""Create the clinical analytics database schema."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS patients (
patient_id INT AUTO_INCREMENT PRIMARY KEY,
first_name VARCHAR(100) NOT NULL,
last_name VARCHAR(100) NOT NULL,
date_of_birth DATE NOT NULL,
gender ENUM('M', 'F', 'O') NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_patient_name (last_name, first_name)
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS providers (
provider_id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(200) NOT NULL,
specialty VARCHAR(100) NOT NULL,
department VARCHAR(100) NOT NULL
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS encounters (
encounter_id INT AUTO_INCREMENT PRIMARY KEY,
patient_id INT NOT NULL,
provider_id INT NOT NULL,
admission_date DATE NOT NULL,
discharge_date DATE,
encounter_type ENUM('emergency', 'inpatient', 'outpatient') NOT NULL,
readmitted_30_day TINYINT DEFAULT 0,
FOREIGN KEY (patient_id) REFERENCES patients(patient_id),
FOREIGN KEY (provider_id) REFERENCES providers(provider_id),
INDEX idx_patient_dates (patient_id, admission_date),
INDEX idx_discharge (discharge_date)
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS diagnoses (
diagnosis_id INT AUTO_INCREMENT PRIMARY KEY,
encounter_id INT NOT NULL,
icd_code VARCHAR(10) NOT NULL,
description VARCHAR(255) NOT NULL,
FOREIGN KEY (encounter_id) REFERENCES encounters(encounter_id),
INDEX idx_encounter (encounter_id),
INDEX idx_icd (icd_code)
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS prescriptions (
prescription_id INT AUTO_INCREMENT PRIMARY KEY,
encounter_id INT NOT NULL,
medication VARCHAR(200) NOT NULL,
dosage VARCHAR(100) NOT NULL,
FOREIGN KEY (encounter_id) REFERENCES encounters(encounter_id),
INDEX idx_rx_encounter (encounter_id)
)
""")
# Create stored procedure to flag readmissions
cursor.execute("DROP PROCEDURE IF EXISTS flag_readmissions")
cursor.execute("""
CREATE PROCEDURE flag_readmissions()
BEGIN
UPDATE encounters e1
JOIN encounters e2
ON e1.patient_id = e2.patient_id
AND e1.encounter_id != e2.encounter_id
AND e1.admission_date > e2.discharge_date
AND DATEDIFF(e1.admission_date, e2.discharge_date) BETWEEN 1 AND 30
SET e1.readmitted_30_day = 1;
END
""")
conn.commit()
cursor.close()
conn.close()
print("Schema created successfully.")
if __name__ == "__main__":
create_tables()Run the schema creation script
Your schema is defined in code. Now run it to create the actual tables in MySQL, then verify they exist with the correct structure.
- Make sure your .env variables are loaded in your terminal session. Source the file by running:
export $(cat .env | xargs)- Execute the schema creation script by running:
python database.py- You should see Schema created successfully. printed in the terminal.
Seeing an Access denied error?
- Check that your .env file has the correct password for the analytics_svc user you created in Step 1.
- Confirm the environment variables are loaded by running echo $DB_USER in your terminal. If it prints nothing, re-run the export command above.
- Make sure MySQL is still running with brew services list. If mysql shows stopped, start it again with brew services start mysql.
Help me debug my MySQL connection error in database.py
Now verify the tables directly in the MySQL CLI.
- Connect to MySQL by running:
mysql -u root- Switch to your database and list the tables by running these commands inside the MySQL prompt:
USE clinical_analytics;
SHOW TABLES;- You should see all 5 tables listed: diagnoses, encounters, patients, prescriptions, and providers.
- Inspect the encounters table to confirm column types, foreign keys, and indexes:
DESCRIBE encounters;- You should see columns including encounter_id (int, PRI), patient_id (int, MUL), encounter_type (enum), and readmitted_30_day (tinyint with default 0).
- Exit the MySQL prompt by typing exit;.
What does MUL mean in the Key column?
MUL stands for "multiple." It means the column is the first column of a non-unique index (your foreign key or named index). PRI means primary key. These Key markers confirm your indexes and foreign keys were created correctly.
Don't see all 5 tables?
- If you see fewer tables, check database.py for syntax errors in the SQL strings. A misplaced comma or missing closing parenthesis will cause MySQL to reject the statement silently.
- Tables must be created in dependency order. patients and providers must come before encounters (which references them). encounters must come before diagnoses and prescriptions.
Help me figure out why my tables were not created
Your database schema is locked in. Five tables with typed columns, foreign key relationships, indexes for performance, and a stored procedure ready to flag readmissions. Next up, you will fill these tables with realistic clinical data.
Populate with Realistic Clinical Data
Your database schema is ready with 5 interconnected tables, foreign keys, and indexes. Now it needs data.
In a real hospital system, clinical data follows patterns. Some patients visit once and never return. Others are "high-utilizers" who cycle through the emergency department every few weeks. In this step, you will generate synthetic data that mirrors these real-world patterns so your ML model has meaningful signals to learn from.
In this step, get ready to:
- Generate 300 patients and 20 providers with realistic demographics.
- Create 800+ hospital encounters with temporal readmission patterns.
- Attach diagnoses and prescriptions, then flag 30-day readmissions.
Generate synthetic patient and provider data
Production databases load data in bulk using batch inserts. Instead of inserting one row at a time (300 separate network round-trips), cursor.executemany() batches all rows into a single optimized INSERT statement.
- Open seed_data.py and add the imports, connection helper, and provider/patient generation by pasting this code:
import random
from datetime import date, timedelta
import mysql.connector
from config import DB_CONFIG
def get_connection():
return mysql.connector.connect(**DB_CONFIG)
def seed():
"""Generate and insert synthetic clinical data."""
conn = get_connection()
cursor = conn.cursor()
specialties = [
("Dr. Sarah Chen", "Cardiology", "Heart & Vascular"),
("Dr. Michael Torres", "Cardiology", "Heart & Vascular"),
("Dr. Priya Patel", "Pulmonology", "Respiratory"),
("Dr. James Wright", "Pulmonology", "Respiratory"),
("Dr. Lisa Kim", "Internal Medicine", "General Medicine"),
("Dr. Robert Johnson", "Internal Medicine", "General Medicine"),
("Dr. Amanda Foster", "Internal Medicine", "General Medicine"),
("Dr. David Okafor", "Emergency Medicine", "Emergency"),
("Dr. Maria Gonzalez", "Emergency Medicine", "Emergency"),
("Dr. Kevin Shah", "Emergency Medicine", "Emergency"),
("Dr. Rachel Adams", "Nephrology", "Kidney & Dialysis"),
("Dr. Thomas Lee", "Nephrology", "Kidney & Dialysis"),
("Dr. Emily Watson", "Endocrinology", "Diabetes & Metabolism"),
("Dr. Carlos Rivera", "Endocrinology", "Diabetes & Metabolism"),
("Dr. Nicole Brown", "Gastroenterology", "Digestive Health"),
("Dr. Anthony Russo", "Gastroenterology", "Digestive Health"),
("Dr. Jennifer Nguyen", "Neurology", "Brain & Spine"),
("Dr. William Park", "Orthopedics", "Musculoskeletal"),
("Dr. Stephanie Davis", "Oncology", "Cancer Center"),
("Dr. Daniel Martinez", "Psychiatry", "Behavioral Health"),
]
cursor.executemany(
"INSERT INTO providers (name, specialty, department) VALUES (%s, %s, %s)",
specialties
)
conn.commit()
first_names_m = ["James", "Robert", "John", "Michael", "David", "William",
"Richard", "Joseph", "Thomas", "Christopher", "Daniel",
"Matthew", "Anthony", "Mark", "Steven"]
first_names_f = ["Mary", "Patricia", "Jennifer", "Linda", "Elizabeth",
"Barbara", "Susan", "Jessica", "Sarah", "Karen",
"Lisa", "Nancy", "Betty", "Margaret", "Dorothy"]
last_names = ["Smith", "Johnson", "Williams", "Brown", "Jones", "Garcia",
"Miller", "Davis", "Rodriguez", "Martinez", "Hernandez",
"Lopez", "Gonzalez", "Wilson", "Anderson", "Thomas",
"Taylor", "Moore", "Jackson", "Martin", "Lee", "Perez",
"Thompson", "White", "Harris", "Sanchez", "Clark",
"Ramirez", "Lewis", "Robinson"]
patients = []
for _ in range(300):
gender = random.choice(["M", "F"])
if gender == "M":
first = random.choice(first_names_m)
else:
first = random.choice(first_names_f)
last = random.choice(last_names)
# Age between 18 and 95
age = random.randint(18, 95)
dob = date.today() - timedelta(days=age * 365 + random.randint(0, 364))
patients.append((first, last, dob.isoformat(), gender))
cursor.executemany(
"INSERT INTO patients (first_name, last_name, date_of_birth, gender) VALUES (%s, %s, %s, %s)",
patients
)
conn.commit()What does this code do?
The specialties list holds 20 providers across 11 departments. Each tuple maps a provider name to a specialty and department, matching the columns in your providers table.
The patient loop generates 300 patients with random genders, names drawn from common first/last name pools, and dates of birth calculated from a random age (18-95). cursor.executemany() batches all 300 rows into a single INSERT call, which is dramatically faster than looping with cursor.execute() 300 times.
Why %s placeholders instead of f-strings?
Parameterized queries (%s placeholders) are the industry standard for preventing SQL injection. The MySQL connector handles escaping and type conversion automatically. Never embed user-supplied values directly into SQL strings with f-strings or string formatting.
Generate encounters with temporal patterns
The readmission prediction model needs a realistic signal to learn from. Some patients should return within 30 days of discharge (the readmission signal), while others have longer gaps between visits.
- Add the encounter generation code directly below the patients section (still inside the seed() function):
diagnosis_options = [
("E11.9", "Type 2 Diabetes Mellitus without complications"),
("I10", "Essential Hypertension"),
("I50.9", "Heart Failure, unspecified"),
("J44.1", "COPD with acute exacerbation"),
("N18.3", "Chronic Kidney Disease, stage 3"),
("I25.10", "Atherosclerotic Heart Disease"),
("E78.5", "Hyperlipidemia, unspecified"),
("J18.9", "Pneumonia, unspecified organism"),
("K21.0", "GERD with esophagitis"),
("M54.5", "Low Back Pain"),
("I48.91", "Atrial Fibrillation, unspecified"),
("E11.65", "Type 2 Diabetes with Hyperglycemia"),
("J96.01", "Acute Respiratory Failure with Hypoxia"),
("N17.9", "Acute Kidney Failure, unspecified"),
("I21.9", "Acute Myocardial Infarction, unspecified"),
]
medication_options = [
("Metformin", "500mg twice daily"),
("Metformin", "1000mg twice daily"),
("Lisinopril", "10mg once daily"),
("Lisinopril", "20mg once daily"),
("Atorvastatin", "40mg once daily"),
("Atorvastatin", "80mg once daily"),
("Amlodipine", "5mg once daily"),
("Metoprolol Succinate", "25mg once daily"),
("Metoprolol Succinate", "50mg once daily"),
("Omeprazole", "20mg once daily"),
("Furosemide", "40mg once daily"),
("Furosemide", "80mg twice daily"),
("Albuterol Inhaler", "2 puffs every 4-6 hours PRN"),
("Aspirin", "81mg once daily"),
("Insulin Glargine", "20 units subcutaneous at bedtime"),
("Warfarin", "5mg once daily"),
("Carvedilol", "12.5mg twice daily"),
("Spironolactone", "25mg once daily"),
("Apixaban", "5mg twice daily"),
("Prednisone", "40mg daily taper"),
]
encounter_types = ["emergency", "inpatient", "outpatient"]
base_date = date(2024, 1, 1)
encounters_data = []
for patient_id in range(1, 301):
# Some patients are high utilizers
num_encounters = random.choices([1, 2, 3, 4, 5, 6], weights=[30, 30, 20, 10, 7, 3])[0]
last_discharge = None
for _ in range(num_encounters):
if last_discharge:
# 25% chance of readmission within 30 days
days_gap = random.choices(
[random.randint(1, 30), random.randint(31, 200)],
weights=[25, 75]
)[0]
admission = last_discharge + timedelta(days=days_gap)
else:
admission = base_date + timedelta(days=random.randint(0, 500))
stay = random.randint(1, 14)
discharge = admission + timedelta(days=stay)
enc_type = random.choices(encounter_types, weights=[25, 50, 25])[0]
provider_id = random.randint(1, 20)
encounters_data.append((
patient_id, provider_id,
admission.isoformat(), discharge.isoformat(), enc_type
))
last_discharge = discharge
cursor.executemany(
"INSERT INTO encounters (patient_id, provider_id, admission_date, discharge_date, encounter_type) VALUES (%s, %s, %s, %s, %s)",
encounters_data
)
conn.commit()How does this model readmission patterns?
The weights=[30, 30, 20, 10, 7, 3] distribution means most patients have 1-2 visits, but a small percentage are high-utilizers with 4-6 encounters. This mirrors real hospital populations.
For returning patients, weights=[25, 75] gives a 25% probability that the next admission falls within 30 days of the prior discharge. This creates the readmission signal your ML model will learn to predict.
The diagnosis_options and medication_options lists are defined here but used in the next section when attaching clinical details to each encounter.
Generate diagnoses and prescriptions, then flag readmissions
Each encounter in a real hospital has diagnoses (coded with ICD-10) and prescriptions attached to it. After inserting all the clinical details, you will call the stored procedure you created in the previous step to automatically flag 30-day readmissions.
- Add this final section to complete the seed() function, followed by the __main__ block:
cursor.execute("SELECT encounter_id FROM encounters")
encounter_ids = [row[0] for row in cursor.fetchall()]
diagnoses_data = []
prescriptions_data = []
for enc_id in encounter_ids:
num_dx = random.randint(1, 4)
for dx in random.sample(diagnosis_options, num_dx):
diagnoses_data.append((enc_id, dx[0], dx[1]))
num_rx = random.randint(1, 5)
for rx in random.sample(medication_options, num_rx):
prescriptions_data.append((enc_id, rx[0], rx[1]))
cursor.executemany(
"INSERT INTO diagnoses (encounter_id, icd_code, description) VALUES (%s, %s, %s)",
diagnoses_data
)
cursor.executemany(
"INSERT INTO prescriptions (encounter_id, medication, dosage) VALUES (%s, %s, %s)",
prescriptions_data
)
conn.commit()
cursor.callproc("flag_readmissions")
conn.commit()
cursor.execute("SELECT COUNT(*) FROM patients")
print(f"Patients: {cursor.fetchone()[0]}")
cursor.execute("SELECT COUNT(*) FROM encounters")
print(f"Encounters: {cursor.fetchone()[0]}")
cursor.execute("SELECT COUNT(*) FROM encounters WHERE readmitted_30_day = 1")
print(f"Readmissions flagged: {cursor.fetchone()[0]}")
cursor.execute("SELECT COUNT(*) FROM diagnoses")
print(f"Diagnoses: {cursor.fetchone()[0]}")
cursor.execute("SELECT COUNT(*) FROM prescriptions")
print(f"Prescriptions: {cursor.fetchone()[0]}")
cursor.close()
conn.close()
print("\nData seeded successfully.")
if __name__ == "__main__":
seed()What does this code do?
The loop queries all encounter IDs, then attaches 1-4 random ICD-10 diagnoses and 1-5 random prescriptions to each one. random.sample() picks without replacement, so no encounter gets duplicate diagnoses.
cursor.callproc("flag_readmissions") executes the stored procedure you created in the previous step. It uses a self-JOIN to find encounters where the same patient was admitted within 30 days of a prior discharge, and sets readmitted_30_day = 1 on those rows.
The summary print statements let you verify the data loaded correctly without opening the MySQL CLI.
- Save seed_data.py.
- Run the seeding script by executing this command in your terminal:
python seed_data.py- You should see output similar to this:
Patients: 300
Encounters: 616
Readmissions flagged: 72
Diagnoses: 1523
Prescriptions: 1845
Data seeded successfully.Why do my numbers differ?
The exact counts for encounters, readmissions, diagnoses, and prescriptions will vary each time you run the script because the data is randomly generated. Patients will always be 300, but expect encounters in the range of 600-900 and readmission flags in the range of 50-150.
Seeing a connection error?
If you see a message about access denied or connection refused, make sure MySQL is running (brew services start mysql) and that the credentials in your .env file match the user you created in the previous step.
If you see a "Not all parameters were used" error, check that each tuple in your data lists has the exact same number of elements as %s placeholders in the corresponding INSERT statement.
help me debug my seed_data.py connection or insert error
- Verify the data in MySQL by running:
mysql -u root -e "USE clinical_analytics; SELECT COUNT(*) FROM encounters WHERE readmitted_30_day = 1;"You should see a count of readmissions matching what your script printed. This confirms the stored procedure ran successfully and flagged the right encounters.
✔️ Awesome, I've got everything!
Great. Double check you have saved seed_data.py and confirmed the readmission flags in MySQL.
ⓧ I'd like to double check the full code
Here is the complete seed_data.py file:
import random
from datetime import date, timedelta
import mysql.connector
from config import DB_CONFIG
def get_connection():
return mysql.connector.connect(**DB_CONFIG)
def seed():
"""Generate and insert synthetic clinical data."""
conn = get_connection()
cursor = conn.cursor()
specialties = [
("Dr. Sarah Chen", "Cardiology", "Heart & Vascular"),
("Dr. Michael Torres", "Cardiology", "Heart & Vascular"),
("Dr. Priya Patel", "Pulmonology", "Respiratory"),
("Dr. James Wright", "Pulmonology", "Respiratory"),
("Dr. Lisa Kim", "Internal Medicine", "General Medicine"),
("Dr. Robert Johnson", "Internal Medicine", "General Medicine"),
("Dr. Amanda Foster", "Internal Medicine", "General Medicine"),
("Dr. David Okafor", "Emergency Medicine", "Emergency"),
("Dr. Maria Gonzalez", "Emergency Medicine", "Emergency"),
("Dr. Kevin Shah", "Emergency Medicine", "Emergency"),
("Dr. Rachel Adams", "Nephrology", "Kidney & Dialysis"),
("Dr. Thomas Lee", "Nephrology", "Kidney & Dialysis"),
("Dr. Emily Watson", "Endocrinology", "Diabetes & Metabolism"),
("Dr. Carlos Rivera", "Endocrinology", "Diabetes & Metabolism"),
("Dr. Nicole Brown", "Gastroenterology", "Digestive Health"),
("Dr. Anthony Russo", "Gastroenterology", "Digestive Health"),
("Dr. Jennifer Nguyen", "Neurology", "Brain & Spine"),
("Dr. William Park", "Orthopedics", "Musculoskeletal"),
("Dr. Stephanie Davis", "Oncology", "Cancer Center"),
("Dr. Daniel Martinez", "Psychiatry", "Behavioral Health"),
]
cursor.executemany(
"INSERT INTO providers (name, specialty, department) VALUES (%s, %s, %s)",
specialties
)
conn.commit()
first_names_m = ["James", "Robert", "John", "Michael", "David", "William",
"Richard", "Joseph", "Thomas", "Christopher", "Daniel",
"Matthew", "Anthony", "Mark", "Steven"]
first_names_f = ["Mary", "Patricia", "Jennifer", "Linda", "Elizabeth",
"Barbara", "Susan", "Jessica", "Sarah", "Karen",
"Lisa", "Nancy", "Betty", "Margaret", "Dorothy"]
last_names = ["Smith", "Johnson", "Williams", "Brown", "Jones", "Garcia",
"Miller", "Davis", "Rodriguez", "Martinez", "Hernandez",
"Lopez", "Gonzalez", "Wilson", "Anderson", "Thomas",
"Taylor", "Moore", "Jackson", "Martin", "Lee", "Perez",
"Thompson", "White", "Harris", "Sanchez", "Clark",
"Ramirez", "Lewis", "Robinson"]
patients = []
for _ in range(300):
gender = random.choice(["M", "F"])
if gender == "M":
first = random.choice(first_names_m)
else:
first = random.choice(first_names_f)
last = random.choice(last_names)
# Age between 18 and 95
age = random.randint(18, 95)
dob = date.today() - timedelta(days=age * 365 + random.randint(0, 364))
patients.append((first, last, dob.isoformat(), gender))
cursor.executemany(
"INSERT INTO patients (first_name, last_name, date_of_birth, gender) VALUES (%s, %s, %s, %s)",
patients
)
conn.commit()
diagnosis_options = [
("E11.9", "Type 2 Diabetes Mellitus without complications"),
("I10", "Essential Hypertension"),
("I50.9", "Heart Failure, unspecified"),
("J44.1", "COPD with acute exacerbation"),
("N18.3", "Chronic Kidney Disease, stage 3"),
("I25.10", "Atherosclerotic Heart Disease"),
("E78.5", "Hyperlipidemia, unspecified"),
("J18.9", "Pneumonia, unspecified organism"),
("K21.0", "GERD with esophagitis"),
("M54.5", "Low Back Pain"),
("I48.91", "Atrial Fibrillation, unspecified"),
("E11.65", "Type 2 Diabetes with Hyperglycemia"),
("J96.01", "Acute Respiratory Failure with Hypoxia"),
("N17.9", "Acute Kidney Failure, unspecified"),
("I21.9", "Acute Myocardial Infarction, unspecified"),
]
medication_options = [
("Metformin", "500mg twice daily"),
("Metformin", "1000mg twice daily"),
("Lisinopril", "10mg once daily"),
("Lisinopril", "20mg once daily"),
("Atorvastatin", "40mg once daily"),
("Atorvastatin", "80mg once daily"),
("Amlodipine", "5mg once daily"),
("Metoprolol Succinate", "25mg once daily"),
("Metoprolol Succinate", "50mg once daily"),
("Omeprazole", "20mg once daily"),
("Furosemide", "40mg once daily"),
("Furosemide", "80mg twice daily"),
("Albuterol Inhaler", "2 puffs every 4-6 hours PRN"),
("Aspirin", "81mg once daily"),
("Insulin Glargine", "20 units subcutaneous at bedtime"),
("Warfarin", "5mg once daily"),
("Carvedilol", "12.5mg twice daily"),
("Spironolactone", "25mg once daily"),
("Apixaban", "5mg twice daily"),
("Prednisone", "40mg daily taper"),
]
encounter_types = ["emergency", "inpatient", "outpatient"]
base_date = date(2024, 1, 1)
encounters_data = []
for patient_id in range(1, 301):
# Some patients are high utilizers
num_encounters = random.choices([1, 2, 3, 4, 5, 6], weights=[30, 30, 20, 10, 7, 3])[0]
last_discharge = None
for _ in range(num_encounters):
if last_discharge:
# 25% chance of readmission within 30 days
days_gap = random.choices(
[random.randint(1, 30), random.randint(31, 200)],
weights=[25, 75]
)[0]
admission = last_discharge + timedelta(days=days_gap)
else:
admission = base_date + timedelta(days=random.randint(0, 500))
stay = random.randint(1, 14)
discharge = admission + timedelta(days=stay)
enc_type = random.choices(encounter_types, weights=[25, 50, 25])[0]
provider_id = random.randint(1, 20)
encounters_data.append((
patient_id, provider_id,
admission.isoformat(), discharge.isoformat(), enc_type
))
last_discharge = discharge
cursor.executemany(
"INSERT INTO encounters (patient_id, provider_id, admission_date, discharge_date, encounter_type) VALUES (%s, %s, %s, %s, %s)",
encounters_data
)
conn.commit()
cursor.execute("SELECT encounter_id FROM encounters")
encounter_ids = [row[0] for row in cursor.fetchall()]
diagnoses_data = []
prescriptions_data = []
for enc_id in encounter_ids:
num_dx = random.randint(1, 4)
for dx in random.sample(diagnosis_options, num_dx):
diagnoses_data.append((enc_id, dx[0], dx[1]))
num_rx = random.randint(1, 5)
for rx in random.sample(medication_options, num_rx):
prescriptions_data.append((enc_id, rx[0], rx[1]))
cursor.executemany(
"INSERT INTO diagnoses (encounter_id, icd_code, description) VALUES (%s, %s, %s)",
diagnoses_data
)
cursor.executemany(
"INSERT INTO prescriptions (encounter_id, medication, dosage) VALUES (%s, %s, %s)",
prescriptions_data
)
conn.commit()
cursor.callproc("flag_readmissions")
conn.commit()
cursor.execute("SELECT COUNT(*) FROM patients")
print(f"Patients: {cursor.fetchone()[0]}")
cursor.execute("SELECT COUNT(*) FROM encounters")
print(f"Encounters: {cursor.fetchone()[0]}")
cursor.execute("SELECT COUNT(*) FROM encounters WHERE readmitted_30_day = 1")
print(f"Readmissions flagged: {cursor.fetchone()[0]}")
cursor.execute("SELECT COUNT(*) FROM diagnoses")
print(f"Diagnoses: {cursor.fetchone()[0]}")
cursor.execute("SELECT COUNT(*) FROM prescriptions")
print(f"Prescriptions: {cursor.fetchone()[0]}")
cursor.close()
conn.close()
print("\nData seeded successfully.")
if __name__ == "__main__":
seed()ⓧ I need to re-run on a fresh database
If you need to re-run the seed script (for example, after fixing a bug), you must clear the existing data first to avoid duplicate key errors. Run this in the MySQL CLI:
mysql -u root -e "USE clinical_analytics; SET FOREIGN_KEY_CHECKS=0; TRUNCATE TABLE prescriptions; TRUNCATE TABLE diagnoses; TRUNCATE TABLE encounters; TRUNCATE TABLE patients; TRUNCATE TABLE providers; SET FOREIGN_KEY_CHECKS=1;"Then run python seed_data.py again.
Your database is now loaded with realistic clinical data. Next up, you will write production SQL queries that power clinical dashboards and extract features for machine learning.
Write Production SQL Queries
Your database is full of clinical data. Patients, encounters, diagnoses, prescriptions, and readmission flags are all sitting in interconnected tables waiting to be queried.
In this step, you'll write the SQL functions that power clinical dashboards and extract machine learning features. These are the exact queries a Clinical Analytics team runs daily to monitor hospital performance.
In this step, get ready to:
- Write dashboard queries for patient census, diagnosis frequency, and readmission rates.
- Write analytical queries that break down readmissions by diagnosis, age group, and patient utilization.
- Build an ML feature extraction query that JOINs all 5 tables into a training-ready dataset.
Write clinical dashboard queries
Clinical dashboards answer the questions hospital leadership asks every morning: how many patients do we have, what are the most common diagnoses, and what is our readmission rate?
- Open queries.py in your editor and add the following code:
from database import get_connection
def patient_census():
"""Get total patient count."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM patients")
count = cursor.fetchone()[0]
cursor.close()
conn.close()
return count
def top_diagnoses(limit=10):
"""Get most common diagnoses."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("""
SELECT icd_code, description, COUNT(*) AS frequency
FROM diagnoses
GROUP BY icd_code, description
ORDER BY frequency DESC
LIMIT %s
""", (limit,))
results = cursor.fetchall()
cursor.close()
conn.close()
return results
def readmission_rate():
"""Calculate overall readmission rate."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("""
SELECT
COUNT(*) AS total_encounters,
SUM(readmitted_30_day) AS readmissions,
ROUND(100.0 * SUM(readmitted_30_day) / COUNT(*), 2) AS rate_pct
FROM encounters
""")
result = cursor.fetchone()
cursor.close()
conn.close()
return resultWhat does this code do?
Each function follows the same production pattern: open a connection, execute a query, fetch results, close everything, and return the data.
- patient_census() runs a simple COUNT(*) to get the total number of patients in the system.
- top_diagnoses() uses GROUP BY + ORDER BY + LIMIT to rank diagnoses by frequency. The %s placeholder is a parameterized query that prevents SQL injection.
- readmission_rate() calculates the percentage of encounters that resulted in a 30-day readmission using SUM() and ROUND().
- Save queries.py.
Why parameterized queries?
The %s placeholder in top_diagnoses() lets MySQL safely handle user-provided values. In production systems, this is mandatory to prevent SQL injection attacks.
Write analytical queries for clinical insights
Dashboard totals are useful, but clinical leaders need to drill deeper. Which diagnoses drive the most readmissions? Are older patients at higher risk? Who are the high-utilization patients consuming the most resources?
- Add the following analytical functions below readmission_rate() in queries.py:
def readmission_by_diagnosis():
"""Readmission rate broken down by primary diagnosis."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("""
SELECT
d.icd_code,
d.description,
COUNT(DISTINCT e.encounter_id) AS encounters,
SUM(e.readmitted_30_day) AS readmissions,
ROUND(100.0 * SUM(e.readmitted_30_day) / COUNT(DISTINCT e.encounter_id), 2) AS rate_pct
FROM encounters e
JOIN diagnoses d ON e.encounter_id = d.encounter_id
GROUP BY d.icd_code, d.description
HAVING encounters >= 10
ORDER BY rate_pct DESC
""")
results = cursor.fetchall()
cursor.close()
conn.close()
return results
def readmission_by_age_group():
"""Readmission rate by age bucket."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("""
SELECT
CASE
WHEN TIMESTAMPDIFF(YEAR, p.date_of_birth, e.admission_date) < 40 THEN '18-39'
WHEN TIMESTAMPDIFF(YEAR, p.date_of_birth, e.admission_date) < 60 THEN '40-59'
WHEN TIMESTAMPDIFF(YEAR, p.date_of_birth, e.admission_date) < 75 THEN '60-74'
ELSE '75+'
END AS age_group,
COUNT(*) AS total,
SUM(e.readmitted_30_day) AS readmissions,
ROUND(100.0 * SUM(e.readmitted_30_day) / COUNT(*), 2) AS rate_pct
FROM encounters e
JOIN patients p ON e.patient_id = p.patient_id
GROUP BY age_group
ORDER BY age_group
""")
results = cursor.fetchall()
cursor.close()
conn.close()
return results
def frequent_flyers():
"""Patients with 4+ encounters in the dataset."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("""
SELECT
p.patient_id,
CONCAT(p.first_name, ' ', p.last_name) AS patient_name,
TIMESTAMPDIFF(YEAR, p.date_of_birth, CURDATE()) AS age,
COUNT(e.encounter_id) AS visit_count,
SUM(e.readmitted_30_day) AS readmission_count
FROM patients p
JOIN encounters e ON p.patient_id = e.patient_id
GROUP BY p.patient_id, patient_name, age
HAVING visit_count >= 4
ORDER BY visit_count DESC
LIMIT 15
""")
results = cursor.fetchall()
cursor.close()
conn.close()
return resultsWhat does this code do?
- readmission_by_diagnosis() uses a JOIN between encounters and diagnoses to find which ICD-10 codes have the highest readmission rates. The HAVING encounters >= 10 clause filters out rare diagnoses that would skew the percentages.
- readmission_by_age_group() uses CASE WHEN to bucket patients into age ranges and TIMESTAMPDIFF(YEAR, ...) to calculate age at admission.
- frequent_flyers() identifies high-utilization patients using HAVING visit_count >= 4. In real hospitals, care managers use this list to proactively coordinate care and reduce preventable visits.
- Save queries.py.
Seeing a syntax error in the SQL?
- Make sure the triple-quoted SQL strings (""") are properly closed. Each query starts and ends with triple quotes.
- Check that HAVING comes after GROUP BY in your queries. Swapping the order causes a MySQL error.
Why is my analytical query returning unexpected results?
Write the ML feature extraction query
This is the most important query in the project. It bridges your SQL database and the machine learning model by producing one row per encounter with all the features the model needs to predict readmission risk.
- Add the extract_features() and run_all() functions below frequent_flyers() in queries.py:
def extract_features():
"""Extract ML features using JOINs across all tables."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("""
SELECT
TIMESTAMPDIFF(YEAR, p.date_of_birth, e.admission_date) AS age,
CASE WHEN p.gender = 'M' THEN 1 ELSE 0 END AS gender_male,
(SELECT COUNT(*) FROM encounters e_prev
WHERE e_prev.patient_id = p.patient_id
AND e_prev.encounter_id < e.encounter_id) AS prior_admissions,
(SELECT COUNT(*) FROM diagnoses d
WHERE d.encounter_id = e.encounter_id) AS diagnosis_count,
(SELECT COUNT(*) FROM prescriptions rx
WHERE rx.encounter_id = e.encounter_id) AS prescription_count,
CASE
WHEN e.encounter_type = 'emergency' THEN 2
WHEN e.encounter_type = 'inpatient' THEN 1
ELSE 0
END AS encounter_type_encoded,
DATEDIFF(e.discharge_date, e.admission_date) AS length_of_stay,
e.readmitted_30_day
FROM encounters e
JOIN patients p ON e.patient_id = p.patient_id
WHERE e.discharge_date IS NOT NULL
""")
results = cursor.fetchall()
cursor.close()
conn.close()
return results
def run_all():
"""Run and display all analytics queries."""
print(f"=== Patient Census: {patient_census()} ===\n")
print("=== Top 10 Diagnoses ===")
for row in top_diagnoses():
print(f" {row[0]} - {row[1]}: {row[2]}")
print(f"\n=== Readmission Rate ===")
total, readmissions, rate = readmission_rate()
print(f" {readmissions}/{total} encounters ({rate}%)")
print(f"\n=== Readmission by Age Group ===")
for row in readmission_by_age_group():
print(f" {row[0]}: {row[3]}% ({row[2]}/{row[1]})")
print(f"\n=== Frequent Flyers (4+ visits) ===")
for row in frequent_flyers():
print(f" {row[1]} (age {row[2]}): {row[3]} visits, {row[4]} readmissions")
if __name__ == "__main__":
run_all()What does extract_features() do?
This single query JOINs all 5 tables to produce a machine-learning-ready dataset. Each row represents one encounter with 7 numeric features and a label:
- age is calculated using TIMESTAMPDIFF(YEAR, ...) from the patient's date of birth to their admission date.
- gender_male encodes gender as 1 or 0 using a CASE WHEN expression.
- prior_admissions uses a correlated subquery to count how many earlier encounters the same patient had.
- diagnosis_count and prescription_count use correlated subqueries against the diagnoses and prescriptions tables.
- length_of_stay uses DATEDIFF() to calculate the number of days between admission and discharge.
- readmitted_30_day is the label your model will learn to predict.
- Save queries.py.
- Run the queries script to verify everything works by running this command in your terminal:
python queries.pyYou should see output showing your patient census, the top 10 diagnoses ranked by frequency, the overall readmission rate as a percentage, readmission rates broken down by age group, and a list of frequent flyer patients.
Seeing a connection error?
- Make sure MySQL is still running. Run brew services start mysql if it stopped.
- Check that your .env file has the correct DB_PASSWORD value matching the password you set for analytics_svc.
- Verify you sourced your environment variables (source .env or export them manually) before running the script.
Why is my queries.py script failing to connect to MySQL?
✔️ Awesome, I've got everything!
Great. Double check you've saved queries.py and that the output looks reasonable.
ⓧ I'd like to double check the full code
Here is the complete queries.py file:
from database import get_connection
def patient_census():
"""Get total patient count."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM patients")
count = cursor.fetchone()[0]
cursor.close()
conn.close()
return count
def top_diagnoses(limit=10):
"""Get most common diagnoses."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("""
SELECT icd_code, description, COUNT(*) AS frequency
FROM diagnoses
GROUP BY icd_code, description
ORDER BY frequency DESC
LIMIT %s
""", (limit,))
results = cursor.fetchall()
cursor.close()
conn.close()
return results
def readmission_rate():
"""Calculate overall readmission rate."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("""
SELECT
COUNT(*) AS total_encounters,
SUM(readmitted_30_day) AS readmissions,
ROUND(100.0 * SUM(readmitted_30_day) / COUNT(*), 2) AS rate_pct
FROM encounters
""")
result = cursor.fetchone()
cursor.close()
conn.close()
return result
def readmission_by_diagnosis():
"""Readmission rate broken down by primary diagnosis."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("""
SELECT
d.icd_code,
d.description,
COUNT(DISTINCT e.encounter_id) AS encounters,
SUM(e.readmitted_30_day) AS readmissions,
ROUND(100.0 * SUM(e.readmitted_30_day) / COUNT(DISTINCT e.encounter_id), 2) AS rate_pct
FROM encounters e
JOIN diagnoses d ON e.encounter_id = d.encounter_id
GROUP BY d.icd_code, d.description
HAVING encounters >= 10
ORDER BY rate_pct DESC
""")
results = cursor.fetchall()
cursor.close()
conn.close()
return results
def readmission_by_age_group():
"""Readmission rate by age bucket."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("""
SELECT
CASE
WHEN TIMESTAMPDIFF(YEAR, p.date_of_birth, e.admission_date) < 40 THEN '18-39'
WHEN TIMESTAMPDIFF(YEAR, p.date_of_birth, e.admission_date) < 60 THEN '40-59'
WHEN TIMESTAMPDIFF(YEAR, p.date_of_birth, e.admission_date) < 75 THEN '60-74'
ELSE '75+'
END AS age_group,
COUNT(*) AS total,
SUM(e.readmitted_30_day) AS readmissions,
ROUND(100.0 * SUM(e.readmitted_30_day) / COUNT(*), 2) AS rate_pct
FROM encounters e
JOIN patients p ON e.patient_id = p.patient_id
GROUP BY age_group
ORDER BY age_group
""")
results = cursor.fetchall()
cursor.close()
conn.close()
return results
def frequent_flyers():
"""Patients with 4+ encounters in the dataset."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("""
SELECT
p.patient_id,
CONCAT(p.first_name, ' ', p.last_name) AS patient_name,
TIMESTAMPDIFF(YEAR, p.date_of_birth, CURDATE()) AS age,
COUNT(e.encounter_id) AS visit_count,
SUM(e.readmitted_30_day) AS readmission_count
FROM patients p
JOIN encounters e ON p.patient_id = e.patient_id
GROUP BY p.patient_id, patient_name, age
HAVING visit_count >= 4
ORDER BY visit_count DESC
LIMIT 15
""")
results = cursor.fetchall()
cursor.close()
conn.close()
return results
def extract_features():
"""Extract ML features using JOINs across all tables."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("""
SELECT
TIMESTAMPDIFF(YEAR, p.date_of_birth, e.admission_date) AS age,
CASE WHEN p.gender = 'M' THEN 1 ELSE 0 END AS gender_male,
(SELECT COUNT(*) FROM encounters e_prev
WHERE e_prev.patient_id = p.patient_id
AND e_prev.encounter_id < e.encounter_id) AS prior_admissions,
(SELECT COUNT(*) FROM diagnoses d
WHERE d.encounter_id = e.encounter_id) AS diagnosis_count,
(SELECT COUNT(*) FROM prescriptions rx
WHERE rx.encounter_id = e.encounter_id) AS prescription_count,
CASE
WHEN e.encounter_type = 'emergency' THEN 2
WHEN e.encounter_type = 'inpatient' THEN 1
ELSE 0
END AS encounter_type_encoded,
DATEDIFF(e.discharge_date, e.admission_date) AS length_of_stay,
e.readmitted_30_day
FROM encounters e
JOIN patients p ON e.patient_id = p.patient_id
WHERE e.discharge_date IS NOT NULL
""")
results = cursor.fetchall()
cursor.close()
conn.close()
return results
def run_all():
"""Run and display all analytics queries."""
print(f"=== Patient Census: {patient_census()} ===\n")
print("=== Top 10 Diagnoses ===")
for row in top_diagnoses():
print(f" {row[0]} - {row[1]}: {row[2]}")
print(f"\n=== Readmission Rate ===")
total, readmissions, rate = readmission_rate()
print(f" {readmissions}/{total} encounters ({rate}%)")
print(f"\n=== Readmission by Age Group ===")
for row in readmission_by_age_group():
print(f" {row[0]}: {row[3]}% ({row[2]}/{row[1]})")
print(f"\n=== Frequent Flyers (4+ visits) ===")
for row in frequent_flyers():
print(f" {row[1]} (age {row[2]}): {row[3]} visits, {row[4]} readmissions")
if __name__ == "__main__":
run_all()Your SQL query layer is complete. Next up, you'll use the extract_features() function to train a machine learning model that predicts which patients are at risk of readmission.
Train the Readmission Prediction Model
Your SQL queries are pulling clinical insights from production data. Now you'll use that same data to train a machine learning model that predicts which patients will be readmitted within 30 days.
The extract_features() query you built in the previous step is the bridge between SQL and ML. It produces one row per encounter with 7 numeric features and a readmission label. In this step, you'll feed that data into a Logistic Regression model, evaluate its performance, and save it for the API to use.
In this step, get ready to:
- Prepare training data by converting SQL results into feature matrices.
- Train and evaluate a Logistic Regression model with clinical performance metrics.
- Save the trained model and examine which clinical factors drive readmission risk.
Prepare training data from SQL
Machine learning models need numeric arrays, not database rows. You'll call your feature extraction query, split the results into features (X) and labels (y), then divide into training and test sets.
- Open train_model.py in your editor and add the following imports and data preparation code:
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import classification_report, accuracy_score
import joblib
from queries import extract_features
def train():
"""Train a readmission prediction model from SQL features."""
data = extract_features()
if not data:
print("No data found. Run seed_data.py first.")
return
# Separate features (all columns except last) from labels (last column)
X = [list(row[:-1]) for row in data]
y = [row[-1] for row in data]
print(f"Total samples: {len(X)}")
print(f"Readmission rate: {sum(y)/len(y)*100:.1f}%")
print(f"Features: age, gender_male, prior_admissions, diagnosis_count, prescription_count, encounter_type, length_of_stay")
print()
# Split 80% training, 20% test
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42
)
print(f"Training: {len(X_train)} | Test: {len(X_test)}")
print()What does this code do?
extract_features() runs the SQL query you built earlier and returns rows of tuples. Each tuple contains 7 feature values followed by the readmission label (0 or 1).
X = [list(row[:-1]) for row in data] takes everything except the last column (the features). y = [row[-1] for row in data] grabs just the last column (the readmission label the model learns to predict).
train_test_split holds back 20% of the data for testing. The model never sees this test data during training, so its performance on the test set reflects how well it generalises to new patients.
Train and evaluate a Logistic Regression model
Logistic Regression predicts the probability of a binary outcome (readmitted or not). It works well for tabular clinical data and produces interpretable coefficients that clinicians can trust.
- Add the following model training and evaluation code directly below the last print() line, still inside the train() function:
# Fit the model on training data
model = LogisticRegression(max_iter=1000, random_state=42)
model.fit(X_train, y_train)
# Evaluate on held-out test data
y_pred = model.predict(X_test)
print("=== Model Performance ===")
print(f"Accuracy: {accuracy_score(y_test, y_pred)*100:.1f}%\n")
print(classification_report(
y_test, y_pred,
target_names=["Not Readmitted", "Readmitted"]
))What do precision and recall mean in healthcare?
classification_report prints precision, recall, and F1-score for each class. In clinical terms, recall on the "Readmitted" class answers: of all patients who actually got readmitted, what percentage did we catch?
Precision on the "Readmitted" class answers: of the patients we flagged as at-risk, how many were truly readmitted? A balance of both means the model catches real readmissions without overwhelming care teams with false alarms.
Save the model and print feature importances
Clinicians need to understand why a model flags a patient. Logistic Regression coefficients show exactly which factors push risk up or down.
- Add the feature importance printing, model saving, and the __main__ block below your evaluation code, still inside train() (except the last two lines which go at module level):
# Show which clinical factors drive readmission risk
feature_names = [
"age", "gender_male", "prior_admissions",
"diagnosis_count", "prescription_count",
"encounter_type", "length_of_stay"
]
print("=== Feature Importances ===")
for name, coef in sorted(
zip(feature_names, model.coef_[0]),
key=lambda x: abs(x[1]), reverse=True
):
direction = "+" if coef > 0 else "-"
print(f" {direction} {name}: {coef:.4f}")
# Persist model for the API to load
joblib.dump(model, "readmission_model.joblib")
print("\nModel saved to readmission_model.joblib")
if __name__ == "__main__":
train()What does this code do?
model.coef_[0] contains one coefficient per feature. A positive coefficient means higher values of that feature increase readmission risk. A negative coefficient means higher values decrease risk.
joblib.dump() serialises the trained model to a file. The API you build in the next step will load this file with joblib.load() and use it to score new patients without retraining.
- Save train_model.py.
✔️ Awesome, I've got everything!
Great. Make sure you've saved the file before running it.
ⓧ I'd like to double check the full code
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import classification_report, accuracy_score
import joblib
from queries import extract_features
def train():
"""Train a readmission prediction model from SQL features."""
data = extract_features()
if not data:
print("No data found. Run seed_data.py first.")
return
X = [list(row[:-1]) for row in data]
y = [row[-1] for row in data]
print(f"Total samples: {len(X)}")
print(f"Readmission rate: {sum(y)/len(y)*100:.1f}%")
print(f"Features: age, gender_male, prior_admissions, diagnosis_count, prescription_count, encounter_type, length_of_stay")
print()
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42
)
print(f"Training: {len(X_train)} | Test: {len(X_test)}")
print()
model = LogisticRegression(max_iter=1000, random_state=42)
model.fit(X_train, y_train)
y_pred = model.predict(X_test)
print("=== Model Performance ===")
print(f"Accuracy: {accuracy_score(y_test, y_pred)*100:.1f}%\n")
print(classification_report(
y_test, y_pred,
target_names=["Not Readmitted", "Readmitted"]
))
feature_names = [
"age", "gender_male", "prior_admissions",
"diagnosis_count", "prescription_count",
"encounter_type", "length_of_stay"
]
print("=== Feature Importances ===")
for name, coef in sorted(
zip(feature_names, model.coef_[0]),
key=lambda x: abs(x[1]), reverse=True
):
direction = "+" if coef > 0 else "-"
print(f" {direction} {name}: {coef:.4f}")
joblib.dump(model, "readmission_model.joblib")
print("\nModel saved to readmission_model.joblib")
if __name__ == "__main__":
train()- Train the model by running this command in your terminal:
python train_model.pyYou should see output showing the dataset stats, accuracy percentage, a precision/recall/F1 table, feature importances sorted by magnitude, and a confirmation that the model was saved.
How to read the feature importances
Features prefixed with + increase readmission risk when their value is higher. For example, + prior_admissions means patients with more prior visits are at higher risk.
Features prefixed with - decrease risk when their value is higher. The magnitude (absolute value) tells you how strongly each factor influences the prediction.
- Confirm the model file was saved by running:
ls readmission_model.joblibYou should see readmission_model.joblib listed in the output. This file contains the trained model that the API will load in the next step.
Seeing "No data found"?
This means extract_features() returned an empty result. Make sure you ran python seed_data.py in a previous step and that MySQL is still running.
Check that your .env credentials match what you set up for the analytics_svc user. You can verify with mysql -u analytics_svc -p clinical_analytics -e "SELECT COUNT(*) FROM encounters;".
Help me debug why my training script returns no data.
Why Logistic Regression instead of a neural network?
For tabular clinical data with a few hundred to a few thousand rows, Logistic Regression performs comparably to complex models. It trains in milliseconds and produces coefficients that clinicians can inspect directly.
In healthcare, interpretability is a regulatory and ethical requirement. A model that says "prior admissions increases risk by 0.3" is one a physician can act on. A black-box neural network that says "risk = 0.72" requires additional explanation tooling before clinical teams will trust it.
You now have a trained readmission prediction model sitting in your project folder. Next up, you'll wrap it in a FastAPI microservice so clinical staff can call it with patient data and get a risk score back in real time.
Build the FastAPI Prediction Microservice
Your model is trained, evaluated, and saved. Now it's time to give clinical staff a way to actually use it.
In this step, you'll wrap your prediction model in a FastAPI microservice. Instead of running Python scripts manually, clinicians will submit patient features through an API and get back a risk score with recommended actions.
In this step, get ready to:
- Create a FastAPI app with Pydantic request and response models.
- Build a prediction endpoint that stratifies patients into risk tiers.
- Test the microservice end-to-end using the interactive API docs.
Create the API with Pydantic models
FastAPI uses Pydantic models to validate incoming request data and shape outgoing responses. You'll define what patient features the API accepts and what it returns.
- In main.py, add the imports, app instantiation, and model loading by pasting this code:
from fastapi import FastAPI
from pydantic import BaseModel
import joblib
import numpy as np
# Initialize the FastAPI app with service metadata
app = FastAPI(
title="Clinical Analytics - Readmission Risk API",
description="Internal microservice for 30-day readmission risk prediction",
version="1.0.0"
)
# Load the trained model once when the server starts
model = joblib.load("readmission_model.joblib")What does this code do?
FastAPI(title=..., version=...) creates the app and populates the auto-generated documentation page with your service name and version.
joblib.load("readmission_model.joblib") loads your trained Logistic Regression model into memory at startup. Every prediction request reuses this loaded model instead of reading from disk each time.
- Below the model loading line, define the PatientFeatures Pydantic model by pasting this code:
class PatientFeatures(BaseModel):
age: int
gender_male: int
prior_admissions: int
diagnosis_count: int
prescription_count: int
encounter_type: int # 0=outpatient, 1=inpatient, 2=emergency
length_of_stay: int
# Provides an example payload in the auto-generated docs
model_config = {
"json_schema_extra": {
"examples": [
{
"age": 72,
"gender_male": 1,
"prior_admissions": 3,
"diagnosis_count": 3,
"prescription_count": 5,
"encounter_type": 2,
"length_of_stay": 7
}
]
}
}What does this code do?
PatientFeatures defines the exact shape of data the API accepts. Each field maps to one of the seven features your model was trained on.
The model_config with json_schema_extra pre-fills the interactive docs with a realistic example. This saves clinicians from having to guess valid input values when testing.
- Below PatientFeatures, define the response model and health endpoint by pasting this code:
class PredictionResponse(BaseModel):
risk_probability: float # Model's predicted probability
risk_level: str # HIGH, MEDIUM, or LOW
recommended_action: str # Clinical guidance for the care team
@app.get("/health")
def health_check():
return {"status": "healthy", "service": "readmission-risk-api", "version": "1.0.0"}What does this code do?
PredictionResponse defines what the API sends back: the raw probability, a human-readable risk level, and a recommended clinical action.
The /health endpoint is a standard pattern for production services. Monitoring tools ping this endpoint to confirm the service is alive and responsive.
- Save main.py.
Seeing red squiggles in your editor?
Make sure your virtual environment is activated. Your editor needs to see the installed packages. If you closed your terminal since the last step, run source venv/bin/activate again from your project folder.
help me fix import errors in my FastAPI project
Build the prediction endpoint
The prediction endpoint is where the model meets the real world. It accepts patient features, runs the model, and translates raw probability into actionable clinical guidance.
- Below the health_check function, add the prediction endpoint by pasting this code:
@app.post("/predict", response_model=PredictionResponse)
def predict_readmission(features: PatientFeatures):
"""Predict 30-day readmission risk for a patient encounter."""
# Build feature array in the same order the model was trained on
X = np.array([[
features.age,
features.gender_male,
features.prior_admissions,
features.diagnosis_count,
features.prescription_count,
features.encounter_type,
features.length_of_stay
]])
# Get probability of readmission (positive class)
probability = float(model.predict_proba(X)[0, 1])
# Stratify into risk tiers with clinical recommendations
if probability >= 0.5:
risk_level = "HIGH"
action = "Activate care coordination. Extended discharge planning, schedule 48-hour follow-up call, refer to transitional care."
elif probability >= 0.3:
risk_level = "MEDIUM"
action = "Schedule follow-up within 7 days. Ensure medication reconciliation complete. Provide patient education materials."
else:
risk_level = "LOW"
action = "Standard discharge protocol. Schedule routine 30-day follow-up."
return PredictionResponse(
risk_probability=round(probability, 4),
risk_level=risk_level,
recommended_action=action
)What does this code do?
model.predict_proba(X)[0, 1] returns the probability that this patient will be readmitted within 30 days. The [0, 1] indexes into the positive class (readmitted = 1).
The risk stratification thresholds (0.5 for HIGH, 0.3 for MEDIUM) translate a raw number into something clinicians can act on. Each tier comes with a specific recommendation so care teams know exactly what to do next.
Why these specific thresholds?
In clinical decision support, 0.5 and 0.3 are common starting points. HIGH-risk patients (50%+ chance of readmission) need intensive intervention. MEDIUM-risk patients (30-50%) need closer follow-up. These thresholds would be tuned with real clinical data in production.
- Save main.py.
Seeing indentation errors?
Make sure the entire predict_readmission function is indented consistently. Python requires either all spaces or all tabs. If you mixed them, select the entire function and re-indent with spaces.
help me fix indentation errors in my Python FastAPI file
✔️ Awesome, I've got everything!
Great. Double check you've saved main.py before moving on.
ⓧ I'd like to double check the full code
from fastapi import FastAPI
from pydantic import BaseModel
import joblib
import numpy as np
app = FastAPI(
title="Clinical Analytics - Readmission Risk API",
description="Internal microservice for 30-day readmission risk prediction",
version="1.0.0"
)
model = joblib.load("readmission_model.joblib")
class PatientFeatures(BaseModel):
age: int
gender_male: int
prior_admissions: int
diagnosis_count: int
prescription_count: int
encounter_type: int
length_of_stay: int
model_config = {
"json_schema_extra": {
"examples": [
{
"age": 72,
"gender_male": 1,
"prior_admissions": 3,
"diagnosis_count": 3,
"prescription_count": 5,
"encounter_type": 2,
"length_of_stay": 7
}
]
}
}
class PredictionResponse(BaseModel):
risk_probability: float
risk_level: str
recommended_action: str
@app.get("/health")
def health_check():
return {"status": "healthy", "service": "readmission-risk-api", "version": "1.0.0"}
@app.post("/predict", response_model=PredictionResponse)
def predict_readmission(features: PatientFeatures):
"""Predict 30-day readmission risk for a patient encounter."""
X = np.array([[
features.age,
features.gender_male,
features.prior_admissions,
features.diagnosis_count,
features.prescription_count,
features.encounter_type,
features.length_of_stay
]])
probability = float(model.predict_proba(X)[0, 1])
if probability >= 0.5:
risk_level = "HIGH"
action = "Activate care coordination. Extended discharge planning, schedule 48-hour follow-up call, refer to transitional care."
elif probability >= 0.3:
risk_level = "MEDIUM"
action = "Schedule follow-up within 7 days. Ensure medication reconciliation complete. Provide patient education materials."
else:
risk_level = "LOW"
action = "Standard discharge protocol. Schedule routine 30-day follow-up."
return PredictionResponse(
risk_probability=round(probability, 4),
risk_level=risk_level,
recommended_action=action
)Test the microservice end-to-end
Time to see the whole system in action. You'll start the development server and submit test predictions through FastAPI's built-in interactive documentation.
- Start the FastAPI development server by running this command:
fastapi dev main.pyYou should see output showing the server is running, including a line like:
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)What does this command do?
fastapi dev starts a local development server with auto-reload enabled. Any changes you save to main.py will automatically restart the server. The dev mode also enables more detailed error pages.
Seeing a ModuleNotFoundError?
If you see an error about joblib or numpy not being found, your virtual environment is not active. Press Ctrl+C to stop the server, run source venv/bin/activate, then try again.
If you see an error about readmission_model.joblib not being found, make sure you're running the command from your project folder where the model file was saved.
help me fix my FastAPI server startup error
- Open your browser and navigate to http://127.0.0.1:8000/docs.
You should see a Swagger UI page titled "Clinical Analytics - Readmission Risk API" with two endpoints listed: GET /health and POST /predict.
- Click on POST /predict to expand it.
- Click Try it out.
- The example payload is already pre-filled. Click Execute.
You should see a JSON response with a risk_probability, risk_level, and recommended_action. The example patient (age 72, 3 prior admissions, emergency encounter) should return a MEDIUM or HIGH risk level.
- Now test a low-risk profile. Replace the request body with this JSON:
{
"age": 35,
"gender_male": 0,
"prior_admissions": 0,
"diagnosis_count": 1,
"prescription_count": 1,
"encounter_type": 0,
"length_of_stay": 2
}- Click Execute again.
This patient (young, no prior admissions, outpatient, short stay) should return a LOW risk level. The contrast between the two results confirms your model is working as expected.
Why test with both profiles?
A model that returns the same risk level for everyone is useless. By testing a clearly high-risk and clearly low-risk patient, you confirm the model differentiates between them. This is a basic sanity check before any clinical deployment.
Secret mission
Build a Risk Stratification Dashboard Endpoint
Your /predict endpoint handles one patient at a time. But a clinical care coordinator needs to see ALL high-risk patients at a glance. Build a dashboard endpoint that queries MySQL directly, runs batch predictions, and returns a ranked list of patients most likely to be readmitted.
Clean Up Your Resources
Clean Up Your Resources
Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely on your local machine, so there are no ongoing cloud costs. MySQL is running as a background service that uses minimal memory when idle.
Resources you used:
- MySQL clinical_analytics database.
- MySQL analytics_svc service account.
- MySQL background service (started via Homebrew).
- Project folder on your Desktop (virtual environment, Python files, trained model).
Keep everything running
No action needed. Choose this if you plan to extend this project or continue experimenting with the database and API.
- MySQL continues running as a background service. It uses minimal resources when idle.
- Your clinical_analytics database and all seeded data remain available.
- Your virtual environment and trained model stay in your project folder.
Pause - I'll come back to this later
Stop the MySQL background service to free up memory. Your database data persists on disk and will be available when you start the service again.
- Stop MySQL by running:
brew services stop mysql- When you want to resume, start MySQL again by running:
brew services start mysqlYour data persists in the MySQL data directory. Nothing is lost by stopping the service.
Delete - I don't want to use this again
Remove all project resources and database objects this project created. This does not uninstall MySQL itself.
- Stop the MySQL service by running:
brew services stop mysql- Connect to MySQL as root by running:
mysql -u root- Drop the database and service account by running these SQL commands:
DROP DATABASE clinical_analytics;
DROP USER 'analytics_svc'@'localhost';
FLUSH PRIVILEGES;
EXIT;- Remove your project folder from the Desktop by running:
cd ~/Desktop
rm -rf clinical-analyticsWhat if your folder has a different name?
Replace clinical-analytics in the command above with the actual name of the project folder you created in Step 1.
Nice Work!
Nice Work!
Nice work! You just shipped your first sprint as a data engineer on the Clinical Analytics team.
You've learned how to:
- Administered a MySQL database end-to-end: created a dedicated service account with scoped permissions, designed a production schema with 5 tables, typed columns, foreign keys, indexes, and a stored procedure.
- Wrote complex SQL queries spanning JOINs, GROUP BY, subqueries, CASE WHEN, and DATEDIFF to extract ML features directly from production data, then trained a Logistic Regression model with clinically interpretable coefficients.
- Deployed a FastAPI microservice with environment-based configuration that serves 30-day readmission risk predictions with stratified clinical recommendations.
- Built a risk stratification dashboard endpoint that runs batch predictions, filters by risk threshold and department, and surfaces each patient's top contributing risk factor.
Ready to quiz yourself?