From 3d7f4d94ec854ed6e62b94c0e3ea5ca4a8d5f7a0 Mon Sep 17 00:00:00 2001 From: Max Wojczuk Date: Fri, 9 May 2025 20:40:42 +0200 Subject: [PATCH] feat: Add fees as crypto cost --- src/crypto.py | 9 +++++---- src/data_sources/crypto_loader/csv_loader.py | 7 ++++--- src/domain/crypto/profit_calculator.py | 13 ++++++++++++- src/domain/transactions/transaction.py | 4 +++- src/plugins/crypto/binance/csv.py | 9 +++++---- src/plugins/crypto/generic_saver.py | 5 +++-- src/plugins/crypto/revolut/row_parser.py | 15 ++++++++++++--- 7 files changed, 44 insertions(+), 18 deletions(-) diff --git a/src/crypto.py b/src/crypto.py index 2a05972..a7d0307 100644 --- a/src/crypto.py +++ b/src/crypto.py @@ -23,7 +23,7 @@ def setup_yearly_profit_calculator(cls) -> YearlyProfitCalculator: def read_transactions(cls, filepaths: List[str]) -> List[Transaction]: transaction_loader = MultiSourcesLoader(Loader()) return transaction_loader.load(filepaths) - + @classmethod def set_log_level(cls, log_level: str): logger.remove() @@ -37,14 +37,15 @@ def set_log_level(cls, log_level: str): @click.option('--deductible-loss', '-l', default=-1, help='Deductible loss from previous years. It overrides calculation of loss by the script', type=float) +@click.option('--include-fees', '-fe', default=True, help='Should fees be included in the loss calculation') @click.option('--log-level', '-ll', default='DEBUG', help='Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)') -def crypto(tax_year: int, filepaths: tuple[str, ...], deductible_loss: float, log_level: str): +def crypto(tax_year: int, filepaths: tuple[str, ...], deductible_loss: float, log_level: str, include_fees: bool): CryptoSetup.set_log_level(log_level) profit_calculator = CryptoSetup.setup_yearly_profit_calculator() all_transactions = CryptoSetup.read_transactions(list(filepaths)) - - profit_per_year = profit_calculator.profit_per_year(all_transactions) + + profit_per_year = profit_calculator.profit_per_year(all_transactions, include_fees) tax_calculator = TaxCalculator() tax_data = tax_calculator.calculate_tax_per_year(profit_per_year, tax_year, deductible_loss) print(tax_data, end='\n\n') diff --git a/src/data_sources/crypto_loader/csv_loader.py b/src/data_sources/crypto_loader/csv_loader.py index beaf243..d111cd1 100644 --- a/src/data_sources/crypto_loader/csv_loader.py +++ b/src/data_sources/crypto_loader/csv_loader.py @@ -30,7 +30,8 @@ def _parse_row(cls, row: dict) -> Optional[Transaction]: asset=cls._asset_value(row), fiat_value=cls._fiat_value(row), action=cls._action(row), - date=cls._datetime(row) + date=cls._datetime(row), + fees=cls._fiat_value(row, "fees") ) return transaction except (ValueError, KeyError) as e: @@ -49,9 +50,9 @@ def _asset_value(cls, row: dict) -> AssetValue: raise ValueError(f"Failed to parse cryptocurrency value: {str(e)}") @classmethod - def _fiat_value(cls, row: dict) -> FiatValue: + def _fiat_value(cls, row: dict, column="fiat_value") -> FiatValue: try: - amount = float(row["fiat_value"]) + amount = float(row[column]) currency = CurrencyBuilder.build(row["currency"]) return FiatValue(amount, currency) except (ValueError, KeyError) as e: diff --git a/src/domain/crypto/profit_calculator.py b/src/domain/crypto/profit_calculator.py index 80c5d9b..7e47aec 100644 --- a/src/domain/crypto/profit_calculator.py +++ b/src/domain/crypto/profit_calculator.py @@ -13,10 +13,15 @@ class YearlyProfitCalculator: def __init__(self, exchanger: Exchanger): self.exchanger = exchanger - def profit_per_year(self, transactions: List[Transaction]) -> ProfitPerYear: + def profit_per_year(self, transactions: List[Transaction], include_fees: bool) -> ProfitPerYear: income = self._sum_transactions_per_year(transactions, Action.SELL) cost = self._sum_transactions_per_year(transactions, Action.BUY) + fees = self._sum_fees_per_year(transactions) profit = ProfitPerYear(income, cost) + if include_fees: + logger.info("Adding fees to the cost calculation") + for year in fees.keys(): + profit.add_cost(year, fees[year]) logger.info(f"Calculated profit per year: {profit}") return profit @@ -31,3 +36,9 @@ def _sum_transactions_per_year(self, transactions: List[Transaction], transactio transactions_sum_per_year[transaction.year()] += transaction_value_in_base_currency return transactions_sum_per_year + + def _sum_fees_per_year(self, transactions: List[Transaction]) -> defaultdict[int, FiatValue]: + fees_sum_per_year: defaultdict[int, FiatValue] = defaultdict(lambda: FiatValue(0)) + for transaction in transactions: + fees_sum_per_year[transaction.year()] += transaction.fees + return fees_sum_per_year diff --git a/src/domain/transactions/transaction.py b/src/domain/transactions/transaction.py index 2f60ab3..203ae65 100644 --- a/src/domain/transactions/transaction.py +++ b/src/domain/transactions/transaction.py @@ -10,11 +10,13 @@ def __init__(self, asset: AssetValue, fiat_value: FiatValue, action: Action, - date: pendulum.DateTime): + date: pendulum.DateTime, + fees: FiatValue = FiatValue(0)): self.fiat_value = fiat_value self.asset = asset self.action = action self.date = date + self.fees = fees def year(self) -> int: return self.date.year diff --git a/src/plugins/crypto/binance/csv.py b/src/plugins/crypto/binance/csv.py index e37f308..34352e1 100644 --- a/src/plugins/crypto/binance/csv.py +++ b/src/plugins/crypto/binance/csv.py @@ -44,14 +44,14 @@ def _process_convert_transactions(self, binance_transactions: List[BinanceTransa fiat_value=FiatValue(abs(fiat_tx.change), fiat_tx.coin), action=Action.SELL if crypto_tx.change < 0 else Action.BUY, # TODO: handle datezone - date=pendulum.instance(current_tx.utc_time) + date=pendulum.instance(current_tx.utc_time), + fees=FiatValue(0) # Default to 0 fees for convert operations ) for current_tx, next_tx in zip(binance_transactions[::2], binance_transactions[1::2]) for fiat_tx, crypto_tx in [(current_tx, next_tx) if current_tx.coin in fiat_currencies_list else (next_tx, current_tx)] ] def _process_transaction_operations(self, binance_transactions: List[BinanceTransaction]) -> List[Transaction]: - # TODO: handle sell operations transactions = [] for group in zip(binance_transactions[::3], binance_transactions[1::3], binance_transactions[2::3]): buy_tx = next(tx for tx in group if tx.operation == "Transaction Buy") @@ -63,11 +63,12 @@ def _process_transaction_operations(self, binance_transactions: List[BinanceTran continue transactions.append(Transaction( - asset=AssetValue(abs(buy_tx.change) + abs(fee_tx.change), buy_tx.coin), + asset=AssetValue(abs(buy_tx.change), buy_tx.coin), fiat_value=FiatValue(abs(spend_tx.change), spend_tx.coin), action=Action.BUY, # TODO: handle datezone - date=pendulum.instance(buy_tx.utc_time) + date=pendulum.instance(buy_tx.utc_time), + fees=FiatValue(abs(fee_tx.change), fee_tx.coin) )) return transactions diff --git a/src/plugins/crypto/generic_saver.py b/src/plugins/crypto/generic_saver.py index 3968f12..70bd46b 100644 --- a/src/plugins/crypto/generic_saver.py +++ b/src/plugins/crypto/generic_saver.py @@ -8,7 +8,7 @@ class GenericCsvSaver: def save(transactions: List[Transaction], file_path: str): with open(file_path, 'w') as csvfile: # todo: move it to domain or data_sources - writer = csv.DictWriter(csvfile, fieldnames=["date", "operation", "amount", "symbol", "fiat_value", "currency"]) + writer = csv.DictWriter(csvfile, fieldnames=["date", "operation", "amount", "symbol", "fiat_value", "currency", "fees"]) writer.writeheader() for transaction in transactions: writer.writerow({ @@ -17,5 +17,6 @@ def save(transactions: List[Transaction], file_path: str): "amount": f"{float(transaction.asset.amount):.8f}", "symbol": transaction.asset.asset_name, "fiat_value": f"{float(transaction.fiat_value.amount):.2f}", - "currency": transaction.fiat_value.currency + "currency": transaction.fiat_value.currency, + "fees": f"{float(transaction.fees.amount):.2f}", }) diff --git a/src/plugins/crypto/revolut/row_parser.py b/src/plugins/crypto/revolut/row_parser.py index 2ba097a..9d9bc89 100644 --- a/src/plugins/crypto/revolut/row_parser.py +++ b/src/plugins/crypto/revolut/row_parser.py @@ -18,11 +18,20 @@ def parse(cls, row: Dict) -> Transaction: if action is None: logger.debug(f"Skipping transaction: {row}") return None + + # Try to parse fees, default to 0 if not present or invalid + try: + fees = cls._fiat_value(row, "Fees") if "Fees" in row else FiatValue(0, Currency.ZLOTY) + except (ValueError, KeyError): + logger.debug(f"Could not parse fees for row: {row}, defaulting to 0") + fees = FiatValue(0, Currency.ZLOTY) + transaction = Transaction( asset=cls._crypto_value(row), fiat_value=cls._fiat_value(row), action=cls._action(row), date=cls._datetime(row), + fees=fees ) logger.info(f"Parsed transaction: {transaction}") return transaction @@ -34,11 +43,11 @@ def _crypto_value(cls, row: dict) -> AssetValue: return AssetValue(float(amount), currency) @classmethod - def _fiat_value(cls, row: dict) -> FiatValue: - if row["Value"] == "": + def _fiat_value(cls, row: dict, column: str="Value") -> FiatValue: + if row[column] == "": return FiatValue(0, Currency.ZLOTY) - value = row["Value"].replace(",", "") + value = row[column].replace(",", "") amount_match = re.search(r'\d+\.?\d{2}', value) if not amount_match: