From a1c2a999e449054d6641bbb633954e45fcd63f90 Mon Sep 17 00:00:00 2001 From: Kristóf Marussy Date: Mon, 25 Jan 2021 01:14:28 +0100 Subject: Add plugins and importers from private config The importers are missing tests, because not having any specifications for the import formats means we must use real, private data as test inputs --- .../plugins/selective_implicit_prices.py | 157 +++++++++++++++++++++ 1 file changed, 157 insertions(+) create mode 100644 beancount_extras_kris7t/plugins/selective_implicit_prices.py (limited to 'beancount_extras_kris7t/plugins/selective_implicit_prices.py') diff --git a/beancount_extras_kris7t/plugins/selective_implicit_prices.py b/beancount_extras_kris7t/plugins/selective_implicit_prices.py new file mode 100644 index 0000000..07dc893 --- /dev/null +++ b/beancount_extras_kris7t/plugins/selective_implicit_prices.py @@ -0,0 +1,157 @@ +"""This plugin synthesizes Price directives for all Postings with a price or +directive or if it is an augmenting posting, has a cost directive. + +Price directives will be synthesized only for commodities with the +implicit-prices: TRUE metadata set. +""" +__copyright__ = "Copyright (C) 2015-2017 Martin Blais, " + \ + "2020 Kristóf Marussy " +__license__ = "GNU GPLv2" + +import collections +from typing import List, Tuple, Set + +from beancount.core.data import Commodity, Entries, Transaction +from beancount.core import data +from beancount.core import amount +from beancount.core import inventory +from beancount.core.position import Cost + +__plugins__ = ('add_implicit_prices',) + + +ImplicitPriceError = collections.namedtuple('ImplicitPriceError', 'source message entry') + + +METADATA_FIELD = "__implicit_prices__" +IMPLICIT_PRICES_META = "implicit-prices" + + +def fetch_commodities(entries: Entries) -> Tuple[Set[str], List[ImplicitPriceError]]: + commodities: Set[str] = set() + errors: List[ImplicitPriceError] = [] + for entry in entries: + if isinstance(entry, Commodity): + implicit_prices = entry.meta.get(IMPLICIT_PRICES_META, False) + if not isinstance(implicit_prices, bool): + errors.append(ImplicitPriceError( + entry.meta, + f'{IMPLICIT_PRICES_META} must be Boolean, got {implicit_prices} instead', + entry)) + if implicit_prices: + commodities.add(entry.currency) + return commodities, errors + + +def add_implicit_prices(entries: Entries, + unused_options_map) -> Tuple[Entries, List[ImplicitPriceError]]: + """Insert implicitly defined prices from Transactions. + + Explicit price entries are simply maintained in the output list. Prices from + postings with costs or with prices from Transaction entries are synthesized + as new Price entries in the list of entries output. + + Args: + entries: A list of directives. We're interested only in the Transaction instances. + unused_options_map: A parser options dict. + Returns: + A list of entries, possibly with more Price entries than before, and a + list of errors. + """ + new_entries: Entries = [] + errors: List[ImplicitPriceError] = [] + + commodities, fetch_errors = fetch_commodities(entries) + errors.extend(fetch_errors) + + # A dict of (date, currency, cost-currency) to price entry. + new_price_entry_map = {} + + balances = collections.defaultdict(inventory.Inventory) + for entry in entries: + # Always replicate the existing entries. + new_entries.append(entry) + + if isinstance(entry, Transaction): + # Inspect all the postings in the transaction. + for posting in entry.postings: + units = posting.units + if units.currency not in commodities: + continue + + cost = posting.cost + + # Check if the position is matching against an existing + # position. + _, booking = balances[posting.account].add_position(posting) + + # Add prices when they're explicitly specified on a posting. An + # explicitly specified price may occur in a conversion, e.g. + # Assets:Account 100 USD @ 1.10 CAD + # or, if a cost is also specified, as the current price of the + # underlying instrument, e.g. + # Assets:Account 100 HOOL {564.20} @ {581.97} USD + if posting.price is not None: + meta = data.new_metadata(entry.meta["filename"], entry.meta["lineno"]) + meta[METADATA_FIELD] = "from_price" + price_entry = data.Price(meta, entry.date, + units.currency, + posting.price) + + # Add costs, when we're not matching against an existing + # position. This happens when we're just specifying the cost, + # e.g. + # Assets:Account 100 HOOL {564.20} + elif (cost is not None and + booking != inventory.MatchResult.REDUCED): + # TODO(blais): What happens here if the account has no + # booking strategy? Do we end up inserting a price for the + # reducing leg? Check. + meta = data.new_metadata(entry.meta["filename"], entry.meta["lineno"]) + meta[METADATA_FIELD] = "from_cost" + if isinstance(cost, Cost): + price_entry = data.Price(meta, entry.date, + units.currency, + amount.Amount(cost.number, cost.currency)) + else: + errors.append( + ImplicitPriceError( + entry.meta, + f'Expected {entry} to have a Cost, got {cost} instead', + entry)) + price_entry = None + else: + price_entry = None + + if price_entry is not None: + key = (price_entry.date, + price_entry.currency, + price_entry.amount.number, # Ideally should be removed. + price_entry.amount.currency) + try: + new_price_entry_map[key] + + # Do not fail for now. We still have many valid use + # cases of duplicate prices on the same date, for + # example, stock splits, or trades on two dates with + # two separate reported prices. We need to figure out a + # more elegant solution for this in the long term. + # Keeping both for now. We should ideally not use the + # number in the de-dup key above. + # + # dup_entry = new_price_entry_map[key] + # if price_entry.amount.number == dup_entry.amount.number: + # # Skip duplicates. + # continue + # else: + # errors.append( + # ImplicitPriceError( + # entry.meta, + # "Duplicate prices for {} on {}".format(entry, + # dup_entry), + # entry)) + except KeyError: + new_price_entry_map[key] = price_entry + new_entries.append(price_entry) + + return new_entries, errors -- cgit v1.2.3-54-g00ecf