#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 农业股票数据采集器 根据项目需求从腾讯行情接口采集农业类上市公司数据 """ import requests import pandas as pd import json import time import logging import schedule from datetime import datetime, timedelta from typing import List, Dict, Optional import mysql.connector # Kafka导入已移除 import configparser # 日志配置 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('stock_crawler.log', encoding='utf-8'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class StockCrawler: """股票数据爬虫类""" def __init__(self, config_file='config.ini'): """初始化爬虫""" self.config = self.load_config(config_file) self.session = requests.Session() self.session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36' }) # 农业类股票代码列表（示例） self.agricultural_stocks = [ 'sz300189', # 神农种业 'sz000713', # 丰乐钟业 'sh600313', # 农发种业 'bj837403', # 康农种业 'sz200505', # 京粮B 'sz300268', # st佳沃 'sz000930', # 中粮科技 'sz002299', # 圣农发展 'sh600371', # 万向德农 'sh600598', # 北大荒 'sh603609', # 禾丰股份 'bj831087', 'sh603363', 'sh603336', 'sh600354', 'sz002385', 'sz000048', 'sh600251', 'sz002321', 'sz000505', 'sz001366', 'sz002772', 'sz002041', 'sh600127', 'sz002548', 'sh605296', 'sh600359', 'sh601952', 'sh600975', # 新增股票代码 'sz300505', 'sh600141', 'sz000893', 'sh600691', 'sh600470', 'sz000408', 'sz000902', 'sz002539', 'sz002599', 'sz002545', 'sz300387', 'sz002274', 'sz002470', 'sz002538', 'sz000731', 'sh600078', 'sh600227', 'sh603395', 'sz002588', 'sz000422', 'sz002556', 'sz002312', 'sz002170', 'sz000792', 'sh600096', # 农药兽药类股票 'sz001231', 'sz002731', 'sz200553', 'sh603810', 'sh603970', 'sz301035', 'sz002868', 'sz003042', 'sz002391', 'bj870866', 'sz301665', 'sh600486', ] # 初始化数据库连接 self.init_db_connection() # Kafka生产者初始化已移除 def load_config(self, config_file: str) -> configparser.ConfigParser: """加载配置文件""" config = configparser.ConfigParser() try: config.read(config_file, encoding='utf-8') except FileNotFoundError: logger.warning(f"配置文件 {config_file} 不存在，使用默认配置") except UnicodeDecodeError: logger.warning(f"配置文件编码错误，使用默认配置") return config def init_db_connection(self): """初始化数据库连接""" try: self.db_connection = mysql.connector.connect( host=self.config.get('database', 'host', fallback='localhost'), port=self.config.getint('database', 'port', fallback=3306), user=self.config.get('database', 'user', fallback='root'), password=self.config.get('database', 'password', fallback='123456'), database=self.config.get('database', 'database', fallback='agricultural_stock'), charset='utf8mb4' ) logger.info("数据库连接成功") except Exception as e: logger.error(f"数据库连接失败: {e}") self.db_connection = None # Kafka生产者初始化方法已移除 def fetch_stock_data(self, stock_code: str) -> Optional[Dict]: """ 从腾讯行情接口获取单个股票数据 Args: stock_code: 股票代码，格式如 'sz300630' 或 'sh600998' Returns: 股票数据字典或None """ url = f"http://qt.gtimg.cn/q={stock_code}" try: response = self.session.get(url, timeout=10) response.raise_for_status() # 解析响应数据 content = response.text.strip() if not content or 'v_' not in content: logger.warning(f"股票 {stock_code} 无数据返回") return None # 提取数据部分 data_part = content.split('="')[1].split('";')[0] fields = data_part.split('~') if len(fields) < 50: logger.warning(f"股票 {stock_code} 数据字段不完整") return None # 构造股票数据字典 stock_data = { 'stock_code': stock_code, 'stock_name': fields[1], 'current_price': float(fields[3]) if fields[3] else 0.0, 'yesterday_close': float(fields[4]) if fields[4] else 0.0, 'open_price': float(fields[5]) if fields[5] else 0.0, 'volume': int(fields[6]) if fields[6] else 0, 'outer_volume': int(fields[7]) if fields[7] else 0, 'inner_volume': int(fields[8]) if fields[8] else 0, 'buy1_price': float(fields[9]) if fields[9] else 0.0, 'buy1_volume': int(fields[10]) if fields[10] else 0, 'buy2_price': float(fields[11]) if fields[11] else 0.0, 'buy2_volume': int(fields[12]) if fields[12] else 0, 'buy3_price': float(fields[13]) if fields[13] else 0.0, 'buy3_volume': int(fields[14]) if fields[14] else 0, 'buy4_price': float(fields[15]) if fields[15] else 0.0, 'buy4_volume': int(fields[16]) if fields[16] else 0, 'buy5_price': float(fields[17]) if fields[17] else 0.0, 'buy5_volume': int(fields[18]) if fields[18] else 0, 'sell1_price': float(fields[19]) if fields[19] else 0.0, 'sell1_volume': int(fields[20]) if fields[20] else 0, 'sell2_price': float(fields[21]) if fields[21] else 0.0, 'sell2_volume': int(fields[22]) if fields[22] else 0, 'sell3_price': float(fields[23]) if fields[23] else 0.0, 'sell3_volume': int(fields[24]) if fields[24] else 0, 'sell4_price': float(fields[25]) if fields[25] else 0.0, 'sell4_volume': int(fields[26]) if fields[26] else 0, 'sell5_price': float(fields[27]) if fields[27] else 0.0, 'sell5_volume': int(fields[28]) if fields[28] else 0, 'latest_deals': fields[29], 'trade_time': fields[30], 'change_amount': float(fields[31]) if fields[31] else 0.0, 'change_percent': float(fields[32]) if fields[32] else 0.0, 'high_price': float(fields[33]) if fields[33] else 0.0, 'low_price': float(fields[34]) if fields[34] else 0.0, 'price_volume_ratio': fields[35], 'volume_ratio': fields[36], 'turnover_rate': float(fields[37]) if fields[37] else 0.0, 'pe_ratio': float(fields[38]) if fields[38] else 0.0, 'pb_ratio': float(fields[46]) if len(fields) > 46 and fields[46] else 0.0, 'market_cap': float(fields[44]) if len(fields) > 44 and fields[44] else 0.0, 'float_market_cap': float(fields[45]) if len(fields) > 45 and fields[45] else 0.0, 'crawl_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S') } # 计算派生字段 if stock_data['current_price'] > 0 and stock_data['yesterday_close'] > 0: stock_data['change_amount'] = stock_data['current_price'] - stock_data['yesterday_close'] stock_data['change_percent'] = (stock_data['change_amount'] / stock_data['yesterday_close']) * 100 # 计算成交额 stock_data['turnover'] = stock_data['volume'] * stock_data['current_price'] logger.info(f"成功获取股票 {stock_code} 数据: {stock_data['stock_name']}") return stock_data except requests.exceptions.RequestException as e: logger.error(f"请求股票 {stock_code} 数据失败: {e}") return None except (ValueError, IndexError) as e: logger.error(f"解析股票 {stock_code} 数据失败: {e}") return None except Exception as e: logger.error(f"获取股票 {stock_code} 数据时出现未知错误: {e}") return None def fetch_all_stocks(self) -> List[Dict]: """获取所有农业股票数据""" all_data = [] for stock_code in self.agricultural_stocks: try: stock_data = self.fetch_stock_data(stock_code) if stock_data: all_data.append(stock_data) # 避免请求过于频繁 time.sleep(0.5) except Exception as e: logger.error(f"处理股票 {stock_code} 时出错: {e}") continue logger.info(f"本次采集完成，共获取 {len(all_data)} 只股票数据") return all_data def is_trading_time(self) -> bool: """检查当前是否为交易时间""" now = datetime.now() # 检查是否为交易日（周一到周五） if now.weekday() >= 5: # 周六、周日 return False # 检查是否在交易时间内 current_time = now.time() morning_start = datetime.strptime("09:30", "%H:%M").time() morning_end = datetime.strptime("12:00", "%H:%M").time() afternoon_start = datetime.strptime("13:00", "%H:%M").time() afternoon_end = datetime.strptime("17:00", "%H:%M").time() return (morning_start <= current_time <= morning_end) or \ (afternoon_start <= current_time <= afternoon_end) def save_to_database(self, stock_data_list: List[Dict]): """保存数据到MySQL数据库""" if not self.db_connection or not stock_data_list: return try: cursor = self.db_connection.cursor() # 获取最新交易日期 today = datetime.now().strftime('%Y-%m-%d') for stock_data in stock_data_list: # 检查是否已有相同的交易数据（避免重复存储相同价格的数据） check_sql = """ SELECT COUNT(*) FROM stock_data WHERE stock_code = %s AND DATE(trade_date) = %s AND close_price = %s AND volume = %s """ cursor.execute(check_sql, ( stock_data['stock_code'], today, stock_data['current_price'], stock_data['volume'] )) count = cursor.fetchone()[0] if count > 0: logger.info(f"股票 {stock_data['stock_code']} 今日数据已存在且相同，跳过") continue # 检查是否有今日其他数据需要更新 update_check_sql = """ SELECT COUNT(*) FROM stock_data WHERE stock_code = %s AND DATE(trade_date) = %s """ cursor.execute(update_check_sql, (stock_data['stock_code'], today)) update_count = cursor.fetchone()[0] if update_count > 0: # 已有数据，使用UPDATE更新 update_sql = """ UPDATE stock_data SET stock_name = %(stock_name)s, open_price = %(open_price)s, close_price = %(current_price)s, high_price = %(high_price)s, low_price = %(low_price)s, volume = %(volume)s, turnover = %(turnover)s, change_percent = %(change_percent)s, change_amount = %(change_amount)s, pe_ratio = %(pe_ratio)s, pb_ratio = %(pb_ratio)s, market_cap = %(market_cap)s, float_market_cap = %(float_market_cap)s, update_time = NOW() WHERE stock_code = %(stock_code)s AND DATE(trade_date) = %(date)s """ cursor.execute(update_sql, {**stock_data, 'date': today}) logger.info(f"更新股票 {stock_data['stock_code']} 数据") else: # 无数据，使用INSERT插入 insert_sql = """ INSERT INTO stock_data ( stock_code, stock_name, open_price, close_price, high_price, low_price, volume, turnover, change_percent, change_amount, pe_ratio, pb_ratio, market_cap, float_market_cap, trade_date, create_time ) VALUES ( %(stock_code)s, %(stock_name)s, %(open_price)s, %(current_price)s, %(high_price)s, %(low_price)s, %(volume)s, %(turnover)s, %(change_percent)s, %(change_amount)s, %(pe_ratio)s, %(pb_ratio)s, %(market_cap)s, %(float_market_cap)s, NOW(), NOW() ) """ cursor.execute(insert_sql, stock_data) logger.info(f"插入股票 {stock_data['stock_code']} 数据") self.db_connection.commit() logger.info(f"成功处理 {len(stock_data_list)} 条股票数据到数据库") except Exception as e: logger.error(f"保存数据到数据库失败: {e}") if self.db_connection: self.db_connection.rollback() finally: if cursor: cursor.close() # Kafka发送方法已移除 def run_once(self): """执行一次完整的数据采集流程""" # 检查是否为交易时间 if not self.is_trading_time(): logger.info("当前非交易时间，跳过数据采集") return logger.info("开始执行股票数据采集...") # 获取股票数据 stock_data_list = self.fetch_all_stocks() if stock_data_list: # 保存到数据库 self.save_to_database(stock_data_list) # Kafka发送已移除 logger.info(f"数据采集完成，共处理 {len(stock_data_list)} 只股票") else: logger.warning("本次采集未获取到任何股票数据") def start_scheduler(self): """启动定时任务""" # 配置定时任务 interval_minutes = self.config.getint('crawler', 'interval_minutes', fallback=5) schedule.every(interval_minutes).minutes.do(self.run_once) logger.info(f"定时任务已启动，每 {interval_minutes} 分钟采集一次") # 立即执行一次 self.run_once() # 开始定时循环 while True: schedule.run_pending() time.sleep(1) def close(self): """关闭连接""" if self.db_connection: self.db_connection.close() # Kafka生产者关闭已移除 def main(): """主函数""" crawler = StockCrawler() try: logger.info("股票数据采集器启动") crawler.start_scheduler() except KeyboardInterrupt: logger.info("收到停止信号，正在关闭...") except Exception as e: logger.error(f"程序运行出错: {e}") finally: crawler.close() logger.info("股票数据采集器已停止") if __name__ == "__main__": main()