简介一套基于Flask框架的城市房价数据分析系统源码主要面向计算机专业毕业设计、课程设计及期末大作业等场景适合需要独立完成Web数据可视化项目的学生。系统后端采用Python与Flask编写前端涵盖HTML、CSS及JavaScript页面数据部分利用链家二手房真实数据集配合Excel表格和CSV文件形成从数据清洗、存储到展示的完整流程项目经过严格调试确保能够直接运行并附带数据库连接文档与项目配置说明便于二次开发。整个资源共248个文件除核心的Python脚本、模板页面与样式文件外还包含ipynb分析脚本、XML/Meta配置、Markdown说明、图片素材等压缩包大小约4.81MB。项目中还提供链家数据表和详情页、列表页等可视化界面读者可据此理解城市房价分析系统的模块划分与实现思路。已有115人参与学习属于98分的毕业设计参考项目特别适合用作课程实践、期末大作业或毕业设计的起点模板。1. 城市房价数据分析系统为什么 Flask 能撑起数据项目城市房价数据分析系统是数据分析入门最常见的实战项目之一但很多人的实现只停在“读 CSV 加画柱状图”离一个能被验收的 Flask 源码项目还差一整套服务化封装。Flask 的定位是轻量级 Web 框架不强加 ORM、不规定目录结构、不绑定前端刚好适合把 pandas 的分析结果以 HTTP 接口和可视化页面暴露出去。这个项目里数据是核心资产Flask 是表现层二者都依赖 Python 生态所以它比 Spring Boot 或 Node 更适合作这类“分析为主、展示为辅”的系统。下文不写教科书直接按“数据清洗 → 接口设计 → 图表联动 → 部署收尾”的路径把一份可运行的代码骨架和参数边界讲清楚。适合准备做毕设或简历项目的开发者参考也适合已经在做数据分析项目的人快速补齐工程化短板。2. 房价数据清洗与预处理入库前先解决脏数据数据分析系统的评分点往往不在界面而在数据质量。如果原始数据没清干净后面任何统计都是错的。做城市房价分析常见的数据来源是爬取房产平台公开列表页或者直接使用开源社区整理的 CSV。我这里推荐先用 CSV 起步把爬虫和清洗拆开避免把爬虫稳定性扯进项目评审里。数据清洗的目标很明确字段类型正确、单价总价一致、区域名称统一、异常样本不影响统计。2.1 房价数据字段设计与来源约定我先约定一个最小可用字段集也方便你对照自己的数据表。下面这张表是设计和清洗时都会反复用到的字段说明字段名类型示例清洗策略districtstr朝阳归一化到标准行政区名communitystr望京某小区去空格、去后缀house_typestr2室1厅正则提取室数areafloat89.5范围校验删除小于5或大于500的值unit_pricefloat58600单位元/平方米过大过小需复核total_pricefloat525单位万元与 unit_price*area/10000 做交叉校验build_yearint2008缺失时置 1970超过当前年份置缺失floorstr中楼层清洗为 high/mid/low 三档字段设计要先想清楚统计维度否则后续 groupby 会做得很别扭。比如“户型”要拆出“室数”而不是整串字符串否则前端按 2室1厅 和 2室2厅 筛选时联动逻辑会复杂很多。我把house_type拆成了rooms和hall两个派生字段分析时直接按rooms聚合阅读性更好。2.1.1 用 pandas 读取和初筛下面的代码重点不是 read_csv 本身而是把 dtype 和 parse_dates 这两个参数用对。import pandas as pd df pd.read_csv( house_data.csv, dtype{ district: string, community: string, house_type: string, area: float64, unit_price: float64, total_price: float64, build_year: Int64, # 可空整型 floor: string, }, keep_default_naFalse, ) print(df.dtypes) print(df.shape)这里有一个容易被忽略的细节build_year用Int64而不是int64因为 pandas 的缺失值在 int 类型下会被转成 float 或丢失。keep_default_naFalse能防止“暂无”“--”等字符串被错误识别为 NaN 后又引发类型混乱。dtype指定后CSV 里的脏值会在读取时直接报错这样能在最早阶段暴露数据格式问题而不是等画图时才发现。2.2 pandas 处理缺失值、重复值与异常值清洗顺序建议是去重 → 处理缺失 → 过滤异常 → 派生字段。去重要用subset锁定业务主键不要全表去重否则可能把相同小区、相同面积但不同朝向的房源也删掉。我会把“小区 户型 面积 总价 建造年份”作为重复判定条件。df df.drop_duplicates( subset[district, community, house_type, area, total_price, build_year], keepfirst, ) # 缺失值处理有明确含义的字段不要直接 drop df[build_year] df[build_year].fillna(1970) df[floor] df[floor].fillna(unknown) df[area] df[area].mask(df[area] 5) df df.dropna(subset[area, unit_price, total_price]) # 异常值过滤单价低于 1000 或高于 20 万大概率是分割符错误 df df[(df[unit_price] 1000) (df[unit_price] 200000)] df df[df[area].between(5, 500)]dropna不要轻易全表执行因为某一列缺失可能只是信息不全不代表整行无效。对build_year用 1970 填充是让它在统计时能落到“老旧房”分组而不是被过滤掉。异常值过滤的阈值要按城市调整一线城市单价上限二三十万很正常三四线城市可能上限十万都不到所以把阈值放到配置项里更合适。2.3 区域归一化与价格单位换算区域是最容易脏的文本字段。“北京市朝阳区”“朝阳区”“北京朝阳”的出现概率差不多直接 groupby 会导致同一个区被统计成三个区。归一化最实用的方法是用字典映射加正则兜底。alias_map { 北京市朝阳区: 朝阳, 北京朝阳: 朝阳, 朝阳区: 朝阳, 上海市浦东新区: 浦东, 上海浦东: 浦东, } def normalize_district(name: str) - str: name str(name).strip() if name in alias_map: return alias_map[name] # 兜底提取常见行政区后缀 for tail in (区, 县, 市): if name.endswith(tail): return name[:-1] return name df[district] df[district].map(normalize_district)关于单位换算我会把总价统一成万元单价统一成元/平方米。如果原始文件给出的是“总价 525.0 万元”但单价算出来对不上就要用交叉校验total_price * 10000 / area与unit_price的偏差超过 5% 时这条记录标记为可疑。这个校验动作特别容易被评分老师抓住提问所以我在项目里单独抽了一个_validate_record函数字段名和阈值的解释都写在注释里。这个清洗层完成后df就可以作为后续 Flask 接口的数据源了。如果你要更进一步把清洗后的结果导出为house_clean.parquet比 CSV 读取快还能保留Int64和string类型。我在接口层读取 parquet 而不是 CSV能明显减少每次启动的加载时间。3. Flask 后端用蓝图拆分数据接口与页面路由Flask 项目如果只把路由堆在 app.py 里逻辑多了以后会很难维护。城市房价数据分析系统的后端可以分成两大部分一是页面路由负责渲染 HTML 模板二是数据接口负责返回 JSON。用 Flask Blueprint 把这两块拆开代码清晰度会提升一个档次。这里我按实际可运行的目录结构来讲而不是贴一个 Flask 官方示例。3.1 项目目录结构与 Flask 扩展装配先给出一个最小但完整的目录布局。这个结构能满足“要源码”和“后续横向扩展”两种需求。house_price_analysis/ ├── app.py # 应用入口 ├── config.py # 配置项包括数据文件路径、缓存超时 ├── requirements.txt ├── data/ │ └── house_clean.parquet ├── services/ │ ├── __init__.py │ ├── data_loader.py # 读 parquet 并暴露全局 df │ └── stats_service.py # 聚合、排序、趋势计算 ├── routes/ │ ├── __init__.py │ ├── pages.py # 页面渲染蓝图 │ └── api.py # 数据接口蓝图 └── templates/ └── dashboard.htmlapp.py只做三件事创建 app、注册蓝图、挂载启动逻辑。不要在app.py里写业务计算。services/data_loader.py在应用启动时加载一次清理好的数据后续接口只做只读查询这样能避免每个请求反复读磁盘。下面是app.py的装配示例from flask import Flask from routes.pages import pages_bp from routes.api import api_bp def create_app(): app Flask(__name__) app.config.from_object(config.Config) app.register_blueprint(pages_bp) app.register_blueprint(api_bp, url_prefix/api) return app app create_app() if __name__ __main__: app.run(host0.0.0.0, port8000, debugFalse)注意url_prefix/api写在了register_blueprint的位置这样api_bp内部路由可以不用重复写/api后面调整前缀只改一个地方。debugFalse是必须的debugTrue不但有重载开销还会把错误堆栈直接暴露给前端项目评审时会被当成安全扣分点。3.2 实现 /api/house/analysis 统计接口这个接口是整个系统的核心它接收district、rooms、start_year、end_year四个参数返回均价趋势和区县分布。前端图表的上方筛选器和它一一对应。from flask import Blueprint, request, jsonify, current_app import pandas as pd from services.stats_service import price_trend_by_district api_bp Blueprint(api, __name__) api_bp.route(/house/analysis) def house_analysis(): district request.args.get(district, typestr, default) rooms request.args.get(rooms, typeint) start_year request.args.get(start_year, typeint, default1990) end_year request.args.get(end_year, typeint, default2025) if start_year end_year: return jsonify(code400, messagestart_year 必须小于 end_year), 400 df current_app.config[HOUSE_DF] result price_trend_by_district( df, districtdistrict, roomsrooms, start_yearstart_year, end_yearend_year ) return jsonify(code0, dataresult, messageok)request.args.get的typeint参数很实用它能直接把字符串转成 int如果转换失败会回落到default比手写 try-except 干净。这种写法对前端传rooms2和roomsabc都能安全处理。current_app.config[HOUSE_DF]是我在启动时塞进去的 DataFrame 引用避免每个函数都去 import 一个全局变量单元测试时也方便替换。3.3 参数校验、异常捕获与统一 JSON 返回接口多了以后统一返回值格式比每个接口自己拼 JSON 更值得花时间。我常用的返回结构是{code, message, data}其中code0表示成功非 0 表示业务异常。参数错误返回 HTTP 400但响应体里同样带message。这样前端只要解析一个结构。下表是这套系统的接口约定方便你对照前端代码路径参数返回/api/house/analysisdistrict, rooms, start_year, end_year{code, data: {trend, districts}, message}/api/house/price_distributiondistrict, rooms{code, data: {buckets, count}, message}/api/house/summary无{code, data: {total_records, avg_price, avg_area}, message}用 Flask 的errorhandler统一处理已知异常避免业务代码里到处写 try-catchapi_bp.errorhandler(ValueError) def handle_value_error(e): return jsonify(code400, messagestr(e)), 400 api_bp.errorhandler(KeyError) def handle_key_error(e): return jsonify(code400, messagef缺少必要字段: {e.args[0]}), 400这里要说明一点errorhandler只对蓝图内部生效如果你放在了pages_bp或全局 app 上作用范围不一样。我习惯把异常分成三层ValueError表示参数问题RuntimeError表示数据加载问题后续如果接 ORM 再增加数据库异常映射。这样评分老师追问“异常怎么处理”时你能说清楚层次而不是一句“try-except 包住”。4. 前端图表联动ECharts 对接 Flask 查询接口后端聚合得再好前端一张静态表格也看不出分析能力。城市房价分析系统里的“可视化”不是只画一张折线图而是要能和后端交互选择区县、户型、年份范围后图表随之更新。我用 ECharts 做图表库因为它的配置项直观且支持按需引入。Flask 侧不需要模板引擎做复杂数据渲染直接把接口返回的 JSON 交给 JavaScript 里的 ECharts 实例。4.1 页面模板与静态资源配置Flask 默认从templates目录渲染模板static目录放 JS 和 CSS。下面是一个简化版dashboard.html的骨架!DOCTYPE html html langzh-CN head meta charsetUTF-8 title城市房价分析面板/title script src{{ url_for(static, filenamejs/echarts.min.js) }}/script style #chart_price { width: 100%; height: 420px; } #chart_district { width: 100%; height: 320px; } /style /head body select iddistrict/select select idrooms option value全部户型/option option value11室/option option value22室/option option value33室/option /select div idchart_price/div div idchart_district/div script src{{ url_for(static, filenamejs/dashboard.js) }}/script /body /htmlurl_for(static, ...)比手写/static/...更安全因为应用可能挂在子路径下。如果项目直接用 CDN 的 ECharts要注意离线演示场景我建议把echarts.min.js下载到本地static/js这份文件本身就是“源码依赖完整”的证明也避免评审现场没有外网。4.2 按区县、户型、年份多维筛选前端筛选器通过fetch请求后端接口并把筛选条件拼到查询字符串里。下面这段dashboard.js是图表联动的最小可运行逻辑async function loadAnalysis() { const params new URLSearchParams(); if (document.getElementById(district).value) { params.append(district, document.getElementById(district).value); } const rooms document.getElementById(rooms).value; if (rooms) params.append(rooms, rooms); params.append(start_year, 1990); params.append(end_year, 2025); const resp await fetch(/api/house/analysis? params.toString()); const result await resp.json(); if (result.code ! 0) { console.error(result.message); return; } renderPriceTrend(result.data.trend); renderDistrictBar(result.data.districts); } document.getElementById(district).addEventListener(change, loadAnalysis); document.getElementById(rooms).addEventListener(change, loadAnalysis); loadAnalysis(); function renderPriceTrend(trend) { const chart echarts.init(document.getElementById(chart_price)); chart.setOption({ xAxis: { type: category, data: trend.map(item item.year) }, yAxis: { type: value, name: 均价元/㎡ }, tooltip: { trigger: axis }, series: [{ type: line, smooth: true, data: trend.map(item item.avg_price) }] }); }URLSearchParams是原生 API不需要引入 jQuery。这里有一个很容易忽略的问题ECharts 实例不能在 DOM 隐藏或销毁后原地复用如果页面有 Tab 切换需要在重新渲染前调用chart.dispose()否则会报 “Cant get DOM width or height”。我一般会在renderPriceTrend开头加chart.dispose()兜底但第一次渲染时没有实例会抛错所以更推荐用全局变量保存实例切换前先判断是否存在。4.3 后端 groupby 聚合与前端数据映射前端要的是“按年份均价”和“按区域均价”两组数据后端不要直接返回原始列表而是应该在stats_service里用 groupby 聚合好。下面这段是price_trend_by_district的典型实现def price_trend_by_district(df, districtNone, roomsNone, start_year1990, end_year2025): query df.copy() if district: query query[query[district] district] if rooms: query query[query[rooms] rooms] query query[(query[build_year] start_year) (query[build_year] end_year)] trend ( query.groupby(build_year)[unit_price] .agg([mean, count]) .reset_index() .rename(columns{build_year: year, mean: avg_price}) ) trend[avg_price] trend[avg_price].round(0) districts ( query.groupby(district)[total_price] .agg([mean, count]) .reset_index() .rename(columns{mean: avg_total}) ) return { trend: trend.to_dict(orientrecords), districts: districts.head(15).to_dict(orientrecords), }第一个关键点是query df.copy()。如果用query df后面query[query[district] district]会产生一个拷贝不会污染原 DataFrame但如果你后续做query.dropna(inplaceTrue)就会影响原数据。所以复制是一个安全的习惯尤其在接口并发场景下。第二个点是把to_dict(orientrecords)放在最外层这样 JSON 序列化时行数据就是 list of dict前端map可以直接消费。如果返回 pandas 的 Series 或 Timestamp 对象Flask 的jsonify会对 numpy 类型报错所以聚合后需要调用round()和reset_index()把 numpy 标量转成 Python 内建类型。5. 高分系统要过的最后一关缓存、日志与 Docker 交付数据分析系统如果每次打开页面都重新 groupby数据量一万条还能忍几十万条时接口响应会达到秒级。评分老师最常问的“性能瓶颈在哪”答案往往不在 DataFrame 本身而在重复计算。下面三个优化点属于典型的“少改代码、收益明显”的部分而且能够在源码层面直接体现工程素养。5.1 用 Flask-Caching 把热点统计提速 10 倍Flask-Caching 是 Flask 生态里的标准缓存扩展支持simple、redis等后端。对于单机演示项目simple缓存就够了不需要额外起 Redis。要点是缓存 key 必须包含筛选条件否则不同区县之间的数据会互相串。我用make_cache_key来自动生成 keyfrom flask_caching import Cache from flask import request cache Cache(config{CACHE_TYPE: simple, CACHE_DEFAULT_TIMEOUT: 120}) def make_data_key(): return request.full_path api_bp.route(/house/analysis) cache.cached(timeout60, key_prefixmake_data_key) def house_analysis(): # 原有逻辑不变request.full_path会带上查询字符串所以“朝阳区 2室”和“海淀区 3室”不会命中同一个缓存。timeout60意味着数据最多有一分钟延迟前端可以接受。如果你想在演示时主动清理缓存重启 Flask 进程是最省事的办法。这里不要写成“session 缓存”那和 HTTP session 概念会混掉。5.2 统一异常日志与 Access 日志分流默认 Flask 不会把每个请求的耗时打出来评审时讲“性能到底多少”全靠截图。用after_request钩子记录耗时是一个通用做法app.after_request def log_access(response): duration round(time.time() - g.start_time, 3) status response.status_code print(f{request.remote_addr} {request.method} {request.path} f- {status} [{duration}s], filesys.stderr) return responseg.start_time在before_request里塞进去不要把日志写到标准输出而是交给系统服务管理。容器场景下标准输出就是 Docker 日志评审时用docker compose logs -f能直接展示请求记录。这样做比直接调用logging.getLogger(werkzeug)更可控因为werkzeug的日志级别默认在开发模式下会刷掉太多无关内容。5.3 Dockerfile 与 docker-compose 一键部署源码交付时最贵的一句话是“在我这边跑不起来”。用 Docker 固定 Python 版本和系统依赖能省掉大量环境问题。下面的Dockerfile适合当前项目FROM python:3.11-slim WORKDIR /app RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD [gunicorn, -w, 2, -b, 0.0.0.0:8000, app:app]pip install这行如果下载慢可以在后面追加-i参数指定一个可用的 Python 包镜像源但不建议把镜像地址写死在源码里换环境后反而容易失败。注意这里用了gunicorn而不是 Flask 自带的run()因为app.run()不适合多进程部署。-w 2表示两个 worker如果机器只有单核建议改成-w 1。如果想先验证容器能启动也可以临时把 CMD 换成[python, app.py]确认页面和接口正常后再换回 gunicorn。docker-compose 只需要把 8000 端口暴露出去再把data/目录用 volume 挂载这样更新房价数据不用重新 build 镜像。最后一个小技巧gunicorn 启动时Flask 的__main__不会被触发所以app.run(use_reloaderFalse)写在入口文件里也不影响容器运行。但如果有人直接python app.py启动reloader 会创建额外监控进程容易让端口冲突。把use_reloaderFalse显式写上既是日志也是给后续接手的人留了“为什么这么做”的注释。每一条请求的耗时会持续打印到 stderr演示时直接贴出这段日志比口头解释“比较快”有说服力得多。本文还有配套的精品资源点击获取