Python сканер подозрительных транзакций: от CSV-выгрузки до velocity-алертов

По данным Association of Certified Fraud Examiners (ACFE — международная ассоциация специалистов по противодействию мошенничеству), бизнес теряет в среднем 5% годовой выручки на фроде. Большая часть потерь — транзакции, которые прошли через платёжную систему без единого алерта. При этом первой линией обороны в реальных платёжных сервисах почти всегда становится не ML-модель, а набор velocity-правил на Python, написанный за пару дней и запущенный через cron прямо из Jupyter Notebook. За три года работы с такими системами я убедился: пять точных правил с порогами ловят больше фрода в первый месяц, чем Random Forest, обученный на синтетике. Здесь построим python сканер подозрительных транзакций с нуля — от парсинга CSV до генерации алертов, где каждое срабатывание сопровождается человекочитаемым объяснением, а не сухим флагом is_fraud=True.

Что такое velocity-фрод и зачем rule-based детект

Velocity-фрод (velocity fraud) — тип мошенничества, при котором злоумышленник совершает аномально много операций за короткий промежуток времени. Типичные сценарии: перебор украденных карт через автоматизированные скрипты (одна мелкая транзакция на каждую карту, чтобы проверить — жива ли), или серия крупных переводов с одного счёта за несколько минут, пока банк не заблокировал. В терминологии MITRE ATT&CK (открытая база тактик и техник атак, где каждая техника имеет T-код — уникальный идентификатор) конечная цель таких действий описывается техникой T1657 Financial Theft (тактика Impact) — вывод денег.

Зачем начинать с правил, а не с машинного обучения? Три причины.

Первая — для ML нужны размеченные данные. Тысячи транзакций с ярлыками «фрод» и «не фрод». У новых сервисов их просто нет. Velocity-правила работают без обучающей выборки — достаточно экспертного порога: «больше 5 операций в час с одной карты — подозрительно».

Вторая — объяснимость. Аналитик открывает алерт и видит: «6 операций за 47 минут, порог — 5». Модель Random Forest выдаёт probability 0.87, и аналитик гадает, что именно сработало. В задачах финансового мониторинга (AML — Anti-Money Laundering, противодействие отмыванию денег) объяснимость алерта — не бонус, а требование регулятора.

Третья — скорость деплоя. Python-скрипт для анализа транзакций можно написать за день и запускать каждые 15 минут. ML-пайплайн с SMOTE (метод синтетической генерации миноритарных примеров для борьбы с дисбалансом классов), валидацией и мониторингом дрифта — это недели работы.

По данным IBM X-Force Threat Intelligence Index 2025, infostealers (32%) стали самым распространённым типом вредоносного ПО, обогнав ransomware. Украденные карточные данные часто монетизируются через velocity-атаки — массовую проверку и обналичивание через подставные мерчанты. Автоматизация анализа финансовых данных на Python нужна даже небольшим платёжным сервисам — вопрос не «зачем», а «почему ещё не сделали».

Требования к окружению и структура CSV-данных

Перед практической частью — фиксируем минимальный стек и формат входных данных.

Требования к окружению:

• Python 3.9 или выше
• Библиотека pandas 2.0+ (установка: pip install pandas)
• ОС: любая — Windows, macOS, Linux
• RAM: от 512 МБ для файлов до 1 млн строк. Для крупных выгрузок (10 млн+ строк) потребуется от 4 ГБ, либо переход на polars
• Сеть: не требуется — обработка полностью локальная
• GPU: не требуется

Транзакционный CSV из процессинга обычно содержит как минимум эти колонки:

Если процессинг выгружает данные в другом формате (JSON-lines, Parquet), конвертация в CSV занимает одну строку: pd.read_json("file.jsonl", lines=True).to_csv("out.csv"). Дальше работаем с CSV — самый распространённый формат выгрузки из банковских систем и платёжных шлюзов.

Парсим CSV-выгрузку и строим признаки для анализа транзакций

Задача: загрузить CSV, обработать временные метки и вычислить два ключевых признака — количество транзакций за временное окно (velocity count) и z-score суммы. Z-score — число стандартных отклонений от средней суммы по карте. Чем выше значение, тем аномальнее операция: z-score 0 — ровно среднее, z-score 3 — сумма больше средней на три стандартных отклонения, что случается менее чем в 1% случаев.

import pandas as pd

df = pd.read_csv("transactions.csv", parse_dates=["timestamp"])
df.sort_values(["card_id", "timestamp"], inplace=True)
# Velocity: количество транзакций за скользящее окно 1 час
grouped = df.set_index("timestamp").groupby("card_id")["amount"]
df["tx_count_1h"] = grouped.rolling("1h").count().droplevel(0).values
# Z-score суммы: отклонение от среднего по карте
df["amount_zscore"] = df.groupby("card_id")["amount"].transform(
    lambda x: (x - x.mean()) / x.std().clip(lower=0.01)
)

Разберём построчно — чтобы было понятно, что происходит на каждом шаге и как убедиться, что всё работает.

pd.read_csv с параметром parse_dates=["timestamp"] загружает CSV и сразу превращает колонку timestamp из строки в объект datetime. Без этого параметра pandas хранит дату как текст, и скользящее окно не заработает. После загрузки проверьте: df["timestamp"].dtype должен вернуть datetime64[ns].

sort_values(["card_id", "timestamp"]) — сортировка по карте, затем по времени. Это обязательно: метод rolling считает окно по последовательным строкам, и без сортировки результат будет мусором.

Конструкция set_index("timestamp").groupby("card_id")["amount"].rolling("1h").count() — для каждой карты подсчитывает, сколько транзакций попало в скользящее окно длиной один час. Результат включает текущую транзакцию. Методы droplevel(0) и .values нужны для технической совместимости — они убирают лишний уровень индекса и позволяют присвоить результат обратно в DataFrame.

Если после запуска в колонке tx_count_1h появились NaN, скорее всего timestamp не распарсился — проверьте формат дат в исходном CSV. Ожидаемые значения — числа от 1 (единственная транзакция в окне) и выше.

Вызов .clip(lower=0.01) в z-score защищает от деления на ноль, когда все транзакции карты одинаковые — без него pandas выдаст inf, и дальнейшая фильтрация сломается.

Типичные ошибки при парсинге транзакционных CSV

Две проблемы, которые встречаются регулярно при работе с транзакционными данными.

Временные зоны. CSV из процессинга может содержать UTC-время, а выгрузка из клиентского интерфейса — локальное. Смешение файлов без нормализации сдвинет velocity-окно на несколько часов. После загрузки приведите к единой зоне: df["timestamp"] = df["timestamp"].dt.tz_localize("UTC"). Если данные уже содержат зону (формат 2025-03-15T14:23:07+03:00), pandas обработает это автоматически.

Дубли строк. Retry-механизмы и reconciliation иногда записывают транзакцию дважды. Дубли завышают velocity-счётчик — и вы получите ложные алерты на ровном месте. Проверка: df.duplicated(subset=["card_id", "timestamp", "amount"]).sum(). Если результат больше нуля — удалите через df.drop_duplicates(subset=["card_id", "timestamp", "amount"], inplace=True) до вычисления признаков.

Velocity-правила детекта мошеннических транзакций

Теперь у нас два признака: tx_count_1h и amount_zscore. Из них строим правила. Каждое правило — проверка порога с генерацией текстового объяснения.

Правила частоты и аномальной суммы

Функция check_velocity принимает одну строку DataFrame и возвращает список причин подозрительности. Пустой список — транзакция чистая.

def check_velocity(row, max_tx=5, z_thresh=3.0):
    reasons = []
    if row["tx_count_1h"] > max_tx:
        reasons.append(
            f"Частота: {int(row['tx_count_1h'])} tx/час (порог: {max_tx})")
    if abs(row["amount_zscore"]) > z_thresh:
        reasons.append(
            f"Сумма {row['amount']:.0f} — отклонение "
            f"{row['amount_zscore']:.1f}σ от среднего")
    return reasons

Параметр max_tx=5: больше пяти транзакций с одной карты за час — подозрительно. z_thresh=3.0: если сумма отклоняется от средней больше чем на 3σ, срабатывает алерт.

Откуда эти числа? Из практики. Порог 5 подходит для розничных карт: обычный покупатель делает 1-3 покупки в день. Для корпоративных карт или B2B-платежей порог поднимают до 15-20. Z-score 3.0 — классический статистический порог: менее 0.3% наблюдений выходят за ±3σ при нормальном распределении. Для агрессивного детекта снижают до 2.5, для уменьшения ложных срабатываний — поднимают до 3.5.

В реальном velocity fraud detection скрипте правил больше. Добавить новое — ещё один if в функции и ещё одна строка в reasons. Примеры:

• Смена IP. Если ip_address текущей транзакции отличается от предыдущей у той же карты — добавляем причину. Для вычисления: df.groupby("card_id")["ip_address"].shift(1) и сравнение с текущей строкой.
• Разнообразие мерчантов. Больше 3 уникальных магазинов за час — нетипично для обычного покупателя. Считается аналогично velocity count, но через nunique вместо count.
• Ночная активность. Транзакция между 01:00 и 05:00 при отсутствии ночных покупок в истории карты.

Архитектура остаётся плоской — никаких вложенных классов и цепочек ответственности. На практике 5-10 правил покрывают 80% детектов в первые месяцы работы антифрод-системы.

Генерация алертов с объяснением срабатывания

Последний шаг — применить правила ко всему DataFrame и вывести результат в формате, который аналитик прочитает без заглядывания в код.

df["reasons"] = df.apply(check_velocity, axis=1)
alerts = df[df["reasons"].apply(len) > 0]
for _, tx in alerts.iterrows():
    print(f"[ALERT] {tx['card_id']} | {tx['timestamp']} | {tx['amount']:.2f}")
    for r in tx["reasons"]:
        print(f"  -> {r}")

Вызов df.apply(check_velocity, axis=1) применяет функцию к каждой строке (параметр axis=1 означает «по строкам»; по умолчанию pandas работает по столбцам). Результат — колонка reasons, где каждая ячейка содержит список строк-объяснений.

Ожидаемый вывод для подозрительной транзакции: [ALERT] card_8827 | 2025-03-15 14:23:07 | 45000.00, под ним строки причин: -> Частота: 8 tx/час (порог: 5) и -> Сумма 45000 — отклонение 4.2σ от среднего. Аналитик сразу видит: карта подозрительна по двум правилам — и по частоте, и по сумме. Одно правило — сигнал слабый. Два и более — приоритет на ручную проверку.

Для отправки алертов в Telegram или Slack замените print на вызов webhook — формат сообщения тот же, один HTTP POST с текстом. Для интеграции с SIEM (Security Information and Event Management — система, которая собирает логи со всей инфраструктуры и ищет в них подозрительные паттерны) сохраните результат в JSON: alerts.to_json("alerts.json", orient="records", date_format="iso") и настройте агент (filebeat, Elastic Agent) на парсинг файла. Подход соответствует рекомендации NIST CSF 2.0 по категории DE.AE-01 — поддержание baseline и обнаружение аномалий в потоках данных.

Ограничения rule-based подхода и когда переходить к ML

Velocity-правила — рабочий инструмент, но с чёткими границами. Знать их нужно до того, как вы запустите скрипт в продакшн.

Ложные срабатывания. Покупатель зашёл в ТЦ и за час оплатил кофе, обед, пару обуви, парковку и бензин — 5 транзакций, порог на грани. Жёсткие пороги не учитывают контекст: день зарплаты, распродажу, отпуск с мелкими покупками. Снижение порогов увеличивает detection rate, но заваливает аналитика ложными алертами.

Правила не учатся. Если мошенники адаптировались и стали делать не 10 транзакций за час, а 4 за 4 часа (каждая на верхней границе «нормальной» суммы), фиксированные пороги их пропустят. ML-модели (Random Forest, XGBoost) ловят комбинации признаков, которые по отдельности не выглядят аномально. Но для этого нужен размеченный датасет.

Когда добавлять ML. Практическое правило: если накопилось больше 500 подтверждённых кейсов фрода и 50 000+ чистых транзакций — время строить ML-модель поверх rule-based системы. Не вместо, а поверх: правила остаются быстрым первым фильтром, модель добирает сложные паттерны. По данным курса DataCamp Fraud Detection in Python, на датасете кредитных карт (284 807 транзакций, из них 492 фродовых — 0.17%) Random Forest с SMOTE показывает recall выше 90%. Но это на размеченных данных, которых на старте нет.

Промежуточный вариант — Isolation Forest (алгоритм обнаружения аномалий из scikit-learn, не требующий меток). Он изолирует «странные» точки через случайные деревья решений. Вызов в коде: [IsolationForest](https://scikit-learn.org/stable/modules/generated/sklearn.ensemble.IsolationForest.html)(contamination=0.01).fit_predict(X), где contamination — ожидаемая доля фрода. Но даже Isolation Forest не даёт explain-вывода — только бинарный флаг «аномалия / норма». В задачах финансового мониторинга, где регулятор требует обоснование каждого алерта, velocity-правила с текстовыми причинами честнее.

Практический совет: экспортируйте результаты velocity-сканера в CSV с колонками is_suspicious и reasons. Используйте этот файл как первичную разметку для будущей ML-модели. Транзакции с 3+ сработавшими правилами — кандидаты на is_fraud=1 после ручной проверки. Так вы постепенно строите обучающую выборку, не останавливая детект. Техника Automated Collection (T1119 в MITRE ATT&CK, тактика Collection) описывает схожий принцип — систематический сбор данных для дальнейшей обработки, только здесь он работает на стороне защиты.

Velocity-детект вызывает у многих начинающих инженеров скепсис: «зачем писать if-ы, когда есть XGBoost». На практике картина повторяется из проекта в проект. Команда тратит два месяца на ML-пайплайн, обучает модель на синтетических данных (потому что реальных фродов ещё нет), получает F1=0.95 на тесте — и через неделю после деплоя выясняется, что модель ловит паттерны генератора Faker, а не реальных мошенников. Параллельно пять velocity-правил, написанных стажёром за день, уже дважды заблокировали реальный carding.

ML не бесполезен. Но порядок имеет значение: сначала правила, потом ML поверх правил, никогда наоборот. Правила — baseline, который работает с первого дня и даёт объяснимые алерты. ML — усиление, когда baseline накопил данные для обучения.

Ещё один момент, который часто упускают: velocity-сканер — это не только про фрод. Те же приёмы (скользящее окно, z-score, explain-вывод) работают для обнаружения аномалий в транзакциях и за пределами платёжных систем: credential stuffing в логах веб-сервера, DDoS-паттерны, аномалии в DNS-запросах. Освоив этот подход, применяйте его к любым временным рядам с подозрительной активностью — архитектура одинаковая. Если хочешь разобраться, как встроить такую аналитику в контекст реальной работы SOC и blue team — на IB Basics показываем что делать в первый месяц после «хочу в ИБ».