┌─────────────────────────────────────────────────────────────────────┐
│                    OpenClaw 數(shù)據(jù)分析能力架構(gòu)                          │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│   ┌─────────────────────────────────────────────────────────────┐   │
│   │                      數(shù)據(jù)輸入層                              │   │
│   │  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐          │   │
│   │  │ 本地文件 │ │ API數(shù)據(jù) │ │ 網(wǎng)頁抓取 │ │ 數(shù)據(jù)庫   │          │   │
│   │  │ CSV/JSON│ │ REST API│ │ Browser │ │ SQLite  │          │   │
│   │  └────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘          │   │
│   └───────┼───────────┼───────────┼───────────┼───────────────┘   │
│           │           │           │           │                    │
│           └───────────┴─────┬─────┴───────────┘                    │
│                             ▼                                      │
│   ┌─────────────────────────────────────────────────────────────┐   │
│   │                      數(shù)據(jù)處理層                              │   │
│   │  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐          │   │
│   │  │ 數(shù)據(jù)清洗 │ │ 數(shù)據(jù)轉(zhuǎn)換 │ │ 統(tǒng)計分析 │ │ 機器學(xué)習(xí) │          │   │
│   │  │ Pandas  │ │ NumPy   │ │ SciPy   │ │ Scikit  │          │   │
│   │  └─────────┘ └─────────┘ └─────────┘ └─────────┘          │   │
│   └─────────────────────────────────────────────────────────────┘   │
│                             │                                      │
│                             ▼                                      │
│   ┌─────────────────────────────────────────────────────────────┐   │
│   │                      輸出呈現(xiàn)層                              │   │
│   │  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐          │   │
│   │  │ 圖表生成 │ │ 報告輸出 │ │ 交互式  │ │ 自動化  │          │   │
│   │  │ Matplotlib│ │ Markdown│ │ Canvas  │ │ Cron   │          │   │
│   │  └─────────┘ └─────────┘ └─────────┘ └─────────┘          │   │
│   └─────────────────────────────────────────────────────────────┘   │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

# 讓 OpenClaw 讀取本地 CSV 文件
# 用戶：幫我分析 /data/sales.csv 文件
# OpenClaw 會執(zhí)行以下操作：
import pandas as pd
# 讀取 CSV 文件
df = pd.read_csv('/data/sales.csv')
# 查看數(shù)據(jù)概覽
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)
# 轉(zhuǎn)換為 DataFrame
df = pd.json_normalize(data['records'])
# 處理嵌套數(shù)據(jù)
df_flat = pd.json_normalize(
    data, 
    record_path=['items'],
    meta=['id', 'name', ['metadata', 'created_at']]
)

# 示例：獲取股票數(shù)據(jù) API
# 用戶：幫我獲取蘋果公司最近30天的股價數(shù)據(jù)并分析趨勢
import requests
import pandas as pd
# OpenClaw 可以直接調(diào)用 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')
# 計算技術(shù)指標(biāo)
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):
    """分析數(shù)據(jù)缺失情況"""
    missing = df.isnull().sum()
    percent = (missing / len(df)) * 100
    missing_df = pd.DataFrame({
        '缺失數(shù)量': missing,
        '缺失比例': percent.round(2)
    })
    return missing_df[missing_df['缺失數(shù)量'] > 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%，使用模型預(yù)測填充
            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):
    """自動推斷和轉(zhuǎn)換數(shù)據(jù)類型"""
    df_converted = df.copy()
    for col in df_converted.columns:
        # 嘗試轉(zhuǎn)換為日期
        try:
            df_converted[col] = pd.to_datetime(df_converted[col])
            continue
        except:
            pass
        # 嘗試轉(zhuǎn)換為數(shù)值
        try:
            df_converted[col] = pd.to_numeric(df_converted[col])
            continue
        except:
            pass
        # 分類數(shù)據(jù)優(yōu)化
        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
# 數(shù)據(jù)透視表示例
def create_pivot_report(df, index, columns, values, aggfunc='sum'):
    """創(chuàng)建透視表報告"""
    pivot = pd.pivot_table(
        df,
        index=index,
        columns=columns,
        values=values,
        aggfunc=aggfunc,
        fill_value=0,
        margins=True,
        margins_name='總計'
    )
    return pivot
# 示例：按地區(qū)和產(chǎn)品類別的銷售透視
# 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'):
    """時間序列數(shù)據(jù)處理"""
    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

# 用戶發(fā)送消息給 OpenClaw：
"幫我分析 /Users/anyi/data/orders.csv 文件，清洗數(shù)據(jù)并生成統(tǒng)計報告"

┌─────────────────────────────────────────────────────────────────────┐
│                    OpenClaw 數(shù)據(jù)處理流程                             │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│   1. 讀取文件                                                        │
│      └── 使用 read 工具讀取 CSV 文件內(nèi)容                              │
│                                                                     │
│   2. 數(shù)據(jù)探索                                                        │
│      └── 分析數(shù)據(jù)結(jié)構(gòu)、類型、缺失情況                                  │
│                                                                     │
│   3. 數(shù)據(jù)清洗                                                        │
│      └── 處理缺失值、異常值、重復(fù)數(shù)據(jù)                                  │
│                                                                     │
│   4. 數(shù)據(jù)轉(zhuǎn)換                                                        │
│      └── 類型轉(zhuǎn)換、特征工程、標(biāo)準化                                   │
│                                                                     │
│   5. 統(tǒng)計分析                                                        │
│      └── 描述統(tǒng)計、相關(guān)性分析、分組統(tǒng)計                               │
│                                                                     │
│   6. 結(jié)果輸出                                                        │
│      └── 生成清洗后的文件和統(tǒng)計報告                                   │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

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):
    """創(chuàng)建多系列折線圖"""
    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):
    """創(chuàng)建柱狀圖"""
    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)
        # 添加數(shù)值標(biāo)簽
        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):
    """創(chuàng)建餅圖"""
    fig, ax = plt.subplots(figsize=(10, 8))
    values = data[values_col]
    labels = data[labels_col]
    # 設(shè)置顏色
    colors = plt.cm.Set3(np.linspace(0, 1, len(values)))
    # 突出顯示最大扇區(qū)
    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):
    """創(chuàng)建散點圖"""
    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):
    """創(chuàng)建熱力圖"""
    fig, ax = plt.subplots(figsize=(12, 10))
    # 計算相關(guān)性矩陣或直接使用數(shù)據(jù)
    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):
    """創(chuàng)建箱線圖"""
    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('')  # 移除默認標(biāo)題
    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):
    """創(chuàng)建多圖表儀表盤"""
    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('區(qū)域占比', 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('價格-銷量關(guān)系', fontsize=12)
    plt.suptitle('數(shù)據(jù)分析儀表盤', 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'):
    """創(chuàng)建交互式圖表 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:
    """圖表導(dǎo)出工具"""
    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):
    """生成數(shù)據(jù)分析報告"""
    report = f"""# 數(shù)據(jù)分析報告
## 1. 數(shù)據(jù)概覽
### 基本信息
| 指標(biāo) | 值 |
|------|-----|
| 數(shù)據(jù)行數(shù) | {len(data):,} |
| 數(shù)據(jù)列數(shù) | {len(data.columns)} |
| 內(nèi)存占用 | {data.memory_usage(deep=True).sum() / 1024 / 1024:.2f} MB |
| 缺失值比例 | {data.isnull().sum().sum() / (len(data) * len(data.columns)) * 100:.2f}% |
### 數(shù)據(jù)類型分布
"""
    # 數(shù)據(jù)類型統(tǒng)計
    dtype_counts = data.dtypes.value_counts()
    for dtype, count in dtype_counts.items():
        report += f"- **{dtype}**: {count} 列\(zhòng)n"
    report += f"""
## 2. 描述性統(tǒng)計
### 數(shù)值型變量統(tǒng)計
"""
    # 數(shù)值型變量統(tǒng)計
    numeric_stats = data.describe().T
    report += "| 變量 | 計數(shù) | 均值 | 標(biāo)準差 | 最小值 | 中位數(shù) | 最大值 |\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. 數(shù)據(jù)質(zhì)量報告
### 缺失值分析
"""
    # 缺失值分析
    missing = data.isnull().sum()
    missing = missing[missing > 0].sort_values(ascending=False)
    if len(missing) > 0:
        report += "| 列名 | 缺失數(shù)量 | 缺失比例 |\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 方法檢測數(shù)值型變量的異常值：
"""
    # 異常值檢測
    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="數(shù)據(jù)分析報告"):
        self.data = data
        self.title = title
        self.sections = []
    def add_section(self, title, content, chart_path=None):
        """添加報告章節(jié)"""
        section = {
            'title': title,
            'content': content,
            'chart_path': chart_path
        }
        self.sections.append(section)
    def add_summary_section(self):
        """添加摘要章節(jié)"""
        summary = f"""## 執(zhí)行摘要
本報告基于 {len(self.data):,} 條數(shù)據(jù)記錄進行分析。
### 關(guān)鍵發(fā)現(xiàn)
"""
        # 自動生成關(guān)鍵發(fā)現(xiàn)
        numeric_cols = self.data.select_dtypes(include=[np.number]).columns
        for col in numeric_cols[:3]:  # 前3個數(shù)值列
            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("執(zhí)行摘要", summary)
    def add_trend_section(self, date_col, value_col):
        """添加趨勢分析章節(jié)"""
        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, "銷售數(shù)據(jù)分析報告")
# generator.add_summary_section()
# generator.add_trend_section('date', 'sales')
# generator.generate('/reports/sales_analysis.md')

# 用戶指令
"分析銷售數(shù)據(jù)，生成報告并發(fā)送到飛書"

┌─────────────────────────────────────────────────────────────────────┐
│                    報告發(fā)送到飛書流程                                  │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│   1. 數(shù)據(jù)分析                                                        │
│      └── 執(zhí)行數(shù)據(jù)處理和統(tǒng)計分析                                      │
│                                                                     │
│   2. 生成圖表                                                        │
│      └── 創(chuàng)建可視化圖表并保存                                        │
│                                                                     │
│   3. 生成報告                                                        │
│      └── 創(chuàng)建 Markdown 格式的分析報告                                │
│                                                                     │
│   4. 發(fā)送到飛書                                                      │
│      ├── 使用 feishu_doc 工具創(chuàng)建文檔                               │
│      └── 或使用 feishu_wiki 添加到知識庫                             │
│                                                                     │
│   5. 通知用戶                                                        │
│      └── 通過飛書/Telegram等通道發(fā)送完成通知                          │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

# OpenClaw 內(nèi)部會調(diào)用飛書工具
# 1. 創(chuàng)建飛書文檔
# feishu_doc action=create title="銷售數(shù)據(jù)分析報告"
# 2. 寫入報告內(nèi)容
# feishu_doc action=write doc_token="xxx" content="# 分析報告..."
# 3. 或者追加到現(xiàn)有文檔
# feishu_doc action=append doc_token="xxx" content="新章節(jié)..."

// ~/.openclaw/openclaw.json
{
  "cron": {
    "jobs": [
      {
        "id": "daily-sales-report",
        "schedule": "0 8 * * *",
        "message": "分析昨天的銷售數(shù)據(jù)，生成日報并發(fā)送到飛書",
        "channel": "feishu",
        "enabled": true
      },
      {
        "id": "weekly-metrics",
        "schedule": "0 9 * * 1",
        "message": "生成上周的關(guān)鍵指標(biāo)分析報告",
        "channel": "telegram",
        "enabled": true
      },
      {
        "id": "monthly-trend",
        "schedule": "0 10 1 * *",
        "message": "分析上月數(shù)據(jù)趨勢，生成月度對比報告",
        "enabled": true
      }
    ]
  }
}

# 添加定時分析任務(wù)
openclaw cron add \
  --schedule "0 8 * * *" \
  --message "每日銷售數(shù)據(jù)分析" \
  --channel "feishu"
# 查看所有定時任務(wù)
openclaw cron list
# 禁用某個任務(wù)
openclaw cron disable daily-sales-report
# 啟用任務(wù)
openclaw cron enable daily-sales-report
# 刪除任務(wù)
openclaw cron remove daily-sales-report

# Heartbeat 任務(wù)清單
## 每次檢查
- [ ] 檢查數(shù)據(jù)源是否有更新
- [ ] 驗證關(guān)鍵指標(biāo)是否在正常范圍
## 定時任務(wù)
### 每4小時
- 檢查系統(tǒng)健康狀態(tài)
- 驗證數(shù)據(jù)同步狀態(tài)
### 每天 9:00
- 生成昨日數(shù)據(jù)摘要
- 檢查異常數(shù)據(jù)
### 每周一
- 生成周報
- 數(shù)據(jù)趨勢分析

class DataMonitor:
    """數(shù)據(jù)監(jiān)控類"""
    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):
        """檢查數(shù)據(jù)新鮮度"""
        latest = self.get_latest_record()
        age = datetime.now() - latest['timestamp']
        if age.total_seconds() / 3600 > max_age_hours:
            return {
                'status': 'warning',
                'message': f'數(shù)據(jù)已過期 {age}'
            }
        return {'status': 'ok'}
    def check_anomalies(self):
        """檢查數(shù)據(jù)異常"""
        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 "? 所有指標(biāo)正常"
        message = "?? 發(fā)現(xiàn)以下異常：\n\n"
        for alert in alerts:
            message += f"- **{alert['metric']}**: 當(dāng)前值 {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:
    """數(shù)據(jù)分析流水線"""
    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):
        """執(zhí)行流水線"""
        print(f"開始執(zhí)行流水線: {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"流水線執(zhí)行完成")
        return True
    def get_results(self):
        """獲取所有結(jié)果"""
        return self.results
# 創(chuàng)建分析流水線
# 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:
    """增量數(shù)據(jù)處理器"""
    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):
        """處理增量數(shù)據(jù)"""
        # 獲取新數(shù)據(jù)
        new_data = self.fetch_new_data(
            data_source, 
            since=self.last_processed
        )
        if new_data.empty:
            print("無新數(shù)據(jù)需要處理")
            return None
        # 處理數(shù)據(jù)
        processed = self.transform(new_data)
        # 更新進度
        self.save_checkpoint(processed['timestamp'].max())
        return processed
    def fetch_new_data(self, source, since):
        """獲取新數(shù)據(jù)（需根據(jù)實際數(shù)據(jù)源實現(xiàn)）"""
        # 示例：從數(shù)據(jù)庫獲取
        # query = f"SELECT * FROM {source} WHERE created_at > '{since}'"
        # return pd.read_sql(query, connection)
        pass
    def transform(self, data):
        """轉(zhuǎn)換數(shù)據(jù)"""
        return data

import sqlite3
import pandas as pd
class DatabaseConnector:
    """數(shù)據(jù)庫連接器"""
    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):
        """執(zhí)行查詢"""
        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):
        """執(zhí)行更新"""
        cursor = self.connection.cursor()
        if params:
            cursor.execute(sql, params)
        else:
            cursor.execute(sql)
        self.connection.commit()
        return cursor.rowcount
    def close(self):
        """關(guān)閉連接"""
        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": "收到數(shù)據(jù)更新通知，執(zhí)行增量分析"
      }
    }
  ]
}

# 用戶指令
"分析 /Users/anyi/data/weekly_sales.csv 的銷售數(shù)據(jù)"
# OpenClaw 自動執(zhí)行
import pandas as pd
import matplotlib.pyplot as plt
# 1. 讀取數(shù)據(jù)
df = pd.read_csv('/Users/anyi/data/weekly_sales.csv')
df['date'] = pd.to_datetime(df['date'])
# 2. 數(shù)據(jù)概覽
print(f"數(shù)據(jù)行數(shù): {len(df)}")
print(f"時間范圍: {df['date'].min()} 至 {df['date'].max()}")
print(f"銷售總額: ${df['amount'].sum():,.2f}")

# 3. 數(shù)據(jù)清洗
df_clean = df.dropna(subset=['amount', 'product'])
df_clean = df_clean[df_clean['amount'] > 0]
# 4. 統(tǒng)計分析
# 按產(chǎn)品類別統(tǒng)計
category_stats = df_clean.groupby('category').agg({
    'amount': ['sum', 'mean', 'count'],
    'quantity': 'sum'
}).round(2)
# 按日期統(tǒng)計趨勢
daily_trend = df_clean.groupby('date').agg({
    'amount': 'sum',
    'order_id': 'nunique'
}).reset_index()
# 5. 計算關(guān)鍵指標(biāo)
total_revenue = df_clean['amount'].sum()
avg_order_value = df_clean['amount'].mean()
top_products = df_clean.groupby('product')['amount'].sum().nlargest(5)

# 6. 創(chuàng)建圖表
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 產(chǎn)品
top_products.plot(kind='bar', ax=axes[1, 0], color='steelblue')
axes[1, 0].set_title('Top 5 產(chǎn)品銷售額')
# 訂單金額分布
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"""# 銷售數(shù)據(jù)分析報告
## 概覽
- 分析周期: {df_clean['date'].min().strftime('%Y-%m-%d')} 至 {df_clean['date'].max().strftime('%Y-%m-%d')}
- 總訂單數(shù): {len(df_clean):,}
- 總銷售額: ${total_revenue:,.2f}
- 平均訂單金額: ${avg_order_value:,.2f}
## 品類分析
{category_stats.to_markdown()}
## Top 5 產(chǎn)品
{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 發(fā)送到飛書
# 使用 feishu_doc 工具創(chuàng)建文檔并寫入內(nèi)容

// ~/.openclaw/openclaw.json
{
  "cron": {
    "jobs": [
      {
        "id": "traffic-monitor",
        "schedule": "0 */4 * * *",
        "message": "檢查網(wǎng)站流量數(shù)據(jù)，如發(fā)現(xiàn)異常自動告警",
        "enabled": true
      }
    ]
  }
}

class TrafficMonitor:
    """網(wǎng)站流量監(jiān)控"""
    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):
        """獲取當(dāng)前指標(biāo)"""
        # 調(diào)用分析 API
        # response = requests.get(
        #     f"https://api.analytics.com/sites/{self.site_id}/metrics",
        #     headers={"Authorization": f"Bearer {self.api_key}"}
        # )
        # return response.json()
        # 模擬數(shù)據(jù)
        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%，當(dāng)前: {metrics['visitors']}"
                })
            if metrics['visitors'] > self.baseline['visitors'] * 2:
                alerts.append({
                    'type': 'traffic_spike',
                    'severity': 'warning',
                    'message': f"訪問量激增，當(dāng)前: {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天數(shù)據(jù)
        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):
        """執(zhí)行檢查"""
        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 發(fā)送告警
#     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):
        """收集數(shù)據(jù)"""
        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):
        """分析部門數(shù)據(jù)"""
        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}
| 指標(biāo) | 數(shù)值 |
|------|------|
| 總?cè)蝿?wù)數(shù) | {analysis['total_tasks']} |
| 已完成 | {analysis['completed']} |
| 進行中 | {analysis['pending']} |
| 完成率 | {analysis['completion_rate']:.1%} |
| 平均得分 | {analysis['avg_score']:.1f} |
"""
    def generate(self):
        """生成完整報告"""
        # 收集數(shù)據(jù)
        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)
        # 添加總結(jié)
        report += """---
## 本周總結(jié)
### 主要成就
- [待補充]
### 待改進事項
- [待補充]
### 下周計劃
- [待補充]
---
*報告由 OpenClaw 自動生成*
"""
        return report
# 配置并運行
# generator = WeeklyReportGenerator(
#     data_sources={
#         '技術(shù)部': {'type': 'csv', 'path': '/data/tech_tasks.csv'},
#         '產(chǎn)品部': {'type': 'csv', 'path': '/data/product_tasks.csv'},
#         '運營部': {'type': 'csv', 'path': '/data/ops_tasks.csv'}
#     },
#     output_config={'channel': 'feishu', 'wiki_id': 'xxx'}
# )
# 
# report = generator.generate()

| 實踐 | 說明 |
|------|------|
| **敏感數(shù)據(jù)脫敏** | 分析前對 PII 數(shù)據(jù)進行脫敏處理 |
| **數(shù)據(jù)隔離** | 使用沙箱環(huán)境處理外部數(shù)據(jù) |
| **訪問控制** | 限制 OpenClaw 對敏感目錄的訪問權(quán)限 |
| **日志審計** | 記錄所有數(shù)據(jù)處理操作 |

# 大數(shù)據(jù)處理優(yōu)化建議
# 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. 使用適當(dāng)?shù)臄?shù)據(jù)類型
def optimize_dtypes(df):
    """優(yōu)化數(shù)據(jù)類型減少內(nèi)存"""
    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. 避免重復(fù)計算
# 使用緩存存儲中間結(jié)果
from functools import lru_cache
@lru_cache(maxsize=128)
def get_aggregated_data(key):
    """緩存聚合結(jié)果"""
    # 執(zhí)行聚合查詢
    pass

┌─────────────────────────────────────────────────────────────────────┐
│                      圖表類型選擇指南                                 │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│   數(shù)據(jù)類型              推薦圖表                                     │
│   ─────────────────────────────────────────────                     │
│   時間序列              折線圖、面積圖                                │
│   分類比較              柱狀圖、條形圖                                │
│   占比關(guān)系              餅圖、環(huán)形圖、堆疊柱狀圖                      │
│   相關(guān)性分析            散點圖、氣泡圖                                │
│   分布情況              直方圖、箱線圖、密度圖                       │
│   多維度對比            熱力圖、雷達圖                                │
│   地理位置              地圖、地理熱力圖                              │
│   流程關(guān)系              ?；鶊D、漏斗圖                                │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

# 圖表設(shè)計最佳實踐
# 1. 使用清晰的標(biāo)題和標(biāo)簽
plt.title('2024年第一季度各產(chǎn)品線銷售額', fontsize=14, fontweight='bold')
plt.xlabel('產(chǎn)品線', fontsize=12)
plt.ylabel('銷售額（萬元）', fontsize=12)
# 2. 合理的顏色選擇
# 使用色盲友好的調(diào)色板
colors = ['#1f77b4', '#ff7f0e', '#2ca02c', '#d62728', '#9467bd']
# 3. 避免過度裝飾
# 保持簡潔，去除不必要的邊框和網(wǎng)格線
plt.gca().spines['top'].set_visible(False)
plt.gca().spines['right'].set_visible(False)
plt.grid(axis='y', alpha=0.3)
# 4. 適當(dāng)?shù)臄?shù)據(jù)標(biāo)簽
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))  # 根據(jù)內(nèi)容量調(diào)整

# [報告標(biāo)題]
## 執(zhí)行摘要
- 關(guān)鍵結(jié)論（1-3條）
- 核心指標(biāo)概覽
## 數(shù)據(jù)概覽
- 數(shù)據(jù)來源說明
- 數(shù)據(jù)范圍和限制
- 數(shù)據(jù)質(zhì)量評估
## 詳細分析
### 分析維度1
### 分析維度2
### 分析維度3
## 可視化展示
[圖表說明]
## 結(jié)論與建議
- 主要發(fā)現(xiàn)
- 改進建議
- 后續(xù)行動項
## 附錄
- 數(shù)據(jù)字典
- 方法說明
- 參考資料

| 檢查項 | 說明 |
|--------|------|
| **數(shù)據(jù)準確性** | 驗證所有數(shù)字和計算 |
| **圖表可讀性** | 確保圖表標(biāo)題、標(biāo)簽清晰 |
| **邏輯連貫性** | 結(jié)論需有數(shù)據(jù)支撐 |
| **時效性** | 注明數(shù)據(jù)截止時間 |
| **可復(fù)現(xiàn)性** | 記錄分析方法，便于追溯 |

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)
        # 發(fā)送告警
        send_alert(error_msg)
        return {'success': False, 'error': error_msg}

# 限制并發(fā)和資源使用
import resource
def limit_memory(max_mb):
    """限制內(nèi)存使用"""
    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("分析超時")

// 自動化任務(wù)監(jiān)控配置
{
  "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"
      }
    ]
  }
}