Work with the new IBKR 2022 data format (#243)

* Work with the new IBKR 2022 data format

Author: oittaa

ibkr_report/definitions.py

@@ -3,7 +3,7 @@
 import os
 from dataclasses import dataclass
 from decimal import Decimal
-from enum import Enum, IntEnum, unique
+from enum import Enum, unique
 from typing import Dict
 
 
@@ -41,19 +41,6 @@ def _strtobool(val: str) -> bool:
 MAX_HTTP_RETRIES = 5
 SAVED_RATES_FILE = "official_ecb_exchange_rates-{0}.json.xz"
 
-_SINGLE_ACCOUNT = (
-    "Trades,Header,DataDiscriminator,Asset Category,Currency,Symbol,Date/Time,Exchange,"
-    "Quantity,T. Price,Proceeds,Comm/Fee,Basis,Realized P/L,Code"
-).split(",")
-_MULTI_ACCOUNT = (
-    "Trades,Header,DataDiscriminator,Asset Category,Currency,Account,Symbol,Date/Time,Exchange,"
-    "Quantity,T. Price,Proceeds,Comm/Fee,Basis,Realized P/L,Code"
-).split(",")
-OFFSET_DICT = {
-    tuple(_SINGLE_ACCOUNT): 0,
-    tuple(_MULTI_ACCOUNT): len(_MULTI_ACCOUNT) - len(_SINGLE_ACCOUNT),
-}
-FIELD_COUNT = len(_SINGLE_ACCOUNT)
 DATE_FORMAT = "%Y-%m-%d"
 TIME_FORMAT = " %H:%M:%S"
 DATE_STR_FORMATS = (
@@ -70,24 +57,24 @@ class StrEnum(str, Enum):
 
 
 @unique
-class Field(IntEnum):
+class Field(StrEnum):
     """CSV indices."""
 
-    TRADES = 0
-    HEADER = 1
-    DATA_DISCRIMINATOR = 2
-    ASSET_CATEGORY = 3
-    CURRENCY = 4
-    SYMBOL = 5
-    DATE_TIME = 6
-    EXCHANGE = 7
-    QUANTITY = 8
-    TRANSACTION_PRICE = 9
-    PROCEEDS = 10
-    COMMISSION_AND_FEES = 11
-    BASIS = 12
-    REALIZED_PL = 13
-    CODE = 14
+    TRADES = "Trades"
+    HEADER = "Header"
+    DATA_DISCRIMINATOR = "DataDiscriminator"
+    ASSET_CATEGORY = "Asset Category"
+    CURRENCY = "Currency"
+    SYMBOL = "Symbol"
+    DATE_TIME = "Date/Time"
+    EXCHANGE = "Exchange"
+    QUANTITY = "Quantity"
+    TRANSACTION_PRICE = "T. Price"
+    PROCEEDS = "Proceeds"
+    COMMISSION_AND_FEES = "Comm/Fee"
+    BASIS = "Basis"
+    REALIZED_PL = "Realized P/L"
+    CODE = "Code"
 
 
 @unique
@@ -130,7 +117,7 @@ class ReportOptions:
 
     report_currency: str
     deemed_acquisition_cost: bool
-    offset: int
+    fields: dict
 
 
 @dataclass

ibkr_report/report.py

@@ -10,8 +10,6 @@
 
 from ibkr_report.definitions import (
     CURRENCY,
-    FIELD_COUNT,
-    OFFSET_DICT,
     USE_DEEMED_ACQUISITION_COST,
     AssetCategory,
     DataDiscriminator,
@@ -67,7 +65,7 @@ def __init__(
         self.options = ReportOptions(
             report_currency=report_currency.upper(),
             deemed_acquisition_cost=use_deemed_acquisition_cost,
-            offset=0,
+            fields={},
         )
         self.rates = ExchangeRates()
         if file:
@@ -85,32 +83,47 @@ def add_trades(self, file: Iterable[bytes]) -> None:
     def is_stock_or_options_trade(self, items: Tuple[str, ...]) -> bool:
         """Checks whether the current row is part of a trade or not."""
         if (
-            len(items) == FIELD_COUNT + self.options.offset
-            and items[Field.TRADES] == FieldValue.TRADES
-            and items[Field.HEADER] == FieldValue.HEADER
-            and items[Field.DATA_DISCRIMINATOR]
+            all(
+                item in self.options.fields
+                for item in [
+                    Field.TRADES,
+                    Field.HEADER,
+                    Field.DATA_DISCRIMINATOR,
+                    Field.ASSET_CATEGORY,
+                ]
+            )
+            and items[self.options.fields[Field.TRADES]] == FieldValue.TRADES
+            and items[self.options.fields[Field.HEADER]] == FieldValue.HEADER
+            and items[self.options.fields[Field.DATA_DISCRIMINATOR]]
             in (DataDiscriminator.TRADE, DataDiscriminator.CLOSED_LOT)
-            and items[Field.ASSET_CATEGORY]
+            and items[self.options.fields[Field.ASSET_CATEGORY]]
             in (AssetCategory.STOCKS, AssetCategory.OPTIONS)
         ):
             return True
         return False
 
     def _handle_one_line(self, items: Tuple[str, ...]) -> None:
-        offset = OFFSET_DICT.get(items)
-        if offset is not None:
-            self.options.offset = offset
+        if len(items) > 2 and items[0] == Field.TRADES and items[1] == Field.HEADER:
+            self.options.fields = {}
             self._trade = None
+            for index, item in enumerate(items):
+                self.options.fields[item] = index
             return
         if self.is_stock_or_options_trade(items):
             self._handle_trade(items)
 
     def _handle_trade(self, items: Tuple[str, ...]) -> None:
         """Parses prices, gains, and losses from trades."""
-        if items[Field.DATA_DISCRIMINATOR] == DataDiscriminator.TRADE:
+        if (
+            items[self.options.fields[Field.DATA_DISCRIMINATOR]]
+            == DataDiscriminator.TRADE
+        ):
             self._trade = Trade(items, self.options, self.rates)
             self.prices += self._trade.total_selling_price
-        if items[Field.DATA_DISCRIMINATOR] == DataDiscriminator.CLOSED_LOT:
+        if (
+            items[self.options.fields[Field.DATA_DISCRIMINATOR]]
+            == DataDiscriminator.CLOSED_LOT
+        ):
             if not self._trade:
                 raise ValueError("Tried to close a lot without trades.")
             details = self._trade.details_from_closed_lot(items)

ibkr_report/trade.py

@@ -40,12 +40,12 @@ def __init__(
         self.rates = rates
         self.data = self._row_data(items)
 
-        fee = decimal_cleanup(items[Field.COMMISSION_AND_FEES + self.options.offset])
+        fee = decimal_cleanup(items[self.options.fields[Field.COMMISSION_AND_FEES]])
         self.fee = fee / self.data.rate
 
         # Sold stocks have a negative value in the "Quantity" column
         if self.data.quantity < Decimal(0):
-            proceeds = decimal_cleanup(items[Field.PROCEEDS + self.options.offset])
+            proceeds = decimal_cleanup(items[self.options.fields[Field.PROCEEDS]])
             self.total_selling_price = proceeds / self.data.rate
         log.debug(
             'Trade: "%s" "%s" %.2f',
@@ -70,7 +70,11 @@ def details_from_closed_lot(self, items: Tuple[str, ...]) -> TradeDetails:
             unit_sell_price, unit_buy_price = unit_buy_price, unit_sell_price
 
         # One option represents 100 shares of the underlying stock
-        multiplier = 100 if items[Field.ASSET_CATEGORY] == AssetCategory.OPTIONS else 1
+        multiplier = (
+            100
+            if items[self.options.fields[Field.ASSET_CATEGORY]] == AssetCategory.OPTIONS
+            else 1
+        )
         lot_sell_price = abs(lot_data.quantity) * unit_sell_price * multiplier
         lot_buy_price = abs(lot_data.quantity) * unit_buy_price * multiplier
         lot_fee = lot_data.quantity * self.fee / self.data.quantity
@@ -102,16 +106,16 @@ def details_from_closed_lot(self, items: Tuple[str, ...]) -> TradeDetails:
         )
 
     def _row_data(self, items: Tuple[str, ...]) -> RowData:
-        symbol = items[Field.SYMBOL + self.options.offset]
-        date_str = items[Field.DATE_TIME + self.options.offset]
+        symbol = items[self.options.fields[Field.SYMBOL]]
+        date_str = items[self.options.fields[Field.DATE_TIME]]
         rate = self.rates.get_rate(
             currency_from=self.options.report_currency,
-            currency_to=items[Field.CURRENCY],
+            currency_to=items[self.options.fields[Field.CURRENCY]],
             date_str=date_str,
         )
-        original_price_per_share = items[Field.TRANSACTION_PRICE + self.options.offset]
+        original_price_per_share = items[self.options.fields[Field.TRANSACTION_PRICE]]
         price_per_share = decimal_cleanup(original_price_per_share) / rate
-        quantity = decimal_cleanup(items[Field.QUANTITY + self.options.offset])
+        quantity = decimal_cleanup(items[self.options.fields[Field.QUANTITY]])
         return RowData(symbol, date_str, rate, price_per_share, quantity)
 
     def _validate_lot(self, lot_data: RowData) -> None:

tests/test-data/data_single_account_2022.csv

@@ -0,0 +1,5 @@
+Trades,Header,DataDiscriminator,Asset Category,Currency,Symbol,Date/Time,Exchange,Quantity,T. Price,C. Price,Proceeds,Comm/Fee,Basis,Realized P/L,MTM P/L,Code
+Trades,Data,Order,Equity and Index Options,USD,SPY 18MAR22 440.0 P,"2022-03-01, 11:02:15",-,-2,14.66,16.7975,2932,-1.4176132,-2483.2827,447.299686,-427.5,C
+Trades,Data,Trade,Equity and Index Options,USD,SPY 18MAR22 440.0 P,"2022-03-01, 11:02:15",MERCURY,-2,14.66,16.7975,2932,-1.4176132,-2483.2827,447.299686,-427.5,C
+Trades,Data,ClosedLot,Equity and Index Options,USD,SPY 18MAR22 440.0 P,2021-12-14,,2,12.4164135,,,,2483.2827,447.299686,,ST
+Trades,SubTotal,,Equity and Index Options,USD,SPY 18MAR22 440.0 P,,,-2,,,2932,-1.4176132,-2483.2827,447.299686,-427.5,

tests/test-data/eurofxref-hist.zip

@@ -39,6 +39,14 @@ def test_report_currency_eur_lowercase(self):
         self.assertEqual(round(report.gains, 2), Decimal("5964.76"))
         self.assertEqual(round(report.losses, 2), Decimal("0.00"))
 
+    def test_report_ibkr_2022_format(self):
+        report = Report()
+        with open("tests/test-data/data_single_account_2022.csv", "rb") as file:
+            report.add_trades(file)
+        self.assertEqual(round(report.prices, 2), Decimal("2626.77"))
+        self.assertEqual(round(report.gains, 2), Decimal("429.65"))
+        self.assertEqual(round(report.losses, 2), Decimal("0.00"))
+
 
 if __name__ == "__main__":
     unittest.main()