invest.section-104-pool Unreviewed

Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.

from typing import Dict, List, Optional, Sequence

from .dates_add_days import epoch_day_from_iso  ← from dates.add-days ^1.0.0 · built alongside by fune
from .invest_section_104_pool_types import Section104Result, ShareDisposal, ShareMatch, ShareTransaction
from .money_amount import Money  ← from money.amount ^1.0.0 · built alongside by fune

#: The identification rules below apply to disposals on or after this date (FA 2008).
_RULES_START = "2008-04-06"


class _DayTotal:
    """One day's purchases or sales, added together as the same-day rule requires."""

    def __init__(self, date: str) -> None:
        self.date = date
        self.day = epoch_day_from_iso(date)
        self.quantity = 0
        self.amount = 0
        # Shares and cost still unmatched (purchases) or unidentified (sales).
        self.remaining = 0
        self.remaining_amount = 0


def _gbp(minor: int) -> Money:
    return Money(minor=minor, currency="GBP")


def _share(a: int, b: int, c: int) -> int:
    """a * b / c rounded half-up, all non-negative."""
    return (2 * a * b + c) // (2 * c)


def _take(total: _DayTotal, quantity: int) -> int:
    """Take `quantity` shares from a purchase (or the pool) and the cost that
    goes with them. Cost is apportioned by number of shares, rounded half-up to
    the penny, and taken from what is left, so the pieces always add back to
    the whole: the last shares out take the last pennies."""
    if quantity == total.remaining:
        cost = total.remaining_amount
    else:
        cost = _share(total.remaining_amount, quantity, total.remaining)
    total.remaining -= quantity
    total.remaining_amount -= cost
    return cost


def _is_int(value: object) -> bool:
    return isinstance(value, int) and not isinstance(value, bool)


def _check(tx: ShareTransaction, previous: Optional[str]) -> None:
    epoch_day_from_iso(tx.date)
    if previous is not None and tx.date < previous:
        raise ValueError("transactions must be in date order: %s comes after %s" % (tx.date, previous))
    if tx.kind not in ("buy", "sell"):
        raise ValueError('kind must be buy or sell, received "%s"' % (tx.kind,))
    if not _is_int(tx.quantity) or tx.quantity <= 0:
        raise ValueError(
            "quantity must be a whole number greater than zero, received %s on %s" % (tx.quantity, tx.date)
        )
    if tx.amount.currency != "GBP":
        raise ValueError("amounts must be in GBP, received %s on %s" % (tx.amount.currency, tx.date))
    if not _is_int(tx.amount.minor) or tx.amount.minor < 0:
        raise ValueError(
            "amount must be a whole number of pence, not negative, received %s on %s" % (tx.amount.minor, tx.date)
        )
    if tx.kind == "sell" and tx.date < _RULES_START:
        raise ValueError(
            "disposals before %s follow earlier identification rules, received %s" % (_RULES_START, tx.date)
        )


def section104_pool(transactions: Sequence[ShareTransaction]) -> Section104Result:
    """Identify each disposal of shares with acquisitions the way TCGA 1992
    s105-106A require for individuals from 6 April 2008: first with shares
    bought the same day, then with shares bought in the 30 days after the
    disposal (earliest first), then with the Section 104 pool at average cost.

    Same-day matching is resolved for every day before any 30-day matching,
    and 30-day matching for every disposal (earliest disposal first) before
    the pool is walked, because a purchase claimed by an earlier rule never
    enters the pool.
    """
    buys: List[_DayTotal] = []
    sells: List[_DayTotal] = []
    previous: Optional[str] = None
    for tx in transactions:
        _check(tx, previous)
        previous = tx.date
        group = buys if tx.kind == "buy" else sells
        if not group or group[-1].date != tx.date:
            group.append(_DayTotal(tx.date))
        last = group[-1]
        last.quantity += tx.quantity
        last.amount += tx.amount.minor
        last.remaining = last.quantity
        last.remaining_amount = last.amount

    # Nobody can sell shares they do not hold at the end of that day.
    held = 0
    b = 0
    for sell in sells:
        while b < len(buys) and buys[b].date <= sell.date:
            held += buys[b].quantity
            b += 1
        held -= sell.quantity
        if held < 0:
            raise ValueError("cannot sell more shares than are held: %d sold on %s" % (sell.quantity, sell.date))

    matches: Dict[str, List[list]] = {sell.date: [] for sell in sells}

    # 1. Same day: TCGA 1992 s105(1)(b).
    for sell in sells:
        buy = next((candidate for candidate in buys if candidate.date == sell.date), None)
        if buy is None:
            continue
        quantity = min(sell.remaining, buy.remaining)
        if quantity == 0:
            continue
        matches[sell.date].append(["same-day", buy.date, quantity, _take(buy, quantity)])
        sell.remaining -= quantity

    # 2. The next 30 days, earliest acquisition first: TCGA 1992 s106A(5).
    for sell in sells:
        for buy in buys:
            if sell.remaining == 0:
                break
            after = buy.day - sell.day
            if after < 1 or after > 30 or buy.remaining == 0:
                continue
            quantity = min(sell.remaining, buy.remaining)
            matches[sell.date].append(["bed-and-breakfast", buy.date, quantity, _take(buy, quantity)])
            sell.remaining -= quantity

    # 3. Everything else through the Section 104 pool, in date order.
    pool = _DayTotal(_RULES_START)
    disposals: List[ShareDisposal] = []
    nxt = 0
    for sell in sells:
        while nxt < len(buys) and buys[nxt].date <= sell.date:
            pool.remaining += buys[nxt].remaining
            pool.remaining_amount += buys[nxt].remaining_amount
            nxt += 1
        pending = matches[sell.date]
        if sell.remaining > 0:
            if sell.remaining > pool.remaining:
                raise ValueError(
                    "cannot sell more shares than are held: %d sold on %s" % (sell.quantity, sell.date)
                )
            pending.append(["section-104", None, sell.remaining, _take(pool, sell.remaining)])
        disposals.append(_dispose(sell, pending))
    while nxt < len(buys):
        pool.remaining += buys[nxt].remaining
        pool.remaining_amount += buys[nxt].remaining_amount
        nxt += 1
    return Section104Result(disposals=disposals, pool_quantity=pool.remaining, pool_cost=_gbp(pool.remaining_amount))


def _dispose(sell: _DayTotal, pending: List[list]) -> ShareDisposal:
    """Apportion the day's proceeds across its matches by shares, the last
    match taking what is left."""
    proceeds_left = sell.amount
    shares_left = sell.quantity
    allowable = 0
    out: List[ShareMatch] = []
    for rule, acquisition_date, quantity, cost in pending:
        if quantity == shares_left:
            proceeds = proceeds_left
        else:
            proceeds = _share(proceeds_left, quantity, shares_left)
        proceeds_left -= proceeds
        shares_left -= quantity
        allowable += cost
        out.append(
            ShareMatch(
                rule=rule,
                acquisition_date=acquisition_date,
                quantity=quantity,
                proceeds=_gbp(proceeds),
                cost=_gbp(cost),
                gain=_gbp(proceeds - cost),
            )
        )
    return ShareDisposal(
        date=sell.date,
        quantity=sell.quantity,
        proceeds=_gbp(sell.amount),
        allowable_cost=_gbp(allowable),
        gain=_gbp(sell.amount - allowable),
        matches=out,
    )