"""CSV / Excel exports. transactions every buy/sell/DRP — the raw ledger holdings current valuation per open position dividends income by payment, with franking credits and the gross-up realised_cgt disposals matched FIFO to their parcels unrealised_cgt what a sale today would look like, per parcel still held performance per-FY invested / value / gain closed_positions one row per instrument fully exited Every money column is AUD at each row's own FX rate unless the header says "native". The discount columns apply the 50% CGT discount per parcel, before losses are netted — the FY report nets losses first and is the number to lodge, and that caveat travels with the file (decisions.md #77). """ from __future__ import annotations import csv import io from decimal import Decimal from sqlalchemy import select from sqlalchemy.orm import Session, selectinload from . import clock, cpi, fyreport, queries from .models import CorporateAction, Dividend, Instrument, Trade ZERO = Decimal(0) Report = tuple[list[str], list[list]] # report key -> (title, builder). `fy` is passed to the ones that take a year. TITLES = { "transactions": "All trades", "holdings": "Current holdings", "dividends": "Dividends and franking", "realised_cgt": "Realised capital gains", "unrealised_cgt": "Unrealised capital gains", "performance": "Performance by financial year", "closed_positions": "Closed positions", } def _d(value: Decimal | float | None, places: str = "0.01"): if value is None: return "" return Decimal(value).quantize(Decimal(places)) def _total(parts) -> Decimal | None: """The sum, or None when any part is unknown: a total short of a part it cannot convert is a wrong number, and a blank is an honest one (#5).""" parts = list(parts) return None if any(p is None for p in parts) else sum(parts, ZERO) def _aud(value: Decimal, fx: Decimal | None) -> Decimal | None: """Convert, or return None so the cell comes out blank. Multiplying by 1 when the rate is unknown would put a native-currency number under a column headed "(AUD)", and a spreadsheet will happily total that column. Blank is the honest answer; the caller resolves what it can through `queries.FxBook` before getting here. """ if fx is None: return None return value * fx # --------------------------------------------------------------------------- # # Reports # --------------------------------------------------------------------------- # def transactions(session: Session, ticker: str | None = None, **_) -> Report: headers = [ "Date", "Ticker", "Name", "Exchange", "Type", "Units", "Price (native)", "Brokerage (native)", "Currency", "FX rate to AUD", "Value (AUD)", "Recorded by", "Note", ] stmt = select(Trade).options(selectinload(Trade.instrument)).order_by(Trade.date, Trade.id) fxbook = queries.FxBook(session) rows = [] for t in session.scalars(stmt): if ticker and t.instrument.ticker != ticker: continue gross = t.quantity * t.unit_price signed = gross + t.brokerage if t.type in ("buy", "drp") else gross - t.brokerage # The FX column shows what the trade actually captured; the AUD column # may additionally fall back to the stored series for that date. rows.append(((t.date, 1, t.id), [ t.date.isoformat(), t.instrument.ticker, t.instrument.name or "", t.instrument.exchange, t.type, _d(t.quantity, "0.00000001"), _d(t.unit_price, "0.000001"), _d(t.brokerage), t.instrument.currency, _d(t.fx_rate, "0.000001"), _d(_aud(signed, fxbook.of(t, t.instrument.currency))), t.user_id or "", t.note or "", ])) # Corporate actions in the same ledger, each before its day's trades (#146). for a in session.scalars(select(CorporateAction) .options(selectinload(CorporateAction.instrument))): if ticker and a.instrument.ticker != ticker: continue money = None if a.amount is None else _aud(a.amount, fxbook.of(a, a.instrument.currency)) rows.append(((a.date, 0, a.id), [ a.date.isoformat(), a.instrument.ticker, a.instrument.name or "", a.instrument.exchange, queries.action_label(a), "", "", "", a.instrument.currency, _d(a.fx_rate, "0.000001"), "" if a.amount is None else _d(money), a.user_id or "", a.note or "", ])) return headers, [row for _order, row in sorted(rows, key=lambda pair: pair[0])] def holdings(session: Session, **_) -> Report: headers = [ "Ticker", "Name", "Class", "Currency", "Units", "Avg price (native)", "Last price (native)", "Price date", "Cost base (AUD)", "Value (AUD)", "Unrealised gain (AUD)", "Gain %", "Dividends (native)", "DRP", "Note", ] open_positions, _closed = queries.split_positions(queries.all_holdings(session)) rows = [] for h in open_positions: rows.append([ h.instrument.ticker, h.instrument.name or "", h.instrument.asset_class, h.instrument.currency, _d(h.units, "0.00000001"), _d(h.avg_price, "0.000001"), _d(h.price, "0.000001"), h.price_date.isoformat() if h.price_date else "", _d(h.cost_aud), _d(h.value_aud), _d(h.gain_aud), _d(h.gain_pct * 100, "0.01") if h.gain_pct is not None else "", _d(h.dividends_cash), "yes" if h.drp else "no", h.note or "", ]) return headers, rows def dividends(session: Session, fy: int | None = None, **_) -> Report: headers = [ "Payment date", "Period ended", "Ticker", "Name", "Cash (native)", "Currency", "FX rate to AUD", "Cash (AUD)", "Franking credits (AUD)", "Foreign tax withheld (native)", "Grossed up (AUD)", "Reinvested", "Financial year", "Note", ] stmt = ( select(Dividend) .options(selectinload(Dividend.instrument)) .order_by(Dividend.date, Dividend.id) ) fxbook = queries.FxBook(session) rows = [] for d in session.scalars(stmt): year = fyreport.current_fy(fyreport.income_date(d)) if fy and year != fy: continue rate = fxbook.of(d, d.instrument.currency) cash_aud = _aud(d.cash_amount, rate) withheld_aud = _aud(d.withholding_tax or ZERO, rate) # NOT `or ZERO`. `franking_credits` is None when no statement has ever # supplied one and 0 when the distribution was unfranked, and ABOUT # (below) promises the reader those are different cells. Flattening # them here turned "we don't know" into "there was none" — a tax figure # somebody would have taken at face value. franking = d.franking_credits rows.append([ d.date.isoformat(), d.period_end.isoformat() if d.period_end else "", d.instrument.ticker, d.instrument.name or "", _d(d.cash_amount, "0.0001"), d.instrument.currency, _d(d.fx_rate, "0.000001"), _d(cash_aud), _d(franking), _d(d.withholding_tax, "0.0001"), # No AUD cash figure means no honest total — and neither does an # unrecorded credit, for the same reason: the grossed-up amount IS # cash + franking + tax withheld, so it cannot be known while part # of it is not. _d(None if cash_aud is None or franking is None else cash_aud + franking + withheld_aud), "yes" if d.reinvest_trade_id else "no", fyreport.fy_label(year), d.note or "", ]) return headers, rows # What made a row of the realised gains: a sale, or a capital change beyond a # parcel's cost base, which sold nothing (decisions.md #146). EVENTS = {"sale": "sale", "capital_return": "return of capital", "cost_base": "cost base decrease"} def realised_cgt(session: Session, fy: int | None = None, **_) -> Report: """One row per parcel consumed by a sale — the tax-time report. "Gain after discount" is the per-parcel figure: half the gain when the parcel was held more than twelve months, and from 1 July 2027 half the gain to 30 June 2027 only (decisions.md #150). The FY report applies losses before the discount, so use its total when lodging. """ headers = [ "Sale date", "Ticker", "Name", "Units", "Acquired", "Days held", "Cost base (AUD)", "Proceeds (AUD)", "Gross gain (AUD)", "Discount eligible", "CGT discount (AUD)", "Gain after discount (AUD)", "Financial year", "Event", "Value at 30 Jun 2027 (AUD)", "Gain to 30 Jun 2027 (AUD)", "Indexation factor", "Indexed cost base (AUD)", "Indexation (AUD)", ] instruments = ( session.scalars(select(Instrument).options(selectinload(Instrument.trades), selectinload(Instrument.actions))) .unique() .all() ) fxbook = queries.FxBook(session) rows = [] for inst in instruments: # A sale, or a capital change that can go beyond a cost base (#146). if not (any(t.type == "sell" for t in inst.trades) or any(a.kind != "split" for a in inst.actions)): continue for disposal in fyreport.instrument_disposals(inst, fxbook): year = disposal.date.year + (0 if disposal.date.month <= 6 else 1) if fy and year != fy: continue for parcel in disposal.parcels: gain = parcel.gain # None when either side had no rate discount = parcel.discount rows.append([ disposal.date.isoformat(), inst.ticker, inst.name or "", _d(parcel.quantity, "0.00000001"), parcel.acquired.isoformat(), (disposal.date - parcel.acquired).days, _d(parcel.cost), _d(parcel.proceeds), _d(gain), "yes" if parcel.discountable else "no", _d(discount), _d(None if gain is None else gain - discount), fyreport.fy_label(year), EVENTS[disposal.event], _d(parcel.deferred_proceeds), _d(parcel.deferred), _d(parcel.factor, "0.001"), _d(parcel.indexed_cost if parcel.factor is not None else None), _d(parcel.indexation), ]) rows.sort(key=lambda r: (r[0], r[1])) return headers, rows def unrealised_cgt(session: Session, **_) -> Report: """What a sale at today's price would realise, per parcel still held: worked as the sale simulator works one, so from 1 July 2027 the latest CPI stands in for the quarter's, which is not out yet (decisions.md #150).""" headers = [ "Ticker", "Name", "Acquired", "Units", "Days held", "Cost base (AUD)", "Market value (AUD)", "Unrealised gain (AUD)", "Discount eligible", "Gain after discount (AUD)", "Indexation (AUD)", ] today = clock.today() instruments = ( session.scalars(select(Instrument).options(selectinload(Instrument.trades), selectinload(Instrument.actions))) .unique() .all() ) fxbook = queries.FxBook(session) fxbook.cpi = cpi.CpiBook(session, estimate=True) rows = [] for inst in instruments: latest = queries._latest_price(session, inst.id) if latest is None: continue price, _date = latest fx_now = fxbook.rate(inst.currency, today) # The same FIFO walk as the CGT engine and the holdings, so a holding's # unrealised gain is the sum of these rows (decisions.md #54). for parcel in fyreport.walk(inst, fxbook).held: units = parcel.remaining # A spreadsheet will happily add a USD number to an AUD column, so # a figure we can't convert is left EMPTY rather than converted at # 1:1 — the same refusal the dashboard makes via `totals.excluded`. cost = parcel.cost_held value = None if fx_now is None else units * price * fx_now # `cost` is None only for a currency with no rate stored at all, # and then there is no rate for `value` either. gain = None if value is None or cost is None else value - cost sold = fyreport.as_if_sold(parcel, value, today, fxbook) rows.append([ inst.ticker, inst.name or "", parcel.acquired.isoformat(), _d(units, "0.00000001"), (today - parcel.acquired).days, _d(cost), _d(value), _d(gain), "yes" if sold.discountable else "no", "" if sold.gain is None else _d(sold.gain - sold.discount), _d(sold.indexation), ]) rows.sort(key=lambda r: (r[0], r[2])) return headers, rows def performance(session: Session, **_) -> Report: headers = [ "Financial year", "Invested in year (AUD)", "Invested cumulative (AUD)", "Value at year end (AUD)", "Total return cumulative (AUD)", "Return %", "Return in year (AUD)", "Year return %", "Note", ] annual = queries.annual_series(session) rows = [] for i, label in enumerate(annual["labels"]): left_out = ", ".join(f"{ticker} ({why})" for ticker, why in annual["left_out"][i]) rows.append([ label, _d(annual["invested_yr"][i]), _d(annual["invested_cum"][i]), _d(annual["value_eofy"][i]), _d(annual["gain_cum"][i]), _d(annual["gain_pct"][i] * 100), _d(annual["gain_yr"][i]), _d(annual["yr_pct"][i] * 100), f"Leaves out {left_out}" if left_out else "", ]) return headers, rows def closed_positions(session: Session, **_) -> Report: """Fully exited instruments — outlay, proceeds and realised profit. Capital-gain columns come from the FIFO engine, so they line up with the realised CGT report rather than being a second opinion. """ headers = [ "Ticker", "Name", "Units sold", "First bought", "Last sold", "Outlay incl. brokerage (AUD)", "Proceeds net of brokerage (AUD)", "Dividends (AUD)", "Gross capital gain (AUD)", "CGT discount (AUD)", "Capital gain after discount (AUD)", "Total return incl. dividends (AUD)", "Note", ] _open, closed = queries.split_positions(queries.all_holdings(session)) prefs = queries.prefs_by_instrument(session) fxbook = queries.FxBook(session) rows = [] for position in closed: inst = position.instrument disposals = fyreport.instrument_disposals(inst, fxbook) # Sales, and not a return of capital beyond the cost base, whose gain # (#146) is in the gross figure but is not a sale's proceeds. sales = [d for d in disposals if d.event == "sale"] if not sales: continue cost_base = _total(d.cost for d in sales) proceeds = _total(d.proceeds for d in sales) gross = _total(d.gain for d in disposals) discount = _total(p.discount for d in disposals for p in d.parcels) # The return is what came back over what was paid: indexing lowers # the gain taxed, not the money made (decisions.md #150). indexed = sum((d.indexation or ZERO for d in disposals), ZERO) units = sum((d.quantity for d in sales), ZERO) dividends_aud = _total( queries.in_aud(dv.cash_amount, inst.currency, fxbook.of(dv, inst.currency)) for dv in inst.dividends ) buys = [t.date for t in inst.trades if t.type in ("buy", "drp")] pref = prefs.get(inst.id) rows.append([ inst.ticker, inst.name or "", _d(units, "0.00000001"), min(buys).isoformat(), # a closed position was bought first max(d.date for d in sales).isoformat(), _d(cost_base), _d(proceeds), _d(dividends_aud), _d(gross), _d(discount), _d(None if gross is None or discount is None else gross - discount), _d(None if gross is None or dividends_aud is None else gross + indexed + dividends_aud), (pref.note if pref else "") or "", ]) rows.sort(key=lambda r: r[0]) return headers, rows BUILDERS = { "transactions": transactions, "holdings": holdings, "dividends": dividends, "realised_cgt": realised_cgt, "unrealised_cgt": unrealised_cgt, "performance": performance, "closed_positions": closed_positions, } # --------------------------------------------------------------------------- # # Rendering # --------------------------------------------------------------------------- # # A spreadsheet runs a cell that starts with any of these as a formula. Notes, # names and symbols are typed by people, so neither renderer trusts them # (decisions.md #142). FORMULA_START = ("=", "+", "-", "@", "\t", "\r") def _inert(value): """Text that starts like a formula, quoted so it is shown, not run. Only text: a negative number is a number and keeps its sign.""" if isinstance(value, str) and value.startswith(FORMULA_START): return "'" + value return value def to_csv(headers: list[str], rows: list[list]) -> bytes: buffer = io.StringIO() writer = csv.writer(buffer, lineterminator="\n") writer.writerow([_inert(h) for h in headers]) writer.writerows([_inert(v) for v in row] for row in rows) # utf-8-sig: Excel on Windows opens a plain utf-8 CSV as mojibake. return buffer.getvalue().encode("utf-8-sig") def to_xlsx(sheets: dict[str, Report], note: str | None = None) -> bytes: from openpyxl import Workbook from openpyxl.styles import Alignment, Font from openpyxl.utils import get_column_letter wb = Workbook() wb.remove(wb.active) for title, (headers, rows) in sheets.items(): # Excel sheet names: 31 chars, no []:*?/\ safe = "".join(c for c in title if c not in "[]:*?/\\")[:31] ws = wb.create_sheet(safe) ws.append(headers) for cell in ws[1]: cell.font = Font(bold=True) cell.alignment = Alignment(horizontal="left") for row in rows: ws.append(row) # Every text cell is text. openpyxl makes a formula of a string that # starts with "=", and XLSX, unlike CSV, has a type to say so, so # nothing is added to what the cell shows. for line in ws.iter_rows(): for cell in line: if isinstance(cell.value, str): cell.data_type = "s" ws.freeze_panes = "A2" for i, header in enumerate(headers, start=1): longest = max( [len(str(header))] + [len(str(r[i - 1])) for r in rows[:200]] or [0] ) ws.column_dimensions[get_column_letter(i)].width = min(40, longest + 3) if note: ws = wb.create_sheet("About") for line in note.splitlines(): ws.append([line]) ws.column_dimensions["A"].width = 100 buffer = io.BytesIO() wb.save(buffer) return buffer.getvalue() ABOUT = """Stocktake export Generated by Stocktake, a self-hosted share portfolio tracker. All money columns are AUD unless the header says "native"; each row is converted at its own FX rate, or where it has none, at the nearest one stored. The FY report in the app names any holding where that is more than a week from the day. Capital gains: parcels are matched FIFO. The "gain after discount" columns halve a gain on a parcel held more than twelve months. The ATO applies capital LOSSES before that discount, the year's own and then those carried from earlier years, which the FY report in the app does — use the FY report's net figure when lodging, and these rows as the working. From 1 July 2027 the discount is replaced by indexation (Treasury Laws Amendment (Tax Reform No. 1) Act 2026). A parcel held through 30 June 2027 is taken to be sold just before 1 July 2027 for its value at the last close and bought again for it: its gain to then is "Gain to 30 Jun 2027", discounted as before. A cost base from 1 July 2027 (that value, for such a parcel) is indexed by the CPI when the parcel was held twelve months, counted from when it was bought: "Indexed cost base". Where it changed after, the factor is the amounts' own factors weighed together. Indexing cannot make a loss. "Cost base" stays what the parcel cost, and "Indexation" is what indexing took off its gain, so the gross gain is the proceeds less both. Unrealised gains are worked as a sale today would be, with the latest CPI for this quarter's, which is not out yet. The gains from 1 July 2027 left after losses carry a 30% minimum tax: the FY report gives them as the minimum tax capital gain. Dividends: franking credits are only present where a statement supplied them; blank means not recorded, not zero. The grossed-up column is cash plus franking plus any foreign tax withheld, so it is blank on those rows too rather than repeating the cash figure. Foreign tax withheld is income as well as cash, and a foreign income tax offset. """ def build(session: Session, reports: list[str], fy: int | None = None, ticker: str | None = None) -> dict[str, Report]: # The route refuses a report it does not know before it gets here. return { TITLES[key]: BUILDERS[key](session, fy=fy, ticker=ticker) for key in reports }