implied volatility trading dashboard
i just wanted to try my luck if anyone would like to review my code and let me know what they think.
i followed a tutorial (roman paolucci) but instead of using ibapi (interactive brokers) i used yfinance (yahoo finance) and i had to remove IBApp class, connection frame, and create_equity_contract function.
heres is my code!
import tkinter as tk
from tkinter import ttk, messagebox, scrolledtext
import threading
import time
import pandas as pd
import numpy as np
from scipy import stats
import matplotlib.pyplot as plt
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
import yfinance as yf
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings("ignore", category=UserWarning, module="matplotlib")
class ImpliedVolatilityDashboard:
def __init__(self, root):
self.root = root
self.root.title("Implied Volatility Trading Dashboard")
self.root.geometry("1400x1200") # window size
self.option_data = None
self.volatility_data = None
self.current_implied_vol = None
# yfinance doesn't require connection management
self.yfinance_available = True
self.vol_annualization = 252 # scaling of implied volatility
self.setup_ui() # buttons, charts, etc
def setup_ui(self):
main_frame = ttk.Frame(self.root, padding="10")
main_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
self.root.columnconfigure(0, weight=1)
self.root.rowconfigure(0, weight=1)
main_frame.columnconfigure(1, weight=1)
main_frame.rowconfigure(5, weight=1)
# yfinance data frame - no connection needed
# data query frame
data_frame = ttk.LabelFrame(main_frame, text="Data Query", padding="5") # NOT pandas data frame, just a section for user input to query data
data_frame.grid(row=1, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(0, 10))
ttk.Label(data_frame, text="Symbol:").grid(row=0, column=0, padx=(0, 5))
self.symbol_var = tk.StringVar(value="SPY") # default symbol for data query
ttk.Entry(data_frame, textvariable=self.symbol_var, width=10).grid(row=0, column=1, padx=(0, 10))
ttk.Label(data_frame, text="Duration:").grid(row=0, column=2, padx=(0, 5))
self.duration_var = tk.StringVar(value="2 Y") # default duration for historical data query
ttk.Entry(data_frame, textvariable=self.duration_var, width=10).grid(row=0, column=3, padx=(0, 10))
self.query_btn = ttk.Button(data_frame, text="Query IV Data", command=self.query_data)
self.query_btn.grid(row=0, column=4, padx=(0, 10))
self.analyze_btn = ttk.Button(data_frame, text="Analyze Implied Vol", command=self.analyze_volatility, state="disabled")
self.analyze_btn.grid(row=0, column=5, padx=(0, 10))
# vol frame
vol_frame = ttk.LabelFrame(main_frame, text="Current Volatility Analysis", padding="5")
vol_frame.grid(row=2, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(0, 10))
ttk.Label(vol_frame, text="Current Volatility:").grid(row=0, column=0, padx=(0, 5))
self.current_vol_label = ttk.Label(vol_frame, text="N/A", font=("Arial", 12, "bold"))
self.current_vol_label.grid(row=0, column=1, padx=(0, 20))
ttk.Label(vol_frame, text="Computation:").grid(row=0, column=2, padx=(0, 5))
self.vol_computation_label = ttk.Label(vol_frame, text="No data", font=("Arial", 10))
self.vol_computation_label.grid(row=0, column=3, padx=(0, 10))
ttk.Label(vol_frame, text="Vol Range:").grid(row=0, column=4, padx=(0, 5))
self.vol_range_label = ttk.Label(vol_frame, text="N/A", font=("Arial", 10))
self.vol_range_label.grid(row=0, column=5)
# regime frame
regime_frame = ttk.LabelFrame(main_frame, text="Volatility Regime Analysis", padding="5")
regime_frame.grid(row=3, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(0, 10))
ttk.Label(regime_frame, text="Current Regime:").grid(row=0, column=0, padx=(0, 5))
self.regime_label = ttk.Label(regime_frame, text="N/A", font=("Arial", 12, "bold"))
self.regime_label.grid(row=0, column=1, padx=(0, 20))
ttk.Label(regime_frame, text="Percentile:").grid(row=0, column=2, padx=(0, 5))
self.percentile_label = ttk.Label(regime_frame, text="N/A", font=("Arial", 10))
self.percentile_label.grid(row=0, column=3, padx=(0, 20))
ttk.Label(regime_frame, text="Mean Reversion Signal:").grid(row=0, column=4, padx=(0, 5))
self.reversion_label = ttk.Label(regime_frame, text="N/A", font=("Arial", 10))
self.reversion_label.grid(row=0, column=5, padx=(0, 20))
status_frame = ttk.LabelFrame(main_frame, text="Status", padding="5")
status_frame.grid(row=4, column=0, sticky=(tk.W, tk.E), pady=(0, 10))
self.status_text = scrolledtext.ScrolledText(status_frame, height=6, width=80)
self.status_text.grid(row=0, column=0, sticky=(tk.W, tk.E))
status_frame.columnconfigure(0, weight=1)
plot_frame = ttk.LabelFrame(main_frame, text="Implied Volatility Analysis Results", padding="5")
plot_frame.grid(row=5, column=0, columnspan=2, sticky=(tk.W, tk.E, tk.N, tk.S))
plot_frame.columnconfigure(0, weight=1)
plot_frame.rowconfigure(0, weight=1)
# ax2 diff regression, ax3 time series of implied vol and percentiles
self.fig, (self.ax1, self.ax2, self.ax3) = plt.subplots(1, 3, figsize=(18, 6))
self.canvas = FigureCanvasTkAgg(self.fig, plot_frame)
self.canvas.get_tk_widget().grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
def log_message(self, message):
timestamp = datetime.now().strftime("%H:%M:%S")
self.status_text.insert(tk.END, f"[{timestamp}] {message}\n")
self.status_text.see(tk.END)
self.root.update_idletasks()
def get_historical_implied_volatility(self, symbol, duration_str="2 Y"):
"""Get historical implied volatility using Yahoo Finance options data"""
try:
ticker = yf.Ticker(symbol)
# parse duration (e.g., "2 Y" -> 2 years)
duration_years = int(duration_str.split()[0]) if 'Y' in duration_str.upper() else 2
start_date = datetime.now() - timedelta(days=duration_years * 365)
self.log_message(f"Fetching {duration_years} years of price data...")
# get historical stock prices
hist_data = ticker.history(start=start_date, interval="1d")
if hist_data.empty:
self.log_message(f"No historical data found for {symbol}")
return None
self.log_message(f"Calculating volatility from {len(hist_data)} price points...")
# calculate historical volatility as proxy for implied volatility
# this is an approximation since yfinance doesn't provide historical IV
returns = np.log(hist_data['Close'] / hist_data['Close'].shift(1))
rolling_vol = returns.rolling(window=30).std() * np.sqrt(252) # Annualized
current_iv = None
# get current options chain for current IV (run in separate try to avoid blocking)
try:
self.log_message("Fetching options data for current IV...")
expiries = ticker.options
if expiries:
# use nearest expiry
nearest_expiry = expiries[0]
opt_chain = ticker.option_chain(nearest_expiry)
# calculate average IV from ATM options
calls = opt_chain.calls
puts = opt_chain.puts
# find at-the-money options
current_price = hist_data['Close'].iloc[-1]
atm_calls = calls[(calls['strike'] >= current_price * 0.95) &
(calls['strike'] <= current_price * 1.05)]
atm_puts = puts[(puts['strike'] >= current_price * 0.95) &
(puts['strike'] <= current_price * 1.05)]
if not atm_calls.empty and 'impliedVolatility' in atm_calls.columns:
current_iv = atm_calls['impliedVolatility'].mean()
elif not atm_puts.empty and 'impliedVolatility' in atm_puts.columns:
current_iv = atm_puts['impliedVolatility'].mean()
elif not calls.empty and 'impliedVolatility' in calls.columns:
current_iv = calls['impliedVolatility'].mean()
if current_iv is not None:
self.log_message(f"Current IV from options: {current_iv:.4f} ({current_iv*100:.2f}%)")
else:
self.log_message("No options data available")
except Exception as e:
self.log_message(f"Could not fetch options data: {e}")
# create DataFrame with volatility data
vol_data = pd.DataFrame({
'date': rolling_vol.index,
'implied_vol': rolling_vol.values
}).dropna()
# scale to match typical IV ranges (adjustment factor)
vol_data['implied_vol'] = vol_data['implied_vol'] * 1.2 # IV is usually higher than HV
# if we have current IV from options, use it for the most recent data point
if current_iv is not None and len(vol_data) > 0:
vol_data.iloc[-1, vol_data.columns.get_loc('implied_vol')] = current_iv
return vol_data
except Exception as e:
self.log_message(f"Error fetching volatility data: {e}")
return None
def query_data(self):
symbol = self.symbol_var.get().upper()
duration = self.duration_var.get()
self.log_message(f"Querying volatility data for {symbol} using Yahoo Finance...")
self.query_btn.config(state="disabled")
self.analyze_btn.config(state="disabled")
# run data fetching in background thread
def fetch_data_thread():
try:
vol_data = self.get_historical_implied_volatility(symbol, duration)
# update UI from main thread
self.root.after(0, lambda: self.process_data_results(vol_data, symbol))
except Exception as e:
self.root.after(0, lambda: self.handle_data_error(e))
thread = threading.Thread(target=fetch_data_thread, daemon=True)
thread.start()
def process_data_results(self, vol_data, symbol):
if vol_data is not None and len(vol_data) > 0:
self.equity_data = vol_data.copy()
self.equity_data['date'] = pd.to_datetime(self.equity_data['date'])
self.equity_data.set_index('date', inplace=True)
self.equity_data['implied_vol'] = self.equity_data['implied_vol']
self.log_message(f"Received {len(self.equity_data)} volatility points for {symbol}")
self.log_message(f"Data range: {self.equity_data.index.min()} to {self.equity_data.index.max()}")
self.log_message(f"Note: Using calculated volatility as IV proxy")
self.process_implied_volatility()
self.analyze_btn.config(state="normal")
else:
self.log_message("No volatility data received")
self.equity_data = None
self.query_btn.config(state="normal")
def handle_data_error(self, error):
self.log_message(f"Error fetching data: {error}")
self.query_btn.config(state="normal")
def process_implied_volatility(self):
if self.equity_data is None:
return
self.log_message("Processing volatility data...")
self.log_message("All volatility data is annualized")
# optimize rolling calculation - use smaller window for faster processing
window_size = min(252, len(self.equity_data) // 4) # Use 252 or 25% of data, whichever is smaller
self.equity_data['iv_percentile'] = self.equity_data['implied_vol'].rolling(window=window_size, min_periods=30).rank(pct=True)
self.current_implied_vol = self.equity_data['implied_vol'].iloc[-1] if len(self.equity_data) > 0 else None
self.volatility_data = self.equity_data[['implied_vol', 'iv_percentile']].copy()
self.update_current_vol_display()
if self.current_implied_vol is not None:
self.log_message(f"Current volatility: {self.current_implied_vol:.4f} ({self.current_implied_vol*100:.2f}%)")
self.log_message(f"Vol Range: {self.equity_data['implied_vol'].min():.4f} - {self.equity_data['implied_vol'].max():.4f}")
else:
self.log_message("Failed to process volatility data")
def update_current_vol_display(self):
if self.current_implied_vol is not None:
self.current_vol_label.config(text=f"{self.current_implied_vol*100:.2f}%")
self.vol_computation_label.config(text="Calculated from historical prices")
if self.equity_data is not None:
vol_min = self.equity_data['implied_vol'].min()
vol_max = self.equity_data['implied_vol'].max()
vol_mean = self.equity_data['implied_vol'].mean()
range_text = f"Min: {vol_min*100:.2f}% | Max: {vol_max*100:.2f}% | Mean: {vol_mean*100:.2f}%"
else:
range_text = "N/A"
self.vol_range_label.config(text=range_text)
# try later: make this based on percentile (25%, 75%)
if self.current_implied_vol > .4:
self.current_vol_label.config(foreground="red")
elif self.current_implied_vol < .15:
self.current_vol_label.config(foreground="green")
else:
self.current_vol_label.config(foreground="black")
self.update_regime_analysis()
else:
self.current_vol_label.config(text="N/A", foreground="black")
self.vol_computation_label.config(text="No data")
self.vol_range_label.config(text="N/A")
self.regime_label.config(text="N/A")
self.percentile_label.config(text="N/A")
self.reversion_label.config(text="N/A")
def update_regime_analysis(self):
if self.volatility_data is None or self.current_implied_vol is None:
return
current_percentile = self.volatility_data['iv_percentile'].iloc[-1]
if current_percentile > .8:
regime = "HIGH VOLATILITY"
color = "red"
elif current_percentile > .6:
regime = "ABOVE AVG VOLATILITY"
color = "orange"
elif current_percentile > .4:
regime = "NORMAL VOLATILITY"
color = "black"
elif current_percentile > .2:
regime = "BELOW AVG VOLATILITY"
color = "blue"
else:
regime = "LOW VOLATILITY"
color = "green"
self.regime_label.config(text=regime, foreground=color)
self.percentile_label.config(text=f"{current_percentile:.1%}")
if current_percentile > .8:
reversion = "EXPECT MEAN REVERSION DOWN"
rcolor = 'blue'
elif current_percentile < .2:
reversion = "EXPECT MEAN REVERSION UP"
rcolor = 'orange'
else:
reversion = "NEUTRAL"
rcolor = 'black'
self.reversion_label.config(text=reversion, foreground=rcolor)
def analyze_volatility(self):
if self.equity_data is None or self.volatility_data is None:
messagebox.showerror("Error", "No volatility data is available for analysis")
return
self.log_message("Analyzing volatility data...")
self.analyze_btn.config(state="disabled")
# run analysis in background thread
def analysis_thread():
try:
# perform heavy calculations
vol_forward_30d = self.volatility_data['implied_vol'].rolling(window=30, min_periods=1).mean().shift(-30)
analysis_df = pd.DataFrame({
'current_iv': self.volatility_data['implied_vol'],
'forward_30d_vol': vol_forward_30d,
'vol_diff': vol_forward_30d - self.volatility_data['implied_vol'],
'vol_percentile': self.volatility_data['iv_percentile']
})
analysis_df = analysis_df.dropna()
if len(analysis_df) < 30:
self.root.after(0, lambda: self.log_message("Insufficient volatility Data for Analysis"))
return
# calculate regressions
slope1, intercept1, r_value1, p_value1, std_error1 = stats.linregress(
analysis_df['current_iv'], analysis_df['forward_30d_vol']
)
slope2, intercept2, r_value2, p_value2, std_error2 = stats.linregress(
analysis_df['current_iv'], analysis_df['vol_diff']
)
if slope1 != 1:
intersection_x = intercept1 / (1-slope1)
else:
intersection_x = analysis_df['current_iv'].median()
high_vol_regime = analysis_df['current_iv'] > intersection_x
low_vol_regime = analysis_df['current_iv'] <= intersection_x
if high_vol_regime.sum() > 10:
slope_high, intercept_high, r_high, p_high, std_error_high = stats.linregress(
analysis_df.loc[high_vol_regime, 'current_iv'], analysis_df.loc[high_vol_regime, 'vol_diff']
)
else:
slope_high = intercept_high = r_high = p_high = std_error_high = None
if low_vol_regime.sum() > 10:
slope_low, intercept_low, r_low, p_low, std_error_low = stats.linregress(
analysis_df.loc[low_vol_regime, 'current_iv'], analysis_df.loc[low_vol_regime, 'vol_diff']
)
else:
slope_low = intercept_low = r_low = p_low = std_error_low = None
# update UI from main thread
self.root.after(0, lambda: self.update_analysis_plots(
analysis_df, slope1, intercept1, r_value1, slope2, intercept2, r_value2,
slope_high, intercept_high, r_high, slope_low, intercept_low, r_low, intersection_x
))
except Exception as e:
self.root.after(0, lambda: self.handle_analysis_error(e))
thread = threading.Thread(target=analysis_thread, daemon=True)
thread.start()
def update_analysis_plots(self, analysis_df, slope1, intercept1, r_value1, slope2, intercept2, r_value2,
slope_high, intercept_high, r_high, slope_low, intercept_low, r_low, intersection_x):
try:
self.ax1.clear()
self.ax2.clear()
self.ax3.clear()
self.ax1.scatter(analysis_df['current_iv'], analysis_df['forward_30d_vol'], alpha=0.6, s=20)
x_range = np.linspace(analysis_df['current_iv'].min(), analysis_df['current_iv'].max(), 100)
y_pred1 = slope1 * x_range + intercept1
self.ax1.plot(x_range, y_pred1, 'r-', linewidth=2, label=f"Regression R^2 = {r_value1**2:.3f}")
min_val = min(analysis_df['current_iv'].min(), analysis_df['forward_30d_vol'].min())
max_val = max(analysis_df['current_iv'].max(), analysis_df['forward_30d_vol'].max())
self.ax1.plot([min_val, max_val], [min_val, max_val], 'k--', linewidth=1, alpha=0.7, label="y=x (No Change)")
self.ax1.set_xlabel("Current Volatility")
self.ax1.set_ylabel("30-Day Forward Average Volatility")
self.ax1.set_title(f"Forward Volatility vs Current Volatility \n y = {slope1:.3f}x + {intercept1}, R^2 = {r_value1**2:.3f}")
self.ax1.legend()
self.ax1.grid(True, alpha=0.3)
# recalculate regimes for plotting
high_vol_regime = analysis_df['current_iv'] > intersection_x
low_vol_regime = analysis_df['current_iv'] <= intersection_x
self.ax2.scatter(analysis_df.loc[high_vol_regime, 'current_iv'],
analysis_df.loc[high_vol_regime, 'vol_diff'],
alpha=0.6, s=20, color='red', label="High Vol Regime")
self.ax2.scatter(analysis_df.loc[low_vol_regime, 'current_iv'],
analysis_df.loc[low_vol_regime, 'vol_diff'],
alpha=0.6, s=20, color='green', label="Low Vol Regime")
if slope_high is not None:
x_high = analysis_df.loc[high_vol_regime, 'current_iv']
if len(x_high) > 0:
x_range_high = np.linspace(x_high.min(), x_high.max(), 100)
y_pred_high = slope_high * x_range_high + intercept_high
self.ax2.plot(x_range_high, y_pred_high, 'r-', linewidth=2,
label=f"High Vol R^2 = {r_high**2:.3f}")
if slope_low is not None:
x_low = analysis_df.loc[low_vol_regime, 'current_iv']
if len(x_low) > 0:
x_range_low = np.linspace(x_low.min(), x_low.max(), 100)
y_pred_low = slope_low * x_range_low + intercept_low
self.ax2.plot(x_range_low, y_pred_low, 'g-', linewidth=2,
label=f"Low Vol R^2 = {r_low**2:.3f}")
self.ax2.axhline(y = 0, color = 'k', linestyle = '--', linewidth=1, alpha=0.7, label="No Change (y=0)")
self.ax2.axvline(x = intersection_x, color = 'g', linestyle = ':', linewidth=1, alpha=0.7,
label=f"Regime Split (Vol={intersection_x:.3f})")
self.ax2.set_xlabel("Current Volatility")
self.ax2.set_ylabel("Vol Difference (Forward - Current)")
self.ax2.set_title("Vol Difference vs Current Vol (Regime Analysis)")
self.ax2.legend()
self.ax2.grid(True, alpha=0.3)
self.ax3.plot(self.volatility_data.index, self.volatility_data['implied_vol'],
label="Implied Volatility", linewidth=1)
vol_75th = self.volatility_data['implied_vol'].quantile(0.75)
vol_25th = self.volatility_data['implied_vol'].quantile(0.25)
self.ax3.axhline(y=vol_75th, color='red', linestyle='--', alpha=0.7, label="75th Percentile")
self.ax3.axhline(y=vol_25th, color='green', linestyle='--', alpha=0.7, label="25th Percentile")
self.ax3.axhline(y=self.volatility_data['implied_vol'].mean(), color='black', linestyle='-', alpha=0.7, label="Mean")
if self.current_implied_vol is not None:
self.ax3.scatter(self.volatility_data.index[-1], self.current_implied_vol,
color='red', s=100, zorder=5, label="Current Volatility")
self.ax3.set_xlabel("Date")
self.ax3.set_ylabel("Volatility")
self.ax3.set_title("Volatility Time Series with Regime Bands")
self.ax3.legend()
self.ax3.grid(True, alpha=0.3)
self.ax3.tick_params(axis='x', rotation=45)
self.fig.tight_layout()
self.canvas.draw()
self.log_message("Analysis complete!")
self.analyze_btn.config(state="normal")
except Exception as e:
self.handle_analysis_error(e)
def handle_analysis_error(self, error):
self.log_message(f"Analysis error: {error}")
self.analyze_btn.config(state="normal")
def main():
root = tk.Tk()
app = ImpliedVolatilityDashboard(root)
root.mainloop()
if __name__ == "__main__":
main()