import pandas as pd
import numpy as np
from io import StringIO

csv_data = """
TransactionID,Date,Product,Category,UnitPrice,Quantity,CustomerType,Region,Discount,PaymentMethod
,2025-01-05,Widget A,Gadgets,100.5,5,Regular,North,0.1,Credit Card
1002,2025-01-06,Widget B,Gadgets,200.0,,VIP,South,0.15, 
1003,invalid_date,Widget C,Accessories,50.25,2,Regular,East,,PayPal
1004,2025-01-08,Widget D,Gadgets,300.0,10,VIP,North,0.2,Credit Card
1005,2025-01-09,Widget E ,Accessories,75.0,3,Regular,West,0.05, 
1006,2025-01-10,Widget F,Gadgets,150.0,8,Regular,South,,Cash
1007,2025-01-11,Widget G,Accessories,25.0,1, ,East,0,
1008,2025-01-12,Widget H,Gadgets,500.0,4,VIP,North,0.3,Credit Card
1009,2025-01-13,Widget I,Accessories,60.0,6,Regular,West,0.1,PayPal
1010,2025-01-14,Widget A,Gadgets,100.5,2,,North,,Credit Card
"""

# ========================================================
# Step 1: 读取（接入层）
# ========================================================
df = pd.read_csv(
    StringIO(csv_data),
    sep=',',
    dtype={
        'TransactionID': 'Int64',    # 👉 核心层：空值友好的整数类型
        'Quantity': 'Int64',
    },
    na_values=['', 'NA', 'null'],
    on_bad_lines='warn',             # 遇到坏行跳过并警告
)
# ========================================================
# Step 2: 探查（贯穿整个流程，属于核心层的数据审视）
# ========================================================
print("=== 概览 ===")
df.info()
print("=== 统计 ===")
df.describe(include='all')

# ========================================================
# Step 3: 清洗
# ========================================================

# 3a 日期清洗 → 无效日期变 NaT          🧹 清洗
df['Date'] = pd.to_datetime(df['Date'], errors='coerce')

# 3b Region 列的空格去掉                🧹 清洗（str 访问器）
df['Region'] = df['Region'].str.strip()

# 3c Product 列空格 + CustomerType 空值  🧹 清洗
df['Product'] = df['Product'].str.strip()
df['CustomerType'] = df['CustomerType'].str.strip().fillna('Regular')

# 3d 统一填充缺失值                      🧹 清洗
df = df.fillna({
    'TransactionID': -1,
    'Quantity': 0,
    'Discount': 0.0,
    'PaymentMethod': 'Unknown',
})

# 3e 去重（假设 TransactionID 是唯一键）  🧹 清洗
df.drop_duplicates(subset=['TransactionID'], keep='first', inplace=True)

# ========================================================
# Step 4: 处理
# ========================================================

# 4a 派生列：收入 = 单价 × 数量          🛠 变换（向量化运算）
df['Revenue'] = df['UnitPrice'] * df['Quantity']

# 4b 分类：根据收入打标签               🛠 变换（apply）
df['RevenueLevel'] = df['Revenue'].apply(
    lambda x: 'High' if x >= 1000 else ('Mid' if x >= 500 else 'Low')
)

# 4c 客户类型列转为 category 类型        👉 核心层：类型优化
df['CustomerType'] = df['CustomerType'].astype('category')

# ========================================================
# Step 5: 分析
# ========================================================

# 5a 按产品类别分组汇总                  📊 进阶层：分组聚合
summary = (
    df.groupby('Category')
    .agg(
        TotalRevenue=('Revenue', 'sum'),
        AvgUnitPrice=('UnitPrice', 'mean'),
        OrderCount=('TransactionID', 'count'),
        AvgDiscount=('Discount', 'mean'),
    )
    .reset_index()
)

# 5b 按区域和客户类型做透视              📊 进阶层：重塑
pivot = df.pivot_table(
    index='Region',
    columns='CustomerType',
    values='Revenue',
    aggfunc='sum',
    fill_value=0,
)

# ========================================================
# Step 6: 窗口计算
# ========================================================

# 按日期排序后，计算销售额的 3 期移动平均  ⏱ 时间序列
df_sorted = df.sort_values('Date').reset_index(drop=True)
df_sorted['Revenue_MA3'] = df_sorted['Revenue'].rolling(window=3).mean()

# ========================================================
# Step 7: 持久化                          📤 持久化
# ========================================================
summary.to_csv('retail_summary.csv', index=False, encoding='utf-8-sig')
df_sorted.to_parquet('retail_full.parquet', index=False)

print("全流程结束，已输出 retail_summary.csv 和 retail_full.parquet")

DataFrame
├── columns: Index(['TransactionID', 'Date', 'Product', ...])   ← 列名也是 Index
├── index: Index([0, 1, 2, 3, ...])                             ← 行标签
├── 'TransactionID': Series (Int64)                              ← 每一列是 Series
├── 'Product': Series (object)
├── 'Date': Series (datetime64)
└── ...

a = pd.DataFrame({'val': [1, 2, 3]}, index=['x', 'y', 'z'])
b = pd.DataFrame({'val': [10, 20, 30]}, index=['z', 'y', 'x'])
a + b  # 结果按 x/y/z 对齐，而不是逐行加
# Output:
#    val
# x   31  ← a的1 + b的30 = 31
# y   22  ← 2 + 20 = 22
# z   13  ← 3 + 10 = 13

df = pd.DataFrame({'A': [1, 2, 3], 'B': [4, 5, 6]})

# 返回视图的情况：选取单列、切片行
view = df['A']       # view 是 df['A'] 的视图
view.iloc[0] = 999   # ⚠️ 会修改原始 df！因为没复制

# 安全的做法：显式复制
safe_view = df['A'].copy()
safe_view.iloc[0] = 999  # ✅ 不影响原 df

# SettingWithCopyWarning 的经典场景
df[df['A'] > 1]['B'] = 10   # ❌ 链式取 → 无法确定是视图还是副本，pandas 报警告
df.loc[df['A'] > 1, 'B'] = 10  # ✅ 单次操作，pandas 明确知道要修改原对象

# 问题：列中有 NaN 时，int64 无法存 NaN，自动转 float64
df = pd.DataFrame({'id': [1, None, 3]})
print(df['id'].dtype)  # float64 ← 1.0, NaN, 3.0，连整数都变成了浮点

# 方案：用可空整数类型 Int64（大写 I）
df['id'] = df['id'].astype('Int64')
print(df['id'].dtype)  # Int64 ← 1, <NA>, 3，保持整数形态

# 读取时直接指定
df = pd.read_csv('file.csv', dtype={'id': 'Int64'})

# 🎯 场景 1：按标签选（.loc）
df.loc[0:3]                        # 标签 0 到 3（包含两端）
df.loc[df['Region'] == 'North']    # 条件筛选
df.loc[df['Revenue'] > 500, ['Product', 'Revenue']]  # 行条件 + 指定列（最常用写法）

# 🎯 场景 2：按位置选（.iloc）
df.iloc[0:3]                       # 前 3 行（不包含 3，和 Python 切片一致）
df.iloc[:, [0, 2, 4]]              # 第 0/2/4 列

# 🎯 场景 3：标量快速取值（.at / .iat）
df.at[2, 'Product']                # 第 2 行 Product 列的值（比 .loc 快）
df.iat[2, 2]                       # 第 2 行第 2 列的值（比 .iloc 快）

# 🎯 推荐用法总结
df[df['A'] > 0]['B']               # ❌ 链式索引 → 可能 SettingWithCopyWarning
df.loc[df['A'] > 0, 'B']           # ✅ 一步到位，清晰安全

# 🛠 选择 1：向量化运算（最推荐，性能最好）
df['Revenue'] = df['UnitPrice'] * df['Quantity']      # 四则运算
df['HighValue'] = df['Revenue'] > 1000                 # 布尔列
df['DiscountedPrice'] = df['UnitPrice'] * (1 - df['Discount'])  # 复合运算

# 🛠 选择 2：map — Series 的一对一映射（适用于枚举翻译）
df['RegionCN'] = df['Region'].map({
    'North': '华北', 'South': '华南', 'East': '华东', 'West': '西部'
})

# 🛠 选择 3：apply — 函数应用（灵活但非向量化）
# 场景 A：单列传入，返回标量
df['RevenueLevel'] = df['Revenue'].apply(lambda x: 'High' if x > 800 else 'Low')

# 场景 B：多列传入，返回标量（axis=1 表示按行）
df['Score'] = df.apply(
    lambda row: row['UnitPrice'] * row['Quantity'] * (1 - row['Discount']),
    axis=1,
)

# 🛠 选择 4：访问器 — .str 和 .dt
df['Product_Upper'] = df['Product'].str.upper()           # 字符串向量化
df['YearMonth'] = df['Date'].dt.strftime('%Y-%m')         # 日期字段提取
df['Region'] = df['Region'].str.strip()                   # 去空格

# 🛠 选择 5：管道（pipe）— 函数式组合
def add_score(df, multiplier=1):
    df['Score'] = df['Revenue'] * multiplier
    return df

df = df.pipe(add_score, multiplier=1.2)

# 🧹 缺失值检测
df.isna().sum()                          # 每列缺失数量
df.isna().mean() * 100                   # 每列缺失百分比

# 🧹 缺失值删除
df.dropna(subset=['TransactionID'])      # 关键列有缺失 → 直接删行
df.dropna(thresh=5)                      # 一行有 5 个以上非空才保留

# 🧹 缺失值填充
df.fillna(0)                                    # 全局填 0
df.fillna({'Quantity': 0, 'Discount': 0.0})      # 逐列填不同值
df.fillna(method='ffill')                        # 前向填充（时间序列常用）
df.fillna(method='bfill')                        # 后向填充

# 🧹 重复值
df.duplicated(subset=['TransactionID'])          # 标记重复
df.drop_duplicates(subset=['TransactionID'], keep='first')  # 删除重复，保留首个

# 🧹 类型治理
df['col'] = df['col'].astype('Int64')            # 类型转换
df['col'] = pd.to_numeric(df['col'], errors='coerce')  # 强制数字，无效变 NaN
df['col'] = pd.to_datetime(df['col'], errors='coerce')  # 强制日期，无效变 NaT

# 🧹 异常值治理
df['Price'].clip(lower=0, upper=10000)           # 截断异常值（如价格负数）
df['Region'] = df['Region'].replace({'': 'Unknown', ' ': 'Unknown'})  # 替换特定值

# 按单个字段分组
df.groupby('Category')['Revenue'].sum()

# 按多个字段分组
df.groupby(['Category', 'Region'])['Revenue'].sum()

# 命名聚合（一次性做多件事，列名自解释）
summary = (
    df.groupby('Category')
    .agg(
        TotalRevenue=('Revenue', 'sum'),
        AvgPrice=('UnitPrice', 'mean'),
        OrderCount=('TransactionID', 'count'),
        MaxQty=('Quantity', 'max'),
    )
    .reset_index()  # 把 Category 从索引变回普通列
)

# transform：把聚合结果广播回原始形状（保持行数不变）
df['CategoryAvgRevenue'] = df.groupby('Category')['Revenue'].transform('mean')

# filter：筛选整个组
df.groupby('Category').filter(lambda g: g['Revenue'].sum() > 2000)

# 窗口函数（rolling）：依赖排序，常用于时间序列
df_sorted['Revenue_MA3'] = df_sorted['Revenue'].rolling(window=3).mean()
df_sorted['Revenue_Cumsum'] = df_sorted['Revenue'].cumsum()

# 🔄 宽表化（pivot）：行×列×值 = 三维透视
# 把 Category 展开为列名，查看每个 Region 的各类别收入
pivot = df.pivot_table(
    index='Region',
    columns='Category',
    values='Revenue',
    aggfunc='sum',
    fill_value=0,
)

# 🔄 长表化（melt）：把宽表还原回长表
melted = pivot.reset_index().melt(
    id_vars=['Region'],
    var_name='Category',
    value_name='Revenue',
)

# 🔗 横向合并（merge）：类似 SQL JOIN
left = df[['TransactionID', 'Product', 'Revenue']]
right = pd.DataFrame({
    'Product': ['Widget A', 'Widget B'],
    'ProductGroup': ['Premium', 'Standard'],
})
merged = pd.merge(left, right, on='Product', how='left')

# 🔗 纵向拼接（concat）：类似 UNION ALL
# 模拟两份数据
q1 = df[df['Date'].dt.quarter == 1].copy()
q2 = df[df['Date'].dt.quarter == 2].copy()
all_data = pd.concat([q1, q2], axis=0, ignore_index=True)

# 🔗 索引连接（join）：按索引对齐，两个索引相同的 df 合并
df_with_index = df.set_index('TransactionID')
meta = pd.DataFrame({'ProductGroup': ['Gadget', 'Accessory']}, index=[1002, 1003])
meta.index.name = 'TransactionID'
joined = df_with_index.join(meta)

# 确保日期列是 datetime 类型
df['Date'] = pd.to_datetime(df['Date'], errors='coerce')

# 排序（时间序列操作大多依赖有序索引）
df = df.sort_values('Date').reset_index(drop=True)

# ⏱ 移动平均（rolling）
df['Revenue_MA3'] = df['Revenue'].rolling(window=3).mean()

# ⏱ 环比差异（shift：把上一行的值拿到当前行）
df['Revenue_Change'] = df['Revenue'] - df['Revenue'].shift(1)

# ⏱ 累计求和（cumsum）
df['Revenue_Cum'] = df['Revenue'].cumsum()

# ⏱ 重采样（resample）：改变时间频率 — 需要时间作为索引
df_ts = df.set_index('Date')
# 按周聚合
weekly = df_ts.resample('W')['Revenue'].sum()
# 按月聚合
monthly = df_ts.resample('M')['Revenue'].agg(['sum', 'mean', 'count'])

# ⚡ 原则 1：向量化 > apply > 循环
# 这是一个数量级的差距
df['Revenue'] = df['UnitPrice'] * df['Quantity']            # ✅ 向量化（最快）
df['RevenueLevel'] = df['Revenue'].apply(lambda x: ...)       # ✅ apply（灵活但慢）
for i in range(len(df)): df.iloc[i, ...] = ...               # ❌ 循环（最慢）

# ⚡ 原则 2：category 类型优化内存
# 例：Region 列只有 4 种取值
df['Region'] = df['Region'].astype('category')
# 内存可以从 object 的 ~300MB 降到 category 的 ~5MB（数据量百万行时差异明显）

# ⚡ 原则 3：只读需要的列
df = pd.read_csv('large.csv', usecols=['Product', 'Sales'])  # ✅ 避免读入无用列

# ⚡ 原则 4：分块处理超大文件
chunks = pd.read_csv('huge_file.csv', chunksize=10000)
result = pd.concat([
    chunk.groupby('Category')['Sales'].sum() for chunk in chunks
]).groupby(level=0).sum()

# ⚡ 原则 5：inplace=True 不一定更快
# inplace 的本意是"就地修改"，但对大部分操作来说 pandas 内部仍然会复制
# 新版 pandas 已标记 inplace 为 deprecated，推荐显式赋值
df = df.drop_duplicates()                # ✅ 推荐
df.drop_duplicates(inplace=True)         # ⚠️ 不推荐（inplace 将被废弃）

# 现象
df.loc[df['Region'] == 'North', 'Revenue']  = 0  # 正常
df[df['Region'] == 'North']['Revenue'] = 0        # ⚠️ SettingWithCopyWarning

# 为什么会发生？
# 链式索引 df[A][B] 等价于：(df[A])[B] → 中间变量 df[A] 可能是视图也可能是副本
# pandas 无法判断你是想改原数据还是改副本 → 于是警告

# 方案：永远用 .loc 单步操作
df.loc[df['Region'] == 'North', 'Revenue'] = 0   # 一步到位，语义明确

# 现象
df = pd.DataFrame({'id': [1, None, 3]})
print(df['id'].dtype)  # float64 → 1.0, NaN, 3.0（整数变浮点）

# 为什么会发生？
# numpy 的 int64 不支持 NaN，pandas 自动提升为 float64（float64 支持 NaN）
# 这是一种"类型提升"行为

# 方案
df['id'] = df['id'].astype('Int64')                # 大写 I — pandas 扩展类型，支持 NaN
df = pd.read_csv('file.csv', dtype={'id': 'Int64'})   # 读取时直接指定

# 现象
orders = pd.DataFrame({'product': ['A', 'B', 'C'], 'price': [10, 20, 30]})
subset = orders[orders['product'] != 'C']   # 取了前两行
subset['price'] = subset['price'] * 2       # 想改 subset
print(orders['price'].values)               # ⚠️ 可能也变了（如果是视图）

# 为什么会发生？
# 布尔索引返回的可能是原始数据的视图（取决于 pandas 内部优化）
# 视图本身是原始数据的一段内存指针，改视图可能改到原数据

# 方案：对子集操作前明确复制
subset = orders[orders['product'] != 'C'].copy()   # ✅ 明确是副本
subset['price'] = subset['price'] * 2               # ✅ 只改副本，不影响原数据

# 现象
df['Category'] = df['Category'].str.strip().str.lower()  # 正常
# 但某个操作后，数据看起来变了，写入文件时发现没变

# 方案：每做完一步检查一下类型
print(df['Category'].dtype)            # 确认类型
print(df[['Category']].head())         # 确认内容

pd.read_csv(
    path,

    # ── 文件结构 ──
    sep=',',            # 分隔符（CSV 用逗号，TSV 用 '\t'）
    header=0,           # 表头在第几行（按 0 计）
    names=None,         # 自定义列名（配合 header=0 可跳过原表头）
    index_col=None,     # 指定某列作为行索引
    skiprows=[],        # 跳过前 N 行或指定行
    nrows=None,         # 只读前 N 行（测试数据时有用）

    # ── 类型控制（最易出错的区域）──
    dtype={},           # 强制指定列类型：dtype={'id': 'Int64', 'price': 'float64'}
    parse_dates=[],     # 需要解析为日期的列：parse_dates=['Date', 'CreatedAt']
    converters={},      # 自定义解析函数：converters={'col': lambda x: x.strip()}

    # ── 缺失值处理 ──
    na_values=[],       # 指定哪些值视为 NaN：na_values=['', 'NA', 'null', 'N/A']
    keep_default_na=True,  # 是否保留默认的 NaN 标记

    # ── 容错 ──
    on_bad_lines='warn',    # 'error'（抛异常，默认）| 'warn'（警告并跳过）| 'skip'（静默跳过）
    encoding='utf-8',       # 文件编码
    encoding_errors='strict',  # 'strict' | 'ignore' | 'replace'

    # ── 性能 ──
    usecols=[],         # 只读指定列，节省内存
    chunksize=None,     # 分块读取整型，每次返回迭代器
)