Merge pull request #114 from tarioch/feature/fidelity_netbenefits_imp…

…orter

Initial version of an importer for netbenefits trx

.pre-commit-config.yaml

@@ -2,7 +2,7 @@ exclude: '^docs/conf.py'
 
 repos:
 - repo: https://github.com/pre-commit/pre-commit-hooks
-  rev: v4.5.0
+  rev: v4.6.0
   hooks:
   - id: trailing-whitespace
   - id: check-added-large-files
@@ -23,19 +23,19 @@ repos:
   - id: isort
 
 - repo: https://github.com/psf/black
-  rev: 23.12.1
+  rev: 24.4.2
   hooks:
   - id: black
     language_version: python3
 
 - repo: https://github.com/asottile/blacken-docs
-  rev: 1.16.0
+  rev: 1.18.0
   hooks:
   - id: blacken-docs
     additional_dependencies: [black]
 
 - repo: https://github.com/PyCQA/flake8
-  rev: 7.0.0
+  rev: 7.1.0
   hooks:
   - id: flake8
     additional_dependencies: [flake8-print]
@@ -49,7 +49,7 @@ repos:
     files: README.rst
 
 - repo: https://github.com/pre-commit/mirrors-mypy
-  rev: v1.8.0
+  rev: v1.10.1
   hooks:
   -   id: mypy
       args: [--install-types, --non-interactive, --ignore-missing-imports]

docs/importers.rst

@@ -402,10 +402,33 @@ Import mt940 from `BCGE <https://www.bcge.ch/>`__
 Swisscard cards
 ---------------
 
-Import Swisscard's `Cashback Cards <https://www.cashback-cards.ch/>` transactions from a CSV export.__
+Import Swisscard's `Cashback Cards <https://www.cashback-cards.ch/>` transactions from a CSV export.
 
 .. code-block:: python
 
   from tariochbctools.importers.swisscard import importer as swisscard
 
   CONFIG = [swisscard.SwisscardImporter("swisscard/.*\.csv", "Liabilities:Cashback")]
+
+Fidelity Netbenefits
+--------------------
+
+Import Fidelity Netbenefits `<https://netbenefits.fidelity.com/>` transactions from a CSV export of the activities.
+
+.. code-block:: python
+
+  from tariochbctools.importers.netbenefits import importer as netbenefits
+
+  CONFIG = [
+      netbenefits.Importer(
+          regexps="Transaction history\.csv",
+          cashAccount="Assets:Netbenefits:USD",
+          investmentAccount="Assets:Netbenefits:SYMBOL",
+          dividendAccount="Income:Interest",
+          taxAccount="Expenses:Tax",
+          capGainAccount="Income:Capitalgain",
+          symbol="SYMBOL",
+          ignoreTypes=["REINVESTMENT REINVEST @ $1.000"],
+          baseCcy="CHF",
+      )
+  ]

setup.py

@@ -6,6 +6,7 @@
     PyScaffold helps you to put up the scaffold of your new Python project.
     Learn more under: https://pyscaffold.org/
 """
+
 from setuptools import setup
 
 if __name__ == "__main__":

src/tariochbctools/importers/general/priceLookup.py

@@ -1,16 +1,25 @@
 from datetime import date
 
 from beancount.core import amount, prices
+from beancount.core.number import D
 
 
 class PriceLookup:
     def __init__(self, existing_entries, baseCcy: str):
-        self.priceMap = prices.build_price_map(existing_entries)
+        if existing_entries:
+            self.priceMap = prices.build_price_map(existing_entries)
+        else:
+            self.priceMap = None
         self.baseCcy = baseCcy
 
     def fetchPriceAmount(self, instrument: str, date: date):
-        price = prices.get_price(self.priceMap, tuple([instrument, self.baseCcy]), date)
-        return price[1]
+        if self.priceMap:
+            price = prices.get_price(
+                self.priceMap, tuple([instrument, self.baseCcy]), date
+            )
+            return price[1]
+        else:
+            return D(1)
 
     def fetchPrice(self, instrument: str, date: date):
         if instrument == self.baseCcy:

src/tariochbctools/importers/netbenefits/importer.py

@@ -0,0 +1,173 @@
+import csv
+from collections.abc import Iterable
+from datetime import date
+from io import StringIO
+
+from beancount.core import amount, data
+from beancount.core.number import D
+from beancount.core.position import CostSpec
+from beancount.ingest import importer
+from beancount.ingest.importers.mixins import identifier
+from dateutil.parser import parse
+
+from tariochbctools.importers.general.priceLookup import PriceLookup
+
+
+class Importer(identifier.IdentifyMixin, importer.ImporterProtocol):
+    """An importer for Fidelity Netbenefits Activity CSV files."""
+
+    def __init__(
+        self,
+        regexps: str | Iterable[str],
+        cashAccount: str,
+        investmentAccount: str,
+        dividendAccount: str,
+        taxAccount: str,
+        capGainAccount: str,
+        symbol: str,
+        ignoreTypes: Iterable[str],
+        baseCcy: str,
+    ):
+        identifier.IdentifyMixin.__init__(self, matchers=[("filename", regexps)])
+        self.cashAccount = cashAccount
+        self.investmentAccount = investmentAccount
+        self.dividendAccount = dividendAccount
+        self.taxAccount = taxAccount
+        self.capGainAccount = capGainAccount
+        self.symbol = symbol
+        self.ignoreTypes = ignoreTypes
+        self.baseCcy = baseCcy
+
+    def name(self):
+        return super().name() + self.cashAccount
+
+    def file_account(self, file):
+        return self.cashAccount
+
+    def extract(self, file, existing_entries):
+        entries = []
+
+        self.priceLookup = PriceLookup(existing_entries, self.baseCcy)
+
+        with StringIO(file.contents()) as csvfile:
+            reader = csv.DictReader(
+                csvfile,
+                [
+                    "Transaction date",
+                    "Transaction type",
+                    "Investment name",
+                    "Shares",
+                    "Amount",
+                ],
+                delimiter=",",
+                skipinitialspace=True,
+            )
+            next(reader)
+            for row in reader:
+                if not row["Transaction type"]:
+                    break
+
+                if row["Transaction type"] in self.ignoreTypes:
+                    continue
+
+                book_date = parse(row["Transaction date"].strip()).date()
+                amt = amount.Amount(D(row["Amount"].replace("$", "")), "USD")
+                shares = None
+                if row["Shares"] != "-":
+                    shares = amount.Amount(D(row["Shares"]), self.symbol)
+
+                metakv = {}
+
+                if not amt and not shares:
+                    continue
+
+                meta = data.new_metadata(file.name, 0, metakv)
+                description = row["Transaction type"].strip()
+
+                if "TAX" in description:
+                    postings = self.__createDividend(amt, book_date, self.taxAccount)
+                elif "DIVIDEND" in description:
+                    postings = self.__createDividend(
+                        amt, book_date, self.dividendAccount
+                    )
+                elif "YOU BOUGHT" in description:
+                    postings = self.__createBuy(amt, shares, book_date)
+                elif "YOU SOLD" in description:
+                    postings = self.__createSell(amt, shares, book_date)
+                else:
+                    postings = [
+                        data.Posting(self.cashAccount, amt, None, None, None, None),
+                    ]
+
+                    if shares is not None:
+                        postings.append(
+                            data.Posting(
+                                self.investmentAccount, shares, None, None, None, None
+                            ),
+                        )
+
+                entry = data.Transaction(
+                    meta,
+                    book_date,
+                    "*",
+                    "",
+                    description,
+                    data.EMPTY_SET,
+                    data.EMPTY_SET,
+                    postings,
+                )
+                entries.append(entry)
+
+        return entries
+
+    def __createBuy(self, amt: amount, shares: amount, book_date: date):
+        price = self.priceLookup.fetchPrice("USD", book_date)
+        cost = CostSpec(
+            number_per=None,
+            number_total=round(-amt.number * price.number, 2),
+            currency=self.baseCcy,
+            date=None,
+            label=None,
+            merge=None,
+        )
+        postings = [
+            data.Posting(self.investmentAccount, shares, cost, None, None, None),
+            data.Posting(self.cashAccount, amt, None, price, None, None),
+        ]
+
+        return postings
+
+    def __createSell(self, amt: amount, shares: amount, book_date: date):
+        price = self.priceLookup.fetchPrice("USD", book_date)
+        cost = CostSpec(
+            number_per=None,
+            number_total=None,
+            currency=None,
+            date=None,
+            label=None,
+            merge=None,
+        )
+        postings = [
+            data.Posting(self.investmentAccount, shares, cost, None, None, None),
+            data.Posting(self.cashAccount, amt, None, price, None, None),
+            data.Posting(self.capGainAccount, None, None, None, None, None),
+        ]
+
+        return postings
+
+    def __createDividend(self, amt: amount, book_date: date, incomeAccount: str):
+        price = self.priceLookup.fetchPrice("USD", book_date)
+        postings = [
+            data.Posting(
+                self.investmentAccount,
+                amount.Amount(D(0), self.symbol),
+                None,
+                None,
+                None,
+                None,
+            ),
+            data.Posting(self.cashAccount, amt, None, price, None, None),
+            data.Posting(incomeAccount, None, None, None, None, None),
+        ]
+
+        return postings

src/tariochbctools/plugins/check_portfolio_sum.py

@@ -1,6 +1,7 @@
 """A plugin that verifies that on each transaction, all the "portfolios" have
 the same weight.
 """
+
 import collections
 from collections import defaultdict
 from decimal import Decimal

src/tariochbctools/plugins/generate_base_ccy_prices.py

@@ -1,6 +1,7 @@
 """A plugin that inserts an additional price to the base rate by applying
 fx rate to a price.
 """
+
 from beancount.core import amount, data, prices
 
 __plugins__ = ["generate"]