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

量化数据开发实战系列第 7 篇涨跌停股池深度实战涨停、跌停数据采集、入库、指标衍生计算前言前面章节已经完成涨停股池采集、日志重试、定时调度、本地交易日历。本篇扩展接入跌停股池接口同时完成涨停、跌停两套股池的采集、清洗、持久化存储。接口仅返回单只标的原始明细炸板率、封板成功率这类市场情绪聚合指标不会直接提供需要读取明细数据通过代码完成聚合运算。本篇将实现双股池流水线并且计算每日市场情绪统计值存入汇总表方便后续时序分析与可视化。一、接口原始字段梳理涨停股池核心字段表格字段说明dm股票代码mc股票名称p价格元 ¥zf涨幅%cje成交额元 ¥lt流通市值元 ¥zsz总市值元 ¥hs换手率%lbc连板数zbc炸板次数fbt首次封板时间lbt最后封板时间zj封板资金元 ¥跌停股池核心字段表格字段说明dm股票代码mc股票名称p价格元 ¥zf跌幅%cje成交额元 ¥lt流通市值元 ¥zsz总市值元 ¥pe动态市盈率hs换手率%lbc连续跌停次数lbt最后封板时间zj封单资金元 ¥fba板上成交额元 ¥zbc开板次数二、自研衍生统计指标涨停家数当日涨停股池记录总条数跌停家数当日跌停股池记录总条数炸板数量涨停池中炸板次数zbc0标的数量炸板率 (%) 炸板数量 / (涨停家数 炸板数量) * 100封板成功率 (%) 涨停家数 / (涨停家数 炸板数量) * 100全部聚合指标需要业务代码计算接口不直接返回统计结果。三、数据表设计在quant.db新增两张数据表dt_pool跌停股池明细表market_emotion_daily每日市场情绪汇总统计表dt_pool 跌停股池表表格字段类型说明idINTEGER自增主键trade_dateTEXT交易日期 yyyy‑MM‑ddstock_codeTEXT股票代码stock_nameTEXT股票名称priceREAL价格 ¥zfREAL涨跌幅 %cje_yiREAL成交额 亿元 ¥ltsz_yiREAL流通市值 亿元 ¥zsz_yiREAL总市值 亿元 ¥peREAL动态市盈率hsREAL换手率 %lbcINTEGER连续跌停次数lbtTEXT最后封板时间zj_yiREAL封单资金 亿元 ¥fba_yiREAL板上成交额 亿元 ¥zbcINTEGER开板次数UNIQUE(trade_date,stock_code)联合唯一约束market_emotion_daily 每日情绪汇总表表格字段类型说明idINTEGER自增主键trade_dateTEXT交易日期up_countINTEGER涨停家数down_countINTEGER跌停家数bomb_countINTEGER炸板数量bomb_rateREAL炸板率 %success_rateREAL封板成功率 %UNIQUE(trade_date)四、完整可运行代码代码调用复用前面章节请求、日志、交易日历逻辑新增跌停股池全流程处理、情绪统计计算改造每日采集任务。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( filenameLOG_FILE, levellogging.INFO, format%(asctime)s %(levelname)s %(message)s, datefmt%Y‑%m‑%d %H:%M:%S, filemodea ) logger logging.getLogger(__name__) # ----------带重试HTTP请求---------- def biying_api_get_retry(full_url, timeout15, max_retry3): for attempt in range(1, max_retry 1): try: resp requests.get(full_url, timeouttimeout) if resp.status_code 200: return resp.json() logger.warning(fHTTP状态码异常{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_namequant.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_namequant.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_namequant.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 fhttp://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], errorscoerce) 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_namequant.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 },inplaceTrue) write_df.to_sql(dt_pool, conn, if_existsappend, indexFalse) conn.close() logger.info(f跌停池入库完成共{len(df)}条) def calc_daily_emotion_stat(trade_date): 计算市场情绪聚合指标 conn sqlite3.connect(DB_PATH) df_zt pd.read_sql(fSELECT * FROM zt_pool WHERE trade_date{trade_date}, conn) df_dt pd.read_sql(fSELECT * 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_try0 else 0.0 success_rate round(up_count / total_try *100,2) if total_try0 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_namequant.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 fhttp://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], errorscoerce) df df.dropna(subset[股票代码,股票名称]) df[交易日期] trade_date df[流通市值_亿] df[流通市值] / 1e8 df[成交额_亿] df[成交额] / 1e8 return df def init_sqlite_db(db_namequant.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_namequant.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 },inplaceTrue) write_df.to_sql(zt_pool,conn,if_existsappend,indexFalse) 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_infoTrue) logger.info( 每日盘后涨跌停池采集任务执行结束 \n) def start_scheduler(): scheduler BackgroundScheduler() scheduler.add_job(daily_collect_work, cron, hour16, minute45) 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, markero, label涨停家数) ax.plot(df[trade_date], df[down_count], color#2a9d8f, markers, label跌停家数) ax.set_title(A股涨跌停家数时序统计, fontsize14) ax.set_xlabel(交易日) ax.set_ylabel(标的数量) ax.legend() ax.grid(alpha0.3) plt.xticks(rotation45) plt.tight_layout() plt.savefig(up_down_trend.png, dpi200) plt.show() # plot_up_down_trend(2026‑07‑01,2026‑08‑25)六、业务关键点接口仅返回个股明细涨停家数、跌停家数、炸板率、封板成功率全部通过代码聚合计算。数据表设置trade_datestock_code联合唯一约束脚本重复运行不会造成重复入库。金额字段原始单位为元代码统一换算为亿元便于分析阅读。market_emotion_daily汇总表存储每日聚合结果时序查询不需要扫描全部明细提升查询性能。七、拓展练习方向SQL 联合查询对比涨停池、跌停池两组标的换手率、流通市值分布筛选高炸板率的交易日观察对应个股特征编写批量回捞脚本批量导入历史涨跌停数据积累长周期样本。下篇预告系列第 8 篇强势股池数据实战新高、量比、涨速指标的二次计算接入强势股池接口完成清洗入库自研新高占比等衍生指标实现多股池联合筛选。免责申明文中所有数据处理逻辑仅为编程演示仅为数据演示不构成投资建议。市场有风险投资需谨慎。