同花顺数据采集实战：Python构建自动化行情入库系统

做股票数据分析的人都会有那么一段被数据源折磨的日子。想验证选股逻辑想看历史回测结果发现数据要么贵得离谱要么散在各大平台里拖着不让下载。我后来把主数据源切到同花顺的API体系上自己搭了一套自动化采集系统每天收盘后自动拉行情、写数据库、跑数据校验一个月下来基本不需要人工干预。这篇文章就把这套系统的完整思路和代码实现拆开讲从接口选型、参数细节到入库调度和避坑适合正在做量化、复盘工具或者个人数据仓库的开发者参考。1. 先想清楚这套采集系统到底要解决什么问题动手写代码之前最关键的是把需求掰开揉碎。数据采集听起来简单实际做起来会牵扯出很多隐藏问题比如接口一天能拉多少条、断线了怎么补、数据库涨太快怎么办。这套系统最初的目标很明确每天收盘后拿到沪深A股的日线行情和基础财务指标存到本地数据库里供后续策略回测和复盘分析使用。1.1 为什么数据采集不能靠手工下载最开始的阶段我也试过从软件里手动导出数据。偶尔一次两次没问题但时间一长就会崩溃。手工下载有几个天生解决不了的问题一是覆盖范围有限几十个股票还能导几百个上千个就完全没法玩二是历史数据不连续间隔一段时间没导出中间就断档了三是数据格式不统一从不同模块导出来的表结构、字段命名都不一样做分析之前还要花大力气清洗。自动化采集的价值不只是在“省时间”这个层面。接口拉到的数据是结构化的可以直接按照统一规则入库存档这样就有了一个标准、可复用的底层数据集。有了这个数据集写选股条件也好做收益归因也好都是一句SQL或者一段pandas代码的事不需要每天手动折腾数据。1.2 技术选型数据源、语言、存储怎么定数据源的选择我对比过几类各有各的优劣数据源类型优点注意点同花顺 iFinD / 开放接口行情覆盖全财务数据丰富终端体验好多数高级接口面向机构客户或有申请门槛同花顺网页端/客户端背后HTTP接口个人账号即可调数据实时性不错非官方正式接口需关注频率限制和服务条款开源数据包如AkShare、Tushare上手快社区活跃免费方案多数据稳定性、更新及时性要看上游限频较严格自建抓取完全可控开发和维护成本高需处理反爬和稳定性我最终采用的方式是“官方接口优先、社区接口补充”。代码层做一层抽象不管上游是官方API还是其他HTTP数据源统一封装成同一种调用入口。这样将来换数据源或者升级权限只需要替换底层实现上层的数据处理逻辑完全不用动。语言方面没有犹豫直接选了Python。数据采集加数据分析这个场景Python的生态优势太明显了requests处理HTTP请求pandas做数据清洗SQLAlchemy或者pymysql操作数据库APScheduler做定时任务每一个环节都有非常成熟的库。存储层我建议从SQLite或者MySQL起步如果只有自己回测用SQLite单文件最省事如果有多人协作或者数据量很大直接上MySQL/PostgreSQL更稳。1.3 系统整体架构与模块划分这套系统的整体结构并不复杂核心是四个模块数据源对接模块、清洗模块、存储模块、调度模块。数据源对接模块负责和接口打交道处理登录鉴权、请求加密、参数拼装、频率控制这些脏活。清洗模块负责把原始返回的数据转成统一格式处理缺失值、复权因子、单位换算。存储模块负责数据库的表结构设计和写入策略。调度模块负责按时间规则触发任务并记录运行日志和告警信息。模块划分清楚之后有个好处就是每个部分都能独立测试。接口挂了我只需要看数据源模块入库报错只需要看存储模块不会出现一个故障牵连一整片的情况。第一版别追求完美把链路跑通比什么都重要后面再逐步优化性能和数据质量。2. 核心接口细节看懂返回值和参数比写代码更重要做数据采集的人普遍有个毛病上来就急着写代码结果被接口返回搞得一头雾水。我的经验是先把接口文档吃透把可能遇到的参数坑提前列出来。同花顺数据接口的形态比较多有的是REST接口有的是客户端内置功能但核心的行情、财务、板块数据都有相对固定的字段结构。2.1 同花顺数据接口的常见形态从实际应用角度看同花顺相关的数据接口大概可以分成几类第一类是行情快照接口返回股票当前最新的价格、涨跌幅、成交量、成交额等实时数据适合盘中做监控和盯盘提醒。这类接口一般对频率非常敏感高频调用很容易触发限制。第二类是历史K线接口返回指定股票在某个时间段的日K、周K、月K数据。每个周期包含开高低收、成交量、成交额这些字段是回测最依赖的数据。要注意的是K线接口通常有单次返回数量上限超过上限要么分批拉要么用带游标的分页参数。第三类是财务数据接口包括资产负债表、利润表、现金流量表以及各类财务指标。这个接口的更新频率比行情慢得多没必要每天全量去拉半个月或者一个月同步一次就够了。第四类是板块成分和概念接口比如某个行业板块包含哪些股票、某只股票属于哪些概念板块。这类数据适合做板块轮动和热点追踪。从个人开发的角度我没有完全依赖某一种接口因为不同接口的维护状态和稳定性差别挺大。正式的做法是优先对接同花顺官方认可的数据服务申请token拿到统一的接口文档和SDK然后在这个基础上做一层缓存。申请不到的情况下再考虑从终端反推或者用社区封装但要对稳定性和使用条款有心理准备。2.2 几个必须提前确认的参数细节接口返回数据的准确性很大程度上取决于参数有没有调对。这几个参数是我实际踩坑之后总结出来的建议在写代码前先确认清楚。代码格式同花顺体系里A股代码经常区分带不带交易所前缀比如“SH600000”和“600000”在有些接口里是两种结果前者明确指定了上交所后者需要接口自己去判断。保险的做法是所有代码统一处理成带前缀的格式避免歧义。复权方式做历史回测必须用复权价格不复权的价格会带着分红送股的“假跳空”导致收益率计算完全失真。接口里一般有fq_type或类似参数常见取值有none不复权、qfq前复权、hfq后复权。我的建议是数据入库的时候按不复权和后复权各存一份回测时再按需切换。前复权价格会随着最新价变化而整体平移不适合作为历史数据库里的长期数据。交易周期要区分是1分钟、5分钟、日K还是周K。分钟级数据的数据量比日线大几个量级如果你的目标是选股策略和日线级别复盘完全用不着分钟级数据不仅拉取慢还占用存储。返回字段类型成交量到底是股还是手成交额单位是元还是万元市盈率里的“TTM”和静态市盈率含义完全不一样。这些问题在接口文档里通常有说明但很多封装好的库不会帮你换算代码里必须显式处理一遍把单位统一好再入库。2.3 数据校验与清洗思路接口返回的数据不能直接信任清洗这步必须做。我遇到的情况主要有三种第一种是停牌导致的缺失。停牌日期的K线记录可能直接不存在也可能返回空数据。处理方式是建一张交易日历表然后用交易日历去对齐缺失的日期补零或者标记为停牌而不是直接跳过否则后续算收益率时会把停牌前后的交易日错误地连在一起。第二种是复权价格与成交价格不一致。这是正常的前复权、后复权价格本来就不是真实成交价只是用来保持收益率连续。如果你写的是盘中监控策略那必须用实时价格或不复权价格如果你在做历史回测那就用复权价格。用错场景会导致策略失真。第三种是字段精度和边界问题。比如有的接口返回high、low但某一天最高价被填成0或者收盘价超出涨跌停限制这类异常值不清理掉入库之后排查起来非常头疼。写一个简单的校验函数把high max(open, close)或者low min(open, close)这类明显矛盾的数据标记出来自动重试或者告警。3. 实操从零搭建一个能跑起来的数据采集服务理论部分聊清楚了直接进入实操。下面这套代码和步骤是完整可跑的方案照着做就能把日线数据采集链路搭起来。我只用通用接口模式做示例真实环境里的接口地址和鉴权字段以你实际申请到的文档为准。3.1 环境准备项目目录与依赖建议用虚拟环境隔离依赖避免和系统Python环境互相污染。创建项目目录和虚拟环境mkdir stock_collector cd stock_collector python3 -m venv venv source venv/bin/activate # Windows下执行 venv\Scripts\activate安装依赖这几个库足够支撑整套流程pip install requests pandas pymysql loguru apscheduler然后创建基础目录结构stock_collector/ ├── config.py # 配置文件密钥、连接信息等 ├── collector.py # 数据源客户端封装 ├── cleaner.py # 数据清洗逻辑 ├── storage.py # 数据入库逻辑 ├── scheduler.py # 定时任务入口 └── logs/ # 日志目录config.py里不要硬编码敏感信息建议通过环境变量读取import os # 从环境变量读取避免把密钥提交到代码仓库 API_TOKEN os.getenv(THS_API_TOKEN, ) API_BASE os.getenv(THS_API_BASE, https://api.example.com) DB_HOST os.getenv(DB_HOST, 127.0.0.1) DB_PORT int(os.getenv(DB_PORT, 3306)) DB_USER os.getenv(DB_USER, root) DB_PASSWORD os.getenv(DB_PASSWORD, ) DB_NAME os.getenv(DB_NAME, stock_data)数据接口的密钥这种东西一旦提交到公开仓库就等于裸奔了这方面的教训不用我多说。3.2 编写请求客户端鉴权、重试、限速这一层是整个系统的核心把所有请求的公共逻辑收敛到一起。设计上要处理三件事鉴权头拼接、超时与重试、请求频率控制。import time import random import requests from loguru import logger class THSDataClient: 同花顺数据源客户端封装 def __init__(self, base_url: str, token: str, max_retries: int 3): self.base_url base_url.rstrip(/) self.token token self.max_retries max_retries self.session requests.Session() self.session.headers.update({ Authorization: fBearer {self.token}, Content-Type: application/json, User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) }) def _request(self, endpoint: str, params: dict): url f{self.base_url}/{endpoint} for attempt in range(self.max_retries): try: resp self.session.get(url, paramsparams, timeout15) if resp.status_code 200: return resp.json() if resp.status_code 429: # 触发限频等待后重试 wait_time 2 ** attempt random.uniform(0, 1) logger.warning(f触发限频等待 {wait_time:.2f}s 后重试) time.sleep(wait_time) continue # 其他状态码抛出方便上层排查 resp.raise_for_status() except requests.exceptions.Timeout: logger.error(f请求超时: {endpoint}, params{params}) time.sleep(2 ** attempt) except requests.exceptions.RequestException as e: logger.error(f请求异常: {e}) time.sleep(2 ** attempt) raise RuntimeError(f接口请求失败多次重试未恢复: {endpoint}) def get_daily_kline(self, stock_code: str, start_date: str, end_date: str, fq_type: str none): 获取日K线数据 params { code: stock_code, start_date: start_date, end_date: end_date, fq_type: fq_type, period: day, } data self._request(kline, params) return data.get(data, [])重试策略用了指数退避第一次失败等2秒第二次等4秒第三次等8秒同时加了一点随机抖动避免多个任务同时重试导致二次踩踏。注意一点接口限频不只是为了保护服务器也是保护你账号的可用性被临时封禁的代价远远大于等一下。3.3 数据入库表结构设计与更新策略拿到数据之后先清洗再入库。清洗逻辑放在独立模块里import pandas as pd def clean_daily_kline(raw_rows: list[dict]) - pd.DataFrame: 清洗日K数据统一字段格式和单位 df pd.DataFrame(raw_rows) if df.empty: return df # 统一字段名为小写 df.columns [c.lower() for c in df.columns] # 统一日期格式 df[date] pd.to_datetime(df[date]).dt.strftime(%Y-%m-%d) # 成交量转成手成交额转成元 if volume in df.columns: df[volume] df[volume].astype(float) if amount in df.columns: df[amount] df[amount].astype(float) numeric_cols [open, high, low, close, volume, amount] for col in numeric_cols: if col in df.columns: df[col] pd.to_numeric(df[col], errorscoerce) # 基础异常校验最高价不能低于开盘收盘价 df df[(df[high] df[[open, close]].max(axis1))] df df[(df[low] df[[open, close]].min(axis1))] return df.drop_duplicates(subset[date], keeplast)数据库建表时主键直接设成代码加日期天然防重复CREATE TABLE IF NOT EXISTS stock_daily ( stock_code VARCHAR(20) NOT NULL, trade_date DATE NOT NULL, open DECIMAL(10,4), high DECIMAL(10,4), low DECIMAL(10,4), close DECIMAL(10,4), volume BIGINT, amount DECIMAL(20,4), fq_type VARCHAR(10) DEFAULT none, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (stock_code, trade_date, fq_type) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;入库时用MySQL的ON DUPLICATE KEY UPDATE已经存在的数据自动更新不会重复插入def upsert_stock_daily(records: list[dict], table_name: str stock_daily): 批量写入日K数据存在则更新 if not records: return 0 df pd.DataFrame(records) # 使用 INSERT ... ON DUPLICATE KEY UPDATE 实现幂等写入 from sqlalchemy.dialects.mysql import insert # 此处省略具体数据库连接初始化假设已有一个 engine # 批量写入的替代方案用 pymysql executemanySQLite下可以用INSERT OR REPLACE INTO达到类似效果。写入逻辑的唯一要求是“幂等”也就是同一批数据重复执行几次最终库里结果一致这样重跑任务才不会有脏数据。3.4 定时调度让采集任务自动跑起来数据源准备好了清洗入库也写好了接下来就是把整个流程挂在定时任务上。我用的APScheduler理由很简单不需要额外部署服务支持cron表达式Python代码直接控制测试也方便。from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger def daily_job(): 每日采集任务 logger.info(开始执行每日行情采集) # 1. 读取股票列表 # 2. 遍历调用 THSDataClient 拉取日K # 3. 清洗后写入数据库 # 4. 记录执行状态 logger.info(每日行情采集完成) if __name__ __main__: scheduler BlockingScheduler(timezoneAsia/Shanghai) # 工作日 15:30 执行等待行情稳定入库 scheduler.add_job( daily_job, triggerCronTrigger(day_of_weekmon-fri, hour15, minute30), iddaily_kline_job, replace_existingTrue, misfire_grace_time3600, # 错过执行后1小时内补跑 ) scheduler.start()如果部署在Linux服务器上很多人会直接用crontab方案更轻。无论哪种方式建议采集时间定在收盘后半小时以上因为接口数据源自身有落地延迟15:00收盘立刻去拉经常出现部分股票数据不完整的情况。宁可晚半小时也不要拿不完整的数据入库。3.5 日常维护增量更新与数据自检采集系统跑起来之后重点就变成了数据质量和任务稳定性。增量更新是必须的全量更新不仅慢还会给接口造成不必要的压力。设计上我是这么做的每天拉取最近5个交易日的日K覆盖更新到库里。为什么是5天而不是1天因为遇到节假日调休、临时停牌或者接口漏数5天的窗口足够把缺失的数据补回来。每周再做一次数据自检脚本检查最近一段时间的数据完整度SELECT stock_code, COUNT(*) AS day_count FROM stock_daily WHERE trade_date DATE_SUB(CURDATE(), INTERVAL 10 DAY) AND fq_type none GROUP BY stock_code HAVING day_count 6;查出明显缺失的股票触发一次补拉任务。有了自检机制系统才能在没人盯的时候自我修复。4. 实战中的高频报错与避坑记录跑数据采集的时间长了什么奇怪的错误都能遇到。我把最常见的几类问题整理成速查表做避坑参考。4.1 常见异常与排查速查表错误现象可能原因处理方法接口返回 HTTP 400参数格式不对代码格式、日期格式有问题先用单个请求逐步检查参数确认code前缀和日期格式返回 HTTP 401 / 403Token失效或没有权限重新申请Token确认账号是否有目标数据权限返回 HTTP 429请求频率超出限制加大请求间隔引入限速器和指数退避重试超时无响应网络问题或并发过高设置15秒超时降低线程数分批次拉取返回数据全为空停牌、新股未上市、代码错误用交易日历对齐标记停牌状态日期字段变成乱码编码问题或时间戳格式不统一统一在清洗层转成“YYYY-MM-DD”字符串除权后历史价格突变前复权/后复权理解有误入库使用不复权回测时动态计算复权因子上面表格里最常见的其实还是400。接口报400先别怀疑接口挂了百分之八十是你参数没传对。我自己排查400的思路是先拿文档里的示例请求原样跑一次确认接口正常再把自己的参数逐步替换进去这样能很快定位到是哪一个字段的问题。4.2 几个容易被忽略但很关键的坑第一个坑是交易日历。很多新手直接用自然日去遍历日期遇到周末和节假日就全部浪费了请求额度还会得到一堆空数据。正确做法是先建一张交易日历表把每年交易日提前维护好循环时只遍历交易日。第二个坑是并发控制。为了提高效率很多人第一反应就是上多线程。但数据接口的限频机制往往和账号绑定线程再多账号被限流了还是白搭。我实测下来日K数据这类接口用3到5个线程并发每个请求之间保持0.2秒左右的间隔比单线程快又不容易触发限频。第三个坑是数据源状态判断。有时候接口返回200但里面是空列表或者错误提示千万不要只判断HTTP状态码要解析返回体里的业务状态码否则你入库的就是一堆空数据还以为是正常情况。第四个坑是增量日期边界。写增量任务时起始日期很多人习惯写“昨天”但遇到周一或节后昨天根本就不是交易日。稳妥的做法是以最近一次库里的最大日期为基准再往前多取一个交易日做覆盖。4.3 安全和合规这个必须单独说最后这点我觉得有必要单独强调。数据接口的使用一定要遵守服务商的使用条款和数据授权协议。申请API的时候文档里通常都写清楚了允许的使用范围、调用频率上限、数据是否可以二次分发。个人学习和研究用途和机构商业用途权限边界是不一样的。在实际操作中有几条底线我建议任何人都别碰第一不要尝试突破接口的权限限制去拉取自己没有权限的数据第二不要修改或者伪造请求绕过服务商的鉴权机制第三不要对接口做超出正常使用范围的并发请求第四拿到数据的Token和密钥绝对不要提交到公开仓库也绝对不要分享给别人。合规不只是在保护服务商更是在保护你自己。数据服务账号一旦被判定为违规使用轻则封接口重则影响账号主体这对长期做数据研究的人来说非常不值。最后再分享一个小经验这套系统从第一版到现在中间改过很多次最大的体会是采集系统真正的难点不在“能跑”而在“长期稳定地跑”。接口参数、数据格式、调度策略这些都可以在短期内搞定但数据链路里各种稀奇古怪的异常只有靠时间才能暴露出来。建议第一次搭的时候先只跑30只股票的日K观察一周把日志和告警都调顺了再逐步扩容到全市场。先让链路稳定再追求覆盖度这个顺序一定不要反过来。数据量上来之后还可以在现有采集系统后面接一层数据处理用大模型自动生成每日复盘摘要但那是下一个阶段的事了先把数据底座打好。