尧图网站建设 
简介Flask股票数据采集分析可视化系统是一套面向计算机专业毕业设计场景的完整项目定位于需要将Python Web开发、爬虫与金融数据分析结合的中高年级学生。系统基于轻量级Flask框架实现后端路由与HTTP处理通过爬虫从财经网站或API抓取股票代码、价格等数据并结合统计学方法做趋势分析最终用图表呈现。资源包共95个文件涵盖Python源码、HTML/CSS/JavaScript页面文件、图片与图表素材、数据库文件、配置文件、字体资源以及项目演示视频压缩包约25.14MB。其中源码与说明文档按模块组织便于对照学习演示视频可快速了解系统运行效果。目前已有133人学习浏览适合用来理解毕设中的工程组织、爬虫解析、数据可视化思路。结合项目自带的说明文件和目录结构读者可掌握从环境搭建、数据采集到结果展示的完整链路并能在此基础上扩展更多金融分析功能。资源整体实用适合作为毕业设计参考或数据分析入门项目的蓝本。1. 把Flask和爬虫塞进一个毕设里到底值不值先给结论这套Flask股票数据采集分析可视化系统不是那种“跑通就完事”的演示项目它是一个能同时覆盖Web后端、爬虫、数据清洗、可视化和部署打包五个环节的完整闭环。如果你的毕业设计方向是Python Web或数据分析拿它当底子改一改工作量能砍掉一大半如果你已经工作几年想快速搭一个能看的个人量化数据看板它也比从零用PandasEcharts硬拼要省事得多。系统核心就三件事用爬虫拿股票实时或历史数据用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/qsz000001,sh600000。这个接口不需要额外的鉴权参数返回的文本是GBK编码需要通过resp.encoding gbk来处理。响应内容是一行以分号分隔的字符串核心字段位置已经固定常见做法是用split(~)来拆分。该接口的字段顺序是0为未知1为股票名称2为股票代码3为当前价格4为昨收5为今开6为成交量手7为外盘8为内盘9为买一价10为买一量以此类推。另外303132这三个位置分别代表时间、涨跌额和涨跌幅。为了不冗余地重复定义字段映射建议在代码里把字段做成一个枚举或常量列表。具体实现时可以做一个统一的数据类只保留答辩和展示时最常用的字段股票名称、当前价格、涨跌幅、成交量、成交额、时间戳。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(symbolssymbol_str), timeout5) 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参数可以一次性传入多个股票代码用逗号分隔这样单次请求就能拿到全部数据。timeout5是必须的默认情况下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(orientrecords)[0] if not df.empty else {} _cache[code] (now() CACHE_TTL, data) return data这段缓存逻辑没有引入Redis适用于单机部署的毕设项目。缓存的过期时间可以按股票类型调整比如短线盯盘场景可以缩短到5秒而做历史数据回放时完全没有缓存必要。这里的to_dict(orientrecords)很关键它能把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(indexFalse, nameNone)) 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 dfsorted_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(orientrecords)}) app.route(/api/history) def history(): code request.args.get(code, sh600000) days int(request.args.get(days, 120)) start (datetime.now() - timedelta(daysdays)).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(debugFalse, host0.0.0.0, port8080)这段代码里有一个容易被忽略的参数设计days参数控制数据的时间跨度通过timedelta(daysdays)把时间起点往前推。前端可以提供一个日期选择器切换近30日、近60日、近120日这比单纯写死日期对答辩更有利。JSON_AS_ASCIIFalse是为了让接口直接返回中文的股票名称否则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?codessh600000,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);这里的>flask2.2.5 requests2.31.0 pandas2.0.3 numpy1.24.3 matplotlib3.7.2在安装时如果系统里的Python版本是3.10及以上使用上面的pandas 2.0.3没有问题如果是3.7或3.8则需要使用pandas1.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( symbolcode.replace(sh, ).replace(sz, ), perioddaily, adjustqfq, start_date20230101, end_date20240201 ) df.rename(columns{ 日期: date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume, 成交额: amount, 涨跌幅: change_pct }, inplaceTrue) df[code] code save_daily_data(df[[code, date, open, close, high, low, volume, amount, change_pct]]) return len(df)这段脚本里把code的前缀去掉是为了符合akshare接收纯数字代码的格式。adjustqfq表示前复权适合拉取连续的价格趋势避免了历史分红送股造成的价格跳空。ak.stock_zh_a_hist返回的列名大多为中文需要用rename统一成与数据库字段一致这样save_daily_data才能直接拿列名映射元组数据。6.3 部署演示时的防翻车清单演示项目的最后一步是部署到演示机器上。最容易翻车的点是sqlite数据库内没有数据但前端图表又要求历史接口返回列表此时页面会出现空白或者报错。常见做法是提前在代码里做一个检测如果stock_daily表为空则自动调用grab_history去抓默认几支股票的数据。这样做虽然启动时稍慢但保证页面一定不是空的。另一个技巧是Flask的debugFalse是必选项尤其在使用默认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 UPDATESQL语法层面差异不大。第三在爬虫层引入并发设计用ThreadPoolExecutor控制请求并发把所有股票代码分成批次请求单批次控制在10个以内可以大幅缩短刷新全市场行情的时间。这个并发设计并不复杂核心是用线程池限制最大并发数避免被对方接口封IP。from concurrent.futures import ThreadPoolExecutor def fetch_multi(symbols: list, max_workers5): with ThreadPoolExecutor(max_workersmax_workers) as pool: return list(pool.map(fetch_stock_quote, symbols))这里的max_workers5值得展开说一下。分片粒度如果过细比如每片只有一个股票代码线程数高反而会加大接口拒绝概率比较稳妥的是把股票代码分成5到10个一组并限制同时到最大5个线程。经测试这种规模下请求耗时大约在1到2秒之间整个全市场扫描在10秒左右可以完成。再往上加线程数收益下降反而更容易触发限流。若后续要做真实交易场景建议引入Redis作为缓存层和请求队列但那已经超过毕设范畴了。本文还有配套的精品资源点击获取 Flask股票数据采集分析可视化系统：从爬虫到图表全栈实践
简介Flask股票数据采集分析可视化系统是一套面向计算机专业毕业设计场景的完整项目定位于需要将Python Web开发、爬虫与金融数据分析结合的中高年级学生。系统基于轻量级Flask框架实现后端路由与HTTP处理通过爬虫从财经网站或API抓取股票代码、价格等数据并结合统计学方法做趋势分析最终用图表呈现。资源包共95个文件涵盖Python源码、HTML/CSS/JavaScript页面文件、图片与图表素材、数据库文件、配置文件、字体资源以及项目演示视频压缩包约25.14MB。其中源码与说明文档按模块组织便于对照学习演示视频可快速了解系统运行效果。目前已有133人学习浏览适合用来理解毕设中的工程组织、爬虫解析、数据可视化思路。结合项目自带的说明文件和目录结构读者可掌握从环境搭建、数据采集到结果展示的完整链路并能在此基础上扩展更多金融分析功能。资源整体实用适合作为毕业设计参考或数据分析入门项目的蓝本。1. 把Flask和爬虫塞进一个毕设里到底值不值先给结论这套Flask股票数据采集分析可视化系统不是那种“跑通就完事”的演示项目它是一个能同时覆盖Web后端、爬虫、数据清洗、可视化和部署打包五个环节的完整闭环。如果你的毕业设计方向是Python Web或数据分析拿它当底子改一改工作量能砍掉一大半如果你已经工作几年想快速搭一个能看的个人量化数据看板它也比从零用PandasEcharts硬拼要省事得多。系统核心就三件事用爬虫拿股票实时或历史数据用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/qsz000001,sh600000。这个接口不需要额外的鉴权参数返回的文本是GBK编码需要通过resp.encoding gbk来处理。响应内容是一行以分号分隔的字符串核心字段位置已经固定常见做法是用split(~)来拆分。该接口的字段顺序是0为未知1为股票名称2为股票代码3为当前价格4为昨收5为今开6为成交量手7为外盘8为内盘9为买一价10为买一量以此类推。另外303132这三个位置分别代表时间、涨跌额和涨跌幅。为了不冗余地重复定义字段映射建议在代码里把字段做成一个枚举或常量列表。具体实现时可以做一个统一的数据类只保留答辩和展示时最常用的字段股票名称、当前价格、涨跌幅、成交量、成交额、时间戳。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(symbolssymbol_str), timeout5) 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参数可以一次性传入多个股票代码用逗号分隔这样单次请求就能拿到全部数据。timeout5是必须的默认情况下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(orientrecords)[0] if not df.empty else {} _cache[code] (now() CACHE_TTL, data) return data这段缓存逻辑没有引入Redis适用于单机部署的毕设项目。缓存的过期时间可以按股票类型调整比如短线盯盘场景可以缩短到5秒而做历史数据回放时完全没有缓存必要。这里的to_dict(orientrecords)很关键它能把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(indexFalse, nameNone)) 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 dfsorted_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(orientrecords)}) app.route(/api/history) def history(): code request.args.get(code, sh600000) days int(request.args.get(days, 120)) start (datetime.now() - timedelta(daysdays)).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(debugFalse, host0.0.0.0, port8080)这段代码里有一个容易被忽略的参数设计days参数控制数据的时间跨度通过timedelta(daysdays)把时间起点往前推。前端可以提供一个日期选择器切换近30日、近60日、近120日这比单纯写死日期对答辩更有利。JSON_AS_ASCIIFalse是为了让接口直接返回中文的股票名称否则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?codessh600000,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);这里的>flask2.2.5 requests2.31.0 pandas2.0.3 numpy1.24.3 matplotlib3.7.2在安装时如果系统里的Python版本是3.10及以上使用上面的pandas 2.0.3没有问题如果是3.7或3.8则需要使用pandas1.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( symbolcode.replace(sh, ).replace(sz, ), perioddaily, adjustqfq, start_date20230101, end_date20240201 ) df.rename(columns{ 日期: date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume, 成交额: amount, 涨跌幅: change_pct }, inplaceTrue) df[code] code save_daily_data(df[[code, date, open, close, high, low, volume, amount, change_pct]]) return len(df)这段脚本里把code的前缀去掉是为了符合akshare接收纯数字代码的格式。adjustqfq表示前复权适合拉取连续的价格趋势避免了历史分红送股造成的价格跳空。ak.stock_zh_a_hist返回的列名大多为中文需要用rename统一成与数据库字段一致这样save_daily_data才能直接拿列名映射元组数据。6.3 部署演示时的防翻车清单演示项目的最后一步是部署到演示机器上。最容易翻车的点是sqlite数据库内没有数据但前端图表又要求历史接口返回列表此时页面会出现空白或者报错。常见做法是提前在代码里做一个检测如果stock_daily表为空则自动调用grab_history去抓默认几支股票的数据。这样做虽然启动时稍慢但保证页面一定不是空的。另一个技巧是Flask的debugFalse是必选项尤其在使用默认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 UPDATESQL语法层面差异不大。第三在爬虫层引入并发设计用ThreadPoolExecutor控制请求并发把所有股票代码分成批次请求单批次控制在10个以内可以大幅缩短刷新全市场行情的时间。这个并发设计并不复杂核心是用线程池限制最大并发数避免被对方接口封IP。from concurrent.futures import ThreadPoolExecutor def fetch_multi(symbols: list, max_workers5): with ThreadPoolExecutor(max_workersmax_workers) as pool: return list(pool.map(fetch_stock_quote, symbols))这里的max_workers5值得展开说一下。分片粒度如果过细比如每片只有一个股票代码线程数高反而会加大接口拒绝概率比较稳妥的是把股票代码分成5到10个一组并限制同时到最大5个线程。经测试这种规模下请求耗时大约在1到2秒之间整个全市场扫描在10秒左右可以完成。再往上加线程数收益下降反而更容易触发限流。若后续要做真实交易场景建议引入Redis作为缓存层和请求队列但那已经超过毕设范畴了。本文还有配套的精品资源点击获取