Всем привет!

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

О чем

В качестве входных данных берутся банковские выписки из мобильных приложений и веб-версий. На текущий момент одна из болей - это сходить в несколько мест и выгрузить вручную. Если у вас есть идеи покруче или примеры подобных проектов - велком в комменты, буду очень благодарна. Проект лежит тут.

Парсинг выгрузок

Так как у каждого банка свой формат, то я написала отдельные обработки. Посмотрим, чем они отличаются:

1. Сбер

Выписка может быть либо по дебетовой карте, либо по счету.

Посмотрев внимательно, можно увидеть, что поля "Категория" и "Название операции" расположены полярно противоположно. Также если в картах поступления идут с плюсом, а расходы без знака, то в счетах все наоборот: поступления без знака, а расходы с минусом.

Я попробовала несколько разных вариантов, но больше всех мне зашла библиотека PyMuPDF. Она точно определяла границы полей, и с ней было легко вытащить то, что нужно.

После предобработок мы получаем такой объект, который будет записываться в базу (об этом далее).

transaction = {
    'bank': 'Sber',
    'trans_datetime': datetime.strptime(' '.join((trans_date, trans_time)),
                                        '%d.%m.%Y %H:%M'),
    'transfer_datetime': datetime.strptime(transfer_date, '%d.%m.%Y'),
    'auth_code': auth_code,
    'category': category if is_debit_card else text,
    'debit': debit,
    'credit': credit,
    'text': text if is_debit_card else category
}

2. Тинькофф

Сначала я попробовала использовать выписки из приложения, но там оказалось сильно мало данных по сравнению с веб-версией (актуально на начало 2023 года):

Особенно печалило отсутствие категории, которой, кстати, нет и в текущей версии:

В этом случае я из веба выгружаю экселечку. Отчет читается в pandas датафрейм df = pd.read_excel(filename, sheet_name='Отчет по операциям', header=0), и все транзакции приводятся к виду:

transaction = {
    'bank': 'Tinkoff',
    'trans_datetime': datetime.strptime(trans_datetime, '%d.%m.%Y %H:%M:%S'),
    'transfer_datetime': None if pd.isna(transfer_datetime)
    else datetime.strptime(transfer_datetime, '%d.%m.%Y'),
    'pan': pan,
    'status': status,
    'debit': trans_sum if trans_sum > 0 else 0,
    'credit': -trans_sum if trans_sum < 0 else 0,
    'trans_currency': trans_currency,
    'pay_sum': pay_sum,
    'pay_currency': pay_currency,
    'cashback': cashback,
    'category': category,
    'mcc': mcc,
    'text': text,
    'bonus': float(bonus),
    'rounding': float(rounding),
    'sum_with_rounding': float(sum_with_rounding)
}

Идею с разделением на дебет и кредит я решила распространить на все банки.

3. Совкомбанк

Тут особая выгрузка в html формате.

Для работы с тегами использовала либу BeautifulSoup, в итоге получаем это:

transaction = {
    'bank': 'Sovcom',
    'trans_datetime': datetime.strptime(tds[0].find('p').get_text(), '%d.%m.%y'),
    'account': tds[1].find('p').get_text(),
    'income_balance': float(tds[2].find('p').get_text().replace(',', '')),
    'debit': float(tds[4].find('p').get_text().replace(',', '')),
    'credit': float(tds[3].find('p').get_text().replace(',', '')),
    'text': tds[5].find('p').get_text()
}

4. ВТБ

Аналогично Сберу, использовала либу PyMuPDF:

transaction = {
    'bank': 'VTB',
    'trans_datetime': datetime.strptime(
        ' '.join((trans_date, trans_time)),
        '%d.%m.%Y %H:%M:%S') if trans_date is not None else None,
    'transfer_datetime': datetime.strptime(transfer_date, '%d.%m.%Y'),
    'card_sum': float(card_sum.replace(' RUB', '')),
    'debit': float(debit),
    'credit': float(credit),
    'text': text[1:].replace(' Спасибо, что Вы с нами! Всегда Ваш, Банк ВТБ (ПАО)', '').strip()
}

Загрузка данных

В проекте используется Docker, который поднимает PostgreSQL, pgAdmin и Metabase. Для работы с базой использовала SQLAlchemy ORM. Есть пара фишек, которые я открыла для себя:

1. Создание схемы и табличек в этой схеме

Передаем через метадату, предварительно проверяя, что такой схемы еще не существует:

db_string = 'postgresql://{}:{}@{}:{}/{}'.format(db_user, db_pass, db_host, db_port, db_name)
engine = create_engine(db_string)
Session = sessionmaker(bind=engine)

if not engine.dialect.has_schema(engine, db_schema):
    engine.execute(CreateSchema(db_schema))

metadata_obj = MetaData(schema=db_schema)
Base = declarative_base(metadata=metadata_obj)

2. UniqueConstraint - ограничения на уникальность полей

Во время анализа данных я обнаружила, что есть транзакции, которые совершились, но еще не обработались банком. У таких транзакций в поле transfer_datetime отсутствует значение. При следующей выгрузке они уже будут идти с заполненной датой и временем обработки, что приведет к дублированию данных. Поэтому необходимо навесить ограничения на группу полей, которые в любом случае не изменятся:

__table_args__ = (
    UniqueConstraint(
        'bank',
        'trans_datetime',
        'category',
        'debit',
        'credit'
    )
)

Соответственно, если мы натыкаемся на дубликат, скипаем его и идем дальше:

for _, (source_transactions, TransactionClass) in transactions.items():
    for transaction in source_transactions:
        logger.debug(transaction)
        session.execute(insert(TransactionClass).values(transaction).on_conflict_do_nothing())
        session.commit()

Про дашборды

Зачем нужно было разбираться со всеми этими данными? Чтобы строить дашборды, смотреть на категории расходов и делать свои выводы.
В качестве BI-инструмента я взяла Metabase, потому что ранее с ним работала и он удобен с точки зрения написания SQL-запросов, а не drag-n-drop UI-элементов.

У меня есть очень частый кейс, когда я перевожу между своими картами. Много раз. Например:

Если брать общие обороты, то выходит, что я потратила 4000 и получила 3000. Поэтому имеет смысл исключать такие транзакции. Но как?
Так как потратить я могу любую сумму, то невозможно по ней определить источник: либо это часть перевода, либо деньги уже были. Поэтому я решила атрибуцировать транзакцию к первой операции - мы знаем, что сумма в переводах всегда одинаковая.

Как найти начало цепочки?
Если в текущем банке нет поступления на сумму перевода.
Как понять, что перевод не просто лежит на другой карте, а был использован?
Если количество операций в последующих банках после перевода четно: поступление-трата-поступление-трата...

Итак, считаем все реальные расходы:

t1.credit > 0
and t1.text != 'Перевод между счетами'
and t1.text not like '%VKLAD%'
--либо не перевод, либо перевод с доп. условиями
and (isTransfer = 0
    or isTransfer = 1
    --нет поступления в текущем банке на дату на ту же сумму, т.е. начало цепочки транзакций
    and not exists (
        select 1 from transactions t2
        where t2.bank = t1.bank
        and cast(t2.trans_datetime as date) = cast(t1.trans_datetime as date)
        and t2.debit = t1.credit
        and t2.isTransfer = 1
        and t2.text not like '%VKLAD%'
    )
    --количество операций в других банках на дату на ту же сумму четно: - +- +- +-
    and exists (
        select 1 from transactions t2
        where t2.bank != t1.bank
            and cast(t2.trans_datetime as date) = cast(t1.trans_datetime as date)
            and (t2.debit = t1.credit or t2.credit = t1.credit)
            and t2.isTransfer = 1
        having mod(case when count(*) != 0 then count(*) else 0 end, 2) = 0
    )
)

Предыдущая таблица с другой стороны:

По факту: по картам 1, 2, 3 я ничего не потратила, с 4 - 1000 рублей.
На дэше: карта 1 - трата 1к, 2, 3, 4 - 0 (есть поступление на ту же сумму и четное количество операций).

Еще и у Metabase есть свои недостатки. Один из них - невозможность использовать одну переменную на нескольких таблицах. То есть, если я хочу поставить фильтр по месяцу (=переменная) на запрос, где я соединяю несколько таблиц, то это невозможно. Приходится изобретать костыли и джойнить каждый подобный запрос с таким сниппетом:

join sber
    on extract(month from t1.trans_datetime) = extract(month from sber.trans_datetime)
    and extract(year from t1.trans_datetime) = extract(year from sber.trans_datetime)

На этом все, спасибо за прочтение!

Код доступен на гитхабе
tg: https://t.me/data_engineerette