Агрегатор личных финансов со всех счетов

Всем привет!

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

О чем

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

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

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

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