transitive_prices.py

from contextlib import contextmanager
import re
import traceback

from beancount.core.data import *
from beancount.core.prices import get_price, build_price_map
from beancount.plugins.implicit_prices import ImplicitPriceError as ImplicitPriceError_t

"""Beancount plugin to make certain prices transitive

EXAMPLE USAGE: display worth statistics in AU despite most underlying data being in USD

    option "operating_currency" "AU"
    include "lmba_gold-pm.beancount.txt"
    plugin "beancount.plugins.implicit_prices"
    plugin "_local.transitive_prices" "USD:AU"
"""

__plugins__ = ['transitive_prices']

def transitive_prices(entries, options_map, arg):
	errors = []
	new_entries = list(_main(entries, options_map, arg, errors_mut=errors))

	return new_entries, errors

def _main(entries, options_map, arg, *, errors_mut):
	m = re.fullmatch(r'(\w+):(\w+)', arg)
	if m:
		commodity2 = m.group(1)
		operating_currencies = [m.group(2)]
	else:
		raise ValueError(arg)

	orig_price_map = build_price_map(entries)

	for entry in entries:
		yield entry

		if (
			isinstance(entry, Price)
			and entry.currency not in operating_currencies
			and entry.amount.currency == commodity2
			# FIXME need to make this idempotent by adding extra check here
		):
			commodity1 = entry.currency
			for commodity3 in operating_currencies:
				with _beancount_catching(entry, errors_mut, ImplicitPriceError_t):
					conv_1_2 = entry.amount.number

					conv_2_3 = get_price(orig_price_map, (commodity2, commodity3), entry.date)[1]
					if conv_2_3 is not None:
						conv_1_3 = conv_1_2 * conv_2_3
					else:
						conv_3_2 = get_price(orig_price_map, (commodity3, commodity2), entry.date)[1]
						if conv_2_3 is not None:
							conv_1_3 = conv_1_2 / conv_3_2
						else:
							raise ValueError(f"Unable to look up {commodity1} in terms of {commodity3} because there is no intermediate entry putting {commodity2} in terms of {commodity3}")

					yield Price(
						meta=entry.meta,
						date=entry.date,
						amount=Amount(currency=commodity3, number=conv_1_3),
						currency=commodity1
					)

@contextmanager
def _beancount_catching(ctx_entry, errors_into, t_bean, *, t_exc=Exception):
	try:
		yield
	except t_exc as exc:
		message = "\n".join(re.sub(r"\n$", "", line) for line in traceback.format_exception(exc))
		errors_into.append(t_bean(ctx_entry.meta, message, ctx_entry))