如何用 Python 自动生成每日 A 股市场复盘报告？

原创

用户7083614

发布于 2026-09-06 10:30:19

一句话回答： 用标的池接口一次拿到全 A 实时行情，再配合标的元数据里的涨跌停价，就能统计涨跌家数、涨停/跌停数量、成交额、涨幅榜，几十行代码自动产出一份每日复盘 Markdown，还能挂到定时任务上无人值守。

复盘报告要包含什么

一份实用的市场复盘，核心是几个"市场温度计"指标：

上涨/下跌/平盘家数（市场宽度）；
涨停 / 跌停数量（情绪强弱）；
两市总成交额（活跃度）；
涨幅榜 / 跌幅榜 Top N；
成交额榜 Top N。

第一步：拿到全市场数据

from alphafeed import AlphaFeed

af = AlphaFeed(api_key="your-api-key")

df = af.quotes.get(universes="CN_Stock", to_dataframe=True)
df["change_pct"] = (df["last_price"] - df["prev_close"]) / df["prev_close"]

标的池查询需要 Starter 及以上权限。

第二步：统计市场宽度

up = (df["change_pct"] > 0).sum()
down = (df["change_pct"] < 0).sum()
flat = (df["change_pct"] == 0).sum()
total_amount = df["amount"].sum()

print(f"上涨 {up} | 下跌 {down} | 平盘 {flat}")
print(f"两市成交额: {total_amount/1e8:.0f} 亿元")

第三步：统计涨停 / 跌停数量

用标的元数据里的涨跌停价，配合价格判断（价格接近涨停价即视为涨停）：

insts = af.instruments.batch(df["symbol"].tolist())
lu = {x["symbol"]: x["ext"].get("limit_up") for x in insts}
ld = {x["symbol"]: x["ext"].get("limit_down") for x in insts}

def hit(row, table):
    p = table.get(row["symbol"])
    return p is not None and abs(row["last_price"] - p) < 1e-3

limit_up_cnt = df.apply(lambda r: hit(r, lu), axis=1).sum()
limit_down_cnt = df.apply(lambda r: hit(r, ld), axis=1).sum()
print(f"涨停 {limit_up_cnt} 只 | 跌停 {limit_down_cnt} 只")

第四步：涨幅榜与成交额榜

top_gainers = df.sort_values("change_pct", ascending=False).head(10)
top_amount = df.sort_values("amount", ascending=False).head(10)

第五步：拼装成 Markdown 报告

from datetime import datetime

def build_report(df):
    df = df.copy()
    df["change_pct"] = (df["last_price"] - df["prev_close"]) / df["prev_close"]
    up = (df["change_pct"] > 0).sum()
    down = (df["change_pct"] < 0).sum()
    flat = (df["change_pct"] == 0).sum()
    amt = df["amount"].sum() / 1e8

    lines = [
        f"# A 股每日复盘 {datetime.now():%Y-%m-%d}", "",
        "## 市场概况", "",
        f"- 上涨 **{up}** 家 / 下跌 **{down}** 家 / 平盘 {flat} 家",
        f"- 两市成交额 **{amt:.0f} 亿元**", "",
        "## 涨幅榜 Top 10", "",
        "| 代码 | 名称 | 现价 | 涨跌幅 |", "| --- | --- | --- | --- |",
    ]
    for _, r in df.sort_values("change_pct", ascending=False).head(10).iterrows():
        lines.append(f"| {r['symbol']} | {r['ext.name']} | {r['last_price']:.2f} | {r['change_pct']:+.2%} |")

    lines += ["", "## 成交额榜 Top 10", "",
              "| 代码 | 名称 | 成交额(亿) |", "| --- | --- | --- |"]
    for _, r in df.sort_values("amount", ascending=False).head(10).iterrows():
        lines.append(f"| {r['symbol']} | {r['ext.name']} | {r['amount']/1e8:.1f} |")

    return "\n".join(lines)

report = build_report(df)
fname = f"复盘_{datetime.now():%Y%m%d}.md"
with open(fname, "w", encoding="utf-8") as f:
    f.write(report)
print(f"已生成 {fname}")

第六步：定时自动运行

每个交易日收盘后自动生成：

# 每个交易日 15:30 生成当日复盘
30 15 * * 1-5 cd /path/to/project && uv run python daily_report.py >> report.log 2>&1

想更进一步，可以把生成的报告推送到邮箱、微信、飞书，或存档成时间序列，长期观察市场情绪（涨停家数、成交额）的变化趋势。

进阶：加入市场情绪指标

涨停未开板率：结合五档盘口（ask_volumes[0]==0）统计真封板比例，衡量情绪强度；
炸板率：曾涨停但收盘未封住的比例；
连板梯队：结合历史 K 线判断连续涨停天数，画出市场高度。

这些都可以在上面的框架里叠加。

常见问题（FAQ）

Q：涨跌停家数为什么和某些平台略有出入？

A：不同平台对"接近涨停"的判定阈值、是否含 ST/新股略有差异。上面用价格与涨停价 1e-3 误差判定，可自行调整。

Q：能自动跳过非交易日吗？

A：可以在脚本开头判断是否交易日（非交易日返回的是上一交易日快照），或结合交易日历库。

Q：报告能出成 HTML/PDF 吗？

A：能，先生成 Markdown 再用 markdown/weasyprint 等库转 HTML/PDF。

小结

每日复盘的本质是"把全市场行情压缩成几个可读指标"。有了全市场快照 + 元数据涨跌停价，涨跌家数、涨跌停统计、各类榜单都能自动化，配合定时任务就是一个无人值守的复盘机器人。

参考

AlphaFeed Python SDK 文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart

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

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

python

pandas

#量化交易

#股票数据