周末花 10 分钟，用 Python 自动复盘上周 A 股行情
周末是复盘的最佳时间。但大多数人的复盘方式是"打开行情软件，翻翻 K 线图，想想下周买什么"——没有结构、没有数据、没有积累。
这篇文章教你写一个周复盘脚本：周末跑一下，自动生成上周的市场概览、板块表现、自选股状态和关键信号，输出一份结构化的复盘报告。
复盘看什么
一份有用的周复盘应该回答这几个问题：
- 上周大盘怎么样？ — 主要指数涨跌、成交额变化
- 哪些板块最强/最弱？ — 板块轮动方向
- 我的自选股表现如何？ — 持仓或关注标的的周度表现
- 有没有值得注意的信号？ — 异动、突破、趋势变化
第一步：大盘周度概览
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}%")
第二步：全市场板块强弱
# 拉全市场行情，按板块分组看周表现
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}")
第三步：自选股周报
# 你的自选股列表
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}")
第四步：关键信号检测
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(" 本周自选股无特殊信号")
第五步：输出完整报告
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()
配合定时任务
# 每周六上午 10 点自动跑复盘脚本
0 10 * * 6 cd /your/path && uv run python weekly_review.py
加上推送，周六早上起来手机就有一份现成的复盘报告。
为什么手动复盘不如自动化
| 手动翻行情软件 | Python 自动复盘 | |
|---|---|---|
| 耗时 | 30-60 分钟 | 10 秒 |
| 覆盖面 | 凭记忆看几只票 | 系统扫描自选股 + 全市场 |
| 一致性 | 每周看的指标不一样 | 固定框架，纵向可比 |
| 积累 | 记在脑子里，下周就忘 | Markdown 文件，永久可追溯 |
| 情绪干扰 | 亏了就不想看 | 不带情绪，客观输出 |
最后一条最重要。亏钱的那一周，恰恰是最需要复盘的一周。 自动化复盘不会因为你心情不好就跳过。
AlphaFeed 在这个场景中的作用
整个脚本的数据获取只有三处：
af.klines.get()— 拉指数 ETF 的 K 线af.quotes.get(universes=...)— 全市场行情做板块分析af.klines.batch()— 批量拉自选股 K 线
三处总共花几秒钟。剩下的时间全部是分析逻辑。如果用循环 + sleep 的方式拉数据，光数据获取就要好几分钟——周末复盘变成了等数据。
- AlphaFeed 官网：https://alphafeed.org/
- Python SDK 文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart