长春绿园网站建设网络营销模式下品牌推广途径
基于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.datereturn 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+环境兼容性
✅ 修复所有已知类型错误
✅ 确保页面加载顺序正确
✅ 支持中文字符显示
✅ 响应式布局适配不同屏幕
"""