Changeset - 6213bc1e5da9
[Not reviewed]
0 2 1
Brett Smith - 4 years ago 2020-06-20 23:04:53
brettcsmith@brettcsmith.org
reports: Add PeriodPostings class.

This is AccountPostings from the ledger report, cleaned up to be more
general.
3 files changed with 126 insertions and 20 deletions:
0 comments (0 inline, 0 general)
conservancy_beancount/reports/core.py
Show inline comments
...
 
@@ -393,6 +393,57 @@ class RelatedPostings(Sequence[data.Posting]):
 
        return {post.meta.get(key, default) for post in self}
 

	
 

	
 
class PeriodPostings(RelatedPostings):
 
    """Postings filtered and balanced over a date range
 

	
 
    Create a subclass with ``PeriodPostings.with_start_date(date)``.
 
    Note that there is no explicit stop date. The expectation is that the
 
    caller has already filtered out posts past the stop date from the input.
 

	
 
    Instances of that subclass will have three Balance attributes:
 

	
 
    * ``start_bal`` is the balance at cost of postings to your start date
 
    * ``period_bal`` is the balance at cost of postings from your start date
 
    * ``stop_bal`` is the balance at cost of all postings
 

	
 
    Use this subclass when your report includes a lot of balances over time to
 
    help you get the math right.
 
    """
 
    __slots__ = (
 
        'begin_bal',
 
        'end_bal',
 
        'period_bal',
 
        'start_bal',
 
        'stop_bal',
 
    )
 
    START_DATE = datetime.date(datetime.MINYEAR, 1, 1)
 

	
 
    def __init__(self,
 
                 source: Iterable[data.Posting]=(),
 
                 *,
 
                 _can_own: bool=False,
 
    ) -> None:
 
        start_posts: List[data.Posting] = []
 
        period_posts: List[data.Posting] = []
 
        for post in source:
 
            if post.meta.date < self.START_DATE:
 
                start_posts.append(post)
 
            else:
 
                period_posts.append(post)
 
        super().__init__(period_posts, _can_own=True)
 
        self.start_bal = RelatedPostings(start_posts, _can_own=True).balance_at_cost()
 
        self.period_bal = self.balance_at_cost()
 
        self.stop_bal = self.start_bal + self.period_bal
 
        # Convenience aliases
 
        self.begin_bal = self.start_bal
 
        self.end_bal = self.stop_bal
 

	
 
    @classmethod
 
    def with_start_date(cls: Type[RelatedType], start_date: datetime.date) -> Type[RelatedType]:
 
        name = f'BalancePostings{start_date.strftime("%Y%m%d")}'
 
        return type(name, (cls,), {'START_DATE': start_date})
 

	
 

	
 
class BaseSpreadsheet(Generic[RT, ST], metaclass=abc.ABCMeta):
 
    """Abstract base class to help write spreadsheets
 

	
conservancy_beancount/reports/ledger.py
Show inline comments
...
 
@@ -85,20 +85,6 @@ PostTally = List[Tuple[int, data.Account]]
 
PROGNAME = 'ledger-report'
 
logger = logging.getLogger('conservancy_beancount.reports.ledger')
 

	
 
class AccountPostings(core.RelatedPostings):
 
    START_DATE: datetime.date
 

	
 
    def __init__(self,
 
                 source: Iterable[data.Posting]=(),
 
                 *,
 
                 _can_own: bool=False,
 
    ) -> None:
 
        super().__init__(source, _can_own=_can_own)
 
        self.start_bal = self.balance_at_cost_by_date(self.START_DATE)
 
        self.stop_bal = self.balance_at_cost()
 
        self.period_bal = self.stop_bal - self.start_bal
 

	
 

	
 
class LedgerODS(core.BaseODS[data.Posting, data.Account]):
 
    CORE_COLUMNS: Sequence[str] = [
 
        'Date',
...
 
@@ -310,9 +296,7 @@ class LedgerODS(core.BaseODS[data.Posting, data.Account]):
 
        self._report_section_balance(key, 'stop')
 

	
 
    def write_row(self, row: data.Posting) -> None:
 
        if row.meta.date not in self.date_range:
 
            return
 
        elif row.cost is None:
 
        if row.cost is None:
 
            amount_cell = odf.table.TableCell()
 
        else:
 
            amount_cell = self.currency_cell(self.norm_func(row.units))
...
 
@@ -378,13 +362,13 @@ class LedgerODS(core.BaseODS[data.Posting, data.Account]):
 
        self._combined_balance_row(balance_accounts, 'stop')
 

	
 
    def write(self, rows: Iterable[data.Posting]) -> None:
 
        AccountPostings.START_DATE = self.date_range.start
 
        self.account_groups = dict(AccountPostings.group_by_account(
 
        related_cls = core.PeriodPostings.with_start_date(self.date_range.start)
 
        self.account_groups = dict(related_cls.group_by_account(
 
            post for post in rows if post.meta.date < self.date_range.stop
 
        ))
 
        self.write_balance_sheet()
 
        tally_by_account_iter = (
 
            (account, sum(1 for post in related if post.meta.date in self.date_range))
 
            (account, len(related))
 
            for account, related in self.account_groups.items()
 
        )
 
        tally_by_account = {
tests/test_reports_period_postings.py
Show inline comments
 
new file 100644
 
"""test_reports_period_postings - Unit tests for PeriodPostings"""
 
# 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 datetime
 
import operator
 

	
 
from decimal import Decimal
 

	
 
import pytest
 

	
 
from . import testutil
 

	
 
from conservancy_beancount import data
 
from conservancy_beancount.reports import core
 

	
 
def check_balance_attrs(balance_postings, start_usd, stop_usd):
 
    if start_usd:
 
        expected = {'USD': testutil.Amount(start_usd)}
 
        assert balance_postings.start_bal == expected
 
        assert balance_postings.begin_bal == expected
 
    else:
 
        assert balance_postings.start_bal.is_zero()
 
        assert balance_postings.begin_bal.is_zero()
 
    expected = {'USD': testutil.Amount(stop_usd)}
 
    assert balance_postings.stop_bal == expected
 
    assert balance_postings.end_bal == expected
 
    expected = {'USD': testutil.Amount(stop_usd - start_usd)}
 
    assert balance_postings.period_bal == expected
 

	
 
@pytest.mark.parametrize('start_date,expect_start_bal', [
 
    (datetime.date(2019, 2, 1), 0),
 
    (datetime.date(2019, 4, 1), 30),
 
    (datetime.date(2019, 6, 1), 120),
 
])
 
def test_balance_postings_attrs(start_date, expect_start_bal):
 
    entries = [testutil.Transaction(date=datetime.date(2019, n, 15), postings=[
 
        ('Income:Donations', -n * 10),
 
        ('Assets:Cash', n * 10),
 
    ]) for n in range(3, 7)]
 
    cls = core.PeriodPostings.with_start_date(start_date)
 
    actual = dict(cls.group_by_account(data.Posting.from_entries(entries)))
 
    assert len(actual) == 2
 
    check_balance_attrs(actual['Assets:Cash'], expect_start_bal, 180)
 
    check_balance_attrs(actual['Income:Donations'], -expect_start_bal, -180)
 

	
 
@pytest.mark.parametrize('start_date,expect_count', [
 
    (datetime.date(2019, 2, 1), 4),
 
    (datetime.date(2019, 4, 1), 3),
 
    (datetime.date(2019, 6, 1), 1),
 
])
 
def test_balance_postings_filter(start_date, expect_count):
 
    entries = [testutil.Transaction(date=datetime.date(2019, n, 15), postings=[
 
        ('Income:Donations', -n * 10),
 
        ('Assets:Cash', n * 10),
 
    ]) for n in range(3, 7)]
 
    cls = core.PeriodPostings.with_start_date(start_date)
 
    for _, related in cls.group_by_account(data.Posting.from_entries(entries)):
 
        assert len(related) == expect_count
0 comments (0 inline, 0 general)