r"""Equity curve over time from `account_snapshots` (requirements §12.3, issue #20).
Split deliberately into two halves:
- Data preparation (`equity_curve_points`, `summarize`) turns
`AccountSnapshot` rows into plain, sorted `EquityPoint`\s and a summary — pure
functions, no rendering, fully unit-testable without a terminal.
- Rendering (`render_sparkline`, `format_report`) turns those points into text.
This stayed an ASCII sparkline even after `trader web` (§12.3) added a
second, interactive renderer over the SAME two functions below — this one is
still useful with no browser (e.g. over SSH into the host running the
daemon), and needed no new dependency when it was built. It is not being
replaced; `trader.reporting.web.charts.equity_curve_figure` is the sibling
that calls `equity_curve_points`/`summarize` for a browser instead.
The ROI-vs-benchmark overlay from the issue's stretch goal is built in
`trader web`'s equity-curve page, not here — this module stays exactly the
narrow ASCII report it was.
"""
from collections.abc import Sequence
from dataclasses import dataclass
from datetime import datetime
from decimal import ROUND_HALF_UP, Decimal
from trader.persistence.models import AccountSnapshot
__all__ = [
"EquityCurveSummary",
"EquityPoint",
"equity_curve_points",
"format_report",
"render_sparkline",
"summarize",
]
#: 8 levels, lowest to highest — a standard block-element sparkline alphabet.
_GLYPHS = "▁▂▃▄▅▆▇█"
#: Glyph used for every point when the series is perfectly flat (`high ==
#: low`), so a flat curve renders as a flat middle line rather than pinning to
#: the top of the scale, which would misleadingly look like a peak.
_FLAT_GLYPH = _GLYPHS[len(_GLYPHS) // 2]
[docs]
@dataclass(frozen=True, slots=True)
class EquityPoint:
"""One plotted point: an account's equity at a point in time."""
captured_at: datetime
equity: Decimal
[docs]
@dataclass(frozen=True, slots=True)
class EquityCurveSummary:
r"""Headline numbers over a series of `EquityPoint`\s."""
start: EquityPoint
end: EquityPoint
high: EquityPoint
low: EquityPoint
change: Decimal
#: `None` when `start.equity` is zero — a percentage change from zero is
#: undefined, not zero, so this is `None` rather than a misleading `0%` or
#: a `ZeroDivisionError` reaching the CLI.
change_pct: Decimal | None
count: int
[docs]
def equity_curve_points(snapshots: Sequence[AccountSnapshot]) -> list[EquityPoint]:
r"""Convert snapshot rows into chronological `EquityPoint`\s.
Sorts defensively by `(captured_at, id)` rather than trusting the caller's
query order, so this function is correct standalone and testable with an
out-of-order fixture — the same tiebreak `SnapshotRepository` uses.
"""
ordered = sorted(snapshots, key=lambda row: (row.captured_at, row.id))
return [
EquityPoint(captured_at=row.captured_at, equity=row.equity) for row in ordered
]
[docs]
def summarize(points: Sequence[EquityPoint]) -> EquityCurveSummary | None:
"""Start/end/high/low/change over a chronological series, or `None` if empty.
`points` is assumed already chronological (as `equity_curve_points`
returns it); this does not re-sort.
"""
if not points:
return None
start = points[0]
end = points[-1]
# `max`/`min` return the FIRST maximal/minimal element for ties when
# iterating a list in order, so on a tie the earlier point wins — the
# point where the extreme was first reached, not the last.
high = max(points, key=lambda p: p.equity)
low = min(points, key=lambda p: p.equity)
change = end.equity - start.equity
change_pct = (change / start.equity * 100) if start.equity != 0 else None
return EquityCurveSummary(
start=start,
end=end,
high=high,
low=low,
change=change,
change_pct=change_pct,
count=len(points),
)
def _resample(points: Sequence[EquityPoint], width: int) -> list[Decimal]:
"""Average `points` down to exactly `width` buckets, in order.
Only called when there are more points than `width`; each bucket is the
Decimal mean of the equities that fall in it, split as evenly as integer
bucket boundaries allow.
"""
n = len(points)
buckets: list[Decimal] = []
for i in range(width):
start_idx = (i * n) // width
end_idx = max(((i + 1) * n) // width, start_idx + 1)
chunk = points[start_idx:end_idx]
total = sum((p.equity for p in chunk), Decimal(0))
buckets.append(total / Decimal(len(chunk)))
return buckets
def _level_index(value: Decimal, low: Decimal, span: Decimal) -> int:
"""Map `value` in `[low, low + span]` to a glyph index, Decimal all the way."""
ratio = (value - low) / span
scaled = (ratio * (len(_GLYPHS) - 1)).quantize(Decimal("1"), rounding=ROUND_HALF_UP)
return min(max(int(scaled), 0), len(_GLYPHS) - 1)
[docs]
def render_sparkline(points: Sequence[EquityPoint], width: int = 60) -> str:
"""Render `points` as a one-line block-character sparkline.
Downsamples via `_resample` when there are more points than `width`;
otherwise renders one glyph per point (a narrower series than `width`
prints shorter than `width`, rather than being stretched).
"""
if not points:
return ""
if width < 1:
raise ValueError(f"width must be >= 1, got {width}")
values = (
_resample(points, width) if len(points) > width else [p.equity for p in points]
)
low = min(values)
high = max(values)
span = high - low
if span == 0:
return _FLAT_GLYPH * len(values)
return "".join(_GLYPHS[_level_index(v, low, span)] for v in values)
def _signed(value: Decimal) -> str:
quantized = value.quantize(Decimal("0.01"))
sign = "+" if quantized >= 0 else ""
return f"{sign}{quantized}"
def _pct(value: Decimal | None) -> str:
if value is None:
return "n/a"
quantized = value.quantize(Decimal("0.01"))
sign = "+" if quantized >= 0 else ""
return f"{sign}{quantized}%"
def _fmt_date(value: datetime) -> str:
return value.strftime("%Y-%m-%d %H:%M")