416 lines
15 KiB
Python
416 lines
15 KiB
Python
"""funds.py - Funds report from Beancount
|
|
|
|
This tool produces a report of account balances for restricted funds,
|
|
including income, expenses, and realized gain/loss over a given time period.
|
|
Restricted funds are designated with the "project" metadata.
|
|
|
|
Specify the date range you want to report with the ``--begin`` and ``--end``
|
|
options.
|
|
|
|
Select the accounts you want to report with the ``--account`` option. You can
|
|
specify this option multiple times. The report will include at least one sheet
|
|
for each account you specify. Subaccounts will be reported on that sheet as
|
|
well.
|
|
|
|
Select the postings you want to report by passing metadata search terms in
|
|
``name=value`` format.
|
|
|
|
Run ``ledger-report --help`` for abbreviations and other options.
|
|
|
|
Examples
|
|
--------
|
|
|
|
Generate a report of all restricted funds in ODS format over the last year::
|
|
|
|
fund-report
|
|
|
|
Query a specific restricted fund and get a quick balance on the terminal::
|
|
|
|
fund-report <PROJECT>
|
|
"""
|
|
# Copyright © 2020 Brett Smith
|
|
#
|
|
# 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/>.
|
|
|
|
import argparse
|
|
import collections
|
|
import datetime
|
|
import enum
|
|
import locale
|
|
import logging
|
|
import sys
|
|
|
|
from typing import (
|
|
Dict,
|
|
Iterable,
|
|
Iterator,
|
|
Mapping,
|
|
Optional,
|
|
Sequence,
|
|
TextIO,
|
|
Tuple,
|
|
Union,
|
|
)
|
|
from ..beancount_types import (
|
|
MetaValue,
|
|
)
|
|
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
|
|
import odf.table # type:ignore[import]
|
|
|
|
from beancount.parser import printer as bc_printer
|
|
|
|
from . import core
|
|
from .. import books
|
|
from .. import cliutil
|
|
from .. import config as configmod
|
|
from .. import data
|
|
|
|
AccountsMap = Mapping[data.Account, core.PeriodPostings]
|
|
FundPosts = Tuple[MetaValue, AccountsMap]
|
|
|
|
EQUITY_ACCOUNTS = ['Equity', 'Income', 'Expenses']
|
|
INFO_ACCOUNTS = [
|
|
'Assets:Receivable',
|
|
'Assets:Prepaid',
|
|
'Liabilities:UnearnedIncome',
|
|
'Liabilities:Payable',
|
|
]
|
|
PROGNAME = 'fund-report'
|
|
UNRESTRICTED_FUND = 'Conservancy'
|
|
logger = logging.getLogger('conservancy_beancount.reports.fund')
|
|
|
|
class ODSReport(core.BaseODS[FundPosts, None]):
|
|
def __init__(self, start_date: datetime.date, stop_date: datetime.date) -> None:
|
|
super().__init__()
|
|
self.start_date = start_date
|
|
self.stop_date = stop_date
|
|
|
|
def section_key(self, row: FundPosts) -> None:
|
|
return None
|
|
|
|
def start_spreadsheet(self, *, expanded: bool=True) -> None:
|
|
headers = [["Fund"], ["Balance as of", self.start_date.isoformat()]]
|
|
if expanded:
|
|
sheet_name = "With Breakdowns"
|
|
headers += [["Income"], ["Expenses"], ["Equity"]]
|
|
else:
|
|
sheet_name = "Fund Report"
|
|
headers += [["Additions"], ["Releases from", "Restrictions"]]
|
|
headers.append(["Balance as of", self.stop_date.isoformat()])
|
|
if expanded:
|
|
headers += [
|
|
["Of which", "Receivable"],
|
|
["Of which", "Prepaid Expenses"],
|
|
["Of which", "Payable"],
|
|
["Of which", "Unearned Income"],
|
|
]
|
|
|
|
self.use_sheet(sheet_name)
|
|
for header in headers:
|
|
first_line = header[0]
|
|
if first_line == 'Fund':
|
|
width = 2.0
|
|
elif first_line == 'Balance as of':
|
|
width = 1.5
|
|
elif first_line == 'Of which':
|
|
width = 1.3
|
|
else:
|
|
width = 1.2
|
|
col_style = self.column_style(width)
|
|
self.sheet.addElement(odf.table.TableColumn(stylename=col_style))
|
|
|
|
center_bold = self.merge_styles(self.style_centertext, self.style_bold)
|
|
row = self.add_row(*(
|
|
self.multiline_cell(header, stylename=center_bold)
|
|
for header in headers
|
|
))
|
|
row.firstChild.setAttribute(
|
|
'stylename', self.merge_styles(self.style_endtext, self.style_bold),
|
|
)
|
|
self.lock_first_row()
|
|
self.lock_first_column()
|
|
self.add_row()
|
|
self.add_row(self.string_cell(
|
|
f"Fund Report From {self.start_date.isoformat()} To {self.stop_date.isoformat()}",
|
|
stylename=center_bold,
|
|
numbercolumnsspanned=6,
|
|
))
|
|
self.add_row()
|
|
|
|
def end_spreadsheet(self) -> None:
|
|
start_sheet = self.sheet
|
|
self.set_open_sheet(self.sheet)
|
|
self.start_spreadsheet(expanded=False)
|
|
bal_indexes = [0, 1, 2, 4]
|
|
totals = [core.MutableBalance() for _ in bal_indexes]
|
|
threshold = Decimal('.5')
|
|
for fund, balances in self.balances.items():
|
|
balances = [balances[index] for index in bal_indexes]
|
|
if (not all(bal.clean_copy(threshold).le_zero() for bal in balances)
|
|
and fund != UNRESTRICTED_FUND):
|
|
self.write_balances(fund, balances)
|
|
for total, bal in zip(totals, balances):
|
|
total += bal
|
|
self.write_balances('', totals, self.merge_styles(
|
|
self.border_style(core.Border.TOP, '.75pt'),
|
|
self.border_style(core.Border.BOTTOM, '1.5pt', 'double'),
|
|
))
|
|
self.document.spreadsheet.childNodes.reverse()
|
|
self.sheet = start_sheet
|
|
|
|
def _row_balances(self, accounts_map: AccountsMap) -> Iterator[core.Balance]:
|
|
acct_order = ['Income', 'Expenses', 'Equity']
|
|
key_order = [core.OPENING_BALANCE_NAME, *acct_order, core.ENDING_BALANCE_NAME]
|
|
balances: Dict[str, core.Balance] = {key: core.MutableBalance() for key in key_order}
|
|
for acct_s, balance in core.account_balances(accounts_map, acct_order):
|
|
if acct_s in balances:
|
|
balances[acct_s] = balance
|
|
else:
|
|
acct_root, _, _ = acct_s.partition(':')
|
|
balances[acct_root] += balance
|
|
for key in key_order:
|
|
if key == 'Expenses':
|
|
yield balances[key]
|
|
else:
|
|
yield -balances[key]
|
|
for info_key in INFO_ACCOUNTS:
|
|
for _, balance in core.account_balances(accounts_map, [info_key]):
|
|
pass
|
|
yield core.normalize_amount_func(info_key)(balance)
|
|
|
|
def write_balances(self,
|
|
fund: str,
|
|
balances: Iterable[core.Balance],
|
|
style: Union[None, str, odf.style.Style]=None,
|
|
) -> odf.table.TableRow:
|
|
return self.add_row(
|
|
self.string_cell(fund, stylename=self.style_endtext),
|
|
*(self.balance_cell(bal, stylename=style) for bal in balances),
|
|
)
|
|
|
|
def write_row(self, row: FundPosts) -> None:
|
|
fund, accounts_map = row
|
|
self.balances[fund] = list(self._row_balances(accounts_map))
|
|
if fund != UNRESTRICTED_FUND:
|
|
self.write_balances(fund, self.balances[fund])
|
|
|
|
def write(self, rows: Iterable[FundPosts]) -> None:
|
|
self.balances: Dict[str, Sequence[core.Balance]] = collections.OrderedDict()
|
|
super().write(rows)
|
|
try:
|
|
unrestricted = self.balances[UNRESTRICTED_FUND]
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
self.add_row()
|
|
self.write_balances("Unrestricted", unrestricted)
|
|
|
|
|
|
class TextReport:
|
|
def __init__(self,
|
|
start_date: datetime.date,
|
|
stop_date: datetime.date,
|
|
out_file: TextIO) -> None:
|
|
self.start_date = start_date
|
|
self.stop_date = stop_date
|
|
self.out_file = out_file
|
|
|
|
def _account_balances(self,
|
|
fund: str,
|
|
account_map: AccountsMap,
|
|
) -> Iterator[Tuple[str, Sequence[str]]]:
|
|
total_fmt = f'{fund} balance as of {{}}'
|
|
for acct_s, balance in core.account_balances(account_map, EQUITY_ACCOUNTS):
|
|
if acct_s is core.OPENING_BALANCE_NAME:
|
|
acct_s = total_fmt.format(self.start_date.isoformat())
|
|
elif acct_s is core.ENDING_BALANCE_NAME:
|
|
acct_s = total_fmt.format(self.stop_date.isoformat())
|
|
yield acct_s, (-balance).format(None, sep='\0').split('\0')
|
|
for _, account in core.sort_and_filter_accounts(account_map, INFO_ACCOUNTS):
|
|
balance = account_map[account].stop_bal
|
|
if not balance.is_zero():
|
|
balance = core.normalize_amount_func(account)(balance)
|
|
yield account, balance.format(None, sep='\0').split('\0')
|
|
|
|
def write(self, rows: Iterable[FundPosts]) -> None:
|
|
output = [
|
|
line
|
|
for fund, account_map in rows
|
|
for line in self._account_balances(fund, account_map)
|
|
]
|
|
acct_width = max(len(acct_s) for acct_s, _ in output) + 2
|
|
bal_width = max(len(s) for _, bal_s in output for s in bal_s)
|
|
bal_width = max(bal_width, 8)
|
|
line_fmt = f'{{:>{acct_width}}} {{:>{bal_width}}}'
|
|
print(line_fmt.replace('{:>', '{:^').format("ACCOUNT", "BALANCE"),
|
|
file=self.out_file)
|
|
fund_start = f' balance as of {self.start_date.isoformat()}'
|
|
for acct_s, bal_seq in output:
|
|
if acct_s.endswith(fund_start):
|
|
print(line_fmt.format('―' * acct_width, '―' * bal_width),
|
|
file=self.out_file)
|
|
bal_iter = iter(bal_seq)
|
|
print(line_fmt.format(acct_s, next(bal_iter)), file=self.out_file)
|
|
for bal_s in bal_iter:
|
|
print(line_fmt.format('', bal_s), file=self.out_file)
|
|
|
|
|
|
class ReportType(enum.Enum):
|
|
TEXT = TextReport
|
|
ODS = ODSReport
|
|
TXT = TEXT
|
|
SPREADSHEET = ODS
|
|
|
|
@classmethod
|
|
def from_arg(cls, s: str) -> 'ReportType':
|
|
try:
|
|
return cls[s.upper()]
|
|
except KeyError:
|
|
raise ValueError(f"no report type matches {s!r}") from None
|
|
|
|
|
|
def parse_arguments(arglist: Optional[Sequence[str]]=None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(prog=PROGNAME)
|
|
cliutil.add_version_argument(parser)
|
|
parser.add_argument(
|
|
'--begin', '--start', '-b',
|
|
dest='start_date',
|
|
metavar='DATE',
|
|
type=cliutil.date_arg,
|
|
help="""Date to start reporting entries, inclusive, in YYYY-MM-DD format.
|
|
The default is one year ago.
|
|
""")
|
|
parser.add_argument(
|
|
'--end', '--stop', '-e',
|
|
dest='stop_date',
|
|
metavar='DATE',
|
|
type=cliutil.date_arg,
|
|
help="""Date to stop reporting entries, exclusive, in YYYY-MM-DD format.
|
|
The default is a year after the start date.
|
|
""")
|
|
parser.add_argument(
|
|
'--report-type', '-t',
|
|
metavar='TYPE',
|
|
type=ReportType.from_arg,
|
|
help="""Type of report to generate. `text` gives a plain two-column text
|
|
report listing accounts and balances over the period, and is the default when
|
|
you search for a specific project/fund. `ods` produces a higher-level
|
|
spreadsheet, meant to provide an overview of all funds, and is the default when
|
|
you don't specify a project/fund.
|
|
""")
|
|
parser.add_argument(
|
|
'--output-file', '-O',
|
|
metavar='PATH',
|
|
type=Path,
|
|
help="""Write the report to this file, or stdout when PATH is `-`.
|
|
The default is stdout for text reports, and a generated filename for ODS
|
|
reports.
|
|
""")
|
|
cliutil.add_loglevel_argument(parser)
|
|
parser.add_argument(
|
|
'search_terms',
|
|
metavar='FILTER',
|
|
type=cliutil.SearchTerm.arg_parser('project', 'rt-id'),
|
|
nargs=argparse.ZERO_OR_MORE,
|
|
help="""Report on postings that match this criteria. The format is
|
|
NAME=TERM. TERM is a link or word that must exist in a posting's NAME
|
|
metadata to match. A single ticket number is a shortcut for
|
|
`rt-id=rt:NUMBER`. Any other word is a shortcut for `project=TERM`.
|
|
""")
|
|
args = parser.parse_args(arglist)
|
|
if args.report_type is None:
|
|
if any(term.meta_key == 'project' for term in args.search_terms):
|
|
args.report_type = ReportType.TEXT
|
|
else:
|
|
args.report_type = ReportType.ODS
|
|
return args
|
|
|
|
def main(arglist: Optional[Sequence[str]]=None,
|
|
stdout: TextIO=sys.stdout,
|
|
stderr: TextIO=sys.stderr,
|
|
config: Optional[configmod.Config]=None,
|
|
) -> int:
|
|
args = parse_arguments(arglist)
|
|
cliutil.set_loglevel(logger, args.loglevel)
|
|
if config is None:
|
|
config = configmod.Config()
|
|
config.load_file()
|
|
|
|
if args.stop_date is None:
|
|
if args.start_date is None:
|
|
args.stop_date = datetime.date.today()
|
|
else:
|
|
args.stop_date = cliutil.diff_year(args.start_date, 1)
|
|
if args.start_date is None:
|
|
args.start_date = cliutil.diff_year(args.stop_date, -1)
|
|
|
|
returncode = 0
|
|
books_loader = config.books_loader()
|
|
if books_loader is None:
|
|
entries, load_errors, _ = books.Loader.load_none(config.config_file_path())
|
|
returncode = cliutil.ExitCode.NoConfiguration
|
|
else:
|
|
entries, load_errors, _ = books_loader.load_fy_range(args.start_date, args.stop_date)
|
|
if load_errors:
|
|
returncode = cliutil.ExitCode.BeancountErrors
|
|
elif not entries:
|
|
returncode = cliutil.ExitCode.NoDataLoaded
|
|
for error in load_errors:
|
|
bc_printer.print_error(error, file=stderr)
|
|
|
|
postings = (
|
|
post
|
|
for post in data.Posting.from_entries(entries)
|
|
if post.meta.date < args.stop_date
|
|
)
|
|
for search_term in args.search_terms:
|
|
postings = search_term.filter_postings(postings)
|
|
fund_postings = {
|
|
key: related
|
|
for key, related in core.RelatedPostings.group_by_meta(postings, 'project')
|
|
if isinstance(key, str)
|
|
}
|
|
period_cls = core.PeriodPostings.with_start_date(args.start_date)
|
|
fund_map = collections.OrderedDict(
|
|
(fund, dict(period_cls.group_by_account(fund_postings[fund])))
|
|
for fund in sorted(fund_postings, key=lambda s: locale.strxfrm(s.casefold()))
|
|
)
|
|
if not fund_map:
|
|
logger.warning("no matching postings found to report")
|
|
returncode = returncode or cliutil.ExitCode.NoDataFiltered
|
|
elif args.report_type is ReportType.TEXT:
|
|
out_file = cliutil.text_output(args.output_file, stdout)
|
|
report = TextReport(args.start_date, args.stop_date, out_file)
|
|
report.write(fund_map.items())
|
|
else:
|
|
ods_report = ODSReport(args.start_date, args.stop_date)
|
|
ods_report.set_common_properties(config.books_repo())
|
|
ods_report.write(fund_map.items())
|
|
if args.output_file is None:
|
|
out_dir_path = config.repository_path() or Path()
|
|
args.output_file = out_dir_path / 'FundReport_{}_{}.ods'.format(
|
|
args.start_date.isoformat(), args.stop_date.isoformat(),
|
|
)
|
|
logger.info("Writing report to %s", args.output_file)
|
|
ods_file = cliutil.bytes_output(args.output_file, stdout)
|
|
ods_report.save_file(ods_file)
|
|
return returncode
|
|
|
|
entry_point = cliutil.make_entry_point(__name__, PROGNAME)
|
|
|
|
if __name__ == '__main__':
|
|
exit(entry_point())
|