我用 Python 自动生成每日复盘报告，每天收盘后 3 分钟出结果

炒股的人都知道要复盘。但大多数人的复盘是这样的：收盘后打开行情软件，翻翻涨幅榜，看看自选股涨了还是跌了，刷两条财经新闻，觉得自己"复盘了"。

真正有用的复盘应该回答几个具体问题：

• 今天市场整体涨跌分布是什么样的？涨的多还是跌的多？
• 我的自选股/持仓相比大盘表现如何？
• 哪些票放量了？哪些票创了新高/新低？
• 最近一周/一个月的趋势是什么方向？

这些问题不难回答，但如果每天手动查，会浪费很多时间，而且容易遗漏。

这篇文章教你用 AlphaFeed + Python 写一个自动复盘脚本。每天收盘后跑一次，3 分钟生成一份结构化的复盘报告，保存成文件，长期积累。

1. 复盘报告包含什么

一份有用的每日复盘报告至少要有这几个板块：

板块	内容
市场概况	涨跌家数、涨停跌停数、成交额
指数表现	上证、深证、创业板当日涨跌
自选股表现	你关注的股票今日涨跌、排名
放量异动	成交量异常放大的标的
趋势状态	各标的当前均线方向（多头/空头）
风险提示	连续下跌、逼近止损位的标的

下面一块一块实现。

2. 获取市场全貌

from alphafeed import AlphaFeed

af = AlphaFeed()

all_cn = af.quotes.get(universes="CN_Stock", to_dataframe=True)

all_cn["change_pct"] = (all_cn["last_price"] - all_cn["prev_close"]) / all_cn["prev_close"]

total = len(all_cn)
up = len(all_cn[all_cn["change_pct"] > 0.001])
down = len(all_cn[all_cn["change_pct"] < -0.001])
flat = total - up - down
limit_up = len(all_cn[all_cn["change_pct"] > 0.095])
limit_down = len(all_cn[all_cn["change_pct"] < -0.095])
total_amount = all_cn["amount"].sum()

print("=" * 50)
print("【市场概况】")
print(f"  总标的: {total} 只")
print(f"  上涨: {up} 只 ({up/total:.1%})")
print(f"  下跌: {down} 只 ({down/total:.1%})")
print(f"  平盘: {flat} 只")
print(f"  涨停: {limit_up} 只")
print(f"  跌停: {limit_down} 只")
print(f"  全市场成交额: {total_amount/1e8:.0f} 亿元")

if up > down * 1.5:
    print(f"  → 今日普涨，{up/total:.0%} 的股票收红")
elif down > up * 1.5:
    print(f"  → 今日普跌，{down/total:.0%} 的股票收绿")
else:
    print(f"  → 今日涨跌参半，市场分化")

这段跑完你就知道今天是普涨、普跌还是分化。比打开软件翻涨幅榜快多了。

3. 指数表现

大盘指数是衡量"市场好不好"的锚。我们拉几个关键指数：

from alphafeed import AlphaFeed

af = AlphaFeed()

indices = {
    "上证指数": "000001.SH",
    "深证成指": "399001.SZ",
    "创业板指": "399006.SZ",
    "沪深300": "000300.SH",
    "中证500": "000905.SH",
}

index_symbols = list(indices.values())
quotes = af.quotes.get(symbols=index_symbols, to_dataframe=True)

print("\n【指数表现】")
for name, sym in indices.items():
    row = quotes[quotes["symbol"] == sym]
    if row.empty:
        continue
    price = row["last_price"].values[0]
    prev = row["prev_close"].values[0]
    chg = (price - prev) / prev
    arrow = "↑" if chg > 0 else ("↓" if chg < 0 else "→")
    print(f"  {name}: {price:.2f}  {arrow} {chg:+.2%}")

4. 自选股表现

这是复盘报告最个人化的部分——你自己关注的股票今天表现怎样：

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

my_watchlist = [
    "600519.SH",  # 贵州茅台
    "000001.SZ",  # 平安银行
    "300750.SZ",  # 宁德时代
    "002594.SZ",  # 比亚迪
    "601318.SH",  # 中国平安
    "000858.SZ",  # 五粮液
    "600036.SH",  # 招商银行
    "AAPL.US",    # 苹果
]

quotes = af.quotes.get(symbols=my_watchlist, to_dataframe=True)
quotes["change_pct"] = (quotes["last_price"] - quotes["prev_close"]) / quotes["prev_close"]
quotes = quotes.sort_values("change_pct", ascending=False)

print("\n【自选股表现】")
for _, row in quotes.iterrows():
    sym = row["symbol"]
    price = row["last_price"]
    chg = row["change_pct"]
    amount = row["amount"]
    tag = "🔴" if chg > 0.03 else ("🟢" if chg < -0.03 else "  ")
    print(f"  {tag} {sym:>12s}  {price:>10.2f}  {chg:>+7.2%}  成交额 {amount/1e8:.1f}亿")

把超涨超跌的标出来，盘后一目了然。

5. 放量异动检测

今天成交量异常放大的票值得特别关注——可能有资金在动作：

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

my_watchlist = [
    "600519.SH", "000001.SZ", "300750.SZ", "002594.SZ",
    "601318.SH", "000858.SZ", "600036.SH",
]

klines = af.klines.batch(
    my_watchlist, period="1d", count=20,
    adjust="forward", to_dataframe=True, show_progress=True,
)

today_quotes = af.quotes.get(symbols=my_watchlist, to_dataframe=True)

print("\n【放量异动】")
alerts = []
for sym, kdf in klines.items():
    if len(kdf) < 10:
        continue
    avg_vol = kdf["volume"].mean()
    today_row = today_quotes[today_quotes["symbol"] == sym]
    if today_row.empty:
        continue
    today_vol = today_row["volume"].values[0]
    vol_ratio = today_vol / avg_vol if avg_vol > 0 else 0

    if vol_ratio > 1.8:
        chg = (today_row["last_price"].values[0] - today_row["prev_close"].values[0]) / today_row["prev_close"].values[0]
        alerts.append({
            "symbol": sym,
            "vol_ratio": vol_ratio,
            "change_pct": chg,
        })

if alerts:
    alerts.sort(key=lambda x: x["vol_ratio"], reverse=True)
    for a in alerts:
        direction = "放量上涨" if a["change_pct"] > 0.01 else ("放量下跌" if a["change_pct"] < -0.01 else "放量横盘")
        print(f"  ⚠️  {a['symbol']}  量比={a['vol_ratio']:.1f}x  涨跌={a['change_pct']:+.2%}  → {direction}")
else:
    print("  今日自选股无明显放量")

6. 均线趋势状态

每只自选股当前处于什么趋势？多头排列还是空头排列？

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

my_watchlist = [
    "600519.SH", "000001.SZ", "300750.SZ", "002594.SZ",
    "601318.SH", "000858.SZ", "600036.SH",
]

klines = af.klines.batch(
    my_watchlist, period="1d", count=80,
    adjust="forward", to_dataframe=True, show_progress=True,
)

print("\n【趋势状态】")
for sym, kdf in klines.items():
    kdf = kdf.sort_values("trade_date").reset_index(drop=True)
    if len(kdf) < 60:
        continue

    ma5 = kdf["close"].rolling(5).mean().iloc[-1]
    ma20 = kdf["close"].rolling(20).mean().iloc[-1]
    ma60 = kdf["close"].rolling(60).mean().iloc[-1]
    close = kdf["close"].iloc[-1]

    if ma5 > ma20 > ma60:
        status = "📈 多头排列"
    elif ma5 < ma20 < ma60:
        status = "📉 空头排列"
    elif close > ma20:
        status = "↗️  偏强（价格在20日线上方）"
    else:
        status = "↘️  偏弱（价格在20日线下方）"

    # 距离20日线的偏离度
    deviation = (close - ma20) / ma20 * 100

    print(f"  {sym:>12s}  {status}  偏离MA20: {deviation:+.1f}%")

如果一只票已经从多头排列转成空头排列，但你还在持有，这就是一个需要认真思考的信号。

7. 风险提示

连续下跌的票、已经亏损较大的持仓、接近关键支撑位的标的——这些需要标红提醒：

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

my_watchlist = [
    "600519.SH", "000001.SZ", "300750.SZ", "002594.SZ",
    "601318.SH", "000858.SZ", "600036.SH",
]

klines = af.klines.batch(
    my_watchlist, period="1d", count=30,
    adjust="forward", to_dataframe=True, show_progress=True,
)

print("\n【风险提示】")
has_warning = False

for sym, kdf in klines.items():
    kdf = kdf.sort_values("trade_date").reset_index(drop=True)
    if len(kdf) < 10:
        continue

    warnings = []

    recent_5d = kdf.tail(5)
    down_days = (recent_5d["close"].pct_change().dropna() < -0.005).sum()
    if down_days >= 4:
        warnings.append(f"近5天有{down_days}天下跌")

    ret_5d = kdf["close"].iloc[-1] / kdf["close"].iloc[-6] - 1 if len(kdf) >= 6 else 0
    if ret_5d < -0.08:
        warnings.append(f"近5日跌幅 {ret_5d:.1%}")

    ret_20d = kdf["close"].iloc[-1] / kdf["close"].iloc[-20] - 1 if len(kdf) >= 20 else 0
    if ret_20d < -0.15:
        warnings.append(f"近20日跌幅 {ret_20d:.1%}")

    high_20d = kdf["high"].iloc[-20:].max()
    drawdown = kdf["close"].iloc[-1] / high_20d - 1
    if drawdown < -0.10:
        warnings.append(f"较20日高点回撤 {drawdown:.1%}")

    if warnings:
        has_warning = True
        print(f"  🚨 {sym}: {'; '.join(warnings)}")

if not has_warning:
    print("  ✅ 自选股暂无明显风险信号")

8. 组装完整复盘脚本

把上面的模块拼成一个完整的脚本，收盘后跑一次，输出到文件：

# daily_review.py
import json
from datetime import datetime
import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

MY_WATCHLIST = [
    "600519.SH", "000001.SZ", "300750.SZ", "002594.SZ",
    "601318.SH", "000858.SZ", "600036.SH",
]

INDICES = {
    "上证指数": "000001.SH",
    "深证成指": "399001.SZ",
    "创业板指": "399006.SZ",
}

def generate_report():
    lines = []
    now = datetime.now()
    lines.append(f"# 每日复盘报告 {now.strftime('%Y-%m-%d %H:%M')}")
    lines.append("")

    # === 市场概况 ===
    all_cn = af.quotes.get(universes="CN_Stock", to_dataframe=True)
    all_cn["change_pct"] = (all_cn["last_price"] - all_cn["prev_close"]) / all_cn["prev_close"]

    total = len(all_cn)
    up = len(all_cn[all_cn["change_pct"] > 0.001])
    down = len(all_cn[all_cn["change_pct"] < -0.001])
    limit_up = len(all_cn[all_cn["change_pct"] > 0.095])
    limit_down = len(all_cn[all_cn["change_pct"] < -0.095])
    total_amount = all_cn["amount"].sum()

    lines.append("## 市场概况")
    lines.append(f"- 上涨 {up} 只 ({up/total:.0%}) | 下跌 {down} 只 ({down/total:.0%})")
    lines.append(f"- 涨停 {limit_up} 只 | 跌停 {limit_down} 只")
    lines.append(f"- 全市场成交额: {total_amount/1e8:.0f} 亿")
    lines.append("")

    # === 指数 ===
    idx_quotes = af.quotes.get(symbols=list(INDICES.values()), to_dataframe=True)
    lines.append("## 指数表现")
    for name, sym in INDICES.items():
        row = idx_quotes[idx_quotes["symbol"] == sym]
        if row.empty:
            continue
        chg = (row["last_price"].values[0] - row["prev_close"].values[0]) / row["prev_close"].values[0]
        lines.append(f"- {name}: {row['last_price'].values[0]:.2f} ({chg:+.2%})")
    lines.append("")

    # === 自选股 ===
    wl_quotes = af.quotes.get(symbols=MY_WATCHLIST, to_dataframe=True)
    wl_quotes["change_pct"] = (wl_quotes["last_price"] - wl_quotes["prev_close"]) / wl_quotes["prev_close"]
    wl_quotes = wl_quotes.sort_values("change_pct", ascending=False)

    lines.append("## 自选股表现")
    for _, row in wl_quotes.iterrows():
        lines.append(f"- {row['symbol']}  {row['last_price']:.2f}  {row['change_pct']:+.2%}")
    lines.append("")

    # === 趋势 & 放量 ===
    klines_data = af.klines.batch(
        MY_WATCHLIST, period="1d", count=60,
        adjust="forward", to_dataframe=True,
    )

    lines.append("## 趋势 & 异动")
    for sym, kdf in klines_data.items():
        kdf = kdf.sort_values("trade_date").reset_index(drop=True)
        if len(kdf) < 20:
            continue

        ma5 = kdf["close"].rolling(5).mean().iloc[-1]
        ma20 = kdf["close"].rolling(20).mean().iloc[-1]
        trend = "多头" if ma5 > ma20 else "空头"

        avg_vol = kdf["volume"].iloc[-20:].mean()
        today_vol = kdf["volume"].iloc[-1]
        vol_ratio = today_vol / avg_vol if avg_vol > 0 else 1
        vol_tag = f" ⚠放量{vol_ratio:.1f}x" if vol_ratio > 1.8 else ""

        lines.append(f"- {sym}: {trend}{vol_tag}")
    lines.append("")

    # === 风险提示 ===
    lines.append("## 风险提示")
    risk_count = 0
    for sym, kdf in klines_data.items():
        kdf = kdf.sort_values("trade_date").reset_index(drop=True)
        if len(kdf) < 20:
            continue
        ret_5d = kdf["close"].iloc[-1] / kdf["close"].iloc[-6] - 1
        high_20d = kdf["high"].iloc[-20:].max()
        dd = kdf["close"].iloc[-1] / high_20d - 1
        if ret_5d < -0.08 or dd < -0.10:
            lines.append(f"- 🚨 {sym}: 5日{ret_5d:+.1%}, 回撤{dd:.1%}")
            risk_count += 1
    if risk_count == 0:
        lines.append("- ✅ 暂无明显风险")

    return "\n".join(lines)


report = generate_report()
print(report)

filename = f"review_{datetime.now().strftime('%Y%m%d')}.md"
with open(filename, "w", encoding="utf-8") as f:
    f.write(report)
print(f"\n已保存到 {filename}")

跑一次大概几秒钟，输出一份 Markdown 格式的复盘报告。你可以用任何 Markdown 阅读器打开，也可以直接贴到笔记软件里。

9. 定时自动化

手动跑也行，但既然都写成脚本了，不如让它每天自动跑。

方式一：crontab

# 每个交易日 15:05 自动执行（收盘后 5 分钟，等数据更新）
5 15 * * 1-5 cd /path/to/project && uv run python daily_review.py >> review.log 2>&1

方式二：盘后手动一键跑

给脚本加个别名，收盘后敲一个命令就出报告：

# 加到 ~/.zshrc 或 ~/.bashrc
alias review="cd /path/to/project && uv run python daily_review.py"

以后收盘后在终端打 review 就行。

10. 进阶：把报告推送到手机

生成报告只是第一步。如果能自动推到手机上，就不需要每天打开电脑看了。

推送到微信（通过 Server 酱）：

import requests

def send_to_wechat(title: str, content: str, sendkey: str):
    url = f"//sctapi.ftqq.com/{sendkey}.send"
    data = {"title": title, "desp": content}
    resp = requests.post(url, data=data)
    print(f"推送结果: {resp.json()}")

report = generate_report()
send_to_wechat(
    title=f"复盘 {datetime.now().strftime('%m/%d')}",
    content=report,
    sendkey="your-sendkey-here",
)

Server 酱（https://sct.ftqq.com/）是一个免费的微信推送服务，注册后拿到 SendKey 就能用。

推送到钉钉群（通过 Webhook）：

import requests
import json

def send_to_dingtalk(content: str, webhook_url: str):
    data = {
        "msgtype": "markdown",
        "markdown": {
            "title": "每日复盘",
            "text": content,
        },
    }
    resp = requests.post(webhook_url, json=data)
    print(f"推送结果: {resp.json()}")

report = generate_report()
send_to_dingtalk(report, webhook_url="//oapi.dingtalk.com/robot/send?access_token=xxx")

这样每天 15:05 自动跑脚本，生成报告后推送到手机，你在地铁上就能看完今天的市场复盘。

11. 持续积累：复盘数据库

每天的复盘报告存下来，一个月后你就有了一个"个人市场日记"。

更进一步，可以把关键数据存成结构化的 JSON，方便后续分析：

import json
from datetime import datetime

def save_structured_data(all_cn, wl_quotes, klines_data):
    today = datetime.now().strftime("%Y%m%d")

    data = {
        "date": today,
        "market": {
            "total": len(all_cn),
            "up": int((all_cn["change_pct"] > 0.001).sum()),
            "down": int((all_cn["change_pct"] < -0.001).sum()),
            "limit_up": int((all_cn["change_pct"] > 0.095).sum()),
            "limit_down": int((all_cn["change_pct"] < -0.095).sum()),
            "total_amount": float(all_cn["amount"].sum()),
        },
        "watchlist": wl_quotes[["symbol", "last_price", "change_pct"]].to_dict("records"),
    }

    filename = f"data_{today}.json"
    with open(filename, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    print(f"结构化数据已保存到 {filename}")

积累一个月的数据后，你可以回答这些问题：

• 这个月有几天是普涨？几天是普跌？
• 我自选股的平均涨跌幅 vs 大盘的差距是多少？
• 哪些票在我自选里呆了一个月但一直在跌？该不该剔除？

结语

复盘的价值不在于"做了"，而在于"做了什么"。

手动翻行情软件 10 分钟，能获取的信息很有限。一个自动化的复盘脚本，3 分钟就能告诉你：市场涨跌分布、指数表现、自选股排名、放量异动、趋势状态、风险信号。这些信息用人眼扫一遍，做到心中有数，就足够了。

AlphaFeed 在这里的角色是"数据自来水"——你不需要关心数据从哪来、格式是什么，只需要关心"我想看什么"。全市场行情一个 API 调用、自选股报价一个 API 调用、历史 K 线一个 API 调用。剩下的就是你用 Python 做加减乘除和 if-else。

每天花 3 分钟看一份自动生成的复盘报告，比花 30 分钟漫无目的地刷行情软件有用得多。

参考文献：

• AlphaFeed 官网：https://alphafeed.org/
• Python SDK 快速开始：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart