简介面向Python股票数据分析与量化入门者这套示例工程完整演示了分时图、K线图的绘制流程以及免费Level2行情的抓取和初步解析代码按模块拆分为请求封装、公共函数、主窗口调度、分时与K线绘图等部分适合需要快速搭建行情可视化原型的开发者。压缩包共20个文件以8个py源码文件为核心辅以xml工程配置、pyc缓存文件整体仅42KB体积小巧、结构清晰便于直接阅读和二次开发。资源已有13510人学习下载热度较高内容覆盖数据获取、清洗、绘图及Level2委托队列展示等关键环节并提供了tushare等免费行情接口的调用示例有助于初学者理解行情数据从请求到呈现的完整链路。该示例既可作为Python股票分析入门参考也能在此基础上扩展策略回测或实时监控功能具有实用价值。1. 为什么我用 Python 自己画分时图、K线图还要接免费 level2 行情做 A 股日内交易分析的人十有八九卡在同一个地方想看分时图的均价线、想复盘当天的十档盘口打开行情软件却只给你五档想看一眼逐笔委托还得单独付费开权限。我当时的想法很简单——既然行情接口能拿到数据Python 又能画图能不能自己拼一套分时图、K线图加免费 level2 行情的分析工具答案是能而且效果超出预期。这套东西做完之后我再也不用依赖第三方软件的截图盘中可以直接跑自己的监控脚本把十档挂单变化、大单异动、均价线偏离全部可视化。这篇笔记就把完整方案拆开讲免费 L2 行情到底能从哪拿、分时图和K线图怎么画、数据链路怎么保持稳定、以及我踩过的那些坑。2. 画图之前先搞清楚分时图和K线图各自的数据要求是什么2.1 分时图的构成与均价线算法分时图是日内交易最基础的一类图它由三部分组成每分钟的成交价格折线、当日累计的均价线、以及底部的成交量柱状图。很多人把分时图当成普通折线图来画结果画出来的图只有价格波动没有那条关键的均价线做日内参考的价值就少了一半。均价线的计算逻辑并不复杂当日累计成交额除以当日累计成交量。这个数据在付费行情里是现成的但免费接口很多只给价格和成交量不给成交额。我一般会退而求其次用成交量加权平均价来近似avg_price sum(price_i * volume_i) / sum(volume_i)这个近似在大部分时候接近真实均价线只有在单笔大单价格偏离较远时会拉开差距但作为日内强弱判断指标完全够用。如果你拿到的接口里有成交额字段直接按累计成交额除以累计成交量来算效果最准。分时图的横轴是交易时间从 09:30 到 11:30、13:00 到 15:00中间 11:30 到 13:00 的午休时段要断开。有些现成绘图函数不会自动处理这个断点画出来的图会把上午和下午连成一条直线这就是很多分时图看起来不对劲的原因。2.2 K线图的两种实现路线现成库与手绘蜡烛图K线图的实现路线比分时图多。业内最省事的做法是用 mplfinance它能直接用 pandas DataFrame 生成蜡烛图参数设置也直观。另一个常见方案是纯手绘用 matplotlib 的 Rectangle 画实体、用 Line2D 画上下影线。手绘的好处是自由度盘中实时刷新的时候你能精确控制每一根K线的新增和更新逻辑不会像现成库那样整个重绘。mplfinance 的用法很固定只要把数据整理成带 DatetimeIndex 的 DataFrame里面有 open、high、low、close、volume 五列就行import pandas as pd import mplfinance as mpf # df 必须包含列: open, high, low, close, volume # index 必须是 DatetimeIndex且按时间升序 df.index pd.to_datetime(df.index) mpf.plot( df, typecandle, styleyahoo, volumeTrue, mav(5, 10, 20), figsize(12, 7) )这段代码里的 typecandle 指定蜡烛图格式styleyahoo 是黑底红绿的经典配色volumeTrue 在底部附带成交量子图mav(5, 10, 20) 是叠加 5/10/20 周期均线。刚开始上手用现成库比较稳等你需要实时增量刷新K线尾部的时候再考虑手绘方案。2.3 绘图库选型对比谁适合回放谁适合盘中实时做这个项目之前我先对比了三个主流方案mplfinance、pyecharts、纯 matplotlib。结论是它们各有适用场景不存在全能的。mplfinance 的优势是专为金融数据设计蜡烛图、成交量、均线这些常规元素一行代码就能出适合离线回放和盘中分钟级刷新。pyecharts 的强项是网页交互鼠标悬停能看到每根K线的 OHLC 明细但它是 JavaScript 渲染适合做 Web 报告不适合在本地脚本里频繁刷新。纯 matplotlib 最灵活分时图和K线图都能画代价是代码量大所有细节都得自己处理。我的建议很直接离线分析用 mplfinanceWeb 展示用 pyecharts盘中实时监控用 matplotlib 手绘。三个方案可以共存共用同一份抓取到的行情数据只是输出端不同。3. 免费 level2 行情的真实边界与数据源选型3.1 标准 L2 行情包含什么免费渠道能拿到什么必须先说清楚一件事交易所官方定义的 level2 行情包含十档买卖盘口、逐笔委托、逐笔成交明细这是收费产品任何渠道都不可能白送给你完整版本。市面上标着“免费 L2”的渠道本质上是三类东西的混称第一类是券商对部分客户免费开放的 L2 权限。某些互联网券商为了拉新资产达标或者开户满三个月就送一年 L2但这种数据通常只能在自家 App 里看不开放外部 API 给你拉取。第二类是财经门户和行情软件提供的十档快照。它们的行情源也是从交易所买的但对外免费开放只是不是逐笔级别的推送而是每隔几秒给一次十档快照。做挂单监控、盘口异动分析这个粒度完全够用。第三类是历史 L2 数据的回放接口。部分量化数据服务商提供过去一段时间的逐笔成交和十档快照下载限制是延迟更新甚至每周更新一次只能做复盘研究不能做盘中实盘。我做的方案用的是第二类。你要明确自己的需求如果只是画分时图和K线图普通行情接口就够了十档快照是加分项如果要做盘口队列分析、大单跟踪就必须找十档快照源。3.2 行情源的采集方式HTTP 轮询与 WebSocket 推送免费 level2 快照的采集方式主要有两种HTTP 轮询和 WebSocket 推送。HTTP 轮询是每隔几秒向服务端请求一次最新快照优点是实现简单、调试直观缺点是拿不到逐笔变动且高频轮询容易被限流。WebSocket 推送是建立长连接服务端把窗口内的盘口变化推给你实时性更好但需要处理心跳、重连、数据乱序这些麻烦事。对于个人开发者我一般建议先用 HTTP 轮询跑通全流程再考虑 WebSocket 升级。盘中抓取快照的频率控制在 3 到 5 秒一次就足够十档挂单在几十秒内不会有太大变化3 秒刷新已经能捕捉到大部分盘口异动。低于 1 秒的轮询免费接口基本都会封你 IP。一个稳定的采集框架至少包含三部分定时器、请求重试、数据落盘。定时器用 schedule 或简单的 time.sleep 循环都可以重试用指数退避数据落盘用 CSV 或 SQLite避免程序崩溃后什么都没留下。3.3 快照数据的结构设计把十档盘口整理成标准格式无论从哪个渠道拿数据最终都要整理成统一的内部结构。我一般按如下字段设计字段类型说明timestampdatetime快照时间用于分时图和K线对齐symbolstr证券代码pricefloat最新成交价volumeint当日累计成交量amountfloat当日累计成交额bid_priceslist十档买价从一档到十档bid_volumeslist十档买量从一档到十档ask_priceslist十档卖价ask_volumeslist十档卖量转换成 DataFrame 之后分时图只需要 timestamp、price、volume 三列K线图需要把连续快照按分钟聚合出 OHLC盘口分析需要 bid_prices 和 ask_prices 做差计算买卖压力。统一格式的好处是换数据源时只改采集层绘图和分析层完全不动。4. 核心代码实现抓取、分时图、K线图一条链路跑通4.1 抓取免费十档快照的最小可运行代码下面这个函数是抓取快照的最小实现我刻意去掉了所有黑魔法保留最核心的请求逻辑import requests import json import time import pandas as pd # 假设你的数据源提供了一个按证券代码返回JSON的接口 # 这里的 URL 是占位符你需要替换成实际申请的地址 API_URL https://your-free-l2-endpoint.example/api/snapshot?symbol{symbol} def fetch_snapshot(symbol): 抓取单只证券的十档快照返回标准化dict url API_URL.format(symbolsymbol) # headers 里通常需要带上客户端标识否则会被网关拒绝 headers { User-Agent: Mozilla/5.0 (research-project), Referer: https://your-data-source.example/ } try: resp requests.get(url, headersheaders, timeout3) resp.raise_for_status() raw resp.json() except requests.exceptions.RequestException as e: # 失败时返回空标记由上层决定是否重试 print(f[{time.strftime(%H:%M:%S)}] request failed: {e}) return None content raw.get(data) or raw snapshot { timestamp: pd.Timestamp.now(), symbol: symbol, price: content[price], volume: content[volume], amount: content.get(amount, 0), bid_prices: content[bid_prices][:10], bid_volumes: content[bid_volumes][:10], ask_prices: content[ask_prices][:10], ask_volumes: content[ask_volumes][:10], } return snapshot这段代码的逻辑不复杂但有几个地方必须说明。timeout3 是硬性约束免费接口经常因为负载高而响应变慢不设超时会让整个采集线程卡死。resp.raise_for_status() 会把 HTTP 4xx/5xx 直接变成异常抛出避免你拿到一个错误页面还在傻乎乎地解析 JSON。bid_prices[:10] 是切一刀防止某些接口返回多于十档的数据统一截断前 10 档保持数据结构稳定。4.2 分时图的绘制代码与均价线计算分时图我用纯 matplotlib 实现因为盘中刷新时只需要更新线条的数据不需要重建整个图。下面的代码把价格线、均价线、成交量一次性画出来import matplotlib.pyplot as plt import matplotlib.dates as mdates import numpy as np def plot_intraday(df): # df 需要包含: timestamp(时间), price(价格), volume(成交量) # 先用成交量加权平均近似均线实际可用成交额/成交量替代 total_vol df[volume].cumsum() weighted_sum (df[price] * df[volume]).cumsum() avg_line weighted_sum / total_vol fig, ax plt.subplots(figsize(12, 6)) ax2 ax.twinx() # 右侧Y轴画成交量 ax.plot(df[timestamp], df[price], color#1f77b4, lw1.2, labelprice) ax.plot(df[timestamp], avg_line, color#ff7f0e, lw1.0, labelavg) # 用淡色填充价格与均线的偏离区间直接反映多空强弱 ax.fill_between( df[timestamp], df[price], avg_line, wheredf[price] avg_line, colorred, alpha0.2 ) ax.fill_between( df[timestamp], df[price], avg_line, wheredf[price] avg_line, colorgreen, alpha0.2 ) # 成交量画成柱状颜色随涨跌变化 colors np.where(df[price] df[price].shift(1), #d62728, #2ca02c) ax2.bar(df[timestamp], df[volume], colorcolors, alpha0.4, width0.02) ax.xaxis.set_major_formatter(mdates.DateFormatter(%H:%M)) ax.set_xlabel(time) ax.set_ylabel(price) ax.legend(locupper left) fig.autofmt_xdate() return fig, ax代码里的 fill_between 是分时图的核心增强点价格在均价线上方且持续扩张时红色区域越来越大说明多头掌控局面反过来绿色区域扩张就是空头压制。颜色的选择df[price] df[price].shift(1)判断当前分钟相对上一分钟是涨是跌用于给成交量柱上色。width0.02 是因为时间轴以天为单位一根分钟柱的合理宽度约等于 1/240 天约 0.0040.02 适合看整体形态你按实际分钟数调整即可。4.3 用 mplfinance 生成K线图的完整流程K线数据需要把快照按分钟聚合每分钟的第一笔价格就是 open最后一笔是 close期间最高最低分别是 high 和 low。下面的代码演示从快照列表到K线图的完整流程import mplfinance as mpf def snapshots_to_ohlc(snapshots): 把快照列表聚合成分钟K线DataFrame df pd.DataFrame(snapshots) df[timestamp] pd.to_datetime(df[timestamp]) df.set_index(timestamp, inplaceTrue) agg df[price].resample(1min).ohlc() agg[volume] df[volume].resample(1min).last() - df[volume].resample(1min).first() agg.dropna(inplaceTrue) return agg def plot_kline(agg): mpf.plot( agg, typecandle, volumeTrue, styleyahoo, mav(5, 10, 20), ylabelprice, ylabel_lowervolume, figsize(12, 7), datetime_format%H:%M )这里最容易出错的地方是成交量的聚合。快照里的 volume 是当日累计值不是单分钟成交量所以必须用每分钟最后一笔的累计值减去第一笔的累计值才能还原出该分钟的成交量增量。如果你直接对 volume 做 resample 的 mean 或 sum得到的量柱和行情软件对不上复盘时差异很明显。4.4 盘中实时刷新定时抓取与增量更新的配合盘中跑实时监控时我一般采用一个主循环控制节奏每 3 秒抓一次快照每 60 秒聚合一次K线每次抓取后同时更新分时图和盘口状态。伪代码结构如下import time INTERVAL_SNAPSHOT 3 # 快照轮询间隔单位秒 INTERVAL_KLINE 60 # K线聚合间隔单位秒 def run_realtime(): snapshots [] last_kline_ts time.time() while True: start time.time() snap fetch_snapshot(600000) if snap is not None: snapshots.append(snap) update_intraday(snap) # 增量更新分时图 update_dom(snap) # 更新十档盘口显示 if time.time() - last_kline_ts INTERVAL_KLINE: latest_minute snapshots_to_ohlc(snapshots) plot_kline(latest_minute) last_kline_ts time.time() # 减去已经消耗的请求时间保证实际间隔接近设定值 elapsed time.time() - start time.sleep(max(0.1, INTERVAL_SNAPSHOT - elapsed))这个循环的要点在于 sleep 的时间是动态的如果请求本身耗时 1.2 秒下一次 sleep 就只睡 1.8 秒保证两次请求的间隔始终接近 3 秒而不是“睡眠 3 秒加请求耗时”越跑越慢。max(0.1, ...)防止某次请求异常快时 sleep 变成负数导致空转。5. 避坑指南免费行情和自绘K线最常遇到的五个问题5.1 免费接口返回的价格单位是“分”而不是“元”现象画出来的分时图价格比行情软件高了 100 倍K线图的高点到了六位数。原因部分财经门户的接口为了省流量价格字段直接用整数保存单位是分。如果你不做转换直接当成元的数值来用必然出现全线漂移。解决在 fetch_snapshot 里做统一换算把 price 字段除以 100 再写入标准结构。所有下游绘图和分析都只认标准结构不再关心原始渠道的单位。5.2 分时均价线在数据缺失时变成“假的直线”现象盘中均价线一段一段水平延伸明显不是平滑的曲线和行情软件对不上。原因免费接口的 amount 字段经常缺失或者为 0只能用成交量加权平均近似。当某分钟成交量很小或接口漏了一次推送时加权均值会突然跳到上一个值看起来像台阶。更隐蔽的问题是当接口返回的累计成交量突然清零比如服务端重启均价线会被瞬间拉回起跑线。解决对累计值做单调性检查发现 volume 比上一次快照还小就丢弃这次快照而不是直接参与计算。均价线计算时只保留 volume 0 的分钟避免零成交量污染权值。5.3 K线图的复权问题导致视觉跳空现象某天K线图上突然出现一个巨大的向下缺口但当天并没有除权除息股价走势也很平稳。原因免费数据源给的原始价格是未复权价格。遇到分红送股当天开盘价会“跳空低开”到除权价但实际上持仓者没有亏损。用未复权数据画长周期K线每个除权日都会出现一个假缺口复盘时会误导判断。解决做长周期分析时用前复权数据也就是把历史价格按最新股本进行缩放。如果你的接口不直接提供复权因子可以拉取当日涨跌幅来反推除权日的前收盘价乘以1 涨跌幅得到当日的理论开盘参考价再据此修正历史价格。5.4 高频轮询触发限流整个采集链路瘫痪现象跑了一上午都正常下午突然所有请求都返回 403 或超时十分钟后恢复。原因免费接口通常有单 IP 的请求频率限制超过阈值会直接封禁一段时间。这类限制不会在文档里写明只能通过日志观察到特征。解决第一优先级是把轮询间隔调到 3 秒以上不要贪快。第二优先级是加随机偏移比如 2.7 到 3.3 秒之间随机避免你的请求节奏被网关识别为机器人。第三优先级是设置失败熔断连续失败 5 次后暂停 60 秒而不是一直高频重试加重封禁。5.5 停牌和异常交易日的空数据导致绘制报错现象证券停牌一天当天没有快照数据。聚合K线时该分钟为空绘图时整张图报 ValueError 或直接崩溃。原因免费接口在停牌时返回空内容或字段缺失你如果按正常格式解析会因为缺 key 而抛 KeyError或者因为读到一个全空的 DataFrame 而让绘图函数无法工作。解决在 fetch_snapshot 里检查返回内容的完整性缺少 price 或 volume 字段时直接返回 None不进入快照列表。在 snapshots_to_ohlc 里给 resample 加 min_count1 参数确保只有真正有数据的分钟窗口才会生成K线。这样停牌日顶多缺一根K线不会让整个程序崩溃。6. 升级方向如何验证你的行情数据正确以及把链路变成可复盘的工具6.1 用总成交量一致性校验抓取质量这是我最先做的一步。免费接口的数据质量参差不齐你可以用分时累计成交量和日线收盘后的官方成交量做对比偏差超过 2% 基本可以判定接口数据有缺失。校验方法很简单把当天所有分钟成交量累加和收盘后从日线接口拿到的 volume 对比。daily_volume fetch_daily_volume(600000) # 收盘后权威成交量 intraday_volume int(df_intraday[volume].iloc[-1]) # 分时累计量 diff_ratio abs(intraday_volume - daily_volume) / daily_volume if diff_ratio 0.02: print(f数据偏差异常: {diff_ratio:.2%})持续记录这个偏差你很快就能知道数据源什么时候开始丢数据。我遇到过某个接口下午 14:30 之后偶尔漏推累计量偏差到 5% 以上后来加了这个校验才抓到规律。6.2 把行情流落地到 SQLite做离线回放盘中抓到的快照如果只存在内存里程序一重启就全没了。我一般把标准化快照直接写入 SQLite一张表存快照明细一张表存分钟聚合。SQLite 不需要额外启服务单文件存取切合个人项目的体量。import sqlite3 conn sqlite3.connect(market_data.db) conn.execute( CREATE TABLE IF NOT EXISTS snapshots ( ts TEXT, symbol TEXT, price REAL, volume INTEGER, amount INTEGER ) ) def save_snapshot(snap): if snap is None: return conn.execute( INSERT INTO snapshots VALUES (?, ?, ?, ?, ?), (str(snap[timestamp]), snap[symbol], snap[price], snap[volume], snap[amount]) ) conn.commit()落库之后你可以随时把某一天的数据重新读出来重新画分时图、重新调节K线的均线参数不用守在盘中等行情。这也是我复盘异动股最常用的方式。6.3 一个值得长期维护的核心习惯做这个项目半年后我的最大体会是绘图代码其实只占工作量的一小部分真正的功夫在数据清洗和边界处理上。分时图的“假均价线”、K线的除权缺口、轮询限流每一个坑都能让产出的图看起来正常但实际是错的。我现在的习惯是每天收盘后跑一次数据校验脚本把偏差异常的日子标记出来复盘时先跳过这些脏数据免得被表面的形态带偏方向。如果你想在这个方向深入下一步可以做的有三件事一是把十档快照的买卖量差做成持续的盘口监控曲线捕捉大单托底和出货的痕迹二是把分钟聚合后的 OHLC 接入到自写的均线策略里做回测三是把看似玄学的量价背离用数据量化比如计算价格新高时成交量是否同步创新高。每一件都能单独成为一个小项目。这套从免费行情抓取到自绘分时图、K线图的链路是它们共同的地基。希望帮到你。本文还有配套的精品资源点击获取