148 lines
5.1 KiB
Python
148 lines
5.1 KiB
Python
"""# 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 dataclasses
|
|
import logging
|
|
import fin_data
|
|
|
|
from .data import TaxReport, BoughtAndSold, BoughtAndNotYetSold, LedgerEntry
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
FIN_DB = fin_data.FinancialDatabase(enable_kucoin=True, enable_nationalbanken_dk=True)
|
|
|
|
|
|
def compute_fifo(transfers: list) -> TaxReport:
|
|
transfers.sort(key=lambda t:t.executed_time)
|
|
|
|
|
|
bought_and_sold_for: list[BoughtAndSold] = []
|
|
bought_and_spent_for: list[BoughtAndSold] = []
|
|
ledger_entries = []
|
|
|
|
prices_bought_for: dict[Asset, deque[BoughtAndNotYetSold]] = {}
|
|
if False:
|
|
prices_bought_for[MPC] = deque()
|
|
prices_bought_for[MPC].append(BoughtAndNotYetSold(AssetAmount(MPC, Decimal(100)), datetime.datetime(2000, 1,1,1,1,1,1)))
|
|
prices_bought_for[USDT] = deque()
|
|
prices_bought_for[USDT].append(BoughtAndNotYetSold(AssetAmount(USDT, Decimal(100)), datetime.datetime(2000, 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.info('%s: %s', variant, amount_to_sell)
|
|
amount = amount_to_sell
|
|
while amount.amount > 0:
|
|
if len(prices_bought_for[amount.asset]) == 0:
|
|
raise Exception('Miscalculation: ' + str(amount))
|
|
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:
|
|
logger.info(transfer)
|
|
|
|
# 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
|
|
|
|
# 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,
|
|
)
|