数据分析助手 – 让 AI 帮你分析业务数据(实战教程)
Excel 数据不会分析?数据趋势不会看?让 AI 帮你自动分析并生成报告!
🤔 数据分析的痛点
痛点 1:数据不会分析
打开 Excel → 一堆数字
想看趋势 → 不会画图表
想分析原因 → 不知道从哪下手
痛点 2:报告不会写
分析完数据 → 不知道怎么写结论
老板要看报告 → 憋半天写不出
写出来了 → 没有洞察力
痛点 3:数据分散
销售数据在 CRM
用户数据在后台
财务数据在 Excel
想综合分析 → 手动合并
痛点 4:报告重复
每天都要日报
每周都要周报
每月都要月报
内容差不多 → 手动复制粘贴
🎯 最终效果
1. 自动读取数据
支持多种数据源:
📊 数据源配置
✅ Excel 文件
✅ CSV 文件
✅ 数据库(MySQL/PostgreSQL)
✅ API 接口
✅ 在线表格(飞书/腾讯文档)
2. 自动生成图表
AI 自动分析并画图:
📈 销售数据分析
【销售趋势】
[自动绘制折线图]
【品类占比】
[自动绘制饼图]
【区域对比】
[自动绘制柱状图]
【关键发现】
1. 本月销售增长 15%
2. A 品类表现最好
3. 华东区域增长最快
3. 自动写分析报告
AI 自动生成洞察:
📊 销售分析报告 - 2026-04
【核心指标】
- 总销售额:¥1,234,567(+15%)
- 订单数:5,678(+12%)
- 客单价:¥217(+3%)
【关键发现】
1. 销售额连续 3 个月增长
2. 新品 A 表现突出,占比 25%
3. 华东区域增长最快(+28%)
【问题识别】
1. 华南区域下滑(-5%)
2. 老品类 B 持续萎缩
3. 周末销售占比偏低
【建议】
1. 加大华东区域投入
2. 优化华南区域策略
3. 推出周末促销活动
4. 定时生成报告
自动推送日报/周报/月报:
⏰ 报告计划
📅 日报:每天早上 9 点
📅 周报:每周一早上 10 点
📅 月报:每月 1 号下午 2 点
自动发送到:
- 老板微信
- 团队飞书群
- 邮箱
🛠️ 开始实战
准备工作
需要的基础:
-
• ✅ 完成入门系列 -
• ✅ 基本的 Python 知识 -
• ✅ 有 Excel/CSV 数据
预计耗时: 90 分钟
难度: ⭐⭐⭐☆☆(进阶级)
📝 第 1 步:数据读取系统
创建数据读取脚本
创建 data-loader.py:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
数据读取系统
"""
import pandas as pd
from pathlib import Path
from datetime import datetime
DATA_DIR = Path("~/workspace/data").expanduser()
DATA_DIR.mkdir(parents=True, exist_ok=True)
def load_excel(file_path: str) -> pd.DataFrame:
"""读取 Excel 文件"""
try:
df = pd.read_excel(file_path)
print(f"✅ 读取成功:{len(df)} 行,{len(df.columns)} 列")
return df
except Exception as e:
print(f"❌ 读取失败:{e}")
return None
def load_csv(file_path: str) -> pd.DataFrame:
"""读取 CSV 文件"""
try:
df = pd.read_csv(file_path, encoding='utf-8')
print(f"✅ 读取成功:{len(df)} 行,{len(df.columns)} 列")
return df
except Exception as e:
print(f"❌ 读取失败:{e}")
return None
def load_from_database(db_url: str, query: str) -> pd.DataFrame:
"""从数据库读取"""
try:
from sqlalchemy import create_engine
engine = create_engine(db_url)
df = pd.read_sql(query, engine)
print(f"✅ 读取成功:{len(df)} 行")
return df
except Exception as e:
print(f"❌ 读取失败:{e}")
return None
def load_from_api(api_url: str, params: dict = None) -> pd.DataFrame:
"""从 API 读取"""
import requests
try:
response = requests.get(api_url, params=params, timeout=10)
data = response.json()
# 转换为 DataFrame
df = pd.DataFrame(data)
print(f"✅ 读取成功:{len(df)} 行")
return df
except Exception as e:
print(f"❌ 读取失败:{e}")
return None
def get_data_summary(df: pd.DataFrame) -> dict:
"""获取数据摘要"""
summary = {
"rows": len(df),
"columns": len(df.columns),
"column_names": list(df.columns),
"dtypes": {col: str(dtype) for col, dtype in df.dtypes.items()},
"null_counts": df.isnull().sum().to_dict(),
"basic_stats": {},
}
# 数值列统计
numeric_cols = df.select_dtypes(include=['number']).columns
for col in numeric_cols:
summary["basic_stats"][col] = {
"mean": df[col].mean(),
"min": df[col].min(),
"max": df[col].max(),
}
return summary
if __name__ == '__main__':
# 测试
df = load_csv("sales_data.csv")
if df is not None:
summary = get_data_summary(df)
print(f"数据摘要:{summary}")
📝 第 2 步:数据分析系统
创建数据分析脚本
创建 data-analyzer.py:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
数据分析系统
"""
import pandas as pd
import matplotlib.pyplot as plt
from pathlib import Path
from datetime import datetime
CHART_DIR = Path("~/workspace/data/charts").expanduser()
CHART_DIR.mkdir(parents=True, exist_ok=True)
def analyze_trend(df: pd.DataFrame, date_col: str, value_col: str) -> dict:
"""分析趋势"""
# 按日期分组
trend = df.groupby(date_col)[value_col].sum().reset_index()
# 计算增长率
trend['growth'] = trend[value_col].pct_change() * 100
# 找出最高和最低
max_day = trend.loc[trend[value_col].idxmax()]
min_day = trend.loc[trend[value_col].idxmin()]
return {
"trend_data": trend,
"max": {"date": max_day[date_col], "value": max_day[value_col]},
"min": {"date": min_day[date_col], "value": min_day[value_col]},
"avg": trend[value_col].mean(),
"growth_rate": trend['growth'].mean(),
}
def analyze_category(df: pd.DataFrame, category_col: str, value_col: str) -> dict:
"""分析品类"""
category_stats = df.groupby(category_col)[value_col].agg([
('total', 'sum'),
('count', 'count'),
('avg', 'mean'),
]).reset_index()
# 计算占比
total = category_stats['total'].sum()
category_stats['percentage'] = category_stats['total'] / total * 100
# 排序
category_stats = category_stats.sort_values('total', ascending=False)
return {
"category_stats": category_stats,
"top_category": category_stats.iloc[0][category_col],
"top_percentage": category_stats.iloc[0]['percentage'],
}
def analyze_region(df: pd.DataFrame, region_col: str, value_col: str) -> dict:
"""分析区域"""
region_stats = df.groupby(region_col)[value_col].sum().reset_index()
region_stats = region_stats.sort_values(value_col, ascending=False)
return {
"region_stats": region_stats,
"top_region": region_stats.iloc[0][region_col],
"bottom_region": region_stats.iloc[-1][region_col],
}
def create_trend_chart(trend_data: pd.DataFrame, date_col: str, value_col: str, title: str):
"""创建趋势图"""
plt.figure(figsize=(12, 6))
plt.plot(trend_data[date_col], trend_data[value_col], marker='o')
plt.title(title)
plt.xlabel('日期')
plt.ylabel(value_col)
plt.xticks(rotation=45)
plt.grid(True, alpha=0.3)
plt.tight_layout()
chart_path = CHART_DIR / f"trend_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
plt.savefig(chart_path, dpi=150)
plt.close()
return str(chart_path)
def create_pie_chart(category_stats: pd.DataFrame, category_col: str, value_col: str, title: str):
"""创建饼图"""
plt.figure(figsize=(10, 8))
plt.pie(
category_stats[value_col],
labels=category_stats[category_col],
autopct='%1.1f%%',
startangle=90
)
plt.title(title)
plt.axis('equal')
chart_path = CHART_DIR / f"pie_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
plt.savefig(chart_path, dpi=150, bbox_inches='tight')
plt.close()
return str(chart_path)
def create_bar_chart(region_stats: pd.DataFrame, region_col: str, value_col: str, title: str):
"""创建柱状图"""
plt.figure(figsize=(12, 6))
plt.bar(region_stats[region_col], region_stats[value_col])
plt.title(title)
plt.xlabel(region_col)
plt.ylabel(value_col)
plt.xticks(rotation=45)
plt.grid(True, alpha=0.3, axis='y')
plt.tight_layout()
chart_path = CHART_DIR / f"bar_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
plt.savefig(chart_path, dpi=150)
plt.close()
return str(chart_path)
if __name__ == '__main__':
# 测试
import pandas as pd
# 创建测试数据
df = pd.DataFrame({
'date': pd.date_range('2026-01-01', periods=30),
'category': ['A', 'B', 'C'] * 10,
'region': ['East', 'South', 'West', 'North'] * 7 + ['East', 'South'],
'sales': [100 + i * 10 for i in range(30)],
})
# 分析
trend = analyze_trend(df, 'date', 'sales')
print(f"趋势分析:{trend}")
📝 第 3 步:报告生成系统
创建报告生成脚本
创建 report-generator.py:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
报告生成系统
"""
from datetime import datetime
from pathlib import Path
REPORTS_DIR = Path("~/workspace/data/reports").expanduser()
REPORTS_DIR.mkdir(parents=True, exist_ok=True)
def generate_daily_report(df: pd.DataFrame, analysis_results: dict) -> str:
"""生成日报"""
today = datetime.now().strftime('%Y-%m-%d')
report = f"""
📊 数据日报 - {today}
【核心指标】
- 总销售额:¥{df['sales'].sum():,}
- 订单数:{len(df):,}
- 客单价:¥{df['sales'].mean():.0f}
【销售趋势】
- 日均销售:¥{analysis_results['trend']['avg']:,.0f}
- 平均增长率:{analysis_results['trend']['growth_rate']:.1f}%
- 最高销售日:{analysis_results['trend']['max']['date']}(¥{analysis_results['trend']['max']['value']:,})
【品类表现】
- 最佳品类:{analysis_results['category']['top_category']}
- 占比:{analysis_results['category']['top_percentage']:.1f}%
【区域表现】
- 最佳区域:{analysis_results['region']['top_region']}
- 待提升区域:{analysis_results['region']['bottom_region']}
【关键发现】
1. (AI 自动生成)
2. (AI 自动生成)
3. (AI 自动生成)
【建议】
1. (AI 自动生成)
2. (AI 自动生成)
3. (AI 自动生成)
---
报告生成时间:{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
"""
return report
def generate_weekly_report(df: pd.DataFrame, analysis_results: dict) -> str:
"""生成周报"""
week_start = datetime.now()
week_num = week_start.isocalendar()[1]
report = f"""
📊 周报 - 2026 年第{week_num}周
【本周核心指标】
- 总销售额:¥{df['sales'].sum():,}
- 日均销售:¥{df['sales'].mean():,.0f}
- 订单数:{len(df):,}
- 客单价:¥{df['sales'].mean():.0f}
【周环比】
- 销售额变化:+X%
- 订单数变化:+X%
- 客单价变化:+X%
【本周亮点】
1. (AI 自动生成)
2. (AI 自动生成)
3. (AI 自动生成)
【问题识别】
1. (AI 自动生成)
2. (AI 自动生成)
【下周计划】
1. (AI 自动生成)
2. (AI 自动生成)
3. (AI 自动生成)
---
报告生成时间:{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
"""
return report
def save_report(report: str, report_type: str = "daily") -> str:
"""保存报告"""
filename = f"{report_type}_{datetime.now().strftime('%Y%m%d')}.md"
report_path = REPORTS_DIR / filename
with open(report_path, 'w', encoding='utf-8') as f:
f.write(report)
return str(report_path)
if __name__ == '__main__':
# 测试
import pandas as pd
df = pd.DataFrame({
'sales': [100, 200, 150, 300, 250],
})
report = generate_daily_report(df, {})
print(report)
📝 第 4 步:配置定时任务
配置日报自动生成
# 每天早上 9 点生成日报
openclaw cron add \
--name "daily-report" \
--cron "0 9 * * *" \
--session isolated \
--message "读取昨日数据,生成日报并发送" \
--announce \
--channel wechat \
--to "user:boss-id"
配置周报自动生成
# 每周一早上 10 点生成周报
openclaw cron add \
--name "weekly-report" \
--cron "0 10 * * 1" \
--session isolated \
--message "读取上周数据,生成周报并发送" \
--announce \
--channel feishu \
--to "team-group-id"
配置月报自动生成
# 每月 1 号下午 2 点生成月报
openclaw cron add \
--name "monthly-report" \
--cron "0 14 1 * *" \
--session isolated \
--message "读取上月数据,生成月报并发送" \
--announce \
--channel email \
--to "management@company.com"
📊 完整数据分析流程
创建工作流脚本
创建 analytics-workflow.py:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
完整数据分析工作流
"""
from data_loader import load_csv, get_data_summary
from data_analyzer import analyze_trend, analyze_category, analyze_region
from data_analyzer import create_trend_chart, create_pie_chart, create_bar_chart
from report_generator import generate_daily_report, save_report
def run_analytics_workflow(data_file: str, report_type: str = "daily"):
"""
运行完整分析流程
1. 读取数据
2. 分析数据
3. 生成图表
4. 生成报告
5. 发送报告
"""
print("🚀 开始数据分析...")
# 第 1 步:读取数据
print("📥 读取数据...")
df = load_csv(data_file)
if df is None:
return {"error": "数据读取失败"}
# 第 2 步:数据摘要
print("📊 数据摘要...")
summary = get_data_summary(df)
# 第 3 步:数据分析
print("🔍 分析数据...")
analysis_results = {
"trend": analyze_trend(df, 'date', 'sales'),
"category": analyze_category(df, 'category', 'sales'),
"region": analyze_region(df, 'region', 'sales'),
}
# 第 4 步:生成图表
print("📈 生成图表...")
charts = {
"trend": create_trend_chart(
analysis_results['trend']['trend_data'],
'date', 'sales', '销售趋势图'
),
"category": create_pie_chart(
analysis_results['category']['category_stats'],
'category', 'total', '品类占比'
),
"region": create_bar_chart(
analysis_results['region']['region_stats'],
'region', 'sales', '区域对比'
),
}
# 第 5 步:生成报告
print("📝 生成报告...")
if report_type == "daily":
report = generate_daily_report(df, analysis_results)
elif report_type == "weekly":
report = generate_weekly_report(df, analysis_results)
# 第 6 步:保存报告
print("💾 保存报告...")
report_path = save_report(report, report_type)
return {
"success": True,
"report_path": report_path,
"charts": charts,
"summary": summary,
}
if __name__ == '__main__':
result = run_analytics_workflow("sales_data.csv", "daily")
print(f"分析完成:{result}")
🎉 总结
通过本文,你学会了:
-
1. ✅ 读取多种数据源(Excel/CSV/数据库/API) -
2. ✅ 自动分析数据趋势、品类、区域 -
3. ✅ 生成可视化图表 -
4. ✅ 自动生成分析报告 -
5. ✅ 定时推送日报/周报/月报
效率提升:
-
• 数据整理:30 分钟 → 0 分钟(自动) -
• 数据分析:60 分钟 → 0 分钟(自动) -
• 图表制作:30 分钟 → 0 分钟(自动) -
• 报告撰写:60 分钟 → 0 分钟(自动)
总效率提升:180 分钟 → 0 分钟(100% 自动化)
🚀 扩展阅读
-
• 进阶 2:《自动化工作流》 -
• 进阶 4:《内容创作助手》 -
• 入门 5:《定时任务自动化》
作者:Mac ⚡\ 编辑:兵哥\ 发布时间:2026-04-23
你的数据分析自动化了吗?欢迎在评论区分享你的分析效率!
夜雨聆风