import pandas as pd import numpy as np import akshare as ak import warnings from datetime import datetime, timedelta warnings.filterwarnings('ignore') # ==================== 数据获取模块 ==================== class IntradayDataFetcher: """30分钟K线数据获取类""" def __init__(self): self.symbol = "399673" # 创业板50指数 def fetch_30min_data(self, start_date=None, end_date=None) -> pd.DataFrame: """获取指定时间范围的30分钟K线数据""" try: if start_date is None: start_date = datetime.now() - timedelta(days=60) if end_date is None: end_date = datetime.now() print(f"正在获取创业板50指数的30分钟K线数据...") print(f"时间范围: {start_date.strftime('%Y-%m-%d')} 至 {end_date.strftime('%Y-%m-%d')}") # 使用数据源连接方式获取更多数据 # 首先尝试获取分钟级数据 try: data = ak.index_zh_a_hist_min_em(symbol=self.symbol, period="30") except Exception as e: print(f"获取30分钟数据失败: {e}") # 如果30分钟数据获取失败，尝试获取日线数据作为备选 print("尝试获取日线数据作为备选方案...") data = ak.index_zh_a_hist_em(symbol=self.symbol) if data.empty: raise ValueError("获取的数据为空") # 重命名列 data.rename(columns={ '时间': 'DateTime', '开盘': 'Open', '收盘': 'Close', '最高': 'High', '最低': 'Low', '成交量': 'Volume', '成交额': 'Amount', '振幅': 'Amplitude', '涨跌幅': 'Change_Pct', '涨跌额': 'Change_Amount', '换手率': 'Turnover', '日期': 'DateTime' # 备用字段名 }, inplace=True) # 设置时间索引 data['DateTime'] = pd.to_datetime(data['DateTime']) data.set_index('DateTime', inplace=True) data.sort_index(inplace=True) # 筛选指定时间范围的数据（使用宽松的开始时间，确保有预热数据） buffer_start = start_date - timedelta(days=60) # 增加60天缓冲 filtered_data = data[(data.index >= buffer_start) & (data.index <= end_date)].copy() if filtered_data.empty: print(f"警告：指定时间范围没有数据，使用所有可用数据") filtered_data = data.copy() # 检查数据量 print(f"获取数据总量: {len(data)}条") print(f"筛选后数据量: {len(filtered_data)}条") if len(filtered_data) < 20: # 放宽最低要求 raise ValueError(f"数据量严重不足：只获取到{len(filtered_data)}条数据，无法进行有效回测") # 计算基础指标 filtered_data['Returns'] = filtered_data['Close'].pct_change() filtered_data['High_Low_Pct'] = (filtered_data['High'] - filtered_data['Low']) / filtered_data['Close'].shift(1) filtered_data['Close_Open_Pct'] = (filtered_data['Close'] - filtered_data['Open']) / filtered_data['Open'] # 处理缺失值 filtered_data.fillna(method='ffill', inplace=True) filtered_data.dropna(inplace=True) print(f"最终可用数据: {len(filtered_data)}条") print(f"数据范围: {filtered_data.index[0]} 到 {filtered_data.index[-1]}") return filtered_data except Exception as e: print(f"获取数据时出错: {str(e)}") raise def calculate_intraday_indicators(self, data: pd.DataFrame) -> pd.DataFrame: """计算30分钟技术指标""" print("正在计算30分钟技术指标...") df = data.copy() # 短期移动平均线 df['MA6'] = df['Close'].rolling(window=6).mean() # 3小时 df['MA12'] = df['Close'].rolling(window=12).mean() # 6小时 df['MA24'] = df['Close'].rolling(window=24).mean() # 12小时(一天) # RSI delta = df['Close'].diff() gain = (delta.where(delta > 0, 0)).rolling(window=14).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean() rs = gain / loss df['RSI'] = 100 - (100 / (1 + rs)) # 布林带 df['BB_middle'] = df['Close'].rolling(window=20).mean() bb_std = df['Close'].rolling(window=20).std() df['BB_upper'] = df['BB_middle'] + (bb_std * 2) df['BB_lower'] = df['BB_middle'] - (bb_std * 2) df['BB_width'] = (df['BB_upper'] - df['BB_lower']) / df['BB_middle'] # MACD exp1 = df['Close'].ewm(span=12, adjust=False).mean() exp2 = df['Close'].ewm(span=26, adjust=False).mean() df['MACD'] = exp1 - exp2 df['MACD_signal'] = df['MACD'].ewm(span=9, adjust=False).mean() df['MACD_hist'] = df['MACD'] - df['MACD_signal'] # KDJ low_9 = df['Low'].rolling(window=9).min() high_9 = df['High'].rolling(window=9).max() rsv = (df['Close'] - low_9) / (high_9 - low_9) * 100 df['K'] = rsv.ewm(com=2, adjust=False).mean() df['D'] = df['K'].ewm(com=2, adjust=False).mean() df['J'] = 3 * df['K'] - 2 * df['D'] # ATR high_low = df['High'] - df['Low'] high_close = abs(df['High'] - df['Close'].shift()) low_close = abs(df['Low'] - df['Close'].shift()) true_range = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1) df['ATR'] = true_range.rolling(window=14).mean() df['ATR_Pct'] = df['ATR'] / df['Close'] # 动量指标 df['Momentum'] = df['Close'] / df['Close'].shift(4) - 1 # 2小时动量 # 成交量变化 df['Volume_MA'] = df['Volume'].rolling(window=12).mean() df['Volume_Ratio'] = df['Volume'] / df['Volume_MA'] # 价格动量 df['Price_Momentum'] = (df['Close'] - df['Close'].shift(6)) / df['Close'].shift(6) print("技术指标计算完成") return df # ==================== 翻转信号生成器 ==================== class ReversalSignalGenerator: """日内翻转信号生成器""" def __init__(self): self.signal_count = 0 def generate_reversal_signals(self, data: pd.DataFrame) -> pd.DataFrame: """生成日内翻转信号""" print("正在生成日内翻转信号...") signals = [] df = data.copy() for i in range(24, len(df)): # 至少需要12小时(24个30分钟)的历史数据 current_bar = df.iloc[i] current_time = df.index[i] # 跳过不适合交易的时间段(如午休时间等) hour = current_time.hour if hour < 9 or hour > 15: # 只在交易时间内 continue # 生成信号 signal = { 'DateTime': current_time, 'Open': current_bar['Open'], 'High': current_bar['High'], 'Low': current_bar['Low'], 'Close': current_bar['Close'], 'Volume': current_bar['Volume'], 'RSI': current_bar['RSI'], 'MACD': current_bar['MACD'], 'MACD_hist': current_bar['MACD_hist'], 'K': current_bar['K'], 'D': current_bar['D'], 'J': current_bar['J'], 'ATR_Pct': current_bar['ATR_Pct'], 'Volume_Ratio': current_bar['Volume_Ratio'], 'Price_Momentum': current_bar['Price_Momentum'], 'Close_Open_Pct': current_bar['Close_Open_Pct'] } # 计算各种翻转信号 reversal_score = 0 reversal_signals = [] # 1. RSI超卖翻转 if current_bar['RSI'] < 30: reversal_score += 2 reversal_signals.append("RSI超卖") elif current_bar['RSI'] < 35: reversal_score += 1 reversal_signals.append("RSI偏弱") # 2. KDJ超卖翻转 if current_bar['K'] < 20 and current_bar['D'] < 20: reversal_score += 2 reversal_signals.append("KDJ超卖") elif current_bar['J'] < 0: reversal_score += 2 reversal_signals.append("KDJ极端超卖") # 3. MACD金叉 if current_bar['MACD_hist'] > 0 and df.iloc[i-1]['MACD_hist'] <= 0: reversal_score += 2 reversal_signals.append("MACD金叉") elif current_bar['MACD_hist'] > df.iloc[i-1]['MACD_hist']: reversal_score += 1 reversal_signals.append("MACD改善") # 4. 价格触及布林带下轨 bb_width = current_bar['BB_width'] if current_bar['Close'] <= current_bar['BB_lower'] * 1.005: reversal_score += 2 reversal_signals.append("触及下轨") elif current_bar['Close'] <= current_bar['BB_lower'] * 1.01: reversal_score += 1 reversal_signals.append("接近下轨") # 5. 连续下跌后的反转 recent_returns = df.iloc[i-6:i]['Returns'] if recent_returns.min() < -0.015: # 最近2小时内有超过1.5%的下跌 consecutive_decline = sum(recent_returns < 0) if consecutive_decline >= 4: # 连续4个周期下跌 reversal_score += 2 reversal_signals.append("连续下跌反转") # 6. 价格动量反转 if current_bar['Price_Momentum'] < -0.02: # 3小时下跌超过2% reversal_score += 1 reversal_signals.append("动量超卖") # 7. 成交量配合 if current_bar['Volume_Ratio'] > 1.2: # 放量 reversal_score += 1 reversal_signals.append("放量配合") # 8. 当日开盘价格关系 daily_high = df[df.index.date == current_time.date()]['High'].max() daily_low = df[df.index.date == current_time.date()]['Low'].min() daily_range = daily_high - daily_low if daily_range > 0: position_in_day = (current_bar['Close'] - daily_low) / daily_range if position_in_day < 0.3: # 在当日低位区域 reversal_score += 1 reversal_signals.append("日内低位") # 设置信号 signal['Reversal_Score'] = reversal_score signal['Reversal_Signals'] = ', '.join(reversal_signals) if reversal_signals else '' # 生成买入信号(阈值降低以增加交易频率) if reversal_score >= 4: signal['Signal'] = 1 signal['Signal_Type'] = '做多翻转' self.signal_count += 1 else: signal['Signal'] = 0 signal['Signal_Type'] = '' signals.append(signal) signals_df = pd.DataFrame(signals) signals_df.set_index('DateTime', inplace=True) print(f"信号生成完成，共产生{self.signal_count}个翻转信号") print(f"信号密度: {self.signal_count/len(signals_df)*100:.2f}%") return signals_df # ==================== 日内交易执行器 ==================== class IntradayReversalExecutor: """日内翻转交易执行器""" def __init__(self, initial_capital=1000000): self.initial_capital = initial_capital self.params = { 'commission_rate': 0.0001, # 万分之一 'slippage_rate': 0.0, # 无滑点 'position_size_pct': 1.0, # 每次开仓100%仓位（满仓） 'stop_loss_pct': 0.008, # 0.8%止损 'take_profit_pct': 0.015, # 1.5%止盈 'max_hold_bars': 16, # 最多持有8小时(16个30分钟) 'min_signal_strength': 4 # 最小信号强度 } def execute_intraday_trades(self, signals_df: pd.DataFrame) -> tuple: """执行日内翻转交易""" print("正在执行日内翻转交易...") df = signals_df.copy() # 初始化 trades = [] capital = self.initial_capital position = 0 entry_price = 0 entry_time = None holding_bars = 0 entry_signals = '' # 添加资金列 df = df.copy() df['capital'] = capital df['position'] = 0 df['net_value'] = capital for i in range(len(df)): current_time = df.index[i] current_bar = df.iloc[i] price = current_bar['Close'] # 更新当前净值 if position > 0: current_value = capital + position * price df.iloc[i, df.columns.get_loc('net_value')] = current_value else: df.iloc[i, df.columns.get_loc('net_value')] = capital # 开仓逻辑 if position == 0 and current_bar['Signal'] == 1: # 开仓 position_size = int((capital * self.params['position_size_pct']) / price) if position_size > 0: cost = position_size * price * (1 + self.params['commission_rate'] + self.params['slippage_rate']) if cost <= capital: position = position_size entry_price = price entry_time = current_time entry_signals = current_bar.get('Reversal_Signals', '') holding_bars = 0 capital -= cost df.iloc[i, df.columns.get_loc('position')] = position # 平仓逻辑 elif position > 0: holding_bars += 1 # 计算止损止盈价格 stop_loss = entry_price * (1 - self.params['stop_loss_pct']) take_profit = entry_price * (1 + self.params['take_profit_pct']) exit_signal = False exit_reason = '' exit_price = price # 止损 if price <= stop_loss: exit_signal = True exit_reason = "止损" exit_price = stop_loss # 止盈 elif price >= take_profit: exit_signal = True exit_reason = "止盈" exit_price = take_profit # 最大持仓时间 elif holding_bars >= self.params['max_hold_bars']: exit_signal = True exit_reason = "时间止损" # 翻转信号消失 elif current_bar['RSI'] > 70: # RSI超买 exit_signal = True exit_reason = "RSI超买平仓" # 执行平仓 if exit_signal: # 计算盈亏 - 修复：包含开仓和平仓的总成本 gross_pnl = (exit_price - entry_price) * position # 开仓成本（已经在开仓时扣除） open_cost = position * entry_price * (self.params['commission_rate'] + self.params['slippage_rate']) # 平仓成本 close_revenue = position * exit_price close_cost = close_revenue * (self.params['commission_rate'] + self.params['slippage_rate']) # 净盈亏 = 价差收益 - 开仓成本 - 平仓成本 pnl = gross_pnl - open_cost - close_cost # 更新资金 capital += close_revenue - close_cost # 记录交易 trade = { '买入时间': entry_time, '卖出时间': current_time, '买入价格': entry_price, '卖出价格': exit_price, '仓位': position, '盈亏金额': pnl, '盈亏百分比': (exit_price - entry_price) / entry_price * 100, '退出原因': exit_reason, '持仓周期数': holding_bars, '持仓小时数': holding_bars * 0.5, '入场信号': entry_signals, '卖出时资金': capital, '开仓市值': position * entry_price } trades.append(trade) # 重置 position = 0 entry_price = 0 entry_time = None holding_bars = 0 # 更新资金 df.iloc[i, df.columns.get_loc('capital')] = capital df.iloc[i, df.columns.get_loc('position')] = position # 强制平仓剩余持仓 - 修复：包含开仓和平仓的总成本 if position > 0: final_price = df.iloc[-1]['Close'] # 计算总盈亏 gross_pnl = (final_price - entry_price) * position open_cost = position * entry_price * (self.params['commission_rate'] + self.params['slippage_rate']) close_revenue = position * final_price close_cost = close_revenue * (self.params['commission_rate'] + self.params['slippage_rate']) pnl = gross_pnl - open_cost - close_cost capital += close_revenue - close_cost trade = { '买入时间': entry_time, '卖出时间': df.index[-1], '买入价格': entry_price, '卖出价格': final_price, '仓位': position, '盈亏金额': pnl, '盈亏百分比': (final_price - entry_price) / entry_price * 100, '退出原因': '强制平仓', '持仓周期数': holding_bars, '持仓小时数': holding_bars * 0.5, '入场信号': entry_signals, '卖出时资金': capital, '开仓市值': position * entry_price } trades.append(trade) trades_df = pd.DataFrame(trades) if len(trades_df) > 0: trades_df['买入时间'] = pd.to_datetime(trades_df['买入时间']) trades_df['卖出时间'] = pd.to_datetime(trades_df['卖出时间']) trades_df = trades_df.sort_values('买入时间') print(f"交易执行完成，共{len(trades_df)}笔交易") return df, trades_df # ==================== 验证分析模块 ==================== def validate_intraday_results(results_df, trades_df, initial_capital): """验证日内交易结果""" print("\n" + "=" * 80) print("日内翻转交易结果验证") print("=" * 80) print(f"\n【基础数据验证】") final_capital = results_df['net_value'].iloc[-1] total_return = (final_capital - initial_capital) / initial_capital * 100 print(f"初始资金: {initial_capital:,.2f}元") print(f"最终资金: {final_capital:,.2f}元") print(f"总收益率: {total_return:.2f}%") print(f"交易次数: {len(trades_df)}笔") if len(trades_df) > 0: print(f"\n【交易统计】") win_trades = trades_df[trades_df['盈亏金额'] > 0] lose_trades = trades_df[trades_df['盈亏金额'] < 0] print(f"盈利交易: {len(win_trades)}笔 ({len(win_trades)/len(trades_df)*100:.1f}%)") print(f"亏损交易: {len(lose_trades)}笔 ({len(lose_trades)/len(trades_df)*100:.1f}%)") print(f"平均持仓时间: {trades_df['持仓小时数'].mean():.1f}小时") print(f"平均收益率: {trades_df['盈亏百分比'].mean():.2f}%") # 按退出原因统计 print(f"\n【退出原因统计】") for reason, count in trades_df['退出原因'].value_counts().items(): percentage = count / len(trades_df) * 100 reason_pnl = trades_df[trades_df['退出原因'] == reason]['盈亏金额'].sum() print(f" {reason}: {count}次 ({percentage:.1f}%) - 总盈亏: {reason_pnl:+,.2f}元") # ==================== 主程序 ==================== def main(): """主程序 - 运行30分钟日内翻转策略""" print("=" * 80) print("创业板50 30分钟日内翻转策略") print("=" * 80) # 策略参数 # 时间配置（调整为akshare数据可用的最近时间范围） BACKTEST_START_DATE = "2026-01-05" # 回测开始日期（调整为最近可用数据） BACKTEST_END_DATE = "2026-01-19" # 回测结束日期（使用当前日期） PREWARMP_DAYS = 10 # 指标预热期天数（调整为10天） INITIAL_CAPITAL = 100000 # 转换日期格式 start_date = datetime.strptime(BACKTEST_START_DATE, "%Y-%m-%d") end_date = datetime.strptime(BACKTEST_END_DATE, "%Y-%m-%d") # 计算数据获取开始时间（回测开始时间 - 预热期） data_start_date = start_date - timedelta(days=PREWARMP_DAYS) print(f"\n策略参数:") print(f" 回测期间: {BACKTEST_START_DATE} 至 {BACKTEST_END_DATE}") print(f" 数据获取期间: {data_start_date.strftime('%Y-%m-%d')} 至 {BACKTEST_END_DATE}") print(f" 指标预热期: {PREWARMP_DAYS}天") print(f" K线周期: 30分钟") print(f" 初始资金: {INITIAL_CAPITAL:,}元") print(f" 标的指数: 创业板50 (399673)") try: # Phase 1: 数据获取 print(f"\n【Phase 1: 30分钟数据获取】") fetcher = IntradayDataFetcher() # 获取包含预热期的完整数据 full_data = fetcher.fetch_30min_data(start_date=data_start_date, end_date=end_date) full_data = fetcher.calculate_intraday_indicators(full_data) # 筛选回测期间的数据 original_len = len(full_data) backtest_data = full_data[(full_data.index >= start_date) & (full_data.index <= end_date)].copy() print(f"筛选回测数据: {original_len} -> {len(backtest_data)} 条") print(f"回测数据范围: {backtest_data.index[0]} 到 {backtest_data.index[-1]}") # Phase 2: 信号生成 print(f"\n【Phase 2: 翻转信号生成】") signal_gen = ReversalSignalGenerator() signals_df = signal_gen.generate_reversal_signals(backtest_data) # Phase 3: 交易执行 print(f"\n【Phase 3: 日内交易执行】") executor = IntradayReversalExecutor(initial_capital=INITIAL_CAPITAL) results_df, trades_df = executor.execute_intraday_trades(signals_df) # Phase 4: 验证分析 print(f"\n【Phase 4: 结果验证与分析】") validate_intraday_results(results_df, trades_df, INITIAL_CAPITAL) # Phase 5: 导出数据 if len(trades_df) > 0: print(f"\n【Phase 5: 导出交易数据】") # 确保时间戳格式精确到分钟 trades_df['买入时间'] = pd.to_datetime(trades_df['买入时间']).dt.strftime('%Y-%m-%d %H:%M:%S') trades_df['卖出时间'] = pd.to_datetime(trades_df['卖出时间']).dt.strftime('%Y-%m-%d %H:%M:%S') # 生成带时间戳的文件名 timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') output_file = f'cyb50_30min_intraday_reversal_trades_{timestamp}.csv' trades_df.to_csv(output_file, index=False, encoding='utf-8-sig') print(f"交易记录已保存到: {output_file}") print(f"时间戳格式: YYYY-MM-DD HH:MM:SS") # 策略总结 print(f"\n" + "=" * 80) print("策略运行总结") print("=" * 80) if len(trades_df) > 0: final_capital = results_df['net_value'].iloc[-1] total_return = (final_capital - INITIAL_CAPITAL) / INITIAL_CAPITAL * 100 print(f"初始资金: {INITIAL_CAPITAL:,.2f}元") print(f"最终资金: {final_capital:,.2f}元") print(f"总收益率: {total_return:.2f}%") print(f"交易次数: {len(trades_df)}笔") print(f"胜率: {(trades_df['盈亏金额'] > 0).sum() / len(trades_df) * 100:.1f}%") print(f"平均收益率: {trades_df['盈亏百分比'].mean():.2f}%") print(f"最大单笔盈利: {trades_df['盈亏金额'].max():+,.2f}元") print(f"最大单笔亏损: {trades_df['盈亏金额'].min():+,.2f}元") print(f"\n[SUCCESS] 策略运行成功！") else: print("未产生任何交易信号") except Exception as e: print(f"\n[ERROR] 策略运行出错: {str(e)}") import traceback traceback.print_exc() finally: print(f"\n" + "=" * 80) if __name__ == "__main__": main()