简介面向量化交易与金融数据分析人员这份压缩包围绕聚宽平台SQL接口的数据传递与中转操作整理了一套可直接参考的轻量资料用于解决从聚宽取数到中转数据库落地的需求。包内共3个文件包含Python脚本、XML配置及rzrk信号文件分别用于自动化执行SQL查询、描述信号数据结构及SQL相关配置、承载聚宽QMT模型的买卖规则压缩包整体仅19KB结构紧凑、便于快速查看。目前已有202人学习适合希望了解聚宽SQL取数流程、QMT信号对接方式或正在搭建本地中转数据库的初学者与开发者这类场景通常涉及接口调用与数据落地的衔接值得对照参考。借助这些内容读者可快速掌握利用SQL脚本筛选金融数据的基本思路理解rzrk文件中买卖逻辑的字段组织并参考Python脚本将聚宽查询结果与后续分析流程衔接起来减少自行摸索和数据接口调试的时间成本。1. 聚宽SQL数据库传递为什么先落中转库而不是直连聚宽SQL数据库传递如果只盯着“SQL”两个字很容易把功夫全花在写 INSERT 语句上却忽略了真正决定成败的中间层。以前我贪省事直接把聚宽返回的 DataFrame 塞进 MySQL 业务库结果某次批量回测把接口配额打满当天数据还没入库其他任务全跟着翻车。后来改成“中转数据库”模式单独建一个 MySQL 库聚宽数据带批次号落进来再由回测、因子分析、报表系统各自去读。这套结构解决的不仅是聚宽延迟和接口限流还把字段映射、重复数据、权限边界这些破事统一收敛到一个出口。适合正在搭量化数据管道的开发者也适合需要把聚宽数据分发给多个策略库的团队。2. 中转库选型与表结构四个必须提前定的字段中转库承担的是“接收、暂存、分发、审计”四个角色所以它不等同于数据仓库不需要复杂分层和分区但必须有清晰的批次标记。建表阶段最要紧的是想明白四个字段trade_date、ts_code、batch_id、update_time。这四个字段缺一个后面增量同步和问题排查都会多花一倍时间。2.1 为什么选 MySQL并发、索引和可调试性中转库我首选 MySQL 5.7 以上版本不依赖商业数据库同步软件直接用 SQLAlchemy 加 pandas 就能跑通。选它而不是 SQLite是因为中转库往往同时有多个写入进程SQLite 在并发写锁上的表现非常差选它而不是 ClickHouse 或数仓是因为大部分量化团队的后续消费端还是关系型 SQLMySQL 在字段类型兼容和索引调优上最容易被接手的人理解。另外 MySQL 的information_schema可以直接查表结构、索引状态、慢查询日志。排查数据写没写进去时一条SHOW INDEX就能看清唯一键是否生效。对中转场景来说可调试性比极致性能更重要。2.2 四张核心表batch、行情、因子和配置建表前先明确中转库不需要一堆视图和存储过程先把sync_batch、stock_daily、factor_daily、sync_config四张基础表建好。前两张是主力第三张管因子第四张用来存股票池和同步参数。CREATE DATABASE IF NOT EXISTS quant_mid DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci; USE quant_mid; CREATE TABLE sync_batch ( batch_id INT AUTO_INCREMENT PRIMARY KEY, source_system VARCHAR(32) NOT NULL DEFAULT JQ, start_date DATE NOT NULL, end_date DATE NOT NULL, row_count INT DEFAULT 0, status TINYINT DEFAULT 0 COMMENT 0 running, 1 done, 2 failed, error_msg VARCHAR(512), cost_ms INT, create_time DATETIME DEFAULT CURRENT_TIMESTAMP, KEY idx_status (status) ) ENGINEInnoDB; CREATE TABLE stock_daily ( id BIGINT AUTO_INCREMENT PRIMARY KEY, ts_code VARCHAR(16) NOT NULL, trade_date DATE NOT NULL, open DECIMAL(10,4), high DECIMAL(10,4), low DECIMAL(10,4), close DECIMAL(10,4), pre_close DECIMAL(10,4), volume DECIMAL(20,0), amount DECIMAL(20,4), paused TINYINT DEFAULT 0, high_limit DECIMAL(10,4), low_limit DECIMAL(10,4), batch_id INT, update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uk_ts_date (ts_code, trade_date), KEY idx_trade_date (trade_date), KEY idx_batch_id (batch_id) ) ENGINEInnoDB; CREATE TABLE factor_daily ( id BIGINT AUTO_INCREMENT PRIMARY KEY, ts_code VARCHAR(16) NOT NULL, trade_date DATE NOT NULL, factor_name VARCHAR(64) NOT NULL, factor_value DECIMAL(20,6), batch_id INT, update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uk_factor_name_date (factor_name, ts_code, trade_date), KEY idx_factor_date (factor_name, trade_date) ) ENGINEInnoDB;这里几个字段类型是血泪经验换来的trade_date必须用DATE不能用DATETIME否则按日期分组查询的时间开销翻倍ts_code用VARCHAR(16)因为聚宽的代码带交易所后缀比如000001.XSHE整型存不下volume用DECIMAL(20,0)而不用BIGINT是为了避免 pandas 的 float 精度把成交量变成科学计数法amount用DECIMAL(20,4)否则某些大额成交会被截断。2.3 索引与唯一键慢 SQL 优化要从建表开始很多人以为慢 SQL 优化是后面的事实际上中转库的慢查询基本都是从建表时漏了索引开始的。stock_daily的UNIQUE KEY uk_ts_date(ts_code, trade_date)非常关键没有它同一标的同一交易日的记录可以无限插入增量重跑一次就多一层脏数据。idx_trade_date则是为了支撑“某一天全市场数据”这种高频查询。如果只建唯一键不建日期索引SELECT * FROM stock_daily WHERE trade_date 2024-01-02会走全表扫描数据量到百万级以后一次查询从毫秒级变成秒级多条并发直接把数据库连接池占满。建表后建议立刻执行SHOW INDEX FROM stock_daily; SHOW INDEX FROM factor_daily;确认uk_ts_date、idx_trade_date、uk_factor_name_date都在。后面所有基于INSERT ... ON DUPLICATE KEY UPDATE的重跑逻辑都依赖这些唯一键。2.4 初始化脚本用 mysql 执行 SQL 脚本而不是让 pandas 自动建表准备一个01_create_tables.sql文件把上面的建表语句放进去然后用命令行执行mysql -h127.0.0.1 -P3306 -uquant -p${MYSQL_PWD} --default-character-setutf8mb4 quant_mid 01_create_tables.sql参数说明-h和-P指定数据库地址端口-uquant是中转库专用账号不要给 root 权限MYSQL_PWD环境变量比命令行明文密码安全--default-character-setutf8mb4是为了让表注释和因子名不乱码。 01_create_tables.sql是 shell 重定向把 SQL 文件喂给 mysql 客户端。为什么不建议直接用pandas.to_sql自动建表因为它会把DATE推断成datetime把DECIMAL推断成DOUBLE还会把唯一键全部丢掉。自动建表省五分钟后面数据校验和去重能让你加班五个小时。初始化阶段慢一点给后面留后悔药。3. 从聚宽到 MySQL增量同步脚本的完整写法结构定好之后剩下就是写同步脚本。我在这个环节的原则是所有参数走环境变量或配置不要写死同步窗口由交易日历驱动不要用自然日硬算每条数据都带上批号方便回溯。下面按顺序拆开讲。3.1 连接与认证环境变量与连接池# sync_to_mysql.py import os import time import pandas as pd from jqdatasdk import auth, get_price, get_trade_days from sqlalchemy import create_engine JQ_USER os.getenv(JQ_USER, ) JQ_PASS os.getenv(JQ_PASS, ) MYSQL_DSN os.getenv( MYSQL_DSN, mysqlpymysql://quant:quant127.0.0.1:3306/quant_mid?charsetutf8mb4 ) auth(JQ_USER, JQ_PASS) engine create_engine(MYSQL_DSN, pool_size5, pool_recycle3600, echoFalse)auth是聚宽 SDK 的登录入口用户名密码一定走环境变量否则脚本一旦提交到共享仓库账号就等于公开了。create_engine里pool_size5限制连接池最多 5 个连接避免多线程重试时把 MySQL 连接数打满pool_recycle3600是防止 MySQL 的wait_timeout把闲置连接断开后SQLAlchemy 还在用失效连接。连接串末尾的?charsetutf8mb4必须加否则中文字段写入时会报编码错误。3.2 交易日历驱动增量窗口先算出来def get_sync_window(): last_df pd.read_sql( SELECT COALESCE(MAX(end_date), 2010-01-01) AS last_date FROM sync_batch WHERE source_systemJQ AND status1, engine ) last_date pd.Timestamp(last_df.iloc[0][last_date]) end_date pd.Timestamp.today().normalize() - pd.Timedelta(days1) trade_days get_trade_days(start_datelast_date, end_dateend_date) todo [d for d in trade_days if d last_date] return last_date, end_date, todo这段逻辑的核心是从sync_batch表读上一次成功批次的end_date而不是从stock_daily里读MAX(trade_date)。原因很简单如果上一批次失败了一半stock_daily里可能有部分新数据但批次没成功按行情表的最大日期来算漏掉的那部分永远不会补回来。按批次表来判断只要失败就整段重跑逻辑更干净。end_date减一天是因为聚宽当天数据往往是盘后才完整盘中拉会拿到半成品或空结果。如果只做历史回测这个保险非常省钱。3.3 行情数据入库单标的循环 批量写入我一般不用聚宽的多标的返回结构因为列索引在单标的和多标的两种模式下不一样写通用解析代码容易踩坑。更稳的做法是单标的循环加上chunksize批量写def sync_stock_daily(ts_code, trade_date, batch_id): df get_price( ts_code, start_datetrade_date, end_datetrade_date, frequencydaily, fields[open, high, low, close, pre_close, volume, money, paused, high_limit, low_limit] ) if df is None or df.empty: return 0 date_col df.index.name or index df df.reset_index().rename(columns{date_col: trade_date}) df[ts_code] ts_code df[trade_date] pd.to_datetime(df[trade_date]).dt.date num_cols [open, high, low, close, pre_close, volume, money, paused, high_limit, low_limit] for col in num_cols: if col not in df.columns: df[col] None df[col] pd.to_numeric(df[col], errorscoerce) df df[[ts_code, trade_date, open, high, low, close, pre_close, volume, money, paused, high_limit, low_limit]] df df.where(pd.notnull(df), None) df[batch_id] batch_id df.to_sql(stock_daily, engine, if_existsappend, indexFalse, chunksize2000) return len(df)这里df.index.name or index是兼容聚宽在不同版本里返回的索引名有时候叫time有时候没有名字。pd.to_numeric加errorscoerce是把个别脏值变成NaN随后df.where(pd.notnull(df), None)把NaN统一转成None这样写入 MySQL 后就是标准的NULL而不是一个字符串nan。chunksize2000是我常用的经验值。太大容易撑高内存峰值太小则插入轮次过多。如果单次同步好几千只股票2000 行一批是网络往返和内存消耗之间的平衡点。3.4 因子数据入库宽表转长表行情表处理完因子数据建议用统一的factor_daily表核心操作是把宽表转成长表def sync_factor_daily(factor_df, batch_id): factor_df factor_df.reset_index() factor_df.rename(columns{factor_df.columns[0]: trade_date}, inplaceTrue) long_df factor_df.melt( id_varstrade_date, var_namefactor_name, value_namefactor_value ) long_df[trade_date] pd.to_datetime(long_df[trade_date]).dt.date long_df[factor_value] pd.to_numeric(long_df[factor_value], errorscoerce) long_df long_df.where(pd.notnull(long_df), None) long_df[batch_id] batch_id long_df.to_sql(factor_daily, engine, if_existsappend, indexFalse, chunksize2000) return len(long_df)宽表转长表的好处是以后每新增一个因子不用修改表结构直接按factor_name过滤就行。查询时的写法统一成WHERE factor_name xxx AND trade_date yyyy-mm-dd配合uk_factor_name_date唯一键天然避免重复。因子值缺失时写NULL而不是 0因为 0 会污染因子排序和统计结果。3.5 批次登记每次同步写一行 sync_batch批次是整个中转库的审计线索建议每次任务开始和结束都登记一下def start_batch(start_date, end_date): sql ( INSERT INTO sync_batch(source_system, start_date, end_date, status) VALUES(JQ, %s, %s, 0) ) with engine.begin() as conn: cur conn.execute(sql, (start_date, end_date)) return cur.lastrowid def finish_batch(batch_id, row_count, elapsed_ms, status1, error_msgNone): sql ( UPDATE sync_batch SET row_count%s, status%s, error_msg%s, cost_ms%s WHERE batch_id%s ) with engine.begin() as conn: conn.execute(sql, (row_count, status, error_msg, elapsed_ms, batch_id))engine.begin()是 SQLAlchemy 的事务上下文只要 block 内抛异常连接自动回滚。这里体现的就是常用的数据库增删改查插入批次、查增量窗口、更新状态、清理失败数据。有了sync_batch后续定位“某天数据为什么没进库”就能直接看批次状态而不是在行情表里大海捞针。4. 中转库避坑实录延迟、精度和重复数据这一章全是实战中真实出现的坑。每一条都按“现象 → 原因 → 解决”写照着排查能省不少时间。4.1 聚宽延迟导致空数据被当成成功批次现象sync_batch里显示status1、row_count0但当天确实有行情。结果按交易日一查库里有空洞。原因聚宽在盘后数据未完全就绪时get_price偶尔会返回空的 DataFrame。原来的代码把空结果当作“今天没数据”直接返回 0批次还被标记成功于是漏数就成了永久缺口。解决把空结果当作异常而不是正常结果增加带退避的重试机制。def get_price_with_retry(ts_code, trade_date, times3, wait30): for attempt in range(times): df get_price( ts_code, start_datetrade_date, end_datetrade_date, frequencydaily, fields[open, high, low, close, volume, money] ) if df is not None and not df.empty: return df time.sleep(wait * (attempt 1)) raise RuntimeError(f聚宽返回空结果: {ts_code} {trade_date})wait * (attempt 1)是指数退避的简化版第一次等 30 秒第二次 60 秒第三次 90 秒。如果重试三次仍为空就抛出异常并让整个批次变成status2不要静默跳过。4.2 字段类型不匹配DECIMAL 溢出和精度漂移现象入库后amount出现NULL或者被截断成99999999.9999和聚宽原始money对不上。抽查某只高价股时close比原始值少 0.0001累计久了偏差明显。原因建表时图省事给金额字段用了DECIMAL(10,4)。10位精度只够存百万级别的数A 股单日成交额经常上亿直接溢出截断。价格字段虽然 4 位小数够用但因子值经常是 0.000001 这种精度必须单独放宽。解决金额、成交量、因子值这三类字段的精度在建表阶段就放宽。ALTER TABLE stock_daily MODIFY amount DECIMAL(20,4); ALTER TABLE factor_daily MODIFY factor_value DECIMAL(20,6);注意大表执行ALTER TABLE会锁写最好在低峰期操作。如果表刚建不久干脆删掉重建别跟锁表较劲。4.3 增量重跑产生重复数据唯一键也没拦住现象stock_daily里同一个ts_code加trade_date出现了两行但表上明明有uk_ts_date唯一键。原因唯一键只会挡住INSERT INT... ON DUPLICATE KEY UPDATE和REPLACE INTO而pandas.to_sql(if_existsappend)不管这一套它只会无脑追加。只要增量脚本重复执行重复行就进去了。解决重跑时先写入同结构的临时表再合并到正式表。CREATE TEMPORARY TABLE tmp_stock_daily LIKE stock_daily; -- Python 里将数据先 to_sql 到 tmp_stock_daily INSERT INTO stock_daily (ts_code, trade_date, open, high, low, close, pre_close, volume, amount, paused, batch_id) SELECT ts_code, trade_date, open, high, low, close, pre_close, volume, amount, paused, batch_id FROM tmp_stock_daily ON DUPLICATE KEY UPDATE open VALUES(open), close VALUES(close), batch_id VALUES(batch_id);这样即使同一批次跑两遍最后也只保留一行。注意VALUES()在 MySQL 8.0.20 之后标记为废弃如果用的是 8.0 以上可以改成AS new的写法但 5.7 用这段没问题。4.4 执行 SQL 脚本和 Navicat 导入时中文乱码现象用 Navicat 导入建表 SQL 后表注释和factor_name里的中文变成????。原因SQL 文件本身是 utf8mb4但客户端连接字符集不是 utf8mb4。Navicat 默认连接参数有时沿用旧库的latin1导致写入的字符串在落库前已经被转坏。解决建库时指定字符集命令行执行时强制--default-character-setutf8mb4Navicat 导入前把连接编码改为 utf8mb4。已经乱掉的表可以用ALTER TABLE stock_daily CONVERT TO CHARACTER SET utf8mb4;如果表里已有乱码数据这条命令只能保证后续不再乱修不了已经被写坏的值。所以初始化脚本这一步不能省宁可开始多检查一眼不然后面全是玄学问题。5. 进阶用配置驱动同步链路十分钟验证一份新数据脚本稳定后最烦的是每次改股票池或者时间区间都要动 Python 代码。我一般会把运行参数抽到 YAML 文件里让同步脚本变成通用执行器。# sync_config.yaml jq_user_env: JQ_USER jq_pass_env: JQ_PASS start_date: 2024-01-01 end_date: 2024-12-31 stock_pool: - 000001.XSHE - 600000.XSHG - 300750.XSHE retry_times: 3 retry_wait_seconds: 30Python 入口只需要读这个配置文件再调用前面写好的函数。以后新增一个数据源不是改同步逻辑而是加一段加载器和一张目标表。另一个 Python 脚本通过命令行参数把区间传进来也比改代码要安全python sync_runner.py --config sync_config.yaml --start 2024-01-01 --end 2024-06-30配置驱动只是第一步真正让链路可信的是十分钟内的验证流程。每次同步完成后我会强制跑一遍下面这套检查。检查一按交易日看行数是否完整。SELECT trade_date, COUNT(*) AS cnt FROM stock_daily WHERE trade_date 2024-01-01 GROUP BY trade_date ORDER BY trade_date DESC LIMIT 10;正常情况下每个交易日的行数应该等于股票池数量。如果某天少了几百行优先查当天sync_batch的error_msg。检查二抽查一只股票的最新收盘价。SELECT close, volume, batch_id FROM stock_daily WHERE ts_code 000001.XSHE AND trade_date 2024-06-03;拿这个值和聚宽页面上的收盘价对一下。价格差超过 0.01 就要怀疑精度设置或者复权因子映射错了。检查三看批次状态和耗时。SELECT batch_id, start_date, end_date, row_count, status, cost_ms FROM sync_batch ORDER BY batch_id DESC LIMIT 5;如果status0说明还在跑status2说明这批失败需要先处理错误再继续。耗时异常偏高时去SHOW PROFILE或者慢查询日志里找具体是哪条 SQL 卡住不必把整个链路都怀疑一遍。最后补一个习惯每次同步前用mysqldump --single-transaction把中转库备份一次。不依赖商业同步软件一条命令就能把后悔药存好。mysqldump -h127.0.0.1 -uquant -p${MYSQL_PWD} \ --single-transaction --default-character-setutf8mb4 \ quant_mid backup_$(date %Y%m%d).sql从那以后我每次新接一个数据源都强制走一遍“建表脚本 → 空跑窗口 → 单票抽查 → 批次登记”四步确认无误后再交给策略组。十分钟能验证一条链路比任何文档都管用希望帮到你。本文还有配套的精品资源点击获取