crypto-tax/crypto_tax/__init__.py

"""# Crypto Tax.

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

Produces an Excel report containing the following sheets:

- **Overview Sheet**: An overview sheet with the suggested report values.
- **Current Assets**: Containing currently held assets.
- **Ledger**: Ledger of all registered transactions.
- Various **FIFO** sheets for the different asset classes.

## Usage

Make a `/secrets` folder, and place relevant secrets there.

Run `python -m crypto_tax`

## Disclaimer

This tool is a work in progress, and may contain outdated or plain faulty
logic. The information gathered from crypto exchanges may be incorrect.

When using the produced reports, you acknowledge that you bear the full
responsiblity of reporting the correct information to SKAT, and that you have
validated the generated output.

The report suggestions are informal suggestions, and is not professional
guidance. I bear no responsiblity for faults in the generated tax reports.

## TODO

This tool is a work in progress:

- [ ] Support Partisia Blockchain
- [ ] Fix Kucoin issues.
- [ ] Adjust the account provider column.
- [X] 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,
    )