#!/usr/bin/env python3 """ 销售数据分析看板 (Ozon 真实数据版) 依赖: streamlit, pandas, plotly, numpy 数据源: ~/Desktop/akasha_data/survey_latest.json (由 akasha_full_handshake.py 生成) 用法: streamlit run sales_dashboard.py """ import streamlit as st import pandas as pd import numpy as np import plotly.express as px import plotly.graph_objects as go from datetime import datetime, timedelta import random import time import json import os # ═══════════════════════════════════════════════════════════════ # 页面配置 # ═══════════════════════════════════════════════════════════════ # Alert acknowledgement state if "alert_acknowledged" not in st.session_state: st.session_state.alert_acknowledged = False st.set_page_config( page_title="销售数据分析看板", page_icon="📊", layout="wide", initial_sidebar_state="expanded", ) # ═══════════════════════════════════════════════════════════════ # 数据加载：优先从 survey_latest.json 读取 # ═══════════════════════════════════════════════════════════════ SURVEY_PATH = os.path.expanduser("~/Desktop/akasha_data/survey_latest.json") CATEGORIES = ["电子产品", "服装", "食品", "家居"] REGIONS = ["华东", "华南", "华北", "西南"] # 各类别的价格分布参数（仅 mock 模式使用） CAT_PRICE = { "电子产品": {"mu": 800, "sigma": 400, "min": 50, "max": 5000}, "服装": {"mu": 200, "sigma": 180, "min": 20, "max": 2000}, "食品": {"mu": 80, "sigma": 60, "min": 5, "max": 800}, "家居": {"mu": 300, "sigma": 250, "min": 30, "max": 3000}, } SURNAMES = ["王", "李", "张", "刘", "陈", "杨", "黄", "赵", "周", "吴", "徐", "孙", "马", "朱", "胡", "郭", "何", "高", "林", "罗"] GIVEN_2 = ["伟", "芳", "娜", "秀英", "敏", "静", "丽", "强", "磊", "军", "洋", "勇", "艳", "杰", "娟", "涛", "明", "超", "秀兰", "霞", "平", "刚", "桂英", "文", "华", "飞", "玉兰", "斌", "玲", "宇"] def make_name(): return random.choice(SURNAMES) + random.choice(GIVEN_2) def make_mock_orders(n=500): """生成模拟销售数据（手写数在无真实数据时使用）""" start = datetime(2025, 1, 1) end = datetime(2025, 12, 31) days = (end - start).days rows = [] for i in range(n): cat = random.choice(CATEGORIES) p = CAT_PRICE[cat] price = np.random.normal(p["mu"], p["sigma"]) price = max(p["min"], min(p["max"], price)) price = round(price, 2) day_off = random.randint(0, days) sd = start + timedelta(days=day_off) if sd.weekday() >= 5: price = round(price * random.uniform(1.0, 1.4), 2) region = random.choice(REGIONS) if random.random() < 0.1: region = random.choices(REGIONS, weights=[4, 3, 2, 1])[0] rows.append({ "订单ID": f"ORD-{20250001 + i}", "客户姓名": make_name(), "产品类别": cat, "销售额": price, "销售日期": sd, "地区": region, }) df = pd.DataFrame(rows) df = df.sort_values("销售日期").reset_index(drop=True) return df def load_data() -> pd.DataFrame: """加载数据：优先来自 JSON，失败则 mock""" if os.path.isfile(SURVEY_PATH): try: with open(SURVEY_PATH, "r", encoding="utf-8") as f: survey = json.load(f) raw = survey.get("ozon_sales_data", []) if raw and len(raw) > 0: df = pd.DataFrame(raw) df["销售日期"] = pd.to_datetime(df["销售日期"]) df = df.sort_values("销售日期").reset_index(drop=True) st.sidebar.success(f"✅ 数据来源: {SURVEY_PATH} ({len(df)} 条)") return df except Exception as e: st.sidebar.warning(f"JSON 读取失败: {e}，使用模拟数据") # fallback: mock data st.sidebar.info("📊 使用模拟数据（未检测到 survey_latest.json）") return make_mock_orders(500) def detect_anomalies(df): """检测月环比下降超过 20% 的月份""" monthly = df.set_index("销售日期").resample("ME")["销售额"].sum() if len(monthly) < 2: return pd.DataFrame(), [] pct_change = monthly.pct_change() * 100 anomaly = pct_change[pct_change < -20] warnings = [] for month, val in anomaly.items(): if not pd.isna(val): label = month.strftime("%Y-%m") warnings.append(f"⚠️ {label} 销售额环比下降 {abs(val):.1f}%") return monthly, warnings # ═══════════════════════════════════════════════════════════════ # Streamlit 主界面 # ═══════════════════════════════════════════════════════════════ st.markdown(""" """, unsafe_allow_html=True) with st.sidebar: st.title("📊 销售看板") st.caption("实时销售数据分析") df = load_data() is_real = os.path.isfile(SURVEY_PATH) and len(df) > 0 # Read alert history from JSON alert_data = [] if is_real: try: with open(SURVEY_PATH) as f2: sv_file = json.load(f2) alert_data = sv_file.get("alert_history", []) except Exception: alert_data = [] has_alerts = any(a.get("level_str") in ("critical", "warning") for a in alert_data) # Alert banner (flashing) if has_alerts and not st.session_state.alert_acknowledged: crit = sum(1 for a in alert_data if a.get("level_str") == "critical") warn = sum(1 for a in alert_data if a.get("level_str") == "warning") lvl = "critical" if crit > 0 else "warning" label = "\u4e25\u91cd" if crit > 0 else "\u8b66\u544a" c1, c2 = st.columns([5, 1]) with c1: st.markdown( "
\ud83d\udea8 \u544a\u8b66\uff1a\u6709 " + str(crit + warn) + " \u6761\u544a\u8b66\u672a\u786e\u8ba4\uff08" + label + "\uff09
", unsafe_allow_html=True, ) with c2: if st.button("\u2714 \u786e\u8ba4", use_container_width=True): st.session_state.alert_acknowledged = True st.rerun() st.subheader("过滤器") categories = st.multiselect( "产品类别", sorted(df["产品类别"].unique()) if not df.empty else CATEGORIES, default=sorted(df["产品类别"].unique()) if not df.empty else CATEGORIES, ) min_date = df["销售日期"].min() if not df.empty else datetime(2025, 1, 1) max_date = df["销售日期"].max() if not df.empty else datetime(2025, 12, 31) date_range = st.slider( "日期范围", min_value=min_date.to_pydatetime(), max_value=max_date.to_pydatetime(), value=(min_date.to_pydatetime(), max_date.to_pydatetime()), format="YYYY-MM-DD", ) st.divider() auto_refresh = st.checkbox("自动刷新（每 60 秒）", value=True) refresh_btn = st.button("🔄 立即刷新") st.subheader("告警消息") alert_placeholder = st.empty() st.divider() st.caption(f"最后更新: {datetime.now().strftime('%H:%M:%S')}") # 主区域 st.title("📈 销售数据分析看板") badge_class = "real" if is_real else "mock" badge_text = "Ozon 真实数据" if is_real else "模拟数据" st.markdown(f'● {badge_text}', unsafe_allow_html=True) # 应用过滤器 mask = df["产品类别"].isin(categories) mask &= df["销售日期"] >= pd.Timestamp(date_range[0]) mask &= df["销售日期"] <= pd.Timestamp(date_range[1]) filtered = df[mask].copy() if filtered.empty: st.warning("所选条件下无数据，请调整过滤器。") st.stop() # 异常检测 monthly_series, warnings = detect_anomalies(filtered) alert_placeholder.empty() if warnings: for w in warnings: alert_placeholder.error(w) else: alert_placeholder.info("✅ 未检测到异常波动") # KPI total_sales = filtered["销售额"].sum() total_orders = len(filtered) avg_order = total_sales / total_orders if total_orders > 0 else 0 top_cat = filtered.groupby("产品类别")["销售额"].sum().idxmax() if not filtered.empty else "—" # 三列 KPI col1, col2, col3 = st.columns(3) with col1: st.markdown(f'
💰 总销售额
¥{total_sales:,.0f}
实时统计
', unsafe_allow_html=True) st.markdown(f'
📦 总订单数
{total_orders:,}
', unsafe_allow_html=True) with col2: st.markdown(f'
🎯 平均客单价
¥{avg_order:,.0f}
', unsafe_allow_html=True) st.markdown(f'
🏆 最畅销品类
{top_cat}
', unsafe_allow_html=True) with col3: pass # 月度趋势 st.subheader("📅 月度销售额趋势") monthly = filtered.set_index("销售日期").resample("ME")["销售额"].sum().reset_index() monthly["月份"] = monthly["销售日期"].dt.strftime("%Y-%m") fig_line = px.line( monthly, x="月份", y="销售额", markers=True, line_shape="spline", color_discrete_sequence=["#38bdf8"], ) fig_line.update_traces( hovertemplate="%{x}
销售额: ¥%{y:,.0f}", line=dict(width=3), marker=dict(size=8), ) mon_s, _ = detect_anomalies(filtered) if not mon_s.empty: pct = mon_s.pct_change() * 100 bad_months = pct[pct < -20] for idx in bad_months.index: val = mon_s.loc[idx] fig_line.add_annotation( x=idx.strftime("%Y-%m"), y=val, text="🔻", showarrow=True, arrowhead=2, arrowsize=1.5, arrowcolor="#ef4444", font=dict(size=16, color="#ef4444"), ax=0, ay=-40, ) fig_line.update_layout( height=350, margin=dict(l=20, r=20, t=10, b=10), paper_bgcolor="#0f172a", plot_bgcolor="#0f172a", font=dict(color="#e2e8f0"), xaxis=dict(gridcolor="#1e293b", title=""), yaxis=dict(gridcolor="#1e293b", title="销售额 (¥)", tickformat=","), hovermode="x unified", ) st.plotly_chart(fig_line, use_container_width=True) # 地区饼图 st.subheader("🌍 各地区销售占比") region_total = filtered.groupby("地区")["销售额"].sum().reset_index() if not region_total.empty: fig_pie = px.pie( region_total, names="地区", values="销售额", color_discrete_sequence=px.colors.sequential.Blues_r, hole=0.4, custom_data=["地区"], ) fig_pie.update_traces( hovertemplate="%{label}
销售额: ¥%{value:,.0f}
占比: %{percent}", textposition="inside", textinfo="label+percent", ) fig_pie.update_layout( height=400, margin=dict(l=20, r=20, t=10, b=10), paper_bgcolor="#0f172a", font=dict(color="#e2e8f0"), legend=dict(font=dict(color="#94a3b8")), ) col_chart, col_drill = st.columns([2, 1]) with col_chart: st.plotly_chart(fig_pie, use_container_width=True, key="pie_main") with col_drill: st.subheader("🔍 类别下钻") drill_region = st.selectbox("选择地区查看类别分布", filtered["地区"].unique()) drill_df = filtered[filtered["地区"] == drill_region] cat_dist = drill_df.groupby("产品类别")["销售额"].sum().reset_index() if not cat_dist.empty: fig_drill = px.bar( cat_dist, x="产品类别", y="销售额", color="产品类别", color_discrete_sequence=px.colors.qualitative.Bold, ) fig_drill.update_layout( height=300, margin=dict(l=10, r=10, t=10, b=10), paper_bgcolor="#0f172a", plot_bgcolor="#0f172a", font=dict(color="#e2e8f0"), xaxis=dict(gridcolor="#1e293b"), yaxis=dict(gridcolor="#1e293b", tickformat=","), showlegend=False, ) st.plotly_chart(fig_drill, use_container_width=True) st.caption(f"{drill_region} 各品类销售额") else: st.info("该地区暂无数据") else: st.info("暂无地区数据") # 数据表格 + CSV 导出 # 告警历史面板 if alert_data: st.subheader("🚨 告警历史（最近10条）") for a_entry in alert_data[-10:][::-1]: lv = a_entry.get("level_str", "info") icons = {"critical": "🔴严重", "warning": "🟡警告", "info": "🔵信息"} ic = icons.get(lv, "🔵") sent = "✅" if a_entry.get("sent") else "❌" msg = a_entry.get("message", "") detail = a_entry.get("detail", "") ts = a_entry.get("time", "")[:19].replace("T", " ") st.markdown( '
' + '' + ic + '' + '' + msg + '' + '
' + ts + " | " + sent + " | " + detail + '' + '
', unsafe_allow_html=True, ) st.divider() st.divider() st.subheader("📋 原始数据预览") col_tab, col_exp = st.columns([4, 1]) with col_exp: csv_data = filtered.head(20).to_csv(index=False, encoding="utf-8-sig") st.download_button( label="📥 下载 CSV", data=csv_data, file_name=f"sales_data_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv", mime="text/csv", use_container_width=True, ) with col_tab: st.dataframe( filtered.head(20), use_container_width=True, height=450, column_config={ "订单ID": st.column_config.TextColumn("订单ID", width="small"), "客户姓名": st.column_config.TextColumn("客户姓名", width="small"), "产品类别": st.column_config.TextColumn("产品类别", width="small"), "销售额": st.column_config.NumberColumn("销售额 (¥)", format="¥%.2f"), "销售日期": st.column_config.DateColumn("销售日期", format="YYYY-MM-DD"), "地区": st.column_config.TextColumn("地区", width="small"), }, ) # 自动刷新 if auto_refresh: time.sleep(60) st.rerun()