Python天气预报系统实践：从数据采集、清洗到可视化分析

简介面向Python学习者或需要快速落地可视化项目的开发者这份天气预报系统资源将爬虫抓取、Tkinter界面设计与数据可视化串联成一个可运行的小型应用支持多城市选择、15天预报查询、绘图分析与结果保存。压缩包共5个文件包含2个Python脚本分别负责数据爬取与界面/绘图逻辑、1个图标、1张界面效果图及1个说明文本整体仅754KB轻量易部署在Python或Jupyter环境中均可直接运行。目前已有21310人学习/下载可见其作为课程设计或入门实践参考的热度。下载后可获得可直接运行的Py文件和必要素材借助清晰的文件分工快速理解爬虫请求、GUI布局与绘图流程适合用于毕业设计、课设或自学练手。1. 先把标题拆开看这不是爬虫课设是一条数据管道「基于python的天气预报系统设计和可视化数据分析」这个标题在课程设计和毕设里出现频率极高但大多数交付物长这样一个能跑一次的爬虫脚本加两张截图——一张天气表格、一张折线图。这样交上去不是不行只是做的人自己心里清楚数据一换城市就要改代码多跑几天就开始重复读写图上出现一个 99.9℃ 的毛刺也不知道该不该信。这个标题真正值得做的地方是把「天气预报系统」理解成一条完整的数据管道拉数、清洗、调度、存储、出图、验证缺一环后面所有分析结论都站不住脚。这篇笔记就按我实际搭这种小系统时会走的路径来讲先选数据源再写调度和清洗最后落到可视化和自检。新手能跟着把一整套跑起来熟手可以直接跳到第 5 章的坑和第 6 章的验证思路。方向不复杂但细节很多尤其是那些「跑一遍看不出问题、跑一周才翻车」的部分。2. 数据源选型与落地免费接口、历史观测集和本地存储怎么配2.1 三个可得的数据源免费实时接口、开源历史数据集、自建采集怎么选做天气预报系统第一步不是写代码是确定数据从哪来。常见的选择有三类免费气象接口、开源历史观测数据集、自建页面采集。我分别说一下它们的实际体验。免费实时接口是最常用的起步选择注册后拿一个访问令牌按城市编码请求当前天气和逐日预报返回 JSON。优点是接入快字段齐全通常包含温度、湿度、气压、风向风速、降水概率缺点是限频严格免费档一般每分钟只能请求几十次且历史数据只保留最近几天拿不到长周期序列。开源历史观测数据集则相反能拿到逐小时或逐日的长期记录适合做「数据分析」那半边但往往只覆盖部分站点且格式不统一有的是 CSV有的是 NetCDF。自建采集风险最高页面结构一变脚本就失效还要考虑合规问题我一般不建议作为主数据源。我的选型结论比较保守实时接口负责「系统」的运转历史数据集负责「数据分析」的厚度。两者用同一套字段标准清洗后合并既满足预报展示又支撑趋势分析。如果你只是做课程设计连数据库都可以省先落 CSV等数据量到几万行再考虑 SQLite。2.2 用 requests 拉取实时天气并落原始 JSON最小可用脚本先写一个最简的拉数函数目标是「能跑、能留痕、能排错」。请求天气接口时三个参数最关键城市编码、访问令牌、单位制。单位制建议固定用公制否则后面所有数值分析和图表都要做单位换算容易埋雷。import requests import json from pathlib import Path def fetch_weather_now(city_code: str, token: str, unit: str metric) - dict: 拉取某城市当前天气失败时返回带 error 键的字典。 参数 city_code: 城市编码来自接口文档的城市列表 token: 在气象数据平台申请的访问令牌 unit: metric 返回摄氏度与 m/simperial 返回华氏与 mph url https://api.example-weather.in/v1/now params { location: city_code, key: token, unit: unit, } try: resp requests.get(url, paramsparams, timeout5) resp.raise_for_status() except requests.Timeout: return {error: timeout, city: city_code} except requests.HTTPError as exc: return {error: fhttp_{exc.response.status_code}, city: city_code} payload resp.json() # 留原始响应排查字段结构变化时用得上 Path(raw_responses).mkdir(exist_okTrue) with open(fraw_responses/{city_code}.json, a, encodingutf-8) as f: f.write(json.dumps(payload, ensure_asciiFalse) \n) return payload逻辑说明函数先组装请求参数再用 timeout5 限制单次请求的最长等待避免某个城市编码失效时整个调度卡死。返回值统一成字典成功时是接口的 JSON 结构失败时是带 error 键的结构这样上层调度不需要关心异常细节。落盘原始 JSON 有两个作用一是排错时能回看接口到底返回了什么二是接口字段如果变更原始留痕能帮你定位是代码问题还是数据源问题。参数说明城市编码不要用中文名直接请求多数接口的城市列表里中文名有歧义比如名称相同的区县会返回错误或错数据单位制尽量显式传入不要依赖接口默认值timeout 建议 5 到 10 秒之间太短容易误判超时太长会拖慢批量任务。注意这段代码里的 URL 是占位写法实际接入时替换成你申请到的接口地址即可。2.3 清洗与入库把字段整理成结构化 DataFrame 的关键参数接口返回的 JSON 通常嵌套较深直接存原始 JSON 没问题但做分析时要反复取嵌套字段很容易写错路径。我的习惯是拉数后马上做一次标准化清洗转成扁平的 DataFrame然后追加到 CSV。清洗这一步的核心是类型转换和时区处理。import pandas as pd from pathlib import Path def normalize_weather(raw: dict, city_code: str) - pd.DataFrame: 把接口原始 JSON 转成一行 DataFrame并追加写入 CSV。 data raw.get(data, {}) if not data or obs_time not in data: return pd.DataFrame() row { city: city_code, obs_time: pd.to_datetime(data[obs_time], units, utcTrue) .tz_convert(Asia/Shanghai), temp: float(data[temp]), feels_like: float(data[feels_like]), humidity: float(data[humidity]), pressure: float(data[pressure]), wind_dir: data.get(wind_dir, ), wind_deg: float(data.get(wind_deg, 0)), wind_scale: float(data.get(wind_scale, 0)), } df pd.DataFrame([row]) out_path Path(fdata/{city_code}.csv) if out_path.exists(): df.to_csv(out_path, modea, headerFalse, indexFalse, encodingutf-8) else: df.to_csv(out_path, indexFalse, encodingutf-8) return df逻辑说明函数把嵌套的 data 字段拍平只保留分析会用到的列。obs_time 先用 units 解析 Unix 时间戳再统一转成东八区时间这一步很重要——接口返回的往往是 UTC 时间直接落库的话后面画折线图时横轴会整体偏移 8 小时。wind_dir 是风向的文字描述如「东南风」wind_deg 是风向角度两者都保留可视化时按需选用。参数说明追加写 CSV 时用 modea 和 headerFalse避免每次重写表头编码固定 utf-8否则 Windows 下 Excel 打开会乱码。为什么起步阶段不直接上 MySQL 或 SQLite因为你的数据规模大概率只有几万行CSV 对 pandas 足够友好还能用记事本直接检查等你要做增量更新和多表关联时再迁到 SQLite 也只需要改读写函数不影响上层分析代码。如果你确定要做长时间自动采集一开始就上 SQLite 更省事一张天气明细表加一张采集日志表就够了。3. 把「预报系统」做真定时调度、多城市任务与异常值拦截3.1 为什么实时天气不等于预报系统调度的业务价值很多人做完 2.2 的小脚本就觉得自己完成了「系统」其实那只是「单次查询工具」。真正的天气预报系统要有两个能力一是能定时自动运行二是能积累多城市的历史序列。有了历史序列「预报」才有评估对象——你昨天报的最高温今天实际是多少误差大不大长期看哪个城市的预报更可信。调度方案有两条路。初学阶段直接用 Python 脚本里的死循环加 time.sleep 就能跑优点是好理解缺点是一旦进程被杀就停更而且重启后容易重复采集。成熟一点的做法是用系统级定时任务让操作系统来管进程生命周期。我给一个 APScheduler 的写法它在 Python 进程内做调度比纯 sleep 更可控也比手写 cron 规则更好调试。from apscheduler.schedulers.blocking import BlockingScheduler from datetime import datetime CITY_CODES [A010, B021, C028] # 换成实际城市编码 TOKEN 从环境变量读取不要硬编码在脚本里 def job_tick(): print(f[{datetime.now():%Y-%m-%d %H:%M}] start collect) for city in CITY_CODES: raw fetch_weather_now(city, TOKEN) df normalize_weather(raw, city) if df.empty: print(f {city} empty or error: {raw.get(error)}) else: print(f {city} ok, temp{df.iloc[0][temp]}) scheduler BlockingScheduler(timezoneAsia/Shanghai) # 每小时的 10 分执行避开整点接口高峰 scheduler.add_job(job_tick, cron, minute10, idweather_hourly) scheduler.start()逻辑说明BlockingScheduler 会阻塞主进程按 cron 表达式触发任务。minute10 表示每小时的第 10 分钟执行一次刻意避开整点。很多免费接口在整点会涌进大量请求你晚 10 分钟去拉成功率更高数据本身也不会有什么变化。参数说明timezone 必须显式指定否则服务器时区一变任务执行时间就跟着漂移add_job 里的 id 要唯一重复添加同名任务会报错。如果你不想依赖 APScheduler用系统 cron 也一样把脚本封装成 python collector.py 后在 crontab 里写10 * * * * cd /path/to/project python collector.py两条路的效果等价选一个就行。3.2 多城市批量拉取时的限频与重试指数退避怎么写免费接口最常见的限制是每分钟请求次数。如果你有 10 个城市每个城市每小时拉一次批量任务会在 1 分钟内发出 10 个请求大概率触发限频。解决办法是控制请求间隔并对失败任务做重试。重试不能用固定间隔——接口刚拒绝你时立刻重试大概率还是被拒指数退避才是通用做法。import time def collect_with_retry(city_code: str, token: str, max_retries: int 3) - bool: 带指数退避的采集函数1 秒、4 秒、16 秒逐级拉长重试间隔。 delay 1 for attempt in range(1, max_retries 1): raw fetch_weather_now(city_code, token) if error not in raw: df normalize_weather(raw, city_code) return not df.empty print(f [{city_code}] attempt {attempt} failed: {raw[error]}) if attempt max_retries: time.sleep(delay) delay * 4 return False逻辑说明每次失败后等待时长按 1 秒、4 秒、16 秒递增而不是每次都等同样长的时间。这样既给了接口恢复的时间又不会因为连续重试把自己的请求额度耗尽。max_retries 设 3 是性价比比较高的选择——3 次重试大概耗时 21 秒加上业务上的意义已经足够超过 3 次很可能是接口或网络本身出了问题重试再多也救不回来。参数说明delay 的倍数可以按你的接口限制调整限制严的就用 delay * 3宽松就用 delay * 2。如果重试三次仍然失败建议把这条失败记录写入一个 error_log.csv内容包括城市、时间、错误类型方便隔天排查而不是静默忽略。很多「数据少了几天」的问题都是失败后没有留痕造成的。3.3 异常数据识别温度跳变、降水为负这类脏数据怎么拦数据管道跑起来后你会很快发现部分记录是脏数据。常见的有三种温度毛刺某个时刻突然变成 99.9 或 -99.9通常是传感器占位值或单位混用、降水为负单位换算错误导致、观测时间乱序接口缓存或本地时区处理出错导致时间戳倒挂。这些脏数据如果不拦截画图时就是一个尖峰做统计时会把均值拉偏。import pandas as pd def check_outliers(df: pd.DataFrame, col: str temp, max_gap: float 3.0) - pd.Series: 标记相邻采样点之间变化幅度超过阈值的记录。 diff df[col].diff().abs() return diff max_gap def check_negative_precip(df: pd.DataFrame) - pd.Series: 降水为负或小于 0 的观测视为异常。 return df[precip] 0逻辑说明温度跳变检测用一阶差分相邻两小时的温度变化超过 3℃ 就标记出来。这个阈值不是拍脑袋定的——正常天气系统里逐小时气温变化几乎不会超过 3℃除非遇到强冷锋过境或雷暴。降水负值直接过滤因为降水量的物理意义不可能为负。参数说明max_gap3.0 是温带地区的经验值如果你采集的是高原或沙漠气候数据昼夜温差大可以放宽到 5.0如果数据是分钟级的相邻两个采样点的温差会更小阈值反而要收紧到 1.0。另一个很隐蔽的坑是「检查完了但没落标记」——建议在清洗函数里增加一列 is_outlier标记完不要直接删行先保留原始记录分析时再过滤这样后悔药随时能吃。4. 可视化数据分析让积累的天气数据真正「说话」4.1 选型Matplotlib 还是 Pyecharts取决于你要交付什么可视化选型是个老话题但针对天气数据有明确的偏向。Matplotlib 的优势是静态图质量稳定、嵌入论文和 Word 文档方便、渲染速度快劣势是交互能力弱鼠标悬停没有提示图例一多就显得拥挤。Pyecharts 的优势是交互体验好图表是 HTML 文件浏览器打开就能缩放、悬停、切换适合演示劣势是初始化环境稍重输出是网页文件而不是图片贴进文档要额外截图。我的建议是如果交付物是报告或课设文档用 Matplotlib如果要做展示或让非技术的人自己看用 Pyecharts。下面两节代码以 Matplotlib 为主因为它的中文字体坑多、参数细是踩坑密度最高的部分Pyecharts 的 API 更友好照着官方例子改就能跑通。4.2 温度趋势与降雨量对比一张能放进报告的趋势图画趋势图之前先处理一个基础问题数据频率。如果你每小时采一次一周就有 168 个点直接画会非常密常见的做法是先按天重采样再做一个移动平均平滑。下面这段代码把多城市的气温曲线画在同一张图上并标出最高温极值是报告里最常用到的图型。import matplotlib.pyplot as plt import pandas as pd plt.rcParams[font.sans-serif] [SimHei, Microsoft YaHei] plt.rcParams[axes.unicode_minus] False def plot_temp_trend(csv_files: dict, window: int 6) - None: csv_files: {城市名: csv路径}window 为移动平均窗口大小。 fig, ax plt.subplots(figsize(12, 6)) for city, path in csv_files.items(): df pd.read_csv(path, parse_dates[obs_time], index_colobs_time) # 重采样到小时均值缺失值用线性插值补齐 hourly df[temp].resample(H).mean().interpolate() ax.plot(hourly.index, hourly.rolling(window).mean(), labelcity, linewidth1.6) ax.set_ylabel(气温℃) ax.set_title(多城市气温趋势对比) ax.grid(alpha0.3) ax.legend() fig.autofmt_xdate() plt.tight_layout() plt.savefig(trend_temp.png, dpi150)逻辑说明read_csv 时用 parse_dates[obs_time] 把时间列转成 datetime 类型再设为索引这是时间序列分析的前提。resample(H) 把数据规整到小时网格并取均值这样即使某小时漏采了一次网格上也有一个位置。interpolate() 用线性插值填补空位避免折线图断裂。rolling(window).mean() 做移动平均window6 表示用前后 6 个小时的平均值来平滑曲线适用于分钟级或小时级数据。参数说明移动平均窗口的大小直接影响曲线形状——窗口越大曲线越平滑但峰值会被抹平。做「趋势」分析时窗口可以大一点做「极值」分析时必须用原始数据否则最高温会被平均掉。中文字体配置这两行必须同时存在否则即使指定了 SimHei负号也会显示成方块或乱码。4.3 风玫瑰图与湿度分布图上最容易翻车的三个细节温度趋势图是入门风玫瑰图才是真正能体现「数据分析」深度的图。它用极坐标展示风向和风速的分布能一眼看出一段时间内的主导风向。画风玫瑰图有三个高频坑风向是中文文字要先映射成角度角度分箱不是按 360 度平均分极坐标图的条形宽度不能用默认值。下面是一个能跑的极坐标实现。import pandas as pd import numpy as np import matplotlib.pyplot as plt df pd.read_csv(data/wind.csv) # 中文风向映射为角度北0东90南180西270 wind_map {北风: 0, 东北风: 45, 东风: 90, 东南风: 135, 南风: 180, 西南风: 225, 西风: 270, 西北风: 315} df[wind_deg] df[wind_dir].map(wind_map) bins np.arange(0, 361, 30) # 12 个分箱每箱 30 度 counts, _ np.histogram(df[wind_deg].dropna(), binsbins) theta np.deg2rad(bins[:-1] 15) fig, ax plt.subplots(subplot_kw{projection: polar}) ax.bar(theta, counts, widthnp.deg2rad(30), alpha0.6) ax.set_theta_direction(-1) ax.set_theta_zero_location(N)逻辑说明风向文字到角度的映射必须放在清洗阶段不要在绘图时才做否则每次画图都要处理一遍脏数据。分箱用 np.arange(0, 361, 30) 得到 12 个区间角度取每个箱子的中心值。theta_direction 设为 -1 是为了让极坐标图按顺时针方向递增符合气象上风向的习惯zero_location 设为 N 表示起点在正北方向。参数说明widthnp.deg2rad(30) 必须和分箱宽度一致否则条形之间会重叠或留白。set_theta_direction 和 set_theta_zero_location 这两行是气象图的默认规范不设的话玫瑰图的方位会和常见气象图反着来读者一眼看不懂。湿度分布建议用直方图或箱线图不要用折线图因为湿度是区间型物理量折线图会制造出「相邻湿度连续变化」的误导。5. 天气预报系统避坑指南五个踩坑现场与修法5.1 拉数阶段的坑空响应、限频请求被拒现象定时任务跑了一周后某天某个城市的数据从某时刻起全部缺失查看日志发现返回体是{error: unauthorized}但检查代码里的令牌明明是没变的。原因免费接口的访问令牌常有有效期有的 7 天有的 30 天过期后接口不会明确提示「令牌过期」而是返回一个通用的鉴权失败错误。再加一个隐蔽因素你拉数的时间正好在令牌刷新时间附近接口端做了临时锁定。解决把令牌读取放到一个独立配置文件中并在调度任务里加令牌有效性检查——每次收到 unauthorized 错误时先用测试请求快速验证令牌状态并写一条醒目日志。不要在一次调度里重试超过 3 次直接告警出来人工换令牌更高效。现象脚本本地跑一切正常部署到服务器后每隔几分钟就被接口拒绝返回 403。原因家用宽带的出口 IP 和你所在机房或实验室的出口 IP 不一样部分免费接口对数据中心 IP 段有更严格的限频策略也可能是服务器上同时跑着多个任务每个任务都在请求同一个接口总 QPS 超了。解决把所有请求收敛到一个统一的数据采集入口模块在模块里做全局限频——用一个全局变量记录上次请求时间间隔不足 2 秒就 sleep 补足。同时给每个请求加上 User-Agent 头避免被误判为脚本攻击。5.2 清洗阶段的坑毛刺值、重复记录现象气温趋势图在某个时间点出现一个孤立尖峰比如 7 月气温突然变成 99.9℃ 然后又恢复正常。原因接口或传感器用 99.9 作为「缺测占位值」正常气温不可能到这个数值。这类脏数据在数值上极端最容易一眼看出来但还有一种更难发现的毛刺是值在正常范围内但变化速率异常比如 10 分钟内温度从 25℃ 跳到 40℃ 再跳回来。解决先做物理范围检查温度限定在 -50 到 50 之间湿度限定在 0 到 100 之间超出直接标记再用 3.3 节的一阶差分法检测变化速率异常。注意标记和过滤要分开分析时过滤原始数据保留。现象同一天同一个城市出现了多条完全相同的记录分不清是重复采集还是数据本身如此。原因调度任务被手动执行了一次同时定时任务也正常触发或者 cron 任务因为系统休眠错过了执行窗口重启后补跑了一次。重复行会让后续统计严重失真——均值会被重复值加权。解决在清洗函数里按城市和 obs_time 做去重保留第一条即可。如果已经积累了重复数据用df.drop_duplicates(subset[city, obs_time], keepfirst)一键清理。更彻底的做法是写入前先查一下目标时间戳是否已存在存在就跳过。5.3 出图阶段的坑中文字体与风向表示现象图上的中文标题和坐标轴标签全部显示成方块或乱码但英文和数字正常。原因Matplotlib 默认字体不含中文字形需要指定系统里已有的中文字体。如果指定了SimHei仍然乱码大概率是系统里没有这个字体——Windows 通常内置但 Linux 服务器多半没装。解决先用fc-list :langzh查看服务器可用中文字体然后换成实际存在的字体名。一个更稳的做法是直接给 Matplotlib 注册字体文件路径不依赖系统字体名。另外别忘了配axes.unicode_minusFalse否则坐标轴上的负号也会变成方块。现象风玫瑰图的方向和气象常识对不上北风画在了右边东南风画在了左边。原因极坐标图默认 0 度在右侧、逆时针增大而气象风向的约定是 0 度为正北、顺时针增大。没有做方向转换就直接画图的方向就会镜像翻转。解决设置set_theta_direction(-1)和set_theta_zero_location(N)两行参数。如果风向数据是中文文字更要在清洗阶段就统一映射成角度不要在绘图函数里反复做。风向角度还有一个隐蔽问题部分接口返回的「风向角」是指风的来向还是去向不同平台定义不同画之前选一组实况数据人工核对一次。6. 进阶验证把天气系统从「能跑」变成「可信」做到这一步你的系统已经能定时采集、清洗、出图。但作为数据分析项目还缺最后一块拼图验证数据可靠性和预报准确度。我一般会加一个自检脚本validate_pipeline.py每次采集后运行一遍输出一份简单的健康报告。校验清单包括四个维度数据文件非空、最近观测时间与当前时间差小于 2 小时、字段缺失率低于 5%、图表文件已生成。def validate_runs() - None: report {} for city, path in csv_files.items(): df pd.read_csv(path, parse_dates[obs_time]) newest df[obs_time].max() age_hours (pd.Timestamp.now(tzAsia/Shanghai) - newest).total_seconds() / 3600 missing_rate df[[temp, humidity, wind_dir]].isna().mean().max() report[city] { age_hours: round(age_hours, 1), missing_rate: round(missing_rate, 3), rows: len(df), } print(f{city}: age{report[city][age_hours]}h, fmissing{report[city][missing_rate]}, rows{report[city][rows]})自检跑通之后可以做一件让这个项目从「课程设计」升级成「数据工具」的事情把预报值和次日实况做误差统计。每天记录接口给的次日最高温预报第二天再回填实际最高温每周计算一次平均绝对误差MAE。这个指标才是真正意义上的「天气预报系统质量评估」。写一个评估脚本从已积累的数据里按城市分组计算 MAE哪个城市误差大哪个城市的预报可信度低一目了然。这也是你做数据分析时最能拿出来讲的东西——不是画了几张图而是能回答「这个数据源到底准不准」。我自己吃过的最大教训是接口返回结构变更。某次它把字段从wind_dir改成了wind_direction我的清洗函数没有做兼容结果出图时所有风向数据都变成了空值且因为采集一直正常我直到画图时才注意到。从那以后我养成一个习惯每次刷新数据后先打印原始 JSON 的前 200 个字符看一眼字段名和值类型有没有变化。这个习惯救了我很多次也推荐给你。把这套验证脚本放进调度里让系统每天自己检查自己比出了问题再翻代码省心得多。希望帮到你。本文还有配套的精品资源点击获取