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 validate_and_calculate_units(initial_value: float, entry_spot: float) -> tuple[float, str | None]: """Validate inputs and calculate underlying units. Returns (units, error_message). If error_message is not None, units is 0.0. """ if initial_value <= 0: return 0.0, "Initial portfolio value must be positive." if entry_spot <= 0: return 0.0, "Cannot calculate units: entry spot is invalid. Please select a valid preset." return initial_value / entry_spot, None 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") ui.label( "Underlying units will be calculated from initial value ÷ entry spot." ).classes("text-xs text-slate-500 dark:text-slate-400") initial_value_input = ui.number("Initial portfolio value ($)", value=default_units * default_entry_spot, min=0.01, step=1000).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") # Calculate underlying units with validation initial_value = float(initial_value_input.value or 0.0) computed_units, units_error = ( validate_and_calculate_units(initial_value, entry_spot) if entry_spot is not None else (0.0, "Entry spot unavailable.") ) with ui.grid(columns=1).classes("w-full gap-4 sm:grid-cols-2 lg:grid-cols-1 xl:grid-cols-2"): cards = [ ( "Initial portfolio value", f"${float(initial_value_input.value or 0.0):,.0f}", ), ("Templates", str(len(selected_template_slugs()))), ("Underlying units", f"{computed_units:,.0f}" if computed_units > 0 else "—"), ("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") # Show validation errors (units_error takes priority, then entry_spot_error) display_error = units_error or entry_spot_error if display_error: tone_class = "text-rose-600 dark:text-rose-300" if "must be positive" in display_error else "text-amber-700 dark:text-amber-300" ui.label(display_error).classes(f"text-sm {tone_class}") 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: # Get initial portfolio value from UI and derive entry spot preview_initial_value = float(initial_value_input.value or 0.0) preview_entry_spot = service.derive_entry_spot( preset_slug=str(option["slug"]), template_slugs=template_slugs, ) # Validate and calculate underlying units preview_units, units_error = validate_and_calculate_units(preview_initial_value, preview_entry_spot) if units_error: metadata_label.set_text(f"Preset: {option['label']} — {option['description']}") scenario_label.set_text(units_error) render_selected_summary(entry_spot=preview_entry_spot, entry_spot_error=units_error) return units_error if workspace_id and config is not None and reseed_units: # Recalculate from workspace config workspace_units = asset_quantity_from_workspace_config( config, entry_spot=preview_entry_spot, symbol="GLD", ) syncing_controls["value"] = True try: initial_value_input.value = workspace_units * preview_entry_spot preview_units = workspace_units preview_initial_value = workspace_units * preview_entry_spot 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 initial_value=%s loan=%s margin_call_ltv=%s", workspace_id, preset_select.value, selected_template_slugs(), initial_value_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: # Get initial portfolio value and calculate underlying units with validation initial_value = float(initial_value_input.value or 0.0) # Get entry spot from preview option = preset_lookup.get(str(preset_select.value or "")) if option is None: validation_label.set_text("Select a preset to run comparison.") return entry_spot = service.derive_entry_spot( preset_slug=str(option["slug"]), template_slugs=template_slugs, ) # Validate and calculate underlying units underlying_units, units_error = validate_and_calculate_units(initial_value, entry_spot) if units_error: validation_label.set_text(units_error) render_result_state("Input validation failed", units_error, tone="warning") return report = service.run_read_only_comparison( preset_slug=str(preset_select.value or ""), template_slugs=template_slugs, underlying_units=underlying_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: 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 initial_value=%s loan=%s margin_call_ltv=%s", workspace_id, preset_select.value, selected_template_slugs(), initial_value_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) drilldown_options = service.drilldown_options(report) initial_drilldown_slug = next(iter(drilldown_options), None) 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()}", ), ("Initial value", f"${float(initial_value_input.value or 0.0):,.0f}"), ("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("Strategy Drilldown").classes("text-lg font-semibold text-slate-900 dark:text-slate-100") ui.label( "Select a ranked strategy to inspect margin-call pressure, payoff realization, and the full seeded daily path." ).classes("text-sm text-slate-500 dark:text-slate-400") drilldown_select = ui.select( drilldown_options, value=initial_drilldown_slug, label="Strategy drilldown", ).classes("w-full") drilldown_container = ui.column().classes("w-full gap-4") def render_drilldown() -> None: drilldown_container.clear() if drilldown_select.value is None: return drilldown = service.drilldown_model(report, template_slug=str(drilldown_select.value)) breach_dates = ", ".join(drilldown.breach_dates) if drilldown.breach_dates else "None" worst_ltv_point = ( f"{drilldown.worst_ltv_date} · {drilldown.worst_ltv_hedged:.1%}" if drilldown.worst_ltv_date is not None else "Unavailable" ) with drilldown_container: ui.label(f"Selected strategy: {drilldown.template_name}").classes( "text-lg font-semibold text-slate-900 dark:text-slate-100" ) ui.label( f"Rank #{drilldown.rank} · {'Survived margin call' if drilldown.survived_margin_call else 'Breached margin threshold'}" ).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 = [ ("Margin-call days", str(drilldown.margin_call_days_hedged)), ("Payoff realized", f"${drilldown.total_option_payoff_realized:,.0f}"), ("Hedge cost", f"${drilldown.hedge_cost:,.0f}"), ("Final equity", f"${drilldown.final_equity:,.0f}"), ] 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.grid(columns=2).classes("w-full gap-4 max-md:grid-cols-1"): 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("Worst LTV point").classes("text-sm text-slate-500 dark:text-slate-400") ui.label(worst_ltv_point).classes( "text-xl font-bold text-slate-900 dark:text-slate-100" ) with ui.card().classes( "rounded-xl border border-amber-200 bg-amber-50 p-4 shadow-none dark:border-amber-900/60 dark:bg-amber-950/30" ): ui.label("Margin threshold breach dates").classes( "text-sm text-amber-700 dark:text-amber-300" ) ui.label(breach_dates).classes( "text-base font-semibold text-amber-800 dark:text-amber-200" ) with ui.card().classes( "w-full rounded-xl border border-slate-200 bg-slate-50 p-4 shadow-none dark:border-slate-800 dark:bg-slate-950" ): ui.label("Daily path details").classes( "text-base 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": "option_market_value", "label": "Option value", "field": "option_market_value", "align": "right", }, { "name": "realized_option_cashflow", "label": "Payoff realized", "field": "realized_option_cashflow", "align": "right", }, { "name": "ltv_hedged", "label": "Hedged LTV", "field": "ltv_hedged", "align": "right", }, { "name": "margin_call_hedged", "label": "Breach", "field": "margin_call_hedged", "align": "center", }, ], rows=[ { "date": row.date, "spot_close": f"${row.spot_close:,.2f}", "net_portfolio_value": f"${row.net_portfolio_value:,.0f}", "option_market_value": f"${row.option_market_value:,.0f}", "realized_option_cashflow": f"${row.realized_option_cashflow:,.0f}", "ltv_hedged": f"{row.ltv_hedged:.1%}", "margin_call_hedged": "Yes" if row.margin_call_hedged else "No", } for row in drilldown.rows ], row_key="date", ).classes("w-full") drilldown_select.on_value_change(lambda _: render_drilldown()) render_drilldown() 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=False) 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()) initial_value_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()