OpenClaw 数据处理流水线：抓取、清洗、分析、可视化全自动
引言
数据分析的日常工作中，大量时间花在了数据获取和清洗上。你可能需要从多个网站抓取数据，把各种格式的 Excel、CSV、JSON 文件统一格式，做统计计算，最后生成图表放到报告里。这些步骤每一步都不难，但串起来就是大量的重复劳动。
OpenClaw 擅长的正是这类多步骤的工具协调任务。它可以用 web_fetch 抓取网页数据，用 run_command 执行 Python 脚本进行清洗和分析，用文件操作工具保存结果。整个流水线配置一次，之后自动运行。
场景描述：竞品价格监控分析
假设你是一家电商公司的数据分析师，需要每天监控竞品在各大平台上的价格变化，生成分析报告给运营团队。传统做法是手动打开多个网站记录价格，然后用 Excel 做对比分析。用 OpenClaw 可以把这个流程完全自动化。
第一步：数据抓取
配置 OpenClaw 的数据抓取工具，从多个数据源获取信息：
# tools/data_fetch.yml
- name: fetch_price_data
description: "从指定 URL 抓取商品价格数据"
type: web_fetch
config:
url: "{{target_url}}"
method: GET
headers:
User-Agent: "Mozilla/5.0 (compatible; DataBot/1.0)"
extract: "json"
parameters:
target_url:
type: string
description: "目标数据 API 地址"
- name: fetch_exchange_rate
description: "获取当前汇率数据"
type: web_fetch
config:
url: "https://api.exchangerate.host/latest?base=CNY"
extract: "json"对于没有公开 API 的数据源，可以使用 OpenClaw 的 browser 工具模拟浏览器访问：
# tools/browser_scrape.yml
- name: scrape_competitor_page
description: "用浏览器抓取竞品页面的价格信息"
type: browser
config:
actions:
- navigate: "{{page_url}}"
- wait: ".price-tag"
- extract:
selector: ".price-tag"
attribute: "text"
- screenshot: "/tmp/competitor_{{date}}.png"第二步：数据清洗与转换
抓取的原始数据格式各异，需要统一处理。我们用 OpenClaw 调用 Python 脚本来完成清洗工作：
# scripts/clean_data.py
import json
import csv
import sys
from datetime import datetime
def clean_price_data(raw_data_path, output_path):
with open(raw_data_path, 'r') as f:
raw = json.load(f)
cleaned = []
for item in raw:
cleaned.append({
'product_name': item.get('name', '').strip(),
'price': float(str(item.get('price', 0))
.replace('¥', '').replace(',', '')),
'platform': item.get('source', 'unknown'),
'timestamp': datetime.now().isoformat(),
'currency': 'CNY'
})
with open(output_path, 'w', newline='') as f:
writer = csv.DictWriter(f,
fieldnames=['product_name', 'price',
'platform', 'timestamp', 'currency'])
writer.writeheader()
writer.writerows(cleaned)
print(f"Cleaned {len(cleaned)} records")
if __name__ == '__main__':
clean_price_data(sys.argv[1], sys.argv[2])在 OpenClaw 任务中调用这个脚本：
# tasks/data_pipeline.yml
name: daily_price_pipeline
schedule: "0 8 * * *" # 每天早上 8 点
prompt: |
执行每日竞品价格数据处理流水线。
第一阶段 - 数据抓取：
1. 用 fetch_price_data 从以下 API 获取数据：
- https://api.platform-a.com/products?category=electronics
- https://api.platform-b.com/search?q=wireless+earbuds
2. 用 scrape_competitor_page 抓取以下页面：
- https://competitor-c.com/products/earbuds
3. 将所有原始数据保存到 /data/raw/{{date}}.json
第二阶段 - 数据清洗：
4. 执行: python3 /scripts/clean_data.py \
/data/raw/{{date}}.json /data/clean/{{date}}.csv
第三阶段 - 数据分析：
5. 读取清洗后的 CSV 文件
6. 计算以下指标：
- 每个平台的平均价格
- 与昨日价格对比的变化幅度
- 最低价和最高价差异
- 价格趋势（近 7 天）
第四阶段 - 报告生成：
7. 生成 Markdown 格式的分析报告
8. 将报告保存到 /data/reports/{{date}}.md
9. 通过 message 工具发送摘要到运营群第三步：分析与可视化
OpenClaw 可以调用 Python 的数据分析和可视化库生成图表。准备一个图表生成脚本：
# scripts/generate_charts.py
import pandas as pd
import matplotlib.pyplot as plt
import matplotlib
matplotlib.use('Agg') # 无界面环境
plt.rcParams['font.sans-serif'] = ['SimHei']
def generate_price_chart(data_dir, output_dir, days=7):
# 读取近 7 天数据
frames = []
for i in range(days):
date = (pd.Timestamp.now()
- pd.Timedelta(days=i)).strftime('%Y-%m-%d')
try:
df = pd.read_csv(f'{data_dir}/{date}.csv')
frames.append(df)
except FileNotFoundError:
continue
if not frames:
print("No data available")
return
all_data = pd.concat(frames)
# 各平台价格趋势图
fig, ax = plt.subplots(figsize=(10, 6))
for platform in all_data['platform'].unique():
pdata = all_data[all_data['platform'] == platform]
daily_avg = pdata.groupby(
pd.to_datetime(pdata['timestamp']).dt.date
)['price'].mean()
ax.plot(daily_avg.index, daily_avg.values,
marker='o', label=platform)
ax.set_title('竞品价格趋势 (近7天)')
ax.set_xlabel('日期')
ax.set_ylabel('平均价格 (元)')
ax.legend()
plt.tight_layout()
plt.savefig(f'{output_dir}/price_trend.png', dpi=150)
print(f"Chart saved to {output_dir}/price_trend.png")实际输出效果
流水线执行完成后，运营团队会在群里收到类似这样的每日摘要：
[竞品价格日报 2026-03-23]
今日监控商品: 15 个 SKU，覆盖 3 个平台
价格变动概要:
- 平台A 无线耳机均价 ¥189，较昨日下降 3.2%
- 平台B 无线耳机均价 ¥205，较昨日持平
- 竞品C 官网均价 ¥179，较昨日下降 5.1% (疑似促销)
值得关注:
- 竞品C 的 Pro 款价格跌破 ¥200，为近 30 天最低
- 平台A 有 3 款商品出现缺货标记
详细报告: /data/reports/2026-03-23.md
趋势图表: /data/charts/price_trend.pngClawBrain 加持效果
数据流水线是典型的多步骤串行任务，任何一个环节失败都会影响后续步骤。在实际运行中，数据源不稳定是最常见的问题：网站改版、API 限流、网络超时都可能导致抓取失败。
ClawBrain 在这个场景下提供了关键的可靠性保障：
- 智能重试：当
web_fetch遇到超时或 429 限流时，ClawBrain 会自动等待适当时间后重试，而不是直接报错中断 - 降级处理：如果某个数据源完全不可用，ClawBrain 会让 OpenClaw 跳过该数据源，用历史数据补充，确保报告仍然能正常生成
- 结果验证：ClawBrain 的验证机制会检查生成的报告是否完整，数据是否合理（比如价格是否出现明显异常值），避免错误数据被推送给运营团队
接入 ClawBrain 后，流水线的每日成功运行率从 85% 提升到 98%。剩下的 2% 失败主要是因为所有数据源同时不可用，这种情况下 ClawBrain 会发送一条异常通知而不是一份不完整的报告。
总结
数据处理流水线的自动化不需要复杂的 ETL 框架。对于中小规模的数据任务，OpenClaw 提供了一种轻量但足够强大的解决方案。它的优势在于灵活性：你可以用自然语言描述数据处理逻辑，随时调整流程，不需要重写代码。
对于数据量较大或数据源较多的场景，建议配合 ClawBrain 使用。它的错误恢复和结果验证机制可以让你的数据流水线在无人值守的情况下稳定运行，减少人工干预的频率。