简介一份面向Python量化与金融可视化入门者的可运行示例工程聚焦股票分时图、K线图的绘制与免费Level 2行情的抓取适合已掌握Python基础、想动手实践行情获取与图表呈现的开发者。压缩包共20个文件、约42KB以8个py源码为主体含行情请求封装、公共方法模块、界面入口以及分时图与K线图绘制脚本5个xml、1个iml与1个name为开发环境与项目配置文件另有5个pyc编译缓存便于直接对照运行。目前已有13564人学习下载。读者可从中获得Tushare等接口取数与鉴权的完整调用链、基于matplotlib与mplfinance的分时线和蜡烛图实现、逐笔成交与买卖五档数据的解析思路以及模块分层与配置管理的组织方式可作为二次开发与行情工具搭建的起点。1. 盘中盯着一根分时线刷半天不如自己拉一份数据画出来前阵子有位做量化的朋友发来一张截图问能不能不打开行情客户端直接用 python 把某只票当天的分时图和最近六十根 K 线图拼到一张画布上顺手把逐笔成交落成 csv。这事儿听着简单真动手就会撞上一串细节分时图的基准价拿哪个、午休那一个半小时横轴怎么接、免费渠道到底能拿到几档盘口。这份资源就是围着这几件事转的一份能跑通的取数加绘图脚本分时图、K线图两条渲染链路各自独立逐笔成交和盘口快照各有入口。适合手里有点 Python 底子、想自己搭复盘工具或者做盘中监控的开发也适合刚碰行情数据、想搞清楚 tick 和 bar 到底怎么换算的人。下面按我拆这份源码的顺序走一遍从目录骨架一直讲到怎么改成定时任务。2. 技术选型与目录结构取数、清洗、渲染三层怎么切开这套脚本最容易写歪的地方是把 requests 请求、pandas 清洗、matplotlib 画图全塞进一个main.py跑通一次之后再想加个指标就得整篇改。原项目的分法是三层fetch只管把原始 JSON 拿回来并落盘calc只管把 tick 聚合成 bar、算均线chart只接受标准化的 DataFrame 出图。三层之间靠一个固定的字段 schema 通信换数据源只要改fetch换画图库只要改chart互不牵连。这种切法看着啰嗦但等你第二天想把手里的数据换成另一个品种或者另一段历史就知道省了多少事。2.1 依赖怎么选requests pandas 打底mplfinance 和 pyecharts 各管一头取数这层没必要上重型框架requests加一个带重试的 Session 就够行情接口基本是 GET JSON异步并发在这里收益不大反而容易触发限频。数据清洗压给pandastick 转 bar 一行resample能解决大半问题。绘图分两条路要出静态 PNG 就走mplfinance它是 matplotlib 的封装K 线、成交量副图、均线叠加都有现成接口要出可以缩放拖拽的 HTML 就走pyecharts适合把图发给不看代码的人。# requirements.txt requests2.31.0 pandas2.1.4 numpy1.26.2 matplotlib3.8.2 mplfinance0.12.10b0 pyecharts2.0.4 APScheduler3.10.4mplfinance的版本号带 beta 后缀别看到就跳过0.12 这一支用了两年多make_addplot、make_marketcolors这些关键接口都在里面0.11 老版本缺参数。pyecharts更要注意1.x 和 2.x 的Grid、Kline初始化方式差得挺远网上抄来的示例有一半跑不通锁死主版本号能省掉很多莫名其妙的报错。APScheduler只在最后一章做定时拉取时才用得上第一遍跑通可以先不装。2.2 目录骨架四个包分开别把取数和画图揉在一个文件里stock_chart/ ├── config.py # 接口域名、UA、股票池、输出路径 ├── fetch/ │ ├── http.py # Session、重试、限速 │ ├── quote.py # 分时、日线、盘口快照三个取数函数 │ └── store.py # 落 csv / parquet ├── calc/ │ ├── resample.py # tick - 1min bar │ └── indicator.py # 均价线、MA5、MA10 ├── chart/ │ ├── theme.py # 红涨绿跌配色、中文字体 │ ├── intraday.py # 分时图 │ └── kline.py # K 线图静态 HTML ├── data/ │ ├── raw/ # 原始 json按 trade_date 分目录 │ └── parquet/ # 标准化后的 bar ├── output/ # png / html └── main.py # 命令行入口data/raw保留原始响应是个不起眼但很值的设计。行情接口偶尔会在盘中修改返回结构或者某天的分笔数据缺了几笔出问题时能拿原始报文对一遍不用重新请求一遍接口再去猜。theme.py单独拎出来是因为配色在两个绘图文件里都要用复制粘贴两份迟早改漏一处。main.py只做参数解析和流程编排不写业务逻辑这样后面接定时任务时可以直接调函数不必subprocess起子进程。2.3 数据结构一根 K 线落库前必须定死的九个字段三层之间传的 DataFrame字段名和类型必须先钉死。我见过有人分时接口返回time: 09:30字符串、日线接口返回2024-01-05混在一个 DataFrame 里排序直接乱套。下面这套 schema 是从原项目抄的够用且不冗余。字段类型含义说明codestr六位代码必须补零000001别变成1dtdatetime64bar 起始时间统一到分钟粒度不带时区openfloat64开盘价分时图的第一个 tick 价highfloat64最高价区间内最大值lowfloat64最低价区间内最小值closefloat64收盘价区间内最后一个 tick 价volumeint64成交量手接口给的常是股记得除 100amountfloat64成交额元均价线的分子prev_closefloat64昨收分时图基准线全表同值import pandas as pd BAR_COLUMNS [code, dt, open, high, low, close, volume, amount, prev_close] def normalize_bar(df: pd.DataFrame, code: str, prev_close: float) - pd.DataFrame: 把任意来源的 bar 数据对齐成统一 schema out df.copy() out[code] str(code).zfill(6) # 代码补零字典 key 也对齐 out[dt] pd.to_datetime(out[dt]) # 字符串时间统一解析 for col in [open, high, low, close, amount]: out[col] pd.to_numeric(out[col], errorscoerce).astype(float64) out[volume] (pd.to_numeric(out[volume], errorscoerce) .fillna(0).astype(int64)) out[prev_close] float(prev_close) return (out[BAR_COLUMNS] .sort_values(dt) .drop_duplicates(dt) # 接口重试可能带来重复行 .reset_index(dropTrue))errorscoerce是关键行情接口偶尔会在某个字段塞-或者空串直接astype(float)会整列报错转成 NaN 之后再决定是丢行还是前值填充。drop_duplicates(dt)防的是请求超时后重试拿回两份重叠数据的情况不去重的话resample出来的成交量会翻倍而且这种错误肉眼很难发现。zfill(6)看着小题大做但股票代码当字典 key 用时1和000001是两个 key缓存会莫名失效。3. 行情抓取分笔、分钟线、盘口快照三条取数链路原项目没有把取数写死在一个库上而是自己封了一层 HTTP 入口好处是接口地址变了只改config.py坏处是得自己处理重试和限频。三条链路的数据形态差别很大分笔成交是几百到几千条流水分钟线是当天固定的 240 个点盘口快照是时间序列上每次请求一个瞬间的切片。三者混在一起处理会很乱分开各写各的反而清楚。3.1 统一请求入口重试、限速、UA 三件套公开行情接口对请求头比对参数敏感得多。少了User-Agent或者Referer对不上很多时候不是返回 403而是返回 200 加一段 HTML 或者空数组排查起来就是个黑匣子。把 Session 抽出来统一配好比在每个请求函数里各写一遍靠谱。import random import time import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry BASE https://api.example-quote.invalid # 换成你实际使用的公开行情域名 UA (Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36) def build_session(pool_size: int 8, retry: int 3) - requests.Session: s requests.Session() cfg Retry( totalretry, # 总重试次数 backoff_factor0.6, # 退避第 n 次等待 0.6 * 2^(n-1) 秒 status_forcelist[429, 500, 502, 503, 504], allowed_methods[GET], ) s.mount(https://, HTTPAdapter(max_retriescfg, pool_maxsizepool_size)) s.headers.update({User-Agent: UA, Referer: BASE /}) return s _session build_session() def get_json(path: str, params: dict | None None, timeout: float 6.0) - dict: resp _session.get(BASE path, paramsparams, timeouttimeout) resp.raise_for_status() time.sleep(random.uniform(0.05, 0.15)) # 抖动别把接口打成刷子 return resp.json()backoff_factor配 0.6意味着第一次失败等 0.6 秒、第二次 1.2 秒、第三次 2.4 秒这个量级对付偶发 5xx 够用再长就拖慢整批任务。pool_maxsize8控制在同一个连接池里的并发数一次刷几十只票时连接复用能明显减少握手开销。末尾那个random.uniform抖动是血泪经验固定间隔的连续请求在某些接口上会被识别成脚本行为随机几十一百毫秒就能规避掉大部分情况。3.2 分笔转分钟线tick 聚合里最容易错的四个细节分笔成交是最原始的粒度一条记录一个成交字段通常是时间、价格、成交量、成交额。要画分时图或者补分钟线就得按时间桶聚合。pandas.resample能做但有几个坑不处理会出脏数据。import pandas as pd def ticks_to_bars(ticks: pd.DataFrame, freq: str 1min) - pd.DataFrame: 分笔成交聚合为分钟 bar df ticks.copy() df[time] pd.to_datetime(df[time]) # 坑1集合竞价的 0 价成交行留着会把 low 拉成 0 df df[(df[price] 0) (df[volume] 0)] df df.set_index(time).sort_index() bars df.resample(freq, labelleft, closedleft).agg( open(price, first), high(price, max), low(price, min), close(price, last), volume(volume, sum), amount(amount, sum), ) # 坑2午休和停牌期间没有成交resample 会造出全 NaN 的空桶 bars bars.dropna(subset[open]) return bars.reset_index().rename(columns{time: dt})labelleft和closedleft这两个参数决定了桶的时间戳标在起点还是终点。用左闭左标09:30:00到09:30:59的成交归到09:30那根 bar跟行情客户端的显示口径一致用默认的右标整张图会偏移一分钟肉眼看不出来但跟其他数据源对不齐。坑一是集合竞价。开盘前那笔虚拟成交价常常是 0 或者等于昨收混进去之后low变 0分时图的纵轴被拉到地板整根线贴着顶端走。过滤条件写price 0比写price prev_close * 0.9更稳后者遇到除权日会误杀。坑二是空桶。resample是等距切分11:30 到 13:00 这段没人交易照样给你切出 90 个空桶dropna掉之后一天刚好 240 根。这一步不做的话分时图的横轴会多出一段空白均价线也跟着算歪。坑三是成交量的单位。接口给的volume经常是「股」而客户端显示的是「手」1 手 100 股。直接画出来的量柱比客户端高 100 倍很多人第一反应是接口数据错了其实就是没除。3.3 免费 level2 的边界五档快照能拿十档逐笔委托基本拿不到标题里写了「免费 level2」这里得说实话公开渠道能稳定拿到的通常是五档买卖盘快照加逐笔成交真正的十档委托、逐笔委托、委托队列这些要么需要券商权限要么得走付费通道。原项目里盘口部分就是按五档写的字段结构固定拿不到更高档位就维持现状不影响分时和 K 线的功能。def parse_snapshot(raw: dict) - dict: 把五档盘口响应压成扁平字典方便直接进 DataFrame snap { code: str(raw[code]).zfill(6), ts: pd.to_datetime(raw[time]), last: float(raw.get(price) or 0), } for i in range(1, 6): # 价格字段可能是字符串成交量字段单位多为「股」 snap[fbid{i}_price] float(raw.get(fbid{i}) or 0) snap[fbid{i}_vol] int(float(raw.get(fbid{i}_vol) or 0)) // 100 snap[fask{i}_price] float(raw.get(fask{i}) or 0) snap[fask{i}_vol] int(float(raw.get(fask{i}_vol) or 0)) // 100 return snapraw.get(...) or 0这个写法是为了兼容两种缺失键不存在返回None键存在但值是空字符串或者0.00。后者float能转但没意义统一归零更省事。买卖五档的价量一共 20 个字段加上最新价和时间戳一次快照 23 列盘中每 3 秒采一次、一天 4 小时就是 4800 行量不大直接追加写 parquet 就行。真要拿十档思路一样只是range(1, 6)改成range(1, 11)前提是你手上那个接口确实返回了这些字段——很多标着「十档」的接口后五档全是 0。4. 绘图核心分时图与 K 线图的两套渲染逻辑出图这块两条链路差别挺大。分时图要的是当天 240 个点的连续曲线加一条均价线横轴是交易时间但必须跳过午休K 线图要的是带实体和影线的蜡烛配上成交量和均线横轴是交易日。共用一套代码会互相迁就分开写各用各的顺手工具反而快。4.1 分时图昨收基准线、均价线、午休断点分时图看着简单三个细节决定成图质量。均价线不是价格的移动平均而是「累计成交额 / 累计成交量」算的是当天所有成交的加权均价所以必须用cumsum。基准线取昨收价格在线上是红、线下是绿视觉上一眼能看出强弱。横轴千万不能用真实时间轴否则午休那 90 分钟会留一大段空白两条曲线被拉断。import matplotlib matplotlib.use(Agg) # 无 GUI 环境跑图必须指定后端 import matplotlib.pyplot as plt import pandas as pd def plot_intraday(minute_df: pd.DataFrame, prev_close: float, code: str, out: str output/intraday.png) - str: df minute_df.sort_values(dt).reset_index(dropTrue) df[avg] df[amount].cumsum() / df[volume].cumsum().replace(0, pd.NA) x range(len(df)) # 用序号当横轴天然跳过午休 ticks [0, 59, 119, 120, 179, 239] # 09:30 / 10:29 / 11:29 / 13:00 / 14:00 / 15:00 labels [09:30, 10:29, 11:29, 13:00, 14:00, 15:00] fig, ax plt.subplots(figsize(12, 5), dpi120) ax.plot(x, df[close], lw1.1, color#1f77b4, labelprice) ax.plot(x, df[avg], lw1.0, color#ff9f1c, labelavg) ax.axhline(prev_close, ls--, lw0.8, color#888888) ax.set_xticks(ticks) ax.set_xticklabels(labels) ax.set_xlim(0, len(df) - 1) ax.set_title(f{code} intraday, fontsize11) ax.grid(axisy, ls:, lw0.5, alpha0.6) fig.tight_layout() fig.savefig(out) plt.close(fig) return outcumsum()那行要注意分母可能为 0开盘第一分钟如果一笔成交都没有直接相除会得到inf或者NaN并把整条线打断用replace(0, pd.NA)挡一下。ticks是按 240 个点硬编码的第 0 个点是 09:30第 59 个是 10:29第 119 个是 11:29第 120 个就是下午 13:00 的第一分钟正好跨过午休。如果某天停牌或者数据缺失导致点数不是 240这套刻度会错位稳妥做法是按df[dt].dt.strftime(%H:%M)反查索引硬编码只适合数据完整的场景。matplotlib.use(Agg)这行在服务器上跑是必须的不指定后端matplotlib 会尝试连图形界面容器里直接抛错。本地开发看着没问题一上服务器就翻车很典型。4.2 K 线图mplfinance 叠加均线与成交量面板K 线用mplfinance比手工画矩形省事太多。它的plot()一个调用能出蜡烛、成交量副图、均线叠加还能直接存文件。唯一要改的是配色默认是欧美那套绿涨红跌A 股用户看着别扭。import mplfinance as mpf import pandas as pd def plot_kline(bars: pd.DataFrame, n: int 60, out: str output/kline.png) - str: df bars.tail(n).copy() df df.set_index(dt)[[open, high, low, close, volume]] df.index.name Date ma5 df[close].rolling(5).mean() ma10 df[close].rolling(10).mean() # 不指定 panel 就叠加在主图指定 panel1 才会画到副图 apds [ mpf.make_addplot(ma5, color#e4572e, width0.9), mpf.make_addplot(ma10, color#4b7bec, width0.9), ] mc mpf.make_marketcolors( up#e4572e, down#26a69a, # 红涨绿跌反着来会被同事吐槽 edgeinherit, wickinherit, volumein, ) style mpf.make_mpf_style( marketcolorsmc, gridstyle--, rc{font.family: SimHei, axes.unicode_minus: False}, ) mpf.plot(df, typecandle, stylestyle, addplotapds, volumeTrue, panel_ratios(4, 1), figratio(14, 8), savefigdict(fnameout, dpi120, bbox_inchestight)) return outpanel_ratios(4, 1)控制主图和成交量副图的高度比4:1 是比较舒服的比例成交量再高就喧宾夺主。volumeTrue会自动开第二个 panel不用自己make_addplot。make_marketcolors里volumein表示量柱颜色跟随当根 K 线的涨跌而不是固定色视觉上更连贯。rc里那两个参数是给中文和负号用的。不设font.family标题里的股票名会变成方框不设axes.unicode_minusFalse纵轴负号显示成方块。这两个问题在本地 Windows 上可能正常换到 Linux 服务器上必现属于跨平台必踩的一脚。4.3 交互式输出pyecharts 出 HTML适合发给不看代码的人静态 PNG 适合存档和批量生成但要给别人看、要缩放看细节HTML 更合适。pyecharts的 K 线组件直接用配一个Grid把成交量和 K 线上下拼起来。from pyecharts import options as opts from pyecharts.charts import Grid, Kline, Line def kline_html(bars, out: str output/kline.html) - str: df bars.tail(120) dates df[dt].dt.strftime(%Y-%m-%d).tolist() # 顺序是 [open, close, low, high]跟直觉不一样写反了图会乱 data df[[open, close, low, high]].round(2).values.tolist() k (Kline() .add_xaxis(dates) .add_yaxis(K, data, itemstyle_optsopts.ItemStyleOpts( color#e4572e, color0#26a69a)) .set_global_opts( xaxis_optsopts.AxisOpts(is_scaleTrue), yaxis_optsopts.AxisOpts(is_scaleTrue, splitarea_optsopts.SplitAreaOpts()), datazoom_opts[opts.DataZoomOpts(range_start60, range_end100)], )) vol (Line() .add_xaxis(dates) .add_yaxis(volume, df[volume].tolist(), yaxis_index1, linestyle_optsopts.LineStyleOpts(width0))) grid (Grid(init_optsopts.InitOpts(width1200px, height620px)) .add(k, grid_optsopts.GridOpts(pos_bottom55%)) .add(vol, grid_optsopts.GridOpts(pos_top55%))) grid.render(out) return out[open, close, low, high]这个顺序是pyecharts的硬要求跟mplfinance的列顺序不同从这边切过去时最容易写反写反的表现是影线戳到错误方向图能出来但完全不对。datazoom_opts的range_start60让 HTML 打开时默认显示后 40% 的数据不用手动拖。成交量用Line加零宽度线型来模拟柱状是Grid组合里的常见凑法比再造一个 Bar 再对齐双轴省事。5. 排查清单这套脚本跑不起来时先看这五处下面五条是我照着原项目复现时真踩过的按「现象 → 原因 → 解决」记下来遇到问题对照着看比自己盯代码快。5.1 取数链路的三次翻车现象一请求返回 200json()解析出来是空的或者是一段 HTML。原因基本可以锁定在请求头。公开行情接口对User-Agent和Referer的校验比参数严格缺了就直接给你一个空壳响应状态码还是 200raise_for_status()拦不住。另一种情况是短时间内连续请求同一只票接口没报错但返回空数组属于软限频。 解决把UA和Referer落实到 Session 的默认 headers所有请求走同一个 Session在get_json末尾加 50 到 150 毫秒的随机 sleep第一次接到空响应时先打印resp.text[:200]看看返回的到底是什么别急着改解析逻辑。现象二成交量比行情客户端大整整 100 倍。原因是接口字段单位是「股」客户端显示「手」。这在分笔和盘口两处都会出现而且两处的字段名还不一样一个叫volume一个叫vol。 解决在normalize_bar里统一做// 100盘口解析里单独处理一次。别在下游画图时再除除以 100 的取整在哪一层做决定了你会不会在分钟线聚合时把零头丢掉——正确做法是原始层保留股落库前一次性转换。现象三分时图纵轴被拉到地板曲线贴在最顶上。这是集合竞价的 0 价成交混进了 tick 数据low被拉成 0。也可能是个别记录的price是空字符串转成 NaN 之后min()忽略了它但fillna(0)又把它补成了 0。 解决聚合前统一过滤price 0同时对volume和amount做非负校验。过滤条件写在resample之前写之后再过滤会把整根 bar 丢掉。5.2 绘图与落库的两处暗坑现象四K 线颜色反了涨的是绿色。mplfinance默认配色是欧美习惯跟 A 股的阅读习惯相反。这个问题不影响功能但发出去的图别人第一眼会觉得数据错了。 解决用make_marketcolors(up#e4572e, down#26a69a, edgeinherit, wickinherit, volumein)显式覆盖edge和wick都用inherit让边框和影线跟随实体颜色。配色定好之后塞进theme.py两个绘图文件统一引用避免改一处漏一处。现象五pyecharts的图打开是空白或者报TypeError。八成是 1.x 和 2.x 的 API 混用了。1.x 里Grid().add()的参数位置和 2.x 不同Kline的数据初始化方式也变了。另一个高频原因是 K 线数据顺序写成了[open, high, low, close]图能渲染但影线和实体全部错位。 解决requirements.txt里锁死pyecharts2.0.4从旧教程抄代码时先看版本数据构造处加一行assert data[0][2] min(data[0][0], data[0][1])之类的校验low必须不大于开收盘价跑不通当场就能发现。提示这套排查顺序建议固定下来——先看原始响应、再看标准化后的 DataFrame、最后看图。跳过前两步直接调绘图参数往往在错误的数据上改样式越改越乱。6. 二次开发把单机脚本改成能定时跑的行情面板原项目跑一次出一次图实际用起来肯定不满足。我一般第一步是把它改成收盘后自动拉全天数据、落 parquet、出图三步走用APScheduler挂一个cron任务就够了。from apscheduler.schedulers.blocking import BlockingScheduler from datetime import datetime def daily_job(codes): for code in codes: bars fetch_and_normalize(code) # 取数 标准化 store.to_parquet(bars, code) # 按 code/日期 落到 data/parquet plot_intraday(bars, bars[prev_close].iloc[-1], code) plot_kline(bars, n60) if __name__ __main__: sched BlockingScheduler(timezoneAsia/Shanghai) # 15:10 跑给行情接口留出收盘数据落定的时间 sched.add_job(daily_job, cron, hour15, minute10, args[[000001, 600000]], misfire_grace_time600) sched.start()misfire_grace_time600这句别省。机器重启或者任务卡住时APScheduler 默认会直接放弃错过的任务加了宽限时间才会补跑。hour15, minute10而不是 15:00是因为收盘瞬间的接口数据还在刷新早跑十分钟经常拿到不完整的最后一根 bar。第二步是增量。分笔数据一天一存历史越攒越多全量重算越来越慢。我的做法是按code分目录、按trade_date分文件每次只补缺失的日期读的时候用pd.read_parquet(..., filters[(dt, , start)])只捞需要的区间。验证方式很土但有效拿同一只票用增量拼出来的日线和一次性全量算出来的做一次assert_frame_equal对得上才算改对了。再往上就是把这些函数包成 Flask 或者 FastAPI 的接口前端挂个页面按代码查图但这属于另一个话题了。我自己的习惯是任何跟行情数据打交道的脚本在接定时任务之前先用三个不同交易日的数据各跑一遍把输出图的横轴刻度、成交量量级、昨收基准线位置逐一看过。这三处不出问题后面加指标加品种基本不会翻车。希望帮到你。本文还有配套的精品资源点击获取