frank-luongt/skills/codex/aws-financial-services

<!-- AUTO-GENERATED by export-skills.py — DO NOT EDIT -->

---

name: aws-financial-services description: AWS financial services AI patterns for fraud detection, document processing, and compliance. Use when building AI agents with Amazon Fraud Detector, Textract for KYC, Clean Rooms for data collaboration, or FinSpace for quantitative analytics.

Platform Note: This skill was designed for multi-agent execution. In Codex, treat sub-agent instructions as sequential steps to complete thoroughly within a single agent context.

AWS Financial Services AI

Build AI-powered financial services applications using AWS purpose-built services: Fraud Detector, Textract for KYC, Clean Rooms for data collaboration, and FinSpace for analytics.

When to Use

• Implementing real-time fraud detection for payments, account creation, or lending
• Automating KYC document processing (ID verification, bank statements, tax forms)
• Setting up privacy-safe data collaboration between financial institutions
• Building quantitative analytics pipelines for financial data
• Integrating AWS financial AI services as tools for Bedrock Agents

AWS Financial Services AI Stack

| Service | Use Case | Key Feature | |---|---|---| | Fraud Detector | Real-time fraud scoring | AutoML + rules engine, no ML expertise needed | | Textract | Document data extraction | Tables, forms, signatures, lending-specific models | | Clean Rooms | Multi-party data analysis | Query data without sharing raw records | | FinSpace | Financial analytics | Time-series, capital markets data management | | Macie | PII detection in S3 | Automated sensitive data discovery | | Comprehend | NLP on financial text | Entity extraction, sentiment, PII redaction |

Patterns

1. Amazon Fraud Detector

import boto3
import json

fraud_detector = boto3.client("frauddetector")

def setup_fraud_model(detector_name: str, event_type: str):
    """Set up a fraud detection model for transaction scoring.

    Fraud Detector uses AutoML -- provide labeled historical data and it
    trains a model automatically.
    """
    # Create event type (defines what data your fraud events contain)
    fraud_detector.put_event_type(
        name=event_type,
        eventVariables=[
            {"name": "ip_address"},
            {"name": "email_address"},
            {"name": "transaction_amount"},
            {"name": "card_bin"},
            {"name": "customer_age_days"},
        ],
        labels=[{"name": "fraud"}, {"name": "legit"}],
        entityTypes=[{"name": "customer"}],
    )

    # Create detector
    fraud_detector.put_detector(
        detectorId=detector_name,
        eventTypeName=event_type,
    )


def score_transaction(detector_id: str, detector_version: str, transaction: dict) -> dict:
    """Score a transaction for fraud risk in real-time.

    Args:
        detector_id: Fraud Detector detector ID
        detector_version: Detector version
        transaction: Transaction data matching event variables
    """
    response = fraud_detector.get_event_prediction(
        detectorId=detector_id,
        detectorVersionId=detector_version,
        eventId=transaction["transaction_id"],
        eventTypeName="online_transaction",
        entities=[{"entityType": "customer", "entityId": transaction["customer_id"]}],
        eventTimestamp=transaction["timestamp"],
        eventVariables={
            "ip_address": transaction["ip_address"],
            "email_address": transaction["email"],
            "transaction_amount": str(transaction["amount"]),
            "card_bin": transaction.get("card_bin", ""),
            "customer_age_days": str(transaction.get("account_age_days", 0)),
        },
    )

    # Get outcome and score
    rule_results = response.get("ruleResults", [])
    model_scores = response.get("modelScores", [])

    outcome = rule_results[0]["outcomes"][0] if rule_results else "review"
    score = model_scores[0]["scores"].get("fraud_score", 0) if model_scores else 0

    return {
        "transaction_id": transaction["transaction_id"],
        "fraud_score": score,
        "outcome": outcome,  # approve, review, block
        "rule_triggered": rule_results[0].get("ruleId") if rule_results else None,
    }

2. Amazon Textract for KYC Documents

import boto3

textract = boto3.client("textract")

def extract_id_document(s3_bucket: str, s3_key: str) -> dict:
    """Extract data from an identity document (passport, driver's license).

    Args:
        s3_bucket: S3 bucket containing the document image
        s3_key: S3 object key for the document
    """
    response = textract.analyze_id(
        DocumentPages=[{
            "S3Object": {"Bucket": s3_bucket, "Name": s3_key}
        }]
    )

    extracted = {}
    for doc in response.get("IdentityDocuments", []):
        for field in doc.get("IdentityDocumentFields", []):
            field_type = field.get("Type", {}).get("Text", "")
            field_value = field.get("ValueDetection", {}).get("Text", "")
            confidence = field.get("ValueDetection", {}).get("Confidence", 0)
            extracted[field_type] = {"value": field_value, "confidence": confidence}

    return extracted


def extract_bank_statement(s3_bucket: str, s3_key: str) -> dict:
    """Extract tables and key-value pairs from a bank statement.

    Uses Textract's Lending API for financial document analysis.
    """
    response = textract.analyze_document(
        Document={"S3Object": {"Bucket": s3_bucket, "Name": s3_key}},
        FeatureTypes=["TABLES", "FORMS"],
    )

    key_values = {}
    tables = []

    for block in response.get("Blocks", []):
        if block["BlockType"] == "KEY_VALUE_SET" and "KEY" in block.get("EntityTypes", []):
            key = _get_text_from_block(block, response["Blocks"])
            value_block = _find_value_block(block, response["Blocks"])
            if value_block:
                value = _get_text_from_block(value_block, response["Blocks"])
                key_values[key] = value

    return {"key_values": key_values, "table_count": len(tables)}


def extract_lending_document(s3_bucket: str, s3_key: str) -> str:
    """Process lending documents (1003, 1099, W2, pay stubs) using Textract Lending.

    Starts an async job for multi-page lending document analysis.
    """
    response = textract.start_lending_analysis(
        DocumentLocation={"S3Object": {"Bucket": s3_bucket, "Name": s3_key}},
    )
    return response["JobId"]

3. AWS Clean Rooms for Data Collaboration

import boto3

cleanrooms = boto3.client("cleanrooms")

def create_collaboration(name: str, members: list[dict]) -> dict:
    """Create a Clean Rooms collaboration for cross-institution fraud analysis.

    Allows multiple banks to query shared fraud signals without exposing
    raw customer data.

    Args:
        name: Collaboration name
        members: List of member configs with account IDs and roles
    """
    response = cleanrooms.create_collaboration(
        name=name,
        description="Cross-institution fraud signal sharing",
        creatorMemberAbilities=["CAN_QUERY", "CAN_RECEIVE_RESULTS"],
        creatorDisplayName="Bank A",
        members=[{
            "accountId": m["account_id"],
            "memberAbilities": m.get("abilities", ["CAN_QUERY"]),
            "displayName": m["name"],
        } for m in members],
        queryLogStatus="ENABLED",  # Required for compliance audit trail
    )
    return response["collaboration"]


def run_fraud_analysis_query(membership_id: str) -> dict:
    """Run a privacy-safe query to identify shared fraud indicators.

    Example: Find transaction patterns that appear across multiple
    institutions without revealing customer identities.
    """
    response = cleanrooms.start_protected_query(
        type="SQL",
        membershipIdentifier=membership_id,
        sqlParameters={
            "queryString": """
                SELECT
                    ip_address_hash,
                    COUNT(DISTINCT institution_id) as institution_count,
                    SUM(flagged_transactions) as total_flags
                FROM fraud_signals
                GROUP BY ip_address_hash
                HAVING COUNT(DISTINCT institution_id) >= 2
                ORDER BY total_flags DESC
                LIMIT 100
            """
        },
        resultConfiguration={
            "outputConfiguration": {
                "s3": {"bucket": "my-cleanrooms-results", "keyPrefix": "fraud-analysis/"}
            }
        },
    )
    return response["protectedQuery"]

4. Amazon Comprehend for Financial NLP

import boto3

comprehend = boto3.client("comprehend")

def redact_pii_from_text(text: str) -> str:
    """Detect and redact PII from financial text before sending to LLMs.

    Args:
        text: Raw text that may contain PII
    """
    response = comprehend.detect_pii_entities(
        Text=text,
        LanguageCode="en",
    )

    redacted = text
    # Process entities in reverse order to preserve positions
    for entity in sorted(response["Entities"], key=lambda e: e["BeginOffset"], reverse=True):
        start = entity["BeginOffset"]
        end = entity["EndOffset"]
        pii_type = entity["Type"]  # SSN, CREDIT_DEBIT_NUMBER, BANK_ACCOUNT_NUMBER, etc.
        redacted = redacted[:start] + f"[{pii_type}]" + redacted[end:]

    return redacted


def analyze_financial_sentiment(text: str) -> dict:
    """Analyze sentiment of financial communications (earnings calls, reports).

    Args:
        text: Financial text to analyze
    """
    response = comprehend.detect_sentiment(Text=text, LanguageCode="en")
    return {
        "sentiment": response["Sentiment"],
        "scores": response["SentimentScore"],
    }

5. Agent Tool Integration

def check_fraud_risk(transaction_id: str, amount: float, ip_address: str, email: str) -> str:
    """Check fraud risk for a transaction using Amazon Fraud Detector.

    Args:
        transaction_id: Unique transaction identifier
        amount: Transaction amount
        ip_address: Customer's IP address
        email: Customer's email address
    """
    result = score_transaction(
        detector_id="payment-fraud-detector",
        detector_version="1",
        transaction={
            "transaction_id": transaction_id,
            "customer_id": "derived-from-email",
            "timestamp": "2026-02-28T12:00:00Z",
            "ip_address": ip_address,
            "email": email,
            "amount": amount,
        },
    )

    score = result["fraud_score"]
    outcome = result["outcome"]

    if outcome == "block":
        return f"BLOCKED: Transaction {transaction_id} scored {score:.0f}/1000. High fraud risk detected. Transaction declined."
    elif outcome == "review":
        return f"REVIEW: Transaction {transaction_id} scored {score:.0f}/1000. Requires manual review before approval."
    else:
        return f"APPROVED: Transaction {transaction_id} scored {score:.0f}/1000. Low risk."


def process_kyc_document(s3_bucket: str, document_key: str, document_type: str) -> str:
    """Process a KYC document (ID, bank statement) using Amazon Textract.

    Args:
        s3_bucket: S3 bucket with the uploaded document
        document_key: S3 key for the document
        document_type: Type: 'id_document', 'bank_statement', 'lending'
    """
    if document_type == "id_document":
        data = extract_id_document(s3_bucket, document_key)
        fields = [f"  {k}: {v['value']} (confidence: {v['confidence']:.1f}%)" for k, v in data.items()]
        return f"ID Document extracted:\n" + "\n".join(fields)
    elif document_type == "bank_statement":
        data = extract_bank_statement(s3_bucket, document_key)
        return f"Bank statement extracted: {len(data['key_values'])} fields, {data['table_count']} tables found."
    else:
        job_id = extract_lending_document(s3_bucket, document_key)
        return f"Lending document analysis started. Job ID: {job_id}. Check status with Textract API."

Anti-Patterns

• Using Fraud Detector without sufficient labeled training data (< 10K events) -- model quality suffers
• Sending raw PII to Bedrock/external LLMs without Comprehend PII redaction -- compliance violation
• Running Textract synchronously on large documents -- use async APIs for multi-page documents
• Not enabling query logging in Clean Rooms -- audit trail required for regulatory compliance
• Hardcoding fraud thresholds -- use Fraud Detector's rules engine for adjustable decisioning

References

• Amazon Fraud Detector
• Amazon Textract
• AWS Clean Rooms
• Amazon Comprehend
• AWS Financial Services

<!-- Source: .faos/custom/skills/cloud/aws/aws-financial-services/SKILL.md -->