Modules

Amortize expenses over a period of months.

amortize(entries, _, config_str)

Amortize expenses over a period of months.

Parameters:

Name	Type	Description	Default
entries	Entries	A list of beancount entries.	required
config_str	str	A string containing the configuration for the plugin.	required

Returns:

Type	Description
tuple[Entries, list[AmortizeError]]	A tuple of the modified entries and a list of errors.

Source code in beancount_blue/amortize.py

def amortize(entries: Entries, _, config_str: str) -> tuple[Entries, list[AmortizeError]]:
    """Amortize expenses over a period of months.

    Args:
        entries: A list of beancount entries.
        config_str: A string containing the configuration for the plugin.

    Returns:
        A tuple of the modified entries and a list of errors.
    """
    config = ast.literal_eval(config_str)
    accounts = config.get("accounts", None)
    if not accounts:
        return entries, [AmortizeError(source=None, message="no accounts defined", entry=None)]

    new_entries = entries[:]

    errors = []
    for config_acct, acct_config in accounts.items():
        if not config_acct.startswith("Expenses:"):
            raise Exception(f"amortize requires Expenses: accounts, got {config_acct}")  # noqa: TRY002, TRY003
        acct = config_acct.replace("Expenses:", "Equity:Amortization:")
        counteraccount = config_acct
        months = acct_config.get("months", None)
        if months is None:
            errors.append(AmortizeError(source=None, message=f"no months for account {config_acct}", entry=None))
        decimals = acct_config.get("decimals", 2)

        print(f"Running amortize for {acct}, counter {counteraccount}, months {months}, decimals {decimals}")

        # Collect all of the trading histories
        cashflow = {}
        for _, entry in enumerate(entries):
            if not isinstance(entry, Transaction):
                continue
            for _, post in enumerate(entry.postings):
                if post.account != config_acct:
                    continue
                if len(entry.tags) > 1:
                    errors.append(AmortizeError(entry=entry, message="must be zero or one tag only", source=None))
                    continue
                if not post.units or not post.units.number:
                    errors.append(
                        AmortizeError(entry=entry, message="cannot amortize a posting without units", source=None)
                    )
                    continue
                tag = next(iter(entry.tags)) if entry.tags else ""
                key = (tag, post.units.currency)
                if key not in cashflow:
                    cashflow[key] = defaultdict(Decimal)
                remaining_amt = -1 * post.units.number
                for i in range(months):
                    cashflow_amt = Decimal(round(remaining_amt / (months - i), decimals))
                    cashflow_date = (
                        entry.date + relativedelta.relativedelta(months=i) + relativedelta.relativedelta(day=31)
                    )
                    cashflow[key][cashflow_date] += cashflow_amt
                    if i == 0:
                        cashflow[key][cashflow_date] += post.units.number
                    remaining_amt -= cashflow_amt

        for key, amts in cashflow.items():
            narration = "Amortization Adjustment"
            if key[0]:
                narration = narration + f" for {key[0]}"
            for date, amt in amts.items():
                new_entries.append(
                    Transaction(
                        date=date,
                        meta={"lineno": 0},
                        flag=FLAG_OKAY,
                        payee="Amortized",
                        narration=narration,
                        tags=frozenset({key[0], "amort"}) if key[0] else frozenset({"amort"}),
                        links=frozenset(),
                        postings=[
                            Posting(acct, Amount(number=amt, currency=key[1]), None, None, None, {}),
                            Posting(counteraccount, Amount(number=-1 * amt, currency=key[1]), None, None, None, {}),
                        ],
                    )
                )

    return new_entries, errors

Calculate capital gains for UK tax purposes.

Account

An account that holds securities.

Source code in beancount_blue/calc_uk_gains.py

class Account:
    """An account that holds securities."""

    def __init__(self, account: str, config: dict):
        """Initialize the account.

        Args:
            account: The name of the account.
            config: The configuration for the account.
        """
        self.account = account
        self.config = config
        self.cost_currency = None
        self.history = {}
        self.last_balance = {}

    def process(self) -> list[Adjustment]:
        """Process the trades in the account.

        Returns:
            A list of adjustments.
        """
        adjustments = []

        method = METHODS.get(self.config.get("method", ""), None)
        if method is None:
            raise Exception(f"Account {self.account} has no valid method, mustbe one of {', '.join(METHODS.keys())}")  # noqa: TRY002, TRY003

        cacct = self.config.get("counterAccount", None)

        # Add in counteraccount configuration
        for trades in self.history.values():
            adjs = method(trades)
            adjustments.extend([a._replace(counterAccount=cacct) for a in adjs])

        return adjustments

    def add_posting(self, postingId: PostingID, entry: Transaction, posting: Posting) -> Optional[str]:
        """Add a posting to the account.

        Args:
            postingId: The ID of the posting.
            entry: The entry containing the posting.
            posting: The posting to add.

        Returns:
            An error message if there was an error, otherwise None.
        """
        # Validate the cost currency
        # TODO: Should be indexed per posting.unit.currency
        if posting.cost is None:
            return f"posting {entry.date} and {posting.account} has no cost"

        if posting.units is None or posting.units.number is None:
            return f"posting {entry.date} and {posting.account} has no units"

        if self.cost_currency is None:
            self.cost_currency = posting.cost.currency
        elif self.cost_currency != posting.cost.currency:
            return (
                f"account {self.account} has inconsistent cost currencies:"
                + f"{self.cost_currency} and {posting.units.currency}"
            )

        if posting.cost.date != entry.date:
            return f"cost date {posting.cost.date} is different" + f"than transaction date {entry.date}"

        # Get the last balance
        balance = self.last_balance.get(posting.units.currency, Decimal(0))

        # Determine if realizing
        # print(posting)
        if (balance > 0 and posting.units.number < 0) or (balance < 0 and posting.units.number > 0):
            realizing = True
        else:
            realizing = False

        # TODO: Validate cost date versus transaction date

        # Add the trade
        price = posting.cost.number_per if isinstance(posting.cost, CostSpec) else posting.cost.number
        if price is None:
            return f"cost {posting.cost} has no price!"

        self.history.setdefault(posting.units.currency, []).append(
            Trade(
                postingId=postingId,
                date=entry.date,
                units=posting.units.number,
                price=price,
                realizing=realizing,
            )
        )

        # Update the last balance
        self.last_balance[posting.units.currency] = balance + posting.units.number

        return None

__init__(account, config)

Initialize the account.

Parameters:

Name	Type	Description	Default
account	str	The name of the account.	required
config	dict	The configuration for the account.	required

Source code in beancount_blue/calc_uk_gains.py

def __init__(self, account: str, config: dict):
    """Initialize the account.

    Args:
        account: The name of the account.
        config: The configuration for the account.
    """
    self.account = account
    self.config = config
    self.cost_currency = None
    self.history = {}
    self.last_balance = {}

add_posting(postingId, entry, posting)

Add a posting to the account.

Parameters:

Name	Type	Description	Default
postingId	PostingID	The ID of the posting.	required
entry	Transaction	The entry containing the posting.	required
posting	Posting	The posting to add.	required

Returns:

Type	Description
Optional[str]	An error message if there was an error, otherwise None.

Source code in beancount_blue/calc_uk_gains.py

def add_posting(self, postingId: PostingID, entry: Transaction, posting: Posting) -> Optional[str]:
    """Add a posting to the account.

    Args:
        postingId: The ID of the posting.
        entry: The entry containing the posting.
        posting: The posting to add.

    Returns:
        An error message if there was an error, otherwise None.
    """
    # Validate the cost currency
    # TODO: Should be indexed per posting.unit.currency
    if posting.cost is None:
        return f"posting {entry.date} and {posting.account} has no cost"

    if posting.units is None or posting.units.number is None:
        return f"posting {entry.date} and {posting.account} has no units"

    if self.cost_currency is None:
        self.cost_currency = posting.cost.currency
    elif self.cost_currency != posting.cost.currency:
        return (
            f"account {self.account} has inconsistent cost currencies:"
            + f"{self.cost_currency} and {posting.units.currency}"
        )

    if posting.cost.date != entry.date:
        return f"cost date {posting.cost.date} is different" + f"than transaction date {entry.date}"

    # Get the last balance
    balance = self.last_balance.get(posting.units.currency, Decimal(0))

    # Determine if realizing
    # print(posting)
    if (balance > 0 and posting.units.number < 0) or (balance < 0 and posting.units.number > 0):
        realizing = True
    else:
        realizing = False

    # TODO: Validate cost date versus transaction date

    # Add the trade
    price = posting.cost.number_per if isinstance(posting.cost, CostSpec) else posting.cost.number
    if price is None:
        return f"cost {posting.cost} has no price!"

    self.history.setdefault(posting.units.currency, []).append(
        Trade(
            postingId=postingId,
            date=entry.date,
            units=posting.units.number,
            price=price,
            realizing=realizing,
        )
    )

    # Update the last balance
    self.last_balance[posting.units.currency] = balance + posting.units.number

    return None

process()

Process the trades in the account.

Returns:

Type	Description
list[Adjustment]	A list of adjustments.

Source code in beancount_blue/calc_uk_gains.py

def process(self) -> list[Adjustment]:
    """Process the trades in the account.

    Returns:
        A list of adjustments.
    """
    adjustments = []

    method = METHODS.get(self.config.get("method", ""), None)
    if method is None:
        raise Exception(f"Account {self.account} has no valid method, mustbe one of {', '.join(METHODS.keys())}")  # noqa: TRY002, TRY003

    cacct = self.config.get("counterAccount", None)

    # Add in counteraccount configuration
    for trades in self.history.values():
        adjs = method(trades)
        adjustments.extend([a._replace(counterAccount=cacct) for a in adjs])

    return adjustments

Adjustment

Bases: NamedTuple

An adjustment to a trade.

Source code in beancount_blue/calc_uk_gains.py

class Adjustment(NamedTuple):
    """An adjustment to a trade."""

    postingId: PostingID
    price: Decimal
    counterAmount: Decimal
    counterAccount: Optional[str]

GainsCalculatorError

Bases: NamedTuple

An error that occurred during capital gains calculation.

Source code in beancount_blue/calc_uk_gains.py

class GainsCalculatorError(NamedTuple):
    """An error that occurred during capital gains calculation."""

    source: Meta
    message: str
    entry: object

Trade

Bases: NamedTuple

A trade in a security.

Source code in beancount_blue/calc_uk_gains.py

class Trade(NamedTuple):
    """A trade in a security."""

    postingId: PostingID
    date: datetime.date
    units: Decimal
    price: Decimal
    realizing: bool

calc_gains(entries, _, config_str)

Calculate capital gains for UK tax purposes.

Parameters:

Name	Type	Description	Default
entries	Entries	A list of beancount entries.	required
config_str	str	A string containing the configuration for the plugin.	required

Returns:

Type	Description
tuple[list[Directive], list[GainsCalculatorError]]	A tuple of the modified entries and a list of errors.

Source code in beancount_blue/calc_uk_gains.py

def calc_gains(entries: Entries, _, config_str: str) -> tuple[list[Directive], list[GainsCalculatorError]]:
    """Calculate capital gains for UK tax purposes.

    Args:
        entries: A list of beancount entries.
        config_str: A string containing the configuration for the plugin.

    Returns:
        A tuple of the modified entries and a list of errors.
    """
    accounts = {}

    config = ast.literal_eval(config_str)
    for acct, acct_config in config.get("accounts", {}).items():
        accounts[acct] = Account(acct, acct_config)

    errors = []

    # Collect all of the trading histories
    for transId, entry in enumerate(entries):
        if not isinstance(entry, Transaction):
            continue
        for postId, post in enumerate(entry.postings):
            if post.account not in accounts:
                continue
            if not post.cost and not post.price:
                continue
            if not post.cost:
                errors.append("missing cost?!?")
                continue
            accounts[post.account].add_posting((transId, postId), entry, post)

    # Collect adjustments for accounts
    adjs = []
    for account in accounts.values():
        adjs.extend(account.process())

    # Apply adjustments to the entries
    new_entries = entries.copy()
    errors = []
    for adj in adjs:
        trans = new_entries[adj.postingId[0]]

        # If there is no counterAccount, we need to report an error
        if adj.counterAccount is None:
            errors.append(
                GainsCalculatorError(trans.meta, f"Calculated cost price is {adj.price}, not matching", trans)
            )
            continue

        # Adjust the price
        trans.postings[adj.postingId[1]] = trans.postings[adj.postingId[1]]._replace(
            cost=trans.postings[adj.postingId[1]].cost._replace(
                number=adj.price,
            )
        )

        # Create the adjustment posting
        trans.postings.append(
            Posting(
                account=adj.counterAccount,
                units=Amount(number=adj.counterAmount, currency=trans.postings[adj.postingId[1]].cost.currency),
                cost=None,
                price=None,
                flag=None,
                meta={"note": "adjusted"},
            )
        )

    return new_entries, errors

cost_avg(trades)

Calculate the average cost of a list of trades.

Parameters:

Name	Type	Description	Default
trades	list[Trade]	A list of trades.	required

Returns:

Type	Description
list[Adjustment]	A list of adjustments.

Source code in beancount_blue/calc_uk_gains.py

def cost_avg(trades: list[Trade]) -> list[Adjustment]:
    """Calculate the average cost of a list of trades.

    Args:
        trades: A list of trades.

    Returns:
        A list of adjustments.
    """
    adjs = []
    total_units = Decimal(0)
    total_cost = Decimal(0)
    for _, trade in enumerate(trades):
        if trade.realizing:
            avg_cost_price = total_cost / total_units
            if trade.price != avg_cost_price:
                adjs.append(
                    Adjustment(
                        postingId=trade.postingId,
                        price=total_cost / total_units,
                        counterAmount=((trade.price * trade.units) - (avg_cost_price * trade.units)),
                        counterAccount=None,
                    )
                )
            total_cost += trade.units * avg_cost_price
        else:
            total_cost += trade.units * trade.price
        total_units += trade.units
    return adjs

Tag transactions based on account.

tag(entries, _, config_str)

Tag transactions based on account.

Parameters:

Name	Type	Description	Default
entries	Entries	A list of beancount entries.	required
config_str	str	A string containing the configuration for the plugin.	required

Returns:

Type	Description
tuple[Entries, list[Any]]	A tuple of the modified entries and a list of errors.

Source code in beancount_blue/tag.py

def tag(entries: Entries, _, config_str: str) -> tuple[Entries, list[Any]]:
    """Tag transactions based on account.

    Args:
        entries: A list of beancount entries.
        config_str: A string containing the configuration for the plugin.

    Returns:
        A tuple of the modified entries and a list of errors.
    """
    config = ast.literal_eval(config_str)
    accounts = config.get("accounts", None)
    if not accounts:
        return entries, ["no accounts defined"]

    new_entries = entries[:]

    errors = []
    for acct, tag in accounts.items():
        print(f"Running tag for {acct}, tag {tag}")

        for transId, entry in enumerate(new_entries):
            if not isinstance(entry, Transaction):
                continue
            if all(post.account != acct for post in entry.postings):
                continue
            new_entries[transId] = entry._replace(tags=frozenset(set(entry.tags).union([tag])))

    return new_entries, errors