# AWS Financial Services > - Skill: `frank-luongt/aws-financial-services-2` (Agent Skill) - Install (CLI): `npx skillmds@latest add frank-luongt/aws-financial-services-2` - Raw SKILL.md: https://api.skillmd.com/api/skills/frank-luongt/aws-financial-services-2/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: DevOps & Infra - Author: frank-luongt (https://skillmd.com/u/frank-luongt) - Updated: 2026-09-21 - Page: https://skillmd.com/skills/frank-luongt/aws-financial-services-2 --- --- 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. tags: [aws, fraud, textract, kyc, finspace, financial-services] --- # 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 ```python 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 ```python 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 ```python 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 ```python 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 ```python 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](https://docs.aws.amazon.com/frauddetector/latest/ug/what-is-frauddetector.html) - [Amazon Textract](https://docs.aws.amazon.com/textract/latest/dg/what-is.html) - [AWS Clean Rooms](https://docs.aws.amazon.com/clean-rooms/latest/userguide/what-is.html) - [Amazon Comprehend](https://docs.aws.amazon.com/comprehend/latest/dg/what-is.html) - [AWS Financial Services](https://aws.amazon.com/financial-services/)