前阵子和几个朋友聊起行情软件他们说每天看那些K线、均线、MACD指标总觉得像在对着一个黑盒——数据怎么来的、指标怎么算的、买卖信号怎么触发的全是封装好的结果。我说那干脆自己搭一套出来把数据抓取、清洗、存储、计算、展示这条链路整个打通一遍于是就有了OpenStock这个项目。这套开源股票数据分析系统主要解决三件事全量行情数据自己管、技术指标自己算、策略信号自己定义。不管你是想入门量化分析、想搞懂数据链路还是单纯不想被商业软件绑住都可以照着这套设计思路和代码从零搭一套属于自己的股票数据分析环境出来。1. 这个项目解决什么问题为什么要自己搭1.1 不满足于只是“看盘”很多人打开行情软件主要就是在界面上看涨跌颜色最多切到K线图划拉几下均线。但如果你往深了问一句“这个MACD参数是谁定的、我的选股条件到底怎么写成代码”市面上大部分软件给不了你答案更给不了你自定义的自由。我搭OpenStock的初衷就是把这几件事从“只能看”变成“能改、能算、能跑”行情数据要落到自己的数据库里想做多少年回测都不受限制指标逻辑完全自己实现不是软件里写死的那几个选项可以把选股条件、买卖信号变成代码一键跑出结果整个链路从数据到界面都看得见摸得着哪天想扩展新功能改代码就完事。说白了商业行情软件解决的是“别人定义好的功能怎么用”OpenStock解决的是“我自己想定义的功能怎么做出来”。两件事的价值完全不在一个量级上。1.2 技术选型把复杂度控制在能驾驭的范围做这个项目之前我认真对比过几套技术方案最后敲定的是Python全家桶Python akshare SQLite Flask ECharts。先说Python这个基本没什么悬念。数据分析领域Python生态最成熟pandas处理行情数据、计算指标简直是为这个场景量身定做的写起来顺、资料也好查。数据源选akshare核心原因是免费且没有历史长度限制A股日线数据可以从上世纪90年代一直拉到现在对个人项目和策略研究来说完全够用。相比某些需要注册申请Token的数据平台akshare开箱即用省了不少事。存储层用SQLite这里可能有人会觉得是不是过于“简陋”了。但OpenStock的定位是个人研究工具不是生产级量化系统。单机、单用户、日线级别数据这个量级SQLite处理起来绰绰有余。一张表存上几千只股票十几年的日线也就几百万行SQLite稳定扛得住。而且用文件当数据库拷到哪都能跑部署成本无限趋近于零这对个人项目来说太重要了。Web层用Flask原因无它就是轻。整个项目需要提供的接口就那么几个Flask几十行代码就能把所有路由和API写完不需要上Django这种重型框架。可视化选ECharts它免费、中文文档友好、K线图开箱即用股票图表所需的蜡烛图、均线、成交量、MACD副图都有现成方案省掉自研前端图表的大把时间。1.3 整体架构四层各干各的活OpenStock整个系统划分成四层各层之间耦合很弱单独拎出来都能用数据采集层负责从数据源拉取日K线行情做字段清洗、去重、入库存储层SQLite数据库保存原始行情数据和同步日志计算引擎层读取行情数据计算均线、MACD等指标输出交易信号展示层Flask提供HTTP接口前端用ECharts渲染K线图和指标图。这个设计思路在动手写代码前一定要想清楚因为分层之后你会发现每个模块都能独立调试。数据采集出问题了不用怀疑指标算错了前端展示不对不用怀疑库里数据有问题排查效率能提升一大截。2. 数据链路从K线数据到数据库2.1 数据源免费但需要接得住变化akshare的数据最终来源于公开财经网页或交易所披露数据它的优点是免费、信息全但缺点也很鲜明——它是靠爬取公开数据实现的所以上游页面结构一变接口就可能报错某些字段名可能也会跟着变。基于这个现实我在写数据采集模块时做了一层字段映射的缓冲。akshare返回的A股日线数据默认是中文列名比如“日期”“开盘”“收盘”“最高”“最低”“成交量”我在代码里统一重命名成英文这样就算哪一天接口返回的列顺序变了只要列名没变映射逻辑依然能正常工作。这个习惯建议所有人保留代码里永远不要直接依赖“列的位置”而是要依赖“列的名字”。日线数据主要用到这些字段字段含义用途code股票代码主键的一部分区分标的trade_date交易日期主键的一部分排序和切片用open / close开盘价 / 收盘价K线绘制、指标计算high / low最高价 / 最低价K线绘制、波动分析volume成交量量能分析、副图展示amount成交额资金流向参考pct_chg涨跌幅筛选、排名这里有个细节值得花点时间说清楚只存前复权数据在个人研究场景下是够用的。前复权数据以当前价格为基准对历史价格做除权修正它最大的好处是价格连续、不会因为分红送股出现跳空适合日常看盘和画图。如果以后要做严肃的策略回测可以再额外把不复权数据和复权因子一起入库按需计算后复权价格。2.2 表结构一张主表加一张日志表数据库设计不搞花活两张表就够用。一张存行情数据一张存同步日志。行情表的主键设计是关键要用“股票代码 交易日期”做联合主键这样做有两个直接好处一是天然去重同一只股票同一天的数据只会存在一行同步脚本重复执行也不怕二是联合主键自带索引按股票代码和时间范围查询数据时可以走索引不用全表扫描。建表语句如下CREATE TABLE IF NOT EXISTS stock_daily ( code TEXT NOT NULL, trade_date TEXT NOT NULL, open REAL, close REAL, high REAL, low REAL, volume INTEGER, amount REAL, pct_chg REAL, PRIMARY KEY (code, trade_date) ); CREATE INDEX IF NOT EXISTS idx_code_date ON stock_daily(code, trade_date); CREATE TABLE IF NOT EXISTS sync_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, code TEXT, sync_date TEXT, rows_added INTEGER, status TEXT, message TEXT, created_at TEXT DEFAULT (datetime(now, localtime)) );日志表的作用很容易被忽略但实际用到后面你会发现它特别重要。行情数据同步是个定时任务哪天没跑成功、跑了多少行、报了什么错全都要有记录不然出了问题你根本无从查起。另外强烈建议在初始化数据库后执行一行PRAGMAPRAGMA journal_modeWAL;这是SQLite的预写日志模式能把并发读写的性能提升一截尤其是后面我们会用Flask的API接口去读数据同时后台定时任务还在写数据WAL模式能有效避免“数据库被锁”这种报错。后面第4章会详细展开这个坑。2.3 复权和指标计算很多人容易搞翻车的地方技术指标计算逻辑本身不难难的是搞清楚“拿什么价来算”。这里必须先聊透复权。A股分红送股后股价会进行除权除息比如一只股票从20元变成10元但你的持仓市值没变因为股数翻了一倍。如果直接拿原始价格算均线或MACDK线上就会出现一个向下跳空的大坑所有指标都会被这个非市场因素干扰产生大量假信号。因此在做指标计算时必须使用复权价格。前复权的基本逻辑是保持最新价格不变将历史价格按分红送股比例等比缩小。你在行情软件里看到的K线默认基本都是前复权的。我在OpenStock里用的是akshare接口直接拉前复权数据入库然后基于复权价计算指标。对于个人做技术面分析来说这个方案简单且可靠。如果你以后要做因子回测、更严格的绩效评估建议改成“存储不复权价格 计算复权因子”的方式数据层面会灵活很多。均线和MACD是整套系统最核心的两个指标代码也直接体现三层计算逻辑def add_indicators(df: pd.DataFrame) - pd.DataFrame: df df.sort_values(trade_date) # 第一层简单移动平均线 for window in [5, 10, 20, 60]: df[fma{window}] df[close].rolling(window).mean() # 第二层MACD 中的快慢线 ema12 df[close].ewm(span12, adjustFalse).mean() ema26 df[close].ewm(span26, adjustFalse).mean() df[dif] ema12 - ema26 # 第三层DEA 线和 MACD 柱 df[dea] df[dif].ewm(span9, adjustFalse).mean() df[macd] (df[dif] - df[dea]) * 2 return df这段代码里用了pandas的rolling和ewm前者是固定窗口均值后者是指数加权均值。ema12代表12日指数移动平均线ema26代表26日指数移动平均线两者相减得到DIF线。DIF再做9日指数平滑得到DEA线MACD柱就是两者差值的两倍。这些参数都是约定俗成的除非你有明确的研究依据否则不建议随便改。3. 手把手搭建从0到跑起来3.1 环境准备先准备好基础环境我用的是Python 3.10版本如果你的机器装的是3.8以上的版本问题都不大。建议用虚拟环境把项目依赖隔离起来避免和系统其他Python包相互污染。创建项目目录和虚拟环境mkdir openstock cd openstock python -m venv venv # Windows venv\Scripts\activate # macOS / Linux source venv/bin/activate安装依赖pip install akshare pandas sqlalchemy flask安装完成后可以把依赖导出到文件里方便别处部署pip freeze requirements.txtakshare的依赖比较多安装时间稍长耐心等它跑完就好。装完以后建议在Python交互环境里先试一把import akshare as ak确认没有报错再往下走。3.2 数据库初始化先建好地基项目里建一个db.py负责数据库连接和初始化。这里用SQLAlchemy的好处是后面如果要迁移到MySQL或PostgreSQL代码改动量很小。from sqlalchemy import create_engine, text DB_PATH data/openstock.db engine create_engine( fsqlite:///{DB_PATH}, connect_args{timeout: 30} ) def init_db(): import os os.makedirs(data, exist_okTrue) with engine.connect() as conn: conn.exec_driver_sql( CREATE TABLE IF NOT EXISTS stock_daily ( code TEXT NOT NULL, trade_date TEXT NOT NULL, open REAL, close REAL, high REAL, low REAL, volume INTEGER, amount REAL, pct_chg REAL, PRIMARY KEY (code, trade_date) ) ) conn.exec_driver_sql( CREATE TABLE IF NOT EXISTS sync_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, code TEXT, sync_date TEXT, rows_added INTEGER, status TEXT, message TEXT, created_at TEXT DEFAULT (datetime(now, localtime)) ) ) conn.exec_driver_sql(PRAGMA journal_modeWAL) conn.commit() if __name__ __main__: init_db() print(数据库初始化完成)注意connect_args{timeout: 30}这个参数它设置了SQLite连接在遇到锁时的等待时间默认值是5秒在后续可能会同时跑Web服务和定时任务的情况下30秒更稳妥。3.3 数据采集模块拉数据、洗数据、存数据数据采集是OpenStock的门面也是整个系统最容易出问题的一环。核心逻辑拆成三步调用akshare接口拉数据、重命名字段、增量入库。我在fetcher.py里实现的逻辑如下import akshare as ak import pandas as pd from sqlalchemy import text from db import engine # 中文字段名和英文字段名的映射 COLUMN_MAP { 日期: trade_date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume, 成交额: amount, 涨跌幅: pct_chg, } def fetch_daily(code: str, start_date: str, end_date: str) - pd.DataFrame: df ak.stock_zh_a_hist( symbolcode, perioddaily, start_datestart_date, end_dateend_date, adjustqfq ) df df.rename(columnsCOLUMN_MAP) df[code] code df[trade_date] pd.to_datetime(df[trade_date]).dt.date.astype(str) return df[[code, trade_date, open, close, high, low, volume, amount, pct_chg]] def sync_stock(code: str, start_date: str, end_date: str) - int: df fetch_daily(code, start_date, end_date) if df.empty: return 0 # 查询库里已有的日期做增量过滤 with engine.connect() as conn: result conn.execute( text(SELECT trade_date FROM stock_daily WHERE code :code), {code: code} ) existing_dates {row[0] for row in result} new_df df[~df[trade_date].isin(existing_dates)] if new_df.empty: return 0 new_df.to_sql(stock_daily, engine, if_existsappend, indexFalse) log_sync(code, len(new_df), success, ) return len(new_df)这里有两个操作是必须的第一是重命名字段把akshare返回的中文列名统一成自己库里的英文列名第二是增量过滤先查库里已有哪些日期只插入不重复的数据这样定时任务重复执行也不会产生脏数据。日志记录也是不能省的我单独写了一个log_sync函数往sync_log表里插一条记录。后面排查问题、确认数据是否同步成功全靠这张日志表。3.4 指标信号引擎把计算和信号分离指标计算单独拆出一个analyzer.py我始终坚持一个原则数据库只管存原始数据计算逻辑放在独立的模块里。这样即使指标算法写成渣了也不会影响底层数据完整性。信号判断我用了两个维度的逻辑一个统计维度均线金叉死叉一个趋势维度MACD柱方向变化def detect_signals(df: pd.DataFrame) - pd.DataFrame: df add_indicators(df) df[signal] 0 # 均线金叉MA5上穿MA20 buy_cross (df[ma5] df[ma20]) (df[ma5].shift(1) df[ma20].shift(1)) # 均线死叉MA5下穿MA20 sell_cross (df[ma5] df[ma20]) (df[ma5].shift(1) df[ma20].shift(1)) # MACD柱由负转正 macd_turn_pos (df[macd] 0) (df[macd].shift(1) 0) # MACD柱由正转负 macd_turn_neg (df[macd] 0) (df[macd].shift(1) 0) df.loc[buy_cross | macd_turn_pos, signal] 1 df.loc[sell_cross | macd_turn_neg, signal] -1 return df这里shift(1)是关键它把整列数据往下挪一行用来获取“前一天的指标值”。判断“今天上穿”必须拿“今天大于昨天”来对比而不是只看今天的大小不然你会把“一直大于”也当成金叉信号那这个信号系统就废了。注意信号计算出来的只是一个参考我个人在项目文档里也明确写了任何信号都不构成投资建议。这个系统的作用是帮你把规则固化成代码而不是替你决定买卖。3.5 Web可视化K线图和指标图Flask部分我建了app.py提供两个接口一个渲染页面一个返回JSON数据。from flask import Flask, jsonify, render_template, request import pandas as pd from sqlalchemy import text from db import engine from analyzer import detect_signals app Flask(__name__) def load_stock_data(code: str, limit: int 250): sql SELECT code, trade_date, open, close, high, low, volume, amount FROM stock_daily WHERE code :code ORDER BY trade_date DESC LIMIT :limit df pd.read_sql(sql, engine, params{code: code, limit: limit}) df df.sort_values(trade_date).reset_index(dropTrue) return detect_signals(df) app.route(/) def index(): return render_template(index.html) app.route(/api/kline) def api_kline(): code request.args.get(code, 000001) df load_stock_data(code) result { dates: df[trade_date].tolist(), kline: df[[open, close, low, high]].values.tolist(), volumes: df[volume].tolist(), ma5: df[ma5].round(3).tolist(), ma20: df[ma20].round(3).tolist(), dif: df[dif].round(3).tolist(), dea: df[dea].round(3).tolist(), macd: df[macd].round(3).tolist(), signals: df[[trade_date, signal]].values.tolist(), } return jsonify(result) if __name__ __main__: app.run(host0.0.0.0, port5000, debugFalse)前端用一个templates/index.html引入ECharts数据直接对接/api/kline接口。ECharts的K线图用candlestick系列均线和MACD用line系列副图用一个独立的grid放置MACD柱状图。前端这部分不展开太多细节核心就是拿到接口数据后用ECharts的setOption渲染。需要注意的一点是ECharts K线图的data顺序是[open, close, lowest, highest]而很多人习惯先写high和low这里很容易搞反。3.6 定时调度让数据每天自动更新手动跑同步脚本不是长久之计数据采集必须做成定时任务。我写了个scheduler.py用APScheduler库实现每天的定时同步from apscheduler.schedulers.blocking import BlockingScheduler from datetime import datetime from fetcher import sync_stock STOCKS [000001, 600519, 000858] # 自选股票池 def job(): today datetime.now().strftime(%Y%m%d) for code in STOCKS: try: rows sync_stock(code, 19900101, today) print(f{code} 同步完成新增 {rows} 行) except Exception as e: print(f{code} 同步失败: {e}) if __name__ __main__: scheduler BlockingScheduler(timezoneAsia/Shanghai) scheduler.add_job(job, cron, day_of_weekmon-fri, hour15, minute30) scheduler.start()调度时间设在交易日的15:30这个点A股已经收盘当日数据完整落地。day_of_weekmon-fri只在工作日跑碰上法定节假日虽然会空跑一次不过增量同步逻辑会直接返回0行不会有任何副作用。如果不想依赖APScheduler直接用系统的cron也能跑Linux/macOS在crontab里写30 15 * * 1-5 cd /path/to/openstock venv/bin/python scheduler.pyWindows就用计划任务。这个项目本身和系统调度器解耦怎么方便怎么来。4. 常见问题与排查实录4.1 接口报错与字段变动akshare这个库最大的特点就是接口会“时不时”变一下。我有一次跑同步脚本发现stock_zh_a_hist突然抛异常查了一下是上游数据结构做了调整akshare发了新版本修复。建议养成两个习惯一是定期pip install akshare --upgrade升级到最新版二是在采集代码外面包一层异常捕获出错了先记录日志不要整个定时任务直接崩掉。我排查这类问题的固定流程是这样的先看sync_log表里的报错信息再拿报错关键词去项目GitHub的Issue区搜最后试升级akshare版本。90%的问题在第二步和第三步就能解决。4.2 SQLite并发写入与锁多进程同时写会报错SQLite的并发写能力确实不如MySQL这类服务型数据库尤其是在Web服务进程和定时同步进程同时运行时“database is locked”这种错我踩过不止一次。解决方案就是前面提到的两板斧连接参数里加timeout30初始化时开启WAL模式。WAL模式允许读写并行读操作不会阻塞写操作写操作之间依然互斥但对OpenStock这种日更一次、偶尔查一下的场景来说已经足够了。另外还有一个容易被忽视的细节不要让定时任务用if __name__ __main__里的那套逻辑反复创建多个进程。APScheduler的BlockingScheduler本身是单进程的不要额外开多进程否则并发写SQLite会给自己找麻烦。4.3 数据对不上复权问题与停牌日数据对不上的问题主要有两类。第一类是复权数据“漂移”。前复权以最新价为基准每次分红除权后历史价格都会重新计算一遍所以不同时间拉取的同一段历史数据价格会不一样。你上周拉的数据和这周拉的数据在除权日之前的价格可能就对不上。这不是bug是前复权本身的特性。如果你要复现历史指标、做严格回测建议改成存不复权数据加复权因子。第二类是停牌日空数据。股票长期停牌期间不会有行情记录这会导致它在那段时间的均线是“断层”的复牌后重新计算均线需要的时间窗口变长。在同步脚本里停牌日直接不插入数据就行指标计算用pandas的rolling会自动跳过缺失值这个不用太担心。4.4 数据库文件与部署运行的环境坑把项目部署到服务器上后遇到最多的问题是路径问题。我的数据库文件用的是相对路径data/openstock.db如果你在项目的不同目录下执行启动命令就会生成多个数据库文件数据就乱套了。解决方式是使用绝对路径比如在db.py里基于文件所在目录拼接import os BASE_DIR os.path.dirname(os.path.abspath(__file__)) DB_PATH os.path.join(BASE_DIR, data, openstock.db)还有一个和代码无关但很实际的点服务器内存。akshare拉全市场数据的时候比较吃内存如果机器配置低建议控制并发数或者分批同步别一次性拉几千只股票。5. 还能往哪些方向扩展OpenStock搭起来之后后续扩展空间其实很大。我个人把目前觉得比较实用的几个方向整理一下第一自选股监控与告警。基于detect_signals算出来的信号把触发买入或卖出信号的股票推送到钉钉群或企业微信机器人这样每天收盘后自动收到一份“今日信号汇总”不用打开软件逐个翻。第二选股策略引擎。目前分析的维度只有均线和MACD可以继续加入RSI、布林带、KDJ等因子再按“全部满足才触发”的方式组合成复合策略比如“MA5上穿MA20 MACD柱转正 成交量放大1.5倍”这种多条件选股。第三真正的回测系统。用OpenStock攒了多年的行情数据配合backtrader这类开源的第三方回测框架把信号逻辑接入回测就能算出策略的历史收益、最大回撤、夏普比率这些关键指标。这一步做好了才是从“看盘工具”到“策略研究工具”的分水岭。我在实际搭建中最大的体会是这套系统最大的价值不在于指标多精确、界面多好看而在于你自己走通了从数据到决策的完整链路后面想加什么功能、改什么逻辑都会有底气。行情软件永远只能给你它想给你的东西而OpenStock这样的自建系统能让你按自己的思路去研究这个市场。
