Implementation Guide: Analyze no-show patterns and recommend intervention outreach for high-risk patients

# hostname, WAN interface, VLANs, firewall policy, and DNS

# Initial access — connect laptop to FortiGate port1, navigate to:
# https://192.168.1.99 (default management IP)
# Default credentials: admin / (blank password)

# CLI: Set hostname
config system global
  set hostname "CLIENTNAME-FG40F"
  set timezone 12
end

# CLI: Configure WAN interface (adjust for ISP settings)
config system interface
  edit "wan1"
    set mode dhcp
    set allowaccess ping https ssh
  next
end

# CLI: Create VLAN for clinical workstations
config system interface
  edit "clinical-vlan"
    set vdom "root"
    set ip 10.10.10.1 255.255.255.0
    set allowaccess ping https ssh
    set interface "internal"
    set vlanid 10
  next
end

# CLI: Create VLAN for reception/IoT (tablets)
config system interface
  edit "reception-vlan"
    set vdom "root"
    set ip 10.10.20.1 255.255.255.0
    set allowaccess ping https
    set interface "internal"
    set vlanid 20
  next
end

# CLI: Enable IPS, AV, Web Filter on LAN-to-WAN policy
config firewall policy
  edit 1
    set name "clinical-to-internet"
    set srcintf "clinical-vlan"
    set dstintf "wan1"
    set srcaddr "all"
    set dstaddr "all"
    set action accept
    set schedule "always"
    set service "ALL"
    set utm-status enable
    set av-profile "default"
    set ips-sensor "default"
    set webfilter-profile "default"
    set ssl-ssh-profile "certificate-inspection"
    set nat enable
    set logtraffic all
  next
end

# CLI: Configure DNS to use FortiGuard
config system dns
  set primary 208.91.112.53
  set secondary 208.91.112.52
end

# Windows: Verify OS version
winver

# Windows: Update Chrome via command line
winget upgrade Google.Chrome

# Windows: Set DNS to FortiGate (if not using DHCP from FortiGate)
netsh interface ip set dns "Ethernet" static 10.10.10.1

# Install requests library
pip install requests

# Test API connectivity
import requests

API_KEY = 'YOUR_CLINIKO_API_KEY'
SHARD = 'api2'  # Check your Cliniko shard
BASE_URL = f'https://{SHARD}.cliniko.com/v1'

headers = {
    'User-Agent': 'YourMSPName (your@email.com)',
    'Accept': 'application/json',
    'Authorization': f'Basic {API_KEY}'  # Base64 encode API_KEY:''
}

# Fetch recent appointments
response = requests.get(
    f'{BASE_URL}/individual_appointments',
    headers=headers,
    params={'page': 1, 'per_page': 100, 'sort': 'appointment_start:desc'}
)
print(response.status_code)
print(response.json()['total_entries'])

# No-Show Rate by Provider
NoShowRate_Provider = 
DIVIDE(
    COUNTROWS(FILTER(Appointments, Appointments[Status] = "No-Show")),
    COUNTROWS(Appointments),
    0
)

# No-Show Rate by Day of Week
NoShowRate_DayOfWeek = 
DIVIDE(
    CALCULATE(COUNTROWS(Appointments), Appointments[Status] = "No-Show"),
    COUNTROWS(Appointments),
    0
)

# Average Lead Time for No-Shows vs Shows
AvgLeadTime_NoShow = 
CALCULATE(
    AVERAGE(Appointments[LeadTimeDays]),
    Appointments[Status] = "No-Show"
)

# Outreach Effectiveness (requires outreach data)
OutreachConversionRate = 
DIVIDE(
    COUNTROWS(FILTER(Outreach, Outreach[PatientResponded] = TRUE())),
    COUNTROWS(Outreach),
    0
)

# SMS Template: Standard Reminder (Tier 1)
# ----------------------------------------
# Hi [FIRST_NAME], this is a reminder from [PRACTICE_NAME] that you have 
# an appointment on [DATE] at [TIME]. Reply YES to confirm, CHANGE to 
# reschedule, or call us at [PHONE]. We look forward to seeing you!

# SMS Template: Medium Risk Follow-Up (Tier 2 - 72hr)
# ----------------------------------------
# Hi [FIRST_NAME], [PRACTICE_NAME] here. Your appointment is coming up on 
# [DATE] at [TIME]. We want to make sure this still works for you. Reply 
# YES to confirm or CHANGE if you need a different time. We're happy to help!

import pandas as pd
import numpy as np
from datetime import datetime

# Load exported appointment data
df = pd.read_csv('appointments_export.csv')

# Standardize column names
df.columns = [c.strip().lower().replace(' ', '_') for c in df.columns]

# Parse dates
df['appointment_date'] = pd.to_datetime(df['appointment_date'])
df['booking_date'] = pd.to_datetime(df['booking_date'])

# Calculate lead time (days between booking and appointment)
df['lead_time_days'] = (df['appointment_date'] - df['booking_date']).dt.days

# Standardize status to binary outcome
status_map = {
    'Attended': 0, 'Completed': 0, 'Showed': 0, 'Arrived': 0,
    'No-Show': 1, 'No Show': 1, 'Missed': 1, 'DNA': 1,
    'Cancelled': np.nan, 'Late Cancel': np.nan,  # Exclude cancellations from no-show model
    'Rescheduled': np.nan
}
df['no_show'] = df['status'].map(status_map)
df_model = df.dropna(subset=['no_show'])

# Feature engineering
df_model['day_of_week'] = df_model['appointment_date'].dt.dayofweek
df_model['hour_of_day'] = df_model['appointment_date'].dt.hour
df_model['is_monday'] = (df_model['day_of_week'] == 0).astype(int)
df_model['is_first_appointment'] = (df_model.groupby('patient_id').cumcount() == 0).astype(int)

# Calculate patient-level historical no-show rate
patient_history = df_model.groupby('patient_id')['no_show'].agg(['sum','count']).reset_index()
patient_history.columns = ['patient_id', 'total_no_shows', 'total_appointments']
patient_history['historical_no_show_rate'] = patient_history['total_no_shows'] / patient_history['total_appointments']
df_model = df_model.merge(patient_history[['patient_id','historical_no_show_rate']], on='patient_id', how='left')

# Summary statistics
print(f'Total appointments: {len(df_model)}')
print(f'No-show rate: {df_model["no_show"].mean():.1%}')
print(f'Unique patients: {df_model["patient_id"].nunique()}')
print(f'Date range: {df_model["appointment_date"].min()} to {df_model["appointment_date"].max()}')
print(f'\nNo-show rate by day of week:')
print(df_model.groupby('day_of_week')['no_show'].mean().round(3))

# Export cleaned data for import into prediction platform
df_model.to_csv('appointments_cleaned.csv', index=False)
print(f'\nCleaned data exported: {len(df_model)} records')

# Daily pilot tracking spreadsheet columns:
# Date | Patient_ID | Appointment_Time | Predicted_Risk | Risk_Tier | Actual_Outcome | Correct_Prediction?

# Calculate pilot accuracy metrics (Python)
import pandas as pd

pilot = pd.read_csv('pilot_tracking.csv')

# Accuracy: How often was the prediction correct?
# For binary: High Risk (>70%) predicted no-show, Low Risk (<40%) predicted show
pilot['predicted_noshow'] = pilot['predicted_risk'] > 0.5
pilot['actual_noshow'] = pilot['actual_outcome'] == 'No-Show'

from sklearn.metrics import classification_report, confusion_matrix
print(classification_report(pilot['actual_noshow'], pilot['predicted_noshow']))
print(confusion_matrix(pilot['actual_noshow'], pilot['predicted_noshow']))

# Target metrics for go-live:
# - Sensitivity (recall for no-shows) > 70%
# - Specificity (recall for shows) > 60%
# - Overall accuracy > 70%
# - False positive rate < 30% (we don't want to over-message patients who would have come)

config webfilter urlfilter
  edit 1
    set name "allowed-healthcare-saas"
    config entries
      edit 1
        set url "*.healow.com"
        set type wildcard
        set action allow
      next
      edit 2
        set url "*.powerbi.com"
        set type wildcard
        set action allow
      next
    end
  next
end

baseline_noshow_rate = 0.28  # 28% pre-implementation (example)
current_noshow_rate = 0.18  # 18% post-implementation (example)
avg_appointment_value = 150  # Average revenue per appointment
weekly_appointments = 120   # Total weekly appointments across all providers

weekly_recovered_appointments = weekly_appointments * (baseline_noshow_rate - current_noshow_rate)
weekly_recovered_revenue = weekly_recovered_appointments * avg_appointment_value
monthly_recovered_revenue = weekly_recovered_revenue * 4.33
annual_recovered_revenue = monthly_recovered_revenue * 12

monthly_system_cost = 249 * 5  # 5 provider seats at $249/mo = $1,245
monthly_net_benefit = monthly_recovered_revenue - monthly_system_cost

print(f'Baseline no-show rate: {baseline_noshow_rate:.0%}')
print(f'Current no-show rate: {current_noshow_rate:.0%}')
print(f'Weekly recovered appointments: {weekly_recovered_appointments:.1f}')
print(f'Monthly recovered revenue: ${monthly_recovered_revenue:,.0f}')
print(f'Monthly system cost: ${monthly_system_cost:,.0f}')
print(f'Monthly NET benefit: ${monthly_net_benefit:,.0f}')
print(f'Annual NET benefit: ${monthly_net_benefit * 12:,.0f}')
print(f'ROI: {(monthly_net_benefit / monthly_system_cost * 100):.0f}%')

# No-Show Risk Scoring Model for Allied & Mental Health Practices

# no_show_prediction_model.py
# No-Show Risk Scoring Model for Allied & Mental Health Practices
# Requirements: pip install pandas numpy scikit-learn xgboost joblib

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.preprocessing import LabelEncoder, StandardScaler
from sklearn.metrics import classification_report, roc_auc_score, precision_recall_curve
from xgboost import XGBClassifier
import joblib
import json
from datetime import datetime, timedelta

class NoShowPredictor:
    """
    Predicts probability of patient no-show for scheduled appointments.
    Designed for allied & mental health practices.
    """
    
    def __init__(self, model_path=None):
        self.model = None
        self.feature_columns = None
        self.label_encoders = {}
        self.scaler = StandardScaler()
        self.risk_thresholds = {'high': 0.70, 'medium': 0.40}
        if model_path:
            self.load_model(model_path)
    
    def prepare_features(self, df):
        """Engineer features from raw appointment data."""
        features = pd.DataFrame()
        
        # Temporal features
        features['day_of_week'] = pd.to_datetime(df['appointment_date']).dt.dayofweek
        features['hour_of_day'] = pd.to_datetime(df['appointment_date']).dt.hour
        features['is_monday'] = (features['day_of_week'] == 0).astype(int)
        features['is_friday'] = (features['day_of_week'] == 4).astype(int)
        features['is_morning'] = (features['hour_of_day'] < 12).astype(int)
        features['is_last_slot'] = (features['hour_of_day'] >= 16).astype(int)
        features['month'] = pd.to_datetime(df['appointment_date']).dt.month
        
        # Lead time (days between booking and appointment)
        if 'booking_date' in df.columns:
            features['lead_time_days'] = (
                pd.to_datetime(df['appointment_date']) - pd.to_datetime(df['booking_date'])
            ).dt.days
            features['lead_time_days'] = features['lead_time_days'].clip(0, 365)
        else:
            features['lead_time_days'] = 7  # Default if not available
        
        # Patient history features
        if 'historical_no_show_rate' in df.columns:
            features['historical_no_show_rate'] = df['historical_no_show_rate'].fillna(0)
        if 'total_appointments' in df.columns:
            features['total_appointments'] = df['total_appointments'].fillna(1)
            features['is_new_patient'] = (df['total_appointments'] <= 1).astype(int)
        if 'total_no_shows' in df.columns:
            features['total_no_shows'] = df['total_no_shows'].fillna(0)
        if 'days_since_last_visit' in df.columns:
            features['days_since_last_visit'] = df['days_since_last_visit'].fillna(30)
        
        # Appointment type (encode categorical)
        if 'appointment_type' in df.columns:
            if 'appointment_type' not in self.label_encoders:
                self.label_encoders['appointment_type'] = LabelEncoder()
                features['appointment_type_encoded'] = self.label_encoders['appointment_type'].fit_transform(
                    df['appointment_type'].fillna('Unknown')
                )
            else:
                # Handle unseen categories
                known = set(self.label_encoders['appointment_type'].classes_)
                features['appointment_type_encoded'] = df['appointment_type'].fillna('Unknown').apply(
                    lambda x: x if x in known else 'Unknown'
                )
                features['appointment_type_encoded'] = self.label_encoders['appointment_type'].transform(
                    features['appointment_type_encoded']
                )
        
        # Provider (encode categorical)
        if 'provider_id' in df.columns:
            if 'provider_id' not in self.label_encoders:
                self.label_encoders['provider_id'] = LabelEncoder()
                features['provider_encoded'] = self.label_encoders['provider_id'].fit_transform(
                    df['provider_id'].astype(str)
                )
            else:
                known = set(self.label_encoders['provider_id'].classes_)
                features['provider_encoded'] = df['provider_id'].astype(str).apply(
                    lambda x: x if x in known else list(known)[0]
                )
                features['provider_encoded'] = self.label_encoders['provider_id'].transform(
                    features['provider_encoded']
                )
        
        # Insurance type
        if 'insurance_type' in df.columns:
            if 'insurance_type' not in self.label_encoders:
                self.label_encoders['insurance_type'] = LabelEncoder()
                features['insurance_encoded'] = self.label_encoders['insurance_type'].fit_transform(
                    df['insurance_type'].fillna('Unknown')
                )
            else:
                known = set(self.label_encoders['insurance_type'].classes_)
                features['insurance_encoded'] = df['insurance_type'].fillna('Unknown').apply(
                    lambda x: x if x in known else 'Unknown'
                )
                features['insurance_encoded'] = self.label_encoders['insurance_type'].transform(
                    features['insurance_encoded']
                )
        
        self.feature_columns = features.columns.tolist()
        return features
    
    def train(self, df, target_column='no_show'):
        """Train the no-show prediction model."""
        print(f'Training on {len(df)} records...')
        print(f'No-show rate in training data: {df[target_column].mean():.1%}')
        
        X = self.prepare_features(df)
        y = df[target_column].values
        
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=42, stratify=y
        )
        
        # Scale features
        X_train_scaled = self.scaler.fit_transform(X_train)
        X_test_scaled = self.scaler.transform(X_test)
        
        # Calculate scale_pos_weight for class imbalance
        n_positive = sum(y_train == 1)
        n_negative = sum(y_train == 0)
        scale_pos_weight = n_negative / n_positive if n_positive > 0 else 1
        
        # Train XGBoost model
        self.model = XGBClassifier(
            n_estimators=200,
            max_depth=5,
            learning_rate=0.1,
            scale_pos_weight=scale_pos_weight,
            subsample=0.8,
            colsample_bytree=0.8,
            random_state=42,
            eval_metric='auc',
            use_label_encoder=False
        )
        
        self.model.fit(
            X_train_scaled, y_train,
            eval_set=[(X_test_scaled, y_test)],
            verbose=False
        )
        
        # Evaluate
        y_pred_proba = self.model.predict_proba(X_test_scaled)[:, 1]
        y_pred = (y_pred_proba > 0.5).astype(int)
        
        print('\n--- Model Performance ---')
        print(classification_report(y_test, y_pred, target_names=['Show', 'No-Show']))
        print(f'AUC-ROC: {roc_auc_score(y_test, y_pred_proba):.3f}')
        
        # Feature importance
        importance = dict(zip(self.feature_columns, self.model.feature_importances_))
        print('\n--- Feature Importance ---')
        for feat, imp in sorted(importance.items(), key=lambda x: x[1], reverse=True):
            print(f'  {feat}: {imp:.3f}')
        
        # Cross-validation
        cv_scores = cross_val_score(
            self.model, self.scaler.transform(X), y, cv=5, scoring='roc_auc'
        )
        print(f'\n5-Fold CV AUC: {cv_scores.mean():.3f} (+/- {cv_scores.std():.3f})')
        
        return {
            'auc_roc': roc_auc_score(y_test, y_pred_proba),
            'cv_auc_mean': cv_scores.mean(),
            'feature_importance': importance
        }
    
    def predict_risk(self, df):
        """Predict no-show risk for upcoming appointments."""
        X = self.prepare_features(df)
        X_scaled = self.scaler.transform(X)
        probabilities = self.model.predict_proba(X_scaled)[:, 1]
        
        results = df.copy()
        results['no_show_probability'] = probabilities
        results['risk_tier'] = pd.cut(
            probabilities,
            bins=[0, self.risk_thresholds['medium'], self.risk_thresholds['high'], 1.0],
            labels=['Low', 'Medium', 'High'],
            include_lowest=True
        )
        results['recommended_action'] = results['risk_tier'].map({
            'Low': 'Standard 24hr SMS reminder',
            'Medium': 'SMS at 72hr + SMS at 24hr + Email at 48hr',
            'High': 'Staff phone call at 72hr + SMS at 48hr + SMS at 24hr + SMS at 2hr'
        })
        
        return results[['patient_id', 'appointment_date', 'no_show_probability', 
                        'risk_tier', 'recommended_action']]
    
    def save_model(self, path='noshow_model'):
        """Save trained model and preprocessing artifacts."""
        joblib.dump({
            'model': self.model,
            'scaler': self.scaler,
            'label_encoders': self.label_encoders,
            'feature_columns': self.feature_columns,
            'risk_thresholds': self.risk_thresholds
        }, f'{path}.joblib')
        print(f'Model saved to {path}.joblib')
    
    def load_model(self, path='noshow_model'):
        """Load a trained model."""
        artifacts = joblib.load(f'{path}.joblib')
        self.model = artifacts['model']
        self.scaler = artifacts['scaler']
        self.label_encoders = artifacts['label_encoders']
        self.feature_columns = artifacts['feature_columns']
        self.risk_thresholds = artifacts['risk_thresholds']
        print(f'Model loaded from {path}.joblib')


# --- USAGE EXAMPLE ---
if __name__ == '__main__':
    # Load cleaned historical data
    df = pd.read_csv('appointments_cleaned.csv')
    
    # Train model
    predictor = NoShowPredictor()
    metrics = predictor.train(df, target_column='no_show')
    predictor.save_model('practice_noshow_model')
    
    # Score upcoming appointments
    upcoming = pd.read_csv('upcoming_appointments.csv')
    risk_scores = predictor.predict_risk(upcoming)
    
    print('\n--- Upcoming Appointment Risk Scores ---')
    print(risk_scores.to_string(index=False))
    
    # Export for outreach system
    high_risk = risk_scores[risk_scores['risk_tier'] == 'High']
    print(f'\nHigh-risk appointments requiring intervention: {len(high_risk)}')
    risk_scores.to_csv('risk_scores_today.csv', index=False)

# Azure Logic App / Power Automate workflow definition (can also be
# implemented as a cron-triggered Python script)

# daily_risk_scoring_pipeline.yaml
# Azure Logic App / Power Automate workflow definition
# Can also be implemented as a cron-triggered Python script

name: daily-noshow-risk-scoring
schedule: "0 6 * * 1-6"  # 6:00 AM Monday through Saturday
timezone: "America/New_York"  # Adjust to practice timezone

steps:
  - id: extract_appointments
    name: "Extract Upcoming Appointments"
    description: "Pull appointments for next 72 hours from EHR"
    # For Cliniko API:
    action: http_request
    config:
      method: GET
      url: "https://api2.cliniko.com/v1/individual_appointments"
      headers:
        Authorization: "Basic ${CLINIKO_API_KEY_BASE64}"
        User-Agent: "MSPName (contact@msp.com)"
        Accept: "application/json"
      params:
        q: "appointment_start:>=${TODAY}&appointment_start:<=${TODAY_PLUS_3}"
        per_page: 100
        sort: "appointment_start:asc"
    output: raw_appointments

  - id: enrich_patient_history
    name: "Enrich with Patient History"
    description: "Calculate historical no-show rate per patient"
    action: python_script
    config:
      script: |
        import pandas as pd
        
        # Load raw appointments from previous step
        appointments = pd.DataFrame(raw_appointments['individual_appointments'])
        
        # Query patient attendance history from local cache/database
        # (maintained by nightly sync from EHR)
        history = pd.read_csv('/data/patient_history_cache.csv')
        
        enriched = appointments.merge(
            history[['patient_id', 'historical_no_show_rate', 
                     'total_appointments', 'total_no_shows',
                     'days_since_last_visit']],
            on='patient_id',
            how='left'
        )
        enriched['historical_no_show_rate'] = enriched['historical_no_show_rate'].fillna(0.15)
    output: enriched_appointments

  - id: score_risk
    name: "Score No-Show Risk"
    description: "Run prediction model on all upcoming appointments"
    action: python_script
    config:
      script: |
        from no_show_prediction_model import NoShowPredictor
        
        predictor = NoShowPredictor(model_path='practice_noshow_model')
        risk_scores = predictor.predict_risk(enriched_appointments)
        
        # Log scoring run
        print(f"Scored {len(risk_scores)} appointments")
        print(f"High risk: {(risk_scores['risk_tier']=='High').sum()}")
        print(f"Medium risk: {(risk_scores['risk_tier']=='Medium').sum()}")
        print(f"Low risk: {(risk_scores['risk_tier']=='Low').sum()}")
    output: risk_scores

  - id: route_outreach
    name: "Route to Outreach Tiers"
    description: "Send high-risk patients to intervention workflows"
    action: python_script
    config:
      script: |
        import json
        import requests
        
        OUTREACH_API_URL = "${HEALOW_OUTREACH_API_URL}"
        OUTREACH_API_KEY = "${HEALOW_API_KEY}"
        
        for _, appt in risk_scores.iterrows():
            payload = {
                'patient_id': appt['patient_id'],
                'appointment_date': str(appt['appointment_date']),
                'risk_score': float(appt['no_show_probability']),
                'risk_tier': appt['risk_tier'],
                'recommended_action': appt['recommended_action']
            }
            
            if appt['risk_tier'] == 'High':
                # Queue for staff phone call + automated SMS sequence
                payload['outreach_type'] = 'tier3_high_touch'
                payload['channels'] = ['phone_staff', 'sms_48hr', 'sms_24hr', 'sms_2hr']
            elif appt['risk_tier'] == 'Medium':
                # Queue for automated multi-touch
                payload['outreach_type'] = 'tier2_multi_touch'
                payload['channels'] = ['sms_72hr', 'email_48hr', 'sms_24hr']
            else:
                # Standard reminder
                payload['outreach_type'] = 'tier1_standard'
                payload['channels'] = ['sms_24hr']
            
            # Send to outreach platform
            resp = requests.post(
                f"{OUTREACH_API_URL}/outreach/queue",
                headers={'Authorization': f'Bearer {OUTREACH_API_KEY}',
                         'Content-Type': 'application/json'},
                json=payload
            )
            print(f"Patient {appt['patient_id']}: {appt['risk_tier']} -> {resp.status_code}")
    output: outreach_results

  - id: update_dashboard
    name: "Update Reception Dashboard"
    description: "Push today's risk scores to the live dashboard"
    action: python_script
    config:
      script: |
        # Export today's scores for Power BI / dashboard consumption
        today_scores = risk_scores[
            risk_scores['appointment_date'].str[:10] == str(pd.Timestamp.today().date())
        ]
        today_scores.to_csv('/data/dashboard/today_risk_scores.csv', index=False)
        today_scores.to_json('/data/dashboard/today_risk_scores.json', orient='records')
        print(f"Dashboard updated with {len(today_scores)} appointments for today")

  - id: generate_staff_report
    name: "Generate Staff Action Report"
    description: "Create a simple report for front desk staff with high-risk patients needing calls"
    action: python_script
    config:
      script: |
        high_risk_today = risk_scores[
            (risk_scores['risk_tier'] == 'High') & 
            (risk_scores['appointment_date'].str[:10] <= str(
                (pd.Timestamp.today() + pd.Timedelta(days=3)).date()
            ))
        ].sort_values('no_show_probability', ascending=False)
        
        report = "=== HIGH-RISK NO-SHOW PATIENTS - ACTION REQUIRED ===\n"
        report += f"Generated: {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M')}\n\n"
        
        for _, row in high_risk_today.iterrows():
            report += f"Patient: {row['patient_id']}\n"
            report += f"  Appointment: {row['appointment_date']}\n"
            report += f"  Risk Score: {row['no_show_probability']:.0%}\n"
            report += f"  Action: Personal phone call required\n\n"
        
        report += f"Total high-risk patients needing calls: {len(high_risk_today)}\n"
        
        # Email report to front desk
        # (Use practice email system or HIPAA-compliant internal messaging)
        with open('/data/reports/daily_action_report.txt', 'w') as f:
            f.write(report)
        print(report)

  - id: log_audit
    name: "HIPAA Audit Log"
    description: "Log all PHI access for HIPAA compliance"
    action: python_script
    config:
      script: |
        import json
        from datetime import datetime
        
        audit_entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'action': 'daily_risk_scoring',
            'system': 'noshow_prediction_pipeline',
            'user': 'automated_system',
            'patients_accessed': len(risk_scores),
            'data_elements': ['patient_id', 'appointment_date', 'appointment_type',
                             'historical_attendance', 'demographics'],
            'purpose': 'treatment_operations_no_show_prediction',
            'hipaa_basis': 'TPO_operations',
            'output': 'risk_scores_and_outreach_queue'
        }
        
        with open('/data/audit/audit_log.jsonl', 'a') as f:
            f.write(json.dumps(audit_entry) + '\n')
        print(f"Audit entry logged: {audit_entry['timestamp']}")

# Cron job setup (Linux/macOS server or Azure VM)
# Add to crontab: crontab -e
0 6 * * 1-6 cd /opt/noshow-pipeline && python daily_pipeline.py >> /var/log/noshow-pipeline.log 2>&1

{
  "bindings": [
    {
      "name": "timer",
      "type": "timerTrigger",
      "direction": "in",
      "schedule": "0 0 10 * * 1-6"
    }
  ]
}

# Patient Outreach Orchestrator Agent

# outreach_orchestrator.py
# Patient Outreach Orchestrator Agent
# Requirements: pip install requests schedule twilio

import json
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from enum import Enum

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger('OutreachOrchestrator')


class Channel(Enum):
    SMS = 'sms'
    EMAIL = 'email'
    PHONE_STAFF = 'phone_staff'
    VOICE_AUTO = 'voice_auto'


class OutreachStatus(Enum):
    PENDING = 'pending'
    SENT = 'sent'
    DELIVERED = 'delivered'
    RESPONDED = 'responded'
    CONFIRMED = 'confirmed'
    RESCHEDULED = 'rescheduled'
    OPTED_OUT = 'opted_out'
    FAILED = 'failed'


@dataclass
class OutreachAction:
    channel: Channel
    hours_before_appointment: int
    template_id: str
    status: OutreachStatus = OutreachStatus.PENDING
    sent_at: Optional[datetime] = None
    response_at: Optional[datetime] = None


@dataclass
class PatientOutreachPlan:
    patient_id: str
    appointment_id: str
    appointment_datetime: datetime
    risk_score: float
    risk_tier: str
    actions: List[OutreachAction] = field(default_factory=list)
    patient_confirmed: bool = False
    patient_opted_out: bool = False


# --- PRIVACY-SAFE MESSAGE TEMPLATES ---
# CRITICAL: These templates must NEVER mention therapy, counseling,
# mental health, psychiatry, or any treatment type.

MESSAGE_TEMPLATES = {
    'sms_standard_24hr': {
        'channel': Channel.SMS,
        'body': 'Hi {first_name}, reminder from {practice_name}: you have an appointment on {date} at {time}. Reply YES to confirm or CHANGE to reschedule. Call {phone} with questions.',
        'max_length': 160
    },
    'sms_medium_72hr': {
        'channel': Channel.SMS,
        'body': 'Hi {first_name}, {practice_name} here. Your appointment is on {date} at {time}. We want to make sure this works for you. Reply YES to confirm or CHANGE to reschedule.',
        'max_length': 160
    },
    'sms_high_48hr': {
        'channel': Channel.SMS,
        'body': 'Hi {first_name}, {provider_name} at {practice_name} wants to check in about your appointment on {date}. We know schedules get busy - reply CHANGE if you need to adjust.',
        'max_length': 160
    },
    'sms_high_2hr': {
        'channel': Channel.SMS,
        'body': 'Hi {first_name}, just a reminder your appointment at {practice_name} is in 2 hours at {time}. See you soon! Reply CHANGE if needed.',
        'max_length': 160
    },
    'email_confirmation': {
        'channel': Channel.EMAIL,
        'subject': 'Appointment Reminder - {practice_name}',
        'body': 'Dear {first_name},\n\nThis is a friendly reminder about your upcoming appointment:\n\nDate: {date}\nTime: {time}\nLocation: {address}\n\nPlease click below to confirm or reschedule:\n{confirm_link}\n{reschedule_link}\n\nIf you have questions, call us at {phone}.\n\nBest,\nThe {practice_name} Team'
    },
    'phone_staff_script': {
        'channel': Channel.PHONE_STAFF,
        'script': 'Hi, may I speak with {first_name}? This is {staff_name} calling from {practice_name}. I\'m calling to confirm your appointment on {date} at {time}. We just want to make sure this time still works for you.',
        'if_hesitant': 'I completely understand. Would you like to reschedule? We have availability on {available_slots}. We\'d love to keep you on the schedule.',
        'do_not': 'Do NOT reference therapy, counseling, diagnosis, mental health, or any treatment details.'
    }
}


class OutreachOrchestrator:
    """
    Manages multi-channel outreach workflows for high-risk no-show patients.
    Ensures HIPAA compliance and mental health privacy protections.
    """
    
    def __init__(self, practice_config: Dict):
        self.practice_name = practice_config['practice_name']
        self.practice_phone = practice_config['phone']
        self.practice_address = practice_config['address']
        self.sms_provider = practice_config.get('sms_provider', 'healow')  # or 'twilio'
        self.opt_out_list = set()  # Patient IDs who opted out
        self.outreach_plans = {}  # appointment_id -> PatientOutreachPlan
        self.audit_log = []
    
    def create_outreach_plan(self, patient_id: str, appointment_id: str,
                             appointment_datetime: datetime, risk_score: float,
                             risk_tier: str) -> PatientOutreachPlan:
        """Create an outreach plan based on risk tier."""
        
        # Check opt-out list
        if patient_id in self.opt_out_list:
            logger.info(f'Patient {patient_id} has opted out - no outreach plan created')
            return None
        
        plan = PatientOutreachPlan(
            patient_id=patient_id,
            appointment_id=appointment_id,
            appointment_datetime=appointment_datetime,
            risk_score=risk_score,
            risk_tier=risk_tier
        )
        
        if risk_tier == 'High':
            plan.actions = [
                OutreachAction(Channel.PHONE_STAFF, 72, 'phone_staff_script'),
                OutreachAction(Channel.SMS, 48, 'sms_high_48hr'),
                OutreachAction(Channel.EMAIL, 48, 'email_confirmation'),
                OutreachAction(Channel.SMS, 24, 'sms_standard_24hr'),
                OutreachAction(Channel.SMS, 2, 'sms_high_2hr'),
            ]
        elif risk_tier == 'Medium':
            plan.actions = [
                OutreachAction(Channel.SMS, 72, 'sms_medium_72hr'),
                OutreachAction(Channel.EMAIL, 48, 'email_confirmation'),
                OutreachAction(Channel.SMS, 24, 'sms_standard_24hr'),
            ]
        else:  # Low
            plan.actions = [
                OutreachAction(Channel.SMS, 24, 'sms_standard_24hr'),
            ]
        
        self.outreach_plans[appointment_id] = plan
        self._audit('plan_created', patient_id, appointment_id, risk_tier)
        
        logger.info(f'Outreach plan created: patient={patient_id}, '
                    f'risk={risk_tier} ({risk_score:.0%}), '
                    f'actions={len(plan.actions)}')
        return plan
    
    def execute_due_actions(self):
        """Check all plans and execute any actions that are due."""
        now = datetime.now()
        
        for appt_id, plan in self.outreach_plans.items():
            if plan.patient_confirmed or plan.patient_opted_out:
                continue
            
            for action in plan.actions:
                if action.status != OutreachStatus.PENDING:
                    continue
                
                trigger_time = plan.appointment_datetime - timedelta(hours=action.hours_before_appointment)
                
                if now >= trigger_time:
                    self._execute_action(plan, action)
    
    def _execute_action(self, plan: PatientOutreachPlan, action: OutreachAction):
        """Execute a single outreach action."""
        template = MESSAGE_TEMPLATES[action.template_id]
        
        if action.channel == Channel.SMS:
            success = self._send_sms(plan, template)
        elif action.channel == Channel.EMAIL:
            success = self._send_email(plan, template)
        elif action.channel == Channel.PHONE_STAFF:
            success = self._queue_staff_call(plan, template)
        else:
            success = False
        
        action.status = OutreachStatus.SENT if success else OutreachStatus.FAILED
        action.sent_at = datetime.now()
        
        self._audit('action_executed', plan.patient_id, plan.appointment_id,
                    f'{action.channel.value}:{action.template_id}:{action.status.value}')
    
    def _send_sms(self, plan: PatientOutreachPlan, template: Dict) -> bool:
        """Send SMS via configured provider. Returns True on success."""
        # Implementation depends on provider (healow native or Twilio)
        logger.info(f'SMS sent to patient {plan.patient_id} '
                    f'for appointment {plan.appointment_id}')
        return True  # Replace with actual API call
    
    def _send_email(self, plan: PatientOutreachPlan, template: Dict) -> bool:
        """Send email via configured provider."""
        logger.info(f'Email sent to patient {plan.patient_id}')
        return True  # Replace with actual API call
    
    def _queue_staff_call(self, plan: PatientOutreachPlan, template: Dict) -> bool:
        """Add to staff call queue (displayed on reception dashboard)."""
        logger.info(f'Staff call queued for patient {plan.patient_id} '
                    f'(risk: {plan.risk_score:.0%})')
        return True  # Replace with dashboard/queue integration
    
    def handle_patient_response(self, appointment_id: str, response: str):
        """Process patient response to outreach."""
        plan = self.outreach_plans.get(appointment_id)
        if not plan:
            logger.warning(f'No plan found for appointment {appointment_id}')
            return
        
        response_upper = response.strip().upper()
        
        if response_upper in ['YES', 'CONFIRM', 'Y', 'C']:
            plan.patient_confirmed = True
            # Cancel remaining pending actions
            for action in plan.actions:
                if action.status == OutreachStatus.PENDING:
                    action.status = OutreachStatus.RESPONDED
            logger.info(f'Patient {plan.patient_id} CONFIRMED appointment')
            
        elif response_upper in ['CHANGE', 'RESCHEDULE', 'R', 'CANCEL']:
            # Flag for staff to follow up with reschedule
            for action in plan.actions:
                if action.status == OutreachStatus.PENDING:
                    action.status = OutreachStatus.RESPONDED
            logger.info(f'Patient {plan.patient_id} requested RESCHEDULE')
            
        elif response_upper in ['STOP', 'OPTOUT', 'OPT OUT', 'UNSUBSCRIBE']:
            plan.patient_opted_out = True
            self.opt_out_list.add(plan.patient_id)
            for action in plan.actions:
                if action.status == OutreachStatus.PENDING:
                    action.status = OutreachStatus.OPTED_OUT
            logger.info(f'Patient {plan.patient_id} OPTED OUT of communications')
        
        self._audit('patient_response', plan.patient_id, appointment_id, response_upper)
    
    def _audit(self, action: str, patient_id: str, appointment_id: str, detail: str):
        """HIPAA audit log entry."""
        entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'action': action,
            'patient_id': patient_id,
            'appointment_id': appointment_id,
            'detail': detail,
            'system': 'outreach_orchestrator'
        }
        self.audit_log.append(entry)
        # In production: write to persistent audit log (database or file)
    
    def get_daily_summary(self) -> Dict:
        """Generate daily outreach summary for practice manager."""
        today = datetime.now().date()
        today_plans = [
            p for p in self.outreach_plans.values()
            if p.appointment_datetime.date() == today
        ]
        
        return {
            'date': str(today),
            'total_appointments': len(today_plans),
            'high_risk': sum(1 for p in today_plans if p.risk_tier == 'High'),
            'medium_risk': sum(1 for p in today_plans if p.risk_tier == 'Medium'),
            'low_risk': sum(1 for p in today_plans if p.risk_tier == 'Low'),
            'confirmed': sum(1 for p in today_plans if p.patient_confirmed),
            'pending': sum(1 for p in today_plans if not p.patient_confirmed and not p.patient_opted_out),
            'opted_out': sum(1 for p in today_plans if p.patient_opted_out)
        }


# --- USAGE ---
if __name__ == '__main__':
    config = {
        'practice_name': 'Wellness Center',  # Generic name - no mental health reference
        'phone': '(555) 123-4567',
        'address': '123 Main St, Suite 200, Anytown, ST 12345',
        'sms_provider': 'healow'
    }
    
    orchestrator = OutreachOrchestrator(config)
    
    # Create plans from daily risk scoring output
    orchestrator.create_outreach_plan(
        patient_id='P001',
        appointment_id='A001',
        appointment_datetime=datetime.now() + timedelta(days=3),
        risk_score=0.82,
        risk_tier='High'
    )
    
    # Execute due actions
    orchestrator.execute_due_actions()
    
    # Handle patient response
    orchestrator.handle_patient_response('A001', 'YES')
    
    # Daily summary
    print(json.dumps(orchestrator.get_daily_summary(), indent=2))

# HIPAA-Compliant EHR Data Extraction for No-Show Prediction

# ehr_data_extractor.py
# HIPAA-Compliant EHR Data Extraction for No-Show Prediction
# Requirements: pip install requests pandas cryptography

import os
import json
import logging
import hashlib
from abc import ABC, abstractmethod
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import pandas as pd
import requests
from base64 import b64encode

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('EHRExtractor')

# --- HIPAA MINIMUM NECESSARY: Only these fields are extracted ---
ALLOWED_FIELDS = [
    'appointment_id', 'patient_id', 'appointment_date', 'appointment_time',
    'appointment_type', 'provider_id', 'provider_name', 'booking_date',
    'status',  # attended, no-show, cancelled, late-cancel
    'patient_age', 'patient_zip', 'insurance_type',
    'appointment_duration_minutes'
]

# --- FIELDS TO NEVER EXTRACT (psychotherapy notes, clinical content) ---
BLOCKED_FIELDS = [
    'notes', 'clinical_notes', 'psychotherapy_notes', 'therapy_notes',
    'diagnosis', 'diagnosis_code', 'icd_code', 'treatment_plan',
    'medication', 'prescription', 'substance_use', 'sud_history',
    'ssn', 'social_security', 'credit_card', 'bank_account',
    'full_address', 'email', 'phone'  # PII not needed for prediction
]


class BaseEHRExtractor(ABC):
    """Base class for EHR data extraction with HIPAA safeguards."""
    
    def __init__(self, practice_id: str, audit_log_path: str = '/data/audit/'):
        self.practice_id = practice_id
        self.audit_log_path = audit_log_path
        os.makedirs(audit_log_path, exist_ok=True)
    
    @abstractmethod
    def extract_appointments(self, start_date: datetime, end_date: datetime) -> pd.DataFrame:
        pass
    
    def filter_to_minimum_necessary(self, df: pd.DataFrame) -> pd.DataFrame:
        """HIPAA Minimum Necessary: Keep only allowed fields, block prohibited ones."""
        # Remove any blocked fields
        blocked_present = [col for col in df.columns if col.lower() in BLOCKED_FIELDS]
        if blocked_present:
            logger.warning(f'BLOCKED FIELDS DETECTED AND REMOVED: {blocked_present}')
            df = df.drop(columns=blocked_present)
        
        # Keep only allowed fields that exist
        allowed_present = [col for col in df.columns if col.lower() in ALLOWED_FIELDS]
        df = df[allowed_present]
        
        logger.info(f'Minimum Necessary filter applied. Retained fields: {list(df.columns)}')
        return df
    
    def pseudonymize_patient_ids(self, df: pd.DataFrame, salt: str) -> pd.DataFrame:
        """Hash patient IDs for additional de-identification (optional layer)."""
        if 'patient_id' in df.columns:
            df['patient_id'] = df['patient_id'].apply(
                lambda x: hashlib.sha256(f'{salt}:{x}'.encode()).hexdigest()[:16]
            )
        return df
    
    def audit_extraction(self, record_count: int, fields: List[str], source: str):
        """Write HIPAA audit log entry."""
        entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'action': 'data_extraction',
            'practice_id': self.practice_id,
            'source_system': source,
            'records_extracted': record_count,
            'fields_extracted': fields,
            'purpose': 'no_show_prediction_model_input',
            'hipaa_basis': 'TPO_health_care_operations',
            'user': os.environ.get('MSP_TECHNICIAN_ID', 'system')
        }
        
        log_file = os.path.join(self.audit_log_path, 'extraction_audit.jsonl')
        with open(log_file, 'a') as f:
            f.write(json.dumps(entry) + '\n')
        
        logger.info(f'Audit logged: {record_count} records from {source}')


class ClinikoExtractor(BaseEHRExtractor):
    """Extract appointment data from Cliniko via REST API."""
    
    def __init__(self, practice_id: str, api_key: str, shard: str = 'api2'):
        super().__init__(practice_id)
        self.base_url = f'https://{shard}.cliniko.com/v1'
        self.headers = {
            'User-Agent': 'MSPNoShowPredictor (contact@yourmsp.com)',
            'Accept': 'application/json',
            'Authorization': f'Basic {b64encode(f"{api_key}:".encode()).decode()}'
        }
    
    def extract_appointments(self, start_date: datetime, end_date: datetime) -> pd.DataFrame:
        """Pull appointments from Cliniko API with pagination."""
        all_appointments = []
        page = 1
        
        while True:
            params = {
                'page': page,
                'per_page': 100,
                'sort': 'appointment_start:asc',
                'q[]': [
                    f'appointment_start:>={start_date.strftime("%Y-%m-%dT00:00:00Z")}',
                    f'appointment_start:<={end_date.strftime("%Y-%m-%dT23:59:59Z")}'
                ]
            }
            
            response = requests.get(
                f'{self.base_url}/individual_appointments',
                headers=self.headers,
                params=params
            )
            response.raise_for_status()
            data = response.json()
            
            appointments = data.get('individual_appointments', [])
            if not appointments:
                break
            
            all_appointments.extend(appointments)
            
            if not data.get('links', {}).get('next'):
                break
            page += 1
        
        if not all_appointments:
            logger.warning('No appointments found in date range')
            return pd.DataFrame()
        
        # Normalize to flat DataFrame
        df = pd.json_normalize(all_appointments)
        
        # Map Cliniko fields to standard schema
        field_map = {
            'id': 'appointment_id',
            'patient.links.self': 'patient_id',
            'appointment_start': 'appointment_date',
            'appointment_type.name': 'appointment_type',
            'practitioner.links.self': 'provider_id',
            'created_at': 'booking_date',
            'did_not_arrive': 'did_not_arrive',
            'cancellation_reason': 'cancellation_reason'
        }
        
        df = df.rename(columns={k: v for k, v in field_map.items() if k in df.columns})
        
        # Derive status
        if 'did_not_arrive' in df.columns:
            df['status'] = df.apply(
                lambda row: 'No-Show' if row.get('did_not_arrive') == True
                else ('Cancelled' if pd.notna(row.get('cancellation_reason'))
                      else 'Attended'),
                axis=1
            )
        
        # Apply Minimum Necessary filter
        df = self.filter_to_minimum_necessary(df)
        
        self.audit_extraction(len(df), list(df.columns), 'Cliniko_API')
        return df


class CSVExtractor(BaseEHRExtractor):
    """Extract appointment data from CSV exports (SimplePractice, TherapyNotes, etc.)."""
    
    def __init__(self, practice_id: str, ehr_system: str = 'SimplePractice'):
        super().__init__(practice_id)
        self.ehr_system = ehr_system
    
    def extract_appointments(self, csv_path: str, 
                             field_mapping: Optional[Dict[str, str]] = None) -> pd.DataFrame:
        """Load and standardize CSV export."""
        if not os.path.exists(csv_path):
            raise FileNotFoundError(f'CSV file not found: {csv_path}')
        
        df = pd.read_csv(csv_path)
        logger.info(f'Loaded {len(df)} records from {csv_path}')
        logger.info(f'Source columns: {list(df.columns)}')
        
        # Apply custom field mapping if provided
        if field_mapping:
            df = df.rename(columns=field_mapping)
        
        # Apply Minimum Necessary filter
        df = self.filter_to_minimum_necessary(df)
        
        self.audit_extraction(len(df), list(df.columns), f'{self.ehr_system}_CSV')
        return df


# --- Field mapping templates for common EHR CSV exports ---
SIMPLEPRACTICE_FIELD_MAP = {
    'Client Name': 'patient_id',  # Will need hashing
    'Appointment Date': 'appointment_date',
    'Appointment Time': 'appointment_time',
    'Service Type': 'appointment_type',
    'Clinician': 'provider_id',
    'Status': 'status',
    'Date Created': 'booking_date'
}

THERAPYNOTES_FIELD_MAP = {
    'Patient': 'patient_id',
    'Date': 'appointment_date',
    'Time': 'appointment_time',
    'Type': 'appointment_type',
    'Therapist': 'provider_id',
    'Attendance': 'status',
    'Created': 'booking_date'
}


# --- USAGE ---
if __name__ == '__main__':
    # Example: Cliniko API extraction
    extractor = ClinikoExtractor(
        practice_id='practice_001',
        api_key=os.environ['CLINIKO_API_KEY'],
        shard='api2'
    )
    
    appointments = extractor.extract_appointments(
        start_date=datetime.now() - timedelta(days=365),
        end_date=datetime.now()
    )
    print(f'Extracted {len(appointments)} appointments from Cliniko')
    print(appointments.head())
    
    # Example: SimplePractice CSV extraction
    csv_extractor = CSVExtractor(
        practice_id='practice_002',
        ehr_system='SimplePractice'
    )
    
    appointments_csv = csv_extractor.extract_appointments(
        csv_path='/data/exports/simplepractice_appointments.csv',
        field_mapping=SIMPLEPRACTICE_FIELD_MAP
    )
    print(f'Extracted {len(appointments_csv)} appointments from CSV')