david/beancount-plugins
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Moved plugins to their own repo

Browse Source
This commit is contained in:
David Duran Rosich
2026-07-19 12:24:59 +02:00
commit a864345529
5 changed files with 526 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

151
joint_account.py Normal file
View File

@@ -0,0 +1,151 @@
#!/usr/bin/env python3
from beancount.core import data
from beancount.core import account
from beancount.core import getters
from beancount.core import amount
from beancount.core import account_types
from beancount.core import realization
from beancount.core.number import D
from beancount.parser import options
__plugins__ = ('joint_account_plugin',)
# DEBT_ACCOUNT = "Deute"
SHARE_ACCOUNT = "Part"
def debt_account_str(acc: str, debtor: str, accs: account_types.AccountTypes) -> str:
return account.join(accs.liabilities, account.sans_root(acc), debtor)
def share_account_str(acc: str, debtor: str) -> str:
# return ":".join((account, SHARE_ACCOUNT + "-" + debtor))
return acc # reimbursment in same account
def debt_account(posting: data.Posting, debtor: str, joint_accounts: dict, accs: account_types.AccountTypes) -> str:
if debtor in joint_accounts.get(posting.account, {}):
return debt_account_str(posting.account, debtor, accs)
if account_types.is_income_statement_account(posting.account, accs):
return share_account_str(posting.account, debtor)
return ""
def is_splitable(posting: data.Posting, accs: account_types.AccountTypes):
if posting.meta.get("joint", "") == "nosplit":
return False
return account_types.is_income_statement_account(posting.account, accs)
def divide_and_flip(posting: data.Posting, n: int = 2) -> data.Posting:
"""Return a new posting with n-th the amount and opposite sign."""
assert(n>0)
old_amount = posting.units
new_amount = amount.div(old_amount, D(-n))
return posting._replace(units=new_amount)
def add_tag(txn: data.Transaction, tag: str) -> data.Transaction:
return txn._replace(tags=txn.tags | set([tag]))
def prepare_posting(posting, user, joint_accounts, accs, n):
return divide_and_flip(posting._replace(account=debt_account(posting, user, joint_accounts, accs), flag=None, meta=None), n)
def split_among_users(postings, users, joint_accounts, accs):
new_postings = []
for user in users:
for posting in postings:
new_postings.append(prepare_posting(posting, user, joint_accounts, accs, 1 + len(users)))
return new_postings
def debt_transaction(txn: data.Transaction, joint_accounts: dict, new_accounts: set, accs: account_types.AccountTypes) -> None | data.Transaction:
# Only two cases are processed:
# a) all accounts are joint with the same people
# b) a single joint account with splitable/unsplitable accounts
# get how many joint accounts there are
user_sets = [(idx, set(joint_accounts[posting.account])) for idx, posting in enumerate(txn.postings) if posting.account in joint_accounts]
new_postings = []
# no joint account -> nothing to do
if len(user_sets) == 0:
return None
# single joint account, assume all expenses/income are split equally between users of the joint account
elif len(user_sets) == 1:
splitable_postings = [posting for posting in txn.postings if is_splitable(posting, accs)]
if len(splitable_postings) == 0:
return None
# as not every posting may be splittable, debt of joint account might need to be rebalanced
new_balance = realization.compute_postings_balance(splitable_postings)
joint_posting = txn.postings[user_sets[0][0]]
joint_posting = joint_posting._replace(units=-new_balance.get_currency_units(joint_posting.units.currency))
new_postings = splitable_postings + [joint_posting]
# all postings are from joint accounts, check they share same users and transfer debt equally
elif len(user_sets) == len(txn.postings):
# check all joint accounts share same users
total_set = set().union(*map(lambda s: s[1], user_sets))
if any((total_set != s[1] for s in user_sets)):
return None
new_postings = txn.postings
# just transfer debt equally
new_postings = split_among_users(new_postings, user_sets[0][1], joint_accounts, accs)
new_accounts |= set((p.account for p in new_postings))
return add_tag(txn._replace(postings=new_postings), "retorn")
def joint_account_plugin(entries, options_map):
errors = []
new_entries = []
new_accounts = set()
accs = options.get_account_types(options_map)
joint_accounts = {}
for entry in entries:
if isinstance(entry, data.Open):
if "joint" in entry.meta:
joint_accounts[entry.account] = entry.meta["joint"].strip().split(" ")
for entry in entries:
if isinstance(entry, data.Transaction):
if any((acc in joint_accounts for acc in getters.get_accounts([entry]))):
new_entry = debt_transaction(entry, joint_accounts, new_accounts, accs)
if new_entry is not None:
new_entries.append(new_entry)
new_entries.append(entry)
# add new_accounts at earliest date possible
earliest_date = getters.get_min_max_dates(new_entries)[0]
new_accounts -= set(getters.get_accounts((e for e in entries if isinstance(e, data.Open))))
for idx, acc in enumerate(new_accounts):
entry = data.Open(data.new_metadata("<joint_account>", idx), earliest_date, acc, None, None)
new_entries.append(entry)
# close associated accounts
for entry in entries:
if isinstance(entry, data.Close):
for user in joint_accounts.get(entry.account, []):
e = entry._replace(account=debt_account_str(entry.account, user, accs))
new_entries.append(e)
return new_entries, []
# def change_to_total_account(txn: data.Transaction, joint_accounts) -> data.Transaction:
# return txn._replace(postings=list(map(lambda p: change_to_total_account_posting(p, joint_accounts), txn.postings)))
# def change_to_total_account_posting(posting: data.Posting, joint_accounts) -> data.Posting:
# return posting._replace(account=total_account_str(posting.account) if posting.account in joint_accounts else posting.account)
# def transform_accounts(entry, joint_accounts):
# if hasattr(entry, "account"):
# if entry.account in joint_accounts:
# return entry._replace(account=total_account_str(entry.account))
# elif isinstance(entry, data.Transaction):
# return change_to_total_account(entry, joint_accounts)
# return entry

62
link_accounts.py Normal file
View File

@@ -0,0 +1,62 @@
#!/usr/bin/env python3
import re
from beancount.core import data
from beancount.core import inventory
from beancount.core import account
from beancount.core import amount
from beancount.core import account_types
from beancount.parser import parser
from beancount.core.number import D
from beancount.parser import options
import itertools as it
from collections import namedtuple
LinkAccount = namedtuple("LinkAccount", ["account", "tags"])
__plugins__ = ('link_account_plugin',)
def add_tags(txn: data.Transaction, *tags: str) -> data.Transaction:
return txn._replace(tags=txn.tags | set(tags))
def link_posting(p: data.Posting, link_accounts: dict[str, LinkAccount]) -> list[data.Posting]:
if p.account in link_accounts:
return [
p._replace(account=link_accounts[p.account].account),
p._replace(units=-p.units)
]
return None
def link_txn(txn: data.Transaction, link_accounts: dict[str, LinkAccount]) -> list[data.Transaction]:
new_postings = [p for p in txn.postings if p.account in link_accounts]
if len(new_postings) == 0:
return []
new_txns = []
get_account = lambda x: x.account
new_postings = sorted(new_postings, key=get_account)
for k, g in it.groupby(new_postings, key=get_account):
new_postings = [p for o in g for p in link_posting(o, link_accounts)]
new_txns.append(add_tags(txn._replace(postings=new_postings), "link", *link_accounts[k].tags))
return new_txns
def link_account_plugin(entries, options_map, config=None):
errors = []
link_accounts = {}
for entry in entries:
if isinstance(entry, data.Open):
if "link-account" in entry.meta:
tags = entry.meta["link-tags"].split(" ") if "link-tags" in entry.meta else []
link_accounts[entry.account] = LinkAccount(account=entry.meta["link-account"], tags=tags)
new_entries = []
for entry in entries:
if isinstance(entry, data.Transaction):
new_entries.extend(link_txn(entry, link_accounts))
new_entries.append(entry)
return new_entries, errors

86
old_split_account.py Normal file
View File

@@ -0,0 +1,86 @@
#!/usr/bin/env python3
import re
from beancount.core import data
from beancount.core import getters
from beancount.core import amount
from beancount.core import account_types
from beancount.core import realization
from beancount.core.number import D
from beancount.parser import options
__plugins__ = ('split_account_plugin',)
def get_group(tag, _pattern=re.compile(r"split-(.*)")):
m = _pattern.match(tag)
if m:
return m.group(1)
return ""
def is_splitable(posting: data.Posting, accs: account_types.AccountTypes) -> bool:
if posting.meta.get("split", "") == "no":
return False
return account_types.is_income_statement_account(posting.account, accs)
def add_tag(txn: data.Transaction, tag: str) -> data.Transaction:
return txn._replace(tags=txn.tags | set([tag]))
def divide_and_flip(posting: data.Posting, n: int = 2) -> data.Posting:
"""Return a new posting with n-th the amount and opposite sign."""
assert(n>0)
old_amount = posting.units
new_amount = amount.div(old_amount, D(-n))
return posting._replace(units=new_amount)
def split_txn(txn: data.Transaction, split_accounts: dict, accs: account_types.AccountTypes) -> None | data.Transaction:
# involves_account = any((acc in split_accounts.items() for acc in getters.get_accounts([entry])))
groups = [g for g in ( get_group(tag) for tag in txn.tags ) if g in split_accounts]
if len(groups) != 1:
return None
group = groups[0]
has_ignore_meta = txn.meta.get("split", "") == "no"
if has_ignore_meta:
return None
# gather all splitable expenses/income, and get their total balance
new_postings = [posting for posting in txn.postings if is_splitable(posting, accs)]
if len(new_postings) == 0:
return None
new_balance = realization.compute_postings_balance(new_postings)
for position in new_balance:
new_postings.append(data.Posting(
account=split_accounts[group],
units=-position.units,
cost=position.cost,
price=None,
flag=None,
meta=None
))
new_postings = [ divide_and_flip(p) for p in new_postings ]
return add_tag(txn._replace(postings=new_postings), "retorn")
def split_account_plugin(entries, options_map):
errors = []
new_entries = []
accs = options.get_account_types(options_map)
split_accounts = {}
for entry in entries:
if isinstance(entry, data.Open):
if "split" in entry.meta:
split_accounts[entry.meta["split"]] = entry.account
for entry in entries:
if isinstance(entry, data.Transaction):
new_txn = split_txn(entry, split_accounts, accs)
if new_txn is not None:
new_entries.append(new_txn)
new_entries.append(entry)
return new_entries, errors

41
optional_balance.py Normal file
View File

@@ -0,0 +1,41 @@
#!/usr/bin/env python3
from beancount.core import data
from beancount.core import account
from beancount.core import getters
from beancount.core import amount
from beancount.core import account_types
from beancount.core import realization
from beancount.core.number import D
from beancount.parser import options
import re
__plugins__ = ('optional_balance_plugin',)
def optional_balance_plugin(entries, options_map, config):
try:
match_enabled = re.compile(config)
except Exception as e:
print(f"exception {e}")
return entries, [e]
errors = []
new_entries = []
for entry in entries:
if isinstance(entry, data.Custom):
if entry.type == "optional-balance":
try:
assert(len(entry.values) == 4)
type, acc, units, curr = entry.values
if not match_enabled.match(type.value):
continue
v = amount.Amount(number=D(units.value), currency=curr.value)
entry = data.Balance(meta=entry.meta, date=entry.date, account=acc.value, amount=v, tolerance=None, diff_amount=None)
except Exception as e:
print(f"exception {e}")
errors.append(e)
new_entries.append(entry)
return new_entries, errors

186
split_account.py Normal file
View File

@@ -0,0 +1,186 @@
#!/usr/bin/env python3
import re
import collections
from typing import Sequence
from beancount.core import data
from beancount.core import inventory
from beancount.core import account
from beancount.core import amount
from beancount.core import getters
from beancount.core import account_types
from beancount.parser import parser
from beancount.core.number import D
from beancount.parser import options
import itertools as it
__plugins__ = ('split_account_plugin',)
ConfigError = collections.namedtuple("ConfigError", "source message entry")
MultipleAccountError = collections.namedtuple("MultipleAccountError", "source message entry")
SplitAccount = collections.namedtuple("SplitAccount", ["account", "users", "tag"])
class MultipleAccountException(Exception):
def __init__(self, split_tags):
self.split_tags = split_tags
def split_posting(
posting: data.Posting,
split: SplitAccount,
accs: account_types.AccountTypes,
main: None | str
) -> list[data.Posting]:
weights = None
if "split" in posting.meta:
assert(set(posting.meta["split"].split(" ")) <= set(split.users))
users = posting.meta["split"].split(" ")
weights = [1 if user in users else 0 for user in split.users]
elif account.leaf(posting.account) in split.users:
user = account.leaf(posting.account)
weights = [1 if u == user else 0 for u in split.users]
else:
weights = [1 for _ in range(len(split.users))]
total_weight = sum(weights)
assert(total_weight > 0)
new_postings = [
posting._replace(
units=amount.mul(posting.units, -D(w) / total_weight),
account=account.join(split.account, split.users[idx]),
)
for idx, w in enumerate(weights)
]
if main in split.users:
main_idx = split.users.index(main)
new_postings.extend([
posting._replace(
units=amount.mul(posting.units, -D(w) / total_weight),
)
for idx, w in enumerate(weights)
if idx != main_idx
])
new_postings.extend([
posting._replace(
units=amount.mul(posting.units, D(w) / total_weight),
account=account.join("Passiu", "Split", split.users[idx]),
)
for idx, w in enumerate(weights)
if idx != main_idx
])
return new_postings
def merge_postings(postings: Sequence[data.Posting]) -> list[data.Posting]:
merged = []
keyfunc = lambda p: p.account
for acc, g in it.groupby(sorted(postings, key=keyfunc), key=keyfunc):
inv = inventory.Inventory()
for p in g:
inv.add_position(p)
for pos in inv.get_positions():
merged.append(data.Posting(
account=acc,
units=pos.units,
cost=pos.cost,
price=None,
flag=None,
meta=None
))
return merged
def split_txn(
txn: data.Transaction,
split_accounts: dict[str, SplitAccount],
split_tags: dict[str, str],
accs: account_types.AccountTypes,
main: None | str
) -> data.Transaction:
involved_accs = set(p.account for p in txn.postings if p.account in split_accounts)
tags = list(tag for tag in txn.tags if tag in split_tags)
split_accs = list(split_tags[tag] for tag in tags)
if len(split_accs) > 1:
raise MultipleAccountException(tags)
elif len(split_accs) == 1:
involved_accs.add(split_accs[0])
if len(involved_accs) == 0:
return txn
# filter postings that involve parent accounts, this way we can preserve all asset buckets
new_postings = [p for p in txn.postings if p.account not in involved_accs]
split_postings = []
for acc in involved_accs:
ps = [(p._replace(account="<invalid>") if p.account in involved_accs else p) for p in txn.postings if p.account != acc]
split_postings.extend(n for p in ps for n in split_posting(p, split_accounts[acc], accs, main))
# merge split_postings grouped by account
new_postings.extend(merge_postings(split_postings))
return txn._replace(postings=new_postings)
def split_account_plugin(entries, options_map, config=None):
errors = []
new_entries = []
accs = options.get_account_types(options_map)
split_accounts = {}
split_tags = {}
for entry in entries:
if isinstance(entry, data.Open):
if "split" in entry.meta:
tag, *users = entry.meta["split"].split(" ")
users = list(set(users)) # remove repeats
if len(users) < 2:
errors.append(ConfigError(
entry.meta,
f"Split account need at least two users, only {len(users)} found: {users}",
entry
))
continue
split_accounts[entry.account] = SplitAccount(users=users, account=entry.account, tag=tag)
split_tags[tag] = entry.account
for user in users:
new_entries.append(
data.Open(
meta=data.new_metadata("<split-account>", 0),
date=entry.date,
account=account.join(entry.account, user),
currencies=entry.currencies,
booking=entry.booking
)
)
old_accounts = getters.get_accounts(new_entries) | getters.get_accounts(entries)
new_accounts = set()
for entry in entries:
if isinstance(entry, data.Transaction):
try:
entry = split_txn(entry, split_accounts, split_tags, accs, config)
new_accounts |= set((p.account for p in entry.postings))
except MultipleAccountException as e:
errors.append(MultipleAccountError(
entry.meta,
f"Transaction has multiple split tags: {e.split_tags}",
entry
))
new_entries.append(entry)
oldest_date = getters.get_min_max_dates(new_entries)[0]
for acc in new_accounts - old_accounts:
new_entries.append(
data.Open(
meta=data.new_metadata("<split-account>", 1),
date=oldest_date,
account=acc,
currencies=None,
booking=None
)
)
return new_entries, errors