Anti-Money Laundering Surveillance Metrics Pipeline Script

Role: Senior Financial Crime Analytics Architect specializing in automated AML surveillance reporting and FinCEN telemetry feeds.

Context

• Monitoring Engine: {{transaction_monitoring_system}}
• Compliance Jurisdiction: {{compliance_jurisdiction}}
• Rolling Time Window: {{lookback_period_days}}
• Alert Threshold Score: {{alert_severity_cutoff}}
• Compliance Mart Schema: {{target_datamart_schema}}
• PII Obfuscation Standard: {{pii_masking_standard}}

Task

Write an enterprise-grade PySpark ETL and surveillance extraction script that aggregates flagged high-risk transaction typologies from {{transaction_monitoring_system}}, applies {{pii_masking_standard}}, and populates daily executive compliance monitoring tables in {{target_datamart_schema}}.

Method

1. Ingest raw transaction events and model anomaly scores from {{transaction_monitoring_system}} across {{lookback_period_days}}.
2. Filter events exceeding {{alert_severity_cutoff}} while tagging structuring, rapid movement of funds, and sanction screening hits.
3. Join customer KYC metadata, calculating cumulative velocity metrics by beneficiary jurisdiction.
4. Apply irreversible cryptographic hashing or pseudonymization under {{pii_masking_standard}}.
5. Standardize typological categorization according to {{compliance_jurisdiction}} regulatory definitions.
6. Execute delta-merge statements to insert and update aggregated surveillance records in {{target_datamart_schema}}.
7. Generate automated data quality verification checks ensuring zero duplicated alert references.

Constraints

• Code MUST use distributed PySpark DataFrame transformations and avoid non-vectorized Python UDFs.
• PII MUST NOT be written to intermediate staging tables in unmasked plaintext.
• All timestamp manipulations MUST standardize on UTC with microsecond precision.
• Pipeline MUST execute within an idempotent transaction boundary with automated failure recovery.

Output format

1. Architecture & Schema Definition (PySpark StructType schemas for input and output datasets)
2. Distributed PySpark ETL Script (Fully written, parameterized batch script with data cleansing, scoring, and masking logic)
3. Target Upsert SQL Script (Idempotent MERGE statement targeted at {{target_datamart_schema}})
4. Data Quality Audit Script (Automated validation script checking record counts, null checks, and hash entropy)

Self-review

• Confirm all 6 variables ({{transaction_monitoring_system}}, {{compliance_jurisdiction}}, {{lookback_period_days}}, {{alert_severity_cutoff}}, {{target_datamart_schema}}, {{pii_masking_standard}}) are implemented.
• Ensure PII masking logic strictly satisfies financial privacy regulations.
• Check that PySpark partition strategies are configured to avoid shuffle skew on high-volume days.