Дроп фрод основные инструменты¶

  • ноутбуки лучше просматривать на Github pages, т.к. при просмотре прямо в репозитории могут быть проблемы с отображением, например, обрезка вывода с широкими датафреймами. Если в адресной строке браузера есть iaroslav-dzh.github.io, то вы уже на Github pages.
    Ссылки:
    • Ссылка на этот ноутбук
    • Ссылка на страницу генератора где есть все ноутбуки

Вступление к генерации дроп фрода
Это начало демонстрации генерации дроп фрода - далее будет еще 2 ноутбука о генерации дроп фрода, т.к. этот раздел получился более обширным.
При написании генерации дроп фрода я полностью решил перейти на классы как основу кода, вместо функций.

Дропов будет два типа:

  • распределители (distributors): поведение основано на регулярном получении денег и их распределении - перевод, снятие, покупка криптовалюты.
  • покупатели (purchasers): поведение основано на регулярном получении денег и покупке на них товаров - отмыв денег.

Информация о ноутбуке

  • В этом ноутбуке базовые части генерации:
    • создание конфиг классов
    • управление счетами
    • генерация сумм транзакций
    • генерация времени транзакций
    • управление поведением дропа
In [4]:
import pandas as pd
import numpy as np
import os
import pyarrow
import yaml
In [7]:
from data_generator.general_time import *
from data_generator.utils import load_configs
In [5]:
np.set_printoptions(suppress=True)
pd.set_option('display.max_columns', None)
In [6]:
os.chdir("..")
os.getcwd()
Out[6]:
'C:\\Users\\iaros\\My_documents\\Education\\projects\\fraud_detection_01'
In [8]:
# Базовые конфиги
base_cfg = load_configs("./config/base.yaml")
# Настройки легальных транзакций
legit_cfg = load_configs("./config/legit.yaml")
# Общие настройки фрода
fraud_cfg = load_configs("./config/fraud.yaml")
# Настройки для дроп фрода
drops_cfg = load_configs("./config/drops.yaml")
# Настройки времени
time_cfg = load_configs("./config/time.yaml") 

# Пути к файлам
data_paths = base_cfg["data_paths"]



Создание конфиг классов с конфигами и данными для генерации¶


1. Класс DropConfigBuilder¶

  • модуль data_generator.fraud.drops.build.config. Ссылка на файл в Github
  • Условия
  • до этого были сгенерированы легальные и compromised client fraud транзакции

Основная логика

  • создает объект DropDistributorCfg или DropPurchaserCfg для дропов распределителей или покупателей соотвественно
  • суть такая же как у конфиг билдеров для легальных и compromised client fraud транзакций
In [9]:
# директорию текущего запуска генератора возьмем из предыдущего ноутбука. Т.к. нужны данные того, что сгенерировано до дроп фрода
run_dir = './data/generated/history/generation_run_2025-07-25_121029'
In [11]:
from data_generator.fraud.drops.build.config import DropConfigBuilder

# объект самого конфиг билдера
cfg_build = DropConfigBuilder(base_cfg=base_cfg, legit_cfg=legit_cfg, time_cfg=time_cfg, fraud_cfg=fraud_cfg, \
                              drop_cfg=drops_cfg, run_dir=run_dir)

# Конфиг класс для дропов распределителей
dist_configs = cfg_build.build_dist_cfg()
# Конфиг класс для дропов покупателей
purch_configs = cfg_build.build_purch_cfg()

Демонстрация¶

In [12]:
# Клиенты семплированные под дропов распределителей т.е. они будут дропами
dist_configs.clients.head(2)
Out[12]:
client_id birth_date sex region city lat lon city_id home_ip
0 4344 1950-06-09 male Свердловская Екатеринбург 56.838633 60.605489 54 2.60.16.18
1 2692 1950-05-24 male Ставропольский Ставрополь 45.044544 41.969017 44 2.60.9.247
In [13]:
# Клиенты семплированные под дропов покупателей т.е. они будут дропами
purch_configs.clients.head(2)
Out[13]:
client_id birth_date sex region city lat lon city_id home_ip
0 901 1944-12-14 male Смоленская Смоленск 54.786717 31.815337 39 2.60.3.87
1 3839 1962-06-29 female Ростовская Таганрог 47.209491 38.935154 65 2.60.14.50
In [14]:
# таблица со счетами клиентов и информацией о том дропы ли они.
# клиент помечается как дроп только когда непосредственно идет генерация его активности
dist_configs.accounts.head(2)
Out[14]:
client_id account_id is_drop
0 1 10000 False
1 2 10001 False



Управление счетами транзакций. Генерация сумм транзакций и времени¶


1. Класс DropAccountHandler¶

  • модуль data_generator.fraud.drops.base. Ссылка на файл в Github
  • Управление счетами переводов, статусами дропов в таблице accounts

Основной функционал

  • запись номера счета текущего дропа в атрибут account
  • случайный выбор счета для исходящего перевода если это дроп распределитель
  • пометка текущего клиента, как дропа в таблице accounts

Демонстрация¶

In [16]:
from data_generator.fraud.drops.base import DropAccountHandler

acc_hand = DropAccountHandler(configs=dist_configs)
# DropAccountHandler нуждается в передаче ему client_id, чтобы знать чей счет искать если нужен счет самого дропа 
# и чей счет исключить если это счет под исходящюю транзакцию
acc_hand.client_id = 1

Найти и записать счет текущего клиента в атрибут account

In [19]:
acc_hand.get_account(own=True)
acc_hand.account
Out[19]:
np.int64(10000)

Выбрать случайный счет под исходящий перевод

In [18]:
acc_hand.get_account()
Out[18]:
np.int64(23260)

Пометить текущего клиента, как дропа

In [20]:
# смотрим до вызова метода. Клиент не помечен как дроп: is_drop = False
accounts = acc_hand.accounts
own_id = acc_hand.client_id
accounts.query("client_id == @own_id")
Out[20]:
client_id account_id is_drop
0 1 10000 False
In [21]:
# Отмечаем клиента как дропа
acc_hand.label_drop()

# Снова смотрим на него в accounts
accounts.query("client_id == @own_id")
Out[21]:
client_id account_id is_drop
0 1 10000 True

2. Класс DropAmountHandler¶

  • модуль data_generator.fraud.drops.base. Ссылка на файл в Github

Основной функционал

  • генерация сумм транзакций: вх./исх. переводов, снятий, покупок:
    • генерирует сумму в соответствии с конфигами в drops.yaml и в соответствии с переданными аргументами, например:
      • генерация сумм для переводов, снятий или покупок частями - в соответствии с лимитами в конфиге drops.yaml
      • генерация сумм для перевода всего баланса
      • уменьшение суммы следующего перевода если текущий перевод отклонен
  • управление балансом текущего дропа:
    • начисление при успешной входящей транзакции
    • списание при успешной исх. транз-ции - перевод, снятие, покупка.
    • неизменение баланса если вх./исх. транз-ция отклонена

Демонстрация¶

In [22]:
from data_generator.fraud.drops.base import DropAmountHandler

amt_hand = DropAmountHandler(configs=dist_configs)

Метод receive

  • генерация суммы входящего перевода
  • изменение баланса на сумму перевода если транз. не отклонена
In [25]:
# текущий баланс
amt_hand.balance
Out[25]:
0
In [27]:
# сумма вх. перевода. Успешная транз
amt_hand.receive(declined=False)
Out[27]:
np.float64(30700.0)
In [28]:
# Баланс
amt_hand.balance
Out[28]:
np.float64(30700.0)
In [29]:
# сумма вх. перевода. Отклоненная транз
amt_hand.receive(declined=True)
Out[29]:
np.float64(34300.0)
In [30]:
# Баланс
amt_hand.balance
Out[30]:
np.float64(30700.0)

Метод one_operation

  • объединяет в себе несколько методов. Является конечным методом для генерации суммы транзакции и обновления баланса
  • возвращает сумму транзакции
In [31]:
# Баланс
amt_hand.balance
Out[31]:
np.float64(30700.0)
In [32]:
# Например текущая транзакция онлайн, она не отклонена и нужно перевести только часть баланса
amt_hand.one_operation(online=True, declined=False, in_chunks=True)
Out[32]:
np.float64(20000.0)
In [33]:
amt_hand.balance
Out[33]:
np.float64(10700.0)

Пример с уменьшением суммы если транзакция отклонена

  • каждая сумма будет меньше предыдущей, но не меньше минимального значения, которые выставлено в конфигах drops.yaml
In [42]:
amt_hand.reset_cache(all=True)
amt_hand.balance = 47000
print(f"Баланс перед запуском цикла: {amt_hand.balance}")

all_ops = []
while True:
    one_op = int(amt_hand.one_operation(online=True, declined=True, in_chunks=True))
    
    all_ops.append(one_op)
    if len(all_ops) == 5:
        break
print(f"""Итоговый баланс: {amt_hand.balance}
Все созданные суммы: {all_ops}""")

Баланс перед запуском цикла: 47000
Итоговый баланс: 47000
Все созданные суммы: [24000, 18000, 12000, 6000, 3000]

3. Класс DropTimeHandler¶

  • модуль data_generator.fraud.drops.time. Ссылка на файл в Github

Основной функционал

  • управление временем транзакций дропа. Генерация времени

Основная логика

  1. время первой транзакции берется случайно из диапазона timestamp-ов
  2. создает время транзакций в соответсвии с лимитами активности для дропов выставленных в drops.yaml
  3. есть лимит на входящие транзакции и исходящие в периоде активности
  4. период активности это когда дроп "работает" с деньгами - переводит, снимает, покупает.
  5. период измеряется по кол-ву сделанных транзакций. Если достигнут какой-то из лимитов - входящие/исходящие - то дроп берет паузу в активности на период установленный в конфигах и +/- случайную дельту времени установленную также в конфигах. Пауза считается от первой транзакции в периоде, например новый период дроп перевел деньги в 12:00, и достиг лимита на исх. операции в текущем периоде. Берется время 12:00 и к нему прибавляется указанный лаг по времени, например 24 часа, и случайная положительная или отрицательная дельта, например -1 час. То есть дроп возобновляет активность в 12:00 + 24 часа + (-1) час = 11:00 следующего дня. Это будет транзакция уже нового периода активности.
  6. если лимит не достигнут то время текущей транзакции также основывается на времени предыдущей: прибавляется случайная положительная дельта времени.

Демонстрация¶

In [49]:
from data_generator.fraud.drops.time import DropTimeHandler

time_hand = DropTimeHandler(configs=dist_configs)

Метод get_txn_time

  • основной метод класса. Генерирует время для вх. и исх. транз-ций
  • внизу будет симулирована генерация под активность дропа. Так время ведет себя и при полноценной генерации активности дропа
  • можно заметить что время между транзакциями time_diff периодами составляет не более 3-х часов и потом следует продолжительный перерыв в 15+ часов. Это и есть перерыв между периодами активности дропа. Условно, дроп "работает" какое-то время и уходит на долгий перерыв.
In [50]:
all_times = []

in_txns = 0
txn_time, _ = time_hand.get_txn_time(receive=True, in_txns=in_txns) # первая транзакция. Входящая
in_txns += 1
all_times.append(txn_time)

for _ in range(12): # цикл исходящих транзакций.
    txn_time, _ = time_hand.get_txn_time(receive=False, in_txns=in_txns)
    all_times.append(txn_time)
    
all_times = pd.DataFrame({"txn_time":all_times})
all_times["time_diff"] = all_times.txn_time - all_times.txn_time.shift(1)
all_times
Out[50]:
txn_time time_diff
0 2025-01-14 09:27:00 NaT
1 2025-01-14 10:35:00 0 days 01:08:00
2 2025-01-14 12:17:00 0 days 01:42:00
3 2025-01-14 13:11:00 0 days 00:54:00
4 2025-01-14 16:10:00 0 days 02:59:00
5 2025-01-14 18:55:00 0 days 02:45:00
6 2025-01-15 10:22:00 0 days 15:27:00
7 2025-01-15 11:52:00 0 days 01:30:00
8 2025-01-15 13:15:00 0 days 01:23:00
9 2025-01-15 15:02:00 0 days 01:47:00
10 2025-01-15 16:58:00 0 days 01:56:00
11 2025-01-16 09:02:00 0 days 16:04:00
12 2025-01-16 10:37:00 0 days 01:35:00



Управление поведением дропов¶


1. Класс DistBehaviorHandler¶

  • модуль data_generator.fraud.drops.behavior. Ссылка на файл в Github

Условия

  • подразумевается, что дроп получает сумму и распределяет её до получения новой

Основной функционал

  • управление поведением дропов распределителей
  • случайный выбор сценария распределения полученных денег - перевод всего, снятие всего, перевод частями, снятие+перевод частями - выбор сценария имеет зависимость от баланса. Например если баланс больше установленной суммы, то не может быть выбран сценарий с одним переводом всей суммы на балансе.
  • случайно определять будет ли транзакция покупкой криптовалюты
  • случайно определять будет ли перевод другому дропу внутри банка - итоговый перевод зависит от кол-ва существующих дропов на момент создания транзакции
  • определение кол-ва попыток совершить операцию после первой отклоненной транзакции
  • вычитание попыток когда дроп пытается совершать операции после первой отклоненной транз.
  • идентификация того нужно ли пытаться еще после последней отклоненной транзакции (это может быть первая отклоненная транз. если выпало 0 попыток)

Демонстрация¶

In [51]:
from data_generator.fraud.drops.behavior import DistBehaviorHandler
from data_generator.fraud.drops.base import DropAmountHandler

amt_hand = DropAmountHandler(configs=dist_configs)
dist_behav = DistBehaviorHandler(configs=dist_configs, amt_hand=amt_hand)

Метод sample_scenario

  • случайный выбор сценария
  • всего есть 4 сценария:
    • split_transfer - перевод полученной суммы по частям - возможны случаи покупки криптовалюты вместо переводов (это перевод на криптобиржу)
    • atm+transfer - снятие случайной доли суммы и перевод остального по частям - тоже возможна криптовалюта
    • transfer - перевод всей суммы баланса либо покупка криптовалюты на всю сумму
    • atm- снятие все суммы баланса

Доступные варианты сценариев зависят от величины баланса и лимитов по суммам операций выставленных в конфиг файле drops.yaml

  • зависимость от лимитов:
    • trf_max - максимальная сумма одного перевода. Если на балансе больше этой суммы, то доступны только варианты: split_transfer, atm+transfer,atm
    • atm_min - минимальная сумма для снятия. Если баланс больше и равен этому лимиту, но меньше trf_max, то доступны варианты: split_transfer, atm+transfer,atm,transfer
    • trf_min - минимальная сумма одного перевода. Если баланс равен или больше trf_min * 2, но меньше atm_min то доступны сценарии: split_transfer, transfer
    • если баланс меньше trf_min * 2, то доступен только transfer

Пример №1

  • баланс больше trf_max
  • демонстрация долей выпадающих сценариев
In [53]:
all_scens = [] # Все случаи выбора сценария
i = 0
while i < 2000:
    amt_hand.reset_cache()
    amt_hand.balance = dist_behav.trf_max + 1000
    assert amt_hand.balance > dist_behav.trf_max, "Balance is below trf_max"
    dist_behav.sample_scenario()
    all_scens.append(dist_behav.scen)
    i += 1
all_scens_ser = pd.Series(all_scens)
all_scens_ser.value_counts(normalize=True)
Out[53]:
split_transfer    0.3475
atm+transfer      0.3430
atm               0.3095
Name: proportion, dtype: float64

Пример №2

  • баланс меньше trf_max, но больше atm_min
  • демонстрация долей выпадающих сценариев
In [54]:
all_scens = []
i = 0
while i < 2000:
    amt_hand.reset_cache()
    amt_hand.balance = dist_behav.atm_min + 1000
    assert amt_hand.balance < dist_behav.trf_max, "Balance exceeds trf_max"
    dist_behav.sample_scenario()
    all_scens.append(dist_behav.scen)
    i += 1
all_scens_ser = pd.Series(all_scens)
all_scens_ser.value_counts(normalize=True)
Out[54]:
atm+transfer      0.3465
split_transfer    0.3430
transfer          0.1555
atm               0.1550
Name: proportion, dtype: float64

Пример №3

  • баланс меньше atm_min, но больше trf_min * 2
  • демонстрация долей выпадающих сценариев
In [55]:
all_scens = []
i = 0
while i < 3000:
    amt_hand.reset_cache()
    amt_hand.balance = dist_behav.trf_min * 2
    assert amt_hand.balance < dist_behav.atm_min, "Balance exceeds atm_min"
    assert amt_hand.balance < dist_behav.trf_max, "Balance exceeds trf_max"
    dist_behav.sample_scenario()
    all_scens.append(dist_behav.scen)
    i += 1
all_scens_ser = pd.Series(all_scens)
all_scens_ser.value_counts(normalize=True)
Out[55]:
split_transfer    0.704
transfer          0.296
Name: proportion, dtype: float64

Пример №4

  • баланс меньше trf_min * 2
  • выпадает только один сценарий
In [60]:
all_scens = []
i = 0
while i < 3000:
    amt_hand.reset_cache()
    amt_hand.balance = dist_behav.trf_min + 1000
    assert amt_hand.balance < dist_behav.trf_min * 2, "Balance exceeds the limit"
    dist_behav.sample_scenario()
    all_scens.append(dist_behav.scen)
    i += 1
all_scens_ser = pd.Series(all_scens)
all_scens_ser.value_counts(normalize=True)
Out[60]:
transfer    1.0
Name: proportion, dtype: float64

2. Класс PurchBehaviorHandler¶

  • модуль data_generator.fraud.drops.behavior. Ссылка на файл в Github
  • Управление поведением дропа покупателя

Условия

  • подразумевается, что дроп получает сумму и тратит её до получения новой

Основной функционал

  • управление поведением дропов покупателей
  • случайный выбор сценария распределения полученных денег - одна покупка на все деньги, несколько покупок частями - выбор сценария имеет зависимость от баланса. Например если баланс больше установленной суммы, то не может быть выбран сценарий с одной покупкой на всю сумму баланса.
  • определение кол-ва попыток совершить операцию после первой отклоненной транзакции
  • вычитание попыток когда дроп пытается совершать операции после первой отклоненной транз.
  • идентификация того нужно ли пытаться еще после последней отклоненной транзакции (это может быть первая отклоненная транз. если выпало 0 попыток)

Демонстрация¶

In [64]:
from data_generator.fraud.drops.behavior import PurchBehaviorHandler
from data_generator.fraud.drops.base import DropAmountHandler

amt_hand = DropAmountHandler(configs=purch_configs)
purch_behav = PurchBehaviorHandler(configs=purch_configs, amt_hand=amt_hand)

Метод sample_scenario

  • случайный выбор сценария
  • всего есть 2 сценария:
    • split_money - трата полученной суммы по частям
    • one_purchase - одна покупка покупка на всю полученную сумму

Доступные варианты сценариев зависят от величины баланса и лимитов по суммам операций выставленных в конфиг файле drops.yaml

  • зависимость от лимитов:
    • amt_max - максимальная сумма одной покупки. Если на балансе больше этой суммы, то доступен только вариант: split_money
    • amt_min
      1. если баланс больше или равен amt_min * 2, но меньше или равен amt_maxто доступны сценарии: split_money и one_purchase
      2. если баланс меньше amt_min * 2, то доступен только one_purchase

Пример №1

  • баланс больше чем amt_max
In [65]:
all_scens = []
i = 0
while i < 2000:
    amt_hand.reset_cache()
    amt_hand.balance = purch_behav.amt_max + 1000
    assert amt_hand.balance > purch_behav.amt_max, "Balance is below amt_max"
    purch_behav.sample_scenario()
    all_scens.append(purch_behav.scen)
    i += 1
    
all_scens_ser = pd.Series(all_scens)
all_scens_ser.value_counts(normalize=True)
Out[65]:
split_money    1.0
Name: proportion, dtype: float64

Пример №2

  • баланс больше чем amt_min * 2, но меньше чем amt_max
In [66]:
all_scens = []
i = 0
while i < 3000:
    amt_hand.reset_cache()
    amt_hand.balance = purch_behav.amt_min * 2
    assert amt_hand.balance < purch_behav.amt_max, "Balance exceeds amt_max"
    purch_behav.sample_scenario()
    all_scens.append(purch_behav.scen)
    i += 1
    
all_scens_ser = pd.Series(all_scens)
all_scens_ser.value_counts(normalize=True)
Out[66]:
split_money     0.689667
one_purchase    0.310333
Name: proportion, dtype: float64

Пример №3

  • баланс меньше чем amt_min * 2
In [67]:
all_scens = []
i = 0
while i < 3000:
    amt_hand.reset_cache()
    amt_hand.balance = purch_behav.amt_min + 1000
    assert amt_hand.balance < purch_behav.amt_min * 2, "Balance exceeds the limit"
    purch_behav.sample_scenario()
    all_scens.append(purch_behav.scen)
    i += 1
    
all_scens_ser = pd.Series(all_scens)
all_scens_ser.value_counts(normalize=True)
Out[67]:
one_purchase    1.0
Name: proportion, dtype: float64