简介：InStock股票系统是专为量化投资者设计的自动化行情工具，聚焦A股与ETF的每日数据抓取，覆盖个股行情、资金流向、基本面指标、龙虎榜、大宗交易及行业概念板块，可有效解决投研中数据获取分散、更新不及时的痛点，适合个人量化研究者与数据爱好者部署使用。资源包共178个文件、约4.01MB，以82个Python脚本作为采集与自动化核心，配合前端展示用的HTML/JS/CSS、定时运行脚本、配置与说明文档，构成一套较完整的本地数据平台。当前已有148人学习。通过部署该系统，可快速建立个人A股数据仓库，跟踪资金动向与机构交易痕迹，理解板块轮动特征；目录结构清晰、模块划分明确，也便于二次开发接入自己的选股策略，是入门A股量化数据处理的实用参考。

1. 把“找数据”这件事自动化：InStock 到底解决什么问题

每天收盘后，我见过太多个人投资者把大量时间花在“找数据”上：打开行情软件看个股涨跌，再去翻资金流向，最后到晚上才想起来龙虎榜和大宗交易还没看，等资料凑齐，已经过九点，根本没精力想第二天的操作。InStock股票系统这个标题里最关键的词不是“量化”，而是“自动化”——它把每日抓取A股市场股票与ETF的关键数据做成了流水线：个股行情、资金流向、基本面指标、龙虎榜、大宗交易、行业与概念板块，全部定时抓取、落库、清洗，再在统一的数据表上跑信号。它适合两类人：一类是有 Python 基础、不想花几万块买终端、想自己搭数据管道的开发者；另一类是已经厌倦手工整理数据、希望用规则生成候选池的投资者。它不预测涨跌，只负责把“数据获取”这个最脏最累的环节固定下来。

2. 数据抓取层怎么搭：日频行情、资金流与龙虎榜的采集架构

这一章先讲数据源选型，再给爬虫最小实现，最后落到存储。搭这套抓取层时，我最关心的三件事：数据源稳不稳定、抓取代码断网后能不能自我恢复、存储结构在数据量涨到几百万行之后还能不能查得快。后面所有信号计算都建立在这一层上，所以这一层不能图省事。

2.1 数据源选型：为什么优先走免费公开接口，而不是买终端

个人做量化，最现实的门槛是数据成本。Wind 终端一年几万，Choice 也要账号费，而且桌面终端的数据导出接口基本是为人工操作设计的，长时间无人值守抓取很容易被限制。InStock 这类项目把数据源放在财经门户的公开 HTTP 接口上，是合理选择：零费用、响应快、字段覆盖日线行情与资金流，足够支撑日频量化。

缺点是接口文档不公开、字段名随时可能变、对访问频率有隐性限制。所以我搭数据源时通常做两手准备：主源抓行情和资金流，备源只抓三个字段——close、volume、pct_change。万一主源改版，备源还能保证当日数据不至于断档。不要一上来就接入多家付费数据商，先把一条免费链路跑通，再在它上面做增量补偿。

备源的价值在自动化任务里会被放大：主源凌晨 1 点维护，备用源可能在凌晨 2 点还能正常返回。数据管道里“最后一条可用路径”远比“字段最全的路径”重要。

2.2 采集器最小实现：用 requests.Session 保证每天稳定拉数据

抓取层最常见的翻车点不是接口参数错了，而是网络抖动和频率限制。为此我给采集函数套一个带退避的重试装饰器，把“请求失败→等待→重试”的逻辑抽出来，避免每个业务函数重复写 try except。

import requests import time import random from functools import wraps UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36" def retry_on_failure(max_retries=3, base_delay=2): """给抓取函数加上指数退避重试。""" def decorator(func): @wraps(func) def wrapper(*args, **kwargs): for attempt in range(1, max_retries + 1): try: return func(*args, **kwargs) except (requests.Timeout, requests.ConnectionError, ValueError): if attempt == max_retries: raise sleep_sec = base_delay * attempt + random.random() time.sleep(sleep_sec) return None return wrapper return decorator @retry_on_failure(max_retries=3, base_delay=2) def fetch_json(session, url, params=None): resp = session.get(url, params=params, timeout=10) resp.raise_for_status() return resp.json()

这段代码里有两个关键设置。第一个是timeout=10，超过 10 秒直接判定失败，避免某个接口长时间卡住拖垮整批任务。第二个是退避公式base_delay * attempt + random.random()，第一次失败等 2 秒，第二次等 4 秒，第三次等 6 秒左右，随机数防止多只标的的重试请求在同一个时间点砸向服务器。

调用时用同一个 Session 对象，连接复用能明显降低被风控盯上的概率：

def daily_job(all_codes): session = requests.Session() # 带 Referer 是为了模拟从页面发起的正常请求，降低被识别为脚本的概率。 session.headers.update({"User-Agent": UA, "Referer": "https://quote.example.com/"}) for code in all_codes: params = {"code": code, "fields": "open,high,low,close,volume,amount"} data = fetch_json(session, "https://quote.example.com/api/stock", params=params) # 每只股票间隔 0.2~0.5 秒，全市场 5000 只标的约半小时跑完。 time.sleep(random.uniform(0.2, 0.5))

间隔的设定逻辑是：如果按 5 个请求每秒跑，5000 只标的约 17 分钟跑完全市场；加上 0.2~0.5 秒随机间隔，时间翻倍但在可接受范围。这个节奏对大多数免费接口来说算安全。如果目标只有沪深 300 成分股，间隔可以放宽到 0.5 秒以上，没必要抢时间。

2.3 存储与更新策略：先用 CSV 落地，再平滑迁移到 MySQL

数据抓下来之后，存储结构直接影响后续开发效率。A 股加 ETF 大约 5500 个标的，日频数据一年约 140 万行，五年约 700 万行。这个量级用 CSV 完全能撑住，一开始不用急着上数据库。

from pathlib import Path import pandas as pd base_dir = Path("./data") date_dir = base_dir / "2025-01-15" date_dir.mkdir(parents=True, exist_ok=True) # quote_df 是当天清洗后的行情表，按日期分目录，回看某一天直接读对应文件。 quote_df.to_csv(date_dir / "daily_quote.csv", index=False)

按交易日分目录的好处是：增量更新天然是“只写当天文件”，不会误碰历史数据；排查问题时也能直接看某个日期目录是否存在。缺点是跨日期查询需要拼文件，所以在数据量上来之后，建议迁移到 MySQL。

CREATE TABLE IF NOT EXISTS daily_quote ( trade_date DATE NOT NULL, code VARCHAR(10) NOT NULL, name VARCHAR(20) DEFAULT NULL, open DECIMAL(10,2) DEFAULT NULL, high DECIMAL(10,2) DEFAULT NULL, low DECIMAL(10,2) DEFAULT NULL, close DECIMAL(10,2) DEFAULT NULL, volume BIGINT DEFAULT NULL, amount DECIMAL(20,2) DEFAULT NULL, pct_change DECIMAL(6,2) DEFAULT NULL, PRIMARY KEY (trade_date, code), KEY idx_code_trade (code, trade_date) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

这张表的设计有一个容易被忽略的点：主键用(trade_date, code)而不是自增 id。复合主键保证同一个交易日同一只股票只会存在一行，重复抓取时执行INSERT ... ON DUPLICATE KEY UPDATE就能幂等更新，不会产生脏数据。索引(code, trade_date)则是为“取某只股票最近 N 天行情”这类高频查询设计的。

从 CSV 迁到 MySQL 时，我一般写一个同步脚本，按日期目录逐批读入，用LOAD DATA LOCAL INFILE导入，导入后立刻跑一行 COUNT 对比两边行数。不要把整个历史数据一次性塞进去，一旦中间断掉很难定位断点。

3. 核心数据模块拆解：从原始接口到可用的量化底表

数据抓到本地只是第一步。接口返回的原始字段往往不是我们想要的样子：代码格式不统一、资金流是嵌套结构、龙虎榜按榜单而不是按单只股票组织。这一章要把这些原始数据归一成能直接参与计算的底表。底表设计得好不好，决定后面写信号时是几行代码还是一大堆 if else。

3.1 股票与 ETF 的统一代码模型：先解决撞码再去谈数据

上证和深证都存在 6 位数字代码，不做处理会撞车。比如 5 开头的沪市 ETF 和 1 开头的深市基金，如果只存 6 位数字，分不清归属交易所；600 和 000 开头的代码倒是能从首位数区分，但依赖人眼判断不能写进程序。抓回来的数据源有的给.SH后缀，有的只给纯数字，第一步必须做标准化。

def normalize_code(raw) -> str: """把不同来源的证券代码统一为 6 位代码.交易所 格式。""" code = str(raw).strip().upper() if "." in code: return code if len(code) != 6 or not code.isdigit(): raise ValueError(f"无法识别代码: {raw}") if code.startswith("6"): return f"{code}.SH" # 沪市主板、科创板均以 6 开头 if code.startswith(("0", "3")): return f"{code}.SZ" # 深市主板 0 开头，创业板 3 开头 if code.startswith("5"): return f"{code}.SH" # 沪市 ETF/LOF if code.startswith("1"): return f"{code}.SZ" # 深市基金 if code.startswith(("4", "8")): return f"{code}.BJ" # 北交所 raise ValueError(f"未知代码段: {raw}")

这段代码的判断顺序不能乱：先处理带后缀的字符串，再判断纯数字。startswith(("0", "3"))用元组参数一次完成两个前缀匹配，比连续两个 if 干净。注意北交所在旧数据源里可能是 8 开头，纳入后要单独判断，不要让它落到默认分支。

统一代码后，所有下游代码都按600000.SH这种格式关联，避免在 join 时出现“一边是 600000，另一边是 600000.SH”的尴尬。

3.2 资金流向与龙虎榜的清洗：把嵌套 JSON 拍平成行

资金流接口返回的往往是树状结构：外层是按日期组织的数据，内层又分主力、超大单、大单、中单、小单多个子对象。如果不拍平，pandas 读进来会得到一堆 dict 嵌套列，后续筛选没法写。

def flatten_flow(raw_item): """把资金流接口返回的嵌套结构转成一行底表记录。""" main = raw_item.get("main", {}) or {} super_ = raw_item.get("super", {}) or {} row = { "code": normalize_code(raw_item["code"]), "trade_date": raw_item["date"], "main_net_inflow": main.get("net_inflow"), "main_net_inflow_pct": main.get("net_inflow_pct"), "super_net_inflow": super_.get("net_inflow"), "huge_net_inflow": raw_item.get("huge", {}).get("net_inflow"), "retail_net_inflow": raw_item.get("retail", {}).get("net_inflow"), } return row

这里使用了dict.get而不是dict["net_inflow"]访问，因为资金流数据经常出现某个子结构缺失——新股上市第一天没有历史资金流，部分北交所标的主力字段一直为空。get方法在字段缺失时返回 None，不会中断整批清洗。

龙虎榜是另一个典型清洗场景。交易所每天收盘后才公布当日上榜名单，返回的 JSON 往往是一个榜单包含多个股票列表。要把它拆成“日期 + 股票 + 买入卖出金额”的明细行：

def clean_lhb(raw_rows): """把龙虎榜按榜单组织的原始数据拆成按股票维度组织的明细。""" records = [] for r in raw_rows: for stock in r.get("stock_list", []): records.append({ "trade_date": r["trade_date"], "code": normalize_code(stock["code"]), "buy_amount": stock.get("buy_amount"), "sell_amount": stock.get("sell_amount"), "net_buy": stock.get("net_buy"), "reason": stock.get("reason"), }) return pd.DataFrame(records)

注意reason字段：龙虎榜上榜原因（比如“日涨幅偏离值达 7%”）是后续做事件驱动信号的重要标签，但同一只股票可能同时因为多个原因上榜，导致出现在同一个榜单的多行里。清洗阶段不要做去重，保留原始一对多关系，信号层再按需求聚合。

3.3 基本面指标与行业板块映射：筛选器的原料从哪里来

行情和资金流是日频数据，基本面指标却不一样：市盈率、市净率、ROE 大多按季度更新，不需要每天抓。把这个认识写进存储模型，能省一大半抓取量。我一般单独建两张表：一张存股票基本信息与最新基本面快照，一张存概念板块映射。

CREATE TABLE stock_basic ( code VARCHAR(10) PRIMARY KEY, name VARCHAR(20) NOT NULL, industry VARCHAR(30), pe DECIMAL(10,2), pb DECIMAL(10,2), roe DECIMAL(6,2), update_date DATE ); CREATE TABLE stock_concept ( code VARCHAR(10) NOT NULL, concept VARCHAR(30) NOT NULL, PRIMARY KEY (code, concept) );

行业字段在stock_basic里，概念映射单独成表，是因为两者语义不同：行业是互斥的，一只股票属于且只属于一个行业分类；概念是叠加的，一只股票可以同时属于“人工智能”“算力”“信创”多个概念。把它们放同一张表会导致更新逻辑复杂，还会在信号层产生重复计数。基本面快照表则只需要保留最近一版，每天抓取后REPLACE INTO覆盖即可，没必要保留历史版本，除非你要做基本面变化分析。

4. 用清洗后的数据生成交易信号：动量、资金流与龙虎榜的交叉验证

底表建好之后，写信号就是顺理成章的事。我常用的一个组合是“20 日动量 + 主力净流入为正 + 龙虎榜净买入为正”。这三个条件分别代表趋势、资金和事件三重确认，单独用哪一个都不稳定：动量追高风险大，资金流有滞后，龙虎榜覆盖范围太小。

4.1 信号计算的核心逻辑：20 日动量叠加主力净流入

动量因子用 pandas 的 groupby 加 rolling 实现，注意要先排序再计算，否则前复权数据错位会让结果完全不可信。

import pandas as pd def add_momentum(df: pd.DataFrame, window: int = 20) -> pd.DataFrame: """在日线数据上追加 20 日动量字段。""" df = df.sort_values(["code", "trade_date"]).reset_index(drop=True) df["ret"] = df.groupby("code")["close"].pct_change() df["momentum"] = df.groupby("code")["ret"].rolling(window).apply( lambda x: (1 + x).prod() - 1, raw=True ).reset_index(level=0, drop=True) return df.drop(columns=["ret"])

这段代码里最容易写错的是reset_index(level=0, drop=True)。groupby 后接 rolling，结果索引是“code + 原始行号”两层结构，不加这一步，新字段的长度和原 DataFrame 会对不上。rolling(window)计算的是最近 20 个交易日的收益率乘积，也就是区间累计涨幅，而不是简单平均，这个区别在动量语境里很关键。

参数window我习惯用 20，对应约一个自然月。A 股短周期轮动快，5 日和 10 日动量噪音太大；60 日动量又太钝，等信号出来趋势差不多结束了。如果做的是 ETF 轮动，窗口可以放宽到 60，因为 ETF 的成分股分散，短期动量不如个股敏感。

4.2 每日自动输出候选池：从数据到可操作清单的脚本

动量和资金流算好后，把它们跟龙虎榜合并，输出候选池。这一步不需要复杂模型，pandas 的 merge 加布尔筛选就够。把“主力净流入为正”和“龙虎榜净买入为正”同时作为条件，过滤后的股票数量通常会从 5000 多只降到 30 到 100 只，正好进入人工复核流程。

def select_pool(quote_df, flow_df, lhb_df, date): """把当日动量、资金流和龙虎榜数据合到一起生成候选池。""" merged = quote_df.merge( flow_df[["code", "main_net_inflow_pct"]], on="code", how="left" ) merged["on_lhb"] = merged["code"].isin( lhb_df.loc[lhb_df["trade_date"] == date, "code"] ) pool = merged[ (merged["momentum"] > 0.08) & (merged["main_net_inflow_pct"] > 0) & (merged["on_lhb"]) ] return pool.sort_values("momentum", ascending=False)

how="left"表示以行情表为主表，资金流缺失的股票保留下来但填充 NaN，这样不会因为某一天资金流接口漏了一条数据就把整只股票从底表里剔除。on_lhb用 isin 判断，比 inner join 更直观。momentum > 0.08这个阈值不是拍脑袋定的，是回测出来的：过去三年，满足这个阈值的股票次日常规收益分布最稳定。你可以先按 0.05 跑一个月，观察数据分布再调，不要一开始就追求精确阈值。

候选池生成后，我会再用基本面表过滤一次，把 PE 为负、ST 状态的股票剔除。基本面表是季度更新，过滤条件不宜设太紧，否则候选池经常整个月都空着。

5. InStock 落地避坑：抓取限制、复权错位与数据不一致的五条记录

这一章写的每一条都是我自己踩过的坑。数据管道跑起来不难，难的是它连续跑三个月不出乱子。以下五条按频率排序，基本覆盖了日频量化工具最常见的故障模式。

5.1 抓取频率太高被封 IP：不要在收盘一小时内并发拉全市场

现象：系统运行到第三到五天，某天早上开始出现大量 403 响应；再往后，同一 IP 下任何接口都返回验证码页面。由于采集脚本只记录成功数据，那天的行情表缺了 200 多只股票，直到两天后做完整性校验才发现。

原因：收盘后所有人的定时任务都集中在 15:30 到 16:30 启动，我的脚本在这个时段用 16 个线程并发拉全市场，单秒请求数超过数据源风控阈值，触发封禁。

解决：把并发改成单线程循环，每只股票间隔 0.2 秒到 0.5 秒，并把任务执行时间从 15:30 改到 17:00 以后，避开高峰。封禁解除后，在抓取函数里加了失败告警：连续 5 次 403 就发邮件，而不是默默跳过。

另外，免费接口的风控通常集中在“单 IP 请求频率”和“请求特征”两个维度。UA 和 Referer 是常数，容易被识别成脚本。我的做法是准备两个用户代理轮流使用，同时在凌晨跑一次小批量验证任务，看 IP 是否被标记。

5.2 前复权和后复权混用：同一个 close 字段引发的信号失真

现象：某只股票在回测里走势跟行情软件对不上，尤其是 2020 年以前的数据，close 序列的涨跌幅明显偏大，导致动量信号在除权日附近频繁误报。

原因：免费接口的复权数据有两种口径，前复权随最新价变动，历史数据会不断重算；后复权相对稳定，适合回测。但同一个接口在不同时间段抓下来的前复权数据基准不同，日期越早偏差越大。我的存储表里只有close一个字段，没记录复权类型，清洗时把两种口径混在一起了。

解决：确定一个主口径，我在日频信号里统一用后复权价计算动量，用原始不复权价计算涨跌幅和成交量。这两类数据分开存储，分别命名close_adj和close_raw。如果只想存一个字段，就选后复权，并定期全量重算一次，保证整个历史区间口径一致。反过来说，如果要计算当日涨跌幅，必须用不复权价，否则除权日当天会算出“跳空下跌”或“跳空上涨”的假信号。

5.3 龙虎榜日期和收盘日期错位：夜盘数据落到哪一天很重要

现象：每天早上看候选池，发现不少股票的“龙虎榜净买入”跟当天涨幅对不上。比如 4 月 10 日早上生成候选池时，标记某股票 4 月 9 日上了龙虎榜，但这只股票 4 月 9 日的行情实际还没收盘。

原因：龙虎榜按交易所规则是 T 日收盘后公布 T 日上榜名单，但部分数据提供方在晚上 10 点前返回的数据用的是 T-1 日榜单；也有少数接口返回的 JSON 里日期字段是“公布日期”而不是“交易日期”。全链路没统一基准，信号层拿到的是两个不同日期的数据。

解决：在清洗层强制统一：龙虎榜数据一律以trade_date字段作为归属日期，落库后写一个校验任务，比对龙虎榜最大日期与行情表最大日期，相差超过 1 天直接告警。信号层关联时，永远用trade_date关联，不允许用“导入时间”或“抓取时间”。

5.4 增量更新覆盖历史数据：缺了主键约束就会静默翻车

现象：某天检查数据库，发现总行数从 500 万跌到 300 万，最初以为是存储问题，后来发现是更新逻辑写错了。

原因：早期用“先删后插”策略做增量更新，删除条件只写了code = ?，没有限定trade_date = ?。某只股票某天数据重跑时，把该股票的全部历史记录删掉了，再插入只有当天一行。

解决：两条硬措施。第一条是表结构加复合主键(trade_date, code)，让重复插入直接报错或走ON DUPLICATE KEY UPDATE，从机制上堵住误删。第二条是废弃“先删后插”，改成按日期分片写入，只操作trade_date = 目标日期的记录。执行完立刻对比当天行数与源文件行数，不一致就报警。

如果表已经建好且没有主键，可以先把重复数据清掉再补主键；有主键约束后，很多误操作在第一步就会被数据库拦住，而不是等几天后数据对不上才发现。

5.5 数据库连接池耗尽：定时任务为什么会半夜失联

现象：连续几天凌晨的定时任务都失败，但白天手动跑同样的脚本完全正常。日志里出现Too many connections，MySQL 拒绝新连接。

原因：采集脚本里每次调用数据库都pymysql.connect新建连接，用完没关。白天任务少，连接数没到上限；凌晨任务一启动，同时并发几十个连接，加上之前的连接没释放，直接打满 MySQL 默认的 151 连接数。更隐蔽的是，程序异常退出时finally里没有close()，连接被占用直到超时。

解决：用连接池管理数据库会话，把连接数上限控制在 20 以内。pymysql本身没有连接池实现，我用DBUtils.PooledDB包了一层，设置maxconnections=20，每次从池里借连接，用完归还。同时给所有数据库操作套上了try/finally，确保即使查询报错也能归还连接。

from dbutils.pooled_db import PooledDB import pymysql pool = PooledDB( creator=pymysql, maxconnections=20, host="127.0.0.1", user="quant", password="your_password", database="stock", charset="utf8mb4", ) def query_one(sql, args=None): conn = pool.connection() try: with conn.cursor() as cur: cur.execute(sql, args) return cur.fetchone() finally: conn.close() # 归还连接，而不是真的关闭

这段代码里容易混淆的是conn.close()：在PooledDB语义下，close()不是关闭底层连接，而是把连接还给池子。如果用原生pymysql写同样的代码，close()就是真的断开连接。所以连接池的封装必须统一入口，不要让部分代码走原生连接、部分走池子，两套混用会把连接数逻辑彻底搞乱。

6. 进阶验证技巧：用一致性校验和回放测试让信号靠谱一点

6.1 每日抓取后先做三行校验

自动化跑一段时间后，比抓取更重要的反而是验证。我每天在落库后跑一段简单的覆盖度检查，用“预期标的集合”对比“实际抓到的标的集合”，缺了就告警。

def check_coverage(expected, actual, date): miss = expected - actual extra = actual - expected if miss: print(f"[{date}] 缺失 {len(miss)} 个标的，前 5 个: {sorted(miss)[:5]}") if extra: print(f"[{date}] 多出 {len(extra)} 个标的，可能代码映射有误") return not miss and not extra

预期标的集合来自上一交易日收盘时的全市场股票快照加当日新上市列表。如果缺失数量突然从个位数变成几十个，大概率不是网络原因，而是代码规则变动。这个校验跑完，我才会信任当天生成的候选池。

6.2 用回放确认信号不是过拟合

参数调多了难免过拟合，我的自我检查方式是把当前参数放回六个月的历史数据里重跑一遍，看信号池的平均表现是否稳定。如果 20 日动量阈值设为 8% 在近三个月有效，但五个月前同一批股票平均收益为负，说明当前参数大概率在拟合近期行情。

回放不追求精确收益，只看分布。我只需要知道：满足条件的股票数量是否稳定、次日平均涨跌幅是否为正、最大回撤是否超出预期。任何一天满足条件的股票少于 5 只，我都会把阈值调低再观察，而不是手动往池子里塞股票。信号的稳定性来源于规则对噪声的不敏感，而不是规则足够巧妙。

我习惯每天把覆盖率数字直接打在任务日志第一行，出现任何异常第一眼就能看到。数据管道这种系统，不怕慢，就怕坏得无声无息；把校验做成强制步骤之后，我的信号质量明显稳了一截。希望这套从抓取到校验的工程思路对你有帮助。

本文还有配套的精品资源，点击获取