Recipe 7.10: Python Implementation Example
Heads up: This is a deliberately simplified, illustrative implementation of the optimal intervention timing pipeline from Recipe 7.10. It demonstrates the core concepts (longitudinal feature engineering, survival modeling, intervention window scoring) using synthetic data and a basic hazard model. It is not production-ready. Real intervention timing systems require validated causal models, clinical oversight, and months of calibration against actual outcomes. Think of this as the sketch on the whiteboard, not the blueprint for construction.
Setup
You'll need the following packages:
pip install boto3 numpy pandas scikit-learn
Your environment needs AWS credentials configured (via environment variables, instance profile, or ~/.aws/credentials). The IAM role or user needs:
sagemaker:InvokeEndpoint(for model inference in the AWS-integrated version)dynamodb:PutItem,dynamodb:GetItem,dynamodb:UpdateItem(for patient state storage)s3:GetObject,s3:PutObject(for timeline data and model artifacts)
For this example, we'll build the entire pipeline locally with synthetic data. The SageMaker integration points are noted where you'd swap in real endpoint calls.
Config and Constants
These thresholds and parameters control the intervention timing logic. In production, these would be tuned based on clinical validation studies and care team feedback. Start conservative (higher thresholds, longer cooldown periods) and loosen as you build confidence in the model's calibration.
import logging from decimal import Decimal # Structured logging. Never log PHI field values (patient names, MRNs, etc.). # Log patient_id only when necessary for debugging, and ensure logs are # stored in HIPAA-compliant destinations with appropriate access controls. logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) # --- Hazard thresholds --- # These define what "high risk" and "moderate risk" mean in terms of daily hazard. # A daily hazard of 0.05 means roughly a 5% chance of the target event on that day, # given the patient has survived to that day. HIGH_RISK_THRESHOLD = 0.05 MODERATE_RISK_THRESHOLD = 0.02 # --- Intervention scoring thresholds --- # These control when the system recommends action vs. continued monitoring. URGENT_THRESHOLD = 80.0 # Score above this = "intervene today or tomorrow" ACTION_THRESHOLD = 40.0 # Score above this = "intervene this week" # --- Intervention fatigue parameters --- # Minimum days between outreach attempts to avoid patient fatigue. # Calling a patient every 3 days trains them to ignore you. MIN_INTERVENTION_GAP_DAYS = 14 FATIGUE_DAMPENING = 0.3 # Multiply score by this if within cooldown period DECLINED_DAMPENING = 0.1 # Multiply score by this if patient recently declined # --- Model parameters --- FORECAST_HORIZON_DAYS = 30 # How many days ahead the model predicts LOOKBACK_DAYS = 730 # How far back to look for patient history (2 years) # --- DynamoDB table name --- PATIENT_STATE_TABLE = "intervention-timing-patient-state"
Step 1: Generate Synthetic Patient Timelines
The main recipe's Step 1 assembles patient timelines from EHR, claims, pharmacy, and lab systems. Here we generate realistic synthetic data that mimics the structure of those timelines. In production, this step would be a Glue ETL job pulling from your actual data sources.
import numpy as np import pandas as pd from datetime import datetime, timedelta, timezone def generate_synthetic_timeline(patient_id: str, risk_profile: str = "rising") -> dict: """ Generate a synthetic patient timeline for demonstration purposes. Creates a realistic sequence of clinical events over 2 years, with different risk profiles to show how the timing model responds to different trajectory shapes. Args: patient_id: Unique patient identifier risk_profile: One of "rising", "stable_high", "sudden_spike", "improving" Controls the shape of the risk trajectory in the synthetic data. Returns: A patient timeline dict matching the schema from the main recipe. """ np.random.seed(hash(patient_id) % 2**32) end_date = datetime(2026, 5, 31) start_date = end_date - timedelta(days=LOOKBACK_DAYS) events = [] # Generate encounters (office visits, ED visits, inpatient stays) # Frequency depends on risk profile: sicker patients have more encounters. if risk_profile == "rising": # Encounters accelerate over time: quarterly early, monthly later encounter_dates = [] current = start_date gap_days = 90 # start with quarterly visits while current < end_date: encounter_dates.append(current) # Gradually decrease gap (more frequent visits as risk rises) gap_days = max(14, gap_days - np.random.randint(5, 15)) current += timedelta(days=gap_days + np.random.randint(-5, 5)) elif risk_profile == "stable_high": # Consistent monthly encounters encounter_dates = [ start_date + timedelta(days=30 * i + np.random.randint(-3, 3)) for i in range(24) ] elif risk_profile == "sudden_spike": # Quarterly visits, then a burst of activity in the last 2 weeks encounter_dates = [ start_date + timedelta(days=90 * i + np.random.randint(-5, 5)) for i in range(8) ] # Add sudden burst for i in range(4): encounter_dates.append(end_date - timedelta(days=np.random.randint(1, 14))) else: # improving # Frequent early, tapering off encounter_dates = [] current = start_date gap_days = 14 while current < end_date: encounter_dates.append(current) gap_days = min(120, gap_days + np.random.randint(3, 10)) current += timedelta(days=gap_days) for enc_date in encounter_dates: # Most encounters are outpatient; sprinkle in ED visits for high-risk profiles enc_type = "outpatient" if risk_profile in ("rising", "stable_high") and np.random.random() < 0.1: enc_type = "ED" elif risk_profile == "sudden_spike" and enc_date > end_date - timedelta(days=14): enc_type = "ED" if np.random.random() < 0.3 else "outpatient" events.append({ "timestamp": enc_date.isoformat(), "event_type": "encounter", "event_subtype": enc_type, "attributes": { "diagnosis_codes": ["E11.9"], # Type 2 diabetes, simplified "provider_type": "endocrinology" if np.random.random() < 0.3 else "primary_care", } }) # Generate A1C lab results (quarterly, with trajectory based on risk profile) a1c_dates = [start_date + timedelta(days=90 * i) for i in range(9)] for i, lab_date in enumerate(a1c_dates): if lab_date > end_date: break if risk_profile == "rising": # A1C drifts upward: 7.2 -> 9.1 over 2 years a1c_value = 7.2 + (i * 0.25) + np.random.normal(0, 0.1) elif risk_profile == "stable_high": # A1C stays elevated around 8.5 a1c_value = 8.5 + np.random.normal(0, 0.3) elif risk_profile == "sudden_spike": # A1C stable then jumps a1c_value = 7.0 + (0.3 if i >= 7 else 0) + np.random.normal(0, 0.1) if i == 8: a1c_value = 9.5 # sudden spike else: # improving a1c_value = 9.0 - (i * 0.2) + np.random.normal(0, 0.1) events.append({ "timestamp": lab_date.isoformat(), "event_type": "lab", "event_subtype": "A1C", "attributes": { "value": round(a1c_value, 1), "reference_low": 4.0, "reference_high": 5.6, "abnormal_flag": "H" if a1c_value > 5.6 else "N", } }) # Generate medication fills (monthly for adherent, gaps for non-adherent) med_date = start_date + timedelta(days=np.random.randint(0, 30)) while med_date < end_date: events.append({ "timestamp": med_date.isoformat(), "event_type": "medication", "event_subtype": "fill", "attributes": { "drug_name": "metformin", "days_supply": 30, "refill_number": len([e for e in events if e["event_type"] == "medication"]), } }) # Introduce gaps for rising-risk patients (missed refills) if risk_profile == "rising" and med_date > end_date - timedelta(days=120): # Increasingly likely to miss refills as risk rises gap = 30 + np.random.randint(5, 20) # overdue refills elif risk_profile == "sudden_spike" and med_date > end_date - timedelta(days=30): gap = 45 # missed a refill entirely else: gap = 30 + np.random.randint(-2, 3) # normal adherence med_date += timedelta(days=gap) # Sort all events chronologically events.sort(key=lambda e: e["timestamp"]) return { "patient_id": patient_id, "timeline": events, "event_count": len(events), "span_days": (end_date - start_date).days, }
Step 2: Engineer Temporal Features
The main recipe's Step 2 transforms raw timelines into features that capture temporal dynamics: velocity, acceleration, gaps, and recency. This is where static data becomes timing-aware.
def engineer_temporal_features(timeline: dict, observation_date: datetime) -> dict: """ Compute temporal features from a patient timeline at a specific observation point. These features capture not just current state but rate of change, acceleration, and gaps. The timing model needs these dynamics to predict when risk will peak. Args: timeline: Patient timeline dict from Step 1 (or from your data pipeline) observation_date: The date at which to compute features (today for inference, historical dates for training) Returns: Dictionary of feature name -> value. Null values indicate missing data (the model should handle nulls gracefully via imputation or masking). """ events = timeline["timeline"] features = {} # Parse timestamps and filter to events before observation_date # (we can't use future events when making predictions) parsed_events = [] for event in events: event_date = datetime.fromisoformat(event["timestamp"]) if event_date <= observation_date: parsed_events.append({**event, "_parsed_date": event_date}) # --- Recency features --- # How long since key event types? Longer gaps often signal disengagement. encounters = [e for e in parsed_events if e["event_type"] == "encounter"] ed_visits = [e for e in parsed_events if e["event_type"] == "encounter" and e["event_subtype"] == "ED"] med_fills = [e for e in parsed_events if e["event_type"] == "medication"] labs = [e for e in parsed_events if e["event_type"] == "lab"] features["days_since_last_encounter"] = ( (observation_date - encounters[-1]["_parsed_date"]).days if encounters else None ) features["days_since_last_ed_visit"] = ( (observation_date - ed_visits[-1]["_parsed_date"]).days if ed_visits else None ) features["days_since_last_med_fill"] = ( (observation_date - med_fills[-1]["_parsed_date"]).days if med_fills else None ) features["days_since_last_lab"] = ( (observation_date - labs[-1]["_parsed_date"]).days if labs else None ) # --- Velocity features: rate of change in key lab values --- # Extract A1C values and compute slope over recent measurements. a1c_results = [ (e["_parsed_date"], e["attributes"]["value"]) for e in parsed_events if e["event_type"] == "lab" and e["event_subtype"] == "A1C" ] if len(a1c_results) >= 2: features["a1c_current"] = a1c_results[-1][1] # Compute slope using last 3 measurements (or all if fewer) recent_a1c = a1c_results[-3:] if len(recent_a1c) >= 2: # Simple linear slope: (last - first) / days between them days_span = (recent_a1c[-1][0] - recent_a1c[0][0]).days if days_span > 0: value_change = recent_a1c[-1][1] - recent_a1c[0][1] features["a1c_slope_per_day"] = value_change / days_span else: features["a1c_slope_per_day"] = 0.0 else: features["a1c_slope_per_day"] = None else: features["a1c_current"] = None features["a1c_slope_per_day"] = None # --- Acceleration features: is utilization rate changing? --- thirty_days_ago = observation_date - timedelta(days=30) sixty_days_ago = observation_date - timedelta(days=60) encounters_last_30 = len([ e for e in encounters if e["_parsed_date"] > thirty_days_ago ]) encounters_prior_30 = len([ e for e in encounters if sixty_days_ago < e["_parsed_date"] <= thirty_days_ago ]) features["encounter_acceleration"] = encounters_last_30 - encounters_prior_30 # --- Gap features: missed expected events --- # Check if the patient is overdue for a medication refill. if med_fills: last_fill = med_fills[-1] days_supply = last_fill["attributes"].get("days_supply", 30) expected_refill = last_fill["_parsed_date"] + timedelta(days=days_supply) if observation_date > expected_refill: features["med_gap_days"] = (observation_date - expected_refill).days else: features["med_gap_days"] = 0 else: features["med_gap_days"] = None # --- Pattern features --- features["total_encounters_180d"] = len([ e for e in encounters if e["_parsed_date"] > observation_date - timedelta(days=180) ]) features["ed_visits_365d"] = len([ e for e in ed_visits if e["_parsed_date"] > observation_date - timedelta(days=365) ]) return features
Step 3: Predict Hazard Trajectory
The main recipe's Step 3 trains an LSTM-based survival model. Here we use a simplified hazard estimation approach that demonstrates the concept without requiring GPU training infrastructure. In production, you'd replace this with a SageMaker-hosted model endpoint.
def predict_hazard_trajectory(features: dict, horizon_days: int = FORECAST_HORIZON_DAYS) -> list: """ Predict the daily hazard trajectory for the next N days. This is a SIMPLIFIED hazard model for demonstration. It uses the temporal features to estimate a baseline hazard and a trajectory shape. A real implementation would call a SageMaker endpoint hosting a trained LSTM or transformer survival model. The key insight this demonstrates: hazard isn't flat. It has a shape over time that depends on the patient's current trajectory. A patient with rising A1C and a missed medication refill has a hazard that accelerates over the next few weeks, not a constant elevated risk. Args: features: Temporal feature dict from Step 2 horizon_days: How many days ahead to forecast Returns: List of daily hazard values (probability of event on each day, conditional on survival to that day). Length = horizon_days. """ # --- Compute baseline hazard from current state --- # In a real model, this comes from the neural network's output layer. # Here we use a simple logistic combination of features. baseline = 0.01 # population average daily hazard # A1C contribution: higher A1C = higher baseline hazard a1c = features.get("a1c_current") if a1c is not None: # Hazard increases exponentially above A1C of 7.0 if a1c > 7.0: baseline += 0.005 * (a1c - 7.0) ** 1.5 # Medication gap contribution: overdue refills increase hazard med_gap = features.get("med_gap_days") if med_gap is not None and med_gap > 0: # Hazard grows with days overdue (non-linear: worse the longer you wait) baseline += 0.002 * (med_gap / 7) ** 1.2 # ED visit recency: recent ED visit indicates instability days_since_ed = features.get("days_since_last_ed_visit") if days_since_ed is not None and days_since_ed < 30: baseline += 0.01 * (1 - days_since_ed / 30) # --- Compute trajectory shape from velocity features --- # The slope determines whether hazard is rising, flat, or falling. a1c_slope = features.get("a1c_slope_per_day", 0) or 0 encounter_accel = features.get("encounter_acceleration", 0) or 0 # Daily hazard growth rate: how much the hazard changes per day # Positive slope + accelerating encounters = rising hazard daily_growth = (a1c_slope * 50) + (encounter_accel * 0.003) # --- Generate the trajectory --- trajectory = [] current_hazard = baseline for day in range(horizon_days): # Apply growth (hazard changes over time based on trajectory) current_hazard += daily_growth # Clamp to reasonable bounds (hazard can't be negative or > 1) current_hazard = max(0.001, min(0.5, current_hazard)) trajectory.append(round(current_hazard, 6)) return trajectory # --- SageMaker integration point --- # In production, replace predict_hazard_trajectory with a call to your # trained model endpoint: # # import boto3 # from botocore.config import Config # # sagemaker_runtime = boto3.client( # "sagemaker-runtime", # config=Config(retries={"max_attempts": 3, "mode": "adaptive"}) # ) # # def predict_hazard_trajectory_sagemaker(features: dict) -> list: # """Call the SageMaker endpoint hosting the trained survival model.""" # import json # response = sagemaker_runtime.invoke_endpoint( # EndpointName="intervention-timing-model-v1", # ContentType="application/json", # Body=json.dumps({"features": features}), # ) # result = json.loads(response["Body"].read()) # return result["hazard_trajectory"]
Step 4: Score Intervention Windows
The main recipe's Step 4 applies decision logic to the hazard trajectory. This is where "risk prediction" becomes "timing recommendation." The scoring function identifies whether the patient is in an optimal intervention window based on the shape of their predicted trajectory.
def score_intervention_window( patient_id: str, hazard_trajectory: list, days_since_last_intervention: int = 30, last_intervention_outcome: str = "engaged", ) -> dict: """ Determine whether now is the right time to intervene for this patient. The core logic: interventions work best when risk is rising but hasn't peaked. You want to catch the patient on the upslope, not at the top (too late) or on the flat (too early, they won't engage). Args: patient_id: Patient identifier hazard_trajectory: List of daily hazard values from Step 3 days_since_last_intervention: Days since care team last reached out last_intervention_outcome: "engaged", "no_answer", or "declined" Returns: Intervention scoring result with recommended action and timing window. """ # Compute trajectory characteristics current_hazard = hazard_trajectory[0] peak_hazard = max(hazard_trajectory) peak_day = hazard_trajectory.index(peak_hazard) # Slope over first 7 days (is risk rising in the near term?) if len(hazard_trajectory) >= 7: hazard_slope = (hazard_trajectory[6] - hazard_trajectory[0]) / 7 else: hazard_slope = 0.0 # --- Decision logic --- intervention_score = 0.0 recommended_action = "monitor" action_window_days = None # Case 1: Rising risk with peak ahead. This is the prime intervention window. # The patient is deteriorating, but we still have time to change the trajectory. # Threshold is lower than the pseudocode's 0.01 because our simplified # hazard model produces smaller absolute slope values than a trained LSTM would. if hazard_slope > 0.001 and 2 < peak_day < 14: # Score scales with how fast risk is rising and how much worse it will get intervention_score = hazard_slope * 100 * (peak_hazard / max(current_hazard, 0.001)) # Case 2: Already at or past peak. Window may be closing. # Intervene urgently if current risk is high, but acknowledge we may be late. elif peak_day <= 2 and current_hazard > HIGH_RISK_THRESHOLD: intervention_score = current_hazard * 50 recommended_action = "urgent_outreach" action_window_days = 1 # Case 3: Flat high risk. Chronically elevated but not acutely changing. # Timing is less critical here; schedule outreach at convenience. elif current_hazard > MODERATE_RISK_THRESHOLD and abs(hazard_slope) < 0.0005: intervention_score = current_hazard * 20 recommended_action = "scheduled_outreach" action_window_days = 7 # --- Apply intervention fatigue dampening --- # Patients who were recently contacted (or who declined) get lower urgency. # This prevents the "boy who cried wolf" problem. if days_since_last_intervention < MIN_INTERVENTION_GAP_DAYS: intervention_score *= FATIGUE_DAMPENING if last_intervention_outcome == "declined": intervention_score *= DECLINED_DAMPENING # --- Determine final recommendation --- if intervention_score > URGENT_THRESHOLD: recommended_action = "immediate_outreach" action_window_days = 2 elif intervention_score > ACTION_THRESHOLD: recommended_action = "outreach_this_week" action_window_days = max(1, peak_day - 1) return { "patient_id": patient_id, "intervention_score": round(intervention_score, 1), "recommended_action": recommended_action, "action_window_days": action_window_days, "current_hazard": round(current_hazard, 4), "predicted_peak_day": peak_day, "peak_hazard": round(peak_hazard, 4), "trajectory_slope": round(hazard_slope, 6), "scored_at": datetime.now(timezone.utc).isoformat(), }
Step 5: Generate Explanations and Recommendations
The main recipe's Step 5 assembles scored patients into an actionable worklist with "why now" explanations. Care managers won't act on a number without understanding what changed.
def generate_explanation(scored_result: dict, features: dict) -> str: """ Build a human-readable explanation of why this patient needs outreach now. This is critical for adoption. A care manager looking at a worklist of 8 patients needs to understand, in 10 seconds, why each patient is there today and not yesterday or next week. The explanation should reference specific clinical changes, not model internals. Args: scored_result: Output from score_intervention_window features: Temporal features from Step 2 (used to identify clinical drivers) Returns: A plain-English explanation string suitable for display in a care management platform. """ parts = [] # Trajectory shape explanation if scored_result["trajectory_slope"] > 0.002: parts.append("Risk trajectory is rising sharply") elif scored_result["trajectory_slope"] > 0.0005: parts.append("Risk trajectory is trending upward") # Peak timing peak_day = scored_result["predicted_peak_day"] if 0 < peak_day < 14: parts.append(f"Predicted risk peak within {peak_day} days") # Current risk level if scored_result["current_hazard"] > HIGH_RISK_THRESHOLD: parts.append("Current risk level is elevated") # Clinical drivers from features a1c = features.get("a1c_current") a1c_slope = features.get("a1c_slope_per_day") if a1c is not None and a1c > 8.0: if a1c_slope and a1c_slope > 0: parts.append(f"A1C at {a1c} and rising") else: parts.append(f"A1C elevated at {a1c}") med_gap = features.get("med_gap_days") if med_gap and med_gap > 7: parts.append(f"Medication refill overdue by {med_gap} days") days_since_enc = features.get("days_since_last_encounter") if days_since_enc and days_since_enc > 90: parts.append(f"No encounter in {days_since_enc} days") return ". ".join(parts) + "." if parts else "Risk score elevated above action threshold." def generate_worklist(scored_patients: list, features_by_patient: dict, capacity: int = 8) -> list: """ Assemble the final care team worklist from scored patients. Applies capacity constraints and generates explanations. The output is what the care manager sees in their workflow tool each morning. Args: scored_patients: List of scoring results from Step 4 features_by_patient: Dict mapping patient_id -> features from Step 2 capacity: Maximum number of recommendations to generate (care team slots) Returns: List of recommendation dicts, sorted by urgency, capped at capacity. """ # Filter to actionable recommendations only actionable = [s for s in scored_patients if s["recommended_action"] != "monitor"] # Sort by intervention score (most urgent first) actionable.sort(key=lambda x: x["intervention_score"], reverse=True) # Cap at team capacity recommendations = actionable[:capacity] # Add explanations for rec in recommendations: patient_features = features_by_patient.get(rec["patient_id"], {}) rec["explanation"] = generate_explanation(rec, patient_features) rec["expires_at"] = ( datetime.now(timezone.utc) + timedelta(days=rec["action_window_days"] or 7) ).isoformat() rec["status"] = "pending" return recommendations
Step 6: Store Results in DynamoDB
In production, recommendations are written to DynamoDB for the care management platform to consume. This step shows the storage pattern with proper Decimal handling and TTL for automatic expiration of stale recommendations.
import boto3 from botocore.config import Config import json BOTO3_RETRY_CONFIG = Config(retries={"max_attempts": 3, "mode": "adaptive"}) dynamodb = boto3.resource("dynamodb", config=BOTO3_RETRY_CONFIG) def store_recommendation(recommendation: dict) -> dict: """ Write an intervention recommendation to DynamoDB. The care management platform reads from this table to populate the care manager's daily worklist. TTL ensures stale recommendations (where the action window has passed) are automatically cleaned up. Args: recommendation: A single recommendation dict from generate_worklist Returns: The stored record (with DynamoDB-compatible types). """ table = dynamodb.Table(PATIENT_STATE_TABLE) # Convert floats to Decimal (DynamoDB requirement). # boto3 will raise TypeError on raw floats in put_item. record = { "patient_id": recommendation["patient_id"], "scored_at": recommendation["scored_at"], "intervention_score": Decimal(str(recommendation["intervention_score"])), "recommended_action": recommendation["recommended_action"], "action_window_days": recommendation.get("action_window_days"), "current_hazard": Decimal(str(recommendation["current_hazard"])), "predicted_peak_day": recommendation["predicted_peak_day"], "peak_hazard": Decimal(str(recommendation["peak_hazard"])), "trajectory_slope": Decimal(str(recommendation["trajectory_slope"])), "explanation": recommendation["explanation"], "expires_at": recommendation["expires_at"], "status": recommendation["status"], } # TTL: DynamoDB will automatically delete expired recommendations. # This prevents stale items from cluttering the worklist. expiry_dt = datetime.fromisoformat(recommendation["expires_at"].replace("Z", "+00:00")) record["ttl"] = int(expiry_dt.timestamp()) table.put_item(Item=record) return record
Putting It All Together
Here's the full pipeline assembled into a single callable function. This is what you'd invoke daily (for batch scoring) or on-demand (when a real-time event triggers rescoring for a specific patient).
def run_intervention_timing_pipeline(patient_ids: list, capacity: int = 8) -> list: """ Run the full optimal intervention timing pipeline for a panel of patients. This orchestrates all steps: timeline assembly, feature engineering, hazard prediction, intervention window scoring, and worklist generation. In production, Steps 1-2 would pull from your data lake (S3/Glue), Step 3 would call a SageMaker endpoint, and Steps 4-6 would run in Lambda. Args: patient_ids: List of patient IDs to score capacity: Care team outreach capacity (max recommendations to generate) Returns: Final worklist of intervention recommendations, sorted by urgency. """ print(f"=== Intervention Timing Pipeline ===") print(f"Scoring {len(patient_ids)} patients, capacity = {capacity} slots\n") scored_patients = [] features_by_patient = {} # Assign risk profiles for demonstration purposes. # In production, you'd just pull real timelines from your data lake. profiles = ["rising", "stable_high", "sudden_spike", "improving", "rising"] for i, patient_id in enumerate(patient_ids): profile = profiles[i % len(profiles)] print(f"Patient {patient_id} (profile: {profile})") # Step 1: Assemble timeline timeline = generate_synthetic_timeline(patient_id, risk_profile=profile) print(f" Timeline: {timeline['event_count']} events over {timeline['span_days']} days") # Step 2: Engineer features at current observation date observation_date = datetime(2026, 5, 31) features = engineer_temporal_features(timeline, observation_date) features_by_patient[patient_id] = features print(f" Features: A1C={features.get('a1c_current')}, " f"med_gap={features.get('med_gap_days')}d, " f"slope={features.get('a1c_slope_per_day')}") # Step 3: Predict hazard trajectory trajectory = predict_hazard_trajectory(features) print(f" Trajectory: current={trajectory[0]:.4f}, " f"peak={max(trajectory):.4f} on day {trajectory.index(max(trajectory))}") # Step 4: Score intervention window # Simulate varying intervention histories days_since = 30 if i % 3 != 0 else 7 # some patients were recently contacted outcome = "engaged" if i % 4 != 0 else "declined" score = score_intervention_window( patient_id, trajectory, days_since_last_intervention=days_since, last_intervention_outcome=outcome, ) scored_patients.append(score) print(f" Score: {score['intervention_score']} -> {score['recommended_action']}") print() # Step 5: Generate worklist with explanations print("=== Generating Worklist ===\n") worklist = generate_worklist(scored_patients, features_by_patient, capacity=capacity) for i, rec in enumerate(worklist, 1): print(f"{i}. [{rec['recommended_action']}] Patient {rec['patient_id']}") print(f" Score: {rec['intervention_score']} | " f"Window: {rec['action_window_days']} days") print(f" Why: {rec['explanation']}") print() print(f"Total actionable: {len(worklist)} of {len(scored_patients)} patients scored") return worklist # --- Run the demo --- if __name__ == "__main__": # Simulate a care manager's panel of 10 patients patient_ids = [f"PAT-{1000 + i}" for i in range(10)] worklist = run_intervention_timing_pipeline(patient_ids, capacity=8) # Print final worklist as JSON (what the care management platform would consume) print("\n=== JSON Output (for care management platform) ===\n") # Convert for JSON serialization (Decimal -> float for display) for rec in worklist: for key, val in rec.items(): if isinstance(val, Decimal): rec[key] = float(val) print(json.dumps(worklist[:3], indent=2)) # show first 3 for brevity
The Gap Between This and Production
This example demonstrates the concepts. It generates synthetic data, computes temporal features, predicts hazard trajectories, and scores intervention windows. But there's a significant distance between this sketch and a system you'd deploy to a care management team. Here's where that gap lives:
The survival model is a heuristic, not a trained model. The predict_hazard_trajectory function uses hand-coded rules to approximate what a trained LSTM survival model would produce. A real implementation trains on thousands of patient timelines with known outcomes, learning the complex non-linear relationships between temporal features and event timing. You'd train this in SageMaker with GPU instances, validate it against held-out data, and deploy it as a real-time endpoint.
No causal inference. This example predicts when events are likely to occur, but it doesn't estimate the causal effect of intervention at different time points. A production system would incorporate inverse probability weighting or G-computation to estimate "if we intervene on day X, how much does the event probability decrease?" That's a fundamentally harder problem requiring careful study design and statistical methodology.
Error handling and retries. Every external call (SageMaker endpoint, DynamoDB, S3) can fail. Production code wraps each call in try/except with specific handling for throttling (exponential backoff), service unavailability (circuit breaker), and malformed responses (graceful degradation). A failed scoring for one patient shouldn't crash the entire batch.
Input validation. This code trusts its inputs completely. Production validates that patient timelines have minimum data density (enough events to make meaningful predictions), that feature values are within expected ranges, and that the model endpoint is healthy before sending requests.
Model monitoring and drift detection. Survival models degrade over time as patient populations change, treatment patterns evolve, and intervention strategies shift. Production systems monitor the C-index on recent predictions, track calibration drift (are predicted probabilities matching observed event rates?), and alert when performance drops below acceptable thresholds.
The feedback loop problem. When your model successfully identifies patients at the right time and you intervene, those patients don't have events. Your next training cycle sees "model flagged, no event" and interprets it as a false positive. Production systems maintain a small randomized holdout (patients who are flagged but not intervened on) to preserve the training signal. This raises ethical questions that require IRB review and clinical leadership buy-in.
IAM least-privilege. The Lambda running this scoring logic needs exactly: sagemaker:InvokeEndpoint on the specific model endpoint, dynamodb:PutItem and dynamodb:GetItem on the specific table, and s3:GetObject on the specific timeline data prefix. Not sagemaker:*. Not dynamodb:*.
VPC and encryption. Patient timelines contain PHI. All data movement stays within a VPC using VPC endpoints for S3, DynamoDB, and SageMaker Runtime. KMS customer-managed keys encrypt data at rest in all stores. TLS encrypts everything in transit. VPC Flow Logs capture all network activity for audit.
Structured logging without PHI. Log the patient_id, scoring timestamp, intervention score, and recommended action. Never log the clinical features themselves (A1C values, medication names, diagnosis codes). Those are PHI. If you need to debug a specific patient's scoring, use a secure, audited query against the feature store, not log grep.
Testing. This example has no tests. Production needs: unit tests for feature engineering (given this timeline, do I get these features?), integration tests for the model endpoint (does it return valid trajectories?), calibration tests (are predicted hazards matching observed event rates in the validation cohort?), and end-to-end tests (does the full pipeline produce reasonable worklists for known patient scenarios?).
Clinical validation. Before deploying intervention timing recommendations to care teams, you need a prospective validation study: run the model in shadow mode (generate recommendations but don't show them), compare model-recommended timing against actual intervention timing, and measure whether model-timed interventions would have been more effective. This takes months and requires clinical research methodology, not just engineering.
Part of the Healthcare AI/ML Cookbook. See Recipe 7.10 for the full architectural walkthrough, pseudocode, and honest take on where this gets hard.