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

量化数据开发实战系列第 10 篇炸板股池实战提取炸板时间、炸板次数统计炸板行为特征前言前面第 7‑9 篇已经完成涨停、跌停、强势股、次新股四大股池的数据采集清洗入库并且搭建起每日市场情绪统计汇总表。本篇接入炸板股池接口炸板股池存放当日曾经冲击涨停但盘中涨停板被打开过的个股原始明细包含首次封板时间、炸板次数、价格量能等字段。接口只返回单只炸板标的原始明细早盘炸板数量、午后炸板数量、平均炸板次数这类聚合统计结果接口不会直接输出全部需要读取明细通过业务代码运算得到。本篇完成炸板股池完整流水线新增数据表、衍生统计指标更新定时采集任务同时实现炸板群体的基础统计分析。一、接口原始字段梳理表格字段说明dm股票代码mc股票名称p价格元 ¥ztp涨停价元 ¥zf涨跌幅%cje成交额元 ¥lt流通市值元 ¥zsz总市值元 ¥zs涨速%hs转手率%tj涨停统计x 天 /y 板fbt首次封板时间HH:mm:sszbc炸板次数二、自研衍生统计指标炸板总家数当日炸板股池标的总条数早盘炸板数量首次封板时间早于 10:30 的炸板标的午后炸板数量首次封板时间大于等于 10:30 的炸板标的平均炸板次数全部炸板标的zbc的算术平均值代表板块分歧程度炸板后平均涨跌幅炸板个股当日收盘涨跌幅均值全部聚合统计由业务代码计算接口不提供统计结果。三、数据表设计在quant.db数据库新增zbgc_pool炸板股池明细表扩展market_emotion_daily每日情绪汇总表增加炸板池相关聚合字段。zbgc_pool 炸板股池明细表表格字段类型说明idINTEGER自增主键trade_dateTEXT交易日期 yyyy‑MM‑ddstock_codeTEXT股票代码stock_nameTEXT股票名称priceREAL现价 ¥zt_priceREAL涨停价 ¥zfREAL涨跌幅 %cje_yiREAL成交额 亿元 ¥ltsz_yiREAL流通市值 亿元 ¥zsz_yiREAL总市值 亿元 ¥speed_zREAL涨速 %hsREAL换手率 %stat_infoTEXT涨停统计字符串first_block_timeTEXT首次封板时间 HH:mm:ssbomb_cntINTEGER炸板次数UNIQUE(trade_date,stock_code)联合唯一约束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( 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 init_zb_table(db_namequant.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 fhttp://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], errorscoerce) 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_namequant.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 },inplaceTrue) write_df.to_sql(zbgc_pool, conn, if_existsappend, indexFalse) conn.close() logger.info(f炸板池入库完成共{len(df)}条记录) def calc_zb_stat(trade_date): 计算炸板池聚合统计指标 conn sqlite3.connect(DB_PATH) df_zb pd.read_sql(fSELECT * 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_namequant.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 fhttp://api.biyingapi.com/hslt/ztgc/{trade_date}/{LICENCE} return biying_api_get_retry(url) def fetch_raw_dt_pool(trade_date): url fhttp://api.biyingapi.com/hslt/dtgc/{trade_date}/{LICENCE} return biying_api_get_retry(url) def fetch_raw_qs_pool(trade_date): url fhttp://api.biyingapi.com/hslt/qsgc/{trade_date}/{LICENCE} return biying_api_get_retry(url) def fetch_raw_cx_pool(trade_date): url fhttp://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], errorscoerce) 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], 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 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], errorscoerce) 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], errorscoerce) 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_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 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() def save_qs_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, 涨停价:zt_price, 涨幅:zf, 成交额_亿:cje_yi, 流通市值_亿:ltsz_yi, 总市值_亿:zsz_yi, 涨速:speed_z, 是否新高:is_new_high, 量比:lb, 换手率:hs, 涨停统计:stat_info },inplaceTrue) write_df.to_sql(qsgc_pool, conn, if_existsappend, indexFalse) conn.close() def save_cx_to_sqlite(df, db_namequant.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 },inplaceTrue) write_df.to_sql(cxgc_pool, conn, if_existsappend, indexFalse) conn.close() 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() def calc_qs_stat(trade_date): conn sqlite3.connect(DB_PATH) df_qs pd.read_sql(fSELECT * 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_count0 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_namequant.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(fSELECT * 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_namequant.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_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_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, alpha0.6, label炸板家数) ax2.plot(df[trade_date], df[avg_zb_cnt], color#e63946, markero, label平均炸板次数) ax1.set_xlabel(交易日) ax1.set_ylabel(炸板家数) ax2.set_ylabel(平均炸板次数) fig.legend(locupper right) plt.title(炸板家数与平均炸板次数时序) plt.xticks(rotation45) plt.tight_layout() plt.savefig(zb_trend.png,dpi200) plt.show() # plot_zb_trend(2026‑07‑01,2026‑08‑25)六、业务关键点炸板接口仅返回个股明细炸板家数、早盘 / 午后炸板数量、平均炸板次数全部由代码聚合计算通过截取HH:mm:ss时间字符串区分早盘、午后封板用来观察资金封板时间偏好数据表设置trade_datestock_code联合唯一约束重复执行脚本不会重复入库统计存入market_emotion_daily汇总表时序分析不需要扫描全量表提升查询性能。七、拓展练习方向SQL 联合查询对比涨停池与炸板池两组标的换手率、流通市值分布筛选高炸板次数标的统计相关交易特征编写批量回捞脚本获取历史炸板数据积累长周期样本。下篇预告系列第 11 篇指数接口实战主要指数列表、指数实时、指数 K 线历史数据存储获取指数基础列表、指数历史 K 线原始行情完成数据表设计与入库自研计算阶段涨跌幅、个股相对指数超额收益。免责申明文中所有数据处理逻辑仅为编程演示仅为数据演示不构成投资建议。市场有风险投资需谨慎。