期权量化策略(二):用Python玩转金融衍生品

def calculate_implied_volatility(S, K, T, r, market_price, option_type='call', max_iter=100, tol=1e-6):    """    使用牛顿法计算隐含波动率    参数:    S: 标的资产价格    K: 行权价    T: 到期时间（年）    r: 无风险利率    market_price: 期权市场价格    option_type: 'call'或'put'    """    # 初始猜测    sigma = 0.2    for i in range(max_iter):        # 计算当前sigma下的理论价格        theoretical_price = calculate_bs_price(S, K, T, r, sigma, option_type)        # 计算vega        d1 = (np.log(S / K) + (r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))        vega = S * stats.norm.pdf(d1) * np.sqrt(T)        # 如果vega太小，避免除零错误        if vega < tol:            break        # 牛顿法更新sigma        price_diff = theoretical_price - market_price        sigma = sigma - price_diff / vega        # 确保sigma为正        sigma = max(sigma, 0.001)        # 检查收敛        if abs(price_diff) < tol:            break    return sigma# 测试隐含波动率计算market_call_price = 0.08  # 市场观察到的看涨期权价格iv_call = calculate_implied_volatility(S, K, T, r, market_call_price, 'call')print(f"\n隐含波动率计算:")print(f"市场看涨期权价格: {market_call_price:.4f}")print(f"计算得到的隐含波动率: {iv_call:.4f} ({iv_call*100:.1f}%)")# 验证：用计算出的IV重新计算价格validated_price = calculate_bs_price(S, K, T, r, iv_call, 'call')print(f"验证价格: {validated_price:.4f}")print(f"价格差异: {abs(validated_price - market_call_price):.6f}")

class CoveredCallStrategy:    """    备兑开仓策略    持有标的资产 + 卖出看涨期权    """    def __init__(self, initial_capital=100000):        self.initial_capital = initial_capital        self.results = []    def execute(self, S, K, T, r, sigma, call_premium, stock_quantity=10000):        """        执行备兑开仓策略        """        # 初始头寸        stock_cost = S * stock_quantity        option_premium = call_premium * stock_quantity        # 初始现金流：买入股票，卖出期权        initial_cashflow = -stock_cost + option_premium        # 计算不同到期价格下的损益        expiration_prices = np.linspace(S * 0.7, S * 1.3, 100)        pnl = []        for St in expiration_prices:            # 股票损益            stock_pnl = (St - S) * stock_quantity            # 期权损益（卖出看涨期权的损益）            # 如果到期价格高于行权价，期权被行权            if St > K:                option_pnl = (call_premium - (St - K)) * stock_quantity            else:                option_pnl = call_premium * stock_quantity            # 总损益            total_pnl = stock_pnl + option_pnl            pnl.append(total_pnl)        # 计算盈亏平衡点        # 股票成本 - 期权权利金 = 到期时股票价值        # S - call_premium = 盈亏平衡价格        breakeven = S - call_premium        # 最大盈利（当价格在行权价或以下时）        max_profit = (K - S + call_premium) * stock_quantity        # 最大亏损（当价格跌到0时）        max_loss = (-S + call_premium) * stock_quantity        return {            'expiration_prices': expiration_prices,            'pnl': pnl,            'breakeven': breakeven,            'max_profit': max_profit,            'max_loss': max_loss,            'initial_cashflow': initial_cashflow        }    def backtest(self, stock_prices, option_params, rebalance_freq=30):        """        回测试兑开仓策略        """        capital = self.initial_capital        capital_history = [capital]        for i in range(0, len(stock_prices) - rebalance_freq, rebalance_freq):            current_price = stock_prices[i]            # 选择行权价（例如，当前价格的105%）            K = current_price * 1.05            # 计算期权理论价格            T = rebalance_freq / 365  # 假设每次持有30天            r = 0.02            sigma = 0.2            call_premium = calculate_bs_price(current_price, K, T, r, sigma, 'call')            # 计算可以买卖的股票数量            stock_quantity = capital // current_price            # 执行策略            result = self.execute(current_price, K, T, r, sigma, call_premium, stock_quantity)            # 模拟到期损益（简化：使用下一个调仓日的价格）            expiration_price = stock_prices[i + rebalance_freq]            if expiration_price > K:                # 期权被行权，以K价格卖出股票                stock_pnl = (K - current_price) * stock_quantity                option_pnl = call_premium * stock_quantity            else:                # 期权未被行权                stock_pnl = (expiration_price - current_price) * stock_quantity                option_pnl = call_premium * stock_quantity            total_pnl = stock_pnl + option_pnl            capital += total_pnl            capital_history.append(capital)            self.results.append({                'period': i,                'stock_price': current_price,                'strike': K,                'call_premium': call_premium,                'expiration_price': expiration_price,                'pnl': total_pnl,                'capital': capital            })        return pd.DataFrame(self.results), capital_history# 测试备兑开仓策略S = 2.5K = 2.6  # 行权价略高于当前价T = 30/365r = 0.02sigma = 0.2call_premium = calculate_bs_price(S, K, T, r, sigma, 'call')covered_call = CoveredCallStrategy(initial_capital=100000)single_trade_result = covered_call.execute(S, K, T, r, sigma, call_premium)print(f"\n备兑开仓单次交易分析:")print(f"股票价格: {S:.2f}")print(f"行权价: {K:.2f}")print(f"看涨期权权利金: {call_premium:.4f}")print(f"盈亏平衡点: {single_trade_result['breakeven']:.4f}")print(f"最大盈利: {single_trade_result['max_profit']:.2f}")print(f"最大亏损: {single_trade_result['max_loss']:.2f}")# 可视化损益图plt.figure(figsize=(10, 6))plt.plot(single_trade_result['expiration_prices'], single_trade_result['pnl'],          linewidth=2, label='备兑开仓损益')plt.axhline(y=0, color='black', linestyle='-', alpha=0.3)plt.axvline(x=single_trade_result['breakeven'], color='red', linestyle='--',            alpha=0.5, label=f'盈亏平衡点: {single_trade_result["breakeven"]:.2f}')plt.axvline(x=S, color='green', linestyle='--', alpha=0.5, label=f'当前价格: {S:.2f}')plt.xlabel('到期时股票价格')plt.ylabel('损益')plt.title('备兑开仓策略损益图', fontsize=14)plt.legend()plt.grid(True, alpha=0.3)plt.tight_layout()plt.show()

class ProtectivePutStrategy:    """    保护性看跌策略    持有标的资产 + 买入看跌期权    """    def __init__(self, initial_capital=100000):        self.initial_capital = initial_capital        self.results = []    def execute(self, S, K, T, r, sigma, put_premium, stock_quantity=10000):        """        执行保护性看跌策略        """        # 初始头寸        stock_cost = S * stock_quantity        option_cost = put_premium * stock_quantity        # 初始现金流：买入股票，买入期权        initial_cashflow = -stock_cost - option_cost        # 计算不同到期价格下的损益        expiration_prices = np.linspace(S * 0.7, S * 1.3, 100)        pnl = []        for St in expiration_prices:            # 股票损益            stock_pnl = (St - S) * stock_quantity            # 期权损益（买入看跌期权的损益）            if St < K:                option_pnl = (K - St - put_premium) * stock_quantity            else:                option_pnl = -put_premium * stock_quantity            # 总损益            total_pnl = stock_pnl + option_pnl            pnl.append(total_pnl)        # 计算盈亏平衡点        breakeven = S + put_premium        # 最大盈利（理论上无限）        max_profit = float('inf')  # 当股票上涨时        # 最大亏损（当价格跌到0时）        max_loss = (-S - put_premium + K) * stock_quantity        return {            'expiration_prices': expiration_prices,            'pnl': pnl,            'breakeven': breakeven,            'max_profit': max_profit,            'max_loss': max_loss,            'initial_cashflow': initial_cashflow        }    def backtest(self, stock_prices, option_params, rebalance_freq=30):        """        回测保护性看跌策略        """        capital = self.initial_capital        capital_history = [capital]        for i in range(0, len(stock_prices) - rebalance_freq, rebalance_freq):            current_price = stock_prices[i]            # 选择行权价（例如，当前价格的95%）            K = current_price * 0.95            # 计算期权理论价格            T = rebalance_freq / 365            r = 0.02            sigma = 0.2            put_premium = calculate_bs_price(current_price, K, T, r, sigma, 'put')            # 计算可以买卖的股票数量（考虑期权成本）            max_stocks = capital // (current_price + put_premium)            stock_quantity = max_stocks            # 执行策略            result = self.execute(current_price, K, T, r, sigma, put_premium, stock_quantity)            # 模拟到期损益            expiration_price = stock_prices[i + rebalance_freq]            # 股票损益            stock_pnl = (expiration_price - current_price) * stock_quantity            # 期权损益            if expiration_price < K:                option_pnl = (K - expiration_price - put_premium) * stock_quantity            else:                option_pnl = -put_premium * stock_quantity            total_pnl = stock_pnl + option_pnl            capital += total_pnl            capital_history.append(capital)            self.results.append({                'period': i,                'stock_price': current_price,                'strike': K,                'put_premium': put_premium,                'expiration_price': expiration_price,                'pnl': total_pnl,                'capital': capital            })        return pd.DataFrame(self.results), capital_history# 测试保护性看跌策略S = 2.5K = 2.4  # 行权价略低于当前价T = 30/365r = 0.02sigma = 0.2put_premium = calculate_bs_price(S, K, T, r, sigma, 'put')protective_put = ProtectivePutStrategy(initial_capital=100000)single_trade_result_put = protective_put.execute(S, K, T, r, sigma, put_premium)print(f"\n保护性看跌单次交易分析:")print(f"股票价格: {S:.2f}")print(f"行权价: {K:.2f}")print(f"看跌期权权利金: {put_premium:.4f}")print(f"盈亏平衡点: {single_trade_result_put['breakeven']:.4f}")print(f"最大盈利: 理论上无限")print(f"最大亏损: {single_trade_result_put['max_loss']:.2f}")# 可视化损益图plt.figure(figsize=(10, 6))plt.plot(single_trade_result_put['expiration_prices'], single_trade_result_put['pnl'],          linewidth=2, label='保护性看跌损益', color='orange')plt.axhline(y=0, color='black', linestyle='-', alpha=0.3)plt.axvline(x=single_trade_result_put['breakeven'], color='red', linestyle='--',            alpha=0.5, label=f'盈亏平衡点: {single_trade_result_put["breakeven"]:.2f}')plt.axvline(x=S, color='green', linestyle='--', alpha=0.5, label=f'当前价格: {S:.2f}')plt.axvline(x=K, color='blue', linestyle='--', alpha=0.5, label=f'行权价: {K:.2f}')plt.xlabel('到期时股票价格')plt.ylabel('损益')plt.title('保护性看跌策略损益图', fontsize=14)plt.legend()plt.grid(True, alpha=0.3)plt.tight_layout()plt.show()# 对比两种策略plt.figure(figsize=(10, 6))plt.plot(single_trade_result['expiration_prices'], single_trade_result['pnl'],          linewidth=2, label='备兑开仓')plt.plot(single_trade_result_put['expiration_prices'], single_trade_result_put['pnl'],          linewidth=2, label='保护性看跌')plt.axhline(y=0, color='black', linestyle='-', alpha=0.3)plt.axvline(x=S, color='green', linestyle='--', alpha=0.5, label=f'当前价格: {S:.2f}')plt.xlabel('到期时股票价格')plt.ylabel('损益')plt.title('备兑开仓 vs 保护性看跌策略对比', fontsize=14)plt.legend()plt.grid(True, alpha=0.3)plt.tight_layout()plt.show()