当前位置：首页 > news >正文

防窜货实时监控系统大屏：用python和Streamlit实现的防窜货大屏

news 来源：原创 2025/5/15 9:59:48

基于python和Streamlit实现的防窜货大屏。时间筛选，窜货区域热度图，窜货趋势分析，不同区域窜货的一个对比，🔎 产品追溯查询，输入产品ID可以查询流向。

import streamlit as st
import pandas as pd
import numpy as np
import plotly.express as px
from datetime import datetime, timedelta


# ==================== 数据生成模块 ====================
@st.cache_data
def generate_anti_leak_data():
    """生成模拟防窜货数据，包含地理坐标和时间序列"""
    np.random.seed(42)
    product_ids = [f"PROD{str(i).zfill(5)}" for i in range(1000)]
    regions = ['华东', '华北', '华南', '华中', '西北']

    # 生成时间偏移量并转换为Python原生int类型
    days_offset = np.random.randint(0, 30, 5000).tolist()  # 转换为list避免numpy类型

    data = pd.DataFrame({
        "product_id": np.random.choice(product_ids, 5000),
        "origin_region": np.random.choice(regions, 5000),
        "sold_region": np.random.choice(
            regions + ['异常区域'],
            5000,
            p=[0.14, 0.14, 0.14, 0.14, 0.14, 0.3]
        ),
        "scan_time": [datetime.now() - timedelta(days=int(x)) for x in days_offset],
        "scan_location": [
            f"{lat:.4f},{lon:.4f}"
            for lat, lon in zip(
                np.random.uniform(28.0, 42.0, 5000),
                np.random.uniform(110.0, 122.0, 5000)
            )
        ],
        "is_leaked": np.random.choice([0, 1], 5000, p=[0.7, 0.3])
    })

    # 处理地理坐标
    data[['lat', 'lon']] = data['scan_location'].str.split(',', expand=True).astype(float)
    return data


# ==================== 数据预处理 ====================
df = generate_anti_leak_data()
df['scan_date'] = df['scan_time'].dt.date  # 添加日期字段用于筛选

# ==================== 界面配置 ====================
st.set_page_config(
    page_title="防窜货智能监控系统",
    layout="wide",
    page_icon="🚨",
    menu_items={
        'Get Help': 'https://example.com',
        'Report a bug': "mailto:support@example.com",
        'About': "V1.0 - 防窜货监控系统"
    }
)

# ==================== 侧边栏控件 ====================
with st.sidebar:
    st.header("🔍 数据筛选")
    selected_regions = st.multiselect(
        "选择来源大区",
        options=df['origin_region'].unique(),
        default=df['origin_region'].unique()
    )

    start_date, end_date = st.date_input(
        "选择日期范围",
        value=(df['scan_date'].min(), df['scan_date'].max()),
        min_value=df['scan_date'].min(),
        max_value=df['scan_date'].max()
    )

# ==================== 数据过滤 ====================
filtered_df = df[
    (df['origin_region'].isin(selected_regions)) &
    (df['scan_date'] >= start_date) &
    (df['scan_date'] <= end_date)
    ]

leak_data = filtered_df[filtered_df['is_leaked'] == 1]

# ==================== 主界面布局 ====================
st.title("📊 防窜货实时监控大屏")
st.markdown("---")

# 第一行：关键指标
col1, col2, col3, col4 = st.columns(4)
with col1:
    st.metric("今日预警数", leak_data[leak_data['scan_date'] == datetime.today().date()].shape[0])
with col2:
    st.metric("总窜货率", f"{leak_data.shape[0] / filtered_df.shape[0] * 100:.1f}%")
with col3:
    st.metric("活跃经销商数", filtered_df['origin_region'].nunique())
with col4:
    st.metric("扫码覆盖率",
              f"{filtered_df[filtered_df['scan_location'].notnull()].shape[0] / filtered_df.shape[0] * 100:.1f}%")

# 第二行：地图和趋势
col_left, col_right = st.columns([2, 1])
with col_left:
    # 地理热力图
    fig_map = px.density_mapbox(
        leak_data,
        lat='lat',
        lon='lon',
        radius=10,
        zoom=4,
        mapbox_style="carto-positron",
        title="<b>窜货事件地理分布</b>",
        height=500,
        color_continuous_scale=px.colors.sequential.Viridis
    )
    st.plotly_chart(fig_map, use_container_width=True)

with col_right:
    # 时间趋势分析
    time_analysis = leak_data.resample('D', on='scan_time').size().reset_index(name='count')
    fig_line = px.area(
        time_analysis,
        x='scan_time',
        y='count',
        title="<b>窜货趋势分析</b>",
        labels={'count': '事件数量'},
        height=400,
        line_shape='spline'
    )
    st.plotly_chart(fig_line, use_container_width=True)

# 第三行：分析图表
tab1, tab2, tab3 = st.tabs(["区域分析", "渠道路径", "原始数据"])

with tab1:
    # 区域对比分析
    regional_data = leak_data['origin_region'].value_counts().reset_index()
    fig_bar = px.bar(
        regional_data,
        x='origin_region',
        y='count',
        color='count',
        title="<b>各区域窜货数量对比</b>",
        labels={'origin_region': '大区', 'count': '数量'},
        height=400
    )
    st.plotly_chart(fig_bar, use_container_width=True)

with tab2:
    # 窜货路径分析
    path_data = leak_data.groupby(['origin_region', 'sold_region']).size().reset_index(name='count')
    fig_sunburst = px.sunburst(
        path_data,
        path=['origin_region', 'sold_region'],
        values='count',
        title="<b>跨区窜货路径分析</b>",
        height=600
    )
    st.plotly_chart(fig_sunburst, use_container_width=True)

with tab3:
    # 原始数据展示
    st.dataframe(
        filtered_df[['product_id', 'origin_region', 'sold_region', 'scan_time']],
        height=500,
        use_container_width=True,
        column_config={
            "scan_time": st.column_config.DatetimeColumn("扫码时间", format="YYYY-MM-DD HH:mm")
        }
    )

# ==================== 产品追溯模块 ====================
st.markdown("---")
with st.expander("🔎 产品追溯查询", expanded=True):
    product_id = st.text_input("输入产品ID查询流向", help="输入格式示例：PROD00001")
    if product_id:
        trace_data = df[df['product_id'] == product_id.upper().strip()]
        if not trace_data.empty:
            col_info, col_map = st.columns([1, 2])
            with col_info:
                st.subheader("产品信息")
                st.metric("生产大区", trace_data['origin_region'].values[0])
                st.metric("销售区域",
                          trace_data['sold_region'].values[0],
                          delta="异常" if trace_data['sold_region'].values[0] == '异常区域' else None)
                st.write(f"**扫码时间**: {trace_data['scan_time'].dt.strftime('%Y-%m-%d %H:%M').values[0]}")

            with col_map:
                st.subheader("地理位置")
                st.map(trace_data[['lat', 'lon']], zoom=5, use_container_width=True)
        else:
            st.warning("未找到该产品信息")

# ==================== 运行说明 ====================
"""
**运行指南**：
1. 安装依赖：`pip install streamlit pandas numpy plotly`
2. 保存文件为 `anti_leak_monitor.py`
3. 运行命令：`streamlit run anti_leak_monitor.py`

**功能说明**：
- 实时监控窜货事件的地理分布和趋势变化
- 支持多维度数据筛选和交互式分析
- 提供完整的产品溯源功能
- 自动生成模拟数据，方便演示测试

**技术亮点**：
✅ 类型安全处理：完全兼容numpy和Python原生类型  
✅ 响应式布局：适配不同屏幕尺寸  
✅ 可视化增强：使用Viridis色系优化地图显示  
✅ 错误预防：输入数据自动标准化处理  
✅ 性能优化：@st.cache_data缓存数据生成
"""

import streamlit as st

# 必须作为第一个Streamlit命令
st.set_page_config(
    page_title="防窜货智能监控系统",
    layout="wide",
    page_icon="🚨",
    menu_items={
        'Get Help': 'https://example.com',
        'Report a bug': "mailto:support@example.com",
        'About': "V1.0 - 防窜货监控系统"
    }
)

# 其他库导入
import pandas as pd
import numpy as np
import plotly.express as px
from datetime import datetime, timedelta

# ==================== 数据生成模块 ====================
@st.cache_data
def generate_anti_leak_data():
    """生成模拟防窜货数据"""
    np.random.seed(42)
    product_ids = [f"PROD{str(i).zfill(5)}" for i in range(1000)]
    regions = ['华东', '华北', '华南', '华中', '西北']
    
    # 生成时间数据（兼容类型处理）
    days_offset = np.random.randint(0, 30, 5000).tolist()
    
    data = pd.DataFrame({
        "product_id": np.random.choice(product_ids, 5000),
        "origin_region": np.random.choice(regions, 5000),
        "sold_region": np.random.choice(
            regions + ['异常区域'], 
            5000,
            p=[0.14, 0.14, 0.14, 0.14, 0.14, 0.3]
        ),
        "scan_time": [datetime.now() - timedelta(days=int(x)) for x in days_offset],
        "scan_location": [
            f"{lat:.4f},{lon:.4f}" 
            for lat, lon in zip(
                np.random.uniform(28.0, 42.0, 5000),
                np.random.uniform(110.0, 122.0, 5000)
            )
        ],
        "is_leaked": np.random.choice([0, 1], 5000, p=[0.7, 0.3])
    })
    
    # 处理地理坐标
    data[['lat', 'lon']] = data['scan_location'].str.split(',', expand=True).astype(float)
    data['scan_date'] = data['scan_time'].dt.date
    return data

# ==================== 初始化数据 ====================
df = generate_anti_leak_data()

# ==================== 侧边栏控件 ====================
with st.sidebar:
    st.header("🔍 数据筛选")
    selected_regions = st.multiselect(
        "选择来源大区",
        options=df['origin_region'].unique(),
        default=df['origin_region'].unique()
    )
    
    start_date, end_date = st.date_input(
        "选择日期范围",
        value=(df['scan_date'].min(), df['scan_date'].max()),
        min_value=df['scan_date'].min(),
        max_value=df['scan_date'].max()
    )

# ==================== 数据过滤 ====================
filtered_df = df[
    (df['origin_region'].isin(selected_regions)) &
    (df['scan_date'] >= start_date) &
    (df['scan_date'] <= end_date)
]
leak_data = filtered_df[filtered_df['is_leaked'] == 1]

# ==================== 主界面布局 ====================
st.title("📊 防窜货实时监控大屏")
st.markdown("---")

# 第一行：关键指标
col1, col2, col3, col4 = st.columns(4)
with col1:
    st.metric("今日预警数", leak_data[leak_data['scan_date'] == datetime.today().date()].shape[0])
with col2:
    st.metric("总窜货率", f"{leak_data.shape[0]/filtered_df.shape[0]*100:.1f}%")
with col3:
    st.metric("活跃经销商数", filtered_df['origin_region'].nunique())
with col4:
    st.metric("扫码覆盖率", f"{filtered_df[filtered_df['scan_location'].notnull()].shape[0]/filtered_df.shape[0]*100:.1f}%")

# 第二行：地图和趋势
col_left, col_right = st.columns([2, 1])
with col_left:
    fig_map = px.density_mapbox(
        leak_data,
        lat='lat',
        lon='lon',
        radius=10,
        zoom=4,
        mapbox_style="carto-positron",
        title="<b>窜货事件地理分布</b>",
        height=500
    )
    st.plotly_chart(fig_map, use_container_width=True)

with col_right:
    time_analysis = leak_data.resample('D', on='scan_time').size().reset_index(name='count')
    fig_line = px.area(
        time_analysis, 
        x='scan_time', 
        y='count',
        title="<b>窜货趋势分析</b>",
        height=400
    )
    st.plotly_chart(fig_line, use_container_width=True)

# 第三行：分析图表
tab1, tab2, tab3 = st.tabs(["区域分析", "渠道路径", "原始数据"])
with tab1:
    fig_bar = px.bar(
        leak_data['origin_region'].value_counts().reset_index(),
        x='origin_region',
        y='count',
        title="<b>各区域窜货数量对比</b>",
        height=400
    )
    st.plotly_chart(fig_bar, use_container_width=True)

with tab2:
    path_data = leak_data.groupby(['origin_region', 'sold_region']).size().reset_index(name='count')
    fig_sunburst = px.sunburst(
        path_data,
        path=['origin_region', 'sold_region'],
        values='count',
        title="<b>跨区窜货路径分析</b>",
        height=600
    )
    st.plotly_chart(fig_sunburst, use_container_width=True)

with tab3:
    st.dataframe(
        filtered_df[['product_id', 'origin_region', 'sold_region', 'scan_time']],
        height=500,
        column_config={
            "scan_time": st.column_config.DatetimeColumn("扫码时间", format="YYYY-MM-DD HH:mm")
        }
    )

# 产品追溯模块
st.markdown("---")
with st.expander("🔎 产品追溯查询"):
    product_id = st.text_input("输入产品ID查询流向")
    if product_id:
        trace_data = df[df['product_id'] == product_id.strip().upper()]
        if not trace_data.empty:
            col_info, col_map = st.columns([1, 2])
            with col_info:
                st.subheader("产品信息")
                st.write(f"**生产大区**: {trace_data['origin_region'].values[0]}")
                st.write(f"**销售区域**: {trace_data['sold_region'].values[0]}")
                st.write(f"**扫码时间**: {trace_data['scan_time'].dt.strftime('%Y-%m-%d %H:%M').values[0]}")
            with col_map:
                st.subheader("地理位置")
                st.map(trace_data[['lat', 'lon']], zoom=5)
        else:
            st.warning("未找到该产品信息")

# ==================== 运行说明 ====================
"""
**运行步骤**：
1. 安装依赖：`pip install streamlit pandas numpy plotly`
2. 保存文件为 `anti_leak.py`
3. 运行命令：`streamlit run anti_leak.py`

**界面功能**：
- 实时更新的监控指标仪表盘
- 交互式地图显示窜货热点区域
- 时间趋势分析图表
- 多维度数据钻取分析
- 完整的产品溯源功能

**技术验证**：
✅ 已测试Python 3.8+环境兼容性  
✅ 修复所有已知类型错误  
✅ 确保页面加载顺序正确  
✅ 支持中文字符显示  
✅ 响应式布局适配不同屏幕
"""