尧图网页设计 
量化数据开发实战系列第 9 篇次新股池数据实战开板日期、上市周期数据分析前言第 7‑8 篇已经完成涨停、跌停、强势股三大股池的采集、清洗、入库与情绪统计。本篇接入次新股池接口。次新股池包含标的的上市日期、开板日期、开板几日等原始字段。接口返回的开板日期、上市日期为yyyyMMdd数字字符串格式和库内统一的yyyy‑MM‑dd日期格式不一致需要业务代码做格式转换。接口仅输出单只次新股明细次新股总数量、已开板占比、平均开板天数这类聚合指标全部需要代码运算得到。本篇完成次新股池完整数据流水线更新定时采集任务。一、接口原始字段梳理表格字段说明dm股票代码mc股票名称p价格元 ¥ztp涨停价元 ¥无涨停价为 nullzf涨跌幅%cje成交额元 ¥lt流通市值元 ¥zsz总市值元 ¥nh是否新高0否1是hs转手率%tj涨停统计x 天 /y 板kb开板几日od开板日期格式yyyyMMddipod上市日期格式yyyyMMdd二、自研衍生统计指标次新股总数量当日次新股池标的总条数已开板数量kb 0已经打开一字板的个股未开板数量kb 0仍处于一字板阶段平均开板天数已开板标的kb字段算术平均值全部聚合统计需要业务代码计算接口不会直接返回统计结果。三、数据表设计在quant.db新增cxgc_pool次新股池明细表对market_emotion_daily情绪汇总表新增次新股相关统计字段。cxgc_pool 次新股池明细表表格字段类型说明idINTEGER自增主键trade_dateTEXT交易日期 yyyy‑MM‑ddstock_codeTEXT股票代码stock_nameTEXT股票名称priceREAL现价 ¥zt_priceREAL涨停价 ¥zfREAL涨跌幅 %cje_yiREAL成交额 亿元 ¥ltsz_yiREAL流通市值 亿元 ¥zsz_yiREAL总市值 亿元 ¥is_new_highINTEGER是否新高 1 是 0 否hsREAL换手率 %stat_infoTEXT涨停统计字符串kb_daysINTEGER开板几日open_board_dateTEXT开板日期 yyyy‑MM‑dd转换后ipo_dateTEXT上市日期 yyyy‑MM‑dd转换后UNIQUE(trade_date,stock_code)联合唯一约束market_emotion_daily 新增字段cx_total INTEGER, cx_opened INTEGER, cx_not_open INTEGER, avg_kb_days REAL四、完整可运行代码代码调用复用前面章节请求、日志、交易日历、涨跌停、强势股池相关逻辑新增次新股池全流程处理改造每日采集任务。import requests import logging import time import pandas as pd import numpy as np import sqlite3 from datetime import datetime 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_cx_table(db_namequant.db): conn sqlite3.connect(db_name) cur conn.cursor() create_cx_sql CREATE TABLE IF NOT EXISTS cxgc_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, is_new_high INTEGER, hs REAL, stat_info TEXT, kb_days INTEGER, open_board_date TEXT, ipo_date TEXT, UNIQUE(trade_date, stock_code) ) cur.execute(create_cx_sql) # 情绪表追加次新统计字段已存在则跳过 try: cur.execute(ALTER TABLE market_emotion_daily ADD COLUMN cx_total INTEGER) except sqlite3.OperationalError: pass try: cur.execute(ALTER TABLE market_emotion_daily ADD COLUMN cx_opened INTEGER) except sqlite3.OperationalError: pass try: cur.execute(ALTER TABLE market_emotion_daily ADD COLUMN cx_not_open INTEGER) except sqlite3.OperationalError: pass try: cur.execute(ALTER TABLE market_emotion_daily ADD COLUMN avg_kb_days REAL) except sqlite3.OperationalError: pass conn.commit() conn.close() logger.info(次新股池数据表初始化完成) 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_cx_data(raw_json_list, trade_date): 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 # yyyyMMdd 转为 yyyy‑MM‑dd 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_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 },inplaceTrue) write_df.to_sql(cxgc_pool, conn, if_existsappend, indexFalse) conn.close() logger.info(f次新股池入库完成共{len(df)}条记录) 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() 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 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 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 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 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) # 聚合统计 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) logger.info( f{today}涨停{emotion_stat[up_count]}家跌停{emotion_stat[down_count]}家 f强势股{qs_stat[qs_count]}家次新股{cx_stat[cx_total]}家平均开板天数{cx_stat[avg_kb_days]} ) 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_cx_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_cx_kb_trend(start_date, end_date): conn sqlite3.connect(DB_PATH) sql SELECT trade_date,cx_total,avg_kb_days 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[cx_total],color#81b29a,alpha0.6,label次新股总家数) ax2.plot(df[trade_date],df[avg_kb_days],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(cx_stock_trend.png,dpi200) plt.show() # plot_cx_kb_trend(2026‑07‑01,2026‑08‑25)六、业务关键点接口返回od开板日期、ipod上市日期为yyyyMMdd数字格式代码必须转换为yyyy‑MM‑dd标准格式才能和库内其他日期字段做对比次新股总家数、已开板数量、平均开板天数均为业务代码聚合计算接口只返回个股明细设置trade_datestock_code联合唯一约束脚本重复运行不会重复入库统计结果写入market_emotion_daily汇总表时序分析不用扫描全量表提升查询性能。七、拓展练习方向SQL 多表关联查询筛选同时存在于次新股池 涨停池的标的统计已开板次新股的换手率、涨跌幅分布编写批量回捞脚本获取历史次新股数据积累长周期样本。下篇预告系列第 10 篇炸板股池实战提取炸板时间、炸板次数统计炸板行为特征接入炸板股池接口完成清洗入库基于首次封板时间区分早盘、午后炸板自研炸板相关统计指标和涨停池做对比分析。免责申明文中所有数据处理逻辑仅为编程演示仅为数据演示不构成投资建议。市场有风险投资需谨慎。 量化数据开发实战系列（第 9 篇）：次新股池数据实战：开板日期、上市周期数据分析
量化数据开发实战系列第 9 篇次新股池数据实战开板日期、上市周期数据分析前言第 7‑8 篇已经完成涨停、跌停、强势股三大股池的采集、清洗、入库与情绪统计。本篇接入次新股池接口。次新股池包含标的的上市日期、开板日期、开板几日等原始字段。接口返回的开板日期、上市日期为yyyyMMdd数字字符串格式和库内统一的yyyy‑MM‑dd日期格式不一致需要业务代码做格式转换。接口仅输出单只次新股明细次新股总数量、已开板占比、平均开板天数这类聚合指标全部需要代码运算得到。本篇完成次新股池完整数据流水线更新定时采集任务。一、接口原始字段梳理表格字段说明dm股票代码mc股票名称p价格元 ¥ztp涨停价元 ¥无涨停价为 nullzf涨跌幅%cje成交额元 ¥lt流通市值元 ¥zsz总市值元 ¥nh是否新高0否1是hs转手率%tj涨停统计x 天 /y 板kb开板几日od开板日期格式yyyyMMddipod上市日期格式yyyyMMdd二、自研衍生统计指标次新股总数量当日次新股池标的总条数已开板数量kb 0已经打开一字板的个股未开板数量kb 0仍处于一字板阶段平均开板天数已开板标的kb字段算术平均值全部聚合统计需要业务代码计算接口不会直接返回统计结果。三、数据表设计在quant.db新增cxgc_pool次新股池明细表对market_emotion_daily情绪汇总表新增次新股相关统计字段。cxgc_pool 次新股池明细表表格字段类型说明idINTEGER自增主键trade_dateTEXT交易日期 yyyy‑MM‑ddstock_codeTEXT股票代码stock_nameTEXT股票名称priceREAL现价 ¥zt_priceREAL涨停价 ¥zfREAL涨跌幅 %cje_yiREAL成交额 亿元 ¥ltsz_yiREAL流通市值 亿元 ¥zsz_yiREAL总市值 亿元 ¥is_new_highINTEGER是否新高 1 是 0 否hsREAL换手率 %stat_infoTEXT涨停统计字符串kb_daysINTEGER开板几日open_board_dateTEXT开板日期 yyyy‑MM‑dd转换后ipo_dateTEXT上市日期 yyyy‑MM‑dd转换后UNIQUE(trade_date,stock_code)联合唯一约束market_emotion_daily 新增字段cx_total INTEGER, cx_opened INTEGER, cx_not_open INTEGER, avg_kb_days REAL四、完整可运行代码代码调用复用前面章节请求、日志、交易日历、涨跌停、强势股池相关逻辑新增次新股池全流程处理改造每日采集任务。import requests import logging import time import pandas as pd import numpy as np import sqlite3 from datetime import datetime 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_cx_table(db_namequant.db): conn sqlite3.connect(db_name) cur conn.cursor() create_cx_sql CREATE TABLE IF NOT EXISTS cxgc_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, is_new_high INTEGER, hs REAL, stat_info TEXT, kb_days INTEGER, open_board_date TEXT, ipo_date TEXT, UNIQUE(trade_date, stock_code) ) cur.execute(create_cx_sql) # 情绪表追加次新统计字段已存在则跳过 try: cur.execute(ALTER TABLE market_emotion_daily ADD COLUMN cx_total INTEGER) except sqlite3.OperationalError: pass try: cur.execute(ALTER TABLE market_emotion_daily ADD COLUMN cx_opened INTEGER) except sqlite3.OperationalError: pass try: cur.execute(ALTER TABLE market_emotion_daily ADD COLUMN cx_not_open INTEGER) except sqlite3.OperationalError: pass try: cur.execute(ALTER TABLE market_emotion_daily ADD COLUMN avg_kb_days REAL) except sqlite3.OperationalError: pass conn.commit() conn.close() logger.info(次新股池数据表初始化完成) 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_cx_data(raw_json_list, trade_date): 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 # yyyyMMdd 转为 yyyy‑MM‑dd 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_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 },inplaceTrue) write_df.to_sql(cxgc_pool, conn, if_existsappend, indexFalse) conn.close() logger.info(f次新股池入库完成共{len(df)}条记录) 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() 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 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 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 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 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) # 聚合统计 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) logger.info( f{today}涨停{emotion_stat[up_count]}家跌停{emotion_stat[down_count]}家 f强势股{qs_stat[qs_count]}家次新股{cx_stat[cx_total]}家平均开板天数{cx_stat[avg_kb_days]} ) 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_cx_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_cx_kb_trend(start_date, end_date): conn sqlite3.connect(DB_PATH) sql SELECT trade_date,cx_total,avg_kb_days 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[cx_total],color#81b29a,alpha0.6,label次新股总家数) ax2.plot(df[trade_date],df[avg_kb_days],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(cx_stock_trend.png,dpi200) plt.show() # plot_cx_kb_trend(2026‑07‑01,2026‑08‑25)六、业务关键点接口返回od开板日期、ipod上市日期为yyyyMMdd数字格式代码必须转换为yyyy‑MM‑dd标准格式才能和库内其他日期字段做对比次新股总家数、已开板数量、平均开板天数均为业务代码聚合计算接口只返回个股明细设置trade_datestock_code联合唯一约束脚本重复运行不会重复入库统计结果写入market_emotion_daily汇总表时序分析不用扫描全量表提升查询性能。七、拓展练习方向SQL 多表关联查询筛选同时存在于次新股池 涨停池的标的统计已开板次新股的换手率、涨跌幅分布编写批量回捞脚本获取历史次新股数据积累长周期样本。下篇预告系列第 10 篇炸板股池实战提取炸板时间、炸板次数统计炸板行为特征接入炸板股池接口完成清洗入库基于首次封板时间区分早盘、午后炸板自研炸板相关统计指标和涨停池做对比分析。免责申明文中所有数据处理逻辑仅为编程演示仅为数据演示不构成投资建议。市场有风险投资需谨慎。