app/api.py
Rama main · réplica de lectura del repositorio original.
"""/api/v1 — machine access to the portfolios a key was given, by API key.
JSON in, JSON out, no cookies and no CSRF: a key is only ever sent
deliberately, so there is no ambient authority to abuse.
`Authorization: Bearer pfk_...` or `X-API-Key: pfk_...`. A key belongs to the
person who made it and reaches the portfolios it was given, each no further
than that person can reach it today (decisions.md #128). A key given more than
one says which with `?portfolio=<id>` on every request; `/portfolios` lists
them. `auth.api_session` decides, and `tenancy` enforces it.
Amounts are AUD unless a `currency` field says otherwise; dates are ISO-8601.
Append-only, deliberately — there is no PATCH or DELETE to find:
decisions.md #88.
"""
from __future__ import annotations
import datetime as dt
from decimal import Decimal
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel, Field
from sqlalchemy import select
from . import auth, clock, fyreport, money, plans, pricefeed, queries
from .models import (
WRITE_ROLES,
CorporateAction,
Dividend,
Instrument,
Portfolio,
Trade,
currency_problem,
exchange_problem,
ticker_problem,
)
from .tenancy import owned
router = APIRouter(prefix="/api/v1", tags=["api"])
def _num(value: Decimal | None) -> float | None:
# Eight places, the finest any column stores (a quantity, to a
# hundred-millionth of a coin), so no stored digit is lost on the way out.
return round(float(value), 8) if value is not None else None
@router.get("/portfolios")
def get_portfolios(request: Request) -> dict:
"""The portfolios this key reaches now, and what it may do in each — how a
client learns the ids to pass as `?portfolio=`. One it was given but whose
creator has since lost access is left out, as it would refuse every call."""
with auth.api_session(request, choose=False) as (access, db):
key = access.key
return {"portfolios": [
{"id": portfolio.id, "name": portfolio.name,
"access": "read,write" if key.can_write and role in WRITE_ROLES else "read"}
for portfolio, role in auth.key_reach(db, key) if role is not None
]}
@router.get("/portfolio")
def get_portfolio(request: Request) -> dict:
"""Headline numbers — what a budgeting app wants for a net-worth line.
`cost` is what the units held cost, the oldest sold first, and `gain` is
the unrealised gain on them (decisions.md #54)."""
with auth.api_session(request) as (access, db):
portfolio = db.get(Portfolio, access.portfolio_id)
holdings = queries.all_holdings(db)
open_positions, closed = queries.split_positions(holdings)
totals = queries.totals(open_positions)
return {
"portfolio": {"id": portfolio.id, "name": portfolio.name,
"income_tax_rate": _num(portfolio.income_tax_rate),
"medicare_levy": _num(portfolio.medicare_levy),
"income_support": portfolio.income_support},
"as_at": clock.today().isoformat(),
"currency": "AUD",
"cost": _num(totals.cost),
"value": _num(totals.value),
"gain": _num(totals.gain),
"gain_pct": _num(totals.gain_pct),
"day_change": _num(totals.day_change),
"day_change_pct": _num(totals.day_pct),
"dividends_received": _num(totals.dividends),
"open_positions": len(open_positions),
"closed_positions": len(closed),
# Tickers with no usable price/FX yet — their value is NOT in the
# totals above, so a consumer can flag rather than silently
# under-report.
"excluded": totals.excluded,
# Left out for a sale of more than was held, which has no cost.
"oversold": totals.oversold,
# Converted at a rate more than a week from the day (#124).
"distant_rates": totals.far,
}
@router.get("/holdings")
def get_holdings(request: Request) -> dict:
"""Open positions. `cost_aud` is what the units held cost, the oldest sold
first, and `gain_aud` the unrealised gain on them (decisions.md #54)."""
with auth.api_session(request) as (access, db):
open_positions, _ = queries.split_positions(queries.all_holdings(db))
return {
"holdings": [
{
"ticker": h.instrument.ticker,
"name": h.instrument.name,
"exchange": h.instrument.exchange,
"asset_class": h.instrument.asset_class,
"currency": h.instrument.currency,
"units": _num(h.units),
"avg_price": _num(h.avg_price),
"last_price": _num(h.price),
"price_date": h.price_date.isoformat() if h.price_date else None,
"cost_aud": _num(h.cost_aud),
"value_aud": _num(h.value_aud),
"gain_aud": _num(h.gain_aud),
"gain_pct": _num(h.gain_pct),
"day_change_aud": _num(h.day_change_aud),
"dividends": _num(h.dividends_cash),
}
for h in open_positions
]
}
@router.get("/trades")
def get_trades(
request: Request, since: str | None = None, ticker: str | None = None,
exchange: str | None = None,
) -> dict:
with auth.api_session(request) as (access, db):
stmt = select(Trade).order_by(Trade.date, Trade.id)
if since:
try:
stmt = stmt.where(Trade.date >= dt.date.fromisoformat(since))
except ValueError:
raise HTTPException(400, "since must be YYYY-MM-DD")
rows = db.scalars(stmt).all()
if ticker:
wanted = ticker.strip().upper()
rows = [t for t in rows if t.instrument.ticker == wanted]
if exchange:
wanted = exchange.strip().upper()
rows = [t for t in rows if t.instrument.exchange == wanted]
return {
"trades": [
{
"id": t.id,
"date": t.date.isoformat(),
"ticker": t.instrument.ticker,
"exchange": t.instrument.exchange,
"type": t.type,
"units": _num(t.quantity),
"unit_price": _num(t.unit_price),
"brokerage": _num(t.brokerage),
"currency": t.instrument.currency,
"fx_rate": _num(t.fx_rate),
"note": t.note,
}
for t in rows
]
}
@router.get("/dividends")
def get_dividends(request: Request, since: str | None = None) -> dict:
with auth.api_session(request) as (access, db):
stmt = select(Dividend).order_by(Dividend.date, Dividend.id)
if since:
try:
stmt = stmt.where(Dividend.date >= dt.date.fromisoformat(since))
except ValueError:
raise HTTPException(400, "since must be YYYY-MM-DD")
return {
"dividends": [
{
"id": d.id,
"date": d.date.isoformat(),
"ticker": d.instrument.ticker,
"exchange": d.instrument.exchange,
"cash_amount": _num(d.cash_amount),
"currency": d.instrument.currency,
"fx_rate": _num(d.fx_rate),
"franking_credits": _num(d.franking_credits),
"period_end": d.period_end.isoformat() if d.period_end else None,
"withholding_tax": _num(d.withholding_tax),
"reinvested": d.reinvest_trade_id is not None,
"note": d.note,
}
for d in db.scalars(stmt).all()
]
}
@router.get("/actions")
def get_actions(request: Request) -> dict:
"""Splits, consolidations, returns of capital and AMIT cost base
adjustments. Every figure the API gives already takes them in (#146): a
client counting units from `/trades` needs these too."""
with auth.api_session(request) as (access, db):
rows = db.scalars(
select(CorporateAction).order_by(CorporateAction.date, CorporateAction.id)).all()
return {
"actions": [
{
"id": a.id,
"date": a.date.isoformat(),
"ticker": a.instrument.ticker,
"exchange": a.instrument.exchange,
"kind": a.kind,
"ratio_new": a.ratio_new,
"ratio_old": a.ratio_old,
"amount": _num(a.amount),
"currency": a.instrument.currency,
"fx_rate": _num(a.fx_rate),
"note": a.note,
}
for a in rows
]
}
@router.get("/fy/{year}")
def get_fy(request: Request, year: int) -> dict:
"""FY summary incl. CGT — the retirement/tax planning view.
`year` is the FY *ending* year: 2026 = 1 Jul 2025 → 30 Jun 2026.
"""
if not 2000 <= year <= 2100:
raise HTTPException(400, "year out of range")
with auth.api_session(request) as (access, db):
r = fyreport.fy_report(db, year)
cgt = r["cgt"]
return {
"fy": r["label"],
"start": r["start"].isoformat(),
"end": r["end"].isoformat(),
"as_at": r["asof"].isoformat(),
"is_current": r["is_current"],
# Holdings every total below leaves out, for want of an exchange
# rate: the figures are short of them, not wrong (decisions.md #5).
"withheld": r["withheld"],
# Holdings converted at a rate more than a week from the day (#124).
"distant_rates": r["far"],
"value": _num(r["total_value"]),
"invested": _num(r["total_invested"]),
# What the units held at the as-of date cost, the oldest sold
# first: their gain is unrealised (decisions.md #54).
"cost_base": _num(r["total_cost_base"]),
"invested_this_fy": _num(r["activity"].invested),
"proceeds_this_fy": _num(r["activity"].proceeds),
"brokerage_this_fy": _num(r["activity"].brokerage),
"income": {
"cash": _num(r["income_cash"]),
"franking_credits": _num(r["income_franking"]),
# Income as well as cash, and a foreign income tax offset (#145).
"foreign_tax_withheld": _num(r["income_withholding"]),
"grossed_up": _num(r["income_cash"] + r["income_franking"]
+ r["income_withholding"]),
"dividends_missing_franking": r["franking_missing"],
},
"cgt": {
# A sale of more units than were held then, which this year's
# gains depend on, refuses them: every figure is null, and the
# sale is named below for a person to fix.
**{key: None if cgt.oversold else value for key, value in {
"disposals": len(cgt.disposals),
"gains_discountable": _num(cgt.gains_discountable),
"gains_other": _num(cgt.gains_other),
# From 1 July 2027 the two above are the gains to 30 June
# 2027, deferred to the sale, and these the gains after it,
# which the 30% minimum tax is worked on once losses are
# off (decisions.md #150).
"gains_from_july_2027": _num(cgt.gains_from_july_2027),
"minimum_tax_gain": _num(cgt.minimum_tax_gain),
"losses": _num(cgt.losses),
"discount": _num(cgt.discount),
"net_capital_gain": _num(cgt.net_capital_gain),
"net_capital_loss": _num(cgt.net_capital_loss),
# Earlier years' net capital losses: at the start of the
# year, used this year, and carried on (s102-5(1)).
"losses_carried_in": _num(cgt.losses_carried_in),
"losses_carried_applied": _num(cgt.losses_carried_applied),
"losses_carried_forward": _num(cgt.losses_carried_out),
# Holdings the totals leave out for want of the CPI for a
# quarter, or of a close in the week before 1 July 2027.
"awaiting_cpi": cgt.awaiting_cpi,
"unvalued": cgt.unvalued,
}.items()},
"oversold": [
{
"ticker": sale.instrument.ticker,
"exchange": sale.instrument.exchange,
"date": sale.trade.date.isoformat(),
"units": _num(sale.trade.quantity),
"held": _num(sale.held),
}
for sale in cgt.oversold
],
},
}
@router.get("/schedule")
def get_plan(request: Request) -> dict:
"""Upcoming scheduled buys — what a budgeting app needs to reserve cash."""
with auth.api_session(request) as (access, db):
sched = plans.schedule(db, upcoming=6)
plan = sched["plan"]
return {
"plan": None
if plan is None
else {
"name": plan.name,
"interval_days": plan.interval_days,
"amount": _num(plan.amount),
"brokerage": _num(plan.brokerage),
"rotation": [e.instrument.ticker for e in plan.entries],
},
"upcoming": [
{
"ticker": b.ticker,
"exchange": b.exchange,
"due_date": b.due_date.isoformat(),
"amount": _num(b.amount),
}
for b in sched["next"]
],
}
# --------------------------------------------------------------------------- #
# Writes (keys with the "write" scope)
# --------------------------------------------------------------------------- #
# Every figure has an upper limit too, taken from its column: pydantic refuses
# NaN and Infinity, but `1E+999999` is finite, and SQLite stores it as a float
# that reads back as Infinity (and then every total fails) while Postgres
# refuses it with an overflow. The same rule the forms use, money.parse.
class NewInstrument(BaseModel):
"""Where an instrument trades: needed for one this portfolio has not got
yet, which is then added as the app's add form adds it. Ignored for an
instrument the shared catalogue already has (#139)."""
# Also needed for a ticker this portfolio holds on more than one exchange.
exchange: str | None = Field(default=None, max_length=Instrument.exchange.type.length)
currency: str | None = Field(default=None, max_length=Instrument.currency.type.length)
asset_class: str = Field(default="share", pattern="^(etf|share|crypto)$")
class TradeIn(NewInstrument):
ticker: str
type: str = Field(pattern="^(buy|sell|drp)$")
date: dt.date
units: Decimal = Field(gt=0, lt=money.limit(Trade.quantity))
# `ge`, not `gt`: a free parcel is a real thing (a bonus issue, a
# reward-plan grant), and the forms accept one. Negative is still refused.
unit_price: Decimal = Field(ge=0, lt=money.limit(Trade.unit_price))
brokerage: Decimal = Field(default=Decimal(0), ge=0,
lt=money.limit(Trade.brokerage))
fx_rate: Decimal | None = Field(default=None, gt=0, lt=money.limit(Trade.fx_rate))
# Text is held to its column too: Postgres refuses anything longer.
note: str | None = Field(default=None, max_length=Trade.note.type.length)
class DividendIn(NewInstrument):
ticker: str
date: dt.date
cash_amount: Decimal = Field(gt=0, lt=money.limit(Dividend.cash_amount))
franking_credits: Decimal | None = Field(
default=None, ge=0, lt=money.limit(Dividend.franking_credits))
fx_rate: Decimal | None = Field(default=None, gt=0, lt=money.limit(Dividend.fx_rate))
# A trust's: the financial year holding it is the one the income is for.
period_end: dt.date | None = None
# Foreign tax kept back before `cash_amount` was paid, in the same currency.
withholding_tax: Decimal | None = Field(
default=None, ge=0, lt=money.limit(Dividend.withholding_tax))
note: str | None = Field(default=None, max_length=Dividend.note.type.length)
def _instrument(request: Request, db, body: TradeIn | DividendIn) -> Instrument:
"""This portfolio's instrument for `body`, adding it if `body` says where it
trades. Whether another portfolio has it already never changes the answer:
saying so would tell this one what the other holds (#139)."""
problem = ticker_problem(body.ticker)
if problem:
raise HTTPException(400, problem)
ticker = body.ticker.strip().upper()
found = queries.held_as(db, ticker, body.exchange)
if len(found) > 1:
# One code on two exchanges here: say which, rather than taking the older.
raise HTTPException(409, f"{found[0].ticker} is held on "
f"{' and '.join(i.exchange for i in found)} — say which in "
"`exchange`.")
if found:
return found[0]
if not request.app.state.settings.imports.allow_new_instruments:
raise HTTPException(404, f"{ticker} isn't in this portfolio. Add it in the app first.")
if not (body.exchange and body.currency):
raise HTTPException(404, f"{ticker} isn't in this portfolio. To add it, give its "
"exchange and currency.")
exchange, currency = body.exchange.strip().upper(), body.currency.strip().upper()
problem = exchange_problem(exchange) or currency_problem(currency)
if problem:
raise HTTPException(400, problem)
inst = db.scalar(select(Instrument).where(Instrument.exchange == exchange,
Instrument.ticker == ticker))
if inst is None:
inst = Instrument(ticker=ticker, exchange=exchange, currency=currency,
asset_class=body.asset_class, drp=False,
yahoo_symbol=pricefeed.yahoo_symbol_for(ticker, exchange))
db.add(inst)
# A removed one comes back, as it does when a trade is typed in.
inst.active = True
db.flush()
return inst
class TaxRatesIn(BaseModel):
# Percent, as the settings form takes them. A field left out is kept, and
# null clears it.
income_tax_rate: Decimal | None = Field(default=None, ge=0, le=100, decimal_places=2)
medicare_levy: Decimal | None = Field(default=None, ge=0, le=100, decimal_places=2)
# Whether the taxpayer gets an income support payment this year, which
# lifts the minimum tax (decisions.md #151). Never null: it is yes or no.
income_support: bool = False
@router.put("/portfolio/tax-rates")
def set_tax_rates(request: Request, body: TaxRatesIn) -> dict:
"""The rates the sale simulator taxes a gain at (decisions.md #147): the
taxpayer's top income tax rate, and the Medicare levy; and whether they get
income support, which lifts the minimum tax (#151). Settings, not records,
so the one write here that is not an append (#88); the owner's, as every
portfolio setting is (#34)."""
with auth.api_session(request, write=True) as (access, db):
if access.role != "owner":
raise HTTPException(403, "only the portfolio's owner can change its settings")
portfolio = db.get(Portfolio, access.portfolio_id)
for field in body.model_fields_set:
setattr(portfolio, field, getattr(body, field))
db.flush()
return {"income_tax_rate": _num(portfolio.income_tax_rate),
"medicare_levy": _num(portfolio.medicare_levy),
"income_support": portfolio.income_support}
@router.post("/trades", status_code=201)
def create_trade(request: Request, body: TradeIn) -> dict:
with auth.api_session(request, write=True) as (access, db):
inst = _instrument(request, db, body)
if body.date > clock.today():
raise HTTPException(400, "date is in the future")
trade = owned(
db,
Trade(
instrument_id=inst.id,
date=body.date,
type=body.type,
quantity=body.units,
unit_price=body.unit_price,
brokerage=body.brokerage,
fx_rate=Decimal(1) if inst.currency == "AUD" else body.fx_rate,
note=body.note or "via API",
),
)
# Same guard, and the same explanation, as the web form: a sell that
# drives the running balance negative would later break the FY report's
# FIFO matcher.
held = db.scalars(select(Trade).where(Trade.instrument_id == inst.id)).all()
actions = db.scalars(select(CorporateAction)
.where(CorporateAction.instrument_id == inst.id)).all()
breach = queries.balance_breach(list(held), trade, actions=list(actions))
if breach is not None:
raise HTTPException(409, queries.breach_message(inst.ticker, breach))
db.add(trade)
db.flush()
return {"id": trade.id, "ticker": inst.ticker, "exchange": inst.exchange,
"date": trade.date.isoformat()}
@router.post("/dividends", status_code=201)
def create_dividend(request: Request, body: DividendIn) -> dict:
problem = fyreport.period_problem(body.period_end, body.date)
if problem:
raise HTTPException(400, problem)
with auth.api_session(request, write=True) as (access, db):
inst = _instrument(request, db, body)
existing = db.scalar(
select(Dividend).where(
Dividend.instrument_id == inst.id,
Dividend.date == body.date,
Dividend.cash_amount == body.cash_amount,
)
)
if existing is not None:
raise HTTPException(
409, f"a {inst.ticker} dividend of {body.cash_amount} on {body.date} exists"
)
dividend = owned(
db,
Dividend(
instrument_id=inst.id,
date=body.date,
cash_amount=body.cash_amount,
fx_rate=Decimal(1) if inst.currency == "AUD" else body.fx_rate,
franking_credits=body.franking_credits,
period_end=body.period_end,
withholding_tax=body.withholding_tax,
note=body.note or "via API",
),
)
db.add(dividend)
db.flush()
return {"id": dividend.id, "ticker": inst.ticker, "exchange": inst.exchange,
"date": body.date.isoformat()}