沪深港通数据实战从北向资金到择时信号的全链路量化第 12 篇综合实战流水线数据调度与北向看板一、前言本季前 11 篇逐个打通了沪深港通 21 个接口与各类自研指标。本篇是收官篇用一个通用落库函数把所有接口统一接入用调度框架按交易日节奏自动抓取并输出一张「北向看板」把分散的表聚合成可读快照。至此《沪深港通数据实战从北向资金到择时信号的全链路量化》12 篇全部完成。二、统一接口清单本季 21 个/ht/nbzj/接口表名接口路径ht_lxgllxglht_bxlsbxls/1ht_nxlsnxls/1ht_bxzlbxzlht_nxzlnxzlht_hgtc / ht_sgtchgtc / sgtcht_ggth / ht_ggtsggth / ggtsht_ahahht_hgts / ht_sgtshgts / sgtsht_hcjd / ht_scjdhcjd / scjdht_bxpm / ht_hgpm / ht_sgpmbxpm/d / hgpm/d / sgpm/dht_hgls / ht_shlshgls / shlsht_ghls / ht_gslsghls / gsls三、自研设计要点动态建表根据接口返回的 JSON key 自动建表无需为每个接口手写列定义天然覆盖全部 21 个接口。统一重试全部走biying_api_get_retry限频失败自动重试。调度 看板用 APScheduler 按盘前/盘后定时抓取不可用时退化为单次顺序执行dashboard()汇总各表行数形成看板。四、数据表设计各接口表由fetch_dynamic按返回字段动态创建核心看板汇总表复用前文ht_composite。五、完整代码importurllib.request,json,sqlite3,time,loggingimportpandasaspd LICENCEYOUR_LICENCE_HEREDB_PATHquant.dbLOG_FILEseason4_article12.logBASEhttps://api.biyingapi.comlogging.basicConfig(filenameLOG_FILE,levellogging.INFO,format%(asctime)s %(levelname)s %(message)s)# 表名 - 接口路径阶段参数已写入路径API_MAP{ht_lxgl:lxgl,ht_bxls:bxls/1,ht_nxls:nxls/1,ht_bxzl:bxzl,ht_nxzl:nxzl,ht_hgtc:hgtc,ht_sgtc:sgtc,ht_ggth:ggth,ht_ggts:ggts,ht_ah:ah,ht_hgts:hgts,ht_sgts:sgts,ht_hcjd:hcjd,ht_scjd:scjd,ht_bxpm:bxpm/d,ht_hgpm:hgpm/d,ht_sgpm:sgpm/d,ht_hgls:hgls,ht_shls:shls,ht_ghls:ghls,ht_gsls:gsls,}defbiying_api_get_retry(url,timeout15,max_retry3):lastNoneforiinrange(max_retry):try:requrllib.request.Request(url,headers{User-Agent:Mozilla/5.0})withurllib.request.urlopen(req,timeouttimeout)asresp:returnjson.loads(resp.read().decode(utf-8))exceptExceptionase:laste logging.warning(retry %d/%d %s %s,i1,max_retry,url,e)time.sleep(1)logging.error(api failed %s %s,url,last)returnNonedef_as_list(data):returndataifisinstance(data,list)else[data]deffetch_dynamic(table,path):通用动态落库按返回 JSON 的 key 自动建表。urlf{BASE}/ht/nbzj/{path}/{LICENCE}databiying_api_get_retry(url)ifnotdata:return0rows_as_list(data)ifnotrows:return0colslist(rows[0].keys())connsqlite3.connect(DB_PATH)coldef,.join(f{c}TEXTforcincols)conn.execute(fCREATE TABLE IF NOT EXISTS{table}({coldef}))ph,.join(?*len(cols))forrinrows:conn.execute(fINSERT INTO{table}VALUES ({ph}),[r.get(c)forcincols])conn.commit();conn.close()returnlen(rows)deffetch_all():total0fortable,pathinAPI_MAP.items():nfetch_dynamic(table,path)totaln logging.info(%s - %d rows,table,n)returntotaldefdashboard():北向看板汇总各表最新状态。connsqlite3.connect(DB_PATH)tables[ht_lxgl,ht_bx_trend,ht_hgtc,ht_sgts,ht_bxpm,ht_hgls,ht_ghls,ht_composite]fortintables:try:npd.read_sql(fSELECT COUNT(*) c FROM{t},conn)[c].iloc[0]print(f{t}:{n}rows)exceptException:print(f{t}: (尚未落库))conn.close()defbuild_pipeline():try:fromapscheduler.schedulers.backgroundimportBackgroundScheduler schedBackgroundScheduler()sched.add_job(fetch_all,cron,hour9,minute30,idfetch_morning)sched.add_job(fetch_all,cron,hour15,minute30,idfetch_close)sched.start()print(APScheduler pipeline started, jobs:,len(sched.get_jobs()))exceptExceptionase:print(APScheduler 不可用退化为单次顺序执行:,e)fetch_all()if__name____main__:nfetch_all()print(本次抓取总行数:,n)dashboard()build_pipeline()六、业务关键点动态建表让新增接口「加一行API_MAP」即可接入是全季最省维护成本的设计。调度框架用cron盘前/盘后各抓一次若运行环境无 APScheduler代码自动退化成单次执行不影响数据落地。dashboard()是轻量看板雏形生产可做 Web 页面或定时推送思路一致。七、拓展练习把API_MAP抽到配置文件实现「改配置不加代码」的热更新。给每个接口加「最后成功时间」字段看板显示各接口新鲜度。把dashboard()输出改成 Markdown / HTML定时邮件推送。八、全系列收官至此《沪深港通数据实战从北向资金到择时信号的全链路量化》12 篇全部完成从资金流向概览第 1 篇、走势深度第 2 篇、成分股与 AH 比价第 3–5 篇、十大成交与周期排名第 6–8 篇、南北长周期历史第 9–10 篇到综合择时第 11 篇与实战流水线第 12 篇形成一条完整的北向/南向数据到择时信号的全链路。免责申明文中所有数据处理逻辑仅为编程演示仅为数据演示不构成投资建议。市场有风险投资需谨慎。