Tutorial: Rule Updates Monitoring with CTWise API

Difficulty: Intermediate Time to Complete: 25 minutes Prerequisites: CTWise API key (Starter tier+), webhook endpoint or email

---

Goal

Learn how to monitor regulatory rule updates and set up proactive alerts for changes. By the end of this tutorial, you'll be able to:

1. Query for recently updated rules
2. Compare rule versions
3. Set up webhook notifications for rule changes
4. Build an automated compliance monitoring system

---

Prerequisites

Before you begin, ensure you have:

• A CTWise API key (Starter tier or higher)
• A webhook endpoint (or willingness to set one up)
• Python 3.8+ or Node.js 18+
• Understanding of Query FDA Rules and ICH Integration

---

Why Monitor Rule Updates?

Regulatory rules change frequently:

Source	Avg. Updates/Month	Critical Changes/Year
FDA	15-25	50+
ICH	2-5	10-15
EMA	10-20	30+

Missing a critical update can result in:

• Clinical trial delays
• Regulatory submission rejections
• Compliance violations
• Patient safety issues

---

Step 1: Query Recently Updated Rules

Find rules updated in the last 30 days:

export CTWISE_API_KEY="your-api-key-here"
export CTWISE_API_BASE="https://kj2xss9yg9.execute-api.us-east-1.amazonaws.com/dev0/v1"

# Get rules updated in the last 30 days
curl -s "${CTWISE_API_BASE}/rules?updated_since=30d&limit=20" \
  -H "X-Api-Key: ${CTWISE_API_KEY}" | jq '.rules[] | {id, title, last_updated, change_type}'

Expected Output:

{"id": "fda-21cfr312", "title": "21 CFR Part 312 - IND Application", "last_updated": "2024-12-15", "change_type": "amendment"}
{"id": "ich-e6-r3", "title": "ICH E6(R3) Good Clinical Practice", "last_updated": "2024-12-10", "change_type": "revision"}
{"id": "fda-guidance-ai-ml", "title": "FDA Guidance on AI/ML in Drug Development", "last_updated": "2024-12-05", "change_type": "new"}

Update Filter Options:

Parameter	Values	Description
updated_since	7d, 30d, 90d, 1y	Time range for updates
change_type	new, amendment, revision, withdrawn	Type of change
source	fda, ich, ema	Filter by source

---

Step 2: Get Change Details

Retrieve detailed change information for a specific rule:

RULE_ID="ich-e6-r3"

curl -s "${CTWISE_API_BASE}/rules/${RULE_ID}/history" \
  -H "X-Api-Key: ${CTWISE_API_KEY}" | jq .

Expected Output:

{
  "rule_id": "ich-e6-r3",
  "current_version": "R3",
  "versions": [
    {
      "version": "R3",
      "effective_date": "2024-06-01",
      "status": "current",
      "change_summary": "Major revision incorporating digital health, risk-based approaches",
      "key_changes": [
        "Added Chapter 4: Digital Health Technologies",
        "Enhanced risk-based quality management",
        "New requirements for decentralized trials"
      ]
    },
    {
      "version": "R2",
      "effective_date": "2016-11-09",
      "status": "superseded",
      "change_summary": "Addendum focusing on electronic records and essential documents"
    },
    {
      "version": "R1",
      "effective_date": "1996-05-01",
      "status": "superseded",
      "change_summary": "Original harmonized GCP guideline"
    }
  ]
}

---

Step 3: Compare Rule Versions

Compare two versions of a rule to see specific differences:

curl -s "${CTWISE_API_BASE}/rules/${RULE_ID}/compare?from=R2&to=R3" \
  -H "X-Api-Key: ${CTWISE_API_KEY}" | jq .

Expected Output:

{
  "rule_id": "ich-e6-r3",
  "from_version": "R2",
  "to_version": "R3",
  "comparison": {
    "sections_added": [
      {"section": "4.0", "title": "Digital Health Technologies"},
      {"section": "5.20", "title": "Decentralized Clinical Trials"}
    ],
    "sections_modified": [
      {"section": "2.0", "title": "Principles of ICH GCP", "change": "Risk-based language added"},
      {"section": "5.0", "title": "Sponsor Responsibilities", "change": "Quality management requirements enhanced"}
    ],
    "sections_removed": [],
    "impact_assessment": {
      "severity": "major",
      "compliance_deadline": "2025-01-01",
      "affected_areas": ["trial conduct", "data management", "quality systems"]
    }
  }
}

---

Step 4: Set Up Webhook Notifications

4.1 Register a Webhook

Register your endpoint to receive rule update notifications:

curl -s -X POST "${CTWISE_API_BASE}/webhooks" \
  -H "X-Api-Key: ${CTWISE_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://your-domain.com/webhooks/ctwise",
    "events": ["rule.updated", "rule.new", "rule.withdrawn"],
    "filters": {
      "sources": ["fda", "ich"],
      "categories": ["clinical_trials", "safety"]
    },
    "secret": "your-webhook-secret"
  }' | jq .

Response:

{
  "webhook_id": "whk_abc123",
  "url": "https://your-domain.com/webhooks/ctwise",
  "events": ["rule.updated", "rule.new", "rule.withdrawn"],
  "status": "active",
  "created_at": "2024-12-30T10:00:00Z"
}

4.2 Webhook Payload Format

When a rule is updated, you'll receive:

{
  "event": "rule.updated",
  "timestamp": "2024-12-30T14:30:00Z",
  "rule": {
    "id": "fda-21cfr312",
    "title": "21 CFR Part 312 - IND Application",
    "source": "fda",
    "category": "clinical_trials",
    "previous_version": "2024-06-01",
    "current_version": "2024-12-15",
    "change_type": "amendment",
    "change_summary": "Updated electronic submission requirements",
    "url": "https://docs.ctwise.ai/rules/fda-21cfr312"
  },
  "signature": "sha256=..."
}

4.3 Verify Webhook Signature

import hmac
import hashlib

def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
    """Verify CTWise webhook signature."""
    expected = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(f"sha256={expected}", signature)

# In your webhook handler
from flask import Flask, request, jsonify

app = Flask(__name__)
WEBHOOK_SECRET = "your-webhook-secret"

@app.route('/webhooks/ctwise', methods=['POST'])
def handle_webhook():
    signature = request.headers.get('X-CTWise-Signature')
    payload = request.get_data()

    if not verify_webhook_signature(payload, signature, WEBHOOK_SECRET):
        return jsonify({"error": "Invalid signature"}), 401

    event = request.json
    print(f"Received {event['event']}: {event['rule']['title']}")

    # Process the update
    process_rule_update(event)

    return jsonify({"status": "received"}), 200

---

Step 5: Build an Automated Monitoring System

Here's a complete Python implementation for monitoring rule updates:

import os
import requests
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import json

class CTWiseRuleMonitor:
    """Monitor CTWise regulatory rules for updates."""

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://kj2xss9yg9.execute-api.us-east-1.amazonaws.com/dev0/v1"
        self.headers = {
            "X-Api-Key": api_key,
            "Content-Type": "application/json"
        }

    def get_recent_updates(
        self,
        days: int = 7,
        sources: Optional[List[str]] = None,
        categories: Optional[List[str]] = None
    ) -> List[Dict]:
        """Get rules updated in the last N days."""
        params = {
            "updated_since": f"{days}d",
            "limit": 100
        }
        if sources:
            params["source"] = ",".join(sources)
        if categories:
            params["category"] = ",".join(categories)

        response = requests.get(
            f"{self.base_url}/rules",
            headers=self.headers,
            params=params
        )
        response.raise_for_status()
        return response.json()["rules"]

    def get_rule_history(self, rule_id: str) -> Dict:
        """Get version history for a rule."""
        response = requests.get(
            f"{self.base_url}/rules/{rule_id}/history",
            headers=self.headers
        )
        response.raise_for_status()
        return response.json()

    def compare_versions(self, rule_id: str, from_version: str, to_version: str) -> Dict:
        """Compare two versions of a rule."""
        response = requests.get(
            f"{self.base_url}/rules/{rule_id}/compare",
            headers=self.headers,
            params={"from": from_version, "to": to_version}
        )
        response.raise_for_status()
        return response.json()

    def generate_compliance_report(self, days: int = 30) -> Dict:
        """Generate a compliance report for recent changes."""
        updates = self.get_recent_updates(days=days)

        report = {
            "report_date": datetime.now().isoformat(),
            "period_days": days,
            "total_updates": len(updates),
            "by_source": {},
            "by_change_type": {},
            "critical_updates": [],
            "action_required": []
        }

        for rule in updates:
            # Count by source
            source = rule.get("source", "unknown")
            report["by_source"][source] = report["by_source"].get(source, 0) + 1

            # Count by change type
            change_type = rule.get("change_type", "unknown")
            report["by_change_type"][change_type] = report["by_change_type"].get(change_type, 0) + 1

            # Identify critical updates (new or major revisions)
            if change_type in ["new", "revision"]:
                report["critical_updates"].append({
                    "rule_id": rule["id"],
                    "title": rule["title"],
                    "change_type": change_type,
                    "last_updated": rule.get("last_updated")
                })

                # Flag for action if compliance deadline is approaching
                if rule.get("compliance_deadline"):
                    deadline = datetime.fromisoformat(rule["compliance_deadline"])
                    if deadline <= datetime.now() + timedelta(days=90):
                        report["action_required"].append({
                            "rule_id": rule["id"],
                            "title": rule["title"],
                            "deadline": rule["compliance_deadline"],
                            "days_remaining": (deadline - datetime.now()).days
                        })

        return report


# Example usage
if __name__ == "__main__":
    monitor = CTWiseRuleMonitor(os.environ["CTWISE_API_KEY"])

    # Get FDA updates from last 7 days
    fda_updates = monitor.get_recent_updates(
        days=7,
        sources=["fda"],
        categories=["clinical_trials"]
    )
    print(f"FDA Clinical Trial Updates (7 days): {len(fda_updates)}")

    # Generate compliance report
    report = monitor.generate_compliance_report(days=30)
    print(f"\n30-Day Compliance Report:")
    print(f"  Total Updates: {report['total_updates']}")
    print(f"  Critical Updates: {len(report['critical_updates'])}")
    print(f"  Action Required: {len(report['action_required'])}")

    # Show action items
    if report["action_required"]:
        print("\n⚠️  Action Required:")
        for item in report["action_required"]:
            print(f"  - {item['title']}: {item['days_remaining']} days until deadline")

---

Step 6: Schedule Automated Checks

Using AWS Lambda + EventBridge

Create a scheduled Lambda function:

# lambda_handler.py
import json
import os
import boto3
from ctwise_monitor import CTWiseRuleMonitor  # The class above

ses = boto3.client('ses')

def lambda_handler(event, context):
    """Daily rule monitoring check."""
    monitor = CTWiseRuleMonitor(os.environ["CTWISE_API_KEY"])

    # Generate daily report
    report = monitor.generate_compliance_report(days=1)

    # If there are critical updates, send email
    if report["critical_updates"]:
        send_alert_email(report)

    return {
        "statusCode": 200,
        "body": json.dumps({
            "updates_found": report["total_updates"],
            "critical": len(report["critical_updates"])
        })
    }

def send_alert_email(report):
    """Send email alert for critical updates."""
    body = f"""
    CTWise Daily Regulatory Update Alert

    Updates Found: {report['total_updates']}
    Critical Updates: {len(report['critical_updates'])}

    Critical Updates:
    """
    for update in report["critical_updates"]:
        body += f"\n  - {update['title']} ({update['change_type']})"

    if report["action_required"]:
        body += "\n\n⚠️ Action Required:\n"
        for item in report["action_required"]:
            body += f"\n  - {item['title']}: {item['days_remaining']} days until deadline"

    ses.send_email(
        Source="alerts@orchestraprime.ai",
        Destination={"ToAddresses": [os.environ["ALERT_EMAIL"]]},
        Message={
            "Subject": {"Data": f"CTWise Alert: {len(report['critical_updates'])} Critical Updates"},
            "Body": {"Text": {"Data": body}}
        }
    )

CloudFormation for Scheduled Lambda

RuleMonitorFunction:
  Type: AWS::Lambda::Function
  Properties:
    FunctionName: ctwise-rule-monitor
    Runtime: python3.12
    Handler: lambda_handler.lambda_handler
    Environment:
      Variables:
        CTWISE_API_KEY: !Ref CTWiseApiKey
        ALERT_EMAIL: !Ref AlertEmail

ScheduleRule:
  Type: AWS::Events::Rule
  Properties:
    Name: ctwise-daily-monitor
    ScheduleExpression: "cron(0 8 * * ? *)"  # 8 AM UTC daily
    Targets:
      - Id: RuleMonitorTarget
        Arn: !GetAtt RuleMonitorFunction.Arn

---

Best Practices

1. Prioritize by Impact

Not all updates are equal. Focus on:

• New rules in your therapeutic area
• Revisions with compliance deadlines
• Amendments affecting ongoing trials

2. Maintain an Audit Trail

Log all rule changes for compliance documentation:

def log_update_to_dynamodb(rule_update: dict):
    """Log rule update to DynamoDB for audit."""
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('ctwise-rule-audit')

    table.put_item(Item={
        'rule_id': rule_update['id'],
        'timestamp': datetime.now().isoformat(),
        'change_type': rule_update['change_type'],
        'previous_version': rule_update.get('previous_version'),
        'current_version': rule_update.get('current_version'),
        'reviewed_by': None,
        'review_status': 'pending'
    })

3. Integrate with Your QMS

Connect rule updates to your Quality Management System:

• Create change control records
• Trigger impact assessments
• Update SOPs automatically

---

Troubleshooting

Issue	Cause	Solution
Webhook not receiving events	Firewall blocking	Whitelist CTWise IPs
Missing updates	Filter too restrictive	Broaden source/category filters
Rate limit errors	Too frequent polling	Use webhooks instead of polling
Signature verification fails	Clock skew	Sync server time with NTP

---

Next Steps

• Query FDA Rules Tutorial - Master FDA rule queries
• ICH Integration Tutorial - Cross-reference with ICH
• API Reference - Full API documentation

---

Feedback

Found an issue with this tutorial? Submit feedback or contact support@orchestraprime.ai.