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

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

1. 概述

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

1.1 核心数据分析能力

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

1.2 分析能力矩阵

1.3 与传统工具的对比

2. 数据处理能力

2.1 数据源接入

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

2.1.1 本地文件读取

# 让 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())

2.1.2 Excel 文件处理

# 读取 Excel 文件import pandas as pd# 单个 Sheetdf = pd.read_excel('/data/report.xlsx', sheet_name='Sales')# 多个 Sheetall_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}")

2.1.3 JSON 数据处理

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

2.1.4 API 数据获取

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

# 示例：获取股票数据 API# 用户：帮我获取苹果公司最近30天的股价数据并分析趋势import requestsimport pandas as pd# OpenClaw 可以直接调用 APIresponse = 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()

2.2 数据清洗

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

2.2.1 缺失值处理

import pandas as pdimport 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

2.2.2 异常值检测与处理

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 outliersdef 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

2.2.3 数据类型转换

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

2.3 数据转换

2.3.1 数据透视与聚合

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'# )

2.3.2 时间序列处理

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

2.4 实际操作示例

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

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

OpenClaw 会自动执行以下步骤：

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

3. 图表生成与可视化

3.1 基础图表生成

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

3.1.1 折线图

import matplotlib.pyplot as pltimport pandas as pdimport numpy as npdef 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')

3.1.2 柱状图

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

3.1.3 饼图

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

3.1.4 散点图

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

3.2 高级可视化

3.2.1 热力图

import seaborn as snsdef 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

3.2.2 箱线图

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

3.2.3 多子图仪表盘

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

3.3 交互式图表（通过 Canvas）

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

# 通过 Canvas 展示 HTML 图表import plotly.express as pximport plotly.graph_objects as godef 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')

3.4 图表保存与输出

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

4. 报告自动生成

4.1 Markdown 报告模板

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

4.2 完整报告生成示例

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')

4.3 通过 Feishu 发送报告

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="新章节..."

5. 自动化数据分析工作流

5.1 使用 Cron 实现定时分析

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

5.1.1 配置定时分析任务

// ~/.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      }    ]  }}

5.1.2 通过 CLI 管理任务

# 添加定时分析任务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

5.2 使用 Heartbeat 实现监控分析

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

5.2.1 配置 Heartbeat

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

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

5.2.2 数据监控脚本

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)

5.3 自动化分析流水线

5.3.1 ETL 流程设计

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()

5.3.2 增量数据处理

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

5.4 与外部系统集成

5.4.1 数据库连接

import sqlite3import pandas as pdclass 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()

5.4.2 Webhook 数据接收

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

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

6. 实战案例

6.1 案例一：销售数据分析

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

步骤一：准备数据

# 用户指令"分析 /Users/anyi/data/weekly_sales.csv 的销售数据"# OpenClaw 自动执行import pandas as pdimport 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 工具创建文档并写入内容

6.2 案例二：网站流量监控

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

配置定时监控任务

// ~/.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)

6.3 案例三：自动化周报生成

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

完整工作流

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()

7. 最佳实践与注意事项

7.1 数据处理最佳实践

7.1.1 数据安全

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

7.1.2 性能优化

# 大数据处理优化建议# 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

7.2 可视化最佳实践

7.2.1 图表选择指南

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

7.2.2 图表设计原则

# 图表设计最佳实践# 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))  # 根据内容量调整

7.3 报告生成最佳实践

7.3.1 报告结构模板

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

7.3.2 报告质量控制

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

7.4 自动化注意事项

7.4.1 错误处理

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}

7.4.2 资源管理

# 限制并发和资源使用import resourcedef 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("分析超时")

7.4.3 监控与告警

// 自动化任务监控配置{  "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 等平台

适用场景

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