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

3. 跨式策略（Straddle）

跨式策略：同时买入（或卖出）相同行权价和到期日的看涨和看跌期权。

class StraddleStrategy:    """    跨式策略    """    def __init__(self, initial_capital=100000):        self.initial_capital = initial_capital        self.results = []    def long_straddle(self, S, K, T, r, sigma, stock_quantity=10000):        """        买入跨式：预期大幅波动        """        # 计算期权价格        call_price = calculate_bs_price(S, K, T, r, sigma, 'call')        put_price = calculate_bs_price(S, K, T, r, sigma, 'put')        total_premium = call_price + put_price        # 初始现金流：买入两个期权        initial_cashflow = -total_premium * stock_quantity        # 计算损益        expiration_prices = np.linspace(S * 0.7, S * 1.3, 100)        pnl = []        for St in expiration_prices:            # 看涨期权损益            if St > K:                call_pnl = (St - K - call_price) * stock_quantity            else:                call_pnl = -call_price * stock_quantity            # 看跌期权损益            if St < K:                put_pnl = (K - St - put_price) * stock_quantity            else:                put_pnl = -put_price * stock_quantity            total_pnl = call_pnl + put_pnl            pnl.append(total_pnl)        # 盈亏平衡点        breakeven_up = K + total_premium        breakeven_down = K - total_premium        # 最大亏损（当价格等于行权价时）        max_loss = -total_premium * stock_quantity        # 最大盈利（理论上无限）        max_profit_up = float('inf')  # 价格上涨时        max_profit_down = (K - total_premium) * stock_quantity  # 价格跌到0时        return {            'expiration_prices': expiration_prices,            'pnl': pnl,            'breakeven_up': breakeven_up,            'breakeven_down': breakeven_down,            'max_loss': max_loss,            'max_profit_up': max_profit_up,            'max_profit_down': max_profit_down,            'total_premium': total_premium,            'initial_cashflow': initial_cashflow        }    def short_straddle(self, S, K, T, r, sigma, stock_quantity=10000):        """        卖出跨式：预期小幅波动        """        # 计算期权价格        call_price = calculate_bs_price(S, K, T, r, sigma, 'call')        put_price = calculate_bs_price(S, K, T, r, sigma, 'put')        total_premium = call_price + put_price        # 初始现金流：卖出两个期权        initial_cashflow = total_premium * stock_quantity        # 计算损益        expiration_prices = np.linspace(S * 0.7, S * 1.3, 100)        pnl = []        for St in expiration_prices:            # 看涨期权损益            if St > K:                call_pnl = (call_price - (St - K)) * stock_quantity            else:                call_pnl = call_price * stock_quantity            # 看跌期权损益            if St < K:                put_pnl = (put_price - (K - St)) * stock_quantity            else:                put_pnl = put_price * stock_quantity            total_pnl = call_pnl + put_pnl            pnl.append(total_pnl)        # 盈亏平衡点        breakeven_up = K + total_premium        breakeven_down = K - total_premium        # 最大盈利（当价格等于行权价时）        max_profit = total_premium * stock_quantity        # 最大亏损（理论上无限）        max_loss_up = float('inf')  # 价格上涨时        max_loss_down = (total_premium - K) * stock_quantity  # 价格跌到0时        return {            'expiration_prices': expiration_prices,            'pnl': pnl,            'breakeven_up': breakeven_up,            'breakeven_down': breakeven_down,            'max_profit': max_profit,            'max_loss_up': max_loss_up,            'max_loss_down': max_loss_down,            'total_premium': total_premium,            'initial_cashflow': initial_cashflow        }# 测试跨式策略S = 2.5K = 2.5  # 平值期权T = 30/365r = 0.02sigma = 0.2straddle = StraddleStrategy()# 买入跨式long_straddle_result = straddle.long_straddle(S, K, T, r, sigma)# 卖出跨式short_straddle_result = straddle.short_straddle(S, K, T, r, sigma)print(f"\n跨式策略分析:")print(f"股票价格: {S:.2f}")print(f"行权价: {K:.2f}")print(f"期权总权利金: {long_straddle_result['total_premium']:.4f}")print(f"\n买入跨式:")print(f"上盈亏平衡点: {long_straddle_result['breakeven_up']:.4f}")print(f"下盈亏平衡点: {long_straddle_result['breakeven_down']:.4f}")print(f"最大亏损: {long_straddle_result['max_loss']:.2f}")print(f"\n卖出跨式:")print(f"上盈亏平衡点: {short_straddle_result['breakeven_up']:.4f}")print(f"下盈亏平衡点: {short_straddle_result['breakeven_down']:.4f}")print(f"最大盈利: {short_straddle_result['max_profit']:.2f}")# 可视化跨式策略fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))# 买入跨式ax1.plot(long_straddle_result['expiration_prices'], long_straddle_result['pnl'],         linewidth=2, label='买入跨式', color='green')ax1.axhline(y=0, color='black', linestyle='-', alpha=0.3)ax1.axvline(x=long_straddle_result['breakeven_up'], color='red', linestyle='--',            alpha=0.5, label=f'上盈亏点: {long_straddle_result["breakeven_up"]:.2f}')ax1.axvline(x=long_straddle_result['breakeven_down'], color='red', linestyle='--',            alpha=0.5, label=f'下盈亏点: {long_straddle_result["breakeven_down"]:.2f}')ax1.axvline(x=S, color='blue', linestyle='--', alpha=0.5, label=f'当前价格: {S:.2f}')ax1.set_xlabel('到期时股票价格')ax1.set_ylabel('损益')ax1.set_title('买入跨式策略损益图', fontsize=14)ax1.legend()ax1.grid(True, alpha=0.3)# 卖出跨式ax2.plot(short_straddle_result['expiration_prices'], short_straddle_result['pnl'],         linewidth=2, label='卖出跨式', color='red')ax2.axhline(y=0, color='black', linestyle='-', alpha=0.3)ax2.axvline(x=short_straddle_result['breakeven_up'], color='green', linestyle='--',            alpha=0.5, label=f'上盈亏点: {short_straddle_result["breakeven_up"]:.2f}')ax2.axvline(x=short_straddle_result['breakeven_down'], color='green', linestyle='--',            alpha=0.5, label=f'下盈亏点: {short_straddle_result["breakeven_down"]:.2f}')ax2.axvline(x=S, color='blue', linestyle='--', alpha=0.5, label=f'当前价格: {S:.2f}')ax2.set_xlabel('到期时股票价格')ax2.set_ylabel('损益')ax2.set_title('卖出跨式策略损益图', fontsize=14)ax2.legend()ax2.grid(True, alpha=0.3)plt.tight_layout()plt.show()

四、波动率交易策略

1. Delta中性策略

Delta中性策略旨在消除方向性风险，专注于交易波动率。

class DeltaNeutralStrategy:    """    Delta中性策略    通过期权和标的资产的组合，使总体Delta接近0    """    def __init__(self, initial_capital=100000):        self.initial_capital = initial_capital        self.results = []    def create_delta_neutral_portfolio(self, S, K, T, r, sigma, option_type='straddle'):        """        创建Delta中性组合        """        if option_type == 'straddle':            # 使用跨式组合            call_price = calculate_bs_price(S, K, T, r, sigma, 'call')            put_price = calculate_bs_price(S, K, T, r, sigma, 'put')            # 计算希腊字母            call_greeks = calculate_greeks(S, K, T, r, sigma, 'call')            put_greeks = calculate_greeks(S, K, T, r, sigma, 'put')            # 跨式组合的Delta            portfolio_delta = call_greeks['Delta'] + put_greeks['Delta']            # 需要多少标的资产来对冲Delta            # 目标Delta = 0，所以需要对冲掉portfolio_delta            hedge_quantity = -portfolio_delta  # 负号表示反向操作            return {                'call_price': call_price,                'put_price': put_price,                'call_delta': call_greeks['Delta'],                'put_delta': put_greeks['Delta'],                'portfolio_delta': portfolio_delta,                'hedge_quantity': hedge_quantity,                'gamma': call_greeks['Gamma'] + put_greeks['Gamma'],                'vega': call_greeks['Vega'] + put_greeks['Vega'],                'theta': call_greeks['Theta'] + put_greeks['Theta']            }        elif option_type == 'call':            # 使用看涨期权            call_price = calculate_bs_price(S, K, T, r, sigma, 'call')            call_greeks = calculate_greeks(S, K, T, r, sigma, 'call')            # 对冲Delta            hedge_quantity = -call_greeks['Delta']            return {                'call_price': call_price,                'call_delta': call_greeks['Delta'],                'portfolio_delta': call_greeks['Delta'],                'hedge_quantity': hedge_quantity,                'gamma': call_greeks['Gamma'],                'vega': call_greeks['Vega'],                'theta': call_greeks['Theta']            }        else:            raise ValueError("option_type必须是'straddle'或'call'")    def gamma_scalping(self, S, K, T, r, sigma, initial_capital=100000, days_to_expire=30):        """        Gamma Scalping策略：通过频繁调整Delta中性来从Gamma中获利        """        # 初始参数        daily_returns = []        portfolio_values = [initial_capital]        delta_history = []        gamma_history = []        # 模拟每天的价格变化        np.random.seed(42)        daily_changes = np.random.randn(days_to_expire) * sigma / np.sqrt(252)        prices = [S]        for i in range(days_to_expire):            # 计算新的价格            new_price = prices[-1] * (1 + daily_changes[i])            prices.append(new_price)            # 重新计算希腊字母            T_remaining = (days_to_expire - i) / 365            # 计算组合希腊字母            portfolio_stats = self.create_delta_neutral_portfolio(                new_price, K, T_remaining, r, sigma, 'straddle'            )            # 记录            delta_history.append(portfolio_stats['portfolio_delta'])            gamma_history.append(portfolio_stats['gamma'])            # 模拟损益（简化）            # Gamma Scalping的收益来自价格变动和Gamma的关系            # 损益 ≈ 0.5 * Gamma * (价格变动)^2            price_change = daily_changes[i] * new_price            gamma_pnl = 0.5 * portfolio_stats['gamma'] * (price_change**2)            # Theta损失（时间价值衰减）            theta_loss = portfolio_stats['theta'] / 365            # 每日净损益            daily_pnl = gamma_pnl + theta_loss            daily_returns.append(daily_pnl)            # 更新组合价值            new_value = portfolio_values[-1] + daily_pnl            portfolio_values.append(new_value)        return {            'prices': prices,            'portfolio_values': portfolio_values,            'daily_returns': daily_returns,            'delta_history': delta_history,            'gamma_history': gamma_history,            'total_return': (portfolio_values[-1] - initial_capital) / initial_capital        }# 测试Delta中性策略S = 2.5K = 2.5T = 30/365r = 0.02sigma = 0.2delta_neutral = DeltaNeutralStrategy()# 创建Delta中性组合portfolio_stats = delta_neutral.create_delta_neutral_portfolio(S, K, T, r, sigma, 'straddle')print(f"\nDelta中性组合分析:")print(f"看涨期权价格: {portfolio_stats['call_price']:.4f}")print(f"看跌期权价格: {portfolio_stats['put_price']:.4f}")print(f"看涨Delta: {portfolio_stats['call_delta']:.4f}")print(f"看跌Delta: {portfolio_stats['put_delta']:.4f}")print(f"组合Delta: {portfolio_stats['portfolio_delta']:.4f}")print(f"对冲所需标的资产数量: {portfolio_stats['hedge_quantity']:.4f}")print(f"组合Gamma: {portfolio_stats['gamma']:.6f}")print(f"组合Vega: {portfolio_stats['vega']:.6f}")print(f"组合Theta: {portfolio_stats['theta']:.6f}")# Gamma Scalping回测gamma_scalping_result = delta_neutral.gamma_scalping(S, K, T, r, sigma, 100000, 30)print(f"\nGamma Scalping策略回测:")print(f"初始资金: 100,000")print(f"最终资金: {gamma_scalping_result['portfolio_values'][-1]:.2f}")print(f"总收益率: {gamma_scalping_result['total_return']:.2%}")print(f"日均收益率: {np.mean(gamma_scalping_result['daily_returns']):.2f}")print(f"收益波动率: {np.std(gamma_scalping_result['daily_returns']):.2f}")# 可视化Gamma Scalping结果fig, axes = plt.subplots(2, 2, figsize=(14, 10))# 价格走势axes[0, 0].plot(gamma_scalping_result['prices'], linewidth=2)axes[0, 0].set_xlabel('交易日')axes[0, 0].set_ylabel('标的价格')axes[0, 0].set_title('标的价格走势', fontsize=12)axes[0, 0].grid(True, alpha=0.3)# 组合价值axes[0, 1].plot(gamma_scalping_result['portfolio_values'], linewidth=2, color='green')axes[0, 1].set_xlabel('交易日')axes[0, 1].set_ylabel('组合价值')axes[0, 1].set_title('Gamma Scalping组合价值', fontsize=12)axes[0, 1].grid(True, alpha=0.3)# Delta变化axes[1, 0].plot(gamma_scalping_result['delta_history'], linewidth=2, color='red')axes[1, 0].axhline(y=0, color='black', linestyle='--', alpha=0.5)axes[1, 0].set_xlabel('交易日')axes[1, 0].set_ylabel('组合Delta')axes[1, 0].set_title('组合Delta变化', fontsize=12)axes[1, 0].grid(True, alpha=0.3)# Gamma变化axes[1, 1].plot(gamma_scalping_result['gamma_history'], linewidth=2, color='purple')axes[1, 1].set_xlabel('交易日')axes[1, 1].set_ylabel('组合Gamma')axes[1, 1].set_title('组合Gamma变化', fontsize=12)axes[1, 1].grid(True, alpha=0.3)plt.tight_layout()plt.show()