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()