Source code for trader.reporting.web.charts
"""Plotly figures for `trader web` — pure functions, no FastAPI coupling.
Mirrors `equity_curve.py`'s split between data and rendering: every function
here takes rows the caller already fetched and returns a
`plotly.graph_objects.Figure`, so each is testable without a running server.
`Decimal` is used throughout Core; every value crosses to `float` only inside
these functions, at the point it is handed to Plotly — the same
presentation-only cast `render_sparkline` already performs internally.
"""
from collections.abc import Callable, Mapping, Sequence
from datetime import datetime
from decimal import Decimal
import plotly.graph_objects as go
from trader.domain import Bar
from trader.persistence.models import RoundTrip, Trade
from trader.reporting.equity_curve import EquityPoint
from trader.reporting.returns import cost_basis, days_held, return_pct
__all__ = [
"candlestick_figure",
"cumulative_pl_by_strategy_figure",
"equity_curve_figure",
"pnl_by_strategy_figure",
"pnl_by_symbol_figure",
"pnl_heatmap_figure",
"trade_distribution_figure",
]
#: Two-point minimum for a benchmark line to mean anything — a lone bar has no
#: "since then" to compare against, same reasoning as `benchmark_return`.
_MIN_BARS_FOR_A_LINE = 2
_LOSS_COLOR = "#d62728"
_GAIN_COLOR = "#2ca02c"
def _value_at_or_before(points: Sequence[EquityPoint], at: datetime) -> Decimal:
"""The equity of the latest point at-or-before `at`.
Falls back to the first point's equity when `at` predates every point —
a trade can fill before the app ever saved its first snapshot (e.g. the
very first `trader account` call lands after the day's earlier fills) —
so a marker still lands somewhere on the line rather than being dropped.
"""
candidates = [p for p in points if p.captured_at <= at]
return candidates[-1].equity if candidates else points[0].equity
def _empty_annotation(fig: go.Figure, text: str) -> None:
"""Add a centered message to an otherwise-empty figure rather than
rendering a blank chart with no explanation — same reasoning
`format_report` uses for a snapshot-less window."""
fig.add_annotation(
text=text,
showarrow=False,
xref="paper",
yref="paper",
x=0.5,
y=0.5,
font={"size": 14},
)
def _add_trade_markers(
fig: go.Figure, points: Sequence[EquityPoint], trades: Sequence[Trade]
) -> None:
"""One marker per filled buy and per filled sell, hover text naming the
ticker, side, quantity and fill price — the concrete answer to "when did
a buy/sell happen and what ticker" that the equity line alone can't give."""
filled = [t for t in trades if t.filled_at is not None]
buys = [t for t in filled if t.side == "buy"]
sells = [t for t in filled if t.side == "sell"]
for side_trades, label, symbol, color in (
(buys, "Buy", "triangle-up", _GAIN_COLOR),
(sells, "Sell", "triangle-down", _LOSS_COLOR),
):
if not side_trades:
continue
fig.add_trace(
go.Scatter(
x=[t.filled_at for t in side_trades],
y=[float(_value_at_or_before(points, t.filled_at)) for t in side_trades], # type: ignore[arg-type]
name=label,
mode="markers",
marker={
"symbol": symbol,
"size": 11,
"color": color,
"line": {"width": 1, "color": "black"},
},
text=[
f"{t.ticker} {label.upper()} {t.filled_qty or t.quantity} @ "
f"{t.filled_avg_price or t.price}"
for t in side_trades
],
hovertemplate="%{text}<br>%{x}<extra></extra>",
)
)
[docs]
def equity_curve_figure(
points: Sequence[EquityPoint],
benchmarks: Mapping[str, Sequence[Bar]],
starting_value: Decimal,
trades: Sequence[Trade] = (),
) -> go.Figure:
"""Account equity over time, one line per benchmark ticker alongside it,
with a marker for every filled buy/sell.
Each benchmark is plotted as what `starting_value` would be worth having
bought and held that ticker instead, scaled from its own first close —
the same "lump sum into a benchmark" framing `trader performance` already
uses via `buy_and_hold`, just continuous instead of first/last only.
`trades` are placed on the equity line itself (via `_value_at_or_before`),
not on a separate axis — hovering answers "what happened here" the same
way the `../backtesting` finplot scripts overlay buy/sell markers on a
price line, just applied to account equity instead of one symbol's price,
since this chart has no single price series to mask against. Unfilled
orders (`filled_at is None`) are excluded — this answers "when did a buy
or sell actually happen", not "when was one submitted."
"""
fig = go.Figure()
if points:
fig.add_trace(
go.Scatter(
x=[p.captured_at for p in points],
y=[float(p.equity) for p in points],
name="Account equity",
mode="lines",
line={"width": 3},
)
)
for ticker, bars in benchmarks.items():
ordered = sorted(bars, key=lambda b: b.timestamp)
if len(ordered) < _MIN_BARS_FOR_A_LINE:
continue
first_close = ordered[0].close
fig.add_trace(
go.Scatter(
x=[b.timestamp for b in ordered],
y=[float(starting_value * (b.close / first_close)) for b in ordered],
name=f"{ticker} (buy & hold)",
mode="lines",
line={"dash": "dot"},
)
)
if points:
_add_trade_markers(fig, points, trades)
if not fig.data:
_empty_annotation(
fig, "No account snapshots yet — run 'trader account' at least twice."
)
fig.update_layout(
title="Account equity vs. benchmarks",
xaxis_title="Date",
yaxis_title="Value ($)",
xaxis={"rangeslider": {"visible": True}},
hovermode="x unified",
)
return fig
def _pnl_bar_figure(
trips: Sequence[RoundTrip],
key: Callable[[RoundTrip], str],
*,
title: str,
xaxis_title: str,
empty_message: str,
) -> go.Figure:
"""Shared bar-chart builder for the by-symbol and by-strategy views.
Sums `realized_pl` per group, then sorts worst-to-best left to right so
the chart reads as a ranking, not an alphabetical list — the ordering
itself is the point of the chart.
"""
totals: dict[str, Decimal] = {}
for trip in trips:
group = key(trip)
totals[group] = totals.get(group, Decimal(0)) + trip.realized_pl
fig = go.Figure()
fig.update_layout(
title=title, xaxis_title=xaxis_title, yaxis_title="Realized P/L ($)"
)
if not totals:
_empty_annotation(fig, empty_message)
return fig
ordered = sorted(totals.items(), key=lambda item: item[1])
labels = [group for group, _ in ordered]
values = [float(pl) for _, pl in ordered]
colors = [_LOSS_COLOR if pl < 0 else _GAIN_COLOR for pl in values]
fig.add_trace(go.Bar(x=labels, y=values, marker_color=colors))
return fig
[docs]
def pnl_by_symbol_figure(trips: Sequence[RoundTrip]) -> go.Figure:
"""Realized P/L summed per symbol, across every *closed* round trip."""
return _pnl_bar_figure(
trips,
key=lambda t: t.symbol,
title="Realized P/L by symbol",
xaxis_title="Symbol",
empty_message="No closed round trips yet.",
)
[docs]
def pnl_by_strategy_figure(trips: Sequence[RoundTrip]) -> go.Figure:
"""Realized P/L summed per strategy, across every *closed* round trip.
`strategy_id` is nullable on `RoundTrip` (rows predating the column, or a
non-strategy path like `acceptance_forced_buy`); those group under
`"(unknown)"` rather than being silently dropped from the total.
"""
return _pnl_bar_figure(
trips,
key=lambda t: t.strategy_id or "(unknown)",
title="Realized P/L by strategy",
xaxis_title="Strategy",
empty_message="No closed round trips yet.",
)
[docs]
def cumulative_pl_by_strategy_figure(trips: Sequence[RoundTrip]) -> go.Figure:
"""Running total of realized P/L over time, one line per strategy.
`pnl_by_strategy_figure` answers "which strategy is ahead right now";
this answers "is that lead a trend or one lucky trade" — the same
"learn from wins and losses as they happen, not just a lagging total"
framing issue #28 was raised for. Each strategy's own trips are ordered
by `closed_at` before the running sum, so two strategies' lines are each
internally chronological even though the account interleaves them.
"""
fig = go.Figure()
fig.update_layout(
title="Cumulative realized P/L by strategy",
xaxis_title="Closed at",
yaxis_title="Cumulative realized P/L ($)",
hovermode="x unified",
)
if not trips:
_empty_annotation(fig, "No closed round trips yet.")
return fig
by_strategy: dict[str, list[RoundTrip]] = {}
for trip in trips:
by_strategy.setdefault(trip.strategy_id or "(unknown)", []).append(trip)
for strategy_id, group in sorted(by_strategy.items()):
ordered = sorted(group, key=lambda t: t.closed_at)
running = Decimal(0)
xs, ys = [], []
for trip in ordered:
running += trip.realized_pl
xs.append(trip.closed_at)
ys.append(float(running))
fig.add_trace(go.Scatter(x=xs, y=ys, name=strategy_id, mode="lines+markers"))
return fig
[docs]
def trade_distribution_figure(trips: Sequence[RoundTrip]) -> go.Figure:
"""One point per closed round trip: return % vs. holding period.
Reveals a pattern a total-return number hides — e.g. "short holds lose,
longer holds win" (or the reverse) — by plotting every trip rather than
only its sum. `return_pct` returns `None` for a zero cost basis (a
corporate-action share grant); those trips are dropped from this chart
rather than plotted at a fabricated 0%, since they would otherwise land
on the axis and read as a real, unremarkable outcome.
"""
fig = go.Figure()
fig.update_layout(
title="Trade P/L distribution",
xaxis_title="Holding period (days)",
yaxis_title="Return (%)",
)
xs, ys, colors, text = [], [], [], []
for trip in trips:
pct = return_pct(trip.realized_pl, cost_basis(trip.entry_price, trip.quantity))
if pct is None:
continue
xs.append(days_held(trip.opened_at, trip.closed_at))
ys.append(float(pct))
colors.append(_LOSS_COLOR if trip.realized_pl < 0 else _GAIN_COLOR)
text.append(
f"{trip.symbol} ({trip.strategy_id or '(unknown)'}): "
f"{trip.realized_pl:+} ({pct:+}%)"
)
if not xs:
_empty_annotation(fig, "No closed round trips with a usable cost basis yet.")
return fig
fig.add_trace(
go.Scatter(
x=xs,
y=ys,
mode="markers",
marker={"color": colors, "size": 10},
text=text,
hovertemplate="%{text}<br>%{x} day(s) held<extra></extra>",
)
)
return fig
[docs]
def pnl_heatmap_figure(trips: Sequence[RoundTrip]) -> go.Figure:
"""Realized P/L summed per (symbol, strategy) cell.
Sized to the real data rather than a fixed grid — a heatmap over mostly
empty cells is worse than no heatmap, and this project's live sample is
currently ~10-20 total round trips (see issue #28's own caution). Rows
and columns are exactly the symbols/strategies that actually appear,
never a padded universe.
"""
fig = go.Figure()
fig.update_layout(title="Realized P/L by symbol × strategy")
if not trips:
_empty_annotation(fig, "No closed round trips yet.")
return fig
symbols = sorted({t.symbol for t in trips})
strategies = sorted({t.strategy_id or "(unknown)" for t in trips})
totals: dict[tuple[str, str], Decimal] = {}
for trip in trips:
key = (trip.symbol, trip.strategy_id or "(unknown)")
totals[key] = totals.get(key, Decimal(0)) + trip.realized_pl
z = [
[float(totals.get((symbol, strategy), Decimal(0))) for strategy in strategies]
for symbol in symbols
]
# `None` (not 0.0) for a combination with no round trips at all, so the
# heatmap's own color scale does not read "never traded together" as
# "traded and broke exactly even" — a real, different claim.
for row_index, symbol in enumerate(symbols):
for col_index, strategy in enumerate(strategies):
if (symbol, strategy) not in totals:
z[row_index][col_index] = None # type: ignore[call-overload]
fig.add_trace(
go.Heatmap(
z=z,
x=strategies,
y=symbols,
colorscale=[[0, _LOSS_COLOR], [0.5, "#ffffff"], [1, _GAIN_COLOR]],
zmid=0,
hovertemplate="%{y} / %{x}: %{z:$.2f}<extra></extra>",
)
)
return fig
[docs]
def candlestick_figure(
symbol: str, bars: Sequence[Bar], trades: Sequence[Trade] = ()
) -> go.Figure:
"""OHLC candlesticks for one symbol, with a marker on every filled
buy/sell against that symbol's own price line.
The previously-deferred "per-symbol candlestick + entry/exit markers"
item (`docs/next-session.md`, 2026-08-18 session) — unblocked by simply
reading live bars through the same `MarketDataProvider.get_history` the
benchmark lines on `/` already call, rather than the backtest-only local
`bars` cache the earlier deferral was about.
"""
fig = go.Figure()
fig.update_layout(
title=f"{symbol} — price history",
xaxis_title="Date",
yaxis_title="Price ($)",
xaxis={"rangeslider": {"visible": True}},
)
if not bars:
_empty_annotation(fig, f"No price history available for {symbol}.")
return fig
ordered = sorted(bars, key=lambda b: b.timestamp)
fig.add_trace(
go.Candlestick(
x=[b.timestamp for b in ordered],
open=[float(b.open) for b in ordered],
high=[float(b.high) for b in ordered],
low=[float(b.low) for b in ordered],
close=[float(b.close) for b in ordered],
name=symbol,
)
)
filled = [t for t in trades if t.filled_at is not None]
for side_trades, label, marker_symbol, color in (
([t for t in filled if t.side == "buy"], "Buy", "triangle-up", _GAIN_COLOR),
([t for t in filled if t.side == "sell"], "Sell", "triangle-down", _LOSS_COLOR),
):
if not side_trades:
continue
fig.add_trace(
go.Scatter(
x=[t.filled_at for t in side_trades],
y=[float(t.filled_avg_price or t.price or 0) for t in side_trades],
name=label,
mode="markers",
marker={
"symbol": marker_symbol,
"size": 12,
"color": color,
"line": {"width": 1, "color": "black"},
},
text=[
f"{label.upper()} {t.filled_qty or t.quantity} @ "
f"{t.filled_avg_price or t.price}"
for t in side_trades
],
hovertemplate="%{text}<br>%{x}<extra></extra>",
)
)
return fig