B8. 电商数据可视化与实时 Dashboard

路径: Path B: 技术人 · 模块: B8 最后更新: 2026-07-31 难度: 中级 预计时间: 每天 1 小时，1-2 周 前置模块: B1 数据采集与处理

章节导航

1. 为什么需要自建 Dashboard · 2. 技术栈选择 · 3. Streamlit 快速上手 · 4. 电商核心 Dashboard 模块 · 5. 多平台数据整合 · 6. AI 增强 Dashboard · 7. 部署与分享 · 8. 常见陷阱 · 9. 完成标志

本模块你将构建

• 一个 Streamlit 电商运营 Dashboard（销售/广告/库存/利润）
• 多平台数据整合视图（Amazon + Shopify + 广告平台）
• AI 增强的异常检测和自动洞察
• 可部署到云端的实时监控系统

核心理念：Amazon Seller Central 和 Shopify 后台的数据分散在不同报告中，无法一眼看到全局。自建 Dashboard 把所有数据汇聚到一个视图，加上 AI 异常检测，让你从“被动看数据“变成“主动发现问题“。

1. 为什么需要自建 Dashboard

1.1 平台后台的局限

2. 技术栈选择

2.1 方案对比

2.2 推荐：Streamlit + Plotly

pip3 install streamlit plotly pandas openpyxl

3. Streamlit 快速上手

3.1 最小可行 Dashboard（10 分钟）

# dashboard.py 电商运营 Dashboard
import streamlit as st
import pandas as pd
import plotly.express as px
from datetime import datetime, timedelta

st.set_page_config(page_title="电商运营 Dashboard", layout="wide")
st.title(" 电商运营 Dashboard")

# 侧边栏：日期选择
with st.sidebar:
    st.header("筛选条件")
    date_range = st.date_input(
        "日期范围",
        value=(datetime.now() - timedelta(days=30), datetime.now())
    )
    marketplace = st.selectbox("市场", ["All", "US", "EU", "JP"])

# 数据加载
@st.cache_data
def load_data():
    # 替换为你的数据源（CSV/API/数据库）
    df = pd.read_csv("sales_data.csv", parse_dates=["date"])
    return df

df = load_data()

# KPI 卡片
col1, col2, col3, col4 = st.columns(4)
col1.metric("总收入", f"${df['revenue'].sum():,.0f}",
            f"{(df['revenue'].sum() / df['revenue_prev'].sum() - 1)*100:+.1f}%")
col2.metric("总订单", f"{df['orders'].sum():,}")
col3.metric("平均客单价", f"${df['revenue'].sum() / df['orders'].sum():.2f}")
col4.metric("广告 ROAS", f"{df['ad_revenue'].sum() / df['ad_spend'].sum():.1f}x")

# 销售趋势图
st.subheader(" 销售趋势")
daily = df.groupby("date").agg({"revenue": "sum", "orders": "sum"}).reset_index()
fig = px.line(daily, x="date", y="revenue", title="日收入趋势")
st.plotly_chart(fig, use_container_width=True)

# 品类分布
col1, col2 = st.columns(2)
with col1:
    st.subheader(" 品类收入分布")
    cat_data = df.groupby("category")["revenue"].sum().reset_index()
    fig2 = px.pie(cat_data, values="revenue", names="category")
    st.plotly_chart(fig2, use_container_width=True)

with col2:
    st.subheader(" 库存健康度")
    inv_data = df.groupby("sku")[["inventory_days", "daily_sales"]].mean().reset_index()
    inv_data["status"] = inv_data["inventory_days"].apply(
        lambda x: " 紧急" if x < 7 else (" 注意" if x < 14 else " 正常")
    )
    st.dataframe(inv_data, use_container_width=True)

运行：streamlit run dashboard.py

4. 电商核心 Dashboard 模块

4.1 模块架构

电商 Dashboard 模块：

Overview（总览）
KPI 卡片（收入/订单/利润/ROAS）
日/周/月趋势图
同比/环比变化

Sales（销售分析）
SKU 级别销售排名
品类/市场分布
新品 vs 老品表现
退货率分析

Advertising（广告分析）
Campaign 表现排名
ACOS/ROAS/TACOS 趋势
关键词表现 Top/Bottom
搜索词发现
预算消耗进度

Inventory（库存管理）
库存健康度（红黄绿灯）
可售天数预警
补货建议
长期仓储费预警

Profitability（利润分析）
SKU 级别真实利润
成本结构分解
利润趋势
盈亏平衡分析

AI Insights（AI 洞察）
异常检测（销量骤降/ACOS 飙升）
趋势预测（未来 7 天预测）
自动优化建议
竞品变化提醒

4.2 广告分析模块代码

def render_advertising_tab(df_ads: pd.DataFrame):
    """广告分析 Tab"""
    st.header(" 广告分析")

    # KPI
    col1, col2, col3, col4 = st.columns(4)
    total_spend = df_ads['spend'].sum()
    total_sales = df_ads['attributed_sales'].sum()
    col1.metric("总花费", f"${total_spend:,.0f}")
    col2.metric("广告销售", f"${total_sales:,.0f}")
    col3.metric("ACOS", f"{total_spend/total_sales*100:.1f}%")
    col4.metric("ROAS", f"{total_sales/total_spend:.1f}x")

    # Campaign 排名
    st.subheader("Campaign 表现排名")
    campaign_data = df_ads.groupby("campaign_name").agg({
        "spend": "sum",
        "attributed_sales": "sum",
        "clicks": "sum",
        "impressions": "sum"
    }).reset_index()
    campaign_data["acos"] = campaign_data["spend"] / campaign_data["attributed_sales"] * 100
    campaign_data["roas"] = campaign_data["attributed_sales"] / campaign_data["spend"]
    campaign_data["ctr"] = campaign_data["clicks"] / campaign_data["impressions"] * 100

    # 颜色编码 ACOS
    st.dataframe(
        campaign_data.sort_values("spend", ascending=False),
        use_container_width=True,
        column_config={
            "acos": st.column_config.ProgressColumn(
                "ACOS %", min_value=0, max_value=100, format="%.1f%%"
            )
        }
    )

    # 关键词散点图（花费 vs 转化）
    st.subheader("关键词表现散点图")
    fig = px.scatter(
        df_ads.groupby("keyword").agg({"spend": "sum", "attributed_sales": "sum", "clicks": "sum"}).reset_index(),
        x="spend", y="attributed_sales", size="clicks",
        hover_name="keyword",
        title="花费 vs 销售（气泡大小=点击量）"
    )
    fig.add_shape(type="line", x0=0, y0=0, x1=df_ads["spend"].max(),
                  y1=df_ads["spend"].max()/0.25, line=dict(dash="dash", color="red"))
    st.plotly_chart(fig, use_container_width=True)

5. 多平台数据整合

真实案例：AWS 电商流量异常检测架构 AWS 官方博客展示了如何自动化电商流量模式的异常检测。早期发现网站页面访问和订单完成等指标的微小异常，帮助组织采取纠正措施，减少对业务 KPI 的负面影响（AWS Architecture Blog）。

Content rephrased for compliance with licensing restrictions.

真实案例：Streamlit BI Dashboard 整合 GA4 + 电商数据 Squadbase 展示了一个综合性的 Streamlit BI Dashboard，整合了 Google Analytics 4（GA4）分析和电商智能两个关键业务领域，提供网站流量、用户行为和转化模式的深度分析（Squadbase）。

Content rephrased for compliance with licensing restrictions.

真实案例：Amazon SP-API Python 数据获取 Andrew Kushnerov 的系列教程展示了如何用 Python 从 Amazon SP-API 获取订单数据和库存/价格数据。关键洞察：订单在创建后会持续更新（状态变化、金额变化），要构建高质量分析需要追踪订单的完整生命周期（Medium - Orders，Medium - Inventory）。

Content rephrased for compliance with licensing restrictions.

5.1 Amazon SP-API 数据获取

# Amazon SP-API 订单数据获取示例
from sp_api.api import Orders, Reports
from sp_api.base import Marketplaces
from datetime import datetime, timedelta

def get_amazon_orders(days_back: int = 30) -> pd.DataFrame:
    """从 Amazon SP-API 获取订单数据"""
    orders_api = Orders(marketplace=Marketplaces.US)

    created_after = (datetime.now() - timedelta(days=days_back)).isoformat()

    all_orders = []
    response = orders_api.get_orders(
        CreatedAfter=created_after,
        OrderStatuses=["Shipped", "Unshipped"]
    )

    all_orders.extend(response.payload.get("Orders", []))

    # 处理分页
    while response.payload.get("NextToken"):
        response = orders_api.get_orders(
            CreatedAfter=created_after,
            NextToken=response.payload["NextToken"]
        )
        all_orders.extend(response.payload.get("Orders", []))

    # 转换为 DataFrame
    df = pd.DataFrame(all_orders)
    df["OrderDate"] = pd.to_datetime(df["PurchaseDate"])
    df["Revenue"] = df["OrderTotal"].apply(
        lambda x: float(x["Amount"]) if isinstance(x, dict) else 0
    )

    return df

def get_amazon_inventory() -> pd.DataFrame:
    """获取 FBA 库存数据"""
    reports_api = Reports(marketplace=Marketplaces.US)

    # 请求 FBA 库存报告
    report = reports_api.create_report(
        reportType="GET_FBA_MYI_UNSUPPRESSED_INVENTORY_DATA"
    )

    # 等待报告生成并下载
    # ... (轮询 report status)

    return pd.read_csv(report_file, sep="\t")

5.2 统一数据模型

# 统一的跨平台销售数据模型
unified_schema = {
    "date": "datetime",
    "platform": "str", # amazon_us / shopify / walmart
    "sku": "str",
    "product_name": "str",
    "revenue": "float",
    "orders": "int",
    "units": "int",
    "refunds": "float",
    "ad_spend": "float",
    "ad_revenue": "float",
    "cogs": "float", # 产品成本
    "fba_fees": "float", # 平台费用
    "net_profit": "float" # 净利润
}

def merge_platforms(amazon_df, shopify_df, walmart_df=None):
    """合并多平台数据到统一格式"""
    dfs = []

    # Amazon
    amazon_df["platform"] = "amazon_us"
    amazon_df = amazon_df.rename(columns={...}) # 映射列名
    dfs.append(amazon_df)

    # Shopify
    shopify_df["platform"] = "shopify"
    shopify_df = shopify_df.rename(columns={...})
    dfs.append(shopify_df)

    if walmart_df is not None:
        walmart_df["platform"] = "walmart"
        dfs.append(walmart_df)

    return pd.concat(dfs, ignore_index=True)

6. AI 增强 Dashboard

6.1 电商核心 KPI 体系

根据行业最佳实践（ThoughtSpot，Feedcast），电商 Dashboard 应该追踪以下 KPI：

Content rephrased for compliance with licensing restrictions.

6.2 异常检测（多种方法）

def detect_anomalies(df: pd.DataFrame, metric: str, threshold: float = 2.0):
    """基于 Z-Score 的异常检测"""
    mean = df[metric].rolling(window=7).mean()
    std = df[metric].rolling(window=7).std()
    z_score = (df[metric] - mean) / std

    anomalies = df[abs(z_score) > threshold].copy()
    anomalies["direction"] = z_score.apply(lambda x: " 异常高" if x > 0 else " 异常低")

    return anomalies

# 在 Dashboard 中显示
anomalies = detect_anomalies(daily_data, "revenue")
if len(anomalies) > 0:
    st.warning(f" 发现 {len(anomalies)} 个异常数据点")
    st.dataframe(anomalies[["date", "revenue", "direction"]])

6.2 异常检测（多种方法）

import numpy as np

# 方法 1：Z-Score 异常检测（简单有效）
def detect_zscore_anomalies(df: pd.DataFrame, metric: str,
                            window: int = 7, threshold: float = 2.0):
    """基于滚动 Z-Score 的异常检测"""
    mean = df[metric].rolling(window=window).mean()
    std = df[metric].rolling(window=window).std()
    z_score = (df[metric] - mean) / std

    anomalies = df[abs(z_score) > threshold].copy()
    anomalies["z_score"] = z_score[abs(z_score) > threshold]
    anomalies["direction"] = anomalies["z_score"].apply(
        lambda x: " 异常高" if x > 0 else " 异常低"
    )
    return anomalies

# 方法 2：IQR 异常检测（对非正态分布更稳健）
def detect_iqr_anomalies(df: pd.DataFrame, metric: str, multiplier: float = 1.5):
    """基于四分位距的异常检测"""
    Q1 = df[metric].quantile(0.25)
    Q3 = df[metric].quantile(0.75)
    IQR = Q3 - Q1

    lower = Q1 - multiplier * IQR
    upper = Q3 + multiplier * IQR

    anomalies = df[(df[metric] < lower) | (df[metric] > upper)].copy()
    anomalies["direction"] = anomalies[metric].apply(
        lambda x: " 异常高" if x > upper else " 异常低"
    )
    return anomalies

# 方法 3：同比/环比异常检测（电商最实用）
def detect_period_anomalies(df: pd.DataFrame, metric: str,
                            threshold_pct: float = 0.3):
    """基于同比/环比变化的异常检测"""
    df = df.copy()
    df['wow_change'] = df[metric].pct_change(periods=7) # 周环比
    df['mom_change'] = df[metric].pct_change(periods=30) # 月环比

    anomalies = df[
        (abs(df['wow_change']) > threshold_pct) |
        (abs(df['mom_change']) > threshold_pct)
    ].copy()

    return anomalies

# 在 Dashboard 中整合
def render_anomaly_alerts(df: pd.DataFrame):
    """在 Dashboard 中显示异常告警"""
    metrics_to_check = {
        "revenue": {"threshold": 2.0, "label": "收入"},
        "orders": {"threshold": 2.0, "label": "订单"},
        "acos": {"threshold": 1.5, "label": "ACOS"},
        "conversion_rate": {"threshold": 2.0, "label": "转化率"}
    }

    all_anomalies = []
    for metric, config in metrics_to_check.items():
        if metric in df.columns:
            anomalies = detect_zscore_anomalies(df, metric, threshold=config["threshold"])
            for _, row in anomalies.iterrows():
                all_anomalies.append({
                    "日期": row["date"],
                    "指标": config["label"],
                    "方向": row["direction"],
                    "值": row[metric],
                    "Z-Score": f"{row['z_score']:.1f}"
                })

    if all_anomalies:
        st.warning(f" 发现 {len(all_anomalies)} 个异常数据点")
        st.dataframe(pd.DataFrame(all_anomalies), use_container_width=True)
    else:
        st.success(" 所有指标正常")

6.3 利润分析模块

def render_profitability_tab(df: pd.DataFrame):
    """利润分析 Tab"""
    st.header(" 利润分析")

    # SKU 级别利润计算
    df['gross_profit'] = df['revenue'] - df['cogs'] - df['fba_fees'] - df['ad_spend']
    df['gross_margin'] = df['gross_profit'] / df['revenue'] * 100
    df['net_profit'] = df['gross_profit'] - df['other_costs']
    df['net_margin'] = df['net_profit'] / df['revenue'] * 100

    # 利润瀑布图
    st.subheader("利润瀑布图（单位经济模型）")
    avg_price = df['revenue'].sum() / df['units'].sum()
    avg_cogs = df['cogs'].sum() / df['units'].sum()
    avg_fba = df['fba_fees'].sum() / df['units'].sum()
    avg_ad = df['ad_spend'].sum() / df['units'].sum()
    avg_other = df['other_costs'].sum() / df['units'].sum()
    avg_profit = avg_price - avg_cogs - avg_fba - avg_ad - avg_other

    waterfall_data = pd.DataFrame({
        'item': ['售价', 'COGS', 'FBA 费用', '广告费', '其他成本', '净利润'],
        'amount': [avg_price, -avg_cogs, -avg_fba, -avg_ad, -avg_other, avg_profit]
    })

    fig = px.bar(waterfall_data, x='item', y='amount',
                 color='amount', color_continuous_scale=['red', 'green'],
                 title=f"单件利润分解（平均净利润: ${avg_profit:.2f}）")
    st.plotly_chart(fig, use_container_width=True)

    # SKU 利润排名
    st.subheader("SKU 利润排名")
    sku_profit = df.groupby('sku').agg({
        'revenue': 'sum',
        'gross_profit': 'sum',
        'net_profit': 'sum',
        'units': 'sum'
    }).reset_index()
    sku_profit['margin'] = sku_profit['net_profit'] / sku_profit['revenue'] * 100
    sku_profit = sku_profit.sort_values('net_profit', ascending=False)

    # 标记亏损 SKU
    st.dataframe(
        sku_profit.style.applymap(
            lambda x: 'color: red' if isinstance(x, (int, float)) and x < 0 else '',
            subset=['net_profit', 'margin']
        ),
        use_container_width=True
    )

6.4 库存健康度模块

def render_inventory_tab(df_inv: pd.DataFrame):
    """库存健康度 Tab"""
    st.header(" 库存健康度")

    # 计算可售天数
    df_inv['days_of_supply'] = df_inv['quantity'] / df_inv['daily_sales'].replace(0, 0.1)

    # 库存状态分类
    def classify_inventory(days):
        if days < 7:
            return " 紧急补货"
        elif days < 14:
            return " 即将缺货"
        elif days < 30:
            return " 需要关注"
        elif days < 90:
            return " 健康"
        else:
            return " 库存过多"

    df_inv['status'] = df_inv['days_of_supply'].apply(classify_inventory)

    # 状态分布
    col1, col2 = st.columns(2)
    with col1:
        status_counts = df_inv['status'].value_counts()
        fig = px.pie(values=status_counts.values, names=status_counts.index,
                     title="库存状态分布")
        st.plotly_chart(fig, use_container_width=True)

    with col2:
        # 紧急补货列表
        urgent = df_inv[df_inv['days_of_supply'] < 14].sort_values('days_of_supply')
        st.subheader(f" 需要补货的 SKU ({len(urgent)} 个)")
        st.dataframe(urgent[['sku', 'product_name', 'quantity',
                             'daily_sales', 'days_of_supply', 'status']],
                     use_container_width=True)

    # 长期仓储费预警
    st.subheader(" 长期仓储费预警")
    long_storage = df_inv[df_inv['days_in_warehouse'] > 180]
    if len(long_storage) > 0:
        estimated_fee = long_storage['quantity'].sum() * 6.90 # $6.90/cubic foot/month
        st.warning(f" {len(long_storage)} 个 SKU 在仓超过 180 天，预估月仓储费: ${estimated_fee:,.0f}")
        st.dataframe(long_storage[['sku', 'quantity', 'days_in_warehouse']])

6.5 AI 自动洞察

def generate_ai_insights(data_summary: dict) -> str:
    """用 LLM 生成数据洞察"""
    prompt = f"""
你是一个电商数据分析专家。以下是过去 7 天的运营数据摘要：

{data_summary}

请生成 3-5 条关键洞察，每条包含：
1. 发现了什么（数据事实）
2. 为什么重要（业务影响）
3. 建议的行动（具体可执行）

用简洁的中文回答，每条不超过 2 句话。
"""
    # 调用 LLM API
    return llm_call(prompt)

7. 部署与分享

7.1 部署选项

7.2 Streamlit Cloud 一键部署

# 1. 确保项目有 requirements.txt
echo "streamlit\nplotly\npandas\nopenpyxl" > requirements.txt

# 2. 推送到 GitHub
git add -A && git commit -m "add dashboard" && git push

# 3. 在 share.streamlit.io 连接 GitHub 仓库
# 选择 dashboard.py 作为入口文件
# 点击 Deploy

8. 常见陷阱

8.1 把所有指标都放上去

看板放了 40 个图，等于没有重点。一个有用的看板回答的是「今天我要不要做什么」，不是「我有多少数据」。

8.2 没有基准线

一个数字单独放着没有意义。同比、环比、目标线、行业基准——至少要有一个参照，否则看的人不知道该不该紧张。

8.3 数据延迟没有标注

看板上的数字是实时的还是昨天的，直接影响决策。没标注就会有人拿 T-1 的数据做当天的调价决策。

8.4 做成只有自己看得懂的样子

看板是给团队用的。字段命名、指标口径、异常颜色的含义，都要写在旁边而不是存在你脑子里。

什么时候这套不管用

• 只有你一个人看。 自建 Dashboard 的成本在维护：数据源变了要改、平台加字段要跟、报错了要修。一个人用的话，一个跑得通的 notebook 或者一张定期刷新的 Google Sheet 通常够了，省下的时间比看板好看带来的价值多。
• 指标定义还没统一。 三个人对“利润率“有三个算法时，做出来的看板会变成争论的战场而不是共识的载体。先把指标口径写下来定死（分子分母各含什么、按什么时间口径），再做可视化。这一步跳过，后面每次看数都要重新对齐一次。
• 数据延迟到看板没有决策价值。 每天早上刷新的看板，对需要小时级反应的场景（大促当天的库存、广告预算耗尽）没用。上看板之前先确认：这些数据多久变一次、你多久看一次、看到之后多久能动。三个频率对不上，看板就只是好看。
• 现成 BI 已经够用。 平台自带的报表、Shopify 的分析、或者便宜的 BI SaaS 覆盖了大部分常规指标。自建的理由应该是“跨平台数据只有我自己能拼“或者“我要的指标没有工具算“，不是“想要自己的看板“。

9. 完成标志

• 构建一个包含 4+ 模块的 Streamlit Dashboard
• 整合至少 2 个平台的数据（Amazon + Shopify）
• 实现异常检测功能（自动标注异常数据点）
• 集成 AI 洞察生成（LLM 自动分析数据）
• 部署到 Streamlit Cloud 或其他平台

(b7-review-nlp-system.md) | Path 总览 | B9 图片 >