From 3f1d5766117a83b92a0d2289c988cb983dd3e63f Mon Sep 17 00:00:00 2001 From: Jon Michael Aanes Date: Sun, 22 Dec 2024 06:10:23 +0100 Subject: [PATCH] Restructure --- crypto_tax/__init__.py | 120 ++++++++++++++++++++++++++++++++++++++ crypto_tax/__main__.py | 127 ++++++----------------------------------- 2 files changed, 138 insertions(+), 109 deletions(-) diff --git a/crypto_tax/__init__.py b/crypto_tax/__init__.py index 42b2e5e..6aa42eb 100644 --- a/crypto_tax/__init__.py +++ b/crypto_tax/__init__.py @@ -2,3 +2,123 @@ WIP """ + +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 + + if True: + logger.info('Bought for:') + for asset, prices in prices_bought_for.items(): + price_sum = sum((p.amount for p in prices), start=AssetAmount.ZERO) + sys.stdout.write(f' - {asset.raw_short_name():10} ({price_sum}): ') + for price in prices: + sys.stdout.write(str(price.amount.amount)) + sys.stdout.write(', ') + del price + sys.stdout.write('\n') + del asset, prices + + return TaxReport( + bought_and_sold_for=bought_and_sold_for, + bought_and_spent_for=bought_and_spent_for, + current_assets=prices_bought_for, + ) diff --git a/crypto_tax/__main__.py b/crypto_tax/__main__.py index 8c22c01..509d5a6 100644 --- a/crypto_tax/__main__.py +++ b/crypto_tax/__main__.py @@ -8,6 +8,9 @@ from collections import deque from fin_depo.data import * import datetime import dataclasses +import logging + +from . import compute_tax KUCOIN_CLIENT = fin_depo.defi_kucoin.KucoinDepoFetcher( secrets.KUCOIN_KEY, @@ -20,120 +23,26 @@ KRAKEN_CLIENT = fin_depo.defi_kraken.KrakenDepoFetcher( secrets.KRAKEN_SECRET, ) - -@dataclasses.dataclass -class BoughtAndSold: - amount: AssetAmount - time_bought: datetime.datetime - time_sold: datetime.datetime - -@dataclasses.dataclass -class BoughtAndNotYetSold: - amount: AssetAmount - time_bought: datetime.datetime - - -def compute_tax(transfers: list): - 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): - print(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() - - print(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 - - print(f' => {amount} ({new_partial_amount}') - - # Re-add partial - if new_partial_amount.amount != 0: - new_partial = BoughtAndNotYetSold(new_partial_amount, partial.time_bought) - print(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): - print(f'Spending: {amount}') - sell(amount, executed_time, spent=True) - - for transfer in transfers: - print(transfer) - # Buy - if output := transfer.output: - print('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: - print('Sell') - sell(_input, transfer.executed_time) - del _input - - if fee := transfer.fee: - print('Sell') - spend(fee, transfer.executed_time) - del transfer, fee - - if True: - print('Bought for:') - for asset, prices in prices_bought_for.items(): - price_sum = sum((p.amount for p in prices), start=AssetAmount.ZERO) - sys.stdout.write(f' - {asset.raw_short_name():10} ({price_sum}): ') - for price in prices: - sys.stdout.write(str(price.amount.amount)) - sys.stdout.write(', ') - del price - sys.stdout.write('\n') - del asset, prices - - - print('-'*80) - - - for bas in bought_and_sold_for: - print(f'{bas.amount} ({bas.time_bought} ----> {bas.time_sold})') - - - for c, l in prices_bought_for.items(): - print(c) - print(l) - def main(): """Main function.""" + logger = logging.getLogger('crypto_tax') + logger.setLevel('INFO') + TRANSFERS = list(KRAKEN_CLIENT._get_double_registers()) - for k in TRANSFERS: - print(k) #TRANSFERS += list(KUCOIN_CLIENT._get_double_registers()) - compute_tax(TRANSFERS) + tax_report = compute_tax(TRANSFERS) + + logger.info('-'*80) + + + for bas in tax_report.bought_and_sold_for: + logger.info(f'{bas.amount} ({bas.time_bought} ----> {bas.time_sold})') + + + for c, prices in tax_report.current_assets.items(): + logger.info('%s', c) + logger.info('%s', prices) if __name__ == '__main__':