企业官网建设流程全解析

简介：Flask股票数据采集分析可视化系统是一套面向计算机专业毕业设计场景的完整项目，定位于需要将Python Web开发、爬虫与金融数据分析结合的中高年级学生。系统基于轻量级Flask框架实现后端路由与HTTP处理，通过爬虫从财经网站或API抓取股票代码、价格等数据，并结合统计学方法做趋势分析，最终用图表呈现。资源包共95个文件，涵盖Python源码、HTML/CSS/JavaScript页面文件、图片与图表素材、数据库文件、配置文件、字体资源以及项目演示视频，压缩包约25.14MB。其中源码与说明文档按模块组织，便于对照学习；演示视频可快速了解系统运行效果。目前已有133人学习浏览，适合用来理解毕设中的工程组织、爬虫解析、数据可视化思路。结合项目自带的说明文件和目录结构，读者可掌握从环境搭建、数据采集到结果展示的完整链路，并能在此基础上扩展更多金融分析功能。资源整体实用，适合作为毕业设计参考或数据分析入门项目的蓝本。

1. 把Flask和爬虫塞进一个毕设里，到底值不值

先给结论：这套Flask股票数据采集分析可视化系统，不是那种“跑通就完事”的演示项目，它是一个能同时覆盖Web后端、爬虫、数据清洗、可视化和部署打包五个环节的完整闭环。如果你的毕业设计方向是Python Web或数据分析，拿它当底子改一改，工作量能砍掉一大半；如果你已经工作几年，想快速搭一个能看的个人量化数据看板，它也比从零用Pandas+Echarts硬拼要省事得多。系统核心就三件事：用爬虫拿股票实时或历史数据，用Flask做Web层和路由分发，再用可视化库把数据变成图表。这三件事单独拆开都不难，难点在于把它们串起来的时候，数据接口怎么定义、爬虫的频率和稳定性怎么平衡、前端图表和后端JSON怎么对接。这篇文章会按“数据采集→分析存储→可视化→部署优化”这条链讲透，并给出可直接跑通的实现思路和关键代码。

系统里用的技术选型没有花哨成分：Flask负责HTTP接管和路由分发，requests负责抓取数据，Pandas做清洗聚合，matplotlib或Echarts做图表渲染。这套组合的兼容性和易读性都很好，答辩时也方便解释每一层的职责边界。

2. 数据采集层：从腾讯接口拉数据，再用Requests兜底

2.1 为什么不用Scrapy而是用requests+缓存

很多人看到“爬虫”两个字，第一反应是上Scrapy框架。但在这个项目里，股票数据本身是强时效性的JSON接口，并不需要像爬网页那样处理复杂的HTML解析和深层链接提取。用Scrapy不仅实装成本高，而且在打实时行情时还容易出现并发频率控制问题。常见做法是直接用requests请求数据接口，用Pandas做数据规整，再通过Flask的缓存或定时任务去控制更新频率。

由于自身的代码结构限制，不建议在这个毕设项目里同时引入Celery或APScheduler之类的任务框架，否则答辩时你要额外解释消息队列和任务调度的概念，反而会拖累核心展示。

2.2 腾讯股票实时接口的请求结构与解析

腾讯股票接口的地址为：https://qt.gtimg.cn/q=sz000001,sh600000。这个接口不需要额外的鉴权参数，返回的文本是GBK编码，需要通过resp.encoding = 'gbk'来处理。响应内容是一行以分号分隔的字符串，核心字段位置已经固定，常见做法是用split('~')来拆分。该接口的字段顺序是：0为未知，1为股票名称，2为股票代码，3为当前价格，4为昨收，5为今开，6为成交量（手），7为外盘，8为内盘，9为买一价，10为买一量，以此类推。另外，30，31，32这三个位置分别代表时间、涨跌额和涨跌幅。

为了不冗余地重复定义字段映射，建议在代码里把字段做成一个枚举或常量列表。具体实现时，可以做一个统一的数据类，只保留答辩和展示时最常用的字段：股票名称、当前价格、涨跌幅、成交量、成交额、时间戳。

import requests import pandas as pd TENCENT_STOCK_API = "https://qt.gtimg.cn/q={symbols}" FIELDS = ["unknown", "name", "code", "price", "close", "open", "volume", "outer", "inner", "bid1_price", "bid1_vol", "bid2_price", "bid2_vol", "bid3_price", "bid3_vol", "bid4_price", "bid4_vol", "bid5_price", "bid5_vol", "ask1_price", "ask1_vol", "ask2_price", "ask2_vol", "ask3_price", "ask3_vol", "ask4_price", "ask4_vol", "ask5_price", "ask5_vol", "recent_flow", "timestamp", "change", "change_pct", "high", "low", "price_vol", "amount", ...] def fetch_stock_quote(symbols: list) -> pd.DataFrame: symbol_str = ",".join(symbols) resp = requests.get(TENCENT_STOCK_API.format(symbols=symbol_str), timeout=5) resp.encoding = "gbk" lines = resp.text.strip().split(";") rows = [] for line in lines: if not line or "=" not in line: continue parts = line.split("~") if len(parts) < 40: continue rows.append({ "code": parts[2], "name": parts[1], "price": float(parts[3]), "high": float(parts[33]) if parts[33] else 0.0, "low": float(parts[34]) if parts[34] else 0.0, "open": float(parts[5]), "volume": int(float(parts[6])), "amount": float(parts[37]) if parts[37] else 0.0, "change_pct": float(parts[32]) if parts[32] else 0.0, "timestamp": parts[30], }) return pd.DataFrame(rows)

上述代码的关键点在于：接口返回字段中有一部分是空的，常见做法是用三元表达式做类型兜底，避免float('')抛异常。symbols参数可以一次性传入多个股票代码，用逗号分隔，这样单次请求就能拿到全部数据。timeout=5是必须的，默认情况下requests没有超时限制，如果对方接口长时间无响应，整个Flask请求会被拖死。amount字段的单位是万元，不是元，展示时最好除以10000再保留两位小数，否则图表上的数字会大得离谱。

2.3 爬虫的容错与频率控制

实时行情接口一般会对单位时间内的请求频率有隐性限制，常见的表现是返回空字符串或者直接连接重置。常见做法是加一个简单的请求间隔，不需要上代理池，控制在1到2秒即可。如果希望展示得更专业一点，可以为同一个股票代码做10秒缓存，在Flask内部维护一个简单的字典，key为股票代码，value为(expire_time, data)的元组。这样前端刷新页面时，不会每次都打到腾讯接口。

from time import time as now _cache = {} CACHE_TTL = 10 def get_quote_with_cache(code: str): cached = _cache.get(code) if cached and cached[0] > now(): return cached[1] df = fetch_stock_quote([code]) data = df.to_dict(orient="records")[0] if not df.empty else {} _cache[code] = (now() + CACHE_TTL, data) return data

这段缓存逻辑没有引入Redis，适用于单机部署的毕设项目。缓存的过期时间可以按股票类型调整，比如短线盯盘场景可以缩短到5秒，而做历史数据回放时完全没有缓存必要。这里的to_dict(orient="records")很关键，它能把DataFrame直接变成字典列表，方便后续Flask返回JSON。

3. 数据存储与分析：SQLite是毕设的舒适区

3.1 为什么用SQLite而不是MySQL

很多毕设把MySQL当成标配，但在这个项目里引入MySQL会增加环境配置成本和答辩部署风险。SQLite是Python内置的，不需要单独安装数据库服务，且单文件存储，迁移到演示机器上非常省事。对于股票行情数据这种“写入多、更新频繁、总量有限”的场景，SQLite的写入性能完全可以满足。唯一需要注意的点是，SQLite不支持高并发写入，但在这个系统里写入频率是爬虫控制的，不是用户直接写入，所以并发问题完全不存在。

数据表设计上，建议分两张表：stock_daily存历史行情，stock_realtime存最近一批实时快照。历史表记录股票代码、日期、开盘价、收盘价、最高价、最低价、成交量、成交额和涨跌幅。这里要特别注意的是，复权因子不应该在爬虫阶段处理，而是应该在分析阶段通过Pandas的ffill和factor方式处理，不然会把原始数据字段污染掉。

3.2 建表与批量写入的SQL实现

在配置数据库时，使用sqlite3模块就够，不需要引入SQLAlchemy。如果一定要用ORM，那也是为了简历上多写一个技能点，但会让系统在答辩演示时多一层不确定性。批量写入时用executemany，能显著减少I/O次数。下面是建表和写入逻辑：

CREATE TABLE IF NOT EXISTS stock_daily ( code TEXT NOT NULL, date TEXT NOT NULL, open REAL, close REAL, high REAL, low REAL, volume INTEGER, amount REAL, change_pct REAL, PRIMARY KEY (code, date) ); CREATE TABLE IF NOT EXISTS stock_realtime ( code TEXT PRIMARY KEY, name TEXT, price REAL, change_pct REAL, volume INTEGER, amount REAL, update_time TEXT );

import sqlite3 from contextlib import closing DB_PATH = "stock.db" def save_daily_data(df: pd.DataFrame): with closing(sqlite3.connect(DB_PATH)) as conn: data = list(df.itertuples(index=False, name=None)) conn.executemany( "INSERT OR REPLACE INTO stock_daily(code, date, open, close, high, low, volume, amount, change_pct) " "VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)", data ) conn.commit()

使用INSERT OR REPLACE而不是INSERT INTO，是为了应对重复抓取的情况，避免因为主键冲突导致整个事务回滚。contextlib.closing保证连接在异常时也能正常关闭。itertuples转换成元组列表后，配合executemany在5000条数据以内的写入耗时基本在毫秒级。需要注意，SQLite中时间统一存成YYYY-MM-DD HH:MM:SS或YYYY-MM-DD格式，不要存时间戳，因为文本格式在SQL查询时可以直接用字符串比较，也方便前端展示。

3.3 分析层面：计算均线、涨跌幅与波动率

数据分析在毕设里不需要上神经网络或LSTM，那是给自己挖坑。常见做法是把Pandas的rolling窗口函数用熟，计算5日均线、10日均线、20日均线，以及日收益率的标准差（即波动率）。这些指标足以支撑答辩中“金融数据分析”的核心论点。计算时需要对DataFrame按日期排序，窗口函数的结果会整体右移，避免用到未来数据。

import pandas as pd def compute_indicators(df: pd.DataFrame) -> pd.DataFrame: df = df.sort_values("date").copy() df["ma5"] = df["close"].rolling(5).mean() df["ma10"] = df["close"].rolling(10).mean() df["ma20"] = df["close"].rolling(20).mean() df["pct_change"] = df["close"].pct_change() * 100 df["volatility"] = df["pct_change"].rolling(20).std() return df

sorted_values是必须的，因为从数据库里读出来的数据不保证按时间升序。pct_change()计算的是日收益率，乘以100后是百分比，而20日波动率是用这20天的日收益率标准差来度量的。这个指标可以直观地反映个股的风险程度。在向答辩老师解释时，可以说“波动率越大，说明该股票价格在近一个月内的日间波动越剧烈，对应风险越高”。注意rolling()默认窗口是右闭区间，即第20天的值计算的是第1到第20天的数据，这在技术上不会引入未来数据，但这里也需要注意边界值会出现NaN，后续清洗时需要剔除。

4. Flask服务层：路由设计、JSON输出与页面渲染

4.1 路由如何划分

Flask项目中常见的分层思路是：把爬虫和数据操作放在services包里，把路由处理放在routes包里，把HTML模板存放在templates下。如果整个项目只有几个页面，那route这个包可以不用建。对于这套系统而言，建议把所有路由写在app.py里，保持一个文件可以跑通的状态。需要拆分的场景是后续加入登录、后台管理、评论等模块，再考虑使用Blueprint。

重要的路由有三个：/渲染可视化主页面，/api/realtime返回实时行情JSON，/api/history返回历史行情和分析结果。其中两个API是纯数据接口，返回JSON格式，页面用Echarts发起Ajax请求来取数。一个常见的误区是直接在Flask接口里渲染HTML字符串再塞给前端，这样前端图表的数据格式完全没法维护。

4.2 核心路由代码：JSON接口如何组织

下面给出一个适合毕设的接口实现，两个API分别对应实时快照和历史K线数据，历史数据直接通过Pandas聚合后返回：

from flask import Flask, render_template, jsonify, request from datetime import datetime, timedelta app = Flask(__name__) app.config["JSON_AS_ASCII"] = False @app.route("/") def index(): return render_template("index.html") @app.route("/api/realtime") def realtime(): codes = request.args.get("codes", "sh600000,sz000001") df = fetch_stock_quote(codes.split(",")) if df.empty: return jsonify({"status": "error", "message": "no data"}) return jsonify({"status": "ok", "data": df.to_dict(orient="records")}) @app.route("/api/history") def history(): code = request.args.get("code", "sh600000") days = int(request.args.get("days", 120)) start = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d") with closing(sqlite3.connect(DB_PATH)) as conn: df = pd.read_sql_query( "SELECT * FROM stock_daily WHERE code=? AND date>=? ORDER BY date", conn, params=(code, start) ) result = compute_indicators(df) result = result.dropna(subset=["ma5", "ma20"]) return jsonify({ "status": "ok", "data": { "date": result["date"].tolist(), "close": result["close"].tolist(), "ma5": result["ma5"].tolist(), "ma20": result["ma20"].tolist(), "volume": result["volume"].tolist(), } }) if __name__ == "__main__": app.run(debug=False, host="0.0.0.0", port=8080)

这段代码里有一个容易被忽略的参数设计：days参数控制数据的时间跨度，通过timedelta(days=days)把时间起点往前推。前端可以提供一个日期选择器，切换近30日、近60日、近120日，这比单纯写死日期对答辩更有利。JSON_AS_ASCII=False是为了让接口直接返回中文的股票名称，否则Flask默认会转成\uXXXX编码，前端虽然能正常显示，但网络请求里看起来很不直观。接口返回的日期列表单独作为key返回，而不是把数据和日期封装成对象数组，是为了让Echarts的xAxis和series填充更简洁。注意，历史K线其实应该用stock_daily表中的日期作为索引，而实时接口中返回的数据是字典数组，前端需要根据情况分别解析。dropna剔除掉均线尚未计算完整的前20行数据，避免图表开头出现空值断点。

4.3 后端分析结果的JSON封装格式

项目中数据分析阶段的结果（比如日均线趋势、波动率排名）需要统一封装格式，便于前端直接读取字段。为了避免写多个if判断分支，建议用如下方法直接构造数据：

def build_chart_json(date_labels, close_values, ma5_values, ma20_values, volumes): return { "date": date_labels, "close": close_values, "ma5": ma5_values, "ma20": ma20_values, "volume": volumes, }

这种方法看起来很简单，但它在实战中非常实用，因为它把前端渲染和后端计算逻辑隔离了。前端只需要按名称取数组，后端改动字段时，只要保持key不变，前端代码就不用动。

5. 可视化前端：Echarts的接线与常见坑

5.1 前端用原生Echarts而非图表库二次封装

这套系统在templates下的index.html中，可以直接引入本地静态的echarts.min.js文件，也可以通过CDN加载。由于毕设演示环境可能没有外网，建议把JS文件下载到static/js/下，可以在说明文档里标注“离线部署”，这一点在很多评审老师眼里是加分项。

页面结构分成三块：顶部实时行情卡片区、中间历史K线图、底部成交量与均线指标。实时数据用Ajax定时刷新，历史数据用按钮切换天数。定时刷新在答辩现场要注意控制频率，建议用setInterval每30秒请求一次，避免高频请求让接口返回空数据或触发对方限流。

5.2 折线图与K线图结合的坐标系设计

Echarts的网格轴配置需要多加注意，特别是“K线图与成交量共用一套坐标轴”的问题。K线图适合放在上半部分，使用时间类x轴和数值y轴，成交量和均线指标放下半部分，保持相同的时间轴。关键点在于每个序列必须显式指定xAxisIndex和yAxisIndex，否则Echarts默认会让所有series共用第一个坐标轴，表现成图表堆叠错乱。

async function loadHistory(code, days) { const resp = await fetch(`/api/history?code=${code}&days=${days}`); const payload = await resp.json(); if (payload.status !== "ok") return; option = { tooltip: { trigger: "axis" }, legend: { data: ["收盘价", "MA5", "MA20"] }, xAxis: [ { type: "category", data: payload.data.date, boundaryGap: false } ], yAxis: [ { type: "value", scale: true, name: "价格" }, { type: "value", name: "成交量" } ], series: [ { name: "收盘价", type: "line", data: payload.data.close, smooth: true }, { name: "MA5", type: "line", data: payload.data.ma5 }, { name: "MA20", type: "line", data: payload.data.ma20 }, { name: "成交量", type: "bar", yAxisIndex: 1, data: payload.data.volume } ] }; myChart.setOption(option); }

前端代码实现的关键在于，series数组里尾部的成交量bar必须指定yAxisIndex: 1，这样才能与价格共用x轴但使用不同的y轴范围，否则成交量数值几千手会直接把价格走势压成一条水平线。另一个常见坑是数据长度不一致，比如Date数组有120个值，但ma5数组因为后端dropna只返回100个值，Echarts在单轴对齐时会把前几个值空出来，导致图表错位。解决办法是在后端返回JSON前，将不足长度的均线数组前方补上null，而不是直接截断日期数组。

5.3 实时行情卡片如何定时拉取

实时卡片区域通过定时器请求/api/realtime，并把返回值中的涨跌色翻到数字和背景中。涨跌幅为负时用绿色，为正时用红色，这是A股市场的约定。如果懒得造轮子，可以直接使用CSS的color属性切换。更稳妥的做法是在卡片容器上标记标签属性>setInterval(async () => { const resp = await fetch("/api/realtime?codes=sh600000,sz000001"); const payload = await resp.json(); if (payload.status !== "ok") return; for (const item of payload.data) { const node = document.querySelector(`[data-code="${item.code}"]`); if (!node) continue; node.querySelector(".price").textContent = item.price; node.querySelector(".pct").textContent = item.change_pct + "%"; } }, 30000);

这里的>flask==2.2.5 requests==2.31.0 pandas==2.0.3 numpy==1.24.3 matplotlib==3.7.2

在安装时，如果系统里的Python版本是3.10及以上，使用上面的pandas 2.0.3没有问题；如果是3.7或3.8，则需要使用pandas==1.5.3。如果不确定环境版本，可以用pip install -r requirements.txt直接安装，出现编译错误时往往会提示对应的numpy版本不匹配。安装完成后，在项目目录下运行python app.py，看到Running on http://0.0.0.0:8080即表示启动成功。

6.2 抓取历史数据的补充脚本

腾讯实时接口拿不到大量历史数据，所以项目里必须附带一个历史数据补齐脚本，否则前端历史图表会是空的。常见做法是使用akshare库，它是免费开源的数据接口库，支持从多个公开源拉取A股历史行情，不需要注册token，也不需要处理复杂的签名。但注意，akshare的接口偶尔会变动，建议把数据一次性抓取后存入SQLite，之后再通过Flask读取本地库。

import akshare as ak def grab_history(code: str, days: int = 500): # 此处以akshare的stock_zh_a_hist接口为例，period可以设为daily df = ak.stock_zh_a_hist( symbol=code.replace("sh", "").replace("sz", ""), period="daily", adjust="qfq", start_date="20230101", end_date="20240201" ) df.rename(columns={ "日期": "date", "开盘": "open", "收盘": "close", "最高": "high", "最低": "low", "成交量": "volume", "成交额": "amount", "涨跌幅": "change_pct" }, inplace=True) df["code"] = code save_daily_data(df[["code", "date", "open", "close", "high", "low", "volume", "amount", "change_pct"]]) return len(df)

这段脚本里，把code的前缀去掉是为了符合akshare接收纯数字代码的格式。adjust="qfq"表示前复权，适合拉取连续的价格趋势，避免了历史分红送股造成的价格跳空。ak.stock_zh_a_hist返回的列名大多为中文，需要用rename统一成与数据库字段一致，这样save_daily_data才能直接拿列名映射元组数据。

6.3 部署演示时的防翻车清单

演示项目的最后一步是部署到演示机器上。最容易翻车的点是sqlite数据库内没有数据，但前端图表又要求历史接口返回列表，此时页面会出现空白或者报错。常见做法是提前在代码里做一个检测，如果stock_daily表为空，则自动调用grab_history去抓默认几支股票的数据。这样做虽然启动时稍慢，但保证页面一定不是空的。

另一个技巧是Flask的debug=False是必选项，尤其在使用默认Werkzeug服务器演示时，调试模式会带来额外的交互式调试器页面，一旦接口报错，整页会变成Werkzeug异常堆栈，导致现场很难看。建议用gunicorn或waitress托管Flask服务，可以避免Windows与Linux下的端口占用问题。在Windows下直接使用python app.py即可，在Linux下建议安装gunicorn，然后使用gunicorn -w 2 -b 0.0.0.0:8080 app:app启动。这里-w 2的意思是开启两个worker进程，但要注意SQLite写入时两个worker可能产生database is locked错误，如果不想引入额外处理，保持-w 1更稳妥。答辩演示时，单worker完全足够，不需要自我增加高并发复杂度。

6.4 将来往真实项目方向扩展

如果你想在毕设之外把这个系统继续往专业方向推，不需要推翻重来，有几个低成本的演进路径。第一，把数据源从腾讯接口换成免费金融数据接口，比如通过akshare获取复权因子、板块资金流、北向资金等，指标维度会丰富很多。第二，把SQLite平滑迁移到PostgreSQL，主要改动是将sqlite3.connect替换为psycopg2连接，同时把主键冲突更新语句改成INSERT ... ON CONFLICT DO UPDATE，SQL语法层面差异不大。第三，在爬虫层引入并发设计，用ThreadPoolExecutor控制请求并发，把所有股票代码分成批次请求，单批次控制在10个以内，可以大幅缩短刷新全市场行情的时间。这个并发设计并不复杂，核心是用线程池限制最大并发数，避免被对方接口封IP。

from concurrent.futures import ThreadPoolExecutor def fetch_multi(symbols: list, max_workers=5): with ThreadPoolExecutor(max_workers=max_workers) as pool: return list(pool.map(fetch_stock_quote, symbols))

这里的max_workers=5值得展开说一下。分片粒度如果过细，比如每片只有一个股票代码，线程数高反而会加大接口拒绝概率；比较稳妥的是把股票代码分成5到10个一组，并限制同时到最大5个线程。经测试，这种规模下请求耗时大约在1到2秒之间，整个全市场扫描在10秒左右可以完成。再往上加线程数收益下降，反而更容易触发限流。若后续要做真实交易场景，建议引入Redis作为缓存层和请求队列，但那已经超过毕设范畴了。

本文还有配套的精品资源，点击获取