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

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

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

# 示例：获取股票数据 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()

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：
"帮我分析 /Users/anyi/data/orders.csv 文件，清洗数据并生成统计报告"

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

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

# 通过 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

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}\n\n"
        report += f"*生成时间: {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')}*\n\n"
        report += "---\n\n"
        for section in self.sections:
            report += section['content'] + "\n\n"
            if section['chart_path']:
                report += f"![{section['title']}]({section['chart_path']})\n\n"
        report += "---\n\n"
        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')

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

┌─────────────────────────────────────────────────────────────────────┐
│                    报告发送到飞书流程                                  │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│   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/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 任务清单
## 每次检查
- [ ] 检查数据源是否有更新
- [ ] 验证关键指标是否在正常范围
## 定时任务
### 每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 = "⚠️ 发现以下异常：\n\n"
        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/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"
      }
    ]
  }
}