量化数据开发实战系列第 8 篇强势股池数据实战新高、量比、涨速指标的二次计算前言第 7 篇完成涨停、跌停股池的采集清洗入库与市场情绪统计。本篇接入强势股池接口强势股池返回当日表现较强个股原始明细包含涨速、量比、是否创阶段新高等原始字段。接口只输出单只个股明细数据新高占比这类聚合统计结果不会直接返回需要基于原始明细通过代码做聚合运算得到。本篇完成强势股池完整流水线获取原始数据、清洗转换、持久化入库新增衍生统计指标更新定时采集任务同时可以和涨停池做联合筛选分析。一、接口原始字段梳理表格字段说明dm股票代码mc股票名称p价格元 ¥ztp涨停价元 ¥zf涨幅%cje成交额元 ¥lt流通市值元 ¥zsz总市值元 ¥zs涨速%nh是否新高0否1是lb量比hs换手率%tj涨停统计x 天 /y 板二、自研衍生统计指标强势股总家数当日强势股池返回标的总记录条数新高标的数量强势股池中nh 1的标的数量新高占比 (%) 新高标的数量 / 强势股总家数 * 100含义强势股群体中创出阶段新高个股占比占比越高代表趋势类强势标的越多。高量比标的数量量比lb 2的标的统计代表短期成交活跃度明显放大。以上聚合统计均需要业务代码计算接口不提供。三、数据表设计在quant.db新增qsgc_pool强势股池明细表对已经存在的market_emotion_daily情绪汇总表增加强势股相关统计字段用于时序分析。qsgc_pool 强势股池明细表表格字段类型说明idINTEGER自增主键trade_dateTEXT交易日期 yyyy‑MM‑ddstock_codeTEXT股票代码stock_nameTEXT股票名称priceREAL现价 ¥zt_priceREAL涨停价 ¥zfREAL涨幅 %cje_yiREAL成交额 亿元 ¥ltsz_yiREAL流通市值 亿元 ¥zsz_yiREAL总市值 亿元 ¥speed_zREAL涨速 %is_new_highINTEGER是否新高 1 是 0 否lbREAL量比hsREAL换手率 %stat_infoTEXT涨停统计字符串UNIQUE(trade_date,stock_code)联合唯一约束market_emotion_daily 新增字段qs_count INTEGER, nh_count INTEGER, nh_ratio 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_qs_table(db_namequant.db): conn sqlite3.connect(db_name) cur conn.cursor() create_qs_sql CREATE TABLE IF NOT EXISTS qsgc_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, is_new_high INTEGER, lb REAL, hs REAL, stat_info TEXT, UNIQUE(trade_date, stock_code) ) cur.execute(create_qs_sql) # 给情绪汇总表追加字段字段已存在会捕获异常跳过 try: cur.execute(ALTER TABLE market_emotion_daily ADD COLUMN qs_count INTEGER) except sqlite3.OperationalError: pass try: cur.execute(ALTER TABLE market_emotion_daily ADD COLUMN nh_count INTEGER) except sqlite3.OperationalError: pass try: cur.execute(ALTER TABLE market_emotion_daily ADD COLUMN nh_ratio REAL) except sqlite3.OperationalError: pass conn.commit() conn.close() logger.info(强势股池数据表初始化完成) def fetch_raw_qs_pool(trade_date): url fhttp://api.biyingapi.com/hslt/qsgc/{trade_date}/{LICENCE} return biying_api_get_retry(url) 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 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() logger.info(f强势股池入库完成共{len(df)}条记录) 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() logger.info(f{orig_stat[trade_date]}强势股统计已合并到情绪表) def data_quality_check(raw_list): if not raw_list: logger.warning(接口返回空列表) return False record_count len(raw_list) if record_count 3: 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 # ---------- 复用涨停、跌停池业务函数来自第7篇 ---------- 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 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 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 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 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) # 聚合统计 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) logger.info( f{today}涨停{emotion_stat[up_count]}家跌停{emotion_stat[down_count]}家 f强势股{qs_stat[qs_count]}家新高占比{qs_stat[nh_ratio]}% ) 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_qs_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_new_high_ratio(start_date, end_date): conn sqlite3.connect(DB_PATH) sql SELECT trade_date,qs_count,nh_count,nh_ratio 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[nh_ratio], color#e63946, markero, label强势股新高占比(%)) ax.set_title(强势股池新高占比时序, fontsize14) ax.set_xlabel(交易日) ax.set_ylabel(占比 %) ax.legend() ax.grid(alpha0.3) plt.xticks(rotation45) plt.tight_layout() plt.savefig(new_high_ratio.png,dpi200) plt.show() # plot_new_high_ratio(2026‑07‑01,2026‑08‑25)六、业务关键点强势股接口只返回个股明细强势股家数、新高占比全部代码聚合计算接口无聚合结果输出。数据表设置trade_datestock_code联合唯一约束重复执行脚本不会重复入库。统计数据写入market_emotion_daily汇总表做时序分析时不用扫描全量表提升查询性能。可以做多池联合查询筛选同时出现在涨停池与强势股池的标的。七、拓展练习方向SQL 多表关联查询筛选当日同时属于涨停池、强势股池的标的集合增加量比筛选逻辑统计高量比强势股的特征编写批量回捞脚本拉取历史强势股池数据积累长周期样本。下篇预告系列第 9 篇次新股池数据实战开板日期、上市周期数据分析解析次新股池原始接口字段处理yyyyMMdd格式的开板、上市日期完成数据表设计、清洗入库统计次新板块开板分布更新定时采集流水线。免责申明文中所有数据处理逻辑仅为编程演示仅为数据演示不构成投资建议。市场有风险投资需谨慎。