hiendat04/fraudstream

# FraudStream：实时金融交易智能平台 FraudStream 是一个数据工程和 MLOps 项目，旨在构建一个生产级的欺诈分析平台。它会模拟杂乱的金融交易数据，保留原始源记录，并为基于 Spark 的 lakehouse 处理、欺诈特征工程、模型训练和运营监控奠定基础。 该项目围绕一个现实世界中常见的问题而构建：在模型训练开始之前，欺诈模型高度依赖于可靠的数据。交易数据可能会延迟到达、出现重复、发生 schema 偏移、在短暂的流量突增期间激增、包含缺失值，或者严重倾斜至少数商家和城市。FraudStream 将这些问题转化为明确的工程需求。 ## 本项目展示的内容 FraudStream 旨在展示实用的数据平台技能： - 合成但逼真的金融交易数据生成 - 源格式的原始数据分区及可复现的生成器配置 - 针对客户、账户、商家、交易和设备实体的高基数标识符 - 为下游 Bronze 和 Silver 处理制造逼真的数据质量问题 - 跨历史源分区的 schema 演进 - 包含 `event_timestamp` 和 `created_ts` 的事件时间和到达时间语义 - 一条面向 lakehouse 的路径，通向 Parquet、Spark、特征表和模型操作 ## 当前实现 当前代码库实现了离线源数据生成器。它会创建分区的 CSV 文件，位于： ``` data/raw_source/offline_transactions/ ``` 生成的源数据集当前包含： | 指标 | 数值 | |---|---:| | 基础交易数 | 500,000 | | 注入重复项后的行数 | 510,000 | | 每日 CSV 分区数 | 180 | | 本地输出大小 | ~85 MB | | 重复行数 | 10,000 | | 延迟到达行数 | 20,574 | | 包含缺失值的行数 | 7,655 | | 格式不一致的行数 | 5,076 | | 欺诈行数 | 7,803 | | 欺诈团伙行数 | 3,060 | | 独立客户数 | 197,404 | | 独立商家数 | 40,000 | | 独立设备数 | 193,863 | 生成器会将证据工件写入源数据旁边： ``` data/raw_source/offline_transactions/_manifest.json data/raw_source/offline_transactions/_quality_summary.json data/raw_source/offline_transactions/_quality_summary.csv ``` ## 系统架构 FraudStream 遵循分层的数据平台设计。当前实现主要生成原始源数据。处理架构设计为可扩展至 Spark、Flink、Airflow、DataHub、MLflow 和监控组件。 ``` flowchart LR generator[Data Generator] raw[Raw Source Files] bronze[Bronze Parquet] silver[Silver Clean Tables] gold[Gold Facts and Features] spark[Spark Jobs] kafka[Kafka] flink[Flink Jobs] airflow[Airflow] mlflow[MLflow] api[Fraud Scoring API] monitor[Monitoring] generator --> raw raw --> spark spark --> bronze bronze --> spark spark --> silver silver --> spark spark --> gold generator --> kafka kafka --> flink flink --> gold airflow --> generator airflow --> spark airflow --> flink gold --> mlflow mlflow --> api gold --> monitor api --> monitor ``` ## 数据湖设计 该项目将源数据与分析存储格式分离开来。 | 层级 | 存储目的 | 表示例表 | |---|---|---| | 原始源 | 源格式 CSV 提取，尽可能保留生产者格式 | `offline_transactions` 文件 | | Bronze | 带有源元数据且进行了最少转换的已摄取 Parquet | `raw_transactions` | | Silver | 经过清理、类型化、去重且 schema 对齐的 Parquet | `stg_transactions`、`stg_customers`、`stg_merchants` | | Gold | 面向业务的事实表、维度表和 ML 特征表 | `fact_transactions`、`dim_customer_scd2`、`feat_customer_transaction_rolling` | 在原始源边界处特意使用 CSV，因为许多生产系统在将其转换为分析格式之前，会先从外部系统摄取文件。Spark 作业应读取原始 CSV 分区，保留源元数据，并写入 Bronze Parquet 以便进行高效的下游处理。 ## 交易源设计 每一生成的行代表源系统粒度下的一笔交易事件。 核心列包括： ``` transaction_id account_id customer_id merchant_id merchant_category amount currency city channel transaction_status is_fraud event_timestamp created_ts ``` 较新的源分区添加了演进的列： ``` device_id ip_address authentication_method risk_signal_version ``` 生成器特意使用了两个时间戳： | 列 | 含义 | |---|---| | `event_timestamp` | 交易实际发生的时间。用于业务逻辑、窗口计算、特征和定点时间（point-in-time）正确性。 | | `created_ts` | 记录生成或到达的时间。用于摄取行为、延迟到达分析和 pipeline 新鲜度检测。 | ## 模拟的数据工程问题 离线生成器为 Spark 和 lakehouse 处理制造了逼真且可解决的数据问题。 | 问题 | FraudStream 如何模拟 | 下游处理方式 | |---|---|---| | 重复项 | 重复受控比例的源行 | 在 Silver 层通过 `transaction_id` 去重 | | 延迟到达 | 将部分 `created_ts` 值延迟超过 60 分钟 | 使用事件时间逻辑和到达时间监控 | | Schema 演进 | 在 `schema_change_date` 之后添加设备/IP/认证列 | 使用可为空的演进列读取新旧分区 | | 倾斜 | 将交易量集中在纽约和在线市场交易中 | 测试 Spark joins、分区和聚合行为 | | 高基数 | 生成数十万的客户/设备 ID | 练习 joins、特征分组和去重计数验证 | | 缺失值 | 留空选定的城市、商家、设备、IP 或认证字段 | 在 Silver 层进行验证和标准化 | | 格式不一致 | 注入带有填充字符的城市、大写状态和小写货币 | 在清理过程中标准化字符串 | | 突发流量 | 将更大比例的记录发送至选定的突发日期 | 测试分区不平衡和工作负载激增 | | 欺诈团伙行为 | 在高风险交易中重用可疑的设备/IP 对 | 围绕实体重用和速度构建欺诈特征 | ## 代码库结构 ``` financial-fraud-detection/ ├── AGENTS.md ├── README.md ├── configs/ │ └── generator/ ├── data/ │ └── raw_source/ ├── docs/ ├── src/ │ └── fraudstream/ │ └── generators/ ├── tests/ │ └── unit/ ├── main.py ├── pyproject.toml └── uv.lock ``` 重要路径： | 路径 | 用途 | |---|---| | `configs/generator/offline_transactions.json` | 离线数据生成的 runtime 配置 | | `src/fraudstream/generators/offline_transactions.py` | 离线交易生成器实现 | | `data/raw_source/offline_transactions/` | 生成的源格式交易文件 | | `docs/01_data_generator.md` | 详细的生成器设计和运行文档 | | `tests/unit/test_offline_transactions.py` | 针对生成器行为和证据工件的单元测试覆盖率 | ## 在本地运行 生成默认的离线交易数据集： ``` PYTHONPATH=src python -m fraudstream.generators.offline_transactions ``` 通过项目入口点运行： ``` PYTHONPATH=src python main.py ``` 将生成的数据写入临时输出目录： ``` PYTHONPATH=src python -m fraudstream.generators.offline_transactions \ --output-dir /tmp/fraudstream_offline_transactions ``` 运行单元测试： ``` PYTHONPATH=src python -m unittest tests.unit.test_offline_transactions ``` 运行语法/导入编译检查： ``` PYTHONPATH=src python -m compileall -q src tests main.py ``` ## 配置 默认的生成器配置位于： ``` configs/generator/offline_transactions.json ``` 关键设置： | 设置 | 用途 | |---|---| | `n_transactions` | 注入重复项之前的唯一基础交易计数 | | `n_customers`、`n_accounts`、`n_merchants` | 控制实体基数 | | `days_history` | 生成的交易日期数 | | `skew_city_ratio` | 控制地理倾斜 | | `skew_merchant_category_ratio` | 控制商家类别倾斜 | | `duplicate_rate` | 控制重复的原始行 | | `late_arrival_rate` | 控制延迟的源记录 | | `missing_value_rate` | 控制选定字段中的空白值 | | `inconsistent_format_rate` | 控制易于清理的字符串格式问题 | | `burst_day_count` | 控制大交易量日期的数量 | | `fraud_ring_count` | 控制可疑的重用设备/IP 对 | | `schema_change_date` | 分割旧的和演进的源 schema | ## 工程方向 下一个工程层是通过 Spark 将原始 CSV 摄取为 Bronze Parquet。该作业应保留源记录，添加摄取元数据，处理 schema 演进，并写入分区的 Parquet。Silver 处理随后应对记录去重、标准化格式、强制执行类型，并为 Gold 特征表准备干净的交易数据。 这使得该项目符合生产数据平台模式： ``` raw source files -> Bronze Parquet -> Silver clean tables -> Gold facts/features -> model training and monitoring ```

标签：MLOps, 数据工程, 数据湖仓, 数据生成, 特征工程, 逆向工具, 金融风控