量化数据开发实战系列（第 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 强势股池明细表
表格
| 字段 | 类型 | 说明 |
|---|---|---|
| id | INTEGER | 自增主键 |
| trade_date | TEXT | 交易日期 yyyy‑MM‑dd |
| 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 | 是否新高 1 是 0 否 |
| lb | REAL | 量比 |
| hs | REAL | 换手率 % |
| stat_info | TEXT | 涨停统计字符串 |
| 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( 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_qs_table(db_name="quant.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 = f"http://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], errors="coerce") 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_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() logger.info(f"强势股池入库完成，共{len(df)}条记录") 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() 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 = 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 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 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 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 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_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_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", marker="o", label="强势股新高占比(%)") ax.set_title("强势股池新高占比时序", fontsize=14) ax.set_xlabel("交易日") ax.set_ylabel("占比 %") ax.legend() ax.grid(alpha=0.3) plt.xticks(rotation=45) plt.tight_layout() plt.savefig("new_high_ratio.png",dpi=200) plt.show() # plot_new_high_ratio("2026‑07‑01","2026‑08‑25")六、业务关键点
- 强势股接口只返回个股明细，强势股家数、新高占比全部代码聚合计算，接口无聚合结果输出。
- 数据表设置
trade_date+stock_code联合唯一约束，重复执行脚本不会重复入库。 - 统计数据写入
market_emotion_daily汇总表，做时序分析时不用扫描全量表，提升查询性能。 - 可以做多池联合查询，筛选同时出现在涨停池与强势股池的标的。
七、拓展练习方向
- SQL 多表关联查询，筛选当日同时属于涨停池、强势股池的标的集合；
- 增加量比筛选逻辑，统计高量比强势股的特征；
- 编写批量回捞脚本，拉取历史强势股池数据积累长周期样本。
下篇预告
系列第 9 篇：次新股池数据实战：开板日期、上市周期数据分析
解析次新股池原始接口字段，处理
yyyyMMdd格式的开板、上市日期，完成数据表设计、清洗入库；统计次新板块开板分布，更新定时采集流水线。
免责申明：文中所有数据处理逻辑仅为编程演示，仅为数据演示，不构成投资建议。市场有风险，投资需谨慎。
