pharmacy-supply-chain

药品供应链

你是药品供应链管理和药品分销方面的专家。你的目标是在确保安全、合规且经济高效地分销药品的同时，维持法规合规、管理受管制药品并防止药品短缺。

初步评估

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

1. 药房类型和范围

   • 药房类型？（医院、零售、专科、邮购、340B）
   • 经营地点数量？（单个地点还是医疗系统）
   • 患者量和处方量？
   • 服务类型？（住院、门诊、输液、专科）

2. 法规环境

   • DEA 注册状态及经手的药品管制级别？
   • 州药房委员会的要求？
   • 是否参与 340B 项目？
   • 是否符合 DSCSA（《药品供应链安全法》）要求？
   • 获得哪些认证？（Joint Commission、ACHC、URAC）

3. 库存和处方目录

   • 处方目录中的药品数量？
   • 库存投入和周转率？
   • 高成本专科药品？
   • 受管制药品的数量？
   • 仿制药与品牌药的比例？

4. 当前挑战

   • 药品短缺的影响？
   • 过期和浪费水平？
   • 受管制药品的转移风险？
   • 340B 合规方面的差距？
   • 分销效率低下？

---

药品供应链框架

药品分销渠道

1. 批发商／分销商模式

• 主要分销渠道（占药品的 80-90%）
• 主要批发商：McKesson、Cardinal Health、AmerisourceBergen
• 优势：选择广泛、每日配送、信用账期
• 注意事项：批发商费用、合同合规

2. 直接从制造商采购

• 专科药品
• 限制分销药品
• 大批量仿制药（节省成本）
• 疫苗和生物制品
• 优势：成本更低、供应保障更好
• 注意事项：最低订购量、配送频率较低

3. 340B 合约药房

• 涵盖实体按 340B 最高限价采购
• 向符合条件的患者配药
• 可大幅节省成本
• 合规要求复杂

4. 专科药房分销

• 高成本、复杂药品
• 限制分销网络
• 患者支持服务
• 事先授权和报销支持

---

DSCSA 合规与追溯

《药品供应链安全法》（DSCSA）

要求：

• 在包装层级追踪产品（序列化）
• 验证产品的合法性
• 检测并应对可疑／非法产品
• 建立追踪系统和流程

时间表：

• 2023 年：加强药品分销安全
• 2024 年：实现全面的电子化互操作追踪（2024 年 11 月）

实施：

import hashlib
from dataclasses import dataclass
from datetime import datetime
from typing import List, Optional

@dataclass
class ProductIdentifier:
    """
    DSCSA-compliant product identifier
    """
    gtin: str  # Global Trade Item Number (NDC in GTIN-14 format)
    serial_number: str
    lot_number: str
    expiration_date: datetime

    def to_serialized_string(self):
        """Convert to DSCSA format"""
        exp_date = self.expiration_date.strftime("%y%m%d")
        return f"(01){self.gtin}(21){self.serial_number}(10){self.lot_number}(17){exp_date}"

    @staticmethod
    def from_2d_barcode(barcode_string):
        """Parse 2D Data Matrix barcode"""
        import re

        patterns = {
            'gtin': r'\(01\)(\d{14})',
            'serial_number': r'\(21\)([A-Za-z0-9]+)',
            'lot_number': r'\(10\)([A-Za-z0-9]+)',
            'expiration_date': r'\(17\)(\d{6})'
        }

        data = {}
        for field, pattern in patterns.items():
            match = re.search(pattern, barcode_string)
            if match:
                value = match.group(1)
                if field == 'expiration_date':
                    data[field] = datetime.strptime(value, "%y%m%d")
                else:
                    data[field] = value

        return ProductIdentifier(**data) if data else None

@dataclass
class TransactionInformation:
    """
    DSCSA Transaction Information (TI)
    """
    product_identifier: ProductIdentifier
    transaction_date: datetime
    ship_from: str  # Business name and address
    ship_to: str
    quantity: int

@dataclass
class TransactionHistory:
    """
    DSCSA Transaction History (TH) - full chain of ownership
    """
    transactions: List[TransactionInformation]

    def add_transaction(self, transaction: TransactionInformation):
        """Add transaction to history"""
        self.transactions.append(transaction)

    def verify_chain_of_custody(self):
        """Verify unbroken chain of custody"""
        if len(self.transactions) < 2:
            return True

        for i in range(len(self.transactions) - 1):
            current = self.transactions[i]
            next_trans = self.transactions[i + 1]

            # Verify ship_to of current matches ship_from of next
            if current.ship_to != next_trans.ship_from:
                return False

        return True

@dataclass
class TransactionStatement:
    """
    DSCSA Transaction Statement (TS) - attestation of legitimacy
    """
    product_identifier: ProductIdentifier
    statement_date: datetime
    authorized_entity: str
    attestation: str = "Product is legitimate and not counterfeit"

class DSCSATraceabilitySystem:
    """
    Manage DSCSA traceability and compliance
    """

    def __init__(self, business_name, dea_number, license_number):
        self.business_name = business_name
        self.dea_number = dea_number
        self.license_number = license_number
        self.inventory = {}
        self.transactions = []

    def receive_product(self, product_id: ProductIdentifier, quantity: int,
                        transaction_info: TransactionInformation,
                        transaction_history: TransactionHistory,
                        transaction_statement: TransactionStatement):
        """
        Receive product with DSCSA documentation
        """

        # Verify transaction history
        if not transaction_history.verify_chain_of_custody():
            raise ValueError("Chain of custody verification failed")

        # Verify product identifier matches
        if product_id.serial_number != transaction_info.product_identifier.serial_number:
            raise ValueError("Product identifier mismatch")

        # Store product in inventory with TI/TH/TS
        inventory_key = f"{product_id.gtin}-{product_id.serial_number}"

        self.inventory[inventory_key] = {
            'product_identifier': product_id,
            'quantity': quantity,
            'received_date': datetime.now(),
            'transaction_information': transaction_info,
            'transaction_history': transaction_history,
            'transaction_statement': transaction_statement,
            'status': 'in_stock'
        }

        return inventory_key

    def dispense_product(self, inventory_key: str, quantity: int,
                         patient_or_customer: str):
        """
        Dispense product to patient or transfer to another entity
        """

        if inventory_key not in self.inventory:
            raise ValueError(f"Product {inventory_key} not found in inventory")

        product_record = self.inventory[inventory_key]

        if product_record['quantity'] < quantity:
            raise ValueError(f"Insufficient quantity. Available: {product_record['quantity']}")

        # Create transaction record
        transaction = {
            'type': 'dispensed',
            'product_identifier': product_record['product_identifier'],
            'quantity': quantity,
            'date': datetime.now(),
            'recipient': patient_or_customer
        }

        self.transactions.append(transaction)

        # Update inventory
        product_record['quantity'] -= quantity

        if product_record['quantity'] == 0:
            product_record['status'] = 'dispensed'

        return transaction

    def verify_product(self, serialized_string: str):
        """
        Verify product legitimacy using DSCSA data
        """

        product_id = ProductIdentifier.from_2d_barcode(serialized_string)

        if not product_id:
            return {'verified': False, 'reason': 'Invalid product identifier'}

        inventory_key = f"{product_id.gtin}-{product_id.serial_number}"

        if inventory_key in self.inventory:
            product = self.inventory[inventory_key]
            return {
                'verified': True,
                'product': product_id,
                'status': product['status'],
                'received_date': product['received_date']
            }
        else:
            return {'verified': False, 'reason': 'Product not found in inventory'}

    def suspect_product_investigation(self, inventory_key: str, reason: str):
        """
        Quarantine and investigate suspect product
        """

        if inventory_key not in self.inventory:
            raise ValueError(f"Product {inventory_key} not found")

        product = self.inventory[inventory_key]
        product['status'] = 'quarantined'
        product['quarantine_reason'] = reason
        product['quarantine_date'] = datetime.now()

        # Notify FDA and trading partners as required
        investigation = {
            'product': product['product_identifier'],
            'reason': reason,
            'date': datetime.now(),
            'actions_taken': 'Product quarantined, investigation initiated'
        }

        return investigation

# Example usage
dscsa_system = DSCSATraceabilitySystem(
    business_name="Memorial Hospital Pharmacy",
    dea_number="FM1234563",
    license_number="PHY-12345"
)

# Create product identifier
product = ProductIdentifier(
    gtin="00300123456789",  # NDC in GTIN-14 format
    serial_number="ABC123XYZ789",
    lot_number="LOT2024-A",
    expiration_date=datetime(2026, 12, 31)
)

# Transaction information
trans_info = TransactionInformation(
    product_identifier=product,
    transaction_date=datetime.now(),
    ship_from="McKesson Corporation, 1234 Distributor Way",
    ship_to="Memorial Hospital Pharmacy, 5678 Hospital Blvd",
    quantity=100
)

# Transaction history
trans_history = TransactionHistory(transactions=[trans_info])

# Transaction statement
trans_statement = TransactionStatement(
    product_identifier=product,
    statement_date=datetime.now(),
    authorized_entity="McKesson Corporation"
)

# Receive product
inventory_key = dscsa_system.receive_product(
    product_id=product,
    quantity=100,
    transaction_info=trans_info,
    transaction_history=trans_history,
    transaction_statement=trans_statement
)

print(f"Product received: {inventory_key}")

# Verify product
barcode = product.to_serialized_string()
verification = dscsa_system.verify_product(barcode)
print(f"Product verified: {verification['verified']}")

---

受管制药品管理

DEA 受管制药品级别

一级： 无公认的医疗用途，滥用风险高

• 示例：海洛因、LSD、大麻（联邦层面）
• 通常不在药房经营

二级： 滥用风险高，有公认的医疗用途

• 示例：羟考酮、吗啡、芬太尼、安非他明、可卡因
• 要求：书面处方（有例外情况），不可续配，安全储存

三级： 滥用风险中等

• 示例：可待因／对乙酰氨基酚、氯胺酮、睾酮
• 要求：书面或电子处方，6 个月内最多续配 5 次

四级： 滥用风险低

• 示例：阿普唑仑、地西泮、曲马多、唑吡坦
• 要求：书面或电子处方，6 个月内最多续配 5 次

五级： 滥用风险最低

• 示例：含可待因 <200mg 的止咳制剂
• 要求：部分州可能允许非处方销售

永续库存系统

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

class ControlledSubstanceSchedule(Enum):
    SCHEDULE_II = "C-II"
    SCHEDULE_III = "C-III"
    SCHEDULE_IV = "C-IV"
    SCHEDULE_V = "C-V"

class TransactionType(Enum):
    RECEIVED = "received"
    DISPENSED = "dispensed"
    WASTED = "wasted"
    RETURNED = "returned"
    TRANSFERRED = "transferred"
    DESTROYED = "destroyed"

class ControlledSubstanceManager:
    """
    Perpetual inventory system for controlled substances
    """

    def __init__(self, pharmacy_name, dea_number):
        self.pharmacy_name = pharmacy_name
        self.dea_number = dea_number
        self.inventory = {}
        self.transactions = []

    def add_drug(self, drug_id, drug_name, ndc, schedule, strength, form):
        """
        Add controlled substance to formulary
        """

        self.inventory[drug_id] = {
            'drug_id': drug_id,
            'drug_name': drug_name,
            'ndc': ndc,
            'schedule': schedule,
            'strength': strength,
            'form': form,
            'quantity_on_hand': 0,
            'unit_of_measure': 'each'
        }

    def record_transaction(self, drug_id, transaction_type, quantity,
                          performer, witness=None, metadata=None):
        """
        Record controlled substance transaction

        Parameters:
        - drug_id: Drug identifier
        - transaction_type: Type of transaction (TransactionType enum)
        - quantity: Quantity (positive for additions, negative for removals)
        - performer: Person performing transaction
        - witness: Witness required for Schedule II (optional for others)
        - metadata: Additional data (Rx number, patient, waste reason, etc.)
        """

        if drug_id not in self.inventory:
            raise ValueError(f"Drug {drug_id} not in inventory")

        drug = self.inventory[drug_id]

        # Schedule II requires witness for dispensing and waste
        if drug['schedule'] == ControlledSubstanceSchedule.SCHEDULE_II:
            if transaction_type in [TransactionType.DISPENSED, TransactionType.WASTED]:
                if not witness:
                    raise ValueError("Witness required for Schedule II dispensing/waste")

        # Create transaction record
        transaction = {
            'transaction_id': len(self.transactions) + 1,
            'timestamp': datetime.now(),
            'drug_id': drug_id,
            'drug_name': drug['drug_name'],
            'ndc': drug['ndc'],
            'schedule': drug['schedule'].value,
            'transaction_type': transaction_type.value,
            'quantity': quantity,
            'balance_before': drug['quantity_on_hand'],
            'balance_after': drug['quantity_on_hand'] + quantity,
            'performer': performer,
            'witness': witness,
            'metadata': metadata or {}
        }

        # Update inventory
        drug['quantity_on_hand'] += quantity

        if drug['quantity_on_hand'] < 0:
            raise ValueError(f"Negative inventory not allowed. Current: {drug['quantity_on_hand']}")

        # Store transaction
        self.transactions.append(transaction)

        return transaction

    def receive_order(self, drug_id, quantity, invoice_number, supplier, received_by):
        """
        Receive controlled substance order
        """

        return self.record_transaction(
            drug_id=drug_id,
            transaction_type=TransactionType.RECEIVED,
            quantity=quantity,
            performer=received_by,
            metadata={
                'invoice_number': invoice_number,
                'supplier': supplier
            }
        )

    def dispense_prescription(self, drug_id, quantity, rx_number, patient_id,
                             pharmacist, technician_witness=None):
        """
        Dispense controlled substance prescription
        """

        return self.record_transaction(
            drug_id=drug_id,
            transaction_type=TransactionType.DISPENSED,
            quantity=-quantity,  # Negative for removal
            performer=pharmacist,
            witness=technician_witness,
            metadata={
                'rx_number': rx_number,
                'patient_id': patient_id
            }
        )

    def waste_medication(self, drug_id, quantity, reason, pharmacist, witness):
        """
        Waste controlled substance (expired, damaged, etc.)
        """

        return self.record_transaction(
            drug_id=drug_id,
            transaction_type=TransactionType.WASTED,
            quantity=-quantity,
            performer=pharmacist,
            witness=witness,
            metadata={'waste_reason': reason}
        )

    def physical_count(self, drug_id, counted_quantity, counted_by, witness=None):
        """
        Perform physical count and reconcile with perpetual inventory
        """

        if drug_id not in self.inventory:
            raise ValueError(f"Drug {drug_id} not in inventory")

        drug = self.inventory[drug_id]
        system_quantity = drug['quantity_on_hand']
        discrepancy = counted_quantity - system_quantity

        count_record = {
            'drug_id': drug_id,
            'drug_name': drug['drug_name'],
            'ndc': drug['ndc'],
            'schedule': drug['schedule'].value,
            'count_date': datetime.now(),
            'system_quantity': system_quantity,
            'physical_count': counted_quantity,
            'discrepancy': discrepancy,
            'counted_by': counted_by,
            'witness': witness
        }

        return count_record

    def perpetual_inventory_report(self, drug_id=None, schedule=None):
        """
        Generate perpetual inventory report
        """

        transactions_df = pd.DataFrame(self.transactions)

        if drug_id:
            transactions_df = transactions_df[transactions_df['drug_id'] == drug_id]

        if schedule:
            transactions_df = transactions_df[transactions_df['schedule'] == schedule.value]

        return transactions_df

    def biennial_inventory_report(self):
        """
        DEA biennial (every 2 years) inventory report
        """

        inventory_list = []

        for drug_id, drug in self.inventory.items():
            inventory_list.append({
                'drug_name': drug['drug_name'],
                'ndc': drug['ndc'],
                'schedule': drug['schedule'].value,
                'strength': drug['strength'],
                'form': drug['form'],
                'quantity_on_hand': drug['quantity_on_hand']
            })

        inventory_df = pd.DataFrame(inventory_list)
        inventory_df = inventory_df.sort_values(['schedule', 'drug_name'])

        report = {
            'pharmacy_name': self.pharmacy_name,
            'dea_number': self.dea_number,
            'report_date': datetime.now(),
            'inventory': inventory_df
        }

        return report

# Example usage
cs_manager = ControlledSubstanceManager(
    pharmacy_name="Memorial Hospital Pharmacy",
    dea_number="FM1234563"
)

# Add controlled substances to formulary
cs_manager.add_drug(
    drug_id='OXY-5MG',
    drug_name='Oxycodone',
    ndc='00406-0505-62',
    schedule=ControlledSubstanceSchedule.SCHEDULE_II,
    strength='5mg',
    form='Tablet'
)

cs_manager.add_drug(
    drug_id='DIAZ-5MG',
    drug_name='Diazepam',
    ndc='00591-3445-01',
    schedule=ControlledSubstanceSchedule.SCHEDULE_IV,
    strength='5mg',
    form='Tablet'
)

# Receive order
cs_manager.receive_order(
    drug_id='OXY-5MG',
    quantity=500,
    invoice_number='INV-123456',
    supplier='McKesson',
    received_by='RPh John Smith'
)

# Dispense prescriptions
cs_manager.dispense_prescription(
    drug_id='OXY-5MG',
    quantity=30,
    rx_number='RX-001234',
    patient_id='PT-567890',
    pharmacist='RPh John Smith',
    technician_witness='Tech Jane Doe'  # Schedule II requires witness
)

# Waste expired medication
cs_manager.waste_medication(
    drug_id='OXY-5MG',
    quantity=10,
    reason='Expired',
    pharmacist='RPh John Smith',
    witness='RPh Mary Johnson'
)

# Physical count
count = cs_manager.physical_count(
    drug_id='OXY-5MG',
    counted_quantity=460,
    counted_by='RPh John Smith',
    witness='RPh Mary Johnson'
)

print(f"Physical Count - Expected: {count['system_quantity']}, Counted: {count['physical_count']}, Discrepancy: {count['discrepancy']}")

# Generate perpetual inventory report
report = cs_manager.perpetual_inventory_report(drug_id='OXY-5MG')
print("\nPerpetual Inventory Report:")
print(report[['timestamp', 'transaction_type', 'quantity', 'balance_after', 'performer']])

---

340B 项目合规

340B 药品定价计划

概述：

• 联邦计划，要求制造商以优惠价格提供门诊药品
• 符合条件的实体：安全网医疗服务提供者（医院、FQHC、Ryan White 诊所）
• 节省幅度可达批发采购成本的 20-50%
• 合规要求复杂

关键合规要求：

1. 确定患者资格
2. 防止药品转移
3. 防止重复折扣
4. 准确记录
5. 合同药房合规

class Patient340BEligibility:
    """
    Determine 340B patient eligibility
    """

    def __init__(self, covered_entity_id, entity_type):
        self.covered_entity_id = covered_entity_id
        self.entity_type = entity_type  # DSH, PED, CAH, FQHC, etc.

    def check_eligibility(self, patient_id, encounter_data):
        """
        Determine if patient is eligible for 340B

        Patient must meet all criteria:
        1. Established relationship with covered entity
        2. Received healthcare service from covered entity
        3. Covered entity has responsibility for care
        4. Drug prescribed by provider of covered entity
        5. Drug dispensed by eligible pharmacy
        """

        criteria = {
            'established_patient': self._check_established_patient(patient_id, encounter_data),
            'received_service': self._check_service_received(encounter_data),
            'provider_responsibility': self._check_provider_responsibility(encounter_data),
            'eligible_prescriber': self._check_eligible_prescriber(encounter_data),
            'eligible_pharmacy': self._check_eligible_pharmacy(encounter_data)
        }

        # All criteria must be met
        eligible = all(criteria.values())

        return {
            'eligible': eligible,
            'criteria_results': criteria,
            'reason': self._eligibility_reason(criteria) if not eligible else 'Meets all criteria'
        }

    def _check_established_patient(self, patient_id, encounter_data):
        """Check if patient has established relationship"""
        # Implementation would check patient registration, prior visits, etc.
        return encounter_data.get('established_patient', False)

    def _check_service_received(self, encounter_data):
        """Check if patient received service from covered entity"""
        return encounter_data.get('service_location') == self.covered_entity_id

    def _check_provider_responsibility(self, encounter_data):
        """Check if covered entity has responsibility for patient care"""
        return encounter_data.get('responsible_entity') == self.covered_entity_id

    def _check_eligible_prescriber(self, encounter_data):
        """Check if prescriber is employed/contracted by covered entity"""
        return encounter_data.get('prescriber_entity') == self.covered_entity_id

    def _check_eligible_pharmacy(self, encounter_data):
        """Check if pharmacy is covered entity or registered contract pharmacy"""
        pharmacy = encounter_data.get('dispensing_pharmacy')
        return pharmacy in [self.covered_entity_id] or \
               pharmacy in self._get_contract_pharmacies()

    def _get_contract_pharmacies(self):
        """Get list of registered contract pharmacies"""
        # Would query HRSA database
        return ['CONTRACT-PHARM-001', 'CONTRACT-PHARM-002']

    def _eligibility_reason(self, criteria):
        """Generate reason for ineligibility"""
        failed = [k for k, v in criteria.items() if not v]
        return f"Failed criteria: {', '.join(failed)}"

class Program340BManager:
    """
    Manage 340B program operations and compliance
    """

    def __init__(self, covered_entity_id):
        self.covered_entity_id = covered_entity_id
        self.eligibility_checker = Patient340BEligibility(covered_entity_id, 'DSH')
        self.purchases = []
        self.dispenses = []

    def process_prescription(self, rx_data, patient_id, encounter_data):
        """
        Process prescription and determine 340B eligibility
        """

        # Check patient eligibility
        eligibility = self.eligibility_checker.check_eligibility(patient_id, encounter_data)

        # Check drug eligibility (some drugs excluded from 340B)
        drug_eligible = self._check_drug_eligibility(rx_data['ndc'])

        # Determine if 340B pricing applies
        use_340b = eligibility['eligible'] and drug_eligible

        dispense_record = {
            'rx_number': rx_data['rx_number'],
            'patient_id': patient_id,
            'ndc': rx_data['ndc'],
            'quantity': rx_data['quantity'],
            'dispense_date': datetime.now(),
            '340b_eligible': use_340b,
            'eligibility_reason': eligibility['reason'],
            'acquisition_cost': self._get_acquisition_cost(rx_data['ndc'], use_340b)
        }

        self.dispenses.append(dispense_record)

        return dispense_record

    def _check_drug_eligibility(self, ndc):
        """
        Check if drug is eligible for 340B pricing

        Exclusions:
        - Orphan drugs for rare diseases (when used for that disease)
        - Drugs for cosmetic purposes
        - Drugs for fertility
        """
        # Simplified - would check against exclusion list
        return True

    def _get_acquisition_cost(self, ndc, use_340b):
        """Get drug acquisition cost (340B or WAC)"""
        # Simplified - would query pricing database
        wac_price = 100.00
        price_340b = wac_price * 0.60  # Typical 40% discount

        return price_340b if use_340b else wac_price

    def duplicate_discount_check(self, rx_data, patient_insurance):
        """
        Prevent duplicate discounts (340B + Medicaid rebate)

        Covered entities cannot receive 340B discount AND Medicaid rebate
        """

        is_medicaid = patient_insurance.get('payer_type') == 'Medicaid'
        is_340b = rx_data.get('340b_eligible', False)

        if is_medicaid and is_340b:
            # Options:
            # 1. Carve-out: Don't bill Medicaid, use 340B
            # 2. Carve-in: Bill Medicaid, don't use 340B
            return {
                'duplicate_risk': True,
                'recommendation': 'Use 340B pricing, do not bill Medicaid for rebate'
            }

        return {'duplicate_risk': False}

    def diversion_audit(self, start_date, end_date):
        """
        Audit for drug diversion (using 340B drugs for ineligible patients)
        """

        dispenses_df = pd.DataFrame(self.dispenses)

        # Filter date range
        dispenses_df = dispenses_df[
            (dispenses_df['dispense_date'] >= start_date) &
            (dispenses_df['dispense_date'] <= end_date)
        ]

        # Identify potential diversion
        ineligible_340b = dispenses_df[
            (dispenses_df['340b_eligible'] == False) &
            (dispenses_df['acquisition_cost'] < 100)  # Using 340B-priced inventory
        ]

        audit_results = {
            'total_dispenses': len(dispenses_df),
            '340b_dispenses': len(dispenses_df[dispenses_df['340b_eligible'] == True]),
            'potential_diversion_events': len(ineligible_340b),
            'diversion_details': ineligible_340b
        }

        return audit_results

# Example usage
program_340b = Program340BManager(covered_entity_id='CE-001')

# Process prescription
rx = {
    'rx_number': 'RX-123456',
    'ndc': '00406-0505-62',
    'quantity': 30
}

encounter = {
    'established_patient': True,
    'service_location': 'CE-001',
    'responsible_entity': 'CE-001',
    'prescriber_entity': 'CE-001',
    'dispensing_pharmacy': 'CE-001'
}

dispense = program_340b.process_prescription(rx, patient_id='PT-789', encounter_data=encounter)
print(f"340B Eligible: {dispense['340b_eligible']}")
print(f"Acquisition Cost: ${dispense['acquisition_cost']:.2f}")

# Duplicate discount check
insurance = {'payer_type': 'Medicaid', 'member_id': 'MCD123456'}
duplicate_check = program_340b.duplicate_discount_check(dispense, insurance)
print(f"Duplicate discount risk: {duplicate_check['duplicate_risk']}")

---

药品短缺管理

短缺应对框架

import pandas as pd
from datetime import datetime, timedelta

class DrugShortageManager:
    """
    Manage drug shortages and implement mitigation strategies
    """

    def __init__(self, pharmacy_name):
        self.pharmacy_name = pharmacy_name
        self.shortages = []
        self.inventory = {}

    def declare_shortage(self, drug_id, drug_name, ndc, shortage_reason,
                        expected_duration_days, therapeutic_alternatives=None):
        """
        Declare drug shortage
        """

        shortage = {
            'shortage_id': f"SHORT-{len(self.shortages)+1}",
            'drug_id': drug_id,
            'drug_name': drug_name,
            'ndc': ndc,
            'declared_date': datetime.now(),
            'shortage_reason': shortage_reason,
            'expected_resolution': datetime.now() + timedelta(days=expected_duration_days),
            'status': 'active',
            'therapeutic_alternatives': therapeutic_alternatives or [],
            'mitigation_actions': []
        }

        self.shortages.append(shortage)

        return shortage

    def mitigation_strategy(self, shortage_id):
        """
        Develop mitigation strategy for shortage

        Strategies:
        1. Therapeutic substitution
        2. Dosage form modification
        3. Conservation (restrict to critical patients)
        4. Alternative suppliers
        5. Compounding
        """

        shortage = next((s for s in self.shortages if s['shortage_id'] == shortage_id), None)

        if not shortage:
            raise ValueError(f"Shortage {shortage_id} not found")

        strategies = []

        # Check for therapeutic alternatives
        if shortage['therapeutic_alternatives']:
            strategies.append({
                'strategy': 'therapeutic_substitution',
                'description': f"Substitute with: {', '.join(shortage['therapeutic_alternatives'])}",
                'priority': 1
            })

        # Check current inventory
        current_qty = self.inventory.get(shortage['drug_id'], {}).get('quantity', 0)
        days_supply = self._calculate_days_supply(shortage['drug_id'], current_qty)

        if days_supply < 7:
            strategies.append({
                'strategy': 'conservation',
                'description': 'Restrict to critical patients only (ICU, life-saving)',
                'priority': 1
            })

        # Alternative sourcing
        strategies.append({
            'strategy': 'alternative_sourcing',
            'description': 'Contact alternative distributors and direct manufacturers',
            'priority': 2
        })

        # Compounding option (if appropriate)
        if self._can_compound(shortage['drug_id']):
            strategies.append({
                'strategy': 'compounding',
                'description': 'Compound in-house or outsource to 503B',
                'priority': 3
            })

        shortage['mitigation_actions'] = strategies

        return strategies

    def _calculate_days_supply(self, drug_id, quantity):
        """Calculate days of supply remaining based on usage"""
        # Simplified - would use historical usage data
        avg_daily_usage = 10
        return quantity / avg_daily_usage if avg_daily_usage > 0 else 0

    def _can_compound(self, drug_id):
        """Check if drug can be compounded"""
        # Simplified - would check formulary
        return True

    def prioritize_patients(self, drug_id, patient_list):
        """
        Prioritize patients for shortage allocation

        Priority levels:
        1. Life-saving/critical care
        2. Prevent serious morbidity
        3. Symptomatic relief
        """

        prioritized = []

        for patient in patient_list:
            # Determine priority based on indication
            indication = patient.get('indication', '')

            if any(critical in indication.lower() for critical in ['sepsis', 'shock', 'arrest', 'seizure']):
                priority = 1
                priority_desc = 'Critical - Life-saving'
            elif any(serious in indication.lower() for serious in ['infection', 'pain-severe', 'surgery']):
                priority = 2
                priority_desc = 'High - Prevent serious morbidity'
            else:
                priority = 3
                priority_desc = 'Medium - Symptomatic relief'

            prioritized.append({
                'patient_id': patient['patient_id'],
                'indication': indication,
                'priority': priority,
                'priority_description': priority_desc,
                'prescriber': patient.get('prescriber'),
                'requested_quantity': patient.get('quantity')
            })

        # Sort by priority
        prioritized.sort(key=lambda x: x['priority'])

        return pd.DataFrame(prioritized)

    def shortage_report(self):
        """
        Generate active shortages report
        """

        active_shortages = [s for s in self.shortages if s['status'] == 'active']

        if not active_shortages:
            return None

        report = []

        for shortage in active_shortages:
            current_qty = self.inventory.get(shortage['drug_id'], {}).get('quantity', 0)
            days_supply = self._calculate_days_supply(shortage['drug_id'], current_qty)

            report.append({
                'shortage_id': shortage['shortage_id'],
                'drug_name': shortage['drug_name'],
                'ndc': shortage['ndc'],
                'declared_date': shortage['declared_date'],
                'expected_resolution': shortage['expected_resolution'],
                'reason': shortage['shortage_reason'],
                'current_inventory': current_qty,
                'days_supply': round(days_supply, 1),
                'alternatives': ', '.join(shortage['therapeutic_alternatives']),
                'mitigation_actions': len(shortage['mitigation_actions'])
            })

        return pd.DataFrame(report)

# Example usage
shortage_mgr = DrugShortageManager("Memorial Hospital Pharmacy")

# Add inventory
shortage_mgr.inventory['PROP-100MG'] = {'quantity': 50}

# Declare shortage
shortage = shortage_mgr.declare_shortage(
    drug_id='PROP-100MG',
    drug_name='Propofol 100mg/10mL',
    ndc='63323-0269-10',
    shortage_reason='Manufacturing delay at primary supplier',
    expected_duration_days=45,
    therapeutic_alternatives=['Etomidate', 'Ketamine']
)

print(f"Shortage declared: {shortage['shortage_id']}")

# Develop mitigation strategy
strategies = shortage_mgr.mitigation_strategy(shortage['shortage_id'])
print("\nMitigation Strategies:")
for strategy in strategies:
    print(f"  {strategy['priority']}. {strategy['strategy']}: {strategy['description']}")

# Prioritize patients
patients = [
    {'patient_id': 'PT-001', 'indication': 'Septic shock - ICU', 'prescriber': 'Dr. Smith', 'quantity': 20},
    {'patient_id': 'PT-002', 'indication': 'Elective surgery - General anesthesia', 'prescriber': 'Dr. Jones', 'quantity': 20},
    {'patient_id': 'PT-003', 'indication': 'Status epilepticus', 'prescriber': 'Dr. Brown', 'quantity': 10}
]

prioritized = shortage_mgr.prioritize_patients('PROP-100MG', patients)
print("\nPrioritized Patient Allocation:")
print(prioritized[['patient_id', 'indication', 'priority_description', 'requested_quantity']])

---

专科药房运营

高价药品管理

class SpecialtyPharmacyManager:
    """
    Manage specialty pharmacy operations for high-cost medications
    """

    def __init__(self, pharmacy_name):
        self.pharmacy_name = pharmacy_name
        self.specialty_drugs = {}
        self.prior_authorizations = {}

    def add_specialty_drug(self, drug_id, drug_name, ndc, indication,
                          unit_cost, limited_distribution=False,
                          rems_required=False, storage_requirements=None):
        """
        Add specialty drug to formulary
        """

        self.specialty_drugs[drug_id] = {
            'drug_id': drug_id,
            'drug_name': drug_name,
            'ndc': ndc,
            'indication': indication,
            'unit_cost': unit_cost,
            'limited_distribution': limited_distribution,
            'rems_required': rems_required,  # Risk Evaluation and Mitigation Strategy
            'storage_requirements': storage_requirements or 'Room temperature'
        }

    def prior_authorization(self, drug_id, patient_id, prescriber,
                           diagnosis_code, clinical_justification,
                           insurance_info):
        """
        Submit prior authorization request
        """

        pa_id = f"PA-{len(self.prior_authorizations)+1}"

        pa_request = {
            'pa_id': pa_id,
            'drug_id': drug_id,
            'patient_id': patient_id,
            'prescriber': prescriber,
            'diagnosis_code': diagnosis_code,
            'clinical_justification': clinical_justification,
            'insurance_info': insurance_info,
            'submission_date': datetime.now(),
            'status': 'pending',
            'determination': None,
            'determination_date': None
        }

        self.prior_authorizations[pa_id] = pa_request

        return pa_request

    def financial_assistance_screening(self, patient_id, drug_id, annual_income,
                                      insurance_coverage):
        """
        Screen for patient financial assistance programs
        """

        drug = self.specialty_drugs.get(drug_id)

        if not drug:
            raise ValueError(f"Drug {drug_id} not found")

        # Calculate estimated patient cost
        if insurance_coverage:
            copay = insurance_coverage.get('copay', drug['unit_cost'] * 0.30)
            coinsurance_pct = insurance_coverage.get('coinsurance', 0)
        else:
            copay = drug['unit_cost']
            coinsurance_pct = 100

        estimated_annual_cost = copay * 12  # Assuming monthly therapy

        # Assistance program eligibility (simplified)
        assistance_programs = []

        # Manufacturer copay assistance
        if insurance_coverage and copay > 50:
            assistance_programs.append({
                'program': 'Manufacturer Copay Card',
                'estimated_savings': min(copay * 0.80, 12000),  # Up to $12K/year typical
                'eligibility': 'Commercial insurance required'
            })

        # Patient assistance program (PAP)
        federal_poverty_level = 30000  # Simplified
        if annual_income < federal_poverty_level * 5:
            assistance_programs.append({
                'program': 'Manufacturer Patient Assistance Program (PAP)',
                'estimated_savings': drug['unit_cost'],
                'eligibility': f'Income <500% FPL (${federal_poverty_level * 5:,.0f})'
            })

        # Foundation assistance
        if estimated_annual_cost > annual_income * 0.10:  # >10% of income
            assistance_programs.append({
                'program': 'Independent Charitable Foundation',
                'estimated_savings': estimated_annual_cost * 0.50,
                'eligibility': 'Disease-specific, income-based'
            })

        return {
            'patient_id': patient_id,
            'drug_name': drug['drug_name'],
            'estimated_annual_cost': round(estimated_annual_cost, 2),
            'pct_of_income': round(estimated_annual_cost / annual_income * 100, 1) if annual_income > 0 else 0,
            'assistance_programs': assistance_programs
        }

# Example usage
specialty_pharm = SpecialtyPharmacyManager("Memorial Specialty Pharmacy")

# Add specialty drug
specialty_pharm.add_specialty_drug(
    drug_id='HUMIRA-40MG',
    drug_name='Adalimumab (Humira)',
    ndc='00074-4339-02',
    indication='Rheumatoid Arthritis, Crohn\'s Disease',
    unit_cost=6000,
    limited_distribution=False,
    rems_required=False,
    storage_requirements='Refrigerated 2-8°C'
)

# Prior authorization
pa = specialty_pharm.prior_authorization(
    drug_id='HUMIRA-40MG',
    patient_id='PT-123456',
    prescriber='Dr. Williams',
    diagnosis_code='M05.79',  # Rheumatoid arthritis
    clinical_justification='Failed methotrexate and sulfasalazine. Significant disease activity.',
    insurance_info={'payer': 'Blue Cross', 'plan': 'PPO', 'member_id': 'BC123456'}
)

print(f"PA submitted: {pa['pa_id']}, Status: {pa['status']}")

# Financial assistance screening
insurance = {'copay': 500, 'coinsurance': 20}
assistance = specialty_pharm.financial_assistance_screening(
    patient_id='PT-123456',
    drug_id='HUMIRA-40MG',
    annual_income=45000,
    insurance_coverage=insurance
)

print(f"\nEstimated annual cost: ${assistance['estimated_annual_cost']:,.2f}")
print(f"Percent of income: {assistance['pct_of_income']}%")
print(f"\nAssistance programs available: {len(assistance['assistance_programs'])}")
for program in assistance['assistance_programs']:
    print(f"  - {program['program']}: Up to ${program['estimated_savings']:,.0f}")

---

药房绩效指标

关键绩效指标

def calculate_pharmacy_kpis(rx_data_df, inventory_data_df, cs_data_df=None):
    """
    Calculate pharmacy supply chain KPIs

    Parameters:
    - rx_data_df: Prescription dispensing data
    - inventory_data_df: Inventory positions
    - cs_data_df: Controlled substance data (optional)
    """

    kpis = {}

    # Fill rate / Service level
    if 'filled' in rx_data_df.columns:
        kpis['fill_rate'] = (rx_data_df['filled'].sum() / len(rx_data_df) * 100)

    # Inventory turns
    if all(col in inventory_data_df.columns for col in ['annual_usage', 'avg_inventory_value']):
        total_usage_value = inventory_data_df['annual_usage'].sum()
        total_avg_inventory = inventory_data_df['avg_inventory_value'].sum()
        kpis['inventory_turns'] = total_usage_value / total_avg_inventory if total_avg_inventory > 0 else 0

    # Expiry waste rate
    if 'expired' in inventory_data_df.columns:
        expired_value = inventory_data_df[inventory_data_df['expired'] == True]['inventory_value'].sum()
        total_value = inventory_data_df['inventory_value'].sum()
        kpis['expiry_waste_pct'] = (expired_value / total_value * 100) if total_value > 0 else 0

    # Generic dispensing rate (cost savings)
    if 'generic' in rx_data_df.columns:
        kpis['generic_dispensing_rate'] = (rx_data_df['generic'].sum() / len(rx_data_df) * 100)

    # Prescription turnaround time
    if 'turnaround_minutes' in rx_data_df.columns:
        kpis['avg_turnaround_minutes'] = rx_data_df['turnaround_minutes'].mean()

    # 340B capture rate (if applicable)
    if '340b_eligible' in rx_data_df.columns and '340b_used' in rx_data_df.columns:
        eligible = rx_data_df['340b_eligible'].sum()
        used = rx_data_df['340b_used'].sum()
        kpis['340b_capture_rate'] = (used / eligible * 100) if eligible > 0 else 0

    # Controlled substance accuracy (if applicable)
    if cs_data_df is not None and not cs_data_df.empty:
        if 'discrepancy' in cs_data_df.columns:
            kpis['cs_inventory_accuracy'] = (
                (len(cs_data_df[cs_data_df['discrepancy'] == 0]) / len(cs_data_df) * 100)
            )

    # DSCSA compliance rate
    if 'dscsa_compliant' in rx_data_df.columns:
        kpis['dscsa_compliance_rate'] = (rx_data_df['dscsa_compliant'].sum() / len(rx_data_df) * 100)

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

    return kpis

# Example data
rx_data = pd.DataFrame({
    'rx_number': range(1, 1001),
    'filled': [True] * 980 + [False] * 20,
    'generic': np.random.choice([True, False], 1000, p=[0.85, 0.15]),
    'turnaround_minutes': np.random.normal(20, 5, 1000),
    '340b_eligible': np.random.choice([True, False], 1000, p=[0.40, 0.60]),
    '340b_used': np.random.choice([True, False], 1000, p=[0.35, 0.65]),
    'dscsa_compliant': np.random.choice([True, False], 1000, p=[0.98, 0.02])
})

# Fix 340B logic (can only use if eligible)
rx_data.loc[~rx_data['340b_eligible'], '340b_used'] = False

inventory_data = pd.DataFrame({
    'drug_id': range(1, 501),
    'annual_usage': np.random.randint(1000, 50000, 500),
    'avg_inventory_value': np.random.randint(500, 10000, 500),
    'inventory_value': np.random.randint(500, 10000, 500),
    'expired': np.random.choice([True, False], 500, p=[0.02, 0.98])
})

kpis = calculate_pharmacy_kpis(rx_data, inventory_data)

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

---

工具与库

药房管理系统

ERP/药房系统：

• EPIC Willow Pharmacy：与 Epic EHR 集成
• Cerner PharmNet：药房管理模块
• Omnicell：自动化配药柜
• BD Pyxis：药品管理系统
• McKesson Pharmacy Systems：STAR、EnterpriseRx
• QS/1：独立药房管理

管制药品追踪：

• Omnicell ControlledRx：管制药品管理
• BD Pyxis CII Safe：管制药品保险柜
• Kit Check：药品追踪与转移监控
• Protenus：AI 驱动的药品转移检测

340B 管理：

• Kalderos：340B 索赔识别
• Macro Helix：340B 计划管理
• Verity Solutions：340B 分账
• RxStrategies：340B 优化

DSCSA 合规：

• TraceLink：DSCSA 合规平台
• SAP Information Collaboration Hub for Life Sciences
• rfxcel：序列化与可追溯性
• Systech UniSecure：产品认证

Python 库

数据分析：

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

法规合规：

• python-barcode：条码生成（NDC）
• qrcode：二维条码生成
• hashlib：数据完整性验证

优化：

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

可视化：

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

---

常见挑战与解决方案

挑战：在 2024 年 11 月前实现 DSCSA 合规

问题：

• 要求进行电子化、互操作的追踪
• 旧系统不合规
• 交易伙伴集成
• 在包装层级进行序列化

解决方案：

• 实施符合 DSCSA 的软件平台
• 尽早测试交易伙伴连接
• 对员工进行扫描要求培训
• 制定可疑产品调查流程
• 定期开展合规审计
• 参与行业试点计划

挑战：管制药品转移

问题：

• 员工转移风险
• 记录缺失
• 库存差异
• 监管处罚

解决方案：

• 自动化永续盘存系统
• 对 Schedule II 药品进行双人核验
• 定期开展随机审计和循环盘点
• 使用分析工具检测异常模式
• 开展背景调查和监控
• 制定明确的政策和纪律处分措施
• 建立保密举报机制

挑战：340B 合规与审计

问题：

• 患者资格规则复杂
• 合同药房合规
• 防止重复折扣
• HRSA 审计

解决方案：

• 自动化患者资格系统
• 与电子健康记录集成
• 针对 Medicaid 进行分账
• 定期开展内部合规审计
• 对员工进行资格标准培训
• 详细记录并保留审计追踪
• 与 340B 顾问/专家合作

挑战：药品短缺

问题：

• 关键药物无法供应
• 对患者护理的影响
• 治疗替代的复杂性
• 替代药物导致成本增加

解决方案：

• 多来源采购策略
• 早期预警监测（FDA 短缺清单）
• 治疗替代方案
• 面向重症患者的节约策略
• 配制替代药物（适用时）
• 与处方医生沟通
• 发挥团体采购组织的优势

挑战：专科药物成本

问题：

• 采购成本极高
• 事先授权流程延误
• 患者负担能力
• 事先授权申请未通过或患者不依从导致的浪费

解决方案：

• 主动申请事先授权
• 协助办理财务援助计划
• 分析买入后计费与白袋配送的利弊
• 初期限制发药数量
• 患者依从性支持服务
• 注册制造商援助计划
• 获得专科药房认证

挑战：过期与浪费

问题：

• 有效期较短的产品
• 周转缓慢的库存
• 存储错误
• 配制药物

解决方案：

• 执行 FEFO（先到期、先出库）
• 自动发送过期提醒
• 合理调整库存规模（基于数据设定 PAR 水平）
• 供应商退货计划
• 在不同地点之间调拨滞销品
• 对周转缓慢的物品实行即时订购
• 优化配制药物的使用期限（BUD）

---

输出格式

药房供应链报告

执行摘要：

• 药房概况和范围
• 与基准相比的关键绩效指标
• 主要合规举措
• 财务影响和机遇

库存优化：

药品类别	项目数	库存价值	年周转次数	库存可供天数	过期率	优化机会
抗生素	145	$125,000	15.2	24	0.5%	减少安全库存
专科药品	42	$850,000	8.1	45	1.2%	买入并开具账单与白袋配送方案的比较
受管制药品	38	$45,000	24.3	15	0.1%	合理
普通口服药	1,250	$180,000	18.5	20	1.8%	标准化

340B 项目绩效：

指标	结果	目标	状态
符合条件的处方	2,450/月	-	-
340B 捕获率	92.3%	95%	⚠
340B 年度节省金额	$1,850,000	$2,000,000	⚠
合同药房合规率	98.5%	100%	⚠
重复折扣事件	0	0	✓

受管制药品合规情况：

分类	项目数	永续库存准确率	实物盘点频率	年初至今差异数	状态
II 类	15	99.8%	每日	2	✓
III 类	12	99.5%	每周	3	✓
IV 类	23	99.2%	每周	5	✓

当前药品短缺：

药品名称	NDC	宣布日期	当前库存	可供天数	缓解策略	状态
Propofol 100mg	63323-0269-10	2024-01-15	50 瓶	5 天	替代来源、治疗替代药物	严重
Cefazolin 1g	00409-1964-50	2024-02-01	200 瓶	20 天	节约用药方案	监控中

关键绩效指标：

指标	当前值	目标	趋势
配药满足率	98.2%	98%	✓
库存周转次数	12.5	12.0	✓
仿制药调配率	89.3%	85%	✓
过期浪费率	1.2%	<1.5%	✓
340B 捕获率	92.3%	95%	⚠
DSCSA 合规率	98.1%	100%	⚠
受管制药品准确率	99.5%	99%	✓

行动事项：

1. 通过加强资格筛查提高 340B 捕获率
2. 完成剩余 1.9% 交易的 DSCSA 合规工作
3. 实施自动化短缺监控系统
4. 通过限制首次配药量减少专科药品浪费
5. 优化周转缓慢的抗生素库存

---

要提出的问题

如需更多背景信息：

1. 属于哪种类型的药房？（医院、零售、专科、邮购）
2. 是否属于 340B 覆盖实体？
3. DSCSA 合规状态如何？
4. 管理哪些受管制药品类别？
5. 当前库存投资额和周转情况如何？
6. 是否遇到特定药品短缺？
7. 使用哪种药房管理系统？
8. 最大的合规顾虑是什么？
9. 是否调配专科药品？
10. 与哪些批发商/分销商合作？

---

相关技能

• hospital-logistics：医院物资管理
• medical-device-distribution：医疗器械物流
• clinical-trial-logistics：临床试验供应链
• inventory-optimization：库存优化技术
• demand-forecasting：库存规划预测
• compliance-management：监管合规系统
• track-and-trace：产品追溯与序列化
• value-analysis：价值分析与成本削减
• quality-management：质量管理体系