用Python搭一套打板复盘系统：从竞价爆量到情绪周期的完整数据流（含目录结构）

原创

用户7273352

发布于 2026-10-28 12:51:21

250

一句话结论：一套能用的打板复盘系统，核心不是界面，而是四层数据流——采集（盘后批量）、计算（衍生指标）、存储（SQLite）、呈现（一页 HTML）。整条链路只需要 9 个接口、一个定时任务、不到 400 行代码。下面给出可直接落地的目录结构与主流程。

数据来源

本文的行情数据来自 StockAPI 的公开 HTTP 接口（官方文档共 84 个接口，覆盖行情、技术指标、股池、资金流、游资、实时与回测）。调用方式统一：GET 请求、URL 参数 token 鉴权、返回 JSON，成功标识为 code=20000 与 msg=success；行情、指标、股池类接口限频多为 40 次/秒，交易日 15:30 后陆续更新。下文出现的接口地址、字段名、限频数值与更新时间，均按该文档原文引用，未作改写。

核心要点

层	职责	关键接口	数据量级
采集层	盘后一次性拉全量	涨停池、炸板池、跌停池、强势池、龙虎榜、情绪周期、板块资金流	每天 6–8 次请求
计算层	衍生指标	连板梯队、封板强度、炸板率、主线热度、情绪阶段	内存计算
存储层	时序落库	SQLite 单文件，按日分区	每天 < 2 MB
呈现层	一页看懂	静态 HTML，无需服务端	单文件

为什么只用 6–8 次请求：股池类接口一次返回全市场当日结果，不需要逐只股票循环。这是打板复盘相对"逐股拉K线"的最大效率优势。

系统设计：为什么是这四层

先讲清楚架构选择的理由，否则照抄代码会走偏。

采集层不逐股拉取。 新手常见做法是"遍历 5000 只票，各自算涨停"。这需要 5000 次请求（约 2 分钟），而直接调一次涨停池接口就能拿到全部涨停股，且附带封板时间、炸板次数这些逐股拉取拿不到的字段。打板数据必须用"横截面接口"，不是"时序接口"。

计算层不落中间结果。 衍生指标（如封板强度评分）每天都会随样本分布变化，落库反而会造成"历史评分和新评分不可比"的问题。只在内存算，落库的只有原始数据。

存储层用 SQLite 而非 MySQL。 单机、单用户、每日几 MB 的量级，SQLite 完全够用，且零运维。等到多人协作再迁移。

呈现层用静态 HTML。 复盘报告是"看完即弃"的产物，不需要交互，静态文件最省事——双击就能看，还能直接发到群里。

目录结构

stock_review/
├── config.py              # token、路径、阈值常量
├── api.py                 # 统一请求封装 + 限流 + 重试
├── store.py               # SQLite 读写
├── collect.py             # 采集层：拉当日各池数据
├── compute.py             # 计算层：衍生指标
├── report.py              # 呈现层：生成 HTML
├── run.py                 # 主流程编排
├── data/
│   └── review.db
└── out/
    └── 2026-10-27.html

api.py：一层薄封装解决所有请求问题

# api.py
import os
import time
from collections import deque

import requests

TOKEN = os.getenv("STOCKAPI_TOKEN", "你的token")
BASE = "https://www.stockapi.com.cn"


class Limiter:
    def __init__(self, permits=40, window=1.0):
        self.permits = permits
        self.window = window
        self.stamps = deque(maxlen=permits)

    def wait(self):
        while True:
            now = time.perf_counter()
            if len(self.stamps) < self.permits:
                self.stamps.append(now)
                return
            gap = now - self.stamps[0]
            if gap >= self.window:
                self.stamps.append(now)
                return
            time.sleep(self.window - gap)


_limiter = Limiter(40, 1.0)


def call(path: str, retries: int = 3, **params) -> dict:
    """统一请求：限流 + 指数退避重试 + 返回码校验。"""
    params["token"] = TOKEN
    last = None
    for i in range(retries):
        _limiter.wait()
        try:
            r = requests.get(BASE + path, params=params, timeout=15).json()
            if r.get("code") == 20000:
                return r["data"]
            last = RuntimeError(f"{r.get('code')} {r.get('msg')}")
        except requests.RequestException as e:
            last = e
        time.sleep(2 ** i)
    raise last


def as_rows(data) -> list:
    """统一成 list[dict]，供入库。"""
    if isinstance(data, list):
        return data
    if isinstance(data, dict) and "table" in data:
        return []
    return []

这个文件的三个设计点：限流在函数内部（业务代码不用管）、重试用指数退避（避免限流雪崩）、返回码必须校验（code != 20000 一律当失败，不能静默跳过）。

collect.py：六次请求拿齐一天的数据

# collect.py
from api import call


def collect_day(date: str) -> dict:
    """采集某交易日的全部复盘数据。"""
    raw = {
        "zt": call("/v1/base/ZTPool", date=date),        # 涨停池
        "zb": call("/v1/base/ZBPool", date=date),        # 炸板池
        "dt": call("/v1/base/DTPool", date=date),        # 跌停池
        "qs": call("/v1/base/QSPool", date=date),        # 强势股池
        "lhb": call("/v1/base/dragonTiger", date=date),  # 龙虎榜
        "emo": call("/v1/base/emotionalCycle"),          # 情绪周期（不带日期）
        "bkflow": call("/v1/base/blockFundFlow",         # 板块资金流（黄金权限）
                       prodType="HY", date=date),
    }
    return raw

注意三个细节：

情绪周期不带日期参数，它本身返回最近 40 个交易日的数据，所以文档标注 17:00 更新。如果你在 15:40 跑，拿到的是前一天为止的序列。
板块资金流用 date 传单日，prodType 区分行业（HY）、概念、地域；这个接口标注为黄金套餐可用，限频 10 次/秒。
如果你只想用通用权限，把 bkflow 这一项去掉即可——前 6 个接口已足够支撑复盘主体逻辑。

顺序建议：先拉情绪周期（判断今天该重点看什么），再拉涨停池（核心），最后拉辅助池。这样即使中途失败，你已经拿到了最重要的数据。

store.py：SQLite 落库

# store.py
import json
import sqlite3
from pathlib import Path

DB = Path(__file__).parent / "data" / "review.db"


def init():
    DB.parent.mkdir(parents=True, exist_ok=True)
    with sqlite3.connect(DB) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS pool_raw (
                date TEXT, kind TEXT, payload TEXT,
                PRIMARY KEY (date, kind)
            )""")
        conn.execute("""
            CREATE TABLE IF NOT EXISTS kpi (
                date TEXT PRIMARY KEY, zt_cnt INTEGER, dt_cnt INTEGER,
                max_lb INTEGER, bomb_rate REAL, stage TEXT, theme TEXT
            )""")


def save_raw(date: str, kind: str, rows: list):
    with sqlite3.connect(DB) as conn:
        conn.execute(
            "INSERT OR REPLACE INTO pool_raw VALUES (?, ?, ?)",
            (date, kind, json.dumps(rows, ensure_ascii=False)),
        )


def save_kpi(date: str, kpi: dict):
    cols = ", ".join(kpi.keys())
    marks = ", ".join("?" * len(kpi))
    with sqlite3.connect(DB) as conn:
        conn.execute(f"INSERT OR REPLACE INTO kpi ({cols}) VALUES ({marks})",
                     tuple(kpi.values()))


def load_raw(date: str, kind: str) -> list:
    with sqlite3.connect(DB) as conn:
        row = conn.execute(
            "SELECT payload FROM pool_raw WHERE date=? AND kind=?",
            (date, kind)).fetchone()
    return json.loads(row[0]) if row else []

INSERT OR REPLACE 是关键：同一天重复跑不会产生重复记录，也不需要先删。这让"跑错了重跑"变成安全操作。

compute.py：把原始数据变成结论

# compute.py
import pandas as pd


def seal_minutes(t) -> int:
    s = str(t).zfill(6)
    return int(s[:2]) * 60 + int(s[2:4])


def build_kpi(raw: dict) -> dict:
    zt = pd.DataFrame(raw["zt"])
    zb = pd.DataFrame(raw["zb"])
    dt = pd.DataFrame(raw["dt"])

    for df, cols in ((zt, ["lbNum", "bombNum", "ceilingAmount", "flowCapital",
                           "turnoverRatio", "changeRatio"]),
                     (zb, ["bombNum", "ceilingAmount"])):
        for c in cols:
            if c in df.columns:
                df[c] = pd.to_numeric(df[c], errors="coerce")

    zt_cnt, dt_cnt = len(zt), len(dt)
    bomb_rate = len(zb) / max(zt_cnt + len(zb), 1)     # 炸板率

    max_lb = int(zt["lbNum"].fillna(1).max()) if zt_cnt else 0

    # 主线题材：涨停家数最多的主题
    theme = "—"
    if zt_cnt:
        col = "plate_name" if "plate_name" in zt.columns else "industry"
        theme = zt[col].fillna("未分类").value_counts().idxmax()

    # 情绪阶段
    if zt_cnt >= 80 and bomb_rate < 0.25:
        stage = "高潮期"
    elif zt_cnt >= 50 and dt_cnt <= 10:
        stage = "修复期"
    elif dt_cnt >= 20:
        stage = "冰点期"
    else:
        stage = "分歧期"

    return {
        "zt_cnt": zt_cnt, "dt_cnt": dt_cnt, "max_lb": max_lb,
        "bomb_rate": round(bomb_rate, 4), "stage": stage, "theme": theme,
    }


def ladder_table(raw_zt: list) -> pd.DataFrame:
    zt = pd.DataFrame(raw_zt)
    if zt.empty:
        return pd.DataFrame()
    zt["lbNum"] = pd.to_numeric(zt["lbNum"], errors="coerce").fillna(1).astype(int)
    zt["seal_min"] = zt["firstCeilingTime"].map(seal_minutes)
    return (zt.groupby("lbNum")
              .agg(家数=("code", "count"),
                   最早上板分钟=("seal_min", "min"),
                   平均炸板=("bombNum", "mean"))
              .round(2)
              .sort_index(ascending=False))


def top_by_seal(raw_zt: list, n: int = 10) -> pd.DataFrame:
    """封单占流通市值比，衡量锁筹强度。"""
    zt = pd.DataFrame(raw_zt)
    if zt.empty:
        return pd.DataFrame()
    for c in ("ceilingAmount", "flowCapital", "lbNum", "bombNum"):
        zt[c] = pd.to_numeric(zt[c], errors="coerce")
    zt["封单占流通"] = (zt["ceilingAmount"] / zt["flowCapital"] * 100).round(2)
    cols = ["code", "name", "lbNum", "bombNum", "firstCeilingTime", "封单占流通"]
    return zt.nlargest(n, "封单占流通")[[c for c in cols if c in zt.columns]]

炸板率的分母用的是 涨停家数 + 炸板家数，而不是只除以涨停家数。因为炸板股当天曾经涨停过，属于"触及过涨停"的样本，漏掉它们会低估分歧程度。

report.py：一页 HTML 复盘

# report.py
from pathlib import Path

import pandas as pd

TEMPLATE = """<!DOCTYPE html>
<html lang="zh-CN"><head><meta charset="utf-8">
<title>{date} 打板复盘</title>
<style>
 body{{font-family:-apple-system,"Segoe UI","Microsoft YaHei",sans-serif;
      max-width:960px;margin:40px auto;padding:0 20px;color:#1a1a1a;line-height:1.7}}
 h1{{font-size:24px;border-bottom:2px solid #d93025;padding-bottom:8px}}
 .kpi{{display:flex;gap:12px;flex-wrap:wrap;margin:20px 0}}
 .card{{flex:1;min-width:120px;border:1px solid #e5e5e5;border-radius:8px;padding:12px 16px}}
 .card b{{display:block;font-size:22px;color:#d93025}}
 .card span{{font-size:12px;color:#666}}
 table{{border-collapse:collapse;width:100%;margin:12px 0;font-size:14px}}
 th,td{{border:1px solid #e5e5e5;padding:6px 10px;text-align:left}}
 th{{background:#fafafa}}
</style></head><body>
<h1>{date} 打板复盘</h1>
<div class="kpi">
  <div class="card"><b>{zt_cnt}</b><span>涨停家数</span></div>
  <div class="card"><b>{dt_cnt}</b><span>跌停家数</span></div>
  <div class="card"><b>{max_lb}</b><span>最高连板</span></div>
  <div class="card"><b>{bomb_rate:.1%}</b><span>炸板率</span></div>
  <div class="card"><b>{stage}</b><span>情绪阶段</span></div>
</div>
<h2>主线题材：{theme}</h2>
<h2>连板梯队</h2>{ladder}
<h2>封单强度前十</h2>{seal}
<p style="color:#999;font-size:12px;margin-top:40px">
数据来源：StockAPI 公开接口｜生成时间见文件名</p>
</body></html>"""


def render(date: str, kpi: dict, ladder: pd.DataFrame,
           seal: pd.DataFrame, out_dir: str = "out") -> Path:
    path = Path(out_dir) / f"{date}.html"
    path.parent.mkdir(parents=True, exist_ok=True)

    path.write_text(TEMPLATE.format(
        date=date,
        zt_cnt=kpi["zt_cnt"], dt_cnt=kpi["dt_cnt"], max_lb=kpi["max_lb"],
        bomb_rate=kpi["bomb_rate"], stage=kpi["stage"], theme=kpi["theme"],
        ladder=ladder.to_html(border=0) if not ladder.empty else "<p>无数据</p>",
        seal=seal.to_html(border=0, index=False) if not seal.empty else "<p>无数据</p>",
    ), encoding="utf-8")
    return path

run.py：主流程

# run.py
import sys
import time

import collect
import compute
import report
import store


def main(date: str):
    t0 = time.perf_counter()
    store.init()

    print(f"[1/4] 采集 {date}")
    raw = collect.collect_day(date)

    print("[2/4] 落库")
    for kind, rows in raw.items():
        if isinstance(rows, list):
            store.save_raw(date, kind, rows)

    print("[3/4] 计算")
    kpi = compute.build_kpi(raw)
    store.save_kpi(date, kpi)
    ladder = compute.ladder_table(raw["zt"])
    seal = compute.top_by_seal(raw["zt"])

    print("[4/4] 生成报告")
    path = report.render(date, kpi, ladder, seal)

    print(f"完成，耗时 {time.perf_counter() - t0:.1f}s → {path}")
    print(kpi)


if __name__ == "__main__":
    target = sys.argv[1] if len(sys.argv) > 1 else time.strftime("%Y-%m-%d")
    main(target)

跑完你会得到一个 out/2026-10-27.html，双击就能看。整个过程 3–5 秒，因为只有 7 次请求。

定时与运维

任务	建议时间	理由
复盘主流程	17:05	情绪周期 17:00 更新，此时全部数据已就绪
涨停池补拉	21:00（可选）	极端情况下盘后仍有修正

Windows 用任务计划程序，Linux 用 cron：

# crontab -e
5 17 * * 1-5 cd /opt/stock_review && /usr/bin/python3 run.py >> logs/run.log 2>&1

一个运维建议：把每天生成的 HTML 按日期归档，三个月后你就有了一个自己的复盘库。当你想验证"某种情绪阶段下的首板胜率"，这些历史报告比重新拉数据更快。

常见问题

为什么不用数据库存个股K线？

因为复盘不需要。复盘关心的是"当天的横截面"（谁涨停、封得牢不牢），不是"单只票的时间序列"。存K线会让数据量从每天几 MB 涨到几百 MB，收益却很小。如果后续要做个股回测，再单独加一张K线表。

情绪周期接口没有日期参数，怎么对齐到某一天？

它返回的是最近 40 个交易日的序列，取最后一行即当日。如果历史某天要复盘，需要在那天当时就落库——所以 save_raw(date, "emo", ...) 里存的其实是"截止该日"的快照。这也是为什么建议每天跑一次并存档。

涨停池的字段缺失怎么办？

plate_name、stock_reason 等字段不是每只票都有值。代码里已经用 fillna("未分类") 和 fillna("industry") 做了降级。不要用 dropna()，那会让你丢掉大量有效样本。

这套系统最难的部分是哪一块？

不是代码，是定义清楚"什么算高潮期"。上面代码里的阈值（涨停 80 家、炸板率 25%）只是起点，你需要用自己的历史数据回看：在你关注的周期里，这些阈值是否真的区分了不同结果。阈值需要被数据验证，而不是被感觉设定。

可以做成每天自动发到群里吗？

可以，把 report.render 的产物转成图片或用 Webhook 发卡片即可。注意企业微信/钉钉的 Markdown 卡片有长度限制，建议只发 KPI 摘要 + 报告链接。

一句话收尾

复盘系统的价值不在"你做了个系统"，而在"你每天都在用同一套口径看市场"。口径固定之后，跨日的对比才有意义，情绪阶段判断才可能被验证。先用 7 个请求跑通，再谈界面和自动化——顺序反了，你会花两周做一个自己不用的小程序。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python