🏦 ETL-пайплайн для обнаружения мошеннических операций в банке

Проект: Система выявления мошенничества на основе транзакций, паспортов и терминалов
Технологии: Python, PostgreSQL, DWH, ETL, SCD Type 2
Автор: [Ваше имя] — Data Engineer

---

📌 Описание

Этот проект реализует ETL-пайплайн, который:

• Загружает данные из внешних файлов (transactions.txt, terminals.xlsx, passport_blacklist.xlsx)
• Хранит данные в хранилище данных (DWH) с поддержкой историзации (SCD Type 2)
• Выявляет 5 типов мошеннических операций
• Генерирует отчёт в таблице rep_fraud
• Обеспечивает идемпотентность, контроль дублирования и логирование

---

🧩 Архитектура

flowchart TD
    A["data.zip"] --> B{Распаковка}
    B --> C["transactions_*.txt"]
    B --> D["terminals_*.xlsx"]
    B --> E["passport_blacklist_*.xlsx"]

    C --> F["stg_transactions"]
    D --> G["stg_terminals"]
    E --> H["stg_passport_blacklist"]

    F --> I["dwh_fact_transactions"]
    G --> J["dwh_dim_terminals_hist (SCD2)"]
    H --> K["dwh_fact_passport_blacklist"]

    I --> L["rep_fraud"]
    J --> L
    K --> L

    M["generate_fraud_report.sql"] --> L
    L --> N["Отчёт о мошенничестве"]

Loading

---

📂 Структура проекта

bank-fraud-detection/
├── py_scripts/                   # Основные Python-скрипты
│   ├── file_load.py              # Основной ETL-процесс
│   ├── load_ddl_dml.py           # Создание DWH + загрузка справочников
│   ├── db_connection.py          # Подключение к БД
│   ├── logger.py                 # Логирование с ротацией
│   └── ...
│
├── sql_scripts/                  # SQL-скрипты
│   ├── create_bank_base.sql      # DDL: создание схемы, таблиц, функций
│   ├── ddl_dml.sql               # DML: INSERT-ы справочников (клиенты, карты)
│   ├── merge_*.sql               # MERGE: STG → DWH (SCD2)
│   ├── truncate_stg_*.sql        # Очистка STG
│   ├── generate_fraud_report.sql # Функция детектирования мошенничества
│   └── log_meta_load.sql         # Логирование факта загрузки
│
├── data/                         # Входящие файлы (data.zip)
│                      
│── archive/                      # Архив обработанных файлов (.backup)
│
├── json/
│   └── cred.json                 # Учётные данные БД (не в Git!)
│
├── debug.log                     # Лог-файл (ротируется)
├── requirements.txt
└── README.md

---

🔐 Учётные данные БД

Файл: json/cred.json

{
  "host": "localhost",
  "port": 5432,
  "database": "bank_dw",
  "user": "dwh_user",
  "password": "secret"
}

⚠️ Никогда не коммитьте cred.json! Добавьте файл в .gitignore.

---

📊 Типы выявляемого мошенничества

№	Правило	Описание
1	Карта выдана до рождения клиента	card_create_dt < client.date_of_birth
2	Паспорт просрочен на момент операции	passport_valid_to < transaction_date
3	Паспорт в чёрном списке	passport_num IN blacklist
4	Два разных города за 1 час	t1.city ≠ t2.city AND time_diff ≤ 1h
5	Подбор суммы (кард-скрэппинг)	3+ отказов с уменьшением суммы → успешная транзакция

---

🔁 ETL-процесс

1. Инициализация DWH

python py_scripts/load_ddl_dml.py

• Выполняет create_bank_base.sql → создаёт схему bank, таблицы, индексы
• Загружает справочники из ddl_dml.sql (клиенты, карты, счета)
• Использует TABLE_MAPPING для маппинга: cards → dwh_dim_cards
• Идемпотентен: игнорирует ошибки already exists (проверка через IF NOT EXISTS)

2. Обработка данных

python py_scripts/file_load.py

1. Ищет data.zip в data/
2. Распаковывает и обрабатывает:
   • transactions_*.txt → CSV → STG → DWH
   • terminals_*.xlsx → Excel → STG → DWH (SCD2)
   • passport_blacklist_*.xlsx → STG → DWH
3. Проверяет дубли через таблицу META_DWH_LOAD_DATES
4. Генерирует отчёт о мошенничестве вызовом generate_fraud_report()
5. Архивирует исходные файлы в archive/ с расширением .backup

---

🛠️ Ключевые функции

Функция	Назначение
file_already_processed()	Проверяет, был ли файл уже обработан (по имени и дате)
report_already_generated()	Проверяет, был ли отчёт уже сгенерирован для даты
load_stg_*()	Загружает данные в STG-таблицы
merge_*()	Применяет изменения к DWH с использованием SCD Type 2
generate_fraud_report()	Выполняет SQL-функцию детектирования мошенничества

---

🚀 Запуск

1. Подготовка

pip install -r requirements.txt

# Создайте cred.json
mkdir -p json
cat > json/cred.json << EOF
{
  "host": "localhost",
  "port": 5432,
  "database": "bank_dw",
  "user": "dwh_user",
  "password": "secret"
}
EOF

2. Вручную

python main.py                       # Основной запуск. Автоматически выполняет ниже перечисленные действия
python py_scripts/load_ddl_dml.py    # Создание DWH и загрузка справочников
python py_scripts/file_load.py       # Загрузка транзакций, терминалов, чёрного списка и генерация отчёта

---

🧪 Тестовые данные

Поместите в data/:

• data.zip, содержащий:
  • transactions_20240301.txt
  • terminals_20240301.xlsx
  • passport_blacklist_20240301.xlsx

✅ Рекомендация: имена файлов должны содержать дату в формате YYYYMMDD.

📁 Форматы входных файлов

Транзакции (transactions_DDMMYYYY.txt)
transaction_id;transaction_date;card_num;oper_type;amount;oper_result;terminal
12345;2024-03-01 10:30:00;1234567812345678;payment;1500.50;SUCCESS;TERM001

---

Черный список паспортов (passport_blacklist_DDMMYYYY.xlsx)

passport_num	entry_dt
1234567890	2024-01-15
0987654321	2024-02-20

Терминалы (terminals_DDMMYYYY.xlsx)

terminal_id	terminal_type	city	address
TERM001	ATM	Moscow	Lenina st. 1
TERM002	POS	SPb	Nevsky st. 25

⚙️ Конфигурация системы

Настройка путей (main.py)

#### Базовые директории
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = os.path.join(BASE_DIR, 'data')          # Входные данные
ARCHIVE_DIR = os.path.join(BASE_DIR, 'archive')    # Архив обработки
JSON_DIR = os.path.join(BASE_DIR, 'json')          # Конфигурации