Skip to content

Dashboard

The dashboard package contains all components for the Streamlit dashboard UI, including the main service layer, plotting functions, and UI widgets.

plot_calendar_heatmap 

plot_calendar_heatmap(market_data: DataFrame) -> go.Figure

Plot the calandar heatmap.

Source code in src/optpricing/dashboard/plots.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def plot_calendar_heatmap(market_data: pd.DataFrame) -> go.Figure:
    """Plot the calandar heatmap."""
    if market_data.empty:
        return go.Figure().update_layout(title="No market data to plot calendar.")
    df = market_data.copy()
    df["expiry_date"] = pd.to_datetime(df["expiry"]).dt.date
    volume_pivot = df.pivot_table(
        index="expiry_date", columns="strike", values="volume", aggfunc="sum"
    )
    fig = go.Figure(
        data=go.Heatmap(
            z=volume_pivot.values,
            x=volume_pivot.columns,
            y=[d.strftime("%Y-%m-%d") for d in volume_pivot.index],
            colorscale="Viridis",
            hovertemplate="Strike: %{x}<br>Expiry: %{y}<br>Total Volume: %{z:,.0f}<extra></extra>",  # noqa E501
        )
    )
    fig.update_layout(
        title="<b>Option Volume Calendar</b>",
        xaxis_title="Strike",
        yaxis_title="Expiry",
        height=600,
    )
    return fig

plot_error_heatmap 

plot_error_heatmap(market_surface: DataFrame, model_surface: DataFrame, model_name: str) -> go.Figure

Plots a heatmap of the pricing error (Model IV - Market IV) in basis points. FIXED: Handles duplicate index entries before pivoting.

Source code in src/optpricing/dashboard/plots.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
def plot_error_heatmap(
    market_surface: pd.DataFrame,
    model_surface: pd.DataFrame,
    model_name: str,
) -> go.Figure:
    """
    Plots a heatmap of the pricing error (Model IV - Market IV) in basis points.
    FIXED: Handles duplicate index entries before pivoting.
    """
    if market_surface.empty or model_surface.empty:
        return go.Figure().update_layout(
            title=f"Not enough data to plot error for {model_name}."
        )

    market_dedup = market_surface.drop_duplicates(subset=["expiry", "strike"])
    model_dedup = model_surface.drop_duplicates(subset=["expiry", "strike"])

    merged = pd.merge(
        market_dedup,
        model_dedup,
        on=["expiry", "strike"],
        suffixes=("_market", "_model"),
    )
    merged["error"] = (
        merged["iv_model"] - merged["iv_market"]
    ) * 10000  # Convert to bps

    pivot_df = merged.pivot(index="expiry", columns="strike", values="error")

    fig = go.Figure(
        data=go.Heatmap(
            z=pivot_df.values,
            x=pivot_df.columns,
            y=[pd.to_datetime(d).strftime("%Y-%m-%d") for d in pivot_df.index],
            colorscale="RdBu",
            zmid=0,
            hovertemplate="Strike: %{x}<br>Expiry: %{y}<br>Error: %{z:.2f} bps<extra></extra>",  # noqa E501
        )
    )

    fig.update_layout(
        title=f"<b>{model_name} vs. Market IV Error</b> (in basis points)",
        xaxis_title="Strike",
        yaxis_title="Expiry",
        height=600,
    )
    return fig

plot_iv_surface_3d 

plot_iv_surface_3d(market_surface: DataFrame, model_surfaces: dict[str, DataFrame]) -> go.Figure

Creates an interactive, INTERPOLATED 3D plot of the volatility surfaces.

Source code in src/optpricing/dashboard/plots.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def plot_iv_surface_3d(
    market_surface: pd.DataFrame,
    model_surfaces: dict[str, pd.DataFrame],
) -> go.Figure:
    """
    Creates an interactive, INTERPOLATED 3D plot of the volatility surfaces.
    """
    fig = go.Figure()

    if market_surface.empty:
        return fig.update_layout(title="No market data for 3D surface.")

    # regular grid to interpolate onto
    grid_t, grid_k = np.mgrid[
        market_surface["maturity"].min() : market_surface["maturity"].max() : 100j,
        market_surface["strike"].min() : market_surface["strike"].max() : 100j,
    ]

    market_points = market_surface[["maturity", "strike"]].values
    market_values = market_surface["iv"].values
    grid_z_market = griddata(
        market_points, market_values, (grid_t, grid_k), method="cubic"
    )

    fig.add_trace(
        go.Surface(
            z=grid_z_market,
            x=grid_t,
            y=grid_k,
            name="Market IV",
            colorscale="Viridis",
            opacity=0.7,
            showscale=False,
        )
    )

    # Interpolate model data
    for name, surface in model_surfaces.items():
        if not surface.empty:
            model_points = surface[["maturity", "strike"]].values
            model_values = surface["iv"].values
            grid_z_model = griddata(
                model_points, model_values, (grid_t, grid_k), method="cubic"
            )
            fig.add_trace(
                go.Surface(
                    z=grid_z_model,
                    x=grid_t,
                    y=grid_k,
                    name=f"{name} IV",
                    opacity=0.9,
                    showscale=False,
                )
            )

    fig.update_layout(
        title="<b>Interpolated Implied Volatility Surface</b>",
        scene=dict(
            xaxis_title="Maturity (Years)",
            yaxis_title="Strike",
            zaxis_title="Implied Volatility",
            zaxis=dict(tickformat=".2%"),
        ),
        height=700,
        margin=dict(l=0, r=0, b=0, t=40),
        legend=dict(yanchor="top", y=0.9, xanchor="left", x=0.01),
    )
    return fig

plot_smiles_by_expiry 

plot_smiles_by_expiry(market_surface: DataFrame, model_surfaces: dict[str, DataFrame]) -> go.Figure

Generates a robust Plotly figure with volatility smiles for key expiries.

Source code in src/optpricing/dashboard/plots.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def plot_smiles_by_expiry(
    market_surface: pd.DataFrame,
    model_surfaces: dict[str, pd.DataFrame],
) -> go.Figure:
    """
    Generates a robust Plotly figure with volatility smiles for key expiries.
    """
    if market_surface.empty:
        return go.Figure().update_layout(title="No market data to plot smiles.")

    expiries = sorted(market_surface["expiry"].unique())
    plot_expiries = expiries[:4]

    fig = make_subplots(
        rows=2,
        cols=2,
        subplot_titles=[
            f"Expiry: {pd.to_datetime(e).strftime('%Y-%m-%d')}" for e in plot_expiries
        ],
        vertical_spacing=0.15,
        horizontal_spacing=0.05,
    )

    for i, expiry in enumerate(plot_expiries):
        row, col = i // 2 + 1, i % 2 + 1
        market_slice = market_surface[market_surface["expiry"] == expiry]

        fig.add_trace(
            go.Scatter(
                x=market_slice["strike"],
                y=market_slice["iv"],
                mode="markers",
                name="Market IV",
                legendgroup="market",
                showlegend=(i == 0),
                marker=dict(symbol="cross", size=8, color="black"),
            ),
            row=row,
            col=col,
        )

        for model_name, model_surface in model_surfaces.items():
            if model_surface.empty:
                continue
            model_slice = model_surface[model_surface["expiry"] == expiry].sort_values(
                "strike"
            )
            fig.add_trace(
                go.Scatter(
                    x=model_slice["strike"],
                    y=model_slice["iv"],
                    mode="lines",
                    name=f"{model_name} IV",
                    legendgroup=model_name,
                    showlegend=(i == 0),
                    line=dict(width=3),
                ),
                row=row,
                col=col,
            )

    # Hide any unused subplots
    for i in range(len(plot_expiries), 4):
        row, col = i // 2 + 1, i % 2 + 1
        fig.update_xaxes(visible=False, row=row, col=col)
        fig.update_yaxes(visible=False, row=row, col=col)

    fig.update_layout(
        height=700,
        title_text="<b>Volatility Smiles by Expiration</b>",
        legend_title_text="Models",
        margin=dict(l=40, r=20, t=60, b=40),
    )
    fig.update_yaxes(title_text="Implied Volatility", tickformat=".2%")
    fig.update_xaxes(title_text="Strike")
    return fig

DashboardService 

DashboardService(ticker: str, snapshot_date: str, model_configs: dict[str, Any], *, settings: CalibrationSettings | None = None, use_tuned: bool = True, allow_mc: bool = False)

Loads a chain (live or snapshot), calibrates the selected models with the offline :class:~optpricing.workflows.DailyWorkflow and builds the market and model implied-volatility surfaces for the plots.

Parameters:

Name Type Description Default
ticker str

Ticker symbol.

 required
snapshot_date str

"Live Data" or a stored snapshot date.

 required
model_configs dict[str, Any]

Recipes keyed by model name.

 required
settings CalibrationSettings | None

Calibration settings for every model (defaults if omitted).

 None
use_tuned bool

Prefer the settings cached by optpricing tune for this ticker when available, by default True.

 True
allow_mc bool

Permit Monte Carlo pricing for simulation-only models.

 False
Source code in src/optpricing/dashboard/service.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def __init__(
    self,
    ticker: str,
    snapshot_date: str,
    model_configs: dict[str, Any],
    *,
    settings: CalibrationSettings | None = None,
    use_tuned: bool = True,
    allow_mc: bool = False,
):
    self.ticker = ticker
    self.snapshot_date = snapshot_date
    self.model_configs = model_configs
    self.settings = settings or CalibrationSettings()
    self.use_tuned = use_tuned
    self.allow_mc = allow_mc
    self._market_data: pd.DataFrame | None = None
    self.calibrated_models: dict[str, Any] = {}
    self.workflows: dict[str, DailyWorkflow] = {}
    self.summary_df: pd.DataFrame | None = None
    self.bucket_tables: dict[str, pd.DataFrame] = {}
    self.market_surface: pd.DataFrame | None = None
    self.model_surfaces: dict[str, pd.DataFrame] = {}
    self.final_rate: Rate | None = None
    self.final_stock: Stock | None = None

calculate_iv_surfaces 

calculate_iv_surfaces() -> None

Market and per-model Black-Scholes implied-volatility surfaces.

Source code in src/optpricing/dashboard/service.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def calculate_iv_surfaces(self) -> None:
    """Market and per-model Black-Scholes implied-volatility surfaces."""
    if not self.final_rate or not self.final_stock:
        msg_ = (
            "Cannot calculate IV surfaces without a"
            + " successful calibration run to determine r and q"
        )
        raise RuntimeError(msg_)

    with st.spinner("Calculating Market IV surface..."):
        self.market_surface = (
            VolatilitySurface(self.market_data)
            .calculate_market_iv(self.final_stock, self.final_rate)
            .surface
        )

    for name, model in self.calibrated_models.items():
        with st.spinner(f"Calculating {name} IV surface..."):
            technique = select_fastest_technique(model)
            try:
                self.model_surfaces[name] = (
                    VolatilitySurface(self.market_data)
                    .calculate_model_iv(
                        self.final_stock,
                        self.final_rate,
                        model,
                        technique,
                        allow_mc=self.allow_mc,
                    )
                    .surface
                )
            except NotImplementedError as e:
                logger.warning("No IV surface for %s: %s", name, e)

run_calibrations 

run_calibrations() -> None

Calibrates every configured model and fills the summary table.

Source code in src/optpricing/dashboard/service.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def run_calibrations(self) -> None:
    """Calibrates every configured model and fills the summary table."""
    all_results = []
    for model_name, config in self.model_configs.items():
        with st.spinner(f"Running workflow for {model_name}..."):
            config = dict(config)
            config["ticker"] = self.ticker
            settings = self.settings_for(model_name)
            workflow = DailyWorkflow(
                market_data=self.market_data,
                model_config=config,
                settings=settings,
                allow_mc=self.allow_mc,
            )
            workflow.run()
            self.workflows[model_name] = workflow
            result = dict(workflow.results)
            result["Model"] = model_name
            result["Settings"] = settings.name
            buckets = result.pop("Buckets", None)
            if isinstance(buckets, pd.DataFrame):
                self.bucket_tables[model_name] = buckets
            all_results.append(result)

            if result.get("Status") == "Success":
                model_class = config["model_class"]
                calibrated_params = result.get("Calibrated Params")
                full_params = dict(model_class.default_params)
                full_params.update(calibrated_params)
                self.calibrated_models[model_name] = model_class(params=full_params)

                if self.final_rate is None and getattr(workflow, "rate", None):
                    self.final_rate = workflow.rate
                    self.final_stock = workflow.stock

    if not all_results:
        raise RuntimeError("Calibration workflows failed to produce any results.")
    frame = pd.DataFrame(all_results).set_index("Model")
    keep = [c for c in SUMMARY_COLUMNS if c in frame.columns]
    frame = frame[keep]
    if "Calibrated Params" in frame.columns:
        frame["Calibrated Params"] = frame["Calibrated Params"].map(
            lambda p: (
                ", ".join(f"{k}={v:.4g}" for k, v in p.items())
                if isinstance(p, dict)
                else p
            )
        )
    if "IV RMSE" in frame.columns:
        frame = frame.sort_values("IV RMSE", na_position="last")
    self.summary_df = frame

settings_for 

settings_for(model_name: str) -> CalibrationSettings

Tuned settings for this ticker/model when requested and cached.

Source code in src/optpricing/dashboard/service.py 
100
101
102
103
104
105
106
def settings_for(self, model_name: str) -> CalibrationSettings:
    """Tuned settings for this ticker/model when requested and cached."""
    if self.use_tuned:
        tuned = load_tuned_settings(self.ticker, model_name)
        if tuned is not None:
            return tuned
    return self.settings

build_sidebar 

build_sidebar() -> tuple[str, str, str]

Builds the consistent sidebar for the dashboard and returns the selections.

Uses st.query_params to manage state for deep-linking.

Returns:

Type Description
tuple[str, str, str]

A tuple containing the selected ticker, snapshot_date, and model_name.
Source code in src/optpricing/dashboard/widgets.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def build_sidebar() -> tuple[str, str, str]:
    """
    Builds the consistent sidebar for the dashboard and returns the selections.

    Uses st.query_params to manage state for deep-linking.

    Returns
    -------
    tuple[str, str, str]
        A tuple containing the selected ticker, snapshot_date, and model_name.
    """
    with st.sidebar:
        st.header("Global Configuration")

        params = st.query_params
        default_ticker = params.get(
            "ticker", _config.get("default_tickers", ["SPY"])[0]
        )

        available_tickers = _config.get(
            "default_tickers",
            ["SPY", "AAPL", "TSLA", "NVDA"],
        )
        ticker_index = (
            available_tickers.index(default_ticker)
            if default_ticker in available_tickers
            else 0
        )
        ticker = st.selectbox(
            "Ticker",
            available_tickers,
            index=ticker_index,
            key="ticker_selector",
        )

        try:
            data_source_options = get_available_snapshot_dates(ticker)
        except FileNotFoundError:
            data_source_options = []

        data_source_options.insert(0, "Live Data")
        default_date = params.get("date", "Live Data")
        date_index = (
            data_source_options.index(default_date)
            if default_date in data_source_options
            else 0
        )
        snapshot_date = st.selectbox(
            "Snapshot Date",
            data_source_options,
            index=date_index,
            key="date_selector",
        )

        available_models = list(ALL_MODEL_CONFIGS.keys())
        default_model = params.get("model", "BSM")
        model_index = (
            available_models.index(default_model)
            if default_model in available_models
            else 0
        )
        model_name = st.selectbox(
            "Model",
            available_models,
            index=model_index,
            key="model_selector",
        )

        st.query_params["ticker"] = ticker
        st.query_params["date"] = snapshot_date
        st.query_params["model"] = model_name

        st.markdown("---")
        st.markdown("**Status:**")
        st.markdown(f"  **Ticker:** `{ticker}`")
        st.markdown(f"  **Date:** `{snapshot_date}`")
        st.markdown(f"  **Model:** `{model_name}`")
        st.link_button(
            "Documentation",
            "https://diljit22.github.io/optpricing/",
            use_container_width=True,
        )

    return ticker, snapshot_date, model_name