Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
#!/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, {})
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment