import pandas as pd
import numpy as np

# 新手做法：直接讀取，導(dǎo)致內(nèi)存撐爆或類型推斷錯(cuò)誤
# df = pd.read_csv('huge_orders.csv')

# 專業(yè)做法：精準(zhǔn)控制列、類型與分塊
# 1. 使用 usecols 只讀需要的列
# 2. 使用 dtype 預(yù)先指定類型，避免 object 和 int64 的內(nèi)存浪費(fèi)
# 3. 使用 parse_dates 直接解析時(shí)間
dtypes = {
    'order_id': 'int32',
    'user_id': 'int32',
    'status': 'category',  # 低基數(shù)枚舉字段使用 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:
    # 在每個(gè) chunk 上進(jìn)行聚合操作，最后合并結(jié)果
    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:
    """數(shù)據(jù)質(zhì)量診斷函數(shù)"""
    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)

# 調(diào)用
diagnose_dataframe(df)

# 查看數(shù)值型統(tǒng)計(jì)量，并調(diào)整顯示精度
pd.set_option('display.float_format', lambda x: '%.2f' % x)
print(df.describe(percentiles=[.05, .25, .5, .75, .95, .99])) # 關(guān)注長尾分布

# 查看類別型字段的基數(shù)和Top頻率
print(df.describe(include=['category', 'object']))

# 1. 業(yè)務(wù)規(guī)則填充：例如“退款金額”缺失代表未退款，填0
df['refund_amount'] = df['refund_amount'].fillna(0)

# 2. 分組填充：用該用戶的歷史均值填充缺失的評(píng)分
df['user_score'] = df.groupby('user_id')['user_score'].transform(
    lambda x: x.fillna(x.mean())
)

# 3. 時(shí)間序列插值：對(duì)于連續(xù)的傳感器/財(cái)務(wù)數(shù)據(jù)，使用線性或樣條插值
df['daily_revenue'] = df['daily_revenue'].interpolate(method='spline', order=2)

def clip_outliers_iqr(series: pd.Series, factor=1.5) -> pd.Series:
    """基于 IQR 的異常值截?cái)啵ㄉw帽法）"""
    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'])

# 假設(shè) df['remark'] 包含類似 "使用優(yōu)惠券:VIP2026, 免郵" 的文本
# ? 慢速做法：df['remark'].apply(lambda x: re.search(...))
# ? 向量化做法：
df['coupon_code'] = df['remark'].str.extract(r'優(yōu)惠券:([A-Z0-9]+)')

# 清理隱藏字符并統(tǒng)一小寫
df['city'] = df['city'].str.strip().str.lower().str.replace(r'\s+', ' ', regex=True)

# 篩選：高價(jià)值、已完成、特定城市的訂單
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()

# 計(jì)算 RFM 模型中的時(shí)間間隔特征
user_stats['recency_days'] = (pd.Timestamp('2026-05-27') - user_stats['last_order']).dt.days

# 寬轉(zhuǎn)長：將 Jan, Feb, Mar 三列銷售額轉(zhuǎn)為一列
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'
)

# 長轉(zhuǎn)寬：生成 地區(qū) x 月份 的交叉報(bào)表
report = pd.pivot_table(
    df_long, 
    values='sales', 
    index='region', 
    columns='month', 
    aggfunc='sum', 
    fill_value=0,
    margins=True # 添加總計(jì)行/列
)

# ? 務(wù)必使用 validate 參數(shù)驗(yàn)證關(guān)聯(lián)鍵的唯一性
# 'm:1' 表示左表多對(duì)一右表，如果右表 key 不唯一，直接報(bào)錯(cuò)攔截！
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')

# 按日重采樣，計(jì)算每日總GMV和訂單數(shù)
daily_report = df_time.resample('D').agg(
    gmv=('amount', 'sum'),
    orders=('order_id', 'count')
).fillna(0) # 補(bǔ)齊沒有交易的日期

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

# 2. 計(jì)算日環(huán)比增長率 (今日 / 昨日 - 1)
daily_report['gmv_dod'] = daily_report['gmv'].pct_change(periods=1)

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

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

# 或者在讀取時(shí)指定
df = pd.read_csv('data.csv', engine='pyarrow', dtype_backend='pyarrow')
# 此時(shí)字符串不再是低效的 object，而是 string[pyarrow]，內(nèi)存直降 70%！

# 開啟 CoW
pd.options.mode.copy_on_write = True

# 現(xiàn)在，切片操作是懶加載的（視圖），只有當(dāng)你真正修改切片數(shù)據(jù)時(shí)，才會(huì)發(fā)生內(nèi)存拷貝。
subset = df[df['amount'] > 100]
subset['new_col'] = 1  # 安全，不會(huì)觸發(fā)警告，也不會(huì)意外修改原 df

# 場景：根據(jù)金額劃分等級(jí) (低、中、高)

# ? 極慢的 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 (適用于連續(xù)數(shù)值分箱)
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]

# 一氣呵成的數(shù)據(jù)流水線
final_df = (
    pd.read_csv('raw_data.csv', parse_dates=['time'])
    .query("status == 'success'")
    .assign(
        # 在鏈中直接創(chuàng)建新列
        revenue = lambda x: x['price'] * x['qty'],
        year_month = lambda x: x['time'].dt.to_period('M')
    )
    .pipe(drop_outliers, col='revenue')  # 接入自定義函數(shù)
    .groupby('year_month')
    .agg(total_rev=('revenue', 'sum'))
    .reset_index()
)