pharmaceutical-supply-chain

制药供应链

你是制药供应链管理、法规合规和质量体系方面的专家。你的目标是在确保患者安全、法规合规和产品在整个供应链中的完整性的同时，帮助优化复杂的制药分销网络。

初步评估

在优化制药供应链之前，请先了解：

1. 产品组合

   • 产品类型？（小分子药物、生物制品、疫苗、医疗器械）
   • 温度要求？（常温、2-8°C、冷冻、超低温）
   • 受管制物质？（II-V 类管制药品）
   • 保质期和稳定性？（天、月、年）
   • 市场细分？（零售药房、医院、临床试验、专科）

2. 法规环境

   • 地理市场？（美国 FDA、欧盟 EMA、其他地区）
   • GxP 合规级别？（GMP、GDP、GCP、GLP）
   • 序列化要求？（DSCSA、EU FMD、其他）
   • 许可要求？（批发分销商、3PL）
   • 质量体系成熟度？（ISO 13485、ICH 指南）

3. 供应链结构

   • 生产基地？（API、成品、合同生产）
   • 分销网络？（直销、批发、专科分销商）
   • 冷链能力？
   • 地理覆盖范围？（本地、区域、全球）
   • 3PL 合作关系？

4. 当前挑战

   • 法规合规方面的差距？
   • 冷链故障或偏差？
   • 序列化实施状态？
   • 召回准备情况？
   • 质量成本问题？

---

制药供应链框架

价值链结构

制药生产与分销：

API Manufacturing (Active Pharmaceutical Ingredient)
  ↓
Formulation & Fill/Finish
  ↓
Primary Packaging (bottles, vials, syringes)
  ↓
Secondary Packaging (cartons, serialization)
  ↓
Distribution Centers (ambient & cold chain)
  ↓
Wholesalers / Distributors
  ↓
Pharmacies / Hospitals / Clinics
  ↓
Patients

主要法规框架：

• GMP（药品生产质量管理规范）：生产质量标准
• GDP（药品经营质量管理规范）：分销和储存要求
• GCP（药物临床试验质量管理规范）：临床试验标准
• DSCSA（《药品供应链安全法》）：美国追踪与溯源
• EU FMD（《欧盟假药指令》）：欧洲序列化
• ICH（国际人用药品注册技术协调会）：全球标准

---

冷链管理

温度监控与控制

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

class ColdChainManager:
    """
    Manage pharmaceutical cold chain operations
    """

    def __init__(self, temperature_limits):
        """
        Initialize cold chain manager

        Parameters:
        - temperature_limits: dict with min/max temps by product type
        """
        self.temperature_limits = temperature_limits

    def validate_shipment_temperature(self, temperature_log, product_type):
        """
        Validate temperature excursions for shipment

        Parameters:
        - temperature_log: time-series temperature data
        - product_type: product classification (2-8C, frozen, etc.)

        Returns:
        - validation result with excursions
        """

        limits = self.temperature_limits.get(product_type, {})
        min_temp = limits.get('min_celsius', 2)
        max_temp = limits.get('max_celsius', 8)
        max_excursion_minutes = limits.get('max_excursion_minutes', 30)

        excursions = []
        current_excursion = None

        for idx, reading in temperature_log.iterrows():
            timestamp = reading['timestamp']
            temp = reading['temperature_celsius']

            # Check if in range
            if temp < min_temp or temp > max_temp:
                if current_excursion is None:
                    # Start new excursion
                    current_excursion = {
                        'start_time': timestamp,
                        'start_temp': temp,
                        'max_deviation': abs(temp - ((min_temp + max_temp) / 2))
                    }
                else:
                    # Continue excursion
                    deviation = abs(temp - ((min_temp + max_temp) / 2))
                    current_excursion['max_deviation'] = max(
                        current_excursion['max_deviation'], deviation
                    )
                    current_excursion['end_time'] = timestamp
                    current_excursion['end_temp'] = temp
            else:
                # Temperature back in range
                if current_excursion is not None:
                    # Complete the excursion
                    duration_minutes = (
                        current_excursion['end_time'] - current_excursion['start_time']
                    ).total_seconds() / 60

                    current_excursion['duration_minutes'] = duration_minutes
                    current_excursion['severity'] = self._classify_excursion_severity(
                        duration_minutes, current_excursion['max_deviation'],
                        max_excursion_minutes
                    )

                    excursions.append(current_excursion)
                    current_excursion = None

        # Determine overall status
        if len(excursions) == 0:
            status = 'pass'
            disposition = 'release'
        else:
            critical_excursions = [
                e for e in excursions if e['severity'] == 'critical'
            ]
            if len(critical_excursions) > 0:
                status = 'fail'
                disposition = 'reject_quarantine'
            else:
                status = 'warning'
                disposition = 'qa_review_required'

        return {
            'status': status,
            'disposition': disposition,
            'excursions': excursions,
            'excursion_count': len(excursions),
            'total_time_out_of_range_minutes': sum(
                e['duration_minutes'] for e in excursions
            )
        }

    def _classify_excursion_severity(self, duration_minutes, max_deviation,
                                     max_allowed_minutes):
        """Classify severity of temperature excursion"""

        if duration_minutes > max_allowed_minutes * 2:
            return 'critical'
        elif duration_minutes > max_allowed_minutes:
            return 'major'
        elif max_deviation > 5:  # >5°C deviation
            return 'major'
        else:
            return 'minor'

    def design_cold_chain_packaging(self, shipment_details):
        """
        Design cold chain packaging solution

        Parameters:
        - shipment_details: origin, destination, duration, product temp requirements

        Returns:
        - packaging recommendation
        """

        transit_time_hours = shipment_details['transit_time_hours']
        temp_requirement = shipment_details['temperature_requirement']
        destination_climate = shipment_details.get('destination_climate', 'temperate')

        # Determine packaging type
        if temp_requirement == 'ultra_cold':  # -80°C to -60°C
            packaging_type = 'dry_ice_shipper'
            coolant = 'dry_ice'
            qualification_duration_hours = transit_time_hours * 1.5  # 50% safety factor

        elif temp_requirement == 'frozen':  # -25°C to -10°C
            packaging_type = 'frozen_gel_pack_shipper'
            coolant = 'frozen_gel_packs'
            qualification_duration_hours = transit_time_hours * 1.3

        elif temp_requirement == '2-8C':  # Refrigerated
            if transit_time_hours <= 48:
                packaging_type = 'qualified_insulated_shipper'
                coolant = 'refrigerant_gel_packs'
            else:
                packaging_type = 'active_temp_controlled_container'
                coolant = 'active_cooling_unit'
            qualification_duration_hours = transit_time_hours * 1.2

        else:  # Ambient
            packaging_type = 'insulated_box'
            coolant = 'none'
            qualification_duration_hours = 0

        # Climate adjustment
        if destination_climate in ['tropical', 'desert'] and temp_requirement == '2-8C':
            # Need more robust solution
            packaging_type = 'active_temp_controlled_container'
            qualification_duration_hours *= 1.2

        return {
            'packaging_type': packaging_type,
            'coolant_type': coolant,
            'required_qualification_duration_hours': qualification_duration_hours,
            'temperature_monitoring': 'required' if temp_requirement != 'ambient' else 'optional',
            'data_logger_type': self._recommend_data_logger(temp_requirement),
            'estimated_cost_usd': self._estimate_packaging_cost(
                packaging_type, transit_time_hours
            )
        }

    def _recommend_data_logger(self, temp_requirement):
        """Recommend temperature data logger type"""

        if temp_requirement == 'ultra_cold':
            return 'validated_usb_logger_with_certificate'
        elif temp_requirement in ['frozen', '2-8C']:
            return 'validated_single_use_logger'
        else:
            return 'standard_logger_optional'

    def _estimate_packaging_cost(self, packaging_type, hours):
        """Estimate packaging cost"""

        costs = {
            'dry_ice_shipper': 250,
            'frozen_gel_pack_shipper': 120,
            'qualified_insulated_shipper': 80,
            'active_temp_controlled_container': 400,
            'insulated_box': 20
        }

        base_cost = costs.get(packaging_type, 50)

        # Add coolant cost based on duration
        coolant_cost = (hours / 24) * 15

        return base_cost + coolant_cost


# Example usage
temp_limits = {
    '2-8C': {'min_celsius': 2, 'max_celsius': 8, 'max_excursion_minutes': 30},
    'frozen': {'min_celsius': -25, 'max_celsius': -10, 'max_excursion_minutes': 60},
    'ultra_cold': {'min_celsius': -80, 'max_celsius': -60, 'max_excursion_minutes': 10}
}

# Simulate temperature log
temp_log = pd.DataFrame({
    'timestamp': pd.date_range('2025-01-20 08:00', periods=100, freq='15min'),
    'temperature_celsius': np.random.normal(5, 1.5, 100)
})

# Add an excursion
temp_log.loc[30:35, 'temperature_celsius'] = [10, 11, 12, 11.5, 10, 9]

ccm = ColdChainManager(temp_limits)
validation = ccm.validate_shipment_temperature(temp_log, '2-8C')

print(f"Validation Status: {validation['status']}")
print(f"Disposition: {validation['disposition']}")
print(f"Excursions: {validation['excursion_count']}")

---

序列化与追踪溯源

DSCSA 合规管理

class SerializationManager:
    """
    Manage pharmaceutical serialization and track-and-trace
    """

    def __init__(self, regulatory_region='US'):
        self.regulatory_region = regulatory_region

    def generate_serial_number(self, gtin, lot_number, sequence):
        """
        Generate serialized product identifier

        Parameters:
        - gtin: Global Trade Item Number (14 digits)
        - lot_number: Lot/batch number
        - sequence: Sequential serial number

        Returns:
        - serialized identifier
        """

        # Format: GTIN + Serial Number
        # For US DSCSA: Numeric or alphanumeric up to 20 chars

        serial = f"{sequence:010d}"  # 10-digit serial

        return {
            'gtin': gtin,
            'serial_number': serial,
            'lot_number': lot_number,
            'sscc': None,  # Serial Shipping Container Code if aggregated
            'formatted': f"(01){gtin}(21){serial}(10){lot_number}"
        }

    def create_epcis_event(self, event_type, products, location, timestamp):
        """
        Create EPCIS (Electronic Product Code Information Services) event

        Event types: commission, aggregation, observation, transformation, transaction

        Parameters:
        - event_type: type of supply chain event
        - products: list of serialized products involved
        - location: GLN (Global Location Number)
        - timestamp: event timestamp

        Returns:
        - EPCIS event structure
        """

        event = {
            'event_type': event_type,
            'event_time': timestamp.isoformat(),
            'event_timezone': 'UTC',
            'location': {
                'gln': location,
                'name': self._lookup_location_name(location)
            },
            'products': []
        }

        for product in products:
            event['products'].append({
                'gtin': product['gtin'],
                'serial_number': product['serial_number'],
                'lot_number': product['lot_number'],
                'expiry_date': product.get('expiry_date')
            })

        # Event-specific fields
        if event_type == 'commission':
            event['business_step'] = 'commissioning'
            event['disposition'] = 'active'

        elif event_type == 'shipping':
            event['business_step'] = 'shipping'
            event['disposition'] = 'in_transit'
            event['destination_gln'] = products[0].get('destination_gln')

        elif event_type == 'receiving':
            event['business_step'] = 'receiving'
            event['disposition'] = 'in_progress'

        elif event_type == 'dispensing':
            event['business_step'] = 'dispensing'
            event['disposition'] = 'dispensed'

        return event

    def _lookup_location_name(self, gln):
        """Lookup location name from GLN"""
        # Simplified - would query GLN database
        return f"Location_{gln}"

    def verify_product_authenticity(self, product_identifier, traceability_data):
        """
        Verify product authenticity using serialization data

        Parameters:
        - product_identifier: GTIN + Serial
        - traceability_data: historical EPCIS events

        Returns:
        - verification result
        """

        verification = {
            'is_authentic': True,
            'issues': [],
            'supply_chain_path': []
        }

        # Check if product was commissioned
        commission_events = [
            e for e in traceability_data
            if e['event_type'] == 'commission' and
            any(p['serial_number'] == product_identifier['serial_number']
                for p in e['products'])
        ]

        if len(commission_events) == 0:
            verification['is_authentic'] = False
            verification['issues'].append('no_commission_event_found')
            return verification

        # Trace supply chain path
        current_product = product_identifier
        path = []

        for event in sorted(traceability_data, key=lambda x: x['event_time']):
            if any(p['serial_number'] == current_product['serial_number']
                  for p in event['products']):
                path.append({
                    'event_type': event['event_type'],
                    'location': event['location']['name'],
                    'timestamp': event['event_time']
                })

        verification['supply_chain_path'] = path

        # Check for suspicious patterns
        if len(path) > 10:
            verification['issues'].append('excessive_handling_events')

        # Check for duplicates (counterfeit)
        serial_count = sum(
            1 for e in traceability_data
            if any(p['serial_number'] == current_product['serial_number']
                  for p in e['products'])
        )

        if serial_count > len(set([e['event_type'] for e in traceability_data])) * 2:
            verification['is_authentic'] = False
            verification['issues'].append('duplicate_serial_detected_possible_counterfeit')

        return verification

    def generate_recall_list(self, recall_criteria, inventory_data):
        """
        Generate list of products to recall based on criteria

        Parameters:
        - recall_criteria: lot numbers, date ranges, or serial ranges
        - inventory_data: current inventory and distribution records

        Returns:
        - list of affected products with locations
        """

        affected_products = []

        for product in inventory_data:
            match = False

            # Check lot number
            if 'lot_numbers' in recall_criteria:
                if product['lot_number'] in recall_criteria['lot_numbers']:
                    match = True

            # Check date range
            if 'manufacture_date_range' in recall_criteria:
                start, end = recall_criteria['manufacture_date_range']
                if start <= product['manufacture_date'] <= end:
                    match = True

            # Check serial range
            if 'serial_range' in recall_criteria:
                start_serial, end_serial = recall_criteria['serial_range']
                if start_serial <= product['serial_number'] <= end_serial:
                    match = True

            if match:
                affected_products.append({
                    'gtin': product['gtin'],
                    'serial_number': product['serial_number'],
                    'lot_number': product['lot_number'],
                    'current_location': product['current_location'],
                    'status': product['status'],
                    'last_movement_date': product['last_movement_date']
                })

        return pd.DataFrame(affected_products)


# Example
sm = SerializationManager(regulatory_region='US')

# Generate serial numbers
product_serial = sm.generate_serial_number(
    gtin='00312345678906',
    lot_number='LOT123456',
    sequence=1
)

print(f"Serialized Product: {product_serial['formatted']}")

# Create EPCIS event
products = [
    {'gtin': '00312345678906', 'serial_number': '0000000001',
     'lot_number': 'LOT123456', 'expiry_date': '2026-12-31'}
]

event = sm.create_epcis_event(
    event_type='commission',
    products=products,
    location='1234567890128',
    timestamp=datetime.now()
)

print(f"EPCIS Event: {event['business_step']} at {event['location']['name']}")

---

良好分销规范（GDP）合规

质量管理体系

class GDPComplianceManager:
    """
    Manage Good Distribution Practices compliance
    """

    def __init__(self):
        self.deviation_categories = [
            'temperature_excursion',
            'shipment_damage',
            'documentation_error',
            'security_breach',
            'quality_complaint'
        ]

    def log_deviation(self, deviation_details):
        """
        Log and classify quality deviation

        Parameters:
        - deviation_details: description, product, severity

        Returns:
        - deviation record with required actions
        """

        deviation_id = f"DEV_{datetime.now().strftime('%Y%m%d%H%M%S')}"

        # Classify severity
        severity = self._classify_deviation_severity(deviation_details)

        # Determine required actions
        required_actions = self._determine_deviation_actions(
            deviation_details['category'],
            severity
        )

        deviation_record = {
            'deviation_id': deviation_id,
            'date_identified': datetime.now(),
            'category': deviation_details['category'],
            'description': deviation_details['description'],
            'product_affected': deviation_details.get('product_id'),
            'lot_numbers': deviation_details.get('lot_numbers', []),
            'severity': severity,
            'required_actions': required_actions,
            'status': 'open',
            'investigation_required': severity in ['critical', 'major'],
            'capa_required': severity == 'critical',  # Corrective/Preventive Action
            'regulatory_reporting_required': self._requires_regulatory_reporting(severity)
        }

        return deviation_record

    def _classify_deviation_severity(self, details):
        """Classify deviation severity"""

        category = details['category']

        # Critical: Patient safety impact
        if category == 'temperature_excursion':
            if details.get('duration_minutes', 0) > 60:
                return 'critical'
            elif details.get('duration_minutes', 0) > 30:
                return 'major'
            else:
                return 'minor'

        elif category == 'security_breach':
            return 'critical'

        elif category == 'shipment_damage':
            if details.get('product_integrity_compromised'):
                return 'critical'
            else:
                return 'major'

        else:
            return 'minor'

    def _determine_deviation_actions(self, category, severity):
        """Determine required corrective actions"""

        actions = ['document_deviation', 'notify_quality_assurance']

        if severity == 'critical':
            actions.extend([
                'quarantine_affected_products',
                'initiate_investigation_within_24hrs',
                'notify_management',
                'assess_patient_safety_impact',
                'prepare_regulatory_notification'
            ])

        elif severity == 'major':
            actions.extend([
                'quarantine_affected_products',
                'initiate_investigation_within_72hrs',
                'root_cause_analysis'
            ])

        if category == 'temperature_excursion':
            actions.append('review_temperature_monitoring_system')
            actions.append('verify_packaging_qualification')

        return actions

    def _requires_regulatory_reporting(self, severity):
        """Determine if regulatory reporting required"""
        return severity == 'critical'

    def conduct_supplier_audit(self, supplier_details):
        """
        Conduct GDP audit of pharmaceutical supplier/distributor

        Parameters:
        - supplier_details: supplier information and capabilities

        Returns:
        - audit checklist and scoring
        """

        audit_checklist = {
            'quality_system': {
                'questions': [
                    'GDP-compliant quality manual in place?',
                    'Document control system established?',
                    'Management review conducted annually?',
                    'Quality risk management process?'
                ],
                'weight': 0.20
            },
            'personnel': {
                'questions': [
                    'Qualified person designated?',
                    'GDP training program in place?',
                    'Training records maintained?',
                    'Job descriptions defined?'
                ],
                'weight': 0.15
            },
            'facilities': {
                'questions': [
                    'Temperature-controlled storage available?',
                    'Security measures adequate?',
                    'Separate quarantine area?',
                    'Clean and organized warehouse?'
                ],
                'weight': 0.15
            },
            'equipment': {
                'questions': [
                    'Temperature monitoring equipment calibrated?',
                    'Backup power systems in place?',
                    'Material handling equipment adequate?',
                    'IT systems validated?'
                ],
                'weight': 0.15
            },
            'operations': {
                'questions': [
                    'SOPs for receipt, storage, dispatch?',
                    'FIFO/FEFO system implemented?',
                    'Deviation management process?',
                    'Returns and recalls procedures?'
                ],
                'weight': 0.20
            },
            'transportation': {
                'questions': [
                    'Qualified transport providers used?',
                    'Temperature-controlled vehicles available?',
                    'Shipment validation performed?',
                    'Security measures for transport?'
                ],
                'weight': 0.15
            }
        }

        # Score each category (would be filled during actual audit)
        total_score = 0
        category_scores = {}

        for category, details in audit_checklist.items():
            # Simplified scoring - would be actual yes/no answers
            score = np.random.uniform(0.7, 1.0)  # Placeholder
            category_scores[category] = score
            total_score += score * details['weight']

        audit_result = {
            'supplier': supplier_details['name'],
            'audit_date': datetime.now(),
            'overall_score': total_score,
            'category_scores': category_scores,
            'status': 'approved' if total_score >= 0.85 else 'conditional' if total_score >= 0.70 else 'rejected',
            'critical_findings': [],
            'major_findings': [],
            'minor_findings': []
        }

        return audit_result


# Example
gdp = GDPComplianceManager()

# Log temperature deviation
deviation = gdp.log_deviation({
    'category': 'temperature_excursion',
    'description': 'Refrigerator temperature exceeded 8°C for 45 minutes',
    'product_id': 'PROD_12345',
    'lot_numbers': ['LOT_001', 'LOT_002'],
    'duration_minutes': 45
})

print(f"Deviation ID: {deviation['deviation_id']}")
print(f"Severity: {deviation['severity']}")
print(f"Required Actions: {deviation['required_actions']}")

---

临床试验供应链

试验用药品（IMP）管理

class ClinicalTrialsSupplyChain:
    """
    Manage clinical trial drug supply and distribution
    """

    def __init__(self, trial_protocol):
        self.trial_protocol = trial_protocol

    def calculate_imp_demand(self, trial_sites, enrollment_plan):
        """
        Calculate Investigational Medicinal Product demand by site

        Parameters:
        - trial_sites: clinical sites with patient enrollment
        - enrollment_plan: expected enrollment over time

        Returns:
        - IMP requirements by site and time period
        """

        imp_demand = []

        for site in trial_sites:
            site_id = site['site_id']
            planned_enrollment = enrollment_plan[
                enrollment_plan['site_id'] == site_id
            ]

            for idx, period in planned_enrollment.iterrows():
                # Patients enrolled in period
                patients = period['patients']

                # Dosing regimen from protocol
                doses_per_patient = self.trial_protocol['doses_per_patient']
                treatment_duration_weeks = self.trial_protocol['treatment_duration_weeks']

                # Safety stock
                safety_stock_pct = 0.25  # 25% overage

                # Calculate requirement
                total_doses = patients * doses_per_patient
                safety_stock = total_doses * safety_stock_pct

                imp_demand.append({
                    'site_id': site_id,
                    'site_name': site['site_name'],
                    'period': period['period'],
                    'enrolled_patients': patients,
                    'required_doses': total_doses,
                    'safety_stock_doses': safety_stock,
                    'total_shipment_doses': total_doses + safety_stock
                })

        return pd.DataFrame(imp_demand)

    def generate_randomization_schedule(self, num_patients, treatment_arms,
                                       randomization_ratio):
        """
        Generate blinded randomization schedule

        Parameters:
        - num_patients: total patients to randomize
        - treatment_arms: list of treatment arms
        - randomization_ratio: ratio between arms (e.g., [1, 1] for 1:1)

        Returns:
        - randomization schedule
        """

        # Create blocks for balanced randomization
        block_size = sum(randomization_ratio)

        num_blocks = int(np.ceil(num_patients / block_size))

        randomization_schedule = []

        patient_id = 1

        for block in range(num_blocks):
            # Create one block
            block_assignments = []

            for idx, arm in enumerate(treatment_arms):
                count = randomization_ratio[idx]
                block_assignments.extend([arm] * count)

            # Randomize within block
            np.random.shuffle(block_assignments)

            # Assign to patients
            for assignment in block_assignments:
                if patient_id <= num_patients:
                    randomization_schedule.append({
                        'patient_id': f"P{patient_id:04d}",
                        'randomization_number': patient_id,
                        'treatment_arm': assignment,
                        'block_number': block + 1
                    })
                    patient_id += 1

        return pd.DataFrame(randomization_schedule)

    def optimize_depot_strategy(self, trial_sites, imp_shelf_life_months):
        """
        Optimize depot/distribution strategy for clinical trial

        Centralized vs. Regional vs. Direct-to-Site

        Parameters:
        - trial_sites: list of clinical sites with locations
        - imp_shelf_life_months: product shelf life

        Returns:
        - recommended distribution strategy
        """

        num_sites = len(trial_sites)
        geographic_spread = self._calculate_geographic_spread(trial_sites)

        # Decision logic
        if num_sites <= 5:
            strategy = 'direct_from_central'
            depots_needed = 0

        elif num_sites <= 30 and geographic_spread < 5000:  # km
            strategy = 'single_regional_depot'
            depots_needed = 1

        else:
            strategy = 'multi_regional_depots'
            depots_needed = int(num_sites / 15)  # ~15 sites per depot

        # Shelf life consideration
        if imp_shelf_life_months < 6:
            # Short shelf life requires more frequent shipments
            recommendation = f"{strategy}_with_weekly_shipments"
        else:
            recommendation = f"{strategy}_with_monthly_shipments"

        return {
            'strategy': recommendation,
            'depots_needed': depots_needed,
            'estimated_inventory_holding': self._estimate_trial_inventory(
                num_sites, strategy
            )
        }

    def _calculate_geographic_spread(self, sites):
        """Calculate geographic spread of sites (simplified)"""
        # Simplified - would use actual geocoding
        return len(sites) * 500  # Placeholder km

    def _estimate_trial_inventory(self, num_sites, strategy):
        """Estimate total inventory in supply chain"""

        if strategy == 'direct_from_central':
            pipeline_weeks = 4
        elif strategy == 'single_regional_depot':
            pipeline_weeks = 2
        else:
            pipeline_weeks = 1.5

        # Weekly demand per site (placeholder)
        weekly_demand_per_site = 50

        total_pipeline = num_sites * weekly_demand_per_site * pipeline_weeks

        return total_pipeline


# Example
trial_protocol = {
    'doses_per_patient': 52,  # Weekly dosing for 1 year
    'treatment_duration_weeks': 52
}

ct_supply = ClinicalTrialsSupplyChain(trial_protocol)

# Randomization
randomization = ct_supply.generate_randomization_schedule(
    num_patients=100,
    treatment_arms=['Drug_A', 'Placebo'],
    randomization_ratio=[1, 1]
)

print(f"Randomized {len(randomization)} patients")
print(randomization.groupby('treatment_arm').size())

---

工具与库

Python 库

供应链优化：

• pulp：分销网络优化
• networkx：供应网络建模
• scipy：稳定性研究的统计分析

数据分析：

• pandas：数据处理
• numpy：数值计算
• matplotlib、seaborn：可视化

序列化：

• python-barcode：条形码生成
• epc-tds：EPCIS 和 EPC 标签数据标准

商业软件

ERP/供应链：

• SAP Pharma：制药供应链套件
• Oracle Agile PLM：生命科学 PLM
• TraceLink：序列化与追踪溯源
• Antares Vision：端到端可追溯性

质量管理：

• Veeva Vault Quality：面向生命科学的云端 QMS
• MasterControl：质量与合规管理
• TrackWise：CAPA 和质量事件
• Sparta Systems：质量管理

冷链：

• Sensitech：温度监控与冷链
• Emerson Cargo Solutions：冷链管理
• ELPRO：温度监控系统
• Tive：实时追踪与监控

临床试验：

• Oracle RTSM：随机化与试验供应管理
• Almac RTSM：临床试验供应
• Marken：临床物流
• World Courier：临床试验运输

---

常见挑战与解决方案

挑战：冷链故障

问题：

• 储存或运输期间发生温度偏差
• 产品完整性受损
• 不符合法规要求
• 产品浪费并危及患者安全

解决方案：

• 包装认证：针对运输线路测试包装
• 持续监控：实时跟踪温度
• 冗余系统：备用制冷和电力系统
• 快速响应：制定温度偏差调查规程
• 供应商认证：审核冷链合作伙伴
• 温度分布测试：验证储存设施
• 季节性测试：针对极端天气进行认证

挑战：序列化实施

问题：

• DSCSA 和 EU FMD 合规要求
• 与旧系统集成
• 实施成本高
• 管理合作伙伴之间的主数据

解决方案：

• 分阶段实施：从高价值产品开始
• 技术选型：选择可扩展的序列化平台
• 合作伙伴协作：与 CMO 和分销商保持一致
• 主数据管理：集中管理 GTIN 和产品数据
• 测试：上线前进行全面的端到端测试
• 培训：培训供应链合作伙伴
• 第三方服务：考虑 TraceLink、rfXcel 平台

挑战：药品短缺管理

问题：

• 生产中断
• 导致停产的法规问题
• 原材料供应受限
• 需求激增（疫情、召回）

解决方案：

• 多地点生产：配置冗余产能
• 安全库存策略：为关键药品储备战略库存
• 供应链可视性：建立预警系统
• 法规沟通：主动通知 FDA/EMA
• 需求管理：优先向关键患者分配
• 替代采购：认证备用 API 供应商
• 库存共享：建立协作式分销网络

挑战：受管制物质管理

问题：

• DEA Schedule II-V 法规
• 盗窃和转移风险
• 文档要求复杂
• 各州规定不一

解决方案：

• 安全设施：围栏、保险库、访问控制
• 持续库存管理：实时跟踪 CS 库存
• 双人控制：交易需两人核验
• 背景调查：筛查员工
• 审计跟踪：完整记录文档
• 法规报告：DEA 222 表格、ARCOS 报告
• 防损措施：安保措施和监控

挑战：加快召回执行

问题：

• 需要在 24-48 小时内召回产品
• 定位已分销的产品
• 与所有相关方沟通
• 确保产品全部追回

解决方案：

• 利用序列化数据：使用追踪数据
• 召回程序：每年进行测试
• 沟通规程：预先建立联系人名单
• 分销记录：以电子方式保存
• 模拟召回：每季度进行演练
• 批次谱系：完整的可追溯记录
• 第三方协调：与批发商保持一致

---

输出格式

制药供应链报告

执行摘要：

• 产品组合概览（生物制品、小分子药物等）
• 法规合规状态
• 质量指标和偏差
• 供应链绩效

冷链绩效：

产品	温度范围	运输批次	温度偏差	温度偏差率	产品损失
Vaccine_A	2-8°C	1,250	8	0.64%	$12,400
Biologic_B	2-8°C	850	3	0.35%	$45,000
Insulin_C	2-8°C	3,200	15	0.47%	$8,200
总计	-	5,300	26	0.49%	$65,600

质量指标：

指标	当前值	目标值	状态
GDP 合规率	98%	100%	⚠ Yellow
序列化覆盖率	95%	100%	⚠ Yellow
偏差关闭率（30 天）	87%	95%	⚠ Yellow
供应商审计合规率	100%	100%	✓ Green
OTIF 交付率	96%	98%	⚠ Yellow

偏差摘要：

严重程度	数量	未关闭	逾期	需制定 CAPA
严重	2	1	0	2
重大	15	4	1	8
轻微	42	12	3	0
总计	59	17	4	10

临床试验状态：

试验 ID	阶段	研究中心	患者人数	IMP 库存（周）	问题
TRIAL_001	III	45	350	8	无
TRIAL_002	II	12	80	4	2 个研究中心库存不足
TRIAL_003	I	3	24	12	无

行动事项：

1. 完成 2 项严重偏差的 CAPA——截止日期为 2 月 15 日
2. 在 DC3 安装备用制冷设备——防止未来发生温度偏差
3. 为剩余 5% 的产品完成序列化部署——在 Q2 前完成
4. 解决试验 002 研究中心的 IMP 库存短缺问题——48 小时内发货
5. 更新适用于欧盟新法规的 GDP 操作规程——于 3 月 1 日前合规

---

可提出的问题

如果需要更多背景信息：

1. 产品属于哪些类型？（小分子药物、生物制品、疫苗）
2. 有哪些温度要求？（2-8°C、冷冻、超低温、常温）
3. 涉及哪些法规市场？（美国、欧盟、其他地区）
4. 当前的序列化状态如何？
5. 是否有需要支持的临床试验？
6. 主要的质量/合规挑战是什么？
7. 分销网络结构如何？
8. 是否涉及受管制药物？

---

相关技能

• clinical-trial-logistics：用于试验用药品管理
• cold-chain：用于温控物流
• quality-management：用于 QMS 和 GxP 合规
• track-and-trace：用于实施序列化
• inventory-optimization：用于安全库存和库存策略
• network-design：用于优化分销网络
• risk-mitigation：用于供应链风险管理
• compliance-management：用于法规合规
• medical-device-distribution：用于医疗器械特定要求
• hospital-logistics：用于医院药房供应链