OpenClaw数据分析与可视化实战案例

目录

• 1. 概述
• 1.1 核心数据分析能力
• 1.2 分析能力矩阵
• 1.3 与传统工具的对比
• 2. 数据处理能力
• 2.1 数据源接入
• 2.1.1 本地文件读取
• 2.1.2 Excel 文件处理
• 2.1.3 JSON 数据处理
• 2.1.4 API 数据获取
• 2.2 数据清洗
• 2.2.1 缺失值处理
• 2.2.2 异常值检测与处理
• 2.2.3 数据类型转换
• 2.3 数据转换
• 2.3.1 数据透视与聚合
• 2.3.2 时间序列处理
• 2.4 实际操作示例
• 3. 图表生成与可视化
• 3.1 基础图表生成
• 3.1.1 折线图
• 3.1.2 柱状图
• 3.1.3 饼图
• 3.1.4 散点图
• 3.2 高级可视化
• 3.2.1 热力图
• 3.2.2 箱线图
• 3.2.3 多子图仪表盘
• 3.3 交互式图表（通过 Canvas）
• 3.4 图表保存与输出
• 4. 报告自动生成
• 4.1 Markdown 报告模板
• 4.2 完整报告生成示例
• 4.3 通过 Feishu 发送报告
• 5. 自动化数据分析工作流
• 5.1 使用 Cron 实现定时分析
• 5.1.1 配置定时分析任务
• 5.1.2 通过 CLI 管理任务
• 5.2 使用 Heartbeat 实现监控分析
• 5.2.1 配置 Heartbeat
• 5.2.2 数据监控脚本
• 5.3 自动化分析流水线
• 5.3.1 ETL 流程设计
• 5.3.2 增量数据处理
• 5.4 与外部系统集成
• 5.4.1 数据库连接
• 5.4.2 Webhook 数据接收
• 6. 实战案例
• 6.1 案例一：销售数据分析
• 步骤一：准备数据
• 步骤二：数据清洗与分析
• 步骤三：生成可视化
• 步骤四：生成报告并发送
• 6.2 案例二：网站流量监控
• 配置定时监控任务
• 监控脚本
• 6.3 案例三：自动化周报生成
• 完整工作流
• 7. 最佳实践与注意事项
• 7.1 数据处理最佳实践
• 7.1.1 数据安全
• 7.1.2 性能优化
• 7.2 可视化最佳实践
• 7.2.1 图表选择指南
• 7.2.2 图表设计原则
• 7.3 报告生成最佳实践
• 7.3.1 报告结构模板
• 7.3.2 报告质量控制
• 7.4 自动化注意事项
• 7.4.1 错误处理
• 7.4.2 资源管理
• 7.4.3 监控与告警
• 总结
• 核心优势
• 适用场景

本文档详细介绍如何利用 OpenClaw 进行数据分析与可视化，包括数据处理、图表生成、报告输出和自动化分析的完整流程与实践。

OpenClaw 作为个人 AI 助手，具备强大的数据处理和分析能力。通过与代码执行环境的深度集成，OpenClaw 可以帮助您完成从数据清洗到可视化呈现的全流程数据分析工作。

┌─────────────────────────────────────────────────────────────────────┐
│                    OpenClaw 数据分析能力架构                          │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│   ┌─────────────────────────────────────────────────────────────┐   │
│   │                      数据输入层                              │   │
│   │  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐          │   │
│   │  │ 本地文件 │ │ API数据 │ │ 网页抓取 │ │ 数据库   │          │   │
│   │  │ CSV/JSON│ │ REST API│ │ Browser │ │ SQLite  │          │   │
│   │  └────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘          │   │
│   └───────┼───────────┼───────────┼───────────┼───────────────┘   │
│           │           │           │           │                    │
│           └───────────┴─────┬─────┴───────────┘                    │
│                             ▼                                      │
│   ┌─────────────────────────────────────────────────────────────┐   │
│   │                      数据处理层                              │   │
│   │  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐          │   │
│   │  │ 数据清洗 │ │ 数据转换 │ │ 统计分析 │ │ 机器学习 │          │   │
│   │  │ Pandas  │ │ NumPy   │ │ SciPy   │ │ Scikit  │          │   │
│   │  └─────────┘ └─────────┘ └─────────┘ └─────────┘          │   │
│   └─────────────────────────────────────────────────────────────┘   │
│                             │                                      │
│                             ▼                                      │
│   ┌─────────────────────────────────────────────────────────────┐   │
│   │                      输出呈现层                              │   │
│   │  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐          │   │
│   │  │ 图表生成 │ │ 报告输出 │ │ 交互式  │ │ 自动化  │          │   │
│   │  │ Matplotlib│ │ Markdown│ │ Canvas  │ │ Cron   │          │   │
│   │  └─────────┘ └─────────┘ └─────────┘ └─────────┘          │   │
│   └─────────────────────────────────────────────────────────────┘   │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

能力类别	具体功能	技术栈
数据获取	文件读取、API调用、网页抓取	read/exec/browser/web_fetch
数据清洗	缺失值处理、异常值检测、格式转换	Python/Pandas
统计分析	描述统计、假设检验、相关性分析	NumPy/SciPy
可视化	静态图表、交互式图表、仪表盘	Matplotlib/Plotly
报告生成	Markdown报告、PDF导出、自动发送	write/feishu_doc
自动化	定时分析、监控告警、增量更新	cron/heartbeat

特性	传统 BI 工具	Jupyter Notebook	OpenClaw
学习成本	高	中	低（自然语言）
交互方式	图形界面	代码	对话式
自动化程度	需配置	需编码	原生支持
多通道输出	单一	单一	多平台
实时性	定时刷新	手动执行	随时触发
协作能力	一般	一般	集成IM平台

OpenClaw 支持多种数据源的接入，可以灵活处理各类数据格式。

# 让 OpenClaw 读取本地 CSV 文件
# 用户：帮我分析 /data/sales.csv 文件
# OpenClaw 会执行以下操作：
import pandas as pd
# 读取 CSV 文件
df = pd.read_csv('/data/sales.csv')
# 查看数据概览
print(df.head())
print(df.info())
print(df.describe())

# 读取 Excel 文件
import pandas as pd
# 单个 Sheet
df = pd.read_excel('/data/report.xlsx', sheet_name='Sales')
# 多个 Sheet
all_sheets = pd.read_excel('/data/report.xlsx', sheet_name=None)
for sheet_name, df in all_sheets.items():
    print(f"Sheet: {sheet_name}, Shape: {df.shape}")

import json
import pandas as pd
# 读取 JSON 文件
with open('/data/api_response.json', 'r') as f:
    data = json.load(f)
# 转换为 DataFrame
df = pd.json_normalize(data['records'])
# 处理嵌套数据
df_flat = pd.json_normalize(
    data, 
    record_path=['items'],
    meta=['id', 'name', ['metadata', 'created_at']]
)

使用 OpenClaw 的 web_fetch 工具获取 API 数据：

# 示例：获取股票数据 API
# 用户：帮我获取苹果公司最近30天的股价数据并分析趋势
import requests
import pandas as pd
# OpenClaw 可以直接调用 API
response = requests.get(
    'https://api.example.com/stocks/AAPL',
    params={'period': '30d'}
)
data = response.json()
df = pd.DataFrame(data['prices'])
df['date'] = pd.to_datetime(df['date'])
df = df.set_index('date')
# 计算技术指标
df['MA5'] = df['close'].rolling(window=5).mean()
df['MA20'] = df['close'].rolling(window=20).mean()

数据清洗是数据分析的基础，OpenClaw 可以智能化地处理各类数据质量问题。

import pandas as pd
import numpy as np
# 检测缺失值
def analyze_missing(df):
    """分析数据缺失情况"""
    missing = df.isnull().sum()
    percent = (missing / len(df)) * 100
    missing_df = pd.DataFrame({
        '缺失数量': missing,
        '缺失比例': percent.round(2)
    })
    return missing_df[missing_df['缺失数量'] > 0]
# 处理缺失值策略
def handle_missing(df, strategy='auto'):
    """智能处理缺失值"""
    df_clean = df.copy()
    for col in df_clean.columns:
        missing_pct = df_clean[col].isnull().sum() / len(df_clean)
        if missing_pct > 0.5:
            # 缺失超过 50%，删除列
            df_clean.drop(col, axis=1, inplace=True)
        elif missing_pct > 0.1:
            # 缺失 10-50%，使用模型预测填充
            if df_clean[col].dtype in ['int64', 'float64']:
                df_clean[col].fillna(df_clean[col].median(), inplace=True)
            else:
                df_clean[col].fillna(df_clean[col].mode()[0], inplace=True)
        else:
            # 缺失少于 10%，使用简单填充
            if df_clean[col].dtype in ['int64', 'float64']:
                df_clean[col].fillna(df_clean[col].mean(), inplace=True)
            else:
                df_clean[col].fillna('Unknown', inplace=True)
    return df_clean

def detect_outliers(df, column, method='iqr'):
    """检测异常值"""
    if method == 'iqr':
        # IQR 方法
        Q1 = df[column].quantile(0.25)
        Q3 = df[column].quantile(0.75)
        IQR = Q3 - Q1
        lower = Q1 - 1.5 * IQR
        upper = Q3 + 1.5 * IQR
        outliers = df[(df[column] < lower) | (df[column] > upper)]
    elif method == 'zscore':
        # Z-score 方法
        from scipy import stats
        z_scores = np.abs(stats.zscore(df[column]))
        outliers = df[z_scores > 3]
    return outliers
def handle_outliers(df, column, method='cap'):
    """处理异常值"""
    df_clean = df.copy()
    Q1 = df_clean[column].quantile(0.25)
    Q3 = df_clean[column].quantile(0.75)
    IQR = Q3 - Q1
    lower = Q1 - 1.5 * IQR
    upper = Q3 + 1.5 * IQR
    if method == 'cap':
        # 盖帽法
        df_clean[column] = df_clean[column].clip(lower, upper)
    elif method == 'remove':
        # 删除异常值
        df_clean = df_clean[
            (df_clean[column] >= lower) & 
            (df_clean[column] <= upper)
        ]
    return df_clean

def auto_convert_dtypes(df):
    """自动推断和转换数据类型"""
    df_converted = df.copy()
    for col in df_converted.columns:
        # 尝试转换为日期
        try:
            df_converted[col] = pd.to_datetime(df_converted[col])
            continue
        except:
            pass
        # 尝试转换为数值
        try:
            df_converted[col] = pd.to_numeric(df_converted[col])
            continue
        except:
            pass
        # 分类数据优化
        if df_converted[col].nunique() / len(df_converted) < 0.5:
            df_converted[col] = df_converted[col].astype('category')
    return df_converted

import pandas as pd
# 数据透视表示例
def create_pivot_report(df, index, columns, values, aggfunc='sum'):
    """创建透视表报告"""
    pivot = pd.pivot_table(
        df,
        index=index,
        columns=columns,
        values=values,
        aggfunc=aggfunc,
        fill_value=0,
        margins=True,
        margins_name='总计'
    )
    return pivot
# 示例：按地区和产品类别的销售透视
# sales_pivot = create_pivot_report(
#     df=sales_data,
#     index='region',
#     columns='product_category',
#     values='sales_amount',
#     aggfunc='sum'
# )

def process_time_series(df, date_col, value_col, freq='D'):
    """时间序列数据处理"""
    df_ts = df.copy()
    df_ts[date_col] = pd.to_datetime(df_ts[date_col])
    df_ts = df_ts.set_index(date_col)
    # 重采样
    df_resampled = df_ts.resample(freq).agg({
        value_col: ['sum', 'mean', 'count', 'std']
    })
    # 时间特征提取
    df_ts['year'] = df_ts.index.year
    df_ts['month'] = df_ts.index.month
    df_ts['day_of_week'] = df_ts.index.dayofweek
    df_ts['is_weekend'] = df_ts.index.dayofweek.isin([5, 6])
    return df_ts, df_resampled

通过 OpenClaw 进行数据清洗的完整流程：

# 用户发送消息给 OpenClaw：
"帮我分析 /Users/anyi/data/orders.csv 文件，清洗数据并生成统计报告"

OpenClaw 会自动执行以下步骤：

┌─────────────────────────────────────────────────────────────────────┐
│                    OpenClaw 数据处理流程                             │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│   1. 读取文件                                                        │
│      └── 使用 read 工具读取 CSV 文件内容                              │
│                                                                     │
│   2. 数据探索                                                        │
│      └── 分析数据结构、类型、缺失情况                                  │
│                                                                     │
│   3. 数据清洗                                                        │
│      └── 处理缺失值、异常值、重复数据                                  │
│                                                                     │
│   4. 数据转换                                                        │
│      └── 类型转换、特征工程、标准化                                   │
│                                                                     │
│   5. 统计分析                                                        │
│      └── 描述统计、相关性分析、分组统计                               │
│                                                                     │
│   6. 结果输出                                                        │
│      └── 生成清洗后的文件和统计报告                                   │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

OpenClaw 可以通过 Python 代码生成各类图表，并通过 Canvas 或文件输出展示。

import matplotlib.pyplot as plt
import pandas as pd
import numpy as np
def create_line_chart(data, x_col, y_cols, title, save_path=None):
    """创建多系列折线图"""
    plt.figure(figsize=(12, 6))
    for y_col in y_cols:
        plt.plot(data[x_col], data[y_col], marker='o', label=y_col)
    plt.title(title, fontsize=14)
    plt.xlabel(x_col, fontsize=12)
    plt.ylabel('Value', fontsize=12)
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.xticks(rotation=45)
    plt.tight_layout()
    if save_path:
        plt.savefig(save_path, dpi=150, bbox_inches='tight')
    return plt
# 示例使用
# df = pd.DataFrame({
#     'date': pd.date_range('2024-01-01', periods=30),
#     'sales': np.random.randint(100, 500, 30),
#     'visitors': np.random.randint(1000, 3000, 30)
# })
# create_line_chart(df, 'date', ['sales', 'visitors'], 'Daily Metrics')

def create_bar_chart(data, x_col, y_col, title, orientation='vertical', save_path=None):
    """创建柱状图"""
    fig, ax = plt.subplots(figsize=(10, 6))
    if orientation == 'vertical':
        bars = ax.bar(data[x_col], data[y_col], color='steelblue')
        ax.set_xlabel(x_col)
        ax.set_ylabel(y_col)
        # 添加数值标签
        for bar in bars:
            height = bar.get_height()
            ax.annotate(f'{height:,.0f}',
                       xy=(bar.get_x() + bar.get_width() / 2, height),
                       ha='center', va='bottom')
    else:
        bars = ax.barh(data[x_col], data[y_col], color='steelblue')
        ax.set_xlabel(y_col)
        ax.set_ylabel(x_col)
    ax.set_title(title, fontsize=14)
    plt.tight_layout()
    if save_path:
        plt.savefig(save_path, dpi=150, bbox_inches='tight')
    return plt

def create_pie_chart(data, labels_col, values_col, title, save_path=None):
    """创建饼图"""
    fig, ax = plt.subplots(figsize=(10, 8))
    values = data[values_col]
    labels = data[labels_col]
    # 设置颜色
    colors = plt.cm.Set3(np.linspace(0, 1, len(values)))
    # 突出显示最大扇区
    explode = [0.05 if v == values.max() else 0 for v in values]
    wedges, texts, autotexts = ax.pie(
        values,
        labels=labels,
        autopct='%1.1f%%',
        colors=colors,
        explode=explode,
        shadow=True,
        startangle=90
    )
    ax.set_title(title, fontsize=14)
    plt.tight_layout()
    if save_path:
        plt.savefig(save_path, dpi=150, bbox_inches='tight')
    return plt

def create_scatter_plot(data, x_col, y_col, title, hue_col=None, save_path=None):
    """创建散点图"""
    fig, ax = plt.subplots(figsize=(10, 8))
    if hue_col:
        categories = data[hue_col].unique()
        colors = plt.cm.viridis(np.linspace(0, 1, len(categories)))
        for cat, color in zip(categories, colors):
            subset = data[data[hue_col] == cat]
            ax.scatter(subset[x_col], subset[y_col], 
                      c=[color], label=cat, alpha=0.6)
        ax.legend()
    else:
        ax.scatter(data[x_col], data[y_col], alpha=0.6)
    ax.set_xlabel(x_col, fontsize=12)
    ax.set_ylabel(y_col, fontsize=12)
    ax.set_title(title, fontsize=14)
    ax.grid(True, alpha=0.3)
    plt.tight_layout()
    if save_path:
        plt.savefig(save_path, dpi=150, bbox_inches='tight')
    return plt

import seaborn as sns
def create_heatmap(data, title, annot=True, save_path=None):
    """创建热力图"""
    fig, ax = plt.subplots(figsize=(12, 10))
    # 计算相关性矩阵或直接使用数据
    if isinstance(data, pd.DataFrame):
        matrix = data.corr()
    else:
        matrix = data
    sns.heatmap(
        matrix,
        annot=annot,
        fmt='.2f',
        cmap='RdYlBu_r',
        center=0,
        square=True,
        linewidths=0.5,
        ax=ax
    )
    ax.set_title(title, fontsize=14)
    plt.tight_layout()
    if save_path:
        plt.savefig(save_path, dpi=150, bbox_inches='tight')
    return plt

def create_boxplot(data, x_col, y_col, title, save_path=None):
    """创建箱线图"""
    fig, ax = plt.subplots(figsize=(12, 6))
    data.boxplot(
        column=y_col,
        by=x_col,
        ax=ax,
        patch_artist=True
    )
    plt.title(title, fontsize=14)
    plt.suptitle('')  # 移除默认标题
    plt.xticks(rotation=45)
    plt.tight_layout()
    if save_path:
        plt.savefig(save_path, dpi=150, bbox_inches='tight')
    return plt

def create_dashboard(data, save_path=None):
    """创建多图表仪表盘"""
    fig, axes = plt.subplots(2, 2, figsize=(14, 10))
    # 子图1：趋势线
    ax1 = axes[0, 0]
    ax1.plot(data['date'], data['sales'], 'b-', linewidth=2)
    ax1.set_title('销售趋势', fontsize=12)
    ax1.tick_params(axis='x', rotation=45)
    # 子图2：柱状图
    ax2 = axes[0, 1]
    categories = data['category'].value_counts()
    ax2.bar(categories.index, categories.values, color='steelblue')
    ax2.set_title('类别分布', fontsize=12)
    ax2.tick_params(axis='x', rotation=45)
    # 子图3：饼图
    ax3 = axes[1, 0]
    region_sales = data.groupby('region')['sales'].sum()
    ax3.pie(region_sales.values, labels=region_sales.index, autopct='%1.1f%%')
    ax3.set_title('区域占比', fontsize=12)
    # 子图4：散点图
    ax4 = axes[1, 1]
    ax4.scatter(data['price'], data['sales'], alpha=0.5)
    ax4.set_xlabel('价格')
    ax4.set_ylabel('销量')
    ax4.set_title('价格-销量关系', fontsize=12)
    plt.suptitle('数据分析仪表盘', fontsize=14, fontweight='bold')
    plt.tight_layout()
    if save_path:
        plt.savefig(save_path, dpi=150, bbox_inches='tight')
    return plt

OpenClaw 的 Canvas 功能支持展示交互式可视化：

# 通过 Canvas 展示 HTML 图表
import plotly.express as px
import plotly.graph_objects as go
def create_interactive_chart(data, chart_type='line'):
    """创建交互式图表 HTML"""
    if chart_type == 'line':
        fig = px.line(data, x='date', y='value', title='交互式趋势图')
        fig.update_traces(mode='lines+markers')
    elif chart_type == 'bar':
        fig = px.bar(data, x='category', y='value', 
                     color='category', title='交互式柱状图')
    elif chart_type == 'scatter':
        fig = px.scatter(data, x='x', y='y', 
                         color='category', size='size',
                         title='交互式散点图')
    # 添加交互功能
    fig.update_layout(
        hovermode='closest',
        dragmode='zoom'
    )
    # 返回 HTML
    return fig.to_html(include_plotlyjs='cdn')

class ChartExporter:
    """图表导出工具"""
    def __init__(self, output_dir):
        self.output_dir = output_dir
    def save_as_png(self, plt, filename, dpi=150):
        """保存为 PNG"""
        path = f"{self.output_dir}/{filename}.png"
        plt.savefig(path, dpi=dpi, bbox_inches='tight', 
                     facecolor='white', edgecolor='none')
        return path
    def save_as_svg(self, plt, filename):
        """保存为 SVG（矢量图）"""
        path = f"{self.output_dir}/{filename}.svg"
        plt.savefig(path, format='svg', bbox_inches='tight')
        return path
    def save_as_html(self, fig, filename):
        """保存为 HTML（交互式）"""
        path = f"{self.output_dir}/{filename}.html"
        fig.write_html(path, include_plotlyjs='cdn')
        return path

OpenClaw 可以基于分析结果自动生成结构化的 Markdown 报告。

def generate_analysis_report(data, output_path):
    """生成数据分析报告"""
    report = f"""# 数据分析报告
## 1. 数据概览
### 基本信息
| 指标 | 值 |
|------|-----|
| 数据行数 | {len(data):,} |
| 数据列数 | {len(data.columns)} |
| 内存占用 | {data.memory_usage(deep=True).sum() / 1024 / 1024:.2f} MB |
| 缺失值比例 | {data.isnull().sum().sum() / (len(data) * len(data.columns)) * 100:.2f}% |
### 数据类型分布
"""
    # 数据类型统计
    dtype_counts = data.dtypes.value_counts()
    for dtype, count in dtype_counts.items():
        report += f"- **{dtype}**: {count} 列n"
    report += f"""
## 2. 描述性统计
### 数值型变量统计
"""
    # 数值型变量统计
    numeric_stats = data.describe().T
    report += "| 变量 | 计数 | 均值 | 标准差 | 最小值 | 中位数 | 最大值 |n"
    report += "|------|------|------|--------|--------|--------|--------|n"
    for col in numeric_stats.index:
        row = numeric_stats.loc[col]
        report += f"| {col} | {row['count']:.0f} | {row['mean']:.2f} | "
        report += f"{row['std']:.2f} | {row['min']:.2f} | {row['50%']:.2f} | {row['max']:.2f} |n"
    report += f"""
## 3. 数据质量报告
### 缺失值分析
"""
    # 缺失值分析
    missing = data.isnull().sum()
    missing = missing[missing > 0].sort_values(ascending=False)
    if len(missing) > 0:
        report += "| 列名 | 缺失数量 | 缺失比例 |n"
        report += "|------|----------|----------|n"
        for col, count in missing.items():
            pct = count / len(data) * 100
            report += f"| {col} | {count} | {pct:.2f}% |n"
    else:
        report += "✅ 无缺失值n"
    report += f"""
### 异常值检测
基于 IQR 方法检测数值型变量的异常值：
"""
    # 异常值检测
    for col in data.select_dtypes(include=[np.number]).columns:
        Q1 = data[col].quantile(0.25)
        Q3 = data[col].quantile(0.75)
        IQR = Q3 - Q1
        outliers = ((data[col] < Q1 - 1.5 * IQR) | (data[col] > Q3 + 1.5 * IQR)).sum()
        if outliers > 0:
            report += f"- **{col}**: {outliers} 个异常值 ({outliers/len(data)*100:.2f}%)n"
    # 保存报告
    with open(output_path, 'w', encoding='utf-8') as f:
        f.write(report)
    return output_path

class ReportGenerator:
    """完整的报告生成器"""
    def __init__(self, data, title="数据分析报告"):
        self.data = data
        self.title = title
        self.sections = []
    def add_section(self, title, content, chart_path=None):
        """添加报告章节"""
        section = {
            'title': title,
            'content': content,
            'chart_path': chart_path
        }
        self.sections.append(section)
    def add_summary_section(self):
        """添加摘要章节"""
        summary = f"""## 执行摘要
本报告基于 {len(self.data):,} 条数据记录进行分析。
### 关键发现
"""
        # 自动生成关键发现
        numeric_cols = self.data.select_dtypes(include=[np.number]).columns
        for col in numeric_cols[:3]:  # 前3个数值列
            mean_val = self.data[col].mean()
            max_val = self.data[col].max()
            min_val = self.data[col].min()
            summary += f"- **{col}**: 平均值 {mean_val:.2f}，范围 [{min_val:.2f}, {max_val:.2f}]n"
        self.add_section("执行摘要", summary)
    def add_trend_section(self, date_col, value_col):
        """添加趋势分析章节"""
        content = f"""## 趋势分析
分析 {value_col} 随时间的变化趋势：
- 整体趋势：{'上升' if self.data[value_col].iloc[-1] > self.data[value_col].iloc[0] else '下降'}
- 平均值：{self.data[value_col].mean():.2f}
- 最大值：{self.data[value_col].max():.2f}
- 最小值：{self.data[value_col].min():.2f}
"""
        self.add_section("趋势分析", content)
    def generate(self, output_path):
        """生成完整报告"""
        report = f"# {self.title}nn"
        report += f"*生成时间: {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')}*nn"
        report += "---nn"
        for section in self.sections:
            report += section['content'] + "nn"
            if section['chart_path']:
                report += f"![{section['title']}]({section['chart_path']})nn"
        report += "---nn"
        report += "*本报告由 OpenClaw 自动生成*n"
        with open(output_path, 'w', encoding='utf-8') as f:
            f.write(report)
        return output_path
# 使用示例
# generator = ReportGenerator(df, "销售数据分析报告")
# generator.add_summary_section()
# generator.add_trend_section('date', 'sales')
# generator.generate('/reports/sales_analysis.md')

OpenClaw 集成了飞书文档功能，可以直接将报告发送到飞书：

# 用户指令
"分析销售数据，生成报告并发送到飞书"

OpenClaw 执行流程：

┌─────────────────────────────────────────────────────────────────────┐
│                    报告发送到飞书流程                                  │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│   1. 数据分析                                                        │
│      └── 执行数据处理和统计分析                                      │
│                                                                     │
│   2. 生成图表                                                        │
│      └── 创建可视化图表并保存                                        │
│                                                                     │
│   3. 生成报告                                                        │
│      └── 创建 Markdown 格式的分析报告                                │
│                                                                     │
│   4. 发送到飞书                                                      │
│      ├── 使用 feishu_doc 工具创建文档                               │
│      └── 或使用 feishu_wiki 添加到知识库                             │
│                                                                     │
│   5. 通知用户                                                        │
│      └── 通过飞书/Telegram等通道发送完成通知                          │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

代码示例：

# OpenClaw 内部会调用飞书工具
# 1. 创建飞书文档
# feishu_doc action=create title="销售数据分析报告"
# 2. 写入报告内容
# feishu_doc action=write doc_token="xxx" content="# 分析报告..."
# 3. 或者追加到现有文档
# feishu_doc action=append doc_token="xxx" content="新章节..."

OpenClaw 的 Cron 功能可以实现定时数据分析：

// ~/.openclaw/openclaw.json
{
  "cron": {
    "jobs": [
      {
        "id": "daily-sales-report",
        "schedule": "0 8 * * *",
        "message": "分析昨天的销售数据，生成日报并发送到飞书",
        "channel": "feishu",
        "enabled": true
      },
      {
        "id": "weekly-metrics",
        "schedule": "0 9 * * 1",
        "message": "生成上周的关键指标分析报告",
        "channel": "telegram",
        "enabled": true
      },
      {
        "id": "monthly-trend",
        "schedule": "0 10 1 * *",
        "message": "分析上月数据趋势，生成月度对比报告",
        "enabled": true
      }
    ]
  }
}

# 添加定时分析任务
openclaw cron add 
  --schedule "0 8 * * *" 
  --message "每日销售数据分析" 
  --channel "feishu"
# 查看所有定时任务
openclaw cron list
# 禁用某个任务
openclaw cron disable daily-sales-report
# 启用任务
openclaw cron enable daily-sales-report
# 删除任务
openclaw cron remove daily-sales-report

Heartbeat 机制可以实现周期性的数据监控：

在 HEARTBEAT.md 中配置监控任务：

# Heartbeat 任务清单
## 每次检查
- [ ] 检查数据源是否有更新
- [ ] 验证关键指标是否在正常范围
## 定时任务
### 每4小时
- 检查系统健康状态
- 验证数据同步状态
### 每天 9:00
- 生成昨日数据摘要
- 检查异常数据
### 每周一
- 生成周报
- 数据趋势分析

class DataMonitor:
    """数据监控类"""
    def __init__(self, data_source, threshold_rules):
        self.data_source = data_source
        self.threshold_rules = threshold_rules
    def check_freshness(self, max_age_hours=24):
        """检查数据新鲜度"""
        latest = self.get_latest_record()
        age = datetime.now() - latest['timestamp']
        if age.total_seconds() / 3600 > max_age_hours:
            return {
                'status': 'warning',
                'message': f'数据已过期 {age}'
            }
        return {'status': 'ok'}
    def check_anomalies(self):
        """检查数据异常"""
        alerts = []
        for rule in self.threshold_rules:
            value = self.get_metric(rule['metric'])
            if value < rule['min'] or value > rule['max']:
                alerts.append({
                    'metric': rule['metric'],
                    'value': value,
                    'expected': f"[{rule['min']}, {rule['max']}]",
                    'severity': rule.get('severity', 'warning')
                })
        return alerts
    def generate_alert_message(self, alerts):
        """生成告警消息"""
        if not alerts:
            return "✅ 所有指标正常"
        message = "⚠️ 发现以下异常：nn"
        for alert in alerts:
            message += f"- **{alert['metric']}**: 当前值 {alert['value']}, "
            message += f"期望范围 {alert['expected']}n"
        return message
# 使用示例
# monitor = DataMonitor(
#     data_source='sales_db',
#     threshold_rules=[
#         {'metric': 'daily_revenue', 'min': 10000, 'max': 100000, 'severity': 'critical'},
#         {'metric': 'conversion_rate', 'min': 0.01, 'max': 0.5, 'severity': 'warning'}
#     ]
# )
# alerts = monitor.check_anomalies()
# message = monitor.generate_alert_message(alerts)

class AnalysisPipeline:
    """数据分析流水线"""
    def __init__(self, name):
        self.name = name
        self.steps = []
        self.results = {}
    def add_step(self, name, function, **kwargs):
        """添加处理步骤"""
        self.steps.append({
            'name': name,
            'function': function,
            'params': kwargs
        })
    def run(self):
        """执行流水线"""
        print(f"开始执行流水线: {self.name}")
        for i, step in enumerate(self.steps):
            print(f"  [{i+1}/{len(self.steps)}] {step['name']}...")
            try:
                result = step['function'](**step['params'])
                self.results[step['name']] = result
                print(f"  ✓ 完成")
            except Exception as e:
                print(f"  ✗ 失败: {e}")
                return False
        print(f"流水线执行完成")
        return True
    def get_results(self):
        """获取所有结果"""
        return self.results
# 创建分析流水线
# pipeline = AnalysisPipeline("每日销售分析")
# 
# pipeline.add_step("extract", extract_sales_data, date='yesterday')
# pipeline.add_step("transform", clean_data, rules=['remove_nulls', 'normalize'])
# pipeline.add_step("analyze", analyze_trends, metrics=['revenue', 'orders'])
# pipeline.add_step("visualize", create_charts, output_dir='/reports')
# pipeline.add_step("report", generate_report, template='daily')
# pipeline.add_step("notify", send_to_feishu, channel='daily-reports')
# 
# pipeline.run()

class IncrementalProcessor:
    """增量数据处理器"""
    def __init__(self, checkpoint_file):
        self.checkpoint_file = checkpoint_file
        self.last_processed = self.load_checkpoint()
    def load_checkpoint(self):
        """加载处理进度"""
        try:
            with open(self.checkpoint_file, 'r') as f:
                return json.load(f)['last_processed']
        except:
            return None
    def save_checkpoint(self, timestamp):
        """保存处理进度"""
        with open(self.checkpoint_file, 'w') as f:
            json.dump({'last_processed': timestamp}, f)
    def process_incremental(self, data_source):
        """处理增量数据"""
        # 获取新数据
        new_data = self.fetch_new_data(
            data_source, 
            since=self.last_processed
        )
        if new_data.empty:
            print("无新数据需要处理")
            return None
        # 处理数据
        processed = self.transform(new_data)
        # 更新进度
        self.save_checkpoint(processed['timestamp'].max())
        return processed
    def fetch_new_data(self, source, since):
        """获取新数据（需根据实际数据源实现）"""
        # 示例：从数据库获取
        # query = f"SELECT * FROM {source} WHERE created_at > '{since}'"
        # return pd.read_sql(query, connection)
        pass
    def transform(self, data):
        """转换数据"""
        return data

import sqlite3
import pandas as pd
class DatabaseConnector:
    """数据库连接器"""
    def __init__(self, db_path):
        self.db_path = db_path
        self.connection = None
    def connect(self):
        """建立连接"""
        self.connection = sqlite3.connect(self.db_path)
    def query(self, sql, params=None):
        """执行查询"""
        if params:
            return pd.read_sql_query(sql, self.connection, params=params)
        return pd.read_sql_query(sql, self.connection)
    def execute(self, sql, params=None):
        """执行更新"""
        cursor = self.connection.cursor()
        if params:
            cursor.execute(sql, params)
        else:
            cursor.execute(sql)
        self.connection.commit()
        return cursor.rowcount
    def close(self):
        """关闭连接"""
        if self.connection:
            self.connection.close()
# 使用示例
# db = DatabaseConnector('/data/analytics.db')
# db.connect()
# df = db.query("SELECT * FROM sales WHERE date > '2024-01-01'")
# db.close()

OpenClaw 可以通过 Webhook 接收外部数据触发分析：

// ~/.openclaw/openclaw.json
{
  "webhooks": [
    {
      "path": "/webhook/data-update",
      "secret": "your-webhook-secret",
      "action": {
        "type": "agent",
        "message": "收到数据更新通知，执行增量分析"
      }
    }
  ]
}

场景：每周自动分析销售数据，生成报告并发送到飞书

# 用户指令
"分析 /Users/anyi/data/weekly_sales.csv 的销售数据"
# OpenClaw 自动执行
import pandas as pd
import matplotlib.pyplot as plt
# 1. 读取数据
df = pd.read_csv('/Users/anyi/data/weekly_sales.csv')
df['date'] = pd.to_datetime(df['date'])
# 2. 数据概览
print(f"数据行数: {len(df)}")
print(f"时间范围: {df['date'].min()} 至 {df['date'].max()}")
print(f"销售总额: ${df['amount'].sum():,.2f}")

# 3. 数据清洗
df_clean = df.dropna(subset=['amount', 'product'])
df_clean = df_clean[df_clean['amount'] > 0]
# 4. 统计分析
# 按产品类别统计
category_stats = df_clean.groupby('category').agg({
    'amount': ['sum', 'mean', 'count'],
    'quantity': 'sum'
}).round(2)
# 按日期统计趋势
daily_trend = df_clean.groupby('date').agg({
    'amount': 'sum',
    'order_id': 'nunique'
}).reset_index()
# 5. 计算关键指标
total_revenue = df_clean['amount'].sum()
avg_order_value = df_clean['amount'].mean()
top_products = df_clean.groupby('product')['amount'].sum().nlargest(5)

# 6. 创建图表
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
# 趋势图
axes[0, 0].plot(daily_trend['date'], daily_trend['amount'])
axes[0, 0].set_title('每日销售趋势')
axes[0, 0].tick_params(axis='x', rotation=45)
# 品类分布
category_revenue = df_clean.groupby('category')['amount'].sum()
axes[0, 1].pie(category_revenue.values, labels=category_revenue.index, autopct='%1.1f%%')
axes[0, 1].set_title('品类收入占比')
# Top 5 产品
top_products.plot(kind='bar', ax=axes[1, 0], color='steelblue')
axes[1, 0].set_title('Top 5 产品销售额')
# 订单金额分布
axes[1, 1].hist(df_clean['amount'], bins=30, color='steelblue', edgecolor='white')
axes[1, 1].set_title('订单金额分布')
plt.tight_layout()
plt.savefig('/Users/anyi/reports/sales_dashboard.png', dpi=150)

# 7. 生成报告
report = f"""# 销售数据分析报告
## 概览
- 分析周期: {df_clean['date'].min().strftime('%Y-%m-%d')} 至 {df_clean['date'].max().strftime('%Y-%m-%d')}
- 总订单数: {len(df_clean):,}
- 总销售额: ${total_revenue:,.2f}
- 平均订单金额: ${avg_order_value:,.2f}
## 品类分析
{category_stats.to_markdown()}
## Top 5 产品
{top_products.to_markdown()}
## 趋势分析
![销售趋势仪表盘](sales_dashboard.png)
---
*报告生成时间: {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')}*
"""
# 保存报告
with open('/Users/anyi/reports/sales_report.md', 'w') as f:
    f.write(report)
# 8. OpenClaw 发送到飞书
# 使用 feishu_doc 工具创建文档并写入内容

场景：每日监控网站流量，自动检测异常并告警

// ~/.openclaw/openclaw.json
{
  "cron": {
    "jobs": [
      {
        "id": "traffic-monitor",
        "schedule": "0 */4 * * *",
        "message": "检查网站流量数据，如发现异常自动告警",
        "enabled": true
      }
    ]
  }
}

class TrafficMonitor:
    """网站流量监控"""
    def __init__(self, api_key, site_id):
        self.api_key = api_key
        self.site_id = site_id
        self.history = []
        self.baseline = None
    def fetch_current_metrics(self):
        """获取当前指标"""
        # 调用分析 API
        # response = requests.get(
        #     f"https://api.analytics.com/sites/{self.site_id}/metrics",
        #     headers={"Authorization": f"Bearer {self.api_key}"}
        # )
        # return response.json()
        # 模拟数据
        return {
            'timestamp': datetime.now(),
            'visitors': 1250,
            'pageviews': 3500,
            'bounce_rate': 0.35,
            'avg_session_duration': 180
        }
    def check_anomalies(self, metrics):
        """检测异常"""
        alerts = []
        if self.baseline:
            # 访问量异常
            if metrics['visitors'] < self.baseline['visitors'] * 0.5:
                alerts.append({
                    'type': 'traffic_drop',
                    'severity': 'critical',
                    'message': f"访问量下降 50%，当前: {metrics['visitors']}"
                })
            if metrics['visitors'] > self.baseline['visitors'] * 2:
                alerts.append({
                    'type': 'traffic_spike',
                    'severity': 'warning',
                    'message': f"访问量激增，当前: {metrics['visitors']}"
                })
            # 跳出率异常
            if metrics['bounce_rate'] > 0.6:
                alerts.append({
                    'type': 'high_bounce',
                    'severity': 'warning',
                    'message': f"跳出率过高: {metrics['bounce_rate']:.1%}"
                })
        return alerts
    def update_baseline(self, metrics):
        """更新基线"""
        self.history.append(metrics)
        # 保留最近7天数据
        if len(self.history) > 7:
            self.history = self.history[-7:]
        # 计算基线
        if len(self.history) >= 3:
            self.baseline = {
                'visitors': np.mean([m['visitors'] for m in self.history]),
                'bounce_rate': np.mean([m['bounce_rate'] for m in self.history])
            }
    def run_check(self):
        """执行检查"""
        metrics = self.fetch_current_metrics()
        alerts = self.check_anomalies(metrics)
        self.update_baseline(metrics)
        return {
            'metrics': metrics,
            'alerts': alerts,
            'status': 'alert' if alerts else 'normal'
        }
# 使用示例
# monitor = TrafficMonitor(api_key='xxx', site_id='xxx')
# result = monitor.run_check()
# 
# if result['alerts']:
#     # 通过 OpenClaw 发送告警
#     for alert in result['alerts']:
#         send_alert_to_feishu(alert)

场景：每周一自动生成分部门绩效周报

class WeeklyReportGenerator:
    """周报生成器"""
    def __init__(self, data_sources, output_config):
        self.data_sources = data_sources
        self.output_config = output_config
    def collect_data(self):
        """收集数据"""
        data = {}
        for name, source in self.data_sources.items():
            if source['type'] == 'csv':
                data[name] = pd.read_csv(source['path'])
            elif source['type'] == 'database':
                # db = DatabaseConnector(source['connection'])
                # data[name] = db.query(source['query'])
                pass
        return data
    def analyze_department(self, dept_data):
        """分析部门数据"""
        return {
            'total_tasks': len(dept_data),
            'completed': (dept_data['status'] == 'completed').sum(),
            'pending': (dept_data['status'] == 'pending').sum(),
            'completion_rate': (dept_data['status'] == 'completed').mean(),
            'avg_score': dept_data['score'].mean()
        }
    def generate_department_report(self, dept_name, analysis):
        """生成部门报告"""
        return f"""### {dept_name}
| 指标 | 数值 |
|------|------|
| 总任务数 | {analysis['total_tasks']} |
| 已完成 | {analysis['completed']} |
| 进行中 | {analysis['pending']} |
| 完成率 | {analysis['completion_rate']:.1%} |
| 平均得分 | {analysis['avg_score']:.1f} |
"""
    def generate(self):
        """生成完整报告"""
        # 收集数据
        data = self.collect_data()
        # 生成报告头
        report = f"""# 周度绩效报告
*报告周期: {(datetime.now() - timedelta(days=7)).strftime('%Y-%m-%d')} 至 {datetime.now().strftime('%Y-%m-%d')}*
---
## 部门绩效概览
"""
        # 分析各部门
        for dept_name, dept_data in data.items():
            analysis = self.analyze_department(dept_data)
            report += self.generate_department_report(dept_name, analysis)
        # 添加总结
        report += """---
## 本周总结
### 主要成就
- [待补充]
### 待改进事项
- [待补充]
### 下周计划
- [待补充]
---
*报告由 OpenClaw 自动生成*
"""
        return report
# 配置并运行
# generator = WeeklyReportGenerator(
#     data_sources={
#         '技术部': {'type': 'csv', 'path': '/data/tech_tasks.csv'},
#         '产品部': {'type': 'csv', 'path': '/data/product_tasks.csv'},
#         '运营部': {'type': 'csv', 'path': '/data/ops_tasks.csv'}
#     },
#     output_config={'channel': 'feishu', 'wiki_id': 'xxx'}
# )
# 
# report = generator.generate()

| 实践 | 说明 |
|------|------|
| **敏感数据脱敏** | 分析前对 PII 数据进行脱敏处理 |
| **数据隔离** | 使用沙箱环境处理外部数据 |
| **访问控制** | 限制 OpenClaw 对敏感目录的访问权限 |
| **日志审计** | 记录所有数据处理操作 |

# 大数据处理优化建议
# 1. 使用分块读取
def process_large_csv(file_path, chunk_size=10000):
    """分块处理大文件"""
    results = []
    for chunk in pd.read_csv(file_path, chunksize=chunk_size):
        # 处理每个块
        processed = chunk.groupby('category').sum()
        results.append(processed)
    return pd.concat(results).groupby(level=0).sum()
# 2. 使用适当的数据类型
def optimize_dtypes(df):
    """优化数据类型减少内存"""
    for col in df.columns:
        if df[col].dtype == 'int64':
            if df[col].min() >= 0:
                if df[col].max() < 255:
                    df[col] = df[col].astype('uint8')
                elif df[col].max() < 65535:
                    df[col] = df[col].astype('uint16')
        elif df[col].dtype == 'float64':
            df[col] = df[col].astype('float32')
    return df
# 3. 避免重复计算
# 使用缓存存储中间结果
from functools import lru_cache
@lru_cache(maxsize=128)
def get_aggregated_data(key):
    """缓存聚合结果"""
    # 执行聚合查询
    pass

┌─────────────────────────────────────────────────────────────────────┐
│                      图表类型选择指南                                 │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│   数据类型              推荐图表                                     │
│   ─────────────────────────────────────────────                     │
│   时间序列              折线图、面积图                                │
│   分类比较              柱状图、条形图                                │
│   占比关系              饼图、环形图、堆叠柱状图                      │
│   相关性分析            散点图、气泡图                                │
│   分布情况              直方图、箱线图、密度图                       │
│   多维度对比            热力图、雷达图                                │
│   地理位置              地图、地理热力图                              │
│   流程关系              桑基图、漏斗图                                │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

# 图表设计最佳实践
# 1. 使用清晰的标题和标签
plt.title('2024年第一季度各产品线销售额', fontsize=14, fontweight='bold')
plt.xlabel('产品线', fontsize=12)
plt.ylabel('销售额（万元）', fontsize=12)
# 2. 合理的颜色选择
# 使用色盲友好的调色板
colors = ['#1f77b4', '#ff7f0e', '#2ca02c', '#d62728', '#9467bd']
# 3. 避免过度装饰
# 保持简洁，去除不必要的边框和网格线
plt.gca().spines['top'].set_visible(False)
plt.gca().spines['right'].set_visible(False)
plt.grid(axis='y', alpha=0.3)
# 4. 适当的数据标签
for bar in bars:
    height = bar.get_height()
    plt.annotate(f'{height:,.0f}',
                xy=(bar.get_x() + bar.get_width() / 2, height),
                ha='center', va='bottom', fontsize=10)
# 5. 合理的图表尺寸
plt.figure(figsize=(12, 6))  # 根据内容量调整

# [报告标题]
## 执行摘要
- 关键结论（1-3条）
- 核心指标概览
## 数据概览
- 数据来源说明
- 数据范围和限制
- 数据质量评估
## 详细分析
### 分析维度1
### 分析维度2
### 分析维度3
## 可视化展示
[图表说明]
## 结论与建议
- 主要发现
- 改进建议
- 后续行动项
## 附录
- 数据字典
- 方法说明
- 参考资料

| 检查项 | 说明 |
|--------|------|
| **数据准确性** | 验证所有数字和计算 |
| **图表可读性** | 确保图表标题、标签清晰 |
| **逻辑连贯性** | 结论需有数据支撑 |
| **时效性** | 注明数据截止时间 |
| **可复现性** | 记录分析方法，便于追溯 |

def safe_analyze(data, analysis_func):
    """安全的分析包装器"""
    try:
        result = analysis_func(data)
        return {'success': True, 'data': result}
    except Exception as e:
        error_msg = f"分析失败: {str(e)}"
        # 记录错误
        log_error(error_msg)
        # 发送告警
        send_alert(error_msg)
        return {'success': False, 'error': error_msg}

# 限制并发和资源使用
import resource
def limit_memory(max_mb):
    """限制内存使用"""
    resource.setrlimit(
        resource.RLIMIT_AS,
        (max_mb * 1024 * 1024, max_mb * 1024 * 1024)
    )
# 使用上下文管理器
class TimeoutContext:
    """超时控制"""
    def __init__(self, seconds):
        self.seconds = seconds
    def __enter__(self):
        self.start_time = time.time()
        return self
    def __exit__(self, *args):
        if time.time() - self.start_time > self.seconds:
            raise TimeoutError("分析超时")

// 自动化任务监控配置
{
  "monitoring": {
    "alert_channels": ["feishu", "telegram"],
    "rules": [
      {
        "type": "execution_time",
        "threshold": 300,
        "action": "alert"
      },
      {
        "type": "memory_usage",
        "threshold": "80%",
        "action": "alert"
      },
      {
        "type": "error_rate",
        "threshold": 0.1,
        "action": "alert_and_pause"
      }
    ]
  }
}

OpenClaw 的数据分析与可视化能力涵盖了从数据获取、处理、分析到报告输出的完整流程。通过自然语言交互，用户可以轻松完成复杂的数据分析任务，并通过多种渠道（飞书、Telegram 等）自动接收分析报告。

1. 自然语言驱动：无需编程知识，用自然语言描述需求即可
2. 多数据源支持：本地文件、API、数据库、网页等多种数据源
3. 智能分析：自动识别数据类型、推荐分析方法
4. 自动化输出：定时任务、心跳监控、自动告警
5. 多通道集成：报告可发送到飞书、Telegram、Discord 等平台

• 📊 业务数据分析：销售报告、运营指标监控
• 📈 趋势分析：市场趋势、用户行为分析
• ⚠️ 异常监控：数据异常检测、告警通知
• 📝 自动化报告：日报、周报、月报自动生成
• 🔍 探索性分析：快速了解新数据集特征

到此这篇关于OpenClaw数据分析与可视化的文章就介绍到这了,更多相关OpenClaw数据分析与可视化内容请搜索风君子博客以前的文章或继续浏览下面的相关文章，希望大家以后多多支持风君子博客！