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,
)