skills/value-analysis/SKILL.md
When the user wants to conduct value analysis, evaluate medical products, establish formularies, or optimize product selection for healthcare. Also use when the user mentions "value analysis committee," "product evaluation," "clinical outcomes analysis," "total cost of ownership," "standardization," "formulary management," "evidence-based sourcing," "product selection," "physician preference items," or "cost-quality optimization." For hospital supply chain operations, see hospital-logistics. For cost analysis in general, see spend-analysis.
npx skillsauth add kishorkukreja/awesome-supply-chain value-analysisInstall this skill globally with one command. Works with Claude Code, Cursor, and Windsurf.
3 of 9 scanners reported clean
Some scanners were skipped, did not run, or reported a non-clean status. Review each row below.
You are an expert in healthcare value analysis and product evaluation. Your goal is to help healthcare organizations make evidence-based product selection decisions that optimize clinical outcomes, cost-effectiveness, and operational efficiency.
Before conducting value analysis, understand:
Organization Context
Product Category Focus
Value Analysis Structure
Goals & Objectives
Value = (Clinical Outcomes + Patient Experience + Staff Experience) / Total Cost
Components:
Membership:
Meeting Cadence:
Step 1: Request Submission
Step 2: Initial Screening
Step 3: Data Gathering
Step 4: Product Trial/Evaluation
Step 5: Financial Analysis
Step 6: Committee Review
Step 7: Implementation & Monitoring
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import List, Optional, Dict
import pandas as pd
import numpy as np
class ProductCategory(Enum):
IMPLANT = "implant"
SURGICAL_SUPPLY = "surgical_supply"
MEDICAL_DEVICE = "medical_device"
CAPITAL_EQUIPMENT = "capital_equipment"
PHARMACEUTICAL = "pharmaceutical"
COMMODITY = "commodity"
class RequestStatus(Enum):
SUBMITTED = "submitted"
SCREENING = "screening"
DATA_GATHERING = "data_gathering"
CLINICAL_TRIAL = "clinical_trial"
FINANCIAL_ANALYSIS = "financial_analysis"
COMMITTEE_REVIEW = "committee_review"
APPROVED = "approved"
DENIED = "denied"
ON_HOLD = "on_hold"
class UrgencyLevel(Enum):
ROUTINE = "routine"
URGENT = "urgent"
EMERGENCY = "emergency"
@dataclass
class ValueAnalysisRequest:
"""Value analysis request structure"""
request_id: str
submission_date: datetime
submitted_by: str
product_name: str
manufacturer: str
category: ProductCategory
clinical_justification: str
estimated_annual_volume: int
estimated_unit_cost: float
urgency: UrgencyLevel
status: RequestStatus
current_alternative: Optional[str] = None
physician_champion: Optional[str] = None
class ValueAnalysisCommittee:
"""
Manage value analysis process and product evaluations
"""
def __init__(self, organization_name):
self.organization_name = organization_name
self.requests = {}
self.evaluations = {}
self.formulary = {}
self.committee_members = []
def submit_request(self, request_data):
"""
Submit new value analysis request
Parameters:
- request_data: Dict with request information
"""
request_id = f"VAR-{len(self.requests)+1:05d}"
request = ValueAnalysisRequest(
request_id=request_id,
submission_date=datetime.now(),
submitted_by=request_data['submitted_by'],
product_name=request_data['product_name'],
manufacturer=request_data['manufacturer'],
category=ProductCategory[request_data['category'].upper()],
clinical_justification=request_data['clinical_justification'],
estimated_annual_volume=request_data['estimated_annual_volume'],
estimated_unit_cost=request_data['estimated_unit_cost'],
urgency=UrgencyLevel[request_data.get('urgency', 'ROUTINE').upper()],
status=RequestStatus.SUBMITTED,
current_alternative=request_data.get('current_alternative'),
physician_champion=request_data.get('physician_champion')
)
self.requests[request_id] = request
return request
def initial_screening(self, request_id):
"""
Conduct initial screening of request
"""
if request_id not in self.requests:
raise ValueError(f"Request {request_id} not found")
request = self.requests[request_id]
# Screening criteria
screening_results = {
'complete_submission': self._check_completeness(request),
'clinical_justification_adequate': len(request.clinical_justification) > 50,
'cost_impact_reasonable': request.estimated_unit_cost > 0,
'duplicate_request': self._check_duplicates(request),
'formulary_already_exists': self._check_formulary_exists(request.product_name)
}
# Determine if passes screening
passes_screening = (
screening_results['complete_submission'] and
screening_results['clinical_justification_adequate'] and
screening_results['cost_impact_reasonable'] and
not screening_results['duplicate_request']
)
if passes_screening:
request.status = RequestStatus.DATA_GATHERING
else:
request.status = RequestStatus.ON_HOLD
return {
'request_id': request_id,
'passes_screening': passes_screening,
'screening_results': screening_results,
'next_steps': 'Proceed to data gathering' if passes_screening else 'Revise and resubmit'
}
def _check_completeness(self, request):
"""Check if request is complete"""
required_fields = [
request.product_name,
request.manufacturer,
request.clinical_justification,
request.estimated_annual_volume,
request.estimated_unit_cost
]
return all(required_fields)
def _check_duplicates(self, request):
"""Check for duplicate requests"""
for req_id, req in self.requests.items():
if (req.product_name == request.product_name and
req.manufacturer == request.manufacturer and
req.status not in [RequestStatus.DENIED, RequestStatus.APPROVED]):
return True
return False
def _check_formulary_exists(self, product_name):
"""Check if product already on formulary"""
return product_name in self.formulary
def conduct_clinical_trial(self, request_id, trial_parameters):
"""
Set up clinical trial/evaluation
Parameters:
- request_id: Value analysis request
- trial_parameters: Dict with trial setup
"""
if request_id not in self.requests:
raise ValueError(f"Request {request_id} not found")
request = self.requests[request_id]
trial = {
'request_id': request_id,
'product_name': request.product_name,
'start_date': trial_parameters.get('start_date', datetime.now()),
'duration_days': trial_parameters.get('duration_days', 30),
'trial_sites': trial_parameters.get('trial_sites', []),
'sample_size': trial_parameters.get('sample_size', 20),
'evaluation_criteria': trial_parameters.get('criteria', []),
'participating_physicians': trial_parameters.get('physicians', []),
'status': 'active'
}
self.evaluations[request_id] = trial
request.status = RequestStatus.CLINICAL_TRIAL
return trial
def collect_trial_results(self, request_id, results_data):
"""
Collect and analyze trial results
Parameters:
- request_id: Request ID
- results_data: List of trial result records
"""
if request_id not in self.evaluations:
raise ValueError(f"No trial found for request {request_id}")
trial = self.evaluations[request_id]
# Aggregate results
results_df = pd.DataFrame(results_data)
# Calculate summary statistics
summary = {
'total_uses': len(results_df),
'clinical_success_rate': (results_df['clinical_success'].sum() / len(results_df) * 100) if len(results_df) > 0 else 0,
'user_satisfaction_avg': results_df['user_satisfaction'].mean() if 'user_satisfaction' in results_df.columns else None,
'complications': results_df['complication'].sum() if 'complication' in results_df.columns else 0,
'ease_of_use_avg': results_df['ease_of_use'].mean() if 'ease_of_use' in results_df.columns else None,
'would_recommend_pct': (results_df['would_recommend'].sum() / len(results_df) * 100) if 'would_recommend' in results_df.columns and len(results_df) > 0 else 0
}
trial['results_summary'] = summary
trial['results_data'] = results_df
trial['status'] = 'completed'
# Update request status
self.requests[request_id].status = RequestStatus.FINANCIAL_ANALYSIS
return summary
def financial_analysis(self, request_id, financial_data):
"""
Conduct total cost of ownership analysis
Parameters:
- request_id: Request ID
- financial_data: Dict with cost components
"""
if request_id not in self.requests:
raise ValueError(f"Request {request_id} not found")
request = self.requests[request_id]
# Total Cost of Ownership (TCO) calculation
tco = self._calculate_tco(request, financial_data)
# Compare to current alternative
if request.current_alternative:
current_tco = financial_data.get('current_alternative_tco', {})
comparison = self._compare_alternatives(tco, current_tco)
else:
comparison = None
analysis = {
'request_id': request_id,
'product_name': request.product_name,
'tco_analysis': tco,
'comparison': comparison,
'budget_impact': self._calculate_budget_impact(tco, request),
'roi_analysis': self._calculate_roi(tco, comparison, financial_data) if comparison else None
}
self.evaluations[request_id]['financial_analysis'] = analysis
return analysis
def _calculate_tco(self, request, financial_data):
"""
Calculate total cost of ownership
TCO Components:
- Acquisition cost
- Training costs
- Storage/handling costs
- Waste/expiry costs
- Complication costs
- Disposal costs
"""
annual_volume = request.estimated_annual_volume
unit_cost = request.estimated_unit_cost
tco = {
'acquisition_cost': annual_volume * unit_cost,
'training_cost': financial_data.get('training_cost', 0),
'storage_cost': financial_data.get('storage_cost_per_unit', 0) * annual_volume,
'waste_cost': financial_data.get('waste_rate', 0.02) * annual_volume * unit_cost,
'complication_cost': financial_data.get('complication_rate', 0) * financial_data.get('complication_cost_per_event', 0) * annual_volume,
'disposal_cost': financial_data.get('disposal_cost_per_unit', 0) * annual_volume
}
tco['total_annual_cost'] = sum(tco.values())
tco['cost_per_use'] = tco['total_annual_cost'] / annual_volume if annual_volume > 0 else 0
return tco
def _compare_alternatives(self, new_tco, current_tco):
"""Compare new product TCO to current alternative"""
new_total = new_tco['total_annual_cost']
current_total = current_tco.get('total_annual_cost', new_total)
comparison = {
'new_product_tco': new_total,
'current_product_tco': current_total,
'annual_difference': new_total - current_total,
'percent_change': ((new_total - current_total) / current_total * 100) if current_total > 0 else 0,
'recommendation': 'Cost savings' if new_total < current_total else 'Cost increase'
}
return comparison
def _calculate_budget_impact(self, tco, request):
"""Calculate budget impact"""
# Assume budget is based on current spend
current_budget = request.estimated_annual_volume * request.estimated_unit_cost
budget_impact = {
'current_budget': current_budget,
'projected_spend': tco['total_annual_cost'],
'budget_variance': tco['total_annual_cost'] - current_budget,
'budget_variance_pct': ((tco['total_annual_cost'] - current_budget) / current_budget * 100) if current_budget > 0 else 0
}
return budget_impact
def _calculate_roi(self, new_tco, comparison, financial_data):
"""Calculate return on investment"""
if not comparison:
return None
annual_savings = -comparison['annual_difference'] # Negative if cost increase
implementation_costs = (
financial_data.get('implementation_cost', 0) +
new_tco.get('training_cost', 0)
)
if annual_savings <= 0:
roi = {
'annual_savings': annual_savings,
'implementation_cost': implementation_costs,
'payback_period_years': None,
'roi_3_year': None,
'recommendation': 'Negative ROI - cost increase'
}
else:
payback_period = implementation_costs / annual_savings if annual_savings > 0 else None
roi_3_year = (annual_savings * 3) - implementation_costs
roi = {
'annual_savings': annual_savings,
'implementation_cost': implementation_costs,
'payback_period_years': round(payback_period, 2) if payback_period else None,
'roi_3_year': round(roi_3_year, 2),
'recommendation': 'Positive ROI' if payback_period and payback_period < 2 else 'ROI marginal'
}
return roi
def committee_vote(self, request_id, vote_data):
"""
Record committee vote on product
Parameters:
- request_id: Request ID
- vote_data: Dict with voting results
"""
if request_id not in self.requests:
raise ValueError(f"Request {request_id} not found")
request = self.requests[request_id]
vote_results = {
'request_id': request_id,
'vote_date': datetime.now(),
'votes_for': vote_data['votes_for'],
'votes_against': vote_data['votes_against'],
'abstentions': vote_data['abstentions'],
'decision': 'approved' if vote_data['votes_for'] > vote_data['votes_against'] else 'denied',
'conditions': vote_data.get('conditions', []),
'implementation_plan': vote_data.get('implementation_plan')
}
self.evaluations[request_id]['committee_vote'] = vote_results
# Update request status
if vote_results['decision'] == 'approved':
request.status = RequestStatus.APPROVED
# Add to formulary
self._add_to_formulary(request)
else:
request.status = RequestStatus.DENIED
return vote_results
def _add_to_formulary(self, request):
"""Add approved product to formulary"""
self.formulary[request.product_name] = {
'product_name': request.product_name,
'manufacturer': request.manufacturer,
'category': request.category,
'approval_date': datetime.now(),
'request_id': request.request_id,
'status': 'active'
}
def generate_executive_summary(self, request_id):
"""
Generate executive summary for committee review
"""
if request_id not in self.requests:
raise ValueError(f"Request {request_id} not found")
request = self.requests[request_id]
evaluation = self.evaluations.get(request_id, {})
summary = {
'request_id': request_id,
'product_name': request.product_name,
'manufacturer': request.manufacturer,
'category': request.category.value,
'submitted_by': request.submitted_by,
'submission_date': request.submission_date,
'clinical_justification': request.clinical_justification,
'trial_results': evaluation.get('results_summary'),
'financial_analysis': evaluation.get('financial_analysis'),
'recommendation': self._generate_recommendation(evaluation),
'status': request.status.value
}
return summary
def _generate_recommendation(self, evaluation):
"""Generate recommendation based on trial and financial data"""
if not evaluation:
return "Insufficient data for recommendation"
trial_results = evaluation.get('results_summary', {})
financial = evaluation.get('financial_analysis', {})
# Clinical criteria
clinical_success = trial_results.get('clinical_success_rate', 0) >= 90
user_satisfaction = trial_results.get('would_recommend_pct', 0) >= 70
# Financial criteria
if financial and financial.get('comparison'):
cost_favorable = financial['comparison']['annual_difference'] <= 0
else:
cost_favorable = True # If no comparison, assume neutral
# Recommendation logic
if clinical_success and user_satisfaction and cost_favorable:
return "APPROVE - Meets clinical and financial criteria"
elif clinical_success and user_satisfaction and not cost_favorable:
return "CONDITIONAL APPROVAL - Clinical benefits may justify cost increase"
elif not clinical_success or not user_satisfaction:
return "DENY - Does not meet clinical criteria"
else:
return "DEFER - Additional evaluation needed"
# Example usage
vac = ValueAnalysisCommittee(organization_name="Memorial Hospital System")
# Submit request
request_data = {
'submitted_by': 'Dr. Sarah Johnson, Orthopedic Surgery',
'product_name': 'NextGen Hip Implant System',
'manufacturer': 'Advanced Orthopedics Inc',
'category': 'IMPLANT',
'clinical_justification': 'New ceramic-on-ceramic bearing surface shows reduced wear in published studies. Lower dislocation rates reported. Improved patient outcomes expected.',
'estimated_annual_volume': 150,
'estimated_unit_cost': 4200,
'urgency': 'ROUTINE',
'current_alternative': 'Current Hip System A',
'physician_champion': 'Dr. Sarah Johnson'
}
request = vac.submit_request(request_data)
print(f"Request submitted: {request.request_id}")
# Initial screening
screening = vac.initial_screening(request.request_id)
print(f"Screening result: {screening['passes_screening']}")
# Conduct trial
trial_params = {
'duration_days': 90,
'trial_sites': ['Main Campus OR'],
'sample_size': 20,
'criteria': ['Clinical success', 'User satisfaction', 'Complications', 'Ease of use'],
'physicians': ['Dr. Johnson', 'Dr. Smith', 'Dr. Williams']
}
trial = vac.conduct_clinical_trial(request.request_id, trial_params)
print(f"Clinical trial initiated: {trial['sample_size']} cases")
# Collect trial results (simulated)
trial_results = [
{
'case_number': i+1,
'clinical_success': True,
'user_satisfaction': np.random.randint(7, 11), # 7-10 scale
'complication': False,
'ease_of_use': np.random.randint(7, 11),
'would_recommend': True
}
for i in range(20)
]
results_summary = vac.collect_trial_results(request.request_id, trial_results)
print(f"\nTrial Results:")
print(f" Clinical success rate: {results_summary['clinical_success_rate']:.1f}%")
print(f" Would recommend: {results_summary['would_recommend_pct']:.1f}%")
# Financial analysis
financial_data = {
'training_cost': 5000,
'storage_cost_per_unit': 10,
'waste_rate': 0.01,
'complication_rate': 0.02,
'complication_cost_per_event': 15000,
'disposal_cost_per_unit': 50,
'current_alternative_tco': {
'total_annual_cost': 150 * 4000 # Current product costs less upfront
},
'implementation_cost': 10000
}
financial_analysis = vac.financial_analysis(request.request_id, financial_data)
print(f"\nFinancial Analysis:")
print(f" New product TCO: ${financial_analysis['tco_analysis']['total_annual_cost']:,.2f}")
print(f" Annual difference: ${financial_analysis['comparison']['annual_difference']:,.2f}")
if financial_analysis['roi_analysis']:
print(f" Payback period: {financial_analysis['roi_analysis']['payback_period_years']} years")
# Committee vote
vote_data = {
'votes_for': 8,
'votes_against': 1,
'abstentions': 1,
'conditions': ['Monitor complications for first 50 cases', 'Quarterly utilization review'],
'implementation_plan': 'Phased rollout - 3 surgeons initially, expand after 30 cases'
}
vote = vac.committee_vote(request.request_id, vote_data)
print(f"\nCommittee Vote: {vote['decision'].upper()}")
print(f" For: {vote['votes_for']}, Against: {vote['votes_against']}, Abstain: {vote['abstentions']}")
# Executive summary
summary = vac.generate_executive_summary(request.request_id)
print(f"\nRecommendation: {summary['recommendation']}")
def analyze_standardization_opportunity(usage_data_df):
"""
Identify product standardization opportunities
Parameters:
- usage_data_df: DataFrame with columns:
- category: Product category
- product_name: Product identifier
- manufacturer: Manufacturer
- annual_volume: Units used
- unit_cost: Cost per unit
- user: Department or physician using
"""
results = []
# Group by category
for category in usage_data_df['category'].unique():
category_data = usage_data_df[usage_data_df['category'] == category]
# Calculate fragmentation
num_products = category_data['product_name'].nunique()
num_manufacturers = category_data['manufacturer'].nunique()
num_users = category_data['user'].nunique()
# Calculate spend
total_spend = (category_data['annual_volume'] * category_data['unit_cost']).sum()
# Identify top products
product_spend = category_data.groupby('product_name').apply(
lambda x: (x['annual_volume'] * x['unit_cost']).sum()
).sort_values(ascending=False)
top_2_spend = product_spend.head(2).sum()
top_2_coverage = (top_2_spend / total_spend * 100) if total_spend > 0 else 0
# Standardization opportunity score
# Higher score = more opportunity
fragmentation_score = min(num_products / 5, 1.0) * 30 # Max 30 points
manufacturer_diversity = min(num_manufacturers / 3, 1.0) * 20 # Max 20 points
spend_concentration = (100 - top_2_coverage) / 100 * 30 # Max 30 points if very fragmented
spend_magnitude = min(total_spend / 100000, 1.0) * 20 # Max 20 points if >$100K
opportunity_score = fragmentation_score + manufacturer_diversity + spend_concentration + spend_magnitude
# Recommendation
if opportunity_score >= 60:
recommendation = "HIGH PRIORITY - Significant standardization opportunity"
elif opportunity_score >= 40:
recommendation = "MEDIUM PRIORITY - Moderate opportunity"
else:
recommendation = "LOW PRIORITY - Already standardized or low impact"
# Potential savings (assume 15% from standardization)
potential_savings = total_spend * 0.15
results.append({
'category': category,
'num_products': num_products,
'num_manufacturers': num_manufacturers,
'num_users': num_users,
'total_annual_spend': round(total_spend, 2),
'top_2_coverage_pct': round(top_2_coverage, 1),
'opportunity_score': round(opportunity_score, 1),
'potential_savings': round(potential_savings, 2),
'recommendation': recommendation
})
results_df = pd.DataFrame(results)
results_df = results_df.sort_values('opportunity_score', ascending=False)
return results_df
# Example usage
usage_data = pd.DataFrame({
'category': ['Surgical Gloves'] * 8 + ['Hip Implants'] * 6 + ['IV Catheters'] * 4,
'product_name': [
'Glove-A', 'Glove-B', 'Glove-C', 'Glove-D', 'Glove-E', 'Glove-F', 'Glove-G', 'Glove-H',
'Hip-System-A', 'Hip-System-B', 'Hip-System-C', 'Hip-System-D', 'Hip-System-E', 'Hip-System-F',
'IV-Cath-A', 'IV-Cath-B', 'IV-Cath-A', 'IV-Cath-B'
],
'manufacturer': [
'MfgA', 'MfgB', 'MfgC', 'MfgD', 'MfgA', 'MfgE', 'MfgF', 'MfgG',
'Ortho-A', 'Ortho-B', 'Ortho-C', 'Ortho-A', 'Ortho-D', 'Ortho-E',
'IV-Mfg-A', 'IV-Mfg-B', 'IV-Mfg-A', 'IV-Mfg-B'
],
'annual_volume': [
10000, 8000, 5000, 3000, 2000, 1500, 1000, 500,
80, 60, 30, 20, 15, 10,
15000, 12000, 10000, 8000
],
'unit_cost': [
0.35, 0.38, 0.34, 0.40, 0.36, 0.39, 0.37, 0.41,
4000, 4200, 3800, 4100, 4500, 3900,
8.50, 8.75, 8.50, 8.75
],
'user': [
'OR', 'OR', 'ER', 'ICU', 'Peds', 'NICU', 'Cath Lab', 'Clinic',
'Dr. Smith', 'Dr. Johnson', 'Dr. Williams', 'Dr. Brown', 'Dr. Davis', 'Dr. Wilson',
'All Units', 'All Units', 'All Units', 'All Units'
]
})
standardization_analysis = analyze_standardization_opportunity(usage_data)
print("Standardization Opportunity Analysis:")
print(standardization_analysis[['category', 'num_products', 'total_annual_spend',
'opportunity_score', 'potential_savings', 'recommendation']])
print(f"\nTotal potential savings: ${standardization_analysis['potential_savings'].sum():,.2f}")
class ClinicalEvidenceEvaluator:
"""
Evaluate clinical evidence for product decisions
"""
def __init__(self):
self.evidence_levels = {
'Level 1': 'Systematic review/meta-analysis of RCTs',
'Level 2': 'Individual randomized controlled trial (RCT)',
'Level 3': 'Controlled trial without randomization',
'Level 4': 'Case-control or cohort study',
'Level 5': 'Systematic review of descriptive/qualitative studies',
'Level 6': 'Single descriptive or qualitative study',
'Level 7': 'Expert opinion'
}
def evaluate_evidence(self, product_name, evidence_list):
"""
Evaluate quality of clinical evidence
Parameters:
- product_name: Product being evaluated
- evidence_list: List of studies/evidence with metadata
"""
if not evidence_list:
return {
'product_name': product_name,
'evidence_strength': 'Insufficient',
'recommendation': 'Additional evidence required',
'studies_count': 0
}
# Categorize by evidence level
evidence_by_level = {}
for evidence in evidence_list:
level = evidence.get('evidence_level', 'Level 7')
if level not in evidence_by_level:
evidence_by_level[level] = []
evidence_by_level[level].append(evidence)
# Determine overall evidence strength
if 'Level 1' in evidence_by_level or (
'Level 2' in evidence_by_level and len(evidence_by_level['Level 2']) >= 2
):
evidence_strength = 'Strong'
recommendation = 'Supported by high-quality evidence'
elif 'Level 2' in evidence_by_level or 'Level 3' in evidence_by_level:
evidence_strength = 'Moderate'
recommendation = 'Supported by moderate evidence'
elif any(level in evidence_by_level for level in ['Level 4', 'Level 5', 'Level 6']):
evidence_strength = 'Weak'
recommendation = 'Limited evidence - proceed with caution'
else:
evidence_strength = 'Insufficient'
recommendation = 'Insufficient evidence - require clinical trial'
evaluation = {
'product_name': product_name,
'evidence_strength': evidence_strength,
'studies_count': len(evidence_list),
'evidence_by_level': {
level: len(studies) for level, studies in evidence_by_level.items()
},
'recommendation': recommendation,
'highest_evidence_level': min(evidence_by_level.keys(), key=lambda x: int(x.split()[1])) if evidence_by_level else None
}
return evaluation
def comparative_effectiveness_analysis(self, product_a, product_b,
outcome_data_a, outcome_data_b):
"""
Compare clinical effectiveness of two products
Parameters:
- product_a, product_b: Product names
- outcome_data_a, outcome_data_b: Clinical outcome data
"""
# Calculate outcome metrics
metrics_a = self._calculate_outcomes(outcome_data_a)
metrics_b = self._calculate_outcomes(outcome_data_b)
# Compare
comparison = {}
for metric in metrics_a.keys():
if metric in metrics_b:
comparison[metric] = {
'product_a': metrics_a[metric],
'product_b': metrics_b[metric],
'difference': metrics_a[metric] - metrics_b[metric],
'better_product': product_a if metrics_a[metric] > metrics_b[metric] else product_b
}
# Determine clinical superiority
a_better_count = sum(1 for c in comparison.values() if c['better_product'] == product_a)
b_better_count = sum(1 for c in comparison.values() if c['better_product'] == product_b)
if a_better_count > b_better_count:
conclusion = f"{product_a} demonstrates superior clinical outcomes"
elif b_better_count > a_better_count:
conclusion = f"{product_b} demonstrates superior clinical outcomes"
else:
conclusion = "Products demonstrate equivalent clinical outcomes"
return {
'product_a': product_a,
'product_b': product_b,
'comparison': comparison,
'conclusion': conclusion
}
def _calculate_outcomes(self, outcome_data):
"""Calculate outcome metrics from raw data"""
if not outcome_data:
return {}
metrics = {
'success_rate': (outcome_data.get('successes', 0) / outcome_data.get('total_cases', 1)) * 100,
'complication_rate': (outcome_data.get('complications', 0) / outcome_data.get('total_cases', 1)) * 100,
'readmission_rate': (outcome_data.get('readmissions', 0) / outcome_data.get('total_cases', 1)) * 100,
'patient_satisfaction': outcome_data.get('satisfaction_score', 0)
}
return metrics
# Example usage
evaluator = ClinicalEvidenceEvaluator()
# Evidence evaluation
evidence = [
{'title': 'RCT of NextGen vs Standard Hip', 'evidence_level': 'Level 2', 'n': 200, 'conclusion': 'Favorable'},
{'title': 'Meta-analysis ceramic bearings', 'evidence_level': 'Level 1', 'n': 1500, 'conclusion': 'Reduced wear'},
{'title': 'Retrospective cohort study', 'evidence_level': 'Level 4', 'n': 500, 'conclusion': 'Lower dislocation'}
]
evidence_eval = evaluator.evaluate_evidence('NextGen Hip Implant', evidence)
print("Clinical Evidence Evaluation:")
print(f" Evidence Strength: {evidence_eval['evidence_strength']}")
print(f" Studies: {evidence_eval['studies_count']}")
print(f" Recommendation: {evidence_eval['recommendation']}")
# Comparative effectiveness
outcomes_new = {
'total_cases': 150,
'successes': 145,
'complications': 3,
'readmissions': 2,
'satisfaction_score': 9.2
}
outcomes_current = {
'total_cases': 150,
'successes': 140,
'complications': 8,
'readmissions': 5,
'satisfaction_score': 8.5
}
comparison = evaluator.comparative_effectiveness_analysis(
'NextGen Hip', 'Current Hip System',
outcomes_new, outcomes_current
)
print(f"\n{comparison['conclusion']}")
Key Principles:
def physician_preference_analysis(ppi_usage_df):
"""
Analyze physician preference item (PPI) variation
Parameters:
- ppi_usage_df: DataFrame with:
- physician: Physician name
- procedure_type: Type of procedure
- product_used: Product selected
- unit_cost: Cost of product
- outcome: Clinical outcome (success/complication)
"""
analysis = []
# Group by procedure type
for procedure in ppi_usage_df['procedure_type'].unique():
proc_data = ppi_usage_df[ppi_usage_df['procedure_type'] == procedure]
# Product variation by physician
product_variation = proc_data.groupby('physician')['product_used'].nunique()
avg_products_per_physician = product_variation.mean()
# Cost variation
cost_by_physician = proc_data.groupby('physician')['unit_cost'].mean()
cost_std_dev = cost_by_physician.std()
cost_range = cost_by_physician.max() - cost_by_physician.min()
# Outcome analysis
outcome_by_product = proc_data.groupby('product_used').apply(
lambda x: (x['outcome'] == 'success').sum() / len(x) * 100
)
# Cost vs. outcome correlation
# Group by product, get avg cost and outcome rate
product_analysis = proc_data.groupby('product_used').agg({
'unit_cost': 'mean',
'outcome': lambda x: (x == 'success').sum() / len(x) * 100
}).reset_index()
product_analysis.columns = ['product', 'avg_cost', 'success_rate']
# Identify opportunities
if cost_range > 1000 and cost_std_dev > 500:
opportunity = "HIGH - Significant cost variation without clinical justification"
priority = 1
elif cost_range > 500:
opportunity = "MEDIUM - Moderate cost variation"
priority = 2
else:
opportunity = "LOW - Minimal variation"
priority = 3
analysis.append({
'procedure_type': procedure,
'num_physicians': proc_data['physician'].nunique(),
'num_products': proc_data['product_used'].nunique(),
'avg_products_per_physician': round(avg_products_per_physician, 1),
'cost_range': round(cost_range, 2),
'cost_std_dev': round(cost_std_dev, 2),
'opportunity': opportunity,
'priority': priority,
'engagement_strategy': 'Physician-led review with outcome data' if priority <= 2 else 'Monitor'
})
analysis_df = pd.DataFrame(analysis)
analysis_df = analysis_df.sort_values('priority')
return analysis_df
# Example usage
ppi_usage = pd.DataFrame({
'physician': ['Dr. Smith'] * 30 + ['Dr. Johnson'] * 30 + ['Dr. Williams'] * 30,
'procedure_type': ['Total Hip Arthroplasty'] * 90,
'product_used': (
['Hip-System-A'] * 30 +
['Hip-System-B'] * 20 + ['Hip-System-C'] * 10 +
['Hip-System-A'] * 15 + ['Hip-System-D'] * 15
),
'unit_cost': (
[4000] * 30 +
[4200] * 20 + [5500] * 10 +
[4000] * 15 + [6000] * 15
),
'outcome': np.random.choice(['success', 'complication'], 90, p=[0.95, 0.05])
})
ppi_analysis = physician_preference_analysis(ppi_usage)
print("Physician Preference Item Analysis:")
print(ppi_analysis[['procedure_type', 'num_physicians', 'num_products',
'cost_range', 'opportunity', 'engagement_strategy']])
def calculate_value_analysis_kpis(va_data, implementation_data):
"""
Calculate value analysis program KPIs
Parameters:
- va_data: Value analysis requests and decisions
- implementation_data: Implementation tracking data
"""
kpis = {}
# Request throughput
if 'submission_date' in va_data.columns and 'decision_date' in va_data.columns:
va_data['cycle_time_days'] = (va_data['decision_date'] - va_data['submission_date']).dt.days
kpis['avg_cycle_time_days'] = va_data['cycle_time_days'].mean()
# Approval rate
if 'decision' in va_data.columns:
kpis['approval_rate'] = (va_data['decision'] == 'approved').sum() / len(va_data) * 100
# Cost savings realized
if 'projected_savings' in implementation_data.columns and 'actual_savings' in implementation_data.columns:
kpis['projected_savings'] = implementation_data['projected_savings'].sum()
kpis['actual_savings'] = implementation_data['actual_savings'].sum()
kpis['savings_realization_rate'] = (kpis['actual_savings'] / kpis['projected_savings'] * 100) if kpis['projected_savings'] > 0 else 0
# Standardization progress
if 'sku_count_before' in implementation_data.columns and 'sku_count_after' in implementation_data.columns:
total_sku_reduction = (implementation_data['sku_count_before'] - implementation_data['sku_count_after']).sum()
kpis['sku_reduction'] = total_sku_reduction
# Contract compliance
if 'contract_compliant' in implementation_data.columns:
kpis['contract_compliance_rate'] = (implementation_data['contract_compliant'].sum() / len(implementation_data) * 100)
# Physician engagement
if 'physician_champion_assigned' in va_data.columns:
kpis['physician_engagement_rate'] = (va_data['physician_champion_assigned'].sum() / len(va_data) * 100)
# Format KPIs
for key in kpis:
if 'rate' in key or key == 'savings_realization_rate':
kpis[key] = round(kpis[key], 2)
elif 'savings' in key:
kpis[key] = round(kpis[key], 2)
elif 'reduction' in key:
kpis[key] = int(kpis[key])
else:
kpis[key] = round(kpis[key], 1)
return kpis
# Example data
va_requests = pd.DataFrame({
'request_id': range(1, 51),
'submission_date': pd.date_range('2023-01-01', periods=50, freq='7D'),
'decision_date': pd.date_range('2023-01-01', periods=50, freq='7D') + pd.Timedelta(days=45),
'decision': np.random.choice(['approved', 'denied'], 50, p=[0.70, 0.30]),
'physician_champion_assigned': np.random.choice([True, False], 50, p=[0.80, 0.20])
})
implementation = pd.DataFrame({
'initiative_id': range(1, 36), # 35 approved items implemented
'projected_savings': np.random.randint(10000, 100000, 35),
'actual_savings': np.random.randint(8000, 95000, 35),
'sku_count_before': np.random.randint(3, 10, 35),
'sku_count_after': np.random.randint(1, 3, 35),
'contract_compliant': np.random.choice([True, False], 35, p=[0.90, 0.10])
})
# Align actual savings to be roughly 80-90% of projected
implementation['actual_savings'] = (implementation['projected_savings'] * np.random.uniform(0.75, 0.95, 35)).astype(int)
kpis = calculate_value_analysis_kpis(va_requests, implementation)
print("Value Analysis Program KPIs:")
for metric, value in kpis.items():
if 'savings' in metric:
print(f" {metric}: ${value:,.2f}")
elif 'rate' in metric:
print(f" {metric}: {value}%")
else:
print(f" {metric}: {value}")
Value Analysis Platforms:
Decision Support:
Data Analytics:
Data Analysis:
pandas: Data manipulationnumpy: Numerical computingscipy: Statistical analysisstatsmodels: Statistical modelingOptimization:
pulp: Linear programmingscipy.optimize: Optimization algorithmsVisualization:
matplotlib, seaborn: Chartsplotly: Interactive dashboardsProblem:
Solutions:
Problem:
Solutions:
Problem:
Solutions:
Problem:
Solutions:
Problem:
Solutions:
Problem:
Solutions:
Product Evaluation Summary
Product: NextGen Hip Implant System Manufacturer: Advanced Orthopedics Inc Request ID: VAR-00245 Submitted by: Dr. Sarah Johnson, Orthopedic Surgery Decision Date: February 15, 2024
Clinical Evidence:
Clinical Trial Results:
Financial Analysis:
| Cost Component | New Product | Current | Difference | |----------------|-------------|---------|------------| | Acquisition Cost | $630,000 | $600,000 | +$30,000 | | Training Cost | $5,000 | $0 | +$5,000 | | Complication Cost | $4,500 | $40,500 | -$36,000 | | Total Annual Cost | $639,500 | $640,500 | -$1,000 | | Cost per Use | $4,263 | $4,270 | -$7 |
ROI Analysis:
Committee Recommendation:
APPROVED (8-1-1 vote)
Rationale:
Conditions:
Implementation Plan:
Projected Impact:
If you need more context:
documentation
When the user wants to optimize yard operations, manage trailer parking, or improve dock door utilization. Also use when the user mentions "yard management," "trailer tracking," "yard jockey," "drop trailer program," "trailer pool," "dock scheduling," or "gate management." For cross-dock operations, see cross-docking. For warehouse design, see warehouse-design.
tools
When the user wants to optimize workforce scheduling, create shift plans, or balance labor demand. Also use when the user mentions "staff scheduling," "labor planning," "shift optimization," "crew scheduling," "roster optimization," or "employee scheduling." For task assignment, see task-assignment-problem. For wave planning labor, see wave-planning-optimization.
testing
When the user wants to optimize pick wave planning, schedule warehouse operations, or improve order fulfillment efficiency. Also use when the user mentions "wave management," "batch picking," "pick wave scheduling," "order release optimization," "wave design," or "pick wave strategy." For order batching, see order-batching-optimization. For workforce scheduling, see workforce-scheduling.
testing
When the user wants to optimize warehouse slot assignments, improve pick efficiency, or design warehouse layouts. Also use when the user mentions "slotting optimization," "slot assignment," "ABC slotting," "pick path optimization," "storage location assignment," "warehouse layout optimization," or "forward pick locations." For picker routing, see picker-routing-optimization. For warehouse design, see warehouse-design.