import csv from datetime import datetime from decimal import Decimal from itertools import groupby import logging from typing import Dict, List from kraken import read_ledger from ledger_action import LedgerAction from ledger_process import LedgerProcess from trade import Trade from trade_queue import FIFOQueue logging.basicConfig( level=logging.WARNING, format="%(asctime)s - %(levelname)s - %(name)s - %(message)s" ) # Set up a dedicated logger for FIFOQueue logger = logging.getLogger("pnlcalc") def generate_report(sale_entries, proceeds: Decimal, crypto_asset, date_sold): report = [] sell_date = datetime.strptime(date_sold, "%Y-%m-%d").strftime("%d.%m.%Y") proceeds = Decimal(proceeds) trade: Trade for trade in sale_entries: buy_date_formatted = datetime.strptime(trade.date, "%Y-%m-%d").strftime( "%d.%m.%Y" ) holding_period = ( datetime.strptime(date_sold, "%Y-%m-%d") - datetime.strptime(trade.date, "%Y-%m-%d") ).days short_or_long = "Short" if holding_period < 365 else "Long" total_cost = trade.total_cost gain_or_loss = proceeds - total_cost report.append( { "Amount": f"{trade.amount:.8f}", "Currency": crypto_asset, "Date Sold": sell_date, "Date Acquired": buy_date_formatted, "Short/Long": short_or_long, "Buy/Input at": "Kraken", "Sell/Output at": "Kraken", "Proceeds": f"{proceeds:.2f}", "Cost Basis": f"{total_cost:.2f}", "Gain/Loss": f"{gain_or_loss:.2f}", } ) return report report = [] def write_report(output_path: str): # Write report to CSV with open(output_path, "w", newline="") as csvfile: fieldnames = [ "Amount", "Currency", "Date Sold", "Date Acquired", "Short/Long", "Buy/Input at", "Sell/Output at", "Proceeds", "Cost Basis", "Gain/Loss", ] writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() writer.writerows(report) # Usage ledger_path = "ledgers.csv" # Replace with your ledger file path output_path = "tax_report.csv" # Replace with your desired output file path actions = read_ledger(ledger_path) lp = LedgerProcess() lp.process_ledger(actions) pass # write_report(output_path)