A set of Beancount importers and scripts for popular Ukrainian banks and more
"""Allows to predict postings of a transaction using machine learning.
Adds the predicted postings inline to the transaction.

This is based on smart_importer project but instead of augmenting importers, it allows to run the prediction on already imported data.

Requires:
* autobean <https://github.com/SEIAROTg/autobean/tree/master/autobean/refactor>
* smart_importer <https://github.com/beancount/smart_importer>
"""

import argparse
import copy
import datetime
import logging
import re

from autobean.refactor import models, parser, printer
from sklearn.pipeline import FeatureUnion, make_pipeline
from sklearn.svm import SVC
from smart_importer.pipelines import get_pipeline

logger = logging.getLogger(__name__)


class Predictor:
    FLAG = "P"
    ignored_accounts = ["Expenses:Fees"]
    weights: dict[str, float] = {}
    attribute: str | None = None

    def __init__(self, main_accounts: list[str]):
        self.main_accounts = main_accounts
        if not getattr(self, "blacklisted_accounts", None):
            self.blacklisted_accounts = [
                "Equity:Opening-Balances",
                "Expenses:Fraud",
            ] + self.ignored_accounts

        self.training_data = []
        self.string_tokenizer = None
        self.is_fitted = False
        self.pipeline = None

    def load_training_data(self, directives):
        """Load training data, i.e., a list of Beancount entries."""
        training_data = directives or []
        training_data = list(
            filter(lambda d: isinstance(d, models.Transaction), directives)
        )
        length_all = len(training_data)
        training_data = [txn for txn in training_data if self.training_data_filter(txn)]
        if not training_data:
            if length_all > 0:
                logger.warning(
                    "Cannot train the machine learning model"
                    "None of the training data matches the accounts"
                )
            else:
                logger.warning(
                    "Cannot train the machine learning model: No training data found"
                )
        else:
            logger.debug(
                "Filtered training data to %s of %s entries.",
                len(training_data),
                length_all,
            )
        self.training_data = training_data

    def training_data_filter(self, txn):
        if "fraud" in txn.tags:
            return False
        num_accounts = 0
        found_import_account = False
        accounts = set()
        for pos in txn.postings:
            accounts.add(pos.account)
            if pos.account in self.main_accounts:
                found_import_account = True
                num_accounts += 1
            elif (
                any(bl in pos.account for bl in self.blacklisted_accounts)
                or pos.flag == self.FLAG
            ):
                num_accounts += 1
        return found_import_account and len(accounts) > num_accounts

    def predict_data_filter(self, txn):
        """Returns True if this transaction needs prediction."""
        if "fraud" in txn.tags:
            return False
        non_ignored_accounts_count = 0
        for p in txn.postings:
            if not any(i in p.account for i in self.ignored_accounts):
                non_ignored_accounts_count += 1
            if (
                p.account not in self.main_accounts
                and not any(i in p.account for i in self.ignored_accounts)
                and not p.flag == self.FLAG
            ):
                return False
        return non_ignored_accounts_count < 2

    def define_pipeline(self):
        """Defines the machine learning pipeline based on given weights."""

        transformers = []
        for attribute in self.weights:
            transformers.append(
                (attribute, get_pipeline(attribute, self.string_tokenizer))
            )

        self.pipeline = make_pipeline(
            FeatureUnion(
                transformer_list=transformers, transformer_weights=self.weights
            ),
            SVC(kernel="linear"),
        )

    def train_pipeline(self):
        """Train the machine learning pipeline."""

        targets = self.targets
        self.is_fitted = False

        if len(set(targets)) == 0:
            logger.warning(
                "Cannot train the machine learning model "
                "because there are no targets."
            )
        elif len(set(targets)) == 1:
            self.is_fitted = True
            logger.debug("Only one target possible.")
        else:
            self.pipeline.fit(self.training_data, targets)
            self.is_fitted = True
            logger.debug("Trained the machine learning model.")

    @property
    def targets(self):
        if not self.attribute:
            raise NotImplementedError
        return [getattr(entry, self.attribute) or "" for entry in self.training_data]

    def __call__(self, directives, predict_start_date=None):
        self.load_training_data(directives)
        self.define_pipeline()
        self.train_pipeline()
        return self.process_entries(directives, predict_start_date=predict_start_date)

    def process_entries(self, directives, predict_start_date=None):
        transactions = [
            d
            for d in directives
            if isinstance(d, models.Transaction)
            and self.predict_data_filter(d)
            and (predict_start_date is None or d.date >= predict_start_date)
        ]
        if len(transactions) == 0:
            logger.info("No transactions that need prediction")
            return []
        predictions = self.pipeline.predict(transactions)
        logger.info("got %d predictions.", len(predictions))
        transactions = [
            self.apply_prediction(entry, prediction)
            for entry, prediction in zip(transactions, predictions)
        ]
        return transactions

    def apply_prediction(self, directive, prediction):
        if not self.attribute:
            raise NotImplementedError
        setattr(directive, self.attribute, prediction)
        return directive


class PostingPredictor(Predictor):
    @property
    def targets(self):
        return [
            " ".join(
                p.account
                for p in txn.postings
                if not (
                    p.account in self.main_accounts
                    or any(bl in p.account for bl in self.blacklisted_accounts)
                )
            )
            for txn in self.training_data
        ]

    def apply_prediction(self, directive: models.Transaction, prediction: str):
        existing_posting = None
        for i, p in enumerate(directive.postings):
            if p.flag == self.FLAG:
                existing_posting = copy.deepcopy(p)
                directive.postings[i] = existing_posting
                break
        if existing_posting is None:
            existing_posting = copy.deepcopy(directive.postings[0])
            directive.postings.append(existing_posting)
        existing_posting.account = str(prediction)
        existing_posting.number = None
        existing_posting.currency = None
        existing_posting.flag = self.FLAG
        existing_posting.price = None
        existing_posting.cost = None
        return directive


class WisePredictor(PostingPredictor):
    ignored_accounts = ["Expenses:Fees:Wise"]
    weights = {"meta.src_category": 0.8, "payee": 0.5, "date.day": 0.1}


class MonobankPredictor(PostingPredictor):
    ignored_accounts = [
        "Expenses:Fees:Wise",
        "Assets:Monobank:Receivable:Cashback",
        "Income:Cashback:Monobank",
        "Expenses:Taxes",
        "Income:Monobank:Interest",
    ]
    weights = {"meta.category": 0.8, "payee": 0.5, "date.day": 0.1}


class SensebankPredictor(PostingPredictor):
    weights = {"meta.category": 0.8, "payee": 0.5, "narration": 0.5, "date.day": 0.1}


class PrivatbankPredictor(PostingPredictor):
    weights = {"meta.category": 0.6, "narration": 0.5, "date.day": 0.1}


PREDICTORS_CONFIG = [
    (r"Wise:Personal", WisePredictor),
    (r"Monobank", MonobankPredictor),
    (r"Alfabank", SensebankPredictor),
    (r"Privat", PrivatbankPredictor),
]


def get_predictor(accounts):
    for regex, predictor in PREDICTORS_CONFIG:
        for account in accounts:
            if re.search(regex, account):
                return predictor


def main():
    logging.basicConfig(level=logging.DEBUG)
    argparser = argparse.ArgumentParser()
    argparser.add_argument("-a", "--account", action="append")
    argparser.add_argument(
        "-d", "--predict-start-date", type=datetime.date.fromisoformat
    )
    argparser.add_argument("fname", type=argparse.FileType("r"))
    args = argparser.parse_args()
    PredictorClass = get_predictor(args.account)
    if PredictorClass is None:
        argparser.error("No predictor found for the given accounts")
    predictor = PredictorClass(args.account)
    beancount_parser = parser.Parser()
    file = beancount_parser.parse(args.fname.read(), models.File)
    predictor(file.directives, predict_start_date=args.predict_start_date)
    printer.print_model(file, open(args.fname.name, "w"))


if __name__ == "__main__":
    main()

scripts/predict_inplace.py
OSadovy/uabean
import sys

from autobean.refactor import models, parser, printer


def is_target_transaction(d: models.Transaction) -> bool:
    for p in d.postings:
        if p.account == "Expenses:Fraud":
            return True
    return False


def modify_transaction(t: models.Transaction) -> None:
    for p in t.postings:
        if p.account == "Expenses:Fees:Wise":
            t.postings.remove(p)
            break


f = parser.Parser().parse(open(sys.argv[1]).read(), models.File)
modified = 0
for d in f.directives:
    if isinstance(d, models.Transaction) and is_target_transaction(d):
        modify_transaction(d)
        modified += 1
print(f"Modified {modified} transactions")
printer.print_model(f, open(sys.argv[1] + ".new", "w"))

scripts/remove_posting.py
OSadovy/uabean
"""Sorts transactions in a ledger file by date and time.

requires: autobean <https://github.com/SEIAROTg/autobean/tree/master/autobean/refactor>
"""

import copy
import datetime
import sys

from autobean.refactor import models
from autobean.refactor.parser import Parser
from autobean.refactor.printer import print_model


def sort_transactions(file, start, end):
    def sortkey(d):
        if isinstance(d, models.IgnoredLine):
            return (datetime.date(1, 1, 1), "")
        return (d.date, d.meta.get("time", ""))

    if (
        sum(
            1 if isinstance(directive, models.Transaction) else 0
            for directive in file.directives[start:end]
        )
        > 1
    ):
        file.directives[start:end] = [
            copy.deepcopy(d) for d in sorted(file.directives[start:end], key=sortkey)
        ]


parser = Parser()
file = parser.parse(open(sys.argv[1]).read(), models.File)
start_range = None
for i, directive in enumerate(file.directives):
    if (
        isinstance(directive, models.Transaction)
        or isinstance(directive, models.IgnoredLine)
        or isinstance(directive, models.Balance)
    ):
        if start_range is None:
            start_range = i
    else:
        if start_range is not None:
            print(type(directive))
            sort_transactions(file, start_range, i)
            start_range = None
if start_range is not None:
    sort_transactions(file, start_range, len(file.directives))
print_model(file, open(sys.argv[1], "w"))

scripts/sorttransactions.py
OSadovy/uabean
"""
    Setup file for uabean.
    Use setup.cfg to configure your project.

    This file was generated with PyScaffold 4.4.1.
    PyScaffold helps you to put up the scaffold of your new Python project.
    Learn more under: https://pyscaffold.org/
"""
from setuptools import setup

if __name__ == "__main__":
    try:
        setup(use_scm_version={"version_scheme": "no-guess-dev"})
    except:  # noqa
        print(
            "\n\nAn error occurred while building the project, "
            "please ensure you have the most updated version of setuptools, "
            "setuptools_scm and wheel with:\n"
            "   pip install -U setuptools setuptools_scm wheel\n\n"
        )
        raise

setup.py
OSadovy/uabean
"""Downloads statements via monobank API and dumps them in CSV.

Expects MONOBANK_TOKEN env var to be set to the valid monobank API token.

Run with --help for usage.
"""


import argparse
import csv
import datetime
import os
import time
from dataclasses import dataclass
from decimal import Decimal

import requests

CURRENCY_CODES = {
    "784": "AED",
    "971": "AFN",
    "008": "ALL",
    "051": "AMD",
    "532": "ANG",
    "973": "AOA",
    "032": "ARS",
    "036": "AUD",
    "533": "AWG",
    "944": "AZN",
    "977": "BAM",
    "052": "BBD",
    "050": "BDT",
    "975": "BGN",
    "048": "BHD",
    "108": "BIF",
    "060": "BMD",
    "096": "BND",
    "068": "BOB",
    "984": "BOV",
    "986": "BRL",
    "044": "BSD",
    "064": "BTN",
    "072": "BWP",
    "933": "BYN",
    "084": "BZD",
    "124": "CAD",
    "976": "CDF",
    "947": "CHE",
    "756": "CHF",
    "948": "CHW",
    "990": "CLF",
    "152": "CLP",
    "156": "CNY",
    "170": "COP",
    "970": "COU",
    "188": "CRC",
    "931": "CUC",
    "192": "CUP",
    "132": "CVE",
    "203": "CZK",
    "262": "DJF",
    "208": "DKK",
    "214": "DOP",
    "012": "DZD",
    "818": "EGP",
    "232": "ERN",
    "230": "ETB",
    "978": "EUR",
    "242": "FJD",
    "238": "FKP",
    "826": "GBP",
    "981": "GEL",
    "936": "GHS",
    "292": "GIP",
    "270": "GMD",
    "324": "GNF",
    "320": "GTQ",
    "328": "GYD",
    "344": "HKD",
    "340": "HNL",
    "191": "HRK",
    "332": "HTG",
    "348": "HUF",
    "360": "IDR",
    "376": "ILS",
    "356": "INR",
    "368": "IQD",
    "364": "IRR",
    "352": "ISK",
    "388": "JMD",
    "400": "JOD",
    "392": "JPY",
    "404": "KES",
    "417": "KGS",
    "116": "KHR",
    "174": "KMF",
    "408": "KPW",
    "410": "KRW",
    "414": "KWD",
    "136": "KYD",
    "398": "KZT",
    "418": "LAK",
    "422": "LBP",
    "144": "LKR",
    "430": "LRD",
    "426": "LSL",
    "434": "LYD",
    "504": "MAD",
    "498": "MDL",
    "969": "MGA",
    "807": "MKD",
    "104": "MMK",
    "496": "MNT",
    "446": "MOP",
    "478": "MRO",
    "929": "MRU",
    "480": "MUR",
    "462": "MVR",
    "454": "MWK",
    "484": "MXN",
    "979": "MXV",
    "458": "MYR",
    "943": "MZN",
    "516": "NAD",
    "566": "NGN",
    "558": "NIO",
    "578": "NOK",
    "524": "NPR",
    "554": "NZD",
    "512": "OMR",
    "590": "PAB",
    "604": "PEN",
    "598": "PGK",
    "608": "PHP",
    "586": "PKR",
    "985": "PLN",
    "600": "PYG",
    "634": "QAR",
    "946": "RON",
    "941": "RSD",
    "643": "RUB",
    "646": "RWF",
    "682": "SAR",
    "090": "SBD",
    "690": "SCR",
    "938": "SDG",
    "752": "SEK",
    "702": "SGD",
    "654": "SHP",
    "694": "SLL",
    "706": "SOS",
    "968": "SRD",
    "728": "SSP",
    "930": "STN",
    "222": "SVC",
    "760": "SYP",
    "748": "SZL",
    "764": "THB",
    "972": "TJS",
    "795": "TMM",
    "934": "TMT",
    "788": "TND",
    "776": "TOP",
    "949": "TRY",
    "780": "TTD",
    "901": "TWD",
    "834": "TZS",
    "980": "UAH",
    "800": "UGX",
    "840": "USD",
    "997": "USN",
    "940": "UYI",
    "858": "UYU",
    "927": "UYW",
    "860": "UZS",
    "928": "VES",
    "704": "VND",
    "548": "VUV",
    "882": "WST",
    "950": "XAF",
    "961": "XAG",
    "959": "XAU",
    "955": "XBA",
    "956": "XBB",
    "957": "XBC",
    "958": "XBD",
    "951": "XCD",
    "960": "XDR",
    "952": "XOF",
    "964": "XPD",
    "953": "XPF",
    "962": "XPT",
    "994": "XSU",
    "963": "XTS",
    "965": "XUA",
    "999": "XXX",
    "886": "YER",
    "710": "ZAR",
    "894": "ZMK",
    "967": "ZMW",
    "932": "ZWL",
}


@dataclass
class Account:
    id: str
    send_id: str
    currency: str
    cashback_type: str
    balance: Decimal
    credit_limit: Decimal
    type: str
    masked_ban: list[str]
    iban: str


@dataclass
class ClientInfo:
    client_id: str
    name: str
    webhook_url: str
    permissions: str
    accounts: list[Account]


@dataclass
class StatementItem:
    id: str
    time: datetime.datetime
    description: str
    mcc: int
    original_mcc: int
    amount: Decimal
    operation_amount: Decimal
    currency: str
    commission_rate: Decimal
    cashback_amount: Decimal
    balance: Decimal
    hold: bool


class MonobankClient:
    BASE_URL = "https://api.monobank.ua"

    def __init__(self):
        self.headers = {"X-Token": os.getenv("MONOBANK_TOKEN")}

    @staticmethod
    def get_currency(code: int) -> str:
        return CURRENCY_CODES[str(code)]

    def client_info(self) -> ClientInfo:
        info = requests.get(
            self.BASE_URL + "/personal/client-info", headers=self.headers
        ).json()
        return ClientInfo(
            info["clientId"],
            info["name"],
            info["webHookUrl"],
            info["permissions"],
            accounts=[self._account(account) for account in info["accounts"]],
        )

    def _account(self, obj: dict) -> Account:
        return Account(
            obj["id"],
            obj["sendId"],
            self.get_currency(obj["currencyCode"]),
            obj["cashbackType"],
            self.get_decimal(obj["balance"]),
            self.get_decimal(obj["creditLimit"]),
            obj["type"],
            obj["maskedPan"],
            obj["iban"],
        )

    @staticmethod
    def get_decimal(s: str) -> Decimal:
        return Decimal(s) / 100

    def personal_statement(self, account_id, from_date, to_date) -> list[StatementItem]:
        response = requests.get(
            self.BASE_URL
            + f"/personal/statement/{account_id}/{int(from_date.timestamp())}/{int(to_date.timestamp())}",
            headers=self.headers,
        )
        response.raise_for_status()
        info = response.json()
        return [
            StatementItem(
                i["id"],
                datetime.datetime.fromtimestamp(i["time"]),
                i["description"],
                i["mcc"],
                i["originalMcc"],
                self.get_decimal(i["amount"]),
                self.get_decimal(i["operationAmount"]),
                self.get_currency(i["currencyCode"]),
                self.get_decimal(i["commissionRate"]),
                self.get_decimal(i["cashbackAmount"]),
                self.get_decimal(i["balance"]),
                i["hold"],
            )
            for i in info
        ]


class MonobankCSVWriter:
    headers = [
        "Дата i час операції",
        "Деталі операції",
        "MCC",
        "Сума в валюті картки ({currency})",
        "Сума в валюті операції",
        "Валюта",
        "Курс",
        "Сума комісій ({currency})",
        "Сума кешбеку ({cashback_type})",
        "Залишок після операції",
    ]

    def __init__(self, writer):
        self.writer = writer

    def write_header(self, account: Account):
        header = [
            h.format(currency=account.currency, cashback_type=account.cashback_type)
            for h in self.headers
        ]
        self.writer.writerow(header)

    def write_statement_item(self, account: Account, i: StatementItem):
        exchange_rate = "—"
        if i.currency != account.currency:
            exchange_rate = round(i.operation_amount / i.amount, 6)
        comission = "—"
        if i.commission_rate != 0:
            comission = i.commission_rate
        cashback = "—"
        if i.cashback_amount != 0:
            cashback = i.cashback_amount
        self.writer.writerow(
            [
                i.time.strftime("%d.%m.%Y %H:%M:%S"),
                i.description.replace("\n", " ").replace("\r", ""),
                i.mcc,
                i.amount,
                i.operation_amount,
                i.currency,
                exchange_rate,
                comission,
                cashback,
                i.balance,
            ]
        )


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "-s",
        "--start-date",
        type=lambda d: datetime.datetime.strptime(d, "%Y-%m-%d").date(),
        required=True,
    )
    parser.add_argument(
        "-e",
        "--end-date",
        type=lambda d: datetime.datetime.strptime(d, "%Y-%m-%d").date(),
        default=datetime.date.today(),
    )
    parser.add_argument("-c", "--currency")
    parser.add_argument("-t", "--account-type")
    parser.add_argument("-o", "--output-dir")
    args = parser.parse_args()
    client = MonobankClient()
    info = client.client_info()
    accounts = info.accounts
    if args.currency:
        accounts = [a for a in accounts if a.currency == args.currency]
    if args.account_type:
        accounts = [a for a in accounts if a.type == args.account_type]
    start_time = datetime.datetime.fromordinal(args.start_date.toordinal())
    end_time = datetime.datetime(
        args.end_date.year, args.end_date.month, args.end_date.day, 23, 59, 59
    )
    end_time = min(end_time, datetime.datetime.now())
    writers = {}
    for account in accounts:
        dt = end_time.strftime("%d-%m-%y_%H-%M-%S")
        writer = csv.writer(
            open(
                os.path.join(
                    args.output_dir,
                    f"monobank-{account.type}-{account.currency}_{dt}.csv",
                ),
                "w",
                encoding="utf-8",
            ),
            quoting=csv.QUOTE_MINIMAL,
        )
        writers[account.id] = MonobankCSVWriter(writer)
        writers[account.id].write_header(account)

    while start_time < end_time:
        for account in accounts:
            print(
                f"Downloading {account.type} {account.currency} for {start_time.isoformat()}"
            )
            while True:
                try:
                    statement = client.personal_statement(
                        account.id,
                        start_time,
                        min(end_time, start_time + datetime.timedelta(days=31)),
                    )
                    break
                except requests.exceptions.HTTPError:
                    print("Rate limit encountered, waiting")
                    time.sleep(15)
            for i in sorted(statement, key=lambda s: s.time):
                writers[account.id].write_statement_item(account, i)
            time.sleep(15)
        start_time += datetime.timedelta(days=31)


if __name__ == "__main__":
    main()

src/uabean/downloaders/monobank.py
OSadovy/uabean
import argparse
import base64
import json
import os
from datetime import date, datetime, timezone

import requests
import rsa
from dateutil.parser import isoparse


def do_sca_challenge(one_time_token):
    with open("wise-private.pem", "rb") as f:
        private_key_data = f.read()

    private_key = rsa.PrivateKey.load_pkcs1(private_key_data, "PEM")

    # Use the private key to sign the one-time-token that was returned
    # in the x-2fa-approval header of the HTTP 403.
    signed_token = rsa.sign(one_time_token.encode("ascii"), private_key, "SHA-256")
    signature = base64.b64encode(signed_token).decode("ascii")
    return signature


BASE_URL = "https://api.transferwise.com"


class WiseClient:
    def __init__(self, api_token, base_url=BASE_URL):
        self.base_url = base_url
        self.headers = {"Authorization": f"Bearer {api_token}"}

    def list_profiles(self):
        return requests.get(self.base_url + "/v1/profiles", headers=self.headers).json()

    def list_accounts(self, profile_id):
        return requests.get(
            self.base_url + "/v1/borderless-accounts",
            params={"profileId": profile_id},
            headers=self.headers,
        ).json()

    def list_balance_accounts(self, profile_id):
        return requests.get(
            self.base_url + f"/v4/profiles/{profile_id}/balances",
            params={"types": "STANDARD"},
            headers=self.headers,
        ).json()

    def get_account_statement(
        self, profile_id, account_id, currency, start_date, end_date, format="csv"
    ):
        headers = dict(self.headers)
        params = {
            "currency": currency,
            "intervalStart": start_date,
            "intervalEnd": end_date,
            "type": "COMPACT",
        }
        url = (
            self.base_url
            + f"/v3/profiles/{profile_id}/borderless-accounts/{account_id}/statement.{format}"
        )
        r = requests.get(url, params=params, headers=headers)
        if r.status_code == 403:
            one_time_token = r.headers["x-2fa-approval"]
            signature = do_sca_challenge(one_time_token)
            headers["x-2fa-approval"] = one_time_token
            headers["x-signature"] = signature
            r = requests.get(url, params=params, headers=headers)
        return r.text

    def get_balance_statement(
        self, profile_id, account_id, currency, start_date, end_date, format="csv"
    ):
        headers = dict(self.headers)
        params = {
            "currency": currency,
            "intervalStart": start_date,
            "intervalEnd": end_date,
            "type": "COMPACT",
        }
        url = (
            self.base_url
            + f"/v1/profiles/{profile_id}/balance-statements/{account_id}/statement.{format}"
        )
        r = requests.get(url, params=params, headers=headers)
        if r.status_code == 403:
            one_time_token = r.headers["x-2fa-approval"]
            signature = do_sca_challenge(one_time_token)
            headers["x-2fa-approval"] = one_time_token
            headers["x-signature"] = signature
            r = requests.get(url, params=params, headers=headers)
        return r.text


def main():
    parser = argparse.ArgumentParser(__name__)
    parser.add_argument(
        "--start-date",
        required=False,
        default=None,
        type=lambda d: datetime.strptime(d, "%Y-%m-%d").date(),
    )
    parser.add_argument(
        "--end-date",
        required=False,
        default=date.today(),
        type=lambda d: datetime.strptime(d, "%Y-%m-%d").date(),
    )
    parser.add_argument("-t", "--account-type", choices=("business", "personal"))
    parser.add_argument("-f", "--format", choices=("csv", "json"))
    parser.add_argument("-c", "--currency")
    parser.add_argument("-o", "--out-dir", default="wise")
    args = parser.parse_args()
    wise = WiseClient(os.getenv("WISE_API_TOKEN"))
    profiles = wise.list_profiles()
    if args.account_type is not None:
        profiles = [p for p in profiles if p["type"] == args.account_type]
    for profile in profiles:
        profile_id = profile["id"]
        accounts = wise.list_accounts(profile_id)
        balance_accounts = wise.list_balance_accounts(profile_id)
        if (
            args.start_date is None
            or isoparse(accounts[0]["creationTime"]).date() > args.start_date
        ):
            start_date = accounts[0]["creationTime"]
        else:
            start_date = datetime.combine(
                args.start_date, datetime.min.time(), timezone.utc
            ).isoformat()
        end_date = datetime.combine(
            args.end_date, datetime.max.time(), timezone.utc
        ).isoformat()
        if isoparse(end_date) < isoparse(start_date):
            continue
        # for account in accounts[0]["balances"]:
        for balance_account in balance_accounts:
            if (
                args.currency is not None
                and args.currency.upper() != balance_account["currency"]
            ):
                continue
            # statement = wise.get_csv_statement(profile_id, account_id, account["currency"], start_date, end_date)
            statement = wise.get_balance_statement(
                profile_id,
                balance_account["id"],
                balance_account["currency"],
                start_date,
                end_date,
                format=args.format,
            )
            if not statement.strip():
                continue
            if args.format == "json":
                statement = json.dumps(json.loads(statement), indent=4)
            sd = isoparse(start_date).strftime("%Y-%m-%d")
            ed = isoparse(end_date).strftime("%Y-%m-%d")
            # fname = f"wise-{profile['type']}-{sd}_{ed}-{account['currency']}.csv"
            fname = f"wise-{profile['type']}-{sd}_{ed}-{balance_account['currency']}.{args.format}"
            with open(os.path.join(args.out_dir, fname), "w") as f:
                f.write(statement)
            print(f"Wrote {fname}")


if __name__ == "__main__":
    main()

src/uabean/downloaders/wise.py
OSadovy/uabean
import itertools
import os
from collections import defaultdict
from decimal import Decimal

from beancount.core.data import Transaction


def detect_transfers(extracted, existing_entries):
    """Merges transactions with single posting that happened close to each other
    in time and have the same absolute value.
    Catches transfers from one card to another within the same bank etc.
    """
    possible_transfers = defaultdict(list)
    for filename, new_entries, account, importer in extracted:
        for entry in new_entries[:]:
            if not isinstance(entry, Transaction):
                continue
            k, v = get_transfer_info(entry)
            found = False
            for other_v, other_entry in possible_transfers[k]:
                if are_values_oposit(v, other_v) and not same_accounts(
                    other_entry, entry
                ):
                    found = True
                    merge_transactions(other_entry, entry)
                    new_entries.remove(entry)
                    break
            if not found:
                possible_transfers[k].append((v, entry))
    return extracted


def get_transfer_info(entry):
    time = "00:00:00"
    if "time" in entry.meta:
        time = to_nearest_time(entry.meta["time"], unit="hour", rnd=1, frm="%H:%M:%S")
    s = sum((p.units.number for p in entry.postings if p.units), Decimal(0))
    return (entry.date, time), s


def are_values_oposit(v1, v2):
    """Returns if v1 and v2 have oposit signs and are within 1% tolerance value."""
    if (v1 > 0 and v2 > 0) or (v1 < 0 and v2 < 0):
        return False
    if v2 == 0:
        return False
    return round(abs(abs(v1 / v2) - 1), 2) <= 0.01


def same_accounts(t1, t2):
    return set(p.account for p in t1.postings) == set(p.account for p in t2.postings)


def merge_transactions(t1, t2):
    t1.postings.extend(t2.postings)


def to_nearest_time(ts, unit="sec", rnd=1, frm=None):
    """round to nearest Time format
    param ts = time string to round in '%H:%M:%S' or '%H:%M' format :
    param unit = specify unit wich must be rounded 'sec' or 'min' or 'hour', default is seconds :
    param rnd = to which number you will round, the default is 1 :
    param frm = the output (return) format of the time string, as default the function take the unit format
    """
    from time import gmtime, strftime

    ts = ts + ":00" if len(ts) == 5 else ts
    if "se" in unit.lower():
        frm = "%H:%M:%S" if frm is None else frm
    elif "m" in unit.lower():
        frm = "%H:%M" if frm is None else frm
        rnd = rnd * 60
    elif "h" in unit.lower():
        frm = "%H" if frm is None else frm
        rnd = rnd * 3600
    secs = sum(int(x) * 60**i for i, x in enumerate(reversed(ts.split(":"))))
    rtm = int(round(secs / rnd, 0) * rnd)
    nt = strftime(frm, gmtime(rtm))
    return nt


def sort_samebank_entries(new_entries_list, existing_entries):
    """Sorts entries that are originated from files with same prefix."""
    new_entries_list.sort(key=lambda e: e[0])
    for k, group in itertools.groupby(
        new_entries_list, key=lambda e: os.path.basename(e[0]).split("-", 1)[0]
    ):
        pass

src/uabean/hooks.py
OSadovy/uabean
from .alfa_business import Importer as AlfaBusinessImporter
from .binance import Importer as BinanceImporter
from .ibkr import Importer as IBKRImporter
from .monobank import Importer as MonobankImporter
from .nexo import Importer as NexoImporter
from .privatbank_xls import Importer as PrivatbankXLSImporter
from .privatbank_xml import Importer as PrivatbankXMLImporter
from .sensebank import Importer as SensebankImporter
from .tronscan import Importer as TronscanImporter
from .ukrsib_business import Importer as UkrsibBusinessImporter
from .wise_csv import Importer as WiseCSVImporter
from .wise_json import Importer as WiseJSONImporter

src/uabean/importers/__init__.py
OSadovy/uabean
"""Imports statement csv files exported from Alfabank Business Online web app.

The CSV header is the following:
Наш рахунок;Наш IBAN;Операція;Рахунок;IBAN;МФО банку контрагента;Найменування контрагента;Код контрагента;Призначення платежу;Дата проведення;Номер документа;Сума;Валюта;Час проведення;Дата документа;Дата архівування;Ід.код;Найменування;МФО
"""

import csv
import re
from collections import defaultdict

import beangulp
from beancount.core import data, flags
from beancount.core.number import D
from beancount.utils.date_utils import parse_date_liberally

from uabean.importers.mixins import IdentifyMixin


class Importer(IdentifyMixin, beangulp.Importer):
    matchers = [
        ("content", __doc__.split("\n")[-2]),
        ("mime", "text/csv"),
    ]
    fee_regexes = ("Погашення комісії",)
    currency_conversion_regex = r"Зарахування коштiв вiд вільного продажу (?P<amount>[\d\.]+) (?P<currency>\w+) по курсу (?P<rate>[\d\.]+).*?Комiсiя банку становить (?P<fee>[\d\.]+) грн\."
    DATE_FIELD = "Дата проведення"

    def __init__(
        self, accounts: dict[tuple[str, str], str], fee_account: str, *args, **kwargs
    ):
        """Initialize an instance of the Importer class.

        Parameters:
        - accounts (dict[tuple[str, str], str]): A dictionary mapping from a tuple of currency and
          IBAN to the associated account name. For instance, it maps pairs of currency and IBAN
          to a specific account string.
        - fee_account (str): Name of the account where transaction fees should be posted.
        """
        self.accounts = accounts
        self.fee_account = fee_account
        super().__init__(*args, **kwargs)

    def get_csv_dict_rows(self, filename):
        with open(filename, encoding="windows-1251") as f:
            return list(csv.DictReader(f, delimiter=";"))

    def get_date_from_row(self, row):
        return parse_date_liberally(row[self.DATE_FIELD], dict(dayfirst=True))

    def get_account_from_row(self, row):
        k = (row["Валюта"], row["Наш IBAN"])
        if k not in self.accounts:
            raise ValueError("Unknown account %s %s" % (row["Валюта"], row["Наш IBAN"]))
        return self.accounts[k]

    def account(self, filename):
        return "alfabank_business"

    def date(self, filename):
        "Get the maximum date from the file."
        max_date = None
        for row in self.get_csv_dict_rows(filename):
            if not row:
                continue
            date = self.get_date_from_row(row)
            if max_date is None or date > max_date:
                max_date = date
        return max_date

    def extract(self, filename, existing_entries=None):
        entries = []
        for index, row in enumerate(self.get_csv_dict_rows(filename), 1):
            if not row:
                continue
            meta = data.new_metadata(filename, index)
            entry = self.get_entry_from_row(row, meta)
            if entry is not None:
                entries.append(entry)

        return self.merge_entries(entries)

    def get_entry_from_row(self, row, meta):
        meta["time"] = row["Час проведення"]
        meta["src_doc_n"] = row["Номер документа"]
        meta["src_purpose"] = row["Призначення платежу"]
        account = self.get_account_from_row(row)
        if account is None:
            return
        payee = row["Найменування контрагента"]
        txn = data.Transaction(
            meta,
            self.get_date_from_row(row),
            flags.FLAG_OKAY,
            payee,
            "",
            data.EMPTY_SET,
            data.EMPTY_SET,
            [],
        )
        sum = D(row["Сума"])
        op = row["Операція"]
        if op == "Дебет":
            sum = -sum
        elif op == "Кредит":
            pass
        else:
            raise RuntimeError(f"unknown operation {op}")
        units = data.Amount(sum, row["Валюта"])
        txn.postings.append(data.Posting(account, units, None, None, None, None))
        for fee_str in self.fee_regexes:
            if re.search(fee_str, row["Призначення платежу"]):
                txn.postings.append(
                    data.Posting(self.fee_account, -units, None, None, None, None)
                )
                break
        return txn

    def merge_entries(self, entries):
        def find_closest(lst, other, predicate):
            for e in lst:
                if not e == other and predicate(e):
                    return e
            return None

        entries_by_date = defaultdict(list)
        for e in entries:
            entries_by_date[e.date].append(e)
        for _date, subentries in entries_by_date.items():
            if len(subentries) == 1:
                continue
            # find & merge same currency transfers
            for e in subentries:
                other = find_closest(
                    subentries, e, lambda x: x.meta["src_doc_n"] == e.meta["src_doc_n"]
                )
                if other is not None:
                    entries.remove(other)
                    subentries.remove(e)
                    e.postings.extend(other.postings)
            # find & merge currency convertions
            for e in subentries:
                m = re.search(self.currency_conversion_regex, e.meta["src_purpose"])
                if not m:
                    continue
                units = data.Amount(D(m.group("amount")), m.group("currency"))
                other = find_closest(
                    subentries, e, lambda x: x.postings[0].units == -units
                )
                if other is None:
                    raise RuntimeError("can't find matching         entry for %s" % e)
                other_posting = other.postings[0]._replace(
                    price=data.Amount(D(m.group("rate")), "UAH")
                )
                e.postings.insert(0, other_posting)
                e.postings.append(
                    data.Posting(
                        self.fee_account,
                        data.Amount(D(m.group("fee")), "UAH"),
                        None,
                        None,
                        None,
                        None,
                    )
                )
                e.postings.extend(other.postings[1:])
                e.meta["other_src_doc_n"] = other.meta["src_doc_n"]
                entries.remove(other)
        return entries


def get_test_importer():
    return Importer(
        {
            (
                "UAH",
                "UA11111111111111111111111111",
            ): "Assets:Alfabank:Business:Cash:UAH",
            (
                "GBP",
                "UA11111111111111111111111111",
            ): "Assets:Alfabank:Business:Cash:GBP",
            ("GBP", "UA222222222222222222222"): "Assets:Alfabank:Business:Transit",
        },
        "Expenses:Fees:Banking:Alfabank",
    )


if __name__ == "__main__":
    from beangulp.testing import main

    main(get_test_importer())

src/uabean/importers/alfa_business.py
OSadovy/uabean
"""Imports Binance statements from archive downloaded at wallet -> Transaction History -> generate all statements.

CSV Header:
User_ID,UTC_Time,Account,Operation,Coin,Change,Remark
"""

import csv
import io
import itertools
import tarfile

import beangulp
import dateutil.parser
from beancount.core import data, flags
from beancount.core.amount import Amount
from beancount.core.number import D


class Importer(beangulp.Importer):
    FLAG = flags.FLAG_OKAY

    def __init__(
        self,
        spot_wallet_account="Assets:Binance:Spot",
        fees_account="Expenses:Fees:Binance",
        p2p_account="Assets:Binance:P2P",
        savings_account="Assets:Binance:Savings",
        income_interest_account="Income:Binance:Interest",
        income_referal_account="Income:Binance:Referal",
        income_distributions_account="Income:Binance:Distribution",
        **kwargs,
    ):
        self.spot_wallet_account = spot_wallet_account
        self.fees_account = fees_account
        self.p2p_account = p2p_account
        self.savings_account = savings_account
        self.income_interest_account = income_interest_account
        self.income_referal_account = income_referal_account
        self.income_distributions_account = income_distributions_account
        super().__init__(**kwargs)

    def identify(self, filename):
        if not (filename.endswith(".tar.gz") or filename.endswith(".csv")):
            return False
        return self.get_csv_reader(filename).fieldnames == [
            "User_ID",
            "UTC_Time",
            "Account",
            "Operation",
            "Coin",
            "Change",
            "Remark",
        ]

    def get_csv_reader(self, filename):
        if filename.endswith(".csv"):
            return csv.DictReader(open(filename, errors="ignore"))
        elif filename.endswith(".tar.gz"):
            tar = tarfile.open(filename, "r:gz")
            for tarinfo in tar:
                f = tar.extractfile(tarinfo.name)
                return csv.DictReader(io.TextIOWrapper(f, "utf-8"))

    def account(self, _):
        return "_".join(p.lower() for p in self.spot_wallet_account.split(":")[1:-1])

    def parse_date(self, s):
        return dateutil.parser.parse(s)

    def extract(self, filename, existing_entries=None):
        entries = []
        prev_rows = []
        reader = self.get_csv_reader(filename)
        for _, rows in itertools.groupby(reader, key=lambda r: r["UTC_Time"]):
            lst_rows = list(rows)
            if {row["Operation"] for row in lst_rows} == {"Transaction Related"}:
                lst_rows = prev_rows + lst_rows
                entries.pop(-1)
            meta = data.new_metadata(filename, reader.line_num)
            transaction = self.transaction_from_rows(lst_rows, meta)
            prev_rows = lst_rows
            entries.append(transaction)
        return entries

    def transaction_from_rows(self, rows, meta):
        ops = {row["Operation"] for row in rows}
        ops = {
            "Small assets exchange" if o.startswith("Small assets exchange") else o
            for o in ops
        }
        accounts = {row["Account"].capitalize() for row in rows}
        dt = self.parse_date(rows[0]["UTC_Time"])
        if (
            accounts == {"Spot"}
            and "Deposit" in ops
            and ops.issubset({"Deposit", "Transaction Related"})
        ):
            narration = "Deposit"
            postings = [
                data.Posting(
                    self.spot_wallet_account,
                    amount_from_row(row),
                    None,
                    None,
                    None,
                    None,
                )
                for row in rows
            ]
        elif accounts == {"Spot"} and all(
            is_buy_row(r)
            or is_sell_row(r)
            or is_transaction_related_row(r)
            or is_fee_row(r)
            or is_referal_row(r)
            for r in rows
        ):
            quantity = D("0")
            coin = None
            sell = any(is_sell_row(r) for r in rows)
            other_leg = filter(lambda r: is_counter_row(r, sell), rows)
            postings = []
            for row in rows:
                units = amount_from_row(row)
                if (is_buy_row(row) and units.number > 0) or (
                    is_sell_row(row) and units.number < 0
                ):
                    quantity += D(row["Change"])
                    if coin is not None and not coin == row["Coin"]:
                        raise RuntimeError(
                            f"multiple coins bought: {coin} and {row['Coin']}. {row}"
                        )
                    coin = row["Coin"]
                    other_row = next(other_leg)
                    price = Amount(
                        abs(
                            round(D(other_row["Change"]) / D(row["Change"]), 6)
                        ).normalize(),
                        other_row["Coin"],
                    )
                    postings.append(
                        data.Posting(
                            self.spot_wallet_account, units, None, price, None, None
                        )
                    )
                elif is_fee_row(row):
                    postings += [
                        data.Posting(
                            self.spot_wallet_account, units, None, None, None, None
                        ),
                        data.Posting(
                            self.fees_account,
                            Amount(-D(row["Change"]), row["Coin"]),
                            None,
                            None,
                            None,
                            None,
                        ),
                    ]
                elif is_referal_row(row):
                    postings += [
                        data.Posting(
                            self.spot_wallet_account, units, None, None, None, None
                        ),
                        data.Posting(
                            self.income_referal_account,
                            Amount(-D(row["Change"]), row["Coin"]),
                            None,
                            None,
                            None,
                            None,
                        ),
                    ]
                else:
                    postings.append(
                        data.Posting(
                            self.spot_wallet_account, units, None, None, None, None
                        )
                    )
            quantity = quantity.normalize()
            if sell:
                narration = f"Sell {-quantity} {coin}"
            else:
                narration = f"Buy {quantity} {coin}"
        elif accounts == {"P2p"}:
            narration = rows[0]["Operation"]
            postings = [
                data.Posting(
                    self.p2p_account,
                    amount_from_row(row),
                    None,
                    None,
                    None,
                    None,
                )
                for row in rows
            ]
        elif accounts == {"Spot"} and ops.issubset(
            {"POS savings purchase", "Simple Earn Locked Subscription"}
        ):
            narration = "Savings"
            postings = []
            for row in rows:
                units = amount_from_row(row)
                postings += [
                    data.Posting(
                        self.spot_wallet_account, units, None, None, None, None
                    ),
                    data.Posting(
                        self.savings_account,
                        Amount(-units.number, units.currency),
                        None,
                        None,
                        None,
                        None,
                    ),
                ]
        elif accounts == {"Earn"} and ops == {"Simple Earn Flexible Redemption"}:
            narration = "Redemption"
            postings = []
            assert len(rows) == 2
            for row in rows:
                units = amount_from_row(row)
                if units.number < 0:
                    acc = self.savings_account
                else:
                    acc = self.spot_wallet_account
                postings.append(data.Posting(acc, units, None, None, None, None))
        elif accounts == {"Spot"} and ops == {"Simple Earn Locked Redemption"}:
            narration = "Redemption"
            assert len(rows) == 1
            units = amount_from_row(rows[0])
            postings = [
                data.Posting(self.savings_account, -units, None, None, None, None),
                data.Posting(self.spot_wallet_account, units, None, None, None, None),
            ]
        elif ops.issubset(
            {
                "POS savings interest",
                "Savings Interest",
                "Simple Earn Locked Rewards",
                "Simple Earn Flexible Interest",
            }
        ):
            narration = "Interest"
            postings = []
            for row in rows:
                units = amount_from_row(row)
                postings += [
                    data.Posting(
                        self.spot_wallet_account, units, None, None, None, None
                    ),
                    data.Posting(
                        self.income_interest_account,
                        Amount(-units.number, units.currency),
                        None,
                        None,
                        None,
                        None,
                    ),
                ]
        elif ops == {"Savings purchase"}:
            narration = "Savings"
            postings = []
            for row in rows:
                units = amount_from_row(row)
                postings.append(
                    data.Posting(
                        self.spot_wallet_account
                        if units.number < 0
                        else self.savings_account,
                        units,
                        None,
                        None,
                        None,
                        None,
                    )
                )
        elif accounts == {"Spot"} and ops == {"Withdraw"}:
            narration = "Withdraw"
            postings = [
                data.Posting(
                    self.spot_wallet_account,
                    amount_from_row(row),
                    None,
                    None,
                    None,
                    None,
                )
                for row in rows
            ]
        elif ops == {"Distribution"}:
            narration = "Distribution"
            postings = []
            for row in rows:
                units = amount_from_row(row)
                postings += [
                    data.Posting(
                        self.spot_wallet_account, units, None, None, None, None
                    ),
                    data.Posting(
                        self.income_distributions_account,
                        Amount(-units.number, units.currency),
                        None,
                        None,
                        None,
                        None,
                    ),
                ]
        else:
            raise RuntimeError(f"don't know how to process rows with operations {ops}")
        meta["time"] = dt.time().strftime("%H:%M:%S")
        return data.Transaction(
            meta,
            dt.date(),
            self.FLAG,
            None,
            narration,
            data.EMPTY_SET,
            data.EMPTY_SET,
            postings,
        )


def is_counter_row(row, sell=False):
    if sell:
        return (is_sell_row(row) or is_transaction_related_row(row)) and D(
            row["Change"]
        ) > 0
    return (is_buy_row(row) or is_transaction_related_row(row)) and D(row["Change"]) < 0


def is_transaction_related_row(row):
    return row["Operation"] in {"Transaction Related", "Transaction Revenue"}


def amount_from_row(row):
    return Amount(D(row["Change"]).normalize(), row["Coin"])


def is_buy_row(row):
    return row["Operation"] == "Buy" or row["Operation"].startswith(
        "Small assets exchange"
    )


def is_sell_row(row):
    return row["Operation"] in {"Sell", "Transaction Sold"}


def is_referal_row(row):
    return row["Operation"] in {
        "Referral Commission",
        "Commission Fee Shared With You",
        "Referral Kickback",
    }


def is_fee_row(row):
    return row["Operation"] == "Fee"


def get_test_importer():
    return Importer()


if __name__ == "__main__":
    from beangulp.testing import main

    main(get_test_importer())

src/uabean/importers/binance.py
OSadovy/uabean
"""
This is a beancount importer for Interactive Brokers.

Inspired by https://github.com/Dr-Nuke/drnuke-bean/blob/master/src/drnukebean/importer/ibkr.py

Supports selling short, corporate actions and more.

It takes FlexQuery xml as input. The report should include all fields for these sections:
* Trades
 * Options:Executions, Closed Lots
* Cash Transactions
 * Options:Dividends, Payment in Lieu of Dividends, Withholding Tax, 871(m) Withholding, Advisor Fees, Other Fees, Deposits/Withdrawals, Carbon Credits, Bill Pay, Broker Interest Paid, Broker Interest Received, Broker Fees, Bond Interest Paid, Bond Interest Received, Price Adjustments, Commission Adjustments, Detail
* Cash Report
 * Options:Currency Breakout
* Corporate Actions
 * Options:Detail

The XML can be downloaded from interactive brokers manually or via an API.
"""

import functools
import re
import warnings
from collections import defaultdict
from datetime import timedelta

import beangulp
from beancount.core import amount, data, flags, position, realization
from beancount.core.number import Decimal
from ibflex import Types, parser
from ibflex.enums import BuySell, CashAction, OpenClose, Reorg

from uabean.importers.mixins import IdentifyMixin


class Importer(IdentifyMixin, beangulp.Importer):
    """
    Beancount Importer for the Interactive Brokers XML FlexQueries
    """

    matchers = [
        ("mime", "application/xml"),
        ("content", "<FlexQueryResponse "),
    ]

    def __init__(
        self,
        cash_account="Assets:Investments:IB:Cash",
        assets_account="Assets:Investments:IB:{symbol}",
        div_account="Income:Investments:IB:{symbol}:Div",
        interest_account="Income:Investments:IB:Interest",
        wht_account="Expenses:Investments:IB:WithholdingTax",
        fees_account="Expenses:IB:Fees",
        pnl_account="Income:Investments:IB:{symbol}:PnL",
        use_existing_holdings=True,
        **kwargs,
    ):
        self.cash_account = cash_account
        self.assets_account = assets_account
        self.div_account = div_account
        self.interest_account = interest_account
        self.wht_account = wht_account
        self.fees_account = fees_account
        self.bnl_account = pnl_account
        self.use_existing_holdings = use_existing_holdings
        self.holdings_map = defaultdict(list)
        super().__init__(**kwargs)

    def get_account_name(self, account_str, symbol=None, currency=None):
        if symbol is not None:
            account_str = account_str.replace("{symbol}", symbol.replace(" ", ""))
        if currency is not None:
            account_str = account_str.replace("{currency}", currency)
        return account_str

    def get_liquidity_account(self, currency):
        return self.get_account_name(self.cash_account, currency=currency)

    def get_div_income_account(self, currency, symbol):
        return self.get_account_name(self.div_account, symbol=symbol, currency=currency)

    def get_interest_income_account(self, currency):
        return self.get_account_name(self.interest_account, currency=currency)

    def get_asset_account(self, symbol):
        return self.get_account_name(self.assets_account, symbol=symbol)

    def get_wht_account(self, symbol):
        return self.get_account_name(self.wht_account, symbol=symbol)

    def get_fees_account(self, currency):
        return self.get_account_name(self.fees_account, currency=currency)

    def get_pnl_account(self, symbol):
        return self.get_account_name(self.bnl_account, symbol=symbol)

    def account(self, filename):
        return "ib"

    def extract(self, filename, existing_entries=None):
        if self.use_existing_holdings and existing_entries is not None:
            self.holdings_map = self.get_holdings_map(existing_entries)
        else:
            self.holdings_map = defaultdict(list)
        statement = parser.parse(open(filename))
        assert isinstance(statement, Types.FlexQueryResponse)
        poi = statement.FlexStatements[0]  # point of interest
        transactions = (
            self.Trades(poi.Trades)
            + self.cash_transactions(poi.CashTransactions)
            + self.Balances(poi.CashReport)
            + self.corporate_actions(poi.CorporateActions)
        )

        transactions = self.merge_dividend_and_withholding(transactions)
        # self.adjust_closing_trade_cost_basis(transactions)
        return self.autoopen_accounts(transactions, existing_entries) + transactions

    def cash_transactions(self, ct):
        transactions = []
        for index, row in enumerate(ct):
            if row.type == CashAction.DEPOSITWITHDRAW:
                transactions.append(self.deposit_from_row(index, row))
            elif row.type in (CashAction.BROKERINTRCVD, CashAction.BROKERINTPAID):
                transactions.append(self.Interest_from_row(index, row))
            elif row.type in (CashAction.FEES, CashAction.COMMADJ):
                transactions.append(self.fee_from_row(index, row))
            elif row.type in (
                CashAction.WHTAX,
                CashAction.DIVIDEND,
                CashAction.PAYMENTINLIEU,
            ):
                transactions.append(
                    self.dividends_and_withholding_tax_from_row(index, row)
                )
            else:
                raise RuntimeError(f"Unknown cash transaction type: {row.type}")
        return transactions

    def fee_from_row(self, idx, row):
        amount_ = amount.Amount(row.amount, row.currency)
        text = row.description
        try:
            month = re.findall(r"\w{3} \d{4}", text)[0]
            narration = " ".join(["Fee", row.currency, month])
        except IndexError:
            narration = text

        # make the postings, two for fees
        postings = [
            data.Posting(
                self.get_fees_account(row.currency), -amount_, None, None, None, None
            ),
            data.Posting(
                self.get_liquidity_account(row.currency),
                amount_,
                None,
                None,
                None,
                None,
            ),
        ]
        meta = data.new_metadata(__file__, 0, {"descr": text})
        return data.Transaction(
            meta,
            row.reportDate,
            flags.FLAG_OKAY,
            "IB",  # payee
            narration,
            data.EMPTY_SET,
            data.EMPTY_SET,
            postings,
        )

    def dividends_and_withholding_tax_from_row(self, idx, row):
        """Converts dividends, payment inlieu of dividends and withholding tax to a beancount transaction.
        Stores div type in metadata for the merge step to be able to match tax withdrawals to the correct div.
        """
        amount_ = amount.Amount(row.amount, row.currency)

        text = row.description
        # Find ISIN in description in parentheses
        isin = re.findall(r"\(([a-zA-Z]{2}[a-zA-Z0-9]{9}\d)\)", text)[0]
        pershare_match = re.search(r"(\d*[.]\d*)(\D*)(PER SHARE)", text, re.IGNORECASE)
        # payment in lieu of a dividend does not have a PER SHARE in description
        pershare = pershare_match.group(1) if pershare_match else ""

        meta = {"isin": isin, "per_share": pershare}
        if row.type == CashAction.WHTAX:
            account = self.get_wht_account(row.symbol)
            type_ = (
                CashAction.PAYMENTINLIEU
                if re.search("payment in lieu of dividend", text, re.IGNORECASE)
                else CashAction.DIVIDEND
            )
        else:
            account = self.get_div_income_account(row.currency, row.symbol)
            type_ = row.type
            meta["div"] = True
        meta["div_type"] = type_.value
        postings = [
            data.Posting(account, -amount_, None, None, None, None),
            data.Posting(
                self.get_liquidity_account(row.currency),
                amount_,
                None,
                None,
                None,
                None,
            ),
        ]
        meta = data.new_metadata(
            "dividend",
            0,
            meta,
        )

        return data.Transaction(
            meta,
            row.reportDate,
            flags.FLAG_OKAY,
            row.symbol,  # payee
            text,
            data.EMPTY_SET,
            data.EMPTY_SET,
            postings,
        )

    def Interest_from_row(self, idx, row):
        amount_ = amount.Amount(row.amount, row.currency)
        text = row.description
        month = re.findall(r"\w{3}-\d{4}", text)[0]

        # make the postings, two for interest payments
        # received and paid interests are booked on the same account
        postings = [
            data.Posting(
                self.get_interest_income_account(row.currency),
                -amount_,
                None,
                None,
                None,
                None,
            ),
            data.Posting(
                self.get_liquidity_account(row.currency),
                amount_,
                None,
                None,
                None,
                None,
            ),
        ]
        meta = data.new_metadata("Interest", 0)
        return data.Transaction(
            meta,
            row.reportDate,
            flags.FLAG_OKAY,
            "IB",  # payee
            " ".join(["Interest ", row.currency, month]),
            data.EMPTY_SET,
            data.EMPTY_SET,
            postings,
        )

    def deposit_from_row(self, idx, row):
        amount_ = amount.Amount(row.amount, row.currency)
        postings = [
            data.Posting(
                self.get_liquidity_account(row.currency),
                amount_,
                None,
                None,
                None,
                None,
            ),
        ]
        meta = data.new_metadata("deposit/withdrawal", 0)
        return data.Transaction(
            meta,
            row.reportDate,
            flags.FLAG_OKAY,
            "self",  # payee
            row.description,
            data.EMPTY_SET,
            data.EMPTY_SET,
            postings,
        )

    def Trades(self, tr):
        # forex transactions
        fx = [t for t in tr if is_forex(t.symbol)]
        # Stocks transactions
        stocks = [t for t in tr if not is_forex(t.symbol)]

        return self.forex(fx) + self.stock_trades(stocks)

    def forex(self, fx):
        transactions = []
        for idx, row in enumerate(fx):
            symbol = row.symbol
            curr_prim, curr_sec = get_forex_currencies(symbol)
            currency_IBcommision = row.ibCommissionCurrency
            proceeds = amount.Amount(row.proceeds, curr_sec)
            quantity = amount.Amount(row.quantity, curr_prim)
            price = amount.Amount(row.tradePrice, curr_sec)
            commission = amount.Amount(row.ibCommission, currency_IBcommision)
            buysell = row.buySell.name

            postings = [
                data.Posting(
                    self.get_liquidity_account(curr_prim),
                    quantity,
                    None,
                    price,
                    None,
                    None,
                ),
                data.Posting(
                    self.get_liquidity_account(curr_sec),
                    proceeds,
                    None,
                    None,
                    None,
                    None,
                ),
                data.Posting(
                    self.get_liquidity_account(currency_IBcommision),
                    commission,
                    None,
                    None,
                    None,
                    None,
                ),
                data.Posting(
                    self.get_fees_account(currency_IBcommision),
                    minus(commission),
                    None,
                    None,
                    None,
                    None,
                ),
            ]

            transactions.append(
                data.Transaction(
                    data.new_metadata("FX Transaction", idx),
                    row.tradeDate,
                    flags.FLAG_OKAY,
                    symbol,  # payee
                    " ".join([buysell, quantity.to_string(), "@", price.to_string()]),
                    data.EMPTY_SET,
                    data.EMPTY_SET,
                    postings,
                )
            )
        return transactions

    def stock_trades(self, trades):
        """Generates transactions for IB stock trades.
        Tries to keep track of available holdings to disambiguate sales when lots are not enough,
        e.g. when there were multiple buys of the same symbol on the specific date.
        Currently, it does not take into account comission when calculating cost for stocks,
        just the trade price. It keeps the "real" cost as "ib_cost" metadata field though, which might be utilized in the future.
        It is mostly because I find the raw unafected price nicer to see in my beancount file.
        It also creates the fee posting for comission with "C" flag to distinguish it from other postings.
        """
        transactions = []
        for row, lots in iter_trades_with_lots(trades):
            if row.buySell in (BuySell.SELL, BuySell.CANCELSELL):
                op = "SELL"
            elif row.buySell in (BuySell.BUY, BuySell.CANCELBUY):
                op = "BUY"
            else:
                raise RuntimeError(f"Unknown buySell value: {row.buySell}")
            currency = row.currency
            currency_IBcommision = row.ibCommissionCurrency
            symbol = row.symbol
            net_cash = amount.Amount(row.netCash, currency)
            commission = amount.Amount(row.ibCommission, currency_IBcommision)
            quantity = amount.Amount(row.quantity, get_currency_from_symbol(symbol))
            price = amount.Amount(row.tradePrice, currency)
            date = row.dateTime.date()

            if row.openCloseIndicator == OpenClose.OPEN:
                self.add_holding(row)
                cost = position.CostSpec(
                    number_per=price.number,
                    number_total=None,
                    currency=currency,
                    date=row.tradeDate,
                    label=None,
                    merge=False,
                )
                lotpostings = [
                    data.Posting(
                        self.get_asset_account(symbol),
                        quantity,
                        cost,
                        price,
                        None,
                        {"ib_cost": row.cost},
                    ),
                ]
            else:
                lotpostings = []
                for clo in lots:
                    try:
                        clo_price = self.get_and_reduce_holding(clo)
                    except ValueError as e:
                        warnings.warn(str(e))
                        clo_price = None
                    cost = position.CostSpec(
                        clo_price,
                        number_total=None,
                        currency=clo.currency,
                        date=clo.openDateTime.date(),
                        label=None,
                        merge=False,
                    )

                    lotpostings.append(
                        data.Posting(
                            self.get_asset_account(symbol),
                            amount.Amount(
                                -clo.quantity, get_currency_from_symbol(clo.symbol)
                            ),
                            cost,
                            price,
                            None,
                            {"ib_cost": clo.cost},
                        )
                    )

                lotpostings.append(
                    data.Posting(
                        self.get_pnl_account(symbol), None, None, None, None, None
                    )
                )
            postings = (
                [
                    data.Posting(
                        self.get_liquidity_account(currency),
                        net_cash,
                        None,
                        None,
                        None,
                        None,
                    )
                ]
                + lotpostings
                + [
                    data.Posting(
                        self.get_fees_account(currency_IBcommision),
                        minus(commission),
                        None,
                        None,
                        "C",
                        None,
                    )
                ]
            )

            transactions.append(
                data.Transaction(
                    data.new_metadata("trade", 0),
                    date,
                    flags.FLAG_OKAY,
                    symbol,  # payee
                    " ".join([op, quantity.to_string(), "@", price.to_string()]),
                    data.EMPTY_SET,
                    data.EMPTY_SET,
                    postings,
                )
            )

        return transactions

    def add_holding(self, row):
        holdings = self.holdings_map[
            (row.dateTime.date(), get_currency_from_symbol(row.symbol))
        ]
        for holding in holdings:
            if holding[2] == row.cost / row.quantity:
                holding[0] += row.quantity
                return
        holdings.append([row.quantity, row.tradePrice, row.cost / row.quantity])

    def get_and_reduce_holding(self, lot):
        holdings = self.holdings_map[
            (lot.openDateTime.date(), get_currency_from_symbol(lot.symbol))
        ]
        for i, holding in enumerate(holdings):
            quantity, price, real_price = holding
            if not (
                round(real_price, 4) == round(lot.cost / lot.quantity, 4)
                or (
                    quantity == lot.quantity
                    and round(real_price, 2) == round(lot.cost / lot.quantity, 2)
                )
            ):
                continue
            if (quantity < 0 and quantity > lot.quantity) or (
                quantity > 0 and quantity < lot.quantity
            ):
                raise ValueError(
                    f"not enough holdings of {lot.symbol} at {lot.openDateTime.date()}: have {quantity}, want {lot.quantity}"
                )
            if quantity == lot.quantity:
                holdings.pop(i)
            else:
                holding[0] -= lot.quantity
            return price
        raise ValueError(
            f"do not have {lot.symbol} bought at {lot.openDateTime.date()}: want {lot.quantity} at {lot.cost} ({lot.cost/lot.quantity} per unit). have {holdings}"
        )

    def Balances(self, cr):
        transactions = []
        for row in cr:
            if row.currency == "BASE_SUMMARY":
                continue  # this is a summary balance that is not needed for beancount
            amount_ = amount.Amount(row.endingCash, row.currency)

            transactions.append(
                data.Balance(
                    data.new_metadata("balance", 0),
                    row.toDate + timedelta(days=1),
                    self.get_liquidity_account(row.currency),
                    amount_,
                    None,
                    None,
                )
            )
        return transactions

    def merge_dividend_and_withholding(self, entries):
        """This merges together transactions for earned dividends with the witholding tax ones,
        as they can be on different lines in the cash transactions statement.
        """
        grouped = defaultdict(list)
        for e in entries:
            if not isinstance(e, data.Transaction):
                continue
            if "div_type" in e.meta and "isin" in e.meta:
                grouped[(e.date, e.payee, e.meta["div_type"])].append(e)
        for group in grouped.values():
            if len(group) < 2:
                continue
            # merge
            try:
                d = [e for e in group if "div" in e.meta][0]
            except IndexError:
                continue
            for e in group:
                if e != d:
                    d.postings.extend(e.postings)
                    entries.remove(e)
            del d.meta["div_type"]
            del d.meta["div"]
            # merge postings with the same account
            grouped_postings = defaultdict(list)
            for p in d.postings:
                grouped_postings[p.account].append(p)
            d.postings.clear()
            for account, postings in grouped_postings.items():
                d.postings.append(
                    data.Posting(
                        account,
                        functools.reduce(amount_add, (p.units for p in postings)),
                        None,
                        None,
                        None,
                        None,
                    )
                )
        return entries

    def corporate_actions(self, actions):
        transactions = []
        actions_map = defaultdict(list)
        for row in actions:
            actions_map[row.actionID].append(row)
        for action_group in actions_map.values():
            row = action_group[0]
            if row.type == Reorg.FORWARDSPLIT:
                assert len(action_group) == 1
                transactions.append(self.process_stock_forwardsplit(row))
            elif row.type == Reorg.MERGER:
                assert len(action_group) == 2
                transactions.append(self.process_stock_merger(action_group))
            elif row.type == Reorg.ISSUECHANGE:
                assert len(action_group) == 2
                transactions.append(self.process_issue_change(action_group))
            else:
                raise RuntimeError(f"unknown corporate action type: {row.type}")
        return transactions

    def process_stock_forwardsplit(self, row):
        symbol = get_currency_from_symbol(row.symbol)
        m = re.search(r"SPLIT (\d+) FOR (\d+)", row.description)
        factor = Decimal(int(m.group(1)) / int(m.group(2)))
        holdings = [(k[0], v) for k, v in self.holdings_map.items() if k[1] == symbol]
        postings = []
        for date, lst in holdings:
            for quantity, price, real_price in lst:
                postings.append(
                    data.Posting(
                        self.get_asset_account(row.symbol),
                        amount.Amount(-quantity, symbol),
                        data.CostSpec(price, None, row.currency, date, None, False),
                        None,
                        None,
                        None,
                    )
                )
                postings.append(
                    data.Posting(
                        self.get_asset_account(row.symbol),
                        amount.Amount(quantity * factor, symbol),
                        data.CostSpec(
                            price / factor, None, row.currency, date, None, False
                        ),
                        None,
                        None,
                        {"ib_cost": round(real_price / factor * quantity, 6)},
                    )
                )
        for date, lst in holdings:
            for i in lst:
                i[0] *= Decimal(factor)
                i[1] /= Decimal(factor)
                i[2] /= Decimal(factor)
        return data.Transaction(
            data.new_metadata("corporateactions", 0),
            row.reportDate,
            flags.FLAG_OKAY,
            row.symbol,
            row.description,
            data.EMPTY_SET,
            data.EMPTY_SET,
            postings,
        )

    def process_stock_merger(self, action_group):
        # This is almost certainly wrong for tax accounting
        row = action_group[0]
        symbol = get_currency_from_symbol(row.symbol)
        holdings = [(k[0], v) for k, v in self.holdings_map.items() if k[1] == symbol]
        postings = []
        for date, lst in holdings:
            for quantity, price, _real_price in lst:
                postings.append(
                    data.Posting(
                        self.get_asset_account(row.symbol),
                        amount.Amount(-quantity, symbol),
                        data.CostSpec(price, None, row.currency, date, None, False),
                        None,
                        None,
                        None,
                    )
                )
        for k in list(self.holdings_map.keys()):
            if k[1] == symbol:
                del self.holdings_map[k]
        postings.append(
            data.Posting(
                self.get_liquidity_account(row.currency),
                amount.Amount(row.proceeds, row.currency),
                None,
                None,
                None,
                None,
            )
        )
        postings.append(
            data.Posting(self.get_pnl_account(symbol), None, None, None, None, None)
        )
        row = action_group[1]
        symbol = get_currency_from_symbol(row.symbol)
        postings.append(
            data.Posting(
                self.get_asset_account(row.symbol),
                amount.Amount(row.quantity, get_currency_from_symbol(row.symbol)),
                data.CostSpec(
                    row.value / row.quantity,
                    None,
                    row.currency,
                    row.reportDate,
                    None,
                    None,
                ),
                None,
                None,
                None,
            )
        )
        self.holdings_map[(row.reportDate, symbol)].append(
            (row.quantity, row.value / row.quantity, row.value / row.quantity)
        )
        return data.Transaction(
            data.new_metadata("corporateactions", 0),
            row.reportDate,
            flags.FLAG_OKAY,
            row.symbol,
            row.description,
            data.EMPTY_SET,
            data.EMPTY_SET,
            postings,
        )

    def process_issue_change(self, action_group):
        row = action_group[0]
        if row.symbol.endswith(".OLD"):
            row = action_group[1]
        old_symbol = re.search(r"(.*?)\(", row.description).group(1)
        holdings = [
            (k[0], v) for k, v in self.holdings_map.items() if k[1] == old_symbol
        ]
        postings = []
        for date, lst in holdings:
            for quantity, price, real_price in lst:
                postings.append(
                    data.Posting(
                        self.get_asset_account(old_symbol),
                        amount.Amount(-quantity, old_symbol),
                        data.CostSpec(price, None, row.currency, date, None, False),
                        None,
                        None,
                        None,
                    )
                )
                postings.append(
                    data.Posting(
                        self.get_asset_account(row.symbol),
                        amount.Amount(quantity, get_currency_from_symbol(row.symbol)),
                        data.CostSpec(price, None, row.currency, date, None, False),
                        None,
                        None,
                        {"ib_cost": quantity * real_price},
                    )
                )
            del self.holdings_map[(date, old_symbol)]
            self.holdings_map[(date, row.symbol)] = lst
        return data.Transaction(
            data.new_metadata("corporateactions", 0),
            row.reportDate,
            flags.FLAG_OKAY,
            row.symbol,
            row.description,
            data.EMPTY_SET,
            data.EMPTY_SET,
            postings,
        )

    def autoopen_accounts(self, entries, existing_entries):
        """Adds open directives for unseen accounts.
        Mostly useful if one prefers to have each stock on a separate beancount subaccount.
        """
        opened_accounts = set()
        if existing_entries is None:
            return []
        for e in existing_entries:
            if not isinstance(e, data.Transaction):
                continue
            for p in e.postings:
                opened_accounts.add(p.account)
        open_entries = []
        for e in entries:
            if not isinstance(e, data.Transaction):
                continue
            for p in e.postings:
                if p.account not in opened_accounts:
                    min_entry = min(
                        (
                            e
                            for e in entries
                            if isinstance(e, data.Transaction)
                            and any(i.account == p.account for i in e.postings)
                        ),
                        key=lambda e: e.date,
                    )
                    if p.units is not None:
                        currency = get_currency_from_symbol(p.units.currency)
                    else:
                        currency = "USD"
                    open_entries.append(
                        data.Open(
                            meta=data.new_metadata("open", 0),
                            date=min_entry.date,
                            account=p.account,
                            currencies=[currency],
                            booking=None,
                        )
                    )
                    opened_accounts.add(p.account)
        return open_entries

    def get_holdings_map(self, entries):
        root = realization.realize(entries)
        account_parts = self.assets_account.split(":")
        for part in account_parts:
            if "{" in part:
                break
            if part not in root:
                return defaultdict(list)
            root = root[part]
        result = defaultdict(list)
        for account in realization.iter_children(root, leaf_only=True):
            for pos in account.balance:
                if pos.cost is None:
                    continue
                for tx in account.txn_postings:
                    real_price = None
                    if not isinstance(tx, data.TxnPosting):
                        continue
                    if (
                        tx.posting.units.currency == pos.units.currency
                        and tx.posting.cost.date == pos.cost.date
                        and tx.posting.cost.number == pos.cost.number
                        and "ib_cost" in tx.posting.meta
                    ):
                        real_price = abs(
                            tx.posting.meta["ib_cost"] / tx.posting.units.number
                        )
                    if real_price is None:
                        continue
                    self._adjust_holding(
                        result,
                        pos.cost.date,
                        pos.units.currency,
                        tx.posting.units.number,
                        tx.posting.cost.number,
                        real_price,
                    )
        return result

    def _adjust_holding(self, holdings_map, date, symbol, quantity, price, real_price):
        lst = holdings_map[(date, symbol)]
        for i, (u, _, rp) in enumerate(lst):
            if round(rp, 4) == round(real_price, 4) or (
                u == quantity and round(rp, 2) == round(real_price, 2)
            ):
                lst[i][0] += quantity
                if lst[i][0] == 0:
                    lst.pop(i)
                return
        holdings_map[(date, symbol)].append([quantity, price, real_price])


def is_forex(symbol):
    # returns True if a transaction is a forex transaction.
    b = re.search(r"(\w{3})[.](\w{3})", symbol)  # find something lile "USD.CHF"
    if b is None:  # no forex transaction, rather a normal stock transaction
        return False
    else:
        return True


def get_forex_currencies(symbol):
    b = re.search(r"(\w{3})[.](\w{3})", symbol)
    c = b.groups()
    return [c[0], c[1]]


def amount_add(A1, A2):
    # add two amounts
    if A1.currency == A2.currency:
        quant = A1.number + A2.number
        return amount.Amount(quant, A1.currency)
    else:
        raise (
            "Cannot add amounts of differnent currencies: {} and {}".format(
                A1.currency, A2.currency
            )
        )


def minus(A):
    # a minus operator
    return amount.Amount(-A.number, A.currency)


def get_currency_from_symbol(symbol):
    symbol = symbol.replace(" ", ".")
    if len(symbol) < 2:
        symbol = symbol + "STOCK"
    return symbol


def iter_trades_with_lots(trades):
    """Yields pairs of (trade, lots)."""
    it = iter(trades)
    trade = None
    lots = []
    while True:
        try:
            t = next(it)
        except StopIteration:
            break
        if isinstance(t, Types.Trade):
            if trade is not None:
                yield trade, lots
                lots = []
            trade = t
        elif isinstance(t, Types.Lot):
            lots.append(t)
        else:
            raise ValueError(f"Unknown trade element: {t}")
    if trade is not None:
        yield trade, lots


def get_test_importer():
    return Importer()


if __name__ == "__main__":
    from beangulp.testing import main

    main(get_test_importer())

src/uabean/importers/ibkr.py
OSadovy/uabean
"""Imports transactions from Kraken crypto exchange ledger csv file.

The CSV header is the following:
"txid","refid","time","type","subtype","aclass","asset","amount","fee","balance"
"""


import csv

import beangulp
import dateutil.parser
from beancount.core import data, flags
from beancount.core.amount import Amount
from beancount.core.number import D

from uabean.importers.mixins import IdentifyMixin


class Importer(IdentifyMixin, beangulp.Importer):
    FLAG = flags.FLAG_OKAY
    matchers = [
        ("content", __doc__.split("\n")[-2]),
        ("mime", "text/csv"),
    ]

    def __init__(
        self,
        spot_account="Assets:Kraken:Spot",
        staking_account="Assets:Kraken:Staking",
        fee_account="Expenses:Fees:Kraken",
        staking_income_account="Income:Staking:Kraken",
    ):
        self.spot_account = spot_account
        self.staking_account = staking_account
        self.fee_account = fee_account
        self.staking_income_account = staking_income_account
        super().__init__()

    def account(self, _):
        return "kraken"

    def parse_date(self, s):
        return dateutil.parser.parse(s)

    def extract(self, filename, existing_entries=None):
        entries = []
        balances = {}
        for index, row in enumerate(csv.DictReader(open(filename)), 1):
            meta = data.new_metadata(filename, index)
            entry = self.get_entry_from_row(row, meta, balances)
            if entry is not None:
                entries.append(entry)
        for (account, currency), (date, balance) in balances.items():
            entries.append(
                data.Balance(
                    data.new_metadata(filename, -1),
                    date.date() + dateutil.relativedelta.relativedelta(days=1),
                    account,
                    Amount(balance, currency),
                    None,
                    None,
                )
            )
        return entries

    def get_entry_from_row(self, row, meta, balances):
        if not row["txid"]:
            return None
        date = self.parse_date(row["time"])
        meta["time"] = date.strftime("%H:%M:%S")
        postings = []
        match (row["type"], row["subtype"]):
            case ("deposit", "") | ("transfer", "spottostaking"):
                postings.append(
                    data.Posting(
                        self.spot_account,
                        self.ammount_from_row(row),
                        None,
                        None,
                        None,
                        None,
                    )
                )
                self.update_balance(
                    balances,
                    date,
                    self.spot_account,
                    row["asset"],
                    D(row["balance"]),
                )
            case ("transfer", "stakingfromspot"):
                postings.append(
                    data.Posting(
                        self.staking_account,
                        self.ammount_from_row(row),
                        None,
                        None,
                        None,
                        None,
                    )
                )
                self.update_balance(
                    balances,
                    date,
                    self.staking_account,
                    row["asset"],
                    D(row["balance"]),
                )
            case ("staking", ""):
                postings.append(
                    data.Posting(
                        self.staking_account,
                        self.ammount_from_row(row),
                        None,
                        None,
                        None,
                        None,
                    )
                )
                postings.append(
                    data.Posting(
                        self.staking_income_account,
                        None,
                        None,
                        None,
                        None,
                        None,
                    )
                )
                self.update_balance(
                    balances,
                    date,
                    self.staking_account,
                    row["asset"],
                    D(row["balance"]),
                )
            case _:
                raise NotImplementedError(
                    f"Unknown transaction type {row['type']} row['subtype']"
                )
        if float(row["fee"]):
            postings.append(
                data.Posting(
                    self.fee_account,
                    self.ammount_from_row(row),
                    None,
                    None,
                    None,
                    None,
                )
            )
        return data.Transaction(
            meta,
            date.date(),
            self.FLAG,
            "",
            "",
            data.EMPTY_SET,
            data.EMPTY_SET,
            postings,
        )

    def ammount_from_row(self, row):
        return Amount(D(row["amount"]), row["asset"])

    def update_balance(self, balances, date, account, currency, balance):
        if (account, currency) not in balances:
            balances[(account, currency)] = (date, balance)
        else:
            balances[(account, currency)] = max(
                balances[(account, currency)], (date, balance)
            )


def get_test_importer():
    return Importer()


if __name__ == "__main__":
    from beangulp.testing import main

    main(get_test_importer())

src/uabean/importers/kraken.py
OSadovy/uabean
import collections
import itertools
import re

from beangulp import cache
from beangulp.importers.mixins.identifier import _PARTS, identify


class IdentifyMixin:
    """Converted for use with new beangulp Importer class."""

    def __init__(self, **kwds):
        """Pull 'matchers' and 'converter' from kwds."""

        self.remap = collections.defaultdict(list)
        matchers = kwds.pop("matchers", [])
        cls_matchers = getattr(self, "matchers", [])
        assert isinstance(matchers, list)
        assert isinstance(cls_matchers, list)
        for part, regexp in itertools.chain(matchers, cls_matchers):
            assert part in _PARTS, repr(part)
            assert isinstance(regexp, str), repr(regexp)
            self.remap[part].append(re.compile(regexp))

        # Converter is a fn(filename: Text) -> contents: Text.
        self.converter = kwds.pop("converter", getattr(self, "converter", None))

        super().__init__(**kwds)

    def identify(self, filepath):
        return identify(self.remap, self.converter, cache.get_file(filepath))

src/uabean/importers/mixins.py
OSadovy/uabean
"""Imports monobank csv statements.

CSV header is as follows:
"Дата i час операції","Деталі операції",MCC,"Сума в валюті картки (UAH)","Сума в валюті операції",Валюта,Курс,"Сума комісій (UAH)","Сума кешбеку (UAH)","Залишок після операції"
"""

import csv
import datetime
import re

import beangulp
import dateutil.parser
import requests
from beancount.core import data, flags
from beancount.core.number import D

from uabean.importers.mixins import IdentifyMixin

mcc_codes_url = "https://raw.githubusercontent.com/Oleksios/Merchant-Category-Codes/main/Without%20groups/mcc-en.json"


class Importer(IdentifyMixin, beangulp.Importer):
    FLAG = flags.FLAG_OKAY
    DATE_COL = 0
    DESCRIPTION_COL = 1
    MCC_COL = 2
    AMOUNT_COL = 3
    ORIG_AMOUNT_COL = 4
    CURRENCY_COL = 5
    EXCHANGE_RATE_COL = 6
    COMISSION_RATE_COL = 7
    CASHBACK_COL = 8
    RUNNING_BALANCE_COL = 9
    NA = "—"
    CASHBACK_RE = r"^Виведення кешбеку ([\d\.]+)"
    INTEREST_RE = r"^Відсотки за .*"

    def __init__(
        self,
        account_config: dict[tuple[str, str], str],
        cashback_income_account="Income:Cashback:Monobank",
        cashback_receivable_account="Assets:Monobank:Receivable:Cashback",
        taxes_expense_account="Expenses:Taxes",
        interest_income_account="Income:Monobank:Interest",
    ):
        self.account_config = account_config
        self.cashback_income_account = cashback_income_account
        self.cashback_receivable_account = cashback_receivable_account
        self.taxes_expense_account = taxes_expense_account
        self.interest_income_account = interest_income_account
        self.matchers = [
            ("mime", "text/csv"),
            (
                "filename",
                f"monobank-({'|'.join('%s-%s' % (type, currency) for type, currency in account_config)})",
            ),
        ]
        super().__init__()

    def _download_mcc_codes(self):
        codes = requests.get(mcc_codes_url).json()
        self.mcc_codes = {c["mcc"]: c["shortDescription"] for c in codes}

    def get_csv_reader(self, filename):
        return csv.reader(open(filename))

    def file_account(self, filename):
        # example: monobank-black-UAH_22-10-22_14-24-57.csv
        parts = filename.split("_")[0].split("-")
        return self.account_config[(parts[1], parts[2])]

    def account(self, _):
        return "monobank"

    def extract(self, filename, existing_entries=None):
        self._download_mcc_codes()
        entries = []
        reader = self.get_csv_reader(filename)
        header = next(reader)
        account = self.file_account(filename)
        account_currency = re.search(r"\((\w+)\)", header[self.AMOUNT_COL]).group(1)
        cashback_currency = re.search(r"\((\w+)\)", header[self.CASHBACK_COL]).group(1)
        for i, row in enumerate(reader, 2):
            meta = data.new_metadata(filename, i)
            entries.append(
                self.entry_from_row(
                    meta, account, account_currency, cashback_currency, row
                )
            )
        if entries:
            meta = data.new_metadata(filename, i)
            entries.append(
                data.Balance(
                    meta,
                    entries[-1].date + datetime.timedelta(days=1),
                    account,
                    data.Amount(D(row[self.RUNNING_BALANCE_COL]), account_currency),
                    None,
                    None,
                )
            )
        return entries

    def entry_from_row(self, meta, account, account_currency, cashback_currency, row):
        meta["category"] = self.mcc_codes[row[self.MCC_COL]]
        postings = []
        dt = self.date_from_str(row[self.DATE_COL])
        meta["time"] = dt.time().strftime("%H:%M:%S")
        price = None
        if row[self.CURRENCY_COL] != account_currency:
            price = data.Amount(
                round(
                    D(row[self.ORIG_AMOUNT_COL]) / D(row[self.AMOUNT_COL]), 6
                ).normalize(),
                row[self.CURRENCY_COL],
            )
            meta["converted"] = row[self.ORIG_AMOUNT_COL] + " " + row[self.CURRENCY_COL]
        description = row[self.DESCRIPTION_COL]
        payee = description
        narration = None
        if m := re.search(self.CASHBACK_RE, description):
            u = data.Amount(-D(m.group(1)), cashback_currency)
            postings += [
                data.Posting(
                    self.cashback_receivable_account, u, None, None, None, None
                ),
                data.Posting(self.taxes_expense_account, None, None, None, None, None),
            ]
            narration, payee = payee, None
            del meta["category"]
        if m := re.search(self.INTEREST_RE, description):
            u = data.Amount(-D(row[self.AMOUNT_COL]), account_currency)
            postings.append(
                data.Posting(self.interest_income_account, u, None, None, None, None)
            )
            narration, payee = payee, None
            del meta["category"]
        postings.append(
            data.Posting(
                account,
                data.Amount(D(row[self.AMOUNT_COL]), account_currency),
                None,
                price,
                None,
                None,
            )
        )
        if row[self.CASHBACK_COL] != self.NA:
            postings += [
                data.Posting(
                    self.cashback_income_account,
                    data.Amount(-D(row[self.CASHBACK_COL]), cashback_currency),
                    None,
                    None,
                    None,
                    None,
                ),
                data.Posting(
                    self.cashback_receivable_account,
                    data.Amount(D(row[self.CASHBACK_COL]), cashback_currency),
                    None,
                    None,
                    None,
                    None,
                ),
            ]
        return data.Transaction(
            meta,
            dt.date(),
            self.FLAG,
            payee,
            narration,
            data.EMPTY_SET,
            data.EMPTY_SET,
            postings,
        )

    def date_from_str(self, s):
        return dateutil.parser.parse(s, dayfirst=True)


def get_test_importer():
    return Importer(
        {
            ("black", "UAH"): "Assets:Monobank:Black:UAH",
            ("black", "USD"): "Assets:Monobank:Black:USD",
            ("white", "UAH"): "Assets:Monobank:White:UAH",
        }
    )


if __name__ == "__main__":
    from beangulp.testing import main

    main(get_test_importer())

src/uabean/importers/monobank.py
OSadovy/uabean
"""Imports transactions csv files exported from Nexo crypto platform.

The CSV header is the following:
Transaction,Type,Input Currency,Input Amount,Output Currency,Output Amount,USD Equivalent,Details,(?:Outstanding Loan,)?Date / Time
"""

import csv

import beangulp
from beancount.core import data, flags
from beancount.core.amount import Amount
from beancount.core.number import D
from beancount.utils.date_utils import parse_date_liberally

from uabean.importers.mixins import IdentifyMixin


class Importer(IdentifyMixin, beangulp.Importer):
    matchers = [
        ("content", __doc__.split("\n")[-2]),
        ("mime", "text/csv"),
    ]

    def __init__(self, asset_account, interest_account, *args, **kwargs):
        self.asset_account = asset_account
        self.interest_account = interest_account
        super().__init__(*args, **kwargs)

    def get_csv_reader(self, filename):
        return csv.DictReader(open(filename))

    def get_date_from_row(self, row):
        return parse_date_liberally(row["Date / Time"])

    def account(self, _):
        return "nexo"

    def extract(self, filename, existing_entries=None):
        entries = []
        for index, row in enumerate(self.get_csv_reader(filename), 1):
            meta = data.new_metadata(filename, index)
            entry = self.get_entry_from_row(row, meta)
            if entry is not None:
                entries.append(entry)
        return entries

    def get_entry_from_row(self, row, meta):
        meta["src_tx_id"] = row["Transaction"]
        meta["src_ts"] = row["Date / Time"]
        txn = data.Transaction(
            meta,
            self.get_date_from_row(row),
            flags.FLAG_OKAY,
            "",
            "",
            data.EMPTY_SET,
            data.EMPTY_SET,
            [],
        )
        if row["Type"] == "Interest":
            txn.postings.append(
                data.Posting(
                    self.asset_account,
                    Amount(D(row["Input Amount"]), row["Input Currency"]),
                    None,
                    None,
                    None,
                    None,
                )
            )
            txn.postings.append(
                data.Posting(
                    self.interest_account,
                    Amount(-D(row["Output Amount"]), row["Output Currency"]),
                    None,
                    None,
                    None,
                    None,
                )
            )
        elif row["Type"] == "Deposit":
            txn.postings.append(
                data.Posting(
                    self.asset_account,
                    Amount(D(row["Input Amount"]), row["Input Currency"]),
                    None,
                    None,
                    None,
                    None,
                )
            )
            meta["src_details"] = row["Details"]
        elif row["Type"] == "Withdrawal":
            txn.postings.append(
                data.Posting(
                    self.asset_account,
                    Amount(D(row["Input Amount"]), row["Input Currency"]),
                    None,
                    None,
                    None,
                    None,
                )
            )
            meta["src_details"] = row["Details"]
        elif row["Type"] == "Exchange":
            txn.postings.append(
                data.Posting(
                    self.asset_account,
                    Amount(D(row["Input Amount"]), row["Input Currency"]),
                    None,
                    None,
                    None,
                    None,
                )
            )
            txn.postings.append(
                data.Posting(
                    self.asset_account,
                    Amount(D(row["Output Amount"]), row["Output Currency"]),
                    None,
                    None,
                    None,
                    None,
                )
            )
            meta["src_details"] = row["Details"]
        else:
            raise RuntimeError("Unknown row: %s" % row)
        return txn


def get_test_importer():
    return Importer("Assets:Nexo", "Income:Nexo:Interest")


if __name__ == "__main__":
    from beangulp.testing import main

    main(get_test_importer())

src/uabean/importers/nexo.py
OSadovy/uabean
"""
Imports transactions from Oschadbank online API.

This module supports two types of authorizations:
1. Using OSCHAD_JSESSIONID_COOKIE environment variable:
   The value can be obtained by logging in to Oschadbank online and copying the
   cookie value from the browser.

2. Using manual login credentials:
   If the OSCHAD_JSESSIONID_COOKIE is not found, the module will prompt the user
   to input their username and password. If the session requires OTP
   (One-Time Password), the user will be asked to provide it.
   Environment variables OSCHAD_LOGIN and OSCHAD_PASSWORD can also be used to
   provide the username and password, respectively.
"""


import getpass

# import json
import os
import re
import sys
from datetime import date, timedelta

import beangulp
import dateutil.parser
import requests
from beancount.core import data, flags
from beancount.core.number import D

from uabean.importers.mixins import IdentifyMixin

API_BASE_URL = "https://online.oschadbank.ua/wb/api/v2"


class Importer(IdentifyMixin, beangulp.Importer):
    FLAG = flags.FLAG_OKAY
    matchers = [
        ("filename", "oschadbank"),
    ]
    DESCRIPTION_CLASSIFIERS = [
        (r"реестров УПСЗН", "Income:SocialInsurance"),
        (r"Видача готівки", "Expenses:Mum"),
    ]

    def __init__(
        self,
        account_config: dict[str, str],
        fees_account="Expenses:Fees:Oschadbank",
        min_date=date(2019, 1, 1),
    ):
        self.account_config = account_config
        self.fees_account = fees_account
        self.min_date = min_date
        self._contracts = {}
        self._accounts = {}
        self._contract_id_to_number = {}
        super().__init__()
        self._session = requests.Session()

    def _login(self):
        if os.getenv("OSCHAD_JSESSIONID_COOKIE"):
            self._session.cookies.set(
                "JSESSIONID", os.getenv("OSCHAD_JSESSIONID_COOKIE")
            )
            return
        response = self._session.get("https://online.oschadbank.ua/wb/")
        response.raise_for_status()

        login = os.getenv("OSCHAD_LOGIN") or input("Enter your username: ")
        password = os.getenv("OSCHAD_PASSWORD") or getpass.getpass(
            "Enter your password: "
        )
        initial_data = {"login": login, "password": password, "captcha": ""}

        headers = {
            "authority": "online.oschadbank.ua",
            "accept": "application/json, text/javascript, */*; q=0.01",
            "accept-language": "ua",
            "content-type": "application/json",
            "origin": "https://online.oschadbank.ua",
            "ow-client-browser": "Chrome",
            "referer": "https://online.oschadbank.ua/wb/",
            "sec-ch-ua": '"Google Chrome";v="113", "Chromium";v="113", "Not-A.Brand";v="24"',
            "sec-ch-ua-mobile": "?0",
            "sec-ch-ua-platform": '"Windows"',
            "sec-fetch-dest": "empty",
            "sec-fetch-mode": "cors",
            "sec-fetch-site": "same-origin",
            "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Safari/537.36",
            "x-requested-with": "XMLHttpRequest",
            "x-xsrf-token": response.cookies["XSRF-TOKEN"],
        }
        response = self._session.post(
            f"{API_BASE_URL}/session",
            json=initial_data,
            headers=headers,
        )
        if response.status_code == 401:
            response_data = response.json()

            otp = input("Enter your OTP: ")
            otp_data = {
                "login": login,
                "password": password,
                "captcha": "",
                "_status": None,
                "confirmation": response_data["confirmation"],
            }
            otp_data["confirmation"]["response"] = otp

            response = self._session.post(
                f"{API_BASE_URL}/session",
                json=otp_data,
                headers=headers,
            )
            response.raise_for_status()

        if "JSESSIONID" in response.cookies:
            self._session.cookies.set("JSESSIONID", response.cookies["JSESSIONID"])
            print(
                f"export OSCHAD_JSESSIONID_COOKIE=f{response.cookies['JSESSIONID']}",
                file=sys.stderr,
            )
        else:
            raise Exception(f"Failed to login: {response.text}")

    def _get_contracts(self):
        # return json.load(open("contracts.json"))
        response = self._session.get(f"{API_BASE_URL}/contracts")
        response.raise_for_status()
        return response.json()

    def _get_history(self, date_from, date_to=None):
        # return json.load(open("history.json"))
        params = {
            "from": date_from.strftime("%Y-%m-%d"),
        }
        if date_to is not None:
            params["to"] = date_to.strftime("%Y-%m-%d")
        response = self._session.get(
            f"{API_BASE_URL}/history",
            params=params,
        )
        response.raise_for_status()
        return response.json()

    def account(self, _):
        return "oschadbank"

    def extract(self, filename, existing_entries=None):
        self._login()
        self._contracts = self._get_contracts()
        contract_number_to_id = {c["number"]: c["id"] for c in self._contracts}
        self._contract_id_to_number = {c["id"]: c["number"] for c in self._contracts}
        self._accounts = {
            contract_number_to_id[k]: v for k, v in self.account_config.items()
        }
        entries = []
        max_date = self.min_date

        # Existing entries are considered for determining the max_date
        if existing_entries is not None:
            for entry in existing_entries:
                if isinstance(entry, data.Transaction) and entry.date > max_date:
                    for posting in entry.postings:
                        if posting.account in self._accounts.values():
                            max_date = entry.date

        # Get history from max_date onwards
        records = self._get_history(max_date)
        for i, record in enumerate(records[::-1]):
            meta = data.new_metadata("API", i)
            transaction = self.entry_from_record(meta, record)
            if transaction is None:
                continue
            self.classify_posting(transaction)
            entries.append(transaction)
        max_date = (entries[-1].date + timedelta(days=1)) if entries else date.today()
        meta = data.new_metadata("API", 0)
        processed_accounts = set()
        for contract_number, account in self.account_config.items():
            if (
                contract_number not in contract_number_to_id
                or account in processed_accounts
            ):
                continue
            contract = [
                c
                for c in self._contracts
                if c["id"] == contract_number_to_id[contract_number]
            ][0]
            entries.append(
                data.Balance(
                    meta,
                    max_date,
                    account,
                    self.amount_from_record(contract["balances"]["available"]),
                    None,
                    None,
                )
            )
            processed_accounts.add(account)
        return entries

    def entry_from_record(self, meta, record):
        if record.get("status") == "failed":
            return
        try:
            account = self._accounts[record["contractId"]]
        except KeyError:
            contract_number = self._contract_id_to_number[record["contractId"]]
            raise ValueError(
                f"Encountered transaction with unknown contract: {contract_number}"
            )
        dt = dateutil.parser.parse(record["operationTime"])
        # convert dt into local timezone
        dt = dt.astimezone()
        meta["time"] = dt.time().strftime("%H:%M:%S")
        narration = record["description"].replace("\\$", ";")
        postings = [
            data.Posting(
                account,
                self.amount_from_record(record["totalAmount"]),
                None,
                None,
                None,
                None,
            )
        ]
        if record["fees"]:
            postings.append(
                data.Posting(
                    self.fees_account,
                    -self.amount_from_record(record["fees"]["totalFee"]),
                    None,
                    None,
                    None,
                    None,
                )
            )
        return data.Transaction(
            meta,
            dt.date(),
            self.FLAG,
            None,
            narration,
            data.EMPTY_SET,
            data.EMPTY_SET,
            postings,
        )

    def amount_from_record(self, r):
        return data.Amount(D(r["value"]), r["currency"])

    def classify_posting(self, transaction):
        for pattern, account in self.DESCRIPTION_CLASSIFIERS:
            if re.search(pattern, transaction.narration, re.I):
                transaction.postings.append(
                    data.Posting(account, None, None, None, None, None)
                )


def get_test_importer():
    return Importer(
        {
            "1234567890": "Assets:Oschadbank:Cash:UAH",
        }
    )


if __name__ == "__main__":
    from beangulp.testing import main

    main(get_test_importer())

src/uabean/importers/oschad_api.py
OSadovy/uabean
"""Imports xls statements from privatbank, optained via privat24 web interface.

The header is as follows:
Дата;Час;Категорія;Картка;Опис операції;Сума в валюті картки;Валюта картки;Сума в валюті транзакції;Валюта транзакції;Залишок на кінець періоду;Валюта залишку
"""

import datetime

import beangulp
import dateutil.parser
import xlrd
from beancount.core import data, flags
from beancount.core.number import D

from uabean.importers.mixins import IdentifyMixin


class Importer(IdentifyMixin, beangulp.Importer):
    FLAG = flags.FLAG_OKAY
    matchers = [
        ("filename", "privat"),
        ("mime", "application/vnd.ms-excel"),
    ]
    unknown_account = "Assets:Unknown"
    DATE_COL = 0
    TIME_COL = 1
    CATEGORY_COL = 2
    CARD_COL = 3
    DESCRIPTION_COL = 4
    CARD_CURRENCY_AMOUNT_COL = 5
    CARD_CURRENCY_COL = 6
    TRANSACTION_AMOUNT_COL = 7
    TRANSACTION_CURRENCY_COL = 8
    BALANCE_COL = 9
    BALANCE_CURRENCY_COL = 10
    CURRENCY_MAP = {"грн": "UAH", "дол": "USD", "євро": "EUR", "PLN": "PLN"}

    def __init__(
        self,
        card_to_account_map,
        *args,
        fee_account="Expenses:Fees:Privatbank",
        **kwargs,
    ):
        self.card_to_account_map = card_to_account_map
        self.fee_account = fee_account
        super().__init__(*args, **kwargs)

    def date_from_row(self, row):
        return dateutil.parser.parse(row[self.DATE_COL].value, dayfirst=True)

    @classmethod
    def get_currency(cls, cell):
        return cls.CURRENCY_MAP[cell.value]

    @staticmethod
    def get_number(cell):
        return D(str(cell.value))

    def account(self, _):
        return "privatbank"

    def extract(self, filename, existing_entries=None):
        entries = []
        workbook = xlrd.open_workbook(filename)
        sheet = workbook.sheet_by_index(0)
        assert "Виписка з Ваших карток за період" in sheet.cell(0, 0).value
        max_date = None
        max_row = None
        for nrow in range(2, sheet.nrows):
            row = sheet.row(nrow)
            meta = data.new_metadata(filename, nrow)
            entries.append(self.entry_from_row(meta, row))
            date = self.date_from_row(row)
            if max_date is None or date > max_date:
                max_date = date
                max_row = row
        if max_row is not None:
            amount = data.Amount(
                self.get_number(max_row[self.BALANCE_COL]),
                self.get_currency(max_row[self.BALANCE_CURRENCY_COL]),
            )
            entries.append(
                data.Balance(
                    data.new_metadata(filename, 0),
                    max_date.date() + datetime.timedelta(days=1),
                    self.card_to_account_map[max_row[self.CARD_COL].value],
                    amount,
                    None,
                    None,
                )
            )
        return entries

    def entry_from_row(self, meta, row):
        dt = self.date_from_row(row)
        meta["time"] = row[self.TIME_COL].value
        meta["category"] = row[self.CATEGORY_COL].value
        account = self.card_to_account_map.get(
            row[self.CARD_COL].value, self.unknown_account
        )
        num = self.get_number(row[self.TRANSACTION_AMOUNT_COL])
        currency = self.get_currency(row[self.TRANSACTION_CURRENCY_COL])
        card_num = self.get_number(row[self.CARD_CURRENCY_AMOUNT_COL])
        card_currency = self.get_currency(row[self.CARD_CURRENCY_COL])
        postings = [
            data.Posting(
                account, data.Amount(card_num, card_currency), None, None, None, None
            )
        ]
        if currency != card_currency:
            meta["converted"] = f"{num} {currency}"
        elif abs(card_num) != num:
            fee_amount = data.Amount(abs(card_num) - num, currency)
            postings.append(
                data.Posting(self.fee_account, fee_amount, None, None, None, None)
            )
        return data.Transaction(
            meta,
            dt.date(),
            self.FLAG,
            None,
            row[self.DESCRIPTION_COL].value,
            data.EMPTY_SET,
            data.EMPTY_SET,
            postings,
        )


def get_test_importer():
    return Importer(
        {
            "1234": "Assets:Privatbank:Universal",
            "5678": "Assets:Privatbank:Social",
        }
    )


if __name__ == "__main__":
    from beangulp.testing import main

    main(get_test_importer())

src/uabean/importers/privatbank_xls.py
OSadovy/uabean
"""Imports xml statements from privatbank, optained via p24-cli.

See https://github.com/dimboknv/p24-cli

The xml format is identical to that of p24 merchant api.
"""

import datetime
from xml.etree import ElementTree as ET

import beangulp
import dateutil.parser
from beancount.core import data, flags
from beancount.core.number import D

from uabean.importers.mixins import IdentifyMixin


class Importer(IdentifyMixin, beangulp.Importer):
    FLAG = flags.FLAG_OKAY
    matchers = [
        ("content", "<statements status="),
        ("mime", "application/xml"),
    ]
    unknown_account = "Assets:Unknown"

    def __init__(
        self,
        card_to_account_map,
        *args,
        fee_account="Expenses:Fees:Privatbank",
        **kwargs
    ):
        self.card_to_account_map = card_to_account_map
        self.fee_account = fee_account
        super().__init__(*args, **kwargs)

    def date_from_elem(self, elem):
        return dateutil.parser.parse(" ".join(elem.attrib["trandate"].split(" ")[:2]))

    def account(self, _):
        return "privatbank"

    def extract(self, filename, existing_entries=None):
        entries = []
        tree = ET.parse(filename)
        root = tree.getroot()
        assert root.tag == "statements"
        max_date = None
        max_elem = None
        for elem in root:
            meta = data.new_metadata(filename, 0)
            entries.append(self.entry_from_elem(meta, elem))
            date = self.date_from_elem(elem)
            if max_date is None or date > max_date:
                max_date = date
                max_elem = elem
        if max_elem is not None:
            rest_num, rest_currency = max_elem.attrib["rest"].split(" ", 1)
            amount = data.Amount(D(rest_num), rest_currency)
            entries.append(
                data.Balance(
                    data.new_metadata(filename, 0),
                    max_date.date() + datetime.timedelta(days=1),
                    self.card_to_account_map[max_elem.attrib["card"]],
                    amount,
                    None,
                    None,
                )
            )
        return entries

    def entry_from_elem(self, meta, elem):
        dt = self.date_from_elem(elem)
        meta["time"] = dt.strftime("%H:%M:%S")
        account = self.card_to_account_map.get(
            elem.attrib["card"], self.unknown_account
        )
        num, currency = elem.attrib["amount"].split(" ", 1)
        num = D(num)
        card_num, card_currency = elem.attrib["cardamount"].split(" ", 1)
        card_num = D(card_num)
        postings = [
            data.Posting(
                account, data.Amount(card_num, card_currency), None, None, None, None
            )
        ]
        if currency != card_currency:
            meta["converted"] = elem.attrib["amount"]
        elif abs(card_num) != num:
            fee_amount = data.Amount(abs(card_num) - num, currency)
            postings.append(
                data.Posting(self.fee_account, fee_amount, None, None, None, None)
            )
        return data.Transaction(
            meta,
            dt.date(),
            self.FLAG,
            None,
            elem.attrib["description"],
            data.EMPTY_SET,
            data.EMPTY_SET,
            postings,
        )


def get_test_importer():
    return Importer(
        {
            "1234": "Assets:Privatbank:Universal",
            "5678": "Assets:Privatbank:Social",
        }
    )


if __name__ == "__main__":
    from beangulp.testing import main

    main(get_test_importer())

src/uabean/importers/privatbank_xml.py
OSadovy/uabean
"""Imports statement csv files exported from Procredit Business Online web app.

The CSV header is the following:
ЄДРПОУ;Код ID НБУ;Рахунок;Валюта;Дата операції;Код операції;МФО банка;Назва банка;Рахунок кореспондента;ЄДРПОУ кореспондента;Кореспондент;Номер документа;Дата документа;Дебет;Кредит;Призначення платежу;Гривневе покриття
"""

import csv
import re
from collections import defaultdict

import beangulp
from beancount.core import data, flags
from beancount.core.number import D
from beancount.utils.date_utils import parse_date_liberally

from uabean.importers.mixins import IdentifyMixin


class Importer(IdentifyMixin, beangulp.Importer):
    matchers = [
        ("content", __doc__.split("\n")[-2]),
        ("mime", "text/csv"),
    ]
    fee_regexes = ("Сплата комісії", "Комісія за переказ в національній валюті")
    currency_conversion_regex = r"Кошти від продажу валюти в сумі (?P<amount>[\d.,]+) (?P<currency>\w+) на МВРУ згідно заявки № \d*\.За курсом (?P<rate>[\d.,]+)\.Банк\. коміс\. грн\.- (?P<fee>[\d.,]+)\."
    DATE_FIELD = "Дата операції"

    def __init__(self, accounts, fee_account, *args, **kwargs):
        self.accounts = accounts
        self.fee_account = fee_account
        super().__init__(*args, **kwargs)

    def get_csv_reader(self, filename):
        return csv.DictReader(open(filename, encoding="windows-1251"), delimiter=";")

    def get_date_from_row(self, row):
        return parse_date_liberally(row[self.DATE_FIELD], dict(dayfirst=True))

    def get_account_from_row(self, row):
        k = (row["Валюта"], row["Рахунок"])
        if k not in self.accounts:
            raise ValueError("Unknown account %s %s" % k)
        return self.accounts[k]

    def date(self, filename):
        "Get the maximum date from the file."
        max_date = None
        for row in self.get_csv_reader(filename):
            if not row:
                continue
            date = self.get_date_from_row(row)
            if max_date is None or date > max_date:
                max_date = date
        return max_date

    def account(self, _):
        return "procreditbank-business"

    def extract(self, filename, existing_entries=None):
        entries = []
        for index, row in enumerate(self.get_csv_reader(filename), 1):
            if not row:
                continue
            meta = data.new_metadata(filename, -index)
            entry = self.get_entry_from_row(row, meta)
            if entry is not None:
                entries.append(entry)

        return self.merge_entries(entries)

    def get_entry_from_row(self, row, meta):
        meta["time"] = row[self.DATE_FIELD].split(" ")[-1]
        meta["src_doc_n"] = f"{row['Код операції']} {row['Номер документа']}"
        meta["src_purpose"] = row["Призначення платежу"]
        account = self.get_account_from_row(row)
        if account is None:
            return
        payee = row["Кореспондент"]
        txn = data.Transaction(
            meta,
            self.get_date_from_row(row),
            flags.FLAG_OKAY,
            payee,
            "",
            data.EMPTY_SET,
            data.EMPTY_SET,
            [],
        )
        sum = -D(row["Дебет"]) if row["Дебет"] else D(row["Кредит"])
        units = data.Amount(sum, row["Валюта"])
        txn.postings.append(data.Posting(account, units, None, None, None, None))
        for fee_str in self.fee_regexes:
            if re.search(fee_str, row["Призначення платежу"]):
                txn.postings.append(
                    data.Posting(self.fee_account, -units, None, None, None, None)
                )
                break
        return txn

    def merge_entries(self, entries):
        def find_closest(lst, other, predicate):
            for e in lst:
                if not e == other and predicate(e):
                    return e
            return None

        entries_by_date = defaultdict(list)
        for e in entries:
            entries_by_date[e.date].append(e)
        for _date, subentries in entries_by_date.items():
            if len(subentries) == 1:
                continue
            # find & merge same currency transfers
            for e in subentries:
                other = find_closest(
                    subentries,
                    e,
                    lambda x: x.meta["src_doc_n"] == e.meta["src_doc_n"]
                    and x.postings[0].units == -e.postings[0].units,
                )
                if other is not None:
                    entries.remove(other)
                    subentries.remove(e)
                    subentries.remove(other)
                    e.postings.extend(other.postings)
            # find & merge currency convertions
            for e in subentries:
                m = re.search(self.currency_conversion_regex, e.meta["src_purpose"])
                if not m:
                    continue
                units = data.Amount(
                    D(m.group("amount").replace(",", ".")), m.group("currency")
                )
                other = find_closest(
                    subentries, e, lambda x: x.postings[0].units == -units
                )
                if other is None:
                    raise RuntimeError(f"can't find matching         entry for {e}")
                other_posting = other.postings[0]._replace(
                    price=data.Amount(D(m.group("rate").replace(",", ".")), "UAH")
                )
                e.postings.insert(0, other_posting)
                e.postings.extend(other.postings[1:])
                e.meta["other_src_doc_n"] = other.meta["src_doc_n"]
                entries.remove(other)
        return entries


def get_test_importer():
    return Importer(
        {
            ("UAH", "UA00000001"): "Assets:ProcreditBank:Business:Cash:UAH",
            ("GBP", "UA00000001"): "Assets:ProcreditBank:Business:Cash:GBP",
            ("GBP", "UA00000007"): "Assets:Procreditbank:Business:Transit",
        },
        "Expenses:Fees:Banking:ProcreditBank",
    )


if __name__ == "__main__":
    from beangulp.testing import main

    main(get_test_importer())

src/uabean/importers/procredit_business.py
OSadovy/uabean
"""Imports xls statements from Procreditbank, optained via web interface.
"""

import re

import beangulp
import dateutil.parser
import xlrd
from beancount.core import data, flags
from beancount.core.number import D

from uabean.importers.mixins import IdentifyMixin


class Importer(IdentifyMixin, beangulp.Importer):
    FLAG = flags.FLAG_OKAY
    matchers = [
        ("filename", "Рух по рахунку_"),
        ("mime", "application/vnd.ms-excel"),
    ]
    deposit_regex = r"Перерахування нарахованих відсотків згідно договору вкладу"

    def __init__(
        self,
        account_config,
        *args,
        fee_account="Expenses:Fees:Procreditbank",
        deposit_income_account="Income:Investments:Deposits",
        **kwargs,
    ):
        self.account_config = account_config
        self.fee_account = fee_account
        self.deposit_income_account = deposit_income_account
        super().__init__(*args, **kwargs)

    def date_from_row(self, row):
        return dateutil.parser.parse(row[1].value, dayfirst=True)

    @staticmethod
    def get_number(cell):
        return D(str(cell.value))

    def account(self, _):
        return "procreditbank"

    def extract(self, filename, existing_entries=None):
        entries = []
        workbook = xlrd.open_workbook(filename)
        sheet = workbook.sheet_by_index(0)
        # assert 'Банк: АТ "ПроКредит Банк"' in sheet.cell(1, 12).value
        account_number = sheet.cell(4, 1).value
        assert account_number.startswith("Рахунок:")
        account_number = account_number.split(" ", 1)[1]
        account_currency = sheet.cell(6, 1).value
        assert "Валюта рахунку:" in account_currency
        account_currency = account_currency.rsplit(" ", 1)[1]
        account = self.account_config.get((account_currency, account_number))
        if account is None:
            raise ValueError(f"Unknown account {account_currency} {account_number}")
        header = None
        header_cols = None
        for nrow in range(10, sheet.nrows):
            row = sheet.row(nrow)
            meta = data.new_metadata(filename, nrow)
            if row[1].value == "Операції по рахунку":
                header = "account_ops"
                continue
            elif row[1].value == "Операції по картам":
                header = "card_ops"
                continue
            elif cell_index := find_cell_starting_with(
                row, "Вихідний залишок по рахунку на кінець періоду"
            ):
                entries.append(
                    self.balance_from_row(
                        meta, account, account_currency, row, cell_index
                    )
                )
                break
            elif str(row[1].value).startswith("Дата "):
                second_row = None
                if header == "card_ops":
                    second_row = sheet.row(nrow + 1)
                header_cols = self.header_cols_from_row(row, second_row)
                continue
            row_starts_with_date = bool(re.search(r"\d{2}\.\d{2}\.\d{4}", row[1].value))
            if row_starts_with_date:
                row_dict = self.row_to_dict(header_cols, row)
                if header == "account_ops":
                    entry = self.account_entry_from_row(
                        meta, account, account_currency, row_dict
                    )
                elif header == "card_ops":
                    entry = self.card_entry_from_row(meta, account, row_dict)
                if entry is not None:
                    entries.append(entry)
        return entries

    @classmethod
    def header_cols_from_row(cls, row, second_row=None):
        result = {}
        for i, col in enumerate(row):
            if col.value:
                if second_row is not None and second_row[i].value:
                    # take two consecutive headers from second row which is subheader row
                    result[f"{row[i].value} {second_row[i].value}"] = i
                    for j in range(i + 1, len(second_row)):
                        if second_row[j].value:
                            result[f"{row[i].value} {second_row[j].value}"] = j
                            break
                else:
                    result[col.value] = i
        return result

    @classmethod
    def row_to_dict(cls, header_cols, row):
        result = {}
        for header, col in header_cols.items():
            result[header] = row[col].value
        return result

    def balance_from_row(self, meta, account, currency, row, index):
        dt = dateutil.parser.parse(row[index].value.split(" ")[-1], dayfirst=True)
        num = self.get_number(next(cell for cell in row[index + 1 :] if cell.value))
        return data.Balance(
            meta,
            dt.date(),
            account,
            data.Amount(num, currency),
            None,
            None,
        )

    def account_entry_from_row(self, meta, account, currency, row_dict):
        date = dateutil.parser.parse(row_dict["Дата операціі"], dayfirst=True).date()
        if row_dict["Видатки"]:
            amount = data.Amount(-D(str(row_dict["Видатки"])), currency)
        else:
            amount = data.Amount(D(str(row_dict["Надходження"])), currency)
        postings = [
            data.Posting(
                account,
                amount,
                None,
                None,
                None,
                None,
            )
        ]
        if row_dict["Комісія"]:
            postings.append(
                data.Posting(
                    self.fee_account,
                    data.Amount(D(row_dict["Комісія"]), currency),
                    None,
                    None,
                    None,
                    None,
                )
            )
        narration = row_dict["Призначення платежу"]
        if re.search(self.deposit_regex, narration):
            postings.append(
                data.Posting(self.deposit_income_account, None, None, None, None, None)
            )
        payee = row_dict["Реквізити контрагента"].split("  ")[0]
        return data.Transaction(
            meta,
            date,
            self.FLAG,
            payee,
            narration,
            data.EMPTY_SET,
            data.EMPTY_SET,
            postings,
        )

    def card_entry_from_row(self, meta, account, row_dict):
        dt = dateutil.parser.parse(row_dict["Дата і час операціі"], dayfirst=True)
        meta["time"] = dt.strftime("%H:%M")
        card_amount = data.Amount(
            D(str(row_dict["Карта сума"])), row_dict["Карта валюта"]
        )
        operation_amount = data.Amount(
            D(str(row_dict["Операція сума"])), row_dict["Операція валюта"]
        )
        if card_amount != operation_amount:
            rate = operation_amount.number / card_amount.number
            meta["converted"] = f"{operation_amount} ({rate})"
        postings = [data.Posting(account, card_amount, None, None, None, None)]
        return data.Transaction(
            meta,
            dt.date(),
            self.FLAG,
            None,
            row_dict["Призначення платежу"],
            data.EMPTY_SET,
            data.EMPTY_SET,
            postings,
        )


def find_cell_starting_with(row, s):
    for i, cell in enumerate(row):
        if str(cell.value).startswith(s):
            return i
    return None


def get_test_importer():
    return Importer(
        {
            ("UAH", "26000000000000"): "Assets:Procreditbank:Cash:UAH",
            ("USD", "26000000000000"): "Assets:Procreditbank:Cash:USD",
        },
    )


if __name__ == "__main__":
    from beangulp.testing import main

    main(get_test_importer())

src/uabean/importers/procredit_xls.py
OSadovy/uabean
"""Imports xls statements from pumb, optained via request to online support.

The header is as follows:
Дата операції	Тип операції	Сума у валюті операції	Опис	Дата списання	Сума операції	Комісія
"""

import datetime
import os

import beangulp
import dateutil.parser
import openpyxl
from beancount.core import data, flags
from beancount.core.number import D

from uabean.importers.mixins import IdentifyMixin


class Importer(IdentifyMixin, beangulp.Importer):
    FLAG = flags.FLAG_OKAY
    matchers = [
        ("filename", "pumb"),
        ("mime", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"),
    ]
    DATE_COL = 0
    TYPE_COL = 1
    TRANSACTION_CURRENCY_AMOUNT_COL = 2
    DESCRIPTION_COL = 3
    SETTLE_DATE_COL = 4
    ACCOUNT_CURRENCY_AMOUNT_COL = 5
    FEE_COL = 6
    POS_PAYMENT_STR_RU = "Oперация покупки через POS-терминал"
    POS_PAYMENT_STR_UA = "Купівля через POS-термінал"
    CASHBACK_STR = "Виплата винагороди кешбек ПУМБ"

    def __init__(
        self,
        account_config,
        *args,
        fee_account="Expenses:Fees:Pumb",
        cashback_income_account="Income:Cashback:Pumb",
        **kwargs,
    ):
        self.account_config = account_config
        self.fee_account = fee_account
        self.cashback_income_account = cashback_income_account
        super().__init__(*args, **kwargs)

    def datetime_from_row(self, row):
        return dateutil.parser.parse(row[self.DATE_COL].value, dayfirst=True)

    @classmethod
    def get_amount(cls, cell, currency=None):
        if currency is None:
            n, currency = cell.value.rsplit(" ", 1)
        else:
            n = cell.value
        return data.Amount(D(n.replace(" ", "").replace(",", ".")), currency)

    def account(self, _):
        return "pumb"

    def extract(self, filename, existing_entries=None):
        entries = []
        workbook = openpyxl.load_workbook(filename)
        sheet = workbook.worksheets[0]
        assert "Виписка по операціям" in sheet.cell(7, 1).value
        account_currency = os.path.basename(filename.split("-")[1]).upper()
        account = self.account_config[account_currency]
        date_row_seen = False
        balance_end_date = None
        balance_entry = None
        for i, row in enumerate(sheet.iter_rows()):
            if not date_row_seen:
                if row[0].value == "Дата операції":
                    date_row_seen = True
                if row[0].value == "Період":
                    balance_end_date = dateutil.parser.parse(
                        row[1].value.split(" - ")[-1], dayfirst=True
                    )
                if row[0].value == "Баланс на кінець періоду":
                    balance_entry = data.Balance(
                        data.new_metadata(filename, i + 1),
                        balance_end_date.date() + datetime.timedelta(days=1),
                        account,
                        self.get_amount(row[1], account_currency),
                        None,
                        None,
                    )
                continue
            if row[0].value == "ВСЬОГО":
                break
            meta = data.new_metadata(filename, i + 1)
            entries.append(self.entry_from_row(meta, account, row))
        if balance_entry is not None:
            entries.append(balance_entry)
        return entries

    def entry_from_row(self, meta, account, row):
        dt = self.datetime_from_row(row)
        meta["time"] = dt.strftime("%H:%M")
        narration = row[self.DESCRIPTION_COL].value
        if self.POS_PAYMENT_STR_RU in narration:
            narration = narration.split("; ")[-1]
        elif self.POS_PAYMENT_STR_UA in narration:
            narration = " ".join(narration.split("; ")[0].split()[4:])
        account_currency_amount = self.get_amount(row[self.ACCOUNT_CURRENCY_AMOUNT_COL])
        transaction_currency_amount = self.get_amount(
            row[self.TRANSACTION_CURRENCY_AMOUNT_COL]
        )
        postings = [
            data.Posting(account, account_currency_amount, None, None, None, None)
        ]
        if (
            account_currency_amount.currency != transaction_currency_amount.currency
            and account_currency_amount.number != transaction_currency_amount.number
        ):
            meta["converted"] = f"{-transaction_currency_amount}"
        elif account_currency_amount.number != transaction_currency_amount.number:
            fee_amount = data.Amount(
                abs(
                    account_currency_amount.number - transaction_currency_amount.number
                ),
                account_currency_amount.currency,
            )
            postings.append(
                data.Posting(self.fee_account, fee_amount, None, None, None, None)
            )
        if self.CASHBACK_STR in narration:
            postings.append(
                data.Posting(
                    self.cashback_income_account,
                    -account_currency_amount,
                    None,
                    None,
                    None,
                    None,
                )
            )
            narration = self.CASHBACK_STR
        return data.Transaction(
            meta,
            dt.date(),
            self.FLAG,
            None,
            narration,
            data.EMPTY_SET,
            data.EMPTY_SET,
            postings,
        )


def get_test_importer():
    return Importer(
        account_config={
            "UAH": "Assets:Pumb:Cash:UAH",
            "EUR": "Assets:EUR:Pumb",
        }
    )


if __name__ == "__main__":
    from beangulp.testing import main

    main(get_test_importer())

src/uabean/importers/pumb_xls.py
OSadovy/uabean
"""Imports csv statements from sensesuperapp.

CSV header is as follows:
Дата і час;Статус;Тип;Деталі;Адреса;MCC;Cума списання;Cума зарахування
"""

import csv
import re

import beangulp
import dateutil.parser
import requests
from beancount.core import data, flags
from beancount.core.number import D

from uabean.importers.mixins import IdentifyMixin

mcc_codes_url = "https://raw.githubusercontent.com/Oleksios/Merchant-Category-Codes/main/Without%20groups/mcc-en.json"


class Importer(IdentifyMixin, beangulp.Importer):
    FLAG = flags.FLAG_OKAY
    matchers = [
        ("content", "Виписка за рахунком;;;;;;;"),
        ("mime", "text/csv"),
    ]
    converter = lambda unused, fname: open(fname, encoding="windows-1251").read()
    DATE_COL = 0
    STATUS_COL = 1
    TYPE_COL = 2
    DETAIL_COL = 3
    ADDRESS_COL = 4
    MCC_COL = 5
    DEBIT_COL = 6
    CREDIT_COL = 7

    def __init__(self, account_config, *args, **kwargs):
        self.account_config = account_config
        self.mcc_codes = {}
        super().__init__(*args, **kwargs)

    def get_csv_reader(self, filename):
        return csv.reader(open(filename, encoding="windows-1251"), delimiter=";")

    def date_from_row(self, row):
        return dateutil.parser.parse(row[self.DATE_COL], dayfirst=True)

    def date(self, filename):
        "Get the maximum date from the file."
        reader = self.get_csv_reader(filename)
        next(reader)
        row = next(reader)
        m = re.search(r"за період (.*?) - (.*)", row[0])
        return dateutil.parser.parse(m.group(2), dayfirst=True)

    def account(self, _):
        return "sensebank"

    def _download_mcc_codes(self):
        codes = requests.get(mcc_codes_url).json()
        self.mcc_codes = {c["mcc"]: c["shortDescription"] for c in codes}

    def extract(self, filename, existing_entries=None):
        entries = []
        self._download_mcc_codes()
        reader = self.get_csv_reader(filename)
        for _ in range(4):
            row = next(reader)
        m = re.search(r"Рахунок (\w+) \((\w+)\)", row[0])
        account = self.account_config[m.group(1)]
        currency = m.group(2)
        for row in reader:
            if row[0].startswith("Деталізація операцій") or row[0].startswith(
                "Дата і час"
            ):
                continue
            meta = data.new_metadata(filename, reader.line_num)
            entries.append(self.entry_from_row(meta, account, currency, row))
        return entries

    def entry_from_row(self, meta, account, currency, row):
        dt = self.date_from_row(row)
        meta["time"] = dt.strftime("%H:%M")
        payee = None
        narration = None
        details = row[self.DETAIL_COL]
        type = row[self.TYPE_COL]
        if type.startswith("Покупка") or type.startswith("POS Purchase"):
            payee = details
        elif type in ("Банкомати", "Комуналка та інші платежі", "Інше", "Перекази"):
            narration = details
        if row[self.MCC_COL]:
            meta["category"] = self.mcc_codes[row[self.MCC_COL]]
        if row[self.ADDRESS_COL] and not type == "Перекази":
            meta["address"] = row[self.ADDRESS_COL]
        amount_str = (
            row[self.DEBIT_COL] if row[self.DEBIT_COL] else row[self.CREDIT_COL]
        )
        units = data.Amount(D(amount_str.replace(",", ".")), currency)
        return data.Transaction(
            meta,
            dt.date(),
            self.FLAG,
            payee,
            narration,
            data.EMPTY_SET,
            data.EMPTY_SET,
            [data.Posting(account, units, None, None, None, None)],
        )


def get_test_importer():
    return Importer(
        {
            "UA111111111111111111111111111": "Assets:Sensebank:Cash:UAH",
        }
    )


if __name__ == "__main__":
    from beangulp.testing import main

    main(get_test_importer())

src/uabean/importers/sensebank.py
OSadovy/uabean
"""Imports statement csv fiels exported from tronscan.org UI.

The CSV header is the following:
transaction_id,block,block_ts,from_address,to_address,confirmed,contractRet,quant
"""

import csv

import beangulp
from beancount.core import data, flags
from beancount.core.amount import Amount
from beancount.core.number import D
from beancount.utils.date_utils import parse_date_liberally

from uabean.importers.mixins import IdentifyMixin


class Importer(IdentifyMixin, beangulp.Importer):
    matchers = [
        ("content", __doc__.split("\n")[-2]),
        ("mime", "text/csv"),
    ]

    def __init__(self, account_prefix, currency, *args, **kwargs):
        self.account_prefix = account_prefix
        self.currency = currency  # currently it is not exported by tronscan ui
        super().__init__()

    def get_csv_reader(self, filename):
        return csv.DictReader(open(filename))

    def get_date_from_row(self, row):
        return parse_date_liberally(row["block_ts"])

    def account(self, _):
        return "tron"

    def extract(self, filename, existing_entries=None):
        self.accounts = {}
        entries = []
        for e in existing_entries:
            if isinstance(e, data.Open):
                if "tron_address" in e.meta:
                    self.accounts[e.meta["tron_address"]] = e
        for index, row in enumerate(self.get_csv_reader(filename), 1):
            meta = data.new_metadata(filename, index)
            entry = self.get_entry_from_row(row, meta)
            if entry is not None:
                entries.append(entry)
        return entries

    def get_entry_from_row(self, row, meta):
        meta["src_tx_id"] = row["transaction_id"]
        meta["src_ts"] = row["block_ts"]
        txn = data.Transaction(
            meta,
            self.get_date_from_row(row),
            flags.FLAG_OKAY,
            "",
            "",
            data.EMPTY_SET,
            data.EMPTY_SET,
            [],
        )
        from_account = self.get_account_for_address(row["from_address"])
        from_account_exists = row["from_address"] in self.accounts
        to_account = self.get_account_for_address(row["to_address"])
        to_account_exists = row["to_address"] in self.accounts
        txn.postings.append(
            data.Posting(
                from_account,
                Amount(-D(row["quant"]), self.currency),
                None,
                None,
                None if from_account_exists else "!",
                None,
            )
        )
        txn.postings.append(
            data.Posting(
                to_account,
                Amount(D(row["quant"]), self.currency),
                None,
                None,
                None if to_account_exists else "!",
                None,
            )
        )
        return txn

    def get_account_for_address(self, address):
        if address in self.accounts:
            return self.accounts[address].account
        return f"{self.account_prefix}:{address}"


def get_test_importer():
    return Importer("Assets:Crypto:Tron", "USDT")


if __name__ == "__main__":
    from beangulp.testing import main

    main(get_test_importer())

src/uabean/importers/tronscan.py
OSadovy/uabean
"""Imports statement csv files exported from Ukrsib Business Online web app.

The CSV header is the following:
"ЄДРПОУ";"МФО";"Рахунок";"Валюта";"Дата операцiї";"Код операцiї";"МФО банка";"Назва банка";"Рахунок кореспондента";"ЄДРПОУ кореспондента";"Кореспондент";"Номер документа";"Дата документа";"Дебет";"Кредит";"Призначення платежу";"Гривневе покриття";
"""

import csv
import re
from collections import defaultdict

import beangulp
from beancount.core import data, flags
from beancount.core.number import D
from beancount.utils.date_utils import parse_date_liberally

from uabean.importers.mixins import IdentifyMixin


class Importer(IdentifyMixin, beangulp.Importer):
    matchers = [
        ("content", __doc__.split("\n")[-2]),
        ("mime", "text/csv"),
    ]
    fee_regexes = ("Комісія по рах", "Списання комісії по рахунку")
    currency_conversion_regex = r"Гр.экв.продажу (?P<amount>[\d\.]+) (?P<currency>\w+) на МВР .*?, ЗГІДНО ЗАЯВИ КЛІЄНТА № \d+ \.Курс (?P<rate>[\d\.]+)\.Ком.банку (?P<fee>[\d\.]+)\."
    DATE_FIELD = "Дата операцiї"

    def __init__(self, accounts, fee_account, *args, **kwargs):
        self.accounts = accounts
        self.fee_account = fee_account
        super().__init__(*args, **kwargs)

    def get_csv_reader(self, filename):
        return csv.DictReader(open(filename, encoding="windows-1251"), delimiter=";")

    def get_date_from_row(self, row):
        return parse_date_liberally(row[self.DATE_FIELD], dict(dayfirst=True))

    def get_account_from_row(self, row):
        k = (row["Валюта"], row["Рахунок"])
        if k not in self.accounts:
            raise ValueError("Unknown account %s %s" % (row["Валюта"], row["Наш IBAN"]))
        return self.accounts[k]

    def date(self, filename):
        "Get the maximum date from the file."
        max_date = None
        for row in self.get_csv_reader(filename):
            if not row:
                continue
            date = self.get_date_from_row(row)
            if max_date is None or date > max_date:
                max_date = date
        return max_date

    def account(self, _):
        return "ukrsib"

    def extract(self, filename, existing_entries=None):
        entries = []
        for index, row in enumerate(self.get_csv_reader(filename), 1):
            if not row:
                continue
            meta = data.new_metadata(filename, -index)
            entry = self.get_entry_from_row(row, meta)
            if entry is not None:
                entries.append(entry)

        return self.merge_entries(entries)

    def get_entry_from_row(self, row, meta):
        meta["time"] = row[self.DATE_FIELD].split(" ")[-1]
        meta["src_doc_n"] = row["Номер документа"]
        meta["src_purpose"] = row["Призначення платежу"]
        account = self.get_account_from_row(row)
        if account is None:
            return
        payee = row["Кореспондент"]
        txn = data.Transaction(
            meta,
            self.get_date_from_row(row),
            flags.FLAG_OKAY,
            payee,
            "",
            data.EMPTY_SET,
            data.EMPTY_SET,
            [],
        )
        sum = -D(row["Дебет"]) if row["Дебет"] else D(row["Кредит"])
        units = data.Amount(sum, row["Валюта"])
        txn.postings.append(data.Posting(account, units, None, None, None, None))
        for fee_str in self.fee_regexes:
            if re.search(fee_str, row["Призначення платежу"]):
                txn.postings.append(
                    data.Posting(self.fee_account, -units, None, None, None, None)
                )
                break
        return txn

    def merge_entries(self, entries):
        def find_closest(lst, other, predicate):
            for e in lst:
                if not e == other and predicate(e):
                    return e
            return None

        entries_by_date = defaultdict(list)
        for e in entries:
            entries_by_date[e.date].append(e)
        for _date, subentries in entries_by_date.items():
            if len(subentries) == 1:
                continue
            # find & merge same currency transfers
            for e in subentries:
                other = find_closest(
                    subentries, e, lambda x: x.meta["src_doc_n"] == e.meta["src_doc_n"]
                )
                if other is not None:
                    entries.remove(other)
                    subentries.remove(e)
                    e.postings.extend(other.postings)
            # find & merge currency convertions
            for e in subentries:
                m = re.search(self.currency_conversion_regex, e.meta["src_purpose"])
                if not m:
                    continue
                units = data.Amount(D(m.group("amount")), m.group("currency"))
                other = find_closest(
                    subentries, e, lambda x: x.postings[0].units == -units
                )
                if other is None:
                    raise RuntimeError("can't find matching         entry for %s" % e)
                other_posting = other.postings[0]._replace(
                    price=data.Amount(D(m.group("rate")) / 100, "UAH")
                )
                e.postings.insert(0, other_posting)
                e.postings.extend(other.postings[1:])
                e.meta["other_src_doc_n"] = other.meta["src_doc_n"]
                entries.remove(other)
        return entries


def get_test_importer():
    return Importer(
        {
            ("UAH", "26000000000000"): "Assets:Ukrsibbank:Business:Cash:UAH",
            ("GBP", "26000000000000"): "Assets:Ukrsibbank:Business:Cash:GBP",
        },
        "Expenses:Fees:Banking:Ukrsibbank",
    )


if __name__ == "__main__":
    from beangulp.testing import main

    main(get_test_importer())

src/uabean/importers/ukrsib_business.py
OSadovy/uabean
from beancount.core import data
from beancount.core.amount import Amount
from beancount.core.number import ZERO, D
from beangulp.importers.csv import Col, CSVImporter


class Importer(CSVImporter):
    def __init__(self, *args, **kwargs):
        if "business" in kwargs["account"].lower():
            fname_pattern = f"wise-business-.*-{kwargs['currency']}"
        else:
            fname_pattern = f"wise-personal-.*-{kwargs['currency']}"
        kwargs["matchers"] = [("filename", fname_pattern)]
        kwargs["dateutil_kwds"] = {"dayfirst": True}
        super().__init__(
            {
                Col.AMOUNT: "Amount",
                Col.BALANCE: "Running Balance",
                Col.DATE: "Date",
                Col.NARRATION: "Description",
                Col.PAYEE: "Payee Name",
            },
            *args,
            **kwargs,
        )

    def call_categorizer(self, txn, row):
        txn.meta["lineno"] = -txn.meta["lineno"]
        if txn.narration == "No information":
            txn = txn._replace(narration="")
        if row[13]:  # Merchant
            txn = txn._replace(payee=row[13])
        if row[4]:  # Description
            txn.meta["src_desc"] = row[4]
        txn.meta["src_id"] = row[0]  # TransferWise ID
        total_fees = D(row[18])
        if not total_fees == ZERO:
            txn.postings.append(
                data.Posting(
                    "Expenses:Fees:Wise",
                    Amount(total_fees, self.currency),
                    None,
                    None,
                    None,
                    None,
                )
            )
        return super().call_categorizer(txn, row)


def get_test_importer():
    return Importer(account="Assets:Wise:EUR", currency="EUR")


if __name__ == "__main__":
    from beangulp.testing import main

    main(get_test_importer())

src/uabean/importers/wise_csv.py
OSadovy/uabean
"""Imports balance statements output by Wise API in json format.
"""

import decimal
import json
from datetime import timedelta

import beangulp
import dateutil.parser
from beancount.core import flags
from beancount.core.amount import Amount
from beancount.core.data import Balance, Posting, Transaction, new_metadata

from uabean.importers.mixins import IdentifyMixin


class Importer(IdentifyMixin, beangulp.Importer):
    FLAG = flags.FLAG_OKAY

    def __init__(self, account, fees_account="Expenses:Fees:Wise", **kwargs):
        self._account = account
        self.fees_account = fees_account
        if "business" in account.lower():
            type_pattern = "business"
        elif "personal" in account.lower():
            type_pattern = "personal"
        else:
            type_pattern = ".*?"
        fname_pattern = "wise-" + type_pattern + r"-.*-\w{3}"
        kwargs["matchers"] = [("filename", fname_pattern)]
        super().__init__(**kwargs)

    def file_account(self, filename):
        # example: wise-business-2022-01-01_2022-10-01-USD.json
        parts = filename.split("-")
        account_type = parts[1].capitalize()
        currency = parts[-1].split(".", 1)[0]
        return self._account.format(type=account_type, currency=currency)

    def account(self, _):
        return "wise"

    @staticmethod
    def date_from_str(s):
        return dateutil.parser.parse(s)

    def data_from_file(self, filename):
        return json.load(open(filename, encoding="utf-8"), parse_float=decimal.Decimal)

    def date(self, filename):
        data = self.data_from_file(filename)
        return data["query"]["intervalEnd"]

    def extract(self, filename, existing_entries=None):
        data = self.data_from_file(filename)
        entries = []
        account = self.file_account(filename)
        for i, t in enumerate(data["transactions"]):
            meta = new_metadata(filename, -(i + 1))
            entries.append(self.entry_from_transaction(meta, account, t))
        if data["transactions"]:
            entries.append(
                Balance(
                    new_metadata(filename, -(i + 1)),
                    (
                        self.date_from_str(data["query"]["intervalEnd"])
                        + timedelta(days=1)
                    ).date(),
                    account,
                    amount_from_obj(data["endOfStatementBalance"]),
                    None,
                    None,
                )
            )
        return entries

    def entry_from_transaction(self, meta, account, t):
        dt = self.date_from_str(t["date"])
        meta["time"] = dt.time().strftime("%H:%M:%S")
        meta["src_id"] = t["referenceNumber"]
        if t["exchangeDetails"] is not None:
            meta[
                "converted"
            ] = f'{t["exchangeDetails"]["toAmount"]["value"]} {t["exchangeDetails"]["toAmount"]["currency"]} ({t["exchangeDetails"]["rate"]})'
        postings = [
            Posting(account, amount_from_obj(t["amount"]), None, None, None, None)
        ]
        if not t["totalFees"]["zero"]:
            postings.append(
                Posting(
                    self.fees_account,
                    amount_from_obj(t["totalFees"]),
                    None,
                    None,
                    None,
                    None,
                )
            )
        match t["details"]["type"]:
            case "TRANSFER":
                payee = t["details"]["recipient"]["name"]
            case "CARD":
                payee = t["details"]["merchant"]["name"]
                meta["src_category"] = t["details"]["category"]
            case "DEPOSIT":
                payee = t["details"]["senderName"]
            case "UNKNOWN":
                payee = None
            case "MONEY_ADDED":
                payee = "self"
            case _:
                raise ValueError(f"unknown transaction type: {t['details']['type']}")
        return Transaction(
            meta, dt.date(), self.FLAG, payee, None, set(), set(), postings
        )


def amount_from_obj(obj):
    return Amount(obj["value"], obj["currency"])


def get_test_importer():
    return Importer(account="Assets:Wise:personal:{currency}")


if __name__ == "__main__":
    from beangulp.testing import main

    main(get_test_importer())

src/uabean/importers/wise_json.py
OSadovy/uabean
import pytest

pytest.register_assert_rewrite("common")

tests/conftest.py
OSadovy/uabean
import csv
import datetime
import random

fname = "binance.csv"

with open(fname) as f:
    data = f.read()

lines = data.strip().split("\n")
reader = csv.reader(lines)
headers = next(reader)
data_rows = list(reader)

# Replace User_ID
dummy_id = "123456789"
for row in data_rows:
    row[0] = dummy_id

# Map original dates to new dates
original_dates = {row[1] for row in data_rows}
base_date = datetime.datetime(2000, 1, 1)
date_mapping = {
    date: (base_date + datetime.timedelta(days=i)).strftime("%Y-%m-%d %H:%M:%S")
    for i, date in enumerate(sorted(original_dates))
}

for row in data_rows:
    row[1] = date_mapping[row[1]]

# Change amounts
for row in data_rows:
    try:
        change = float(row[5])
        scale = abs(change) * 0.05  # Up to 5% of original amount
        random_change = random.uniform(-scale, scale)
        new_change = change + random_change

        # Ensure signs stay consistent
        if (change < 0 and new_change > 0) or (change > 0 and new_change < 0):
            new_change = -new_change

        row[5] = "{:.8f}".format(new_change)
    except ValueError:
        pass

# Combine the data and print
output = [headers] + data_rows
with open(fname, "w") as f:
    for row in output:
        print(",".join(row), file=f)

tests/importers/binance/anonymize.py
OSadovy/uabean
import os
from collections import namedtuple

from beancount import loader
from beangulp import extract
from beangulp.testing import _run, compare_expected

Context = namedtuple("Context", ["importers"])


def run_importer_test(importer, capsys):
    _run(
        Context([importer]),
        [os.path.abspath(f"tests/importers/{importer.account('')}/")],
        "",
        0,
        0,
    )
    captured = capsys.readouterr()
    assert "PASSED" in captured.out
    assert "ERROR" not in captured.out


def run_importer_test_with_existing_entries(
    importer, document, existing_entries_filename
):
    base_path = os.path.abspath(f"tests/importers/{importer.account('')}")
    expected_filename = os.path.join(base_path, f"{document}.beancount")
    document = os.path.join(base_path, document)
    existing_entries = loader.load_file(
        os.path.join(base_path, existing_entries_filename)
    )[0]
    account = importer.account(document)
    date = importer.date(document)
    name = importer.filename(document)
    entries = extract.extract_from_file(importer, document, existing_entries)
    diff = compare_expected(expected_filename, account, date, name, entries)
    assert diff == []

tests/importers/common.py
OSadovy/uabean
import re
import sys
from datetime import datetime
from xml.etree import ElementTree as ET


def adjust_date(date_str, offset, date_format="%Y%m%d"):
    return (datetime.strptime(date_str, date_format) + offset).strftime(date_format)


def adjust_datetime(datetime_str, offset, date_format="%Y%m%d"):
    if ";" not in datetime_str:
        return adjust_date(datetime_str, offset, date_format)
    date_part, time_part = datetime_str.split(";")
    adjusted_date = adjust_date(date_part, offset, date_format)
    return f"{adjusted_date};{time_part}"


def adjust_account_id(element, id_mapping):
    account_id = element.get("accountId")
    if account_id not in id_mapping:
        id_mapping[account_id] = f"DUMMYID{len(id_mapping)+1}"
    element.set("accountId", id_mapping[account_id])


def adjust_unique_refs(element, unique_refs, id_mapping):
    for ref in unique_refs:
        original_id = element.get(ref)
        if original_id:
            if original_id not in id_mapping:
                id_mapping[original_id] = f"DUMMYTRADEID{len(id_mapping)+1}"
            element.set(ref, id_mapping[original_id])


def adjust_dates_and_times(element, offset):
    for attr in element.attrib:
        if not element.get(attr):
            continue
        if "datetime" in attr.lower():
            element.set(attr, adjust_datetime(element.get(attr), offset))
        elif "date" in attr.lower():
            element.set(attr, adjust_date(element.get(attr), offset))


def anonymize_dates_in_description(description, offset):
    # Match common date patterns e.g. APR-2020, 2020-04-03, 2020/04/03 etc.
    date_patterns = [
        (r"(\b(JAN|FEB|MAR|APR|MAY|JUN|JUL|AUG|SEP|OCT|NOV|DEC)-\d{4}\b)", "%b-%Y"),
        (r"(\b(JAN|FEB|MAR|APR|MAY|JUN|JUL|AUG|SEP|OCT|NOV|DEC) \d{4}\b)", "%b %Y"),
        (r"(\b\d{4}-\d{2}-\d{2}\b)", "%Y-%m-%d"),
        (r"(\b\d{4}/\d{2}/\d{2}\b)", "%Y/%m/%d"),
    ]

    for pattern, date_format in date_patterns:
        matches = re.findall(pattern, description, re.IGNORECASE)
        for match in matches:
            try:
                original_date_str = match[
                    0
                ]  # Match tuple's first element is the complete matched string
                adjusted_date_str = adjust_date(original_date_str, offset, date_format)
                description = description.replace(original_date_str, adjusted_date_str)
            except ValueError:
                # If a date string does not match the expected format, we'll skip it.
                pass
    return description


def adjust_cash_transaction(cash_transaction, id_mapping, trade_id_mapping, offset):
    adjust_account_id(cash_transaction, id_mapping)
    adjust_unique_refs(
        cash_transaction, ["tradeID", "transactionID", "actionID"], trade_id_mapping
    )
    adjust_dates_and_times(cash_transaction, offset)
    # Anonymize date references in the description
    description = cash_transaction.get("description", "")
    cash_transaction.set(
        "description", anonymize_dates_in_description(description, offset)
    )


def anonymize_xml(xml_data):
    tree = ET.ElementTree(ET.fromstring(xml_data))
    root = tree.getroot()

    earliest_date = min(
        [elem.get("fromDate") for elem in root.findall(".//FlexStatement")]
        + [elem.get("toDate") for elem in root.findall(".//FlexStatement")]
    )
    base_date = "20000101"
    date_format = "%Y%m%d"
    offset = datetime.strptime(base_date, date_format) - datetime.strptime(
        earliest_date, date_format
    )

    id_mapping = {}
    trade_id_mapping = {}

    for flex_statement in root.findall(".//FlexStatement"):
        adjust_account_id(flex_statement, id_mapping)
        adjust_dates_and_times(flex_statement, offset)

    for report_currency in root.findall(".//CashReportCurrency"):
        adjust_account_id(report_currency, id_mapping)
        adjust_dates_and_times(report_currency, offset)

    trade_and_lot_elements = root.findall(".//Trade") + root.findall(".//Lot")
    for trade in trade_and_lot_elements:
        adjust_account_id(trade, id_mapping)
        adjust_unique_refs(
            trade,
            [
                "tradeID",
                "transactionID",
                "ibOrderID",
                "ibExecID",
                "brokerageOrderID",
                "extExecID",
            ],
            trade_id_mapping,
        )
        adjust_dates_and_times(trade, offset)

    for corp_action in root.findall(".//CorporateAction"):
        adjust_account_id(corp_action, id_mapping)
        adjust_unique_refs(corp_action, ["transactionID", "actionID"], trade_id_mapping)
        adjust_dates_and_times(corp_action, offset)

    for cash_transaction in root.findall(".//CashTransaction"):
        adjust_cash_transaction(cash_transaction, id_mapping, trade_id_mapping, offset)

    return ET.tostring(root).decode()


anonymized_xml = anonymize_xml(open(sys.argv[1]).read())
print(anonymized_xml)

tests/importers/ib/anonymize.py
OSadovy/uabean
import csv
import re
import sys
from datetime import datetime


def redact_dates_in_description(desc):
    return re.sub(
        r"від \d+ (січня|лютого|березня|квітня|травня|червня|липня|серпня|вересня|жовтня|листопада|грудня) \d+ р\.",
        "REDACTED",
        desc,
    )


data = open(sys.argv[1], encoding="cp1251").read()

# Convert the data to rows using csv
lines = list(csv.reader(data.splitlines(), delimiter=";"))

# Initialize dictionaries and counters for mappings
account_mappings = {}
correspondent_mappings = {}
edrpou_mappings = {}
nabu_id_mappings = {}
iban_counter = 1
correspondent_counter = 1
edrpou_counter = 1
nabu_id_counter = 1


# Functions to get new values
def get_new_iban():
    global iban_counter
    iban = f"UA{iban_counter:08}"
    iban_counter += 1
    return iban


def get_new_correspondent():
    global correspondent_counter
    name = f"Correspondent{correspondent_counter}"
    correspondent_counter += 1
    return name


def get_new_edrpou():
    global edrpou_counter
    edrpou = f"{edrpou_counter:010}"  # Assuming it's a 10-digit number
    edrpou_counter += 1
    return edrpou


def get_new_nabu_id():
    global nabu_id_counter
    nabu_id = f"{nabu_id_counter:06}"  # Assuming it's a 6-digit number
    nabu_id_counter += 1
    return nabu_id


# Calculate date offset based on the earliest date in CSV
def get_offset_date(original_date, offset_days):
    if not original_date:
        return ""
    if " " in original_date:
        original_datetime = datetime.strptime(original_date, "%d.%m.%Y %H:%M:%S")
    else:
        original_datetime = datetime.strptime(original_date, "%d.%m.%Y")
    offset_datetime = original_datetime - offset_days
    return offset_datetime.strftime("%d.%m.%Y %H:%M:%S")


dates = [line[4] for line in lines[1:] if line[4]]
min_date = min(datetime.strptime(date, "%d.%m.%Y %H:%M:%S") for date in dates)
date_offset = min_date - datetime(2000, 1, 1)

# Anonymize data
for i, line in enumerate(lines[1:], 1):  # Skip header
    if line[0]:  # If the line is not empty
        line[0] = edrpou_mappings.setdefault(line[0], get_new_edrpou())
        line[2] = account_mappings.setdefault(line[2], get_new_iban())
        line[8] = account_mappings.setdefault(line[8], get_new_iban())
        line[4] = get_offset_date(line[4], date_offset)
        line[12] = get_offset_date(line[12], date_offset)
        line[10] = correspondent_mappings.setdefault(line[10], get_new_correspondent())
        line[9] = edrpou_mappings.setdefault(line[9], get_new_edrpou())
        line[1] = nabu_id_mappings.setdefault(line[1], get_new_nabu_id())
        # Redact dates in payment description
        line[15] = redact_dates_in_description(line[15])
        if "nvoice" in line[15]:
            line[15] = "REDACTED"

# Convert the data back to string
output_data = "\n".join([";".join(line) for line in lines])
sys.stdout.buffer.write(output_data.encode("cp1251"))

tests/importers/procreditbank-business/anonymize.py
OSadovy/uabean
from common import run_importer_test

from uabean.importers.alfa_business import get_test_importer


def test_alfabank_business_importer(capsys):
    run_importer_test(get_test_importer(), capsys)

tests/importers/test_alfa_business.py
OSadovy/uabean
from common import run_importer_test

from uabean.importers.binance import get_test_importer


def test_binance_importer(capsys):
    run_importer_test(get_test_importer(), capsys)

tests/importers/test_binance.py
OSadovy/uabean
from common import run_importer_test_with_existing_entries

from uabean.importers.ibkr import get_test_importer


def test_ibkr_importer():
    run_importer_test_with_existing_entries(
        get_test_importer(), "beancount.xml", "existing.beancount"
    )

tests/importers/test_ibkr.py
OSadovy/uabean
from common import run_importer_test

from uabean.importers.kraken import get_test_importer


def test_kraken_importer(capsys):
    run_importer_test(get_test_importer(), capsys)

tests/importers/test_kraken.py
OSadovy/uabean
from common import run_importer_test

from uabean.importers.monobank import get_test_importer


def test_monobank_importer(capsys):
    importer = get_test_importer()
    importer._download_mcc_codes = lambda: None
    importer.mcc_codes = {
        "5817": "Applications",
        "4121": "Taxi",
        "4829": "Money transfer",
    }
    run_importer_test(importer, capsys)

tests/importers/test_monobank.py
OSadovy/uabean
from common import run_importer_test

from uabean.importers.nexo import get_test_importer


def test_nexo_importer(capsys):
    run_importer_test(get_test_importer(), capsys)

tests/importers/test_nexo.py
OSadovy/uabean
from common import run_importer_test

from uabean.importers.procredit_business import get_test_importer


def test_procreditbank_business_importer(capsys):
    run_importer_test(get_test_importer(), capsys)

tests/importers/test_procredit_business.py
OSadovy/uabean
from common import run_importer_test

from uabean.importers.sensebank import get_test_importer


def test_sensebank_importer(capsys):
    importer = get_test_importer()
    importer._download_mcc_codes = lambda: None
    importer.mcc_codes = {
        "4829": "Money transfer",
        "4900": "Utilities",
        "7311": "Advertising",
        "6011": "Сashier's office",
    }
    run_importer_test(importer, capsys)

tests/importers/test_sensebank.py
OSadovy/uabean