diff options
| -rw-r--r-- | bla.py | 22 | ||||
| -rw-r--r-- | kraken.py | 22 | ||||
| -rw-r--r-- | test_kraken.py | 138 |
3 files changed, 162 insertions, 20 deletions
@@ -1,11 +1,11 @@ import csv -from collections import defaultdict 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 trade import Trade from trade_queue import FIFOQueue @@ -51,24 +51,6 @@ def generate_report(sale_entries, proceeds: float | Decimal, crypto_asset, date_ return report -def parse_kraken_row(row: dict) -> LedgerAction: - date = row["time"].split(" ")[0] - return LedgerAction( - type=row["type"], - asset=row["asset"], - amount=Decimal(row["amount"]), - fee=Decimal(row.get("fee", "0")), - refid=row.get("refid", ""), - date=date, - ) - - -def read_kraken_ledger(csv_path: str) -> List[LedgerAction]: - with open(csv_path, "r") as file: - reader = csv.DictReader(file) - return list(map(parse_kraken_row, reader)) - - fifo_queues: Dict[str, FIFOQueue] = {} # Separate FIFO queue per cryptocurrency report = [] @@ -167,5 +149,5 @@ logging.basicConfig(level=logging.DEBUG) # Usage ledger_path = "kraken_ledger.csv" # Replace with your ledger file path output_path = "tax_report.csv" # Replace with your desired output file path -actions = read_kraken_ledger(ledger_path) +actions = read_ledger(ledger_path) process_ledger(actions, output_path) diff --git a/kraken.py b/kraken.py new file mode 100644 index 0000000..c420d89 --- /dev/null +++ b/kraken.py @@ -0,0 +1,22 @@ +import csv +from decimal import Decimal +from typing import List +from ledger_action import LedgerAction + + +def parse_row(row: dict[str,str]) -> LedgerAction: + date = row["time"].split(" ")[0] + return LedgerAction( + type=row["type"], + asset=row["asset"], + amount=Decimal(row["amount"]), + fee=Decimal(row.get("fee", "0")), + refid=row.get("refid", ""), + date=date, + ) + + +def read_ledger(csv_path: str) -> List[LedgerAction]: + with open(csv_path, "r") as file: + reader = csv.DictReader(file) + return list(map(parse_row, reader)) diff --git a/test_kraken.py b/test_kraken.py new file mode 100644 index 0000000..7b3776a --- /dev/null +++ b/test_kraken.py @@ -0,0 +1,138 @@ +import os +import tempfile +import unittest +from decimal import Decimal +from unittest.mock import mock_open, patch + +from kraken import parse_row, read_ledger +from ledger_action import LedgerAction + + +class TestKrakenFunctions(unittest.TestCase): + def test_parse_kraken_row_valid_input(self): + row = { + "type": "deposit", + "asset": "BTC", + "amount": "0.5", + "fee": "0.001", + "time": "2025-04-16 12:00:00", + "refid": "12345", + } + expected = LedgerAction( + type="deposit", + asset="BTC", + amount=Decimal("0.5"), + fee=Decimal("0.001"), + refid="12345", + date="2025-04-16", + ) + self.assertEqual(parse_row(row), expected) + + def test_parse_kraken_row_missing_fields(self): + row = { + "type": "trade", + "asset": "ETH", + "amount": "2.0", + "time": "2025-04-16 15:00:00", + } + expected = LedgerAction( + type="trade", + asset="ETH", + amount=Decimal("2.0"), + fee=Decimal("0"), # Default fee + refid="", # Default refid + date="2025-04-16", + ) + self.assertEqual(parse_row(row), expected) + + @patch( + "builtins.open", + new_callable=mock_open, + read_data='"txid","refid","time","type","subtype","aclass","asset","wallet","amount","fee","balance"\n"bla","67890","2025-04-16 09:00:00","trade","bla","currency","BTC","main","1.5","0.01"\n', + ) + def test_read_kraken_ledger(self, mock_file): + expected = [ + LedgerAction( + type="trade", + asset="BTC", + amount=Decimal("1.5"), + fee=Decimal("0.01"), + refid="67890", + date="2025-04-16", + ) + ] + self.assertEqual(read_ledger("dummy_path.csv"), expected) + + +class TestKrakenFunctionsRealFiles(unittest.TestCase): + + def setUp(self): + # Create a temporary CSV file for tests + self.temp_file = tempfile.NamedTemporaryFile( + delete=False, mode="w", suffix=".csv" + ) + self.temp_file.write( + '"txid","refid","time","type","subtype","aclass","asset","wallet","amount","fee","balance"\n' + '"","","2024-07-01 00:00:00","deposit","","currency","EUR","spot / main",1000.0000,0,1000.0000\n' + '"","d1e57f","2024-07-12 16:26:22","trade","tradespot","currency","EUR","spot / main",-130.4204,0,0.0000\n' + '"","d1e57f","2024-07-12 16:26:22","trade","tradespot","currency","DOGE","spot / main",1321.95097670,5.28780391,1316.66317279\n' + '"","","2024-07-12 16:36:49","withdrawal","","currency","DOGE","spot / main",-1312.66317279,4.00000000,0.00000000\n' + '"","","2024-08-02 14:24:30","deposit","","currency","EUR","spot / main",100.0000,0,100.0000\n' + ) + self.temp_file.close() # Close the file so it can be read later + + def tearDown(self): + # Remove the temporary file after tests are done + os.unlink(self.temp_file.name) + + def test_read_kraken_ledger_with_real_file(self): + # Define the expected result + expected = [ + LedgerAction( + type="deposit", + asset="EUR", + amount=Decimal("1000.0000"), + fee=Decimal("0"), + refid="", + date="2024-07-01", + ), + LedgerAction( + type="trade", + asset="EUR", + amount=Decimal("-130.4204"), + fee=Decimal("0"), + refid="d1e57f", + date="2024-07-12", + ), + LedgerAction( + type="trade", + asset="DOGE", + amount=Decimal("1321.95097670"), + fee=Decimal("5.28780391"), + refid="d1e57f", + date="2024-07-12", + ), + LedgerAction( + type="withdrawal", + asset="DOGE", + amount=Decimal("-1312.66317279"), + fee=Decimal("4.00000000"), + refid="", + date="2024-07-12", + ), + LedgerAction( + type="deposit", + asset="EUR", + amount=Decimal("100.0000"), + fee=Decimal("0"), + refid="", + date="2024-08-02", + ), + ] + + # Test the function with the temporary file + self.assertEqual(read_ledger(self.temp_file.name), expected) + + +if __name__ == "__main__": + unittest.main() |