from __future__ import annotations import logging from fastapi.responses import RedirectResponse from nicegui import ui from app.domain.portfolio_math import resolve_portfolio_spot_from_quote from app.models.workspace import get_workspace_repository from app.pages.common import ( dashboard_page, demo_spot_price, portfolio_snapshot, split_page_panes, strategy_metrics, ) from app.services.runtime import get_data_service from app.services.strategy_templates import StrategyTemplateService logger = logging.getLogger(__name__) def _cost_benefit_options(metrics: dict) -> dict: return { "tooltip": {"trigger": "axis"}, "grid": {"left": 64, "right": 24, "top": 24, "bottom": 48}, "xAxis": { "type": "category", "data": [f"${point['price']:.0f}" for point in metrics["scenario_series"]], "name": "Collateral spot", }, "yAxis": {"type": "value", "name": "Net hedge benefit / oz"}, "series": [ { "type": "bar", "data": [point["benefit"] for point in metrics["scenario_series"]], "itemStyle": { "color": "#0ea5e9", }, "markLine": { "symbol": "none", "lineStyle": {"color": "#94a3b8", "type": "dashed"}, "data": [{"yAxis": 0}], }, } ], } def _waterfall_options(metrics: dict) -> dict: steps = metrics["waterfall_steps"] values: list[dict[str, object]] = [] for label, amount in steps: color = "#0ea5e9" if label == "Net equity" else ("#22c55e" if amount >= 0 else "#ef4444") values.append({"value": amount, "itemStyle": {"color": color}}) return { "tooltip": {"trigger": "axis", "axisPointer": {"type": "shadow"}}, "grid": {"left": 80, "right": 24, "top": 24, "bottom": 48}, "xAxis": {"type": "category", "data": [label for label, _ in steps]}, "yAxis": {"type": "value", "name": "USD"}, "series": [ { "type": "bar", "data": values, "label": {"show": True, "position": "top", "formatter": "{c}"}, }, ], } @ui.page("/{workspace_id}/hedge") async def workspace_hedge_page(workspace_id: str) -> None: repo = get_workspace_repository() if not repo.workspace_exists(workspace_id): return RedirectResponse(url="/", status_code=307) await _render_hedge_page(workspace_id=workspace_id) async def _resolve_hedge_spot(workspace_id: str | None = None) -> tuple[dict[str, float], str, str]: """Resolve hedge page spot price using the same quote-unit seam as overview.""" repo = get_workspace_repository() config = repo.load_portfolio_config(workspace_id) if workspace_id else None if config is None: return {"spot_price": demo_spot_price()}, "demo", "" try: data_service = get_data_service() underlying = config.underlying or "GLD" quote = await data_service.get_quote(underlying) spot, source, updated_at = resolve_portfolio_spot_from_quote( config, quote, fallback_symbol=underlying ) portfolio = portfolio_snapshot(config, runtime_spot_price=spot) return portfolio, source, updated_at except Exception as exc: logger.warning("Falling back to configured hedge spot for workspace %s: %s", workspace_id, exc) portfolio = portfolio_snapshot(config) return portfolio, "configured_entry_price", "" async def _render_hedge_page(workspace_id: str | None = None) -> None: portfolio, quote_source, quote_updated_at = await _resolve_hedge_spot(workspace_id) template_service = StrategyTemplateService() strategies_state = {"items": template_service.catalog_items()} def strategy_map() -> dict[str, str]: return {strategy["label"]: strategy["template_slug"] for strategy in strategies_state["items"]} selected = { "strategy": strategies_state["items"][0]["template_slug"], "label": strategies_state["items"][0]["label"], "scenario_pct": 0, } if quote_source == "configured_entry_price": spot_label = f"Current spot reference: ${portfolio['spot_price']:,.2f}/oz (configured entry price)" else: spot_label = ( f"Current spot reference: ${portfolio['spot_price']:,.2f}/oz (converted collateral spot via {quote_source})" ) updated_label = f"Quote timestamp: {quote_updated_at}" if quote_updated_at else "Quote timestamp: unavailable" # Get underlying for display underlying = "GLD" if workspace_id: try: repo = get_workspace_repository() config = repo.load_portfolio_config(workspace_id) underlying = config.underlying or "GLD" except Exception: pass with dashboard_page( "Hedge Analysis", f"Compare hedge structures across scenarios, visualize cost-benefit tradeoffs, and inspect net equity impacts for {underlying}.", "hedge", workspace_id=workspace_id, ): with ui.row().classes("w-full items-center justify-between gap-4 max-md:flex-col max-md:items-start"): ui.label(f"Active underlying: {underlying}").classes("text-sm text-slate-500 dark:text-slate-400") left_pane, right_pane = split_page_panes( left_testid="hedge-left-pane", right_testid="hedge-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("Strategy Controls").classes("text-lg font-semibold text-slate-900 dark:text-slate-100") selector = ui.select( list(strategy_map().keys()), value=selected["label"], label="Strategy selector" ).classes("w-full") slider_value = ui.label("Scenario move: +0%").classes("text-sm text-slate-500 dark:text-slate-400") slider = ui.slider(min=-25, max=25, value=0, step=5).classes("w-full") ui.label(spot_label).classes("text-sm text-slate-500 dark:text-slate-400") ui.label(updated_label).classes("text-xs text-slate-500 dark:text-slate-400") if workspace_id: ui.label(f"Workspace route: /{workspace_id}/hedge").classes( "text-xs text-slate-500 dark:text-slate-400" ) else: ui.label(f"Demo spot reference: ${demo_spot_price():,.2f}").classes( "text-xs text-slate-500 dark:text-slate-400" ) with ( ui.card() .classes( "w-full rounded-2xl border border-slate-200 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-900" ) .props("data-testid=strategy-builder-card") ): ui.label("Strategy Builder").classes("text-lg font-semibold text-slate-900 dark:text-slate-100") ui.label( "Save a custom protective put or equal-weight two-leg ladder for reuse across hedge, backtests, and event comparison." ).classes("text-sm text-slate-500 dark:text-slate-400") builder_type_options = {"Protective put": "protective_put", "Two-leg ladder": "laddered_put"} builder_name = ui.input("Template name", placeholder="Crash Guard 97").classes("w-full") builder_type = ui.select( list(builder_type_options.keys()), value="Protective put", label="Strategy type", ).classes("w-full") builder_expiry_days = ui.number("Expiration days", value=365, min=30, step=30).classes("w-full") builder_primary_strike = ui.number( "Primary strike (% of spot)", value=100, min=1, max=150, step=1, ).classes("w-full") builder_secondary_strike = ui.number( "Secondary strike (% of spot)", value=95, min=1, max=150, step=1, ).classes("w-full") ui.label("Two-leg ladders currently save with equal 50/50 weights.").classes( "text-xs text-slate-500 dark:text-slate-400" ) builder_status = ui.label("").classes("text-sm text-slate-600 dark:text-slate-300") save_template_button = ui.button("Save template").props("color=primary outline") 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: scenario_results = ui.card().classes( "w-full rounded-2xl border border-slate-200 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-900" ) with ui.row().classes("w-full gap-6 max-xl:flex-col"): initial_metrics = strategy_metrics(selected["strategy"], selected["scenario_pct"], portfolio=portfolio) cost_chart = ui.echart(_cost_benefit_options(initial_metrics)).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" ) waterfall_chart = ui.echart(_waterfall_options(initial_metrics)).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" ) syncing_controls = {"value": False} def refresh_available_strategies() -> None: strategies_state["items"] = template_service.catalog_items() options = strategy_map() syncing_controls["value"] = True try: selector.options = list(options.keys()) if selected["label"] not in options: first_label = next(iter(options)) selected["label"] = first_label selected["strategy"] = options[first_label] selector.value = first_label selector.update() finally: syncing_controls["value"] = False def render_summary() -> None: metrics = strategy_metrics(selected["strategy"], selected["scenario_pct"], portfolio=portfolio) strategy = metrics["strategy"] summary.clear() with summary: ui.label("Scenario Summary").classes("text-lg font-semibold text-slate-900 dark:text-slate-100") ui.label(f"Selected template: {strategy['label']}").classes( "text-sm text-slate-500 dark:text-slate-400" ) ui.label(strategy["description"]).classes("text-sm text-slate-600 dark:text-slate-300") with ui.grid(columns=1).classes("w-full gap-4 sm:grid-cols-2 lg:grid-cols-1 xl:grid-cols-2"): cards = [ ("Start value", f"${portfolio['gold_value']:,.0f}"), ("Start price", f"${portfolio['spot_price']:,.2f}/oz"), ("Weight", f"{portfolio['gold_units']:,.0f} oz"), ("Loan amount", f"${portfolio['loan_amount']:,.0f}"), ("Margin call LTV", f"{portfolio['margin_call_ltv']:.1%}"), ("Monthly hedge budget", f"${portfolio['hedge_budget']:,.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-2xl font-bold text-slate-900 dark:text-slate-100") scenario_results.clear() with scenario_results: ui.label("Scenario Results").classes("text-lg font-semibold text-slate-900 dark:text-slate-100") with ui.grid(columns=2).classes("w-full gap-4 max-md:grid-cols-1"): result_cards = [ ("Scenario spot", f"${metrics['scenario_price']:,.2f}"), ("Hedge cost", f"${float(strategy.get('estimated_cost', 0.0)):,.2f}/oz"), ("Unhedged equity", f"${metrics['unhedged_equity']:,.0f}"), ("Hedged equity", f"${metrics['hedged_equity']:,.0f}"), ("Net hedge benefit", f"${metrics['hedged_equity'] - metrics['unhedged_equity']:,.0f}"), ("Scenario move", f"{selected['scenario_pct']:+d}%"), ] for label, value in result_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") cost_chart.options.clear() cost_chart.options.update(_cost_benefit_options(metrics)) cost_chart.update() waterfall_chart.options.clear() waterfall_chart.options.update(_waterfall_options(metrics)) waterfall_chart.update() def refresh_from_selector(event) -> None: if syncing_controls["value"]: return selected["label"] = str(event.value) selected["strategy"] = strategy_map()[selected["label"]] render_summary() def refresh_from_slider(event) -> None: selected["scenario_pct"] = int(event.value) sign = "+" if selected["scenario_pct"] >= 0 else "" slider_value.set_text(f"Scenario move: {sign}{selected['scenario_pct']}%") render_summary() def save_template() -> None: builder_status.set_text("") try: builder_kind = builder_type_options[str(builder_type.value)] strikes = (float(builder_primary_strike.value or 0.0) / 100.0,) weights: tuple[float, ...] | None = None if builder_kind == "laddered_put": strikes = ( float(builder_primary_strike.value or 0.0) / 100.0, float(builder_secondary_strike.value or 0.0) / 100.0, ) weights = (0.5, 0.5) template = template_service.create_custom_template( display_name=str(builder_name.value or ""), template_kind=builder_kind, target_expiry_days=int(builder_expiry_days.value or 0), strike_pcts=strikes, weights=weights, ) except (ValueError, KeyError) as exc: builder_status.set_text(str(exc)) return refresh_available_strategies() selected["label"] = template.display_name selected["strategy"] = template.slug syncing_controls["value"] = True try: selector.value = template.display_name selector.update() finally: syncing_controls["value"] = False builder_status.set_text( f"Saved template {template.display_name}. Reusable on hedge, backtests, and event comparison." ) render_summary() selector.on_value_change(refresh_from_selector) slider.on_value_change(refresh_from_slider) save_template_button.on_click(lambda: save_template()) render_summary()