Python数据分析进阶课程从入门到精通实战案例解析与行业应用指南

引言：为什么需要从入门到精通的Python数据分析？

在当今数据驱动的时代，Python已成为数据分析领域的首选语言。根据2023年Kaggle调查，Python在数据科学工具中使用率高达83.7%，远超R语言的42.1%。然而，许多初学者在掌握了基础语法和Pandas基础操作后，常常陷入“瓶颈期”——知道如何读取数据、做简单统计，却无法应对真实业务场景中的复杂分析需求。

本指南将系统性地带你从Python数据分析的入门阶段跨越到精通阶段，通过完整的实战案例解析，让你掌握处理真实世界数据的完整流程，并了解在不同行业的应用场景。我们将重点解决以下问题：

• 如何处理大规模、不规则的真实数据？
• 如何进行高级数据清洗和特征工程？
• 如何运用统计学和机器学习方法进行深度分析？
• 如何将分析结果转化为商业洞察？

第一部分：Python数据分析基础回顾与进阶准备

1.1 核心库的深度理解

在进阶之前，我们需要确保对基础库有扎实的理解。以下是三个核心库的进阶用法：

Pandas：超越基础操作

import pandas as pd
import numpy as np

# 基础操作回顾
df = pd.DataFrame({
    '产品ID': [1001, 1002, 1003, 1004],
    '销售额': [2500, 3200, 1800, 4100],
    '地区': ['华东', '华北', '华南', '华东'],
    '日期': pd.date_range('2023-01-01', periods=4)
})

# 进阶操作1：多级索引（MultiIndex）
# 创建多级索引DataFrame
arrays = [['华东', '华东', '华北', '华北'], 
          ['上海', '杭州', '北京', '天津']]
tuples = list(zip(*arrays))
index = pd.MultiIndex.from_tuples(tuples, names=['地区', '城市'])
df_multi = pd.DataFrame({
    '销售额': [2500, 1800, 3200, 2800],
    '利润': [500, 360, 640, 560]
}, index=index)

# 进阶操作2：分组聚合的高级用法
# 使用agg函数进行多维度聚合
sales_data = pd.DataFrame({
    '产品': ['A', 'B', 'A', 'C', 'B', 'C'],
    '销售额': [100, 150, 200, 80, 120, 90],
    '利润': [20, 30, 40, 15, 25, 18],
    '地区': ['华东', '华北', '华东', '华南', '华北', '华南']
})

result = sales_data.groupby('产品').agg({
    '销售额': ['sum', 'mean', 'max'],
    '利润': ['sum', 'mean']
})
print(result)

# 进阶操作3：时间序列处理
# 生成时间序列数据
dates = pd.date_range('2023-01-01', periods=100, freq='D')
ts_data = pd.DataFrame({
    'date': dates,
    'value': np.random.randn(100).cumsum() + 100
})

# 重采样（Resampling）- 按周求平均
weekly_avg = ts_data.set_index('date').resample('W').mean()
print(weekly_avg.head())

NumPy：高性能数值计算

# NumPy广播机制详解
import numpy as np

# 基础广播示例
a = np.array([1, 2, 3])
b = np.array([4, 5, 6])
print(a + b)  # [5 7 9]

# 高级广播：不同维度数组运算
matrix = np.array([[1, 2, 3], [4, 5, 6], [7, 8, 9]])
vector = np.array([10, 20, 30])
result = matrix + vector  # 每行都加上vector
print(result)
# [[11 22 33]
#  [14 25 36]
#  [17 28 39]]

# 向量化操作 vs 循环性能对比
import time

# 循环方式
def loop_sum(arr):
    total = 0
    for i in range(len(arr)):
        total += arr[i]
    return total

# 向量化方式
def vectorized_sum(arr):
    return np.sum(arr)

# 性能测试
large_array = np.random.rand(1000000)

start = time.time()
loop_result = loop_sum(large_array)
loop_time = time.time() - start

start = time.time()
vectorized_result = vectorized_sum(large_array)
vectorized_time = time.time() - start

print(f"循环方式耗时: {loop_time:.4f}秒")
print(f"向量化方式耗时: {vectorized_time:.4f}秒")
print(f"性能提升: {loop_time/vectorized_time:.1f}倍")

Matplotlib/Seaborn：高级可视化

import matplotlib.pyplot as plt
import seaborn as sns

# 设置中文字体（Windows系统）
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

# 创建复杂可视化
fig, axes = plt.subplots(2, 2, figsize=(12, 10))

# 1. 热力图
corr_matrix = np.random.rand(5, 5)
sns.heatmap(corr_matrix, annot=True, cmap='coolwarm', ax=axes[0, 0])
axes[0, 0].set_title('相关性热力图')

# 2. 分布图
data = np.random.normal(0, 1, 1000)
sns.histplot(data, kde=True, ax=axes[0, 1])
axes[0, 1].set_title('正态分布直方图')

# 3. 散点图矩阵
iris = sns.load_dataset('iris')
sns.scatterplot(data=iris, x='sepal_length', y='sepal_width', 
                hue='species', ax=axes[1, 0])
axes[1, 0].set_title('鸢尾花数据散点图')

# 4. 箱线图
sns.boxplot(data=iris, x='species', y='petal_length', ax=axes[1, 1])
axes[1, 1].set_title('鸢尾花花瓣长度箱线图')

plt.tight_layout()
plt.show()

1.2 环境配置与工具链

虚拟环境管理

# 使用conda创建环境
conda create -n data_analysis python=3.9
conda activate data_analysis

# 安装核心库
conda install pandas numpy matplotlib seaborn scikit-learn
pip install jupyterlab  # 推荐使用Jupyter Lab

# 导出环境配置
conda env export > environment.yml

Jupyter Notebook进阶技巧

# 在Jupyter中使用魔法命令
%load_ext autoreload
%autoreload 2  # 自动重新加载模块

# 显示所有输出
from IPython.core.interactiveshell import InteractiveShell
InteractiveShell.ast_node_interactivity = "all"

# 进度条显示
from tqdm import tqdm
import time

for i in tqdm(range(100)):
    time.sleep(0.01)

第二部分：数据清洗与预处理的高级技巧

2.1 处理缺失值的多种策略

import pandas as pd
import numpy as np
from sklearn.impute import KNNImputer, IterativeImputer

# 创建包含缺失值的数据集
np.random.seed(42)
data = pd.DataFrame({
    '年龄': [25, np.nan, 30, 35, np.nan, 40, 28, 32],
    '收入': [50000, 60000, np.nan, 70000, 80000, np.nan, 55000, 65000],
    '教育程度': ['本科', '硕士', '本科', '博士', '硕士', '本科', np.nan, '硕士'],
    '地区': ['北京', '上海', '广州', '深圳', '北京', '上海', '广州', '深圳']
})

print("原始数据：")
print(data)
print("\n缺失值统计：")
print(data.isnull().sum())

# 策略1：简单填充
# 均值填充
data_mean = data.copy()
data_mean['年龄'] = data_mean['年龄'].fillna(data_mean['年龄'].mean())
data_mean['收入'] = data_mean['收入'].fillna(data_mean['收入'].mean())

# 众数填充
data_mode = data.copy()
data_mode['教育程度'] = data_mode['教育程度'].fillna(data_mode['教育程度'].mode()[0])

# 策略2：基于模型的填充（KNN）
# 注意：KNN需要数值型数据
data_numeric = data[['年龄', '收入']].copy()
knn_imputer = KNNImputer(n_neighbors=2)
data_knn = pd.DataFrame(knn_imputer.fit_transform(data_numeric), 
                        columns=data_numeric.columns)

# 策略3：多重插补（MICE）
mice_imputer = IterativeImputer(random_state=42)
data_mice = pd.DataFrame(mice_imputer.fit_transform(data_numeric), 
                         columns=data_numeric.columns)

# 策略4：前向/后向填充（时间序列数据）
time_data = pd.DataFrame({
    '时间': pd.date_range('2023-01-01', periods=10, freq='D'),
    '温度': [20, 21, np.nan, np.nan, 23, 24, np.nan, 26, 27, 28]
})
time_data['温度_ffill'] = time_data['温度'].fillna(method='ffill')
time_data['温度_bfill'] = time_data['温度'].fillna(method='bfill')
time_data['温度_interpolate'] = time_data['温度'].interpolate(method='linear')

print("\n时间序列填充对比：")
print(time_data)

2.2 异常值检测与处理

# 异常值检测方法
from scipy import stats

# 创建包含异常值的数据
np.random.seed(42)
normal_data = np.random.normal(0, 1, 100)
outliers = np.array([10, -8, 15, -12])
data_with_outliers = np.concatenate([normal_data, outliers])

# 方法1：Z-score方法
z_scores = np.abs(stats.zscore(data_with_outliers))
threshold = 3
outliers_z = data_with_outliers[z_scores > threshold]
print(f"Z-score方法检测到的异常值：{outliers_z}")

# 方法2：IQR方法（四分位距）
Q1 = np.percentile(data_with_outliers, 25)
Q3 = np.percentile(data_with_outliers, 75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR
outliers_iqr = data_with_outliers[(data_with_outliers < lower_bound) | 
                                  (data_with_outliers > upper_bound)]
print(f"IQR方法检测到的异常值：{outliers_iqr}")

# 方法3：孤立森林（Isolation Forest）
from sklearn.ensemble import IsolationForest

# 准备数据（需要2D数组）
data_2d = data_with_outliers.reshape(-1, 1)
iso_forest = IsolationForest(contamination=0.1, random_state=42)
outliers_iso = iso_forest.fit_predict(data_2d)
outliers_iso_values = data_with_outliers[outliers_iso == -1]
print(f"孤立森林检测到的异常值：{outliers_iso_values}")

# 异常值处理策略
def handle_outliers(df, column, method='clip'):
    """
    处理异常值的多种方法
    """
    Q1 = df[column].quantile(0.25)
    Q3 = df[column].quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR
    
    if method == 'clip':
        # 截断法：将异常值替换为边界值
        df[column] = df[column].clip(lower_bound, upper_bound)
    elif method == 'remove':
        # 删除法：删除包含异常值的行
        df = df[(df[column] >= lower_bound) & (df[column] <= upper_bound)]
    elif method == 'log_transform':
        # 对数变换：减少异常值影响
        df[column] = np.log1p(df[column])
    
    return df

# 示例：处理销售数据中的异常值
sales_data = pd.DataFrame({
    '销售额': [100, 150, 200, 5000, 120, 180, 250, 8000, 130, 160]
})

print("\n原始销售额：")
print(sales_data['销售额'].values)

# 使用截断法处理
sales_clipped = handle_outliers(sales_data.copy(), '销售额', method='clip')
print("\n截断法处理后：")
print(sales_clipped['销售额'].values)

# 使用对数变换
sales_log = handle_outliers(sales_data.copy(), '销售额', method='log_transform')
print("\n对数变换后：")
print(sales_log['销售额'].values)

2.3 特征工程高级技术

# 特征编码技术
from sklearn.preprocessing import LabelEncoder, OneHotEncoder
from category_encoders import TargetEncoder, CatBoostEncoder

# 创建分类特征数据
data = pd.DataFrame({
    '城市': ['北京', '上海', '广州', '深圳', '北京', '上海'],
    '产品类别': ['电子', '服装', '电子', '食品', '服装', '食品'],
    '销售额': [1000, 1500, 800, 1200, 1100, 900]
})

# 1. 标签编码（Label Encoding）
le = LabelEncoder()
data['城市编码'] = le.fit_transform(data['城市'])
print("标签编码：")
print(data[['城市', '城市编码']].drop_duplicates())

# 2. 独热编码（One-Hot Encoding）
ohe = OneHotEncoder(sparse=False)
city_encoded = ohe.fit_transform(data[['城市']])
city_encoded_df = pd.DataFrame(city_encoded, columns=ohe.get_feature_names_out(['城市']))
print("\n独热编码：")
print(city_encoded_df)

# 3. 目标编码（Target Encoding）- 适用于高基数特征
target_encoder = TargetEncoder(cols=['城市'])
data['城市目标编码'] = target_encoder.fit_transform(data['城市'], data['销售额'])
print("\n目标编码：")
print(data[['城市', '城市目标编码']].drop_duplicates())

# 4. CatBoost编码（避免数据泄露）
catboost_encoder = CatBoostEncoder(cols=['产品类别'])
data['产品类别编码'] = catboost_encoder.fit_transform(data['产品类别'], data['销售额'])
print("\nCatBoost编码：")
print(data[['产品类别', '产品类别编码']].drop_duplicates())

# 特征交互
# 创建交互特征
data['城市_产品'] = data['城市'] + '_' + data['产品类别']
print("\n交互特征：")
print(data[['城市', '产品类别', '城市_产品']])

# 数值特征的多项式特征
from sklearn.preprocessing import PolynomialFeatures

# 创建数值特征
numeric_data = pd.DataFrame({
    '收入': [50000, 60000, 70000, 80000, 90000],
    '年龄': [25, 30, 35, 40, 45]
})

poly = PolynomialFeatures(degree=2, include_bias=False)
poly_features = poly.fit_transform(numeric_data)
poly_columns = [f'收入_{i}' for i in range(1, 3)] + [f'年龄_{i}' for i in range(1, 3)] + ['收入_年龄']
poly_df = pd.DataFrame(poly_features, columns=poly_columns)
print("\n多项式特征：")
print(poly_df)

第三部分：统计分析与假设检验

3.1 描述性统计的深度分析

import pandas as pd
import numpy as np
import scipy.stats as stats

# 创建销售数据集
np.random.seed(42)
sales_data = pd.DataFrame({
    '产品ID': range(1, 101),
    '销售额': np.random.normal(1000, 200, 100),
    '利润': np.random.normal(200, 50, 100),
    '地区': np.random.choice(['华东', '华北', '华南', '华西'], 100),
    '月份': np.random.choice(range(1, 13), 100)
})

# 多维度描述性统计
print("整体描述性统计：")
print(sales_data[['销售额', '利润']].describe())

# 分组描述性统计
grouped_stats = sales_data.groupby('地区')[['销售额', '利润']].agg([
    'mean', 'std', 'min', 'max', 'count'
])
print("\n分地区描述性统计：")
print(grouped_stats)

# 偏度和峰度分析
print("\n偏度和峰度：")
print(f"销售额偏度: {stats.skew(sales_data['销售额']):.3f}")
print(f"销售额峰度: {stats.kurtosis(sales_data['销售额']):.3f}")
print(f"利润偏度: {stats.skew(sales_data['利润']):.3f}")
print(f"利润峰度: {stats.kurtosis(sales_data['利润']):.3f}")

# 分布可视化
import matplotlib.pyplot as plt
import seaborn as sns

fig, axes = plt.subplots(2, 2, figsize=(12, 10))

# 直方图与KDE
sns.histplot(sales_data['销售额'], kde=True, ax=axes[0, 0])
axes[0, 0].set_title('销售额分布')

# Q-Q图
stats.probplot(sales_data['销售额'], dist="norm", plot=axes[0, 1])
axes[0, 1].set_title('销售额Q-Q图')

# 箱线图
sns.boxplot(data=sales_data, x='地区', y='销售额', ax=axes[1, 0])
axes[1, 0].set_title('各地区销售额箱线图')

# 小提琴图
sns.violinplot(data=sales_data, x='地区', y='销售额', ax=axes[1, 1])
axes[1, 1].set_title('各地区销售额小提琴图')

plt.tight_layout()
plt.show()

3.2 假设检验实战

# T检验：比较两组均值差异
# 场景：比较两种营销策略的效果
group_A = np.random.normal(100, 15, 50)  # 策略A
group_B = np.random.normal(110, 15, 50)  # 策略B

# 独立样本T检验
t_stat, p_value = stats.ttest_ind(group_A, group_B)
print(f"独立样本T检验结果：")
print(f"T统计量: {t_stat:.3f}, P值: {p_value:.3f}")
if p_value < 0.05:
    print("结论：两种策略存在显著差异（p<0.05）")
else:
    print("结论：两种策略无显著差异")

# 配对样本T检验（同一组用户在不同时间点的比较）
before = np.random.normal(100, 10, 30)
after = before + np.random.normal(5, 2, 30)  # 假设干预后提升5个单位

t_stat, p_value = stats.ttest_rel(before, after)
print(f"\n配对样本T检验结果：")
print(f"T统计量: {t_stat:.3f}, P值: {p_value:.3f}")

# 方差分析（ANOVA）：比较多个组的均值
# 场景：比较不同地区的销售额
regions = ['华东', '华北', '华南', '华西']
region_data = []
for region in regions:
    region_data.append(np.random.normal(1000, 200, 30))

f_stat, p_value = stats.f_oneway(*region_data)
print(f"\n单因素方差分析结果：")
print(f"F统计量: {f_stat:.3f}, P值: {p_value:.3f}")
if p_value < 0.05:
    print("结论：不同地区销售额存在显著差异")
    
    # 事后检验（Tukey HSD）
    from statsmodels.stats.multicomp import pairwise_tukeyhsd
    df = pd.DataFrame({
        '销售额': np.concatenate(region_data),
        '地区': np.repeat(regions, 30)
    })
    tukey = pairwise_tukeyhsd(df['销售额'], df['地区'], alpha=0.05)
    print("\nTukey HSD事后检验结果：")
    print(tukey)

3.3 相关性分析与因果推断

# 相关性分析
# 创建相关性数据
np.random.seed(42)
n = 200
x1 = np.random.normal(0, 1, n)
x2 = x1 + np.random.normal(0, 0.5, n)  # 与x1相关
x3 = np.random.normal(0, 1, n)  # 独立
y = 2*x1 + 1.5*x2 + 0.5*x3 + np.random.normal(0, 0.5, n)

data_corr = pd.DataFrame({'x1': x1, 'x2': x2, 'x3': x3, 'y': y})

# 皮尔逊相关系数
corr_matrix = data_corr.corr()
print("皮尔逊相关系数矩阵：")
print(corr_matrix)

# 斯皮尔曼秩相关系数（非参数）
spearman_corr = data_corr.corr(method='spearman')
print("\n斯皮尔曼秩相关系数矩阵：")
print(spearman_corr)

# 偏相关分析（控制其他变量的影响）
from pingouin import partial_corr

# 计算x1和y的偏相关（控制x2和x3）
partial_corr_result = partial_corr(data=data_corr, x='x1', y='y', covar=['x2', 'x3'])
print("\n偏相关分析结果：")
print(partial_corr_result)

# 因果推断：双重差分法（DID）
# 场景：评估政策实施效果
# 创建模拟数据
np.random.seed(42)
n = 1000
time = np.repeat([0, 1], n//2)  # 0=政策前，1=政策后
treated = np.concatenate([np.zeros(n//4), np.ones(n//4), 
                          np.zeros(n//4), np.ones(n//4)])  # 0=对照组，1=实验组
# 模拟结果变量（政策使实验组在政策后提升10个单位）
y = 50 + 10*time + 15*treated + 10*time*treated + np.random.normal(0, 5, n)

did_data = pd.DataFrame({'time': time, 'treated': treated, 'y': y})

# 计算DID估计量
# 第一步：计算各组各期的均值
means = did_data.groupby(['treated', 'time'])['y'].mean().unstack()
print("\n各组各期的均值：")
print(means)

# 第二步：计算DID
did_estimate = (means.loc[1, 1] - means.loc[1, 0]) - (means.loc[0, 1] - means.loc[0, 0])
print(f"\n双重差分估计量: {did_estimate:.2f}")
print(f"真实效应: 10.00")
print(f"估计误差: {abs(did_estimate - 10):.2f}")

第四部分：机器学习在数据分析中的应用

4.1 监督学习：预测模型

from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LinearRegression, LogisticRegression
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.metrics import mean_squared_error, r2_score, accuracy_score, classification_report

# 场景1：回归预测 - 房价预测
# 加载数据集
from sklearn.datasets import fetch_california_housing
housing = fetch_california_housing()
X, y = housing.data, housing.target
feature_names = housing.feature_names

# 数据预处理
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)

# 模型训练与评估
models = {
    '线性回归': LinearRegression(),
    '随机森林': RandomForestRegressor(n_estimators=100, random_state=42),
    '梯度提升': GradientBoostingRegressor(n_estimators=100, random_state=42)
}

results = {}
for name, model in models.items():
    model.fit(X_train_scaled, y_train)
    y_pred = model.predict(X_test_scaled)
    mse = mean_squared_error(y_test, y_pred)
    r2 = r2_score(y_test, y_pred)
    results[name] = {'MSE': mse, 'R2': r2}
    print(f"{name}: MSE={mse:.3f}, R2={r2:.3f}")

# 特征重要性分析（以随机森林为例）
rf_model = models['随机森林']
feature_importance = pd.DataFrame({
    '特征': feature_names,
    '重要性': rf_model.feature_importances_
}).sort_values('重要性', ascending=False)

print("\n特征重要性排序：")
print(feature_importance)

# 场景2：分类预测 - 客户流失预测
from sklearn.datasets import make_classification

# 创建模拟数据
X_class, y_class = make_classification(n_samples=1000, n_features=10, 
                                       n_informative=5, n_redundant=2, 
                                       random_state=42)

X_train_c, X_test_c, y_train_c, y_test_c = train_test_split(X_class, y_class, 
                                                            test_size=0.3, random_state=42)

# 逻辑回归
log_reg = LogisticRegression(random_state=42)
log_reg.fit(X_train_c, y_train_c)
y_pred_c = log_reg.predict(X_test_c)

print("\n逻辑回归分类结果：")
print(f"准确率: {accuracy_score(y_test_c, y_pred_c):.3f}")
print("\n分类报告：")
print(classification_report(y_test_c, y_pred_c))

# 交叉验证
cv_scores = cross_val_score(log_reg, X_class, y_class, cv=5, scoring='accuracy')
print(f"\n5折交叉验证准确率: {cv_scores.mean():.3f} (+/- {cv_scores.std():.3f})")

4.2 无监督学习：聚类与降维

from sklearn.cluster import KMeans, DBSCAN
from sklearn.decomposition import PCA
from sklearn.preprocessing import StandardScaler

# 场景：客户细分
# 创建模拟客户数据
np.random.seed(42)
n_samples = 500
# 三个不同的客户群体
cluster1 = np.random.normal(0, 1, (n_samples//3, 2))
cluster2 = np.random.normal(5, 1, (n_samples//3, 2))
cluster3 = np.random.normal(10, 1, (n_samples//3, 2))
X_customer = np.vstack([cluster1, cluster2, cluster3])

# 标准化
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X_customer)

# K-means聚类
kmeans = KMeans(n_clusters=3, random_state=42)
kmeans_labels = kmeans.fit_predict(X_scaled)

# DBSCAN聚类
dbscan = DBSCAN(eps=0.5, min_samples=5)
dbscan_labels = dbscan.fit_predict(X_scaled)

# 降维可视化
pca = PCA(n_components=2)
X_pca = pca.fit_transform(X_scaled)

# 可视化
fig, axes = plt.subplots(1, 3, figsize=(15, 5))

# 原始数据
axes[0].scatter(X_pca[:, 0], X_pca[:, 1], alpha=0.6)
axes[0].set_title('原始数据（PCA降维）')

# K-means结果
scatter = axes[1].scatter(X_pca[:, 0], X_pca[:, 1], c=kmeans_labels, cmap='viridis', alpha=0.6)
axes[1].set_title('K-means聚类结果')
plt.colorbar(scatter, ax=axes[1])

# DBSCAN结果
scatter = axes[2].scatter(X_pca[:, 0], X_pca[:, 1], c=dbscan_labels, cmap='viridis', alpha=0.6)
axes[2].set_title('DBSCAN聚类结果')
plt.colorbar(scatter, ax=axes[2])

plt.tight_layout()
plt.show()

# 聚类评估指标
from sklearn.metrics import silhouette_score, calinski_harabasz_score

print("聚类评估：")
print(f"K-means轮廓系数: {silhouette_score(X_scaled, kmeans_labels):.3f}")
print(f"K-means Calinski-Harabasz指数: {calinski_harabasz_score(X_scaled, kmeans_labels):.3f}")
print(f"DBSCAN轮廓系数: {silhouette_score(X_scaled, dbscan_labels):.3f}")

4.3 时间序列分析

import pandas as pd
import numpy as np
from statsmodels.tsa.seasonal import seasonal_decompose
from statsmodels.tsa.stattools import adfuller, acf, pacf
from statsmodels.tsa.arima.model import ARIMA
from pmdarima import auto_arima

# 创建时间序列数据
np.random.seed(42)
dates = pd.date_range('2023-01-01', periods=365, freq='D')
# 包含趋势、季节性和噪声
trend = np.linspace(100, 200, 365)
seasonal = 10 * np.sin(2 * np.pi * np.arange(365) / 30)  # 月度季节性
noise = np.random.normal(0, 5, 365)
ts_data = pd.DataFrame({
    'date': dates,
    'value': trend + seasonal + noise
}).set_index('date')

# 时间序列分解
result = seasonal_decompose(ts_data['value'], model='additive', period=30)
fig = result.plot()
fig.set_size_inches(12, 8)
plt.show()

# 平稳性检验（ADF检验）
adf_result = adfuller(ts_data['value'])
print("ADF检验结果：")
print(f"ADF统计量: {adf_result[0]:.3f}")
print(f"P值: {adf_result[1]:.3f}")
print(f"临界值: {adf_result[4]}")

# 自相关和偏自相关
acf_vals = acf(ts_data['value'], nlags=40)
pacf_vals = pacf(ts_data['value'], nlags=40)

fig, axes = plt.subplots(1, 2, figsize=(12, 5))
axes[0].stem(range(len(acf_vals)), acf_vals)
axes[0].set_title('自相关函数(ACF)')
axes[1].stem(range(len(pacf_vals)), pacf_vals)
axes[1].set_title('偏自相关函数(PACF)')
plt.tight_layout()
plt.show()

# ARIMA模型
# 使用auto_arima自动选择参数
model = auto_arima(ts_data['value'], seasonal=False, trace=True, 
                   error_action='ignore', suppress_warnings=True)
print(f"\n最佳ARIMA参数: {model.order}")

# 拟合模型
arima_model = ARIMA(ts_data['value'], order=model.order)
fitted_model = arima_model.fit()
print(fitted_model.summary())

# 预测
forecast = fitted_model.get_forecast(steps=30)
forecast_mean = forecast.predicted_mean
forecast_ci = forecast.conf_int()

# 可视化预测
plt.figure(figsize=(12, 6))
plt.plot(ts_data['value'], label='历史数据')
plt.plot(forecast_mean, label='预测', color='red')
plt.fill_between(forecast_ci.index, 
                 forecast_ci.iloc[:, 0], 
                 forecast_ci.iloc[:, 1], 
                 color='pink', alpha=0.3)
plt.title('ARIMA模型预测')
plt.legend()
plt.show()

第五部分：实战案例解析

5.1 电商销售分析案例

# 模拟电商销售数据
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

np.random.seed(42)

# 生成模拟数据
n_orders = 10000
start_date = datetime(2023, 1, 1)

orders = pd.DataFrame({
    '订单ID': range(1, n_orders + 1),
    '用户ID': np.random.randint(1, 1000, n_orders),
    '产品ID': np.random.randint(1, 100, n_orders),
    '数量': np.random.randint(1, 10, n_orders),
    '单价': np.random.uniform(10, 500, n_orders),
    '下单时间': [start_date + timedelta(days=np.random.randint(0, 365)) 
                for _ in range(n_orders)],
    '地区': np.random.choice(['华东', '华北', '华南', '华西', '东北'], n_orders),
    '支付方式': np.random.choice(['支付宝', '微信', '信用卡', '花呗'], n_orders)
})

# 计算销售额
orders['销售额'] = orders['数量'] * orders['单价']

# 数据清洗
# 处理异常值（销售额为0或负数）
orders = orders[orders['销售额'] > 0]

# 添加时间特征
orders['下单日期'] = orders['下单时间'].dt.date
orders['月份'] = orders['下单时间'].dt.month
orders['季度'] = orders['下单时间'].dt.quarter
orders['星期'] = orders['下单时间'].dt.dayofweek

# 分析1：销售趋势分析
monthly_sales = orders.groupby('月份')['销售额'].sum()
print("月度销售趋势：")
print(monthly_sales)

# 分析2：地区销售分析
region_sales = orders.groupby('地区')['销售额'].agg(['sum', 'count', 'mean'])
print("\n地区销售分析：")
print(region_sales)

# 分析3：用户价值分析（RFM模型）
# 计算RFM指标
snapshot_date = orders['下单时间'].max() + timedelta(days=1)
rfm = orders.groupby('用户ID').agg({
    '下单时间': lambda x: (snapshot_date - x.max()).days,  # Recency
    '订单ID': 'count',  # Frequency
    '销售额': 'sum'  # Monetary
}).rename(columns={'下单时间': 'Recency', '订单ID': 'Frequency', '销售额': 'Monetary'})

# RFM评分
rfm['R_Score'] = pd.qcut(rfm['Recency'], 5, labels=[5, 4, 3, 2, 1])
rfm['F_Score'] = pd.qcut(rfm['Frequency'].rank(method='first'), 5, labels=[1, 2, 3, 4, 5])
rfm['M_Score'] = pd.qcut(rfm['Monetary'], 5, labels=[1, 2, 3, 4, 5])

# 客户分群
rfm['RFM_Score'] = rfm['R_Score'].astype(str) + rfm['F_Score'].astype(str) + rfm['M_Score'].astype(str)
rfm['Segment'] = '其他'
rfm.loc[rfm['RFM_Score'].isin(['555', '554', '545', '544']), 'Segment'] = '重要价值客户'
rfm.loc[rfm['RFM_Score'].str.startswith('5') & ~rfm['Segment'].isin(['重要价值客户']), 'Segment'] = '重要保持客户'
rfm.loc[rfm['RFM_Score'].str.endswith('5') & ~rfm['Segment'].isin(['重要价值客户']), 'Segment'] = '重要发展客户'
rfm.loc[rfm['RFM_Score'].str.startswith('1'), 'Segment'] = '流失客户'

print("\n客户分群统计：")
print(rfm['Segment'].value_counts())

# 分析4：产品关联分析（购物篮分析）
from mlxtend.frequent_patterns import apriori, association_rules

# 准备购物篮数据
basket = orders.groupby(['订单ID', '产品ID'])['数量'].sum().unstack().fillna(0)
basket = basket.applymap(lambda x: 1 if x > 0 else 0)

# 挖掘频繁项集
frequent_itemsets = apriori(basket, min_support=0.01, use_colnames=True)
print("\n频繁项集（支持度>0.01）：")
print(frequent_itemsets)

# 生成关联规则
rules = association_rules(frequent_itemsets, metric="lift", min_threshold=1.2)
print("\n关联规则（提升度>1.2）：")
print(rules[['antecedents', 'consequents', 'support', 'confidence', 'lift']])

5.2 金融风控案例

# 信用卡欺诈检测
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, confusion_matrix, roc_auc_score
from imblearn.over_sampling import SMOTE

# 创建模拟信用卡交易数据
np.random.seed(42)
n_samples = 10000
n_fraud = int(n_samples * 0.01)  # 1%欺诈率

# 正常交易
normal_data = pd.DataFrame({
    '交易金额': np.random.exponential(100, n_samples - n_fraud),
    '交易时间': np.random.randint(0, 24, n_samples - n_fraud),
    '商户类型': np.random.choice(['餐饮', '购物', '娱乐', '交通'], n_samples - n_fraud),
    '交易地点': np.random.randint(1, 10, n_samples - n_fraud),
    '是否欺诈': 0
})

# 欺诈交易（金额更大，时间更异常）
fraud_data = pd.DataFrame({
    '交易金额': np.random.exponential(500, n_fraud),
    '交易时间': np.random.choice([0, 1, 2, 22, 23], n_fraud),  # 深夜交易
    '商户类型': np.random.choice(['赌博', '虚拟货币'], n_fraud),
    '交易地点': np.random.randint(1, 10, n_fraud),
    '是否欺诈': 1
})

# 合并数据
data = pd.concat([normal_data, fraud_data], ignore_index=True)

# 特征工程
# 1. 交易金额分箱
data['金额分箱'] = pd.cut(data['交易金额'], bins=[0, 50, 100, 200, 500, np.inf], 
                         labels=['极低', '低', '中', '高', '极高'])

# 2. 交易时间特征
data['是否深夜'] = (data['交易时间'] <= 6) | (data['交易时间'] >= 22)
data['是否工作时间'] = (data['交易时间'] >= 9) & (data['交易时间'] <= 18)

# 3. 商户类型编码
data['商户类型编码'] = data['商户类型'].map({
    '餐饮': 0, '购物': 1, '娱乐': 2, '交通': 3, '赌博': 4, '虚拟货币': 5
})

# 准备训练数据
X = data[['交易金额', '交易时间', '交易地点', '是否深夜', '是否工作时间', '商户类型编码']]
y = data['是否欺诈']

# 处理类别不平衡
smote = SMOTE(random_state=42)
X_resampled, y_resampled = smote.fit_resample(X, y)

# 数据分割
X_train, X_test, y_train, y_test = train_test_split(X_resampled, y_resampled, 
                                                    test_size=0.3, random_state=42)

# 标准化
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)

# 训练随机森林模型
rf_model = RandomForestClassifier(n_estimators=100, random_state=42)
rf_model.fit(X_train_scaled, y_train)

# 预测
y_pred = rf_model.predict(X_test_scaled)
y_pred_proba = rf_model.predict_proba(X_test_scaled)[:, 1]

# 评估
print("欺诈检测模型评估：")
print(f"准确率: {accuracy_score(y_test, y_pred):.3f}")
print(f"AUC-ROC: {roc_auc_score(y_test, y_pred_proba):.3f}")
print("\n分类报告：")
print(classification_report(y_test, y_pred))

# 混淆矩阵
cm = confusion_matrix(y_test, y_pred)
print("\n混淆矩阵：")
print(cm)

# 特征重要性
feature_importance = pd.DataFrame({
    '特征': X.columns,
    '重要性': rf_model.feature_importances_
}).sort_values('重要性', ascending=False)

print("\n特征重要性：")
print(feature_importance)

5.3 医疗健康数据分析案例

# 患者健康数据分析
import pandas as pd
import numpy as np
from sklearn.cluster import KMeans
from sklearn.decomposition import PCA
from sklearn.preprocessing import StandardScaler
import matplotlib.pyplot as plt
import seaborn as sns

# 创建模拟患者数据
np.random.seed(42)
n_patients = 1000

# 基本信息
patient_data = pd.DataFrame({
    '患者ID': range(1, n_patients + 1),
    '年龄': np.random.randint(20, 80, n_patients),
    '性别': np.random.choice(['男', '女'], n_patients),
    'BMI': np.random.normal(24, 4, n_patients),
    '收缩压': np.random.normal(120, 15, n_patients),
    '舒张压': np.random.normal(80, 10, n_patients),
    '血糖': np.random.normal(5.5, 1.5, n_patients),
    '胆固醇': np.random.normal(4.5, 1.2, n_patients),
    '吸烟史': np.random.choice([0, 1], n_patients, p=[0.7, 0.3]),
    '饮酒史': np.random.choice([0, 1], n_patients, p=[0.8, 0.2]),
    '运动频率': np.random.choice(['很少', '偶尔', '经常'], n_patients)
})

# 添加疾病风险标签（基于规则）
def calculate_risk(row):
    risk_score = 0
    if row['年龄'] > 60: risk_score += 2
    if row['BMI'] > 28: risk_score += 2
    if row['收缩压'] > 140: risk_score += 3
    if row['血糖'] > 6.1: risk_score += 3
    if row['胆固醇'] > 5.2: risk_score += 2
    if row['吸烟史'] == 1: risk_score += 2
    if row['饮酒史'] == 1: risk_score += 1
    if row['运动频率'] == '很少': risk_score += 2
    
    if risk_score >= 8:
        return '高风险'
    elif risk_score >= 4:
        return '中风险'
    else:
        return '低风险'

patient_data['疾病风险'] = patient_data.apply(calculate_risk, axis=1)

# 数据分析
print("患者基本信息统计：")
print(patient_data.describe())

# 风险分布
print("\n疾病风险分布：")
print(patient_data['疾病风险'].value_counts())

# 按风险分组统计
risk_groups = patient_data.groupby('疾病风险').agg({
    '年龄': 'mean',
    'BMI': 'mean',
    '收缩压': 'mean',
    '血糖': 'mean',
    '胆固醇': 'mean'
})
print("\n各风险组平均指标：")
print(risk_groups)

# 聚类分析（患者分群）
# 准备数值特征
numeric_features = ['年龄', 'BMI', '收缩压', '舒张压', '血糖', '胆固醇']
X_cluster = patient_data[numeric_features].copy()

# 标准化
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X_cluster)

# K-means聚类
kmeans = KMeans(n_clusters=3, random_state=42)
clusters = kmeans.fit_predict(X_scaled)
patient_data['Cluster'] = clusters

# PCA降维可视化
pca = PCA(n_components=2)
X_pca = pca.fit_transform(X_scaled)

plt.figure(figsize=(12, 8))
scatter = plt.scatter(X_pca[:, 0], X_pca[:, 1], c=clusters, cmap='viridis', 
                      alpha=0.6, s=50)
plt.colorbar(scatter, label='聚类')
plt.title('患者聚类分析（PCA降维）')
plt.xlabel('主成分1')
plt.ylabel('主成分2')
plt.show()

# 分析聚类特征
cluster_profile = patient_data.groupby('Cluster')[numeric_features].mean()
print("\n各聚类平均特征：")
print(cluster_profile)

# 风险与聚类的关系
risk_cluster_crosstab = pd.crosstab(patient_data['疾病风险'], patient_data['Cluster'])
print("\n风险与聚类交叉表：")
print(risk_cluster_crosstab)

# 统计检验：不同风险组的指标差异
from scipy import stats

high_risk = patient_data[patient_data['疾病风险'] == '高风险']
low_risk = patient_data[patient_data['疾病风险'] == '低风险']

print("\n高风险与低风险组的指标差异检验：")
for feature in numeric_features:
    t_stat, p_val = stats.ttest_ind(high_risk[feature], low_risk[feature])
    print(f"{feature}: t={t_stat:.3f}, p={p_val:.3f}")

第六部分：行业应用指南

6.1 电商行业应用

核心分析场景：

1. 用户行为分析：通过漏斗分析、留存分析、路径分析理解用户行为
2. 商品推荐系统：基于协同过滤、内容推荐、混合推荐
3. 价格弹性分析：通过A/B测试分析价格变动对销量的影响
4. 库存优化：基于时间序列预测需求，优化库存水平

关键指标：

• GMV（总成交额）
• 转化率（CVR）
• 客单价（AOV）
• 复购率
• 用户生命周期价值（LTV）

Python实现示例：

# 电商用户留存分析
def calculate_retention(df, user_col, date_col, periods=[1, 7, 30]):
    """
    计算用户留存率
    """
    df = df.sort_values([user_col, date_col])
    df['first_date'] = df.groupby(user_col)[date_col].transform('min')
    df['days_since_first'] = (df[date_col] - df['first_date']).dt.days
    
    retention = {}
    for period in periods:
        retained_users = df[df['days_since_first'] == period][user_col].unique()
        first_users = df[df['days_since_first'] == 0][user_col].unique()
        retention[period] = len(set(retained_users) & set(first_users)) / len(first_users)
    
    return retention

# 示例使用
# 假设orders是订单数据，包含user_id和order_date
# retention = calculate_retention(orders, 'user_id', 'order_date')

6.2 金融行业应用

核心分析场景：

1. 信用评分：基于用户行为、交易记录、社交数据构建评分模型
2. 欺诈检测：实时交易监控，异常模式识别
3. 投资组合优化：基于马科维茨模型、Black-Litterman模型
4. 风险价值（VaR）计算：历史模拟法、蒙特卡洛模拟法

关键指标：

• 违约率（PD）
• 违约损失率（LGD）
• 违约风险暴露（EAD）
• 夏普比率
• 最大回撤

Python实现示例：

# 信用评分模型（逻辑回归）
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import roc_curve, auc

def build_credit_score_model(X, y):
    """
    构建信用评分模型
    """
    # 数据分割
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)
    
    # 模型训练
    model = LogisticRegression(random_state=42)
    model.fit(X_train, y_train)
    
    # 预测
    y_pred_proba = model.predict_proba(X_test)[:, 1]
    
    # 评估
    fpr, tpr, _ = roc_curve(y_test, y_pred_proba)
    auc_score = auc(fpr, tpr)
    
    return model, auc_score, (fpr, tpr)

# 示例使用
# X是特征矩阵，y是违约标签（0=正常，1=违约）
# model, auc_score, roc_data = build_credit_score_model(X, y)

6.3 医疗健康行业应用

核心分析场景：

1. 疾病预测：基于患者历史数据预测疾病风险
2. 医疗资源优化：基于患者流量预测优化医生排班
3. 药物疗效分析：通过A/B测试分析不同治疗方案效果
4. 医学影像分析：使用深度学习进行病灶检测

关键指标：

• 疾病发病率
• 治愈率
• 住院时长
• 医疗成本
• 患者满意度

Python实现示例：

# 疾病风险预测模型
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.model_selection import cross_val_score

def predict_disease_risk(patient_features, disease_labels):
    """
    预测疾病风险
    """
    # 模型训练
    model = GradientBoostingClassifier(n_estimators=100, random_state=42)
    
    # 交叉验证
    cv_scores = cross_val_score(model, patient_features, disease_labels, 
                                cv=5, scoring='roc_auc')
    
    # 完整训练
    model.fit(patient_features, disease_labels)
    
    return model, cv_scores.mean()

# 示例使用
# patient_features是患者特征矩阵，disease_labels是疾病标签
# model, auc_score = predict_disease_risk(patient_features, disease_labels)

6.4 制造业应用

核心分析场景：

1. 预测性维护：基于设备传感器数据预测故障
2. 质量控制：通过统计过程控制（SPC）监控生产质量
3. 供应链优化：需求预测、库存优化、供应商评估
4. 能源管理：优化能源消耗，降低生产成本

关键指标：

• 设备综合效率（OEE）
• 首次通过率（FPY）
• 生产周期时间
• 库存周转率
• 能源效率

Python实现示例：

# 预测性维护模型
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler

def predict_equipment_failure(sensor_data):
    """
    预测设备故障
    """
    # 数据预处理
    scaler = StandardScaler()
    X_scaled = scaler.fit_transform(sensor_data)
    
    # 异常检测（孤立森林）
    iso_forest = IsolationForest(contamination=0.05, random_state=42)
    predictions = iso_forest.fit_predict(X_scaled)
    
    # -1表示异常（可能故障）
    failure_risk = (predictions == -1).astype(int)
    
    return failure_risk, iso_forest

# 示例使用
# sensor_data是设备传感器数据矩阵
# failure_risk, model = predict_equipment_failure(sensor_data)

第七部分：性能优化与工程实践

7.1 大数据处理优化

# 使用Dask处理大数据
import dask.dataframe as dd
import pandas as pd
import numpy as np

# 创建模拟大数据集
def create_large_dataset(n_rows=10_000_000):
    """创建1000万行的模拟数据"""
    data = {
        'user_id': np.random.randint(1, 1_000_000, n_rows),
        'timestamp': pd.date_range('2023-01-01', periods=n_rows, freq='S'),
        'value': np.random.randn(n_rows)
    }
    return pd.DataFrame(data)

# 使用Dask处理大数据
def process_with_dask(df_path):
    """使用Dask处理大数据"""
    # 读取数据（分块）
    ddf = dd.read_csv(df_path, blocksize='256MB')
    
    # 惰性计算
    result = ddf.groupby('user_id')['value'].mean()
    
    # 执行计算
    computed_result = result.compute()
    return computed_result

# 使用Pandas优化技巧
def optimize_pandas_memory(df):
    """优化Pandas内存使用"""
    # 优化数据类型
    for col in df.select_dtypes(include=['object']).columns:
        df[col] = df[col].astype('category')
    
    # 优化数值类型
    for col in df.select_dtypes(include=['int']).columns:
        df[col] = pd.to_numeric(df[col], downcast='integer')
    
    for col in df.select_dtypes(include=['float']).columns:
        df[col] = pd.to_numeric(df[col], downcast='float')
    
    return df

# 并行处理
from concurrent.futures import ProcessPoolExecutor
import multiprocessing

def parallel_processing(data_chunks, func):
    """并行处理数据块"""
    with ProcessPoolExecutor(max_workers=multiprocessing.cpu_count()) as executor:
        results = list(executor.map(func, data_chunks))
    return results

7.2 代码性能优化

# 向量化操作 vs 循环
import numpy as np
import time

# 创建大型数组
size = 10_000_000
arr = np.random.rand(size)

# 方法1：循环（慢）
def loop_method(arr):
    result = np.zeros_like(arr)
    for i in range(len(arr)):
        result[i] = arr[i] * 2 + 1
    return result

# 方法2：向量化（快）
def vectorized_method(arr):
    return arr * 2 + 1

# 性能对比
start = time.time()
loop_result = loop_method(arr)
loop_time = time.time() - start

start = time.time()
vectorized_result = vectorized_method(arr)
vectorized_time = time.time() - start

print(f"循环方法耗时: {loop_time:.4f}秒")
print(f"向量化方法耗时: {vectorized_time:.4f}秒")
print(f"性能提升: {loop_time/vectorized_time:.1f}倍")

# 使用Numba加速
from numba import jit

@jit(nopython=True)
def numba_method(arr):
    result = np.zeros_like(arr)
    for i in range(len(arr)):
        result[i] = arr[i] * 2 + 1
    return result

# 首次调用会编译
start = time.time()
numba_result = numba_method(arr)
numba_time = time.time() - start

print(f"Numba方法耗时: {numba_time:.4f}秒")
print(f"性能提升: {loop_time/numba_time:.1f}倍")

7.3 代码可维护性与最佳实践

# 1. 模块化设计
# data_processing.py
"""
数据处理模块
"""
import pandas as pd
import numpy as np

class DataCleaner:
    """数据清洗类"""
    
    def __init__(self, df):
        self.df = df.copy()
    
    def handle_missing_values(self, strategy='mean'):
        """处理缺失值"""
        if strategy == 'mean':
            self.df = self.df.fillna(self.df.mean())
        elif strategy == 'median':
            self.df = self.df.fillna(self.df.median())
        return self.df
    
    def remove_outliers(self, column, method='iqr'):
        """移除异常值"""
        if method == 'iqr':
            Q1 = self.df[column].quantile(0.25)
            Q3 = self.df[column].quantile(0.75)
            IQR = Q3 - Q1
            lower_bound = Q1 - 1.5 * IQR
            upper_bound = Q3 + 1.5 * IQR
            self.df = self.df[(self.df[column] >= lower_bound) & 
                             (self.df[column] <= upper_bound)]
        return self.df

# 2. 配置管理
import yaml

class Config:
    """配置管理类"""
    
    def __init__(self, config_path):
        with open(config_path, 'r') as f:
            self.config = yaml.safe_load(f)
    
    def get(self, key, default=None):
        """获取配置项"""
        keys = key.split('.')
        value = self.config
        for k in keys:
            if isinstance(value, dict) and k in value:
                value = value[k]
            else:
                return default
        return value

# 3. 日志记录
import logging
import sys

def setup_logging(log_file='data_analysis.log'):
    """设置日志"""
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        handlers=[
            logging.FileHandler(log_file),
            logging.StreamHandler(sys.stdout)
        ]
    )
    return logging.getLogger(__name__)

# 4. 单元测试
import unittest

class TestDataCleaner(unittest.TestCase):
    """数据清洗器测试"""
    
    def setUp(self):
        self.df = pd.DataFrame({
            'A': [1, 2, np.nan, 4],
            'B': [5, 6, 7, 8]
        })
        self.cleaner = DataCleaner(self.df)
    
    def test_handle_missing_values(self):
        """测试缺失值处理"""
        result = self.cleaner.handle_missing_values(strategy='mean')
        self.assertEqual(result['A'].iloc[2], 2.3333333333333335)
    
    def test_remove_outliers(self):
        """测试异常值移除"""
        df_with_outliers = pd.DataFrame({
            'value': [1, 2, 3, 100, 4, 5, 6]
        })
        cleaner = DataCleaner(df_with_outliers)
        result = cleaner.remove_outliers('value')
        self.assertEqual(len(result), 6)  # 移除了一个异常值

# 运行测试
if __name__ == '__main__':
    unittest.main()

第八部分：持续学习与资源推荐

8.1 学习路径建议

1. 基础阶段（1-3个月）

   • Python基础语法
   • NumPy、Pandas基础操作
   • Matplotlib/Seaborn基础可视化
   • 基础统计学知识

2. 进阶阶段（3-6个月）

   • Pandas高级操作（多级索引、时间序列）
   • 数据清洗与特征工程
   • 统计分析与假设检验
   • 机器学习基础（监督学习、无监督学习）

3. 精通阶段（6-12个月）

   • 大数据处理（Dask、PySpark）
   • 深度学习在数据分析中的应用
   • 领域专业知识（电商、金融、医疗等）
   • 工程化实践（代码优化、项目管理）

8.2 推荐资源

在线课程：

• Coursera: “Applied Data Science with Python”（密歇根大学）
• edX: “Data Science MicroMasters”（UC San Diego）
• Kaggle Learn: 免费实战课程

书籍推荐：

• 《Python数据分析基础》（Wes McKinney）
• 《利用Python进行数据分析》（Wes McKinney）
• 《统计学习方法》（李航）
• 《机器学习》（周志华）

开源项目：

• Pandas官方文档
• Scikit-learn官方文档
• Kaggle竞赛项目
• GitHub上的数据分析项目

社区与论坛：

• Kaggle社区
• Stack Overflow
• 知乎数据分析话题
• GitHub开源项目

8.3 持续学习建议

1. 保持实践：每周至少完成一个小项目
2. 参与竞赛：Kaggle竞赛是提升实战能力的最佳途径
3. 阅读论文：关注顶级会议（KDD、ICML、NeurIPS）的最新研究
4. 建立作品集：将项目整理到GitHub，建立个人品牌
5. 加入社区：参与线下/线上技术分享，交流学习心得

结语

Python数据分析从入门到精通是一个循序渐进的过程，需要理论学习与实践相结合。通过本指南的系统学习，你将掌握从数据清洗、统计分析到机器学习应用的完整技能链，并能在不同行业场景中灵活应用。

记住，数据分析的核心不仅是技术，更是业务理解与问题解决能力。保持好奇心，持续学习，你一定能在数据分析领域取得成功！

最后建议：选择一个你感兴趣的行业领域，深入研究其业务场景和数据特点，将技术能力与业务知识相结合，成为既懂技术又懂业务的数据分析专家。