简介：面向聚宽平台量化开发与金融数据分析人员，资料包集中梳理SQL数据库传递与中转的实操要点，借助标准SQL语句从聚宽获取金融数据，完成查询、筛选、排序及入中转库等操作，适合已有数据库基础、希望提升量化数据处理效率的学习者。压缩包共3个文件，整体仅19KB，覆盖rzrk、xml、py三种类型：rzrk文件承载QMT信号买卖逻辑与规则，xml文件描述SQL配置或信号数据结构，py脚本则用于自动化执行SQL查询并处理结果。目前已有202人学习下载。通过对照这些文件，可快速理解QMT买卖信号如何从聚宽传递至中转数据库，掌握用Python封装SQL查询、解析XML信号数据的基本方法。对于需要搭建可复用的聚宽数据通道、实现信号驱动交易或离线分析的场景，这套小体积资料提供了清晰的设计思路与可直接套用的代码骨架。
1. 聚宽SQL数据库传递在做什么：把平台数据变成本地可控的量化底座
聚宽平台的数据很全，但真正卡住开发过程的并不是拿不到数据，而是数据拿到本地以后没地方放、格式不统一、取用还要看接口脸色。所谓“聚宽SQL数据库传递”，简单说就是通过聚宽的数据接口，把行情、交易日历这类资料成批拉取到一台你完全掌控的SQL中转数据库里，让本地回测、统计报表、实盘提醒都从中转库取数，而不是各自去请求平台接口。中转库里存的是标准化表，上游只管“拉数入库”，下游只管“查询计算”，两边互不干扰。
适合搭这套链路的是两类人：一类是策略已经迭代稳定、需要把研究环境搬到本地来跑的开发者；另一类是团队里多个程序依赖同一份行情数据、不想每个人各调一次接口被逐个限流的场景。如果你只是偶尔看几条行情，直接查平台接口就是最省事的方案，SQL中转反而是负担；一旦发现同一段历史数据被反复导出、接口响应慢到影响回测心情，就值得花一天时间把中转库建起来。
2. 直连聚宽还是SQL中转：两种取数链路的结构与取舍
2.1 为什么多数团队最终选择中转：直连的问题不在慢，在耦合
策略程序直接调聚宽接口取数，在小批量研究场景下体验很好；等数据和程序数量上去以后，问题就出来了。每次取数都是一次真实网络请求，取数时间从几百毫秒到几秒波动，盘中还可能碰到限流；更麻烦的是数据格式是内存里的DataFrame，每个脚本各自解析一次，判断缺失、调整字段名、处理停牌的逻辑散落各处。同一个“今日涨跌幅”的定义，A脚本按前收盘算，B脚本按昨收算，最后对不上账。
这并不代表平台接口质量不行，而是“取数”和“用数”的节奏天生不同。取数是低频、大批量、可容忍延迟的，使用则是高频、小窗口、对延迟敏感的。SQL中转数据库正好在这两者之间做一个缓冲：低频拉取、毫秒读取、统一格式、按库表提供契约。把这一步做完，取数逻辑就从各个策略脚本里被剥离出来，集中到一个同步程序里维护，改动一行代码就能影响所有人。
之所以选择SQL而不是继续用CSV或Parquet文件共享，是因为文件方案缺少并发控制和事务保证。多个程序同时读一个CSV问题不大，但一个程序在替换文件时另一个程序正在读取，就会读到半个文件；SQL表天然解决这类问题，还顺手把增量更新、按日期过滤、JOIN交易日历这些高频操作都变成了标准SQL。至于市面上那些数据库同步软件，它们适合异构库之间的同步，这里的数据源是平台接口并不直接暴露表结构，还是要靠脚本先拉数，再写库。
| 对比项 | 直连聚宽接口 | 走SQL中转库 |
|---|---|---|
| 取数耗时 | 每次请求几毫秒到数秒波动 | 本地查询毫秒级稳定 |
| 限流影响 | 直接中断正在跑的任务 | 只影响同步任务，读库不受影响 |
| 多程序协同 | 每个程序维护自己的取数逻辑 | 统一读写同一组表 |
| 历史数据复用 | 每次都要重新拉取 | 拉一次存一份，之后直查 |
| 数据一致性 | 各程序取数时间不同，结果可能不一致 | 同一份落库数据，读取永远一致 |
2.2 MySQL、PostgreSQL、SQLite：中转库选型的三个方向
选哪个库，取决于团队规模和并发程度。先给结论：如果是五个人以内的本地量化团队，MySQL是投入产出比最高的选择，坑少、文档多、备份简单；如果团队已经在用PostgreSQL，完全没有必要为了“统一”而换，尤其在需要存数组字段和JSON元数据的场景里；SQLite适合个人调试、轻量演示和离线环境，别把它当多程序共享库用。
对量化中转这种“写入集中、查询分散”的负载，MySQL的范围分区加上普通索引已经够用。日线全市场拉十年，也就是百万行级别；分钟线全市场两年，接近千万行，用InnoDB跑日常条件查询没有压力。担心数据量膨胀的话，先按月分区，查询时带上日期条件就能走分区裁剪，不需要一开始就上重型列存库。中转库的核心是把数据“转”过来，不是把数据“算”出来。
PostgreSQL的优势在于复杂查询和数据完整性约束更强，如果后续要做财务快照、因子研究这类多表关联密集的工作，PG的窗口函数和物化视图会顺手一些。SQLite则胜在零运维，单个文件拷走就能用，适合在没装数据库的机器上快速做数据验证。选型时还要关注团队里谁负责维护这个库，没人愿意半夜起来处理一个自己不熟的服务。
| 维度 | MySQL | PostgreSQL | SQLite |
|---|---|---|---|
| 安装运维 | 简单，主流云均有托管 | 略重，需要定期维护 | 零安装，单文件 |
| 并发读写 | 支持连接池，日常够用 | 并发更好，复杂查询强 | 写并发弱，多进程写会锁库 |
| 分区与索引 | 支持范围分区 | 分区能力更丰富 | 基本只能靠分表 |
| 典型场景 | 日线/分钟线中转 | 财务快照、因子研究 | 单机调试、Demo演示 |
2.3 建表设计：交易日历放第一，行情表主键必须业务化
先把交易日历建出来，它是后面所有增量任务的计算基础。平台返回的交易日是标准日期，本地存成DATE类型即可。建表SQL我习惯写成这样，同时把行情表一起建好：
CREATE DATABASE IF NOT EXISTS quant_transit DEFAULT CHARACTER SET utf8mb4; USE quant_transit; CREATE TABLE IF NOT EXISTS trade_calendar ( calendar_date DATE NOT NULL COMMENT '交易日', is_open TINYINT NOT NULL DEFAULT 1 COMMENT '1开市 0休市', PRIMARY KEY (calendar_date) ) ENGINE=InnoDB COMMENT='交易日历'; CREATE TABLE IF NOT EXISTS daily_price ( security_code VARCHAR(12) NOT NULL COMMENT '证券代码，如000001.XSHE', trade_date DATE NOT NULL COMMENT '交易日期', open_price DECIMAL(10, 3) NULL COMMENT '开盘价', high_price DECIMAL(10, 3) NULL COMMENT '最高价', low_price DECIMAL(10, 3) NULL COMMENT '最低价', close_price DECIMAL(10, 3) NULL COMMENT '收盘价', volume BIGINT NULL COMMENT '成交量(股)', amount DECIMAL(20, 3) NULL COMMENT '成交额(元)', PRIMARY KEY (security_code, trade_date), KEY idx_trade_date (trade_date) ) ENGINE=InnoDB COMMENT='日线行情';类型和主键是这里最值得花时间的两个地方。价格用DECIMAL(10,3)而不是FLOAT，是为了避开浮点精度问题，金融价格保留三位小数足够，回测时四舍五入的偏差累积起来很要命。成交量用BIGINT，因为大市值股票的日成交量会超过INT上限。主键用（security_code, trade_date）组合，从业务上保证同一只股票同一交易日只有一条记录，等于做了一层天然去重。
索引只需要单独建在trade_date上，因为最常见的查询是按日期范围过滤，再匹配具体股票代码。不要给open_price、close_price这类列盲目加索引，中转库的查询模式本来就固定，多一个索引就多一份写入成本。如果你打算同步分钟线，额外建一张minute_price表，把trade_date改成分钟级时间戳，主键变成（security_code, time_stamp），思路完全一致。
2.4 代码、字段名和时间基准：让中转库成为团队约定
聚宽接口返回的证券代码带市场后缀，上证是.XSHG，深证是.XSHE，例如000001.XSHE。本地其他数据源可能用“000001.SZ”这类写法，入库前必须统一。我的做法是以聚宽原始代码为准存库，不做缩写转换，因为只要源头唯一，后续做代码映射表就很简单。行情表里的字段名也尽量和平台保持一致，只做必要的重命名，比如money字段在MySQL里叫amount更直观，这个转换集中在一个清洗函数里完成。
时间基准同样要提前定死。无论本地服务器在哪个时区，入库的日期统一按北京时间处理，并且以字符串形式写入DATE列。下面的字段映射表是daily_price入库前后的标准对应关系，建议直接贴在项目文档里：
| 聚宽返回字段 | 入库字段 | 入库类型 | 说明 |
|---|---|---|---|
| open | open_price | DECIMAL(10,3) | 开盘价 |
| high | high_price | DECIMAL(10,3) | 最高价 |
| low | low_price | DECIMAL(10,3) | 最低价 |
| close | close_price | DECIMAL(10,3) | 收盘价 |
| volume | volume | BIGINT | 成交量 |
| money | amount | DECIMAL(20,3) | 成交额 |
3. 聚宽到SQL：从拉数、清洗到落库的最小实现
3.1 环境准备：SDK安装与本地MySQL基础配置
开始之前，确认本地已安装MySQL 5.7以上版本，并建好一个专用账号，不要用root跑定时同步。接着安装聚宽SDK、pandas以及MySQL驱动：
pip install jqdatasdk pandas pymysql sqlalchemyjqdatasdk是聚宽官方提供的Python SDK，pymysql是纯Python的MySQL驱动，配合sqlalchemy可以省去手写连接代码的工作量。安装完成后，先验证账号能正常取数，再进入下一步。网络环境特殊的话，提前配好pip镜像源，避免安装过程反复超时。
3.2 从聚宽取一段日线数据：参数怎么设才不踩复权坑
最小取数代码是下面这段，先跑通再扩展：
from jqdatasdk import auth, get_price import pandas as pd auth('139****0000', 'your_password') df = get_price( '000001.XSHE', start_date='2024-01-01', end_date='2024-06-30', frequency='daily', fields=['open', 'high', 'low', 'close', 'volume', 'money'], skip_paused=True, fq='pre' ) print(df.info()) print(df.head())这里有几个参数直接影响后续入库质量。frequency='daily'是日线，要取分钟线可以改成'1m'、'5m'，但分钟线的单次请求条数和体积限制比日线严格，建议在确认日线链路稳定之后再上。fields里只取了六个基础列，实际还有前收盘、后收盘、涨跌幅等字段，按需要再加。skip_paused=True会在停牌日跳过该股票的行情，如果后续要和全市场股票列表做对齐，改成False并在清洗阶段自行过滤。
fq='pre'表示前复权，这个参数必须一锤定音。入库时统一用前复权，或者统一用不复权，绝不能混合。很多回测结果对不上账，追到最后都是因为中间某段数据混用了不同的复权基准。get_price默认返回的索引是时间序列，写库前先reset_index()，再把证券代码补成一列，后续处理才方便。
3.3 写库：DataFrame入库的两种方式，用参数化INSERT代替to_sql
最省事的做法是用pandas的to_sql直接推送DataFrame，但它有两个问题：字段类型由pandas自动推断，可能把close_price推到DOUBLE，和之前建的DECIMAL列冲突；写入过程中出错时，不容易定位是哪一行有问题。所以这里用参数化INSERT加executemany的方式，类型可控、异常好排查：
import numpy as np from sqlalchemy import create_engine engine = create_engine( 'mysql+pymysql://quant:quant_pass@localhost:3306/quant_transit?charset=utf8mb4', pool_size=5, pool_recycle=3600 ) df = df.reset_index() df['security_code'] = '000001.XSHE' df['trade_date'] = df['index'].astype(str).str[:10] df.rename(columns={ 'open': 'open_price', 'high': 'high_price', 'low': 'low_price', 'close': 'close_price', 'volume': 'volume', 'money': 'amount' }, inplace=True) df.replace([np.inf, -np.inf], np.nan, inplace=True) df = df.where(pd.notnull(df), None) rows = df[['security_code', 'trade_date', 'open_price', 'high_price', 'low_price', 'close_price', 'volume', 'amount']].values.tolist() insert_sql = """ INSERT INTO daily_price (security_code, trade_date, open_price, high_price, low_price, close_price, volume, amount) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) """ with engine.begin() as conn: for i in range(0, len(rows), 5000): conn.execute(insert_sql, rows[i:i+5000])engine.begin()是事务型连接，只要中间有任一批失败，整体回滚，不会留下半个区间的数据。每次提交5000行，兼顾数据库锁资源和单次网络开销。trade_date转成字符串再截前10位，是为了去掉时间索引里的时分秒，和DATE列对齐。这里用参数化写法还有一个附带好处：避免手拼SQL带来的转义问题，也更安全。
3.4 验证链路：用SQL和Python双重确认数据落地
数据入库后，先用SQL查一下行数和日期区间：
SELECT trade_date, security_code, close_price FROM daily_price WHERE security_code = '000001.XSHE' ORDER BY trade_date LIMIT 5;再用Python把入库结果取回来，和原始DataFrame对比，确认没有丢行：
import pandas as pd back = pd.read_sql( "SELECT * FROM daily_price WHERE security_code = '000001.XSHE' ORDER BY trade_date", engine ) assert len(back) == len(df), f"行数不一致: {len(back)} vs {len(df)}"需要注意对比前提：拉数时用了skip_paused=True，平台返回的就是过滤停牌后的行数，数据库里也应该是这个行数。断言的目的是确认传输过程没丢行，不是确认该股票所有交易日都有数据。跑完这一步，聚宽SQL数据库传递的最小闭环就建立了。
4. 同步任务怎么调：拉数窗口、批次与调度参数的平衡
4.1 拉数端参数：日期窗口切分与复权基准固定
把同步任务变成长期运行的程序之前，先解决取数的分片问题。聚宽接口对单次取数的条数有限制，与其赌边界值，不如把取数区间切小，失败重跑的成本也低。我一般会把区间切成半年一段：
from datetime import date, timedelta def fetch_range(security, start_date, end_date, freq='daily'): # 按180天切分拉数区间，避免单次请求过大 step = timedelta(days=180) cursor = start_date batches = [] while cursor < end_date: end = min(cursor + step, end_date) batches.append((security, cursor, end)) cursor = end + timedelta(days=1) return batches切分窗口的意义不只是规避接口限制，更重要的是让增量同步天然支持断点续传。比如拉到一半程序挂了，重启后只需要从上次失败的那个窗口继续，而不是全量重跑。复权基准也在这一层固定下来，所有批次一律使用同一个fq参数。
4.2 写库端参数：批次大小、临时索引与写入并发
前文提到的5000行一批是经验值，MySQL 5.7/8.0下5000到10000行一提交比较合适。批次太小比如100行，网络往返开销占比高，整体速度上不去；批次太大比如十万行，事务周期变长，binlog和锁压力都会上来。我用Python的timeit对比过不同批次耗时，5000行和20000行差距很小，再大则明显变慢，所以取中间值最稳。
表数据量大以后，新增大批量数据之前可以临时去掉非主键索引，插入完成后再加回来。这个操作对日线表频率不高，但对分钟线入库效果明显：
ALTER TABLE daily_price DROP INDEX idx_trade_date; -- 执行大批量INSERT ALTER TABLE daily_price ADD INDEX idx_trade_date (trade_date);注意这个操作只能在维护窗口执行，不要在业务查询高峰期做，否则那几分钟所有读这个表的任务都会变慢。
4.3 增量更新：用交易日历计算起始日期，不给跨周留坑
聚宽的行情数据更新有时间差，盘中拉当日数据经常出现部分股票字段缺失的情况。所以同步策略要保守一点：默认同步截止到昨日的数据，当天的数据留到第二天晚上再补。增量起点用“本地交易日历推演”，而不是简单的last_date加一天：
import pandas as pd from datetime import timedelta last_date = pd.read_sql( "SELECT MAX(trade_date) AS max_date FROM daily_price WHERE security_code='000001.XSHE'", engine ).iloc[0]['max_date'] next_open_date = pd.read_sql( """ SELECT MIN(calendar_date) AS next_date FROM trade_calendar WHERE calendar_date > %s AND is_open = 1 """, engine, params=[last_date] ).iloc[0]['next_date']用交易日历而不是自然日推算，能避开国庆、春节这类长假带来的空跑。如果直接用last_date加一天，休市期间拉回来的都是空结果，既浪费请求次数，还容易让人误以为数据缺失。
4.4 调度频率：每天一次全量补昨日数据，比盘中多次刷新更可靠
同步任务的调度我建议放在收盘后，每天一次。一个比较稳妥的cron写法是：
0 18 * * 1-5 cd /opt/quant/sync && python sync_daily.py >> logs/sync.log 2>&118点执行、周一至周五，这个时间是等平台当天数据基本稳定后再拉。如果当天同步失败，第二天同一时间会再跑一次，配合日志能及时发现。日志落盘非常重要，脚本调完接口后把当次的起止日期、行数、耗时都写进日志，后面排查问题全靠它。如果策略确实需要盘中数据，刷新间隔不要小于15分钟，也不要同时并发拉取大量股票，触发限流以后整批任务都会被丢。
5. 聚宽SQL中转的5个高频坑：现象、原因、处置
5.1 当日数据拉下来却是残缺的，隔天对不上账
现象：下午五点多跑同步，当天的close_price和volume看起来正常，但第二天和平台核对，发现少了几只股票的行情。
原因：平台数据更新存在延迟，不同品种的结算数据落地时间并不一致。下午五点的“当日数据”并不一定收齐，盘中数据更是如此。
处置：把同步策略改成“永远同步截止昨日的数据”，当天的数据留到次日晚上再补。如果策略必须当天数据，就把同步时间推迟到晚上八点以后，并且把当日数据标记为临时数据，不放进主表长期依赖。
5.2 MySQL报DataError：NaN或Infinity写不进DECIMAL
现象：脚本执行到一半，报错 DataError: 1265, Data truncated for column 'volume'。
原因：聚宽接口对停牌或无成交的字段返回NaN，pandas把它按浮点数处理，MySQL的BIGINT列不接受NaN，Infinity更是直接报错。
处置：写库前统一处理，把Inf替换成NaN，再把数值列转换补零：
df['volume'] = pd.to_numeric(df['volume'], errors='coerce').fillna(0).astype('int64') df['amount'] = pd.to_numeric(df['amount'], errors='coerce').fillna(0.0)pd.to_numeric先把脏数据转成NaN，fillna再兜底补零，这样无论原数据是空字符串、None还是非数字，都不会让写库中断。
5.3 重复执行同步脚本，数据翻倍或主键冲突
现象：第二次跑同步不报错，但表里同一只股票同一日期出现了两条记录，或者插入时报Duplicate entry错误。
原因：同步逻辑没有做幂等控制，每次都是简单append，历史区间被重复写入。
处置：采用“先删后插”，并且删除和插入必须在同一个事务里：
START TRANSACTION; DELETE FROM daily_price WHERE security_code = '000001.XSHE' AND trade_date BETWEEN '2024-04-01' AND '2024-04-30'; -- 此处由脚本执行插入 COMMIT;删除和插入在同一个事务里，才能避免删完还没插入时其他查询读到空窗口。如果是在两个独立事务里做，中间任何一次查询都会看到数据缺失。
5.4 用自然日窗口拉数据，跨长假时产生大量无效请求
现象：国庆假期后第一天跑增量，start_date按last_date加一天计算，拉出来的区间横跨整个休市周，接口返回大量空结果，同步耗时比平时多出一倍。
原因：没有用交易日历推演增量起始日，直接把自然日当成有效交易日。
处置：用本地trade_calendar表计算下一个交易日，前面第4.3节已经写了完整SQL。把这个逻辑封装成函数，所有增量任务统一调用，不要再出现手工写日期的同步脚本。
5.5 时区偏移导致日期错位，最后一天不是12月31日
现象：同一批数据，有的行落在2024-06-03，有的落在2024-06-02，按日期排序后最后一行不是预期的年末。
原因：服务器操作系统时区设置成了UTC或本地时区，与API返回的北京时区在边界处出现偏移。
处置：入库前统一把时间戳格式化成“YYYY-MM-DD”字符串，不直接把datetime对象交给MySQL。同时给连接加上会话时区设置：
engine = create_engine( 'mysql+pymysql://quant:quant_pass@localhost:3306/quant_transit?charset=utf8mb4', connect_args={'init_command': "SET time_zone = '+08:00'"} )这个设置解决的是数据库会话解释日期的方式，真正干净的做法还是入库前把日期转成固定格式字符串。
6. 进入稳定期：快照、对账与一个持续受益的小习惯
6.1 增量表加月度快照，给回测一个冻结基线
当daily_price表完整度提高之后，会面临一个新问题：复权因子调整会让历史价格发生变化，同一个回测在不同时间跑，结果可能不一样。为了解决这个不确定性，我习惯每月底把全市场日线复制一份到独立的快照表，表名带上日期后缀：
monthly_snapshot = pd.read_sql( "SELECT * FROM daily_price WHERE trade_date >= '2024-06-01'", engine ) monthly_snapshot.to_sql( 'daily_price_snapshot_20240630', engine, if_exists='replace', index=False )生成快照以后，回测任务指定用某张快照表，后续复权调整不会影响已经跑出来的结果。团队协作时这个做法尤其有用，两个人回测同一时段，只要指定同一张快照表，结果就可以互相比较。
6.2 对账脚本：一行SQL找出同步遗漏
每天晚上同步完成后，跑一遍全市场核对脚本，这是最直接有效的健康检查：
SELECT COUNT(DISTINCT security_code) AS stock_count, COUNT(*) AS row_count, MIN(trade_date) AS min_date, MAX(trade_date) AS max_date FROM daily_price WHERE trade_date = '2024-06-28';拿stock_count和当天平台公开的上市证券数量做比较，差异超过阈值就告警。不需要搭建复杂的监控平台，cron定时跑这个SQL，把异常结果追加到日志文件，早晨检查一眼就行。
6.3 一个值得坚持的习惯：每天看同步日志的三个字段
把聚宽SQL数据库传递的链路稳定跑起来之后，最明显的收益是团队里不再有人重复写“从平台取数、清洗、转格式”的代码，新同事取数只面对数据库表结构。接口限流的影响被隔离在同步任务内部，查询性能从“看运气”变成“看本地索引”。
我自己最后留下的习惯是：每天早上先看同步日志里前一日的“行数、最大日期、耗时”三个字段，再决定是否需要回补。这个习惯看起来原始，但确实帮我躲过了好几次复权基准调错导致的历史数据偏差，也让数据问题在一早就被发现，而不是等策略回测结果异常了才回头查库。希望帮到你。
本文还有配套的精品资源，点击获取
