Skip to content

Instantly share code, notes, and snippets.

@rainsunny
Forked from wzyboy/amortize_over.py
Created November 6, 2019 01:51
Show Gist options
  • Save rainsunny/55fb044836a9d7c0c7bbfe05cb8a4907 to your computer and use it in GitHub Desktop.
Save rainsunny/55fb044836a9d7c0c7bbfe05cb8a4907 to your computer and use it in GitHub Desktop.
amortize_over beancount plugin
# Copyright (c) 2017 Cary Kempston
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
from collections import namedtuple
from beancount.core.data import Transaction
from beancount.core.amount import Amount
from datetime import date
from dateutil.relativedelta import relativedelta
__plugins__ = ('amortize_over',)
AmortizationError = namedtuple('AmortizationError', 'source message entry')
def amortize_over(entries, unused_options_map):
"""Repeat a transaction based on metadata.
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 and a list of errors.
Example use:
This plugin will convert the following transactions
2017-06-01 * "Pay car insurance"
Assets:Bank:Checking -600.00 USD
Assets:Prepaid-Expenses
2017-06-01 * "Amortize car insurance over six months"
amortize_months: 3
Assets:Prepaid-Expenses -600.00 USD
Expenses:Insurance:Auto
into the following transactions over six months:
2017/06/01 * Pay car insurance
Assets:Bank:Checking -600.00 USD
Assets:Prepaid-Expenses 600.00 USD
2017/06/01 * Amortize car insurance over six months
Assets:Prepaid-Expenses -200.00 USD
Expenses:Insurance:Auto 200.00 USD
2017/07/01 * Amortize car insurance over six months
Assets:Prepaid-Expenses -200.00 USD
Expenses:Insurance:Auto 200.00 USD
2017/08/01 * Amortize car insurance over six months
Assets:Prepaid-Expenses -200.00 USD
Expenses:Insurance:Auto 200.00 USD
Note that transactions are not included past today's date. For example,
if the above transactions are processed on a date of 2017/07/25, the
transaction dated 2017/08/01 is not included.
"""
new_entries = []
errors = []
for entry in entries:
if isinstance(entry, Transaction) and 'amortize_months' in entry.meta:
a_entires, a_errors = amortize_transaction(entry)
new_entries.extend(a_entires)
errors.extend(a_errors)
else:
# Always replicate the existing entries - unless 'amortize_months'
# is in the metadata
new_entries.append(entry)
return new_entries, errors
def split_amount(amount, periods):
if periods == 1:
return [amount]
amount_this_period = amount / periods
amount_this_period = amount_this_period.quantize(amount)
return [amount_this_period] + split_amount(amount - amount_this_period, periods - 1)
def amortize_transaction(entry):
new_entries = []
errors = []
if len(entry.postings) != 2:
error = AmortizationError(
entry.meta,
'Amortized transactions must have exactly two postings.',
entry
)
errors.append(error)
return new_entries, errors
periods = entry.meta['amortize_months']
amount = abs(entry.postings[0].units.number)
currency = entry.postings[0].units.currency
monthly_amounts = split_amount(amount, periods)
for (n_month, monthly_number) in enumerate(monthly_amounts):
new_postings = []
for posting in entry.postings:
new_monthly_number = monthly_number
if posting.units.number < 0:
new_monthly_number = -monthly_number
new_posting = posting._replace(units=Amount(number=new_monthly_number,
currency=currency))
new_postings.append(new_posting)
new_narration = f'{entry.narration} ({n_month + 1}/{periods})'
new_entry = entry._replace(
narration=new_narration,
postings=new_postings,
date=entry.date + relativedelta(months=n_month),
)
if new_entry.date <= date.today():
new_entries.append(new_entry)
return new_entries, errors
plugin "amortize_over"
2017-06-01 open Assets:Bank:Checking
2017-06-01 open Assets:Prepaid-Expenses
2017-06-01 open Expenses:Insurance:Auto
2017-06-01 * "Pay car insurance"
Assets:Bank:Checking -600.00 USD
Assets:Prepaid-Expenses
2017-06-01 * "Amortize car insurance over six months"
amortize_months: 6
Assets:Prepaid-Expenses -600.00 USD
Expenses:Insurance:Auto
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment