Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# Compute the list of entries for the opening balances sheet.
self.opening_entries = (self.entries[:self.begin_index]
if self.begin_index is not None
else [])
# Compute the list of entries that includes transfer entries of the
# income/expenses amounts to the balance sheet's equity (as "net
# income"). This is used to render the end-period balance sheet, with
# the current period's net income, closing the period.
self.closing_entries = summarize.cap_opt(self.entries, options_map)
# Realize the three sets of entries.
account_types = options.get_account_types(options_map)
with misc_utils.log_time('realize_opening', logging.info):
self.opening_real_accounts = realization.realize(self.opening_entries,
account_types)
with misc_utils.log_time('realize', logging.info):
self.real_accounts = realization.realize(self.entries,
account_types)
with misc_utils.log_time('realize_closing', logging.info):
self.closing_real_accounts = realization.realize(self.closing_entries,
account_types)
assert self.real_accounts is not None
assert self.closing_real_accounts is not None
def merge_postings(entries, accounts):
"""Merge all the postings from the given account names.
Args:
entries: A list of directives.
accounts: A list of account strings to get the balances for.
Yields:
A list of TxnPosting's for all the accounts, in sorted order.
"""
real_root = realization.realize(entries)
merged_postings = []
for account in accounts:
real_account = realization.get(real_root, account)
if real_account is None:
continue
merged_postings.extend(txn_posting
for txn_posting in real_account.txn_postings
if isinstance(txn_posting, data.TxnPosting))
merged_postings.sort(key=lambda txn_posting: txn_posting.txn.date)
return merged_postings
# income"). This is used to render the end-period balance sheet, with
# the current period's net income, closing the period.
self.closing_entries = summarize.cap_opt(self.entries, options_map)
# Realize the three sets of entries.
account_types = options.get_account_types(options_map)
with misc_utils.log_time('realize_opening', logging.info):
self.opening_real_accounts = realization.realize(self.opening_entries,
account_types)
with misc_utils.log_time('realize', logging.info):
self.real_accounts = realization.realize(self.entries,
account_types)
with misc_utils.log_time('realize_closing', logging.info):
self.closing_real_accounts = realization.realize(self.closing_entries,
account_types)
assert self.real_accounts is not None
assert self.closing_real_accounts is not None
def _apply_filters(self):
self.entries = self.all_entries
for filter in self.filters.values():
self.entries = filter.apply(self.entries, self.options)
self.closing_entries = summarize.cap_opt(self.entries, self.options)
self.root_account = realization.realize(self.entries,
self.account_types)
self.title = self.options['title']
if self.options['render_commas']:
self.format_string = '{:,f}'
self.default_format_string = '{:,.2f}'
else:
self.format_string = '{:f}'
self.default_format_string = '{:.2f}'
self.dcontext = self.options['dcontext']
self.active_years = list(getters.get_active_years(self.all_entries))
self.active_tags = list(getters.get_all_tags(self.all_entries))
self.active_payees = list(getters.get_all_payees(self.all_entries))
self.queries = self._entries_filter_type(self.all_entries, Query)
self.all_root_account = realization.realize(self.all_entries,
self.account_types)
self.all_accounts = self._all_accounts()
self.all_accounts_leaf_only = self._all_accounts(leaf_only=True)
self._apply_filters()
def create_report(entries, options_map):
real_root = realization.realize(entries)
# Find the institutions from the data.
groups, ignored_accounts = find_institutions(entries, options_map)
# List all the asset accounts which aren't included in the report.
oc_map = getters.get_account_open_close(entries)
open_map = {acc: open_entry for acc, (open_entry, _) in oc_map.items()}
for acc in sorted(ignored_accounts):
logging.info("Ignored account: %s", acc)
# Gather missing fields and create a report object.
institutions = []
for name, accounts in sorted(groups.items()):
# Get the institution fields, which is the union of the fields for all
# the accounts with the institution fields.
institution_accounts = [acc for acc in accounts
# income"). This is used to render the end-period balance sheet, with
# the current period's net income, closing the period.
self.closing_entries = summarize.cap_opt(self.entries, options_map)
# Realize the three sets of entries.
account_types = options.get_account_types(options_map)
with misc_utils.log_time('realize_opening', logging.info):
self.opening_real_accounts = realization.realize(self.opening_entries,
account_types)
with misc_utils.log_time('realize', logging.info):
self.real_accounts = realization.realize(self.entries,
account_types)
with misc_utils.log_time('realize_closing', logging.info):
self.closing_real_accounts = realization.realize(self.closing_entries,
account_types)
assert self.real_accounts is not None
assert self.closing_real_accounts is not None
def load_file(self):
"""Load the main file and all included files and set attributes."""
# use the internal function to disable cache
if not self._is_encrypted:
# pylint: disable=protected-access
self.all_entries, self.errors, self.options = loader._load(
[(self.beancount_file_path, True)], None, None, None
)
else:
self.all_entries, self.errors, self.options = loader.load_file(
self.beancount_file_path
)
self.account_types = get_account_types(self.options)
self.price_map = prices.build_price_map(self.all_entries)
self.all_root_account = realization.realize(
self.all_entries, self.account_types
)
entries_by_type = collections.defaultdict(list)
for entry in self.all_entries:
entries_by_type[type(entry)].append(entry)
self.all_entries_by_type = entries_by_type
self.accounts = _AccountDict()
for entry in entries_by_type[Open]:
self.accounts.setdefault(entry.account).meta = entry.meta
for entry in entries_by_type[Close]:
self.accounts.setdefault(entry.account).close_date = entry.date
self.fava_options, errors = parse_options(entries_by_type[Custom])
self.errors.extend(errors)
else [])
# Compute the list of entries that includes transfer entries of the
# income/expenses amounts to the balance sheet's equity (as "net
# income"). This is used to render the end-period balance sheet, with
# the current period's net income, closing the period.
self.closing_entries = summarize.cap_opt(self.entries, options_map)
# Realize the three sets of entries.
account_types = options.get_account_types(options_map)
with misc_utils.log_time('realize_opening', logging.info):
self.opening_real_accounts = realization.realize(self.opening_entries,
account_types)
with misc_utils.log_time('realize', logging.info):
self.real_accounts = realization.realize(self.entries,
account_types)
with misc_utils.log_time('realize_closing', logging.info):
self.closing_real_accounts = realization.realize(self.closing_entries,
account_types)
assert self.real_accounts is not None
assert self.closing_real_accounts is not None
holdings. If left unspecified, we use the latest price information.
Returns:
A list of dicts, with the following fields:
"""
# Remove the entries inserted by unrealized gains/losses. Those entries do
# affect asset accounts, and we don't want them to appear in holdings.
#
# Note: Perhaps it would make sense to generalize this concept of "inserted
# unrealized gains."
simple_entries = [entry
for entry in entries
if (not isinstance(entry, data.Transaction) or
entry.flag != flags.FLAG_UNREALIZED)]
# Realize the accounts into a tree (because we want the positions by-account).
root_account = realization.realize(simple_entries)
# For each account, look at the list of positions and build a list.
holdings = []
for real_account in sorted(list(realization.iter_children(root_account)),
key=lambda ra: ra.account):
if included_account_types:
# Skip accounts of invalid types, we only want to reflect the requested
# account types, typically assets and liabilities.
account_type = account_types.get_account_type(real_account.account)
if account_type not in included_account_types:
continue
for pos in real_account.balance.get_positions():
if pos.cost is not None:
# Get price information if we have a price_map.