Integrating Real-Time Sanctions Screening into Your Fintech Stack with SanctionShield AI

pip install requests python-dotenv fastapi uvicorn

export RAPIDAPI_KEY="YOUR_API_KEY"

import requests
import os
from typing import Dict, Any, Optional, List
from enum import Enum
 
RAPIDAPI_KEY = os.environ["RAPIDAPI_KEY"]
SANCTIONSHIELD_HOST = "sanctionshield-ai.p.rapidapi.com"
 
HEADERS = {
    "X-RapidAPI-Key": RAPIDAPI_KEY,
    "X-RapidAPI-Host": SANCTIONSHIELD_HOST,
    "Content-Type": "application/json"
}
 
 
class ScreeningDecision(Enum):
    CLEAR = "clear"           # No matches found
    REVIEW = "review"         # Potential match — manual review required
    BLOCK = "block"           # High-confidence match — block transaction
 
 
def screen_entity(
    name: str,
    entity_type: str = "individual",  # "individual" | "organization" | "vessel" | "aircraft"
    additional_fields: Optional[Dict] = None,
    lists: Optional[List[str]] = None
) -> Dict[str, Any]:
    """
    Screen an individual or organization against sanctions lists.
    Additional fields (date_of_birth, nationality, address) improve match accuracy.
    """
    payload = {
        "name": name,
        "entity_type": entity_type,
        "lists": lists or ["OFAC_SDN", "OFAC_CONS", "EU_CONSOLIDATED", "UN_CONSOLIDATED", "UK_HMT"],
        "fuzzy_matching": True,
        "match_threshold": 85  # Minimum similarity score to flag (0-100)
    }
 
    if additional_fields:
        payload.update(additional_fields)
 
    response = requests.post(
        f"https://{SANCTIONSHIELD_HOST}/screen",
        headers=HEADERS,
        json=payload
    )
    response.raise_for_status()
    return response.json()

from dataclasses import dataclass
from datetime import datetime
 
 
@dataclass
class OnboardingResult:
    applicant_name: str
    decision: str          # "approve" | "review" | "decline"
    screening_id: str
    matches: List[Dict]
    timestamp: str
    reason: Optional[str] = None
 
 
def screen_onboarding_applicant(
    first_name: str,
    last_name: str,
    date_of_birth: Optional[str] = None,  # ISO format: YYYY-MM-DD
    nationality: Optional[str] = None,     # ISO 3166-1 alpha-2: "US", "DE", etc.
    country_of_residence: Optional[str] = None,
    id_number: Optional[str] = None       # Passport or national ID
) -> OnboardingResult:
    """
    Screen a new user during onboarding. Returns decision and audit record.
    """
    full_name = f"{first_name} {last_name}"
 
    additional = {}
    if date_of_birth:
        additional["date_of_birth"] = date_of_birth
    if nationality:
        additional["nationality"] = nationality
    if country_of_residence:
        additional["country_of_residence"] = country_of_residence
    if id_number:
        additional["id_number"] = id_number
 
    result = screen_entity(
        name=full_name,
        entity_type="individual",
        additional_fields=additional if additional else None
    )
 
    # Map API result to onboarding decision
    if result["match_status"] == "no_match":
        decision = "approve"
        reason = None
    elif result["highest_match_score"] >= 95:
        decision = "decline"
        reason = "sanctions_match"
    else:
        decision = "review"
        reason = "potential_sanctions_match"
 
    return OnboardingResult(
        applicant_name=full_name,
        decision=decision,
        screening_id=result["screening_id"],
        matches=result.get("matches", []),
        timestamp=result["screened_at"],
        reason=reason
    )
 
 
# FastAPI endpoint integration
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
 
app = FastAPI()
 
 
class OnboardingRequest(BaseModel):
    first_name: str
    last_name: str
    date_of_birth: Optional[str] = None
    nationality: Optional[str] = None
    country_of_residence: Optional[str] = None
 
 
@app.post("/api/onboard")
async def create_account(request: OnboardingRequest):
    """Customer onboarding endpoint with integrated sanctions screening."""
    screening = screen_onboarding_applicant(
        first_name=request.first_name,
        last_name=request.last_name,
        date_of_birth=request.date_of_birth,
        nationality=request.nationality,
        country_of_residence=request.country_of_residence
    )
 
    # Log for audit trail regardless of outcome
    log_screening_event(screening)
 
    if screening.decision == "decline":
        raise HTTPException(
            status_code=403,
            detail="Account application cannot be processed at this time."
            # Note: Do NOT reveal that the reason is a sanctions match
        )
 
    elif screening.decision == "review":
        # Queue for manual review — don't approve or decline immediately
        create_manual_review_task(screening)
        return {
            "status": "pending_review",
            "message": "Your application is under review. We'll contact you within 2 business days."
        }
 
    else:
        # Proceed with account creation
        account = create_account_in_db(request)
        return {"status": "approved", "account_id": account.id}

def screen_payment(
    beneficiary_name: str,
    beneficiary_account: str,
    beneficiary_bank_country: str,
    amount: float,
    currency: str,
    originator_name: str
) -> Dict[str, Any]:
    """
    Screen a pending payment for sanctions exposure.
    Screens both beneficiary and originator.
    """
    # Screen beneficiary
    beneficiary_result = screen_entity(
        name=beneficiary_name,
        entity_type="individual",
        additional_fields={
            "country": beneficiary_bank_country,
            "account_reference": beneficiary_account
        }
    )
 
    # Screen originator (in correspondent banking scenarios)
    originator_result = screen_entity(
        name=originator_name,
        entity_type="individual"
    )
 
    # Determine overall payment decision
    has_block = (
        beneficiary_result["match_status"] == "match" or
        originator_result["match_status"] == "match"
    )
    has_review = (
        beneficiary_result["match_status"] == "potential_match" or
        originator_result["match_status"] == "potential_match"
    )
 
    return {
        "payment_cleared": not has_block and not has_review,
        "decision": "block" if has_block else ("review" if has_review else "clear"),
        "beneficiary_screening": beneficiary_result,
        "originator_screening": originator_result,
        "screening_ids": [
            beneficiary_result["screening_id"],
            originator_result["screening_id"]
        ],
        "screened_at": datetime.utcnow().isoformat()
    }
 
 
def process_payment_with_screening(payment_data: Dict) -> Dict:
    """Payment processing pipeline with mandatory sanctions screening."""
 
    # Screen before processing
    screening = screen_payment(
        beneficiary_name=payment_data["to_name"],
        beneficiary_account=payment_data["to_account"],
        beneficiary_bank_country=payment_data["to_country"],
        amount=payment_data["amount"],
        currency=payment_data["currency"],
        originator_name=payment_data["from_name"]
    )
 
    if screening["decision"] == "block":
        # Log the blocked payment for compliance reporting
        record_blocked_payment(payment_data, screening)
        return {
            "status": "declined",
            "reason": "payment_screening_failed",
            "reference": screening["screening_ids"][0]
        }
 
    elif screening["decision"] == "review":
        # Hold payment for compliance officer review
        hold_payment_for_review(payment_data, screening)
        return {
            "status": "pending",
            "reason": "manual_review_required",
            "reference": screening["screening_ids"][0]
        }
 
    # Process the payment
    result = execute_payment(payment_data)
    return {
        "status": "processed",
        "transaction_id": result["id"],
        "screening_reference": screening["screening_ids"][0]
    }

import schedule
import time
 
 
def batch_rescreen_customers(customer_batch: List[Dict]) -> Dict[str, Any]:
    """
    Re-screen a batch of existing customers against current sanctions lists.
    Designed to run nightly or on list update events.
    """
    response = requests.post(
        f"https://{SANCTIONSHIELD_HOST}/screen/batch",
        headers=HEADERS,
        json={
            "entities": [
                {
                    "entity_id": c["id"],
                    "name": c["full_name"],
                    "entity_type": "individual",
                    "date_of_birth": c.get("date_of_birth"),
                    "nationality": c.get("nationality")
                }
                for c in customer_batch
            ],
            "lists": ["OFAC_SDN", "OFAC_CONS", "EU_CONSOLIDATED", "UN_CONSOLIDATED"],
            "match_threshold": 85
        }
    )
    response.raise_for_status()
    return response.json()
 
 
def run_nightly_rescreening():
    """Pull all active customers and re-screen against current lists."""
    customers = get_all_active_customers()  # Your DB query
    chunk_size = 1000
    newly_flagged = []
 
    for i in range(0, len(customers), chunk_size):
        chunk = customers[i:i + chunk_size]
        results = batch_rescreen_customers(chunk)
 
        for result in results["results"]:
            if result["match_status"] in ("match", "potential_match"):
                # Check if this customer was previously clear
                if was_previously_clear(result["entity_id"]):
                    newly_flagged.append(result)
                    escalate_to_compliance(result)
 
    print(f"Rescreened {len(customers)} customers. "
          f"Newly flagged: {len(newly_flagged)}")
    return newly_flagged
 
 
# Schedule nightly re-screening
schedule.every().day.at("02:00").do(run_nightly_rescreening)

import json
from pathlib import Path
 
 
def log_screening_event(screening_data: Dict[str, Any], event_type: str = "screen"):
    """
    Write screening event to audit log. Must include:
    - Who was screened (anonymized or pseudonymized)
    - What lists were checked
    - The result
    - Timestamp
    - Screening ID (for cross-reference with API provider)
    """
    audit_entry = {
        "event_type": event_type,
        "screening_id": screening_data.get("screening_id"),
        "entity_type": screening_data.get("entity_type", "individual"),
        "lists_checked": screening_data.get("lists_checked", []),
        "match_status": screening_data.get("match_status"),
        "highest_score": screening_data.get("highest_match_score"),
        "decision": screening_data.get("decision"),
        "timestamp": datetime.utcnow().isoformat(),
        "operator": "automated_system"
    }
 
    # Append to daily audit log file
    log_file = Path(f"/var/log/compliance/sanctions-{datetime.utcnow().date()}.jsonl")
    with log_file.open("a") as f:
        f.write(json.dumps(audit_entry) + "\n")