import numpy as npimport pandas as pdimport matplotlib.pyplot as pltimport seaborn as snsfrom scipy import statsfrom scipy.optimize import minimizeimport warningswarnings.filterwarnings('ignore')# 设置中文字体plt.rcParams['font.sans-serif'] = ['SimHei']plt.rcParams['axes.unicode_minus'] = False# 初始化tushare proimport tushare as tsts.set_token('你的token')pro = ts.pro_api()def get_underlying_data(symbol='510050.SH', start_date='20240101', end_date='20240601'): """ 获取标的资产数据（50ETF） """ df = pro.fund_daily(ts_code=symbol, start_date=start_date, end_date=end_date) df['trade_date'] = pd.to_datetime(df['trade_date']) df = df.sort_values('trade_date') df.set_index('trade_date', inplace=True) # 计算收益率 df['returns'] = df['close'].pct_change() df['log_returns'] = np.log(df['close'] / df['close'].shift(1)) return dfdef get_option_data(underlying='510050.SH', trade_date='20240601'): """ 获取特定日期的期权合约数据 注意：期权数据需要专业权限，这里使用模拟数据 """ # 模拟期权数据 np.random.seed(42) # 模拟标的资产价格 S0 = 2.5 # 50ETF当前价格 # 生成不同行权价和到期日的期权 strike_prices = np.arange(2.3, 2.8, 0.05) # 行权价从2.3到2.8，步长0.05 expiration_days = [7, 14, 30, 60, 90] # 不同到期日 option_data = [] for strike in strike_prices: for days in expiration_days: # 计算理论价格（使用Black-Scholes模型简化版） # 实际应用中应该从市场获取真实报价 # 假设波动率曲面 moneyness = strike / S0 if moneyness < 0.95: iv = 0.25 # 实值看跌/虚值看涨，波动率高 elif moneyness > 1.05: iv = 0.25 # 实值看涨/虚值看跌，波动率高 else: iv = 0.20 # 平值期权，波动率较低 # 计算理论价格 call_price = calculate_bs_price(S0, strike, days/365, 0.02, iv, 'call') put_price = calculate_bs_price(S0, strike, days/365, 0.02, iv, 'put') option_data.append({ 'trade_date': trade_date, 'underlying': underlying, 'strike': strike, 'expiration_days': days, 'call_price': call_price, 'put_price': put_price, 'implied_vol': iv, 'moneyness': moneyness }) return pd.DataFrame(option_data)# 获取标的资产数据underlying_data = get_underlying_data('510050.SH', '20240101', '20240601')print("标的资产数据示例:")print(underlying_data[['open', 'high', 'low', 'close', 'vol']].head())# 获取期权数据示例option_data = get_option_data('510050.SH', '20240601')print(f"\n期权数据示例 (共{len(option_data)}个合约):")print(option_data.head())
def calculate_bs_price(S, K, T, r, sigma, option_type='call'): """ 计算Black-Scholes期权价格 参数: S: 标的资产当前价格 K: 行权价 T: 到期时间（年） r: 无风险利率 sigma: 波动率 option_type: 'call'或'put' """ from scipy.stats import norm # 计算d1和d2 d1 = (np.log(S / K) + (r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T)) d2 = d1 - sigma * np.sqrt(T) # 计算期权价格 if option_type == 'call': price = S * norm.cdf(d1) - K * np.exp(-r * T) * norm.cdf(d2) elif option_type == 'put': price = K * np.exp(-r * T) * norm.cdf(-d2) - S * norm.cdf(-d1) else: raise ValueError("option_type必须是'call'或'put'") return pricedef calculate_greeks(S, K, T, r, sigma, option_type='call'): """ 计算期权希腊字母 """ from scipy.stats import norm d1 = (np.log(S / K) + (r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T)) d2 = d1 - sigma * np.sqrt(T) # Delta if option_type == 'call': delta = norm.cdf(d1) else: # put delta = norm.cdf(d1) - 1 # Gamma（看涨看跌相同） gamma = norm.pdf(d1) / (S * sigma * np.sqrt(T)) # Theta if option_type == 'call': theta = - (S * norm.pdf(d1) * sigma) / (2 * np.sqrt(T)) - r * K * np.exp(-r * T) * norm.cdf(d2) else: # put theta = - (S * norm.pdf(d1) * sigma) / (2 * np.sqrt(T)) + r * K * np.exp(-r * T) * norm.cdf(-d2) # Vega（看涨看跌相同） vega = S * norm.pdf(d1) * np.sqrt(T) # Rho if option_type == 'call': rho = K * T * np.exp(-r * T) * norm.cdf(d2) else: # put rho = -K * T * np.exp(-r * T) * norm.cdf(-d2) return { 'Delta': delta, 'Gamma': gamma, 'Theta': theta, 'Vega': vega, 'Rho': rho }# 测试BS模型S = 2.5 # 标的资产价格K = 2.5 # 行权价T = 30/365 # 30天到期r = 0.02 # 无风险利率2%sigma = 0.2 # 波动率20%call_price = calculate_bs_price(S, K, T, r, sigma, 'call')put_price = calculate_bs_price(S, K, T, r, sigma, 'put')print(f"\nBlack-Scholes模型测试:")print(f"标的资产价格: {S}")print(f"行权价: {K}")print(f"到期时间: {T:.3f}年 ({int(T*365)}天)")print(f"无风险利率: {r:.1%}")print(f"波动率: {sigma:.1%}")print(f"看涨期权价格: {call_price:.4f}")print(f"看跌期权价格: {put_price:.4f}")# 计算希腊字母call_greeks = calculate_greeks(S, K, T, r, sigma, 'call')put_greeks = calculate_greeks(S, K, T, r, sigma, 'put')print("\n看涨期权希腊字母:")for greek, value in call_greeks.items(): print(f" {greek}: {value:.6f}")print("\n看跌期权希腊字母:")for greek, value in put_greeks.items(): print(f" {greek}: {value:.6f}")# 可视化期权价格与希腊字母def visualize_option_pricing(S, K_range, T, r, sigma): """ 可视化期权价格与希腊字母 """ fig, axes = plt.subplots(2, 3, figsize=(15, 10)) # 计算不同行权价下的价格和希腊字母 call_prices = [] put_prices = [] deltas_call = [] deltas_put = [] gammas = [] thetas_call = [] thetas_put = [] vegas = [] for K in K_range: call_prices.append(calculate_bs_price(S, K, T, r, sigma, 'call')) put_prices.append(calculate_bs_price(S, K, T, r, sigma, 'put')) greeks_call = calculate_greeks(S, K, T, r, sigma, 'call') greeks_put = calculate_greeks(S, K, T, r, sigma, 'put') deltas_call.append(greeks_call['Delta']) deltas_put.append(greeks_put['Delta']) gammas.append(greeks_call['Gamma']) # Gamma相同 thetas_call.append(greeks_call['Theta']) thetas_put.append(greeks_put['Theta']) vegas.append(greeks_call['Vega']) # Vega相同 # 期权价格 axes[0, 0].plot(K_range, call_prices, label='看涨期权', linewidth=2) axes[0, 0].plot(K_range, put_prices, label='看跌期权', linewidth=2) axes[0, 0].axvline(x=S, color='red', linestyle='--', alpha=0.5, label='当前价格') axes[0, 0].set_xlabel('行权价') axes[0, 0].set_ylabel('期权价格') axes[0, 0].set_title('期权价格 vs 行权价') axes[0, 0].legend() axes[0, 0].grid(True, alpha=0.3) # Delta axes[0, 1].plot(K_range, deltas_call, label='看涨Delta', linewidth=2) axes[0, 1].plot(K_range, deltas_put, label='看跌Delta', linewidth=2) axes[0, 1].axvline(x=S, color='red', linestyle='--', alpha=0.5) axes[0, 1].set_xlabel('行权价') axes[0, 1].set_ylabel('Delta') axes[0, 1].set_title('Delta vs 行权价') axes[0, 1].legend() axes[0, 1].grid(True, alpha=0.3) # Gamma axes[0, 2].plot(K_range, gammas, label='Gamma', linewidth=2, color='green') axes[0, 2].axvline(x=S, color='red', linestyle='--', alpha=0.5) axes[0, 2].set_xlabel('行权价') axes[0, 2].set_ylabel('Gamma') axes[0, 2].set_title('Gamma vs 行权价') axes[0, 2].legend() axes[0, 2].grid(True, alpha=0.3) # Theta axes[1, 0].plot(K_range, thetas_call, label='看涨Theta', linewidth=2) axes[1, 0].plot(K_range, thetas_put, label='看跌Theta', linewidth=2) axes[1, 0].axvline(x=S, color='red', linestyle='--', alpha=0.5) axes[1, 0].set_xlabel('行权价') axes[1, 0].set_ylabel('Theta') axes[1, 0].set_title('Theta vs 行权价') axes[1, 0].legend() axes[1, 0].grid(True, alpha=0.3) # Vega axes[1, 1].plot(K_range, vegas, label='Vega', linewidth=2, color='purple') axes[1, 1].axvline(x=S, color='red', linestyle='--', alpha=0.5) axes[1, 1].set_xlabel('行权价') axes[1, 1].set_ylabel('Vega') axes[1, 1].set_title('Vega vs 行权价') axes[1, 1].legend() axes[1, 1].grid(True, alpha=0.3) # 隐含波动率微笑（简化） axes[1, 2].plot(K_range/S, [sigma]*len(K_range), linewidth=2, color='orange') axes[1, 2].axvline(x=1, color='red', linestyle='--', alpha=0.5, label='平值') axes[1, 2].set_xlabel('行权价/标的价格') axes[1, 2].set_ylabel('隐含波动率') axes[1, 2].set_title('波动率微笑（简化）') axes[1, 2].legend() axes[1, 2].grid(True, alpha=0.3) plt.tight_layout() plt.show()# 生成行权价范围K_range = np.linspace(2.0, 3.0, 50)visualize_option_pricing(S, K_range, T, r, sigma)
