from typing import Dict, List, Optional, Sequence from .dates_add_days import epoch_day_from_iso from .invest_section_104_pool_types import Section104Result, ShareDisposal, ShareMatch, ShareTransaction from .money_amount import Money #: 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, )