Education· Last updated April 13, 2026
Build a GENIUS Act Stablecoin Compliance Checker in Python with SanctionShield AI
Step-by-step guide to building an automated stablecoin transaction compliance system in Python. Covers OFAC sanctions screening, AML flagging, and audit trail generation for GENIUS Act compliance.
On April 10, 2026, the U.S. Department of the Treasury published proposed rules in the Federal Register requiring all Permitted Payment Stablecoin Issuers to implement AML programs and real-time sanctions screening equivalent to bank-level compliance obligations. The rules, issued jointly by FinCEN and OFAC under the GENIUS Act signed in July 2025, establish a compliance deadline of January 18, 2027.
For fintech engineers and compliance developers building stablecoin infrastructure, this is an actionable deadline — not an abstract regulatory development. This guide walks through building a production-ready GENIUS Act compliance checker in Python using the SanctionShield AI API.
What GENIUS Act Compliance Requires
The proposed rules impose five core obligations on stablecoin issuers:
- Sanctions screening — every transaction involving a stablecoin issuance, redemption, or transfer must be screened against OFAC's SDN list and other applicable sanctions lists before settlement
- AML program — issuers must maintain a risk-based AML/CFT program meeting Bank Secrecy Act standards
- Transaction monitoring — automated monitoring for suspicious activity patterns with SAR filing capability
- Technical blocking capability — issuers must maintain the technical infrastructure to block, freeze, or reject flagged transactions in real time
- Compliance officer designation — a named US-based compliance officer with no history of fraud or cybercrime
This guide focuses on items 1 and 4 — the sanctions screening and technical blocking infrastructure — which requires real-time API integration with a sanctions data provider.
Architecture Overview
The compliance checker we will build has three components:
┌─────────────────────┐ ┌──────────────────────┐ ┌───────────────────┐
│ Transaction Event │────▶│ Compliance Checker │────▶│ Audit Log Store │
│ (wallet address, │ │ (Python service) │ │ (PostgreSQL or │
│ counterparty, │ │ │ │ S3 + DynamoDB) │
│ amount, type) │ │ 1. Wallet screening │ └───────────────────┘
└─────────────────────┘ │ 2. Entity screening │
│ 3. Risk scoring │ ┌───────────────────┐
│ 4. Block/pass logic │────▶│ Block/Pass Signal│
└──────────────────────┘ │ (to settlement │
│ infrastructure) │
└───────────────────┘
Prerequisites
pip install requests python-dotenv psycopg2-binaryCreate a .env file:
SANCTIONSHIELD_API_KEY=YOUR_API_KEY
DATABASE_URL=postgresql://user:password@localhost/compliance_db
Step 1: Build the Sanctions Screening Client
# compliance/screening_client.py
import requests
import os
from dataclasses import dataclass
from typing import Optional
@dataclass
class ScreeningResult:
screened_value: str
screening_type: str # "wallet" or "entity"
match_found: bool
confidence: float
matched_name: Optional[str]
matched_list: Optional[str]
screening_id: str
list_version: str
class SanctionShieldClient:
BASE_URL = "https://apivult.com/api/sanctionshield"
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"X-RapidAPI-Key": api_key,
"Content-Type": "application/json"
}
def screen_wallet(self, wallet_address: str, chain: str = "ethereum") -> ScreeningResult:
"""Screen a crypto wallet address against OFAC and other sanctions lists."""
response = requests.post(
f"{self.BASE_URL}/screen-wallet",
headers=self.headers,
json={
"wallet_address": wallet_address,
"chain": chain,
"lists": ["OFAC_SDN", "OFAC_CYBER", "OFAC_DPRK", "OFAC_RANSOMWARE"]
},
timeout=5.0
)
response.raise_for_status()
data = response.json()
return ScreeningResult(
screened_value=wallet_address,
screening_type="wallet",
match_found=data["match"],
confidence=data.get("confidence", 0.0),
matched_name=data.get("matched_entity"),
matched_list=data.get("matched_list"),
screening_id=data["screening_id"],
list_version=data["list_version"]
)
def screen_entity(self, name: str, country: Optional[str] = None) -> ScreeningResult:
"""Screen an entity name against global sanctions lists."""
payload = {
"name": name,
"lists": ["OFAC_SDN", "UN_CONSOLIDATED", "EU_RESTRICTIVE", "HMT_CONSOLIDATED"],
"threshold": 85
}
if country:
payload["country"] = country
response = requests.post(
f"{self.BASE_URL}/screen",
headers=self.headers,
json=payload,
timeout=5.0
)
response.raise_for_status()
data = response.json()
return ScreeningResult(
screened_value=name,
screening_type="entity",
match_found=data["match"],
confidence=data.get("confidence", 0.0),
matched_name=data.get("matched_entity"),
matched_list=data.get("matched_list"),
screening_id=data["screening_id"],
list_version=data["list_version"]
)Step 2: Build the Transaction Compliance Checker
# compliance/transaction_checker.py
import logging
from dataclasses import dataclass
from typing import Optional
from datetime import datetime, timezone
from .screening_client import SanctionShieldClient, ScreeningResult
logger = logging.getLogger(__name__)
@dataclass
class Transaction:
transaction_id: str
sender_wallet: str
receiver_wallet: str
counterparty_name: Optional[str]
counterparty_country: Optional[str]
amount_usd: float
transaction_type: str # "issuance", "redemption", "transfer"
chain: str = "ethereum"
@dataclass
class ComplianceDecision:
transaction_id: str
decision: str # "PASS", "BLOCK", "REVIEW"
reason: Optional[str]
wallet_screening: Optional[ScreeningResult]
entity_screening: Optional[ScreeningResult]
timestamp: str
compliance_version: str = "GENIUS_ACT_2026"
class TransactionComplianceChecker:
COMPLIANCE_VERSION = "GENIUS_ACT_2026_v1"
def __init__(self, api_key: str, alert_threshold: float = 85.0):
self.client = SanctionShieldClient(api_key)
self.alert_threshold = alert_threshold
def check_transaction(self, tx: Transaction) -> ComplianceDecision:
"""
Run full GENIUS Act compliance check on a stablecoin transaction.
Returns a ComplianceDecision with PASS, BLOCK, or REVIEW status.
"""
timestamp = datetime.now(timezone.utc).isoformat()
wallet_result = None
entity_result = None
# Screen receiver wallet address
try:
wallet_result = self.client.screen_wallet(tx.receiver_wallet, tx.chain)
if wallet_result.match_found and wallet_result.confidence >= self.alert_threshold:
return ComplianceDecision(
transaction_id=tx.transaction_id,
decision="BLOCK",
reason=f"Receiver wallet {tx.receiver_wallet} matched {wallet_result.matched_list} "
f"(entity: {wallet_result.matched_name}, confidence: {wallet_result.confidence:.1f}%)",
wallet_screening=wallet_result,
entity_screening=None,
timestamp=timestamp,
compliance_version=self.COMPLIANCE_VERSION
)
except Exception as e:
logger.error(f"Wallet screening failed for {tx.receiver_wallet}: {e}")
# Fail closed on screening errors — block the transaction
return ComplianceDecision(
transaction_id=tx.transaction_id,
decision="BLOCK",
reason=f"Screening service error — transaction blocked per fail-safe policy",
wallet_screening=None,
entity_screening=None,
timestamp=timestamp,
compliance_version=self.COMPLIANCE_VERSION
)
# Screen counterparty entity name if provided
if tx.counterparty_name:
try:
entity_result = self.client.screen_entity(
tx.counterparty_name,
tx.counterparty_country
)
if entity_result.match_found:
if entity_result.confidence >= self.alert_threshold:
return ComplianceDecision(
transaction_id=tx.transaction_id,
decision="BLOCK",
reason=f"Counterparty '{tx.counterparty_name}' matched {entity_result.matched_list} "
f"(confidence: {entity_result.confidence:.1f}%)",
wallet_screening=wallet_result,
entity_screening=entity_result,
timestamp=timestamp,
compliance_version=self.COMPLIANCE_VERSION
)
else:
# Below block threshold but above zero — flag for human review
return ComplianceDecision(
transaction_id=tx.transaction_id,
decision="REVIEW",
reason=f"Potential match for '{tx.counterparty_name}' on {entity_result.matched_list} "
f"(confidence: {entity_result.confidence:.1f}%) — requires analyst review",
wallet_screening=wallet_result,
entity_screening=entity_result,
timestamp=timestamp,
compliance_version=self.COMPLIANCE_VERSION
)
except Exception as e:
logger.error(f"Entity screening failed for {tx.counterparty_name}: {e}")
return ComplianceDecision(
transaction_id=tx.transaction_id,
decision="BLOCK",
reason="Screening service error — transaction blocked per fail-safe policy",
wallet_screening=wallet_result,
entity_screening=None,
timestamp=timestamp,
compliance_version=self.COMPLIANCE_VERSION
)
# All checks passed
return ComplianceDecision(
transaction_id=tx.transaction_id,
decision="PASS",
reason=None,
wallet_screening=wallet_result,
entity_screening=entity_result,
timestamp=timestamp,
compliance_version=self.COMPLIANCE_VERSION
)Step 3: Build the Audit Trail Logger
GENIUS Act compliance requires documented evidence of every screening decision. The audit trail logger writes immutable records to a database.
# compliance/audit_logger.py
import json
import psycopg2
import os
from dataclasses import asdict
from .transaction_checker import ComplianceDecision, Transaction
CREATE_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS compliance_audit_log (
id SERIAL PRIMARY KEY,
transaction_id VARCHAR(255) NOT NULL,
decision VARCHAR(20) NOT NULL,
reason TEXT,
wallet_screening JSONB,
entity_screening JSONB,
transaction_data JSONB,
screened_at TIMESTAMPTZ NOT NULL,
compliance_version VARCHAR(50) NOT NULL,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_audit_transaction_id ON compliance_audit_log(transaction_id);
CREATE INDEX IF NOT EXISTS idx_audit_decision ON compliance_audit_log(decision);
CREATE INDEX IF NOT EXISTS idx_audit_screened_at ON compliance_audit_log(screened_at);
"""
class AuditLogger:
def __init__(self, database_url: str):
self.database_url = database_url
self._init_db()
def _init_db(self):
with psycopg2.connect(self.database_url) as conn:
with conn.cursor() as cur:
cur.execute(CREATE_TABLE_SQL)
conn.commit()
def log_decision(self, tx: Transaction, decision: ComplianceDecision):
"""Write an immutable audit record for a compliance screening decision."""
with psycopg2.connect(self.database_url) as conn:
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO compliance_audit_log
(transaction_id, decision, reason, wallet_screening,
entity_screening, transaction_data, screened_at, compliance_version)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
""",
(
decision.transaction_id,
decision.decision,
decision.reason,
json.dumps(asdict(decision.wallet_screening)) if decision.wallet_screening else None,
json.dumps(asdict(decision.entity_screening)) if decision.entity_screening else None,
json.dumps(asdict(tx)),
decision.timestamp,
decision.compliance_version
)
)
conn.commit()Step 4: Putting It Together — The Compliance Service
# compliance/service.py
import os
from dotenv import load_dotenv
from .transaction_checker import Transaction, TransactionComplianceChecker
from .audit_logger import AuditLogger
load_dotenv()
class GeniusActComplianceService:
def __init__(self):
self.checker = TransactionComplianceChecker(
api_key=os.environ["SANCTIONSHIELD_API_KEY"],
alert_threshold=85.0
)
self.audit_logger = AuditLogger(
database_url=os.environ["DATABASE_URL"]
)
def process_transaction(self, tx: Transaction) -> dict:
"""
Full GENIUS Act compliance check with audit logging.
Returns the compliance decision. Raises no exceptions — all errors result in BLOCK.
"""
decision = self.checker.check_transaction(tx)
self.audit_logger.log_decision(tx, decision)
return {
"transaction_id": decision.transaction_id,
"decision": decision.decision,
"reason": decision.reason,
"timestamp": decision.timestamp,
"blocked": decision.decision == "BLOCK"
}
# Example usage
if __name__ == "__main__":
service = GeniusActComplianceService()
# Test with a sample transaction
tx = Transaction(
transaction_id="txn_20260413_001",
sender_wallet="0xYourWalletAddress",
receiver_wallet="0xReceiverWalletAddress",
counterparty_name="Acme Trading LLC",
counterparty_country="US",
amount_usd=50000.00,
transaction_type="issuance",
chain="ethereum"
)
result = service.process_transaction(tx)
print(f"Transaction {result['transaction_id']}: {result['decision']}")
if result['reason']:
print(f"Reason: {result['reason']}")Step 5: Add a FastAPI Endpoint for Real-Time Screening
# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
from compliance.service import GeniusActComplianceService
from compliance.transaction_checker import Transaction
app = FastAPI(title="GENIUS Act Compliance API")
service = GeniusActComplianceService()
class TransactionRequest(BaseModel):
transaction_id: str
sender_wallet: str
receiver_wallet: str
counterparty_name: Optional[str] = None
counterparty_country: Optional[str] = None
amount_usd: float
transaction_type: str
chain: str = "ethereum"
@app.post("/compliance/check")
def check_transaction(request: TransactionRequest):
tx = Transaction(**request.model_dump())
result = service.process_transaction(tx)
if result["blocked"]:
# Return 403 for blocked transactions — settlement infrastructure
# should treat any non-200 as a block signal
raise HTTPException(status_code=403, detail=result)
return resultRunning the Compliance Checker
uvicorn main:app --host 0.0.0.0 --port 8000Screen a transaction:
curl -X POST http://localhost:8000/compliance/check \
-H "Content-Type: application/json" \
-d '{
"transaction_id": "txn_20260413_001",
"sender_wallet": "0xYourWallet",
"receiver_wallet": "0xReceiverWallet",
"counterparty_name": "Acme Trading LLC",
"amount_usd": 50000,
"transaction_type": "issuance"
}'GENIUS Act Compliance Checklist
Before the January 18, 2027 deadline, stablecoin issuers need to validate:
- Real-time screening — every transaction screened before settlement
- Multi-list coverage — OFAC SDN plus sector-specific lists (DPRK, ransomware)
- Wallet address screening — not just entity name matching
- Technical blocking — infrastructure can reject flagged transactions before settlement
- Fail-safe behavior — screening service outages result in block, not pass-through
- Audit trail — every screening decision logged with list version, confidence score, and disposition
- Review workflow — ambiguous matches routed to compliance analyst, not auto-blocked or auto-passed
- SAR filing capability — suspicious transactions can be reported to FinCEN
Key Design Decisions
Why fail closed? The compliance checker blocks transactions when the screening service is unavailable. This is the correct behavior for GENIUS Act compliance — a momentary service outage does not constitute authorization to proceed with an unscreened transaction.
Why 85% confidence threshold for auto-block? A lower threshold generates excessive false positives that overwhelm compliance analysts. A higher threshold risks missing high-confidence matches. 85% is a common starting point; organizations should calibrate based on their risk tolerance and analyst capacity.
Why immutable audit logs? OFAC examinations and FinCEN reviews require evidence that screening occurred at the time of the transaction, against a specific list version, and that the disposition was documented. Mutable logs cannot provide this assurance.
What's Next
This implementation covers the core screening requirement. Production deployments should add:
- Continuous monitoring — re-screen existing customer wallets daily against updated sanctions lists
- SAR workflow integration — route REVIEW decisions and high-risk patterns to a SAR filing workflow
- Rate limiting and circuit breaking — protect the screening service from volume spikes
- Webhook callbacks for async batch screening during customer onboarding
Get started with SanctionShield AI at APIVult. The API supports free-tier evaluation and scales to production GENIUS Act compliance volumes with pay-per-call pricing.
More Articles
Real-Time AML Sanctions Screening in Python: A Complete Integration Guide
Real-time sanctions screening against OFAC, UN, EU lists. Integrate SanctionShield API for AML/KYC in Python.
March 31, 2026
How to Screen Cryptocurrency Wallet Addresses for Sanctions Compliance in Python
Build a production-ready crypto sanctions screening pipeline using SanctionShield AI API — check wallet addresses against OFAC, EU, and UN watchlists in real time.
April 4, 2026