引言：为什么需要进阶Python数据分析？
Python作为数据科学领域的首选语言，其生态系统已经发展得非常成熟。对于已经掌握基础数据分析技能的学习者来说，进阶到精通水平意味着能够更高效地处理复杂数据、构建自动化分析流程，并通过实际项目解决真实世界的问题。
本课程将带你从基础的Pandas操作进阶到高级数据处理技巧，从简单的可视化到构建交互式仪表板，从单机分析到大数据处理。我们将通过详尽的代码示例和完整的实战项目，帮助你真正掌握Python数据分析的核心技巧。
第一部分：Pandas高级数据处理技巧
1.1 高效的数据读取与内存优化
在处理大型数据集时，内存优化是首要考虑的问题。让我们看看如何通过优化数据类型来显著减少内存占用。
import pandas as pd
import numpy as np
# 创建一个示例数据集
def create_large_dataset(rows=1000000):
return pd.DataFrame({
'id': range(rows),
'category': np.random.choice(['A', 'B', 'C', 'D'], rows),
'value': np.random.randn(rows),
'timestamp': pd.date_range('2020-01-01', periods=rows, freq='T')
})
# 原始数据类型检查
df = create_large_dataset()
print("原始内存使用:")
print(df.info(memory_usage='deep'))
# 优化数据类型
def optimize_memory(df):
# 优化整数类型
for col in df.select_dtypes(include=['int64']).columns:
df[col] = pd.to_numeric(df[col], downcast='integer')
# 优化浮点数类型
for col in df.select_dtypes(include=['float64']).columns:
df[col] = pd.to_numeric(df[col], downcast='float')
# 优化对象类型（分类数据）
for col in df.select_dtypes(include=['object']).columns:
if len(df[col].unique()) / len(df[col]) < 0.5: # 如果唯一值少于50%
df[col] = df[col].astype('category')
return df
df_optimized = optimize_memory(df.copy())
print("\n优化后内存使用:")
print(df_optimized.info(memory_usage='deep'))
1.2 高效的分组与聚合操作
在处理大规模数据时，groupby操作可能会非常耗时。以下是一些优化技巧：
# 创建一个包含多个分组键的大数据集
np.random.seed(42)
large_df = pd.DataFrame({
'region': np.random.choice(['North', 'South', 'East', 'West'], 1000000),
'product': np.random.choice(['P1', 'P2', 'P3', 'P4', 'P5'], 1000000),
'sales': np.random.exponential(scale=100, size=1000000),
'quantity': np.random.randint(1, 100, 1000000)
})
# 方法1: 基础groupby
result1 = large_df.groupby(['region', 'product']).agg({
'sales': ['sum', 'mean', 'count'],
'quantity': ['mean', 'sum']
})
# 方法2: 使用numba加速（需要安装numba）
from numba import jit
@jit(nopython=True)
def compute_metrics(sales, quantity):
total_sales = np.sum(sales)
avg_sales = np.mean(sales)
total_qty = np.sum(quantity)
avg_qty = np.mean(quantity)
count = len(sales)
return total_sales, avg_sales, total_qty, avg_qty, count
# 方法3: 使用pandas的transform进行高效计算
large_df['sales_rank'] = large_df.groupby('region')['sales'].rank(ascending=False)
large_df['regional_avg_sales'] = large_df.groupby('region')['sales'].transform('mean')
# 性能比较
import time
start = time.time()
result_method1 = large_df.groupby(['region', 'product']).agg({
'sales': ['sum', 'mean'],
'quantity': ['mean']
})
time_method1 = time.time() - start
print(f"方法1耗时: {time_method1:.4f}秒")
print("方法1结果形状:", result_method1.shape)
1.3 复杂索引操作与多级索引处理
多级索引（MultiIndex）是Pandas中强大但常被忽视的功能，特别适合处理高维数据。
# 创建多级索引数据
arrays = [
['bar', 'bar', 'baz', 'baz', 'foo', 'foo', 'qux', 'qux'],
['one', 'two', 'one', 'two', 'one', 'two', 'one', 'two']
]
tuples = list(zip(*arrays))
index = pd.MultiIndex.from_tuples(tuples, names=['first', 'second'])
df_multi = pd.DataFrame(np.random.randn(8, 4), index=index, columns=['A', 'B', 'C', 'D'])
# 多级索引的切片操作
print("选择第一层为'bar'的数据:")
print(df_multi.loc['bar'])
print("\n选择第一层为'bar'且第二层为'one'的数据:")
print(df_multi.loc[('bar', 'one')])
# 多级索引的交换层级
df_swapped = df_multi.swaplevel(0, 1)
print("\n交换层级后的数据:")
print(df_swapped.sort_index())
# 多级索引的聚合
print("\n按第一层索引求和:")
print(df_multi.groupby(level=0).sum())
# 创建透视表（多级索引的另一种形式）
pivot_df = pd.DataFrame({
'A': ['one', 'one', 'two', 'two'],
'B': ['x', 'y', 'x', 'y'],
'C': [1, 2, 3, 4],
'D': [5, 6, 7, 8]
})
pivot_table = pivot_df.pivot_table(index='A', columns='B', values=['C', 'D'])
print("\n透视表结果:")
print(pivot_table)
第二部分：高级数据可视化
2.1 Matplotlib高级定制
import matplotlib.pyplot as plt
import matplotlib.gridspec as gridspec
from matplotlib.patches import Rectangle
# 创建复杂子图布局
fig = plt.figure(figsize=(15, 10))
gs = gridspec.GridSpec(2, 3, figure=fig)
ax1 = fig.add_subplot(gs[0, :2]) # 第一行，前两列
ax2 = fig.add_subplot(gs[0, 2]) # 第一行，第三列
ax3 = fig.add_subplot(gs[1, :]) # 第二行，所有列
# 数据准备
x = np.linspace(0, 10, 100)
y1 = np.sin(x)
y2 = np.cos(x)
y3 = np.exp(-x/5) * np.sin(x)
# 绘制第一个子图：带标注的线图
ax1.plot(x, y1, 'b-', linewidth=2, label='sin(x)')
ax1.plot(x, y2, 'r--', linewidth=2, label='cos(x)')
ax1.axhline(y=0, color='k', linestyle=':', alpha=0.3)
ax1.axvline(x=np.pi, color='g', linestyle=':', alpha=0.3)
ax1.text(np.pi, 0.5, 'π', fontsize=12, ha='center')
ax1.set_title('Trigonometric Functions', fontsize=14, fontweight='bold')
ax1.legend()
ax1.grid(True, alpha=0.3)
# 绘制第二个子图：散点图
np.random.seed(42)
x_scatter = np.random.normal(0, 1, 200)
y_scatter = np.random.normal(0, 1, 200)
colors = np.random.rand(200)
sizes = 100 * np.random.rand(200)
scatter = ax2.scatter(x_scatter, y_scatter, c=colors, s=sizes,
alpha=0.6, cmap='viridis')
ax2.set_title('Scatter Plot with Colorbar', fontsize=14, fontweight='bold')
plt.colorbar(scatter, ax=ax2, label='Color Value')
# 绘制第三个子图：柱状图与误差线
categories = ['A', 'B', 'C', 'D', 'E']
values = [25, 40, 30, 35, 28]
errors = [2, 3, 2.5, 2, 3]
bars = ax3.bar(categories, values, yerr=errors, capsize=5,
color='skyblue', edgecolor='navy', linewidth=1.5)
ax3.set_title('Bar Chart with Error Bars', fontsize=14, fontweight='bold')
ax3.set_ylabel('Value', fontsize=12)
# 在柱子上添加数值标签
for bar, value in zip(bars, values):
height = bar.get_height()
ax3.text(bar.get_x() + bar.get_width()/2., height + 1,
f'{value}', ha='center', va='bottom', fontsize=10)
plt.tight_layout()
plt.show()
2.2 交互式可视化：Plotly高级应用
import plotly.graph_objects as go
from plotly.subplots import make_subplots
import plotly.express as px
# 创建复杂的交互式图表
def create_interactive_dashboard():
# 准备数据
np.random.seed(42)
dates = pd.date_range('2023-01-01', periods=100, freq='D')
stock_data = pd.DataFrame({
'Date': dates,
'Stock_A': 100 + np.cumsum(np.random.randn(100) * 0.5),
'Stock_B': 150 + np.cumsum(np.random.randn(100) * 0.8),
'Stock_C': 80 + np.cumsum(np.random.randn(100) * 0.3)
})
# 创建子图
fig = make_subplots(
rows=2, cols=2,
subplot_titles=('Stock A Performance', 'Stock B Performance',
'Stock C Performance', 'Volume Comparison'),
specs=[[{"secondary_y": False}, {"secondary_y": False}],
[{"secondary_y": False}, {"type": "bar"}]]
)
# 添加折线图
fig.add_trace(
go.Scatter(x=stock_data['Date'], y=stock_data['Stock_A'],
mode='lines+markers', name='Stock A',
line=dict(color='blue', width=2),
marker=dict(size=4)),
row=1, col=1
)
fig.add_trace(
go.Scatter(x=stock_data['Date'], y=stock_data['Stock_B'],
mode='lines', name='Stock B',
line=dict(color='green', width=2)),
row=1, col=2
)
fig.add_trace(
go.Scatter(x=stock_data['Date'], y=stock_data['Stock_C'],
mode='lines', name='Stock C',
line=dict(color='red', width=2)),
row=2, col=1
)
# 添加柱状图
volumes = np.random.randint(1000, 5000, 100)
fig.add_trace(
go.Bar(x=stock_data['Date'], y=volumes,
name='Volume', marker_color='purple'),
row=2, col=2
)
# 更新布局
fig.update_layout(
height=800,
title_text="Interactive Stock Dashboard",
showlegend=True,
hovermode='x unified'
)
# 添加范围滑块
fig.update_xaxes(rangeslider_visible=True)
return fig
# 创建并显示仪表板
dashboard = create_interactive_dashboard()
dashboard.show()
2.3 Seaborn统计可视化高级技巧
import seaborn as sns
from scipy import stats
# 设置风格
sns.set_style("whitegrid")
# 创建复杂的统计可视化
def create_statistical_plots():
# 生成示例数据
np.random.seed(42)
data = pd.DataFrame({
'group': np.repeat(['A', 'B', 'C', 'D'], 100),
'value': np.concatenate([
np.random.normal(0, 1, 100),
np.random.normal(1, 1.5, 100),
np.random.normal(2, 2, 100),
np.random.normal(0.5, 1, 100)
]),
'category': np.random.choice(['X', 'Y'], 400)
})
# 创建复杂图表布局
fig, axes = plt.subplots(2, 2, figsize=(16, 12))
# 1. 小提琴图 + 散点图
sns.violinplot(data=data, x='group', y='value', ax=axes[0,0],
palette="Set2", inner="quartile")
sns.stripplot(data=data, x='group', y='value', ax=axes[0,0],
color='black', alpha=0.3, size=3)
axes[0,0].set_title('Violin Plot with Strip Overlay')
# 2. 箱线图 + 箱线图
sns.boxplot(data=data, x='group', y='value', hue='category',
ax=axes[0,1], palette="Set3")
axes[0,1].set_title('Box Plot by Group and Category')
# 3. 联合分布图
sns.jointplot(data=data, x='value', y='group', kind='reg',
height=6, ax=axes[1,0])
axes[1,0].set_title('Joint Distribution')
# 4. 热力图（相关性矩阵）
corr_data = np.random.randn(100, 5)
corr_matrix = np.corrcoef(corr_data.T)
sns.heatmap(corr_matrix, annot=True, cmap='coolwarm', center=0,
ax=axes[1,1], xticklabels=['Var1', 'Var2', 'Var3', 'Var4', 'Var5'],
yticklabels=['Var1', 'Var2', '3', 'Var4', 'Var5'])
axes[1,1].set_title('Correlation Heatmap')
plt.tight_layout()
plt.show()
create_statistical_plots()
第三部分：时间序列分析进阶
3.1 时间序列数据处理与重采样
# 创建复杂的时间序列数据
def create_time_series_data():
np.random.seed(42)
dates = pd.date_range('2023-01-01', '2023-12-31', freq='H')
# 创建基础趋势
trend = np.linspace(100, 150, len(dates))
# 添加季节性成分
seasonal = 10 * np.sin(2 * np.pi * np.arange(len(dates)) / 24)
# 添加周期性成分（每周模式）
weekly = 5 * np.sin(2 * np.pi * np.arange(len(dates)) / (24 * 7))
# 添加随机噪声
noise = np.random.normal(0, 2, len(dates))
# 组合成分
values = trend + seasonal + weekly + noise
return pd.DataFrame({
'timestamp': dates,
'value': values,
'hour': dates.hour,
'day_of_week': dates.dayofweek,
'month': dates.month
})
ts_data = create_time_series_data()
# 高级重采样操作
def advanced_resampling(df):
# 1. 小时到日均值（包含多种统计量）
daily_stats = df.resample('D', on='timestamp').agg({
'value': ['mean', 'std', 'min', 'max', 'count']
})
# 2. 小时到周均值，保留周几的信息
df['week_number'] = df['timestamp'].dt.isocalendar().week
df['year'] = df['timestamp'].dt.year
weekly_pattern = df.groupby(['year', 'week_number', 'hour']).agg({
'value': 'mean'
}).reset_index()
# 3. 计算移动窗口统计量
df['rolling_mean_24h'] = df['value'].rolling(window=24).mean()
df['rolling_std_24h'] = df['value'].rolling(window=24).std()
df['rolling_min_24h'] = df['value'].rolling(window=24).min()
df['rolling_max_24h'] = df['value'].rolling(window=24).max()
# 4. 指数加权移动平均
df['ewm_24h'] = df['value'].ewm(span=24).mean()
# 5. 计算变化率
df['pct_change'] = df['value'].pct_change()
df['diff'] = df['value'].diff()
return df, daily_stats, weekly_pattern
ts_advanced, daily_stats, weekly_pattern = advanced_resampling(ts_data.copy())
print("时间序列数据高级处理结果:")
print(ts_advanced.head(10))
print("\n日均统计:")
print(daily_stats.head())
3.2 时间序列分解与预测
from statsmodels.tsa.seasonal import seasonal_decompose
from statsmodels.tsa.arima.model import ARIMA
import warnings
warnings.filterwarnings('ignore')
def time_series_decomposition_forecast():
# 使用更短的时间序列用于演示
dates = pd.date_range('2023-01-01', periods=365, freq='D')
np.random.seed(42)
# 创建时间序列
trend = np.linspace(100, 150, 365)
seasonal = 10 * np.sin(2 * np.pi * np.arange(365) / 30)
noise = np.random.normal(0, 3, 365)
series = trend + seasonal + noise
ts = pd.Series(series, index=dates)
# 时间序列分解
decomposition = seasonal_decompose(ts, model='additive', period=30)
# 绘制分解结果
fig, axes = plt.subplots(4, 1, figsize=(12, 10))
decomposition.observed.plot(ax=axes[0], legend=False)
axes[0].set_title('Observed')
decomposition.trend.plot(ax=axes[1], legend=False)
axes[1].set_title('Trend')
decomposition.seasonal.plot(ax=axes[2], legend=False)
axes[2].set_title('Seasonal')
decomposition.resid.plot(ax=axes[3], legend=False)
axes[3].set_title('Residual')
plt.tight_layout()
plt.show()
# ARIMA预测
# 简单示例：使用最后100个数据点进行预测
train_data = ts[-100:]
# 自动选择ARIMA参数（简化版）
best_aic = np.inf
best_order = None
for p in range(3):
for d in range(2):
for q in range(3):
try:
model = ARIMA(train_data, order=(p, d, q))
model_fit = model.fit()
if model_fit.aic < best_aic:
best_aic = model_fit.aic
best_order = (p, d, q)
except:
continue
print(f"Best ARIMA order: {best_order} with AIC: {best_aic:.2f}")
# 使用最佳参数拟合模型
if best_order:
model = ARIMA(train_data, order=best_order)
model_fit = model.fit()
# 预测未来30天
forecast = model_fit.forecast(steps=30)
forecast_index = pd.date_range(train_data.index[-1] + pd.Timedelta(days=1), periods=30, freq='D')
forecast_series = pd.Series(forecast, index=forecast_index)
# 绘制预测结果
plt.figure(figsize=(12, 6))
plt.plot(train_data.index, train_data.values, label='Historical Data', color='blue')
plt.plot(forecast_series.index, forecast_series.values, label='Forecast', color='red', linestyle='--')
plt.title('ARIMA Time Series Forecast')
plt.xlabel('Date')
plt.ylabel('Value')
plt.legend()
plt.grid(True, alpha=0.3)
plt.show()
return model_fit, forecast_series
return None, None
model, forecast = time_series_decomposition_forecast()
第四部分：大数据处理与性能优化
4.1 使用Dask处理超出内存的数据
# 注意：Dask需要单独安装：pip install dask[complete]
import dask.dataframe as dd
import os
def dask_large_data_processing():
# 创建一个大型CSV文件用于演示
def create_large_csv(filename, rows=5000000):
if not os.path.exists(filename):
print(f"Creating large CSV file with {rows} rows...")
chunk_size = 100000
for i in range(0, rows, chunk_size):
chunk = pd.DataFrame({
'id': range(i, min(i + chunk_size, rows)),
'category': np.random.choice(['A', 'B', 'C', 'D', 'E'], min(chunk_size, rows - i)),
'value': np.random.randn(min(chunk_size, rows - i)),
'timestamp': pd.date_range('2020-01-01', periods=min(chunk_size, rows - i), freq='T')
})
chunk.to_csv(filename, mode='a', header=i==0, index=False)
print("CSV file created.")
filename = 'large_dataset.csv'
create_large_csv(filename, rows=5000000)
# 使用Dask读取和处理大数据
print("Loading data with Dask...")
ddf = dd.read_csv(filename, parse_dates=['timestamp'])
# Dask的延迟计算特性
# 这些操作不会立即执行，只是构建计算图
result = ddf.groupby('category').agg({
'value': ['mean', 'sum', 'count']
})
# 执行计算并获取结果
print("Computing results...")
computed_result = result.compute()
print("Dask computation result:")
print(computed_result)
# 复杂操作：按时间窗口聚合
ddf['hour'] = ddf['timestamp'].dt.hour
hourly_stats = ddf.groupby(['category', 'hour']).value.mean().compute()
print("\nHourly statistics by category:")
print(hourly_stats.head(10))
# 清理大文件
# os.remove(filename)
return computed_result, hourly_stats
# dask_result, hourly_stats = dask_large_data_processing()
# print("Dask processing completed.")
4.2 并行处理与多进程优化
import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import time
def parallel_processing_demo():
# 模拟一个耗时的计算任务
def heavy_computation(data_chunk):
# 模拟复杂计算
result = np.linalg.eigvals(np.random.rand(100, 100))
return np.sum(result) + data_chunk
# 单进程版本
def single_process(data):
results = []
for item in data:
results.append(heavy_computation(item))
return results
# 多进程版本
def multi_process(data, n_workers=4):
with ProcessPoolExecutor(max_workers=n_workers) as executor:
results = list(executor.map(heavy_computation, data))
return results
# 多线程版本（适用于I/O密集型任务）
def multi_thread(data, n_workers=4):
with ThreadPoolExecutor(max_workers=n_workers) as executor:
results = list(executor.map(heavy_computation, data))
return results
# 生成测试数据
test_data = list(range(100))
# 性能比较
print("性能测试开始...")
start = time.time()
result_single = single_process(test_data)
time_single = time.time() - start
print(f"单进程耗时: {time_single:.2f}秒")
start = time.time()
result_multi = multi_process(test_data, n_workers=4)
time_multi = time.time() - start
print(f"多进程(4核)耗时: {time_multi:.2f}秒")
start = time.time()
result_thread = multi_thread(test_data, n_workers=4)
time_thread = time.time() - start
print(f"多线程(4线程)耗时: {时间_thread:.2f}秒")
print(f"\n加速比: {time_single/time_multi:.2f}x")
return {
'single': time_single,
'multi': time_multi,
'thread': time_thread
}
# parallel_results = parallel_processing_demo()
第五部分：实战项目 - 构建完整的数据分析管道
5.1 项目1：电商销售分析仪表板
def ecommerce_analysis_project():
"""
完整的电商销售分析项目
包括数据加载、清洗、分析、可视化和报告生成
"""
# 1. 数据生成与加载
np.random.seed(42)
n_rows = 10000
ecommerce_data = pd.DataFrame({
'order_id': range(1, n_rows + 1),
'customer_id': np.random.randint(1, 2000, n_rows),
'product_id': np.random.randint(1, 100, n_rows),
'category': np.random.choice(['Electronics', 'Clothing', 'Books', 'Home', 'Sports'], n_rows),
'price': np.random.uniform(10, 500, n_rows).round(2),
'quantity': np.random.randint(1, 10, n_rows),
'order_date': pd.date_range('2023-01-01', periods=n_rows, freq='H'),
'region': np.random.choice(['North', 'South', 'East', 'West'], n_rows),
'customer_type': np.random.choice(['New', 'Returning'], n_rows, p=[0.3, 0.7])
})
# 计算总价
ecommerce_data['total_price'] = ecommerce_data['price'] * ecommerce_data['quantity']
# 2. 数据清洗
print("原始数据形状:", ecommerce_data.shape)
print("缺失值检查:")
print(ecommerce_data.isnull().sum())
# 3. 核心分析
# 3.1 销售趋势分析
monthly_sales = ecommerce_data.groupby(
ecommerce_data['order_date'].dt.to_period('M')
)['total_price'].sum()
# 3.2 品类分析
category_analysis = ecommerce_data.groupby('category').agg({
'total_price': ['sum', 'mean', 'count'],
'quantity': 'sum'
}).round(2)
# 3.3 区域分析
region_analysis = ecommerce_data.groupby('region').agg({
'total_price': 'sum',
'order_id': 'count'
})
region_analysis['avg_order_value'] = region_analysis['total_price'] / region_analysis['order_id']
# 3.4 客户价值分析（RFM分析）
# 最近购买时间（Recency）
max_date = ecommerce_data['order_date'].max()
recency = ecommerce_data.groupby('customer_id')['order_date'].max()
recency = (max_date - recency).dt.days
# 购买频率（Frequency）
frequency = ecommerce_data.groupby('customer_id')['order_id'].count()
# 消费金额（Monetary）
monetary = ecommerce_data.groupby('customer_id')['total_price'].sum()
rfm = pd.DataFrame({
'recency': recency,
'frequency': frequency,
'monetary': monetary
})
# RFM评分
rfm['R_score'] = pd.qcut(rfm['recency'], 4, labels=[4, 3, 2, 1]) # 越小越好
rfm['F_score'] = pd.qcut(rfm['frequency'].rank(method='first'), 4, labels=[1, 2, 3, 4])
rfm['M_score'] = pd.qcut(rfm['monetary'], 4, labels=[1, 2, 3, 4])
rfm['RFM_score'] = rfm['R_score'].astype(str) + rfm['F_score'].astype(str) + rfm['M_score'].astype(str)
# 4. 可视化
fig = plt.figure(figsize=(20, 15))
gs = fig.add_gridspec(3, 3)
# 4.1 月度销售趋势
ax1 = fig.add_subplot(gs[0, :2])
monthly_sales.plot(ax=ax1, marker='o', linewidth=2, markersize=6)
ax1.set_title('Monthly Sales Trend', fontsize=14, fontweight='bold')
ax1.set_ylabel('Total Sales ($)')
ax1.grid(True, alpha=0.3)
# 4.2 品类销售占比
ax2 = fig.add_subplot(gs[0, 2])
category_sum = category_analysis[('total_price', 'sum')]
ax2.pie(category_sum.values, labels=category_sum.index, autopct='%1.1f%%')
ax2.set_title('Sales by Category')
# 4.3 区域销售对比
ax3 = fig.add_subplot(gs[1, 0])
region_analysis['total_price'].plot(ax=ax3, kind='bar', color='skyblue')
ax3.set_title('Sales by Region')
ax3.set_ylabel('Total Sales ($)')
# 4.4 客户类型分布
ax4 = fig.add_subplot(gs[1, 1])
customer_type_dist = ecommerce_data['customer_type'].value_counts()
ax4.pie(customer_type_dist.values, labels=customer_type_dist.index,
autopct='%1.1f%%', colors=['lightcoral', 'lightgreen'])
ax4.set_title('Customer Type Distribution')
# 4.5 RFM分布
ax5 = fig.add_subplot(gs[1, 2])
rfm['RFM_score'].value_counts().head(10).plot(ax=ax5, kind='bar')
ax5.set_title('Top 10 RFM Segments')
ax5.tick_params(axis='x', rotation=45)
# 4.6 价格与数量的关系
ax6 = fig.add_subplot(gs[2, :])
scatter = ax6.scatter(ecommerce_data['price'], ecommerce_data['quantity'],
c=ecommerce_data['total_price'], cmap='viridis', alpha=0.6)
ax6.set_xlabel('Price ($)')
ax6.set_ylabel('Quantity')
ax6.set_title('Price vs Quantity (color = Total Price)')
plt.colorbar(scatter, ax=ax6, label='Total Price')
plt.tight_layout()
plt.show()
# 5. 生成分析报告
report = {
'total_revenue': ecommerce_data['total_price'].sum(),
'total_orders': len(ecommerce_data),
'avg_order_value': ecommerce_data['total_price'].mean(),
'top_category': category_analysis[('total_price', 'sum')].idxmax(),
'best_region': region_analysis['total_price'].idxmax(),
'unique_customers': ecommerce_data['customer_id'].nunique(),
'top_rfm_segment': rfm['RFM_score'].value_counts().index[0]
}
print("\n=== 电商销售分析报告 ===")
for key, value in report.items():
print(f"{key.replace('_', ' ').title()}: {value:,.2f}" if isinstance(value, (int, float)) else f"{key.replace('_', ' ').title()}: {value}")
return ecommerce_data, report
# 执行项目
ecommerce_data, analysis_report = ecommerce_analysis_project()
5.2 项目2：时间序列预测与异常检测
def time_series_anomaly_detection():
"""
时间序列异常检测项目
包括趋势分析、季节性分解、异常点识别
"""
# 1. 创建带有异常的时间序列数据
np.random.seed(42)
dates = pd.date_range('2023-01-01', periods=500, freq='D')
# 基础趋势 + 季节性 + 噪声
trend = np.linspace(100, 200, 500)
seasonal = 20 * np.sin(2 * np.pi * np.arange(500) / 50)
noise = np.random.normal(0, 5, 500)
# 添加异常点
ts_values = trend + seasonal + noise
anomaly_indices = [50, 150, 250, 350, 450]
ts_values[anomaly_indices] += [50, -60, 80, -40, 70] # 添加异常值
ts_data = pd.DataFrame({
'date': dates,
'value': ts_values
}).set_index('date')
# 2. 异常检测算法
def detect_anomalies_zscore(data, threshold=3):
"""基于Z-score的异常检测"""
mean = data.mean()
std = data.std()
z_scores = (data - mean) / std
anomalies = np.abs(z_scores) > threshold
return anomalies, z_scores
def detect_anomalies_iqr(data):
"""基于IQR的异常检测"""
Q1 = data.quantile(0.25)
Q3 = data.quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR
anomalies = (data < lower_bound) | (data > upper_bound)
return anomalies
def detect_anomalies_rolling(data, window=30, threshold=3):
"""基于滚动统计的异常检测"""
rolling_mean = data.rolling(window=window).mean()
rolling_std = data.rolling(window=window).std()
# 避免窗口期的NaN值
valid_idx = ~rolling_mean.isna()
z_scores = (data[valid_idx] - rolling_mean[valid_idx]) / rolling_std[valid_idx]
anomalies = np.abs(z_scores) > threshold
return anomalies, rolling_mean, rolling_std
# 3. 应用检测算法
anomalies_z, z_scores = detect_anomalies_zscore(ts_data['value'])
anomalies_iqr = detect_anomalies_iqr(ts_data['value'])
anomalies_rolling, rolling_mean, rolling_std = detect_anomalies_rolling(ts_data['value'])
# 4. 可视化结果
fig, axes = plt.subplots(2, 2, figsize=(16, 10))
# 原始数据与Z-score检测
ax1 = axes[0, 0]
ax1.plot(ts_data.index, ts_data['value'], 'b-', label='Original', alpha=0.7)
ax1.scatter(ts_data.index[anomalies_z], ts_data['value'][anomalies_z],
color='red', s=50, label='Anomalies (Z-score)')
ax1.set_title('Anomaly Detection: Z-Score Method')
ax1.legend()
ax1.grid(True, alpha=0.3)
# 原始数据与IQR检测
ax2 = axes[0, 1]
ax2.plot(ts_data.index, ts_data['value'], 'b-', label='Original', alpha=0.7)
ax2.scatter(ts_data.index[anomalies_iqr], ts_data['value'][anomalies_iqr],
color='orange', s=50, label='Anomalies (IQR)')
ax2.set_title('Anomaly Detection: IQR Method')
ax2.legend()
ax2.grid(True, alpha=0.3)
# 滚动窗口检测
ax3 = axes[1, 0]
ax3.plot(ts_data.index, ts_data['value'], 'b-', label='Original', alpha=0.5)
ax3.plot(rolling_mean.index, rolling_mean, 'g--', label='Rolling Mean')
ax3.fill_between(rolling_mean.index,
rolling_mean - 3 * rolling_std,
rolling_mean + 3 * rolling_std,
alpha=0.2, color='gray', label='±3σ Range')
ax3.scatter(ts_data.index[anomalies_rolling], ts_data['value'][anomalies_rolling],
color='red', s=50, label='Anomalies')
ax3.set_title('Anomaly Detection: Rolling Window Method')
ax3.legend()
ax3.grid(True, alpha=0.3)
# Z-score分布
ax4 = axes[1, 1]
ax4.hist(z_scores, bins=30, alpha=0.7, color='purple')
ax4.axvline(x=3, color='red', linestyle='--', label='Threshold = 3')
ax4.axvline(x=-3, color='red', linestyle='--')
ax4.set_title('Z-Score Distribution')
ax4.legend()
ax4.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
# 5. 生成异常报告
anomaly_report = pd.DataFrame({
'Date': ts_data.index[anomalies_z],
'Value': ts_data['value'][anomalies_z].values,
'Z_Score': z_scores[anomalies_z].values,
'Severity': ['High' if abs(z) > 4 else 'Medium' for z in z_scores[anomalies_z]]
})
print("\n=== 异常检测报告 ===")
print(f"检测到的异常点数量: {len(anomaly_report)}")
print("\n异常点详情:")
print(anomaly_report.to_string(index=False))
return ts_data, anomaly_report
# 执行时间序列异常检测项目
ts_data, anomaly_report = time_series_anomaly_detection()
第六部分：高级技巧与最佳实践
6.1 内存优化技巧
def advanced_memory_optimization():
"""
深入的内存优化技巧
"""
# 1. 使用category类型优化分类数据
df = pd.DataFrame({
'id': range(1000000),
'country': np.random.choice(['USA', 'China', 'India', 'Germany', 'Brazil'], 1000000),
'city': np.random.choice(['New York', 'Beijing', 'Mumbai', 'Berlin', 'São Paulo'], 1000000),
'status': np.random.choice(['Active', 'Inactive', 'Pending'], 1000000),
'value': np.random.randn(1000000)
})
print("原始内存使用:")
print(df.info(memory_usage='deep'))
# 优化前
df_original = df.copy()
# 优化后
df_optimized = df.copy()
df_optimized['country'] = df_optimized['country'].astype('category')
df_optimized['city'] = df_optimized['city'].astype('category')
df_optimized['status'] = df_optimized['status'].astype('category')
print("\n优化后内存使用:")
print(df_optimized.info(memory_usage='deep'))
# 2. 使用稀疏矩阵处理大量零值
from scipy import sparse
# 创建稀疏矩阵
dense_matrix = np.random.choice([0, 1, 2], size=(1000, 1000), p=[0.9, 0.05, 0.05])
sparse_matrix = sparse.csr_matrix(dense_matrix)
print(f"\n稀疏矩阵内存节省: {dense_matrix.nbytes / sparse_matrix.data.nbytes:.1f}x")
# 3. 分块处理大数据
def process_large_file_in_chunks(filename, chunk_size=100000):
results = []
for chunk in pd.read_csv(filename, chunksize=chunk_size):
# 处理每个chunk
processed = chunk.groupby('category')['value'].sum()
results.append(processed)
# 合并结果
final_result = pd.concat(results).groupby(level=0).sum()
return final_result
return df_optimized, sparse_matrix
# advanced_memory_optimization()
6.2 代码性能分析与优化
import cProfile
import pstats
from functools import wraps
def performance_analysis_demo():
"""
性能分析与优化演示
"""
# 装饰器：用于测量函数执行时间
def timer(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.time()
result = func(*args, **kwargs)
end = time.time()
print(f"{func.__name__} executed in {end - start:.4f} seconds")
return result
return wrapper
# 慢速版本
@timer
def slow_version():
result = []
for i in range(100000):
result.append(i ** 2)
return result
# 快速版本（向量化）
@timer
def fast_version():
return [i ** 2 for i in range(100000)]
# 最快版本（numpy向量化）
@timer
def fastest_version():
return np.arange(100000) ** 2
# 比较性能
print("性能比较测试:")
slow_result = slow_version()
fast_result = fast_version()
fastest_result = fastest_version()
# 使用cProfile进行详细分析
print("\n详细性能分析 (cProfile):")
profiler = cProfile.Profile()
profiler.enable()
# 执行需要分析的代码
for _ in range(1000):
np.random.randn(100, 100).sum()
profiler.disable()
stats = pstats.Stats(profiler)
stats.sort_stats('cumulative')
stats.print_stats(10) # 显示前10个最耗时的函数
return fastest_result
# performance_analysis_demo()
结论
通过本课程的学习，你已经掌握了Python数据分析的核心进阶技巧。从Pandas高级操作到复杂可视化，从时间序列分析到大数据处理，再到实战项目，这些技能将帮助你在实际工作中更高效地解决问题。
记住以下关键要点：
- 内存优化：始终关注数据类型的优化，使用category类型处理分类数据
- 向量化操作：避免使用循环，尽可能使用NumPy和Pandas的向量化函数
- 分块处理：对于超出内存的数据，使用分块读取或Dask
- 并行计算：合理使用多进程和多线程提升计算效率
- 可视化：选择合适的图表类型，清晰传达数据洞察
- 项目实践：将技能应用到实际项目中，不断积累经验
持续学习和实践是掌握数据分析的关键。建议你：
- 定期阅读Pandas和NumPy的官方文档
- 参与开源数据分析项目
- 在Kaggle等平台练习真实数据集
- 关注数据科学领域的最新发展
祝你在数据分析的道路上越走越远！