用 DuckDB 搭建竞品监控 SaaS：从数据采集到自动化报告的完整指南

手把手教你用 DuckDB 搭建竞品价格监控系统，自动采集、分析、告警、生成报告，打造月入 ¥3000+ 的数据情报服务产品。含完整代码和变现策略。

用 DuckDB 搭建竞品监控 SaaS：从数据采集到自动化报告的完整指南

在电商和 SaaS 行业，谁掌握竞品信息谁就掌握定价权。但现实中，大部分中小卖家和创业者只能靠手动截图、Excel 汇总来跟踪对手——效率低、遗漏多、决策慢。

今天，我们用 DuckDB 搭建一套完整的竞品监控系统：自动采集 → 本地存储 → SQL 分析 → 智能告警 → 报告生成。这套系统你可以自己用，也可以打包成 SaaS 产品，每月向客户收取 ¥299-999 的订阅费。

竞品监控系统架构

一、为什么竞品监控能赚钱？

市场痛点

中小企业在做产品定价、功能规划时，最常参考的就是竞争对手。但传统做法有三个致命问题：

信息分散：竞品价格散落在官网、电商页面、第三方平台，没有统一视图
变化频繁：竞品经常调价、上新、改策略，手动跟踪根本跟不上
分析成本高：即使拿到数据，怎么对比、怎么发现趋势，也需要大量时间

商业工具（如 JADU、Price2Spy）年费 ¥5000-20000，数据源有限，定制化差。而自建系统的边际成本几乎为零。

变现路径

模式	定价	目标客户	月收入预估
个人自用	免费	自己	节省 ¥5000+/年工具费
小商户订阅	¥299/月	中小型电商卖家	10 客户 = ¥3,000
企业定制	¥999/月	品牌方、代理商	5 客户 = ¥5,000
SaaS 产品化	¥199/月起	批量复制	50 客户 = ¥10,000+

二、系统架构

整个系统分为四层：

数据采集层 → 存储计算层 → 分析引擎层 → 输出交付层
    ↓              ↓              ↓              ↓
 爬虫/API      DuckDB 数据库     SQL 聚合视图    报告/API/看板

为什么选 DuckDB？

维度	Python + Pandas	DuckDB
内存占用	全部加载到内存	列式压缩，按需读取
SQL 能力	需要写循环和条件逻辑	直接写 SQL 聚合、JOIN
文件格式支持	需要额外库	原生 CSV/JSON/Parquet
脏数据处理	容易整批崩溃	`RETURN_NULL_ON_ERROR` 优雅容错
部署成本	需要 Python 环境	单文件二进制，零依赖

DuckDB 的核心优势在于：你能用 SQL 完成 80% 的数据清洗和聚合工作，而且速度极快。

三、数据层：采集与存储

3.1 数据采集（模拟 + 真实方案）

实际项目中，你可以用爬虫或 API 获取数据。这里先用模拟数据验证逻辑：

import duckdb
import pandas as pd
from pathlib import Path
from datetime import datetime, timedelta
import random

# 创建项目目录
Path("competitor_monitor/data").mkdir(parents=True, exist_ok=True)

def generate_competitor_data(days: int = 90):
    """生成模拟的竞品监控数据"""
    
    # 竞品 A：高端品牌，价格波动小
    comp_a_prices = []
    base_price_a = 1299
    for day in range(days):
        date = (datetime.now() - timedelta(days=day)).strftime("%Y-%m-%d")
        price = base_price_a + random.randint(-50, 50)
        if day % 14 == 0:  # 每两周促销
            price = base_price_a - 200
        comp_a_prices.append({
            "competitor": "竞品A",
            "date": date,
            "product": "Pro版",
            "price": price,
            "url": "https://competitor-a.com/pricing",
            "stock_status": "有货" if random.random() > 0.1 else "缺货"
        })
    
    # 竞品 B：中端品牌，频繁调价
    comp_b_prices = []
    base_price_b = 799
    for day in range(days):
        date = (datetime.now() - timedelta(days=day)).strftime("%Y-%m-%d")
        price = base_price_b + random.randint(-100, 100)
        if day % 7 == 0:
            price -= 50
        comp_b_prices.append({
            "competitor": "竞品B",
            "date": date,
            "product": "标准版",
            "price": price,
            "url": "https://competitor-b.com/pricing",
            "stock_status": "有货"
        })
    
    # 竞品 C：新兴品牌，激进定价
    comp_c_prices = []
    base_price_c = 499
    for day in range(days):
        date = (datetime.now() - timedelta(days=day)).strftime("%Y-%m-%d")
        price = max(base_price_c - (days - day) // 10, 299)
        comp_c_prices.append({
            "competitor": "竞品C",
            "date": date,
            "product": "基础版",
            "price": price,
            "url": "https://competitor-c.com/pricing",
            "stock_status": "有货"
        })
    
    # 保存到 CSV
    pd.DataFrame(comp_a_prices + comp_b_prices + comp_c_prices).sort_values("date").to_csv(
        "competitor_monitor/data/competitor_prices.csv", index=False, encoding="utf-8-sig"
    )
    
    print(f"✅ 价格数据已生成：{len(comp_a_prices) + len(comp_b_prices) + len(comp_c_prices)} 条")
    return "competitor_monitor/data/competitor_prices.csv"

generate_competitor_data()

3.2 DuckDB 数据库建模

import duckdb
from pathlib import Path

class CompetitorMonitor:
    """竞品监控分析引擎"""
    
    def __init__(self, db_path: str = "competitor_monitor/monitor.db"):
        self.con = duckdb.connect(db_path)
        self._setup_schema()
    
    def _setup_schema(self):
        """创建数据库 schema"""
        self.con.execute("""
            CREATE TABLE IF NOT EXISTS competitor_prices (
                competitor VARCHAR,
                date DATE,
                product VARCHAR,
                price DECIMAL(10,2),
                url VARCHAR,
                stock_status VARCHAR
            )
        """)
        
        self.con.execute("""
            CREATE TABLE IF NOT EXISTS feature_updates (
                competitor VARCHAR,
                update_date DATE,
                update_type VARCHAR,
                description VARCHAR,
                source VARCHAR
            )
        """)
        
        self.con.execute("""
            CREATE TABLE IF NOT EXISTS reviews (
                competitor VARCHAR,
                date DATE,
                platform VARCHAR,
                rating INTEGER,
                sentiment VARCHAR,
                summary VARCHAR
            )
        """)

3.3 数据导入

def ingest_data(self, csv_path: str):
    """将 CSV 数据导入 DuckDB"""
    self.con.execute(f"""
        COPY competitor_prices 
        FROM '{csv_path}' 
        (FORMAT CSV, HEADER, DELIMITER ',')
    """)
    print(f"✅ 已导入 {self.con.execute('SELECT COUNT(*) FROM competitor_prices').fetchone()[0]} 条价格记录")

四、分析引擎：SQL 驱动的深度分析

4.1 价格趋势分析视图

def _setup_views(self):
    """创建分析视图"""
    self.con.execute("""
        CREATE OR REPLACE VIEW v_price_analysis AS
        SELECT 
            competitor,
            product,
            COUNT(*) AS data_points,
            ROUND(AVG(price), 2) AS avg_price,
            ROUND(MIN(price), 2) AS min_price,
            ROUND(MAX(price), 2) AS max_price,
            ROUND(STDDEV(price), 2) AS price_volatility,
            ROUND(
                100.0 * (MAX(price) - MIN(price)) / NULLIF(AVG(price), 0), 
                2
            ) AS price_variance_pct,
            MAX(date) AS latest_date
        FROM competitor_prices
        GROUP BY competitor, product
    """)
    
    # 日环比视图
    self.con.execute("""
        CREATE OR REPLACE VIEW v_price_daily AS
        SELECT 
            competitor,
            date,
            product,
            price,
            LAG(price) OVER (
                PARTITION BY competitor, product 
                ORDER BY date
            ) AS prev_price,
            price - LAG(price) OVER (
                PARTITION BY competitor, product 
                ORDER BY date
            ) AS price_change,
            ROUND(
                100.0 * (price - LAG(price) OVER (
                    PARTITION BY competitor, product 
                    ORDER BY date
                )) / NULLIF(LAG(price) OVER (
                    PARTITION BY competitor, product 
                    ORDER BY date
                ), 0),
                2
            ) AS price_change_pct
        FROM competitor_prices
    """)
    
    # 滚动平均视图
    self.con.execute("""
        CREATE OR REPLACE VIEW v_price_trend AS
        SELECT 
            competitor,
            date,
            product,
            price,
            ROUND(AVG(price) OVER (
                PARTITION BY competitor, product 
                ORDER BY date 
                ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
            ), 2) AS ma_7day,
            ROUND(AVG(price) OVER (
                PARTITION BY competitor, product 
                ORDER BY date 
                ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
            ), 2) AS ma_30day
        FROM competitor_prices
    """)

4.2 实时查询示例

# 查看各竞品价格统计
print("📊 竞品价格概览：")
print(self.con.execute("SELECT * FROM v_price_analysis ORDER BY avg_price").fetchdf())

# 查看今日价格变动
print("\n📈 今日价格变动：")
today = datetime.now().strftime("%Y-%m-%d")
print(self.con.execute(f"""
    SELECT competitor, product, price, price_change_pct
    FROM v_price_daily 
    WHERE date = '{today}' 
    AND price_change != 0
    ORDER BY ABS(price_change_pct) DESC
""").fetchdf())

# 查看 7 日均线趋势
print("\n📉 价格趋势（7日均线）：")
print(self.con.execute(f"""
    SELECT competitor, date, product, price, ma_7day
    FROM v_price_trend 
    WHERE date >= '{(datetime.now()-timedelta(days=7)).strftime("%Y-%m-%d")}'
    ORDER BY competitor, date
""").fetchdf())

4.3 高级分析：异常检测

def detect_price_anomalies(self, threshold_pct: float = 5.0):
    """检测价格异常波动"""
    result = self.con.execute(f"""
        SELECT 
            competitor,
            date,
            product,
            price,
            price_change_pct,
            '价格异常' AS alert_type
        FROM v_price_daily
        WHERE ABS(price_change_pct) >= {threshold_pct}
        ORDER BY ABS(price_change_pct) DESC
    """).fetchdf()
    return result

# 使用示例
anomalies = monitor.detect_price_anomalies(5.0)
if len(anomalies) > 0:
    print(f"⚠️ 发现 {len(anomalies)} 条价格异常：")
    print(anomalies.head(10))

五、告警与报告生成

5.1 智能告警系统

def check_alerts(self) -> list:
    """检查并生成告警"""
    alerts = []
    
    # 价格大幅下降告警
    price_drop = self.con.execute("""
        SELECT competitor, date, product, price, price_change_pct
        FROM v_price_daily
        WHERE date = (SELECT MAX(date) FROM v_price_daily)
        AND price_change_pct <= -5
    """).fetchall()
    
    for row in price_drop:
        alerts.append({
            "type": "PRICE_DROP",
            "competitor": row[0],
            "date": row[1],
            "product": row[2],
            "price": row[3],
            "change_pct": row[4],
            "message": f"⚠️ {row[0]} 的 {row[2]} 降价 {abs(row[4]):.1f}%，现价为 ¥{row[3]}"
        })
    
    # 缺货告警
    out_of_stock = self.con.execute("""
        SELECT competitor, date, product
        FROM competitor_prices
        WHERE stock_status = '缺货'
        AND date = (SELECT MAX(date) FROM competitor_prices)
    """).fetchall()
    
    for row in out_of_stock:
        alerts.append({
            "type": "OUT_OF_STOCK",
            "competitor": row[0],
            "date": row[1],
            "product": row[2],
            "message": f"📦 {row[0]} 的 {row[2]} 已缺货"
        })
    
    return alerts

5.2 报告生成

def generate_report(self, output_path: str = "report.html"):
    """生成 HTML 格式的分析报告"""
    from jinja2 import Template
    
    template_str = """
    <!DOCTYPE html>
    <html>
    <head>
        <meta charset="utf-8">
        <title>竞品监控日报 - {{ date }}</title>
        <style>
            body { font-family: -apple-system, sans-serif; max-width: 900px; margin: 0 auto; padding: 20px; }
            h1 { color: #1a1a2e; border-bottom: 3px solid #4facfe; padding-bottom: 10px; }
            h2 { color: #16213e; margin-top: 30px; }
            table { width: 100%; border-collapse: collapse; margin: 15px 0; }
            th { background: #1a1a2e; color: white; padding: 10px; text-align: left; }
            td { padding: 8px 10px; border-bottom: 1px solid #ddd; }
            tr:hover { background: #f5f5f5; }
            .alert { background: #fff3cd; border-left: 4px solid #ffc107; padding: 10px; margin: 10px 0; }
            .good { color: #28a745; }
            .bad { color: #dc3545; }
            .neutral { color: #6c757d; }
        </style>
    </head>
    <body>
        <h1>📊 竞品监控日报 - {{ date }}</h1>
        
        <h2>🔔 告警信息</h2>
        {% for alert in alerts %}
        <div class="alert">{{ alert.message }}</div>
        {% endfor %}
        {% if not alerts %}
        <p class="good">✅ 今日无异常告警</p>
        {% endif %}
        
        <h2>📈 价格概览</h2>
        <table>
            <tr><th>竞品</th><th>产品</th><th>平均价</th><th>最低价</th><th>最高价</th><th>波动率</th></tr>
            {% for row in price_analysis %}
            <tr>
                <td>{{ row[0] }}</td>
                <td>{{ row[1] }}</td>
                <td>¥{{ row[2] }}</td>
                <td>¥{{ row[3] }}</td>
                <td>¥{{ row[4] }}</td>
                <td>{{ row[7] }}%</td>
            </tr>
            {% endfor %}
        </table>
        
        <h2>📉 最新动态</h2>
        <table>
            <tr><th>竞品</th><th>日期</th><th>产品</th><th>当前价</th><th>变化</th></tr>
            {% for row in daily_changes %}
            <tr>
                <td>{{ row[0] }}</td>
                <td>{{ row[1] }}</td>
                <td>{{ row[2] }}</td>
                <td>¥{{ row[3] }}</td>
                <td class="{% if row[5] < 0 %}bad{% elif row[5] > 0 %}good{% else %}neutral{% endif %}">
                    {{ '%+.2f%%'|format(row[5]) }}
                </td>
            </tr>
            {% endfor %}
        </table>
        
        <p style="color: #999; font-size: 12px; margin-top: 30px;">
            本报告由 DuckDB 竞品监控系统自动生成 | 数据来源：网络公开信息
        </p>
    </body>
    </html>
    """
    
    template = Template(template_str)
    today = datetime.now().strftime("%Y-%m-%d")
    
    alerts = self.check_alerts()
    price_analysis = self.con.execute("SELECT * FROM v_price_analysis ORDER BY avg_price").fetchall()
    daily_changes = self.con.execute(f"""
        SELECT competitor, date, product, price, prev_price, price_change_pct
        FROM v_price_daily 
        WHERE date = '{today}'
        ORDER BY ABS(price_change_pct) DESC
    """).fetchall()
    
    html = template.render(
        date=today,
        alerts=alerts,
        price_analysis=price_analysis,
        daily_changes=daily_changes
    )
    
    Path(output_path).write_text(html, encoding="utf-8")
    print(f"✅ 报告已生成：{output_path}")
    return output_path

六、完整运行流程

# 主程序入口
if __name__ == "__main__":
    # 1. 初始化监控引擎
    monitor = CompetitorMonitor()
    
    # 2. 生成/导入数据
    csv_path = generate_competitor_data()
    monitor.ingest_data(csv_path)
    
    # 3. 创建分析视图
    monitor._setup_views()
    
    # 4. 运行分析
    print("\n" + "="*50)
    print("竞品价格分析报告")
    print("="*50)
    
    alerts = monitor.check_alerts()
    if alerts:
        print(f"\n⚠️ 发现 {len(alerts)} 条告警：")
        for alert in alerts:
            print(f"  - {alert['message']}")
    else:
        print("\n✅ 无异常告警")
    
    # 5. 生成报告
    monitor.generate_report("competitor_monitor/report.html")
    
    # 6. 导出 CSV 供进一步分析
    monitor.con.execute("""
        COPY (SELECT * FROM v_price_analysis) 
        TO 'competitor_monitor/price_summary.csv' 
        (HEADER, DELIMITER ',')
    """)
    print("\n✅ 数据已导出：competitor_monitor/price_summary.csv")

七、从个人工具到 SaaS 产品

7.1 产品化升级路径

阶段	功能	技术栈	定价
MVP	本地 CSV 监控	DuckDB + Python	免费自用
v1	Web 界面 + 邮件报告	DuckDB + Flask + SMTP	¥99/月
v2	API 服务 + 多租户	DuckDB + FastAPI + PostgreSQL	¥299/月
v3	完整 SaaS	DuckDB + React + 云部署	¥999/月

7.2 关键代码片段：API 服务化

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="竞品监控 API")

class PriceAlert(BaseModel):
    competitor: str
    min_price: float
    max_price: float
    alert_email: str

@app.post("/alerts")
def set_price_alert(alert: PriceAlert):
    """设置价格告警"""
    # DuckDB 查询当前价格
    current = monitor.con.execute(f"""
        SELECT price FROM competitor_prices 
        WHERE competitor = '{alert.competitor}'
        AND date = (SELECT MAX(date) FROM competitor_prices)
    """).fetchone()
    
    if current and (current[0] <= alert.min_price or current[0] >= alert.max_price):
        # 发送告警邮件
        send_alert_email(alert.alert_email, alert.competitor, current[0])
        return {"status": "alert_sent", "price": current[0]}
    
    return {"status": "no_alert", "current_price": current[0] if current else None}

7.3 自动化调度

# 使用 cron 每日运行
0 9 * * * cd /path/to/competitor_monitor && python3 monitor.py >> logs/cron.log 2>&1

# 或使用 Python schedule 库
import schedule
import time

schedule.every().day.at("09:00").do(run_monitor)
schedule.every().hour.do(check_alerts)

while True:
    schedule.run_pending()
    time.sleep(60)

八、与传统方案对比

功能	商业工具 (JADU等)	Python + Pandas	DuckDB 方案
价格监控	✅	✅	✅
历史趋势	✅ (有限)	✅	✅ (SQL 聚合)
异常检测	❌	需自行实现	✅ (内置窗口函数)
告警通知	✅ (付费)	需自行实现	✅ (简单代码)
报告生成	✅ (模板固定)	需自行实现	✅ (Jinja2 模板)
多数据源	❌	✅	✅ (ATTACH)
部署成本	¥5000+/年	服务器 + 维护	免费/低成本
定制化	❌	✅	✅
学习曲线	低	中	低 (SQL 为主)

九、变现建议

9.1 快速启动步骤

第一周：搭建本地监控系统，用模拟数据验证逻辑
第二周：接入真实数据源（爬虫或 API），覆盖 3-5 个竞品
第三周：生成第一份自动化报告，发给 2-3 个潜在客户试用
第四周：根据反馈优化，正式定价并开始推广

9.2 定价策略

基础版（¥199/月）：每日价格监控 + 周报
专业版（¥499/月）：实时监控 + 告警 + 月报 + 1 个竞品类别
企业版（¥999/月）：全功能 + 多品类 + API 访问 + 定制报告

9.3 获客渠道

知乎/掘金发技术文章引流
微信公众号发布行业洞察
闲鱼/淘宝挂服务链接
社群口碑传播

十、总结

用 DuckDB 搭建竞品监控系统，核心价值在于：

SQL 驱动：90% 的分析逻辑可以用 SQL 表达，无需复杂代码
轻量部署：单文件二进制，无需 Docker/K8s
可扩展：从本地 CSV 到 Parquet，从单机到多租户，平滑升级
零成本起步：开源免费，边际成本趋近于零

这套系统你已经可以跑了。接下来就是：接入真实数据 → 找第一个付费客户 → 迭代产品。

📖 详细图文教程见 duckdblab.org

📺 Watch video tutorials → Olap Studio YouTube

Subscribe for more DuckDB & AI automation tutorials

⚠️ 本站为独立社区项目，与 DuckDB 基金会及 DuckDB 官方项目无任何从属、背书或赞助关系。

"DuckDB" 是 DuckDB 基金会的注册商标，本站仅以事实描述方式使用该名称。

本站内容仅供教育与社区推广用途，不构成任何商业服务。