frank-luongt/skills/codex/gcp-aml-ai

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

---

name: gcp-aml-ai description: Google Cloud Anti Money Laundering AI for financial crime detection. Use when implementing AML transaction monitoring, risk scoring, SAR prioritization, or integrating Google's pre-trained AML models with existing compliance systems.

Google Cloud AML AI

Leverage Google Cloud's purpose-built Anti Money Laundering AI for transaction monitoring, risk scoring, and suspicious activity detection with explainable results.

When to Use

• Implementing or modernizing AML transaction monitoring systems
• Replacing rules-based AML alerting with ML-based risk scoring
• Reducing false positives in existing AML alert pipelines (up to 60% reduction)
• Integrating AML risk scores into AI agent workflows for compliance investigation
• Building SAR prioritization and investigation tools

Google AML AI Capabilities

| Capability | Description | |---|---| | Risk Scoring | ML-based customer and transaction risk scores (0-1 scale) | | Alert Prioritization | Rank alerts by actual risk vs. false positives | | Network Analysis | Detect suspicious fund flow patterns across entities | | Explainability | Human-readable reasons for each risk score | | Back-testing | Validate model performance against historical SARs | | Continuous Learning | Model improves with feedback from investigation outcomes |

Patterns

1. AML AI Instance Setup

from google.cloud import financialservices_v1

def create_aml_instance(project_id: str, location: str, instance_id: str) -> dict:
    """Create an AML AI instance for a financial institution."""
    client = financialservices_v1.AMLClient()

    instance = financialservices_v1.Instance(
        display_name=f"AML Instance - {instance_id}",
        kms_key=f"projects/{project_id}/locations/{location}/keyRings/aml-keyring/cryptoKeys/aml-key",
    )

    operation = client.create_instance(
        parent=f"projects/{project_id}/locations/{location}",
        instance=instance,
        instance_id=instance_id,
    )

    return operation.result()

2. Ingest Transaction Data

from google.cloud import bigquery

def prepare_aml_dataset(project_id: str, dataset_id: str):
    """Prepare BigQuery dataset with AML AI required schema.

    AML AI expects transaction data in a specific schema including:
    - Party table (customers/entities)
    - Transaction table (financial transactions)
    - Account table (bank accounts)
    - Party-Account link table
    """
    bq_client = bigquery.Client(project=project_id)

    # Party (customer) table
    party_schema = [
        bigquery.SchemaField("party_id", "STRING", mode="REQUIRED"),
        bigquery.SchemaField("type", "STRING"),  # INDIVIDUAL, ORGANIZATION
        bigquery.SchemaField("registration_date", "TIMESTAMP"),
        bigquery.SchemaField("country", "STRING"),
        bigquery.SchemaField("risk_category", "STRING"),
        bigquery.SchemaField("pep_status", "BOOLEAN"),
    ]

    # Transaction table
    transaction_schema = [
        bigquery.SchemaField("transaction_id", "STRING", mode="REQUIRED"),
        bigquery.SchemaField("type", "STRING"),  # WIRE, ACH, CASH, CHECK
        bigquery.SchemaField("direction", "STRING"),  # DEBIT, CREDIT
        bigquery.SchemaField("amount", "FLOAT64"),
        bigquery.SchemaField("currency", "STRING"),
        bigquery.SchemaField("timestamp", "TIMESTAMP"),
        bigquery.SchemaField("originator_account", "STRING"),
        bigquery.SchemaField("beneficiary_account", "STRING"),
        bigquery.SchemaField("originator_country", "STRING"),
        bigquery.SchemaField("beneficiary_country", "STRING"),
    ]

    table_configs = [
        (f"{dataset_id}.party", party_schema),
        (f"{dataset_id}.transaction", transaction_schema),
    ]

    for table_id, schema in table_configs:
        table = bigquery.Table(f"{project_id}.{table_id}", schema=schema)
        bq_client.create_table(table, exists_ok=True)

3. Run Risk Scoring

from google.cloud import financialservices_v1

def run_risk_scoring(
    project_id: str,
    location: str,
    instance_id: str,
    dataset: str,
    engine_config_id: str,
) -> dict:
    """Execute AML AI risk scoring on transaction data.

    Returns risk scores for all parties with explainability signals.
    """
    client = financialservices_v1.AMLClient()

    parent = f"projects/{project_id}/locations/{location}/instances/{instance_id}"

    # Create engine config (defines model and scoring parameters)
    engine_config = financialservices_v1.EngineConfig(
        display_name="Production AML Scoring",
        engine_version="latest",
        dataset=f"projects/{project_id}/locations/{location}/instances/{instance_id}/datasets/{dataset}",
    )

    # Run prediction
    prediction_result = client.create_prediction_result(
        parent=parent,
        prediction_result=financialservices_v1.PredictionResult(
            engine_config=f"{parent}/engineConfigs/{engine_config_id}",
        ),
    )

    return prediction_result.result()


def get_risk_scores(project_id: str, prediction_table: str, threshold: float = 0.7) -> list[dict]:
    """Query high-risk scores from AML AI prediction results.

    Args:
        project_id: GCP project ID
        prediction_table: BigQuery table with prediction results
        threshold: Risk score threshold (0-1, default 0.7 for high risk)
    """
    from google.cloud import bigquery

    bq = bigquery.Client(project=project_id)
    query = f"""
        SELECT
            party_id,
            risk_score,
            risk_category,
            explanation,
            top_signals
        FROM `{prediction_table}`
        WHERE risk_score >= {threshold}
        ORDER BY risk_score DESC
        LIMIT 100
    """
    results = bq.query(query).result()
    return [dict(row) for row in results]

4. Integrate Risk Scores into Agent Workflow

def check_customer_aml_risk(customer_id: str) -> str:
    """Check AML risk score for a customer from Google AML AI.

    Args:
        customer_id: The customer/party identifier
    """
    from google.cloud import bigquery

    bq = bigquery.Client()
    query = """
        SELECT risk_score, risk_category, explanation, scored_at
        FROM `project.aml_results.latest_scores`
        WHERE party_id = @customer_id
        ORDER BY scored_at DESC
        LIMIT 1
    """
    job_config = bigquery.QueryJobConfig(
        query_parameters=[bigquery.ScalarQueryParameter("customer_id", "STRING", customer_id)]
    )
    rows = list(bq.query(query, job_config=job_config).result())

    if not rows:
        return f"No AML risk score found for customer {customer_id}."

    row = rows[0]
    risk_level = "HIGH" if row["risk_score"] >= 0.7 else "MEDIUM" if row["risk_score"] >= 0.4 else "LOW"

    result = (
        f"Customer {customer_id} AML Risk Assessment:\n"
        f"  Risk Score: {row['risk_score']:.3f} ({risk_level})\n"
        f"  Category: {row['risk_category']}\n"
        f"  Last Scored: {row['scored_at']}\n"
        f"  Explanation: {row['explanation']}"
    )

    if risk_level == "HIGH":
        result += "\n  ACTION REQUIRED: Escalate to compliance officer for enhanced due diligence."

    return result

5. Back-Testing Model Performance

def backtest_aml_model(project_id: str, location: str, instance_id: str, engine_config_id: str) -> dict:
    """Run back-test to validate AML AI model against historical SARs.

    Compares model predictions against known suspicious activity to measure:
    - True positive rate (caught SARs)
    - False positive reduction vs. rules-based system
    - Coverage of different typologies
    """
    client = financialservices_v1.AMLClient()

    parent = f"projects/{project_id}/locations/{location}/instances/{instance_id}"

    backtest = client.create_backtest_result(
        parent=parent,
        backtest_result=financialservices_v1.BacktestResult(
            engine_config=f"{parent}/engineConfigs/{engine_config_id}",
        ),
    )

    result = backtest.result()
    return {
        "true_positive_rate": result.metrics.get("true_positive_rate"),
        "false_positive_reduction": result.metrics.get("false_positive_reduction"),
        "coverage": result.metrics.get("typology_coverage"),
    }

Architecture

Core Banking / Payment System
  --> BigQuery (transaction + party data, AML AI schema)
    --> Google AML AI (ML risk scoring + explainability)
      --> BigQuery (risk scores + signals)
        --> AI Agent (investigation assistant, SAR drafting)
          --> Compliance Dashboard (human review + filing)

Anti-Patterns

• Using AML AI scores as final decisions without human review -- regulatory requirement for human oversight
• Skipping back-testing before production deployment -- model validation is a regulatory expectation
• Not feeding investigation outcomes back to the model -- reduces model improvement over time
• Storing raw PII in AML AI without encryption -- use CMEK and data masking
• Running AML scoring infrequently (monthly) -- high-risk customers need near-real-time monitoring

References

• Google Cloud AML AI
• AML AI API Reference
• AML AI Architecture
• FATF Guidance on AI/ML in AML

<!-- Source: .faos/custom/skills/cloud/gcp/gcp-aml-ai/SKILL.md -->