Python Pandas数据分析的使用完整教学

import pandas as pd
import numpy as np

# 新手做法：直接读取，导致内存撑爆或类型推断错误
# df = pd.read_csv('huge_orders.csv')

# 专业做法：精准控制列、类型与分块
# 1. 使用 usecols 只读需要的列
# 2. 使用 dtype 预先指定类型，避免 object 和 int64 的内存浪费
# 3. 使用 parse_dates 直接解析时间
dtypes = {
    'order_id': 'int32',
    'user_id': 'int32',
    'status': 'category',  # 低基数枚举字段使用 category
    'amount': 'float32'
}

df = pd.read_csv(
    'huge_orders.csv',
    usecols=['order_id', 'user_id', 'status', 'amount', 'order_time'],
    dtype=dtypes,
    parse_dates=['order_time']
)

chunk_iter = pd.read_csv('billion_rows.csv', chunksize=100_000)
total_revenue = 0

for chunk in chunk_iter:
    # 在每个 chunk 上进行聚合操作，最后合并结果
    total_revenue += chunk['amount'].sum()

# 写入 Parquet (使用 pyarrow 引擎)
df.to_parquet('orders_clean.parquet', engine='pyarrow', compression='snappy')

# 读取 Parquet (Pandas 2.0+ 推荐开启 PyArrow 后端)
df_arrow = pd.read_parquet('orders_clean.parquet', dtype_backend='pyarrow')

def diagnose_dataframe(df: pd.DataFrame) -> pd.DataFrame:
    """数据质量诊断函数"""
    diag = pd.DataFrame({
        'Dtype': df.dtypes,
        'Non_Null': df.count(),
        'Null_Count': df.isnull().sum(),
        'Null_Rate(%)': (df.isnull().sum() / len(df) * 100).round(2),
        'Unique_Count': df.nunique(),
        'Memory_MB': df.memory_usage(deep=True) / (1024**2)
    })
    return diag.sort_values('Null_Rate(%)', ascending=False)

# 调用
diagnose_dataframe(df)

# 查看数值型统计量，并调整显示精度
pd.set_option('display.float_format', lambda x: '%.2f' % x)
print(df.describe(percentiles=[.05, .25, .5, .75, .95, .99])) # 关注长尾分布

# 查看类别型字段的基数和Top频率
print(df.describe(include=['category', 'object']))

# 1. 业务规则填充：例如“退款金额”缺失代表未退款，填0
df['refund_amount'] = df['refund_amount'].fillna(0)

# 2. 分组填充：用该用户的历史均值填充缺失的评分
df['user_score'] = df.groupby('user_id')['user_score'].transform(
    lambda x: x.fillna(x.mean())
)

# 3. 时间序列插值：对于连续的传感器/财务数据，使用线性或样条插值
df['daily_revenue'] = df['daily_revenue'].interpolate(method='spline', order=2)

def clip_outliers_iqr(series: pd.Series, factor=1.5) -> pd.Series:
    """基于 IQR 的异常值截断（盖帽法）"""
    Q1 = series.quantile(0.25)
    Q3 = series.quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - factor * IQR
    upper_bound = Q3 + factor * IQR
    return series.clip(lower=lower_bound, upper=upper_bound)

df['amount_clean'] = clip_outliers_iqr(df['amount'])

# 假设 df['remark'] 包含类似 "使用优惠券:VIP2026, 免邮" 的文本
# ❌ 慢速做法：df['remark'].apply(lambda x: re.search(...))
# ✅ 向量化做法：
df['coupon_code'] = df['remark'].str.extract(r'优惠券:([A-Z0-9]+)')

# 清理隐藏字符并统一小写
df['city'] = df['city'].str.strip().str.lower().str.replace(r'\s+', ' ', regex=True)

# 筛选：高价值、已完成、特定城市的订单
target_cities = ['北京', '上海', '广州', '深圳']

# 使用 @ 引用外部变量
df_filtered = df.query(
    "amount > 1000 and status == 'completed' and city in @target_cities"
)

user_stats = df.groupby('user_id').agg(
    total_orders=('order_id', 'count'),
    total_spend=('amount', 'sum'),
    avg_spend=('amount', 'mean'),
    first_order=('order_time', 'min'),
    last_order=('order_time', 'max')
).reset_index()

# 计算 RFM 模型中的时间间隔特征
user_stats['recency_days'] = (pd.Timestamp('2026-05-27') - user_stats['last_order']).dt.days

# 宽转长：将 Jan, Feb, Mar 三列销售额转为一列
df_long = pd.melt(
    df_wide, 
    id_vars=['store_id', 'region'], 
    value_vars=['Jan_Sales', 'Feb_Sales', 'Mar_Sales'],
    var_name='month', 
    value_name='sales'
)

# 长转宽：生成 地区 x 月份 的交叉报表
report = pd.pivot_table(
    df_long, 
    values='sales', 
    index='region', 
    columns='month', 
    aggfunc='sum', 
    fill_value=0,
    margins=True # 添加总计行/列
)

# ✅ 务必使用 validate 参数验证关联键的唯一性
# 'm:1' 表示左表多对一右表，如果右表 key 不唯一，直接报错拦截！
df_final = pd.merge(
    orders, 
    users, 
    on='user_id', 
    how='left',
    validate='m:1', 
    suffixes=('_order', '_user')
)

# 确保索引是 DatetimeIndex
df_time = df.set_index('order_time')

# 按日重采样，计算每日总GMV和订单数
daily_report = df_time.resample('D').agg(
    gmv=('amount', 'sum'),
    orders=('order_id', 'count')
).fillna(0) # 补齐没有交易的日期

# 1. 7日移动平均 GMV
daily_report['gmv_ma7'] = daily_report['gmv'].rolling(window=7).mean()

# 2. 计算日环比增长率 (今日 / 昨日 - 1)
daily_report['gmv_dod'] = daily_report['gmv'].pct_change(periods=1)

# 3. 计算周同比 (今日 / 7天前 - 1)
daily_report['gmv_wow'] = daily_report['gmv'].pct_change(periods=7)

# 开启全局 PyArrow 后端 (Pandas 2.0+)
pd.options.mode.dtype_backend = "pyarrow"

# 或者在读取时指定
df = pd.read_csv('data.csv', engine='pyarrow', dtype_backend='pyarrow')
# 此时字符串不再是低效的 object，而是 string[pyarrow]，内存直降 70%！

# 开启 CoW
pd.options.mode.copy_on_write = True

# 现在，切片操作是懒加载的（视图），只有当你真正修改切片数据时，才会发生内存拷贝。
subset = df[df['amount'] > 100]
subset['new_col'] = 1  # 安全，不会触发警告，也不会意外修改原 df

# 场景：根据金额划分等级 (低、中、高)

# ❌ 极慢的 apply
def get_level(x):
    if x < 100: return 'Low'
    elif x < 1000: return 'Mid'
    else: return 'High'
df['level'] = df['amount'].apply(get_level)

# ✅ 向量化方案 1：np.select (推荐，逻辑清晰)
conditions = [df['amount'] < 100, df['amount'] < 1000]
choices = ['Low', 'Mid']
df['level'] = np.select(conditions, choices, default='High')

# ✅ 向量化方案 2：pd.cut (适用于连续数值分箱)
df['level_bin'] = pd.cut(df['amount'], bins=[0, 100, 1000, np.inf], labels=['Low', 'Mid', 'High'])

def drop_outliers(df: pd.DataFrame, col: str) -> pd.DataFrame:
    """自定义清洗管道"""
    limit = df[col].quantile(0.99)
    return df[df[col] <= limit]

# 一气呵成的数据流水线
final_df = (
    pd.read_csv('raw_data.csv', parse_dates=['time'])
    .query("status == 'success'")
    .assign(
        # 在链中直接创建新列
        revenue = lambda x: x['price'] * x['qty'],
        year_month = lambda x: x['time'].dt.to_period('M')
    )
    .pipe(drop_outliers, col='revenue')  # 接入自定义函数
    .groupby('year_month')
    .agg(total_rev=('revenue', 'sum'))
    .reset_index()
)