1. 从数学建模到数据获取为什么需要爬取全球疫情数据如果你正在准备数学建模竞赛或者从事相关的数据分析研究那么“数据”一定是绕不开的第一个坎。一个再精妙的模型如果没有高质量、结构化的数据支撑也只是一个空中楼阁。在涉及公共卫生、社会经济影响分析等领域的建模题目中全球COVID-19数据常常是核心的分析对象。官方数据源如约翰斯·霍普金斯大学JHU、世界卫生组织WHO等虽然权威但其提供的数据接口、文件格式可能并不直接适配你的建模环境或者你需要更灵活、定制化的数据抓取方案。这时自己动手写一个Python爬虫就成了从“被动等待数据”到“主动获取数据”的关键一步。这不仅仅是下载一个CSV文件那么简单。一个健壮的爬虫意味着你可以按需获取特定国家、特定时间范围的数据可以自动化更新你的本地数据集可以处理数据源网站结构变化带来的风险更可以将数据清洗、转换的流程与爬取过程无缝衔接形成一套可复用的数据流水线。对于建模而言时间就是生命线一个提前准备好的、稳定可靠的数据获取工具能让你在赛题发布后的第一时间就投入到核心的模型构建中而不是手忙脚乱地到处找数据。接下来我将以一个实战者的角度带你一步步构建一个用于获取全球COVID-19数据的Python爬虫并分享其中那些教科书上不会写的“坑”与技巧。2. 目标数据源分析与爬虫策略制定在动手写代码之前花时间分析数据源是至关重要的一步这直接决定了后续爬虫的稳定性、效率和合规性。对于全球疫情数据有几个主流且可靠的公开数据源。2.1 主流数据源对比与选型理由我们首先对比几个常见的数据源并说明为什么选择其中一个作为本次爬虫的目标。约翰斯·霍普金斯大学JHUCSSE数据仓库这可能是最知名、使用最广泛的数据源。其数据以GitHub仓库的形式维护提供了按国家/地区、按美国州县划分的累计确诊、死亡、康复病例的时间序列数据。优点是历史数据完整、更新及时、社区活跃。但它的数据文件CSV通常很大且数据结构如列名可能随着疫情发展而调整。世界卫生组织WHOCOVID-19数据作为权威国际机构WHO的数据具有很高的公信力。它提供每日疫情报告和数据库。然而其API或数据下载方式可能不如JHU的GitHub仓库那样对程序员友好有时需要处理PDF报告或复杂的数据库查询界面。Our World in Data (OWID)这是一个非常优秀的数据聚合与可视化项目。它不仅仅提供疫情数据还将疫苗接种、检测、政策严格性指数、社会经济指标等数据关联在一起对于进行多因素影响的建模分析极具价值。其数据通常以结构良好的CSV文件提供并且有清晰的文档。对于数学建模场景我个人的首选是OWID数据。原因如下第一数据“干净”已经过一定程度的清洗和标准化例如国家/地区名称统一减少了我们数据预处理的工作量。第二数据维度丰富除了核心的病例数、死亡数还包含住院、检测、疫苗接种等关键指标为构建更复杂的模型如SEIR模型加入干预措施、评估疫苗效果提供了可能。第三它提供完整的、每日更新的CSV文件通过一个固定的URL即可获取爬取策略简单直接。因此本爬虫将以抓取OWID的“COVID-19 - All Countries”数据集为例进行展开。2.2 爬虫策略请求、解析与存储确定了OWID作为数据源后我们需要制定具体的爬虫策略。OWID的数据文件通常托管在GitHub上我们可以直接通过其原始文件Raw链接进行访问。请求策略采用简单的HTTP GET请求即可。我们需要找到数据文件的稳定URL。例如OWID的核心疫情数据文件可能位于类似https://raw.githubusercontent.com/owid/covid-19-data/master/public/data/owid-covid-data.csv的地址。我们将使用Python的requests库来发送请求并获取数据。解析策略由于目标数据是CSV格式我们几乎不需要复杂的HTML解析。获取到数据后直接使用pandas库的read_csv函数甚至可以指定URL作为参数一步完成下载和加载为DataFrame。这比解析HTML表格要稳定和高效得多。存储策略对于建模而言我们可能需要多次运行模型进行调参和验证。因此不建议每次运行爬虫都重新从网络下载除非需要最新数据。一个良好的实践是首次运行爬虫将数据下载到本地如data/raw/owid_covid_latest.csv后续运行脚本时先检查本地文件是否存在以及其“新鲜度”例如文件修改时间是否超过24小时再决定是否重新下载。这既尊重了数据源的服务器负载也加快了本地开发调试的速度。伦理与合规策略这是必须严肃对待的一环。在爬取任何公开数据前务必查看网站的robots.txt文件例如访问https://ourworldindata.org/robots.txt和使用条款。对于OWID和JHU这类明确鼓励数据用于研究、教育目的的项目通常没有问题但我们仍应做到a) 设置合理的请求间隔如time.sleep(1)避免对服务器造成压力b) 在代码中注明数据来源c) 不将数据用于商业牟利等违反条款的用途。3. 爬虫实战从零构建可复用的数据获取模块现在我们开始动手编写代码。我将构建一个名为CovidDataFetcher的类它将封装数据获取、缓存、更新的所有逻辑方便在建模的主程序中调用。3.1 环境准备与依赖安装首先确保你的Python环境建议3.7以上中安装了必要的库。打开终端或命令提示符执行以下命令pip install requests pandasrequests用于发送HTTP请求获取网络数据。pandas数据处理的利器不仅能轻松读取CSV还能进行复杂的数据清洗、转换和分析是数学建模的必备工具。3.2 核心爬虫类设计与实现我们将创建一个Python脚本文件例如covid_data_fetcher.py。import os import time from datetime import datetime, timedelta import pandas as pd import requests class CovidDataFetcher: 一个用于获取和缓存Our World in Data全球COVID-19数据集的爬虫类。 def __init__(self, data_urlNone, cache_dir./data, cache_hours24): 初始化数据获取器。 参数: data_url (str): OWID数据文件的URL。如果为None则使用默认URL。 cache_dir (str): 本地缓存数据的目录路径。 cache_hours (int): 缓存有效的小时数。超过此时间将重新下载。 self.default_url https://raw.githubusercontent.com/owid/covid-19-data/master/public/data/owid-covid-data.csv self.data_url data_url if data_url else self.default_url self.cache_dir cache_dir self.cache_hours cache_hours self.raw_data_path os.path.join(cache_dir, raw, owid_covid_data.csv) # 确保缓存目录存在 os.makedirs(os.path.dirname(self.raw_data_path), exist_okTrue) def _is_cache_valid(self): 检查本地缓存文件是否有效存在且未过期。 if not os.path.exists(self.raw_data_path): return False file_mod_time datetime.fromtimestamp(os.path.getmtime(self.raw_data_path)) return datetime.now() - file_mod_time timedelta(hoursself.cache_hours) def _download_data(self): 从网络下载数据并保存到本地。 print(f正在从 {self.data_url} 下载数据...) try: # 添加一个简单的请求头模拟浏览器访问避免被某些服务器拒绝 headers {User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36} response requests.get(self.data_url, headersheaders, timeout30) response.raise_for_status() # 如果状态码不是200抛出HTTPError异常 # 保存原始数据 with open(self.raw_data_path, wb) as f: f.write(response.content) print(f数据已保存至: {self.raw_data_path}) return True except requests.exceptions.RequestException as e: print(f下载数据时发生错误: {e}) # 这里可以添加更复杂的错误处理例如重试机制、备用URL等 return False except Exception as e: print(f发生未知错误: {e}) return False def get_data(self, force_downloadFalse): 获取数据。优先使用本地缓存除非缓存过期或强制下载。 参数: force_download (bool): 是否强制重新下载忽略缓存。 返回: pandas.DataFrame: 包含COVID-19数据的DataFrame如果失败则返回None。 # 检查是否需要下载 need_download force_download or not self._is_cache_valid() if need_download: success self._download_data() if not success: # 如果下载失败但缓存文件存在尝试使用旧的缓存并给出警告 if os.path.exists(self.raw_data_path): print(网络下载失败将使用过期的本地缓存数据。请注意数据可能不是最新的。) else: print(无法获取数据请检查网络连接或URL。) return None else: # 下载成功等待一秒体现友好爬虫行为 time.sleep(1) else: print(使用有效的本地缓存数据。) # 加载数据到pandas DataFrame try: df pd.read_csv(self.raw_data_path) print(f数据加载成功共 {df.shape[0]} 行{df.shape[1]} 列。) # 查看前几行和列名帮助了解数据结构 print(\n数据前5行预览:) print(df.head()) print(\n数据列名:) print(df.columns.tolist()) return df except Exception as e: print(f加载CSV数据时发生错误: {e}) return None # 示例如何使用这个类 if __name__ __main__: # 创建一个数据获取器实例缓存设置为12小时 fetcher CovidDataFetcher(cache_hours12) # 获取数据会自动判断是否需要下载 covid_df fetcher.get_data() if covid_df is not None: # 在这里进行你的数据分析或建模预处理 # 例如筛选出中国的数据 china_data covid_df[covid_df[location] China] print(f\n中国数据共有 {china_data.shape[0]} 条记录。) # 查看中国最新的数据 print(china_data[[date, total_cases, total_deaths, people_vaccinated]].tail())代码关键点解析缓存机制 (_is_cache_valid): 这是提升体验的核心。它检查文件是否存在以及修改时间是否在设定的cache_hours内。避免了每次运行脚本都发起网络请求对于快速迭代的建模工作流非常友好。错误处理 (_download_data中的 try-except): 网络请求充满不确定性。我们捕获了requests库可能抛出的各种异常如连接超时、URL错误、HTTP状态码错误等并给出了友好的提示。在下载失败时如果本地有旧缓存会降级使用并警告用户保证了程序的鲁棒性。用户代理 (User-Agent): 有些服务器会拒绝没有标准浏览器标识的请求。添加一个常见的User-Agent头可以避免被简单的反爬机制拦截。延迟 (time.sleep(1)): 在成功下载后等待1秒是一个简单的“礼貌性”爬虫行为表明我们无意对服务器进行高频攻击。数据预览: 在get_data方法中加载数据后打印了数据维度、前几行和列名。这对于初次使用该数据源的人来说非常有用可以快速了解数据结构。4. 数据清洗与预处理为建模准备“干净”的输入直接从网络获取的原始数据很少能直接扔进模型。对于OWID的数据虽然已经很干净但我们仍需要根据具体的建模目标进行预处理。这一步通常在获取DataFrame之后在建模之前完成。4.1 常见的数据问题与处理假设我们的建模目标是分析几个主要国家疫情发展的时序规律并进行简单的预测。我们需要对数据进行以下处理def preprocess_data(df): 对获取的COVID-19数据进行清洗和预处理。 参数: df (pandas.DataFrame): 原始的OWID数据DataFrame。 返回: pandas.DataFrame: 处理后的DataFrame。 # 1. 处理缺失值 print(处理缺失值...) # 对于数值型列我们可以用前向填充ffill或后向填充bfill但疫情数据用0填充新增可能更合理 # 这里我们区分对待累计数据向前填充每日新增数据用0填充假设当天无新增 cols_to_ffill [total_cases, total_deaths, people_vaccinated] # 累计数据 cols_to_fill_zero [new_cases, new_deaths] # 每日新增数据 for col in cols_to_ffill: if col in df.columns: df[col] df[col].fillna(methodffill).fillna(0) # 先向前填充再对开头缺失的填0 for col in cols_to_fill_zero: if col in df.columns: df[col] df[col].fillna(0) # 2. 日期格式转换 print(转换日期格式...) df[date] pd.to_datetime(df[date]) # 3. 筛选感兴趣的国家和时期 print(筛选数据...) countries_of_interest [United States, India, Brazil, United Kingdom, China] # 假设我们分析2020-03-01之后的数据 start_date 2020-03-01 filtered_df df[(df[location].isin(countries_of_interest)) (df[date] start_date)].copy() # 4. 按国家和日期排序确保时间序列正确 filtered_df.sort_values(by[location, date], inplaceTrue) # 5. 计算一些衍生指标例如7日移动平均平滑每日新增数据的噪声 print(计算7日移动平均...) filtered_df[new_cases_smoothed] filtered_df.groupby(location)[new_cases].transform( lambda x: x.rolling(window7, min_periods1).mean() ) filtered_df[new_deaths_smoothed] filtered_df.groupby(location)[new_deaths].transform( lambda x: x.rolling(window7, min_periods1).mean() ) # 6. 重置索引 filtered_df.reset_index(dropTrue, inplaceTrue) print(f预处理完成。处理后数据形状: {filtered_df.shape}) return filtered_df # 在主程序中使用 if __name__ __main__: fetcher CovidDataFetcher() raw_df fetcher.get_data() if raw_df is not None: processed_df preprocess_data(raw_df) # 现在 processed_df 就可以用于后续的建模分析了 print(processed_df[[location, date, new_cases, new_cases_smoothed]].tail(10))预处理步骤的“为什么”缺失值处理疫情数据中某些国家在某些日期可能没有报告数据导致NaN。对于total_cases这样的累计指标用前一天的数值填充前向填充是合理的因为它代表累计值未更新。对于new_cases这样的日增指标填充为0比填充为均值或中位数更符合实际意味着当天报告的新增为0。这里有一个坑要小心区分数据缺失是“未报告”还是“真为零”。OWID的数据通常已经做了处理将缺失的日增数据标记为0或留空我们的填充策略是基于对数据背景的理解。日期转换将date列从字符串转换为datetime类型是进行任何时间序列分析的前提否则无法进行日期比较、排序、重采样等操作。数据筛选全量数据非常庞大。根据建模目标提前筛选出关键国家和时间段可以极大减少后续计算的内存占用和处理时间让分析焦点更集中。移动平均每日新增病例和死亡数据波动很大受检测量、周末报告延迟等因素影响。计算7日移动平均可以平滑掉这些“噪声”更清晰地展示疫情发展的趋势这是时间序列分析和可视化中非常常用的技巧。groupby(location)确保了移动平均是在每个国家内部独立计算的。5. 数据验证与异常检测确保爬取结果的可靠性数据爬取和预处理后绝不能直接相信数据是完全正确的。必须进行验证否则“垃圾进垃圾出”模型结果毫无意义。5.1 基础统计与合理性检查def validate_data(df): 对处理后的数据进行基本验证和异常检测。 print(\n 数据验证报告 ) # 1. 检查关键字段是否存在 essential_cols [location, date, total_cases, new_cases] missing_cols [col for col in essential_cols if col not in df.columns] if missing_cols: print(f警告缺失关键字段 {missing_cols}) else: print(关键字段完整性检查通过。) # 2. 检查数据时间范围和国家数量 print(f数据时间范围: {df[date].min()} 到 {df[date].max()}) print(f包含国家/地区数量: {df[location].nunique()}) print(f包含的国家/地区: {df[location].unique().tolist()}) # 3. 检查逻辑一致性每日新增病例之和应大致等于累计病例的增量允许由于数据修订产生的微小误差 print(\n检查累计病例与新增病例的逻辑一致性抽样...) sample_countries df[location].unique()[:3] # 抽查前3个国家 for country in sample_countries: country_df df[df[location] country].sort_values(date) # 计算通过新增病例累加出来的“计算累计值” country_df[calculated_total] country_df[new_cases].cumsum() # 与官方报告的累计值比较取最后一天 official_final country_df[total_cases].iloc[-1] calculated_final country_df[calculated_total].iloc[-1] diff_ratio abs(official_final - calculated_final) / official_final if official_final 0 else 0 print(f {country}: 官方累计 {official_final:.0f}, 计算累计 {calculated_final:.0f}, 差异比率 {diff_ratio:.2%}) if diff_ratio 0.05: # 如果差异超过5%发出警告 print(f **警告{country}的累计与新增数据可能存在较大不一致**) # 4. 检测异常值例如某天的新增病例数为负值数据修正可能导致 negative_new_cases df[df[new_cases] 0] if not negative_new_cases.empty: print(f\n发现 {len(negative_new_cases)} 条新增病例为负的记录通常是数据修正。) print(例如) print(negative_new_cases[[location, date, new_cases]].head()) # 5. 检查移动平均列的单调性7日平均应比原始数据平滑 # 可以通过简单可视化或计算方差来检查这里打印一个示例 print(\n平滑效果示例以某个国家最后10天为例:) example_country sample_countries[0] example_data df[df[location] example_country].tail(10)[[date, new_cases, new_cases_smoothed]] print(example_data.to_string(indexFalse))验证的意义字段检查防止因为数据源列名变更导致后续代码报错。逻辑一致性检查这是最核心的验证。如果new_cases的累计和与total_cases的终值相差太大说明数据在某个环节可能出了问题例如累计值被修订过但每日新增值未同步更新。OWID的数据质量很高通常差异很小但养成检查的习惯至关重要。异常值检测负的新增病例在现实数据中确实存在通常是之前某天的数据高报后来进行了向下修正。发现它们不是要删除而是要理解其含义并在建模时考虑是否需要进行处理例如将负值视为0或者使用修正后的累计数据。6. 集成到建模工作流与高级技巧一个孤立的爬虫脚本价值有限只有将其无缝嵌入到你的建模分析管道中才能发挥最大效用。6.1 构建模块化的数据管道建议将上述功能组织成一个小的数据获取与预处理包。目录结构可以如下your_model_project/ ├── data/ │ ├── raw/ # 存放爬取的原始CSV │ └── processed/ # 存放清洗后的数据可保存为pickle或feather格式加载更快 ├── src/ │ ├── data_fetcher.py # 包含CovidDataFetcher类 │ ├── data_preprocessor.py # 包含preprocess_data, validate_data函数 │ └── main.py # 主分析或建模脚本 └── requirements.txt在main.py中你可以这样调用from src.data_fetcher import CovidDataFetcher from src.data_preprocessor import preprocess_data, validate_data import matplotlib.pyplot as plt def main(): # 1. 获取数据 fetcher CovidDataFetcher(cache_dir../data, cache_hours6) raw_data fetcher.get_data() if raw_data is None: print(无法获取数据退出。) return # 2. 预处理数据 processed_data preprocess_data(raw_data) # 3. 验证数据 validate_data(processed_data) # 4. 进行简单的可视化示例 countries processed_data[location].unique() fig, axes plt.subplots(len(countries), 1, figsize(12, 3*len(countries)), sharexTrue) if len(countries) 1: axes [axes] for ax, country in zip(axes, countries): country_df processed_data[processed_data[location] country] ax.plot(country_df[date], country_df[new_cases_smoothed], label7天平均新增, linewidth2) ax.set_title(f{country} - 每日新增病例7日平滑) ax.legend() ax.grid(True, alpha0.3) plt.tight_layout() plt.savefig(../output/covid_trends.png, dpi150) plt.show() # 5. 将处理好的数据保存供后续建模使用 processed_data.to_pickle(../data/processed/covid_processed.pkl) print(\n数据预处理完成已保存至 ../data/processed/covid_processed.pkl) if __name__ __main__: main()6.2 应对数据源变更的策略公开数据源的结构或URL可能会变化。一个健壮的爬虫需要有一定的适应性。URL失效将数据源URL作为配置项甚至可以提供一个备选URL列表。在_download_data方法中如果主URL失败可以尝试备用URL。数据结构变化这是更常见的问题。OWID的列名可能会增加如增加新的疫苗指标。我们的预处理代码如果指定了列名可能会因为列不存在而报错。改进方法是在读取数据后打印所有列名并与我们预期的“必需列”进行对比。如果某些必需列缺失尝试用已知的旧列名或类似的列名进行映射或者用默认值创建该列并记录警告。使用try-except块包裹对特定列的操作。# 在preprocess_data函数开头添加列名检查 expected_cols {location, date, total_cases, new_cases, total_deaths, new_deaths} missing_cols expected_cols - set(df.columns) if missing_cols: print(f警告数据源缺少预期列 {missing_cols}。这可能导致后续处理出错。) # 尝试创建缺失的列并填充NaN或0 for col in missing_cols: df[col] 06.3 性能与扩展考量增量更新对于超大规模的历史数据每次都下载整个文件可能低效。如果数据源提供按日期或按国家查询的API可以考虑实现增量爬取只获取最新的数据并与本地历史数据合并。OWID的CSV文件虽然整体不小但对于建模场景通常可以接受全量下载。并发与速率限制如果未来需要从多个不同的API端点获取数据例如分别获取病例数据、疫苗数据、移动数据可以考虑使用concurrent.futures模块进行有限的并发请求但务必遵守目标网站的速率限制并在请求间添加延迟。数据存储格式对于处理好的数据保存为pickle或feather格式比CSV加载速度更快特别是当DataFrame很大时。这在频繁读取中间数据的建模迭代中能节省大量时间。构建这样一个爬虫其价值远不止于获取一份疫情数据。它锻炼了你从需求分析、工具选型、代码实现、错误处理到数据验证的完整数据处理能力。在数学建模竞赛中这份能力能让你在面对任何陌生的数据需求时都能快速构建出可靠的数据获取通道将更多精力专注于模型本身。 Python爬虫实战：构建全球疫情数据获取与预处理管道
1. 从数学建模到数据获取为什么需要爬取全球疫情数据如果你正在准备数学建模竞赛或者从事相关的数据分析研究那么“数据”一定是绕不开的第一个坎。一个再精妙的模型如果没有高质量、结构化的数据支撑也只是一个空中楼阁。在涉及公共卫生、社会经济影响分析等领域的建模题目中全球COVID-19数据常常是核心的分析对象。官方数据源如约翰斯·霍普金斯大学JHU、世界卫生组织WHO等虽然权威但其提供的数据接口、文件格式可能并不直接适配你的建模环境或者你需要更灵活、定制化的数据抓取方案。这时自己动手写一个Python爬虫就成了从“被动等待数据”到“主动获取数据”的关键一步。这不仅仅是下载一个CSV文件那么简单。一个健壮的爬虫意味着你可以按需获取特定国家、特定时间范围的数据可以自动化更新你的本地数据集可以处理数据源网站结构变化带来的风险更可以将数据清洗、转换的流程与爬取过程无缝衔接形成一套可复用的数据流水线。对于建模而言时间就是生命线一个提前准备好的、稳定可靠的数据获取工具能让你在赛题发布后的第一时间就投入到核心的模型构建中而不是手忙脚乱地到处找数据。接下来我将以一个实战者的角度带你一步步构建一个用于获取全球COVID-19数据的Python爬虫并分享其中那些教科书上不会写的“坑”与技巧。2. 目标数据源分析与爬虫策略制定在动手写代码之前花时间分析数据源是至关重要的一步这直接决定了后续爬虫的稳定性、效率和合规性。对于全球疫情数据有几个主流且可靠的公开数据源。2.1 主流数据源对比与选型理由我们首先对比几个常见的数据源并说明为什么选择其中一个作为本次爬虫的目标。约翰斯·霍普金斯大学JHUCSSE数据仓库这可能是最知名、使用最广泛的数据源。其数据以GitHub仓库的形式维护提供了按国家/地区、按美国州县划分的累计确诊、死亡、康复病例的时间序列数据。优点是历史数据完整、更新及时、社区活跃。但它的数据文件CSV通常很大且数据结构如列名可能随着疫情发展而调整。世界卫生组织WHOCOVID-19数据作为权威国际机构WHO的数据具有很高的公信力。它提供每日疫情报告和数据库。然而其API或数据下载方式可能不如JHU的GitHub仓库那样对程序员友好有时需要处理PDF报告或复杂的数据库查询界面。Our World in Data (OWID)这是一个非常优秀的数据聚合与可视化项目。它不仅仅提供疫情数据还将疫苗接种、检测、政策严格性指数、社会经济指标等数据关联在一起对于进行多因素影响的建模分析极具价值。其数据通常以结构良好的CSV文件提供并且有清晰的文档。对于数学建模场景我个人的首选是OWID数据。原因如下第一数据“干净”已经过一定程度的清洗和标准化例如国家/地区名称统一减少了我们数据预处理的工作量。第二数据维度丰富除了核心的病例数、死亡数还包含住院、检测、疫苗接种等关键指标为构建更复杂的模型如SEIR模型加入干预措施、评估疫苗效果提供了可能。第三它提供完整的、每日更新的CSV文件通过一个固定的URL即可获取爬取策略简单直接。因此本爬虫将以抓取OWID的“COVID-19 - All Countries”数据集为例进行展开。2.2 爬虫策略请求、解析与存储确定了OWID作为数据源后我们需要制定具体的爬虫策略。OWID的数据文件通常托管在GitHub上我们可以直接通过其原始文件Raw链接进行访问。请求策略采用简单的HTTP GET请求即可。我们需要找到数据文件的稳定URL。例如OWID的核心疫情数据文件可能位于类似https://raw.githubusercontent.com/owid/covid-19-data/master/public/data/owid-covid-data.csv的地址。我们将使用Python的requests库来发送请求并获取数据。解析策略由于目标数据是CSV格式我们几乎不需要复杂的HTML解析。获取到数据后直接使用pandas库的read_csv函数甚至可以指定URL作为参数一步完成下载和加载为DataFrame。这比解析HTML表格要稳定和高效得多。存储策略对于建模而言我们可能需要多次运行模型进行调参和验证。因此不建议每次运行爬虫都重新从网络下载除非需要最新数据。一个良好的实践是首次运行爬虫将数据下载到本地如data/raw/owid_covid_latest.csv后续运行脚本时先检查本地文件是否存在以及其“新鲜度”例如文件修改时间是否超过24小时再决定是否重新下载。这既尊重了数据源的服务器负载也加快了本地开发调试的速度。伦理与合规策略这是必须严肃对待的一环。在爬取任何公开数据前务必查看网站的robots.txt文件例如访问https://ourworldindata.org/robots.txt和使用条款。对于OWID和JHU这类明确鼓励数据用于研究、教育目的的项目通常没有问题但我们仍应做到a) 设置合理的请求间隔如time.sleep(1)避免对服务器造成压力b) 在代码中注明数据来源c) 不将数据用于商业牟利等违反条款的用途。3. 爬虫实战从零构建可复用的数据获取模块现在我们开始动手编写代码。我将构建一个名为CovidDataFetcher的类它将封装数据获取、缓存、更新的所有逻辑方便在建模的主程序中调用。3.1 环境准备与依赖安装首先确保你的Python环境建议3.7以上中安装了必要的库。打开终端或命令提示符执行以下命令pip install requests pandasrequests用于发送HTTP请求获取网络数据。pandas数据处理的利器不仅能轻松读取CSV还能进行复杂的数据清洗、转换和分析是数学建模的必备工具。3.2 核心爬虫类设计与实现我们将创建一个Python脚本文件例如covid_data_fetcher.py。import os import time from datetime import datetime, timedelta import pandas as pd import requests class CovidDataFetcher: 一个用于获取和缓存Our World in Data全球COVID-19数据集的爬虫类。 def __init__(self, data_urlNone, cache_dir./data, cache_hours24): 初始化数据获取器。 参数: data_url (str): OWID数据文件的URL。如果为None则使用默认URL。 cache_dir (str): 本地缓存数据的目录路径。 cache_hours (int): 缓存有效的小时数。超过此时间将重新下载。 self.default_url https://raw.githubusercontent.com/owid/covid-19-data/master/public/data/owid-covid-data.csv self.data_url data_url if data_url else self.default_url self.cache_dir cache_dir self.cache_hours cache_hours self.raw_data_path os.path.join(cache_dir, raw, owid_covid_data.csv) # 确保缓存目录存在 os.makedirs(os.path.dirname(self.raw_data_path), exist_okTrue) def _is_cache_valid(self): 检查本地缓存文件是否有效存在且未过期。 if not os.path.exists(self.raw_data_path): return False file_mod_time datetime.fromtimestamp(os.path.getmtime(self.raw_data_path)) return datetime.now() - file_mod_time timedelta(hoursself.cache_hours) def _download_data(self): 从网络下载数据并保存到本地。 print(f正在从 {self.data_url} 下载数据...) try: # 添加一个简单的请求头模拟浏览器访问避免被某些服务器拒绝 headers {User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36} response requests.get(self.data_url, headersheaders, timeout30) response.raise_for_status() # 如果状态码不是200抛出HTTPError异常 # 保存原始数据 with open(self.raw_data_path, wb) as f: f.write(response.content) print(f数据已保存至: {self.raw_data_path}) return True except requests.exceptions.RequestException as e: print(f下载数据时发生错误: {e}) # 这里可以添加更复杂的错误处理例如重试机制、备用URL等 return False except Exception as e: print(f发生未知错误: {e}) return False def get_data(self, force_downloadFalse): 获取数据。优先使用本地缓存除非缓存过期或强制下载。 参数: force_download (bool): 是否强制重新下载忽略缓存。 返回: pandas.DataFrame: 包含COVID-19数据的DataFrame如果失败则返回None。 # 检查是否需要下载 need_download force_download or not self._is_cache_valid() if need_download: success self._download_data() if not success: # 如果下载失败但缓存文件存在尝试使用旧的缓存并给出警告 if os.path.exists(self.raw_data_path): print(网络下载失败将使用过期的本地缓存数据。请注意数据可能不是最新的。) else: print(无法获取数据请检查网络连接或URL。) return None else: # 下载成功等待一秒体现友好爬虫行为 time.sleep(1) else: print(使用有效的本地缓存数据。) # 加载数据到pandas DataFrame try: df pd.read_csv(self.raw_data_path) print(f数据加载成功共 {df.shape[0]} 行{df.shape[1]} 列。) # 查看前几行和列名帮助了解数据结构 print(\n数据前5行预览:) print(df.head()) print(\n数据列名:) print(df.columns.tolist()) return df except Exception as e: print(f加载CSV数据时发生错误: {e}) return None # 示例如何使用这个类 if __name__ __main__: # 创建一个数据获取器实例缓存设置为12小时 fetcher CovidDataFetcher(cache_hours12) # 获取数据会自动判断是否需要下载 covid_df fetcher.get_data() if covid_df is not None: # 在这里进行你的数据分析或建模预处理 # 例如筛选出中国的数据 china_data covid_df[covid_df[location] China] print(f\n中国数据共有 {china_data.shape[0]} 条记录。) # 查看中国最新的数据 print(china_data[[date, total_cases, total_deaths, people_vaccinated]].tail())代码关键点解析缓存机制 (_is_cache_valid): 这是提升体验的核心。它检查文件是否存在以及修改时间是否在设定的cache_hours内。避免了每次运行脚本都发起网络请求对于快速迭代的建模工作流非常友好。错误处理 (_download_data中的 try-except): 网络请求充满不确定性。我们捕获了requests库可能抛出的各种异常如连接超时、URL错误、HTTP状态码错误等并给出了友好的提示。在下载失败时如果本地有旧缓存会降级使用并警告用户保证了程序的鲁棒性。用户代理 (User-Agent): 有些服务器会拒绝没有标准浏览器标识的请求。添加一个常见的User-Agent头可以避免被简单的反爬机制拦截。延迟 (time.sleep(1)): 在成功下载后等待1秒是一个简单的“礼貌性”爬虫行为表明我们无意对服务器进行高频攻击。数据预览: 在get_data方法中加载数据后打印了数据维度、前几行和列名。这对于初次使用该数据源的人来说非常有用可以快速了解数据结构。4. 数据清洗与预处理为建模准备“干净”的输入直接从网络获取的原始数据很少能直接扔进模型。对于OWID的数据虽然已经很干净但我们仍需要根据具体的建模目标进行预处理。这一步通常在获取DataFrame之后在建模之前完成。4.1 常见的数据问题与处理假设我们的建模目标是分析几个主要国家疫情发展的时序规律并进行简单的预测。我们需要对数据进行以下处理def preprocess_data(df): 对获取的COVID-19数据进行清洗和预处理。 参数: df (pandas.DataFrame): 原始的OWID数据DataFrame。 返回: pandas.DataFrame: 处理后的DataFrame。 # 1. 处理缺失值 print(处理缺失值...) # 对于数值型列我们可以用前向填充ffill或后向填充bfill但疫情数据用0填充新增可能更合理 # 这里我们区分对待累计数据向前填充每日新增数据用0填充假设当天无新增 cols_to_ffill [total_cases, total_deaths, people_vaccinated] # 累计数据 cols_to_fill_zero [new_cases, new_deaths] # 每日新增数据 for col in cols_to_ffill: if col in df.columns: df[col] df[col].fillna(methodffill).fillna(0) # 先向前填充再对开头缺失的填0 for col in cols_to_fill_zero: if col in df.columns: df[col] df[col].fillna(0) # 2. 日期格式转换 print(转换日期格式...) df[date] pd.to_datetime(df[date]) # 3. 筛选感兴趣的国家和时期 print(筛选数据...) countries_of_interest [United States, India, Brazil, United Kingdom, China] # 假设我们分析2020-03-01之后的数据 start_date 2020-03-01 filtered_df df[(df[location].isin(countries_of_interest)) (df[date] start_date)].copy() # 4. 按国家和日期排序确保时间序列正确 filtered_df.sort_values(by[location, date], inplaceTrue) # 5. 计算一些衍生指标例如7日移动平均平滑每日新增数据的噪声 print(计算7日移动平均...) filtered_df[new_cases_smoothed] filtered_df.groupby(location)[new_cases].transform( lambda x: x.rolling(window7, min_periods1).mean() ) filtered_df[new_deaths_smoothed] filtered_df.groupby(location)[new_deaths].transform( lambda x: x.rolling(window7, min_periods1).mean() ) # 6. 重置索引 filtered_df.reset_index(dropTrue, inplaceTrue) print(f预处理完成。处理后数据形状: {filtered_df.shape}) return filtered_df # 在主程序中使用 if __name__ __main__: fetcher CovidDataFetcher() raw_df fetcher.get_data() if raw_df is not None: processed_df preprocess_data(raw_df) # 现在 processed_df 就可以用于后续的建模分析了 print(processed_df[[location, date, new_cases, new_cases_smoothed]].tail(10))预处理步骤的“为什么”缺失值处理疫情数据中某些国家在某些日期可能没有报告数据导致NaN。对于total_cases这样的累计指标用前一天的数值填充前向填充是合理的因为它代表累计值未更新。对于new_cases这样的日增指标填充为0比填充为均值或中位数更符合实际意味着当天报告的新增为0。这里有一个坑要小心区分数据缺失是“未报告”还是“真为零”。OWID的数据通常已经做了处理将缺失的日增数据标记为0或留空我们的填充策略是基于对数据背景的理解。日期转换将date列从字符串转换为datetime类型是进行任何时间序列分析的前提否则无法进行日期比较、排序、重采样等操作。数据筛选全量数据非常庞大。根据建模目标提前筛选出关键国家和时间段可以极大减少后续计算的内存占用和处理时间让分析焦点更集中。移动平均每日新增病例和死亡数据波动很大受检测量、周末报告延迟等因素影响。计算7日移动平均可以平滑掉这些“噪声”更清晰地展示疫情发展的趋势这是时间序列分析和可视化中非常常用的技巧。groupby(location)确保了移动平均是在每个国家内部独立计算的。5. 数据验证与异常检测确保爬取结果的可靠性数据爬取和预处理后绝不能直接相信数据是完全正确的。必须进行验证否则“垃圾进垃圾出”模型结果毫无意义。5.1 基础统计与合理性检查def validate_data(df): 对处理后的数据进行基本验证和异常检测。 print(\n 数据验证报告 ) # 1. 检查关键字段是否存在 essential_cols [location, date, total_cases, new_cases] missing_cols [col for col in essential_cols if col not in df.columns] if missing_cols: print(f警告缺失关键字段 {missing_cols}) else: print(关键字段完整性检查通过。) # 2. 检查数据时间范围和国家数量 print(f数据时间范围: {df[date].min()} 到 {df[date].max()}) print(f包含国家/地区数量: {df[location].nunique()}) print(f包含的国家/地区: {df[location].unique().tolist()}) # 3. 检查逻辑一致性每日新增病例之和应大致等于累计病例的增量允许由于数据修订产生的微小误差 print(\n检查累计病例与新增病例的逻辑一致性抽样...) sample_countries df[location].unique()[:3] # 抽查前3个国家 for country in sample_countries: country_df df[df[location] country].sort_values(date) # 计算通过新增病例累加出来的“计算累计值” country_df[calculated_total] country_df[new_cases].cumsum() # 与官方报告的累计值比较取最后一天 official_final country_df[total_cases].iloc[-1] calculated_final country_df[calculated_total].iloc[-1] diff_ratio abs(official_final - calculated_final) / official_final if official_final 0 else 0 print(f {country}: 官方累计 {official_final:.0f}, 计算累计 {calculated_final:.0f}, 差异比率 {diff_ratio:.2%}) if diff_ratio 0.05: # 如果差异超过5%发出警告 print(f **警告{country}的累计与新增数据可能存在较大不一致**) # 4. 检测异常值例如某天的新增病例数为负值数据修正可能导致 negative_new_cases df[df[new_cases] 0] if not negative_new_cases.empty: print(f\n发现 {len(negative_new_cases)} 条新增病例为负的记录通常是数据修正。) print(例如) print(negative_new_cases[[location, date, new_cases]].head()) # 5. 检查移动平均列的单调性7日平均应比原始数据平滑 # 可以通过简单可视化或计算方差来检查这里打印一个示例 print(\n平滑效果示例以某个国家最后10天为例:) example_country sample_countries[0] example_data df[df[location] example_country].tail(10)[[date, new_cases, new_cases_smoothed]] print(example_data.to_string(indexFalse))验证的意义字段检查防止因为数据源列名变更导致后续代码报错。逻辑一致性检查这是最核心的验证。如果new_cases的累计和与total_cases的终值相差太大说明数据在某个环节可能出了问题例如累计值被修订过但每日新增值未同步更新。OWID的数据质量很高通常差异很小但养成检查的习惯至关重要。异常值检测负的新增病例在现实数据中确实存在通常是之前某天的数据高报后来进行了向下修正。发现它们不是要删除而是要理解其含义并在建模时考虑是否需要进行处理例如将负值视为0或者使用修正后的累计数据。6. 集成到建模工作流与高级技巧一个孤立的爬虫脚本价值有限只有将其无缝嵌入到你的建模分析管道中才能发挥最大效用。6.1 构建模块化的数据管道建议将上述功能组织成一个小的数据获取与预处理包。目录结构可以如下your_model_project/ ├── data/ │ ├── raw/ # 存放爬取的原始CSV │ └── processed/ # 存放清洗后的数据可保存为pickle或feather格式加载更快 ├── src/ │ ├── data_fetcher.py # 包含CovidDataFetcher类 │ ├── data_preprocessor.py # 包含preprocess_data, validate_data函数 │ └── main.py # 主分析或建模脚本 └── requirements.txt在main.py中你可以这样调用from src.data_fetcher import CovidDataFetcher from src.data_preprocessor import preprocess_data, validate_data import matplotlib.pyplot as plt def main(): # 1. 获取数据 fetcher CovidDataFetcher(cache_dir../data, cache_hours6) raw_data fetcher.get_data() if raw_data is None: print(无法获取数据退出。) return # 2. 预处理数据 processed_data preprocess_data(raw_data) # 3. 验证数据 validate_data(processed_data) # 4. 进行简单的可视化示例 countries processed_data[location].unique() fig, axes plt.subplots(len(countries), 1, figsize(12, 3*len(countries)), sharexTrue) if len(countries) 1: axes [axes] for ax, country in zip(axes, countries): country_df processed_data[processed_data[location] country] ax.plot(country_df[date], country_df[new_cases_smoothed], label7天平均新增, linewidth2) ax.set_title(f{country} - 每日新增病例7日平滑) ax.legend() ax.grid(True, alpha0.3) plt.tight_layout() plt.savefig(../output/covid_trends.png, dpi150) plt.show() # 5. 将处理好的数据保存供后续建模使用 processed_data.to_pickle(../data/processed/covid_processed.pkl) print(\n数据预处理完成已保存至 ../data/processed/covid_processed.pkl) if __name__ __main__: main()6.2 应对数据源变更的策略公开数据源的结构或URL可能会变化。一个健壮的爬虫需要有一定的适应性。URL失效将数据源URL作为配置项甚至可以提供一个备选URL列表。在_download_data方法中如果主URL失败可以尝试备用URL。数据结构变化这是更常见的问题。OWID的列名可能会增加如增加新的疫苗指标。我们的预处理代码如果指定了列名可能会因为列不存在而报错。改进方法是在读取数据后打印所有列名并与我们预期的“必需列”进行对比。如果某些必需列缺失尝试用已知的旧列名或类似的列名进行映射或者用默认值创建该列并记录警告。使用try-except块包裹对特定列的操作。# 在preprocess_data函数开头添加列名检查 expected_cols {location, date, total_cases, new_cases, total_deaths, new_deaths} missing_cols expected_cols - set(df.columns) if missing_cols: print(f警告数据源缺少预期列 {missing_cols}。这可能导致后续处理出错。) # 尝试创建缺失的列并填充NaN或0 for col in missing_cols: df[col] 06.3 性能与扩展考量增量更新对于超大规模的历史数据每次都下载整个文件可能低效。如果数据源提供按日期或按国家查询的API可以考虑实现增量爬取只获取最新的数据并与本地历史数据合并。OWID的CSV文件虽然整体不小但对于建模场景通常可以接受全量下载。并发与速率限制如果未来需要从多个不同的API端点获取数据例如分别获取病例数据、疫苗数据、移动数据可以考虑使用concurrent.futures模块进行有限的并发请求但务必遵守目标网站的速率限制并在请求间添加延迟。数据存储格式对于处理好的数据保存为pickle或feather格式比CSV加载速度更快特别是当DataFrame很大时。这在频繁读取中间数据的建模迭代中能节省大量时间。构建这样一个爬虫其价值远不止于获取一份疫情数据。它锻炼了你从需求分析、工具选型、代码实现、错误处理到数据验证的完整数据处理能力。在数学建模竞赛中这份能力能让你在面对任何陌生的数据需求时都能快速构建出可靠的数据获取通道将更多精力专注于模型本身。