基于Python+MySQL+Docker的股票数据采集与回测系统搭建实践

1. 从标题到落地OpenStock到底是个什么东西先说我为什么折腾OpenStock。过去几年我一直在做个人股票数据研究市面上的股票软件和在线工具说实话功能都很全但我真正需要的几个能力几乎没人能同时给我拿到干净的历史行情数据、按自己的逻辑做筛选和回测、把结果存到自己库里随时查询。各家软件要么数据封闭要么策略写死在产品里要么导出数据还要买会员总之就是“别人给的都不顺手”。后来我决定自己动手搭了一套开源的个人股票数据终端取名OpenStock。它解决的问题非常简单把免费公开的行情数据源通过一套自动化的采集任务拉取下来存入本地数据库再做统一的查询、策略回测和信号提醒。做完之后我再也不用天天打开网页手动刷新也不用为了验证一个选股逻辑去Excel里手工比对几千个文件。这篇文章我不打算空谈概念而是把这套系统的完整搭建过程写出来从表结构设计、数据采集、任务调度到策略回测和容器化部署每一步都给出可以直接照着抄的方案和代码。适合的人群也很明确有一定编程经验、想把股票数据分析做得更自动化的开发者和研究者。零基础的读者也能看懂思路但动手复现时至少需要一点Python基础。在开始之前有个很重要的前提必须先说明OpenStock只做数据和策略研究工作所有行情数据来自公开合规的数据接口不涉及任何内幕信息和非法采集。用它跑出来的任何结论都仅供个人研究参考不构成投资建议。这个边界从一开始就要划清楚。2. 整体设计与模块拆解不是堆功能是梳理流程很多人一上手就急着写爬虫、画K线结果三天热度一过代码和零散数据堆积如山最后什么也没跑通。我最初也犯过这个毛病后来把需求想清楚了才重写。OpenStock的设计核心其实就一条主线围绕“数据从哪来、存哪里、怎么用”这条链路来组织模块。2.1 系统功能定位一个个人投研数据库而不是交易软件OpenStock不是交易客户端它不做下单买卖也不接券商的实盘交易通道。它更像一个“个人研究助理”帮你把行情数据整理好、把选股想法跑出结果、把异常波动及时通知到你手机上。这个定位非常重要想清楚边界之后很多技术选型就变得顺理成章。系统实际拆成四块数据采集模块、数据存储层、策略引擎、信号通知模块。数据采集模块负责从免费公开接口抓取日线数据和基础信息存储层用MySQL保存历史行情和筛选结果策略引擎跑我自己的选股逻辑和回测信号通知在发现符合条件的标的后自动推送到企业微信或邮件。这样的拆分带来一个直接好处模块之间通过数据库表结构解耦哪个环节有问题就修哪个环节不会一崩全崩。后来我在给系统加新功能时完全不需要改动数据采集的代码只是新建了几张策略结果表而已。2.2 技术选型考虑为什么是Python MySQL Docker技术栈的选择我犹豫过很长时间。刚开始用的爬虫脚本直接跑在笔记本的Python环境里数据存在SQLite中。后来数据量上来SQLite频繁锁库查询越来越慢我决定正式重写时才对技术栈做了梳理。最终定下来的是四个核心组件Python 3.10负责核心逻辑MySQL 8.0作为存储引擎Redis做轻量任务队列Docker Compose做一键部署。选Python是因为数据生态最全以akshare为核心的免费行情接口本身就是Python库选MySQL是因为历史数据天然适合关系型存储按股票代码和时间联合查询非常方便Redis则用来做任务去重和状态记录避免重复跑采集任务。部署方式我选择了Docker这是让我后期省心最多的一步。以前手工配环境换台机器要折腾半天兼容性问题层出不穷。用Compose之后一条命令就能把MySQL、Redis、采集服务全部拉起来。对于个人项目来说这套组合的成本几乎为零但稳定性和可维护性远远超出了我的预期。3. 数据层设计一张好表胜过后端十倍补丁数据层是整个OpenStock的地基地基没打好的话后续策略写得再漂亮都是空中楼阁。我在第一版设计中就犯过一个典型错误为了灵活把行情数据设计成“日期股票代码指标名指标值”的长表结构。查询单个股票的历史走势时要把几十行指标行再拼回去效率极其低下后来才意识到对量化场景来说宽表才是行情数据的正确打开方式。3.1 行情数据表与基础信息表怎么设计OpenStock里面最关键的一张表是daily_quote用于存储所有股票的日线行情。字段设计看起来不起眼但每一列都和后面的策略查询息息相关字段名类型说明idBIGINT主键自增stock_codeVARCHAR(10)股票代码如600519stock_nameVARCHAR(50)股票名称冗余存储便于展示trade_dateDATE交易日期openDECIMAL(10,2)开盘价highDECIMAL(10,2)当日最高价lowDECIMAL(10,2)当日最低价closeDECIMAL(10,2)收盘价volumeBIGINT成交量股amountDECIMAL(20,2)成交额元amplitudeDECIMAL(8,2)振幅%pct_changeDECIMAL(8,3)涨跌幅%updated_atDATETIME记录更新时间这张表建好之后我对stock_code和trade_date建了联合唯一索引从源头防止同一个股票同一天的数据重复入库。实际项目里这个唯一索引帮我拦截掉了大量由于接口重复推送导致的数据重复问题。基础信息表stock_basic则独立存放股票的静态信息包括股票代码、名称、上市日期、所属行业、总市值、流通市值、PE、PB等。把基础信息和行情数据分开有两个好处一是行情数据按交易日增长体量远大于基础信息分开存储能让备份策略更灵活二是基础信息更新频率低并不需要像行情表那样频繁写入。3.2 分区表与索引策略让千万级数据查询依然秒回当数据积累到一定程度只靠索引不够查询性能还是会下降。我目前的库已经存了三年全市场日线数据单表接近千万行。这个量级对MySQL不算压力但如果不做分区查询一只股票三年的K线仍要扫全表耗时不可接受。我的方案是对daily_quote表按年份做RANGE分区。具体来说用trade_date作为分区键每一年一个分区。这样查询“某只股票2022年以来的全部行情”时MySQL只需要定位到对应的三个分区扫描的数据量直接缩小一个量级。建表时分区定义大致是这样CREATE TABLE daily_quote ( id BIGINT NOT NULL AUTO_INCREMENT, stock_code VARCHAR(10) NOT NULL, stock_name VARCHAR(50), trade_date DATE NOT NULL, open DECIMAL(10,2), high DECIMAL(10,2), low DECIMAL(10,2), close DECIMAL(10,2), volume BIGINT, amount DECIMAL(20,2), amplitude DECIMAL(8,2), pct_change DECIMAL(8,3), updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id, stock_code, trade_date) ) PARTITION BY RANGE (YEAR(trade_date)) ( PARTITION p2020 VALUES LESS THAN (2021), PARTITION p2021 VALUES LESS THAN (2022), PARTITION p2022 VALUES LESS THAN (2023), PARTITION p2023 VALUES LESS THAN (2024), PARTITION p2024 VALUES LESS THAN (2025) );这里要注意一个细节主键必须包含所有分区键字段所以我把stock_code和trade_date都放进了主键里。第一次建分区时我已经踩过这个坑MySQL会直接报“A PRIMARY KEY must include all columns in the tables partitioning function”。4. 数据采集与调度让系统和交易日历一样准时有了表结构之后最核心的工作就是写采集程序。行情数据的获取是整个项目的基础拿不到干净、连续的数据后面无论策略多花哨都没有意义。我用的数据接口是akshare它属于开源免费的数据接口库聚合了全市场公开的行情数据比如某交易所的日行情数据和上市公司的基本信息等研究用途完全够用。4.1 核心采集逻辑与增量更新方案采集模块的代码逻辑并不复杂核心是一个循环遍历股票代码列表逐个拉取行情数据再逐批写入数据库。但真正写起来有几个隐藏的坑。第一次跑全量历史数据时我用了同步循环方式也就是读一只股票的全部历史行情再读下一只。全市场五千多只股票一个一个同步拉预计要跑十几个小时而且中途一旦网络抖动整个任务就断了。第二次我改成了多线程并发采用线程池来控制并发数每只股票一个任务实测下来速度提升非常明显不到两个小时就能跑完全市场的历史日线。采集线程的关键代码大致是这个样子import akshare as ak from concurrent.futures import ThreadPoolExecutor, as_completed def fetch_daily(stock_code: str, start_date: str, end_date: str): df ak.stock_zh_a_hist(symbolstock_code, perioddaily, start_datestart_date, end_dateend_date, adjustqfq) if df is None or df.empty: return # 统一列名转成符合daily_quote表结构的DataFrame df.rename(columns{ 日期: trade_date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume, 成交额: amount, 振幅: amplitude, 涨跌幅: pct_change }, inplaceTrue) df[stock_code] stock_code df[stock_name] get_stock_name(stock_code) return df[[stock_code, stock_name, trade_date, open, high, low, close, volume, amount, amplitude, pct_change]] def run_full_update(stock_list, start_date, end_date, max_workers16): results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: future_map {executor.submit(fetch_daily, code, start_date, end_date): code for code in stock_list} for future in as_completed(future_map): try: df future.result() if df is not None: results.append(df) except Exception as e: print(f抓取失败{future_map[future]}错误{e}) return results增量更新则是另一个思路。每个交易日收盘后只需要抓取当天的最新行情。我写了一个增量任务先查询daily_quote表里每个股票代码的最大交易日然后只拉取这个日期之后一天到当天的数据。在实际操作中大多数股票每天只会新增一行几千只股票加起来也只是一次非常轻量的请求。4.2 定时任务与幂等设计每天16点半自动跑更新定时任务我用了系统自带的定时器服务写了一个shell脚本每天下午五点半触发增量更新。为什么选五点半而不选收盘的三点整因为公开数据源的数据落库有时间延迟三点收盘后接口马上请求经常出现某些股票当天的数据还没更新完。推迟到五点半跑数据就比较完整了。任务调度的核心设计原则是“幂等”。同一个任务重复执行不能产生重复数据不能因为中断导致任务状态错乱。幂等在增量更新场景里有两个层面的保障在数据入库时INSERT ... ON DUPLICATE KEY UPDATE利用唯一索引处理重复写入在任务调度层面Redis里存一个task:last_run_date的键每次启动任务时先检查今天是否已经跑过避免误触发重跑。调度脚本本身不复杂核心就是在系统定时器里加一条任务配置30 17 * * 1-5 cd /opt/openstock /usr/local/bin/docker-compose exec -T worker python main.py --mode incremental /var/log/openstock/update.log 21注意这里我用了docker-compose exec -T好处是采集任务和服务本身一样在容器里运行环境隔离、依赖完整。实测跑了大半年几乎没有因为系统环境问题出过错。4.3 限流与数据完整性保护别为了快让IP被封多线程加速很快但代价是很容易触发公开数据源的限制策略。我第一次用32个线程跑全量更新跑了不到十分钟请求就开始大量报错连续重试甚至导致封了一段IP。后来我把线程数压到16个并且在每次请求之间加了微小的随机延迟模拟人工操作节奏问题才得到缓解。除了控制请求频率数据完整性也被我列在了重要位置。行情数据有一个特殊之处历史数据可能会被数据源修正。比如某天发生除权除息如果采集时用的是前复权行情那么全历史的价格都可能发生变化。如果不做处理库里存的历史数据会和最新一次抓取的数据对不上做回测时就会出现明显错误。我的对策是每周日跑一次全量的重刷任务对最近250个交易日的行情做整体刷新用最新数据覆盖旧数据。这样既不用每天全量重刷又能保证策略依赖的近期数据是比较准确的。5. 从数据到决策策略回测与信号通知模块的实现数据采集只是起点真正的重头戏在策略引擎。我搭建OpenStock的最终目的是把脑子里那些“如果……就……”的投资想法变成代码跑出历史数据上的结果再看值不值得在实盘中关注。整个过程离不开两个能力回测和信号提醒。5.1 回测框架集成用历史数据验证你的选股逻辑回测我用了Backtrader这个Python回测框架它最大的优势是策略逻辑和数据加载完全分离。我只需要实现一个最简单的双均线策略来演示OpenStock怎么和Backtrader对接。回测的数据加载有个关键步骤要把OpenStock库里的数据导成Backtrader认识的格式。较简单的方式是把daily_quote表的数据读成Pandas Dataframe然后按Backtrader要求的列名重命名再传入bt.feeds.PandasData。双均线策略核心代码如下import backtrader as bt class DualMovingAverageStrategy(bt.Strategy): params dict( fast_period5, slow_period20 ) def __init__(self): self.fast_ma bt.indicators.SMA(self.data.close, periodself.p.fast_period) self.slow_ma bt.indicators.SMA(self.data.close, periodself.p.slow_period) self.cross bt.indicators.CrossOver(self.fast_ma, self.slow_ma) def next(self): if not self.position: if self.cross 0: self.buy() elif self.cross 0: self.close() def run_backtest(stock_code: str, start: str, end: str): data_df load_daily_from_db(stock_code, start, end) data_df.columns [datetime, open, high, low, close, volume, openinterest] data_df[datetime] pd.to_datetime(data_df[datetime]) data_df.set_index(datetime, inplaceTrue) cerebro bt.Cerebro() data_feed bt.feeds.PandasData(datanamedata_df) cerebro.adddata(data_feed) cerebro.addstrategy(DualMovingAverageStrategy) cerebro.broker.setcash(100000.0) cerebro.broker.setcommission(commission0.0003) print(f初始资金: {cerebro.broker.getvalue():.2f}) results cerebro.run() print(f期末资金: {cerebro.broker.getvalue():.2f})这里有一个非常重要的原则回测和实盘行情的数据必须同源同口径。如果回测时用前复权数据而你在看盘软件里看到的是不复权价格两者之间的差异会让策略信号完全失真。我的做法是OpenStock统一使用前复权数据并且在前端展示和策略参数中明确标注这个口径。5.2 选股信号入库把每日筛选结果固化下来每交易日收盘后我会跑一批选股条件把符合形态的股票筛选出来存入stock_signal表。这个表的结构比行情表简单得多关键字段包括运行批次ID、股票代码、股票名称、信号类型、触发价、收盘价、信号日期和附加信息。信号入库的价值在于“可复盘”。每天筛选出的标的如果只放在日志里过一个月就再也找不回来了。存进数据库之后我可以回溯某天的信号看它后续的走势不断调整筛选条件。我把策略结果和数据源放在同一套MySQL里备份和查找都方便。5.3 通知推送企业微信机器人极简接入信号有了如果不开OpenStock系统就看不到结果那自动化的意义就少了一半。我接入了企业微信的群机器人来做推送配置起来非常简单在目标群里添加一个自定义机器人拿到Webhook地址然后用requests.post把消息发过去即可。推送代码大致如下import requests import json def send_wechat_message(webhook_url: str, content: str): headers {Content-Type: application/json} payload {msgtype: text, text: {content: content}} resp requests.post(webhook_url, headersheaders, datajson.dumps(payload)) return resp.status_code 200每日收盘后OpenStock会把当天所有信号整理成一条格式化消息推送到群里内容包括股票代码、名称、信号类型和触发价格。这样我不用打开电脑也能及时知道当天有哪些标的需要关注。用企业微信机器人而不是自己搭推送服务成本几乎为零稳定性还特别好跑了半年没有一次漏推。6. 部署上车与例行维护Docker Compose让搭建十分钟搞定系统本地跑通之后下一步就是正式部署。我强烈建议直接用Docker Compose把整套服务打包起来别在自己电脑上裸奔部署。裸奔的问题是你换了机器、改了系统、升级了Python都可能把环境搞挂。容器化之后整个系统的运行环境被固化成镜像文件不管到哪台机器都能一致运行。6.1 Compose编排实战一条命令拉起全套服务我的项目目录结构非常直观openstock/ ├── docker-compose.yml ├── mysql/ │ └── init.sql ├── worker/ │ ├── Dockerfile │ ├── main.py │ ├── collector.py │ ├── strategy.py │ └── requirements.txt └── .env核心的docker-compose.yml只需要三个服务定义分别负责数据库、缓存和采集任务version: 3.8 services: mysql: image: mysql:8.0 container_name: openstock-mysql restart: always environment: MYSQL_ROOT_PASSWORD: ${MYSQL_ROOT_PASSWORD} MYSQL_DATABASE: openstock ports: - 3306:3306 volumes: - mysql_data:/var/lib/mysql - ./mysql/init.sql:/docker-entrypoint-initdb.d/init.sql:ro redis: image: redis:7-alpine container_name: openstock-redis restart: always ports: - 6379:6379 volumes: - redis_data:/data worker: build: ./worker container_name: openstock-worker restart: always depends_on: - mysql - redis environment: DB_HOST: mysql DB_PORT: 3306 DB_USER: root DB_PASSWORD: ${MYSQL_ROOT_PASSWORD} DB_NAME: openstock REDIS_HOST: redis REDIS_PORT: 6379 volumes: - ./worker:/app command: python main.py --mode service volumes: mysql_data: redis_data:这里有个小细节值得一说worker服务把宿主机的./worker目录挂载到了容器内的/app。这样开发调试时改了代码不用重新构建镜像就能生效非常方便。但生产环境里这个用法要慎用毕竟挂载了源码目录后“环境一致性”就只能靠自觉了。启动服务只需一条命令docker-compose up -d首次启动时MySQL容器会自动执行init.sql完成建库建表不用手动连MySQL执行脚本。Docker官方镜像支持把初始化脚本放在/docker-entrypoint-initdb.d目录下自动执行这个机制帮我省了不少事。6.2 例行维护事项备份、监控和日志部署完成不等于服务能一直稳定跑下去。我维护OpenStock的日常事项主要有三件备份数据库、检查采集日志、定期更新依赖。数据库备份是我踩过坑之后才高度重视的。有一次我手滑在测试环境误删了一张表当时还没开启自动备份只能靠着之前的人工导出恢复数据丢失了最近两周的行情。后来我写了一个简单的备份脚本每天凌晨三点用mysqldump把整个openstock库导出到备份目录然后保留最近30天的备份文件。平时不觉得它重要真要恢复数据的时候才知道它的价值。采集日志的检查也很有必要。OpenStock的所有任务运行日志统一写到/var/log/openstock/目录下按日期命名。我会每天早上看一遍昨天的日志有没有异常报错特别关注“抓取失败”和“写入失败”这两类错误。大多数失败都是数据源临时不可用重试即可解决但如果连续几天同一批股票都抓取失败就要考虑是不是接口数据结构发生变化了。6.3 常见问题速查表从莫名报错到数据不连续运行了这段时间我把最常遇到的问题整理成了一张速查表希望能帮后来者少走弯路。问题现象可能原因解决方法某只股票历史行情缺失该股票可能刚上市历史区间过短检查起止日期确认上市日期抓取任务频繁报错请求过于频繁触发数据源限流降低并发数增加随机延迟数据库连接数耗尽采集线程和查询服务共用连接池为worker单独配置连接池上限回测结果与预期偏差大没有处理除权除息数据统一使用前复权数据口径容器启动后MySQL无法初始化初始化脚本存在语法错误查看容器日志定位具体错误行采集到重复数据没有建唯一索引或索引失效检查联合唯一索引是否生效某天行情数据大面积缺失数据源当日数据更新延迟推迟定时任务触发时间手动补跑速查表看起来简短但每条都是真金白银踩出来的。比如“数据库连接数耗尽”这条最初采集任务和Web查询服务共享同一个连接池全量刷新时瞬时连接数几十个直接把MySQL连接数顶满了。后来把worker单独配置连接池参数问题彻底解决。7. 踩坑实录三个最让我印象深刻的低级错误除了上面的速查表有三个坑我必须单独拿出来说因为它们不是一查就知道的配置问题而是根深蒂固的“认知坑”。第一个坑是前复权数据的“中心漂移”问题。前复权数据以最新价格为基准调整历史价格所以每一次除权除息之后前复权的历史价格都会整体变化。我用OpenStock跑了一个长期策略的回测隔一段时间再跑结果竟然不一样当时还以为程序有bug。后来才意识到是前复权数据本身在变化。这个问题的标准解法是回测时固定某个时点的前复权数据或者用不复权数据自己做除权调整。我目前的做法是每次回测前从OpenStock拉取数据并打上数据版本号算完结果一起保存避免“同一个策略、同一天、结果对不上”的尴尬。第二个坑是时区问题。虽然我们做的是A股数据只涉及东八区但Docker容器默认时区是UTC。采集任务里如果用datetime.now()拿当前时间在容器里拿到的和宿主机差了八个小时。最直接的后果是定时任务的日志时间戳全乱了排查问题时特别绕。解决办法是在Dockerfile里设置时区或者用第三方库时区对象统一处理时间。我最后在docker-compose.yml里给所有服务添加了TZ: Asia/Shanghai环境变量一劳永逸。第三个坑是MySQL的ON DUPLICATE KEY UPDATE的“副作用”。用这个语法处理重复数据时如果只是重复插入MySQL会把updated_at字段自动更新为当前时间。这看起来没问题但它会导致一个问题当你全量重刷某段历史数据时即使数据没有变化这些记录的updated_at也被刷新了。如果后续有基于updated_at做数据增量同步的逻辑就会把它搞糊涂。我的处理方式是给幂等写入的更新语句显式排除updated_at列的更新或者另外加一个字段专门记录数据实际变更时间。8. 下一步还能怎么玩给系统的可扩展性留好口子OpenStock现在已经稳定运行了大半年我也在这套系统上验证了多种不同的数据思路。它最大的价值不是帮我“赚了多少钱”而是让我有了一个随时可以试验想法的平台。有些思路在回测里表现不错放到实际行情里可能又不一样但起码我能用历史数据快速过滤掉明显不靠谱的想法这在没有系统之前是做不到的。如果你也想搭一套类似的东西我的建议是不要一开始就追求大而全先把“单只股票的数据入库、查询、简单策略”这条最小链路跑通再逐步加功能。一个能跑的最小系统胜过十个只停留在设计文档里的完美架构。未来的扩展方向其实不少。比如把分钟级数据也纳入采集范围做更精细的交易策略研究或者加入更多的技术指标计算库直接在数据库层面完成指标预计算加速回测。数据源方面也可以接入更多公开合规的信息源把财务数据、公告信息都整合进来形成更完整的研究维度。但无论怎么扩展数据层的规范设计和采集任务的稳定调度永远是整个系统的根基。根据我个人经验自建这套系统的过程给我最大的收获反而不是代码本身而是对数据质量的敬畏之心。一个数字的偏差、一天数据源的延迟、一次任务的中断都可能在某个角落放大成完全不可信的结论。如果你决定动手搭建请一定在数据完整性和可重复性上多花时间这比调出一个完美的策略参数重要得多。