Python
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import random
import time
import threading
from matplotlib.animation import FuncAnimation
from matplotlib.gridspec import GridSpec
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore')
plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False
class AdvancedDataAnalyzer:
def __init__(self):
self.data = pd.DataFrame()
self.is_running = True
self.fig = plt.figure(figsize=(16, 10), facecolor='#f8f9fa')
self.gs = GridSpec(3, 3, figure=self.fig, hspace=0.3, wspace=0.3)
self.real_time_data = []
self.categories = ['产品A', '产品B', '产品C', '产品D', '产品E']
self.regions = ['华北', '华东', '华南', '西南', '西北', '东北']
self.stats_history = {
'mean': [], 'max': [], 'min': [], 'std': []
}
self.init_subplots()
def init_subplots(self):
self.ax1 = self.fig.add_subplot(self.gs[0, 0])
self.ax2 = self.fig.add_subplot(self.gs[0, 1])
self.ax3 = self.fig.add_subplot(self.gs[0, 2])
self.ax4 = self.fig.add_subplot(self.gs[1, 0])
self.ax5 = self.fig.add_subplot(self.gs[1, 1])
self.ax6 = self.fig.add_subplot(self.gs[1, 2])
self.ax7 = self.fig.add_subplot(self.gs[2, :])
self.ax1.set_title('实时销售趋势', fontsize=12, fontweight='bold', color='#2c3e50')
self.ax2.set_title('产品销量分布', fontsize=12, fontweight='bold', color='#2c3e50')
self.ax3.set_title('区域销售占比', fontsize=12, fontweight='bold', color='#2c3e50')
self.ax4.set_title('价格波动曲线', fontsize=12, fontweight='bold', color='#2c3e50')
self.ax5.set_title('销量热力分布', fontsize=12, fontweight='bold', color='#2c3e50')
self.ax6.set_title('统计指标变化', fontsize=12, fontweight='bold', color='#2c3e50')
self.ax7.set_title('综合数据分析报告', fontsize=14, fontweight='bold', color='#2c3e50')
for ax in [self.ax1, self.ax2, self.ax3, self.ax4, self.ax5, self.ax6, self.ax7]:
ax.grid(True, alpha=0.3, linestyle='--')
ax.set_facecolor('#ffffff')
for spine in ax.spines.values():
spine.set_color('#bdc3c7')
def generate_mock_data(self):
start_date = datetime.now() - timedelta(days=30)
dates = [start_date + timedelta(days=i) for i in range(30)]
data_list = []
for date in dates:
for category in self.categories:
for region in self.regions:
sales = random.randint(100, 1000)
price = round(random.uniform(20, 200), 2)
cost = round(price * random.uniform(0.4, 0.7), 2)
profit = round((price - cost) * sales, 2)
data_list.append({
'日期': date,
'产品': category,
'区域': region,
'销量': sales,
'单价': price,
'成本': cost,
'利润': profit
})
self.data = pd.DataFrame(data_list)
return self.data
def generate_real_time_data(self):
while self.is_running:
new_sales = random.randint(300, 800)
self.real_time_data.append(new_sales)
if len(self.real_time_data) > 50:
self.real_time_data.pop(0)
time.sleep(0.5)
def update_stats(self):
if len(self.data) > 0:
current_mean = self.data['销量'].mean()
current_max = self.data['销量'].max()
current_min = self.data['销量'].min()
current_std = self.data['销量'].std()
self.stats_history['mean'].append(current_mean)
self.stats_history['max'].append(current_max)
self.stats_history['min'].append(current_min)
self.stats_history['std'].append(current_std)
for key in self.stats_history:
if len(self.stats_history[key]) > 20:
self.stats_history[key].pop(0)
def update_charts(self, frame):
if len(self.data) == 0:
self.generate_mock_data()
self.update_stats()
data = self.data
real_time = self.real_time_data[-50:] if len(self.real_time_data) >= 50 else self.real_time_data
self.ax1.clear()
self.ax1.plot(range(len(real_time)), real_time, color='#3498db', linewidth=2, marker='o', markersize=3)
self.ax1.set_title('实时销售趋势', fontsize=12, fontweight='bold', color='#2c3e50')
self.ax1.fill_between(range(len(real_time)), real_time, alpha=0.2, color='#3498db')
self.ax1.grid(True, alpha=0.3)
self.ax2.clear()
product_sales = data.groupby('产品')['销量'].sum().sort_values(ascending=True)
colors = plt.cm.viridis(np.linspace(0, 1, len(product_sales)))
bars = self.ax2.barh(product_sales.index, product_sales.values, color=colors)
self.ax2.set_title('产品销量分布', fontsize=12, fontweight='bold', color='#2c3e50')
for bar in bars:
width = bar.get_width()
self.ax2.text(width + 10, bar.get_y() + bar.get_height()/2, f'{int(width)}', ha='left', va='center', fontsize=9)
self.ax3.clear()
region_sales = data.groupby('区域')['销量'].sum()
wedges, texts, autotexts = self.ax3.pie(region_sales.values, labels=region_sales.index,
autopct='%1.1f%%', startangle=90,
colors=plt.cm.Set3(np.linspace(0, 1, len(region_sales))))
self.ax3.set_title('区域销售占比', fontsize=12, fontweight='bold', color='#2c3e50')
self.ax4.clear()
price_trend = data.groupby(data['日期'].dt.date)['单价'].mean()
self.ax4.plot(price_trend.index, price_trend.values, color='#e74c3c', linewidth=2)
self.ax4.set_title('价格波动曲线', fontsize=12, fontweight='bold', color='#2c3e50')
self.ax4.tick_params(axis='x', rotation=45, labelsize=8)
self.ax5.clear()
pivot_data = data.pivot_table(index='区域', columns='产品', values='销量', aggfunc='sum')
im = self.ax5.imshow(pivot_data.values, cmap='YlOrRd', aspect='auto')
self.ax5.set_xticks(range(len(self.categories)))
self.ax5.set_xticklabels(self.categories, rotation=45)
self.ax5.set_yticks(range(len(self.regions)))
self.ax5.set_yticklabels(self.regions)
self.ax5.set_title('销量热力分布', fontsize=12, fontweight='bold', color='#2c3e50')
self.ax6.clear()
if len(self.stats_history['mean']) > 0:
self.ax6.plot(self.stats_history['mean'], label='平均销量', color='#2ecc71', linewidth=2)
self.ax6.plot(self.stats_history['max'], label='最高销量', color='#f39c12', linewidth=2)
self.ax6.plot(self.stats_history['min'], label='最低销量', color='#9b59b6', linewidth=2)
self.ax6.legend(fontsize=8)
self.ax6.set_title('统计指标变化', fontsize=12, fontweight='bold', color='#2c3e50')
self.ax7.clear()
total_sales = int(data['销量'].sum())
total_profit = round(data['profit'].sum(), 2) if 'profit' in data.columns else 0
avg_price = round(data['单价'].mean(), 2)
best_product = data.groupby('产品')['销量'].sum().idxmax()
best_region = data.groupby('区域')['销量'].sum().idxmax()
report_text = f"""
【实时数据分析报告】
总销量：{total_sales:,} 件 总利润：{total_profit:,} 元
平均单价：{avg_price} 元 最佳产品：{best_product}
最佳区域：{best_region} 数据更新时间：{datetime.now().strftime('%H:%M:%S')}
系统状态：运行中 ● 数据维度：7大指标 ● 覆盖区域：6个 ● 产品种类：5种
"""
self.ax7.text(0.05, 0.5, report_text, transform=self.ax7.transAxes, fontsize=12,
verticalalignment='center', bbox=dict(boxstyle='round,pad=1', facecolor='#ecf0f1', alpha=0.8))
self.ax7.set_title('综合数据分析报告', fontsize=14, fontweight='bold', color='#2c3e50')
self.ax7.set_xticks([])
self.ax7.set_yticks([])
plt.tight_layout()
return [self.ax1, self.ax2, self.ax3, self.ax4, self.ax5, self.ax6, self.ax7]
def start_analysis(self):
print("=" * 60)
print("🚀 高级Python数据分析可视化系统启动中...")
print("📊 功能：实时数据生成 | 多维度图表 | 智能分析 | 动态刷新")
print("=" * 60)
data_thread = threading.Thread(target=self.generate_real_time_data, daemon=True)
data_thread.start()
ani = FuncAnimation(self.fig, self.update_charts, interval=1000, cache_frame_data=False)
try:
plt.show()
except KeyboardInterrupt:
print("\n✅ 系统已安全退出")
finally:
self.is_running = False
if __name__ == "__main__":
analyzer = AdvancedDataAnalyzer()
analyzer.start_analysis()