commodity_par.py

#!/usr/bin/env python3

# Copyright (C) 2021 Xidorn Quan
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.

__plugins__ = ['commodity_par']

from typing import List, Union

from beancount.core.data import Amount, Balance, Commodity, Cost, CostSpec, \
        Directive, Open, Posting, Transaction

def commodity_par(entries, options_map):
    """Replace commodities with par with their par value.

    Example:

    Given a commodity like:

        2020-06-30 commodity EDR.PT
            name: "Everyday Rewards Point"
            par: 0.005 AUD

    This plugin will convert the following entries

        2020-06-30 open Assets:EverydayRewards:Points       EDR.PT
        2020-06-30 open Income:Woolworths:Points:Bonus      EDR.PT

        2020-08-04 * "Got bonus points from Woolworths"
            Assets:EverydayRewards:Points             950 EDR.PT
            Income:Woolworths:Points:Bonus

    into the following entries

        2020-06-30 open Assets:EverydayRewards:Points       AUD
        2020-06-30 open Income:Woolworths:Points:Bonus      AUD

        2020-08-04 * "Got bonus points from Woolworths"
            Assets:EverydayRewards:Points               4.75 AUD
            Income:Woolworths:Points:Bonus

    It will also check that transactions are valid before this change.
    """

    processor = CommodityParProcessor(entries)
    new_entries = processor.process_all()
    return new_entries, []

class CommodityParProcessor:
    def __init__(self, entries: List[Directive]):
        self.entries = entries
        self.account_currencies = {}
        self.pars = {
            entry.currency: entry.meta['par']
            for entry in entries
            if isinstance(entry, Commodity) and 'par' in entry.meta
        }

    def process_all(self):
        return [self._process_entry(entry) for entry in self.entries]

    def _process_entry(self, entry: Directive):
        if isinstance(entry, Open):
            return self._process_open(entry)
        if isinstance(entry, Transaction):
            return self._process_transaction(entry)
        if isinstance(entry, Balance):
            return self._process_balance(entry)
        return entry

    def _process_open(self, entry: Open):
        if not entry.currencies:
            return entry
        self.account_currencies[entry.account] = set(entry.currencies)
        currencies = [
            self.pars[currency].currency
            if currency in self.pars else currency
            for currency in entry.currencies
        ]
        return entry._replace(currencies=currencies)

    def _process_transaction(self, entry: Transaction):
        if not any(self._posting_needs_processing(p) for p in entry.postings):
            return entry
        postings = [
            self._process_posting(posting)
            for posting in entry.postings
        ]
        return entry._replace(postings=postings)

    def _posting_needs_processing(self, posting: Posting):
        return posting.units.currency in self.pars \
            or (posting.cost and posting.cost.currency in self.pars) \
            or (posting.price and posting.price.currency in self.pars)

    def _process_posting(self, posting: Posting):
        units = posting.units
        cost = posting.cost
        price = posting.price

        if posting.units.currency in self.pars:
            par = self.pars[units.currency]
            units = Amount(units.number * par.number, par.currency)
            if cost:
                if isinstance(cost, Cost):
                    number = cost.number / par.number
                    cost = cost._replace(number=number)
                elif isinstance(cost, CostSpec):
                    number_per = cost.number_per and \
                        cost.number_per / par.number
                    cost = cost._replace(number_per=number_per)
            if price:
                number = price.number / par.number
                price = price._replace(number=number)

        if cost and cost.currency in self.pars:
            par = self.pars[cost.currency]
            if isinstance(cost, Cost):
                number = cost.number * par.number
                return cost._replace(number=number, currency=par.currency)
            if isinstance(cost, CostSpec):
                number_per = cost.number_per and cost.number_per * par.number
                number_total = cost.number_total and cost.number_total * par.number
                return cost._replace(
                    number_per=number_per,
                    number_total=number_total,
                    currency=par.currency
                )
            raise TypeError(f"Unknown type {type(cost)} for cost")

        if price and price.currency in self.pars:
            par = self.pars[price.currency]
            price = Amount(price.number * par.number, par.currency)

        return posting._replace(units=units, cost=cost, price=price)

    def _process_balance(self, balance: Balance):
        amount = balance.amount
        if amount.currency not in self.pars:
            return balance
        par = self.pars[amount.currency]
        amount = Amount(amount.number * par.number, par.currency)
        return balance._replace(amount=amount)

if __name__ == '__main__':
    import sys
    from beancount import loader

    entries, errors, options = loader.load_file(sys.argv[1], log_errors=sys.stderr)
    commodity_par(entries, {})