"""FastAPI app serving interactive equity-curve and P&L reports (issue #20).
Read-only: no route here writes to the database or submits an order.
`create_app` takes already-built collaborators — the same shape as every
`build_*` construction seam in `trader.cli.main` — so tests substitute fakes
and nothing about the trading daemon's own construction changes because this
module exists.
"""
import html
import json
from collections.abc import Mapping, Sequence
from datetime import UTC, date, datetime, time, timedelta
from decimal import Decimal
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from trader.config.schema import ReportingConfig
from trader.domain import Bar
from trader.errors import TraderError
from trader.marketdata.base import MarketDataProvider
from trader.persistence.decisions import DecisionRepository
from trader.persistence.models import Decision, Trade
from trader.persistence.outcomes import OutcomeRepository
from trader.persistence.snapshots import SnapshotRepository
from trader.persistence.trades import TradeRepository
from trader.reporting.equity_curve import equity_curve_points
from trader.reporting.returns import PERCENT_QUANTUM, return_pct
from trader.reporting.web.charts import (
candlestick_figure,
cumulative_pl_by_strategy_figure,
equity_curve_figure,
pnl_by_strategy_figure,
pnl_by_symbol_figure,
pnl_heatmap_figure,
trade_distribution_figure,
)
from trader.reporting.window import resolve_report_start
__all__ = ["create_app"]
_PAGE_SHELL = """<!doctype html>
<html>
<head>
<meta charset="utf-8">
<title>{title} — trader web</title>
<style>
body {{ font-family: -apple-system, BlinkMacSystemFont, sans-serif; margin: 2rem; }}
nav a {{ margin-right: 1.5rem; font-weight: 600; }}
h1 {{ font-size: 1.25rem; color: #444; }}
h2 {{ font-size: 1rem; color: #444; margin-top: 2rem; }}
table {{ border-collapse: collapse; }}
th, td {{ text-align: left; padding: 0.3rem 1rem 0.3rem 0; border-bottom: 1px solid #ddd; vertical-align: top; }}
td.why {{ max-width: 48ch; white-space: normal; color: #333; }}
</style>
</head>
<body>
<nav><a href="/">Equity curve</a><a href="/pnl">Realized P&L</a><a href="/candles">Price history</a></nav>
<h1>{title}</h1>
{body}
</body>
</html>"""
def _page(title: str, body: str) -> HTMLResponse:
return HTMLResponse(_PAGE_SHELL.format(title=title, body=body))
def _parse_date(value: str | None) -> date | None:
"""`None` or an empty string both mean "not given"; anything else must be
a real `YYYY-MM-DD` or the mistake should be visible, not silently
swallowed into "no filter"."""
if not value:
return None
return date.fromisoformat(value)
def _filters_form(
start: date | None, end: date | None, symbol: str | None, symbols: Sequence[str]
) -> str:
"""A plain GET form for `?start=&end=&symbol=` — native date pickers and
a dropdown, no JS.
Exact windowing/symbol only (§ the operator's own choice on start/end:
manual, per view, no auto-derived default — extended the same way to
symbol) — this just makes typing the URL by hand unnecessary, it does not
add a smart default.
"""
start_value = f' value="{start.isoformat()}"' if start else ""
end_value = f' value="{end.isoformat()}"' if end else ""
options = ['<option value="">All symbols</option>']
for candidate in symbols:
selected = " selected" if candidate == symbol else ""
options.append(f'<option value="{candidate}"{selected}>{candidate}</option>')
clear_link = '<a href="/">clear</a>' if start or end or symbol else ""
return (
'<form method="get" style="margin-bottom:1rem;">'
f'<label>Start <input type="date" name="start"{start_value}></label> '
f'<label>End <input type="date" name="end"{end_value}></label> '
f'<label>Symbol <select name="symbol">{"".join(options)}</select></label> '
'<button type="submit">Apply</button> '
f"{clear_link}"
"</form>"
)
def _fetch_benchmarks(
provider: MarketDataProvider,
tickers: list[str],
window_start: datetime,
window_end: datetime,
) -> dict[str, list[Bar]]:
"""One ticker's outage must not blank the whole page — same isolation
discipline `trader performance` already applies per benchmark ticker."""
benchmarks: dict[str, list[Bar]] = {}
for ticker in tickers:
try:
benchmarks[ticker] = provider.get_history(
ticker, window_start.date(), window_end.date()
)
except TraderError:
continue
return benchmarks
def _trades_in_window(
trade_repo: TradeRepository,
window_start: datetime | None,
window_end: datetime | None,
symbol: str | None = None,
) -> list[Trade]:
"""Filled trades within `[window_start, window_end]`, chronological,
optionally narrowed to one `symbol`.
`None` for either bound means "no floor"/"no ceiling" — same convention
`SnapshotRepository.snapshots_in_range` uses, and the reason this takes
the query's own effective bounds rather than `points[0]`/`points[-1]`'s
captured_at: snapshots are taken periodically, not continuously, so a
fill between two snapshots — or after the last one, same day — would
otherwise fall outside a window derived from snapshot instants and
silently vanish from the table. `symbol` is matched case-insensitively
against `Trade.ticker`, which is always stored uppercase, so a
lower-cased URL param still matches.
Filters `TradeRepository.all()` in this layer rather than adding a
date-ranged repository method: this is a display concern (which trades
fall inside whatever window the page is currently showing), the same
reasoning `pnl_by_symbol_figure`'s grouping lives in `charts.py` rather
than on `OutcomeRepository`. A personal account's trade volume (dozens to
low hundreds) makes reading the whole table cheap.
"""
return sorted(
(
t
for t in trade_repo.all()
if t.filled_at is not None
and (window_start is None or t.filled_at >= window_start)
and (window_end is None or t.filled_at <= window_end)
and (symbol is None or t.ticker == symbol.upper())
),
key=lambda t: t.filled_at, # type: ignore[arg-type,return-value]
)
def _traded_symbols(trade_repo: TradeRepository) -> list[str]:
"""Every distinct ticker with at least one filled trade, sorted.
Independent of any window/symbol filter already applied — deliberately:
the dropdown's own option list must stay stable while narrowing the date
range, or a symbol traded outside the current window would silently
disappear from the choices instead of just filtering to nothing.
"""
return sorted({t.ticker for t in trade_repo.all() if t.filled_at is not None})
_NO_DECISION_WHY = (
"No recorded decision (likely a protective stop fill, or a manual/forced trade)"
)
def _decision_why(decision: Decision | None) -> str:
"""The best human-readable "why" for a decision.
Prefers the LLM's own stated reason at evaluation time
(`inputs_json.raw_response.reason`) over the `reasoning` COLUMN — for a
buy/sell, `reasoning` is deliberately overwritten at order time with an
execution note ("Bought 26 at limit 374.47. evaluator skipped: ...", see
`run_once.py`'s `record("buy", f"Bought {qty} at limit {price}. ...")`),
which answers "what happened", not "why". The ORIGINAL LLM rationale
survives only inside `inputs_json`, captured before the order was ever
placed. A rule-based strategy (turtle/RSI/Bollinger) has no
`raw_response` at all — its `reasoning` already IS the real answer
("RSI 61.8 within thresholds"), so that's the fallback.
`decision is None` means no `Decision` row references this trade at
all — most often a protective stop firing, which this app records as a
fill, never as a fresh decision.
"""
if decision is None:
return _NO_DECISION_WHY
if decision.inputs_json:
try:
inputs = json.loads(decision.inputs_json)
except ValueError:
inputs = None
if isinstance(inputs, dict):
raw_response = inputs.get("raw_response")
if isinstance(raw_response, dict):
reason = raw_response.get("reason")
if isinstance(reason, str) and reason.strip():
return reason
return decision.reasoning or "(no reasoning recorded)"
def _trades_table_html(
trades: Sequence[Trade], decisions_by_trade: Mapping[int, Decision]
) -> str:
"""A plain chronological table — SYMBOL, SIDE, QTY, PRICE, FILLED AT, WHY
— the unambiguous answer to "what tickers were bought/sold, when, and
why" that hovering over a chart marker can't give at a glance.
Every cell is `html.escape`d: SIDE/QTY/PRICE/timestamps are internal and
safe either way, but WHY is free text an LLM generated from real news
headlines — untrusted content by the time it reaches this page."""
if not trades:
return "<p>No filled trades in this window.</p>"
rows = "".join(
"<tr>"
f"<td>{html.escape(t.ticker)}</td><td>{html.escape(t.side.upper())}</td>"
f"<td>{html.escape(str(t.filled_qty or t.quantity))}</td>"
f"<td>{html.escape(str(t.filled_avg_price or t.price))}</td>"
f"<td>{html.escape(str(t.filled_at))}</td>"
f'<td class="why">{html.escape(_decision_why(decisions_by_trade.get(t.id)))}</td>'
"</tr>"
for t in trades
)
return (
"<table><thead><tr><th>Symbol</th><th>Side</th><th>Qty</th>"
"<th>Fill price</th><th>Filled at (UTC)</th><th>Why</th></tr></thead>"
f"<tbody>{rows}</tbody></table>"
)
[docs]
def create_app(
snapshot_repo: SnapshotRepository,
outcome_repo: OutcomeRepository,
reporting_config: ReportingConfig,
provider: MarketDataProvider,
trade_repo: TradeRepository,
decision_repo: DecisionRepository,
) -> FastAPI:
app = FastAPI(title="trader web")
@app.get("/", response_class=HTMLResponse)
def equity_curve(
days: int = 0,
start: str | None = None,
end: str | None = None,
symbol: str | None = None,
) -> HTMLResponse:
"""Account equity vs. benchmarks.
`?start=YYYY-MM-DD` and/or `?end=YYYY-MM-DD` narrow to an exact
window, both bounds inclusive — same semantics as `trader performance
--start/--end`, so a dead window before real trading began (e.g. a
one-off `acceptance_forced_buy` test predating the daemon's first
real cycle) can be excluded from the comparison by hand. Picking only
one bound leaves the other open — an unset start means "from the
earliest snapshot", an unset end means "through the latest". Takes
priority over `?days=N`, which narrows by "last N days from now"
instead — same convention as `trader chart` — when either is given.
Omitting all three means all time.
`?symbol=TICKER` narrows the buy/sell markers and the trades table to
one ticker — e.g. "when was GOOG traded" — WITHOUT changing the
account equity or benchmark lines, which stay whole-account; a symbol
filter on a single-symbol curve wouldn't mean anything.
`start`/`end`/`symbol` are plain strings, not FastAPI's own typed
query params: the filters `<form>` always sends every field (a plain
GET form has no concept of "this one wasn't touched"), so a blank
date or the "All symbols" option arrives as `""`, not an absent
param — a typed `date | None`/enum would reject that as invalid
rather than reading it as "not given."
"""
start_date = _parse_date(start)
end_date = _parse_date(end)
# Validated against real traded tickers, not just normalized: an
# unrecognised value falls back to "no filter" rather than being
# trusted as-is — this is also what keeps `symbol_filter` safe to
# interpolate into the page title below (`table_title`), since it can
# only ever be `None` or a ticker this app actually saw in `trades`.
known_symbols = _traded_symbols(trade_repo)
symbol_filter = (
symbol.upper() if symbol and symbol.upper() in known_symbols else None
)
if start_date is not None or end_date is not None:
window_start = (
datetime.combine(start_date, time.min, tzinfo=UTC) if start_date else None
)
window_end = (
datetime.combine(end_date, time.min, tzinfo=UTC) if end_date else None
)
snapshots = snapshot_repo.snapshots_in_range(window_start, window_end)
else:
window_start = datetime.now(UTC) - timedelta(days=days) if days > 0 else None
window_end = None
snapshots = snapshot_repo.snapshots_since(window_start)
points = equity_curve_points(snapshots)
starting_value = points[0].equity if points else Decimal(0)
benchmarks = (
_fetch_benchmarks(
provider,
reporting_config.benchmark_tickers,
points[0].captured_at,
points[-1].captured_at,
)
if points
else {}
)
# The trade window is the REQUEST's own bounds, not points[0]/[-1]'s
# captured_at — see `_trades_in_window`'s docstring for why that would
# silently drop same-day fills that landed after the last snapshot.
window_trades = _trades_in_window(
trade_repo, window_start, window_end, symbol=symbol_filter
)
decisions_by_trade = decision_repo.by_trade_ids([t.id for t in window_trades])
fig = equity_curve_figure(
points, benchmarks, starting_value, trades=window_trades
)
table_title = (
f"Filled trades in this window ({symbol_filter})"
if symbol_filter
else "Filled trades in this window"
)
body = (
_filters_form(start_date, end_date, symbol_filter, known_symbols)
+ fig.to_html(include_plotlyjs="cdn", full_html=False)
+ f"<h2>{table_title}</h2>"
+ _trades_table_html(window_trades, decisions_by_trade)
)
return _page("Equity curve", body)
@app.get("/pnl", response_class=HTMLResponse)
def pnl(since_creation: bool = False) -> HTMLResponse:
"""Realized P/L broken down by symbol, by strategy, over time, and
by trade — "how are we doing" plus "where is it coming from"
(issue #28).
**Default start date (issue #57).** Without `?since_creation=1`, this
page windows to `config/reporting.yaml`'s `default_report_start_date`
onward, same as `trader outcomes` — a deployment-specific, dated
finding (`docs/operational-timeline.md`), not a code constant. UX
decision (this issue's own judgment call, documented here since this
route has no CLI flag to carry the explanation): a query param
(`?since_creation=1`) rather than a visible toggle control, because a
plain link ("show full history" / "apply default window") is exactly
as discoverable with a fraction of the code, and this page has no
other filter UI to match a toggle's visual weight against. If
`default_report_start_date` is unset in config, this page's
behaviour is unchanged from before issue #57: full history, and the
query param has no effect since there is no default to opt out of.
"""
window_start = resolve_report_start(
explicit_start=None,
since_creation=since_creation,
default_start_date=reporting_config.default_report_start_date,
)
trips = outcome_repo.recent(limit=10_000, start=window_start)
summary = outcome_repo.outcome_summary(start=window_start)
if reporting_config.default_report_start_date is not None:
window_note = (
f'<p>Full history (<a href="/pnl">apply default window, '
f"{reporting_config.default_report_start_date} onward</a>).</p>"
if window_start is None
else (
f"<p>Since {window_start.date()} "
'(<a href="/pnl?since_creation=1">show full history</a>).</p>'
)
)
else:
window_note = ""
total_pct = return_pct(summary.total_realized_pl, summary.total_cost_basis)
win_rate = (
(Decimal(summary.win_count) / summary.trip_count * 100).quantize(
PERCENT_QUANTUM
)
if summary.trip_count
else None
)
summary_html = (
f"<p>{summary.trip_count} closed round trips — "
f"total realized P/L <strong>{summary.total_realized_pl:+}</strong> "
f"({'n/a' if total_pct is None else f'{total_pct:+}%'}), "
f"win rate {'n/a' if win_rate is None else f'{win_rate}%'}.</p>"
)
by_symbol = pnl_by_symbol_figure(trips).to_html(
include_plotlyjs="cdn", full_html=False
)
by_strategy = pnl_by_strategy_figure(trips).to_html(
include_plotlyjs=False, full_html=False
)
cumulative = cumulative_pl_by_strategy_figure(trips).to_html(
include_plotlyjs=False, full_html=False
)
distribution = trade_distribution_figure(trips).to_html(
include_plotlyjs=False, full_html=False
)
heatmap = pnl_heatmap_figure(trips).to_html(
include_plotlyjs=False, full_html=False
)
return _page(
"Realized P&L",
window_note
+ summary_html
+ by_symbol
+ by_strategy
+ "<h2>Cumulative P/L over time, by strategy</h2>"
+ cumulative
+ "<h2>Trade P/L distribution</h2>"
+ distribution
+ "<h2>P/L by symbol × strategy</h2>"
+ heatmap,
)
@app.get("/candles", response_class=HTMLResponse)
def candles(
symbol: str | None = None,
start: str | None = None,
end: str | None = None,
) -> HTMLResponse:
"""Per-symbol candlestick history with entry/exit markers.
Separate page from `/`: the equity curve is whole-account and a
symbol filter there only narrows the trades table and markers, never
the line itself — a price chart is inherently per-symbol, so it
earns its own page rather than overloading `/`'s meaning further.
Live bars via `provider.get_history`, the same seam `/`'s benchmark
lines already use — never the backtest-only local `bars` cache.
"""
known_symbols = _traded_symbols(trade_repo)
symbol_filter = (
symbol.upper() if symbol and symbol.upper() in known_symbols else None
)
start_date = _parse_date(start)
end_date = _parse_date(end)
window_start = (
datetime.combine(start_date, time.min, tzinfo=UTC) if start_date else None
)
window_end = (
datetime.combine(end_date, time.min, tzinfo=UTC) if end_date else None
)
if symbol_filter is None:
body = _filters_form(start_date, end_date, None, known_symbols)
note = (
"<p>Pick a symbol above to see its price history.</p>"
if known_symbols
else "<p>No filled trades yet — nothing to chart.</p>"
)
return _page("Price history", body + note)
history_start = (window_start or datetime.now(UTC) - timedelta(days=180)).date()
history_end = (window_end or datetime.now(UTC)).date()
try:
bars = provider.get_history(symbol_filter, history_start, history_end)
except TraderError:
bars = []
window_trades = _trades_in_window(
trade_repo, window_start, window_end, symbol=symbol_filter
)
fig = candlestick_figure(symbol_filter, bars, trades=window_trades)
body = _filters_form(
start_date, end_date, symbol_filter, known_symbols
) + fig.to_html(include_plotlyjs="cdn", full_html=False)
return _page(f"Price history ({symbol_filter})", body)
return app