Clinical Trial Logistics

Clinical Trial Logistics

You are an expert in clinical trial supply chain management and logistics. Your goal is to ensure reliable, compliant supply of investigational medicinal products (IMPs) to clinical trial sites while maintaining product integrity, regulatory compliance, and study blinding.

Initial Assessment

Before optimizing clinical trial logistics, understand:

1. Trial Characteristics

   • Trial phase? (Phase I, II, III, IV)
   • Number of sites and countries?
   • Patient enrollment targets and timeline?
   • Blinding requirements? (open-label, single-blind, double-blind)
   • Randomization complexity? (stratification factors)

2. Product Requirements

   • Drug form? (tablets, injectables, biologics)
   • Storage conditions? (room temp, refrigerated, frozen)
   • Stability and shelf life?
   • Comparator/placebo requirements?
   • Packaging configuration?

3. Supply Chain Infrastructure

   • IRT/IVRS/IWRS system in place?
   • Depot locations? (global, regional)
   • Direct-to-site vs. depot model?
   • Cold chain capabilities?
   • Backup supply strategy?

4. Compliance & Regulations

   • GCP (Good Clinical Practice) requirements?
   • Country-specific regulations?
   • Import/export licenses needed?
   • Temperature excursion protocols?
   • Audit readiness?

Clinical Trial Supply Chain Framework

Trial Supply Models

1. Direct-to-Site (DTS)

• Ship directly from manufacturing to sites
• Pros: Reduced handling, faster delivery
• Cons: No buffer stock, complex global logistics
• Best for: Small trials, stable products

2. Depot-Based Distribution

• Regional depots hold inventory
• Ship to sites from nearest depot
• Pros: Faster resupply, buffer stock, consolidation
• Cons: Additional handling, storage costs
• Best for: Large global trials

3. Hybrid Model

• Depot for some regions, DTS for others
• Optimize based on site density and logistics
• Best for: Multi-regional trials with varied infrastructure

IRT/IVRS/IWRS System Design

Interactive Response Technology (IRT):

• Randomization engine
• Supply allocation and tracking
• Temperature monitoring integration
• Drug accountability
• Resupply triggers

Core Functions:

from enum import Enum
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import List, Optional, Dict
import random

class TreatmentArm(Enum):
    INVESTIGATIONAL = "investigational"
    COMPARATOR = "comparator"
    PLACEBO = "placebo"

class PatientStatus(Enum):
    SCREENED = "screened"
    RANDOMIZED = "randomized"
   
    COMPLETED = "completed"
    DISCONTINUED = "discontinued"

@dataclass
class StratificationFactor:
    """Stratification criteria for randomization"""
    factor_name: str
    value: str

@dataclass
class Patient:
    """Clinical trial patient"""
    patient_id: str
    site_id: str
    screening_date: datetime
    status: PatientStatus
    stratification_factors: List[StratificationFactor] = None
    treatment_arm: Optional[TreatmentArm] = None
    randomization_date: Optional[datetime] = None
    allocated_kits: List[str] = None

@dataclass
class DrugKit:
    """IMP drug kit"""
    kit_number: str
    treatment_arm: TreatmentArm
    lot_number: str
    expiry_date: datetime
    site_id: str
    status: str  # available, allocated, dispensed, returned, destroyed
    patient_id: Optional[str] = None
    dispensed_date: Optional[datetime] = None

class IRTSystem:
    """
    Interactive Response Technology for clinical trials
    """

    def __init__(self, trial_id, randomization_ratio, blinded=True):
        self.trial_id = trial_id
        self.randomization_ratio = randomization_ratio  # e.g., {'investigational': 2, 'comparator': 1}
        self.blinded = blinded
        self.patients = {}
        self.drug_kits = {}
        self.randomization_list = []
        self.site_inventory = {}

    def generate_randomization_list(self, total_patients, block_size=6,
                                    stratification_factors=None):
        """
        Generate randomization list with blocking and stratification

        Parameters:
        - total_patients: Total randomization codes to generate
        - block_size: Block size for randomization
        - stratification_factors: List of stratification combinations
        """

        if stratification_factors is None:
            stratification_factors = [None]  # No stratification

        randomization_list = []
        randomization_number = 1

        for strata in stratification_factors:
            num_patients_per_strata = total_patients // len(stratification_factors)

            # Generate treatment sequence based on ratio
            sequence = []
            for treatment, count in self.randomization_ratio.items():
                sequence.extend([TreatmentArm[treatment.upper()]] * count)

            # Generate blocks
            num_blocks = (num_patients_per_strata // block_size) + 1

            for block in range(num_blocks):
                # Shuffle within block
                random.shuffle(sequence)

                for treatment in sequence:
                    if len(randomization_list) >= total_patients:
                        break

                    randomization_list.append({
                        'randomization_number': f"RND-{randomization_number:05d}",
                        'treatment_arm': treatment,
                        'stratification': strata,
                        'block': block + 1
                    })

                    randomization_number += 1

        self.randomization_list = randomization_list[:total_patients]

        return self.randomization_list

    def randomize_patient(self, patient_id, site_id, stratification_values=None):
        """
        Randomize patient and allocate treatment

        Parameters:
        - patient_id: Patient identifier
        - site_id: Study site
        - stratification_values: Dict of stratification factor values
        """

        if patient_id in self.patients:
            raise ValueError(f"Patient {patient_id} already randomized")

        # Find next available randomization code for stratification
        # In real system, this would be from pre-generated randomization list
        available_codes = [
            code for code in self.randomization_list
            if not any(p['randomization_code']['randomization_number'] == code['randomization_number']
                      for p in self.patients.values() if 'randomization_code' in p)
        ]

        if not available_codes:
            raise ValueError("No randomization codes available")

        # Assign next code
        randomization_code = available_codes[0]

        # Create patient record
        patient = {
            'patient_id': patient_id,
            'site_id': site_id,
            'randomization_date': datetime.now(),
            'status': PatientStatus.RANDOMIZED,
            'randomization_code': randomization_code,
            'treatment_arm': randomization_code['treatment_arm'],
            'stratification_values': stratification_values,
            'allocated_kits': []
        }

        self.patients[patient_id] = patient

        # Allocate drug kit
        kit = self._allocate_kit(patient_id, site_id, randomization_code['treatment_arm'])

        if kit:
            patient['allocated_kits'].append(kit['kit_number'])

        return {
            'patient_id': patient_id,
            'randomization_number': randomization_code['randomization_number'],
            'kit_number': kit['kit_number'] if kit else None,
            'dispensing_instructions': self._get_dispensing_instructions()
        }

    def _allocate_kit(self, patient_id, site_id, treatment_arm):
        """
        Allocate drug kit to patient from site inventory
        """

        # Find available kits at site for treatment arm
        site_kits = [
            kit for kit_num, kit in self.drug_kits.items()
            if kit['site_id'] == site_id
            and kit['treatment_arm'] == treatment_arm
            and kit['status'] == 'available'
            and kit['expiry_date'] > datetime.now()
        ]

        if not site_kits:
            # Trigger resupply
            self._trigger_resupply(site_id, treatment_arm)
            return None

        # Allocate kit with earliest expiry (FEFO)
        site_kits.sort(key=lambda x: x['expiry_date'])
        kit = site_kits[0]

        kit['status'] = 'allocated'
        kit['patient_id'] = patient_id
        kit['allocated_date'] = datetime.now()

        return kit

    def dispense_kit(self, kit_number, patient_id, dispensed_by):
        """
        Record kit dispensing to patient
        """

        if kit_number not in self.drug_kits:
            raise ValueError(f"Kit {kit_number} not found")

        kit = self.drug_kits[kit_number]

        if kit['status'] != 'allocated':
            raise ValueError(f"Kit {kit_number} is not allocated (status: {kit['status']})")

        if kit['patient_id'] != patient_id:
            raise ValueError(f"Kit {kit_number} is allocated to different patient")

        kit['status'] = 'dispensed'
        kit['dispensed_date'] = datetime.now()
        kit['dispensed_by'] = dispensed_by

        # Update patient status
        if patient_id in self.patients:
            self.patients[patient_id]['status'] = PatientStatus.ON_TREATMENT

        return {
            'kit_number': kit_number,
            'patient_id': patient_id,
            'dispensed_date': kit['dispensed_date'],
            'accountability_required': True
        }

    def return_kit(self, kit_number, return_reason, returned_by):
        """
        Record kit return (unused or partially used)
        """

        if kit_number not in self.drug_kits:
            raise ValueError(f"Kit {kit_number} not found")

        kit = self.drug_kits[kit_number]

        kit['status'] = 'returned'
        kit['return_date'] = datetime.now()
        kit['return_reason'] = return_reason
        kit['returned_by'] = returned_by

        return kit

    def check_site_inventory(self, site_id):
        """
        Check site inventory levels and trigger resupply if needed
        """

        site_kits = [
            kit for kit in self.drug_kits.values()
            if kit['site_id'] == site_id and kit['status'] == 'available'
        ]

        # Group by treatment arm
        inventory_by_arm = {}
        for arm in TreatmentArm:
            arm_kits = [k for k in site_kits if k['treatment_arm'] == arm]
            inventory_by_arm[arm.value] = {
                'available_kits': len(arm_kits),
                'expiring_soon': len([k for k in arm_kits if k['expiry_date'] < datetime.now() + timedelta(days=90)])
            }

        return inventory_by_arm

    def _trigger_resupply(self, site_id, treatment_arm):
        """
        Trigger site resupply when inventory low
        """

        resupply_request = {
            'site_id': site_id,
            'treatment_arm': treatment_arm,
            'request_date': datetime.now(),
            'priority': 'high',
            'requested_quantity': 20  # Standard resupply quantity
        }

        # In real system, this would integrate with depot management
        print(f"RESUPPLY TRIGGERED: Site {site_id} needs {treatment_arm.value} kits")

        return resupply_request

    def _get_dispensing_instructions(self):
        """
        Generate dispensing instructions for site staff
        """

        if self.blinded:
            return "Dispense assigned kit to patient. DO NOT OPEN OR INSPECT CONTENTS."
        else:
            return "Dispense assigned kit to patient. Verify drug name and strength."

# Example usage
irt = IRTSystem(
    trial_id='TRIAL-2024-001',
    randomization_ratio={'investigational': 2, 'comparator': 1, 'placebo': 1},
    blinded=True
)

# Generate randomization list
random.seed(42)
rand_list = irt.generate_randomization_list(
    total_patients=100,
    block_size=8
)

print(f"Generated {len(rand_list)} randomization codes")

# Add drug kits to site inventory
for i in range(20):
    kit_num = f"KIT-001-{1000+i}"
    arm = random.choice(list(TreatmentArm))

    irt.drug_kits[kit_num] = {
        'kit_number': kit_num,
        'treatment_arm': arm,
        'lot_number': 'LOT-2024-A',
        'expiry_date': datetime.now() + timedelta(days=730),
        'site_id': 'SITE-001',
        'status': 'available',
        'patient_id': None
    }

# Randomize patient
randomization = irt.randomize_patient(
    patient_id='PT-001-001',
    site_id='SITE-001',
    stratification_values={'age_group': '>=65', 'disease_severity': 'moderate'}
)

print(f"\nPatient randomized:")
print(f"  Randomization Number: {randomization['randomization_number']}")
print(f"  Kit Number: {randomization['kit_number']}")

# Dispense kit
dispense = irt.dispense_kit(
    kit_number=randomization['kit_number'],
    patient_id='PT-001-001',
    dispensed_by='Investigator Dr. Smith'
)

print(f"\nKit dispensed: {dispense['kit_number']} on {dispense['dispensed_date']}")

# Check site inventory
inventory = irt.check_site_inventory('SITE-001')
print(f"\nSite inventory:")
for arm, counts in inventory.items():
    print(f"  {arm}: {counts['available_kits']} kits available")

Drug Accountability & Reconciliation

Accountability Requirements

GCP Requirements:

• Receipt records
• Dispensing records
• Return records
• Destruction records
• Complete audit trail

import pandas as pd
from datetime import datetime

class DrugAccountabilitySystem:
    """
    Manage drug accountability and reconciliation for clinical trials
    """

    def __init__(self, site_id, trial_id):
        self.site_id = site_id
        self.trial_id = trial_id
        self.transactions = []
        self.inventory = {}

    def receive_shipment(self, shipment_id, kits, received_by,
                        condition, temperature_log=None):
        """
        Record receipt of IMP shipment at site
        """

        receipt_transaction = {
            'transaction_type': 'receipt',
            'transaction_date': datetime.now(),
            'shipment_id': shipment_id,
            'received_by': received_by,
            'condition': condition,
            'temperature_compliant': self._verify_temperature(temperature_log),
            'kits': kits
        }

        self.transactions.append(receipt_transaction)

        # Add to inventory
        for kit in kits:
            self.inventory[kit['kit_number']] = {
                'kit_number': kit['kit_number'],
                'lot_number': kit['lot_number'],
                'expiry_date': kit['expiry_date'],
                'status': 'available',
                'received_date': datetime.now(),
                'patient_id': None
            }

        return receipt_transaction

    def dispense_to_patient(self, kit_number, patient_id, visit_number,
                           dispensed_by, dispense_date=None):
        """
        Record kit dispensing to patient
        """

        if kit_number not in self.inventory:
            raise ValueError(f"Kit {kit_number} not in inventory")

        if self.inventory[kit_number]['status'] != 'available':
            raise ValueError(f"Kit {kit_number} is not available")

        dispense_transaction = {
            'transaction_type': 'dispensed',
            'transaction_date': dispense_date or datetime.now(),
            'kit_number': kit_number,
            'patient_id': patient_id,
            'visit_number': visit_number,
            'dispensed_by': dispensed_by
        }

        self.transactions.append(dispense_transaction)

        # Update inventory
        self.inventory[kit_number]['status'] = 'dispensed'
        self.inventory[kit_number]['patient_id'] = patient_id
        self.inventory[kit_number]['dispensed_date'] = dispense_date or datetime.now()

        return dispense_transaction

    def return_from_patient(self, kit_number, patient_id, return_date,
                           units_returned, units_used, returned_by):
        """
        Record kit return from patient (compliance check)
        """

        if kit_number not in self.inventory:
            raise ValueError(f"Kit {kit_number} not in inventory")

        return_transaction = {
            'transaction_type': 'returned_from_patient',
            'transaction_date': return_date or datetime.now(),
            'kit_number': kit_number,
            'patient_id': patient_id,
            'units_returned': units_returned,
            'units_used': units_used,
            'compliance_pct': (units_used / (units_used + units_returned) * 100) if (units_used + units_returned) > 0 else 0,
            'returned_by': returned_by
        }

        self.transactions.append(return_transaction)

        self.inventory[kit_number]['status'] = 'returned_from_patient'
        self.inventory[kit_number]['units_returned'] = units_returned
        self.inventory[kit_number]['units_used'] = units_used

        return return_transaction

    def quarantine_kit(self, kit_number, reason, quarantined_by):
        """
        Quarantine kit (temperature excursion, damaged, etc.)
        """

        if kit_number not in self.inventory:
            raise ValueError(f"Kit {kit_number} not in inventory")

        quarantine_transaction = {
            'transaction_type': 'quarantined',
            'transaction_date': datetime.now(),
            'kit_number': kit_number,
            'reason': reason,
            'quarantined_by': quarantined_by
        }

        self.transactions.append(quarantine_transaction)

        self.inventory[kit_number]['status'] = 'quarantined'
        self.inventory[kit_number]['quarantine_reason'] = reason

        return quarantine_transaction

    def destroy_kit(self, kit_number, destruction_method, witnessed_by,
                   destruction_certificate=None):
        """
        Record kit destruction
        """

        if kit_number not in self.inventory:
            raise ValueError(f"Kit {kit_number} not in inventory")

        destruction_transaction = {
            'transaction_type': 'destroyed',
            'transaction_date': datetime.now(),
            'kit_number': kit_number,
            'destruction_method': destruction_method,
            'witnessed_by': witnessed_by,
            'destruction_certificate': destruction_certificate
        }

        self.transactions.append(destruction_transaction)

        self.inventory[kit_number]['status'] = 'destroyed'
        self.inventory[kit_number]['destruction_date'] = datetime.now()

        return destruction_transaction

    def return_to_sponsor(self, kit_numbers, shipment_id, shipped_by,
                         tracking_number, return_reason):
        """
        Return kits to sponsor/depot
        """

        return_transaction = {
            'transaction_type': 'returned_to_sponsor',
            'transaction_date': datetime.now(),
            'kit_numbers': kit_numbers,
            'shipment_id': shipment_id,
            'shipped_by': shipped_by,
            'tracking_number': tracking_number,
            'return_reason': return_reason
        }

        self.transactions.append(return_transaction)

        for kit_number in kit_numbers:
            if kit_number in self.inventory:
                self.inventory[kit_number]['status'] = 'returned_to_sponsor'
                self.inventory[kit_number]['return_date'] = datetime.now()

        return return_transaction

    def reconcile_inventory(self, physical_count_by_kit):
        """
        Reconcile physical inventory with system records
        """

        discrepancies = []

        for kit_number, physical_status in physical_count_by_kit.items():
            system_status = self.inventory.get(kit_number, {}).get('status', 'NOT_IN_SYSTEM')

            if physical_status != system_status:
                discrepancies.append({
                    'kit_number': kit_number,
                    'system_status': system_status,
                    'physical_status': physical_status,
                    'discrepancy_type': self._classify_discrepancy(system_status, physical_status)
                })

        reconciliation = {
            'reconciliation_date': datetime.now(),
            'site_id': self.site_id,
            'total_kits_system': len(self.inventory),
            'total_kits_physical': len(physical_count_by_kit),
            'discrepancies': discrepancies,
            'reconciliation_status': 'CLEAN' if len(discrepancies) == 0 else 'DISCREPANCIES_FOUND'
        }

        return reconciliation

    def _classify_discrepancy(self, system_status, physical_status):
        """Classify type of discrepancy"""
        if system_status == 'NOT_IN_SYSTEM':
            return 'EXTRA_KIT_FOUND'
        elif physical_status == 'NOT_FOUND':
            return 'KIT_MISSING'
        else:
            return 'STATUS_MISMATCH'

    def _verify_temperature(self, temperature_log):
        """Verify temperature compliance during shipment"""
        if not temperature_log:
            return None

        # Check all readings in range
        compliant = all(
            log['min_temp'] <= log['temperature'] <= log['max_temp']
            for log in temperature_log
        )

        return compliant

    def accountability_report(self, report_date=None):
        """
        Generate drug accountability report
        """

        report_date = report_date or datetime.now()

        # Group inventory by status
        inventory_df = pd.DataFrame(self.inventory.values())

        if len(inventory_df) == 0:
            return None

        status_summary = inventory_df.groupby('status').size().to_dict()

        # Recent transactions
        recent_transactions = [
            t for t in self.transactions
            if t['transaction_date'] >= report_date - timedelta(days=30)
        ]

        report = {
            'site_id': self.site_id,
            'trial_id': self.trial_id,
            'report_date': report_date,
            'inventory_summary': status_summary,
            'total_kits': len(inventory_df),
            'dispensed_kits': len(inventory_df[inventory_df['status'] == 'dispensed']),
            'available_kits': len(inventory_df[inventory_df['status'] == 'available']),
            'recent_transactions': len(recent_transactions),
            'transactions': recent_transactions
        }

        return report

# Example usage
accountability = DrugAccountabilitySystem(
    site_id='SITE-001',
    trial_id='TRIAL-2024-001'
)

# Receive shipment
kits = [
    {'kit_number': 'KIT-001-1001', 'lot_number': 'LOT-A', 'expiry_date': datetime(2026, 12, 31)},
    {'kit_number': 'KIT-001-1002', 'lot_number': 'LOT-A', 'expiry_date': datetime(2026, 12, 31)},
    {'kit_number': 'KIT-001-1003', 'lot_number': 'LOT-A', 'expiry_date': datetime(2026, 12, 31)}
]

receipt = accountability.receive_shipment(
    shipment_id='SHIP-2024-0015',
    kits=kits,
    received_by='Study Coordinator Jane Doe',
    condition='Good',
    temperature_log=[{'temperature': 6, 'min_temp': 2, 'max_temp': 8}]
)

print(f"Received {len(kits)} kits")

# Dispense to patient
dispense = accountability.dispense_to_patient(
    kit_number='KIT-001-1001',
    patient_id='PT-001-001',
    visit_number='Visit 2',
    dispensed_by='Investigator Dr. Smith'
)

print(f"Dispensed kit {dispense['kit_number']} to patient {dispense['patient_id']}")

# Generate accountability report
report = accountability.accountability_report()
print(f"\nAccountability Report:")
print(f"  Total kits: {report['total_kits']}")
print(f"  Available: {report['available_kits']}")
print(f"  Dispensed: {report['dispensed_kits']}")
print(f"  Inventory summary: {report['inventory_summary']}")

Temperature-Controlled Logistics

Cold Chain Management for Clinical Trials

Temperature Ranges:

• Room temperature: 15-25°C (59-77°F)
• Refrigerated: 2-8°C (36-46°F)
• Frozen: -20°C (-4°F)
• Ultra-cold: -80°C (-112°F)

class ClinicalColdChainManager:
    """
    Manage temperature-controlled shipments for clinical trials
    """

    def __init__(self, trial_id):
        self.trial_id = trial_id
        self.shipments = {}
        self.excursions = []

    def create_shipment(self, shipment_id, product_name, from_location,
                       to_location, temp_requirement, packaging_type):
        """
        Create temperature-controlled shipment
        """

        shipment = {
            'shipment_id': shipment_id,
            'product_name': product_name,
            'from_location': from_location,
            'to_location': to_location,
            'temp_requirement': temp_requirement,  # e.g., (2, 8) for 2-8°C
            'packaging_type': packaging_type,
            'ship_date': None,
            'delivery_date': None,
            'temperature_log': [],
            'excursions_detected': [],
            'status': 'pending'
        }

        self.shipments[shipment_id] = shipment

        return shipment

    def ship(self, shipment_id, data_logger_id, carrier, tracking_number):
        """
        Ship temperature-controlled package
        """

        if shipment_id not in self.shipments:
            raise ValueError(f"Shipment {shipment_id} not found")

        shipment = self.shipments[shipment_id]

        shipment['ship_date'] = datetime.now()
        shipment['data_logger_id'] = data_logger_id
        shipment['carrier'] = carrier
        shipment['tracking_number'] = tracking_number
        shipment['status'] = 'in_transit'

        return shipment

    def record_temperature(self, shipment_id, timestamp, temperature, location='in transit'):
        """
        Record temperature reading from data logger
        """

        if shipment_id not in self.shipments:
            raise ValueError(f"Shipment {shipment_id} not found")

        shipment = self.shipments[shipment_id]
        min_temp, max_temp = shipment['temp_requirement']

        reading = {
            'timestamp': timestamp,
            'temperature': temperature,
            'location': location,
            'in_range': min_temp <= temperature <= max_temp
        }

        shipment['temperature_log'].append(reading)

        # Check for excursion
        if not reading['in_range']:
            excursion = {
                'shipment_id': shipment_id,
                'timestamp': timestamp,
                'temperature': temperature,
                'required_range': shipment['temp_requirement'],
                'deviation': abs(temperature - ((min_temp + max_temp) / 2)),
                'location': location
            }

            shipment['excursions_detected'].append(excursion)
            self.excursions.append(excursion)

            # Trigger alert
            self._temperature_excursion_alert(excursion)

        return reading

    def deliver_shipment(self, shipment_id, received_by, condition_assessment):
        """
        Record shipment delivery
        """

        if shipment_id not in self.shipments:
            raise ValueError(f"Shipment {shipment_id} not found")

        shipment = self.shipments[shipment_id]

        shipment['delivery_date'] = datetime.now()
        shipment['received_by'] = received_by
        shipment['condition_assessment'] = condition_assessment
        shipment['status'] = 'delivered'

        # Analyze temperature compliance
        compliance = self._analyze_temperature_compliance(shipment)

        shipment['temperature_compliance'] = compliance

        return {
            'shipment_id': shipment_id,
            'delivery_date': shipment['delivery_date'],
            'temperature_compliant': compliance['compliant'],
            'excursions': len(shipment['excursions_detected']),
            'disposition': self._determine_disposition(compliance)
        }

    def _analyze_temperature_compliance(self, shipment):
        """
        Analyze temperature compliance for shipment
        """

        temp_log = shipment['temperature_log']

        if not temp_log:
            return {'compliant': None, 'reason': 'No temperature data'}

        total_readings = len(temp_log)
        compliant_readings = sum(1 for r in temp_log if r['in_range'])
        compliance_rate = (compliant_readings / total_readings * 100) if total_readings > 0 else 0

        num_excursions = len(shipment['excursions_detected'])

        # Determine compliance
        compliant = (compliance_rate >= 95 and num_excursions == 0)

        return {
            'compliant': compliant,
            'compliance_rate': round(compliance_rate, 2),
            'num_excursions': num_excursions,
            'total_readings': total_readings,
            'compliant_readings': compliant_readings
        }

    def _determine_disposition(self, compliance):
        """
        Determine product disposition based on compliance
        """

        if compliance['compliant']:
            return 'ACCEPT - Use per protocol'
        elif compliance['num_excursions'] > 0:
            return 'QUARANTINE - Investigate excursion, stability assessment required'
        else:
            return 'QUARANTINE - Temperature compliance review required'

    def _temperature_excursion_alert(self, excursion):
        """
        Send alert for temperature excursion
        """

        print(f"⚠ TEMPERATURE EXCURSION ALERT")
        print(f"  Shipment: {excursion['shipment_id']}")
        print(f"  Temperature: {excursion['temperature']}°C")
        print(f"  Required range: {excursion['required_range']}")
        print(f"  Time: {excursion['timestamp']}")

    def excursion_investigation_report(self, shipment_id):
        """
        Generate excursion investigation report
        """

        if shipment_id not in self.shipments:
            raise ValueError(f"Shipment {shipment_id} not found")

        shipment = self.shipments[shipment_id]

        if not shipment['excursions_detected']:
            return {'investigation_required': False}

        # Analyze excursions
        excursions_df = pd.DataFrame(shipment['excursions_detected'])

        report = {
            'shipment_id': shipment_id,
            'product_name': shipment['product_name'],
            'investigation_required': True,
            'num_excursions': len(shipment['excursions_detected']),
            'excursion_details': excursions_df.to_dict('records'),
            'total_time_out_of_range': 'Calculate from timestamp data',
            'max_deviation': excursions_df['deviation'].max() if len(excursions_df) > 0 else 0,
            'recommended_action': self._excursion_recommended_action(shipment),
            'stability_data_required': True,
            'sponsor_notification_required': True
        }

        return report

    def _excursion_recommended_action(self, shipment):
        """
        Recommend action for excursion
        """

        excursions = shipment['excursions_detected']

        if not excursions:
            return 'No action required'

        max_deviation = max(e['deviation'] for e in excursions)

        if max_deviation > 10:
            return 'REJECT - Significant excursion, product unusable'
        elif max_deviation > 5:
            return 'QUARANTINE - Contact sponsor, stability data review required'
        else:
            return 'QUARANTINE - Minor excursion, sponsor review required'

# Example usage
cold_chain = ClinicalColdChainManager(trial_id='TRIAL-2024-001')

# Create shipment
shipment = cold_chain.create_shipment(
    shipment_id='SHIP-2024-0020',
    product_name='Investigational Biologic ABC-123',
    from_location='Depot - Amsterdam',
    to_location='Site 105 - Memorial Hospital',
    temp_requirement=(2, 8),  # 2-8°C
    packaging_type='Qualified shipper with dry ice'
)

# Ship
cold_chain.ship(
    shipment_id='SHIP-2024-0020',
    data_logger_id='LOGGER-5678',
    carrier='FedEx Priority Overnight',
    tracking_number='FX123456789'
)

# Simulate temperature readings
import numpy as np
np.random.seed(42)

for hour in range(36):  # 36 hours in transit
    temp = 5 + np.random.normal(0, 1.5)

    # Simulate excursion at hour 20
    if hour == 20:
        temp = 12  # Excursion

    cold_chain.record_temperature(
        shipment_id='SHIP-2024-0020',
        timestamp=datetime.now() + timedelta(hours=hour),
        temperature=round(temp, 1)
    )

# Deliver
delivery = cold_chain.deliver_shipment(
    shipment_id='SHIP-2024-0020',
    received_by='Study Coordinator at Site 105',
    condition_assessment='Package intact, data logger attached'
)

print(f"Shipment delivered:")
print(f"  Temperature compliant: {delivery['temperature_compliant']}")
print(f"  Excursions: {delivery['excursions']}")
print(f"  Disposition: {delivery['disposition']}")

# Excursion investigation
if delivery['excursions'] > 0:
    investigation = cold_chain.excursion_investigation_report('SHIP-2024-0020')
    print(f"\n Excursion Investigation Required:")
    print(f"  Number of excursions: {investigation['num_excursions']}")
    print(f"  Max deviation: {investigation['max_deviation']:.1f}°C")
    print(f"  Recommended action: {investigation['recommended_action']}")

Comparator Sourcing & Management

Commercial Comparator Procurement

Challenges:

• Sourcing commercial drugs in different countries
• Ensuring consistent quality across batches
• Managing expiry dates
• Blinding/overencapsulation
• Import/export compliance

class ComparatorManager:
    """
    Manage comparator drug sourcing and inventory
    """

    def __init__(self, trial_id):
        self.trial_id = trial_id
        self.comparators = {}
        self.procurement_orders = []

    def add_comparator(self, comparator_id, drug_name, strength,
                      countries_needed, blinding_required):
        """
        Add comparator to trial requirements
        """

        self.comparators[comparator_id] = {
            'comparator_id': comparator_id,
            'drug_name': drug_name,
            'strength': strength,
            'countries_needed': countries_needed,
            'blinding_required': blinding_required,
            'sourcing_strategy': self._determine_sourcing_strategy(countries_needed)
        }

    def _determine_sourcing_strategy(self, countries_needed):
        """
        Determine optimal sourcing strategy
        """

        if len(countries_needed) == 1:
            return 'local_sourcing'
        elif len(countries_needed) <= 5:
            return 'regional_sourcing'
        else:
            return 'global_sourcing_multiple_suppliers'

    def create_procurement_order(self, comparator_id, country, quantity,
                                target_delivery_date, supplier=None):
        """
        Create procurement order for comparator
        """

        if comparator_id not in self.comparators:
            raise ValueError(f"Comparator {comparator_id} not defined")

        comparator = self.comparators[comparator_id]

        order = {
            'order_id': f"PO-{len(self.procurement_orders)+1:05d}",
            'comparator_id': comparator_id,
            'drug_name': comparator['drug_name'],
            'country': country,
            'quantity': quantity,
            'target_delivery_date': target_delivery_date,
            'supplier': supplier or 'To be determined',
            'order_date': datetime.now(),
            'status': 'pending',
            'import_license_required': self._check_import_requirements(country),
            'blinding_required': comparator['blinding_required']
        }

        self.procurement_orders.append(order)

        return order

    def _check_import_requirements(self, country):
        """Check if import license required"""
        # Simplified - would check regulatory database
        controlled_countries = ['US', 'CA', 'AU', 'JP']
        return country in controlled_countries

    def quality_assessment(self, order_id, batch_number, test_results):
        """
        Record quality assessment of received comparator
        """

        order = next((o for o in self.procurement_orders if o['order_id'] == order_id), None)

        if not order:
            raise ValueError(f"Order {order_id} not found")

        assessment = {
            'order_id': order_id,
            'batch_number': batch_number,
            'assessment_date': datetime.now(),
            'test_results': test_results,
            'acceptable': all(t['result'] == 'pass' for t in test_results),
            'release_status': 'released' if all(t['result'] == 'pass' for t in test_results) else 'rejected'
        }

        order['quality_assessment'] = assessment
        order['status'] = assessment['release_status']

        return assessment

# Example usage
comparator_mgr = ComparatorManager(trial_id='TRIAL-2024-001')

# Add comparator requirement
comparator_mgr.add_comparator(
    comparator_id='COMP-001',
    drug_name='Lipitor (Atorvastatin) 40mg',
    strength='40mg',
    countries_needed=['US', 'UK', 'Germany', 'France', 'Japan'],
    blinding_required=True
)

# Create procurement orders
order_us = comparator_mgr.create_procurement_order(
    comparator_id='COMP-001',
    country='US',
    quantity=5000,
    target_delivery_date=datetime.now() + timedelta(days=90),
    supplier='Cardinal Health'
)

print(f"Procurement order created: {order_us['order_id']}")
print(f"  Import license required: {order_us['import_license_required']}")
print(f"  Blinding required: {order_us['blinding_required']}")

# Quality assessment
test_results = [
    {'test': 'Identification', 'result': 'pass'},
    {'test': 'Assay', 'result': 'pass', 'value': '99.2% (90-110% spec)'},
    {'test': 'Dissolution', 'result': 'pass'},
    {'test': 'Uniformity of Dosage Units', 'result': 'pass'}
]

qa = comparator_mgr.quality_assessment(
    order_id=order_us['order_id'],
    batch_number='BATCH-US-2024-A',
    test_results=test_results
)

print(f"\nQuality Assessment:")
print(f"  Acceptable: {qa['acceptable']}")
print(f"  Release status: {qa['release_status']}")

Clinical Trial Supply Chain Metrics

Key Performance Indicators

def calculate_clinical_trial_kpis(supply_data, enrollment_data, shipment_data):
    """
    Calculate clinical trial supply chain KPIs

    Parameters:
    - supply_data: Site inventory and supply data
    - enrollment_data: Patient enrollment data
    - shipment_data: Shipment performance data
    """

    kpis = {}

    # Stockout rate (sites without adequate supply)
    if 'stockout_event' in supply_data.columns:
        kpis['stockout_rate'] = (supply_data['stockout_event'].sum() / len(supply_data) * 100)

    # Depot-to-site delivery time
    if 'delivery_time_days' in shipment_data.columns:
        kpis['avg_delivery_time_days'] = shipment_data['delivery_time_days'].mean()

    # Temperature compliance rate
    if 'temp_compliant' in shipment_data.columns:
        kpis['temp_compliance_rate'] = (shipment_data['temp_compliant'].sum() / len(shipment_data) * 100)

    # Drug accountability compliance
    if 'accountability_complete' in supply_data.columns:
        kpis['accountability_compliance'] = (supply_data['accountability_complete'].sum() / len(supply_data) * 100)

    # Expiry waste rate
    if 'expired_kits' in supply_data.columns and 'total_kits' in supply_data.columns:
        total_expired = supply_data['expired_kits'].sum()
        total_kits = supply_data['total_kits'].sum()
        kpis['expiry_waste_rate'] = (total_expired / total_kits * 100) if total_kits > 0 else 0

    # Randomization-to-supply time (time to get drug after randomization)
    if 'randomization_to_supply_hours' in enrollment_data.columns:
        kpis['avg_randomization_to_supply_hours'] = enrollment_data['randomization_to_supply_hours'].mean()


…(truncated)