企业官网建设流程全解析

你是不是也有过这种经历：上午还能跑的行情脚本，下午接口就报错了；好不容易把全市场数据抓到本地，算历史回撤时发现日期字段对不上；想找一个免费工具把自选股的K线、均线、成交额排行放到一个页面里，翻来翻去总有不满意的地方。OpenStock就是在这种背景下开始折腾的一个开源项目，目标很朴素：用自己的代码跑通一条完整的行情数据链路——从数据采集、存储、指标计算，再到Web可视化展示。这篇文章会按我实际搭建的过程来讲，从架构设计到每一段关键代码，再到运行中踩过的坑。如果你正在做个人投资分析、准备入门量化，或者想找一个Python后端练手项目，这套思路可以直接参考。

1. OpenStock到底解决什么问题

1.1 免费接口的痛点和自建系统的价值

用现成行情软件和免费接口的时候，最让人头疼的一点是“数据不自由”。行情软件展示的指标是别人定义好的，你想把自定义的均线周期、换手率分位数、各行业涨跌统计放在同一个视图里，几乎做不到。网页接口呢，通常只提供一段时间的日线，历史分钟线要么收费，要么需要积分，字段还经常变动。靠手动导出Excel维护数据，时间一长基本都会放弃。

自己搭OpenStock这一类系统，核心价值在于三点。第一，数据是资产的观念，采集到的历史行情存放在自己的数据库里，格式由自己定义，以后做任何分析都不会被上游限制；第二，数据链路是透明的，从抓取、清洗、落库到计算展示，每一步都可以追溯，出了问题可以直接查日志和数据表；第三，扩展性是无限逼近自己需求的，想加北向资金、龙虎榜、行业板块，只需要扩展采集模块，不需要迁就别人的产品设计。

1.2 整体架构和四个核心模块

OpenStock最简版本可以拆成四个模块，对应一条完整的数据流：

• 数据采集层：从开源数据源获取股票列表、日线行情、实时报价
• 数据存储层：把采集结果写入数据库，支持增量更新
• 业务计算层：计算技术指标、价格排名、涨跌分布等
• Web展示层：提供REST API和可交互的前端页面

我强烈建议第一次搭建时不要一上来就搞微服务、消息队列、分布式任务调度。以“能跑通全链路”为第一目标，最简形态就是单机上的Python进程加SQLite文件加一个FastAPI服务。跑通之后再在某些模块上做优化，比如把SQLite换成PostgreSQL、把定时任务拆成独立worker，这样心里有底，知道每一步优化的动机是什么。

1.3 技术选型背后的理由

OpenStack选择Python作为主语言，可以说没有任何悬念：pandas和numpy让行情数据处理非常顺手，AKShare/Tushare这类数据接口也是Python生态。Web框架我选了FastAPI而不是Flask或Django，主要看中三点：原生异步支持让WebSocket推送实时行情非常方便，自带Swagger文档调试接口很直观，以及基于Pydantic的请求校验让接口健壮性有了保障。

存储层面，日线数据用SQLite足够应付个人使用的体量。全市场5000多只股票，按每天5000行、每年约250个交易日计算，一年的日线数据也就120万行左右，SQLite完全扛得住。只有当计划存储全市场的分钟级数据时，才需要认真考虑PostgreSQL或时序数据库。前端K线图推荐Lightweight Charts，样式接近专业行情软件，滚动流畅；排行榜和统计图则用ECharts，配置灵活、图表类型丰富。

2. 环境初始化：先把项目骨架和数据库建好

2.1 项目目录结构与虚拟环境配置

一个清晰的目录结构能省掉很多后期维护的麻烦。我的OpenStock项目目录是这样的：

openstock/ ├── app/ │ ├── api/ # FastAPI路由 │ ├── core/ # 配置项、数据库连接 │ ├── models/ # ORM模型 │ ├── services/ # 业务逻辑 │ ├── collectors/ # 数据采集模块 │ └── indicators/ # 技术指标计算 ├── scripts/ # 初始化脚本、一键更新脚本 ├── data/ # SQLite文件、日志文件 ├── requirements.txt └── docker-compose.yml

用Python虚拟环境隔离依赖是做这个项目的第一件正事。Python 3.10以上版本都自带venv模块，操作很简单：

mkdir openstock && cd openstock python -m venv .venv source .venv/bin/activate pip install fastapi uvicorn sqlalchemy pandas akshare tushare apscheduler

为什么一定要虚拟环境？因为AKShare和Tushare这类数据源库更新很频繁，依赖的pandas、requests版本经常会变动。如果直接用系统Python，一段时间后很可能出现“升级了一个包，结果另一个项目跑不起来”的连锁反应。

requirements.txt建议把版本号也打上，我在实际工作中吃过亏，某次AKShare新版本调整了一个接口的参数名，结果整个采集任务在凌晨定时执行时静默失败，第二天早上看到空数据才发现。锁定版本至少能保证上线环境可控，升级时再做针对性的回归测试。

2.2 数据库表结构：日线数据与股票基础信息

数据库是OpenStock的核心资产，表设计得好不好直接决定后续分析的效率。最开始只需要两张表：股票基础信息表和每日行情表。

先看股票基础信息表的设计思路：

from sqlalchemy import Column, Integer, String, Date, Float, BigInteger, UniqueConstraint from sqlalchemy.orm import declarative_base Base = declarative_base() class StockInfo(Base): __tablename__ = "stock_info" id = Column(Integer, primary_key=True, autoincrement=True) symbol = Column(String(16), unique=True, index=True, nullable=False) name = Column(String(32), nullable=False) exchange = Column(String(8), default="") list_date = Column(Date, nullable=True)

每日行情表是数据量最大的表，字段设计上要把常用查询场景想清楚：

class StockDaily(Base): __tablename__ = "stock_daily" id = Column(Integer, primary_key=True, autoincrement=True) symbol = Column(String(16), nullable=False, index=True) trade_date = Column(Date, nullable=False, index=True) open = Column(Float) high = Column(Float) low = Column(Float) close = Column(Float) volume = Column(BigInteger) amount = Column(Float) __table_args__ = ( UniqueConstraint("symbol", "trade_date", name="uq_symbol_trade_date"), )

这里有两个细节值得展开。一是symbol统一用纯数字字符串，不带交易所前缀，只在展示层根据需要拼接成“sh600000”或“600000.SH”这类格式。二是增加symbol和trade_date的联合唯一索引，这个约束能防止采集任务重复执行时写入重复数据，配合“插入时先删后插”或“INSERT OR REPLACE”的策略，可以保证历史数据始终干净。

2.3 数据源选型：AKShare与Tushare对比

数据源是OpenStock的命脉。免费可用的方案里，AKShare和Tushare是最常用的两个选择。简单对比一下：

数据源这块我个人的建议是：初版用AKShare，因为它零门槛，拿到就能跑；但要在代码里做一个“数据源适配层”，把所有对AKShare的调用封装到collectors模块里，不在业务逻辑中散落。这样哪天想切换到Tushare，只要改适配层的实现，上层API和指标计算完全不受影响。

配置文件我用一个简单的config.py管理，方便在不同环境里调整参数：

# config.py DATABASE_URL = "sqlite:///data/openstock.db" DATA_SOURCE = "akshare" REQUEST_INTERVAL = 1.0 # 采集请求间隔，单位秒 REQUEST_TIMEOUT = 15 STOCK_LIST_CACHE_TTL = 3600

REQUEST_INTERVAL这个参数很关键。AKShare虽然免费，但免费的东西往往有隐性频率限制。请求太快容易被远端限制，建议抓日线时每只股票之间至少间隔0.5到1秒，实时行情推送则不要直接发HTTP请求，而是用WebSocket长连接接收数据，这会在后面的章节展开。

3. 核心模块实现：采集、计算与实时推送

3.1 行情采集：全量历史抓取与增量更新策略

OpenStock的采集任务分两个层次：全量历史抓取和每日增量更新。第一次跑的时候需要全量抓取，之后每天只要做增量就好。

全量抓取的第一步是获取股票列表。AKShare提供现成接口，返回的是DataFrame：

import akshare as ak import pandas as pd def fetch_stock_list(): df = ak.stock_info_a_code_name() df = df.rename(columns={"code": "symbol", "name": "name"}) return df[["symbol", "name"]]

抓日线数据时，建议用前复权方式获取，因为计算均线、MACD这类技术指标时，如果遇到分红送股而不复权，K线上会出现断崖式的跳空，指标会严重失真。

def fetch_daily_bars(symbol, start_date, end_date): df = ak.stock_zh_a_hist( symbol=symbol, period="daily", start_date=start_date, end_date=end_date, adjust="qfq", ) if df is None or df.empty: return None df["symbol"] = symbol df["trade_date"] = pd.to_datetime(df["日期"]).dt.date df = df.rename(columns={ "开盘": "open", "最高": "high", "最低": "low", "收盘": "close", "成交量": "volume", "成交额": "amount", }) return df[["symbol", "trade_date", "open", "high", "low", "close", "volume", "amount"]]

增量更新的逻辑是每次先查数据库里每只股票的最大交易日期，再从这个日期加一天开始抓。A股市场有休市日，所以不能简单地“今天减一天”，而要看数据源返回了几天数据、最新日期是不是上一个交易日，交给数据源判断就行。

这里要特别提一个复权基准的坑。前复权价格会随着每次新数据加入而变化，举个例子：一只股票在2023年6月分红除权，你在2023年7月抓到的2020年历史价格，和2024年1月抓到的同一历史日期价格，数值可能不一样。如果只是用来展示K线，问题不大；但如果用这些数据做策略回测，历史价格随时变化会导致回测结果无法复现。更稳的做法是：存储原始不复权价格，再把复权因子也存下来，计算时自行决定用哪种复权。这样数据是稳定的，公式是透明的。

3.2 技术指标计算：MA、MACD的实现细节

有了干净的日线数据，指标计算就是顺理成章的事。我习惯把指标计算做成独立的服务函数，输入DataFrame，输出带指标列的DataFrame，方便随时全量重算。

均线是最基础的指标，pandas的rolling一行就能算：

def add_ma(df, windows=(5, 10, 20, 60)): for w in windows: df[f"ma{w}"] = df["close"].rolling(window=w).mean() return df

MACD的实现稍微复杂一点，但理解了原理代码也不长。MACD由三部分组成：DIF是快线EMA12与慢线EMA26的差；DEA是DIF的9日EMA；MACD柱状图等于DIF与DEA差值的两倍：

def add_macd(df, fast=12, slow=26, signal=9): df["ema_fast"] = df["close"].ewm(span=fast, adjust=False).mean() df["ema_slow"] = df["close"].ewm(span=slow, adjust=False).mean() df["dif"] = df["ema_fast"] - df["ema_slow"] df["dea"] = df["dif"].ewm(span=signal, adjust=False).mean() df["macd"] = (df["dif"] - df["dea"]) * 2 return df

为什么不直接用现成的技术指标库？库用起来虽然省事，但遇到数据缺失和停牌空值时，库内部的处理方式未必符合你的预期。手写rolling和ewm能让你对每一步的输入输出都有掌控。等后续确实需要布林带、KDJ、RSI时，再引入talib或pandas-ta也不迟。

指标计算环节还有一个容易忽略的点：计算指标前要把数据按trade_date升序排列，并且要处理停牌缺口。A股停牌期间数据源可能直接不返回记录，直接用rolling计算会导致停牌前后被当作相邻交易日，MA和MACD的值都会偏差。严格的做法是先补全交易日历，把缺失交易日记成NaN，再计算指标，最后展示时再决定是隐藏NaN还是画断线。

3.3 实时行情推送：从轮询到WebSocket长连接

OpenStock的实时行情展示，最初我用的方案是前端每3秒调用一次后端HTTP接口，后端每3秒主动去数据源抓一次最新价格。这个方案实现最简单，但有两个问题：前端轮询压力大时后端响应变慢，而且每次都重新抓数据源会触发限流。后来改成WebSocket推送，体验好了很多。

FastAPI的WebSocket支持非常简洁：

from fastapi import FastAPI, WebSocket app = FastAPI() @app.websocket("/ws/quote") async def websocket_endpoint(websocket: WebSocket): await websocket.accept() while True: quote = await latest_quote_queue.get() await websocket.send_json(quote)

关键在于后端维护一个asyncio.Queue，后台任务定时从数据源获取整个自选股列表的最新行情，然后推送到队列里，所有已连接的WebSocket客户端都从这个队列取数据。这样的设计有几个好处：不重复请求数据源，节省接口配额；推送的内容对所有客户端一致；实现简单，代码可读性高。

有个细节要注意：WebSocket连接需要处理心跳。有些反向代理或网关会在一段时间没有数据传输后主动断开连接，所以每隔15秒左右主动发一个ping或者一个象征性的数据帧，能有效避免断线。另外，断线重连的前端逻辑也别忘了，浏览器端在onclose事件里重新调用connect，并且加一个指数退避，避免刷新风暴。

4. Web展示层：让数据变成能看的K线与榜单

4.1 API接口设计与统一返回结构

OpenStock的后端API建议设计成扁平、单一职责的风格，前端调用容易理解：

响应结构统一用这样的格式：

{ "code": 0, "message": "ok", "data": { "symbol": "600000", "name": "浦发银行", "kline": [] } }

这么做的好处是前端可以统一处理错误和加载状态，后端错误信息也能通过message字段传得比较清楚。

4.2 K线图组件选型与前端渲染

K线图是行情系统最核心的展示组件。我对比过Lightweight Charts和ECharts，两者各有优势：

所以我最后采用的是组合方案：主K线图用Lightweight Charts，页面上的成交额排行榜、涨跌分布直方图用ECharts。两者的CDN引入方式都很简单，用ESModule的方式引入：

import { createChart } from 'lightweight-charts'; const chart = createChart(document.getElementById('kline'), { width: 800, height: 450, }); const series = chart.addCandlestickSeries(); series.setData(klineData);

前端页面不需要做得很复杂，第一版能展示行情列表、K线详情、排行榜三块就够了。更多精力建议放在数据质量和后端稳定性上，这也是我踩过坑之后最大的体会——页面好看救不了底层数据混乱。

4.3 定时任务与数据刷新时间窗

行情数据更新有自己的节奏：盘中要高频，盘后要完整。OpenStock用APScheduler来管理定时任务。

from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger scheduler = BackgroundScheduler() def start_scheduler(): scheduler.add_job( run_daily_update, CronTrigger(timezone="Asia/Shanghai", day_of_week="mon-fri", hour=16, minute=5), id="daily_update", max_instances=1, replace_existing=True, ) scheduler.start()

收盘后为什么要等到16:05再跑？A股15:00收盘，行情数据源不会在收盘那一刻马上把所有数据整理完毕，通常需要半小时左右。如果设置15:05跑，大概率抓不到完整数据，或者抓到的是未完成结算的临时数据。多等一分钟换个数据完整度，非常划算。

盘中如果需要实时刷新，可以再加一个小时级的任务：每个交易日10:30、13:30各更新一次当日分钟线，并入当日汇总。但这部分会让数据量快速增长，建议在OpenStock运行稳定后再加。

5. OpenStock实测踩坑：这些坑我替你先踩了

5.1 数据源接口频繁变动：重试与兜底缺一不可

AKShare接口变动频繁这件事，用过的朋友都知道。接口名、参数、返回字段都可能突然变更，而且常常是周六周日改完，周一开盘才发现。所以我写了统一的调用包装函数，所有对数据源的访问都通过这个函数走：

import time import logging logger = logging.getLogger("openstock.collector") def safe_call(func, *args, **kwargs): for attempt in range(4): try: return func(*args, **kwargs) except Exception as e: logger.warning("第%s次调用失败: %s", attempt + 1, e) time.sleep(2 + attempt * 3) raise RuntimeError(f"数据源调用最终失败: {func.__name__}")

这里用了线性退避而不是固定等待。第一次失败等2秒，第二次5秒，第三次8秒，给对方服务一点恢复时间，也避免自己撞上限流。还有一点很重要：每次采集任务结束后，把成功和失败的股票数写入日志。如果某次任务失败率异常高，说明上游接口可能变了，这时候最好触发一个告警通知，而不是任由任务“静默失败”。

5.2 股票代码格式与时区错乱

A股有很多种代码格式：AKShare返回的上海股票是"600000"，纯数字不带前缀；有些数据源返回的是"sh600000"，带交易所前缀；还有的用"600000.SH"。如果OpenStock采集模块存了一种格式，后来接入另一个数据源又存了另一种格式，前端关联时就会出问题。

我的建议是统一规定：数据库里只存纯数字字符串，代码里用exchange字段区分交易所，展示层再拼接前缀。

时区问题则是另一个容易被忽视的坑。如果服务器时区设置为UTC，Python的datetime.now()取出来的是UTC时间，在和交易日的日期做比较时可能差8小时，导致增量更新的日期范围判断出错，某些交易日的行情被漏掉。OpenStock全项目统一用北京时间：

from datetime import datetime import pytz CN_TZ = pytz.timezone("Asia/Shanghai") def now_cn() -> datetime: return datetime.now(CN_TZ)

所有涉及“今天”“最近交易日”的判断，都走now_cn()，数据库里存储的trade_date用date类型而不是datetime，天然避免时区混淆。

5.3 停牌缺失与分钟级数据膨胀

A股股票停牌期间，日线接口通常不会返回记录。这带来两个问题：一是K线图上会断开，二是计算指标时停牌前后被连成相邻数据。前者是展示问题，前端可以用前向填充的方式补全显示；后者是数据质量问题，更严谨的做法是维护一份交易日历表，用交易日做左连接，缺失的日子填NaN，再算指标。

分钟级数据要特别注意膨胀问题。全市场5000多只股票，每分钟一条数据，一天交易240分钟，理论上每天会生成约120万行记录，一年接近3亿行。这还没算上高频的快照数据，SQLite根本处理不了。我的方案是分层次保留：最近5个交易日的分钟线完整保留，用于短线复盘；更早的分钟线只保留聚合后的15分钟、60分钟K线，再往前的自动归档或删除。这个策略可以在“数据可用性”和“存储成本”之间找到平衡点，动手采集分钟数据之前一定要先把这部分设计好，不然跑一个月就会面临数据库膨胀问题。

6. 进阶：从本地跑通到长期稳定运行

6.1 Docker Compose一键部署

OpenStock本地跑通之后，下一步是部署到一台常年开机的服务器上，让定时任务稳定运行。Docker Compose是比较轻量的部署方案，一个文件搞定服务编排：

version: "3.8" services: db: image: postgres:15-alpine environment: POSTGRES_USER: openstock POSTGRES_PASSWORD: openstock POSTGRES_DB: openstock volumes: - pgdata:/var/lib/postgresql/data healthcheck: test: ["CMD-SHELL", "pg_isready -U openstock"] interval: 10s timeout: 5s retries: 5 web: build: . ports: - "8000:8000" depends_on: db: condition: service_healthy environment: DATABASE_URL: postgresql://openstock:openstock@db:5432/openstock volumes: pgdata:

容器化部署的好处是环境依赖全部固化，换机器部署时只要装了Docker，基本就是一条docker compose up -d的事。从SQLite迁移到PostgreSQL时，SQLAlchemy层几乎不用改代码，只改一个DATABASE_URL即可。要注意的是pandas的read_sql和DataFrame的to_sql在两者之间有一些日期和浮点精度的差异，迁移后要跑一遍完整的增量任务做验证。

6.2 日志、健康检查与失败告警

长期稳定运行的关键不是“写更多代码”，而是“能及时发现问题”。OpenStock里我加了三层保障：

第一层是结构化日志。用loguru替换标准logging，输出格式带上时间、模块、任务ID，排查问题的时候能快速定位是哪只股票、哪次任务、哪个环节出了问题。

第二层是健康检查接口：

@app.get("/healthz") def healthz(): db_ok = check_database() last_update = get_last_update_time() return { "status": "ok" if db_ok else "error", "last_update": last_update, "stale_minutes": minutes_since(last_update), }

这个接口不仅能被Docker的healthcheck用来判断容器状态，还能接外部监控平台做周期性探测。只要数据超过某个阈值没有更新，就说明定时任务可能挂了。

第三层是失败告警。采集任务失败后，除了记日志，还可以通过简单的方式发一个通知。在Python里调用Webhook发送告警非常简单，几十行代码就能接上钉钉、企微或者普通邮件。这样即使凌晨定时任务失败，第二天早上也能在手机上看到消息，而不是等打开页面才发现数据是空的。

6.3 后续扩展思路：回测、因子与多市场

OpenStock的基础版本稳定运行之后，可以沿着几个方向继续扩展。

回测方向是最自然的一步。日线数据和技术指标已经齐全，接上backtrader或vectorbt这类回测框架，就能验证交易策略。数据格式要做一层适配，把OpenStock的字段映射成回测框架的OHLCV格式，基本就能跑起来。

因子选股方向也很有意思。在基础行情数据之上，增加财务数据、行业板块、北向持仓等字段，就可以计算估值因子、动量因子、质量因子，每天对全市场股票打分排序。这个方向的难点不在计算，而在于数据源的质量和历史覆盖度。

多市场覆盖方面，港股和美股在AKShare里也有对应的免费接口。多市场数据的关键问题是交易日历不同、汇率换算、代码格式差异更大。如果打算覆盖多市场，建议在数据模型上提前加一个market字段，隔离各市场的配置和数据表，不要混在同一个表里。

写在最后

我自己在实际搭建OpenStock的过程中，最大的体会是：真正有价值的不是跑通的那一刻，而是后续不断修修补补、让系统稳定运行的过程。数据源接口会变、服务器会宕机、数据库会长大，所有这些都是在真实项目中才会遇到的问题，也是最好的学习机会。最后再分享一个小技巧：把数据采集和指标计算彻底解耦，采集只负责把原始数据写进库，指标计算随时可以从库里重新读取重算。这样即使你改了指标公式，也不需要回补历史行情数据，一条命令就能重算所有结果。这套架构风格同样适用于很多数据类项目，希望这篇记录能帮你少踩几个坑。