medical-device-distribution

医療機器の流通

医療機器の流通とサプライチェーン管理の専門家です。トレーサビリティを維持し、規制要件を管理し、在庫コストを最適化しながら、医療機器を法令に準拠して効率的に流通させることが目標です。

初期評価

医療機器の流通を最適化する前に、次の点を把握してください。

1. 機器のカテゴリー

   • 機器の分類は何ですか？（クラス I、II、III）
   • 製品の種類は何ですか？（インプラント、設備機器、ディスポーザブル製品、貸出セット）
   • 高価値製品と大量流通製品のどちらですか？
   • 滅菌要件と非滅菌要件は何ですか？

2. 規制環境

   • FDA 登録の状況はどうなっていますか？
   • UDI への準拠要件は何ですか？
   • QMS（品質マネジメントシステム）は整備されていますか？
   • 海外にも流通していますか？（CE マーク、その他の国）

3. 流通ネットワーク

   • 直接流通ですか、それとも販売代理店経由ですか？
   • 流通センターはいくつありますか？
   • 顧客の種類は何ですか？（病院、クリニック、手術センター）
   • 地理的な対象範囲はどこですか？

4. 現在の課題

   • トレーサビリティに不備はありますか？
   • リコールへの備えはできていますか？
   • 委託在庫を管理できていますか？
   • 使用期限切れや陳腐化の問題はありますか？
   • コンプライアンス違反や警告書はありますか？

---

医療機器流通のフレームワーク

機器の分類と要件

FDA の機器クラス：

クラス I（低リスク）：

• 例：包帯、検査用手袋、ハンドヘルド器具
• 要件：一般規制。大半は 510(k) の対象外です
• 流通：標準的なサプライチェーンで、特別な取り扱いは最小限です

クラス II（中リスク）：

• 例：電動車椅子、輸液ポンプ、手術用ドレープ
• 要件：一般規制と特別規制。通常、510(k) の認可が必要です
• 流通：トレーサビリティを強化し、多くの場合 UDI が必要です

クラス III（高リスク）：

• 例：ペースメーカー、心臓弁、植込み型除細動器
• 要件：市販前承認（PMA）と最も厳格な規制
• 流通：完全なトレーサビリティ、シリアル化、一般的な委託販売

機器固有識別子（UDI）への準拠

UDI の要件：

• 機器のラベルと包装に固有の識別子を記載する
• 植込み型機器に直接マーキングする
• FDA の GUDID（Global Unique Device Identification Database）に登録する
• 流通チェーン全体を通じて追跡する

UDI の構造：

UDI = Device Identifier (DI) + Production Identifier (PI)

DI: Identifies the specific device (model, version)
PI: Identifies the production unit (lot, serial number, expiry, manufacture date)

導入：

import re
from dataclasses import dataclass
from datetime import datetime
from typing import Optional

@dataclass
class UDI:
    """
    Unique Device Identification structure
    """
    device_identifier: str  # DI - identifies device model
    lot_number: Optional[str] = None
    serial_number: Optional[str] = None
    manufacturing_date: Optional[datetime] = None
    expiration_date: Optional[datetime] = None
    donation_id: Optional[str] = None  # For blood/tissue

    def to_gs1_string(self):
        """Convert to GS1 format"""
        udi_string = f"(01){self.device_identifier}"

        if self.lot_number:
            udi_string += f"(10){self.lot_number}"

        if self.serial_number:
            udi_string += f"(21){self.serial_number}"

        if self.expiration_date:
            exp_date = self.expiration_date.strftime("%y%m%d")
            udi_string += f"(17){exp_date}"

        if self.manufacturing_date:
            mfg_date = self.manufacturing_date.strftime("%y%m%d")
            udi_string += f"(11){mfg_date}"

        return udi_string

    @staticmethod
    def parse_gs1(gs1_string):
        """Parse GS1 UDI string"""
        # Application Identifiers (AI)
        patterns = {
            'device_identifier': r'\(01\)(\d{14})',
            'lot_number': r'\(10\)([A-Za-z0-9]+)',
            'serial_number': r'\(21\)([A-Za-z0-9]+)',
            'expiration_date': r'\(17\)(\d{6})',
            'manufacturing_date': r'\(11\)(\d{6})'
        }

        udi_data = {}

        for field, pattern in patterns.items():
            match = re.search(pattern, gs1_string)
            if match:
                value = match.group(1)
                if 'date' in field:
                    # Convert YYMMDD to datetime
                    udi_data[field] = datetime.strptime(value, "%y%m%d")
                else:
                    udi_data[field] = value

        return UDI(**udi_data)

# Example usage
device_udi = UDI(
    device_identifier="10884521123456",
    lot_number="LOT2024A",
    serial_number="SN123456789",
    expiration_date=datetime(2027, 12, 31)
)

gs1_string = device_udi.to_gs1_string()
print(f"GS1 UDI: {gs1_string}")

# Parse back
parsed_udi = UDI.parse_gs1(gs1_string)
print(f"Parsed Serial: {parsed_udi.serial_number}")
print(f"Parsed Expiry: {parsed_udi.expiration_date}")

---

トレーサビリティとシリアル化

エンドツーエンドのトレーサビリティシステム

import pandas as pd
from datetime import datetime
from enum import Enum

class EventType(Enum):
    MANUFACTURED = "manufactured"
    SHIPPED = "shipped"
    RECEIVED = "received"
    IMPLANTED = "implanted"
    RETURNED = "returned"
    RECALLED = "recalled"

class TraceabilitySystem:
    """
    Track medical devices through supply chain
    """

    def __init__(self):
        self.events = []

    def record_event(self, event_type, device_id, serial_number,
                     location, operator, timestamp=None, metadata=None):
        """
        Record a traceability event

        Parameters:
        - event_type: Type of event (EventType enum)
        - device_id: Device identifier (DI)
        - serial_number: Unique serial number
        - location: Where event occurred
        - operator: Who performed action
        - timestamp: When it occurred (defaults to now)
        - metadata: Additional data (patient ID, lot, etc.)
        """

        if timestamp is None:
            timestamp = datetime.now()

        event = {
            'event_type': event_type.value,
            'device_id': device_id,
            'serial_number': serial_number,
            'location': location,
            'operator': operator,
            'timestamp': timestamp,
            'metadata': metadata or {}
        }

        self.events.append(event)

        return event

    def trace_device_history(self, serial_number):
        """
        Get complete history for a device by serial number
        """

        device_events = [
            e for e in self.events
            if e['serial_number'] == serial_number
        ]

        # Sort by timestamp
        device_events.sort(key=lambda x: x['timestamp'])

        return pd.DataFrame(device_events)

    def find_devices_by_lot(self, lot_number):
        """
        Find all devices from a specific lot (for recalls)
        """

        devices = [
            e for e in self.events
            if e.get('metadata', {}).get('lot_number') == lot_number
        ]

        # Get unique serial numbers
        serial_numbers = list(set(e['serial_number'] for e in devices))

        return serial_numbers

    def implanted_devices_report(self, start_date, end_date):
        """
        Report of devices implanted in date range
        """

        implanted = [
            e for e in self.events
            if e['event_type'] == EventType.IMPLANTED.value
            and start_date <= e['timestamp'] <= end_date
        ]

        return pd.DataFrame(implanted)

    def audit_trail(self, device_id=None, location=None, date_range=None):
        """
        Generate audit trail for compliance
        """

        filtered_events = self.events

        if device_id:
            filtered_events = [e for e in filtered_events if e['device_id'] == device_id]

        if location:
            filtered_events = [e for e in filtered_events if e['location'] == location]

        if date_range:
            start, end = date_range
            filtered_events = [
                e for e in filtered_events
                if start <= e['timestamp'] <= end
            ]

        return pd.DataFrame(filtered_events)

# Example usage
traceability = TraceabilitySystem()

# Manufacturing event
traceability.record_event(
    EventType.MANUFACTURED,
    device_id="10884521123456",
    serial_number="SN123456789",
    location="Manufacturing Plant A",
    operator="Production Line 3",
    metadata={'lot_number': 'LOT2024A', 'manufacturing_date': '2024-01-15'}
)

# Shipment to distributor
traceability.record_event(
    EventType.SHIPPED,
    device_id="10884521123456",
    serial_number="SN123456789",
    location="Distribution Center East",
    operator="Warehouse Operator J.Smith",
    metadata={'tracking_number': 'TRACK123', 'carrier': 'FedEx'}
)

# Receipt at hospital
traceability.record_event(
    EventType.RECEIVED,
    device_id="10884521123456",
    serial_number="SN123456789",
    location="Memorial Hospital",
    operator="Materials Manager",
    metadata={'purchase_order': 'PO-2024-5678'}
)

# Implantation
traceability.record_event(
    EventType.IMPLANTED,
    device_id="10884521123456",
    serial_number="SN123456789",
    location="Memorial Hospital OR-3",
    operator="Dr. Johnson",
    metadata={
        'patient_id': 'PT-987654',  # PHI - must be secured
        'procedure_code': 'CPT-33206',
        'surgeon': 'Dr. Johnson'
    }
)

# Get device history
history = traceability.trace_device_history("SN123456789")
print("Device History:")
print(history[['timestamp', 'event_type', 'location', 'operator']])

---

委託在庫管理

委託在庫の最適化モデル

委託在庫の特徴：

• 使用されるまでベンダーが在庫を所有する
• 顧客の施設（病院）に置かれる
• 病院は消費した分のみ支払う
• ベンダーが補充を担当する
• 高価値のインプラントで一般的

import numpy as np
import pandas as pd

class ConsignmentInventoryManager:
    """
    Manage consignment inventory for medical devices
    """

    def __init__(self, location_name, par_level_policy=None):
        self.location_name = location_name
        self.inventory = {}  # {serial_number: device_info}
        self.par_levels = par_level_policy or {}
        self.usage_history = []

    def add_device(self, device_id, serial_number, cost, vendor, metadata=None):
        """Add device to consignment inventory"""

        self.inventory[serial_number] = {
            'device_id': device_id,
            'serial_number': serial_number,
            'cost': cost,
            'vendor': vendor,
            'status': 'available',
            'received_date': datetime.now(),
            'metadata': metadata or {}
        }

    def consume_device(self, serial_number, patient_id, procedure_info):
        """
        Record device consumption (trigger payment to vendor)
        """

        if serial_number not in self.inventory:
            raise ValueError(f"Device {serial_number} not in consignment inventory")

        device = self.inventory[serial_number]

        if device['status'] != 'available':
            raise ValueError(f"Device {serial_number} is not available (status: {device['status']})")

        # Record usage
        usage_record = {
            'serial_number': serial_number,
            'device_id': device['device_id'],
            'vendor': device['vendor'],
            'cost': device['cost'],
            'consumption_date': datetime.now(),
            'patient_id': patient_id,
            'procedure_info': procedure_info,
            'days_in_consignment': (datetime.now() - device['received_date']).days
        }

        self.usage_history.append(usage_record)

        # Update device status
        device['status'] = 'consumed'
        device['consumption_date'] = datetime.now()

        return usage_record

    def check_par_levels(self):
        """
        Check if consignment inventory meets PAR levels
        Trigger vendor replenishment if needed
        """

        replenishment_needed = []

        for device_id, par_level in self.par_levels.items():
            # Count available devices
            available_count = sum(
                1 for device in self.inventory.values()
                if device['device_id'] == device_id and device['status'] == 'available'
            )

            if available_count < par_level:
                shortage = par_level - available_count

                replenishment_needed.append({
                    'device_id': device_id,
                    'current_qty': available_count,
                    'par_level': par_level,
                    'replenishment_qty': shortage
                })

        return replenishment_needed

    def vendor_reconciliation_report(self, vendor, month):
        """
        Generate monthly reconciliation report for vendor billing
        """

        usage_df = pd.DataFrame(self.usage_history)

        if len(usage_df) == 0:
            return pd.DataFrame()

        # Filter by vendor and month
        vendor_usage = usage_df[
            (usage_df['vendor'] == vendor) &
            (usage_df['consumption_date'].dt.to_period('M') == month)
        ]

        # Aggregate
        summary = vendor_usage.groupby('device_id').agg({
            'serial_number': 'count',
            'cost': 'sum'
        }).rename(columns={'serial_number': 'quantity', 'cost': 'total_cost'})

        summary['average_cost'] = summary['total_cost'] / summary['quantity']

        return summary

    def aging_report(self):
        """
        Report devices sitting in consignment for extended periods
        (potential obsolescence risk)
        """

        aging_data = []

        for serial, device in self.inventory.items():
            if device['status'] == 'available':
                days_on_hand = (datetime.now() - device['received_date']).days

                aging_data.append({
                    'serial_number': serial,
                    'device_id': device['device_id'],
                    'vendor': device['vendor'],
                    'cost': device['cost'],
                    'days_on_hand': days_on_hand,
                    'risk_level': self._aging_risk(days_on_hand)
                })

        aging_df = pd.DataFrame(aging_data)

        if len(aging_df) > 0:
            aging_df = aging_df.sort_values('days_on_hand', ascending=False)

        return aging_df

    @staticmethod
    def _aging_risk(days):
        """Categorize aging risk"""
        if days > 365:
            return 'Critical'
        elif days > 180:
            return 'High'
        elif days > 90:
            return 'Medium'
        else:
            return 'Low'

# Example usage
consignment = ConsignmentInventoryManager(
    location_name="Memorial Hospital",
    par_level_policy={
        'Hip-Implant-A': 5,
        'Knee-Implant-B': 8,
        'Cardiac-Stent-C': 10
    }
)

# Add devices to consignment
for i in range(5):
    consignment.add_device(
        device_id='Hip-Implant-A',
        serial_number=f'HIP-SN-{1000+i}',
        cost=3500,
        vendor='Orthopedic Devices Inc',
        metadata={'lot': 'LOT-2024-A'}
    )

# Consume a device
usage = consignment.consume_device(
    serial_number='HIP-SN-1000',
    patient_id='PT-123456',
    procedure_info={'procedure': 'Total Hip Arthroplasty', 'surgeon': 'Dr. Smith'}
)

print(f"Device consumed: {usage['serial_number']}")
print(f"Cost: ${usage['cost']:,.2f}")

# Check PAR levels
replenishment = consignment.check_par_levels()
if replenishment:
    print("\nReplenishment needed:")
    for item in replenishment:
        print(f"  {item['device_id']}: Need {item['replenishment_qty']} units")

---

貸出セット管理

手術用貸出セットの追跡

貸出セット：

• 特殊な器具をまとめたトレイ
• 特定の手術のためにメーカーから貸し出される
• 使用後に返却する必要がある
• 高価（セットあたり 50K～500K ドル）
• 紛失請求を避けるため、追跡が重要です

from datetime import datetime, timedelta

class LoanerSetManager:
    """
    Track loaner instrument sets for surgeries
    """

    def __init__(self):
        self.loaner_sets = {}
        self.bookings = []
        self.movements = []

    def register_loaner_set(self, set_id, vendor, description, value, contents):
        """
        Register a loaner set in the system
        """

        self.loaner_sets[set_id] = {
            'set_id': set_id,
            'vendor': vendor,
            'description': description,
            'value': value,
            'contents': contents,
            'status': 'available',
            'location': 'Vendor',
            'registered_date': datetime.now()
        }

    def book_loaner_set(self, set_id, procedure_date, surgeon, procedure_type, patient_id):
        """
        Book loaner set for upcoming surgery
        """

        if set_id not in self.loaner_sets:
            raise ValueError(f"Loaner set {set_id} not registered")

        booking = {
            'booking_id': f"BOOK-{len(self.bookings)+1}",
            'set_id': set_id,
            'procedure_date': procedure_date,
            'surgeon': surgeon,
            'procedure_type': procedure_type,
            'patient_id': patient_id,
            'booking_date': datetime.now(),
            'status': 'booked'
        }

        self.bookings.append(booking)

        # Update set status
        self.loaner_sets[set_id]['status'] = 'booked'

        return booking

    def receive_loaner_set(self, set_id, received_by, condition='good'):
        """
        Record receipt of loaner set at hospital
        """

        if set_id not in self.loaner_sets:
            raise ValueError(f"Loaner set {set_id} not registered")

        movement = {
            'set_id': set_id,
            'event_type': 'received',
            'timestamp': datetime.now(),
            'location': 'Hospital Central Sterile',
            'operator': received_by,
            'condition': condition
        }

        self.movements.append(movement)

        # Update set status and location
        self.loaner_sets[set_id]['status'] = 'in_house'
        self.loaner_sets[set_id]['location'] = 'Hospital Central Sterile'
        self.loaner_sets[set_id]['received_date'] = datetime.now()

    def send_to_sterile_processing(self, set_id, processor):
        """
        Send loaner set for sterilization
        """

        movement = {
            'set_id': set_id,
            'event_type': 'sent_to_sterilization',
            'timestamp': datetime.now(),
            'location': 'Sterile Processing',
            'operator': processor
        }

        self.movements.append(movement)

        self.loaner_sets[set_id]['location'] = 'Sterile Processing'
        self.loaner_sets[set_id]['status'] = 'processing'

    def ready_for_surgery(self, set_id, sterilization_lot):
        """
        Mark loaner set as sterile and ready
        """

        movement = {
            'set_id': set_id,
            'event_type': 'sterilization_complete',
            'timestamp': datetime.now(),
            'location': 'Sterile Storage',
            'sterilization_lot': sterilization_lot
        }

        self.movements.append(movement)

        self.loaner_sets[set_id]['location'] = 'Sterile Storage'
        self.loaner_sets[set_id]['status'] = 'ready'

    def use_in_surgery(self, set_id, procedure_info):
        """
        Record use in surgery
        """

        movement = {
            'set_id': set_id,
            'event_type': 'used_in_surgery',
            'timestamp': datetime.now(),
            'location': f"OR-{procedure_info.get('or_room', 'Unknown')}",
            'procedure_info': procedure_info
        }

        self.movements.append(movement)

        self.loaner_sets[set_id]['status'] = 'used'
        self.loaner_sets[set_id]['last_used_date'] = datetime.now()

    def return_to_vendor(self, set_id, carrier, tracking_number):
        """
        Ship loaner set back to vendor
        """

        movement = {
            'set_id': set_id,
            'event_type': 'returned_to_vendor',
            'timestamp': datetime.now(),
            'location': 'In Transit to Vendor',
            'carrier': carrier,
            'tracking_number': tracking_number
        }

        self.movements.append(movement)

        self.loaner_sets[set_id]['status'] = 'returned'
        self.loaner_sets[set_id]['location'] = 'In Transit to Vendor'
        self.loaner_sets[set_id]['return_date'] = datetime.now()

    def overdue_loaners_report(self, days_threshold=14):
        """
        Identify loaner sets held beyond expected return date
        """

        overdue = []

        for set_id, loaner in self.loaner_sets.items():
            if loaner['status'] in ['available', 'returned']:
                continue

            if 'received_date' in loaner:
                days_on_hand = (datetime.now() - loaner['received_date']).days

                if days_on_hand > days_threshold:
                    overdue.append({
                        'set_id': set_id,
                        'vendor': loaner['vendor'],
                        'description': loaner['description'],
                        'value': loaner['value'],
                        'days_on_hand': days_on_hand,
                        'status': loaner['status'],
                        'location': loaner['location']
                    })

        return pd.DataFrame(overdue)

    def loaner_utilization_report(self):
        """
        Analyze loaner set utilization and costs
        """

        utilization_data = []

        for set_id, loaner in self.loaner_sets.items():
            # Count uses
            uses = [m for m in self.movements if m['set_id'] == set_id and m['event_type'] == 'used_in_surgery']

            # Calculate average time on-site
            received_events = [m for m in self.movements if m['set_id'] == set_id and m['event_type'] == 'received']
            returned_events = [m for m in self.movements if m['set_id'] == set_id and m['event_type'] == 'returned_to_vendor']

            cycles = min(len(received_events), len(returned_events))
            if cycles > 0:
                avg_days_on_site = sum([
                    (returned_events[i]['timestamp'] - received_events[i]['timestamp']).days
                    for i in range(cycles)
                ]) / cycles
            else:
                avg_days_on_site = 0

            utilization_data.append({
                'set_id': set_id,
                'vendor': loaner['vendor'],
                'description': loaner['description'],
                'value': loaner['value'],
                'times_used': len(uses),
                'avg_days_on_site': round(avg_days_on_site, 1),
                'cycles': cycles
            })

        return pd.DataFrame(utilization_data)

# Example usage
loaner_mgr = LoanerSetManager()

# Register loaner sets
loaner_mgr.register_loaner_set(
    set_id='LOANER-SPINE-001',
    vendor='Spinal Devices Corp',
    description='Lumbar Fusion Instrument Set',
    value=125000,
    contents=['Pedicle Screws', 'Rods', 'Insertion Tools', 'Measurement Guides']
)

# Book for surgery
booking = loaner_mgr.book_loaner_set(
    set_id='LOANER-SPINE-001',
    procedure_date=datetime.now() + timedelta(days=7),
    surgeon='Dr. Williams',
    procedure_type='Lumbar Fusion L4-L5',
    patient_id='PT-789012'
)

# Track through process
loaner_mgr.receive_loaner_set('LOANER-SPINE-001', received_by='Materials Clerk')
loaner_mgr.send_to_sterile_processing('LOANER-SPINE-001', processor='SPD Tech A')
loaner_mgr.ready_for_surgery('LOANER-SPINE-001', sterilization_lot='STER-20240215-001')
loaner_mgr.use_in_surgery('LOANER-SPINE-001', {'or_room': '5', 'surgeon': 'Dr. Williams'})
loaner_mgr.return_to_vendor('LOANER-SPINE-001', carrier='FedEx', tracking_number='TRACK123456')

print("Loaner set lifecycle completed")

---

リコール管理

医療機器のリコールプロセス

FDA のリコール分類：

• クラス I：深刻な健康被害または死亡に至る合理的な可能性がある
• クラス II：一時的、または医学的に回復可能な健康被害がある
• クラス III：健康被害を引き起こす可能性は低い

class RecallManager:
    """
    Manage medical device recalls
    """

    def __init__(self, traceability_system):
        self.traceability_system = traceability_system
        self.recalls = []

    def initiate_recall(self, recall_id, device_id, lot_numbers, reason,
                        recall_class, recall_date, corrective_action):
        """
        Initiate a device recall
        """

        recall = {
            'recall_id': recall_id,
            'device_id': device_id,
            'lot_numbers': lot_numbers,
            'reason': reason,
            'recall_class': recall_class,  # I, II, or III
            'recall_date': recall_date,
            'corrective_action': corrective_action,
            'status': 'active'
        }

        self.recalls.append(recall)

        return recall

    def identify_affected_devices(self, recall_id):
        """
        Identify all devices affected by recall using traceability
        """

        recall = next((r for r in self.recalls if r['recall_id'] == recall_id), None)

        if not recall:
            raise ValueError(f"Recall {recall_id} not found")

        affected_devices = []

        for lot_number in recall['lot_numbers']:
            # Find devices from affected lots
            serial_numbers = self.traceability_system.find_devices_by_lot(lot_number)

            for serial in serial_numbers:
                device_history = self.traceability_system.trace_device_history(serial)

                # Get current location/status
                latest_event = device_history.iloc[-1] if len(device_history) > 0 else None

                affected_devices.append({
                    'serial_number': serial,
                    'lot_number': lot_number,
                    'current_location': latest_event['location'] if latest_event is not None else 'Unknown',
                    'status': latest_event['event_type'] if latest_event is not None else 'Unknown',
                    'last_event_date': latest_event['timestamp'] if latest_event is not None else None
                })

        return pd.DataFrame(affected_devices)

    def identify_implanted_devices(self, recall_id):
        """
        Identify devices from recall that have been implanted (most critical)
        """

        affected = self.identify_affected_devices(recall_id)

        # Filter to implanted devices
        implanted = affected[affected['status'] == 'implanted']

        return implanted

    def customer_notification_list(self, recall_id):
        """
        Generate list of customers to notify about recall
        """

        affected = self.identify_affected_devices(recall_id)

        # Group by current location (customer)
        customers = affected.groupby('current_location').agg({
            'serial_number': 'count',
            'status': lambda x: ', '.join(x.unique())
        }).rename(columns={'serial_number': 'affected_devices'})

        return customers

    def recall_effectiveness_check(self, recall_id):
        """
        Track recall effectiveness (how many devices recovered)
        """

        affected = self.identify_affected_devices(recall_id)
        total_affected = len(affected)

        # Devices that have been returned/recovered
        recovered = affected[affected['status'] == 'returned']
        recovered_count = len(recovered)

        # Devices still implanted
        implanted = affected[affected['status'] == 'implanted']
        implanted_count = len(implanted)

        # Devices unaccounted for
        unaccounted = affected[~affected['status'].isin(['returned', 'implanted'])]
        unaccounted_count = len(unaccounted)

        effectiveness = {
            'recall_id': recall_id,
            'total_affected': total_affected,
            'recovered': recovered_count,
            'recovery_rate': (recovered_count / total_affected * 100) if total_affected > 0 else 0,
            'still_implanted': implanted_count,
            'unaccounted': unaccounted_count
        }

        return effectiveness

# Example usage
# Assuming traceability system from earlier example
recall_mgr = RecallManager(traceability)

# Initiate recall
recall = recall_mgr.initiate_recall(
    recall_id='RECALL-2024-001',
    device_id='10884521123456',
    lot_numbers=['LOT2024A', 'LOT2024B'],
    reason='Potential battery malfunction',
    recall_class='I',  # Class I - serious
    recall_date=datetime.now(),
    corrective_action='Return device to manufacturer for inspection and replacement'
)

print(f"Recall initiated: {recall['recall_id']}")
print(f"Class: {recall['recall_class']}")
print(f"Affected lots: {', '.join(recall['lot_numbers'])}")

# Identify affected devices
affected = recall_mgr.identify_affected_devices('RECALL-2024-001')
print(f"\nTotal affected devices: {len(affected)}")

# Identify implanted (most critical)
implanted = recall_mgr.identify_implanted_devices('RECALL-2024-001')
print(f"Implanted devices requiring patient notification: {len(implanted)}")

# Customer notification list
customers = recall_mgr.customer_notification_list('RECALL-2024-001')
print("\nCustomers to notify:")
print(customers)

---

コールドチェーンと温度管理

温度管理された流通

要件：

• 特定の生物製剤および組織由来の機器
• 流通全体を通じて温度範囲を維持する
• 継続的に監視する
• 温度逸脱を記録する

import numpy as np

class ColdChainMonitor:
    """
    Monitor temperature-controlled shipments
    """

    def __init__(self, shipment_id, product_id, temp_range):
        self.shipment_id = shipment_id
        self.product_id = product_id
        self.min_temp, self.max_temp = temp_range
        self.temperature_log = []
        self.excursions = []

    def record_temperature(self, timestamp, temperature, location, recorder):
        """
        Record temperature reading
        """

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

        self.temperature_log.append(reading)

        # Check for excursion
        if not reading['in_range']:
            excursion = {
                'excursion_start': timestamp,
                'temperature': temperature,
                'location': location,
                'severity': self._calculate_severity(temperature)
            }
            self.excursions.append(excursion)

        return reading

    def _calculate_severity(self, temperature):
        """
        Determine excursion severity
        """

        if temperature < self.min_temp:
            deviation = self.min_temp - temperature
        else:
            deviation = temperature - self.max_temp

        if deviation <= 2:
            return 'Minor'
        elif deviation <= 5:
            return 'Moderate'
        else:
            return 'Major'

    def compliance_report(self):
        """
        Generate temperature compliance report
        """

        if not self.temperature_log:
            return None

        df = pd.DataFrame(self.temperature_log)

        total_readings = len(df)
        compliant_readings = df['in_range'].sum()
        compliance_rate = (compliant_readings / total_readings * 100) if total_readings > 0 else 0

        report = {
            'shipment_id': self.shipment_id,
            'product_id': self.product_id,
            'required_range': f"{self.min_temp}°F - {self.max_temp}°F",
            'total_readings': total_readings,
            'compliant_readings': compliant_readings,
            'compliance_rate': round(compliance_rate, 2),
            'num_excursions': len(self.excursions),
            'min_temp_recorded': df['temperature'].min(),
            'max_temp_recorded': df['temperature'].max(),
            'avg_temp': round(df['temperature'].mean(), 2)
        }

        return report

    def excursion_summary(self):
        """
        Summarize temperature excursions
        """

        if not self.excursions:
            return pd.DataFrame()

        return pd.DataFrame(self.excursions)

# Example usage
cold_chain = ColdChainMonitor(
    shipment_id='SHIP-2024-001',
    product_id='Tissue-Graft-A',
    temp_range=(36, 46)  # 36-46°F required
)

# Simulate temperature readings
np.random.seed(42)
base_temp = 40
for hour in range(48):
    temp = base_temp + np.random.normal(0, 2)

    # Simulate excursion at hour 24
    if hour == 24:
        temp = 50  # Excursion

    cold_chain.record_temperature(
        timestamp=datetime.now() + timedelta(hours=hour),
        temperature=round(temp, 1),
        location='In Transit',
        recorder='Data Logger SN12345'
    )

# Generate reports
compliance = cold_chain.compliance_report()
print("Cold Chain Compliance:")
print(f"  Compliance Rate: {compliance['compliance_rate']}%")
print(f"  Excursions: {compliance['num_excursions']}")
print(f"  Temp Range: {compliance['min_temp_recorded']}°F - {compliance['max_temp_recorded']}°F")

if compliance['num_excursions'] > 0:
    excursions = cold_chain.excursion_summary()
    print("\nTemperature Excursions:")
    print(excursions)

---

流通パフォーマンス指標

主要業績評価指標（KPI）

def calculate_distribution_kpis(shipment_data_df, inventory_data_df, recall_data_df=None):
    """
    Calculate medical device distribution KPIs

    Parameters:
    - shipment_data_df: Shipment transactions
    - inventory_data_df: Inventory positions
    - recall_data_df: Recall events (optional)
    """

    kpis = {}

    # On-Time Delivery Rate
    if 'on_time' in shipment_data_df.columns:
        kpis['otd_rate'] = (shipment_data_df['on_time'].sum() / len(shipment_data_df) * 100)

    # Perfect Order Rate (on-time, complete, damage-free, correct documentation)
    if all(col in shipment_data_df.columns for col in ['on_time', 'complete', 'damage_free', 'docs_correct']):
        perfect_orders = shipment_data_df[
            shipment_data_df['on_time'] &
            shipment_data_df['complete'] &
            shipment_data_df['damage_free'] &
            shipment_data_df['docs_correct']
        ]
        kpis['perfect_order_rate'] = (len(perfect_orders) / len(shipment_data_df) * 100)

    # Inventory Accuracy
    if 'physical_count' in inventory_data_df.columns and 'system_count' in inventory_data_df.columns:
        accurate_items = inventory_data_df[
            inventory_data_df['physical_count'] == inventory_data_df['system_count']
        ]
        kpis['inventory_accuracy'] = (len(accurate_items) / len(inventory_data_df) * 100)

    # UDI Compliance Rate
    if 'udi_compliant' in inventory_data_df.columns:
        kpis['udi_compliance_rate'] = (inventory_data_df['udi_compliant'].sum() / len(inventory_data_df) * 100)

    # Average Recall Response Time
    if recall_data_df is not None and not recall_data_df.empty:
        if 'response_hours' in recall_data_df.columns:
            kpis['avg_recall_response_hours'] = recall_data_df['response_hours'].mean()

    # Traceability Completeness
    if 'traceability_complete' in shipment_data_df.columns:
        kpis['traceability_completeness'] = (
            shipment_data_df['traceability_complete'].sum() / len(shipment_data_df) * 100
        )

    # Cold Chain Compliance (if applicable)
    if 'temp_compliant' in shipment_data_df.columns:
        cold_chain_shipments = shipment_data_df[shipment_data_df['requires_cold_chain'] == True]
        if len(cold_chain_shipments) > 0:
            kpis['cold_chain_compliance'] = (
                cold_chain_shipments['temp_compliant'].sum() / len(cold_chain_shipments) * 100
            )

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

    return kpis

# Example usage
shipment_data = pd.DataFrame({
    'shipment_id': range(1, 101),
    'on_time': np.random.choice([True, False], 100, p=[0.95, 0.05]),
    'complete': np.random.choice([True, False], 100, p=[0.98, 0.02]),
    'damage_free': np.random.choice([True, False], 100, p=[0.99, 0.01]),
    'docs_correct': np.random.choice([True, False], 100, p=[0.97, 0.03]),
    'traceability_complete': np.random.choice([True, False], 100, p=[0.99, 0.01]),
    'requires_cold_chain': np.random.choice([True, False], 100, p=[0.15, 0.85]),
    'temp_compliant': np.random.choice([True, False], 100, p=[0.98, 0.02])
})

inventory_data = pd.DataFrame({
    'item_id': range(1, 501),
    'physical_count': np.random.randint(0, 100, 500),
    'system_count': np.random.randint(0, 100, 500),
    'udi_compliant': np.random.choice([True, False], 500, p=[0.95, 0.05])
})

# Make some matching for accuracy
inventory_data.loc[0:450, 'physical_count'] = inventory_data.loc[0:450, 'system_count']

kpis = calculate_distribution_kpis(shipment_data, inventory_data)

print("Medical Device Distribution KPIs:")
for metric, value in kpis.items():
    print(f"  {metric}: {value}{'%' if 'rate' in metric or 'compliance' in metric else ''}")

---

ツールとライブラリ

医療機器ソフトウェアシステム

ERP/QMS システム：

• TrackWise：品質管理とコンプライアンス
• MasterControl：文書管理と品質管理
• Veeva Vault：規制対象コンテンツの管理
• SAP for Medical Devices：医療機器に特化した機能を備えた ERP
• Oracle Agile PLM：製品ライフサイクル管理

トレーサビリティとシリアライゼーション：

• TraceLink：ネットワークベースの追跡とトレース
• Systech UniSecure：シリアライゼーションとトレーサビリティ
• Optel Vision：エンドツーエンドのトレーサビリティ
• Antares Vision：追跡とトレースのソリューション
• rfxcel：サプライチェーンのトレーサビリティプラットフォーム

委託在庫と貸出品の管理：

• iMDsoft：インプラントと委託在庫の管理
• Attainia：医療サプライチェーン管理
• SmartTray：貸出資産の追跡
• Surgical Information Systems (SIS)：手術室資産の追跡

コールドチェーンのモニタリング：

• Sensitech：温度モニタリングソリューション
• Emerson Cargo Solutions：コールドチェーンの可視化
• Tive：センサーによるリアルタイム追跡
• Controlant：温度モニタリングプラットフォーム

Python ライブラリ

シリアライゼーションとバーコード：

• python-barcode：バーコードの生成
• qrcode：QR コードの生成
• pylibdmtx：Data Matrix バーコードの読み取り
• pyzbar：バーコード／QR コードの読み取り

データ分析：

• pandas：データ操作
• numpy：数値演算
• scipy：統計分析
• matplotlib, seaborn：可視化

データベースと追跡：

• sqlalchemy：データベース ORM
• pymongo：MongoDB インターフェース
• redis-py：キャッシュ用 Redis
• flask/fastapi：API 開発

---

よくある課題と解決策

課題：UDI コンプライアンスの実装

問題：

• 複雑なラベリング要件
• 複数の規格（GS1、HIBCC、ICCBBA）
• 既存システムとの統合
• 履歴データの欠落

解決策：

• 機器クラスごとに段階的に実装する
• GS1 規格（最も一般的）を導入する
• ラベル生成システムを自動化する
• 可能な場合は履歴データを補完する
• ラベリング業者と提携する
• スキャン要件についてスタッフを訓練する
• コンプライアンス監査を定期的に実施する

課題：埋め込み型機器のトレーサビリティ

問題：

• 病院の EMR との統合が必要
• 手術記録の欠落
• PHI のセキュリティ上の懸念
• リアルタイムのデータ取得が困難

解決策：

• 自動取得のための手術室統合システムを導入する
• 使用時点でバーコードをスキャンする
• HL7/FHIR 経由で EMR と統合する
• 安全で HIPAA に準拠したシステムを使用する
• 外科医向けのデータ取得プロセスを簡素化する
• バックアップとして専任のデータ入力スタッフを配置する
• 照合作業の監査を定期的に実施する

課題：委託在庫の最適化

問題：

• 委託在庫の配置先で在庫過剰が発生する
• 入手可能性とコストのバランス調整が難しい
• 在庫の長期滞留リスク
• ベンダーとの関係管理

解決策：

• データに基づいて PAR レベルを最適化する
• 配置先ごとの使用状況を定期的に分析する
• 補充率に関するベンダー評価表を作成する
• 動きの遅い在庫を拠点間で移動する
• 手術の需要を予測する
• 長期滞留在庫に関する契約条件を定める
• VMI（ベンダー管理在庫）プログラムを導入する

課題：貸出セットの追跡と返却

問題：

• 返却品の紛失や返却遅延
• 多額の金銭的負担
• 手作業による追跡プロセス
• 滅菌処理の遅延

解決策：

• 貸出品の追跡システムを自動化する
• 貸出セットに RFID タグを付ける
• 返却期限超過のアラートを設定する
• 貸出品コーディネーターの専任職を設ける
• 返却用の運送業者契約を結ぶ
• 滅菌プロセスを効率化する
• 集荷が遅れた場合にベンダーへペナルティを課す

課題：リコール実施の迅速化

問題：

• 影響を受けるすべての機器の所在特定が難しい
• トレーサビリティデータが不完全
• 顧客への通知が遅れる
• 規制当局への報告要件

解決策：

• 最初から堅牢なトレーサビリティシステムを構築する
• リコール対象の特定を自動化する
• 顧客通知用のテンプレートを事前に作成する
• リコールチームの訓練と演習を実施する
• FDA のリコールシステムと統合する
• トレーサビリティ監査を定期的に実施する
• 対応準備のために模擬リコールを実施する

課題：コールドチェーンコンプライアンス

問題：

• 輸送中の温度逸脱
• データロガーの故障
• 季節による気象の変化
• 国際輸送

解決策：

• 検証済みの包装システム
• 冗長化された温度監視
• 温度逸脱に対するリアルタイムアラート
• 季節に応じた輸送計画（極端な気象を回避）
• 適格性が確認された運送業者と輸送経路
• 代替輸送手段
• 温度逸脱の調査手順

---

出力形式

医療機器流通レポート

エグゼクティブサマリー：

• 流通ネットワークの概要
• 主要なコンプライアンス指標
• 目標に対する実績
• 対応が必要な重大な問題

UDI・トレーサビリティのコンプライアンス：

医療機器カテゴリ	クラス	UDIコンプライアンス	トレーサビリティの完全性	ギャップ分析
心臓インプラント	III	100%	98%	手術データが2%欠落
整形外科用インプラント	III	100%	95%	EMR統合に5%のギャップ
輸液ポンプ	II	98%	100%	旧式製品が2%
手術器具	I	85%	N/A	ラベル更新が15%保留中

委託在庫の概要：

病院	医療機器カテゴリ	手持ち数量	手持ち在庫金額	月間使用量	回転率	180日超の滞留
メモリアル病院	心臓ステント	45	$67,500	8	2.1	3点
リージョナル・メディカル	股関節インプラント	62	$217,000	15	2.9	0点
ユニバーシティ病院	脊椎セット	38	$133,000	6	1.9	8点

実施中のリコール：

リコールID	医療機器	クラス	対象ロット	対象機器の合計	埋め込み済み	回収済み	回収率
REC-2024-001	ペースメーカー Model X	I	LOT-A, LOT-B	247	89	158	64%
REC-2024-003	IV Pump Series 3	II	LOT-2024-C	1,205	0	1,089	90%

流通パフォーマンス指標：

指標	現在	目標	状況
納期遵守率	96.2%	95%	✓
完全注文率	92.8%	95%	⚠
在庫精度	99.1%	98%	✓
UDIコンプライアンス	97.5%	100%	⚠
トレーサビリティの完全性	98.3%	99%	⚠
コールドチェーンコンプライアンス	99.8%	99%	✓
リコール対応時間	6.2 hrs	< 8 hrs	✓

アクション項目：

1. クラス I/II の医療機器の残り2.5%についてUDIラベルの貼付を完了する
2. EMR統合を改善し、トレーサビリティのギャップを解消する
3. クラス I のリコール REC-2024-001 の回収を加速する
4. ユニバーシティ病院の委託在庫の滞留を減らす
5. 完全注文率のギャップに対処する（書類の正確性）

---

確認事項

さらに詳しい情報が必要な場合：

1. どのクラスの医療機器を流通させていますか？（I、II、III）
2. UDIコンプライアンスの現在の状況はどうなっていますか？
3. トレーサビリティシステムを導入していますか？
4. 委託販売と購入が占める割合はそれぞれどのくらいですか？
5. 年間で管理する実施中のリコールは何件ですか？
6. 温度に敏感な製品を流通させていますか？
7. 現在の在庫精度はどのくらいですか？
8. 国際流通を行っていますか？
9. どのQMSシステムを使用していますか？
10. コンプライアンスに関する最大の懸念事項やギャップは何ですか？

---

関連スキル

• hospital-logistics：病院内のサプライチェーン管理
• pharmacy-supply-chain：医薬品の流通とロジスティクス
• clinical-trial-logistics：臨床試験のサプライチェーン管理
• compliance-management：規制コンプライアンスと品質システム
• track-and-trace：製品の追跡とシリアル化
• quality-management：品質管理システム
• inventory-optimization：在庫最適化の手法
• warehouse-design：配送センターの設計
• risk-mitigation：サプライチェーンのリスク管理