412 lines
21 KiB
Python
412 lines
21 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import date, datetime
|
|
|
|
from fastapi.responses import RedirectResponse
|
|
from nicegui import ui
|
|
|
|
from app.domain.backtesting_math import asset_quantity_from_workspace_config
|
|
from app.models.workspace import get_workspace_repository
|
|
from app.pages.common import dashboard_page, split_page_panes
|
|
from app.services.backtesting.ui_service import BacktestPageRunResult, BacktestPageService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _chart_options(result: BacktestPageRunResult) -> dict:
|
|
template_result = result.run_result.template_results[0]
|
|
return {
|
|
"tooltip": {"trigger": "axis"},
|
|
"legend": {"data": ["Spot", "LTV hedged", "LTV unhedged"]},
|
|
"xAxis": {"type": "category", "data": [point.date.isoformat() for point in template_result.daily_path]},
|
|
"yAxis": [
|
|
{"type": "value", "name": "Spot"},
|
|
{"type": "value", "name": "LTV", "min": 0, "max": 1, "axisLabel": {"formatter": "{value}"}},
|
|
],
|
|
"series": [
|
|
{
|
|
"name": "Spot",
|
|
"type": "line",
|
|
"smooth": True,
|
|
"data": [round(point.spot_close, 2) for point in template_result.daily_path],
|
|
"lineStyle": {"color": "#0ea5e9"},
|
|
},
|
|
{
|
|
"name": "LTV hedged",
|
|
"type": "line",
|
|
"yAxisIndex": 1,
|
|
"smooth": True,
|
|
"data": [round(point.ltv_hedged, 4) for point in template_result.daily_path],
|
|
"lineStyle": {"color": "#22c55e"},
|
|
},
|
|
{
|
|
"name": "LTV unhedged",
|
|
"type": "line",
|
|
"yAxisIndex": 1,
|
|
"smooth": True,
|
|
"data": [round(point.ltv_unhedged, 4) for point in template_result.daily_path],
|
|
"lineStyle": {"color": "#ef4444"},
|
|
},
|
|
],
|
|
}
|
|
|
|
|
|
@ui.page("/{workspace_id}/backtests")
|
|
def workspace_backtests_page(workspace_id: str) -> None:
|
|
repo = get_workspace_repository()
|
|
if not repo.workspace_exists(workspace_id):
|
|
return RedirectResponse(url="/", status_code=307)
|
|
_render_backtests_page(workspace_id=workspace_id)
|
|
|
|
|
|
def _render_backtests_page(workspace_id: str | None = None) -> None:
|
|
service = BacktestPageService()
|
|
template_options = service.template_options("GLD")
|
|
select_options = {str(option["slug"]): str(option["label"]) for option in template_options}
|
|
default_template_slug = str(template_options[0]["slug"]) if template_options else None
|
|
repo = get_workspace_repository()
|
|
config = repo.load_portfolio_config(workspace_id) if workspace_id else None
|
|
default_entry_spot = service.derive_entry_spot("GLD", date(2024, 1, 2), date(2024, 1, 8))
|
|
default_units = (
|
|
asset_quantity_from_workspace_config(config, entry_spot=default_entry_spot, symbol="GLD")
|
|
if config is not None and default_entry_spot > 0
|
|
else 1000.0
|
|
)
|
|
default_loan = float(config.loan_amount) if config else 68000.0
|
|
default_margin_call_ltv = float(config.margin_threshold) if config else 0.75
|
|
|
|
with dashboard_page(
|
|
"Backtests",
|
|
"Run a thin read-only historical scenario over the BT-001 synthetic engine and inspect the daily path.",
|
|
"backtests",
|
|
workspace_id=workspace_id,
|
|
):
|
|
left_pane, right_pane = split_page_panes(
|
|
left_testid="backtests-left-pane",
|
|
right_testid="backtests-right-pane",
|
|
)
|
|
|
|
with left_pane:
|
|
with ui.card().classes(
|
|
"w-full rounded-2xl border border-slate-200 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-900"
|
|
):
|
|
ui.label("Scenario Form").classes("text-lg font-semibold text-slate-900 dark:text-slate-100")
|
|
ui.label(
|
|
"Entry spot is auto-derived from the first historical close in the selected window so the scenario stays consistent with BT-001 entry timing."
|
|
).classes("text-sm text-slate-500 dark:text-slate-400")
|
|
ui.label("BT-001A currently supports GLD only for this thin read-only page.").classes(
|
|
"text-sm text-slate-500 dark:text-slate-400"
|
|
)
|
|
if workspace_id:
|
|
ui.label("Workspace defaults seed underlying units, loan amount, and margin threshold.").classes(
|
|
"text-sm text-slate-500 dark:text-slate-400"
|
|
)
|
|
symbol_input = ui.input("Symbol", value="GLD").props("readonly").classes("w-full")
|
|
start_input = ui.input("Start date", value="2024-01-02").classes("w-full")
|
|
end_input = ui.input("End date", value="2024-01-08").classes("w-full")
|
|
template_select = ui.select(select_options, value=default_template_slug, label="Template").classes(
|
|
"w-full"
|
|
)
|
|
units_input = ui.number("Underlying units", value=default_units, min=0.0001, step=1).classes("w-full")
|
|
loan_input = ui.number("Loan amount", value=default_loan, min=0, step=1000).classes("w-full")
|
|
ltv_input = ui.number(
|
|
"Margin call LTV",
|
|
value=default_margin_call_ltv,
|
|
min=0.01,
|
|
max=0.99,
|
|
step=0.01,
|
|
).classes("w-full")
|
|
entry_spot_hint = ui.label("Entry spot will be auto-derived on run.").classes(
|
|
"text-sm text-slate-500 dark:text-slate-400"
|
|
)
|
|
validation_label = ui.label("").classes("text-sm text-rose-600 dark:text-rose-300")
|
|
run_button = ui.button("Run backtest").props("color=primary")
|
|
|
|
seeded_summary = ui.card().classes(
|
|
"w-full rounded-2xl border border-slate-200 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-900"
|
|
)
|
|
|
|
with right_pane:
|
|
result_panel = ui.column().classes("w-full gap-6")
|
|
|
|
def parse_iso_date(raw: object, field_name: str) -> date:
|
|
try:
|
|
return datetime.strptime(str(raw), "%Y-%m-%d").date()
|
|
except ValueError as exc:
|
|
raise ValueError(f"{field_name} must be in YYYY-MM-DD format") from exc
|
|
|
|
def derive_entry_spot() -> tuple[float | None, str | None]:
|
|
try:
|
|
resolved_entry_spot = service.derive_entry_spot(
|
|
str(symbol_input.value or "GLD"),
|
|
parse_iso_date(start_input.value, "Start date"),
|
|
parse_iso_date(end_input.value, "End date"),
|
|
)
|
|
except (ValueError, KeyError) as exc:
|
|
return None, str(exc)
|
|
return resolved_entry_spot, None
|
|
|
|
def render_seeded_summary(*, entry_spot: float | None = None, entry_spot_error: str | None = None) -> None:
|
|
seeded_summary.clear()
|
|
resolved_entry_spot = entry_spot
|
|
resolved_error = entry_spot_error
|
|
if resolved_entry_spot is None and resolved_error is None:
|
|
resolved_entry_spot, resolved_error = derive_entry_spot()
|
|
with seeded_summary:
|
|
ui.label("Scenario Summary").classes("text-lg font-semibold text-slate-900 dark:text-slate-100")
|
|
with ui.grid(columns=1).classes("w-full gap-4 sm:grid-cols-2 lg:grid-cols-1 xl:grid-cols-2"):
|
|
cards = [
|
|
("Template", select_options.get(str(template_select.value), str(template_select.value or "—"))),
|
|
("Underlying units", f"{float(units_input.value or 0.0):,.0f}"),
|
|
("Loan amount", f"${float(loan_input.value or 0.0):,.0f}"),
|
|
("Margin call LTV", f"{float(ltv_input.value or 0.0):.1%}"),
|
|
(
|
|
"Date range",
|
|
f"{str(start_input.value or '—')} → {str(end_input.value or '—')}",
|
|
),
|
|
(
|
|
"Entry spot",
|
|
f"${resolved_entry_spot:,.2f}" if resolved_entry_spot is not None else "Unavailable",
|
|
),
|
|
]
|
|
for label, value in cards:
|
|
with ui.card().classes(
|
|
"rounded-xl border border-slate-200 bg-slate-50 p-4 shadow-none dark:border-slate-800 dark:bg-slate-950"
|
|
):
|
|
ui.label(label).classes("text-sm text-slate-500 dark:text-slate-400")
|
|
ui.label(value).classes("text-xl font-bold text-slate-900 dark:text-slate-100")
|
|
if resolved_error:
|
|
ui.label(resolved_error).classes("text-sm text-amber-700 dark:text-amber-300")
|
|
|
|
def render_result_state(title: str, message: str, *, tone: str = "info") -> None:
|
|
tone_classes = {
|
|
"info": "border-sky-200 bg-sky-50 dark:border-sky-900/60 dark:bg-sky-950/30",
|
|
"warning": "border-amber-200 bg-amber-50 dark:border-amber-900/60 dark:bg-amber-950/30",
|
|
"error": "border-rose-200 bg-rose-50 dark:border-rose-900/60 dark:bg-rose-950/30",
|
|
}
|
|
text_classes = {
|
|
"info": "text-sky-800 dark:text-sky-200",
|
|
"warning": "text-amber-800 dark:text-amber-200",
|
|
"error": "text-rose-800 dark:text-rose-200",
|
|
}
|
|
result_panel.clear()
|
|
with result_panel:
|
|
with ui.card().classes(
|
|
f"w-full rounded-2xl border shadow-sm {tone_classes.get(tone, tone_classes['info'])}"
|
|
):
|
|
ui.label(title).classes("text-lg font-semibold text-slate-900 dark:text-slate-100")
|
|
ui.label(message).classes(f"text-sm {text_classes.get(tone, text_classes['info'])}")
|
|
|
|
def mark_results_stale() -> None:
|
|
render_result_state(
|
|
"Results out of date",
|
|
"Inputs changed. Run backtest again to refresh charts and daily results for the current scenario.",
|
|
tone="info",
|
|
)
|
|
|
|
def render_result(result: BacktestPageRunResult) -> None:
|
|
result_panel.clear()
|
|
template_result = result.run_result.template_results[0]
|
|
summary = template_result.summary_metrics
|
|
entry_spot_hint.set_text(f"Auto-derived entry spot: ${result.entry_spot:,.2f}")
|
|
render_seeded_summary(entry_spot=result.entry_spot)
|
|
with result_panel:
|
|
with ui.card().classes(
|
|
"w-full rounded-2xl border border-slate-200 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-900"
|
|
):
|
|
ui.label("Scenario Results").classes("text-lg font-semibold text-slate-900 dark:text-slate-100")
|
|
ui.label(f"Template: {template_result.template_name}").classes(
|
|
"text-sm text-slate-500 dark:text-slate-400"
|
|
)
|
|
with ui.grid(columns=4).classes("w-full gap-4 max-lg:grid-cols-2 max-sm:grid-cols-1"):
|
|
cards = [
|
|
("Start value", f"${summary.start_value:,.0f}"),
|
|
("End value hedged", f"${summary.end_value_hedged_net:,.0f}"),
|
|
("Max LTV hedged", f"{summary.max_ltv_hedged:.1%}"),
|
|
("Hedge cost", f"${summary.total_hedge_cost:,.0f}"),
|
|
("Margin call days hedged", str(summary.margin_call_days_hedged)),
|
|
("Margin call days unhedged", str(summary.margin_call_days_unhedged)),
|
|
(
|
|
"Hedged survived",
|
|
"Yes" if not summary.margin_threshold_breached_hedged else "No",
|
|
),
|
|
(
|
|
"Unhedged breached",
|
|
"Yes" if summary.margin_threshold_breached_unhedged else "No",
|
|
),
|
|
]
|
|
for label, value in cards:
|
|
with ui.card().classes(
|
|
"rounded-xl border border-slate-200 bg-slate-50 p-4 shadow-none dark:border-slate-800 dark:bg-slate-950"
|
|
):
|
|
ui.label(label).classes("text-sm text-slate-500 dark:text-slate-400")
|
|
ui.label(value).classes("text-2xl font-bold text-slate-900 dark:text-slate-100")
|
|
|
|
ui.echart(_chart_options(result)).classes(
|
|
"h-96 w-full rounded-2xl border border-slate-200 bg-white p-4 shadow-sm dark:border-slate-800 dark:bg-slate-900"
|
|
)
|
|
|
|
with ui.card().classes(
|
|
"w-full rounded-2xl border border-slate-200 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-900"
|
|
):
|
|
ui.label("Daily Results").classes("text-lg font-semibold text-slate-900 dark:text-slate-100")
|
|
ui.table(
|
|
columns=[
|
|
{"name": "date", "label": "Date", "field": "date", "align": "left"},
|
|
{"name": "spot_close", "label": "Spot", "field": "spot_close", "align": "right"},
|
|
{
|
|
"name": "net_portfolio_value",
|
|
"label": "Net hedged",
|
|
"field": "net_portfolio_value",
|
|
"align": "right",
|
|
},
|
|
{
|
|
"name": "ltv_unhedged",
|
|
"label": "LTV unhedged",
|
|
"field": "ltv_unhedged",
|
|
"align": "right",
|
|
},
|
|
{"name": "ltv_hedged", "label": "LTV hedged", "field": "ltv_hedged", "align": "right"},
|
|
{
|
|
"name": "margin_call_hedged",
|
|
"label": "Hedged breach",
|
|
"field": "margin_call_hedged",
|
|
"align": "center",
|
|
},
|
|
],
|
|
rows=[
|
|
{
|
|
"date": point.date.isoformat(),
|
|
"spot_close": f"${point.spot_close:,.2f}",
|
|
"net_portfolio_value": f"${point.net_portfolio_value:,.0f}",
|
|
"ltv_unhedged": f"{point.ltv_unhedged:.1%}",
|
|
"ltv_hedged": f"{point.ltv_hedged:.1%}",
|
|
"margin_call_hedged": "Yes" if point.margin_call_hedged else "No",
|
|
}
|
|
for point in template_result.daily_path
|
|
],
|
|
row_key="date",
|
|
).classes("w-full")
|
|
|
|
def validate_current_scenario(*, entry_spot: float | None = None) -> str | None:
|
|
try:
|
|
service.validate_preview_inputs(
|
|
symbol=str(symbol_input.value or ""),
|
|
start_date=parse_iso_date(start_input.value, "Start date"),
|
|
end_date=parse_iso_date(end_input.value, "End date"),
|
|
template_slug=str(template_select.value or ""),
|
|
underlying_units=float(units_input.value or 0.0),
|
|
loan_amount=float(loan_input.value or 0.0),
|
|
margin_call_ltv=float(ltv_input.value or 0.0),
|
|
entry_spot=entry_spot,
|
|
)
|
|
except (ValueError, KeyError) as exc:
|
|
return str(exc)
|
|
except Exception:
|
|
logger.exception(
|
|
"Backtest preview failed for workspace=%s symbol=%s start=%s end=%s template=%s units=%s loan=%s margin_call_ltv=%s",
|
|
workspace_id,
|
|
symbol_input.value,
|
|
start_input.value,
|
|
end_input.value,
|
|
template_select.value,
|
|
units_input.value,
|
|
loan_input.value,
|
|
ltv_input.value,
|
|
)
|
|
return "Backtest preview failed. Please verify the scenario inputs and try again."
|
|
return None
|
|
|
|
def refresh_workspace_seeded_units() -> None:
|
|
validation_label.set_text("")
|
|
entry_spot, entry_error = derive_entry_spot()
|
|
if workspace_id and config is not None and config.gold_value is not None and entry_spot is not None:
|
|
units_input.value = asset_quantity_from_workspace_config(config, entry_spot=entry_spot, symbol="GLD")
|
|
entry_spot_hint.set_text(f"Auto-derived entry spot: ${entry_spot:,.2f}")
|
|
else:
|
|
entry_spot_hint.set_text("Entry spot unavailable until the scenario dates are valid.")
|
|
render_seeded_summary(entry_spot=entry_spot, entry_spot_error=entry_error)
|
|
if entry_error:
|
|
validation_label.set_text(entry_error)
|
|
render_result_state("Scenario validation failed", entry_error, tone="warning")
|
|
return
|
|
validation_error = validate_current_scenario(entry_spot=entry_spot)
|
|
if validation_error:
|
|
validation_label.set_text(validation_error)
|
|
render_result_state("Scenario validation failed", validation_error, tone="warning")
|
|
else:
|
|
mark_results_stale()
|
|
|
|
def on_form_change() -> None:
|
|
validation_label.set_text("")
|
|
entry_spot, entry_error = derive_entry_spot()
|
|
render_seeded_summary(entry_spot=entry_spot, entry_spot_error=entry_error)
|
|
if entry_error:
|
|
entry_spot_hint.set_text("Entry spot unavailable until the scenario dates are valid.")
|
|
validation_label.set_text(entry_error)
|
|
render_result_state("Scenario validation failed", entry_error, tone="warning")
|
|
return
|
|
validation_error = validate_current_scenario(entry_spot=entry_spot)
|
|
if validation_error:
|
|
validation_label.set_text(validation_error)
|
|
render_result_state("Scenario validation failed", validation_error, tone="warning")
|
|
else:
|
|
mark_results_stale()
|
|
|
|
def run_backtest() -> None:
|
|
validation_label.set_text("")
|
|
try:
|
|
result = service.run_read_only_scenario(
|
|
symbol=str(symbol_input.value or ""),
|
|
start_date=parse_iso_date(start_input.value, "Start date"),
|
|
end_date=parse_iso_date(end_input.value, "End date"),
|
|
template_slug=str(template_select.value or ""),
|
|
underlying_units=float(units_input.value or 0.0),
|
|
loan_amount=float(loan_input.value or 0.0),
|
|
margin_call_ltv=float(ltv_input.value or 0.0),
|
|
)
|
|
except (ValueError, KeyError) as exc:
|
|
entry_spot, entry_error = derive_entry_spot()
|
|
render_seeded_summary(entry_spot=entry_spot, entry_spot_error=entry_error)
|
|
if entry_spot is None:
|
|
entry_spot_hint.set_text("Entry spot unavailable until the scenario dates are valid.")
|
|
validation_label.set_text(str(exc))
|
|
render_result_state("Scenario validation failed", str(exc), tone="warning")
|
|
return
|
|
except Exception:
|
|
entry_spot, entry_error = derive_entry_spot()
|
|
render_seeded_summary(entry_spot=entry_spot, entry_spot_error=entry_error)
|
|
if entry_spot is None:
|
|
entry_spot_hint.set_text("Entry spot unavailable until the scenario dates are valid.")
|
|
message = "Backtest failed. Please verify the scenario inputs and try again."
|
|
logger.exception(
|
|
"Backtest page run failed for workspace=%s symbol=%s start=%s end=%s template=%s units=%s loan=%s margin_call_ltv=%s",
|
|
workspace_id,
|
|
symbol_input.value,
|
|
start_input.value,
|
|
end_input.value,
|
|
template_select.value,
|
|
units_input.value,
|
|
loan_input.value,
|
|
ltv_input.value,
|
|
)
|
|
validation_label.set_text(message)
|
|
render_result_state("Backtest failed", message, tone="error")
|
|
return
|
|
render_result(result)
|
|
|
|
if workspace_id:
|
|
start_input.on_value_change(lambda _event: refresh_workspace_seeded_units())
|
|
end_input.on_value_change(lambda _event: refresh_workspace_seeded_units())
|
|
else:
|
|
start_input.on_value_change(lambda _event: on_form_change())
|
|
end_input.on_value_change(lambda _event: on_form_change())
|
|
template_select.on_value_change(lambda _event: on_form_change())
|
|
units_input.on_value_change(lambda _event: on_form_change())
|
|
loan_input.on_value_change(lambda _event: on_form_change())
|
|
ltv_input.on_value_change(lambda _event: on_form_change())
|
|
run_button.on_click(lambda: run_backtest())
|
|
render_seeded_summary(entry_spot=default_entry_spot)
|
|
run_backtest()
|