def pseudoVIX(symb ='^SPX', risk_free_rate=0.035): ticker = yf.Ticker(symb) # prezzo corrente del sottostante. underlying_price = ticker.history(period='1d').Close.iloc[-1] def get_options_data(expiration_date): """ expiration_date: data di scadenza (None = la più vicina disponibile) Ritorna: DataFrame con le opzioni calls e puts filtrate,per una specifica scadenza """ expirations = ticker.options # Ottieni tutte le scadenze disponibili if not expirations: raise ValueError(f"Nessuna opzione disponibile per {symb}") # Se non specificata, usa la scadenza più vicina if expiration_date is None: expiration_date = expirations[0] elif expiration_date not in expirations: raise ValueError(f"Scadenza {expiration_date} non disponibile. Scadenze: {expirations}") chain = ticker.option_chain(expiration_date) # Scarica la catena di opzioni # Unisce calls e puts in un unico DataFrame con flag per il tipo calls = chain.calls.copy() calls['type'] = 'call' puts = chain.puts.copy() puts['type'] = 'put' options = pd.concat([calls, puts], ignore_index=True) # Calcola il prezzo medio (bid+ask)/2 options['mid_price'] = (options['bid'] + options['ask']) / 2 # Filtra opzioni illiquide (bid=0) options = options[options['bid'] > 0] # giorni alla scadenza in anni today = datetime.now().date() expiry = datetime.strptime(expiration_date, '%Y-%m-%d').date() time_to_expiry = (expiry - today).days / 365.25 return options, time_to_expiry def forward(): """ Calcola il prezzo forward usando la put-call parity. Trova lo strike dove call e put hanno prezzo più vicino. """ # Separa calls e puts calls = options[options.type == 'call'] puts = options[options.type == 'put'] # Trova strike comuni common_strikes = set(calls['strike']).intersection(set(puts['strike'])) if not common_strikes: # nel caso non ci siano strike comuni # se le opzioni sono illiquide o nei giorni di chiusura delle contrattazioni # per semplicità possiamo restituire NaN o generare un errore più specifico # restituire NaN si propagherà e probabilmente causerà l'errore nel calcolo del VIX # che è un comportamento accettabile se i dati sono insufficienti. forward_price, forward_strike = np.nan, np.nan return np.nan # Per ogni strike comune, calcola |call - put| diff_prices = [] for strike in common_strikes: call_price = calls[calls['strike'] == strike]['mid_price'].values[0] put_price = puts[puts['strike'] == strike]['mid_price'].values[0] diff = abs(call_price - put_price) diff_prices.append((strike, diff, call_price, put_price)) # Trova lo strike con minima differenza best_strike, _, call_best, put_best = min(diff_prices, key=lambda x: x[1]) # Calcola forward: F = strike + e^(rT) * (call - put) T, r = time_to_expiry, risk_free_rate F = best_strike + np.exp(r * T) * (call_best - put_best) return F, best_strike def implied_variance_single_expiry(): """ Implementa la formula: σ² = (2/T) * Σ (ΔK/K²) * e^(rT) * Q(K) - (1/T) * (F/K₀ - 1)² """ forward_price, forward_strike = forward() if np.isnan(forward_price): return np.nan options_otm = [] # Raccoglie le opzioni OTM for _, option in options.iterrows(): strike = option['strike'] mid = option['mid_price'] opt_type = option['type'] # Selezione OTM if strike < forward_price and opt_type == 'put': options_otm.append((strike, mid)) elif strike > forward_price and opt_type == 'call': options_otm.append((strike, mid)) # Per lo strike K₀, prendi la media (solo se disponibile) elif abs(strike - forward_price) < 1e-6: # Prendi put e call per questo strike puts_at_k0 = options[(options['strike'] == strike) & (options['type'] == 'put')] calls_at_k0 = options[(options['strike'] == strike) & (options['type'] == 'call')] if not puts_at_k0.empty and not calls_at_k0.empty: avg_price = (puts_at_k0['mid_price'].values[0] + calls_at_k0['mid_price'].values[0]) / 2 options_otm.append((strike, avg_price)) # Ordina per strike options_otm.sort(key=lambda x: x[0]) strikes = np.array([x[0] for x in options_otm]) prices = np.array([x[1] for x in options_otm]) # Calcola ΔK per ogni strike delta_K = np.zeros_like(strikes) for i in range(len(strikes)): if i == 0: # Primo strike: differenza tra primo e secondo delta_K[i] = strikes[1] - strikes[0] elif i == len(strikes) - 1: # Ultimo strike: differenza tra ultimo e penultimo delta_K[i] = strikes[-1] - strikes[-2] else: # Strike intermedi: metà distanza tra precedente e successivo delta_K[i] = (strikes[i+1] - strikes[i-1]) / 2 # Calcola la varianza implicita T, r, F, K0 = time_to_expiry, risk_free_rate, forward_price, forward_strike # Termine di somma: Σ (ΔK/K²) * e^(rT) * Q(K) sum_term = 0 for i in range(len(strikes)): K = strikes[i] Q = prices[i] dK = delta_K[i] weight = (dK / (K * K)) * np.exp(r * T) * Q sum_term += weight # Correzione: (1/T) * (F/K₀ - 1)² correction = (1/T) * ((F / K0) - 1)**2 # Varianza implicita implied_variance = (2/T) * sum_term - correction return implied_variance days_to_target = 30 """ Calcola l'indice di volatilità stile VIX a 30 giorni. Usa interpolazione tra due scadenze se disponibili. """ expirations = ticker.options if len(expirations) < 2: # Solo una scadenza disponibile options, time_to_expiry = get_options_data(expirations[0]) σ1 = implied_variance_single_expiry() if np.isnan(σ1): return np.nan N1 = time_to_expiry * 365 # giorni vix = np.sqrt(σ1) * 100 return vix # Seleziona l'ultima scadenza <= 23 giorni exp1 = [exp for exp in expirations if (datetime.strptime(exp, '%Y-%m-%d').date() - datetime.now().date()).days >= 23][-1] # Seleziona la prima scadenza >= 37 giorni exp2 = [exp for exp in expirations if (datetime.strptime(exp, '%Y-%m-%d').date() - datetime.now().date()).days >= 37][0] # Calcola varianza implicita per exp1 options, time_to_expiry = get_options_data(exp1) σ1 = implied_variance_single_expiry() N1 = time_to_expiry * 365 # giorni # Calcola varianza implicita per exp2 options, time_to_expiry = get_options_data(exp2) σ2 = implied_variance_single_expiry() N2 = time_to_expiry * 365 # giorni # Se non è stato possibile calcolare una delle due varianze, restituisci NaN if np.isnan(σ1) or np.isnan(σ2): return np.nan # Interpolazione a 30 giorni N_target = days_to_target # Verifica che target sia tra le due scadenze if not (N1 < N_target < N2): print(f"Attenzione: target {N_target} giorni non è tra {N1} e {N2}") # Usa la scadenza più vicina if abs(N1 - N_target) < abs(N2 - N_target): return np.sqrt(σ1) * 100 else: return np.sqrt(σ2) * 100 # Pesi per interpolazione w1 = N1 * (N2 - N_target) / (N2 - N1) w2 = N2 * (N_target - N1) / (N2 - N1) # Interpolazione delle varianze var_target = (σ1 * w1 + σ2 * w2) / N_target # Indice stile VIX vix_like = np.sqrt(var_target) * 100 return vix_like pseudoVIX(symb='^SPX'), yf.Ticker('^VIX').history().Close.iloc[-1]