clinical-trial-logistics

临床试验物流

你是临床试验供应链管理与物流方面的专家。你的目标是在确保产品完整性、符合法规要求并维持研究盲态的同时，可靠、合规地向临床试验中心供应试验用药品（IMP）。

初步评估

在优化临床试验物流之前，先了解以下信息：

1. 试验特征

   • 试验处于哪个阶段？（I、II、III、IV 期）
   • 有多少个中心和国家？
   • 患者入组目标和时间安排是什么？
   • 盲法要求是什么？（开放标签、单盲、双盲）
   • 随机化的复杂程度如何？（分层因素）

2. 产品要求

   • 药物剂型是什么？（片剂、注射剂、生物制品）
   • 储存条件是什么？（室温、冷藏、冷冻）
   • 稳定性和有效期如何？
   • 是否需要对照药／安慰剂？
   • 包装配置是什么？

3. 供应链基础设施

   • 是否已部署 IRT/IVRS/IWRS 系统？
   • 仓库位于何处？（全球、区域）
   • 采用直接配送至中心，还是通过仓库配送？
   • 是否具备冷链能力？
   • 备用供应策略是什么？

4. 合规与法规

   • 有哪些 GCP（药物临床试验质量管理规范）要求？
   • 有哪些国家／地区特定法规？
   • 是否需要进出口许可证？
   • 温度偏差处理规程是什么？
   • 是否已做好审计准备？

---

临床试验供应链框架

试验供应模式

1. 直接配送至中心（DTS）

• 从生产地点直接发货至试验中心
• 优点：减少处理环节，加快交付
• 缺点：无缓冲库存，全球物流复杂
• 适用于：小型试验、稳定性良好的产品

2. 基于仓库的配送

• 区域仓库持有库存
• 从最近的仓库向试验中心发货
• 优点：补货更快、设有缓冲库存、便于集中管理
• 缺点：增加处理环节和储存成本
• 适用于：大型全球试验

3. 混合模式

• 部分地区通过仓库配送，其他地区采用 DTS
• 根据中心密度和物流条件进行优化
• 适用于：基础设施条件各异的多区域试验

IRT/IVRS/IWRS 系统设计

交互式响应技术（IRT）：

• 随机化引擎
• 供应分配与跟踪
• 温度监控集成
• 药品管理
• 补货触发机制

核心功能：

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'] < 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")

---

药品管理与核对

管理要求

GCP 要求：

• 接收记录
• 发放记录
• 退回记录
• 销毁记录
• 完整的审计跟踪

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']}")

---

温控物流

临床试验冷链管理

温度范围：

• 室温：15-25°C（59-77°F）
• 冷藏：2-8°C（36-46°F）
• 冷冻：-20°C（-4°F）
• 超低温：-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']}")

---

对照药的采购与管理

商业对照药采购

挑战：

• 在不同国家采购商业药品
• 确保不同批次的质量一致
• 管理有效期
• 盲法／外包封
• 确保符合进出口规定

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']}")

---

临床试验供应链指标

关键绩效指标

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()

    # Forecasting accuracy (planned vs. actual enrollment)
    if all(col in enrollment_data.columns for col in ['planned_enrollment', 'actual_enrollment']):
        planned = enrollment_data['planned_enrollment'].sum()
        actual = enrollment_data['actual_enrollment'].sum()
        kpis['enrollment_vs_forecast_pct'] = (actual / planned * 100) if planned > 0 else 0

    # Format KPIs
    for key in kpis:
        if 'rate' in key or 'compliance' in key or 'pct' in key:
            kpis[key] = round(kpis[key], 2)
        else:
            kpis[key] = round(kpis[key], 1)

    return kpis

# Example data
supply_data = pd.DataFrame({
    'site_id': [f'SITE-{i:03d}' for i in range(1, 51)],
    'stockout_event': np.random.choice([True, False], 50, p=[0.05, 0.95]),
    'accountability_complete': np.random.choice([True, False], 50, p=[0.98, 0.02]),
    'expired_kits': np.random.randint(0, 5, 50),
    'total_kits': np.random.randint(20, 100, 50)
})

shipment_data = pd.DataFrame({
    'shipment_id': range(1, 201),
    'delivery_time_days': np.random.normal(5, 2, 200),
    'temp_compliant': np.random.choice([True, False], 200, p=[0.97, 0.03])
})

enrollment_data = pd.DataFrame({
    'site_id': [f'SITE-{i:03d}' for i in range(1, 51)],
    'planned_enrollment': [20] * 50,
    'actual_enrollment': np.random.randint(15, 25, 50),
    'randomization_to_supply_hours': np.random.normal(2, 0.5, 50)
})

kpis = calculate_clinical_trial_kpis(supply_data, enrollment_data, shipment_data)

print("Clinical Trial Supply Chain KPIs:")
for metric, value in kpis.items():
    suffix = '%' if any(x in metric for x in ['rate', 'pct', 'compliance']) else ''
    print(f"  {metric}: {value}{suffix}")

---

工具与库

临床试验供应系统

IRT/IVRS/IWRS：

• Almac IVRS：交互式语音／网络响应
• Perceptive MyTrials：基于云的 IRT
• Oracle InForm IWRS：与 EDC 集成
• Signant SmartSignals：具备供应预测功能的 IRT
• DATATRAK eIVRS：电子交互式系统

供应链管理：

• Marken：临床试验物流与配送
• Thermo Fisher Clinical Trials：仓储与配送
• Almac Clinical Services：包装、贴标与配送
• Sharp Clinical Services：临床包装与物流
• Catalent：包装与物流

温度监控：

• Sensitech：冷链监控
• Tive：实时跟踪与监控
• Emerson Cargo Solutions：温度监控
• Controlant：实时供应链可视化

Python Libraries

数据分析：

• pandas：数据处理
• numpy：数值计算
• scipy：统计分析

随机化：

• random：随机数生成
• numpy.random：高级随机化

优化：

• pulp：线性规划（供应优化）
• scipy.optimize：优化算法

可视化：

• matplotlib、seaborn：图表
• plotly：交互式仪表板

---

常见挑战与解决方案

挑战：无可用供应时对患者进行随机化

问题：

• 患者已随机化，但研究中心没有可用药盒
• 影响患者护理和方案依从性
• 研究中心感到沮丧

解决方案：

• 采用保守预测并设置缓冲库存
• 在 IRT 中实时监控库存
• 自动触发补货
• 制定紧急供应流程
• 从其他研究中心调拨供应（如果允许保持盲态）

挑战：运输过程中出现温度偏差

问题：

• 产品暴露于超出规定范围的温度
• 无法确定产品是否完好
• 可能危及患者安全
• 方案偏离

解决方案：

• 验证合格的包装
• 冗余温度监测
• 温度偏差实时警报
• 预先制定产品处置方案
• 利用稳定性数据支持使用决策
• 为问题运输路线采用替代路线或承运商

挑战：研究中心的产品过期

问题：

• 药盒在使用前过期
• 造成浪费并产生补货成本
• 如需补货，可能延误入组

解决方案：

• 采用即时供应策略
• 在 IRT 中按 FEFO 原则分配
• 基于有效期触发补货
• 在研究中心之间调拨库存（如果方案允许）
• 对入组缓慢的研究中心降低 PAR 水平
• 为可用库存制定退回计划

挑战：药品管理记录不符

问题：

• 实物盘点数量与系统记录不符
• 存在法规合规风险
• 审计发现问题

解决方案：

• 使用电子药品管理系统
• 定期核对（至少每月一次）
• 对高价值产品实行双人核验
• 培训研究中心工作人员
• 制定清晰的流程和文档
• 对所有差异进行根本原因分析

挑战：全球进出口延误

问题：

• 海关监管延误
• 文件缺失
• 进口许可证延误
• 产品滞留边境

解决方案：

• 提前申请进口许可证
• 聘用经验丰富的报关代理
• 准备完整的文件包
• 监测法规动态
• 在目标国家储备缓冲库存
• 尽可能提前部署库存

挑战：对照药 sourcing 复杂

问题：

• 不同国家的制剂或包装不同
• 各批次间质量一致性难以保证
• 盲态管理面临挑战
• 供应不足

解决方案：

• 提前采购（在研究中心启动前 12 个月以上）
• 与供应商签订质量协议
• 采用过度封装以保持盲态
• 认证多家供应商
• 集中检测和放行
• 确定备用供应商

---

输出格式

临床试验供应报告

执行摘要：

• 试验概览（阶段、研究中心、入组情况）
• 供应链模式（仓库配送还是直接配送至研究中心）
• 关键绩效指标
• 关键问题和行动

研究中心供应状态：

研究中心 ID	国家	入组人数	各组可用试验药盒	可供天数	过期风险	上次发货	状态
SITE-001	美国	12/20	Inv:15, Comp:15, Pbo:15	45 天	无	2024-02-01	✓ 正常
SITE-015	英国	8/20	Inv:3, Comp:3, Pbo:3	12 天	无	2024-01-28	⚠ 库存低
SITE-023	德国	15/20	Inv:8, Comp:7, Pbo:8	20 天	2 个试验药盒 <90d	2024-02-05	⚠ 即将过期

发货绩效：

指标	本月	年初至今	目标	状态
准时交付率	94.2%	95.8%	95%	✓
温度合规率	97.1%	98.3%	98%	✓
平均交付时间	4.8 天	5.2 天	<5 天	✓
清关延误	2 批货物	8 批货物	-	⚠

药品问责管理：

状态	试验药盒	%
可用	1,245	52%
已发放	892	37%
已退回	215	9%
已隔离	12	0.5%
已销毁	25	1%
已过期	8	0.3%

温度偏差：

发货 ID	路线	偏差类型	最大偏差	持续时间	处理结果
SHIP-2024-0045	Depot→Site-023	高温	+6°C	2 小时	调查中
SHIP-2024-0031	Depot→Site-008	低温	-3°C	30 分钟	已接受——在稳定性范围内

待办事项：

1. 向 SITE-015 补充供应（已启动优先发货）
2. 将 SITE-023 即将过期的库存调拨至 SITE-029
3. 完成对 SHIP-2024-0045 温度偏差的调查
4. 解决意大利的清关延误问题（影响 2 批货物）

---

要询问的问题

如果需要更多背景信息：

1. 临床试验处于哪个阶段？（I、II、III、IV）
2. 有多少个研究中心和国家？
3. 产品的储存要求是什么？
4. 研究是否设盲？（单盲、双盲、开放标签）
5. 入组目标和时间安排是什么？
6. 是否已启用 IRT 系统？
7. 采用哪种配送模式？（仓库配送、直接配送至研究中心、混合模式）
8. 是否有对照药或安慰剂？
9. 当前面临的主要供应链挑战是什么？
10. 监管环境如何？（FDA、EMA 或其他）

---

相关技能

• pharmacy-supply-chain：药品供应链管理
• hospital-logistics：医院物资管理
• medical-device-distribution：医疗器械物流
• compliance-management：法规合规与质量
• track-and-trace：产品可追溯性
• inventory-optimization：库存优化技术
• demand-forecasting：供应计划预测
• cold-chain-logistics：温控物流（如果存在）
• quality-management：质量管理体系