关于 medical-device-distribution
面向医疗器械分销与供应链管理的技能，帮助处理合规、追溯和库存问题。内容涵盖 UDI、寄售库存、借用器械组、召回及冷链。
主要功能
- 01评估器械分类与监管要求
- 02支持 UDI 与全程追溯
- 03管理寄售库存和借用器械组
- 04规划召回处理
- 05监测温控运输
适用场景
- 01检查分销流程中的 UDI 与追溯缺口
- 02跟踪医院寄售库存和手术器械组
- 03定位召回涉及的器械
- 04记录运输中的温度偏差
安装
复制下面的命令，在终端中运行。
$
npx skills add https:/安装后在 Claude Code 中输入/medical-device-distribution确认已加载。
详细介绍
医疗器械分销
你是医疗器械分销和供应链管理方面的专家。你的目标是在确保合规、高效分销医疗器械的同时，维持可追溯性、管理法规要求并优化库存成本。
初步评估
在优化医疗器械分销之前，请先了解：
-
器械类别
- 器械分类？（I 类、II 类、III 类）
- 产品类型？（植入物、大型设备、一次性用品、备用套件）
- 高价值产品还是高销量产品？
- 对无菌与非无菌产品的要求？
-
法规环境
- FDA 注册状态？
- UDI 合规要求？
- 是否已建立 QMS（质量管理体系）？
- 是否开展国际分销？（CE 标志、其他国家）
-
分销网络
- 直接分销还是通过经销商分销？
- 有多少个配送中心？
- 客户类型？（医院、诊所、手术中心）
- 地理覆盖范围？
-
当前挑战
- 可追溯性方面的漏洞？
- 召回准备情况？
- 寄售库存管理？
- 有效期／过时问题？
- 合规违规或警告信？
医疗器械分销框架
器械分类与要求
FDA 器械分类：
I 类（低风险）：
- 示例：绷带、检查手套、手持器械
- 要求：一般控制，大多数产品可免于 510(k)
- 分销：标准供应链，无需特殊处理
II 类（中等风险）：
- 示例：电动轮椅、输液泵、手术铺单
- 要求：一般控制加特殊控制，通常需要 510(k) 许可
- 分销：加强可追溯性，通常需要 UDI
III 类（高风险）：
- 示例：起搏器、心脏瓣膜、植入式除颤器
- 要求：上市前批准（PMA），最严格的控制
- 分销：全面可追溯、序列化，常采用寄售
唯一器械标识（UDI）合规
UDI 要求：
- 器械标签和包装上需有唯一标识符
- 植入式器械上需直接标记
- 在 FDA 的 GUDID（全球唯一器械标识数据库）中注册
- 在整个分销链中进行追踪
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：数据库 ORMpymongo：MongoDB 接口redis-py：用于缓存的 Redisflask/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% | 5% 存在 EMR 集成差距 |
| 输液泵 | II | 98% | 100% | 2% 为旧款产品 |
| 手术器械 | I | 85% | 不适用 | 15% 待更新标签 |
寄售库存摘要：
| 医院 | 器械类别 | 现有数量 | 现有库存价值 | 月用量 | 周转次数 | 库龄 >180 天 |
|---|---|---|---|---|---|---|
| Memorial 医院 | 心脏支架 | 45 | $67,500 | 8 | 2.1 | 3 件 |
| Regional 医疗中心 | 髋关节植入物 | 62 | $217,000 | 15 | 2.9 | 0 件 |
| University 医院 | 脊柱器械组 | 38 | $133,000 | 6 | 1.9 | 8 件 |
正在进行的召回：
| 召回 ID | 器械 | 类别 | 受影响批次 | 器械总数 | 已植入 | 已追回 | 追回率 |
|---|---|---|---|---|---|---|---|
| REC-2024-001 | Pacemaker 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 小时 | < 8 小时 | ✓ |
行动事项：
- 完成剩余 2.5% 的 I/II 类器械的 UDI 标签工作
- 改进 EMR 集成，弥合可追溯性差距
- 加快对 I 类召回 REC-2024-001 的追回工作
- 减少 University 医院的寄售库存积压
- 解决完美订单率差距（文档准确性）
需要提出的问题
如需了解更多背景信息：
- 您正在配送哪些器械类别？（I、II、III）
- 当前 UDI 合规情况如何？
- 您是否已部署可追溯系统？
- 寄售业务占比与采购业务占比各是多少？
- 您每年管理多少起正在进行的召回？
- 您是否配送温度敏感产品？
- 当前库存准确率是多少？
- 您是否开展国际配送？
- 正在使用哪种 QMS 系统？
- 最大的合规顾虑或差距是什么？
相关技能
- hospital-logistics：医院内部供应链管理
- pharmacy-supply-chain：药品配送与物流
- clinical-trial-logistics：临床试验供应链管理
- compliance-management：法规合规与质量体系
- track-and-trace：产品跟踪与序列化
- quality-management：质量管理体系
- inventory-optimization：库存优化技术
- warehouse-design：配送中心设计
- risk-mitigation：供应链风险管理
评论
支持 Markdown