diff --git a/crypto_tax/__init__.py b/crypto_tax/__init__.py index 3902f38..de3c223 100644 --- a/crypto_tax/__init__.py +++ b/crypto_tax/__init__.py @@ -28,81 +28,107 @@ import dataclasses import logging import fin_data -from .data import TaxReport, BoughtAndSold, BoughtAndNotYetSold +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_tax(transfers: list) -> TaxReport: +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 True: + 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}') + 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() - 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: + 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): - logger.info(f'Spending: {amount}') - sell(amount, executed_time, spent=True) + 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) - # 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) + 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: - logger.info('Sell') spend(fee, transfer.executed_time) - del transfer, fee + del fee + + del transfer def exchange_rate_at_time(a, b, t): try: @@ -117,4 +143,5 @@ def compute_tax(transfers: list) -> TaxReport: 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, ) diff --git a/crypto_tax/__main__.py b/crypto_tax/__main__.py index 2166315..e909e7b 100644 --- a/crypto_tax/__main__.py +++ b/crypto_tax/__main__.py @@ -11,7 +11,7 @@ import dataclasses import logging from pathlib import Path -from . import compute_tax, output_excel +from . import compute_fifo, output_excel KUCOIN_CLIENT = fin_depo.defi_kucoin.KucoinDepoFetcher( secrets.KUCOIN_KEY, @@ -31,28 +31,13 @@ def main(): logger = logging.getLogger('crypto_tax') logger.setLevel('INFO') - TRANSFERS = list(KRAKEN_CLIENT._get_double_registers()) - TRANSFERS += list(KUCOIN_CLIENT._get_double_registers()) - tax_report = compute_tax(TRANSFERS) + TRANSFERS = [] + TRANSFERS += list(KRAKEN_CLIENT._get_double_registers()) + #TRANSFERS += list(KUCOIN_CLIENT._get_double_registers()) + tax_report = compute_fifo(TRANSFERS) logger.info('-'*80) - if True: - logger.info('Bought for:') - for asset, prices in tax_report.current_assets.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 - - - for bas in tax_report.bought_and_sold_for: - sys.stdout.write(f'{bas.amount} ({bas.time_bought} ----> {bas.time_sold})\n') - output_path = Path('./output/report.xlsx') output_excel.produce_excel_report(tax_report, output_path) diff --git a/crypto_tax/data.py b/crypto_tax/data.py index 7354142..0b88b11 100644 --- a/crypto_tax/data.py +++ b/crypto_tax/data.py @@ -23,12 +23,19 @@ class BoughtAndNotYetSold: time_bought: datetime.datetime +@dataclasses.dataclass +class LedgerEntry: + time: datetime.datetime + type: str + account_provider: str + amount: AssetAmount + balance: AssetAmount + + @dataclasses.dataclass class TaxReport: bought_and_sold_for: list[BoughtAndSold] bought_and_spent_for: list[BoughtAndSold] current_assets: dict[Asset, deque[BoughtAndNotYetSold]] exchange_rate_at_time: Callable[[Asset,Asset, datetime.datetime ], Decimal] - - - + ledger_entries: list[LedgerEntry] diff --git a/crypto_tax/output_excel.py b/crypto_tax/output_excel.py index bfd9525..bdd416e 100644 --- a/crypto_tax/output_excel.py +++ b/crypto_tax/output_excel.py @@ -51,7 +51,32 @@ def write_current_assets(sheet, tax_report: TaxReport): del asset, positions, row, total_amount +def write_ledger_sheet(sheet, tax_report: TaxReport): + column_headers = [ + 'Tidspunkt (UTC)', + 'Type', + 'Værdipapir', + 'Mængde', + 'Balance', + 'Kontoudbyder', + ] + sheet.append(column_headers) + row_idx = 2 + for ledger_entry in tax_report.ledger_entries: + row = [ + ledger_entry.time.replace(tzinfo=None), + ledger_entry.type, + ledger_entry.amount.asset.raw_short_name(), + ledger_entry.amount.amount, + ledger_entry.balance.amount, + ledger_entry.account_provider, + ] + sheet.append(row) + row_idx+= 1 + del ledger_entry, row + def write_fifo_sheet(sheet, tax_report: TaxReport): + # TODO: Account for transfers column_headers = [ 'Værdipapir', 'Mængde', 'Anskaffelsesværdi (DKK)', 'Købsdato (UTC)', 'Salgsværdi (DKK)', 'Salgsdato (UTC)', 'Profit (DKK)', 'Tab (DKK)', @@ -89,7 +114,10 @@ def produce_excel_report(report: TaxReport, output_path: Path): workbook.active.title = 'Current Assets' write_current_assets(workbook.active, report) - sheet_buy_sales = workbook.create_sheet('FIFO') - write_fifo_sheet(sheet_buy_sales, report) + sheet_ledger = workbook.create_sheet('Ledger') + write_ledger_sheet(sheet_ledger , report) + + sheet_fifo = workbook.create_sheet('FIFO') + write_fifo_sheet(sheet_fifo, report) workbook.save(output_path)