企业官网建设流程全解析

量化数据开发实战系列（第 10 篇）：炸板股池实战：提取炸板时间、炸板次数，统计炸板行为特征

前言

前面第 7‑9 篇，已经完成涨停、跌停、强势股、次新股四大股池的数据采集清洗入库，并且搭建起每日市场情绪统计汇总表。本篇接入炸板股池接口，炸板股池存放当日曾经冲击涨停，但盘中涨停板被打开过的个股原始明细，包含首次封板时间、炸板次数、价格量能等字段。

接口只返回单只炸板标的原始明细，早盘炸板数量、午后炸板数量、平均炸板次数这类聚合统计结果，接口不会直接输出，全部需要读取明细通过业务代码运算得到。本篇完成炸板股池完整流水线，新增数据表、衍生统计指标，更新定时采集任务，同时实现炸板群体的基础统计分析。

一、接口原始字段梳理

表格

二、自研衍生统计指标

1. 炸板总家数：当日炸板股池标的总条数
2. 早盘炸板数量：首次封板时间早于 10:30 的炸板标的
3. 午后炸板数量：首次封板时间大于等于 10:30 的炸板标的
4. 平均炸板次数：全部炸板标的zbc的算术平均值，代表板块分歧程度
5. 炸板后平均涨跌幅：炸板个股当日收盘涨跌幅均值

全部聚合统计由业务代码计算，接口不提供统计结果。

三、数据表设计

在quant.db数据库新增zbgc_pool炸板股池明细表；扩展market_emotion_daily每日情绪汇总表，增加炸板池相关聚合字段。

zbgc_pool 炸板股池明细表

表格

market_emotion_daily 新增字段：zb_total INTEGER, zb_morning INTEGER, zb_afternoon INTEGER, avg_zb_cnt REAL, avg_zb_zf REAL

四、完整可运行代码

代码调用复用前面章节请求、日志、交易日历、各股池相关逻辑；新增炸板股池全流程处理，改造每日采集任务。

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 init_zb_table(db_name="quant.db"): conn = sqlite3.connect(db_name) cur = conn.cursor() create_zb_sql = """ CREATE TABLE IF NOT EXISTS zbgc_pool ( id INTEGER PRIMARY KEY AUTOINCREMENT, trade_date TEXT, stock_code TEXT, stock_name TEXT, price REAL, zt_price REAL, zf REAL, cje_yi REAL, ltsz_yi REAL, zsz_yi REAL, speed_z REAL, hs REAL, stat_info TEXT, first_block_time TEXT, bomb_cnt INTEGER, UNIQUE(trade_date, stock_code) ) """ cur.execute(create_zb_sql) # 情绪汇总表追加炸板统计字段，已存在直接跳过 try: cur.execute("ALTER TABLE market_emotion_daily ADD COLUMN zb_total INTEGER") except sqlite3.OperationalError: pass try: cur.execute("ALTER TABLE market_emotion_daily ADD COLUMN zb_morning INTEGER") except sqlite3.OperationalError: pass try: cur.execute("ALTER TABLE market_emotion_daily ADD COLUMN zb_afternoon INTEGER") except sqlite3.OperationalError: pass try: cur.execute("ALTER TABLE market_emotion_daily ADD COLUMN avg_zb_cnt REAL") except sqlite3.OperationalError: pass try: cur.execute("ALTER TABLE market_emotion_daily ADD COLUMN avg_zb_zf REAL") except sqlite3.OperationalError: pass conn.commit() conn.close() logger.info("炸板股池数据表初始化完成") def fetch_raw_zb_pool(trade_date): url = f"http://api.biyingapi.com/hslt/zbgc/{trade_date}/{LICENCE}" return biying_api_get_retry(url) def clean_zb_data(raw_json_list, trade_date): df = pd.DataFrame(raw_json_list) keep_cols = ["dm","mc","p","ztp","zf","cje","lt","zsz","zs","hs","tj","fbt","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 return df def save_zb_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", "涨停价":"zt_price", "涨跌幅":"zf", "成交额_亿":"cje_yi", "流通市值_亿":"ltsz_yi", "总市值_亿":"zsz_yi", "涨速":"speed_z", "换手率":"hs", "涨停统计":"stat_info", "首次封板时间":"first_block_time", "炸板次数":"bomb_cnt" },inplace=True) write_df.to_sql("zbgc_pool", conn, if_exists="append", index=False) conn.close() logger.info(f"炸板池入库完成，共{len(df)}条记录") def calc_zb_stat(trade_date): """计算炸板池聚合统计指标""" conn = sqlite3.connect(DB_PATH) df_zb = pd.read_sql(f"SELECT * FROM zbgc_pool WHERE trade_date='{trade_date}'", conn) conn.close() zb_total = len(df_zb) if zb_total == 0: return { "zb_total":0,"zb_morning":0,"zb_afternoon":0,"avg_zb_cnt":0.0,"avg_zb_zf":0.0 } # 截取时间字符串前5位，按10:30区分早盘、午后 df_zb["hour_str"] = df_zb["first_block_time"].str.slice(0,5) zb_morning = len(df_zb[df_zb["hour_str"] < "10:30"]) zb_afternoon = zb_total - zb_morning avg_zb_cnt = round(df_zb["bomb_cnt"].mean(),2) avg_zb_zf = round(df_zb["zf"].mean(),2) return { "zb_total":zb_total, "zb_morning":zb_morning, "zb_afternoon":zb_afternoon, "avg_zb_cnt":avg_zb_cnt, "avg_zb_zf":avg_zb_zf } def merge_zb_stat(trade_date, zb_stat, db_name="quant.db"): """把炸板统计更新写入情绪汇总表""" conn = sqlite3.connect(db_name) cur = conn.cursor() sql = """ UPDATE market_emotion_daily SET zb_total=?, zb_morning=?, zb_afternoon=?, avg_zb_cnt=?, avg_zb_zf=? WHERE trade_date=? """ cur.execute(sql,( zb_stat["zb_total"], zb_stat["zb_morning"], zb_stat["zb_afternoon"], zb_stat["avg_zb_cnt"], zb_stat["avg_zb_zf"], trade_date )) conn.commit() conn.close() logger.info(f"{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 fetch_raw_dt_pool(trade_date): url = f"http://api.biyingapi.com/hslt/dtgc/{trade_date}/{LICENCE}" return biying_api_get_retry(url) def fetch_raw_qs_pool(trade_date): url = f"http://api.biyingapi.com/hslt/qsgc/{trade_date}/{LICENCE}" return biying_api_get_retry(url) def fetch_raw_cx_pool(trade_date): url = f"http://api.biyingapi.com/hslt/cxgc/{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 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 clean_qs_data(raw_json_list, trade_date): df = pd.DataFrame(raw_json_list) keep_cols = ["dm","mc","p","ztp","zf","cje","lt","zsz","zs","nh","lb","hs","tj"] 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 return df def clean_cx_data(raw_json_list, trade_date): from datetime import datetime df = pd.DataFrame(raw_json_list) keep_cols = ["dm","mc","p","ztp","zf","cje","lt","zsz","nh","hs","tj","kb","od","ipod"] df = df[keep_cols].copy() df.columns = [ "股票代码","股票名称","价格","涨停价","涨跌幅","成交额","流通市值","总市值", "是否新高","换手率","涨停统计","开板几日","开板日期_raw","上市日期_raw" ] 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 def convert_ymd(raw_val): if pd.isna(raw_val): return None s = str(int(raw_val)) try: return datetime.strptime(s,"%Y%m%d").strftime("%Y‑%m‑%d") except Exception: return None df["open_board_date"] = df["开板日期_raw"].apply(convert_ymd) df["ipo_date"] = df["上市日期_raw"].apply(convert_ymd) return df 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 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() def save_qs_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", "涨停价":"zt_price", "涨幅":"zf", "成交额_亿":"cje_yi", "流通市值_亿":"ltsz_yi", "总市值_亿":"zsz_yi", "涨速":"speed_z", "是否新高":"is_new_high", "量比":"lb", "换手率":"hs", "涨停统计":"stat_info" },inplace=True) write_df.to_sql("qsgc_pool", conn, if_exists="append", index=False) conn.close() def save_cx_to_sqlite(df, db_name="quant.db"): conn = sqlite3.connect(db_name) write_df = df[[ "交易日期","股票代码","股票名称","价格","涨停价","涨跌幅", "成交额_亿","流通市值_亿","总市值_亿","是否新高","换手率", "涨停统计","开板几日","open_board_date","ipo_date" ]].copy() write_df.rename(columns={ "交易日期":"trade_date", "股票代码":"stock_code", "股票名称":"stock_name", "价格":"price", "涨停价":"zt_price", "涨跌幅":"zf", "成交额_亿":"cje_yi", "流通市值_亿":"ltsz_yi", "总市值_亿":"zsz_yi", "是否新高":"is_new_high", "换手率":"hs", "涨停统计":"stat_info", "开板几日":"kb_days", "open_board_date":"open_board_date", "ipo_date":"ipo_date" },inplace=True) write_df.to_sql("cxgc_pool", conn, if_exists="append", index=False) conn.close() 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() def calc_qs_stat(trade_date): conn = sqlite3.connect(DB_PATH) df_qs = pd.read_sql(f"SELECT * FROM qsgc_pool WHERE trade_date='{trade_date}'", conn) conn.close() qs_count = len(df_qs) nh_count = len(df_qs[df_qs["is_new_high"] == 1]) nh_ratio = round(nh_count / qs_count * 100,2) if qs_count>0 else 0.0 return {"qs_count": qs_count, "nh_count": nh_count, "nh_ratio": nh_ratio} def merge_emotion_stat(orig_stat, qs_stat, db_name="quant.db"): conn = sqlite3.connect(db_name) cur = conn.cursor() sql = """ UPDATE market_emotion_daily SET qs_count=?, nh_count=?, nh_ratio=? WHERE trade_date=? """ cur.execute(sql,(qs_stat["qs_count"],qs_stat["nh_count"],qs_stat["nh_ratio"],orig_stat["trade_date"])) conn.commit() conn.close() def calc_cx_stat(trade_date): conn = sqlite3.connect(DB_PATH) df_cx = pd.read_sql(f"SELECT * FROM cxgc_pool WHERE trade_date='{trade_date}'", conn) conn.close() cx_total = len(df_cx) if cx_total ==0: return {"cx_total":0,"cx_opened":0,"cx_not_open":0,"avg_kb_days":0.0} cx_opened = len(df_cx[df_cx["kb_days"] > 0]) cx_not_open = len(df_cx[df_cx["kb_days"] == 0]) opened_df = df_cx[df_cx["kb_days"]>0] avg_kb_days = round(opened_df["kb_days"].mean(),2) if len(opened_df)>0 else 0.0 return {"cx_total":cx_total,"cx_opened":cx_opened,"cx_not_open":cx_not_open,"avg_kb_days":avg_kb_days} def merge_cx_stat_to_emotion(trade_date, cx_stat, db_name="quant.db"): conn = sqlite3.connect(db_name) cur = conn.cursor() sql = """ UPDATE market_emotion_daily SET cx_total=?, cx_opened=?, cx_not_open=?, avg_kb_days=? WHERE trade_date=? """ cur.execute(sql,(cx_stat["cx_total"],cx_stat["cx_opened"],cx_stat["cx_not_open"],cx_stat["avg_kb_days"],trade_date)) conn.commit() 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 强势股池 raw_qs = fetch_raw_qs_pool(today) if data_quality_check(raw_qs): df_qs_clean = clean_qs_data(raw_qs, today) save_qs_to_sqlite(df_qs_clean) # 4 次新股池 raw_cx = fetch_raw_cx_pool(today) if data_quality_check(raw_cx): df_cx_clean = clean_cx_data(raw_cx, today) save_cx_to_sqlite(df_cx_clean) # 5 炸板股池【本篇新增】 raw_zb = fetch_raw_zb_pool(today) if data_quality_check(raw_zb): df_zb_clean = clean_zb_data(raw_zb, today) save_zb_to_sqlite(df_zb_clean) # 聚合统计 emotion_stat = calc_daily_emotion_stat(today) save_emotion_stat(emotion_stat) qs_stat = calc_qs_stat(today) merge_emotion_stat(emotion_stat, qs_stat) cx_stat = calc_cx_stat(today) merge_cx_stat_to_emotion(today, cx_stat) zb_stat = calc_zb_stat(today) merge_zb_stat(today, zb_stat) logger.info( f"{today}｜涨停{emotion_stat['up_count']}家，跌停{emotion_stat['down_count']}家，" f"强势股{qs_stat['qs_count']}家，次新股{cx_stat['cx_total']}家，炸板{zb_stat['zb_total']}家" ) 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_zb_table() # 取消注释，手动执行一次测试 # daily_collect_work() start_scheduler()

五、可视化：炸板数量与平均炸板次数时序绘图

import matplotlib.pyplot as plt plt.rcParams["font.sans-serif"] = ["SimHei"] plt.rcParams["axes.unicode_minus"] = False def plot_zb_trend(start_date, end_date): conn = sqlite3.connect(DB_PATH) sql = """ SELECT trade_date,zb_total,avg_zb_cnt,avg_zb_zf 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, ax1 = plt.subplots(figsize=(14,6)) ax2 = ax1.twinx() ax1.bar(df["trade_date"], df["zb_total"], color="#f4a261", alpha=0.6, label="炸板家数") ax2.plot(df["trade_date"], df["avg_zb_cnt"], color="#e63946", marker="o", label="平均炸板次数") ax1.set_xlabel("交易日") ax1.set_ylabel("炸板家数") ax2.set_ylabel("平均炸板次数") fig.legend(loc="upper right") plt.title("炸板家数与平均炸板次数时序") plt.xticks(rotation=45) plt.tight_layout() plt.savefig("zb_trend.png",dpi=200) plt.show() # plot_zb_trend("2026‑07‑01","2026‑08‑25")

六、业务关键点

1. 炸板接口仅返回个股明细，炸板家数、早盘 / 午后炸板数量、平均炸板次数全部由代码聚合计算；
2. 通过截取HH:mm:ss时间字符串区分早盘、午后封板，用来观察资金封板时间偏好；
3. 数据表设置trade_date+stock_code联合唯一约束，重复执行脚本不会重复入库；
4. 统计存入market_emotion_daily汇总表，时序分析不需要扫描全量表，提升查询性能。

七、拓展练习方向

1. SQL 联合查询，对比涨停池与炸板池两组标的换手率、流通市值分布；
2. 筛选高炸板次数标的，统计相关交易特征；
3. 编写批量回捞脚本，获取历史炸板数据积累长周期样本。

下篇预告

系列第 11 篇：指数接口实战：主要指数列表、指数实时、指数 K 线历史数据存储

获取指数基础列表、指数历史 K 线原始行情；完成数据表设计与入库；自研计算阶段涨跌幅、个股相对指数超额收益。

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