|
#!/usr/bin/env -S uv run --script |
|
# /// script |
|
# requires-python = ">=3.11" |
|
# dependencies = [ |
|
# "pandas", |
|
# "scipy", |
|
# "yfinance", |
|
# ] |
|
# /// |
|
"""Annualised real (inflation-adjusted) performance of an ETF holding. |
|
|
|
Fetches prices and distributions from Yahoo Finance and Australian CPI |
|
(All Groups, quarterly) from the RBA's Statistical Table G1, builds the |
|
series of dated cashflows implied by the supplied purchases, and reports |
|
both the nominal and real money-weighted return (XIRR). |
|
|
|
Prices are taken in the currency of the chosen ticker; the default |
|
(VGS.AX) is quoted in AUD on the ASX, which matches the AUD CPI series. |
|
If you change --ticker to a foreign listing, be aware that the CPI used |
|
assumes AUD cashflows. |
|
""" |
|
|
|
from __future__ import annotations |
|
|
|
import argparse |
|
import csv |
|
import io |
|
import sys |
|
import urllib.request |
|
from dataclasses import dataclass |
|
from datetime import date, datetime |
|
from pathlib import Path |
|
|
|
import pandas as pd |
|
import yfinance as yf |
|
from scipy.optimize import brentq |
|
|
|
|
|
@dataclass |
|
class Purchase: |
|
on: date |
|
units: float |
|
|
|
|
|
def parse_purchase(s: str) -> Purchase: |
|
sep = ":" if ":" in s else "," |
|
parts = [p.strip() for p in s.split(sep)] |
|
if len(parts) != 2: |
|
raise argparse.ArgumentTypeError( |
|
f"expected DATE:UNITS or DATE,UNITS, got {s!r}" |
|
) |
|
return Purchase( |
|
on=datetime.strptime(parts[0], "%Y-%m-%d").date(), |
|
units=float(parts[1]), |
|
) |
|
|
|
|
|
def load_purchases_csv(path: Path) -> list[Purchase]: |
|
out: list[Purchase] = [] |
|
with path.open() as f: |
|
for row in csv.reader(f): |
|
if not row or row[0].lower().startswith("date"): |
|
continue |
|
out.append( |
|
Purchase( |
|
on=datetime.strptime(row[0].strip(), "%Y-%m-%d").date(), |
|
units=float(row[1].strip()), |
|
) |
|
) |
|
return out |
|
|
|
|
|
def xnpv(rate: float, cashflows: list[tuple[date, float]]) -> float: |
|
t0 = cashflows[0][0] |
|
return sum( |
|
cf / (1.0 + rate) ** ((d - t0).days / 365.0) for d, cf in cashflows |
|
) |
|
|
|
|
|
def xirr(cashflows: list[tuple[date, float]]) -> float: |
|
cashflows = sorted(cashflows, key=lambda x: x[0]) |
|
return brentq(lambda r: xnpv(r, cashflows), -0.9999, 100.0, maxiter=500) |
|
|
|
|
|
RBA_G1_URL = "https://www.rba.gov.au/statistics/tables/csv/g1-data.csv" |
|
|
|
|
|
def fetch_cpi(url: str = RBA_G1_URL, series_id: str = "GCPIAG") -> pd.Series: |
|
"""Fetch Australian quarterly CPI index from the RBA's Table G1. |
|
|
|
Defaults to series GCPIAG: Consumer Price Index, All Groups, original, |
|
sourced from the ABS via the RBA. |
|
""" |
|
with urllib.request.urlopen(url, timeout=30) as resp: |
|
raw = resp.read().decode("utf-8-sig") |
|
lines = raw.splitlines() |
|
header_idx = next( |
|
i |
|
for i, line in enumerate(lines) |
|
if line.lstrip().lower().startswith("series id") |
|
) |
|
df = pd.read_csv(io.StringIO(raw), skiprows=header_idx, header=0) |
|
if series_id not in df.columns: |
|
raise RuntimeError( |
|
f"series {series_id!r} not present in {url} (have {list(df.columns)[:6]}...)" |
|
) |
|
date_col = df.columns[0] |
|
df[date_col] = pd.to_datetime(df[date_col], errors="coerce", dayfirst=True) |
|
df = df.dropna(subset=[date_col]).set_index(date_col).sort_index() |
|
s = pd.to_numeric(df[series_id], errors="coerce").dropna() |
|
s.index = s.index.date |
|
return s |
|
|
|
|
|
def cpi_at(cpi: pd.Series, d: date) -> float: |
|
earlier = cpi[[i for i in cpi.index if i <= d]] |
|
if earlier.empty: |
|
return float(cpi.iloc[0]) |
|
return float(earlier.iloc[-1]) |
|
|
|
|
|
def build_cashflows( |
|
ticker: str, purchases: list[Purchase], as_of: date |
|
) -> tuple[list[tuple[date, float]], float, float, date]: |
|
purchases = sorted(purchases, key=lambda p: p.on) |
|
first = purchases[0].on |
|
|
|
hist = yf.Ticker(ticker).history( |
|
start=first.isoformat(), |
|
end=(as_of + pd.Timedelta(days=1)).isoformat(), |
|
auto_adjust=False, |
|
) |
|
if hist.empty: |
|
raise RuntimeError(f"no price data for {ticker}") |
|
hist.index = pd.to_datetime(hist.index).date |
|
prices, dividends = hist["Close"], hist["Dividends"] |
|
|
|
def price_on_or_after(d: date) -> tuple[date, float]: |
|
for i in prices.index: |
|
if i >= d: |
|
return i, float(prices[i]) |
|
raise RuntimeError(f"no price on or after {d}") |
|
|
|
events: list[tuple[date, str, float]] = [] |
|
for p in purchases: |
|
events.append((p.on, "buy", p.units)) |
|
for d, v in dividends[dividends > 0].items(): |
|
events.append((d, "div", float(v))) |
|
events.sort(key=lambda x: (x[0], 0 if x[1] == "buy" else 1)) |
|
|
|
cashflows: list[tuple[date, float]] = [] |
|
units_held = 0.0 |
|
for d, kind, v in events: |
|
if kind == "buy": |
|
trade_date, px = price_on_or_after(d) |
|
cashflows.append((trade_date, -v * px)) |
|
units_held += v |
|
else: |
|
if units_held > 0: |
|
cashflows.append((d, units_held * v)) |
|
|
|
last_date = prices.index[-1] |
|
last_price = float(prices.iloc[-1]) |
|
cashflows.append((last_date, units_held * last_price)) |
|
return cashflows, units_held, last_price, last_date |
|
|
|
|
|
def deflate( |
|
cashflows: list[tuple[date, float]], cpi: pd.Series, target: date |
|
) -> list[tuple[date, float]]: |
|
target_cpi = cpi_at(cpi, target) |
|
return [(d, cf * target_cpi / cpi_at(cpi, d)) for d, cf in cashflows] |
|
|
|
|
|
def main() -> int: |
|
ap = argparse.ArgumentParser( |
|
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter |
|
) |
|
ap.add_argument( |
|
"--ticker", |
|
default="VGS.AX", |
|
help="Yahoo Finance ticker (default: VGS.AX)", |
|
) |
|
ap.add_argument( |
|
"--purchase", |
|
action="append", |
|
type=parse_purchase, |
|
default=[], |
|
metavar="DATE:UNITS", |
|
help="a purchase as YYYY-MM-DD:UNITS (repeatable)", |
|
) |
|
ap.add_argument( |
|
"--purchases", |
|
type=Path, |
|
metavar="FILE", |
|
help="CSV file with 'date,units' rows (header optional)", |
|
) |
|
ap.add_argument( |
|
"--as-of", |
|
type=lambda s: datetime.strptime(s, "%Y-%m-%d").date(), |
|
default=date.today(), |
|
help="valuation date (default: today)", |
|
) |
|
ap.add_argument( |
|
"--cpi-series", |
|
default="GCPIAG", |
|
help="RBA Table G1 series id (default: GCPIAG — All Groups CPI)", |
|
) |
|
args = ap.parse_args() |
|
|
|
purchases = list(args.purchase) |
|
if args.purchases: |
|
purchases.extend(load_purchases_csv(args.purchases)) |
|
if not purchases: |
|
ap.error("no purchases provided; use --purchase or --purchases") |
|
|
|
cashflows, units, last_price, last_date = build_cashflows( |
|
args.ticker, purchases, args.as_of |
|
) |
|
|
|
invested = -sum(cf for _, cf in cashflows if cf < 0) |
|
distributions = sum(cf for d, cf in cashflows[:-1] if cf > 0) |
|
final_value = cashflows[-1][1] |
|
|
|
nominal = xirr(cashflows) |
|
cpi = fetch_cpi(series_id=args.cpi_series) |
|
real = xirr(deflate(cashflows, cpi, last_date)) |
|
|
|
print(f"Ticker: {args.ticker}") |
|
print(f"Valuation date: {last_date}") |
|
print(f"Units held: {units:,.4f}") |
|
print(f"Last price: {last_price:,.4f}") |
|
print(f"Total invested: {invested:,.2f}") |
|
print(f"Total distributions: {distributions:,.2f}") |
|
print(f"Current value: {final_value:,.2f}") |
|
print(f"Nominal return: {nominal * 100:.2f}% p.a.") |
|
print(f"Real return: {real * 100:.2f}% p.a.") |
|
return 0 |
|
|
|
|
|
if __name__ == "__main__": |
|
sys.exit(main()) |