tests/test_series_properties.py
Rama main · réplica de lectura del repositorio original.
"""The daily series behind every chart, against a working-out of its own.
A mutation run left the series' arithmetic almost entirely unguarded: a US
sale's proceeds or a dividend could be divided by its rate, and the holding
left out of a day it cannot be valued on could keep its proceeds, with every
test green. Worked out again here, day by day, from the rules in
`portfolio_series`'s docstring and decisions.md #5 and #54:
- invested is the cumulative cost of buys with brokerage, at each trade's rate;
- value is units at the latest close on or before the day, at that day's rate;
- a holding with units but no close yet leaves the day whole — its cost, its
proceeds and its income — rather than charting as a total loss;
- a currency with no stored rate at all is left out of everything, by name.
A chart split by ticker or class follows the same rules per line, so its lines
add up to the total. It did not: a holding not yet priced charted as a loss of
its whole cost in the split, the cliff the total was fixed for.
"""
from __future__ import annotations
import datetime as dt
import pytest
from hypothesis import given
from app import queries, tenancy
from histories import ZERO, History, fx_of, histories, load, stored_rate
def _worked(history: History, group=None) -> dict:
"""The series, by day, as {label: {"value": [...], ...}} plus the dates."""
excluded = sorted(h.ticker for h in history.holdings
if h.currency != "AUD" and not history.usd)
priced = [h for h in history.holdings if h.ticker not in excluded]
events = sorted([(t.date, "t", h, t) for h in priced for t in h.trades]
+ [(d.date, "d", h, d) for h in priced for d in h.dividends],
key=lambda e: e[0])
if not events:
return {"dates": [], "lines": {}, "excluded": excluded}
# Every stored close makes a day, an excluded holding's included.
dates = sorted({d for h in history.holdings for d, _c in h.closes if d >= events[0][0]})
label = (lambda h: "all") if group is None else group
lines = {label(h): {k: [] for k in ("value", "invested", "gain", "gain_pct", "flow_in",
"cash_div")} for h in priced}
units, spent, taken, earned = {}, {}, {}, {}
applied = 0
for day in dates:
flow_in = {k: ZERO for k in lines}
income = {k: ZERO for k in lines}
while applied < len(events) and events[applied][0] <= day:
_when, kind, h, row = events[applied]
applied += 1
rate = fx_of(history, h.currency, row)
if kind == "d":
cash = row.cash * rate
earned[h.ticker] = earned.get(h.ticker, ZERO) + cash
income[label(h)] += cash
elif row.type == "sell":
units[h.ticker] = units.get(h.ticker, ZERO) - row.quantity
got = (row.quantity * row.unit_price - row.brokerage) * rate
taken[h.ticker] = taken.get(h.ticker, ZERO) + got
else:
units[h.ticker] = units.get(h.ticker, ZERO) + row.quantity
if row.type == "buy":
cost = (row.quantity * row.unit_price + row.brokerage) * rate
spent[h.ticker] = spent.get(h.ticker, ZERO) + cost
flow_in[label(h)] += cost
value = {k: ZERO for k in lines}
kept = {k: [ZERO, ZERO, ZERO] for k in lines} # invested, proceeds, income
for h in priced:
u = units.get(h.ticker, ZERO)
usable = [c for d, c in h.closes if d <= day]
if u and not usable:
continue # not valued today, so not counted at all
part = kept[label(h)]
part[0] += spent.get(h.ticker, ZERO)
part[1] += taken.get(h.ticker, ZERO)
part[2] += earned.get(h.ticker, ZERO)
if u:
value[label(h)] += u * usable[-1] * stored_rate(history, h.currency, day)
for k, line in lines.items():
invested, proceeds, cash = kept[k]
gain = value[k] + proceeds + cash - invested
line["value"].append(round(float(value[k]), 2))
line["invested"].append(round(float(invested), 2))
line["gain"].append(round(float(gain), 2))
line["gain_pct"].append(round(float(gain / invested), 4) if invested else 0.0)
line["flow_in"].append(round(float(flow_in[k]), 2))
line["cash_div"].append(round(float(income[k]), 2))
return {"dates": [d.isoformat() for d in dates], "lines": lines, "excluded": excluded}
@given(history=histories())
def test_the_portfolio_series_agrees_with_the_working_out(session_factory, portfolio, history):
s = session_factory()
tenancy.bind(s, *portfolio)
try:
load(s, history)
got = queries.portfolio_series(s)
want = _worked(history)
assert got["dates"] == want["dates"]
assert got["excluded"] == want["excluded"]
line = want["lines"].get("all")
for key in ("value", "invested", "gain", "gain_pct", "flow_in", "cash_div"):
assert got[key] == (line[key] if line else []), key
finally:
s.rollback()
s.close()
@pytest.mark.parametrize("by", ["ticker", "currency"])
@given(history=histories())
def test_a_split_chart_follows_the_same_rules_and_adds_up(session_factory, portfolio, history,
by):
s = session_factory()
tenancy.bind(s, *portfolio)
try:
load(s, history)
got = queries.grouped_series(s, by)
want = _worked(history, group=lambda h: getattr(h, by))
assert got["dates"] == want["dates"]
assert got["excluded"] == want["excluded"]
live = {k: v for k, v in want["lines"].items()
if any(any(series) for series in v.values())}
assert got["groups"] == live
total = queries.portfolio_series(s)
for key in ("value", "invested", "gain"):
for i in range(len(got["dates"])):
split = sum(line[key][i] for line in got["groups"].values())
assert abs(split - total[key][i]) <= 0.005 * max(len(got["groups"]), 1) + 1e-9
finally:
s.rollback()
s.close()
def test_a_holding_not_yet_priced_is_left_out_of_its_line_too(session_factory, portfolio):
"""The case the total was fixed for, in a split: ZULU bought on the 2nd and
first priced on the 4th is not a loss of its whole cost on the 3rd."""
import factories as fac
s = session_factory()
tenancy.bind(s, *portfolio)
try:
alpha, zulu = fac.make_instrument(s, "ALPHA"), fac.make_instrument(s, "ZULU")
fac.add_trade(s, alpha, "2026-03-02", "buy", 10, "1.00")
fac.add_trade(s, zulu, "2026-03-02", "buy", 10, "5.00")
fac.add_prices(s, alpha, [(dt.date(2026, 3, d), "1.00") for d in (2, 3, 4)])
fac.add_prices(s, zulu, [(dt.date(2026, 3, 4), "5.50")])
lines = queries.grouped_series(s, "ticker")["groups"]
assert lines["ZULU"]["gain"] == [0.0, 0.0, 5.0]
assert lines["ZULU"]["invested"] == [0.0, 0.0, 50.0]
assert queries.portfolio_series(s)["gain"] == [0.0, 0.0, 5.0]
finally:
s.rollback()
s.close()