AI 自动开发 Python 量化程序（专业进阶）

Python 量化基石：全市场多因子扫描、同步/异步双引擎、自动复权聚合与 ./mydb 毫秒级落盘（写入实时可读）。

专业量化 · 同步/异步双支持
1. 本地目录分工与文件职责
清晰架构 · 严禁自建第三方库
📁 ./data ～ ./dataN
多数据源分区存储：底层并行挂载多分卷历史行情与多源数据，只读管理，由数据更新程序维护。
💾 ./mydb (私有写入)
专属持久化空间（WAL保障），存储自定义因子、选股结果与策略状态。高速写入，实时可读。
📦 ./pybao/
Python SDK 核心库，内含 安装.py 与 C++ 跨平台加速核心库（Win 为 .pyd / Linux & Mac 为 .so）。
📑 ./调用方式/python/
官方实战工程代码示范目录，包含 sdk_test.py、rd_test.py 等。
2. 前置准备与 pybao 全局安装
运行一次 · 全局可用
① 运行 ./pybao/安装.py

运行一次 python 安装.py 自动注册到 Python Path，相当于全局 pip install。后续在任何项目路径下均可直接 from stock_sdk import *！

② Python 3.8+ / 3.14t+ 支持

• Python 3.8+ 标准版：默认使用 stockdb.pyd / .so
• Python 3.14t+ 自由线程：将 3.14t+stockdb.pyd (或 .so) 重命名为 stockdb.pyd (或 .so) 即可享受无 GIL 极致并发！

3. 核心内置能力（批量写入 · 批量查询 · 批量计算 · 多向映射）
C++ 底层加速 · 零冗余依赖
💾 1. 【批量写入】rd.set() & rd.pipe() 毫秒级落盘 · 写入立即可读
私有存储落盘 (./mydb)

• 实时可读：数据写入本地 ./mydb 独立目录，WAL 机制确保异常不丢数据，写入完成后立即可读、无缓存延迟。
• 单条与批量：单条直接调用 rd.set(...)；全市场多因子或海量选股评分，统一组装 rd.pipe() 一次性提交批量落盘。

# 单条写入：rd.set('我的策略', '600633', '20270101', {"score": 95.5}) pipeline = rd.pipe() for item in results: pipeline.mset("策略因子", item["code"], item["date"], item["score"]) pipeline.do() # 批量一次性落盘，写入立即可查
📈 2. 【批量查询】rd.get_data(...) 全周期自动复权与聚合
单次支持数千只股票

• 批量并行：单次直接传入代码列表 ["600633", "000001", ...]，底层并行提取，彻底杜绝在 Python 循环中逐股低效请求。
• 自动复权：原生支持 qfq/hfq/None，内置全量历史除权除息数据（亦可通过 rd.get("复权*") 查询底层数据）。
• 周期聚合：原生支持 1d/1w/1M 与 1m/5m/15m/30m/60m；支持 as_df=True 一键直接输出 Pandas DataFrame。

# 批量前复权 5分钟K线查询并转 DataFrame k = rd.get_data(["600633", "000001"], start="20260713", end="20260715", frequency="5m", fq="qfq", as_df=True)
⚡ 3. 【批量计算】zb.get(...) 原生 Rust 比 Pandas 快 3 倍+
Rust 核心加速 · 免装 TA-Lib

• 5 种平滑算法：原生 Rust 核心加速计算，内置支持 5 种指数/39种指标平滑计算方法（method=39），实测速度比原生 Pandas 计算快 3 倍以上。
• 多指标并发：单次支持多指标 + 多股票批量并发，输出 MACD、KDJ、RSI、BOLL、ATR、CCI、BIAS、DMI 等全套指标与金叉死叉信号矩阵（cross=True）。

# 批量计算全市场标的的 MACD 与 KDJ 指标 macd = zb.get(["macd", "kdj"], hs[10:13], start="20260710", end="20260725", frequency="1d", fq="qfq")
🧩 4. 【多向映射】bk.get(...) 内存倒排索引 O(1) 毫秒检索
全量概念与申万行业

• 全量板块内置：预加载内存倒排索引（亦可通过 rd.get("板块*") 获取全量分类字典，rd.get("股票代码") 获取全量标的字典）。
• 多向毫秒检索：支持 个股 → 查所属板块（bk.get('600633')）与 板块 → 查全量成分股（bk.get('5G', 0, 'symbols')），及 模糊多向匹配（如查“电池”系列题材）。

concepts = bk.get('600633')[-10:] # 个股查所属概念板块 symbols = bk.get('5G', 0, 'symbols') # 板块查全部成分股代码列表 fuzzy = bk.get('电池', 0, 'name') # 模糊匹配: ['BC电池', 'HJT电池', ...]
4. SDK 全能实战示范（./调用方式/python/sdk_test.py）
9 大全能实战代码
from stock_sdk import * # 1. 原生 rd 区间与分钟通配查询 print("原生rd日:", rd.get("日k", "600633", "20260727>20260729")) print("原生分钟:", rd.get("分钟k", "600633", "20260727093*")) # 2. 板块概念多向映射 (bk.get) print("板块映射:", bk.get("600633")[-10:]) # 3. 全市场标的代码分层提取 d = rd.get("股票代码") hs = [v for v in d['6'] if '60' == v[:2]] + d['0'] # 快速提取 3100+ 只主板标的 # 4. 全周期批量自动复权查询 (rd.get_data) k = rd.get_data(hs[:3], start="20260713", end="20260715", frequency="1d", fq="qfq", as_df=False) # 5. 多指标批量计算 (zb.get) macd = zb.get(["macd", "kdj"], hs[10:13], start="20260710", end="20260725", frequency="1d", fq="qfq") # ========================================================================= # 【全量在线 API 支持】完整在线文档：https://a.123128.xyz/docs/index.html # ========================================================================= # 6. 在线财务 API 查询 (get_fundamentals，支持全市场财务、资金流、资产负债等) df = get_fundamentals(query(cash_flow).filter(cash_flow.code == '000001.XSHE'), statDate='2024q4') # 7. 在线实时 Tick 盘口切片 (get_last_tick) t = get_last_tick('000001', count=10) # 8. 保存私有数据到 ./mydb (Value 支持 dict/list/int/str/DataFrame) rd.set('我的数据', '600633', '20270101', {"score": 95.5, "tags": ["金叉", "放量"], "params": {"n": 20}}) # 9. 读取私有数据 ./mydb loaded = rd.get('我的数据', '600633', '20270101')
📌 rd.get_data(...) 常用参数速查：
• code: 单股 "600633" 或 批量列表 ["600633", "000001"]
• frequency: "1d"(日K), "1m/5m/15m/30m/60m"(分钟K), "1w"(周K), "1M"(月K)
• fq: "qfq"(前复权), "hfq"(后复权), None(不复权) ｜ as_df: True(直接返回 Pandas DataFrame)
5. 同步 vs 异步双模式与 Pipeline 批量落盘（rd_test.py）
吞吐 >260万条/秒
🔹 同步模式 (Sync · Pipeline 极速批量写入)
# 批量写入必须使用 Pipeline (严禁循环单条 set) pipeline = rd.pipe() for item in results: pipeline.mset("策略因子", item["code"], item["date"], item["score"]) pipeline.do() # 同步一次性落盘，吞吐超 260 万条/秒
⚡ 异步模式 (Async · 高频并发调度)
# 异步高频非阻塞 async def main(): rows = await rd.vals("日k", "6*", "202606*").get("close") pipeline = rd.pipe() pipeline.mset("实时缓存", "600519", "20260625", data) await pipeline
6. 全量 API 与全维度数据体系（本地 + 在线无缝直调）
在线 API 完整文档 (50+ 接口)

Python SDK 完整覆盖并支持 全量在线 API（部分接口需联网自动调用），策略无需自建爬虫与数据管道，一键调取全市场 7 大全维度金融数据：

📈 行情数据
日K/分钟K/分时/Tick
全周期聚合/自动前后复权
📊 财务指标
资产负债表/利润表/现金流
估值指标/成长盈利能力
🌾 期货衍生
期货全合约/主力连续合约
升贴水/持仓排名与基差
🏛️ 指数体系
全市场指数成分股与权重
指数风格暴露 (Barra体系)
🧠 因子看板
Alpha 101 / Alpha 191
动量/波动/估值多因子值
⚡ Ticks 盘口
实时盘口切片/五档买卖
逐笔成交历史回放
🔴 实时特色数据
实时行情切片 · 资金流向 (大单/超大单) · 龙虎榜明细 · 集合竞价 · 融资融券余额
7. 让 AI 编写 Python 策略流程 (3 步极速落地)：
1

后台打开 stockdb 数据引擎

双击运行 stockdb（小狗图标），本地数据引擎就绪（127.0.0.1:7899）。

第一步
2

将 Python 接口文档与需求发给 AI

发给 Cursor / WorkBuddy / Claude 等任意 AI 工具生成 .py 策略脚本

第二步
请阅读 stockdb Python开发文档：http://a.123128.xyz/docs/AI策略python开发接口文档.md ，编写 Python 量化程序： 1、核心接口统一通过 from stock_sdk import * 导入； 2、全市场多因子选股并计算收益率回测； 3、将计算结果与动态因子使用 pipeline 实时写入本地私有缓存 ./mydb 空间； 请帮我编写一个完整的高性能 Python 策略脚本。
3

运行脚本，秒级获取回测结果并落盘

保存为 strategy.py 并运行，毫秒级完成全市场运算并存入 ./mydb。

第三步 · 完成

立即开始：一键复制发给 AI

查看完整官方 Python 开发文档.md →