from __future__ import annotations from typing import Any from nicegui import ui from app.components import GreeksTable from app.pages.common import dashboard_page, strategy_catalog from app.services.runtime import get_data_service @ui.page("/options") async def options_page() -> None: chain_data = await get_data_service().get_options_chain("GLD") chain = list(chain_data.get("rows") or [*chain_data.get("calls", []), *chain_data.get("puts", [])]) expiries = list(chain_data.get("expirations") or sorted({row["expiry"] for row in chain})) strike_values = sorted({float(row["strike"]) for row in chain}) selected_expiry = {"value": expiries[0] if expiries else None} strike_range = { "min": strike_values[0] if strike_values else 0.0, "max": strike_values[-1] if strike_values else 0.0, } selected_strategy = {"value": strategy_catalog()[0]["label"]} chosen_contracts: list[dict[str, Any]] = [] with dashboard_page( "Options Chain", "Browse GLD contracts, filter by expiry and strike range, inspect Greeks, and attach contracts to hedge workflows.", "options", ): with ui.row().classes("w-full gap-6 max-lg:flex-col"): 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("Filters").classes("text-lg font-semibold text-slate-900 dark:text-slate-100") expiry_select = ui.select(expiries, value=selected_expiry["value"], label="Expiry").classes("w-full") min_strike = ui.number( "Min strike", value=strike_range["min"], min=strike_values[0] if strike_values else 0.0, max=strike_values[-1] if strike_values else 0.0, step=5, ).classes("w-full") max_strike = ui.number( "Max strike", value=strike_range["max"], min=strike_values[0] if strike_values else 0.0, max=strike_values[-1] if strike_values else 0.0, step=5, ).classes("w-full") strategy_select = ui.select( [item["label"] for item in strategy_catalog()], value=selected_strategy["value"], label="Add to hedge strategy", ).classes("w-full") source_label = f"Source: {chain_data.get('source', 'unknown')}" if chain_data.get("updated_at"): source_label += f" · Updated {chain_data['updated_at']}" ui.label(source_label).classes("text-xs text-slate-500 dark:text-slate-400") if chain_data.get("error"): ui.label(f"Options data unavailable: {chain_data['error']}").classes( "text-xs text-amber-700 dark:text-amber-300" ) selection_card = ui.card().classes( "w-full rounded-2xl border border-slate-200 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-900" ) chain_table = ui.html("").classes("w-full") greeks = GreeksTable([]) def filtered_rows() -> list[dict[str, Any]]: if not selected_expiry["value"]: return [] return [ row for row in chain if row["expiry"] == selected_expiry["value"] and strike_range["min"] <= float(row["strike"]) <= strike_range["max"] ] def render_selection() -> None: selection_card.clear() with selection_card: ui.label("Strategy Integration").classes("text-lg font-semibold text-slate-900 dark:text-slate-100") ui.label(f"Target strategy: {selected_strategy['value']}").classes( "text-sm text-slate-500 dark:text-slate-400" ) if not chosen_contracts: ui.label("No contracts added yet.").classes("text-sm text-slate-500 dark:text-slate-400") return for contract in chosen_contracts[-3:]: ui.label( f"{contract['symbol']} · premium ${float(contract['premium']):.2f} · IV {float(contract.get('impliedVolatility', 0.0)):.1%}" ).classes("text-sm text-slate-600 dark:text-slate-300") def add_to_strategy(contract: dict[str, Any]) -> None: chosen_contracts.append(contract) render_selection() greeks.set_options(chosen_contracts[-6:]) ui.notify( f"Added {contract['symbol']} to {selected_strategy['value']}", color="positive", ) def render_chain() -> None: rows = filtered_rows() chain_table.content = ( """
""" + "".join(f""" """ for row in rows) + ( "" if rows else "" ) + """
Contract Type Strike Bid / Ask Last IV Greeks Action
{row['symbol']} {row['type'].upper()} ${float(row['strike']):.2f} ${float(row['bid']):.2f} / ${float(row['ask']):.2f} ${float(row.get('lastPrice', row.get('premium', 0.0))):.2f} {float(row.get('impliedVolatility', 0.0)):.1%} Δ {float(row.get('delta', 0.0)):+.3f} · Γ {float(row.get('gamma', 0.0)):.3f} · Θ {float(row.get('theta', 0.0)):+.3f} Use quick-add buttons below
No contracts match the current filter.
""" ) chain_table.update() quick_add.clear() with quick_add: ui.label("Quick add to hedge").classes( "text-sm font-semibold uppercase tracking-wide text-slate-500 dark:text-slate-400" ) with ui.row().classes("w-full gap-2 max-sm:flex-col"): for row in rows[:6]: ui.button( f"Add {row['type'].upper()} {float(row['strike']):.0f}", on_click=lambda _, contract=row: add_to_strategy(contract), ).props("outline color=primary") greeks.set_options(rows[:6]) quick_add = ui.card().classes( "w-full rounded-2xl border border-slate-200 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-900" ) def update_filters() -> None: selected_expiry["value"] = expiry_select.value strike_range["min"] = float(min_strike.value or 0.0) strike_range["max"] = float(max_strike.value or 0.0) if strike_range["min"] > strike_range["max"]: strike_range["min"], strike_range["max"] = ( strike_range["max"], strike_range["min"], ) min_strike.value = strike_range["min"] max_strike.value = strike_range["max"] render_chain() expiry_select.on_value_change(lambda _: update_filters()) min_strike.on_value_change(lambda _: update_filters()) max_strike.on_value_change(lambda _: update_filters()) def on_strategy_change(event) -> None: selected_strategy["value"] = event.value # type: ignore[assignment] render_selection() strategy_select.on_value_change(on_strategy_change) render_selection() render_chain()