ADD support for new coinbase CSV format (#172)

src/book.py

@@ -287,7 +287,7 @@ def _read_binance(self, file_path: Path, version: int = 1) -> None:
     def _read_binance_v2(self, file_path: Path) -> None:
         self._read_binance(file_path=file_path, version=2)
 
-    def _read_coinbase(self, file_path: Path) -> None:
+    def _read_coinbase(self, file_path: Path, version: int = 1) -> None:
         platform = "coinbase"
         operation_mapping = {
             "Receive": "Deposit",
@@ -302,18 +302,39 @@ def _read_coinbase(self, file_path: Path) -> None:
 
             # Skip header.
             try:
-                assert next(reader)  # header line
-                assert next(reader) == []
-                assert next(reader) == []
-                assert next(reader) == []
-                assert next(reader) == ["Transactions"]
-                assert next(reader)  # user row
-                assert next(reader) == []
+                if version == 4:
+                    assert next(reader) == []
+                    assert next(reader) == ["Transactions"]
+                    assert next(reader)  # user row
+                else:
+                    assert next(reader)  # header line
+                    assert next(reader) == []
+                    assert next(reader) == []
+                    assert next(reader) == []
+                    assert next(reader) == ["Transactions"]
+                    assert next(reader)  # user row
+                    assert next(reader) == []
 
                 fields = next(reader)
                 num_columns = len(fields)
-                # Coinbase export format from late 2021 and ongoing
-                if num_columns == 10:
+                # Coinbase export format from 2023/2024 and ongoing
+                if num_columns == 11:
+                    assert version == 4
+                    assert fields == [
+                        "ID",
+                        "Timestamp",
+                        "Transaction Type",
+                        "Asset",
+                        "Quantity Transacted",
+                        "Price Currency",
+                        "Price at Transaction",
+                        "Subtotal",
+                        "Total (inclusive of fees and/or spread)",
+                        "Fees and/or Spread",
+                        "Notes",
+                    ]
+                # Coinbase export format from late 2021 until 2023/2024
+                elif num_columns == 10:
                     assert fields == [
                         "Timestamp",
                         "Transaction Type",
@@ -367,8 +388,28 @@ def _read_coinbase(self, file_path: Path) -> None:
 
             for columns in reader:
 
-                # Coinbase export format from late 2021 and ongoing
-                if num_columns == 10:
+                # Coinbase export format from 2023/2024 and ongoing
+                if num_columns == 11:
+                    (
+                        _id,
+                        _utc_time,
+                        operation,
+                        coin,
+                        _change,
+                        _currency_spot,
+                        _eur_spot,
+                        _eur_subtotal,
+                        _eur_total,
+                        _eur_fee,
+                        remark,
+                    ) = columns
+                    _eur_spot = _eur_spot.replace("€", "")
+                    _eur_subtotal = _eur_subtotal.replace("€", "")
+                    _eur_total = _eur_total.replace("€", "")
+                    _eur_fee = _eur_fee.replace("€", "")
+
+                # Coinbase export format from late 2021 until 2023/2024
+                elif num_columns == 10:
                     (
                         _utc_time,
                         operation,
@@ -400,12 +441,22 @@ def _read_coinbase(self, file_path: Path) -> None:
                 row = reader.line_num
 
                 # Parse data.
-                utc_time = datetime.datetime.strptime(_utc_time, "%Y-%m-%dT%H:%M:%SZ")
+                if version == 4:
+                    utc_time = datetime.datetime.strptime(
+                        _utc_time, "%Y-%m-%d %H:%M:%S UTC"
+                    )
+                else:
+                    utc_time = datetime.datetime.strptime(
+                        _utc_time, "%Y-%m-%dT%H:%M:%SZ"
+                    )
                 utc_time = utc_time.replace(tzinfo=datetime.timezone.utc)
                 operation = operation_mapping.get(operation, operation)
                 change = misc.force_decimal(_change)
                 # `eur_subtotal` and `eur_fee` are None for withdrawals.
                 eur_subtotal = misc.xdecimal(_eur_subtotal)
+                if version == 4:
+                    change = abs(change)
+                    eur_subtotal = abs(eur_subtotal)
                 if eur_subtotal is None:
                     # Cost without fees from CSV is missing. This can happen for
                     # old transactions (<2018), event though something was bought.
@@ -442,6 +493,8 @@ def _read_coinbase(self, file_path: Path) -> None:
                     convert_coin = match.group("coin")
 
                     eur_total = misc.force_decimal(_eur_total)
+                    if version == 4:
+                        eur_total = abs(eur_total)
                     convert_eur_spot = eur_total / convert_change
 
                     self.append_operation(
@@ -502,10 +555,13 @@ def _read_coinbase(self, file_path: Path) -> None:
                     )
 
     def _read_coinbase_v2(self, file_path: Path) -> None:
-        self._read_coinbase(file_path=file_path)
+        self._read_coinbase(file_path=file_path, version=2)
 
     def _read_coinbase_v3(self, file_path: Path) -> None:
-        self._read_coinbase(file_path=file_path)
+        self._read_coinbase(file_path=file_path, version=3)
+
+    def _read_coinbase_v4(self, file_path: Path) -> None:
+        self._read_coinbase(file_path=file_path, version=4)
 
     def _read_coinbase_pro(self, file_path: Path) -> None:
         platform = "coinbase_pro"
@@ -1307,6 +1363,7 @@ def detect_exchange(self, file_path: Path) -> Optional[str]:
                 "coinbase": 1,
                 "coinbase_v2": 1,
                 "coinbase_v3": 1,
+                "coinbase_v4": 4,
                 "coinbase_pro": 1,
                 "kraken_ledgers_old": 1,
                 "kraken_ledgers": 1,
@@ -1355,6 +1412,19 @@ def detect_exchange(self, file_path: Path) -> Optional[str]:
                     "and Donations are taxable events. "
                     "For final tax obligations, please consult your tax advisor."
                 ],
+                "coinbase_v4": [
+                    "ID",
+                    "Timestamp",
+                    "Transaction Type",
+                    "Asset",
+                    "Quantity Transacted",
+                    "Price Currency",
+                    "Price at Transaction",
+                    "Subtotal",
+                    "Total (inclusive of fees and/or spread)",
+                    "Fees and/or Spread",
+                    "Notes",
+                ],
                 "coinbase_pro": [
                     "portfolio",
                     "trade id",