crypto-tax/crypto_tax/__init__.py

"""# Crypto Tax.

Tool to automatically perform tax computations of crypto transactions from the danish perspective.

## TODO

This tool is a work in progress:

- [ ] Support Partisia Blockchain
- [ ] Fix Kucoin issues.
- [ ] Produce Excel output file:
  * Sheet 1: Gives a basic overview, including computed tax values from the rest of
    the sheet, and reasonings for these. Gives an overview of the other sheets.
  * Sheet 2: Current assets, including FIFO acquisition prices.
  * Sheet 3: Historic sales, with FIFO.
"""

import sys
import requests
import fin_depo
from . import secrets
from fin_defs import CryptoCurrency, AssetAmount, MPC, Asset, USDT
from decimal import Decimal
from collections import deque
from fin_depo.data import *
import datetime
import dataclassabc
import dataclasses
import logging
import fin_data

from .data import TaxReport, BoughtAndSold, BoughtAndNotYetSold, LedgerEntry

try:
    import logging_color

    logging_color.monkey_patch()
except ImportError as e:
    logging.warning('Color logging not enabled')
    del e

logger = logging.getLogger(__name__)

FIN_DB = fin_data.FinancialDatabase(enable_kucoin=True, enable_nationalbanken_dk=True)

@dataclassabc.dataclassabc(frozen=True)
class TransferDetails(fin_depo.data.DoubleRegister):
    withdrawal: fin_depo.data.WithdrawalDetails
    deposit: fin_depo.data.DepositDetails

    @property
    def fee(self) -> AssetAmount | None:
        return self.withdrawal.fee # TODO?

    @property
    def executed_time(self) -> datetime.datetime:
        return self.withdrawal.executed_time

    @property
    def input(self) -> None:
        return None

    @property
    def output(self) -> None:
        return None


def collate_interaccount_transfers(transfers: list[fin_depo.data.DoubleRegister]) -> list[fin_depo.data.DoubleRegister]:
    new_transfers = []
    for t in transfers:
        if isinstance(t, fin_depo.data.DepositDetails):
            prev_transfer = new_transfers[-1] if len(new_transfers) > 0 else None
            if isinstance(prev_transfer , fin_depo.data.WithdrawalDetails):
                if prev_transfer.withdrawn == t.deposit:
                    del new_transfers[-1]
                    new_transfers.append(TransferDetails(prev_transfer, t))
                    continue
            del prev_transfer

        new_transfers.append(t)
        del t

    return new_transfers


def compute_fifo(transfers: list) -> TaxReport:
    transfers.sort(key=lambda t:t.executed_time)
    transfers = collate_interaccount_transfers(transfers)


    bought_and_sold_for: list[BoughtAndSold] = []
    bought_and_spent_for: list[BoughtAndSold] = []
    ledger_entries = []

    prices_bought_for: dict[Asset, deque[BoughtAndNotYetSold]] = {}
    if True:
        prices_bought_for[MPC] = deque()
        # TODO:
        prices_bought_for[MPC].append(BoughtAndNotYetSold(AssetAmount(MPC, Decimal(80)), datetime.datetime(2024, 4,1,1,1,1,1)))
        prices_bought_for[USDT] = deque()
        prices_bought_for[USDT].append(BoughtAndNotYetSold(AssetAmount(USDT, Decimal(20)), datetime.datetime(2020, 1,1,1,1,1,1)))

    def current_balance(asset: Asset) -> AssetAmount:
        return sum((p.amount for p in prices_bought_for[asset]), start=AssetAmount.ZERO)

    def sell(amount_to_sell: AssetAmount, executed_time: datetime.datetime, variant: str = 'SELL'):
        logger.debug('%s: %s', variant, amount_to_sell)
        amount = amount_to_sell
        while amount.amount > 0:
            if len(prices_bought_for[amount.asset]) == 0:
                msg = f'Miscalculation: {amount} of {amount_to_sell}'
                raise Exception(msg)
            partial = prices_bought_for[amount.asset].popleft()

            amount_covered_by_partial = min(partial.amount, amount)
            amount -= amount_covered_by_partial
            new_partial_amount = partial.amount - amount_covered_by_partial

            # Re-add partial
            if new_partial_amount.amount != 0:
                new_partial = BoughtAndNotYetSold(new_partial_amount, partial.time_bought)
                prices_bought_for[amount.asset].appendleft(new_partial)
                del new_partial

            # Add FIFO like
            if variant == 'FEE':
                bought_and_spent_for.append(BoughtAndSold(amount_covered_by_partial, partial.time_bought, executed_time))
            else:
                bought_and_sold_for.append(BoughtAndSold(amount_covered_by_partial, partial.time_bought, executed_time))

            del partial, amount_covered_by_partial

        ledger_entries.append(LedgerEntry(
            time = executed_time,
            type = variant,
            account_provider = 'Kraken (TODO)',
            amount = -amount_to_sell,
            balance = current_balance(amount.asset),
        ))

    def spend(amount: AssetAmount, executed_time: datetime.datetime):
        if amount.amount > 0:
            sell(amount, executed_time, 'FEE')

    def buy(bought: AssetAmount, executed_time: datetime.datetime, variant: str):
        logger.debug('Bought: %s', bought)
        if bought.asset not in prices_bought_for:
            prices_bought_for[bought.asset] = deque()
        prices_bought_for[bought.asset].append(BoughtAndNotYetSold(bought, executed_time))

        ledger_entries.append(LedgerEntry(
            time = executed_time,
            type = variant,
            account_provider = 'Kraken (TODO)',
            amount = bought,
            balance = current_balance(bought.asset),
        ))


    for transfer in transfers:

        # Sell
        if _input := transfer.input:
            variant = 'WITHDRAW' if transfer.output is None else 'SELL'
            sell(_input, transfer.executed_time, variant)
            del variant
        del _input

        # Buy
        if output := transfer.output:
            variant = 'DEPOSIT' if transfer.input is None else 'BUY'
            buy(output, transfer.executed_time, variant)
            del variant
        del output

        # TODO: Output line for transfer.

        # Fee
        if fee := transfer.fee:
            spend(fee, transfer.executed_time)
        del fee

        del transfer

    def exchange_rate_at_time(a, b, t):
        try:
            sample = FIN_DB.get_exchange_rate_at_time(a, b, t)
            return sample.average
        except Exception:
            logger.exception('Error when fetching exchange rate (%s,%s) at %s', a, b, t)
            return None

    return TaxReport(
            bought_and_sold_for=bought_and_sold_for,
            bought_and_spent_for=bought_and_spent_for,
            current_assets=prices_bought_for,
            exchange_rate_at_time=exchange_rate_at_time,
            ledger_entries = ledger_entries,
    )