tests/test_fy_report_properties.py
Rama main · réplica de lectura del repositorio original.
"""The FY report against a working-out of its own, over generated histories.
`test_cgt.py` works its figures by hand, on round numbers. A mutation run
showed how much that leaves open: the snapshot's gain could be `value +
invested`, a US sale's proceeds could be divided by the rate, the year's
activity could count trades from other years, and the price could come from
after the year ended, all with every test green.
Each case here is a history from `histories.py`. Every financial year it
reaches is reported by the app and worked out again below from the rules in
fyreport.py's docstring and decisions.md (#5, #7, #93), and the two must agree
to the cent.
"""
from __future__ import annotations
import datetime as dt
from decimal import Decimal
import pytest
from freezegun import freeze_time
from hypothesis import given
from hypothesis import strategies as st
from sqlalchemy import select
import factories as fac
from app import fyreport, tenancy
from app.models import Portfolio, User
from histories import (
CENTS,
ONE,
TODAY,
ZERO,
History,
close_on_or_before,
fx_of,
fy_bounds,
fy_of,
histories,
known,
load,
stored_rate,
worked_disposals,
worked_held,
)
EXACT = Decimal("0.00000001")
def _session(session_factory, portfolio):
s = session_factory()
tenancy.bind(s, *portfolio)
return s
# --------------------------------------------------------------------------- #
# The working-out
# --------------------------------------------------------------------------- #
def _year(history: History, fy: int, carried) -> dict:
start, end = fy_bounds(fy)
disposals = sorted((d for d in worked_disposals(history) if start <= d["date"] <= end),
key=lambda d: d["date"])
gains_disc = gains_other = losses = ZERO
for d in disposals:
if not known(d):
continue # listed, and left out of every figure (decisions.md #5)
for p in d["parcels"]:
gain = p["proceeds"] - p["cost_base"]
if gain < 0:
losses -= gain
elif p["discountable"]:
gains_disc += gain
else:
gains_other += gain
# This year's losses come off the gains that get no discount first, then
# the rest; earlier years' losses next, in the same order; and the
# discount applies to what is left (s102-5(1)).
off_other = min(losses, gains_other)
off_disc = min(losses - off_other, gains_disc)
unapplied = losses - off_other - off_disc
other, disc = gains_other - off_other, gains_disc - off_disc
carried_other = min(carried, other)
carried_disc = min(carried - carried_other, disc)
disc -= carried_disc
net_gain = ZERO if unapplied > 0 else (other - carried_other) + disc / 2
applied = carried_other + carried_disc
return {"disposals": disposals, "gains_discountable": gains_disc,
"gains_other": gains_other, "losses": losses, "discount": disc / 2,
"net_capital_gain": net_gain, "net_capital_loss": max(unapplied, ZERO),
"losses_carried_in": carried, "losses_carried_applied": applied,
"losses_carried_out": carried - applied + max(unapplied, ZERO),
"withheld": sorted({d["ticker"] for d in disposals if not known(d)})}
def _cgt(history: History, fy: int) -> dict:
"""Every year from the first sale, each carrying its net capital loss on."""
sales = [t.date for h in history.holdings for t in h.trades if t.type == "sell"]
first = min(fy_of(min(sales)), fy) if sales else fy
carried, left_out = ZERO, set()
for year in range(first, fy + 1):
out = _year(history, year, carried)
carried = out["losses_carried_out"]
if year < fy:
left_out |= set(out["withheld"])
out["withheld"] = sorted(set(out["withheld"]) | left_out)
return out
def _report(history: History, fy: int) -> dict:
start, end = fy_bounds(fy)
asof = min(end, TODAY)
snapshot, activity = {}, dict(invested=ZERO, buys=0, sells=0, brokerage=ZERO,
proceeds=ZERO)
# decisions.md #5: an AUD figure a row has no rate for is left out of it,
# and the holding named, never booked at 1:1.
withheld = set()
for h in history.holdings:
units, invested = ZERO, ZERO
for t in h.trades:
if t.date > asof:
continue
rate = fx_of(history, h.currency, t)
units += -t.quantity if t.type == "sell" else t.quantity
if t.type == "buy":
invested = (None if invested is None or rate is None else
invested + (t.quantity * t.unit_price + t.brokerage) * rate)
if not start <= t.date <= end or t.type == "drp":
continue
if rate is None:
withheld.add(h.ticker)
else:
activity["brokerage"] += t.brokerage * rate
if t.type == "buy":
activity["buys"] += 1
if rate is not None:
activity["invested"] += (t.quantity * t.unit_price + t.brokerage) * rate
else:
activity["sells"] += 1
if rate is not None:
activity["proceeds"] += (t.quantity * t.unit_price - t.brokerage) * rate
if units > 0:
close = close_on_or_before(h, asof)
rate = stored_rate(history, h.currency, asof)
# What the units held then cost, the oldest sold first (#54).
held = worked_held(history, h, asof)
cost_base = (None if any(p["cost_held"] is None for p in held)
else sum((p["cost_held"] for p in held), ZERO))
if cost_base is None or (close and rate is None):
withheld.add(h.ticker)
# No rate stored for the currency is no AUD value: decisions.md #5.
value = units * close[0] * rate if close and rate is not None else None
gain = None if value is None or cost_base is None else value - cost_base
snapshot[h.ticker] = {
"units": units, "invested": invested, "cost_base": cost_base,
"price": close[0] if close else None, "price_date": close[1] if close else None,
"value": value, "gain": gain,
# A percentage of a cost of nothing does not exist; one of an
# unknown amount exists but is not known.
"gain_pct": gain / cost_base if gain is not None and cost_base else None,
"applies": cost_base is None or bool(cost_base)}
income = {}
for h in history.holdings:
paid = [d for d in h.dividends if start <= d.date <= end]
if paid:
rates = [fx_of(history, h.currency, d) for d in paid]
cash = None if None in rates else sum((d.cash * r for d, r in zip(paid, rates)), ZERO)
if cash is None:
withheld.add(h.ticker)
income[h.ticker] = {
"cash": cash,
"franking": sum((d.franking or ZERO for d in paid), ZERO),
"missing": sum(1 for d in paid if d.franking is None)}
cgt = _cgt(history, fy)
return {"snapshot": snapshot, "activity": activity, "income": income, "cgt": cgt,
"withheld": sorted(withheld | set(cgt["withheld"]))}
# --------------------------------------------------------------------------- #
# What the app says, in the same shape
# --------------------------------------------------------------------------- #
def _round(v):
return None if v is None else Decimal(v).quantize(EXACT)
def _app(report: dict) -> dict:
a = report["activity"]
cgt = report["cgt"]
return {
"snapshot": {r.instrument.ticker: {
"units": _round(r.units), "invested": _round(r.invested_cum),
"cost_base": _round(r.cost_base), "price": r.price,
"price_date": r.price_date, "value": _round(r.value_aud), "gain": _round(r.gain_aud),
"gain_pct": _round(r.gain_pct), "applies": r.percentage_applies}
for r in report["snapshot"]},
"activity": {"invested": _round(a.invested), "buys": a.buys, "sells": a.sells,
"brokerage": _round(a.brokerage), "proceeds": _round(a.proceeds)},
"income": {r.instrument.ticker: {"cash": _round(r.cash), "franking": _round(r.franking),
"missing": r.franking_missing}
for r in report["income"]},
"cgt": {"disposals": [{"ticker": d.instrument.ticker, "date": d.date,
"quantity": _round(d.quantity), "proceeds": d.proceeds,
"parcels": [{"acquired": p.acquired, "quantity": _round(p.quantity),
"cost_base": p.cost_base, "discountable": p.discountable,
"proceeds": p.proceeds} for p in d.parcels]}
for d in cgt.disposals],
"gains_discountable": cgt.gains_discountable, "gains_other": cgt.gains_other,
"losses": cgt.losses, "discount": cgt.discount,
"net_capital_gain": cgt.net_capital_gain,
"net_capital_loss": cgt.net_capital_loss,
"losses_carried_in": cgt.losses_carried_in,
"losses_carried_applied": cgt.losses_carried_applied,
"losses_carried_out": cgt.losses_carried_out, "withheld": cgt.withheld},
"withheld": report["withheld"],
}
def _rounded(expected: dict) -> dict:
"""The working-out in the comparison's shape (quantities and money to 1e-8)."""
snap = {t: {k: (_round(v) if k in ("units", "invested", "cost_base", "value", "gain",
"gain_pct") else v)
for k, v in row.items()} for t, row in expected["snapshot"].items()}
act = {k: (_round(v) if isinstance(v, Decimal) else v)
for k, v in expected["activity"].items()}
inc = {t: {"cash": _round(r["cash"]), "franking": _round(r["franking"]),
"missing": r["missing"]} for t, r in expected["income"].items()}
cgt = dict(expected["cgt"])
cgt["disposals"] = [{**d, "quantity": _round(d["quantity"]),
"parcels": [{**p, "quantity": _round(p["quantity"])}
for p in d["parcels"]]} for d in cgt["disposals"]]
return {"snapshot": snap, "activity": act, "income": inc, "cgt": cgt,
"withheld": expected["withheld"]}
@given(history=histories())
def test_every_financial_year_agrees_with_the_working_out(session_factory, portfolio, history):
s = _session(session_factory, portfolio)
try:
load(s, history)
with freeze_time(TODAY):
years = fyreport.available_fys(s)
first = min(t.date for h in history.holdings for t in h.trades)
assert years == list(range(fy_of(TODAY), fy_of(first) - 1, -1))
for fy in years:
report = fyreport.fy_report(s, fy)
dates = [d.date for d in report["cgt"].disposals]
assert dates == sorted(dates), "the schedule reads in date order"
tickers = [r.instrument.ticker for r in report["income"]]
assert tickers == sorted(tickers)
got = _app(report)
want = _rounded(_report(history, fy))
# Sales on one day in different holdings come in no set order.
for side in (got, want):
side["cgt"]["disposals"].sort(key=lambda d: (d["date"], d["ticker"]))
for part in ("snapshot", "activity", "income", "cgt", "withheld"):
assert got[part] == want[part], f"FY{fy} {part}"
finally:
s.rollback()
s.close()
@given(history=histories())
def test_the_parts_of_every_disposal_add_up(session_factory, portfolio, history):
"""Independent of the working-out: each sale's parcels add to its proceeds,
and a parcel used up has billed exactly its cost, to the cent."""
s = _session(session_factory, portfolio)
try:
rows = load(s, history)
for h in history.holdings:
inst = s.get(type(rows[h.ticker]), rows[h.ticker].id)
s.refresh(inst, ["trades"])
disposals = fyreport.instrument_disposals(inst, _book(s))
for d in disposals:
if d.proceeds is not None:
assert sum((p.proceeds for p in d.parcels), ZERO) == d.proceeds
assert sum((p.quantity for p in d.parcels), ZERO) == d.quantity
billed: dict[dt.date, Decimal] = {}
for d in disposals:
for p in d.parcels:
if p.cost_base is not None:
billed[p.acquired] = billed.get(p.acquired, ZERO) + p.cost_base
sold = sum((t.quantity for t in h.trades if t.type == "sell"), ZERO)
used = ZERO
for t in h.trades:
if t.type == "sell":
continue
if used + t.quantity > sold:
break # FIFO: nothing after this is used up
used += t.quantity
rate = fx_of(history, h.currency, t)
same_day = [x for x in h.trades if x.type != "sell" and x.date == t.date]
if rate is not None and len(same_day) == 1: # one day's parcels share a key
cost = ((t.quantity * t.unit_price + t.brokerage) * rate).quantize(CENTS)
assert billed.get(t.date, ZERO) == cost, t
finally:
s.rollback()
s.close()
def _book(s):
from app import queries
return queries.FxBook(s)
# --------------------------------------------------------------------------- #
# The boundaries the mutation run found nothing guarding
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("day, fy", [
("2026-06-30", 2026), ("2026-07-01", 2027), ("2026-01-01", 2026),
("2025-12-31", 2026), ("2024-02-29", 2024), ("2026-05-31", 2026),
])
def test_the_financial_year_a_day_falls_in(day, fy):
assert fyreport.current_fy(dt.date.fromisoformat(day)) == fy
@given(st.dates(dt.date(2000, 1, 1), dt.date(2099, 12, 31)))
def test_every_day_falls_inside_its_own_financial_year(day):
start, end = fyreport.fy_bounds(fyreport.current_fy(day))
assert start <= day <= end
assert (end - start).days in (364, 365)
def test_a_first_trade_in_june_starts_the_list_at_that_june(session_factory, portfolio):
s = _session(session_factory, portfolio)
try:
inst = fac.make_instrument(s, "ALPHA")
fac.add_trade(s, inst, "2024-06-30", "buy", 1, "1.00")
fac.add_trade(s, inst, "2023-06-30", "buy", 1, "1.00") # earliest, added last
with freeze_time(TODAY):
assert fyreport.available_fys(s) == [2027, 2026, 2025, 2024, 2023]
finally:
s.rollback()
s.close()
def test_selling_a_fraction_more_than_is_held_is_refused():
inst = SimpleInstrument("BTC", [("2026-01-02", "buy", "1.5"), ("2026-02-02", "sell", "1.75")])
with pytest.raises(ValueError, match="exceeds held parcels"):
fyreport.instrument_disposals(inst)
def test_a_net_loss_of_cents_is_still_a_loss(session_factory, portfolio):
s = _session(session_factory, portfolio)
try:
inst = fac.make_instrument(s, "ALPHA")
fac.add_trade(s, inst, "2025-08-01", "buy", 10, "1.00")
fac.add_trade(s, inst, "2025-09-01", "sell", 10, "0.95") # 50 cents down
out = fyreport.fy_cgt(s, 2026)
assert (out.net_capital_loss, out.net_capital_gain) == (Decimal("0.50"), ZERO)
finally:
s.rollback()
s.close()
def test_a_close_on_the_last_day_of_the_year_is_its_price(session_factory, portfolio):
s = _session(session_factory, portfolio)
try:
inst = fac.make_instrument(s, "ALPHA")
fac.add_trade(s, inst, "2025-01-06", "buy", 10, "1.00")
fac.add_prices(s, inst, [("2025-06-27", "1.10"), ("2025-06-30", "1.20"),
("2025-07-01", "1.30")])
with freeze_time(TODAY):
row = fyreport.fy_report(s, 2025)["snapshot"][0]
assert (row.price, row.price_date) == (Decimal("1.20"), dt.date(2025, 6, 30))
finally:
s.rollback()
s.close()
@pytest.mark.parametrize("today, current", [("2026-06-30", 2026), ("2026-07-01", 2027)])
def test_the_last_day_of_june_is_still_in_the_year(session_factory, portfolio, today, current):
s = _session(session_factory, portfolio)
try:
inst = fac.make_instrument(s, "ALPHA")
fac.add_trade(s, inst, "2025-01-06", "buy", 10, "1.00")
with freeze_time(today):
assert fyreport.available_fys(s)[0] == current
assert fyreport.fy_report(s, 2026)["is_current"] is (current == 2026)
finally:
s.rollback()
s.close()
def test_each_table_sorts_by_its_own_column(session_factory, portfolio):
from app import sorting
s = _session(session_factory, portfolio)
try:
for ticker, bought, sold in (("ALPHA", "1.00", "3.00"), ("BETAX", "1.00", "1.50"),
("GAMMA", "1.00", "2.00")):
inst = fac.make_instrument(s, ticker)
fac.add_trade(s, inst, "2025-08-01", "buy", 20, bought)
fac.add_trade(s, inst, "2025-09-01", "sell", 10, sold)
fac.add_prices(s, inst, [("2026-06-30", sold)])
fac.add_dividend(s, inst, "2025-10-01", sold)
with freeze_time(TODAY):
report = fyreport.fy_report(s, 2026)
query = {"disposals_sort": "gain", "disposals_dir": "desc",
"snapshot_sort": "value_aud", "income_sort": "cash", "income_dir": "desc"}
sorts = {t: sorting.read(query, t, keys) for t, keys in fyreport.SORTABLE.items()}
fyreport.sort_tables(report, sorts)
assert [d.instrument.ticker for d in report["cgt"].disposals] == ["ALPHA", "GAMMA",
"BETAX"]
assert [r.instrument.ticker for r in report["snapshot"]] == ["BETAX", "GAMMA", "ALPHA"]
assert [r.instrument.ticker for r in report["income"]] == ["ALPHA", "GAMMA", "BETAX"]
finally:
s.rollback()
s.close()
class SimpleInstrument:
"""Enough of an Instrument for `instrument_disposals`, with no database."""
def __init__(self, ticker, trades):
from types import SimpleNamespace as NS
self.ticker, self.currency = ticker, "AUD"
self.trades = [NS(id=i + 1, date=dt.date.fromisoformat(d), time=None, type=k,
quantity=Decimal(q), unit_price=ONE, brokerage=ZERO, fx_rate=ONE)
for i, (d, k, q) in enumerate(trades)]
self.actions = []
def test_the_owner_fixture_is_what_the_cases_are_bound_to(session_factory, portfolio):
with session_factory() as s:
assert s.get(Portfolio, portfolio[0]) is not None
assert s.scalar(select(User.email)) == "owner@example.test"