# 深度实值买购和卖购组合的牛差策略 # 参考文档: Lib/Options/README.md - 策略1 from jqdata import * import pandas as pd import numpy as np import matplotlib.pyplot as plt import tqdm from datetime import datetime, timedelta import warnings warnings.filterwarnings('ignore') class DeepITMBullSpreadStrategy: """深度实值买购和卖购组合的牛差策略""" def __init__(self, underlying='510300.XSHG', start_date='2024-01-01', end_date='2025-06-30'): self.underlying = underlying # 标的ETF self.start_date = start_date self.end_date = end_date # 策略参数设置 self.params = { '一组张数': 30, '最小权利金': {'沪深300ETF': 0.03, '上证50ETF': 0.05}, '最少开仓日期': 20, # 距离到期日天数 '买购时间价值阈值': 0.0001, '卖购平仓权利金阈值': 0.0005, '合约到期移仓日期最大': 7, # 交易日 '合约到期移仓日期最小': 2, # 交易日 '加仓窗口阈值': {'沪深300ETF': 0.2, '上证50ETF': 0.1} } # 交易记录 self.positions = [] # 持仓记录 self.trades = [] # 交易记录 self.daily_pnl = [] # 每日损益 # 获取交易日历 self.trade_days = get_trade_days(start_date, end_date) def get_etf_name(self): """根据标的代码获取ETF名称""" if self.underlying == '510300.XSHG': return '沪深300ETF' elif self.underlying == '510050.XSHG': return '上证50ETF' else: return '未知ETF' def get_option_contracts(self, date, contract_type='CO', min_days_to_expire=20): """获取期权合约信息""" # 查询期权合约 q = query(opt.OPT_CONTRACT_INFO.code, opt.OPT_CONTRACT_INFO.trading_code, opt.OPT_CONTRACT_INFO.name, opt.OPT_CONTRACT_INFO.exercise_price, opt.OPT_CONTRACT_INFO.last_trade_date, opt.OPT_CONTRACT_INFO.list_date ).filter( opt.OPT_CONTRACT_INFO.underlying_symbol == self.underlying, opt.OPT_CONTRACT_INFO.contract_type == contract_type, opt.OPT_CONTRACT_INFO.list_date <= date, opt.OPT_CONTRACT_INFO.last_trade_date > date ) contracts = opt.run_query(q) print(f" 查询到{len(contracts)}个{contract_type}类型的期权合约") # 过滤距离到期日至少min_days_to_expire天的合约 valid_contracts = [] for _, contract in contracts.iterrows(): days_to_expire = (pd.to_datetime(contract['last_trade_date']) - pd.to_datetime(date)).days if days_to_expire >= min_days_to_expire: valid_contracts.append(contract) print(f" 过滤后剩余{len(valid_contracts)}个距离到期日至少{min_days_to_expire}天的合约") return pd.DataFrame(valid_contracts) if valid_contracts else pd.DataFrame() def get_option_price(self, option_code, date): """获取期权价格""" try: q = query(opt.OPT_DAILY_PRICE.close).filter( opt.OPT_DAILY_PRICE.code == option_code, opt.OPT_DAILY_PRICE.date == date ) result = opt.run_query(q) return result['close'].iloc[0] if not result.empty else None except: return None def calculate_time_value(self, option_price, intrinsic_value): """计算时间价值""" return max(0, option_price - max(0, intrinsic_value)) def select_call_option_to_buy(self, date, etf_price): """选择深度实值买购期权""" contracts = self.get_option_contracts(date, 'CO') if contracts.empty: print(f" 买购选择失败: 未找到任何认购期权合约") return None print(f" 找到{len(contracts)}个认购期权合约") # 筛选深度实值期权（时间价值 <= 阈值） suitable_contracts = [] checked_count = 0 for _, contract in contracts.iterrows(): option_price = self.get_option_price(contract['code'], date) checked_count += 1 if option_price is None: continue intrinsic_value = max(0, etf_price - contract['exercise_price']) time_value = self.calculate_time_value(option_price, intrinsic_value) if time_value <= self.params['买购时间价值阈值']: suitable_contracts.append({ 'code': contract['code'], 'exercise_price': contract['exercise_price'], 'option_price': option_price, 'time_value': time_value, 'last_trade_date': contract['last_trade_date'] }) print(f" 检查了{checked_count}个合约，找到{len(suitable_contracts)}个符合时间价值条件的买购期权") # 选择时间价值最小的 if suitable_contracts: return min(suitable_contracts, key=lambda x: x['time_value']) return None def select_call_option_to_sell(self, date, etf_price): """选择平值卖购期权""" contracts = self.get_option_contracts(date, 'CO') if contracts.empty: print(f" 卖购选择失败: 未找到任何认购期权合约") return None # 找到最接近平值的期权 etf_name = self.get_etf_name() min_premium = self.params['最小权利金'].get(etf_name, 0.03) print(f" 最小权利金要求: {min_premium}") suitable_contracts = [] checked_count = 0 for _, contract in contracts.iterrows(): option_price = self.get_option_price(contract['code'], date) checked_count += 1 if option_price is None or option_price < min_premium: continue price_diff = abs(contract['exercise_price'] - etf_price) suitable_contracts.append({ 'code': contract['code'], 'exercise_price': contract['exercise_price'], 'option_price': option_price, 'price_diff': price_diff, 'last_trade_date': contract['last_trade_date'] }) print(f" 检查了{checked_count}个合约，找到{len(suitable_contracts)}个符合权利金条件的卖购期权") # 选择最接近平值且权利金较高的 if suitable_contracts: # 先按价格差排序，再按权利金排序 suitable_contracts.sort(key=lambda x: (x['price_diff'], -x['option_price'])) return suitable_contracts[0] return None def open_bull_spread_position(self, date, etf_price, quantity=None): """开仓牛差组合""" if quantity is None: quantity = self.params['一组张数'] print(f"尝试开仓: 日期={date}, ETF价格={etf_price:.4f}") # 选择买购期权（深度实值） buy_call = self.select_call_option_to_buy(date, etf_price) if buy_call is None: print(f" 失败: 未找到合适的买购期权") return None print(f" 找到买购期权: 代码={buy_call['code']}, 行权价={buy_call['exercise_price']:.4f}, 权利金={buy_call['option_price']:.4f}") # 选择卖购期权（平值） sell_call = self.select_call_option_to_sell(date, etf_price) if sell_call is None: print(f" 失败: 未找到合适的卖购期权") return None print(f" 找到卖购期权: 代码={sell_call['code']}, 行权价={sell_call['exercise_price']:.4f}, 权利金={sell_call['option_price']:.4f}") # 确保买购行权价 < 卖购行权价 if buy_call['exercise_price'] >= sell_call['exercise_price']: print(f" 失败: 买购行权价({buy_call['exercise_price']:.4f}) >= 卖购行权价({sell_call['exercise_price']:.4f})") return None # 计算单张最大盈利和最小盈利 max_profit_per_contract = (sell_call['exercise_price'] - buy_call['exercise_price'] - buy_call['option_price'] + sell_call['option_price']) min_profit_per_contract = sell_call['option_price'] position = { 'open_date': date, 'etf_price': etf_price, 'buy_call_code': buy_call['code'], 'buy_call_strike': buy_call['exercise_price'], 'buy_call_price': buy_call['option_price'], 'sell_call_code': sell_call['code'], 'sell_call_strike': sell_call['exercise_price'], 'sell_call_price': sell_call['option_price'], 'quantity': quantity, 'max_profit_per_contract': max_profit_per_contract, 'min_profit_per_contract': min_profit_per_contract, 'max_profit_total': max_profit_per_contract * quantity * 10000, 'min_profit_total': min_profit_per_contract * quantity * 10000, 'expire_date': sell_call['last_trade_date'], 'status': 'open', 'is_additional': False # 是否为加仓 } self.positions.append(position) # 记录交易 self.trades.append({ 'date': date, 'action': 'open_bull_spread', 'details': position }) print(f" 成功开仓: 数量={quantity}张, 最大盈利={max_profit_per_contract:.4f}元/张") return position def should_close_position(self, position, date, etf_price): """判断是否应该平仓""" # 1. 检查是否接近到期 days_to_expire = (pd.to_datetime(position['expire_date']) - pd.to_datetime(date)).days if days_to_expire <= self.params['合约到期移仓日期最大']: return True, "approaching_expiry" # 2. 检查ETF大涨情况（接近最大盈利） current_max_profit = max(0, min(etf_price - position['buy_call_strike'], position['sell_call_strike'] - position['buy_call_strike'])) if current_max_profit >= position['max_profit_per_contract'] * 0.9: return True, "max_profit_reached" # 3. 检查ETF下跌情况（卖购权利金剩余低于阈值） sell_call_price = self.get_option_price(position['sell_call_code'], date) if sell_call_price is not None and sell_call_price <= self.params['卖购平仓权利金阈值']: return True, "etf_declined" return False, None def close_position(self, position, date, reason): """平仓操作""" buy_call_price = self.get_option_price(position['buy_call_code'], date) sell_call_price = self.get_option_price(position['sell_call_code'], date) if buy_call_price is None or sell_call_price is None: return None # 计算平仓收益 buy_call_pnl = (buy_call_price - position['buy_call_price']) * position['quantity'] * 10000 sell_call_pnl = (position['sell_call_price'] - sell_call_price) * position['quantity'] * 10000 total_pnl = buy_call_pnl + sell_call_pnl # 更新持仓状态 position['status'] = 'closed' position['close_date'] = date position['close_reason'] = reason position['close_buy_call_price'] = buy_call_price position['close_sell_call_price'] = sell_call_price position['realized_pnl'] = total_pnl # 记录交易 self.trades.append({ 'date': date, 'action': 'close_bull_spread', 'reason': reason, 'pnl': total_pnl, 'details': position }) return total_pnl def should_add_position(self, etf_price): """判断是否应该加仓""" # 检查是否有开仓的主仓位 main_positions = [p for p in self.positions if p['status'] == 'open' and not p['is_additional']] if not main_positions: return False # 获取最近的主仓位 latest_main = max(main_positions, key=lambda x: x['open_date']) # 检查价格下跌幅度 etf_name = self.get_etf_name() threshold = self.params['加仓窗口阈值'].get(etf_name, 0.2) if latest_main['etf_price'] - etf_price >= threshold: # 检查是否已经在这个价格水平加过仓 existing_additional = [p for p in self.positions if p['status'] == 'open' and p['is_additional'] and abs(p['etf_price'] - etf_price) < threshold * 0.5] if not existing_additional: return True return False def run_backtest(self): """运行回测""" print(f"开始回测: {self.start_date} 到 {self.end_date}") print(f"标的: {self.underlying} ({self.get_etf_name()})") # 获取ETF价格数据 etf_prices = get_price(self.underlying, self.start_date, self.end_date, fields=['close'])['close'] # 初始化 portfolio_value = [] etf_benchmark = [] with tqdm.tqdm(self.trade_days, desc="回测进度") as pbar: for date in pbar: if date not in etf_prices.index: continue etf_price = etf_prices[date] # 检查现有持仓是否需要平仓 open_positions = [p for p in self.positions if p['status'] == 'open'] for position in open_positions: should_close, reason = self.should_close_position(position, date, etf_price) if should_close: self.close_position(position, date, reason) # 检查是否需要开新仓 open_positions = [p for p in self.positions if p['status'] == 'open'] if not open_positions: # 没有持仓时开仓 self.open_bull_spread_position(date, etf_price) elif self.should_add_position(etf_price): # 加仓 additional_pos = self.open_bull_spread_position(date, etf_price) if additional_pos: additional_pos['is_additional'] = True # 计算当日组合价值 daily_pnl = self.calculate_daily_pnl(date) portfolio_value.append(daily_pnl) # ETF基准收益 if len(etf_benchmark) == 0: etf_benchmark.append(0) else: etf_return = (etf_price / etf_prices[self.trade_days[0]] - 1) * 100000 # 假设初始投资10万 etf_benchmark.append(etf_return) # 生成回测报告 self.generate_detailed_report() return portfolio_value, etf_benchmark def calculate_daily_pnl(self, date): """计算每日损益""" total_pnl = 0 for position in self.positions: if position['status'] == 'closed': if 'realized_pnl' in position: total_pnl += position['realized_pnl'] elif position['status'] == 'open': # 计算未实现损益 buy_call_price = self.get_option_price(position['buy_call_code'], date) sell_call_price = self.get_option_price(position['sell_call_code'], date) if buy_call_price is not None and sell_call_price is not None: buy_call_pnl = (buy_call_price - position['buy_call_price']) * position['quantity'] * 10000 sell_call_pnl = (position['sell_call_price'] - sell_call_price) * position['quantity'] * 10000 total_pnl += buy_call_pnl + sell_call_pnl return total_pnl def export_data_to_csv(self, filename_prefix="bull_spread_data"): """导出数据到CSV文件，用于线下分析""" # 1. 导出持仓数据 positions_data = [] for pos in self.positions: positions_data.append({ 'open_date': pos['open_date'], 'etf_price': pos['etf_price'], 'buy_call_code': pos['buy_call_code'], 'buy_call_strike': pos['buy_call_strike'], 'buy_call_price': pos['buy_call_price'], 'sell_call_code': pos['sell_call_code'], 'sell_call_strike': pos['sell_call_strike'], 'sell_call_price': pos['sell_call_price'], 'quantity': pos['quantity'], 'max_profit_total': pos['max_profit_total'], 'expire_date': pos['expire_date'], 'status': pos['status'], 'is_additional': pos['is_additional'], 'close_date': pos.get('close_date', ''), 'close_reason': pos.get('close_reason', ''), 'realized_pnl': pos.get('realized_pnl', 0) }) positions_df = pd.DataFrame(positions_data) positions_df.to_csv(f"{filename_prefix}_positions.csv", index=False, encoding='utf-8-sig') # 2. 导出交易记录 trades_data = [] for trade in self.trades: trades_data.append({ 'date': trade['date'], 'action': trade['action'], 'pnl': trade.get('pnl', 0), 'reason': trade.get('reason', ''), 'details': str(trade.get('details', '')) }) trades_df = pd.DataFrame(trades_data) trades_df.to_csv(f"{filename_prefix}_trades.csv", index=False, encoding='utf-8-sig') # 3. 导出每日损益数据 daily_data = [] etf_prices = get_price(self.underlying, self.start_date, self.end_date, fields=['close'])['close'] for date in self.trade_days: if date not in etf_prices.index: continue etf_price = etf_prices[date] # 计算买购和卖购分别的损益 buy_call_pnl = 0 sell_call_pnl = 0 total_positions = 0 for position in self.positions: if position['open_date'] <= date and (position['status'] == 'open' or position.get('close_date', date) >= date): total_positions += 1 if position['status'] == 'closed' and position.get('close_date') == date: # 已平仓的实现损益 buy_call_pnl += position.get('realized_pnl', 0) / 2 # 简化分配 sell_call_pnl += position.get('realized_pnl', 0) / 2 elif position['status'] == 'open': # 未实现损益 buy_call_price = self.get_option_price(position['buy_call_code'], date) sell_call_price = self.get_option_price(position['sell_call_code'], date) if buy_call_price is not None: buy_call_pnl += (buy_call_price - position['buy_call_price']) * position['quantity'] * 10000 if sell_call_price is not None: sell_call_pnl += (position['sell_call_price'] - sell_call_price) * position['quantity'] * 10000 # ETF基准收益 etf_return = (etf_price / etf_prices.iloc[0] - 1) * 100000 if len(etf_prices) > 0 else 0 daily_data.append({ 'date': date, 'etf_price': etf_price, 'etf_return': etf_return, 'buy_call_pnl': buy_call_pnl, 'sell_call_pnl': sell_call_pnl, 'total_pnl': buy_call_pnl + sell_call_pnl, 'total_positions': total_positions }) daily_df = pd.DataFrame(daily_data) daily_df.to_csv(f"{filename_prefix}_daily.csv", index=False, encoding='utf-8-sig') print(f"数据已导出到以下文件:") print(f"- {filename_prefix}_positions.csv (持仓数据)") print(f"- {filename_prefix}_trades.csv (交易记录)") print(f"- {filename_prefix}_daily.csv (每日损益)") return positions_df, trades_df, daily_df def generate_detailed_report(self): """生成详细的分析报告，分别分析买购和卖购收益""" print("\n" + "="*80) print("深度实值牛差策略详细回测报告") print("="*80) # 获取ETF价格数据 etf_prices = get_price(self.underlying, self.start_date, self.end_date, fields=['close'])['close'] # 计算分项收益 buy_call_returns = [] sell_call_returns = [] combined_returns = [] etf_returns = [] dates = [] initial_etf_price = etf_prices.iloc[0] if len(etf_prices) > 0 else 1 for date in self.trade_days: if date not in etf_prices.index: continue etf_price = etf_prices[date] dates.append(date) # 计算买购和卖购分别的损益 buy_call_pnl = 0 sell_call_pnl = 0 for position in self.positions: if position['open_date'] <= date: if position['status'] == 'closed': if 'realized_pnl' in position: # 简化处理：假设买购和卖购各承担一半损益 buy_call_pnl += position['realized_pnl'] / 2 sell_call_pnl += position['realized_pnl'] / 2 elif position['status'] == 'open': # 计算未实现损益 buy_call_price = self.get_option_price(position['buy_call_code'], date) sell_call_price = self.get_option_price(position['sell_call_code'], date) if buy_call_price is not None: buy_call_pnl += (buy_call_price - position['buy_call_price']) * position['quantity'] * 10000 if sell_call_price is not None: sell_call_pnl += (position['sell_call_price'] - sell_call_price) * position['quantity'] * 10000 buy_call_returns.append(buy_call_pnl) sell_call_returns.append(sell_call_pnl) combined_returns.append(buy_call_pnl + sell_call_pnl) # ETF基准收益 etf_return = (etf_price / initial_etf_price - 1) * 100000 # 假设10万本金 etf_returns.append(etf_return) # 基本统计 total_trades = len([t for t in self.trades if t['action'] == 'close_bull_spread']) winning_trades = len([t for t in self.trades if t['action'] == 'close_bull_spread' and t['pnl'] > 0]) if total_trades > 0: win_rate = winning_trades / total_trades * 100 total_pnl = sum([t['pnl'] for t in self.trades if t['action'] == 'close_bull_spread']) avg_pnl = total_pnl / total_trades else: win_rate = 0 total_pnl = 0 avg_pnl = 0 print(f"交易次数: {total_trades}") print(f"胜率: {win_rate:.2f}%") print(f"总收益: {total_pnl:.2f}元") print(f"平均每笔收益: {avg_pnl:.2f}元") # 分项收益统计和对比分析 if len(buy_call_returns) > 0: final_buy_call = buy_call_returns[-1] final_sell_call = sell_call_returns[-1] final_combined = combined_returns[-1] final_etf = etf_returns[-1] print(f"\n=== 分项收益分析 ===") print(f"买购期权收益: {final_buy_call:.2f}元") print(f"卖购期权收益: {final_sell_call:.2f}元") print(f"组合总收益: {final_combined:.2f}元") print(f"ETF基准收益: {final_etf:.2f}元") print(f"\n=== 与ETF基准对比分析 ===") print(f"1. 牛差策略 vs ETF: {final_combined:.2f} vs {final_etf:.2f} = {final_combined - final_etf:+.2f}元") print(f"2. 买购策略 vs ETF: {final_buy_call:.2f} vs {final_etf:.2f} = {final_buy_call - final_etf:+.2f}元") print(f"3. 卖购策略 vs ETF: {final_sell_call:.2f} vs {final_etf:.2f} = {final_sell_call - final_etf:+.2f}元") # 持仓统计 open_positions = len([p for p in self.positions if p['status'] == 'open']) closed_positions = len([p for p in self.positions if p['status'] == 'closed']) print(f"\n=== 持仓统计 ===") print(f"当前持仓: {open_positions}个") print(f"已平仓位: {closed_positions}个") # 绘制分项收益曲线 if len(dates) > 0: self.plot_detailed_performance(dates, buy_call_returns, sell_call_returns, combined_returns, etf_returns) self.plot_strategy_vs_etf_comparison(dates, buy_call_returns, sell_call_returns, combined_returns, etf_returns) return { 'dates': dates, 'buy_call_returns': buy_call_returns, 'sell_call_returns': sell_call_returns, 'combined_returns': combined_returns, 'etf_returns': etf_returns } def plot_detailed_performance(self, dates, buy_call_returns, sell_call_returns, combined_returns, etf_returns): """绘制详细的策略表现图，分别显示买购和卖购收益""" # 设置中文字体 plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS', 'DejaVu Sans'] plt.rcParams['axes.unicode_minus'] = False plt.figure(figsize=(15, 12)) # 第一个子图：分项收益对比 plt.subplot(3, 1, 1) plt.plot(dates, buy_call_returns, label='买购期权收益', linewidth=2, color='blue') plt.plot(dates, sell_call_returns, label='卖购期权收益', linewidth=2, color='red') plt.plot(dates, combined_returns, label='组合总收益', linewidth=2, color='green') plt.plot(dates, etf_returns, label=f'{self.get_etf_name()}基准', linewidth=2, color='orange') plt.axhline(0, color='gray', linestyle='--', alpha=0.5) plt.title('策略分项收益对比') plt.ylabel('收益(元)') plt.legend() plt.grid(True, alpha=0.3) # 第二个子图：累计收益率 plt.subplot(3, 1, 2) if len(combined_returns) > 0 and len(etf_returns) > 0: combined_cumret = [(x / 100000) * 100 for x in combined_returns] # 转换为百分比 etf_cumret = [(x / 100000) * 100 for x in etf_returns] plt.plot(dates, combined_cumret, label='策略累计收益率', linewidth=2, color='green') plt.plot(dates, etf_cumret, label='ETF累计收益率', linewidth=2, color='orange') plt.axhline(0, color='gray', linestyle='--', alpha=0.5) plt.title('累计收益率对比') plt.ylabel('收益率(%)') plt.legend() plt.grid(True, alpha=0.3) # 第三个子图：持仓数量变化 plt.subplot(3, 1, 3) position_counts = [] for date in dates: count = len([p for p in self.positions if p['open_date'] <= date and (p['status'] == 'open' or p.get('close_date', date) >= date)]) position_counts.append(count) plt.plot(dates, position_counts, label='持仓数量', linewidth=2, color='purple') plt.title('持仓数量变化') plt.ylabel('持仓数量') plt.xlabel('日期') plt.legend() plt.grid(True, alpha=0.3) plt.tight_layout() plt.show() def plot_strategy_vs_etf_comparison(self, dates, buy_call_returns, sell_call_returns, combined_returns, etf_returns): """绘制三个策略分别与ETF的对比图""" # 设置中文字体 plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS', 'DejaVu Sans'] plt.rcParams['axes.unicode_minus'] = False plt.figure(figsize=(15, 12)) # 第一个子图：牛差策略 vs ETF plt.subplot(3, 1, 1) plt.plot(dates, combined_returns, label='牛差策略收益', linewidth=2, color='green') plt.plot(dates, etf_returns, label='ETF基准收益', linewidth=2, color='orange') plt.axhline(0, color='gray', linestyle='--', alpha=0.5) plt.title('牛差策略 vs ETF基准收益对比') plt.ylabel('收益(元)') plt.legend() plt.grid(True, alpha=0.3) # 第二个子图：买购策略 vs ETF plt.subplot(3, 1, 2) plt.plot(dates, buy_call_returns, label='买购期权收益', linewidth=2, color='blue') plt.plot(dates, etf_returns, label='ETF基准收益', linewidth=2, color='orange') plt.axhline(0, color='gray', linestyle='--', alpha=0.5) plt.title('买购期权 vs ETF基准收益对比') plt.ylabel('收益(元)') plt.legend() plt.grid(True, alpha=0.3) # 第三个子图：卖购策略 vs ETF plt.subplot(3, 1, 3) plt.plot(dates, sell_call_returns, label='卖购期权收益', linewidth=2, color='red') plt.plot(dates, etf_returns, label='ETF基准收益', linewidth=2, color='orange') plt.axhline(0, color='gray', linestyle='--', alpha=0.5) plt.title('卖购期权 vs ETF基准收益对比') plt.ylabel('收益(元)') plt.xlabel('日期') plt.legend() plt.grid(True, alpha=0.3) plt.tight_layout() plt.show() def plot_performance(self, portfolio_value, etf_benchmark): """绘制策略表现图""" plt.figure(figsize=(12, 8)) dates = self.trade_days[:len(portfolio_value)] plt.subplot(2, 1, 1) plt.plot(dates, portfolio_value, label='牛差策略', linewidth=2) plt.plot(dates, etf_benchmark, label=f'{self.get_etf_name()}基准', linewidth=2) plt.title('策略收益对比') plt.ylabel('收益(元)') plt.legend() plt.grid(True, alpha=0.3) plt.subplot(2, 1, 2) # 绘制持仓数量变化 position_counts = [] for date in dates: count = len([p for p in self.positions if p['open_date'] <= date and (p['status'] == 'open' or p.get('close_date', date) >= date)]) position_counts.append(count) plt.plot(dates, position_counts, label='持仓数量', linewidth=2, color='orange') plt.title('持仓数量变化') plt.ylabel('持仓数量') plt.xlabel('日期') plt.legend() plt.grid(True, alpha=0.3) plt.tight_layout() plt.show() def print_position_details(self): """打印持仓详情""" print("\n" + "="*80) print("持仓详情") print("="*80) for i, pos in enumerate(self.positions): print(f"\n持仓 {i+1}:") print(f" 开仓日期: {pos['open_date']}") print(f" ETF价格: {pos['etf_price']:.4f}") print(f" 买购: {pos['buy_call_strike']:.2f} @ {pos['buy_call_price']:.4f}") print(f" 卖购: {pos['sell_call_strike']:.2f} @ {pos['sell_call_price']:.4f}") print(f" 数量: {pos['quantity']}张") print(f" 单张最大盈利: {pos['max_profit_per_contract']:.4f}") print(f" 总最大盈利: {pos['max_profit_total']:.2f}元") print(f" 状态: {pos['status']}") print(f" 是否加仓: {'是' if pos['is_additional'] else '否'}") if pos['status'] == 'closed': print(f" 平仓日期: {pos['close_date']}") print(f" 平仓原因: {pos['close_reason']}") print(f" 实现损益: {pos['realized_pnl']:.2f}元") class OptionsAnalyzer: """期权分析工具类""" @staticmethod def analyze_options(*options): """ 统一的期权分析方法 参数: *options: 一个或多个期权，每个期权格式为 (direction, option_type, premium, strike_price, quantity) """ if not options: raise ValueError("请至少提供一个期权") # 解析期权数据 option_list = [] all_strikes = [] for i, opt in enumerate(options): if len(opt) != 5: raise ValueError(f"期权{i+1}格式错误，应为(direction, option_type, premium, strike_price, quantity)") direction, option_type, premium, strike_price, quantity = opt option_list.append({ 'direction': direction, 'option_type': option_type, 'premium': premium, 'strike_price': strike_price, 'quantity': quantity }) all_strikes.append(strike_price) # 确定价格分析区间 min_strike = min(all_strikes) max_strike = max(all_strikes) price_min = min_strike * 0.7 price_max = max_strike * 1.3 # 生成价格序列 gap = (price_max - price_min) / 1000 prices = np.arange(price_min, price_max + gap, gap) # 计算每个期权的收益 results = {'price': prices} for i, opt in enumerate(option_list): profits = [] for price in prices: profit = OptionsAnalyzer._calculate_profit(opt, price) profits.append(profit) results[f'opt{i+1}'] = profits # 计算组合收益 if len(option_list) > 1: combined_profits = [] for j in range(len(prices)): total = sum(results[f'opt{i+1}'][j] for i in range(len(option_list))) combined_profits.append(total) results['combined'] = combined_profits # 绘制图表 OptionsAnalyzer._plot_results(results, option_list, prices) # 打印分析报告 OptionsAnalyzer._print_report(results, option_list, prices) return pd.DataFrame(results) @staticmethod def _calculate_profit(option, price): """计算单个期权在特定价格下的收益""" direction = option['direction'] option_type = option['option_type'] premium = option['premium'] strike_price = option['strike_price'] quantity = option['quantity'] if direction == 'buy' and option_type == 'call': # 买入认购 if price > strike_price: return (price - strike_price - premium) * quantity else: return -premium * quantity elif direction == 'sell' and option_type == 'call': # 卖出认购 if price > strike_price: return -(price - strike_price - premium) * quantity else: return premium * quantity elif direction == 'buy' and option_type == 'put': # 买入认沽 if price < strike_price: return (strike_price - price - premium) * quantity else: return -premium * quantity elif direction == 'sell' and option_type == 'put': # 卖出认沽 if price < strike_price: return -(strike_price - price - premium) * quantity else: return premium * quantity return 0 @staticmethod def _plot_results(results, option_list, prices): """绘制分析图表""" # 设置中文字体 plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS', 'DejaVu Sans'] plt.rcParams['axes.unicode_minus'] = False plt.figure(figsize=(14, 10)) colors = ['blue', 'green', 'orange', 'purple', 'brown'] # 绘制单个期权曲线 for i in range(len(option_list)): opt = option_list[i] opt_name = f'opt{i+1}' strategy_name = f"{opt['direction'].upper()} {opt['option_type'].upper()}" color = colors[i % len(colors)] plt.plot(prices, results[opt_name], '--', color=color, linewidth=2, alpha=0.7, label=f'{opt_name}: {strategy_name} (行权价:{opt["strike_price"]})') # 绘制组合曲线 if 'combined' in results: plt.plot(prices, results['combined'], 'r-', linewidth=3, label='组合收益') # 添加零线和行权价线 plt.axhline(0, color='gray', linestyle='-', alpha=0.5) for opt in option_list: plt.axvline(opt['strike_price'], color='gray', linestyle='--', alpha=0.3) # 找到并标注关键点 if 'combined' in results: OptionsAnalyzer._mark_key_points(results['combined'], prices, '组合') elif len(option_list) == 1: OptionsAnalyzer._mark_key_points(results['opt1'], prices, '期权') plt.xlabel('标的资产价格', fontsize=12) plt.ylabel('收益/损失', fontsize=12) if len(option_list) == 1: opt = option_list[0] title = f'{opt["direction"].upper()} {opt["option_type"].upper()} 期权分析' else: title = f'期权组合分析 ({len(option_list)}个期权)' plt.title(title, fontsize=14, weight='bold') plt.grid(True, alpha=0.3) plt.legend() plt.tight_layout() plt.show() @staticmethod def _mark_key_points(profits, prices, label_prefix): """标注关键点：盈亏平衡点、最大收益/损失边界点""" # 标注盈亏平衡点 for i in range(len(profits) - 1): if profits[i] * profits[i + 1] <= 0: # 符号改变 # 线性插值找到精确平衡点 p1, profit1 = prices[i], profits[i] p2, profit2 = prices[i + 1], profits[i + 1] if profit2 != profit1: breakeven_price = p1 - profit1 * (p2 - p1) / (profit2 - profit1) plt.plot(breakeven_price, 0, 'ro', markersize=10) plt.annotate(f'平衡点: {breakeven_price:.3f}', xy=(breakeven_price, 0), xytext=(breakeven_price + (prices.max() - prices.min()) * 0.05, max(profits) * 0.1), arrowprops=dict(arrowstyle='->', color='red'), fontsize=11, color='red', weight='bold') @staticmethod def _print_report(results, option_list, prices): """打印分析报告""" print("=" * 60) print("期权分析报告") print("=" * 60) # 期权基本信息 for i, opt in enumerate(option_list): print(f"期权{i+1}: {opt['direction'].upper()} {opt['option_type'].upper()}") print(f" 行权价: {opt['strike_price']}") print(f" 权利金: {opt['premium']}") print(f" 数量: {opt['quantity']}手") # 分析关键指标 if 'combined' in results: profits = results['combined'] print(f"\n【组合分析】") else: profits = results['opt1'] print(f"\n【单期权分析】") max_profit = max(profits) min_profit = min(profits) max_idx = profits.tolist().index(max_profit) if hasattr(profits, 'tolist') else profits.index(max_profit) min_idx = profits.tolist().index(min_profit) if hasattr(profits, 'tolist') else profits.index(min_profit) print(f"最大收益: {max_profit:.4f} (标的价格: {prices[max_idx]:.4f})") print(f"最大损失: {min_profit:.4f} (标的价格: {prices[min_idx]:.4f})") print(f"一单最大收益: {max_profit * 10000:.2f}元") print(f"一单最大亏损: {abs(min_profit) * 10000:.2f}元") # 找盈亏平衡点 breakeven_points = [] for i in range(len(profits) - 1): if profits[i] * profits[i + 1] <= 0: p1, profit1 = prices[i], profits[i] p2, profit2 = prices[i + 1], profits[i + 1] if profit2 != profit1: bp = p1 - profit1 * (p2 - p1) / (profit2 - profit1) breakeven_points.append(bp) if breakeven_points: print(f"盈亏平衡点: {[f'{bp:.4f}' for bp in breakeven_points]}") else: print("无盈亏平衡点") print("=" * 60) def test_opening_logic(): """专门测试开仓逻辑""" print("="*60) print("测试开仓逻辑") print("="*60) # 创建策略实例 strategy = DeepITMBullSpreadStrategy( underlying='510300.XSHG', # 300ETF start_date='2024-01-01', end_date='2025-06-30' ) # 获取ETF价格数据 try: etf_prices = get_price(strategy.underlying, strategy.start_date, strategy.end_date, fields=['close'])['close'] print(f"成功获取ETF价格数据，共{len(etf_prices)}个交易日") # 测试前几个交易日的开仓逻辑 test_dates = strategy.trade_days[:10] # 只测试前10个交易日 for i, date in enumerate(test_dates): if date not in etf_prices.index: continue etf_price = etf_prices[date] print(f"\n第{i+1}个交易日测试: {date}, ETF价格: {etf_price:.4f}") # 尝试开仓 position = strategy.open_bull_spread_position(date, etf_price) if position: print(f"✓ 成功开仓!") print(f" 持仓总数: {len(strategy.positions)}") break else: print(f"✗ 开仓失败") # 打印最终结果 print(f"\n测试结果:") print(f"总持仓数: {len(strategy.positions)}") print(f"总交易数: {len(strategy.trades)}") if strategy.positions: print(f"第一个持仓详情:") pos = strategy.positions[0] for key, value in pos.items(): print(f" {key}: {value}") return strategy except Exception as e: print(f"测试过程中出现错误: {e}") import traceback traceback.print_exc() return None def test_strategy(): """测试策略""" print("开始测试深度实值牛差策略...") # 创建策略实例 strategy = DeepITMBullSpreadStrategy( underlying='510300.XSHG', # 300ETF start_date='2024-01-01', end_date='2025-06-30' ) # 运行回测 try: print("正在运行回测...") strategy.run_backtest() print("正在生成详细报告...") # 生成详细报告（包含买购和卖购分项分析） detailed_results = strategy.generate_detailed_report() print("正在导出数据...") # 导出数据到CSV positions_df, trades_df, daily_df = strategy.export_data_to_csv("bull_spread_300etf") print("正在打印持仓详情...") # 打印详细持仓信息 strategy.print_position_details() return strategy, detailed_results except Exception as e: print(f"回测过程中出现错误: {e}") import traceback traceback.print_exc() return None, None def analyze_bull_spread_example(): """分析牛差策略示例""" print("\n" + "="*60) print("牛差策略期权组合分析示例") print("="*60) # 假设当前300ETF价格为4.0 underlying_price = 4.0 # 根据策略逻辑选择期权 buy_strike = underlying_price * 0.90 # 深度实值买购 sell_strike = underlying_price * 1.00 # 平值卖购 # 估算权利金（简化） buy_premium = 0.44 # 深度实值期权，主要是内在价值 sell_premium = 0.20 # 平值期权，主要是时间价值 print(f"当前标的价格: {underlying_price}") print(f"买购期权: 行权价 {buy_strike:.2f}, 权利金 {buy_premium:.4f}") print(f"卖购期权: 行权价 {sell_strike:.2f}, 权利金 {sell_premium:.4f}") # 使用期权分析工具 analyzer = OptionsAnalyzer() result = analyzer.analyze_options( ('buy', 'call', buy_premium, buy_strike, 1), ('sell', 'call', sell_premium, sell_strike, 1) ) return result def compare_with_etf_holding(): """三个策略分别与ETF持有收益对比""" print("\n" + "="*80) print("三个策略分别与ETF持有收益对比") print("="*80) initial_price = 4.0 initial_investment = 100000 # 牛差策略参数 buy_strike = initial_price * 0.90 sell_strike = initial_price * 1.00 buy_premium = 0.44 sell_premium = 0.20 # 计算可开仓张数 net_premium = buy_premium - sell_premium contracts = int(initial_investment / (net_premium * 10000)) print(f"初始投资: {initial_investment:,.0f}元") print(f"初始ETF价格: {initial_price:.2f}") print(f"牛差组合: 买购{buy_strike:.2f}@{buy_premium:.4f}, 卖购{sell_strike:.2f}@{sell_premium:.4f}") print(f"可开仓张数: {contracts}张") # 模拟不同价格变化下的收益 price_changes = np.arange(-20, 21, 5) results = [] for change in price_changes: new_price = initial_price * (1 + change / 100) # ETF持有收益 etf_return = (new_price - initial_price) / initial_price * initial_investment # 分别计算三个策略的收益 buy_call_value = max(0, new_price - buy_strike) sell_call_value = max(0, new_price - sell_strike) # 1. 买购期权收益 buy_call_pnl = (buy_call_value - buy_premium) * contracts * 10000 # 2. 卖购期权收益 sell_call_pnl = (sell_premium - sell_call_value) * contracts * 10000 # 3. 牛差组合收益 bull_spread_return = buy_call_pnl + sell_call_pnl results.append({ 'price_change': change, 'new_price': new_price, 'etf_return': etf_return, 'buy_call_return': buy_call_pnl, 'sell_call_return': sell_call_pnl, 'bull_spread_return': bull_spread_return, 'bull_vs_etf': bull_spread_return - etf_return, 'buy_vs_etf': buy_call_pnl - etf_return, 'sell_vs_etf': sell_call_pnl - etf_return }) # 创建对比表格 df = pd.DataFrame(results) print(f"\n=== 详细收益对比表 ===") print("价格变化(%) | 新价格 | ETF收益 | 买购收益 | 卖购收益 | 牛差收益") print("-" * 75) for _, row in df.iterrows(): print(f"{row['price_change']:>8.0f}% | {row['new_price']:>6.2f} | " f"{row['etf_return']:>7.0f} | {row['buy_call_return']:>8.0f} | " f"{row['sell_call_return']:>8.0f} | {row['bull_spread_return']:>8.0f}") print(f"\n=== 与ETF基准的差异对比 ===") print("价格变化(%) | 牛差-ETF | 买购-ETF | 卖购-ETF") print("-" * 50) for _, row in df.iterrows(): print(f"{row['price_change']:>8.0f}% | {row['bull_vs_etf']:>8.0f} | " f"{row['buy_vs_etf']:>8.0f} | {row['sell_vs_etf']:>8.0f}") # 绘制三个对比图 # 设置中文字体 plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS', 'DejaVu Sans'] plt.rcParams['axes.unicode_minus'] = False plt.figure(figsize=(15, 12)) # 第一个子图：牛差策略 vs ETF plt.subplot(3, 1, 1) plt.plot(df['price_change'], df['etf_return'], 'o-', linewidth=2, label='ETF持有', color='orange') plt.plot(df['price_change'], df['bull_spread_return'], 's-', linewidth=2, label='牛差策略', color='green') plt.axhline(0, color='gray', linestyle='--', alpha=0.5) plt.xlabel('价格变化 (%)') plt.ylabel('收益 (元)') plt.title('牛差策略 vs ETF持有收益对比') plt.legend() plt.grid(True, alpha=0.3) # 第二个子图：买购策略 vs ETF plt.subplot(3, 1, 2) plt.plot(df['price_change'], df['etf_return'], 'o-', linewidth=2, label='ETF持有', color='orange') plt.plot(df['price_change'], df['buy_call_return'], '^-', linewidth=2, label='买购期权', color='blue') plt.axhline(0, color='gray', linestyle='--', alpha=0.5) plt.xlabel('价格变化 (%)') plt.ylabel('收益 (元)') plt.title('买购期权 vs ETF持有收益对比') plt.legend() plt.grid(True, alpha=0.3) # 第三个子图：卖购策略 vs ETF plt.subplot(3, 1, 3) plt.plot(df['price_change'], df['etf_return'], 'o-', linewidth=2, label='ETF持有', color='orange') plt.plot(df['price_change'], df['sell_call_return'], 'v-', linewidth=2, label='卖购期权', color='red') plt.axhline(0, color='gray', linestyle='--', alpha=0.5) plt.xlabel('价格变化 (%)') plt.ylabel('收益 (元)') plt.title('卖购期权 vs ETF持有收益对比') plt.legend() plt.grid(True, alpha=0.3) plt.tight_layout() plt.show() return df if __name__ == "__main__": print("深度实值牛差策略综合测试工具") print("="*50) # 0. 首先测试开仓逻辑 print("\n0. 测试开仓逻辑:") test_strategy_instance = test_opening_logic() if test_strategy_instance and len(test_strategy_instance.positions) > 0: print("✓ 开仓逻辑测试通过，继续完整回测") # 1. 运行策略回测 print("\n1. 运行策略回测:") strategy, detailed_results = test_strategy() # 2. 分析期权组合 print("\n2. 期权组合分析:") option_analysis = analyze_bull_spread_example() # 3. 与ETF持有对比 print("\n3. 与ETF持有收益对比:") comparison_results = compare_with_etf_holding() print(f"\n测试完成！数据已导出到CSV文件，可用于进一步分析。") else: print("✗ 开仓逻辑测试失败，请检查期权数据和策略参数") print("建议检查以下几点:") print("1. 期权合约数据是否可用") print("2. 策略参数设置是否合理") print("3. 时间价值阈值是否过于严格")