FDA 483 Compliance Workflows

Practical workflows for using CTWise 483 Intelligence in your quality and compliance operations.

---

Workflow 1: Supplier Due Diligence

Scenario: Your sourcing team is evaluating a new contract manufacturing organization (CMO). You need to assess their FDA inspection track record and compliance posture.

Step 1: Get Facility Profile

import requests

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://api.ctwise.ai/v1"

# Lookup facility by FEI number (provided by supplier)
fei_number = "3016004437"
response = requests.get(
    f"{BASE_URL}/483/facilities/{fei_number}",
    headers={"X-Api-Key": API_KEY}
)

facility = response.json()
print(f"Facility: {facility['legal_name']}")
print(f"Location: {facility['city']}, {facility['state']}")
print(f"Total Inspections: {facility['total_inspections']}")
print(f"\nClassification Breakdown:")
print(f"  OAI: {facility['classification_breakdown']['oai_count']}")
print(f"  VAI: {facility['classification_breakdown']['vai_count']}")
print(f"  NAI: {facility['classification_breakdown']['nai_count']}")

Step 2: Calculate Risk Score

# Get risk score
response = requests.get(
    f"{BASE_URL}/483/risk-scores/{fei_number}",
    headers={"X-Api-Key": API_KEY}
)

risk = response.json()
print(f"\nRisk Score: {risk['risk_score']:.1f}/100")
print(f"Risk Level: {risk['risk_level']}")
print(f"\nFactor Breakdown:")
for factor, value in risk['factors'].items():
    print(f"  {factor}: {value:.1f}")

Step 3: Review Citation History

# Get last 20 citations
response = requests.get(
    f"{BASE_URL}/483/facilities/{fei_number}/citations",
    headers={"X-Api-Key": API_KEY},
    params={"limit": 20, "sort_order": "desc"}
)

citations = response.json()
print(f"\nRecent Citations ({citations['total']} total):")
for citation in citations["results"][:10]:
    print(f"\n{citation['inspection_end_date']}:")
    print(f"  CFR: {citation['act_cfr_number']}")
    print(f"  Category: {citation['category']}")
    print(f"  {citation['short_description'][:80]}...")

Step 4: Compare to Benchmarks

# Get peer benchmarks
response = requests.get(
    f"{BASE_URL}/483/analytics/benchmarks",
    headers={"X-Api-Key": API_KEY},
    params={"product_type": facility['product_type']}
)

benchmarks = response.json()
print(f"\nPeer Comparison:")
print(f"  This facility: {risk['risk_score']:.1f}")
print(f"  Peer median: {benchmarks['risk_score_stats']['median']:.1f}")
print(f"  Peer 75th percentile: {benchmarks['risk_score_stats']['p75']:.1f}")

# Decision logic
if risk['risk_score'] > benchmarks['risk_score_stats']['p75']:
    print("\n⚠️ HIGH RISK: Above peer 75th percentile - requires audit")
elif risk['risk_score'] > benchmarks['risk_score_stats']['median']:
    print("\n⚡ MODERATE RISK: Above peer median - requires detailed review")
else:
    print("\n✅ LOW RISK: Below peer median - standard onboarding")

Decision Matrix

Risk Score	OAI Count	Last OAI Age	Recommendation
Less than 25	0	N/A	✅ Approve - low risk
25-50	0-1	Over 2 years	⚡ Approve with monitoring
50-75	1-2	Under 2 years	⚠️ Conditional approval - require audit
Over 75	2+	Under 1 year	❌ Reject or escalate to senior management

---

Workflow 2: Proactive Facility Risk Monitoring

Scenario: Your quality team manages a watchlist of 50 key suppliers and internal facilities. You want to monitor them for new inspection activity and trending violations.

Step 1: Create Watchlist

# Add facilities to watchlist
facilities_to_monitor = [
    {"fei": "3016004437", "notes": "Key API supplier - sterile manufacturing"},
    {"fei": "3005012345", "notes": "Contract packager - secondary packaging"},
    {"fei": "3012007890", "notes": "Internal facility - Boston biologics plant"}
]

for facility in facilities_to_monitor:
    response = requests.post(
        f"{BASE_URL}/483/watchlist",
        headers={
            "X-Api-Key": API_KEY,
            "Content-Type": "application/json"
        },
        json={
            "fei_number": facility['fei'],
            "notes": facility['notes'],
            "alert_on_new_inspection": True
        }
    )
    print(f"Added: {facility['notes']}")

Step 2: Weekly Risk Check

# Get current watchlist
response = requests.get(
    f"{BASE_URL}/483/watchlist",
    headers={"X-Api-Key": API_KEY}
)

watchlist = response.json()
print(f"Watchlist Summary ({watchlist['total']} facilities):\n")

high_risk_count = 0
for entry in watchlist["results"]:
    if entry['current_risk_score'] > 50:
        high_risk_count += 1
        print(f"⚠️ {entry['facility_name']} (FEI: {entry['fei_number']})")
        print(f"   Risk: {entry['current_risk_score']:.1f}/100")
        print(f"   Last inspection: {entry['last_inspection_date']}")
        print(f"   Notes: {entry['notes']}\n")

if high_risk_count > 0:
    print(f"\n📧 ALERT: {high_risk_count} high-risk facilities require attention")

Step 3: Trending Citation Analysis

# Check for trending violations
response = requests.get(
    f"{BASE_URL}/483/citations/trending",
    headers={"X-Api-Key": API_KEY},
    params={"trend": "increasing", "limit": 10}
)

trending = response.json()
print("\n📈 Trending Violations (Increasing Frequency):\n")

for item in trending["results"]:
    print(f"{item['act_cfr_number']}: {item['description']}")
    print(f"  Recent: {item['recent_count']} | Previous: {item['previous_count']}")
    print(f"  Trend: +{item['trend_percentage']:.1f}%\n")

Automation: Weekly Report

import datetime

def generate_weekly_report():
    """Generate automated weekly 483 monitoring report"""

    print(f"FDA 483 Intelligence Weekly Report")
    print(f"Generated: {datetime.datetime.now().strftime('%Y-%m-%d')}\n")
    print("=" * 60)

    # Section 1: Watchlist Risk Summary
    watchlist = requests.get(
        f"{BASE_URL}/483/watchlist",
        headers={"X-Api-Key": API_KEY}
    ).json()

    risk_counts = {"low": 0, "medium": 0, "high": 0, "critical": 0}
    for entry in watchlist["results"]:
        score = entry['current_risk_score']
        if score < 25: risk_counts["low"] += 1
        elif score < 50: risk_counts["medium"] += 1
        elif score < 75: risk_counts["high"] += 1
        else: risk_counts["critical"] += 1

    print("\n1. WATCHLIST RISK DISTRIBUTION")
    print(f"   Total facilities: {watchlist['total']}")
    print(f"   Low risk: {risk_counts['low']}")
    print(f"   Medium risk: {risk_counts['medium']}")
    print(f"   High risk: {risk_counts['high']} ⚠️")
    print(f"   Critical risk: {risk_counts['critical']} 🚨")

    # Section 2: Trending Citations
    trending = requests.get(
        f"{BASE_URL}/483/citations/trending",
        headers={"X-Api-Key": API_KEY},
        params={"trend": "increasing", "limit": 5}
    ).json()

    print("\n2. TRENDING CITATIONS (Top 5 Increasing)")
    for i, item in enumerate(trending["results"][:5], 1):
        print(f"   {i}. {item['act_cfr_number']} (+{item['trend_percentage']:.1f}%)")

    # Section 3: Action Items
    print("\n3. RECOMMENDED ACTIONS")
    if risk_counts["critical"] > 0:
        print(f"   🚨 Immediate: Review {risk_counts['critical']} critical-risk facilities")
    if risk_counts["high"] > 0:
        print(f"   ⚠️ This week: Assess {risk_counts['high']} high-risk facilities")

    print("\n" + "=" * 60)

# Run weekly (via cron job or scheduled Lambda)
generate_weekly_report()

---

Workflow 3: Pre-Inspection Preparation

Scenario: Your Boston manufacturing facility is scheduled for an FDA pre-approval inspection (PAI) in 4 weeks. You want to prepare by understanding common inspection findings for similar facilities.

Step 1: Find Similar Facilities

# Search for observations from similar facilities (drug manufacturing, sterile products)
response = requests.post(
    f"{BASE_URL}/483/observations/search",
    headers={
        "X-Api-Key": API_KEY,
        "Content-Type": "application/json"
    },
    json={
        "query": "sterile drug manufacturing aseptic processing",
        "filters": {
            "product_type": "drug",
            "classification": ["OAI", "VAI"],
            "program_area": "Human Drugs"
        },
        "top_k": 50
    }
)

observations = response.json()
print(f"Found {observations['total']} relevant observations\n")

Step 2: Analyze Common Citations

from collections import Counter

# Count CFR citations
cfr_counts = Counter()
category_counts = Counter()

for obs in observations["results"]:
    cfr_counts[obs['act_cfr_number']] += 1
    category_counts[obs['category']] += 1

print("Most Common CFR Citations:")
for cfr, count in cfr_counts.most_common(10):
    print(f"  {cfr}: {count} occurrences")

print("\nMost Common Categories:")
for category, count in category_counts.most_common(5):
    print(f"  {category}: {count} occurrences")

Step 3: Build Inspection Prep Checklist

# Get detailed findings for top CFRs
top_cfrs = [cfr for cfr, _ in cfr_counts.most_common(5)]

print("\nINSPECTION PREP CHECKLIST\n")
print("=" * 60)

for cfr in top_cfrs:
    # Get CFR details
    response = requests.get(
        f"{BASE_URL}/483/cfr-references/{cfr}",
        headers={"X-Api-Key": API_KEY},
        params={"cfr": cfr}
    )

    cfr_detail = response.json()
    print(f"\n{cfr}: {cfr_detail['short_description']}")
    print(f"Frequency: {cfr_detail['occurrence_count']} citations")

    # Search for specific observations
    response = requests.post(
        f"{BASE_URL}/483/observations/search",
        headers={
            "X-Api-Key": API_KEY,
            "Content-Type": "application/json"
        },
        json={
            "query": f"{cfr} violations",
            "filters": {"cfr_reference": cfr},
            "top_k": 3
        }
    )

    examples = response.json()
    print("Common issues:")
    for example in examples["results"][:3]:
        print(f"  - {example['short_description'][:80]}...")

    print("\n✅ Recommended actions:")
    # Placeholder - in production, this would map to internal SOP numbers
    print(f"  □ Review SOPs for {cfr}")
    print(f"  □ Audit compliance with {cfr}")
    print(f"  □ Document corrective actions for any gaps")

print("\n" + "=" * 60)

---

Workflow 4: Root Cause Investigation

Scenario: Your facility just received an OAI classification with 8 observations citing 21 CFR 211.113 (aseptic processing). You need to understand industry patterns and develop a comprehensive response.

Step 1: Research Industry Precedent

# Search for similar violations
response = requests.post(
    f"{BASE_URL}/483/observations/search",
    headers={
        "X-Api-Key": API_KEY,
        "Content-Type": "application/json"
    },
    json={
        "query": "aseptic processing environmental monitoring 21 CFR 211.113",
        "filters": {
            "classification": ["OAI"],
            "category": "Sterility Assurance"
        },
        "top_k": 20
    }
)

similar = response.json()
print(f"Similar Industry Violations: {similar['total']}\n")

for obs in similar["results"][:5]:
    print(f"{obs['facility_name']} ({obs['inspection_end_date']})")
    print(f"  {obs['short_description'][:100]}...")
    print(f"  Similarity: {obs['similarity_score']:.2f}\n")

Step 2: Analyze Regulatory Mapping

# Get ICH E6(R3) cross-references
response = requests.get(
    f"{BASE_URL}/483/regulatory-mapping/21 CFR 211.113",
    headers={"X-Api-Key": API_KEY}
)

mapping = response.json()
print("Regulatory Context:")
print(f"CFR: {mapping['cfr_section']}")
print(f"Title: {mapping['cfr_title']}")
print(f"\nRelated ICH E6(R3) Requirements:")

for ich in mapping["ich_mappings"]:
    print(f"  {ich['ich_section']}: {ich['matched_requirement']}")
    print(f"  Relevance: {ich['relevance_score']:.2f}\n")

Step 3: Build Response Strategy

print("CORRECTIVE ACTION PLAN BUILDER\n")
print("=" * 60)

# Categorize similar violations by root cause
root_causes = {}
for obs in similar["results"]:
    # Simple keyword-based categorization (in production, use NLP)
    description = obs['short_description'].lower()
    if "calibration" in description:
        root_causes.setdefault("Calibration", []).append(obs)
    elif "monitoring" in description or "sampling" in description:
        root_causes.setdefault("Environmental Monitoring", []).append(obs)
    elif "validation" in description:
        root_causes.setdefault("Process Validation", []).append(obs)
    else:
        root_causes.setdefault("Other", []).append(obs)

for cause, observations in root_causes.items():
    print(f"\n{cause} ({len(observations)} industry examples)")
    print(f"  Recommended CAPA:")
    if cause == "Calibration":
        print(f"    1. Review calibration program for all equipment in ISO 5 areas")
        print(f"    2. Update calibration SOPs with frequency requirements")
        print(f"    3. Implement automated calibration tracking system")
    elif cause == "Environmental Monitoring":
        print(f"    1. Enhance environmental monitoring program")
        print(f"    2. Increase viable sampling frequency during operations")
        print(f"    3. Implement real-time particulate monitoring")

print("\n" + "=" * 60)

---

Best Practices

1. API Rate Limiting

import time

def rate_limited_batch_query(fei_numbers, delay=0.5):
    """Query multiple facilities with rate limiting"""
    results = []
    for fei in fei_numbers:
        response = requests.get(
            f"{BASE_URL}/483/facilities/{fei}",
            headers={"X-Api-Key": API_KEY}
        )
        results.append(response.json())
        time.sleep(delay)  # Avoid rate limits
    return results

2. Error Handling

def safe_api_call(url, **kwargs):
    """API call with retry logic and error handling"""
    max_retries = 3
    for attempt in range(max_retries):
        try:
            response = requests.get(url, **kwargs, timeout=30)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.Timeout:
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # Exponential backoff
                continue
            raise
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 429:  # Rate limit
                time.sleep(60)
                continue
            raise

3. Data Caching

import json
from datetime import datetime, timedelta

def cache_risk_scores(fei_numbers, cache_hours=24):
    """Cache risk scores locally to reduce API calls"""
    cache_file = "risk_scores_cache.json"

    # Load existing cache
    try:
        with open(cache_file, 'r') as f:
            cache = json.load(f)
    except FileNotFoundError:
        cache = {}

    # Check cache freshness
    results = {}
    for fei in fei_numbers:
        if fei in cache:
            cached_time = datetime.fromisoformat(cache[fei]['cached_at'])
            if datetime.now() - cached_time < timedelta(hours=cache_hours):
                results[fei] = cache[fei]['data']
                continue

        # Fetch from API
        response = requests.get(
            f"{BASE_URL}/483/risk-scores/{fei}",
            headers={"X-Api-Key": API_KEY}
        )
        data = response.json()

        # Update cache
        cache[fei] = {
            'data': data,
            'cached_at': datetime.now().isoformat()
        }
        results[fei] = data

    # Save cache
    with open(cache_file, 'w') as f:
        json.dump(cache, f)

    return results

---

Next Steps

• 483 Quickstart - Get started in 5 minutes
• Risk Scoring Guide - Deep dive into risk methodology
• Compliance Dashboard Tutorial - Build a monitoring dashboard
• API Reference - Complete endpoint documentation