"""The fields a custom chart can be built from. A chart spec is four choices: a dimension for the X axis, one or more measures, optionally a second dimension to split into series, and a filter. This declares the vocabulary; `charts_build.py` executes a spec against it. Three grains, which cannot be mixed because they answer different questions: timeseries — one row per day. X is always the date. positions — one row per holding, as it stands now. X is a holding attribute, so these aggregate. periods — one row per performance window. Fixed rows, chosen columns. `grain` on every field is what stops the builder offering "market value by ticker over time" and then having to explain why it cannot draw it. """ from __future__ import annotations from dataclasses import dataclass @dataclass(frozen=True) class Field: key: str label: str grain: str # timeseries | positions role: str # dimension | measure kind: str = "money" # money | percent | number | text | date blurb: str = "" # Measures only: how rows combine when several fall in one bucket. agg: str = "sum" FIELDS: tuple[Field, ...] = ( # ---- daily series ----------------------------------------------------- Field("date", "Date", "timeseries", "dimension", "date", "Every day the market produced a close"), Field("value", "Market value", "timeseries", "measure", "money", "What the holdings were worth that day", agg="last"), Field("invested", "Invested (cumulative)", "timeseries", "measure", "money", "Everything paid in to that day, incl. brokerage", agg="last"), Field("gain", "Total return", "timeseries", "measure", "money", "Value + realised proceeds + cash income − invested", agg="last"), Field("gain_pct", "Return %", "timeseries", "measure", "percent", "Total return over what was put in", agg="last"), Field("flow_in", "Bought", "timeseries", "measure", "money", "Money put in that day", agg="sum"), Field("cash_div", "Cash income", "timeseries", "measure", "money", "Distributions taken as cash, not reinvested", agg="sum"), # Derived measures. Each is a named computation over the base series # (see charts_build._TS_MEASURES) rather than a formula language — the # vocabulary stays small enough that every chart in it can be explained. Field("value_gain", "Value gain", "timeseries", "measure", "money", "Market value less what was put in — unrealised, excludes income", agg="derived"), Field("invested_delta", "Invested in period", "timeseries", "measure", "money", "How much more went in during each bucket", agg="derived"), Field("gain_delta", "Return in period", "timeseries", "measure", "money", "How much the total return moved during each bucket", agg="derived"), Field("period_return", "Return in period %", "timeseries", "measure", "percent", "Gain made in the bucket over what was invested by its end", agg="derived"), # ---- holdings --------------------------------------------------------- Field("ticker", "Ticker", "positions", "dimension", "text"), Field("asset_class", "Asset class", "positions", "dimension", "text", "ETF, share or crypto"), Field("currency", "Currency", "positions", "dimension", "text"), Field("exchange", "Exchange", "positions", "dimension", "text"), Field("pos_value", "Market value", "positions", "measure", "money", "Current value in AUD"), Field("pos_cost", "Cost base", "positions", "measure", "money", "What the units held cost, incl. brokerage, in AUD"), Field("pos_gain", "Unrealised gain", "positions", "measure", "money", "Value less cost base, in AUD"), Field("pos_gain_pct", "Unrealised gain %", "positions", "measure", "percent", "Unrealised gain over cost base", agg="ratio"), Field("pos_dividends", "Dividends", "positions", "measure", "money", "Distributions received, in AUD"), Field("pos_units", "Units", "positions", "measure", "number"), Field("pos_day_change", "Change today", "positions", "measure", "money", "Move since the previous close, in AUD"), Field("pos_day_pct", "Change today %", "positions", "measure", "percent", agg="ratio"), # ---- performance windows --------------------------------------------- # The rows are fixed (1 day … all time); these are the columns. Making them # real fields is what lets the performance summary be edited like anything # else instead of being a black box. Field("period", "Period", "periods", "dimension", "text", "1 day, 1 month, 1 year … all time"), Field("p_from", "Window start", "periods", "measure", "text", "The date the window opens", agg="first"), Field("p_value_then", "Value then", "periods", "measure", "money", "What the portfolio was worth when the window opened", agg="first"), Field("p_contributed", "You invested", "periods", "measure", "money", "Money put in during the window", agg="first"), Field("p_growth", "Market + income", "periods", "measure", "money", "What the market and distributions made, contributions removed", agg="first"), Field("p_on_money_in", "On money in %", "periods", "measure", "percent", "Growth over the capital at work — the familiar figure", agg="first"), Field("p_twr", "Investments %", "periods", "measure", "percent", "Time-weighted: how the holdings performed, contributions removed", agg="first"), Field("p_twr_annual", "Per year %", "periods", "measure", "percent", "The time-weighted return restated per year", agg="first"), ) BY_KEY = {f.key: f for f in FIELDS} # Chart types the builder can draw, and which grains suit them. CHART_TYPES = ( {"key": "line", "label": "Line", "grains": ["timeseries"]}, # "Area" is the same chart with the space under the line filled in, which # was not obvious from the word alone — the label says so. {"key": "area", "label": "Line, filled", "grains": ["timeseries"]}, {"key": "bar", "label": "Bar", "grains": ["timeseries", "positions", "periods"]}, {"key": "hbar", "label": "Horizontal bar", "grains": ["positions", "periods"]}, {"key": "doughnut", "label": "Doughnut", "grains": ["positions"]}, {"key": "table", "label": "Table only", "grains": ["timeseries", "positions", "periods"]}, ) # How a daily series is bucketed along the X axis. BUCKETS = ( {"key": "day", "label": "Day"}, {"key": "week", "label": "Week"}, {"key": "month", "label": "Month"}, {"key": "quarter", "label": "Quarter"}, {"key": "fy", "label": "Financial year"}, ) def fields_for(grain: str) -> list[Field]: return [f for f in FIELDS if f.grain == grain] def catalogue() -> dict: """The vocabulary, shaped for the builder UI.""" return { "grains": [ { "key": "timeseries", "label": "Over time", "blurb": "One row per day — how the whole portfolio moved", }, { "key": "positions", "label": "By holding", "blurb": "One row per holding — how they compare right now", }, { "key": "periods", "label": "By period", "blurb": "One row per window, 1 day to all time — the performance summary", }, ], "fields": [ { "key": f.key, "label": f.label, "grain": f.grain, "role": f.role, "kind": f.kind, "blurb": f.blurb, } for f in FIELDS ], "chart_types": list(CHART_TYPES), "buckets": list(BUCKETS), # Dimensions a chart can be filtered on. Values come from the # portfolio, so the builder fills them at render time. "filterable": ["asset_class", "ticker", "currency", "exchange"], } def validate(spec: dict) -> str | None: """Why this spec can't be drawn, or None if it can. Called before saving and before executing, so a bad spec is refused at the door rather than producing an empty chart nobody can explain. """ # The builder only ever sends strings and a list of them. Anything else is # a hand-made request, and `x in BY_KEY` on a list was a 500. if not isinstance(spec, dict): return "That is not a chart." if any(not isinstance(spec.get(k), (str, type(None))) for k in ("grain", "x", "type", "split")): return "That is not a chart." measures = spec.get("measures") if measures is not None and (not isinstance(measures, list) or not all(isinstance(m, str) for m in measures)): return "That is not a chart." grain = spec.get("grain") if grain not in ("timeseries", "positions", "periods"): return "Pick what the chart is about: over time, by holding, or by period." x = spec.get("x") if x not in BY_KEY or BY_KEY[x].grain != grain or BY_KEY[x].role != "dimension": return "Drop a field on the X axis." measures = [m for m in (spec.get("measures") or []) if m in BY_KEY] if not measures: return "Drop at least one field on Measures." if any(BY_KEY[m].grain != grain or BY_KEY[m].role != "measure" for m in measures): return "Those measures don't belong to the same kind of chart as the X axis." chart_type = spec.get("type", "line") match = next((c for c in CHART_TYPES if c["key"] == chart_type), None) if match is None or grain not in match["grains"]: return f"A {chart_type} chart can't show that kind of data." if chart_type == "doughnut" and len(measures) > 1: return "A doughnut shows one measure at a time." split = spec.get("split") if split and (split not in BY_KEY or BY_KEY[split].role != "dimension"): return "That split isn't a dimension." if split and grain == "timeseries" and BY_KEY[split].grain != "positions": return "Split a daily series by a holding attribute (ticker, class…)." if split: if grain == "timeseries": if BY_KEY[split].key == "date": return "Split by a holding attribute — the date is already the X axis." if len(measures) > 1: return ("Pick one measure when splitting into series — several " "measures across every holding is more lines than anyone " "can read.") if chart_type not in ("line", "area", "bar", "table"): return f"A {chart_type} chart can't show one series per group." elif grain == "positions": if split == x: return "The split and the X axis are the same field." if len(measures) > 1: return "Pick one measure when splitting into series." if chart_type in ("doughnut",): return "A doughnut can't be split — use a stacked bar." elif grain == "periods": # Not a chart this app declines to draw — not a chart. Before # this rule the split was accepted and silently dropped: # decisions.md #32. return "Performance windows can't be split — each row is one window." return None