vabhishek8/aml-transaction-monitoring-pipeline

# AML 交易监控 Pipeline [](https://github.com/vabhishek8/aml-transaction-monitoring-pipeline/actions/workflows/pipeline.yml) 一个基于规则的交易监控 pipeline，用于在合成的零售银行交易数据流中检测五种真实的 AML 类型（Structuring、Layering、Impossible-travel、统计金额异常值、 High-risk-corridor routing）——通过案例级别的检测测试进行验证，而不仅仅是“能跑通”。 **[实时仪表盘 →](https://vabhishek8.github.io/aml-transaction-monitoring-pipeline/)** 本项目作为 [azure-medallion-weather-pipeline](https://github.com/vabhishek8/azure-medallion-weather-pipeline) 的姊妹篇而构建， 专门针对银行/金融服务数据工程：检测逻辑围绕 AML/BSA 合规的实际需求设计， 且 Azure IaC 的架构设计符合银行 InfoSec 审查的门槛要求（无公共网络暴露、 不可变的审计存储、集中的治理），而不是为了实现最低成本。 ## 本项目存在的原因 “我在 Kaggle 信用卡数据集上训练了一个欺诈模型”并不能证明你具备银行领域的 判断力——它只能证明你会调用 `.fit()`。本项目的范围围绕交易监控工程师 实际需要回答的问题展开： - 当生产环境中没有真实标签（ground truth）时，检测规则的*证据*应该是什么样的？ - 如何在不让下游调查员被误报（false positives）淹没的情况下， 验证规则能捕捉到其设计针对的模式？ - 监管记录保存（不可变性、血缘、留存）从何处开始成为 一种数据架构需求，而不是合规的事后补救？ 这里没有使用也不可能使用真实的交易数据——正确的公开数据根本不存在。 取而代之的是，`src/generate_transactions.py` 合成了一批真实的交易数据，并 *刻意注入*了五种已知的 AML 类型及其记录在案的真实标签（ground truth）， 因此检测 SQL（`src/generate_transactions.py`）可以像真实的交易监控团队在将新规则 投入生产数据前验证它们那样进行验证：已知场景注入、测量的召回率和误报率。 ## 架构 ``` flowchart LR subgraph GEN["Synthetic core-banking feed"] G["600 customers x ~90 days
5 injected AML typologies"] end subgraph BRONZE["Bronze — raw"] B["Raw transaction batch (JSONL)"] end subgraph SILVER["Silver — validated"] QC{{"Quality gate
schema · nulls · domain · range · dupes"}} S["transactions.parquet"] end subgraph GOLD["Gold — risk-scored"] D1["Structuring
(48h rolling window SQL)"] D2["Layering
(self-join, wire in/out pairing)"] D3["Impossible travel
(LAG + haversine distance)"] D4["Amount outlier
(per-customer robust z-score)"] D5["High-risk corridor
(jurisdiction/category rule)"] SCORE["Composite risk_score (0-100)"] end subgraph SERVE["Serve"] ALERTS["Alert queue
(SAR-candidate list)"] DASH["Static dashboard
(GitHub Pages)"] end G --> B --> QC QC -- pass --> S QC -- fail: abort write --> FAIL["Non-zero exit, CI fails"] S --> D1 & D2 & D3 & D4 & D5 --> SCORE --> ALERTS --> DASH ``` 每个检测信号都在 `tests/test_gold.py` 中针对注入的真实标签进行了独立的单元测试—— 而且该测试文件正是本仓库中最值得优先阅读的部分。 ## 检测质量（经过测量，而非断言） 召回率是在**案例级别**进行测量的：该方案是否在其任意一笔交易中触发了警报？ 真实的监控系统会在模式积累了足够的证据后发出一次警报，而不是对其每一个环节进行追溯警报—— 测试行级别的召回率测试的是错误的对象。 | 类型 | 案例级召回率 | 备注 | |---|---|---| | Structuring | 100% | 当累计低于 CTR 阈值的存款在滚动的 48 小时窗口内超过约 90% 的 10,000 澳元时触发警报 | | Layering | 100% | 自连接（Self-join）将汇入款项与同一客户在 6 小时内汇出的金额相近的款项配对 | | Impossible travel | 100% | Haversine 距离 / 流逝时间对比该客户紧邻的前一笔交易，阈值设定在高于商业航班速度以上 | | Amount outlier | 100% | 每个客户的 **median/MAD** z-score——见下文，这取代了原先简单的 mean/stddev 版本 | | High-risk corridor | 100% | 确定性的司法管辖区/商户类别规则 | 干净（非注入）交易的误报率：总体为 **7.9%**， 但这些“误报”中有 86% 是 `high_risk_corridor` 正确地在真正的高风险司法管辖区交易上触发， 这些交易只是不属于预设的场景——并不是错误。`impossible_travel` 规则产生了真实的、 可解释的约 1.5% 的偶然误报率，这是由于完全随机的合成时间戳偶尔发生巧合聚集造成的； 在生产环境中，这将根据实际的客户旅行历史和卡在场（card-present）认证信号进行收紧， 而不是针对合成数据的假象进行调优。 ### 抓到的一个真实 Bug：异常值检测基于自身异常值进行测量 第一版 `amount_outlier` 使用了基于客户级别的 mean/stddev z-score。召回率只有 20%——因为异常值交易本身会抬高用于测量它的样本标准差，对于交易记录较少的客户 （小样本的 Bessel 校正放大了这种效应）来说情况最糟。改用 **median/MAD（绝对中位差）** 稳健统计量（能够抵抗少量极端数值对基线的污染）后，召回率从 20% 提升到了 100%，同时干净数据的误报率为 0.01%。`tests/test_gold.py::test_amount_outlier_uses_robust_stat_not_skewed_by_own_outlier` 是专门针对这种失败模式的回归防护，主要针对低交易量客户，因为这种问题最容易在那里再次出现。 ## 生产级 Azure 映射 `infra/main.bicep` 将相同的 bronze/silver/gold 设计转换为受治理的 Azure 资产—— 但这里的架构决策不同于普通的“部署到 Azure”模板，因为数据类别有所不同： | 决策 | 原因 | |---|---| | 所有服务均无公共网络访问（存储、Key Vault、ADF、Synapse） | 每个承载数据的服务都位于专用 VNet 中的 private endpoint 之后。这是银行 InfoSec 审查期望的默认姿态，而不是可选的加固步骤。 | | 开启版本控制的账户级不可变存储 | 交易记录以及从中得出的警报需要具有无可辩驳的监管链——AML/CTF 法案和 SAR 记录保存义务是数据架构层面的要求，而不仅仅是政策文件。 | | Microsoft Purview | 跨 bronze → silver → gold 的集中化血缘/分类。BCBS 239 的风险数据聚合原则从根本上说是关于可证明的数据血缘和所有权，而不是建模的准确性。 | | RBAC 授权的 Key Vault、开启清除保护、90 天软删除 | 没有遗留的访问策略；每个身份都被授予限定在单一资源范围内的最低角色（`Key Vault Secrets User`、`Storage Blob Data Contributor`）。 | | Synapse **managed virtual network** | 即使是服务内部调用，计算到存储的流量也永远不会流经公共互联网。 | | 生产环境中 365 天的 Log Analytics 留存（相比之下，仅针对运营的基线为 60 天） | 这是审计追踪，而不仅仅是运营遥测数据。 | | 不进行常驻部署及空转 | 一个持有交易型数据（即使是合成的）且没有活跃监控负责人的常驻环境，在大多数银行安全审查中本身就是一项发现——这与姊妹篇天气 pipeline 项目的成本控制理由相同，只是增加了合规视角。 | 已使用 `bicep build` 验证（0 个错误，25 个资源）。按需部署： ``` az deployment group create \ --resource-group rg-aml-pipeline-dev \ --template-file infra/main.bicep \ --parameters environment=dev alertEmail=you@example.com ``` ## 本地运行 ``` python -m venv .venv && source .venv/bin/activate pip install -r requirements.txt python src/pipeline.py # generate -> silver -> gold -> dashboard PYTHONPATH=src pytest tests/ -v # 29 tests: quality gate, generator invariants, detection recall/FP rate open docs/index.html ``` ## 仓库结构 ``` src/ generate_transactions.py synthetic core-banking feed, 5 injected AML typologies with recorded ground truth quality_checks.py the silver quality gate transform.py bronze -> silver: parse, validate, dedupe gold.py risk-scoring SQL: structuring, layering, impossible travel, outlier, corridor dashboard.py renders gold -> static Plotly HTML (risk distribution, alert queue) pipeline.py orchestrates all four stages tests/ 29 pytest cases -- including case-level recall/FP-rate assertions against ground truth infra/main.bicep production Azure IaC: private-endpoint-only, immutable storage, Purview, RBAC .github/workflows/ scheduled CI: test -> run -> commit refreshed gold data ``` ## 技术栈 Python · pandas · DuckDB（窗口函数、自连接、haversine SQL） · Plotly · pytest · GitHub Actions · Bicep（VNet + private endpoints、ADLS Gen2 不可变存储、Key Vault RBAC、 Azure Data Factory、Synapse managed VNet、Microsoft Purview、Log Analytics） 由 [Abhishek Vadlamudi](https://abhishekvadlamudi.com) 构建——一位正致力于在金融服务领域发展 Azure Data Engineering 的高级 BI 工程师。

标签：Azure, SQL, 反洗钱, 安全规则引擎, 异常检测, 数据工程, 系统审计, 逆向工具