20d - AI 数据分析与报告

你的场景是什么？
├── 企业级、大数据量、实时分析 → ThoughtSpot
├── 中小团队、需要完整 BI 平台 → Metabase（自托管免费）
├── 个人/小团队、快速数据探索 → Julius AI 或 Fabi.ai
├── 开发者、需要 API 集成 → Text2SQL.ai 或 Chat2DB
├── 微软生态用户 → Power BI + Copilot
└── 预算极低、开源优先 → Metabase 开源版 + Chat2DB

用户行为：Plausible Analytics（隐私友好，$9/月）
产品指标：自建 SQLite / PostgreSQL + 简单 API
收入数据：Stripe Dashboard（免费）
错误追踪：Sentry 免费层
AI 查询层：Chat2DB（免费）或 Text2SQL.ai（$12/月）
可视化：Metabase 开源版（免费自托管）或 Grafana（免费）

用户自然语言问题
    ↓
Schema 提取（表结构、字段名、关系）
    ↓
Prompt 组装（Schema + 问题 + 约束规则）
    ↓
LLM 生成 SQL
    ↓
SQL 验证与安全检查
    ↓
执行查询
    ↓
结果格式化返回

"""
NL-to-SQL 引擎 —— 使用 OpenAI API + SQLite
适用于 Solo Founder 的轻量级数据查询方案
"""
import sqlite3
import json
from openai import OpenAI
 
client = OpenAI()  # 需设置 OPENAI_API_KEY 环境变量
 
def get_schema(db_path: str) -> str:
    """从 SQLite 数据库提取 schema 信息"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
 
    # 获取所有表
    cursor.execute(
        "SELECT name FROM sqlite_master WHERE type='table'"
    )
    tables = cursor.fetchall()
 
    schema_parts = []
    for (table_name,) in tables:
        cursor.execute(f"PRAGMA table_info({table_name})")
        columns = cursor.fetchall()
        col_defs = [
            f"  {col[1]} {col[2]}"
            + (" PRIMARY KEY" if col[5] else "")
            + (" NOT NULL" if col[3] else "")
            for col in columns
        ]
        schema_parts.append(
            f"CREATE TABLE {table_name} (\n"
            + ",\n".join(col_defs)
            + "\n);"
        )
 
    conn.close()
    return "\n\n".join(schema_parts)
 
 
def nl_to_sql(question: str, db_path: str) -> dict:
    """将自然语言问题转换为 SQL 并执行"""
    schema = get_schema(db_path)
 
    system_prompt = f"""你是一个 SQL 专家。根据用户的自然语言问题，
生成对应的 SQLite SQL 查询。
 
数据库 Schema：
{schema}
 
规则：
1. 只生成 SELECT 查询，禁止 INSERT/UPDATE/DELETE
2. 使用标准 SQLite 语法
3. 对日期字段使用 SQLite 日期函数
4. 返回 JSON 格式：{{"sql": "...", "explanation": "..."}}
5. 如果问题无法用当前 schema 回答，返回
   {{"sql": null, "explanation": "无法回答的原因"}}"""
 
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": question},
        ],
        response_format={"type": "json_object"},
        temperature=0,
    )
 
    result = json.loads(response.choices[0].message.content)
 
    if result.get("sql"):
        # 安全检查：确保是 SELECT 查询
        sql_upper = result["sql"].strip().upper()
        if not sql_upper.startswith("SELECT"):
            return {
                "error": "安全拦截：只允许 SELECT 查询",
                "sql": result["sql"],
            }
 
        # 执行查询
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()
        try:
            cursor.execute(result["sql"])
            columns = [desc[0] for desc in cursor.description]
            rows = cursor.fetchall()
            return {
                "sql": result["sql"],
                "explanation": result["explanation"],
                "columns": columns,
                "data": [dict(zip(columns, row)) for row in rows],
                "row_count": len(rows),
            }
        except Exception as e:
            return {"error": str(e), "sql": result["sql"]}
        finally:
            conn.close()
 
    return result
 
 
# 使用示例
if __name__ == "__main__":
    result = nl_to_sql(
        "上个月新注册用户中，有多少人在注册后 7 天内付费？",
        "app.db",
    )
    print(json.dumps(result, ensure_ascii=False, indent=2))

/**
 * NL-to-SQL API 服务 —— 使用 Hono + OpenAI + better-sqlite3
 * 可部署为独立微服务或集成到现有后端
 */
import { Hono } from "hono";
import OpenAI from "openai";
import Database from "better-sqlite3";
 
const app = new Hono();
const openai = new OpenAI();
 
interface NLQueryRequest {
  question: string;
  dbPath?: string;
}
 
interface SQLResult {
  sql: string | null;
  explanation: string;
  columns?: string[];
  data?: Record<string, unknown>[];
  rowCount?: number;
  error?: string;
}
 
function getSchema(dbPath: string): string {
  const db = new Database(dbPath, { readonly: true });
  const tables = db
    .prepare(
      `SELECT name FROM sqlite_master
       WHERE type='table' AND name NOT LIKE 'sqlite_%'`
    )
    .all() as { name: string }[];
 
  const schemaParts = tables.map((t) => {
    const cols = db
      .prepare(`PRAGMA table_info(${t.name})`)
      .all() as { name: string; type: string; pk: number }[];
    const colDefs = cols
      .map((c) => `  ${c.name} ${c.type}${c.pk ? " PRIMARY KEY" : ""}`)
      .join(",\n");
    return `CREATE TABLE ${t.name} (\n${colDefs}\n);`;
  });
 
  db.close();
  return schemaParts.join("\n\n");
}
 
app.post("/api/query", async (c) => {
  const { question, dbPath = "app.db" } = await c.req.json<NLQueryRequest>();
  const schema = getSchema(dbPath);
 
  const completion = await openai.chat.completions.create({
    model: "gpt-4o",
    messages: [
      {
        role: "system",
        content: `你是 SQL 专家。根据自然语言问题生成 SQLite SQL。
Schema：\n${schema}\n
规则：只生成 SELECT，返回 JSON {"sql":"...","explanation":"..."}`,
      },
      { role: "user", content: question },
    ],
    response_format: { type: "json_object" },
    temperature: 0,
  });
 
  const parsed = JSON.parse(
    completion.choices[0].message.content ?? "{}"
  );
 
  if (!parsed.sql) {
    return c.json<SQLResult>({
      sql: null,
      explanation: parsed.explanation ?? "无法生成查询",
    });
  }
 
  // 安全检查
  if (!parsed.sql.trim().toUpperCase().startsWith("SELECT")) {
    return c.json<SQLResult>({
      sql: parsed.sql,
      explanation: "安全拦截：只允许 SELECT 查询",
      error: "BLOCKED",
    });
  }
 
  // 执行查询
  const db = new Database(dbPath, { readonly: true });
  try {
    const stmt = db.prepare(parsed.sql);
    const rows = stmt.all() as Record<string, unknown>[];
    const columns = stmt.columns().map((c) => c.name);
    return c.json<SQLResult>({
      sql: parsed.sql,
      explanation: parsed.explanation,
      columns,
      data: rows,
      rowCount: rows.length,
    });
  } catch (e) {
    return c.json<SQLResult>({
      sql: parsed.sql,
      explanation: parsed.explanation,
      error: String(e),
    });
  } finally {
    db.close();
  }
});
 
export default app;

┌─────────────────────────────────────────────────┐
│              自动周报生成管线                      │
├─────────────────────────────────────────────────┤
│                                                   │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐       │
│  │Plausible │  │ Stripe   │  │ Sentry   │       │
│  │  API     │  │  API     │  │  API     │       │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘       │
│       │              │              │              │
│       └──────────┬───┴──────────────┘              │
│                  ↓                                  │
│       ┌──────────────────┐                         │
│       │   数据收集层      │                         │
│       │  (n8n / Python)  │                         │
│       └────────┬─────────┘                         │
│                ↓                                    │
│       ┌──────────────────┐                         │
│       │   AI 分析层       │                         │
│       │  (GPT-4o/Claude) │                         │
│       │  · 指标对比       │                         │
│       │  · 异常检测       │                         │
│       │  · 趋势分析       │                         │
│       │  · 行动建议       │                         │
│       └────────┬─────────┘                         │
│                ↓                                    │
│       ┌──────────────────┐                         │
│       │   输出层          │                         │
│       │  · 邮件周报       │                         │
│       │  · Notion 存档    │                         │
│       │  · Slack 通知     │                         │
│       └──────────────────┘                         │
└─────────────────────────────────────────────────┘

"""
AI 周报生成器 —— 自动收集数据、AI 分析、生成 Markdown 报告
适用于 Solo Founder 的自动化数据分析管线
"""
import os
import json
import httpx
from datetime import datetime, timedelta
from openai import OpenAI
 
client = OpenAI()
 
# ── 数据收集层 ──────────────────────────────────
 
def collect_plausible_data(
    site_id: str, period: str = "7d"
) -> dict:
    """从 Plausible Analytics 收集网站数据"""
    base = os.getenv("PLAUSIBLE_URL", "https://plausible.io")
    token = os.getenv("PLAUSIBLE_TOKEN")
    headers = {"Authorization": f"Bearer {token}"}
 
    # 聚合指标
    agg = httpx.get(
        f"{base}/api/v1/stats/aggregate",
        params={
            "site_id": site_id,
            "period": period,
            "metrics": "visitors,pageviews,bounce_rate,"
                       "visit_duration,visits",
        },
        headers=headers,
    ).json()
 
    # 流量来源
    sources = httpx.get(
        f"{base}/api/v1/stats/breakdown",
        params={
            "site_id": site_id,
            "period": period,
            "property": "visit:source",
            "limit": "10",
        },
        headers=headers,
    ).json()
 
    # 热门页面
    pages = httpx.get(
        f"{base}/api/v1/stats/breakdown",
        params={
            "site_id": site_id,
            "period": period,
            "property": "event:page",
            "limit": "10",
        },
        headers=headers,
    ).json()
 
    return {
        "aggregate": agg.get("results", {}),
        "top_sources": sources.get("results", []),
        "top_pages": pages.get("results", []),
    }
 
 
def collect_stripe_data() -> dict:
    """从 Stripe 收集收入数据"""
    import stripe
    stripe.api_key = os.getenv("STRIPE_SECRET_KEY")
 
    now = datetime.now()
    week_ago = now - timedelta(days=7)
    ts_start = int(week_ago.timestamp())
    ts_end = int(now.timestamp())
 
    # 本周收入
    charges = stripe.Charge.list(
        created={"gte": ts_start, "lte": ts_end},
        limit=100,
    )
    revenue = sum(c.amount for c in charges if c.paid) / 100
 
    # 新订阅
    subs = stripe.Subscription.list(
        created={"gte": ts_start, "lte": ts_end},
        limit=100,
    )
 
    # MRR（活跃订阅的月度总额）
    active_subs = stripe.Subscription.list(
        status="active", limit=100
    )
    mrr = sum(
        s.items.data[0].price.unit_amount
        * s.items.data[0].quantity
        for s in active_subs
    ) / 100
 
    return {
        "weekly_revenue": revenue,
        "new_subscriptions": len(subs.data),
        "mrr": mrr,
        "total_charges": len(charges.data),
    }
 
 
def collect_sentry_data(org: str, project: str) -> dict:
    """从 Sentry 收集错误数据"""
    token = os.getenv("SENTRY_TOKEN")
    headers = {"Authorization": f"Bearer {token}"}
 
    # 最近 7 天的 Issue 统计
    issues = httpx.get(
        f"https://sentry.io/api/0/projects/{org}/{project}/issues/",
        params={"query": "is:unresolved", "limit": "25"},
        headers=headers,
    ).json()
 
    return {
        "unresolved_issues": len(issues),
        "top_issues": [
            {
                "title": i.get("title", ""),
                "count": i.get("count", "0"),
                "level": i.get("level", "error"),
            }
            for i in issues[:5]
        ],
    }
 
 
# ── AI 分析层 ──────────────────────────────────
 
def generate_weekly_report(
    plausible_data: dict,
    stripe_data: dict,
    sentry_data: dict,
    last_week_data: dict | None = None,
) -> str:
    """使用 AI 生成周报"""
    data_summary = json.dumps(
        {
            "website": plausible_data,
            "revenue": stripe_data,
            "errors": sentry_data,
            "last_week": last_week_data,
        },
        ensure_ascii=False,
        indent=2,
    )
 
    today = datetime.now()
    week_ago = today - timedelta(days=7)
    period = (
        f"{week_ago.strftime('%Y.%m.%d')}"
        f" ~ {today.strftime('%Y.%m.%d')}"
    )
 
    prompt = f"""你是一个资深数据分析师。基于以下业务数据，
生成一份结构化的中文周报。
 
数据时间范围：{period}
 
原始数据：
{data_summary}
 
请生成包含以下部分的 Markdown 周报：
 
## 关键指标
用表格展示本周 vs 上周的关键指标对比（如有上周数据），
包含变化百分比和趋势箭头（↑/↓）。
 
## 异常检测
识别任何显著变化（>20% 的波动），分析可能原因。
 
## 趋势分析
基于数据识别增长或下降趋势，给出简短分析。
 
## 行动建议
基于数据给出 3 个具体、可执行的行动建议，
按优先级排序。
 
## 错误概况
总结本周错误情况，标注需要优先处理的问题。
 
格式要求：使用 Markdown，表格对齐，emoji 标注趋势。"""
 
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": "你是一个专业的数据分析师，"
                "擅长从业务数据中提取洞察并给出行动建议。",
            },
            {"role": "user", "content": prompt},
        ],
        temperature=0.3,
    )
 
    title = f"# 产品周报 - {period}\n\n"
    return title + response.choices[0].message.content
 
 
# ── 输出层 ──────────────────────────────────────
 
def send_email_report(report: str, to: str):
    """通过 Resend 发送邮件周报"""
    import resend
    resend.api_key = os.getenv("RESEND_API_KEY")
 
    resend.Emails.send({
        "from": "reports@yourdomain.com",
        "to": [to],
        "subject": f"📊 产品周报 - {datetime.now().strftime('%Y.%m.%d')}",
        "html": markdown_to_html(report),
    })
 
 
def save_to_notion(report: str, database_id: str):
    """将周报存档到 Notion 数据库"""
    from notion_client import Client
    notion = Client(auth=os.getenv("NOTION_TOKEN"))
 
    notion.pages.create(
        parent={"database_id": database_id},
        properties={
            "Name": {
                "title": [
                    {
                        "text": {
                            "content": f"周报 {datetime.now().strftime('%Y-%m-%d')}"
                        }
                    }
                ]
            },
            "Date": {
                "date": {"start": datetime.now().isoformat()}
            },
        },
        children=[
            {
                "object": "block",
                "type": "paragraph",
                "paragraph": {
                    "rich_text": [
                        {"type": "text", "text": {"content": report}}
                    ]
                },
            }
        ],
    )
 
 
def markdown_to_html(md: str) -> str:
    """简单的 Markdown 转 HTML（生产环境建议用 markdown 库）"""
    import markdown
    return markdown.markdown(md, extensions=["tables", "fenced_code"])
 
 
# ── 主流程 ──────────────────────────────────────
 
if __name__ == "__main__":
    print("📊 开始生成周报...")
 
    plausible = collect_plausible_data("yoursite.com")
    stripe = collect_stripe_data()
    sentry = collect_sentry_data("your-org", "your-project")
 
    report = generate_weekly_report(plausible, stripe, sentry)
 
    # 保存本地
    filename = f"reports/weekly_{datetime.now().strftime('%Y%m%d')}.md"
    os.makedirs("reports", exist_ok=True)
    with open(filename, "w") as f:
        f.write(report)
    print(f"✅ 周报已保存: {filename}")
 
    # 发送邮件
    send_email_report(report, "you@example.com")
    print("📧 邮件已发送")

n8n 工作流：每周日晚 21:00 自动运行

节点 1：Schedule Trigger
├── 每周日 21:00 触发

节点 2：并行数据收集（HTTP Request 节点 ×4）
├── Plausible API → 本周访问量、来源、热门页面
├── Stripe API → 本周收入、新订阅、MRR、流失
├── Sentry API → 本周错误数量和类型
└── GitHub API → 本周 Issue 和 PR 数据

节点 3：Merge 节点
├── 合并所有数据源的结果

节点 4：AI Agent 节点（Claude/GPT-4o）
├── System Prompt：你是一个数据分析师
├── 输入：合并后的所有数据
├── 输出：
│   - 关键指标周报（与上周对比）
│   - 异常检测（哪些指标有显著变化）
│   - 趋势分析（增长/下降趋势）
│   - 行动建议（基于数据的 3 个建议）
└── 格式：结构化 Markdown

节点 5：并行输出
├── Send Email 节点 → 发送周报邮件
├── Notion API 节点 → 存档到 Notion 数据库
└── Slack/Telegram 节点 → 发送摘要通知

# 产品周报 - 2026.02.15 ~ 02.21
 
## 关键指标
 
| 指标 | 本周 | 上周 | 变化 | 趋势 |
|------|------|------|------|------|
| 网站访问 | 2,340 | 1,890 | +23.8% | 📈 |
| 新注册 | 45 | 38 | +18.4% | 📈 |
| 活跃用户 | 312 | 298 | +4.7% | 📈 |
| MRR | $1,240 | $1,180 | +5.1% | 📈 |
| 错误率 | 0.3% | 0.5% | -40.0% | ✅ |
| 客户流失 | 2 | 1 | +100% | ⚠️ |
 
## 异常检测
 
⚠️ **流量激增**：来自 Hacker News 的流量激增 340%
（周三有人分享了你的产品）。转化率仅 2.1%，
低于平均 4.5%，说明 Landing Page 未针对 HN 用户优化。
 
⚠️ **客户流失翻倍**：本周流失 2 个付费用户（上周 1 个）。
需要检查流失原因（功能缺失？竞品？价格？）。
 
## 趋势分析
 
📈 整体增长健康，MRR 连续 4 周正增长。
📊 日本用户占比从 3% 增长到 8%，呈明显上升趋势。
📉 移动端跳出率持续偏高（68%），需要关注移动体验。
 
## 行动建议
 
1. **🔴 高优先级**：优化 Landing Page 的 CTA，
   针对 HN 流量添加开发者导向的价值主张
2. **🟡 中优先级**：联系流失用户了解原因，
   考虑添加退出调查
3. **🟢 低优先级**：日本用户增长明显，
   考虑添加日语支持（可用 AI 翻译快速实现）
 
## 错误概况
 
- 未解决 Issue：8 个（上周 5 个）
- 🔴 P0："同步失败"相关 Issue 增加 3 个，影响核心功能
- 🟡 P1：iOS Safari 下的 CSS 渲染问题（2 个报告）
- 建议优先修复同步失败问题

"""
AI 驱动的业务仪表板 —— 使用 Streamlit
运行：streamlit run dashboard.py
"""
import streamlit as st
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
from openai import OpenAI
from datetime import datetime, timedelta
import sqlite3
 
client = OpenAI()
 
st.set_page_config(
    page_title="📊 AI 业务仪表板",
    layout="wide",
)
st.title("📊 AI 业务仪表板")
 
# ── 数据加载 ──────────────────────────────────
 
@st.cache_data(ttl=3600)
def load_metrics(db_path: str = "app.db") -> pd.DataFrame:
    conn = sqlite3.connect(db_path)
    df = pd.read_sql(
        """
        SELECT date, visitors, signups, active_users,
               mrr, error_rate, churn_count
        FROM daily_metrics
        ORDER BY date DESC
        LIMIT 90
        """,
        conn,
    )
    conn.close()
    df["date"] = pd.to_datetime(df["date"])
    return df.sort_values("date")
 
df = load_metrics()
 
# ── KPI 卡片 ──────────────────────────────────
 
col1, col2, col3, col4 = st.columns(4)
 
latest = df.iloc[-1]
prev_week = df.iloc[-8] if len(df) > 7 else df.iloc[0]
 
with col1:
    delta = (
        (latest["visitors"] - prev_week["visitors"])
        / prev_week["visitors"] * 100
    )
    st.metric("日访问量", f"{latest['visitors']:,.0f}",
              f"{delta:+.1f}%")
 
with col2:
    delta = (
        (latest["signups"] - prev_week["signups"])
        / max(prev_week["signups"], 1) * 100
    )
    st.metric("日注册", f"{latest['signups']:,.0f}",
              f"{delta:+.1f}%")
 
with col3:
    st.metric("MRR", f"${latest['mrr']:,.0f}",
              f"{latest['mrr'] - prev_week['mrr']:+,.0f}")
 
with col4:
    st.metric("错误率", f"{latest['error_rate']:.2%}",
              f"{(latest['error_rate'] - prev_week['error_rate']):.2%}",
              delta_color="inverse")
 
# ── 趋势图 ──────────────────────────────────
 
st.subheader("📈 关键指标趋势")
 
tab1, tab2, tab3 = st.tabs(["流量", "收入", "质量"])
 
with tab1:
    fig = px.line(
        df, x="date", y=["visitors", "signups"],
        title="网站流量与注册趋势",
        labels={"value": "数量", "date": "日期"},
    )
    st.plotly_chart(fig, use_container_width=True)
 
with tab2:
    fig = go.Figure()
    fig.add_trace(go.Scatter(
        x=df["date"], y=df["mrr"],
        mode="lines+markers", name="MRR ($)",
        fill="tozeroy",
    ))
    fig.update_layout(title="MRR 增长趋势")
    st.plotly_chart(fig, use_container_width=True)
 
with tab3:
    fig = px.area(
        df, x="date", y="error_rate",
        title="错误率趋势",
        labels={"error_rate": "错误率", "date": "日期"},
    )
    st.plotly_chart(fig, use_container_width=True)
 
# ── AI 数据问答 ──────────────────────────────
 
st.subheader("🤖 AI 数据问答")
question = st.text_input(
    "用自然语言提问：",
    placeholder="例如：最近一周哪天的注册量最高？",
)
 
if question:
    with st.spinner("AI 分析中..."):
        data_context = df.tail(30).to_csv(index=False)
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "system",
                    "content": f"你是数据分析师。基于以下 CSV 数据回答问题。"
                    f"用中文回答，简洁明了。\n\n{data_context}",
                },
                {"role": "user", "content": question},
            ],
            temperature=0.3,
        )
        st.markdown(response.choices[0].message.content)

# grafana-alerting-rule.yaml
# Grafana 告警规则示例：AI 异常检测
apiVersion: 1
groups:
  - name: ai-anomaly-detection
    folder: AI Alerts
    interval: 5m
    rules:
      - title: MRR 异常波动
        condition: C
        data:
          - refId: A
            datasourceUid: prometheus
            model:
              expr: sum(mrr_total)
              intervalMs: 60000
          - refId: B
            datasourceUid: __expr__
            model:
              type: classic_conditions
              conditions:
                - evaluator:
                    type: outside_range
                    params: [900, 1500]  # MRR 正常范围
        annotations:
          summary: "MRR 出现异常波动：当前值 {{ $value }}"
          description: "MRR 超出正常范围 $900-$1500，请检查"

异常检测方法选择：
├── 简单阈值法（适合起步）
│   └── 指标超过固定阈值时告警
├── 百分比变化法（推荐 Solo Founder）
│   └── 与上周/上月对比，变化超过 X% 时告警
├── 移动平均法（适合有历史数据后）
│   └── 偏离 N 日移动平均 M 个标准差时告警
└── AI 异常检测（适合数据量大时）
    └── 使用 ML 模型学习正常模式，自动识别异常

"""
业务指标异常检测 Agent
可作为 cron job 每日运行，发现异常时主动告警
"""
import json
import sqlite3
from datetime import datetime, timedelta
from openai import OpenAI
 
client = OpenAI()
 
 
def detect_anomalies(db_path: str = "app.db") -> list[dict]:
    """基于百分比变化的异常检测"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
 
    # 获取最近 2 周的日数据
    cursor.execute("""
        SELECT date, visitors, signups, active_users,
               mrr, error_rate, churn_count
        FROM daily_metrics
        ORDER BY date DESC
        LIMIT 14
    """)
    rows = cursor.fetchall()
    columns = [desc[0] for desc in cursor.description]
    conn.close()
 
    if len(rows) < 8:
        return []
 
    # 计算本周 vs 上周的平均值
    this_week = rows[:7]
    last_week = rows[7:14]
 
    anomalies = []
    metrics = ["visitors", "signups", "active_users",
               "mrr", "error_rate", "churn_count"]
 
    for metric in metrics:
        idx = columns.index(metric)
        this_avg = sum(r[idx] for r in this_week) / len(this_week)
        last_avg = sum(r[idx] for r in last_week) / len(last_week)
 
        if last_avg == 0:
            continue
 
        change_pct = (this_avg - last_avg) / last_avg * 100
 
        # 阈值：变化超过 25% 视为异常
        threshold = 25
        if abs(change_pct) > threshold:
            anomalies.append({
                "metric": metric,
                "this_week_avg": round(this_avg, 2),
                "last_week_avg": round(last_avg, 2),
                "change_pct": round(change_pct, 1),
                "severity": "high" if abs(change_pct) > 50
                           else "medium",
            })
 
    return anomalies
 
 
def investigate_anomaly(anomalies: list[dict]) -> str:
    """使用 AI 调查异常原因并给出建议"""
    if not anomalies:
        return "✅ 本周所有指标正常，无异常检测到。"
 
    prompt = f"""你是一个业务数据分析师。以下是本周检测到的异常指标：
 
{json.dumps(anomalies, ensure_ascii=False, indent=2)}
 
请对每个异常：
1. 分析可能的原因（列出 2-3 个最可能的原因）
2. 评估影响程度（对业务的潜在影响）
3. 给出具体的调查步骤和行动建议
 
用中文回答，格式清晰，按严重程度排序。"""
 
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": "你是资深业务分析师，"
                "擅长从数据异常中发现问题并给出行动建议。",
            },
            {"role": "user", "content": prompt},
        ],
        temperature=0.3,
    )
 
    return response.choices[0].message.content
 
 
def send_alert(message: str, channel: str = "slack"):
    """发送异常告警"""
    import httpx
 
    if channel == "slack":
        webhook_url = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"
        httpx.post(webhook_url, json={"text": f"🚨 异常告警\n\n{message}"})
    elif channel == "telegram":
        bot_token = "YOUR_BOT_TOKEN"
        chat_id = "YOUR_CHAT_ID"
        httpx.post(
            f"https://api.telegram.org/bot{bot_token}/sendMessage",
            json={"chat_id": chat_id, "text": f"🚨 异常告警\n\n{message}"},
        )
 
 
if __name__ == "__main__":
    anomalies = detect_anomalies()
    if anomalies:
        report = investigate_anomaly(anomalies)
        print(report)
        send_alert(report)
    else:
        print("✅ 所有指标正常")

多渠道反馈收集 Agent（n8n 工作流）：

触发器：多渠道监听
├── Discord Webhook → 用户消息
├── GitHub Issues Webhook → Bug 报告和功能请求
├── 邮件（IMAP 监听）→ 用户邮件
└── Twitter/X 提及监听 → 社交媒体反馈
    ↓
节点 1：AI 分类
├── Bug 报告 → 创建 GitHub Issue + 标签
├── 功能请求 → 添加到 Linear Backlog
├── 使用问题 → 触发客服 agent 回复
├── 正面反馈 → 收集到"好评库"（用于营销素材）
└── 负面反馈 → 高优先级通知创始人
    ↓
节点 2：自动回复
├── Bug：感谢报告，已创建 Issue #xxx，我们会尽快修复
├── 功能请求：好主意！已记录到我们的 roadmap
├── 问题：[AI 生成的解答]
└── 反馈：感谢你的反馈！
    ↓
节点 3：周汇总
├── 本周收到的反馈分类统计
├── 最常被请求的功能 Top 5
├── 用户情感趋势（正面/中性/负面比例）
└── 需要创始人亲自处理的高优先级项

角色：你是一个资深数据分析师，擅长从业务数据中提取洞察。

任务：分析以下 [数据类型] 数据，提供深度洞察。

数据：
[粘贴数据或 CSV]

分析要求：
1. 数据概览：关键统计量（均值、中位数、标准差）
2. 趋势分析：识别增长/下降趋势和拐点
3. 异常检测：标注任何异常数据点及可能原因
4. 相关性分析：不同指标之间的关联关系
5. 行动建议：基于分析给出 3-5 个具体建议

输出格式：
- 使用 Markdown 表格展示数据
- 用 emoji 标注趋势方向
- 建议按优先级排序
- 每个建议包含"做什么"和"为什么"

角色：你是 [产品名] 的数据分析师，负责生成周度业务报告。

数据来源：
- 网站分析：[Plausible/GA 数据]
- 收入数据：[Stripe 数据]
- 错误数据：[Sentry 数据]
- 用户反馈：[反馈汇总]

报告要求：
1. 执行摘要（3 句话总结本周表现）
2. 关键指标表格（本周 vs 上周，含变化百分比）
3. 异常与风险（任何需要关注的问题）
4. 增长机会（数据揭示的增长点）
5. 本周行动项（3 个最重要的待办事项）

语气：专业但简洁，面向创始人/CEO 阅读。
格式：Markdown，适合邮件发送。

角色：你是一个业务数据侦探，擅长追踪数据异常的根因。

异常描述：
- 指标：[指标名称]
- 正常范围：[正常值范围]
- 当前值：[异常值]
- 变化幅度：[百分比变化]
- 发生时间：[时间点/时间段]

已知上下文：
- [任何已知的变更，如产品发布、营销活动、外部事件]

请执行以下调查：
1. 列出 5 个最可能的原因（按可能性排序）
2. 对每个原因，给出验证方法（需要查看什么数据）
3. 评估影响：这个异常对业务的短期和长期影响
4. 给出应急措施和长期解决方案
5. 建议设置什么监控来防止类似问题

输出格式：结构化 Markdown，每个原因一个小节。

工具选择：
├── 网站分析：Plausible Analytics（$9/月，隐私友好）
├── 产品指标：SQLite 数据库 + 自建 API
│   └── 表：daily_metrics（date, visitors, signups,
│       active_users, mrr, error_rate, churn_count）
├── 收入：Stripe Dashboard（免费）
├── 错误：Sentry 免费层
├── 查询：Chat2DB（免费，本地 AI 查询）
└── 可视化：Metabase 开源版（Docker 自托管）

月成本：$9（仅 Plausible）

问题："最近 30 天每天的新注册用户数是多少？"
→ AI 生成 SQL：
  SELECT date, signups
  FROM daily_metrics
  WHERE date >= date('now', '-30 days')
  ORDER BY date;

问题："哪个月的 MRR 增长最快？"
→ AI 生成 SQL：
  SELECT strftime('%Y-%m', date) AS month,
         MAX(mrr) - MIN(mrr) AS mrr_growth
  FROM daily_metrics
  GROUP BY month
  ORDER BY mrr_growth DESC
  LIMIT 5;

# crontab -e
# 每周日 21:00 运行周报生成器
0 21 * * 0 cd /app && python weekly_report.py >> /var/log/report.log 2>&1

# 每天早上 9:00 运行异常检测
0 9 * * * cd /app && python anomaly_detector.py >> /var/log/anomaly.log 2>&1

Metabase 仪表板布局：

┌─────────────────────────────────────────┐
│  KPI 卡片行                              │
│  [日访问] [新注册] [MRR] [错误率]         │
├──────────────────┬──────────────────────┤
│  流量趋势图       │  收入趋势图           │
│  （30 天折线图）   │  （MRR 面积图）       │
├──────────────────┼──────────────────────┤
│  流量来源饼图      │  热门页面排行          │
├──────────────────┼──────────────────────┤
│  错误趋势图       │  用户反馈分类          │
└──────────────────┴──────────────────────┘