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

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

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

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

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

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

## 1. 复盘报告包含什么

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

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

下面一块一块实现。

## 2. 获取市场全貌

```python
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. 指数表现

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

```python
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. 自选股表现

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

```python
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. 放量异动检测

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

```python
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. 均线趋势状态

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

```python
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. 风险提示

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

```python
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. 组装完整复盘脚本

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

```python
# 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**

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

**方式二：盘后手动一键跑**

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

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

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

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

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

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

```python
import requests

def send_to_wechat(title: str, content: str, sendkey: str):
    url = f"https://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）：**

```python
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="https://oapi.dingtalk.com/robot/send?access_token=xxx")
```

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

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

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

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

```python
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


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

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

人工智能

编程语言

大数据

Python自动生成股票复盘报告，3分钟获取市场概况、指数表现、自选股排名、放量异动等关键数据。通过AlphaFeed+Python实现自动化分析，每日定时推送至手机，帮助投资者高效复盘，积累市场数据。文章提供完整代码示例，适合量化交易和股票投资人群。

结构化数据

数据库

Python

JSON

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

我用 Python 自动生成每日复盘报告，每天收盘后 3 分钟出结果-腾讯云开发者社区-腾讯云

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

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

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐