from __future__ import annotations import logging 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.event_comparison_ui import EventComparisonPageService logger = logging.getLogger(__name__) def _chart_options(dates: tuple[str, ...], series: tuple[dict[str, object], ...]) -> dict: return { "tooltip": {"trigger": "axis"}, "legend": {"type": "scroll"}, "xAxis": {"type": "category", "data": list(dates)}, "yAxis": {"type": "value", "name": "Net value"}, "series": [ { "name": item["name"], "type": "line", "smooth": True, "data": item["values"], } for item in series ], } @ui.page("/{workspace_id}/event-comparison") def workspace_event_comparison_page(workspace_id: str) -> None: repo = get_workspace_repository() if not repo.workspace_exists(workspace_id): return RedirectResponse(url="/", status_code=307) _render_event_comparison_page(workspace_id=workspace_id) def _render_event_comparison_page(workspace_id: str | None = None) -> None: service = EventComparisonPageService() preset_options = service.preset_options("GLD") template_options = service.template_options("GLD") repo = get_workspace_repository() config = repo.load_portfolio_config(workspace_id) if workspace_id else None default_preset_slug = str(preset_options[0]["slug"]) if preset_options else None default_template_slugs = list(preset_options[0]["default_template_slugs"]) if preset_options else [] default_entry_spot = 100.0 if default_preset_slug is not None: default_preview = service.preview_scenario( preset_slug=default_preset_slug, template_slugs=tuple(default_template_slugs), underlying_units=1.0, loan_amount=0.0, margin_call_ltv=0.75, ) default_entry_spot = default_preview.initial_portfolio.entry_spot 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 preset_select_options = {str(option["slug"]): str(option["label"]) for option in preset_options} template_select_options = {str(option["slug"]): str(option["label"]) for option in template_options} preset_lookup = {str(option["slug"]): option for option in preset_options} with dashboard_page( "Event Comparison", "Thin BT-003A read-only UI over EventComparisonService with deterministic seeded GLD presets.", "event-comparison", workspace_id=workspace_id, ): left_pane, right_pane = split_page_panes( left_testid="event-comparison-left-pane", right_testid="event-comparison-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("Comparison Form").classes("text-lg font-semibold text-slate-900 dark:text-slate-100") ui.label( "Preset selection is deterministic and read-only in the sense that runs reuse seeded event windows and existing BT-003 ranking logic." ).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" ) preset_select = ui.select( preset_select_options, value=default_preset_slug, label="Event preset", ).classes("w-full") template_select = ui.select( template_select_options, value=default_template_slugs, label="Strategy templates", multiple=True, ).classes("w-full") ui.label( "Changing the preset resets strategy templates to that preset's default comparison set." ).classes("text-xs text-slate-500 dark:text-slate-400") 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") metadata_label = ui.label("").classes("text-sm text-slate-500 dark:text-slate-400") scenario_label = ui.label("").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 comparison").props("color=primary") selected_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") syncing_controls = {"value": False} def selected_template_slugs() -> tuple[str, ...]: raw_value = template_select.value or [] if isinstance(raw_value, str): return (raw_value,) if raw_value else () return tuple(str(item) for item in raw_value if item) def render_selected_summary(entry_spot: float | None = None, entry_spot_error: str | None = None) -> None: selected_summary.clear() with selected_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 = [ ( "Preset", preset_select_options.get(str(preset_select.value), str(preset_select.value or "—")), ), ("Templates", str(len(selected_template_slugs()))), ("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%}"), ("Entry spot", f"${entry_spot:,.2f}" if 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 entry_spot_error: ui.label(entry_spot_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 comparison again to refresh rankings and portfolio paths for the current scenario.", tone="info", ) def refresh_preview(*, reset_templates: bool = False, reseed_units: bool = False) -> str | None: option = preset_lookup.get(str(preset_select.value or "")) if option is None: metadata_label.set_text("") scenario_label.set_text("") render_selected_summary(entry_spot=None) return None if reset_templates: syncing_controls["value"] = True try: template_select.value = list(service.default_template_selection(str(option["slug"]))) finally: syncing_controls["value"] = False template_slugs = selected_template_slugs() try: preview_units = float(units_input.value or 0.0) if workspace_id and config is not None and reseed_units: preview_scenario = service.preview_scenario( preset_slug=str(option["slug"]), template_slugs=template_slugs, underlying_units=1.0, loan_amount=float(loan_input.value or 0.0), margin_call_ltv=float(ltv_input.value or 0.0), ) preview_units = asset_quantity_from_workspace_config( config, entry_spot=float(preview_scenario.initial_portfolio.entry_spot), symbol="GLD", ) syncing_controls["value"] = True try: units_input.value = preview_units finally: syncing_controls["value"] = False scenario = service.preview_scenario( preset_slug=str(option["slug"]), template_slugs=template_slugs, underlying_units=preview_units, loan_amount=float(loan_input.value or 0.0), margin_call_ltv=float(ltv_input.value or 0.0), ) except (ValueError, KeyError) as exc: metadata_label.set_text(f"Preset: {option['label']} — {option['description']}") scenario_label.set_text(str(exc)) render_selected_summary(entry_spot=None, entry_spot_error=str(exc)) return str(exc) except Exception: logger.exception( "Event comparison preview failed for workspace=%s preset=%s templates=%s units=%s loan=%s margin_call_ltv=%s", workspace_id, preset_select.value, selected_template_slugs(), units_input.value, loan_input.value, ltv_input.value, ) message = "Event comparison preview failed. Please verify the seeded inputs and try again." metadata_label.set_text(f"Preset: {option['label']} — {option['description']}") scenario_label.set_text(message) render_selected_summary(entry_spot=None, entry_spot_error=message) return message preset = service.event_preset_service.get_preset(str(option["slug"])) metadata_label.set_text(f"Preset: {option['label']} — {option['description']}") scenario_label.set_text( "Scenario preview: " f"{scenario.start_date.isoformat()} → {scenario.end_date.isoformat()}" + ( f" · Anchor date: {preset.anchor_date.isoformat()}" if preset.anchor_date is not None else " · Anchor date: none" ) + f" · Entry spot: ${scenario.initial_portfolio.entry_spot:,.2f}" ) render_selected_summary(entry_spot=float(scenario.initial_portfolio.entry_spot)) return None def render_report() -> None: validation_label.set_text("") result_panel.clear() template_slugs = selected_template_slugs() try: report = service.run_read_only_comparison( preset_slug=str(preset_select.value or ""), template_slugs=template_slugs, 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: validation_label.set_text(str(exc)) render_result_state("Scenario validation failed", str(exc), tone="warning") return except Exception: message = "Event comparison failed. Please verify the seeded inputs and try again." logger.exception( "Event comparison page run failed for workspace=%s preset=%s templates=%s units=%s loan=%s margin_call_ltv=%s", workspace_id, preset_select.value, selected_template_slugs(), units_input.value, loan_input.value, ltv_input.value, ) validation_label.set_text(message) render_result_state("Event comparison failed", message, tone="error") return preset = report.event_preset scenario = report.scenario metadata_label.set_text( f"Preset: {preset.display_name} ({preset.event_type}) · Tags: {', '.join(preset.tags) or 'none'}" ) scenario_label.set_text( "Scenario dates used: " f"{scenario.start_date.isoformat()} → {scenario.end_date.isoformat()} · " f"Entry spot: ${scenario.initial_portfolio.entry_spot:,.2f}" ) render_selected_summary(entry_spot=float(scenario.initial_portfolio.entry_spot)) chart_model = service.chart_model(report) 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") with ui.grid(columns=4).classes("w-full gap-4 max-lg:grid-cols-2 max-sm:grid-cols-1"): cards = [ ("Symbol", scenario.symbol), ("Event window", f"{preset.window_start.isoformat()} → {preset.window_end.isoformat()}"), ( "Anchor date", preset.anchor_date.isoformat() if preset.anchor_date is not None else "None", ), ( "Scenario dates used", f"{scenario.start_date.isoformat()} → {scenario.end_date.isoformat()}", ), ("Underlying units", f"{scenario.initial_portfolio.underlying_units:,.0f}"), ("Loan amount", f"${scenario.initial_portfolio.loan_amount:,.0f}"), ("Margin call LTV", f"{scenario.initial_portfolio.margin_call_ltv:.1%}"), ("Templates compared", str(len(report.rankings))), ] 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") 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("Ranked Results").classes("text-lg font-semibold text-slate-900 dark:text-slate-100") ui.table( columns=[ {"name": "rank", "label": "Rank", "field": "rank", "align": "right"}, {"name": "template_name", "label": "Template", "field": "template_name", "align": "left"}, { "name": "survived_margin_call", "label": "Survived margin call", "field": "survived_margin_call", "align": "center", }, { "name": "margin_call_days_hedged", "label": "Hedged margin call days", "field": "margin_call_days_hedged", "align": "right", }, { "name": "max_ltv_hedged", "label": "Max hedged LTV", "field": "max_ltv_hedged", "align": "right", }, {"name": "hedge_cost", "label": "Hedge cost", "field": "hedge_cost", "align": "right"}, { "name": "final_equity", "label": "Final equity", "field": "final_equity", "align": "right", }, ], rows=[ { "rank": item.rank, "template_name": item.template_name, "survived_margin_call": "Yes" if item.survived_margin_call else "No", "margin_call_days_hedged": item.margin_call_days_hedged, "max_ltv_hedged": f"{item.max_ltv_hedged:.1%}", "hedge_cost": f"${item.hedge_cost:,.0f}", "final_equity": f"${item.final_equity:,.0f}", } for item in report.rankings ], row_key="rank", ).classes("w-full") 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("Portfolio Value Paths").classes( "text-lg font-semibold text-slate-900 dark:text-slate-100" ) ui.label( "Baseline series shows the unhedged collateral value path for the same seeded event window." ).classes("text-sm text-slate-500 dark:text-slate-400") ui.echart( _chart_options( chart_model.dates, tuple({"name": item.name, "values": list(item.values)} for item in chart_model.series), ) ).classes("h-96 w-full") def on_preset_change() -> None: if syncing_controls["value"]: return validation_label.set_text("") preview_error = refresh_preview(reset_templates=True, reseed_units=True) if preview_error: validation_label.set_text(preview_error) render_result_state("Scenario validation failed", preview_error, tone="warning") else: mark_results_stale() def on_preview_input_change() -> None: if syncing_controls["value"]: return validation_label.set_text("") preview_error = refresh_preview() if preview_error: validation_label.set_text(preview_error) render_result_state("Scenario validation failed", preview_error, tone="warning") else: mark_results_stale() preset_select.on_value_change(lambda _: on_preset_change()) template_select.on_value_change(lambda _: on_preview_input_change()) units_input.on_value_change(lambda _: on_preview_input_change()) loan_input.on_value_change(lambda _: on_preview_input_change()) ltv_input.on_value_change(lambda _: on_preview_input_change()) run_button.on_click(lambda: render_report()) refresh_preview(reset_templates=False, reseed_units=False) render_report()