import akshare as ak import pandas as pd from datetime import datetime, timedelta, time import time as time_module import traceback from pytz import timezone import warnings warnings.filterwarnings('ignore') class ChinaStockData: """A股数据获取类""" def __init__(self): self.tz = timezone('Asia/Shanghai') # A股交易时间配置 self.trading_hours = { 'morning': {'start': '09:30', 'end': '11:30'}, 'afternoon': {'start': '13:00', 'end': '15:00'} } def get_stock_list(self): """获取A股股票列表""" try: import requests # 设置较短的超时时间，避免长时间等待 import akshare as ak print("正在获取A股股票列表...") # 尝试获取沪深A股实时行情，设置超时时间 try: # 临时设置requests的默认超时 original_timeout = getattr(requests, 'timeout', None) requests.timeout = 10 # 10秒超时 stock_info = ak.stock_zh_a_spot_em() # 恢复原始超时设置 if original_timeout: requests.timeout = original_timeout else: delattr(requests, 'timeout') except Exception as network_error: print(f"网络请求失败: {network_error}") # 网络失败时返回空列表，让调用方使用备用方案 return [] if stock_info is None or len(stock_info) == 0: print("获取到的股票数据为空") return [] # 增加到前2000只股票，提供更多选择 stock_list = [] for index, row in stock_info.head(2000).iterrows(): try: # 过滤掉ST股票和停牌股票 stock_name = str(row['名称']) if 'ST' not in stock_name and '*' not in stock_name: stock_list.append({ 'symbol': row['代码'], 'name': row['名称'], 'price': float(row['最新价']) if pd.notna(row['最新价']) else 0.0, 'change_pct': float(row['涨跌幅']) if pd.notna(row['涨跌幅']) else 0.0, 'volume': float(row['成交量']) if pd.notna(row['成交量']) else 0.0, 'amount': float(row['成交额']) if pd.notna(row['成交额']) else 0.0 }) except Exception as row_error: print(f"处理股票数据行时出错: {row_error}") continue # 按成交金额排序，优先显示活跃股票 stock_list.sort(key=lambda x: x['amount'], reverse=True) print(f"成功获取 {len(stock_list)} 只股票") return stock_list except Exception as e: print(f"获取股票列表失败: {e}") return [] def get_popular_stocks(self): """获取热门A股股票代码列表 - 扩展版本，按行业分类""" return [ # 包装引印刷 {'symbol': '002836', 'name': '新宏泽', 'sector': '包装印刷'}, # 银行股 {'symbol': '600036', 'name': '招商银行', 'sector': '银行'}, {'symbol': '000001', 'name': '平安银行', 'sector': '银行'}, {'symbol': '600000', 'name': '浦发银行', 'sector': '银行'}, {'symbol': '002142', 'name': '宁波银行', 'sector': '银行'}, {'symbol': '600016', 'name': '民生银行', 'sector': '银行'}, {'symbol': '601288', 'name': '农业银行', 'sector': '银行'}, {'symbol': '601398', 'name': '工商银行', 'sector': '银行'}, {'symbol': '601328', 'name': '交通银行', 'sector': '银行'}, # 白酒股 {'symbol': '600519', 'name': '贵州茅台', 'sector': '白酒'}, {'symbol': '000858', 'name': '五粮液', 'sector': '白酒'}, {'symbol': '002304', 'name': '洋河股份', 'sector': '白酒'}, {'symbol': '000596', 'name': '古井贡酒', 'sector': '白酒'}, {'symbol': '603369', 'name': '今世缘', 'sector': '白酒'}, {'symbol': '000799', 'name': '酒鬼酒', 'sector': '白酒'}, {'symbol': '600809', 'name': '山西汾酒', 'sector': '白酒'}, # 科技股 {'symbol': '002415', 'name': '海康威视', 'sector': '科技'}, {'symbol': '000063', 'name': '中兴通讯', 'sector': '科技'}, {'symbol': '002475', 'name': '立讯精密', 'sector': '科技'}, {'symbol': '300059', 'name': '东方财富', 'sector': '科技'}, {'symbol': '000725', 'name': '京东方A', 'sector': '科技'}, {'symbol': '002230', 'name': '科大讯飞', 'sector': '科技'}, {'symbol': '300433', 'name': '蓝思科技', 'sector': '科技'}, {'symbol': '002236', 'name': '大华股份', 'sector': '科技'}, # 新能源 {'symbol': '300750', 'name': '宁德时代', 'sector': '新能源'}, {'symbol': '002594', 'name': '比亚迪', 'sector': '新能源'}, {'symbol': '300274', 'name': '阳光电源', 'sector': '新能源'}, {'symbol': '002460', 'name': '赣锋锂业', 'sector': '新能源'}, {'symbol': '300014', 'name': '亿纬锂能', 'sector': '新能源'}, {'symbol': '600884', 'name': '杉杉股份', 'sector': '新能源'}, {'symbol': '002812', 'name': '恩捷股份', 'sector': '新能源'}, # 房地产 {'symbol': '000002', 'name': '万科A', 'sector': '房地产'}, {'symbol': '000858', 'name': '五粮液', 'sector': '房地产'}, {'symbol': '600048', 'name': '保利发展', 'sector': '房地产'}, {'symbol': '001979', 'name': '招商蛇口', 'sector': '房地产'}, {'symbol': '600606', 'name': '绿地控股', 'sector': '房地产'}, # 消费股 {'symbol': '600887', 'name': '伊利股份', 'sector': '消费'}, {'symbol': '000568', 'name': '泸州老窖', 'sector': '消费'}, {'symbol': '600600', 'name': '青岛啤酒', 'sector': '消费'}, {'symbol': '000895', 'name': '双汇发展', 'sector': '消费'}, {'symbol': '002304', 'name': '洋河股份', 'sector': '消费'}, {'symbol': '600779', 'name': '水井坊', 'sector': '消费'}, # 医药股 {'symbol': '600196', 'name': '复星医药', 'sector': '医药'}, {'symbol': '000661', 'name': '长春高新', 'sector': '医药'}, {'symbol': '300015', 'name': '爱尔眼科', 'sector': '医药'}, {'symbol': '002821', 'name': '凯莱英', 'sector': '医药'}, {'symbol': '300760', 'name': '迈瑞医疗', 'sector': '医药'}, {'symbol': '600276', 'name': '恒瑞医药', 'sector': '医药'}, # 证券股 {'symbol': '000776', 'name': '广发证券', 'sector': '证券'}, {'symbol': '600030', 'name': '中信证券', 'sector': '证券'}, {'symbol': '000166', 'name': '申万宏源', 'sector': '证券'}, {'symbol': '601688', 'name': '华泰证券', 'sector': '证券'}, {'symbol': '600837', 'name': '海通证券', 'sector': '证券'}, # 化工股 {'symbol': '600309', 'name': '万华化学', 'sector': '化工'}, {'symbol': '002352', 'name': '顺丰控股', 'sector': '化工'}, {'symbol': '600346', 'name': '恒力石化', 'sector': '化工'}, {'symbol': '000792', 'name': '盐湖股份', 'sector': '化工'}, # 汽车股 {'symbol': '600104', 'name': '上汽集团', 'sector': '汽车'}, {'symbol': '000625', 'name': '长安汽车', 'sector': '汽车'}, {'symbol': '601633', 'name': '长城汽车', 'sector': '汽车'}, {'symbol': '002049', 'name': '紫光国微', 'sector': '汽车'}, # 军工股 {'symbol': '002179', 'name': '中航光电', 'sector': '军工'}, {'symbol': '600893', 'name': '航发动力', 'sector': '军工'}, {'symbol': '000768', 'name': '中航飞机', 'sector': '军工'}, # 基建股 {'symbol': '601186', 'name': '中国铁建', 'sector': '基建'}, {'symbol': '601390', 'name': '中国中铁', 'sector': '基建'}, {'symbol': '000001', 'name': '平安银行', 'sector': '基建'}, # 煤炭股 {'symbol': '601225', 'name': '陕西煤业', 'sector': '煤炭'}, {'symbol': '600188', 'name': '兖矿能源', 'sector': '煤炭'}, {'symbol': '601898', 'name': '中煤能源', 'sector': '煤炭'}, # 钢铁股 {'symbol': '000717', 'name': '韶钢松山', 'sector': '钢铁'}, {'symbol': '600019', 'name': '宝钢股份', 'sector': '钢铁'}, {'symbol': '000708', 'name': '中信特钢', 'sector': '钢铁'}, ] def timeframe_to_period(self, timeframe): """将时间周期转换为akshare的period参数""" mapping = { '1m': '1', # 1分钟 '3m': '5', # 3分钟 (A股不支持3分钟，使用5分钟替代) '5m': '5', # 5分钟 '15m': '15', # 15分钟 '30m': '30', # 30分钟 '1h': '60', # 60分钟 '2h': '60', # 2小时 (A股不支持2小时，使用1小时替代) '4h': '60', # 4小时 (A股不支持4小时，使用1小时替代) '6h': '60', # 6小时 (A股不支持6小时，使用1小时替代) '8h': '60', # 8小时 (A股不支持8小时，使用1小时替代) '12h': '60', # 12小时 (A股不支持12小时，使用1小时替代) '1d': 'daily', # 日线 '3d': 'daily', # 3日 (A股不支持3日，使用日线替代) '1w': 'weekly',# 周线 '1M': 'monthly'# 月线 } return mapping.get(timeframe, 'daily') def get_kl_data(self, symbol, timeframe='1d', start_date=None, end_date=None, limit=1000): """ 获取A股K线数据 - 支持分批次获取突破单次限制 :param symbol: 股票代码，如 '000001' :param timeframe: 时间周期，如 '1d', '1h', '5m' :param start_date: 开始日期，格式 'YYYY-MM-DD' :param end_date: 结束日期，格式 'YYYY-MM-DD' :param limit: 数据条数限制 :return: DataFrame """ try: period = self.timeframe_to_period(timeframe) # 处理时间参数 if start_date is None: # 默认获取最近一年的数据 start_date = (datetime.now() - timedelta(days=365)).strftime('%Y%m%d') else: # 将 YYYY-MM-DD 格式转换为 YYYYMMDD if '-' in start_date: start_date = start_date.replace('-', '') if end_date is None: end_date = datetime.now().strftime('%Y%m%d') else: if '-' in end_date: end_date = end_date.replace('-', '') print(f"获取A股数据: {symbol}, 周期: {timeframe}, 开始: {start_date}, 结束: {end_date}") # 分批次获取数据以突破单次限制 all_data = [] current_start = start_date # 计算时间间隔（根据时间周期调整批次大小） if period in ['1', '5', '15', '30']: # 分钟级数据，每次获取7天 batch_days = 7 elif period == '60': # 小时级数据（所有小时级别都映射到60分钟），每次获取30天 batch_days = 30 else: # 日线及以上，每次获取365天 batch_days = 365 max_iterations = 20 # 最大迭代次数，防止无限循环 iteration_count = 0 while current_start <= end_date and iteration_count < max_iterations: iteration_count += 1 # 计算当前批次的结束时间 current_start_dt = datetime.strptime(current_start, '%Y%m%d') current_end_dt = current_start_dt + timedelta(days=batch_days) current_end = min(current_end_dt.strftime('%Y%m%d'), end_date) print(f"批次 {iteration_count}: 获取 {current_start} 到 {current_end} 的数据") try: # 根据时间周期选择不同的API df_batch = None if period in ['1', '5', '15', '30', '60']: # 分钟级数据 df_batch = ak.stock_zh_a_hist_min_em(symbol=symbol, period=period, start_date=current_start, end_date=current_end) if df_batch is not None and len(df_batch) > 0: # 重命名列 df_batch = df_batch.rename(columns={ '时间': 'date', '开盘': 'open', '收盘': 'close', '最高': 'high', '最低': 'low', '成交量': 'volume' }) else: # 日线、周线、月线数据 df_batch = ak.stock_zh_a_hist(symbol=symbol, period=period, start_date=current_start, end_date=current_end) if df_batch is not None and len(df_batch) > 0: # 重命名列 df_batch = df_batch.rename(columns={ '日期': 'date', '开盘': 'open', '收盘': 'close', '最高': 'high', '最低': 'low', '成交量': 'volume' }) if df_batch is not None and len(df_batch) > 0: # 转换时间格式 df_batch['date'] = pd.to_datetime(df_batch['date']) # 根据A股交易时间调整时间戳 df_batch = self.adjust_timestamp_for_trading_hours(df_batch, timeframe) all_data.append(df_batch) print(f"批次 {iteration_count}: 获取到 {len(df_batch)} 条记录") else: print(f"批次 {iteration_count}: 未获取到数据") except Exception as e: print(f"批次 {iteration_count} 获取失败: {e}") # 继续下一个批次 # 更新下一批次的开始时间 current_start = (current_end_dt + timedelta(days=1)).strftime('%Y%m%d') # 防止API请求过于频繁 time_module.sleep(0.5) # 合并所有批次的数据 if not all_data: print(f"未获取到任何数据: {symbol}") return None # 合并DataFrame df = pd.concat(all_data, ignore_index=True) # 数据清洗和格式化 df = df.dropna() # 删除空值 df = df.drop_duplicates(subset=['date']) # 删除重复数据 df = df.sort_values('date').reset_index(drop=True) # 按时间排序 # A股特有的数据清理和时间处理 df = self.clean_a_stock_data(df, timeframe) # 限制数据条数 - 只有在没有指定明确时间范围时才应用 # 如果用户指定了start_date和end_date，应该返回该时间范围内的所有数据 if limit is not None and len(df) > limit: # 检查是否指定了明确的时间范围 if start_date and end_date: # 如果指定了时间范围，优先返回完整的时间范围数据 print(f"用户指定了时间范围 {start_date} 到 {end_date}，返回完整数据 {len(df)} 条记录") if len(df) > 10000: # 防止数据量过大，设置一个合理的上限 print(f"警告：数据量过大({len(df)}条)，为保证性能将限制为最新的10000条记录") df = df.tail(10000).reset_index(drop=True) else: # 如果没有指定时间范围，使用默认的limit限制 print(f"未指定明确时间范围，应用默认限制，返回最新的 {limit} 条记录") df = df.tail(limit).reset_index(drop=True) elif limit is None and len(df) > 10000: # 即使没有limit限制，也要防止数据量过大影响性能 print(f"无limit限制但数据量过大({len(df)}条)，为保证性能将限制为最新的10000条记录") df = df.tail(10000).reset_index(drop=True) # 添加技术指标 df = self.add_indicators(df) # 最终数据验证 - 确保没有NaN值 import numpy as np # 检查并处理任何剩余的NaN值 if df.isnull().any().any(): print("警告：发现NaN值，正在清理...") # 对于数值列，用0填充NaN numeric_cols = df.select_dtypes(include=[np.number]).columns for col in numeric_cols: if col in ['volume_ratio']: df[col] = df[col].fillna(1.0) else: df[col] = df[col].fillna(0) # 删除仍然包含NaN的行 df = df.dropna() # 确保所有数值都是有限的 for col in df.select_dtypes(include=[np.number]).columns: df[col] = df[col].replace([np.inf, -np.inf], 0 if col != 'volume_ratio' else 1.0) print(f"成功获取A股数据: {len(df)} 条记录 (共 {len(all_data)} 个批次)") return df except Exception as e: print(f"获取A股数据失败: {e}") traceback.print_exc() return None def add_indicators(self, df): """添加技术指标""" try: import talib.abstract as ta import numpy as np # MACD指标 fast = 8 slow = 16 period = 6 macd = ta.MACD(df, fastperiod=fast, slowperiod=slow, signalperiod=period) df['macd'] = macd['macd'].fillna(0) df['macdsignal'] = macd['macdsignal'].fillna(0) df['macdhist'] = macd['macdhist'].fillna(0) # 移动平均线 df['ma5'] = ta.MA(df, timeperiod=5).fillna(0) df['ma10'] = ta.MA(df, timeperiod=10).fillna(0) df['ma30'] = ta.EMA(df, timeperiod=30).fillna(0) df['ma250'] = ta.MA(df, timeperiod=250).fillna(0) # RSI指标 df['rsi'] = ta.RSI(df, timeperiod=14).fillna(0) # 成交量指标 df['avg_volume'] = df['volume'].rolling(10).mean().fillna(0) df['volume_ratio'] = (df['volume'] / df['avg_volume']).fillna(1.0) # 处理Infinity和-Infinity值 df['volume_ratio'] = df['volume_ratio'].replace([float('inf'), float('-inf')], 1.0) # 确保所有指标列都不包含NaN或无限值 indicator_columns = ['macd', 'macdsignal', 'macdhist', 'ma5', 'ma10', 'ma30', 'ma250', 'rsi', 'avg_volume', 'volume_ratio'] for col in indicator_columns: if col in df.columns: # 替换NaN、inf、-inf为合理的默认值 df[col] = df[col].replace([np.nan, np.inf, -np.inf], 0 if col != 'volume_ratio' else 1.0) return df except Exception as e: print(f"添加指标失败: {e}") return df def search_stock(self, keyword): """搜索股票 - 支持代码和名称模糊搜索""" try: if not keyword or len(keyword.strip()) == 0: return [] keyword = keyword.strip().upper() results = [] # 从热门股票中搜索 popular_stocks = self.get_popular_stocks() for stock in popular_stocks: if (keyword in stock['symbol'] or keyword.lower() in stock['name'].lower() or stock['symbol'].startswith(keyword)): results.append({ 'symbol': stock['symbol'], 'name': stock['name'], 'sector': stock.get('sector', ''), 'source': '热门股票' }) # 如果热门股票中找到的结果少于10个，从完整股票列表中搜索 if len(results) < 10: try: # 获取完整股票列表进行搜索 stock_info = ak.stock_zh_a_spot_em() # 搜索前1000只活跃股票 for index, row in stock_info.head(1000).iterrows(): stock_code = str(row['代码']) stock_name = str(row['名称']) # 过滤ST股票 if 'ST' in stock_name or '*' in stock_name: continue # 检查是否已经在结果中 if any(r['symbol'] == stock_code for r in results): continue # 搜索匹配 if (keyword in stock_code or keyword.lower() in stock_name.lower() or stock_code.startswith(keyword)): results.append({ 'symbol': stock_code, 'name': stock_name, 'price': float(row['最新价']) if pd.notna(row['最新价']) else 0.0, 'change_pct': float(row['涨跌幅']) if pd.notna(row['涨跌幅']) else 0.0, 'source': '全市场搜索' }) # 限制结果数量 if len(results) >= 30: break except Exception as e: print(f"全市场搜索失败: {e}") # 排序：优先显示代码匹配的结果 def sort_key(item): if item['symbol'].startswith(keyword): return (0, item['symbol']) # 代码开头匹配优先级最高 elif keyword in item['symbol']: return (1, item['symbol']) # 代码包含匹配次之 else: return (2, item['symbol']) # 名称匹配最后 results.sort(key=sort_key) # 限制返回结果数量 return results[:20] except Exception as e: print(f"搜索股票失败: {e}") return [] def get_stock_by_sector(self, sector=None): """根据行业获取股票列表""" try: popular_stocks = self.get_popular_stocks() if sector: return [stock for stock in popular_stocks if stock.get('sector', '') == sector] else: # 按行业分组 sectors = {} for stock in popular_stocks: sector_name = stock.get('sector', '其他') if sector_name not in sectors: sectors[sector_name] = [] sectors[sector_name].append(stock) return sectors except Exception as e: print(f"获取行业股票失败: {e}") return {} if sector is None else [] def get_all_sectors(self): """获取所有行业分类""" try: popular_stocks = self.get_popular_stocks() sectors = set() for stock in popular_stocks: sector = stock.get('sector', '其他') sectors.add(sector) return sorted(list(sectors)) except Exception as e: print(f"获取行业分类失败: {e}") return [] def is_trading_day(self, date): """判断是否为交易日（排除周末和节假日）""" try: # 将日期转换为datetime对象 if isinstance(date, str): date = datetime.strptime(date.split()[0], '%Y-%m-%d') elif isinstance(date, pd.Timestamp): date = date.to_pydatetime() # 周末不是交易日 if date.weekday() >= 5: # 5=周六, 6=周日 return False # 这里可以进一步添加节假日判断 # 目前暂时只过滤周末 return True except Exception as e: print(f"判断交易日失败: {e}") return True # 默认返回True，避免过度过滤 def is_trading_time(self, dt): """判断是否为交易时间""" try: if isinstance(dt, str): dt = pd.to_datetime(dt) time_str = dt.strftime('%H:%M') # 上午交易时间：09:30-11:30 morning_start = self.trading_hours['morning']['start'] morning_end = self.trading_hours['morning']['end'] # 下午交易时间：13:00-15:00 afternoon_start = self.trading_hours['afternoon']['start'] afternoon_end = self.trading_hours['afternoon']['end'] return ((morning_start <= time_str <= morning_end) or (afternoon_start <= time_str <= afternoon_end)) except Exception as e: print(f"判断交易时间失败: {e}") return True # 默认返回True，避免过度过滤 def adjust_timestamp_for_trading_hours(self, df, timeframe): """根据A股交易时间调整时间戳""" try: if df is None or len(df) == 0: return df # 确保date列是datetime类型 if 'date' in df.columns: df['date'] = pd.to_datetime(df['date']) # 对于日线数据，设置为收盘时间（15:00） if timeframe == '1d': df['date'] = df['date'].dt.normalize() + pd.Timedelta(hours=15) # 对于分钟级数据，过滤非交易时间的数据 elif timeframe in ['1m', '3m', '5m', '15m', '30m', '1h', '2h', '4h', '6h', '8h', '12h']: # 过滤交易日 df = df[df['date'].apply(self.is_trading_day)] # 过滤交易时间（只在有足够数据时进行） if len(df) > 10: # 避免过度过滤导致数据不足 df = df[df['date'].apply(self.is_trading_time)] # 重新计算时间戳 if 'date' in df.columns: # 将时间转换为上海时区 df['date'] = df['date'].dt.tz_localize('Asia/Shanghai', ambiguous='infer', nonexistent='shift_forward') # 转换为毫秒时间戳 df['timestamp'] = df['date'].astype('int64') // 10**6 return df.reset_index(drop=True) except Exception as e: print(f"调整A股时间戳失败: {e}") traceback.print_exc() return df def get_trading_calendar(self, start_date, end_date): """获取交易日历（简化版本）""" try: # 使用akshare获取交易日历 trading_calendar = ak.tool_trade_date_hist_sina() # 过滤指定日期范围 start_dt = pd.to_datetime(start_date) end_dt = pd.to_datetime(end_date) trading_days = [] for _, row in trading_calendar.iterrows(): trade_date = pd.to_datetime(row['trade_date']) if start_dt <= trade_date <= end_dt: trading_days.append(trade_date.strftime('%Y-%m-%d')) return trading_days except Exception as e: print(f"获取交易日历失败: {e}") # 如果获取失败，生成简单的工作日列表（排除周末） trading_days = [] current = pd.to_datetime(start_date) end = pd.to_datetime(end_date) while current <= end: if current.weekday() < 5: # 周一到周五 trading_days.append(current.strftime('%Y-%m-%d')) current += timedelta(days=1) return trading_days def fill_trading_gaps(self, df, timeframe): """填补A股交易时间间隙，确保图表连续性""" try: if df is None or len(df) == 0: return df # 对于日线数据，不需要填补间隙，因为本来就是每日一个数据点 if timeframe == '1d': return df # 对于分钟级数据，创建完整的交易时间序列 if timeframe in ['1m', '3m', '5m', '15m', '30m', '1h', '2h', '4h', '6h', '8h', '12h']: # 获取数据的开始和结束时间 start_date = df['date'].min().date() end_date = df['date'].max().date() # 创建完整的交易时间序列 complete_times = [] current_date = start_date # 获取时间间隔（分钟）- 注意A股新增的时间周期都映射到了已有的时间间隔 freq_map = { '1m': 1, '3m': 5, '5m': 5, '15m': 15, '30m': 30, '1h': 60, '2h': 60, '4h': 60, '6h': 60, '8h': 60, '12h': 60 } freq_minutes = freq_map.get(timeframe, 5) while current_date <= end_date: # 只处理交易日 if self.is_trading_day(current_date): # 上午交易时间 - 使用datetime.time而不是pd.Time morning_start = pd.Timestamp.combine(current_date, time(9, 30)) morning_end = pd.Timestamp.combine(current_date, time(11, 30)) # 下午交易时间 afternoon_start = pd.Timestamp.combine(current_date, time(13, 0)) afternoon_end = pd.Timestamp.combine(current_date, time(15, 0)) # 生成上午时间序列 current_time = morning_start while current_time <= morning_end: complete_times.append(current_time) current_time += pd.Timedelta(minutes=freq_minutes) # 生成下午时间序列 current_time = afternoon_start while current_time <= afternoon_end: complete_times.append(current_time) current_time += pd.Timedelta(minutes=freq_minutes) current_date += timedelta(days=1) # 创建完整时间序列的DataFrame if complete_times: complete_df = pd.DataFrame({'date': complete_times}) complete_df['date'] = complete_df['date'].dt.tz_localize('Asia/Shanghai') complete_df['timestamp'] = complete_df['date'].astype('int64') // 10**6 # 将原始数据合并到完整时间序列 # 使用时间戳进行合并，避免时区问题 df_merged = pd.merge(complete_df, df, on='timestamp', how='left', suffixes=('', '_orig')) # 保持原有date列 df_merged['date'] = df_merged['date'] # 对于缺失的OHLCV数据，使用前向填充 price_cols = ['open', 'high', 'low', 'close'] for col in price_cols: if col in df_merged.columns: df_merged[col] = df_merged[col].ffill() # 成交量缺失时设为0 if 'volume' in df_merged.columns: df_merged['volume'] = df_merged['volume'].fillna(0) # 删除辅助列 cols_to_drop = [col for col in df_merged.columns if col.endswith('_orig')] df_merged = df_merged.drop(columns=cols_to_drop) return df_merged return df except Exception as e: print(f"填补交易时间间隙失败: {e}") traceback.print_exc() return df def clean_a_stock_data(self, df, timeframe): """清理A股数据，处理异常值和时间问题""" try: if df is None or len(df) == 0: return df import numpy as np # 首先删除所有包含NaN的行 df = df.dropna() # 删除价格异常的数据 price_cols = ['open', 'high', 'low', 'close'] for col in price_cols: if col in df.columns: # 删除价格为0、负数、NaN、inf的记录 df = df[df[col] > 0] df = df[np.isfinite(df[col])] # 检查OHLC逻辑合理性 if all(col in df.columns for col in price_cols): # high应该是最高价 df = df[df['high'] >= df['open']] df = df[df['high'] >= df['close']] # low应该是最低价 df = df[df['low'] <= df['open']] df = df[df['low'] <= df['close']] # high应该大于等于low df = df[df['high'] >= df['low']] # 删除成交量异常的数据 if 'volume' in df.columns: # 删除成交量为负数、NaN、inf的记录 df = df[df['volume'] >= 0] df = df[np.isfinite(df['volume'])] # 确保所有数值列都不包含NaN或无限值 numeric_cols = df.select_dtypes(include=[np.number]).columns for col in numeric_cols: # 替换NaN、inf、-inf为0（除了价格列，价格列的异常值已经被过滤掉了） if col not in price_cols: df[col] = df[col].replace([np.nan, np.inf, -np.inf], 0) # 确保时间序列连续性（仅对分钟级数据） if timeframe in ['1m', '3m', '5m', '15m', '30m', '1h', '2h', '4h', '6h', '8h', '12h']: df = self.fill_trading_gaps(df, timeframe) # 最后再次检查并清理任何剩余的NaN值 df = df.dropna() return df.reset_index(drop=True) except Exception as e: print(f"清理A股数据失败: {e}") return df