# 周末花 10 分钟，用 Python 自动复盘上周 A 股行情

周末是复盘的最佳时间。但大多数人的复盘方式是"打开行情软件，翻翻 K 线图，想想下周买什么"——没有结构、没有数据、没有积累。

这篇文章教你写一个周复盘脚本：周末跑一下，自动生成上周的市场概览、板块表现、自选股状态和关键信号，输出一份结构化的复盘报告。

## 复盘看什么

一份有用的周复盘应该回答这几个问题：

1. **上周大盘怎么样？** — 主要指数涨跌、成交额变化
2. **哪些板块最强/最弱？** — 板块轮动方向
3. **我的自选股表现如何？** — 持仓或关注标的的周度表现
4. **有没有值得注意的信号？** — 异动、突破、趋势变化

## 第一步：大盘周度概览

```python
from alphafeed import AlphaFeed
import pandas as pd
import numpy as np
from datetime import date, timedelta

af = AlphaFeed()

# 主要指数 ETF（用 ETF 代替指数，能拿到成交额）
indices = {
    "沪深300": "510300.SH",
    "中证500": "510500.SH",
    "创业板": "159915.SZ",
    "科创50": "588000.SH",
}

print("=" * 60)
print(f"  A 股周复盘报告 | {date.today()}")
print("=" * 60)

print("\n📊 主要指数周度表现")
print("-" * 50)

for name, symbol in indices.items():
    df = af.klines.get(symbol, period="1d", count=10,
                        adjust="forward", to_dataframe=True)
    if len(df) < 5:
        continue

    # 取最近 5 个交易日（一周）
    week = df.tail(5)
    prev_week = df.iloc[-10:-5] if len(df) >= 10 else df.head(5)

    week_open = week["open"].iloc[0]
    week_close = week["close"].iloc[-1]
    week_high = week["high"].max()
    week_low = week["low"].min()
    week_ret = (week_close - week_open) / week_open
    week_amount = week["amount"].sum()
    prev_amount = prev_week["amount"].sum() if len(prev_week) == 5 else week_amount

    amount_change = (week_amount - prev_amount) / prev_amount if prev_amount > 0 else 0

    print(f"  {name:<6}  周涨跌: {week_ret*100:>+6.2f}%  "
          f"最高: {week_high:.3f}  最低: {week_low:.3f}  "
          f"成交额变化: {amount_change*100:>+5.1f}%")
```

## 第二步：全市场板块强弱

```python
# 拉全市场行情，按板块分组看周表现
all_quotes = af.quotes.get(universes="CN_Stock", to_dataframe=True)
all_quotes["change_rate"] = all_quotes["change_rate"].astype(float)
all_quotes["amount"] = all_quotes["amount"].astype(float)
all_quotes["last_price"] = all_quotes["last_price"].astype(float)

# 用代码前缀做简易分板块
def get_board(symbol):
    code = symbol.split(".")[0]
    if code.startswith("688"):
        return "科创板"
    elif code.startswith("300") or code.startswith("301"):
        return "创业板"
    elif code.startswith("002"):
        return "中小板"
    elif code.startswith(("600", "601", "603", "605")):
        return "沪主板"
    elif code.startswith(("000", "001")):
        return "深主板"
    return "其他"

all_quotes["board"] = all_quotes["symbol"].apply(get_board)

# 从每个板块抽取前 20 只（按成交额），批量拉周 K 线看周涨幅
board_symbols = {}
for board in ["沪主板", "深主板", "创业板", "科创板", "中小板"]:
    bdf = all_quotes[all_quotes["board"] == board]
    top = bdf.nlargest(20, "amount")["symbol"].tolist()
    board_symbols[board] = top

all_syms = []
for syms in board_symbols.values():
    all_syms.extend(syms)

dfs = af.klines.batch(list(set(all_syms)), period="1d", count=5,
                       adjust="forward", to_dataframe=True, show_progress=True)

print("\n📈 板块周度表现（按龙头股均值）")
print("-" * 50)

board_results = {}
for board, syms in board_symbols.items():
    rets = []
    for s in syms:
        if s in dfs and len(dfs[s]) >= 2:
            kdf = dfs[s]
            week_ret = (kdf["close"].iloc[-1] - kdf["open"].iloc[0]) / kdf["open"].iloc[0]
            rets.append(week_ret)

    avg_ret = np.mean(rets) if rets else 0
    board_results[board] = avg_ret

for board, ret in sorted(board_results.items(), key=lambda x: x[1], reverse=True):
    bar = "█" * int(max(0, ret * 200))
    neg_bar = "░" * int(max(0, -ret * 200))
    print(f"  {board:<6}  {ret*100:>+6.2f}%  {bar}{neg_bar}")
```

## 第三步：自选股周报

```python
# 你的自选股列表
my_watchlist = [
    ("600519.SH", "贵州茅台"),
    ("000001.SZ", "平安银行"),
    ("300750.SZ", "宁德时代"),
    ("601318.SH", "中国平安"),
    ("002594.SZ", "比亚迪"),
    ("000858.SZ", "五粮液"),
    ("600036.SH", "招商银行"),
    ("601012.SH", "隆基绿能"),
]

symbols = [s for s, _ in my_watchlist]
name_map = {s: n for s, n in my_watchlist}

# 批量拉最近 25 天数据（4 周 + 余量）
dfs = af.klines.batch(symbols, period="1d", count=25,
                       adjust="forward", to_dataframe=True)

print("\n📋 自选股周度表现")
print("-" * 60)

watchlist_summary = []
for symbol in symbols:
    if symbol not in dfs or len(dfs[symbol]) < 5:
        continue

    kdf = dfs[symbol]
    week = kdf.tail(5)
    name = name_map.get(symbol, symbol)

    week_ret = (week["close"].iloc[-1] - week["open"].iloc[0]) / week["open"].iloc[0]
    week_high = week["high"].max()
    week_low = week["low"].min()
    week_vol_avg = week["volume"].mean()

    # 对比上周量能
    prev_week = kdf.iloc[-10:-5] if len(kdf) >= 10 else None
    vol_change = ""
    if prev_week is not None and len(prev_week) == 5:
        prev_vol_avg = prev_week["volume"].mean()
        if prev_vol_avg > 0:
            vc = (week_vol_avg - prev_vol_avg) / prev_vol_avg
            vol_change = f"{vc*100:>+5.0f}%"

    # MA20 位置
    if len(kdf) >= 20:
        ma20 = kdf["close"].rolling(20).mean().iloc[-1]
        above_ma20 = "✅" if week["close"].iloc[-1] > ma20 else "❌"
    else:
        above_ma20 = "—"

    watchlist_summary.append({
        "symbol": symbol, "name": name,
        "week_ret": week_ret, "above_ma20": above_ma20,
        "vol_change": vol_change,
    })

    print(f"  {name:<6} {symbol}  "
          f"周涨跌: {week_ret*100:>+6.2f}%  "
          f"站上MA20: {above_ma20}  "
          f"量能变化: {vol_change}")
```

## 第四步：关键信号检测

```python
print("\n🔔 关键信号")
print("-" * 60)

signals = []

for symbol in symbols:
    if symbol not in dfs or len(dfs[symbol]) < 20:
        continue

    kdf = dfs[symbol]
    name = name_map.get(symbol, symbol)
    close = kdf["close"].values
    volume = kdf["volume"].values
    high = kdf["high"].values

    # 信号 1：周线突破 20 日新高
    high_20 = high[-21:-1].max() if len(high) > 21 else high[:-1].max()
    if close[-1] > high_20:
        signals.append(f"  🔺 {name}({symbol}) 突破 20 日新高")

    # 信号 2：MA5 金叉 MA20（本周发生）
    ma5_arr = pd.Series(close).rolling(5).mean().values
    ma20_arr = pd.Series(close).rolling(20).mean().values
    for i in range(-5, 0):
        if (i-1 >= -len(ma5_arr) and
            ma5_arr[i-1] <= ma20_arr[i-1] and
            ma5_arr[i] > ma20_arr[i]):
            signals.append(f"  ⬆️  {name}({symbol}) 本周 MA5 金叉 MA20")
            break
        if (i-1 >= -len(ma5_arr) and
            ma5_arr[i-1] >= ma20_arr[i-1] and
            ma5_arr[i] < ma20_arr[i]):
            signals.append(f"  ⬇️  {name}({symbol}) 本周 MA5 死叉 MA20")
            break

    # 信号 3：周成交量异常放大
    if len(volume) > 10:
        avg_vol_prev = volume[-15:-5].mean()
        avg_vol_week = volume[-5:].mean()
        if avg_vol_prev > 0 and avg_vol_week > avg_vol_prev * 2:
            signals.append(f"  📊 {name}({symbol}) 本周成交量是前两周均值的 "
                          f"{avg_vol_week/avg_vol_prev:.1f} 倍")

    # 信号 4：连续下跌后止稳
    if len(close) >= 8:
        prev_5 = [close[i] < close[i-1] for i in range(-6, -1)]
        if sum(prev_5) >= 4 and close[-1] > close[-2]:
            signals.append(f"  🔄 {name}({symbol}) 连续下跌后本周尾盘企稳")

if signals:
    for s in signals:
        print(s)
else:
    print("  本周自选股无特殊信号")
```

## 第五步：输出完整报告

```python
def generate_weekly_report():
    """生成完整周复盘报告并保存为 Markdown"""
    lines = []
    lines.append(f"# A 股周复盘 | {date.today()}\n")

    # 大盘部分
    lines.append("## 主要指数\n")
    lines.append("| 指数 | 周涨跌 | 周最高 | 周最低 |")
    lines.append("| --- | --- | --- | --- |")
    for name, symbol in indices.items():
        df = af.klines.get(symbol, period="1d", count=5,
                            adjust="forward", to_dataframe=True)
        if len(df) < 2:
            continue
        ret = (df["close"].iloc[-1] - df["open"].iloc[0]) / df["open"].iloc[0]
        lines.append(f"| {name} | {ret*100:+.2f}% | {df['high'].max():.3f} "
                     f"| {df['low'].min():.3f} |")

    # 自选股部分
    lines.append("\n## 自选股\n")
    lines.append("| 名称 | 代码 | 周涨跌 | MA20 | 量能变化 |")
    lines.append("| --- | --- | --- | --- | --- |")
    for item in watchlist_summary:
        lines.append(f"| {item['name']} | {item['symbol']} | "
                     f"{item['week_ret']*100:+.2f}% | "
                     f"{item['above_ma20']} | {item['vol_change']} |")

    # 信号部分
    if signals:
        lines.append("\n## 本周信号\n")
        for s in signals:
            lines.append(f"- {s.strip()}")

    report = "\n".join(lines)

    filename = f"weekly_review_{date.today():%Y%m%d}.md"
    with open(filename, "w", encoding="utf-8") as f:
        f.write(report)
    print(f"\n报告已保存到 {filename}")
    return report

report = generate_weekly_report()
```

## 配合定时任务

```bash
# 每周六上午 10 点自动跑复盘脚本
0 10 * * 6 cd /your/path && uv run python weekly_review.py
```

加上推送，周六早上起来手机就有一份现成的复盘报告。

## 为什么手动复盘不如自动化

| | 手动翻行情软件 | Python 自动复盘 |
| --- | --- | --- |
| 耗时 | 30-60 分钟 | 10 秒 |
| 覆盖面 | 凭记忆看几只票 | 系统扫描自选股 + 全市场 |
| 一致性 | 每周看的指标不一样 | 固定框架，纵向可比 |
| 积累 | 记在脑子里，下周就忘 | Markdown 文件，永久可追溯 |
| 情绪干扰 | 亏了就不想看 | 不带情绪，客观输出 |

最后一条最重要。**亏钱的那一周，恰恰是最需要复盘的一周。** 自动化复盘不会因为你心情不好就跳过。

## AlphaFeed 在这个场景中的作用

整个脚本的数据获取只有三处：

1. `af.klines.get()` — 拉指数 ETF 的 K 线
2. `af.quotes.get(universes=...)` — 全市场行情做板块分析
3. `af.klines.batch()` — 批量拉自选股 K 线

三处总共花几秒钟。剩下的时间全部是分析逻辑。如果用循环 + sleep 的方式拉数据，光数据获取就要好几分钟——周末复盘变成了等数据。

---

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


周末是复盘的最佳时间。但大多数人的复盘方式是"打开行情软件，翻翻 K 线图，想想下周买什么"——没有结构、没有数据、没有积累。

周末花 10 分钟，用 Python 自动复盘上周 A 股行情

编程语言

人工智能

周末用Python 10分钟自动化复盘A股：抓取指数、板块、自选股数据，生成结构化周报，检测突破、金叉、放量等关键信号，支持定时运行和推送，替代手动翻盘，提升复盘效率与一致性。

代码解释

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

周末花 10 分钟，用 Python 自动复盘上周 A 股行情-腾讯云开发者社区-腾讯云

周末花 10 分钟，用 Python 自动复盘上周 A 股行情

周末花 10 分钟，用 Python 自动复盘上周 A 股行情

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐