1
0
crypto-tax/crypto_tax/__init__.py
Jon Michael Aanes cd630de70b
Some checks failed
Run Python tests (through Pytest) / Test (push) Failing after 24s
Verify Python project can be installed, loaded and have version checked / Test (push) Failing after 22s
Plans for the future
2024-12-22 06:25:07 +01:00

125 lines
4.2 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
logger = logging.getLogger(__name__)
@dataclasses.dataclass
class BoughtAndSold:
amount: AssetAmount
time_bought: datetime.datetime
time_sold: datetime.datetime
@dataclasses.dataclass
class BoughtAndNotYetSold:
amount: AssetAmount
time_bought: datetime.datetime
@dataclasses.dataclass
class TaxReport:
bought_and_sold_for: list[BoughtAndSold]
bought_and_spent_for: list[BoughtAndSold]
current_assets: dict[Asset, deque[BoughtAndNotYetSold]]
def compute_tax(transfers: list) -> TaxReport:
transfers.sort(key=lambda t:t.executed_time)
bought_and_sold_for: list[BoughtAndSold] = []
bought_and_spent_for: list[BoughtAndSold] = []
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 sell(amount: AssetAmount, executed_time: datetime.datetime, spent=False):
logger.info(f'Selling: {amount}')
while amount.amount > 0:
if len(prices_bought_for[amount.asset]) == 0:
raise Exception('Miscalculation: ' + str(amount))
partial = prices_bought_for[amount.asset].popleft()
logger.info(f'Fuck : {amount} - {partial.amount}')
amount_covered_by_partial = min(partial.amount, amount)
amount -= amount_covered_by_partial
new_partial_amount = partial.amount - amount_covered_by_partial
logger.info(f' => {amount} ({new_partial_amount}')
# Re-add partial
if new_partial_amount.amount != 0:
new_partial = BoughtAndNotYetSold(new_partial_amount, partial.time_bought)
logger.info(f' => new partial: {new_partial}')
prices_bought_for[amount.asset].appendleft(new_partial)
del new_partial
# Add FIFO like
if spent:
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
def spend(amount: AssetAmount, executed_time: datetime.datetime):
logger.info(f'Spending: {amount}')
sell(amount, executed_time, spent=True)
for transfer in transfers:
logger.info(transfer)
# Buy
if output := transfer.output:
logger.info('Buy')
if output.asset not in prices_bought_for:
prices_bought_for[output.asset] = deque()
prices_bought_for[output.asset].append(BoughtAndNotYetSold(output, transfer.executed_time))
del output
# Sell
if _input := transfer.input:
logger.info('Sell')
sell(_input, transfer.executed_time)
del _input
if fee := transfer.fee:
logger.info('Sell')
spend(fee, transfer.executed_time)
del transfer, fee
return TaxReport(
bought_and_sold_for=bought_and_sold_for,
bought_and_spent_for=bought_and_spent_for,
current_assets=prices_bought_for,
)