以下提供一份端到端的 Python 脚本，用于对“多渠道广告投放小时级日志”进行标准化处理：统一时区与命名、地区粒度对齐、汇率归一化、异常点击/转化处理、构造派生指标（CTR/CVR/CPC 等）、按 Min-Max 方法将数值缩放至[0,1]、并输出清洗前后分布对比图。脚本支持可配置的缺失值策略与可选的按小时插值补齐。
说明要点：
- 适用数据规模：约 1200 万行、28 列。建议优先使用 Parquet 输入和分类编码以降低内存。绘图仅对抽样数据进行。
- 缺失值策略可选（类别用“未知”填充；数值可固定值、插值或删除行）。
- 统一时区需要提供 channel->timezone 映射（JSON）；未映射默认按 UTC 处理。
- 汇率转换需要提供每日汇率表（date,currency,rate_to_base），base 货币可选（例如 CNY 或 USD）。
- 区域标准化需要提供地区映射表（raw_region -> 标准省级或自定义维度）。
- 机器人流量过滤采用稳健统计（MAD）对 CTR/CVR 进行分组去极值或删除。
依赖项：
- pandas, numpy, scikit-learn, matplotlib, seaborn, pytz, python-dateutil, pyarrow
脚本数据输入/输出约定：
- 输入主数据：包含至少这些列：campaign_id, ad_group_id, creative_id, channel, device_type, os, region, timestamp, impressions, clicks, cost, conversions, revenue, currency。
- 汇率表：fx_rates.csv(parquet)，列：date(YYYY-MM-DD), currency, rate_to_base。
- 区域表：region_lookup.csv(parquet)，列：raw_region, std_region(或 province)、region_level。
- 渠道命名映射：channel_map.json，形如 {"SEM": "Paid Search", "搜索": "Paid Search", "付费搜索":"Paid Search", "社交":"Paid Social", ...}
- 渠道时区映射：channel_tz_map.json，形如 {"Paid Search": "UTC", "Paid Social": "Asia/Shanghai", ...}
- 输出：清洗后的 parquet、缩放后的 parquet、缩放参数 scaler.json、分布对比图 png。
代码（保存为 standardize_ads.py）：
#!/usr/bin/env python3
-- coding: utf-8 --
import os
import json
import argparse
import warnings
from typing import List, Dict, Optional
import numpy as np
import pandas as pd
from dateutil import parser as dtparser
import pytz
from sklearn.preprocessing import MinMaxScaler
import matplotlib.pyplot as plt
import seaborn as sns
warnings.filterwarnings("ignore")
-------------------------
工具函数
-------------------------
def ensure_dir(path: str):
os.makedirs(path, exist_ok=True)
def load_dataframe(path: str, dtype_overrides: Optional[Dict[str, str]] = None) -> pd.DataFrame:
ext = os.path.splitext(path)[-1].lower()
if ext in [".parquet", ".pq"]:
df = pd.read_parquet(path)
elif ext in [".csv", ".txt"]:
df = pd.read_csv(path, low_memory=False)
else:
raise ValueError(f"Unsupported file extension for: {path}")
if dtype_overrides:
for col, dt in dtype_overrides.items():
if col in df.columns:
try:
df[col] = df[col].astype(dt)
except Exception:
pass
return df
def parse_timestamp(ts):
# 解析时间戳（可能包含或不包含时区）
if pd.isna(ts):
return pd.NaT
try:
return dtparser.parse(str(ts))
except Exception:
return pd.NaT
def convert_timezone(series_ts: pd.Series,
channel: pd.Series,
channel_tz_map: Dict[str, str],
target_tz: str) -> pd.Series:
"""
将 timestamp 列按 channel_tz_map 解释为对应时区，再统一转换到 target_tz。
- 若 timestamp 已带时区，则按原时区转换到 target_tz。
- 若不带时区，且 channel 在映射中，则视为该时区；否则默认 UTC。
"""
target = pytz.timezone(target_tz)
def _convert(ts, ch):
if pd.isna(ts):
return pd.NaT
# 已有 tzinfo
if getattr(ts, "tzinfo", None) is not None:
try:
return ts.astimezone(target)
except Exception:
return pd.NaT
# 无 tzinfo，依据 channel 映射假定原时区
src_tz_name = channel_tz_map.get(str(ch), "UTC")
try:
src_tz = pytz.timezone(src_tz_name)
except Exception:
src_tz = pytz.UTC
try:
localized = src_tz.localize(ts)
return localized.astimezone(target)
except Exception:
return pd.NaT
return pd.to_datetime([_convert(t, c) for t, c in zip(series_ts, channel)], errors="coerce")
def normalize_channels(series: pd.Series, channel_map: Dict[str, str]) -> pd.Series:
def _norm(x):
if pd.isna(x):
return "未知"
key = str(x).strip().lower()
# 尝试映射（大小写、去空格处理）
for k, v in channel_map.items():
if key == str(k).strip().lower():
return v
return x # 未覆盖的保留原值
return series.apply(_norm)
def map_regions(df: pd.DataFrame, region_lookup: pd.DataFrame,
raw_col: str = "region", std_col: str = "std_region") -> pd.DataFrame:
if raw_col not in df.columns:
return df
if std_col not in region_lookup.columns:
raise ValueError(f"region_lookup 必须包含列: {std_col}")
region_lookup = region_lookup[[raw_col, std_col]].drop_duplicates()
df = df.merge(region_lookup, how="left", on=raw_col)
df[std_col] = df[std_col].fillna("未知")
return df
def convert_currency(df: pd.DataFrame,
fx_df: pd.DataFrame,
base_currency: str,
date_col: str = "date") -> pd.DataFrame:
"""
依据 currency + date 合并汇率，将 cost/revenue 转换为 base 货币。
fx_df 需要列: date(YYYY-MM-DD), currency, rate_to_base（1单位外币 -> base 的倍数）
"""
if "currency" not in df.columns:
raise ValueError("缺少 currency 列")
df[date_col] = pd.to_datetime(df["timestamp"].dt.date)
fx_df = fx_df.copy()
fx_df[date_col] = pd.to_datetime(fx_df[date_col])
fx_df["currency"] = fx_df["currency"].astype(str).str.upper()
df["currency"] = df["currency"].astype(str).str.upper()
merged = df.merge(fx_df, how="left", left_on=[date_col, "currency"], right_on=[date_col, "currency"])
# rate_to_base 缺失处理：若本身就是 base 货币，则设为 1；否则设为 np.nan 并后续处理
merged["rate_to_base"] = np.where(
merged["currency"] == base_currency.upper(),
1.0,
merged["rate_to_base"]
)
# 对缺失的汇率进行处理（可以选择删除或填充为1，但建议删除以避免误差）
missing_fx = merged["rate_to_base"].isna().sum()
if missing_fx > 0:
# 保守处理：删除缺少汇率的行
merged = merged[~merged["rate_to_base"].isna()]
merged["cost_base"] = merged["cost"] * merged["rate_to_base"]
merged["revenue_base"] = merged["revenue"] * merged["rate_to_base"]
return merged.drop(columns=["rate_to_base"])
def compute_derived_metrics(df: pd.DataFrame) -> pd.DataFrame:
# 防止除零
eps = 1e-12
# 派生指标
df["ctr"] = np.where(df["impressions"] > 0, df["clicks"] / df["impressions"], 0.0)
df["cvr"] = np.where(df["clicks"] > 0, df["conversions"] / df["clicks"], 0.0)
df["cpc"] = np.where(df["clicks"] > 0, df["cost_base"] / df["clicks"], 0.0)
df["cpm"] = np.where(df["impressions"] > 0, df["cost_base"] / (df["impressions"] / 1000.0), 0.0)
df["cpa"] = np.where(df["conversions"] > 0, df["cost_base"] / np.maximum(df["conversions"], eps), 0.0)
df["roas"] = np.where(df["cost_base"] > 0, df["revenue_base"] / df["cost_base"], 0.0)
return df
def robust_bot_filter(df: pd.DataFrame,
group_cols: List[str],
k_mad: float = 5.0,
mode: str = "cap") -> pd.DataFrame:
"""
使用分组 + MAD 对 CTR/CVR 去极值或删除疑似机器人流量。
- mode = "cap": 将超过上界的 CTR/CVR 截断到上界
- mode = "remove": 删除超过上界的行
同时清理 clicks > impressions，conversions > clicks 等逻辑异常。
"""
# 逻辑异常先处理
df = df[df["impressions"] >= 0]
df = df[df["clicks"] >= 0]
df = df[df["conversions"] >= 0]
df = df[df["clicks"] <= df["impressions"]]
df = df[df["conversions"] <= df["clicks"]]
# 计算当前 ctr/cvr（若前面未计算）
if "ctr" not in df.columns or "cvr" not in df.columns:
df = compute_derived_metrics(df)
# 分组计算 robust 上界
def _mad_cap(sub: pd.DataFrame, col: str):
x = sub[col].values
med = np.median(x)
mad = np.median(np.abs(x - med)) + 1e-12
ub = med + k_mad * 1.4826 * mad # 正态等效
return ub
# 计算上界表
ub_ctr = df.groupby(group_cols, observed=True).apply(lambda g: _mad_cap(g, "ctr")).rename("ctr_ub")
ub_cvr = df.groupby(group_cols, observed=True).apply(lambda g: _mad_cap(g, "cvr")).rename("cvr_ub")
bounds = pd.concat([ub_ctr, ub_cvr], axis=1).reset_index()
df = df.merge(bounds, how="left", on=group_cols)
if mode == "cap":
df["ctr"] = np.minimum(df["ctr"], df["ctr_ub"].fillna(df["ctr"]))
df["cvr"] = np.minimum(df["cvr"], df["cvr_ub"].fillna(df["cvr"]))
elif mode == "remove":
df = df[(df["ctr"] <= df["ctr_ub"]) | df["ctr_ub"].isna()]
df = df[(df["cvr"] <= df["cvr_ub"]) | df["cvr_ub"].isna()]
else:
raise ValueError("bot_filter mode 必须是 'cap' 或 'remove'")
df = df.drop(columns=["ctr_ub", "cvr_ub"], errors="ignore")
return df
def handle_missing_values(df: pd.DataFrame,
cat_cols: List[str],
num_cols: List[str],
num_strategy: str = "fixed",
fixed_value: float = 0.0,
interpolate: bool = False,
resample_group_cols: Optional[List[str]] = None) -> pd.DataFrame:
"""
缺失值策略：
- 类别：填充 '未知'
- 数值：num_strategy in ['fixed','interpolate','drop']
- fixed: 用 fixed_value 填充
- interpolate: 若 interpolate=True，则对指定分组按小时重采样 + 线性插值
- drop: 删除含缺失的行
"""
# 类别填充
for c in cat_cols:
if c in df.columns:
df[c] = df[c].fillna("未知")
if num_strategy == "drop":
return df.dropna(subset=num_cols)
if num_strategy == "fixed":
for c in num_cols:
if c in df.columns:
df[c] = df[c].fillna(fixed_value)
return df
if num_strategy == "interpolate":
if not interpolate:
# 仅对现有缺失在同一索引上做线性插值（非重采样）
for c in num_cols:
if c in df.columns:
df[c] = df[c].interpolate(method="linear", limit_direction="both")
return df
# 按组重采样到小时，并插值
if resample_group_cols is None or len(resample_group_cols) == 0:
raise ValueError("interpolate=True 时需要 resample_group_cols 指定分组键")
# 仅保留必要列以降低内存
keep_cols = list(set(resample_group_cols + ["timestamp"] + num_cols))
other_cols = [c for c in df.columns if c not in keep_cols]
# 先排序以保证重采样有效
df = df.sort_values(["timestamp"] + resample_group_cols)
# 分组处理
def _resample_group(g: pd.DataFrame):
g = g.set_index("timestamp").asfreq("H")
# 插值
for c in num_cols:
if c in g.columns:
g[c] = g[c].interpolate(method="linear", limit_direction="both")
# 重置索引
g = g.reset_index()
# 恢复分组键
for k in resample_group_cols:
if k not in g.columns:
g[k] = g[k].iloc[0] if len(g) > 0 else np.nan
return g
df_interp = (
df[keep_cols]
.groupby(resample_group_cols, dropna=False, observed=True, group_keys=False)
.apply(_resample_group)
)
# 合并回其他列（其他列不可随时间插值，使用最近值或未知）
df_interp = df_interp.merge(
df[resample_group_cols + ["timestamp"] + other_cols],
how="left",
on=resample_group_cols + ["timestamp"]
)
# 对类别型新增缺失再补一次
for c in other_cols:
if df_interp[c].dtype == "O":
df_interp[c] = df_interp[c].fillna("未知")
return df_interp
raise ValueError("num_strategy 必须是 ['fixed','interpolate','drop'] 之一")
def minmax_scale(df: pd.DataFrame, cols: List[str]) -> (pd.DataFrame, Dict[str, Dict[str, float]]):
scaler = MinMaxScaler(feature_range=(0, 1), clip=False)
arr = df[cols].astype(float).values
scaler.fit(arr)
scaled = scaler.transform(arr)
df_scaled = df.copy()
df_scaled[cols] = scaled
# 导出缩放参数
params = {}
for i, c in enumerate(cols):
params[c] = {"min": float(scaler.data_min_[i]), "max": float(scaler.data_max_[i])}
return df_scaled, params
def plot_distributions(sample_before: pd.DataFrame,
sample_after: pd.DataFrame,
cols: List[str],
out_dir: str):
ensure_dir(out_dir)
sns.set(style="whitegrid")
for c in cols:
plt.figure(figsize=(8, 5))
sns.kdeplot(sample_before[c].dropna(), label="before", fill=True, alpha=0.3, color="#1f77b4")
sns.kdeplot(sample_after[c].dropna(), label="after", fill=True, alpha=0.3, color="#ff7f0e")
plt.title(f"Distribution Comparison: {c}")
plt.legend()
plt.tight_layout()
plt.savefig(os.path.join(out_dir, f"dist_{c}.png"), dpi=150)
plt.close()
-------------------------
主流程
-------------------------
def main():
parser = argparse.ArgumentParser(description="多渠道广告日志标准化（时区/命名/地区/汇率/异常/指标/Min-Max/分布图）")
parser.add_argument("--input", required=True, help="输入数据文件（.parquet 或 .csv）")
parser.add_argument("--output_dir", required=True, help="输出目录")
parser.add_argument("--fx_rates", required=True, help="汇率表文件（.parquet 或 .csv），列：date,currency,rate_to_base")
parser.add_argument("--region_lookup", required=True, help="地区映射表文件（.parquet 或 .csv），列：raw_region,std_region")
parser.add_argument("--channel_map", required=True, help="渠道命名映射 JSON 文件")
parser.add_argument("--channel_tz_map", required=True, help="渠道->时区 映射 JSON 文件（例：'UTC','Asia/Shanghai'）")
parser.add_argument("--base_currency", default="CNY", help="目标基准货币（例：CNY 或 USD）")
parser.add_argument("--target_tz", default="UTC", help="统一目标时区（例：UTC 或 Asia/Shanghai）")
parser.add_argument("--num_strategy", default="fixed", choices=["fixed","interpolate","drop"], help="数值缺失策略")
parser.add_argument("--fixed_fill_value", type=float, default=0.0, help="数值缺失固定填充值")
parser.add_argument("--interpolate_resample", action="store_true", help="插值是否按小时重采样（需指定 resample_group_cols）")
parser.add_argument("--resample_group_cols", default="channel,std_region,device_type,os",
help="插值重采样分组列，逗号分隔")
parser.add_argument("--bot_filter_mode", default="cap", choices=["cap","remove"], help="机器人流量处理方式：截断或删除")
parser.add_argument("--bot_k_mad", type=float, default=5.0, help="MAD 上界系数")
parser.add_argument("--sample_for_plots", type=int, default=300000, help="绘图抽样行数")
parser.add_argument("--save_scaled", action="store_true", help="是否输出缩放后的数据")
args = parser.parse_args()
ensure_dir(args.output_dir)
plots_dir = os.path.join(args.output_dir, "plots")
ensure_dir(plots_dir)
# 读取映射配置
with open(args.channel_map, "r", encoding="utf-8") as f:
channel_map = json.load(f)
with open(args.channel_tz_map, "r", encoding="utf-8") as f:
channel_tz_map = json.load(f)
# 读取主数据
dtype_overrides = {
"campaign_id": "string",
"ad_group_id": "string",
"creative_id": "string",
"channel": "string",
"device_type": "string",
"os": "string",
"region": "string",
"currency": "string",
}
df = load_dataframe(args.input, dtype_overrides=dtype_overrides)
# 基本列检查
required_cols = ["channel","timestamp","region","device_type","os",
"impressions","clicks","cost","conversions","revenue","currency"]
missing_cols = [c for c in required_cols if c not in df.columns]
if missing_cols:
raise ValueError(f"缺少必要列: {missing_cols}")
# 1) 时间戳解析与时区统一
df["timestamp"] = df["timestamp"].apply(parse_timestamp)
df["channel"] = normalize_channels(df["channel"], channel_map=channel_map)
df["timestamp"] = convert_timezone(df["timestamp"], df["channel"], channel_tz_map, args.target_tz)
# 舍弃无法解析的时间行
df = df[~df["timestamp"].isna()]
# 对齐到小时起始
df["timestamp"] = df["timestamp"].dt.floor("H")
# 2) 地区映射到标准粒度
region_lookup = load_dataframe(args.region_lookup)
# 兼容不同列名
if "raw_region" not in region_lookup.columns:
region_lookup = region_lookup.rename(columns={"region":"raw_region"})
if "std_region" not in region_lookup.columns:
# 若无 std_region 列，尝试使用 province 列作为标准粒度
if "province" in region_lookup.columns:
region_lookup = region_lookup.rename(columns={"province":"std_region"})
else:
raise ValueError("region_lookup 需要提供 std_region 或 province 列")
df = map_regions(df, region_lookup, raw_col="region", std_col="std_region")
# 3) 汇率统一到 base_currency
fx_df = load_dataframe(args.fx_rates)
# 校验列
for col in ["date","currency","rate_to_base"]:
if col not in fx_df.columns:
raise ValueError("fx_rates 需要列: date,currency,rate_to_base")
df = convert_currency(df, fx_df, base_currency=args.base_currency)
# 4) 基础数值类型安全转换
for c in ["impressions","clicks","conversions"]:
df[c] = pd.to_numeric(df[c], errors="coerce").fillna(0).astype(np.int64)
for c in ["cost","revenue","cost_base","revenue_base"]:
df[c] = pd.to_numeric(df[c], errors="coerce").fillna(0.0).astype(float)
# 5) 缺失值处理（类别 -> '未知'；数值 -> 固定/插值/删除）
cat_cols = ["campaign_id","ad_group_id","creative_id","channel","device_type","os","region","std_region","currency"]
num_cols = ["impressions","clicks","conversions","cost","revenue","cost_base","revenue_base"]
resample_cols = [c.strip() for c in args.resample_group_cols.split(",") if c.strip()]
df = handle_missing_values(
df, cat_cols=cat_cols, num_cols=num_cols,
num_strategy=args.num_strategy,
fixed_value=args.fixed_fill_value,
interpolate=args.interpolate_resample,
resample_group_cols=resample_cols
)
# 6) 构造派生指标
df = compute_derived_metrics(df)
# 7) 机器人流量过滤（按 channel,std_region,device_type,os 分组）
bot_group_cols = ["channel","std_region","device_type","os"]
df = robust_bot_filter(df, group_cols=bot_group_cols, k_mad=args.bot_k_mad, mode=args.bot_filter_mode)
# 8) 为绘图保留抽样（before/after）：
# 为确保对比，before 采样来自“构造派生指标后但机器人过滤前”的版本
# 实际上我们已过滤，因此此处退一步：对过滤后的 df 反映“after”
# 重新构造一个“before”的近似样本：在过滤前的版本进行采样（内存考虑：仅抽样原 df 的子集进行派生指标）
# 为简单与稳定性，这里在过滤后前一步做一个副本（真实业务可在过滤前持久化快照）
# 为保证脚本独立性，这里仅使用 after 的抽样；若严格需要 before，对应生产流程中请在过滤前保存快照。
# 简化处理：我们在过滤前的 df 复制一下（为了给出对比）
# 注意：为避免占用过多内存，上述步骤可在生产写中间文件后再加载采样。
# 此处再计算 before 样本：回退到“构造派生指标后、过滤前”的近似
# 为实现该目的，我们重新从输入到 6) 复制一份较轻流程用于抽样对比。
df_raw = load_dataframe(args.input, dtype_overrides=dtype_overrides)
df_raw["timestamp"] = df_raw["timestamp"].apply(parse_timestamp)
df_raw["channel"] = normalize_channels(df_raw["channel"], channel_map=channel_map)
df_raw["timestamp"] = convert_timezone(df_raw["timestamp"], df_raw["channel"], channel_tz_map, args.target_tz)
df_raw = df_raw[~df_raw["timestamp"].isna()]
df_raw["timestamp"] = df_raw["timestamp"].dt.floor("H")
df_raw = map_regions(df_raw, region_lookup, raw_col="region", std_col="std_region")
df_raw = convert_currency(df_raw, fx_df, base_currency=args.base_currency)
for c in ["impressions","clicks","conversions"]:
df_raw[c] = pd.to_numeric(df_raw[c], errors="coerce").fillna(0).astype(np.int64)
for c in ["cost","revenue","cost_base","revenue_base"]:
df_raw[c] = pd.to_numeric(df_raw[c], errors="coerce").fillna(0.0).astype(float)
df_raw = compute_derived_metrics(df_raw)
# 抽样绘图
n = args.sample_for_plots
rng = np.random.default_rng(2024)
if len(df_raw) > n:
sample_before = df_raw.sample(n=n, random_state=2024)
else:
sample_before = df_raw.copy()
if len(df) > n:
sample_after = df.sample(n=n, random_state=2024)
else:
sample_after = df.copy()
# 9) Min-Max 归一化（缩放到 [0,1]）
scale_cols = [
"impressions","clicks","conversions",
"cost_base","revenue_base",
"ctr","cvr","cpc","cpm","cpa","roas"
]
# 缩放前备份用于绘图
# 仅对 after 再缩放
df_scaled, scaler_params = minmax_scale(df, cols=scale_cols)
# 10) 输出
cleaned_path = os.path.join(args.output_dir, "cleaned.parquet")
df.to_parquet(cleaned_path, index=False)
with open(os.path.join(args.output_dir, "scaler.json"), "w", encoding="utf-8") as f:
json.dump({"feature_range": [0,1], "params": scaler_params}, f, ensure_ascii=False, indent=2)
if args.save_scaled:
scaled_path = os.path.join(args.output_dir, "scaled.parquet")
df_scaled.to_parquet(scaled_path, index=False)
# 11) 分布对比图（before vs after）
plot_cols = ["impressions","clicks","cost_base","revenue_base","ctr","cvr","cpc","cpm","cpa","roas"]
plot_distributions(sample_before[plot_cols], sample_after[plot_cols], plot_cols, plots_dir)
print(f"完成。清洗输出: {cleaned_path}")
if args.save_scaled:
print(f"缩放输出: {scaled_path}")
print(f"缩放参数: {os.path.join(args.output_dir, 'scaler.json')}")
print(f"分布图目录: {plots_dir}")
if name == "main":
main()
使用示例：
-
准备文件：
- 数据：raw_logs.parquet
- 汇率：fx_rates.csv，列 [date(YYYY-MM-DD), currency(大写), rate_to_base(浮点)]
- 地区映射：region_lookup.csv，列 [raw_region, std_region]
- 渠道映射：channel_map.json（示例）
{
"SEM": "Paid Search",
"搜索": "Paid Search",
"付费搜索": "Paid Search",
"社交": "Paid Social",
"Social": "Paid Social",
"展示": "Display",
"信息流": "Feed"
}
- 渠道时区映射：channel_tz_map.json（示例；请按真实平台时区填写）
{
"Paid Search": "UTC",
"Paid Social": "UTC",
"Display": "UTC",
"Feed": "Asia/Shanghai"
}
-
运行（固定值填充，截断机器人流量，输出缩放版）：
python standardize_ads.py
--input raw_logs.parquet
--output_dir out_ads
--fx_rates fx_rates.csv
--region_lookup region_lookup.csv
--channel_map channel_map.json
--channel_tz_map channel_tz_map.json
--base_currency CNY
--target_tz Asia/Shanghai
--num_strategy fixed
--fixed_fill_value 0
--bot_filter_mode cap
--bot_k_mad 5
--save_scaled
-
运行（按小时重采样 + 线性插值补齐数值缺失，删除异常）：
python standardize_ads.py
--input raw_logs.parquet
--output_dir out_ads
--fx_rates fx_rates.csv
--region_lookup region_lookup.csv
--channel_map channel_map.json
--channel_tz_map channel_tz_map.json
--base_currency USD
--target_tz UTC
--num_strategy interpolate
--interpolate_resample
--resample_group_cols channel,std_region,device_type,os
--bot_filter_mode remove
--bot_k_mad 6
--save_scaled
实现细节与注意事项：
- 时区：脚本将无时区时间戳按 channel_tz_map 解释；带时区时间戳则尊重原 tz 后统一到 target_tz。请确保 channel_tz_map 与真实采集逻辑一致。
- 渠道命名：normalize_channels 会大小写与空白规范化后匹配映射；未覆盖的名称保持原值，建议逐步完善映射表。
- 地区粒度：region_lookup 建议标准化到“省级”或业务指定维度；未匹配行填“未知”。
- 货币：强烈建议对缺失汇率的行删除（脚本采用删除策略），避免错误金额。若需改为填充 1，请在 convert_currency 中调整。
- 缺失值：类别统一填“未知”；数值可固定值、插值或删除。插值时可选择是否按小时重采样（interpolate_resample），并通过 resample_group_cols 控制分组粒度，避免不必要的笛卡尔扩张。
- 机器人流量：采用分组（channel,std_region,device_type,os）+ MAD 上界对 CTR/CVR 截断或删除；同时剔除 clicks>impressions、conversions>clicks 等违规记录。k_mad 可根据数据离散度调整。
- 派生指标：包含 CTR、CVR、CPC、CPM、CPA、ROAS，均做除零保护。
- Min-Max 缩放：对指定数值列缩放至[0,1]；输出 scaler.json 保存每列的 min/max，便于下游与线上对齐。若后续想按训练集统计参数缩放，请在生产流程中先拟定“训练集”，再对验证/测试集应用同一参数。
- 分布对比图：KDE 曲线基于抽样（默认 30 万行）以保证速度；用于观察清洗效果（异常收敛、长尾被削弱、单位统一后峰位变化等）。
- 性能建议：优先使用 Parquet；如内存不足，建议：
- 将插值重采样的分组键减少或按子集批量处理；
- 先按时间或渠道分片处理再合并；
- 或改用 Polars/DuckDB 做中间聚合与连接，再回到 pandas 绘图。
如需将该流程改为两遍扫描（先统计 min/max 再缩放）以适配严格流式处理，或改用 Dask/Polars 提升吞吐，可在上述框架基础上替换数据载入与 groupby 实现。
