How to Screen Cryptocurrency Wallet Addresses for Sanctions Compliance in Python

pip install requests python-dotenv

# .env
APIVULT_API_KEY=YOUR_API_KEY

import os
import requests
from dotenv import load_dotenv
 
load_dotenv()
 
API_KEY = os.getenv("APIVULT_API_KEY")
BASE_URL = "https://apivult.com/api/sanctionshield"
 
def screen_wallet_address(address: str, asset_type: str = "crypto") -> dict:
    """
    Screen a single wallet address against sanctions lists.
 
    Args:
        address: Crypto wallet address (ETH, BTC, etc.)
        asset_type: Type of address ("crypto", "eth", "btc")
 
    Returns:
        Screening result with match status and details
    """
    headers = {
        "X-RapidAPI-Key": API_KEY,
        "Content-Type": "application/json"
    }
 
    payload = {
        "query": address,
        "entity_type": asset_type,
        "lists": ["OFAC_SDN", "EU_CONSOLIDATED", "UN_CONSOLIDATED"]
    }
 
    response = requests.post(
        f"{BASE_URL}/screen",
        json=payload,
        headers=headers,
        timeout=10
    )
    response.raise_for_status()
    return response.json()
 
# Example usage
result = screen_wallet_address("0x7F367cC41522cE07553e823bf3be79A889DEBE1B")
print(f"Status: {result['status']}")
print(f"Match score: {result.get('match_score', 0)}")
print(f"Matches found: {len(result.get('matches', []))}")

{
  "status": "MATCH",
  "match_score": 0.98,
  "matches": [
    {
      "list": "OFAC_SDN",
      "entity_name": "LAZARUS GROUP",
      "match_type": "exact",
      "designation_date": "2019-09-13",
      "reason": "North Korea - Cyber Operations"
    }
  ],
  "screened_at": "2026-04-04T10:23:41Z"
}

import asyncio
import aiohttp
from typing import List
 
async def screen_wallet_batch_async(
    addresses: List[str],
    session: aiohttp.ClientSession,
    max_concurrent: int = 10
) -> List[dict]:
    """
    Screen multiple wallet addresses concurrently.
    """
    semaphore = asyncio.Semaphore(max_concurrent)
 
    async def screen_single(address: str) -> dict:
        async with semaphore:
            payload = {
                "query": address,
                "entity_type": "crypto",
                "lists": ["OFAC_SDN", "EU_CONSOLIDATED", "UN_CONSOLIDATED"]
            }
            headers = {"X-RapidAPI-Key": API_KEY}
 
            async with session.post(
                f"{BASE_URL}/screen",
                json=payload,
                headers=headers
            ) as resp:
                data = await resp.json()
                data["address"] = address
                return data
 
    tasks = [screen_single(addr) for addr in addresses]
    return await asyncio.gather(*tasks, return_exceptions=True)
 
async def batch_screen(addresses: List[str]) -> dict:
    """
    Screen a batch of addresses and categorize results.
    """
    async with aiohttp.ClientSession() as session:
        results = await screen_wallet_batch_async(addresses, session)
 
    cleared = []
    blocked = []
    errors = []
 
    for result in results:
        if isinstance(result, Exception):
            errors.append(str(result))
            continue
 
        if result.get("status") == "CLEAR":
            cleared.append(result["address"])
        else:
            blocked.append({
                "address": result["address"],
                "matches": result.get("matches", []),
                "score": result.get("match_score", 0)
            })
 
    return {
        "total": len(addresses),
        "cleared": len(cleared),
        "blocked": len(blocked),
        "errors": len(errors),
        "blocked_addresses": blocked
    }
 
# Screen a batch of wallet addresses from KYC onboarding
wallets_to_screen = [
    "0x3cD751E6b0078Be393132286c442345e5DC49699",
    "bc1qxy2kgdygjrsqtzq2n0yrf2493p83kkfjhx0wlh",
    "0x7F367cC41522cE07553e823bf3be79A889DEBE1B",
    # ... more addresses
]
 
summary = asyncio.run(batch_screen(wallets_to_screen))
print(f"Screened {summary['total']} addresses")
print(f"Cleared: {summary['cleared']}")
print(f"Blocked: {summary['blocked']}")

from dataclasses import dataclass
from enum import Enum
import logging
 
logger = logging.getLogger(__name__)
 
class TransactionDecision(Enum):
    APPROVE = "approve"
    BLOCK = "block"
    REVIEW = "review"
 
@dataclass
class TransactionScreeningResult:
    decision: TransactionDecision
    match_score: float
    reason: str
    matches: list
 
def screen_transaction(
    sender_address: str,
    recipient_address: str,
    amount_usd: float
) -> TransactionScreeningResult:
    """
    Screen both parties in a transaction before processing.
    High-value transactions use stricter thresholds.
    """
    # Stricter threshold for high-value transactions
    match_threshold = 0.70 if amount_usd >= 10000 else 0.85
 
    parties = {
        "sender": screen_wallet_address(sender_address),
        "recipient": screen_wallet_address(recipient_address)
    }
 
    for party_name, result in parties.items():
        score = result.get("match_score", 0)
        matches = result.get("matches", [])
 
        if score >= 0.90:
            # High confidence match — hard block
            logger.warning(
                f"SANCTIONS BLOCK: {party_name} {sender_address} "
                f"score={score:.2f} matches={[m['entity_name'] for m in matches]}"
            )
            return TransactionScreeningResult(
                decision=TransactionDecision.BLOCK,
                match_score=score,
                reason=f"Sanctions match on {party_name}: {matches[0]['entity_name'] if matches else 'unknown'}",
                matches=matches
            )
 
        elif score >= match_threshold:
            # Possible match — route to manual review
            return TransactionScreeningResult(
                decision=TransactionDecision.REVIEW,
                match_score=score,
                reason=f"Possible sanctions match on {party_name} — manual review required",
                matches=matches
            )
 
    return TransactionScreeningResult(
        decision=TransactionDecision.APPROVE,
        match_score=0.0,
        reason="All parties cleared",
        matches=[]
    )
 
# Integrate into your transaction handler
def process_crypto_transaction(sender, recipient, amount_usd):
    screening = screen_transaction(sender, recipient, amount_usd)
 
    if screening.decision == TransactionDecision.BLOCK:
        raise ValueError(f"Transaction blocked: {screening.reason}")
 
    elif screening.decision == TransactionDecision.REVIEW:
        queue_for_manual_review(sender, recipient, amount_usd, screening)
        return {"status": "pending_review", "message": "Transaction queued for compliance review"}
 
    # Proceed with transaction
    return execute_transaction(sender, recipient, amount_usd)

import psycopg2
from datetime import datetime, timedelta
 
def get_customer_wallets(conn) -> List[dict]:
    """Fetch all customer wallet addresses from your database."""
    with conn.cursor() as cur:
        cur.execute("""
            SELECT user_id, wallet_address, last_screened_at
            FROM user_wallets
            WHERE is_active = TRUE
            ORDER BY last_screened_at ASC NULLS FIRST
        """)
        return cur.fetchall()
 
def run_daily_monitoring(db_connection_string: str):
    """
    Re-screen all active customer wallets.
    Run this as a daily scheduled job.
    """
    conn = psycopg2.connect(db_connection_string)
    wallets = get_customer_wallets(conn)
 
    blocked_customers = []
 
    for user_id, address, last_screened in wallets:
        result = screen_wallet_address(address)
 
        # Update last screened timestamp
        with conn.cursor() as cur:
            cur.execute(
                "UPDATE user_wallets SET last_screened_at = %s WHERE wallet_address = %s",
                (datetime.utcnow(), address)
            )
 
        if result.get("status") == "MATCH" and result.get("match_score", 0) >= 0.85:
            blocked_customers.append({
                "user_id": user_id,
                "address": address,
                "match_score": result["match_score"],
                "matches": result.get("matches", [])
            })
 
            # Freeze account immediately
            freeze_customer_account(user_id, result)
            alert_compliance_team(user_id, address, result)
 
    conn.commit()
 
    print(f"Daily monitoring complete: {len(wallets)} wallets screened")
    print(f"Accounts frozen: {len(blocked_customers)}")
    return blocked_customers

import json
from datetime import datetime
 
def log_screening_decision(
    address: str,
    result: dict,
    decision: str,
    triggered_by: str
):
    """
    Write an immutable compliance audit log entry.
    Store these logs for minimum 5 years per FinCEN requirements.
    """
    log_entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "address": address,
        "decision": decision,
        "match_score": result.get("match_score", 0),
        "lists_checked": result.get("lists_checked", []),
        "matches": result.get("matches", []),
        "triggered_by": triggered_by,
        "api_request_id": result.get("request_id")
    }
 
    # Write to append-only audit log table
    write_audit_log("sanctions_screening_log", log_entry)
 
    # Also log to SIEM for real-time compliance monitoring
    logger.info(f"COMPLIANCE_AUDIT {json.dumps(log_entry)}")