量化数据开发实战系列（第 7 篇）：涨跌停股池深度实战：涨停、跌停数据采集、入库、指标衍生计算

量化数据开发实战系列（第 7 篇）：涨跌停股池深度实战：涨停、跌停数据采集、入库、指标衍生计算

前言

前面章节已经完成涨停股池采集、日志重试、定时调度、本地交易日历。本篇扩展接入跌停股池接口，同时完成涨停、跌停两套股池的采集、清洗、持久化存储。

接口仅返回单只标的原始明细，炸板率、封板成功率这类市场情绪聚合指标不会直接提供，需要读取明细数据通过代码完成聚合运算。本篇将实现双股池流水线，并且计算每日市场情绪统计值，存入汇总表，方便后续时序分析与可视化。

一、接口原始字段梳理

涨停股池核心字段

表格

跌停股池核心字段

表格

二、自研衍生统计指标

1. 涨停家数：当日涨停股池记录总条数
2. 跌停家数：当日跌停股池记录总条数
3. 炸板数量：涨停池中炸板次数zbc>0标的数量
4. 炸板率 (%)= 炸板数量 / (涨停家数 + 炸板数量) * 100
5. 封板成功率 (%)= 涨停家数 / (涨停家数 + 炸板数量) * 100

全部聚合指标需要业务代码计算，接口不直接返回统计结果。

三、数据表设计

在quant.db新增两张数据表：

1. dt_pool：跌停股池明细表
2. market_emotion_daily：每日市场情绪汇总统计表

dt_pool 跌停股池表

表格

market_emotion_daily 每日情绪汇总表

表格

四、完整可运行代码

代码调用复用前面章节请求、日志、交易日历逻辑；新增跌停股池全流程处理、情绪统计计算，改造每日采集任务。

import requests import logging import time import pandas as pd import numpy as np import sqlite3 from apscheduler.schedulers.background import BackgroundScheduler # ========== 全局配置 ========== LICENCE = "你的licence" DB_PATH = "quant.db" LOG_FILE = "quant_collect.log" # ----------日志初始化---------- logging.basicConfig( filename=LOG_FILE, level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", datefmt="%Y‑%m‑%d %H:%M:%S", filemode="a" ) logger = logging.getLogger(__name__) # ----------带重试HTTP请求---------- def biying_api_get_retry(full_url, timeout=15, max_retry=3): for attempt in range(1, max_retry + 1): try: resp = requests.get(full_url, timeout=timeout) if resp.status_code == 200: return resp.json() logger.warning(f"HTTP状态码异常：{resp.status_code}，第{attempt}次重试") except Exception as e: logger.warning(f"网络请求异常，第{attempt}次重试，错误信息：{str(e)}") time.sleep(2) logger.error("达到最大重试次数，接口请求失败") return [] # ----------交易日历函数（复用第6篇）---------- def is_trade_day(dt_str: str, db_name="quant.db") -> bool | None: conn = sqlite3.connect(db_name) sql = "SELECT is_trade FROM trade_calendar WHERE dt = ?" df = pd.read_sql(sql, conn, params=(dt_str,)) conn.close() if len(df) == 0: logger.warning(f"交易日历中没有该日期记录:{dt_str}") return None return bool(df.iloc[0]["is_trade"]) def get_trade_day_list(start_dt, end_dt, db_name="quant.db"): conn = sqlite3.connect(db_name) sql = """ SELECT dt FROM trade_calendar WHERE dt >= ? AND dt <= ? AND is_trade =1 ORDER BY dt ASC """ df = pd.read_sql(sql, conn, params=(start_dt, end_dt)) conn.close() return df["dt"].tolist() # =====================本篇新增：跌停池、情绪统计===================== def init_new_tables(db_name="quant.db"): conn = sqlite3.connect(db_name) cur = conn.cursor() # 跌停股池 sql_dt = """ CREATE TABLE IF NOT EXISTS dt_pool ( id INTEGER PRIMARY KEY AUTOINCREMENT, trade_date TEXT, stock_code TEXT, stock_name TEXT, price REAL, zf REAL, cje_yi REAL, ltsz_yi REAL, zsz_yi REAL, pe REAL, hs REAL, lbc INTEGER, lbt TEXT, zj_yi REAL, fba_yi REAL, zbc INTEGER, UNIQUE(trade_date, stock_code) ) """ cur.execute(sql_dt) # 每日市场情绪汇总 sql_emotion = """ CREATE TABLE IF NOT EXISTS market_emotion_daily ( id INTEGER PRIMARY KEY AUTOINCREMENT, trade_date TEXT, up_count INTEGER, down_count INTEGER, bomb_count INTEGER, bomb_rate REAL, success_rate REAL, UNIQUE(trade_date) ) """ cur.execute(sql_emotion) conn.commit() conn.close() logger.info("跌停池、市场情绪汇总表初始化完成") def fetch_raw_dt_pool(trade_date): url = f"http://api.biyingapi.com/hslt/dtgc/{trade_date}/{LICENCE}" return biying_api_get_retry(url) def clean_dt_data(raw_json_list, trade_date): df = pd.DataFrame(raw_json_list) keep_cols = ["dm","mc","p","zf","cje","lt","zsz","pe","hs","lbc","lbt","zj","fba","zbc"] df = df[keep_cols].copy() df.columns = ["股票代码","股票名称","价格","涨跌幅","成交额","流通市值","总市值","动态市盈率","换手率","连续跌停数","最后封板时间","封单资金","板上成交额","开板次数"] df = df.replace([None,"null",""], np.nan) num_cols = ["价格","涨跌幅","成交额","流通市值","总市值","动态市盈率","换手率","连续跌停数","开板次数","封单资金","板上成交额"] for col in num_cols: df[col] = pd.to_numeric(df[col], errors="coerce") df = df.dropna(subset=["股票代码","股票名称"]) df["交易日期"] = trade_date df["成交额_亿"] = df["成交额"] / 1e8 df["流通市值_亿"] = df["流通市值"] / 1e8 df["总市值_亿"] = df["总市值"] / 1e8 df["封单资金_亿"] = df["封单资金"] / 1e8 df["板上成交额_亿"] = df["板上成交额"] / 1e8 return df def save_dt_to_sqlite(df, db_name="quant.db"): conn = sqlite3.connect(db_name) write_df = df[["交易日期","股票代码","股票名称","价格","涨跌幅","成交额_亿","流通市值_亿","总市值_亿","动态市盈率","换手率","连续跌停数","最后封板时间","封单资金_亿","板上成交额_亿","开板次数"]].copy() write_df.rename(columns={ "交易日期":"trade_date", "股票代码":"stock_code", "股票名称":"stock_name", "价格":"price", "涨跌幅":"zf", "成交额_亿":"cje_yi", "流通市值_亿":"ltsz_yi", "总市值_亿":"zsz_yi", "动态市盈率":"pe", "换手率":"hs", "连续跌停数":"lbc", "最后封板时间":"lbt", "封单资金_亿":"zj_yi", "板上成交额_亿":"fba_yi", "开板次数":"zbc" },inplace=True) write_df.to_sql("dt_pool", conn, if_exists="append", index=False) conn.close() logger.info(f"跌停池入库完成，共{len(df)}条") def calc_daily_emotion_stat(trade_date): """计算市场情绪聚合指标""" conn = sqlite3.connect(DB_PATH) df_zt = pd.read_sql(f"SELECT * FROM zt_pool WHERE trade_date='{trade_date}'", conn) df_dt = pd.read_sql(f"SELECT * FROM dt_pool WHERE trade_date='{trade_date}'", conn) conn.close() up_count = len(df_zt) down_count = len(df_dt) bomb_count = len(df_zt[df_zt["zbc"] > 0]) total_try = up_count + bomb_count bomb_rate = round(bomb_count / total_try *100,2) if total_try>0 else 0.0 success_rate = round(up_count / total_try *100,2) if total_try>0 else 0.0 return { "trade_date":trade_date, "up_count":up_count, "down_count":down_count, "bomb_count":bomb_count, "bomb_rate":bomb_rate, "success_rate":success_rate } def save_emotion_stat(stat_dict, db_name="quant.db"): conn = sqlite3.connect(db_name) cur = conn.cursor() sql = """INSERT OR REPLACE INTO market_emotion_daily (trade_date,up_count,down_count,bomb_count,bomb_rate,success_rate) VALUES (?,?,?,?,?,?)""" cur.execute(sql,( stat_dict["trade_date"],stat_dict["up_count"],stat_dict["down_count"], stat_dict["bomb_count"],stat_dict["bomb_rate"],stat_dict["success_rate"] )) conn.commit() conn.close() logger.info(f"{stat_dict['trade_date']}市场情绪统计入库完成") def data_quality_check(raw_list): if not raw_list: logger.warning("接口返回空列表") return False record_count = len(raw_list) if record_count <2: logger.warning(f"返回记录数量过少:{record_count}") df_check = pd.DataFrame(raw_list) null_code_cnt = df_check["dm"].isna().sum() null_rate = null_code_cnt / len(df_check) if null_rate >0.2: logger.error(f"股票代码空值占比过高 {null_rate:.2%}") return False return True # ----------复用涨停池相关函数---------- def fetch_raw_zt_pool(trade_date): url = f"http://api.biyingapi.com/hslt/ztgc/{trade_date}/{LICENCE}" return biying_api_get_retry(url) def clean_zt_data(raw_json_list, trade_date): df = pd.DataFrame(raw_json_list) keep_cols = ["dm","mc","p","zf","cje","lt","hs","lbc","zbc","fbt","lbt","zj"] df = df[keep_cols].copy() df.columns = ["股票代码","股票名称","价格","涨幅","成交额","流通市值","换手率","连板数","炸板次数","首次封板时间","最后封板时间","封板资金"] df = df.replace([None,"null",""], np.nan) num_cols = ["价格","涨幅","成交额","流通市值","换手率","连板数","炸板次数"] for col in num_cols: df[col] = pd.to_numeric(df[col], errors="coerce") df = df.dropna(subset=["股票代码","股票名称"]) df["交易日期"] = trade_date df["流通市值_亿"] = df["流通市值"] / 1e8 df["成交额_亿"] = df["成交额"] / 1e8 return df def init_sqlite_db(db_name="quant.db"): conn = sqlite3.connect(db_name) cur = conn.cursor() create_sql = """ CREATE TABLE IF NOT EXISTS zt_pool ( id INTEGER PRIMARY KEY AUTOINCREMENT, trade_date TEXT, stock_code TEXT, stock_name TEXT, price REAL, zf REAL, cje_yi REAL, ltsz_yi REAL, hs REAL, lbc INTEGER, zbc INTEGER, fbt TEXT, lbt TEXT, zj_yi REAL, UNIQUE(trade_date, stock_code) ) """ cur.execute(create_sql) conn.commit() conn.close() def save_zt_to_sqlite(df, db_name="quant.db"): conn = sqlite3.connect(db_name) write_df = df[["交易日期","股票代码","股票名称","价格","涨幅","成交额_亿","流通市值_亿","换手率","连板数","炸板次数","首次封板时间","最后封板时间","封板资金"]].copy() write_df.rename(columns={ "交易日期":"trade_date","股票代码":"stock_code","股票名称":"stock_name", "价格":"price","涨幅":"zf","成交额_亿":"cje_yi","流通市值_亿":"ltsz_yi", "换手率":"hs","连板数":"lbc","炸板次数":"zbc","首次封板时间":"fbt","最后封板时间":"lbt","封板资金":"zj_yi" },inplace=True) write_df.to_sql("zt_pool",conn,if_exists="append",index=False) conn.close() # ----------------改造每日采集任务：同时采集涨停+跌停+情绪统计---------------- def daily_collect_work(): logger.info("==== 开始执行每日盘后涨跌停池采集任务 ====") today = time.strftime("%Y‑%m‑%d") try: trade_flag = is_trade_day(today) if trade_flag is None: logger.warning(f"{today} 未在交易日历找到记录，跳过采集") return if not trade_flag: logger.info(f"{today} 判定为非交易日，直接跳过采集") return # 1 涨停池 raw_zt = fetch_raw_zt_pool(today) if data_quality_check(raw_zt): df_zt_clean = clean_zt_data(raw_zt, today) save_zt_to_sqlite(df_zt_clean) # 2 跌停池 raw_dt = fetch_raw_dt_pool(today) if data_quality_check(raw_dt): df_dt_clean = clean_dt_data(raw_dt, today) save_dt_to_sqlite(df_dt_clean) # 3 计算并保存当日市场情绪聚合统计 emotion_stat = calc_daily_emotion_stat(today) save_emotion_stat(emotion_stat) logger.info( f"{today}统计：涨停{emotion_stat['up_count']}家，跌停{emotion_stat['down_count']}家，" f"炸板率{emotion_stat['bomb_rate']}%，封板成功率{emotion_stat['success_rate']}%" ) except Exception as e: logger.error(f"每日采集流程发生未知异常：{str(e)}", exc_info=True) logger.info("==== 每日盘后涨跌停池采集任务执行结束 ====\n") def start_scheduler(): scheduler = BackgroundScheduler() scheduler.add_job(daily_collect_work, "cron", hour=16, minute=45) scheduler.start() logger.info("定时任务已启动，每日16:45采集涨跌停股池与情绪统计") try: while True: time.sleep(60) except KeyboardInterrupt: scheduler.shutdown() logger.info("接收到中断信号，调度器已关闭") if __name__ == "__main__": init_sqlite_db() init_new_tables() # 取消注释，手动执行一次测试 # daily_collect_work() start_scheduler()

五、可视化：涨跌停数量时序绘图

import matplotlib.pyplot as plt plt.rcParams["font.sans-serif"] = ["SimHei"] plt.rcParams["axes.unicode_minus"] = False def plot_up_down_trend(start_date, end_date): conn = sqlite3.connect(DB_PATH) sql = """ SELECT trade_date,up_count,down_count,bomb_rate,success_rate FROM market_emotion_daily WHERE trade_date >= ? AND trade_date <= ? ORDER BY trade_date """ df = pd.read_sql(sql, conn, params=(start_date, end_date)) conn.close() if len(df) ==0: print("暂无情绪统计数据") return fig, ax = plt.subplots(figsize=(14,6)) ax.plot(df["trade_date"], df["up_count"], color="#e63946", marker="o", label="涨停家数") ax.plot(df["trade_date"], df["down_count"], color="#2a9d8f", marker="s", label="跌停家数") ax.set_title("A股涨跌停家数时序统计", fontsize=14) ax.set_xlabel("交易日") ax.set_ylabel("标的数量") ax.legend() ax.grid(alpha=0.3) plt.xticks(rotation=45) plt.tight_layout() plt.savefig("up_down_trend.png", dpi=200) plt.show() # plot_up_down_trend("2026‑07‑01","2026‑08‑25")

六、业务关键点

1. 接口仅返回个股明细，涨停家数、跌停家数、炸板率、封板成功率全部通过代码聚合计算。
2. 数据表设置trade_date+stock_code联合唯一约束，脚本重复运行不会造成重复入库。
3. 金额字段原始单位为元，代码统一换算为亿元，便于分析阅读。
4. market_emotion_daily汇总表存储每日聚合结果，时序查询不需要扫描全部明细，提升查询性能。

七、拓展练习方向

1. SQL 联合查询，对比涨停池、跌停池两组标的换手率、流通市值分布；
2. 筛选高炸板率的交易日，观察对应个股特征；
3. 编写批量回捞脚本，批量导入历史涨跌停数据积累长周期样本。

下篇预告

系列第 8 篇：强势股池数据实战：新高、量比、涨速指标的二次计算

接入强势股池接口，完成清洗入库；自研新高占比等衍生指标，实现多股池联合筛选。

免责申明：文中所有数据处理逻辑仅为编程演示，仅为数据演示，不构成投资建议。市场有风险，投资需谨慎。