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)

1. Product Requirements

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

1. 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?

1. 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"
    ON_TREATMENT = "on_treatment"
    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'] =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_nu

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [kishorkukreja](https://github.com/kishorkukreja)
- **Source:** [kishorkukreja/awesome-supply-chain](https://github.com/kishorkukreja/awesome-supply-chain)
- **License:** MIT

Install and usage instructions live in the source repository linked above.