乐于分享
好东西不私藏

数据分析助手 – 让 AI 帮你分析业务数据(实战教程)

数据分析助手 – 让 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. 1. ✅ 读取多种数据源(Excel/CSV/数据库/API)
  2. 2. ✅ 自动分析数据趋势、品类、区域
  3. 3. ✅ 生成可视化图表
  4. 4. ✅ 自动生成分析报告
  5. 5. ✅ 定时推送日报/周报/月报

效率提升:

  • • 数据整理:30 分钟 → 0 分钟(自动)
  • • 数据分析:60 分钟 → 0 分钟(自动)
  • • 图表制作:30 分钟 → 0 分钟(自动)
  • • 报告撰写:60 分钟 → 0 分钟(自动)

总效率提升:180 分钟 → 0 分钟(100% 自动化)


🚀 扩展阅读

  • • 进阶 2:《自动化工作流》
  • • 进阶 4:《内容创作助手》
  • • 入门 5:《定时任务自动化》

作者:Mac ⚡\ 编辑:兵哥\ 发布时间:2026-04-23


不凡工具  免费寸照  一拍即印

你的数据分析自动化了吗?欢迎在评论区分享你的分析效率!