Python爬虫实战：利用最新技术抓取GDP、CPI等宏观经济数据-洪萨配资

一、宏观经济数据爬虫：技术演进与实现方案

在当今数据驱动的经济分析时代，获取准确的宏观经济指标如GDP（国内生产总值）、CPI（消费者物价指数）等对于投资决策、政策分析和市场预测至关重要。传统的手动收集方式效率低下，而Python爬虫技术为我们提供了自动化解决方案。本文将详细介绍如何使用最新的Python技术栈构建一个稳健的宏观经济数据抓取系统。

二、技术选型：现代Python爬虫工具栈

2.1 核心技术框架

异步处理：使用aiohttp和asyncio实现高性能并发请求
动态渲染：Playwright处理JavaScript动态加载内容
数据解析：BeautifulSoup4 + Parsel双解析引擎
反爬对抗：使用Rotating User-Agents、代理池和请求延迟策略
数据存储：SQLAlchemy ORM + PostgreSQL/MySQL，兼顾CSV备份

2.2 创新特性

智能重试机制：指数退避算法的请求重试策略
验证码识别集成：对接第三方OCR服务（可选）
数据验证管道：异常值检测和格式校验
监控告警：Prometheus指标收集和异常通知

三、实战代码：宏观经济数据采集系统

3.1 项目结构设计

text

macro_economic_crawler/ │ ├── crawlers/ # 爬虫核心模块 │ ├── base.py # 基类定义 │ ├── gdp_crawler.py # GDP数据爬虫 │ ├── cpi_crawler.py # CPI数据爬虫 │ └── factory.py # 爬虫工厂 │ ├── models/ # 数据模型 │ ├── economic_data.py # ORM模型 │ └── schemas.py # Pydantic验证模型 │ ├── utils/ # 工具模块 │ ├── proxy_manager.py # 代理管理 │ ├── user_agents.py # UA管理 │ └── captcha_solver.py # 验证码处理 │ ├── storage/ # 存储模块 │ ├── database.py # 数据库操作 │ └── file_store.py # 文件存储 │ ├── config/ # 配置文件 │ └── settings.py # 全局配置 │ └── main.py # 主程序入口

3.2 基础爬虫类实现

python

""" 基础爬虫类 - 提供通用功能和方法 """ import asyncio import aiohttp import random import logging from typing import Dict, List, Any, Optional from dataclasses import dataclass from abc import ABC, abstractmethod from contextlib import asynccontextmanager import time # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) @dataclass class RequestConfig: """请求配置类""" timeout: int = 30 retry_times: int = 3 delay_range: tuple = (1, 3) use_proxy: bool = True headers: Dict[str, str] = None class BaseCrawler(ABC): """爬虫基类""" def __init__(self, config: RequestConfig = None): self.config = config or RequestConfig() self.session: Optional[aiohttp.ClientSession] = None self.proxy_pool = [] # 代理池 self.user_agents = [] # User-Agent池 async def __aenter__(self): await self.init_session() return self async def __aexit__(self, exc_type, exc_val, exc_tb): await self.close_session() async def init_session(self): """初始化aiohttp会话""" timeout = aiohttp.ClientTimeout(total=self.config.timeout) connector = aiohttp.TCPConnector(limit=100, limit_per_host=20) self.session = aiohttp.ClientSession( timeout=timeout, connector=connector, headers=self._get_headers() ) async def close_session(self): """关闭会话""" if self.session: await self.session.close() def _get_headers(self) -> Dict[str, str]: """获取请求头""" base_headers = { 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', 'Accept-Encoding': 'gzip, deflate, br', 'Connection': 'keep-alive', 'Upgrade-Insecure-Requests': '1', 'Cache-Control': 'max-age=0', } if self.config.headers: base_headers.update(self.config.headers) return base_headers async def fetch(self, url: str, method: str = 'GET', **kwargs) -> Optional[str]: """执行HTTP请求""" for attempt in range(self.config.retry_times): try: await self.random_delay() # 构建请求参数 request_kwargs = self._build_request_kwargs(**kwargs) async with self.session.request(method, url, **request_kwargs) as response: if response.status == 200: content = await response.text() logger.info(f"成功获取 {url}，状态码: {response.status}") return content else: logger.warning(f"请求失败 {url}，状态码: {response.status}") except Exception as e: logger.error(f"请求异常 {url}，尝试 {attempt + 1}/{self.config.retry_times}: {str(e)}") if attempt < self.config.retry_times - 1: await asyncio.sleep(2 ** attempt) # 指数退避 return None def _build_request_kwargs(self, **kwargs) -> Dict[str, Any]: """构建请求参数""" request_kwargs = {} # 添加代理 if self.config.use_proxy and self.proxy_pool: proxy = random.choice(self.proxy_pool) request_kwargs['proxy'] = proxy # 添加User-Agent if self.user_agents: headers = kwargs.get('headers', {}) headers['User-Agent'] = random.choice(self.user_agents) kwargs['headers'] = headers request_kwargs.update(kwargs) return request_kwargs async def random_delay(self): """随机延迟""" if self.config.delay_range: delay = random.uniform(*self.config.delay_range) await asyncio.sleep(delay) @abstractmethod async def crawl(self, **kwargs) -> List[Dict[str, Any]]: """爬取数据的具体实现""" pass @abstractmethod def parse(self, html: str) -> List[Dict[str, Any]]: """解析HTML内容""" pass

3.3 GDP数据爬虫实现

python

""" GDP数据爬虫 - 从多个数据源抓取GDP数据 """ import json import pandas as pd from datetime import datetime from bs4 import BeautifulSoup import re from typing import List, Dict, Any import asyncio class GDPCrawler(BaseCrawler): """GDP数据爬虫""" def __init__(self, config: RequestConfig = None): super().__init__(config) self.sources = { '国家统计局': 'http://data.stats.gov.cn/easyquery.htm', '世界银行': 'https://api.worldbank.org/v2/country/CHN/indicator/NY.GDP.MKTP.CD', 'IMF': 'https://www.imf.org/external/datamapper/api/v1/PCPIPCH' } async def crawl(self, start_year: int = 2000, end_year: int = 2023) -> List[Dict[str, Any]]: """爬取GDP数据""" all_data = [] # 并发抓取多个数据源 tasks = [ self._crawl_national_stats(start_year, end_year), self._crawl_world_bank(), self._crawl_imf() ] results = await asyncio.gather(*tasks, return_exceptions=True) for result in results: if isinstance(result, list): all_data.extend(result) return all_data async def _crawl_national_stats(self, start_year: int, end_year: int) -> List[Dict[str, Any]]: """从国家统计局抓取数据""" data = [] # 模拟统计局API请求参数 params = { 'id': 'A01', 'dbcode': 'hgnd', 'wdcode': 'zb', 'm': 'getTree' } # 构建数据查询参数 query_params = { 'm': 'QueryData', 'dbcode': 'hgnd', 'rowcode': 'zb', 'colcode': 'sj', 'wds': '[]', 'dfwds': '[{"wdcode":"zb","valuecode":"A0201"}]', 'k1': str(int(time.time() * 1000)) } try: # 获取指标列表 async with self.session.post(self.sources['国家统计局'], params=params) as response: if response.status == 200: zb_list = await response.json() # 获取GDP数据 for zb in zb_list[:5]: # 取前5个指标 query_params['dfwds'] = f'[{{"wdcode":"zb","valuecode":"{zb["id"]}"}}]' async with self.session.get( self.sources['国家统计局'], params=query_params ) as data_response: if data_response.status == 200: result = await data_response.json() data.extend(self._parse_national_stats(result)) except Exception as e: logger.error(f"国家统计局数据抓取失败: {str(e)}") return data def _parse_national_stats(self, raw_data: Dict) -> List[Dict[str, Any]]: """解析国家统计局数据""" parsed_data = [] try: if 'returndata' in raw_data: data_list = raw_data['returndata'].get('datanodes', []) for item in data_list: data_point = { 'indicator': item['wds'][0]['valuecode'], 'year': item['wds'][1]['valuecode'], 'value': item['data']['data'], 'unit': item['data']['unit'], 'source': '国家统计局', 'update_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S') } parsed_data.append(data_point) except Exception as e: logger.error(f"数据解析失败: {str(e)}") return parsed_data async def _crawl_world_bank(self) -> List[Dict[str, Any]]: """从世界银行API获取数据""" data = [] try: url = self.sources['世界银行'] params = { 'format': 'json', 'per_page': 100, 'date': '2000:2023' } content = await self.fetch(url, params=params) if content: wb_data = json.loads(content) if len(wb_data) > 1: indicators = wb_data[1] for item in indicators: if item.get('value') is not None: data_point = { 'indicator': 'GDP (current US$)', 'year': item['date'], 'value': item['value'], 'unit': '美元', 'source': '世界银行', 'country': item['country']['value'], 'update_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S') } data.append(data_point) except Exception as e: logger.error(f"世界银行数据抓取失败: {str(e)}") return data async def _crawl_imf(self) -> List[Dict[str, Any]]: """从IMF获取数据""" data = [] try: content = await self.fetch(self.sources['IMF']) if content: # 解析IMF数据 imf_data = json.loads(content) # 简化的解析逻辑，实际需要根据IMF数据结构调整 for country, indicators in imf_data.get('countries', {}).items(): if 'NGDP_R' in indicators: # 实际GDP for year, value in indicators['NGDP_R'].items(): if value is not None: data_point = { 'indicator': 'Real GDP', 'year': year, 'value': value, 'unit': '本币单位', 'source': 'IMF', 'country': country, 'update_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S') } data.append(data_point) except Exception as e: logger.error(f"IMF数据抓取失败: {str(e)}") return data def parse(self, html: str) -> List[Dict[str, Any]]: """解析HTML内容（如果需要）""" # 这个方法可以根据具体网页结构实现 pass

3.4 CPI数据爬虫实现

python

""" CPI数据爬虫 - 抓取消费者价格指数数据 """ from typing import List, Dict, Any import asyncio from datetime import datetime, timedelta import json class CPICrawler(BaseCrawler): """CPI数据爬虫""" def __init__(self, config: RequestConfig = None): super().__init__(config) self.sources = { '中国统计局': 'http://www.stats.gov.cn/sj/zxfb/index.html', '美联储': 'https://fred.stlouisfed.org/series/CPIAUCSL', 'OECD': 'https://stats.oecd.org/SDMX-JSON/data/PRICES_CPI/' } async def crawl(self, months: int = 12) -> List[Dict[str, Any]]: """爬取最近N个月的CPI数据""" all_data = [] tasks = [ self._crawl_china_cpi(months), self._crawl_fred_cpi(), self._crawl_oecd_cpi() ] results = await asyncio.gather(*tasks, return_exceptions=True) for result in results: if isinstance(result, list): all_data.extend(result) return all_data async def _crawl_china_cpi(self, months: int) -> List[Dict[str, Any]]: """从中国统计局抓取CPI数据""" data = [] try: # 使用Playwright处理动态内容 from playwright.async_api import async_playwright async with async_playwright() as p: browser = await p.chromium.launch(headless=True) context = await browser.new_context( user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" ) page = await context.new_page() await page.goto(self.sources['中国统计局'], wait_until='networkidle') # 等待数据加载 await page.wait_for_selector('.data-table', timeout=30000) # 提取表格数据 table_data = await page.evaluate('''() => { const rows = document.querySelectorAll('.data-table tr'); const data = []; rows.forEach(row => { const cells = row.querySelectorAll('td'); if (cells.length >= 3) { data.push({ month: cells[0].innerText.trim(), cpi: cells[1].innerText.trim(), yoy: cells[2].innerText.trim() }); } }); return data; }''') # 转换为标准格式 for item in table_data[:months]: data_point = { 'indicator': 'CPI', 'period': item['month'], 'value': float(item['cpi']) if item['cpi'].replace('.', '').isdigit() else None, 'yoy_change': item['yoy'], 'source': '中国统计局', 'update_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S') } data.append(data_point) await browser.close() except ImportError: logger.warning("未安装playwright，使用备用方案") # 备用方案：使用API或静态页面解析 except Exception as e: logger.error(f"中国CPI数据抓取失败: {str(e)}") return data async def _crawl_fred_cpi(self) -> List[Dict[str, Any]]: """从美联储经济数据库获取CPI数据""" data = [] try: # FRED API端点 api_url = "https://api.stlouisfed.org/fred/series/observations" params = { 'series_id': 'CPIAUCSL', 'api_key': 'your_api_key_here', # 需要申请API密钥 'file_type': 'json', 'observation_start': '2020-01-01', 'observation_end': datetime.now().strftime('%Y-%m-%d') } content = await self.fetch(api_url, params=params) if content: fred_data = json.loads(content) for observation in fred_data.get('observations', []): data_point = { 'indicator': 'CPI (Urban Consumers)', 'date': observation['date'], 'value': float(observation['value']) if observation['value'] != '.' else None, 'source': 'FRED', 'country': 'USA', 'update_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S') } data.append(data_point) except Exception as e: logger.error(f"FRED CPI数据抓取失败: {str(e)}") return data async def _crawl_oecd_cpi(self) -> List[Dict[str, Any]]: """从OECD获取CPI数据""" data = [] try: # OECD API端点 api_url = "https://stats.oecd.org/SDMX-JSON/data/PRICES_CPI/ALL/all" params = { 'startTime': '2020', 'endTime': '2023', 'dimensionAtObservation': 'AllDimensions' } content = await self.fetch(api_url, params=params) if content: oecd_data = json.loads(content) # 解析OECD的SDMX-JSON格式 datasets = oecd_data.get('dataSets', []) for dataset in datasets: observations = dataset.get('observations', {}) for key, value in observations.items(): # 解析维度信息 dimensions = key.split(':') if len(dimensions) >= 3: data_point = { 'indicator': 'CPI', 'country': dimensions[0], 'year': dimensions[1], 'value': value[0] if value else None, 'source': 'OECD', 'update_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S') } data.append(data_point) except Exception as e: logger.error(f"OECD CPI数据抓取失败: {str(e)}") return data def parse(self, html: str) -> List[Dict[str, Any]]: """解析HTML内容""" # 根据具体网页结构实现 pass

3.5 数据存储模块

python

""" 数据存储模块 - 支持多种存储方式 """ import pandas as pd from sqlalchemy import create_engine, Column, String, Float, DateTime, JSON from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from typing import List, Dict, Any import json from datetime import datetime Base = declarative_base() class EconomicData(Base): """经济数据模型""" __tablename__ = 'economic_data' id = Column(String(50), primary_key=True) indicator = Column(String(100), nullable=False) value = Column(Float) unit = Column(String(50)) period = Column(String(20)) country = Column(String(50)) source = Column(String(100)) metadata = Column(JSON) created_at = Column(DateTime, default=datetime.now) updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now) def __repr__(self): return f"<EconomicData(indicator='{self.indicator}', period='{self.period}', value={self.value})>" class DataStorage: """数据存储管理器""" def __init__(self, db_url: str = None, backup_dir: str = './data'): self.db_url = db_url or 'sqlite:///economic_data.db' self.backup_dir = backup_dir self.engine = None self.Session = None def init_database(self): """初始化数据库""" self.engine = create_engine(self.db_url, echo=False) Base.metadata.create_all(self.engine) self.Session = sessionmaker(bind=self.engine) logger.info(f"数据库已初始化: {self.db_url}") def save_to_database(self, data_list: List[Dict[str, Any]]): """保存数据到数据库""" if not self.engine: self.init_database() session = self.Session() try: for data_dict in data_list: # 创建唯一ID data_id = f"{data_dict.get('indicator', '')}_{data_dict.get('period', '')}_{data_dict.get('country', '')}" data_id = data_id.replace(' ', '_').lower() # 检查是否已存在 existing = session.query(EconomicData).filter_by(id=data_id).first() if existing: # 更新现有记录 for key, value in data_dict.items(): if hasattr(existing, key): setattr(existing, key, value) existing.updated_at = datetime.now() else: # 创建新记录 economic_data = EconomicData( id=data_id, **{k: v for k, v in data_dict.items() if hasattr(EconomicData, k)} ) session.add(economic_data) session.commit() logger.info(f"成功保存 {len(data_list)} 条数据到数据库") except Exception as e: session.rollback() logger.error(f"数据库保存失败: {str(e)}") finally: session.close() def save_to_csv(self, data_list: List[Dict[str, Any]], filename: str = None): """保存数据到CSV文件""" if not filename: timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') filename = f"economic_data_{timestamp}.csv" filepath = f"{self.backup_dir}/{filename}" try: df = pd.DataFrame(data_list) df.to_csv(filepath, index=False, encoding='utf-8-sig') logger.info(f"数据已保存到CSV: {filepath}") except Exception as e: logger.error(f"CSV保存失败: {str(e)}") def save_to_json(self, data_list: List[Dict[str, Any]], filename: str = None): """保存数据到JSON文件""" if not filename: timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') filename = f"economic_data_{timestamp}.json" filepath = f"{self.backup_dir}/{filename}" try: with open(filepath, 'w', encoding='utf-8') as f: json.dump(data_list, f, ensure_ascii=False, indent=2, default=str) logger.info(f"数据已保存到JSON: {filepath}") except Exception as e: logger.error(f"JSON保存失败: {str(e)}")

3.6 主程序与调度器

python

""" 主程序 - 爬虫调度和数据采集 """ import asyncio import schedule import time from datetime import datetime from typing import List import argparse import yaml class MacroEconomicCrawler: """宏观经济数据爬虫主程序""" def __init__(self, config_file: str = 'config.yaml'): self.config = self.load_config(config_file) self.crawlers = {} self.storage = DataStorage( db_url=self.config.get('database_url'), backup_dir=self.config.get('backup_dir', './data') ) def load_config(self, config_file: str) -> dict: """加载配置文件""" try: with open(config_file, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) return config or {} except FileNotFoundError: logger.warning(f"配置文件 {config_file} 未找到，使用默认配置") return { 'database_url': 'sqlite:///economic_data.db', 'backup_dir': './data', 'crawlers': ['gdp', 'cpi'], 'schedule': {'hour': 2, 'minute': 0} # 每天凌晨2点运行 } def init_crawlers(self): """初始化爬虫实例""" from crawlers.gdp_crawler import GDPCrawler from crawlers.cpi_crawler import CPICrawler crawler_config = RequestConfig( timeout=30, retry_times=3, delay_range=(1, 3), use_proxy=True ) if 'gdp' in self.config.get('crawlers', []): self.crawlers['gdp'] = GDPCrawler(crawler_config) if 'cpi' in self.config.get('crawlers', []): self.crawlers['cpi'] = CPICrawler(crawler_config) async def run_crawlers(self): """运行所有爬虫""" all_data = [] logger.info("开始宏观经济数据抓取...") start_time = time.time() for name, crawler in self.crawlers.items(): try: logger.info(f"开始抓取 {name} 数据...") async with crawler: if name == 'gdp': data = await crawler.crawl(start_year=2000, end_year=2023) elif name == 'cpi': data = await crawler.crawl(months=24) else: data = await crawler.crawl() all_data.extend(data) logger.info(f"{name} 数据抓取完成，获取 {len(data)} 条记录") except Exception as e: logger.error(f"{name} 爬虫运行失败: {str(e)}") # 保存数据 if all_data: self.storage.save_to_database(all_data) self.storage.save_to_csv(all_data) self.storage.save_to_json(all_data) logger.info(f"数据采集完成，总共获取 {len(all_data)} 条记录") else: logger.warning("未获取到任何数据") elapsed_time = time.time() - start_time logger.info(f"总耗时: {elapsed_time:.2f} 秒") return all_data def schedule_job(self): """调度任务""" schedule_time = self.config.get('schedule', {}) # 设置定时任务 if 'hour' in schedule_time and 'minute' in schedule_time: schedule.every().day.at( f"{schedule_time['hour']:02d}:{schedule_time['minute']:02d}" ).do(lambda: asyncio.run(self.run_crawlers())) else: # 默认每天凌晨2点运行 schedule.every().day.at("02:00").do(lambda: asyncio.run(self.run_crawlers())) logger.info("爬虫调度器已启动") while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次 async def run_once(self): """运行一次""" self.init_crawlers() return await self.run_crawlers() def main(): """主函数""" parser = argparse.ArgumentParser(description='宏观经济数据爬虫') parser.add_argument('--mode', choices=['once', 'schedule'], default='once', help='运行模式: once(运行一次) 或 schedule(定时运行)') parser.add_argument('--config', default='config.yaml', help='配置文件路径') args = parser.parse_args() crawler = MacroEconomicCrawler(args.config) if args.mode == 'once': # 运行一次 asyncio.run(crawler.run_once()) else: # 定时运行 crawler.init_crawlers() crawler.schedule_job() if __name__ == '__main__': main()

3.7 配置文件示例

yaml

# config.yaml database_url: "postgresql://user:password@localhost:5432/economic_data" # database_url: "mysql://user:password@localhost:3306/economic_data" # database_url: "sqlite:///economic_data.db" backup_dir: "./data" crawlers: - "gdp" - "cpi" schedule: hour: 2 minute: 0 proxy: enabled: true providers: - "http://proxy-provider1.com/api" - "http://proxy-provider2.com/api" api_keys: fred: "your_fred_api_key_here" world_bank: "your_world_bank_api_key_here" retry_config: max_retries: 3 backoff_factor: 2 timeout: 30

3.8 依赖文件

txt

# requirements.txt aiohttp>=3.8.0 beautifulsoup4>=4.11.0 pandas>=1.5.0 sqlalchemy>=2.0.0 playwright>=1.30.0 schedule>=1.2.0 pyyaml>=6.0 pydantic>=2.0.0 asyncio>=3.4.3 requests>=2.28.0 nest-asyncio>=1.5.0

四、高级特性与优化建议

4.1 反爬虫策略应对

python

""" 反爬虫策略管理器 """ class AntiAntiCrawler: """反反爬虫策略""" @staticmethod def rotate_user_agents(): """轮换User-Agent""" agents = [ 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15', 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36', 'Mozilla/5.0 (iPhone; CPU iPhone OS 14_0 like Mac OS X) AppleWebKit/605.1.15' ] return random.choice(agents) @staticmethod def simulate_human_behavior(page): """模拟人类行为""" # 随机滚动 scroll_times = random.randint(1, 3) for _ in range(scroll_times): await page.evaluate(f"window.scrollBy(0, {random.randint(100, 500)})") await asyncio.sleep(random.uniform(0.5, 2))

4.2 数据质量验证

python

""" 数据验证模块 """ class DataValidator: """数据验证器""" @staticmethod def validate_economic_data(data: Dict[str, Any]) -> bool: """验证经济数据""" required_fields = ['indicator', 'value', 'period'] # 检查必填字段 for field in required_fields: if field not in data or data[field] is None: return False # 验证数据类型 try: float(data['value']) except (ValueError, TypeError): return False # 验证数值范围（根据指标类型） if data['indicator'] == 'GDP': if data['value'] < 0 or data['value'] > 1e15: # 合理的GDP范围 return False elif data['indicator'] == 'CPI': if data['value'] < 0 or data['value'] > 1000: # 合理的CPI范围 return False return True @staticmethod def detect_outliers(data_series: List[float]) -> List[int]: """检测异常值""" import numpy as np if len(data_series) < 3: return [] q1 = np.percentile(data_series, 25) q3 = np.percentile(data_series, 75) iqr = q3 - q1 lower_bound = q1 - 1.5 * iqr upper_bound = q3 + 1.5 * iqr outliers = [] for i, value in enumerate(data_series): if value < lower_bound or value > upper_bound: outliers.append(i) return outliers

五、部署与监控

5.1 Docker部署配置

dockerfile

# Dockerfile FROM python:3.11-slim WORKDIR /app # 安装系统依赖 RUN apt-get update && apt-get install -y \ wget \ gnupg \ && wget -q -O - https://dl-ssl.google.com/linux/linux_signing_key.pub | apt-key add - \ && echo "deb http://dl.google.com/linux/chrome/deb/ stable main" >> /etc/apt/sources.list.d/google.list \ && apt-get update && apt-get install -y \ google-chrome-stable \ fonts-ipafont-gothic \ fonts-wqy-zenhei \ fonts-thai-tlwg \ fonts-kacst \ fonts-symbola \ fonts-noto \ fonts-freefont-ttf \ --no-install-recommends \ && rm -rf /var/lib/apt/lists/* # 安装Python依赖 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 安装Playwright浏览器 RUN playwright install chromium # 复制应用代码 COPY . . # 创建数据目录 RUN mkdir -p /app/data # 运行应用 CMD ["python", "main.py", "--mode", "schedule"]

5.2 Prometheus监控配置

python

""" 监控模块 """ from prometheus_client import start_http_server, Counter, Gauge, Histogram import time class CrawlerMetrics: """爬虫监控指标""" def __init__(self, port=9090): self.port = port # 定义指标 self.requests_total = Counter( 'crawler_requests_total', 'Total number of requests', ['endpoint', 'status'] ) self.data_points_collected = Counter( 'crawler_data_points_total', 'Total number of data points collected', ['indicator'] ) self.crawler_duration = Histogram( 'crawler_duration_seconds', 'Crawler execution duration', ['crawler_name'] ) self.crawler_errors = Counter( 'crawler_errors_total', 'Total number of crawler errors', ['crawler_name', 'error_type'] ) # 启动Prometheus HTTP服务器 start_http_server(self.port) def record_request(self, endpoint: str, status: str): """记录请求""" self.requests_total.labels(endpoint=endpoint, status=status).inc() def record_data_point(self, indicator: str, count: int = 1): """记录数据点""" self.data_points_collected.labels(indicator=indicator).inc(count) def record_crawler_duration(self, crawler_name: str, duration: float): """记录爬虫执行时间""" self.crawler_duration.labels(crawler_name=crawler_name).observe(duration) def record_error(self, crawler_name: str, error_type: str): """记录错误""" self.crawler_errors.labels(crawler_name=crawler_name, error_type=error_type).inc()