"""FY report: financial-year income, CGT and a year-end snapshot, plus the AU tax pieces (CGT + dividend/franking income). CGT method: **FIFO parcel matching** (every buy and DRP allocation is a parcel; sells consume the oldest first). The engine keeps per-parcel detail so specific identification can be offered later without schema changes. Rules applied: * cost base = parcel outlay incl. buy brokerage, at trade-date fx * proceeds = sale amount less sell brokerage (apportioned), at sale-date fx * 50% discount for parcels held > 12 months (per parcel, individuals) * within a FY, losses offset non-discountable gains first, then discountable, and the discount applies to what remains (the taxpayer-favourable order) * a year's net capital loss carries forward, and is applied in later years after their own losses and before the discount (s102-5(1)); the app sees only the sales recorded in it * from 1 July 2027 (decisions.md #150): a parcel held through 30 June 2027 is taken to be sold just before 1 July 2027 for its market value and bought again on it, the gain to then deferred to the real sale and discounted; after it, the cost base is indexed by the CPI instead of discounted, and the year's losses reduce the deferred gains first (s102-5(1) as amended) """ from __future__ import annotations import datetime as dt from dataclasses import dataclass, field from decimal import Decimal from sqlalchemy import select from sqlalchemy.orm import Session, selectinload from . import clock, cpi, queries from .models import CorporateAction, Instrument, Price, Trade ZERO = Decimal(0) CENTS = Decimal("0.01") # The day the 2027 rules start (Treasury Laws Amendment (Tax Reform No. 1) Act # 2026): an asset held through 30 June 2027 is taken to be sold just before it # and bought again on it, and a cost base is indexed from it. decisions.md #150. NEW_RULES = dt.date(2027, 7, 1) def fy_bounds(fy_end_year: int) -> tuple[dt.date, dt.date]: """FY2026 = 1 Jul 2025 → 30 Jun 2026.""" return dt.date(fy_end_year - 1, 7, 1), dt.date(fy_end_year, 6, 30) def current_fy(today: dt.date) -> int: """The financial year `today` falls in, named by the year it ends. The AU boundary, and one of the several places T28b's jurisdiction work has to reach — a US install's year ends on 31 December. """ return today.year if today.month <= 6 else today.year + 1 def fy_label(fy_end_year: int) -> str: return f"FY{str(fy_end_year - 1)[-2:]}/{str(fy_end_year)[-2:]}" def income_date(dividend) -> dt.date: """The date that decides a distribution's financial year: its period's end where it has one, as a trust's income belongs to the trust's year (decisions.md #144), else the day it was paid.""" return dividend.period_end or dividend.date def period_problem(period_end: dt.date | None, paid: dt.date) -> str | None: """Why `period_end` cannot be the end of a distribution paid on `paid`.""" if period_end is not None and (period_end > paid or _one_year_after(period_end) < paid): return ("A distribution's period ends on or before its payment date, and within " "a year of it.") return None def _rate(obj, inst: Instrument, fx_book) -> Decimal | None: """AUD per 1 unit for a trade or dividend, or None when none is known. The rate captured on the row, then the stored FX series at its date. An AUD row needs neither (decisions.md #6). A foreign one with no rate anywhere is left out of every total and named on the page — never booked at 1:1, which counts a US dollar as an Australian one (#5). """ if obj.fx_rate is not None: return obj.fx_rate if inst.currency == queries.REPORTING_CURRENCY: return Decimal(1) return fx_book.rate(inst.currency, obj.date, inst.ticker) if fx_book is not None else None def _one_year_after(d: dt.date) -> dt.date: try: return d.replace(year=d.year + 1) except ValueError: # 29 Feb return d.replace(year=d.year + 1, day=28) # ---------- CGT ---------- @dataclass class ParcelUse: acquired: dt.date quantity: Decimal # AUD; None when the parcel's purchase had no rate. From 1 July 2027, for a # parcel held through 30 June 2027, what it was worth then (#150). cost_base: Decimal | None proceeds: Decimal | None # AUD; None when the sale had no rate discountable: bool # held > 12 months at disposal # A sale from 1 July 2027 (decisions.md #150): new_rules: bool = False held_through_2027: bool = False deferred_proceeds: Decimal | None = None # its value just before 1 July 2027 deferred_cost: Decimal | None = None # its cost base then factor: Decimal | None = None # indexation; None when not indexed awaiting_cpi: dt.date | None = None # a quarter whose CPI is not out yet @property def deferred(self) -> Decimal | None: """The gain or loss to 30 June 2027, taxed when the parcel is sold.""" if self.deferred_proceeds is None or self.deferred_cost is None: return None return self.deferred_proceeds - self.deferred_cost @property def indexed_cost(self) -> Decimal | None: if self.cost_base is None or self.factor is None: return self.cost_base return (self.cost_base * self.factor).quantize(CENTS) @property def indexed_gain(self) -> Decimal | None: """The gain from 1 July 2027, over the indexed cost base. Indexing cannot make a loss: a loss is worked from the cost base as it was, and between the two there is neither (s 110-36(1A), the reduced cost base).""" if self.cost_base is None or self.proceeds is None or self.awaiting_cpi: return None if self.proceeds >= self.indexed_cost: return self.proceeds - self.indexed_cost if self.proceeds < self.cost_base: return self.proceeds - self.cost_base return ZERO @property def gain(self) -> Decimal | None: if not self.new_rules: if self.cost_base is None or self.proceeds is None: return None return self.proceeds - self.cost_base after = self.indexed_gain if after is None or not self.held_through_2027: return after return None if self.deferred is None else self.deferred + after @property def cost(self) -> Decimal | None: """What the units cost, as their holding counts it. A parcel held through 30 June 2027 has a cost base from its value then, so this differs from it; proceeds − cost − indexation is the gain (#150).""" if not self.held_through_2027: return self.cost_base if None in (self.deferred_cost, self.cost_base, self.deferred_proceeds): return None return self.deferred_cost + self.cost_base - self.deferred_proceeds @property def indexation(self) -> Decimal | None: """How far indexing lowered the gain, which can be less than it raised the cost base; None when the cost base was not indexed.""" if self.factor is None or self.indexed_gain is None: return None return self.proceeds - self.cost_base - self.indexed_gain @property def discount(self) -> Decimal | None: """The 50% discount on this parcel's own gain, before any loss: from 1 July 2027, on its gain to 30 June 2027 only (#150).""" if self.gain is None: return None halved = self.deferred if self.new_rules else self.gain return halved / 2 if self.discountable and halved and halved > 0 else ZERO @dataclass class Disposal: instrument: Instrument date: dt.date quantity: Decimal proceeds: Decimal | None # AUD, net of sale brokerage; None without a rate parcels: list[ParcelUse] # "sale", or the corporate action whose reduction went beyond a parcel's # cost base: what went beyond it is a gain on the day (CGT events G1 and # E10), and `quantity` is the parcel's units, none of them sold (#146). event: str = "sale" @property def known(self) -> bool: """Whether every figure converts: the sale and each parcel it used had a rate. One that does not is listed but left out of the totals.""" return self.proceeds is not None and all(p.gain is not None for p in self.parcels) @property def cost_base(self) -> Decimal | None: if any(p.cost_base is None for p in self.parcels): return None return sum((p.cost_base for p in self.parcels), ZERO) @property def cost(self) -> Decimal | None: """What its units cost, as their holding counts it (`ParcelUse.cost`).""" if any(p.cost is None for p in self.parcels): return None return sum((p.cost for p in self.parcels), ZERO) @property def gain(self) -> Decimal | None: return sum((p.gain for p in self.parcels), ZERO) if self.known else None @property def deferred_proceeds(self) -> Decimal | None: """What its parcels held through 30 June 2027 were worth then (#150); None when it had none, or one could not be valued.""" held = [p for p in self.parcels if p.held_through_2027] if not held or any(p.deferred_proceeds is None for p in held): return None return sum((p.deferred_proceeds for p in held), ZERO) @property def indexation(self) -> Decimal | None: """How far indexing lowered its gain; None when none of it was indexed.""" indexed = [p.indexation for p in self.parcels if p.factor is not None] if not indexed or None in indexed: return None return sum(indexed, ZERO) @dataclass class _Share: """An amount a parcel's units owe between them, billed by running total as they go so the pieces add up to the whole (see `_Parcel`).""" total: Decimal | None allotted: Decimal = ZERO @property def held(self) -> Decimal | None: return None if self.total is None else self.total - self.allotted def bill(self, consumed: Decimal, quantity: Decimal) -> Decimal | None: if self.total is None: return None owed = (self.total * consumed / quantity).quantize(CENTS) billed = owed - self.allotted self.allotted = owed return billed def restart(self, change: Decimal | None = ZERO) -> Decimal: """Begin again from what is held, moved by `change` (None: unknown). Returns how far that took it below nil, and it is then nil.""" held = self.held self.total = None if held is None or change is None else held + change self.allotted = ZERO if self.total is not None and self.total < 0: beyond, self.total = -self.total, ZERO return beyond return ZERO @dataclass class _Deemed: """A parcel held through 30 June 2027, as taken to be sold just before 1 July 2027 and bought again on it (ITAA 1997 s 112-155).""" value: _Share # its market value then: the deferred gain's proceeds cost: _Share # its cost base then: the deferred gain's cost after: _Share # its cost base from 1 July 2027: that value, moved by later changes @dataclass class _Billed: """What a parcel's units owe as they go: the cost (what was paid), and for a parcel held through 30 June 2027 its three 2027 amounts.""" cost: Decimal | None value: Decimal | None = None deferred_cost: Decimal | None = None after: Decimal | None = None @dataclass class _Parcel: """An acquisition still holding units, and what has been billed from it. `cost_allotted` is why this is a class rather than three loose values. **Rounding each use of a parcel on its own loses cents.** A parcel bought for 10.00 and sold one unit at a time reports 3.33 three times, so the schedule claims a 9.99 cost base against money that was 10.00 — and a cost base a cent light is a gain a cent heavy, on a document somebody hands to an accountant. The running total fixes it: each use is billed the DIFFERENCE between what the parcel owes once that use is included and what it has already given up. Rounding still happens, but it happens once against the total instead of independently per slice, so the pieces always sum to the whole. """ acquired: dt.date remaining: Decimal quantity: Decimal # as acquired; never changes cost: Decimal | None # exact AUD outlay for `quantity`, unrounded; None without a rate cost_allotted: Decimal = ZERO native: Decimal = ZERO # the same outlay in the instrument's own currency trade_id: int | None = None deemed: _Deemed | None = None # The amounts in the cost base from 1 July 2027, as (when, AUD, indexed): # what its indexation factor is weighed over (#150). None: one is unknown. spent: list[tuple[dt.date, Decimal, bool]] | None = field(default_factory=list) @property def cost_held(self) -> Decimal | None: """The AUD cost of the units still held: what has not been billed yet.""" return None if self.cost is None else self.cost - self.cost_allotted @property def native_held(self) -> Decimal: return self.native * self.remaining / self.quantity def restart(self, units: Decimal, native: Decimal = ZERO, cost: Decimal | None = ZERO, on: dt.date | None = None) -> Decimal: """Begin again from what is still held, as `units` of it, with its cost moved by `native` and `cost`, the AUD change (None: unknown) made on `on`. The date stays, so the discount clock does. Returns how far the cost base went below nil, which is a capital gain, and it is then nil: from 1 July 2027 a deemed parcel's cost base is the one from then.""" held = self.cost_held share = self.remaining / self.quantity self.native = max(self.native_held + native, ZERO) self.cost = None if held is None or cost is None else held + cost self.quantity = self.remaining = units self.cost_allotted = ZERO beyond = ZERO if self.cost is not None and self.cost < 0: beyond, self.cost = -self.cost, ZERO if self.spent is not None: self.spent = [(when, amount * share, indexed) for when, amount, indexed in self.spent] if cost is None: self.spent = None elif cost and on is not None and on >= NEW_RULES: # What is spent is indexed; what comes back is not (#150). self.spent.append((on, cost, cost > 0)) if self.deemed is None: return beyond self.deemed.value.restart() self.deemed.cost.restart() return self.deemed.after.restart(cost) def absorb(self, other: _Parcel) -> None: """Take on the cost of a parcel a consolidation left without a unit.""" self.native += other.native self.cost = None if self.cost is None or other.cost is None else self.cost + other.cost if self.spent is not None: self.spent = None if other.spent is None else self.spent + other.spent if self.deemed is not None and other.deemed is not None: for mine, theirs in ((self.deemed.value, other.deemed.value), (self.deemed.cost, other.deemed.cost), (self.deemed.after, other.deemed.after)): mine.total = (None if mine.total is None or theirs.held is None else mine.total + theirs.held) def take(self, units: Decimal) -> _Billed: """Consume `units` from this parcel and bill them their cost (None when it is not known), and their 2027 amounts if it has them.""" self.remaining -= units consumed = self.quantity - self.remaining billed = _Billed(None) if self.cost is not None: owed = (self.cost * consumed / self.quantity).quantize(CENTS) billed.cost = owed - self.cost_allotted self.cost_allotted = owed if self.deemed is not None: billed.value = self.deemed.value.bill(consumed, self.quantity) billed.deferred_cost = self.deemed.cost.bill(consumed, self.quantity) billed.after = self.deemed.after.bill(consumed, self.quantity) return billed @dataclass(frozen=True) class Oversold: """A sale of more units than were held at the time. The trade forms and the importer refuse one (`queries.balance_breach`), but a ledger written before they did can hold one. What it sold has no parcel to match, and neither does any later sale of the holding. """ instrument: Instrument trade: Trade held: Decimal # units held just before it def sentence(self) -> str: return (f"{self.instrument.ticker}'s sell of {queries._units(self.trade.quantity)} " f"units on {self.trade.date.isoformat()} is more than the " f"{queries._units(self.held)} units held then.") class OversoldError(ValueError): """An `Oversold` sale. Matching only the parcels there are would understate the gain, so the engine stops instead of guessing at the missing one.""" def __init__(self, sale: Oversold): super().__init__(f"{sale.instrument.ticker}: sell of {sale.trade.quantity} on " f"{sale.trade.date} exceeds held parcels") self.sale = sale @dataclass class Walk: disposals: list[Disposal] held: list[_Parcel] # what is left after every sale, oldest first def instrument_disposals(inst: Instrument, fx_book=None, until: dt.date | None = None) -> list[Disposal]: """This instrument's sales, each matched FIFO to its parcels: `walk`.""" return walk(inst, fx_book, until).disposals def walk(inst: Instrument, fx_book=None, until: dt.date | None = None) -> Walk: """FIFO-match this instrument's sells against its acquisition parcels, and keep what is left: the cost of the units still held, which is what a holding's unrealised gain is measured from (decisions.md #54). `fx_book` is a `queries.FxBook` used to resolve a trade whose own `fx_rate` was never captured. Without one — the direct call in a test — only the rates captured on the trades are known. A parcel or a sale with no rate keeps its units in the matching, and its AUD figures are None. Every session-taking entry point here passes one. `until` ends the walk after that day, so a later `Oversold` sale does not stop a year that sold nothing after it. """ fifo: list[_Parcel] = [] disposals: list[Disposal] = [] # Timeline order, not entry order: two parcels bought on one day at # different prices must be consumed in the order they were acquired, or a # later sale is matched against the wrong cost base. deemed = False for t in queries.timeline(inst.trades, inst.actions): if until is not None and t.date > until: break if not deemed and t.date >= NEW_RULES: _deem(fifo, inst, fx_book) deemed = True if isinstance(t, CorporateAction): disposals += _act(t, fifo, inst, fx_book) continue fx = _rate(t, inst, fx_book) if t.type in ("buy", "drp"): outlay = t.quantity * t.unit_price + t.brokerage cost = None if fx is None else outlay * fx fifo.append( _Parcel( acquired=t.date, remaining=t.quantity, quantity=t.quantity, cost=cost, native=outlay, trade_id=t.id, # Bought from 1 July 2027: indexed from its own quarter. spent=(None if cost is None else [(t.date, cost, True)] if t.date >= NEW_RULES else []), ) ) continue # sell: oldest parcels consumed first. Both sides of a parcel use are # apportioned rather than rounded on their own — see `_Parcel.take` for # the cost base, and the loop below for the proceeds. net_proceeds = (None if fx is None else ((t.quantity * t.unit_price - t.brokerage) * fx).quantize(CENTS)) remaining = t.quantity uses: list[ParcelUse] = [] while remaining > 0 and fifo: parcel = fifo[0] take = min(parcel.remaining, remaining) billed = parcel.take(take) use = ParcelUse( acquired=parcel.acquired, quantity=take, cost_base=billed.cost, proceeds=ZERO, # apportioned below, once they are all known discountable=t.date > _one_year_after(parcel.acquired), ) if t.date >= NEW_RULES: _split(use, parcel, billed, t.date, fx_book) uses.append(use) remaining -= take if parcel.remaining <= 0: fifo.pop(0) if remaining > 0: raise OversoldError(Oversold(inst, t, held=t.quantity - remaining)) # Running-total differencing, so the parts sum to the whole by # construction. Per-unit-then-round does not — decisions.md #93. sold = ZERO allotted = ZERO for use in uses: sold += use.quantity if net_proceeds is None: use.proceeds = None continue owed = (net_proceeds * sold / t.quantity).quantize(CENTS) use.proceeds = owed - allotted allotted = owed disposals.append( Disposal( instrument=inst, date=t.date, quantity=t.quantity, # The money received, not a sum of rounded parts. The parcels # now add to exactly this; `test_cgt.py` asserts they do. proceeds=net_proceeds, parcels=uses, ) ) # Held past 1 July 2027 with nothing since: what is held is still deemed, # for whatever sells it later. if not deemed and (until or clock.today()) >= NEW_RULES: _deem(fifo, inst, fx_book) return Walk(disposals, fifo) def _deem(fifo: list[_Parcel], inst, fx_book) -> None: """Every parcel held through 30 June 2027, taken to be sold just before 1 July 2027 for its market value and bought again on it (ITAA 1997 s 112-155): its units at the last close in the week before, at that day's rate. Unknown without one, which the year names (#150).""" eve = NEW_RULES - dt.timedelta(days=1) close = fx_book.close_before(inst.id, NEW_RULES) if fx_book is not None else None rate = fx_book.rate(inst.currency, eve, inst.ticker) if fx_book is not None else None for parcel in fifo: parcel.restart(parcel.remaining) value = None if close is None or rate is None else parcel.remaining * close * rate parcel.deemed = _Deemed(value=_Share(value), cost=_Share(parcel.cost_held), after=_Share(value)) parcel.spent = None if value is None else [(NEW_RULES, value, True)] def _split(use: ParcelUse, parcel: _Parcel, billed: _Billed, sold_on: dt.date, fx_book) -> None: """A sale from 1 July 2027: the parcel's deferred gain to then, if it was held through 30 June 2027, and its cost base from then, indexed if it was acquired a year before the sale, counting from the real acquisition (s 114-10(9)).""" use.new_rules = True if parcel.deemed is not None: use.held_through_2027 = True use.deferred_proceeds, use.deferred_cost = billed.value, billed.deferred_cost use.cost_base = billed.after if use.discountable: use.factor, use.awaiting_cpi = _indexation(parcel, sold_on, fx_book) def as_if_sold(parcel: _Parcel, value: Decimal | None, on: dt.date, fx_book) -> ParcelUse: """A parcel still held, worked out as if sold whole on `on` for `value` (AUD), by the rules of that day: what a sale would realise.""" use = ParcelUse(acquired=parcel.acquired, quantity=parcel.remaining, cost_base=parcel.cost_held, proceeds=value, discountable=on > _one_year_after(parcel.acquired)) if on >= NEW_RULES: deemed = parcel.deemed billed = _Billed(parcel.cost_held) if deemed is None else _Billed( parcel.cost_held, deemed.value.held, deemed.cost.held, deemed.after.held) _split(use, parcel, billed, on, fx_book) return use def _indexation(parcel: _Parcel, sold_on: dt.date, fx_book) -> tuple[Decimal | None, dt.date | None]: """The factor a parcel's cost base from 1 July 2027 is indexed by: each amount spent by its own quarter's, weighed by the amounts, and what came back not at all. With it, the quarter whose CPI is missing, if one is.""" if parcel.spent is None: return None, None book = fx_book.cpi if fx_book is not None else None weighed = total = ZERO for when, amount, indexed in parcel.spent: factor = Decimal(1) if indexed: factor = None if book is None else book.factor(when, sold_on) if factor is None: missing = (cpi.quarter(sold_on) if book is None or book.index(sold_on) is None else cpi.quarter(when)) return None, missing weighed += amount * factor total += amount if total <= 0: return None, None return weighed / total, None def _act(action: CorporateAction, fifo: list[_Parcel], inst: Instrument, fx_book) -> list[Disposal]: """A corporate action on the parcels held on its day (decisions.md #146). A split shares the holding's new units between them in proportion, with their costs and dates as they were. A return of capital lowers each one's cost by its share of the units, and an AMIT adjustment raises or lowers it; whatever takes a cost below nil is a gain on the day, as a Disposal.""" held = sum((p.remaining for p in fifo), ZERO) if not held: return [] if action.kind == "split": total, seen, given = queries.split_units(held, action), ZERO, ZERO for i, parcel in enumerate(fifo): seen += parcel.remaining share = (total if i == len(fifo) - 1 else (total * seen / held).quantize(queries.UNIT)) - given given += share parcel.restart(share) # A parcel too small to keep a unit after a consolidation gives its # cost to the next one, so none of the cost is lost. for i, parcel in reversed(list(enumerate(fifo))): if not parcel.remaining and len(fifo) > 1: into = fifo[i + 1] if i + 1 < len(fifo) else fifo[i - 1] into.absorb(parcel) fifo.pop(i) return [] # Paid back, the cash lowers the cost base; an adjustment carries its sign. change = -action.amount if action.kind == "capital_return" else action.amount rate = _rate(action, inst, fx_book) gains, seen, given = [], ZERO, ZERO for parcel in fifo: seen += parcel.remaining part = change * seen / held - given given += part # A reduction with no AUD figure on either side may or may not go # below nil: listed unpriced, so the year names it, as a sale (#5). base = parcel.deemed.after.held if parcel.deemed is not None else parcel.cost_held unpriced = part < 0 and (rate is None or base is None) beyond = parcel.restart(parcel.remaining, part, None if rate is None else part * rate, on=action.date) if beyond or unpriced: gain = beyond.quantize(CENTS) if beyond else None gains.append(Disposal( instrument=inst, date=action.date, quantity=parcel.remaining, proceeds=gain, parcels=[ParcelUse(acquired=parcel.acquired, quantity=parcel.remaining, cost_base=None if unpriced else ZERO, proceeds=gain, discountable=action.date > _one_year_after(parcel.acquired), new_rules=action.date >= NEW_RULES)], event=action.kind)) return gains @dataclass class CgtSummary: disposals: list[Disposal] = field(default_factory=list) # Gross gains on >12m parcels, and the rest. From 1 July 2027 these are the # deferred gains, to 30 June 2027, and the gains after it are apart: # indexed where the parcel was held a year, and never discounted (#150). gains_discountable: Decimal = ZERO gains_other: Decimal = ZERO gains_from_july_2027: Decimal = ZERO losses: Decimal = ZERO # positive number losses_applied_other: Decimal = ZERO losses_applied_discountable: Decimal = ZERO losses_applied_from_july_2027: Decimal = ZERO discount: Decimal = ZERO net_capital_gain: Decimal = ZERO net_capital_loss: Decimal = ZERO # >0 when the FY nets to a loss # Earlier years' net capital losses: what was left at the start of this # year, what this year used of it, and what goes on to the next, this # year's own loss included (s102-5(1)). losses_carried_in: Decimal = ZERO losses_carried_applied: Decimal = ZERO losses_carried_out: Decimal = ZERO # The gains from 1 July 2027 left after the losses: what the 30% minimum # tax is worked on (ITAA 1997 s 119-5). Nil before 1 July 2027. minimum_tax_gain: Decimal = ZERO # Holdings with a sale this year that has no exchange rate behind it: the # sale is listed, and every figure above leaves it out (decisions.md #5). withheld: list[str] = field(default_factory=list) # The same, for want of the CPI of a quarter not published yet, or of a # close in the week before 1 July 2027 to value a parcel at (#150). awaiting_cpi: list[str] = field(default_factory=list) unvalued: list[str] = field(default_factory=list) # A sale of more than was held, which this year's gains depend on. When # there is one, every figure above is zero: the year's gains are refused, # not worked out without it, and the sale is named for somebody to fix. oversold: list[Oversold] = field(default_factory=list) def cgt_instruments(session: Session) -> list[Instrument]: """Every instrument with its trades and actions: what `fy_cgt` walks.""" return session.scalars( select(Instrument).options(selectinload(Instrument.trades), selectinload(Instrument.actions)) ).unique().all() def fy_cgt(session: Session, fy_end_year: int, fx_book=None, instruments=None) -> CgtSummary: """This financial year's capital gains, with every earlier year's net capital loss carried in: each year is worked out in turn from the first sale, so a loss left over reaches the years after it (decisions.md #77). `instruments` replaces what is walked: the sale simulator passes stand-ins carrying a sell that is never recorded (decisions.md #148).""" if instruments is None: instruments = cgt_instruments(session) fx_book = fx_book or queries.FxBook(session) sales = [t.date for inst in instruments for t in inst.trades if t.type == "sell"] first = min(current_fy(min(sales)), fy_end_year) if sales else fy_end_year carried = ZERO left_out = {"withheld": set(), "awaiting_cpi": set(), "unvalued": set()} for year in range(first, fy_end_year + 1): out = _year_cgt(instruments, year, fx_book, carried) if out.oversold: # Refused, and so is every year after it: whether it made a loss # to carry on is not known. return out carried = out.losses_carried_out if year < fy_end_year: for name, tickers in left_out.items(): tickers.update(getattr(out, name)) # A sale left out of an earlier year's figures is left out of the loss it # carried too. for name, tickers in left_out.items(): setattr(out, name, sorted(set(getattr(out, name)) | tickers)) return out def _year_cgt(instruments, fy_end_year: int, fx_book, carried: Decimal) -> CgtSummary: start, end = fy_bounds(fy_end_year) out = CgtSummary(losses_carried_in=carried) for inst in instruments: # A sale, or an action that can take a cost base below nil (#146). if not (any(t.type == "sell" and start <= t.date <= end for t in inst.trades) or any(a.kind != "split" and start <= a.date <= end for a in inst.actions)): continue try: found = instrument_disposals(inst, fx_book, until=end) except OversoldError as err: out.oversold.append(err.sale) continue for d in found: if not (start <= d.date <= end): continue out.disposals.append(d) if not d.known: missing = (out.awaiting_cpi if any(p.awaiting_cpi for p in d.parcels) else out.unvalued if any(p.held_through_2027 and p.deferred_proceeds is None for p in d.parcels) else out.withheld) if inst.ticker not in missing: missing.append(inst.ticker) continue for p in d.parcels: for amount, bucket in _parts(p): if amount >= 0: name = f"gains_{bucket}" setattr(out, name, getattr(out, name) + amount) else: out.losses += -amount if out.oversold: # Refused: gains without the oversold holding are not the year's gains. return CgtSummary(oversold=sorted(out.oversold, key=lambda sale: sale.trade.date)) # This year's losses reduce its gains with no discount first, then the # discountable ones, then those from 1 July 2027; earlier years' losses come # next, in the same order; and only then is the discount applied. Before # 1 July 2027 that is the taxpayer-favourable order of s102-5(1), and from # it the order the amended s102-5(1) sets, the deferred gains first (#150). order = ("other", "discountable", "from_july_2027") left = {b: getattr(out, f"gains_{b}") for b in order} rest = out.losses for b in order: used = min(rest, left[b]) setattr(out, f"losses_applied_{b}", used) left[b] -= used rest -= used unapplied, carry = rest, carried for b in order: used = min(carry, left[b]) left[b] -= used carry -= used out.losses_carried_applied = carried - carry out.discount = left["discountable"] / 2 if unapplied > 0: out.net_capital_loss = unapplied else: out.net_capital_gain = (left["other"] + left["discountable"] - out.discount + left["from_july_2027"]) out.minimum_tax_gain = left["from_july_2027"] out.losses_carried_out = carried - out.losses_carried_applied + unapplied out.disposals.sort(key=lambda d: d.date) for names in (out.withheld, out.awaiting_cpi, out.unvalued): names.sort() return out def _parts(use: ParcelUse) -> list[tuple[Decimal, str]]: """A known use's gains or losses, each with the bucket it goes in: before 1 July 2027 one, discountable or not; from it, the gain to 30 June 2027 if the parcel was held then, deferred and discountable as before, and the indexed gain after (#150).""" if not use.new_rules: return [(use.gain, "discountable" if use.discountable else "other")] parts = [] if use.held_through_2027: parts.append((use.deferred, "discountable" if use.discountable else "other")) parts.append((use.indexed_gain, "from_july_2027")) return parts # ---------- dividend / franking income ---------- @dataclass class IncomeRow: instrument: Instrument cash: Decimal | None # None when a dividend this FY has no exchange rate franking: Decimal franking_missing: int # dividends this FY with no franking data yet # AUD foreign tax withheld, which is income as well (decisions.md #145). # None, as `cash` is, when a dividend this FY has no exchange rate. withholding: Decimal | None = ZERO @property def grossed_up(self) -> Decimal | None: if self.cash is None or self.withholding is None: return None return self.cash + self.franking + self.withholding def fy_income(session: Session, fy_end_year: int, fx_book=None) -> list[IncomeRow]: start, end = fy_bounds(fy_end_year) rows: list[IncomeRow] = [] instruments = session.scalars( select(Instrument).options(selectinload(Instrument.dividends)) ).unique().all() fx_book = fx_book or queries.FxBook(session) for inst in instruments: divs = [d for d in inst.dividends if start <= income_date(d) <= end] if not divs: continue rates = [_rate(d, inst, fx_book) for d in divs] cash = (None if None in rates else sum((d.cash_amount * rate for d, rate in zip(divs, rates)), ZERO)) withheld = (None if None in rates else sum(((d.withholding_tax or ZERO) * rate for d, rate in zip(divs, rates)), ZERO)) franking = sum((d.franking_credits or ZERO for d in divs), ZERO) rows.append( IncomeRow( instrument=inst, cash=cash, franking=franking, franking_missing=sum(1 for d in divs if d.franking_credits is None), withholding=withheld, ) ) rows.sort(key=lambda r: r.instrument.ticker) return rows # ---------- FY holdings snapshot + activity ---------- @dataclass class SnapshotRow: instrument: Instrument units: Decimal invested_cum: Decimal | None # AUD, buys to date (sheet semantics); None if one had no rate price: Decimal | None # native, close on/before the as-of date price_date: dt.date | None value_aud: Decimal | None gain_aud: Decimal | None # unrealised: value less `cost_base` # AUD, what the units held then cost, the oldest sold first (decisions.md # #54). None without a rate, or for a sale of more than was held. cost_base: Decimal | None = None @property def gain_pct(self) -> Decimal | None: """Unrealised gain as a fraction of what the units held cost. Derived rather than stored, so it cannot drift from the two numbers it comes from. **None covers two situations and the template tells them apart**: a cost base of nothing, so a percentage cannot exist — renders **N/A**; or an unpriced holding, where it is computable in principle and simply not known yet — renders an em dash. Returning zero would claim the holding broke even, which is a statement about the data rather than its absence. """ if self.gain_aud is None or not self.cost_base: return None return self.gain_aud / self.cost_base @property def percentage_applies(self) -> bool: """Whether a gain percentage is a meaningful thing to ask for here. A cost base that is unknown for want of a rate still has one.""" return self.cost_base is None or bool(self.cost_base) def _price_at(session: Session, instrument_id: int, asof: dt.date): return session.execute( select(Price.close, Price.date) .where(Price.instrument_id == instrument_id, Price.date <= asof) .order_by(Price.date.desc()) .limit(1) ).first() @dataclass class FyActivity: invested: Decimal = ZERO # AUD buys incl. brokerage buys: int = 0 sells: int = 0 brokerage: Decimal = ZERO proceeds: Decimal = ZERO def fy_report(session: Session, fy_end_year: int) -> dict: start, end = fy_bounds(fy_end_year) today = clock.today() asof = min(end, today) instruments = session.scalars( select(Instrument).options(selectinload(Instrument.trades), selectinload(Instrument.actions)) ).unique().all() snapshot: list[SnapshotRow] = [] activity = FyActivity() fx_book = queries.FxBook(session) # Holdings some AUD figure on this page leaves out for want of a rate, # named once at the top of the page (decisions.md #5). withheld: set[str] = set() # Held at the as-of date with a sale of more than was held behind them. oversold: set[str] = set() for inst in instruments: units = ZERO invested: Decimal | None = ZERO for t in queries.timeline(inst.trades, inst.actions): if t.date > asof: continue if isinstance(t, CorporateAction): if t.kind == "split": units = queries.split_units(units, t) continue fx = _rate(t, inst, fx_book) in_fy = start <= t.date <= end if t.type in ("buy", "drp"): units += t.quantity if t.type == "buy": outlay = t.quantity * t.unit_price + t.brokerage if fx is None: invested = None elif invested is not None: invested += outlay * fx if in_fy: activity.buys += 1 if fx is None: withheld.add(inst.ticker) else: activity.invested += outlay * fx activity.brokerage += t.brokerage * fx else: units -= t.quantity if in_fy: activity.sells += 1 if fx is None: withheld.add(inst.ticker) else: activity.proceeds += (t.quantity * t.unit_price - t.brokerage) * fx activity.brokerage += t.brokerage * fx if units <= 0: continue # What the units held then cost, from the same FIFO walk as the gains. try: held = walk(inst, fx_book, until=asof).held except OversoldError: held = None oversold.add(inst.ticker) cost_base = (None if held is None or any(p.cost_held is None for p in held) else sum((p.cost_held for p in held), ZERO)) price_row = _price_at(session, inst.id, asof) price, price_date = (price_row.close, price_row.date) if price_row else (None, None) # The stored rate nearest the date, and no value at all for a currency # with none: never 1:1 (decisions.md #5). # Named only for a value it converts: with no price there is none. rate = fx_book.rate(inst.currency, asof, inst.ticker if price is not None else None) if (cost_base is None and held is not None) or (price is not None and rate is None): withheld.add(inst.ticker) value = gain = None if price is not None and rate is not None: value = units * price * rate gain = None if cost_base is None else value - cost_base snapshot.append( SnapshotRow( instrument=inst, units=units, invested_cum=invested, price=price, price_date=price_date, value_aud=value, gain_aud=gain, cost_base=cost_base, ) ) income = fy_income(session, fy_end_year, fx_book) cgt = fy_cgt(session, fy_end_year, fx_book) withheld.update(r.instrument.ticker for r in income if r.cash is None) withheld.update(cgt.withheld) return { "fy_end_year": fy_end_year, "label": fy_label(fy_end_year), "start": start, "end": end, "asof": asof, "is_current": end >= today, "snapshot": snapshot, "total_value": sum((r.value_aud for r in snapshot if r.value_aud is not None), ZERO), "total_invested": sum((r.invested_cum for r in snapshot if r.invested_cum is not None), ZERO), "total_cost_base": sum((r.cost_base for r in snapshot if r.cost_base is not None), ZERO), "oversold_held": sorted(oversold), "activity": activity, "income": income, "income_cash": sum((r.cash for r in income if r.cash is not None), ZERO), "income_franking": sum((r.franking for r in income), ZERO), "income_withholding": sum((r.withholding for r in income if r.withholding is not None), ZERO), "franking_missing": sum(r.franking_missing for r in income), "cgt": cgt, "withheld": sorted(withheld), # From 1 July 2027 (#150): the year's rules, and the sales its gains # leave out for want of a quarter's CPI or of a June 2027 close. "new_rules": start >= NEW_RULES, "awaiting_cpi": cgt.awaiting_cpi, "unvalued": cgt.unvalued, # Converted at a rate more than a week from the day: a hole in the # stored series, or one that does not reach back that far (#124). "far": sorted(fx_book.far), } def available_fys(session: Session) -> list[int]: first = session.execute(select(Trade.date).order_by(Trade.date.asc()).limit(1)).scalar() if first is None: return [] return list(range(current_fy(clock.today()), current_fy(first) - 1, -1)) # --------------------------------------------------------------------------- # # Sorting the report's tables # --------------------------------------------------------------------------- # # The key names appear in the URL, so this is an allow-list as well as a lookup. SORTABLE: dict[str, dict[str, object]] = { "snapshot": { "ticker": lambda r: r.instrument.ticker, "units": lambda r: r.units, "price": lambda r: r.price, "cost_base": lambda r: r.cost_base, "value_aud": lambda r: r.value_aud, "gain_aud": lambda r: r.gain_aud, "gain_pct": lambda r: r.gain_pct, }, # Franking and grossed_up are deliberately absent: the FY page no longer # shows those columns, and this is an allow-list, so leaving them would let # a query reorder the table by a column nobody can see. "income": { "ticker": lambda r: r.instrument.ticker, "cash": lambda r: r.cash, "withholding": lambda r: r.withholding, }, "disposals": { "date": lambda d: d.date, "ticker": lambda d: d.instrument.ticker, "quantity": lambda d: d.quantity, "cost_base": lambda d: d.cost, "proceeds": lambda d: d.proceeds, "gain": lambda d: d.gain, }, } def sort_tables(report: dict, sorts: dict) -> None: """Reorder the report's tables in place, per `sorts`. In place because the report is already built and the totals underneath it are computed from the same rows — reordering must not be able to change a number, and rebinding a list cannot. """ from . import sorting for table, sort in sorts.items(): if not sort.key: continue getter = SORTABLE[table][sort.key] if table == "disposals": report["cgt"].disposals = sorting.apply( report["cgt"].disposals, getter, descending=sort.descending) else: report[table] = sorting.apply( report[table], getter, descending=sort.descending)