本文介绍如何用AKShare免费数据接口+Python，搭建一套A股每日自动复盘系统。从数据采集到分析报告一键生成，把每天30分钟的复盘工作压缩到3秒。所有代码经过实盘验证，可直接运行。## 一、为什么需要自动化复盘？做A股短线的朋友应该都有体会：每天收盘后的复盘工作繁琐但必不可少——要看大盘走势、板块轮动、涨停分析、龙虎榜、资金流向，少说也要看20-30分钟。我的痛点：- 每天重复打开5-6个网站，手动记录数据- 关键数据容易遗漏（比如北向资金突然大幅流出）- 复盘报告写完后经常忘记存档，回头看不了解决方案： 用Python写一个自动化复盘脚本，每天收盘后运行一次，自动生成包含所有关键数据的复盘报告。> 核心思路：数据采集（AKShare）→ 关键指标计算 → 报告生成 → 消息推送## 二、AKShare：免费A股数据的瑞士军刀AKShare是一个开源的Python财经数据库，封装了东方财富、新浪、同花顺等多个数据源，完全免费、无需注册、无需Token。### 2.1 安装bashpip install akshare -i https://pypi.tuna.tsinghua.edu.cn/simple### 2.2 能获取什么数据？| 数据类型 | 接口 | 说明 ||---------|------|------|| 大盘行情 | ak.stock_zh_index_daily() | 上证/深证/创业板日K || 涨停板 | ak.stock_zt_pool_em() | 涨停股票池 || 龙虎榜 | ak.stock_lhb_detail_em() | 近5日龙虎榜 || 板块资金 | ak.stock_sector_fund_flow_rank() | 行业资金流向排名 || 北向资金 | ak.stock_hsgt_north_net_flow_in_em() | 沪深港通净流入 || 个股资金流 | ak.stock_individual_fund_flow() | 主力/散户净流入 |## 三、核心代码：每日复盘数据采集### 3.1 大盘总览pythonimport akshare as akimport pandas as pdfrom datetime import datetime, timedeltadef get_market_overview(): """获取大盘指数走势""" indices = { '上证指数': 'sh000001', '深证成指': 'sz399001', '创业板指': 'sz399006', } results = [] for name, code in indices.items(): df = ak.stock_zh_index_daily(symbol=code) latest = df.iloc[-1] prev = df.iloc[-2] change_pct = (latest['close'] - prev['close']) / prev['close'] * 100 results.append({ '指数': name, '收盘': round(latest['close'], 2), '涨跌幅': f'{change_pct:+.2f}%', '成交额(亿)': round(latest['volume'] / 1e8, 0) if 'volume' in latest else '-' }) return pd.DataFrame(results)print(get_market_overview())### 3.2 涨停板分析pythondef get_limit_up_analysis(date=None): """涨停板数据分析""" if date is None: date = datetime.now().strftime('%Y%m%d') df = ak.stock_zt_pool_em(date=date) # 关键统计 total = len(df) # 连板统计 if '连板数' in df.columns: lianban = df[df['连板数'] > 1] max_lianban = df['连板数'].max() if total > 0 else 0 else: lianban = pd.DataFrame() max_lianban = 0 # 行业分布（取前5） if '所属行业' in df.columns: industry_dist = df['所属行业'].value_counts().head(5) else: industry_dist = {} return { '涨停总数': total, '连板数': len(lianban), '最高连板': max_lianban, '热门行业': dict(industry_dist), '详细数据': df }zt_data = get_limit_up_analysis()print(f"今日涨停{zt_data['涨停总数']}只，最高{zt_data['最高连板']}连板")print(f"热门行业：{zt_data['热门行业']}")### 3.3 板块资金流向pythondef get_sector_flow(): """获取行业板块资金流向""" df = ak.stock_sector_fund_flow_rank(indicator="今日") # 取流入前5和流出前5 cols = ['名称', '今日涨跌幅', '今日主力净流入-净额', '今日主力净流入-净占比'] available_cols = [c for c in cols if c in df.columns] top5 = df.head(5)[available_cols] bottom5 = df.tail(5)[available_cols] return {'流入前5': top5, '流出前5': bottom5}flow = get_sector_flow()print("\n资金流入前5板块：")print(flow['流入前5'].to_string(index=False))print("\n资金流出前5板块：")print(flow['流出前5'].to_string(index=False))### 3.4 北向资金监控pythondef get_north_flow(): """获取北向资金净流入""" df = ak.stock_hsgt_north_net_flow_in_em() latest = df.iloc[-1] prev5 = df.tail(5) today_flow = latest.iloc[-1] # 最新一天数据 week_sum = prev5.iloc[:, -1].sum() # 近5日累计 return { '今日净流入(亿)': round(today_flow / 1e4, 2), '近5日累计(亿)': round(week_sum / 1e4, 2), '趋势': '持续流入' if week_sum > 0 else '持续流出' }north = get_north_flow()print(f"\n北向资金：今日{north['今日净流入(亿)']}亿，近5日{north['近5日累计(亿)']}亿")## 四、自动生成复盘报告把上面的数据采集函数整合起来，一键生成完整的复盘报告：pythondef generate_daily_report(): """生成每日复盘报告""" today = datetime.now().strftime('%Y-%m-%d') report = f"""# A股每日复盘 - {today}## 一、大盘概况{get_market_overview().to_string(index=False)}## 二、涨停分析- 涨停总数：{zt_data['涨停总数']}只- 连板股数：{zt_data['连板数']}只- 最高连板：{zt_data['最高连板']}板- 热门行业：{zt_data['热门行业']}## 三、板块资金流向### 资金流入前5{flow['流入前5'].to_string(index=False)}### 资金流出前5{flow['流出前5'].to_string(index=False)}## 四、北向资金- 今日净流入：{north['今日净流入(亿)']}亿- 近5日累计：{north['近5日累计(亿)']}亿- 趋势判断：{north['趋势']}## 五、操作提示> 以上数据仅供参考，不构成投资建议。""" # 保存到文件 filename = f'daily_report_{datetime.now().strftime("%Y%m%d")}.md' with open(filename, 'w', encoding='utf-8') as f: f.write(report) print(f"复盘报告已生成：{filename}") return report# 一键运行report = generate_daily_report()print(report)## 五、进阶：消息推送到手机复盘报告生成后，可以通过企业微信Webhook自动推送到手机：pythonimport requestsdef push_to_wechat(report_content, webhook_url): """推送到企业微信""" data = { "msgtype": "markdown", "markdown": { "content": report_content[:4096] # 企业微信限制4096字 } } resp = requests.post(webhook_url, json=data) return resp.json()# 使用示例# push_to_wechat(report, "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=你的KEY")也支持钉钉机器人、飞书机器人，原理类似，换个Webhook地址就行。## 六、踩坑记录1. AKShare接口不稳定：封装的是网页爬虫，数据源网站改版就会导致接口失效。解决办法：保持pip install akshare --upgrade及时更新版本。2. 频率限制：调用间隔建议1-2秒，太快会被数据源封IP。批量采集时用time.sleep(1)控制节奏。3. 非交易时间数据：周末和节假日调用返回的是上一个交易日的数据，需要在代码中判断日期。4. 字段名变化：AKShare不同版本返回的DataFrame列名可能不同，建议用df.columns先检查可用列名再取值，别硬编码。5. 数据缓存：建议把每日数据存入SQLite本地数据库，避免重复采集，也方便历史回溯。## 七、总结| 项目 | 手动复盘 | 自动化复盘 ||------|---------|-----------|| 时间 | 30分钟 | 3秒 || 遗漏率 | 高 | 接近0 || 历史存档 | 经常丢失 | 自动归档 || 扩展性 | 低 | 可叠加任意指标 |这套系统我已经用了两个月，每天收盘后跑一次脚本，3秒钟拿到所有关键数据。省下来的时间可以专注做交易决策，而不是当数据搬运工。> 代码已验证可运行，直接复制即可使用。如果觉得有用，点赞收藏支持下～