""" 数据爬取服务 用于从外部网站获取期货数据 """ import requests from bs4 import BeautifulSoup import pandas as pd import logging from datetime import datetime import re logger = logging.getLogger(__name__) class FutureDataScraper: """期货数据爬取类""" def __init__(self, url=None): """初始化爬虫""" # 从配置服务获取数据源URL if url is None: try: from app.services.config_service import get_str_config url = get_str_config('data_source_url', 'http://121.37.80.177/fees.html') except Exception as e: logger.warning(f"获取配置失败，使用默认URL: {e}") url = "http://121.37.80.177/fees.html" self.url = url logger.info(f"期货数据爬虫初始化，URL: {self.url}") def fetch_future_daily(self): """ 从网页爬取期货每日数据 返回一个包含期货数据的DataFrame """ try: # 检查当前时间是否在交易时间内 current_time = datetime.now().time() # 从配置服务获取交易时间 try: from app.services.config_service import get_str_config trading_start = get_str_config('trading_start_time', '09:00') trading_end = get_str_config('trading_end_time', '17:00') except Exception as e: logger.warning(f"获取交易时间配置失败，使用默认值: {e}") trading_start = '09:00' trading_end = '17:00' is_trading_hours = ( current_time >= datetime.strptime(trading_start, '%H:%M').time() and current_time <= datetime.strptime(trading_end, '%H:%M').time() ) logger.debug(f"当前时间: {current_time}, 交易时间: {trading_start}-{trading_end}, 是否在交易时间内: {is_trading_hours}") # 发送HTTP请求 logger.debug(f"开始请求URL: {self.url}") response = requests.get(self.url) response.raise_for_status() # 如果请求失败则抛出异常 # 尝试多种编码方式 try: from app.services.config_service import get_str_config encoding_list = get_str_config('encoding_attempts', 'gb2312,gbk,gb18030,utf-8') encodings = [enc.strip() for enc in encoding_list.split(',')] except Exception as e: logger.warning(f"获取编码配置失败，使用默认值: {e}") encodings = ['gb2312', 'gbk', 'gb18030', 'utf-8'] html_text = None for encoding in encodings: try: html_text = response.content.decode(encoding) print(f'成功使用 {encoding} 编码解析') break except UnicodeDecodeError: continue if html_text is None: html_text = response.content.decode('utf-8', errors='ignore') print('使用UTF-8忽略错误模式解析') # 使用BeautifulSoup解析HTML soup = BeautifulSoup(html_text, 'html.parser') # 查找表格 table = soup.find('table') if not table: logger.error("未找到数据表格") logger.debug(f"页面内容前100字符: {html_text[:100]}") return None logger.debug("找到数据表格，开始解析") # 解析表格数据 headers = [] header_row = table.find('tr') if header_row: headers = [th.text.strip() for th in header_row.find_all('th')] logger.debug(f"表头: {headers}") # 检查表头是否包含预期的中文字段 if not any(('交易所' in h or '合约' in h) for h in headers): logger.warning("表头可能存在编码问题，没有找到预期的中文字段") else: logger.warning("未找到表头行") rows = [] data_rows = table.find_all('tr')[1:] # 跳过表头行 logger.debug(f"找到 {len(data_rows)} 行数据") # 用于存储每个品种的所有合约数据 product_contracts = {} # 查找持仓量列的位置 open_interest_col = -1 for i, header in enumerate(headers): if '持仓量' in header: open_interest_col = i break # 如果没找到持仓量列，使用经验位置 if open_interest_col == -1: try: from app.services.config_service import get_int_config default_col = get_int_config('open_interest_column_position', 21) except Exception as e: logger.warning(f"获取持仓量列配置失败，使用默认值: {e}") default_col = 21 open_interest_col = default_col if len(headers) > default_col else len(headers) - 2 logger.debug(f"未找到持仓量列，使用经验位置 {open_interest_col} (配置默认值: {default_col})") else: logger.debug(f"找到持仓量列位置: {open_interest_col}") # 第一遍遍历：收集所有合约数据 for row in data_rows: cols = row.find_all('td') if not cols: continue # 获取合约代码和品种代码 contract_code = cols[1].text.strip() product_code = ''.join([c for c in contract_code if c.isalpha()]).upper() # 获取持仓量（用于判断主力合约） open_interest = 0 if open_interest_col >= 0 and len(cols) > open_interest_col: oi_text = cols[open_interest_col].text.strip() try: # 清理持仓量文本，移除逗号等 import re oi_clean = re.sub(r'[^\d]', '', oi_text) if oi_clean: open_interest = int(oi_clean) except (ValueError, TypeError): open_interest = 0 # 将合约信息存储到对应品种的列表中 if product_code not in product_contracts: product_contracts[product_code] = [] product_contracts[product_code].append((contract_code, open_interest)) # 确定主力合约（根据持仓量） product_main_contracts = {} for product_code, contracts in product_contracts.items(): if contracts: # 按持仓量排序，取持仓量最大的合约 main_contract = max(contracts, key=lambda x: x[1])[0] max_oi = max(contracts, key=lambda x: x[1])[1] product_main_contracts[product_code] = main_contract # logger.debug(f"根据持仓量确定主力合约: {product_code} {main_contract} (持仓量: {max_oi})") logger.debug(f"识别出 {len(product_main_contracts)} 个主力合约") # 第二遍遍历：处理所有数据行 for row in data_rows: cols = row.find_all('td') if not cols: continue # 获取合约代码和品种代码 contract_code = cols[1].text.strip() product_code = ''.join([c for c in contract_code if c.isalpha()]).upper() # 判断是否为主力合约 is_main = False if product_code in product_main_contracts: if contract_code == product_main_contracts[product_code]: is_main = True # logger.debug(f"标记主力合约: {product_code} {contract_code}") # 收集行数据 row_data = [col.text.strip() for col in cols] row_data.append(is_main) # 添加主力合约标志 rows.append(row_data) # 创建DataFrame headers.append('is_main_contract') # 添加主力合约标志列 # 确保所有行的长度与表头一致 for i, row in enumerate(rows): if len(row) != len(headers): logger.warning(f"第{i+1}行数据长度({len(row)})与表头长度({len(headers)})不一致，进行调整") # 如果行长度不够，用空字符串填充 if len(row) < len(headers): row.extend([''] * (len(headers) - len(row))) # 如果行长度超过，截断 elif len(row) > len(headers): row = row[:len(headers)] rows[i] = row df = pd.DataFrame(rows, columns=headers) # 记录日志 logger.debug(f"成功获取期货数据，共{len(df)}条记录") logger.debug(f"数据前5行: {df.head()}") return df except Exception as e: logger.error(f"获取期货数据失败: {str(e)}", exc_info=True) return None def update_future_daily(self, db_session, FutureDaily): """ 更新数据库中的future_daily表 参数: db_session: 数据库会话 FutureDaily: 期货日数据模型类 返回: 更新的记录数量 """ try: # 获取数据 df = self.fetch_future_daily() if df is None or df.empty: logger.error("无法更新期货日数据: 未获取到数据") return 0 # 打印表头查看具体的字段名 logger.debug(f"表格的字段名: {list(df.columns)}") # 清空当前数据表 db_session.query(FutureDaily).delete() # 创建新记录 records = [] update_time = datetime.now() # 遍历DataFrame中的每一行 for idx, row in df.iterrows(): try: # 提取合约代码和品种代码 contract_code = row.get('合约代码', '') if not contract_code: logger.warning(f"第{idx+1}行没有合约代码，跳过") continue # 提取品种代码（合约代码中的字母部分） product_code = ''.join([c for c in contract_code if c.isalpha()]) # 记录处理的行号和关键字段 # logger.debug(f"处理第{idx+1}行，合约代码: {contract_code}, 产品代码: {product_code}") # 创建新记录 record = FutureDaily( exchange=row.get('交易所', ''), contract_code=contract_code, contract_name=row.get('合约名称', ''), product_code=product_code, product_name=row.get('品种名称', ''), contract_multiplier=self._safe_float(row.get('合约乘数', 0)), price_tick=self._safe_float(row.get('最小跳动', 0)), open_fee_rate=self._safe_float(row.get('开仓费率', 0)), open_fee=self._safe_float(row.get('开仓费用/手', 0)), close_fee_rate=self._safe_float(row.get('平仓费率', 0)), close_fee=self._safe_float(row.get('平仓费用/手', 0)), close_today_fee_rate=self._safe_float(row.get('平今费率', 0)), close_today_fee=self._safe_float(row.get('平今费用/手', 0)), long_margin_rate=self._safe_float(row.get('做多保证金率', 0)), long_margin_fee=self._safe_float(row.get('做多保证金/手', 0)), short_margin_rate=self._safe_float(row.get('做空保证金率', 0)), short_margin_fee=self._safe_float(row.get('做空保证金/手', 0)), latest_price=self._safe_float(row.get('最新价', 0)), open_interest=self._safe_int(row.get('持仓量', 0)), volume=self._safe_int(row.get('成交量', 0)), is_main_contract=row.get('is_main_contract', False), update_time=update_time ) if row.get('is_main_contract', False): logger.debug(f"保存主力合约到数据库: {product_code} {contract_code}") records.append(record) except Exception as e: logger.error(f"解析期货日数据行失败(行号:{idx+1}): {str(e)}", exc_info=True) continue # 批量添加记录 if records: db_session.add_all(records) db_session.commit() logger.debug(f"成功更新期货日数据，共{len(records)}条记录") return len(records) else: logger.warning("无期货日数据可更新") return 0 except Exception as e: logger.error(f"更新期货日数据失败: {str(e)}", exc_info=True) db_session.rollback() return 0 def _normalize_contract_code(self, contract_code): """ 标准化合约代码格式 例如：将 'AP505' 转换为 'AP2505'，同时确保字母部分为大写 """ try: if not contract_code: return contract_code # 提取字母部分和数字部分，并将字母转换为大写 letters = ''.join(c for c in contract_code if c.isalpha()).upper() numbers = ''.join(c for c in contract_code if c.isdigit()) # 如果数字部分是3位数，在前面加上2 if len(numbers) == 3: numbers = '2' + numbers return letters + numbers except Exception as e: logger.error(f"合约代码格式转换失败: {str(e)}") return contract_code def update_future_info_from_daily(self, db_session, FutureInfo, FutureDaily): """ 根据future_daily表更新future_info表的数据 参数: db_session: 数据库会话 FutureInfo: 期货基础信息模型类 FutureDaily: 期货日数据模型类 返回: 更新的记录数量 """ try: # 获取当前数据库中的所有期货基础信息 # 将contract_letter转换为大写用于统一比较 futures = {f.contract_letter.upper(): f for f in db_session.query(FutureInfo).all()} logger.debug(f"从future_info表获取到{len(futures)}个期货品种") # 获取最新的future_daily数据的所有产品代码 product_data = {} # 记录每个品种的主力合约 product_main_contracts = {} main_contract_records = db_session.query(FutureDaily).filter(FutureDaily.is_main_contract == True).all() logger.debug(f"查询到{len(main_contract_records)}条主力合约记录") for daily in main_contract_records: product_code_upper = daily.product_code.upper() if product_code_upper not in product_main_contracts: product_main_contracts[product_code_upper] = self._normalize_contract_code(daily.contract_code) # logger.debug(f"从数据库获取主力合约: {product_code_upper} -> {daily.contract_code}") logger.debug(f"整理得到{len(product_main_contracts)}个主力合约: {product_main_contracts}") # 如果没有找到主力合约，尝试根据持仓量重新识别 if not product_main_contracts: logger.debug("未找到主力合约，尝试根据持仓量重新识别...") # 按品种分组，找出每个品种持仓量最大的合约 product_contracts = {} for daily in db_session.query(FutureDaily).all(): product_code_upper = daily.product_code.upper() if product_code_upper not in product_contracts: product_contracts[product_code_upper] = [] product_contracts[product_code_upper].append((daily.contract_code, daily.open_interest or 0)) # 确定每个品种的主力合约 for product_code, contracts in product_contracts.items(): if contracts: # 按持仓量排序，取持仓量最大的合约 main_contract = max(contracts, key=lambda x: x[1])[0] max_oi = max(contracts, key=lambda x: x[1])[1] product_main_contracts[product_code] = self._normalize_contract_code(main_contract) logger.debug(f"根据持仓量识别主力合约: {product_code} {main_contract} (持仓量: {max_oi})") logger.debug(f"重新识别出 {len(product_main_contracts)} 个主力合约") # 只获取主力合约的数据用于更新future_info表 main_daily_records = db_session.query(FutureDaily).filter(FutureDaily.is_main_contract == True).all() logger.debug(f"查询主力合约数据: 找到{len(main_daily_records)}条记录") for daily in main_daily_records: product_code_upper = daily.product_code.upper() product_data[product_code_upper] = daily # logger.debug(f"获取主力合约数据: {product_code_upper} -> {daily.contract_code}") logger.debug(f"从future_daily表获取到{len(product_data)}个主力合约的数据: {list(product_data.keys())}") # 更新计数器 updated_count = 0 not_found_count = 0 # 更新期货基础信息 logger.debug(f"开始匹配和更新期货基础信息: future_info有{len(futures)}个品种，product_data有{len(product_data)}个品种") for contract_letter, future in futures.items(): contract_letter_upper = contract_letter.upper() # logger.debug(f"检查期货品种: {contract_letter_upper} -> 是否在product_data中: {contract_letter_upper in product_data}") if contract_letter_upper in product_data: daily = product_data[contract_letter_upper] # 记录更新前的值 old_values = { 'exchange': future.exchange, 'contract_multiplier': future.contract_multiplier, 'long_margin_rate': future.long_margin_rate, 'short_margin_rate': future.short_margin_rate, 'open_fee': future.open_fee, 'close_fee': future.close_fee, 'close_today_rate': future.close_today_rate, 'close_today_fee': future.close_today_fee, 'current_main_contract': future.current_main_contract } # 更新字段 future.exchange = daily.exchange future.contract_multiplier = daily.contract_multiplier future.long_margin_rate = daily.long_margin_rate future.short_margin_rate = daily.short_margin_rate future.open_fee = daily.open_fee future.close_fee = daily.close_fee future.close_today_rate = daily.close_today_fee_rate future.close_today_fee = daily.close_today_fee # 根据当前价格和保证金率计算保证金金额 if hasattr(daily, 'latest_price') and daily.latest_price and future.contract_multiplier: if future.long_margin_rate: future.long_margin_amount = daily.latest_price * future.contract_multiplier * future.long_margin_rate if future.short_margin_rate: future.short_margin_amount = daily.latest_price * future.contract_multiplier * future.short_margin_rate # 更新主力合约 if contract_letter_upper in product_main_contracts: future.current_main_contract = product_main_contracts[contract_letter_upper] # 检查是否有实际更新 has_changes = False changes = [] for field, old_value in old_values.items(): new_value = getattr(future, field) if old_value != new_value: has_changes = True changes.append(f"{field}: {old_value} -> {new_value}") if has_changes: logger.debug(f"更新期货 {future.contract_letter} ({future.name}): {', '.join(changes)}") updated_count += 1 else: not_found_count += 1 logger.warning(f"未找到期货 {future.contract_letter} ({future.name}) 的每日数据") # 更新不在future_info中的主力合约 for product_code, contract_code in product_main_contracts.items(): if product_code not in futures: # 创建新的期货基础信息记录 daily = product_data.get(product_code) if daily: future_info = FutureInfo( contract_letter=product_code, name=daily.product_name, market=0, # 默认为国内市场 exchange=daily.exchange, contract_multiplier=daily.contract_multiplier, long_margin_rate=daily.long_margin_rate, short_margin_rate=daily.short_margin_rate, open_fee=daily.open_fee, close_fee=daily.close_fee, close_today_rate=daily.close_today_fee_rate, close_today_fee=daily.close_today_fee, # 计算保证金金额 long_margin_amount=(daily.latest_price * daily.contract_multiplier * daily.long_margin_rate) if daily.latest_price and daily.long_margin_rate else None, short_margin_amount=(daily.latest_price * daily.contract_multiplier * daily.short_margin_rate) if daily.latest_price and daily.short_margin_rate else None, current_main_contract=contract_code ) db_session.add(future_info) updated_count += 1 logger.debug(f"新增期货品种 {product_code} ({daily.product_name})") # 提交更改 db_session.commit() logger.debug(f"根据期货日数据成功更新{updated_count}条期货基础信息，{not_found_count}个期货未找到对应数据") return updated_count except Exception as e: logger.error(f"根据期货日数据更新期货基础信息失败: {str(e)}") db_session.rollback() return 0 def update_future_info(self, db_session, FutureInfo): """ 更新数据库中的期货基础信息 (直接从网页获取) 参数: db_session: 数据库会话 FutureInfo: 期货基础信息模型类 返回: 更新的记录数量 """ try: # 获取数据 df = self.fetch_future_daily() if df is None or df.empty: logger.error("无法更新期货基础信息: 未获取到数据") return 0 # 获取当前数据库中的所有期货基础信息 futures = {f.contract_letter: f for f in db_session.query(FutureInfo).all()} # 更新计数器 updated_count = 0 # 遍历DataFrame中的每一行 for _, row in df.iterrows(): try: # 提取合约字母 contract_code = row.get('合约代码', '') if not contract_code: continue # 假设合约代码的前1-2位是合约字母 contract_letter = ''.join([c for c in contract_code if c.isalpha()]) # 如果合约字母在数据库中存在，则更新相应字段 if contract_letter in futures: future = futures[contract_letter] # 更新字段 future.exchange = row.get('交易所', '') future.contract_multiplier = self._safe_float(row.get('合约乘数', 0)) future.long_margin_rate = self._safe_float(row.get('做多保证金率', 0)) future.short_margin_rate = self._safe_float(row.get('做空保证金率', 0)) future.open_fee = self._safe_float(row.get('开仓费用/手', 0)) future.close_fee = self._safe_float(row.get('平仓费用/手', 0)) future.close_today_rate = self._safe_float(row.get('平今费率', 0)) future.close_today_fee = self._safe_float(row.get('平今费用/手', 0)) # 更新保证金金额字段（从网站的"做多1手保证金"和"做空1手保证金"） future.long_margin_amount = self._safe_float(row.get('做多1手保证金', 0)) future.short_margin_amount = self._safe_float(row.get('做空1手保证金', 0)) # 如果网站字段名不同，尝试其他可能的字段名 if future.long_margin_amount is None: future.long_margin_amount = self._safe_float(row.get('1手做多保证金', 0)) if future.short_margin_amount is None: future.short_margin_amount = self._safe_float(row.get('1手做空保证金', 0)) # 如果是主连合约，更新主力合约字段 if row.get('is_main_contract', False): future.current_main_contract = contract_code updated_count += 1 except Exception as e: logger.error(f"更新单个期货信息失败: {str(e)}") continue # 提交更改 db_session.commit() logger.debug(f"成功更新{updated_count}条期货基础信息") return updated_count except Exception as e: logger.error(f"更新期货基础信息失败: {str(e)}") db_session.rollback() return 0 def _safe_float(self, value): """安全地转换为浮点数""" try: if pd.isna(value): return None return float(value) except (ValueError, TypeError): return None def _safe_int(self, value): """安全地转换为整数""" try: if pd.isna(value): return None return int(value) except (ValueError, TypeError): return None