EducationApril 18, 2026· Last updated April 18, 2026

How to Screen Crypto Transactions for OFAC Sanctions in Python with SanctionShield AI

Step-by-step Python guide to building a real-time crypto transaction OFAC sanctions screening system. Covers wallet address screening, counterparty name matching, and GENIUS Act compliance logging.

APIVult Team

@apivult

How to Screen Crypto Transactions for OFAC Sanctions in Python with SanctionShield AI

The FinCEN and OFAC Notice of Proposed Rulemaking published on April 10, 2026 sets a clear deadline: stablecoin issuers must implement real-time sanctions screening before January 2027. But the technical requirement — screen every transaction before it settles, maintain an audit trail, handle fuzzy name matching across multilingual datasets — is not trivial to build from scratch.

This tutorial shows you exactly how to build a production-ready crypto transaction OFAC sanctions screening system in Python using the SanctionShield AI API. By the end, you'll have a working pipeline that screens wallet addresses, counterparty names, and transaction metadata against 40+ global watchlists and produces the structured audit logs that OFAC's proposed compliance program requires.

What You'll Build

A Python screening pipeline that:

Screens counterparty wallet addresses against OFAC's blocked address list
Screens counterparty names using fuzzy matching against the full SDN list
Applies jurisdiction-based risk scoring (Iran, Sudan, Nicaragua, Russia risk tiers)
Generates a compliance audit record for every transaction, pass or fail
Integrates with a FastAPI endpoint for real-time pre-settlement screening

Prerequisites

Python 3.12+
SanctionShield AI API key from RapidAPI
requests, fastapi, pydantic, python-dotenv packages

pip install requests fastapi uvicorn pydantic python-dotenv

Step 1: Set Up Your Environment

# config.py
import os
from dotenv import load_dotenv
 
load_dotenv()
 
SANCTIONSHIELD_API_KEY = os.environ["SANCTIONSHIELD_API_KEY"]
SANCTIONSHIELD_BASE_URL = "https://apivult.com/api/sanctionshield"
SCREENING_THRESHOLD = 0.82  # Fuzzy match confidence threshold
HIGH_RISK_JURISDICTIONS = {"IR", "SY", "KP", "CU", "SD", "NI", "RU", "BY"}

Step 2: Build the Core Screening Client

# screener.py
import requests
import logging
from datetime import datetime, timezone
from dataclasses import dataclass, asdict
from typing import Optional
from config import SANCTIONSHIELD_API_KEY, SANCTIONSHIELD_BASE_URL, SCREENING_THRESHOLD
 
logger = logging.getLogger(__name__)
 
@dataclass
class ScreeningResult:
    transaction_id: str
    timestamp: str
    counterparty_name: str
    wallet_address: Optional[str]
    jurisdiction: Optional[str]
    sanctions_hit: bool
    pep_hit: bool
    match_details: Optional[dict]
    risk_score: int
    decision: str  # "APPROVED", "BLOCKED", "REVIEW"
    audit_trail: dict
 
class SanctionShieldClient:
    def __init__(self):
        self.headers = {
            "X-RapidAPI-Key": SANCTIONSHIELD_API_KEY,
            "Content-Type": "application/json"
        }
 
    def screen_entity(self, name: str, wallet: Optional[str] = None,
                       country: Optional[str] = None) -> dict:
        """Screen a counterparty entity: name + optional wallet address."""
        payload = {
            "query": name,
            "threshold": SCREENING_THRESHOLD,
            "screening_types": ["SANCTIONS", "PEP"],
            "lists": ["OFAC_SDN", "OFAC_CONS", "EU", "UN", "OFSI"],
            "include_reason_codes": True
        }
        if wallet:
            payload["wallet_address"] = wallet
        if country:
            payload["country"] = country
 
        response = requests.post(
            f"{SANCTIONSHIELD_BASE_URL}/screen",
            headers=self.headers,
            json=payload,
            timeout=5
        )
        response.raise_for_status()
        return response.json()
 
    def get_jurisdiction_risk(self, country_code: str) -> dict:
        """Get OFAC jurisdiction risk assessment for a country."""
        response = requests.get(
            f"{SANCTIONSHIELD_BASE_URL}/jurisdiction/{country_code}",
            headers=self.headers,
            timeout=3
        )
        response.raise_for_status()
        return response.json()

Step 3: Build the Transaction Screening Pipeline

# pipeline.py
import uuid
from datetime import datetime, timezone
from typing import Optional
from screener import SanctionShieldClient, ScreeningResult
from config import HIGH_RISK_JURISDICTIONS
 
client = SanctionShieldClient()
 
def screen_transaction(
    counterparty_name: str,
    amount_usd: float,
    wallet_address: Optional[str] = None,
    jurisdiction: Optional[str] = None,
    transaction_id: Optional[str] = None
) -> ScreeningResult:
    """
    Screen a crypto transaction before settlement.
    Returns ScreeningResult with decision and full audit trail.
    """
    tx_id = transaction_id or str(uuid.uuid4())
    timestamp = datetime.now(timezone.utc).isoformat()
 
    # Run the sanctions + PEP screen
    screen_response = client.screen_entity(
        name=counterparty_name,
        wallet=wallet_address,
        country=jurisdiction
    )
 
    sanctions_hit = screen_response.get("sanctions", {}).get("match_found", False)
    pep_hit = screen_response.get("pep", {}).get("is_pep", False)
    match_details = screen_response.get("sanctions", {}).get("match", None)
    base_risk_score = screen_response.get("risk_score", 0)
 
    # Apply jurisdiction uplift
    jurisdiction_risk = 0
    if jurisdiction and jurisdiction.upper() in HIGH_RISK_JURISDICTIONS:
        jurisdiction_info = client.get_jurisdiction_risk(jurisdiction.upper())
        jurisdiction_risk = jurisdiction_info.get("risk_uplift", 20)
 
    total_risk_score = min(100, base_risk_score + jurisdiction_risk)
 
    # Determine decision
    if sanctions_hit:
        decision = "BLOCKED"
    elif total_risk_score >= 75:
        decision = "REVIEW"
    elif pep_hit and amount_usd >= 10000:
        decision = "REVIEW"  # EDD required for PEPs above threshold
    else:
        decision = "APPROVED"
 
    audit_trail = {
        "transaction_id": tx_id,
        "screened_at": timestamp,
        "api_response_id": screen_response.get("response_id"),
        "lists_checked": screen_response.get("lists_checked", []),
        "threshold_used": 0.82,
        "jurisdiction_risk_uplift": jurisdiction_risk,
        "compliance_rule_version": "GENIUS_ACT_NPRM_2026_04_10"
    }
 
    return ScreeningResult(
        transaction_id=tx_id,
        timestamp=timestamp,
        counterparty_name=counterparty_name,
        wallet_address=wallet_address,
        jurisdiction=jurisdiction,
        sanctions_hit=sanctions_hit,
        pep_hit=pep_hit,
        match_details=match_details,
        risk_score=total_risk_score,
        decision=decision,
        audit_trail=audit_trail
    )

Step 4: Wrap in a FastAPI Pre-Settlement Endpoint

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional
from pipeline import screen_transaction
import logging
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
app = FastAPI(
    title="Crypto Transaction OFAC Screening API",
    description="Pre-settlement sanctions screening for crypto transactions",
    version="1.0.0"
)
 
class TransactionScreenRequest(BaseModel):
    counterparty_name: str = Field(..., min_length=2, max_length=500)
    amount_usd: float = Field(..., gt=0)
    wallet_address: Optional[str] = Field(None, pattern=r"^0x[a-fA-F0-9]{40}$")
    jurisdiction: Optional[str] = Field(None, min_length=2, max_length=2)
    transaction_id: Optional[str] = None
 
class TransactionScreenResponse(BaseModel):
    transaction_id: str
    decision: str
    risk_score: int
    sanctions_hit: bool
    pep_hit: bool
    match_details: Optional[dict]
    audit_trail: dict
 
@app.post("/screen/transaction", response_model=TransactionScreenResponse)
async def screen_pre_settlement(request: TransactionScreenRequest):
    """
    Screen a crypto transaction before settlement.
    Returns APPROVED, BLOCKED, or REVIEW with full audit trail.
    """
    result = screen_transaction(
        counterparty_name=request.counterparty_name,
        amount_usd=request.amount_usd,
        wallet_address=request.wallet_address,
        jurisdiction=request.jurisdiction,
        transaction_id=request.transaction_id
    )
 
    logger.info(
        "Transaction screened",
        extra={
            "transaction_id": result.transaction_id,
            "decision": result.decision,
            "risk_score": result.risk_score,
            "sanctions_hit": result.sanctions_hit
        }
    )
 
    if result.decision == "BLOCKED":
        # Log for SAR consideration — do not raise exception,
        # return the BLOCKED decision so the caller can handle it
        logger.warning(f"BLOCKED transaction {result.transaction_id}: {result.match_details}")
 
    return TransactionScreenResponse(
        transaction_id=result.transaction_id,
        decision=result.decision,
        risk_score=result.risk_score,
        sanctions_hit=result.sanctions_hit,
        pep_hit=result.pep_hit,
        match_details=result.match_details,
        audit_trail=result.audit_trail
    )
 
@app.get("/health")
async def health():
    return {"status": "ok"}

Step 5: Test Your Implementation

# test_screening.py
import pytest
from pipeline import screen_transaction
 
def test_clean_transaction_approved():
    result = screen_transaction(
        counterparty_name="Acme Software GmbH",
        amount_usd=5000,
        jurisdiction="DE"
    )
    assert result.decision == "APPROVED"
    assert result.sanctions_hit is False
    assert result.audit_trail["compliance_rule_version"] == "GENIUS_ACT_NPRM_2026_04_10"
 
def test_iran_jurisdiction_triggers_review():
    result = screen_transaction(
        counterparty_name="Unknown Trading Company",
        amount_usd=1000,
        jurisdiction="IR"
    )
    # Iran jurisdiction should trigger review regardless of name match
    assert result.decision in ["REVIEW", "BLOCKED"]
    assert result.risk_score >= 50
 
def test_audit_trail_completeness():
    result = screen_transaction(
        counterparty_name="Test Entity",
        amount_usd=100
    )
    assert "screened_at" in result.audit_trail
    assert "lists_checked" in result.audit_trail
    assert "api_response_id" in result.audit_trail
    assert "threshold_used" in result.audit_trail
 
def test_pep_large_amount_requires_review():
    # PEP + $10K+ amount should trigger Enhanced Due Diligence review
    # This test uses a mock — in production, a known PEP name would trigger this
    pass  # Integration test against live API

Step 6: Run It

# Set your API key
export SANCTIONSHIELD_API_KEY=your_key_here
 
# Run the API server
uvicorn main:app --host 0.0.0.0 --port 8000
 
# Test a transaction
curl -X POST http://localhost:8000/screen/transaction \
  -H "Content-Type: application/json" \
  -d '{
    "counterparty_name": "Acme Trading LLC",
    "amount_usd": 25000,
    "wallet_address": "0x742d35Cc6634C0532925a3b8D4C9C4564e0b8e5C",
    "jurisdiction": "US"
  }'

What the OFAC Compliance Requirements Mean for This Code

The GENIUS Act NPRM requires stablecoin issuers to build compliance programs around five elements. Here's how this implementation addresses each:

OFAC Requirement	Implementation
Senior management commitment	Documented policy in `compliance_rule_version` field
Risk assessment	`jurisdiction_risk_uplift` + base risk scoring
Internal controls	Pre-settlement block on `sanctions_hit`; `REVIEW` queue for elevated risk
Testing	pytest suite in `test_screening.py`
Training	Audit trail provides evidence for training record documentation

The audit_trail dictionary returned with every screening result is the foundation of your compliance evidence. Store it — alongside the API's response_id — in a append-only log table (PostgreSQL with row-level security, or a dedicated compliance data store). OFAC expects organizations to maintain screening records for a minimum of five years.

Next Steps

Add batch screening for existing customer portfolios when new OFAC designations are published
Implement webhook notifications from SanctionShield AI to trigger retroactive screening when entities are newly designated
Integrate the REVIEW queue with a case management system for Enhanced Due Diligence workflows

Get started with SanctionShield AI on RapidAPI →

GENIUS Act AML Rules: FinCEN and OFAC Propose Bank-Level Stablecoin Compliance — April 2026

FinCEN and OFAC jointly published proposed AML and sanctions compliance rules for stablecoin issuers on April 10, 2026, bringing crypto firms under bank-level obligations by January 2027.

April 18, 2026

Best Crypto Sanctions Screening APIs in 2026: Complete Comparison

Compare the top crypto and blockchain sanctions screening APIs in 2026. Features, pricing, OFAC coverage, and real-time performance benchmarks for crypto exchanges, fintech, and DeFi.

April 11, 2026