下面提供一个可运行的 Python 脚本，用于对住院就诊记录（patient_visits.parquet）执行缺失值清理，满足以下要求：

• 避免目标泄露：任何插补不使用 readmission_30d。
• 保留主键列与诊断列：visit_id、patient_id、diagnosis_code 不会被删除。
• 支持策略：KNN 插补、模型预测插补、数值中位数、类别众数、分组前向填充、按阈值删除高缺失列。
• 处理哨兵值：将 lab_*（如 lab_glucose、lab_creatinine）中的 0 和 -1 视为缺失。
• 拥有命令行参数、日志、随机种子、可复现实验配置，输出清洗摘要与字段字典。

使用说明：

• 依赖：pandas, numpy, pyarrow, scikit-learn, joblib
• 示例： python clean_patient_visits.py --input patient_visits.parquet --output-dir ./cleaned --seed 42 --enable-knn --ffill-cols BMI,smoker_status --drop-threshold 0.95 --model-impute-cols "" 若使用模型插补某些列（如 BMI）：--model-impute-cols BMI

代码（保存为 clean_patient_visits.py）：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
清理住院就诊记录缺失值的脚本

功能概要：
- 加载 Parquet 数据
- 统一类型与基础清洗（去重、范围校验、哨兵值处理）
- 按患者-时间顺序对选定列进行前向填充（避免目标泄露）
- 可选 KNN 插补数值列（不含目标、键、日期等）
- 数值列用中位数兜底，类别列用众数兜底
- 可选对指定列执行模型预测插补（RandomForest），特征不包含 readmission_30d
- 可选删除高缺失列（保护 visit_id, patient_id, diagnosis_code, admission_date, discharge_date, readmission_30d）
- 输出：清洗后的数据、清洗摘要（JSON/CSV）、字段字典（JSON/CSV）、生效配置（JSON）、日志

运行示例：
python clean_patient_visits.py \
  --input patient_visits.parquet \
  --output-dir ./cleaned \
  --seed 42 \
  --enable-knn \
  --n-neighbors 5 \
  --ffill-cols BMI,smoker_status \
  --drop-threshold 0.95 \
  --model-impute-cols "" \
  --save-csv 0

注意：
- 脚本不会在任何插补步骤中使用 readmission_30d 作为特征（避免目标泄露）
"""

import argparse
import json
import logging
import os
import sys
import time
import random
from datetime import datetime

import numpy as np
import pandas as pd

from sklearn.impute import KNNImputer, SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.ensemble import RandomForestRegressor, RandomForestClassifier
from sklearn.utils.validation import check_is_fitted

# ------------------------------
# 实用函数
# ------------------------------

def setup_logging(output_dir: str, log_level: str = "INFO"):
    os.makedirs(output_dir, exist_ok=True)
    log_path = os.path.join(output_dir, f"cleaning_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log")

    logger = logging.getLogger()
    logger.setLevel(getattr(logging, log_level.upper(), logging.INFO))

    fmt = logging.Formatter("[%(asctime)s] %(levelname)s - %(message)s")

    fh = logging.FileHandler(log_path, encoding="utf-8")
    fh.setLevel(getattr(logging, log_level.upper(), logging.INFO))
    fh.setFormatter(fmt)
    logger.addHandler(fh)

    sh = logging.StreamHandler(sys.stdout)
    sh.setLevel(getattr(logging, log_level.upper(), logging.INFO))
    sh.setFormatter(fmt)
    logger.addHandler(sh)

    logging.info(f"日志输出：{log_path}")


def set_global_seed(seed: int = 42):
    np.random.seed(seed)
    random.seed(seed)
    try:
        import torch  # noqa
        import torch.backends.cudnn as cudnn  # noqa
        torch.manual_seed(seed)
        torch.cuda.manual_seed_all(seed)
        cudnn.deterministic = True
        cudnn.benchmark = False
    except Exception:
        pass


def load_parquet(path: str) -> pd.DataFrame:
    try:
        df = pd.read_parquet(path, engine="pyarrow")
    except Exception as e:
        logging.warning(f"pyarrow 读取失败，尝试 fastparquet。错误：{e}")
        try:
            df = pd.read_parquet(path, engine="fastparquet")
        except Exception as ee:
            logging.error(f"无法读取 Parquet 文件：{ee}")
            raise
    return df


def to_datetime_safe(df: pd.DataFrame, cols):
    for c in cols:
        if c in df.columns:
            df[c] = pd.to_datetime(df[c], errors="coerce")
    return df


def compute_missing_rates(df: pd.DataFrame) -> pd.Series:
    return df.isna().mean().sort_values(ascending=False)


def detect_column_roles(df: pd.DataFrame):
    # 预定义角色
    key_cols = [c for c in ["visit_id", "patient_id"] if c in df.columns]
    target_col = "readmission_30d" if "readmission_30d" in df.columns else None
    date_cols = [c for c in ["admission_date", "discharge_date"] if c in df.columns]

    # 文本列（自由文本）
    text_cols = [c for c in df.columns if "free_text" in c.lower() or df[c].dtype == "object" and c.lower().endswith("notes")]
    # 诊断列
    diag_cols = [c for c in ["diagnosis_code"] if c in df.columns]

    # 推断类别列：object 或 pandas Categorical
    cat_cols = [c for c in df.columns if c not in key_cols + diag_cols + (date_cols or []) and
                (str(df[c].dtype) == "object" or "category" in str(df[c].dtype))]
    # 指定一些已知类别列加入（如果未被检测）
    for c in ["gender", "smoker_status"]:
        if c in df.columns and c not in cat_cols and c not in key_cols:
            cat_cols.append(c)

    # 数值列
    num_cols = [c for c in df.select_dtypes(include=[np.number]).columns if c not in (key_cols + ([target_col] if target_col else []))]

    # 保护列（不删除）
    protected_cols = list(set(key_cols + diag_cols + date_cols + ([target_col] if target_col else [])))

    roles = {
        "key_cols": key_cols,
        "target_col": target_col,
        "date_cols": date_cols,
        "diag_cols": diag_cols,
        "text_cols": text_cols,
        "cat_cols": cat_cols,
        "num_cols": num_cols,
        "protected_cols": protected_cols,
    }
    return roles


def enforce_types_and_basic_rules(df: pd.DataFrame):
    # 主键为字符串
    for c in ["visit_id", "patient_id"]:
        if c in df.columns:
            df[c] = df[c].astype(str)

    # 日期
    df = to_datetime_safe(df, [c for c in ["admission_date", "discharge_date"] if c in df.columns])

    # 范围规则：age [0, 120]，BMI>0，sbp/dbp > 0
    if "age" in df.columns:
        df.loc[(df["age"] < 0) | (df["age"] > 120), "age"] = np.nan
    if "BMI" in df.columns:
        df.loc[(df["BMI"] <= 0) | (df["BMI"] > 1000), "BMI"] = np.nan  # 上界极大值仅用于屏蔽异常
    for bp in ["sbp", "dbp"]:
        if bp in df.columns:
            df.loc[(df[bp] <= 0) | (df[bp] > 4000), bp] = np.nan  # 宽松上界

    return df


def apply_lab_sentinel_to_nan(df: pd.DataFrame):
    # 对 lab_* 列，将 0 和 -1 视为缺失
    lab_cols = [c for c in df.columns if c.startswith("lab_")]
    for c in lab_cols:
        df.loc[df[c].isin([0, -1]), c] = np.nan

    # 显式处理常见实验室列
    for c in ["lab_glucose", "lab_creatinine"]:
        if c in df.columns:
            df.loc[df[c].isin([0, -1]), c] = np.nan
    return df


def drop_high_missing_columns(df: pd.DataFrame, threshold: float, protected_cols):
    missing = compute_missing_rates(df)
    drop_candidates = [c for c, r in missing.items() if r > threshold and c not in protected_cols]
    kept_before = df.shape[1]
    df = df.drop(columns=drop_candidates, errors="ignore")
    logging.info(f"删除高缺失列（阈值 {threshold}）：{drop_candidates}，删除列数：{kept_before - df.shape[1]}")
    return df, drop_candidates


def forward_fill_within_patient(df: pd.DataFrame, group_col: str, order_col: str, cols: list):
    cols = [c for c in cols if c in df.columns]
    if not cols or group_col not in df.columns or order_col not in df.columns:
        return df, []
    # 排序后分组前向填充
    df = df.sort_values([group_col, order_col])
    filled_counts = {}
    for c in cols:
        before = df[c].isna().sum()
        df[c] = df.groupby(group_col, sort=False)[c].ffill()
        after = df[c].isna().sum()
        filled_counts[c] = int(before - after)
    logging.info(f"分组前向填充完成：{filled_counts}")
    return df, filled_counts


def knn_impute_numeric(df: pd.DataFrame, num_cols: list, exclude_cols: list, n_neighbors: int = 5, scale: bool = True, seed: int = 42):
    use_cols = [c for c in num_cols if c not in exclude_cols and c in df.columns]
    if not use_cols:
        return df, {}

    # 若缺失很少则可能无需 KNN
    if df[use_cols].isna().sum().sum() == 0:
        logging.info("数值列无缺失，跳过 KNN 插补。")
        return df, {c: 0 for c in use_cols}

    # KNNImputer 是确定性的（同数据与顺序），设置随机种子以统一其他流程
    imputer = KNNImputer(n_neighbors=n_neighbors, weights="uniform", metric="nan_euclidean")
    filled_counts = {}
    values_before = df[use_cols].isna().sum().to_dict()

    if scale:
        scaler = StandardScaler(with_mean=True, with_std=True)
        X = df[use_cols].values
        # fit on non-NaN rows? StandardScaler 能处理 NaN，scikit-learn 1.4+ 会忽略 NaN；为兼容性，分列处理
        # 简化实现：对每列手动标准化（忽略 NaN）
        X_scaled = np.zeros_like(X, dtype=float)
        means = {}
        stds = {}
        for idx, c in enumerate(use_cols):
            col = df[c].astype(float).values
            m = np.nanmean(col)
            s = np.nanstd(col)
            if not np.isfinite(s) or s == 0:
                s = 1.0
            means[c] = m
            stds[c] = s
            X_scaled[:, idx] = (col - m) / s
        X_imp = imputer.fit_transform(X_scaled)
        # 逆缩放
        for idx, c in enumerate(use_cols):
            X_imp[:, idx] = X_imp[:, idx] * stds[c] + means[c]
        df.loc[:, use_cols] = X_imp
    else:
        df.loc[:, use_cols] = imputer.fit_transform(df[use_cols].values)

    values_after = df[use_cols].isna().sum().to_dict()
    for c in use_cols:
        filled_counts[c] = int(values_before.get(c, 0) - values_after.get(c, 0))
    logging.info(f"KNN 插补完成，邻居数={n_neighbors}，插补计数：{filled_counts}")
    return df, filled_counts


def median_mode_fallback(df: pd.DataFrame, num_cols: list, cat_cols: list):
    filled = {}

    # 数值中位数兜底
    for c in [x for x in num_cols if x in df.columns]:
        before = int(df[c].isna().sum())
        if before > 0:
            med = df[c].median()
            df[c] = df[c].fillna(med)
        after = int(df[c].isna().sum())
        filled[c] = before - after

    # 类别众数兜底
    for c in [x for x in cat_cols if x in df.columns]:
        before = int(df[c].isna().sum())
        if before > 0:
            try:
                mode_val = df[c].mode(dropna=True).iloc[0]
            except Exception:
                mode_val = None
            if mode_val is not None:
                df[c] = df[c].fillna(mode_val)
        after = int(df[c].isna().sum())
        filled[c] = filled.get(c, 0) + (before - after)

    logging.info(f"中位数/众数兜底插补完成：{filled}")
    return df, filled


def model_impute_columns(df: pd.DataFrame, cols: list, roles: dict, seed: int = 42, n_estimators: int = 200):
    cols = [c for c in cols if c in df.columns]
    if not cols:
        return df, {}

    key_cols = roles["key_cols"]
    date_cols = roles["date_cols"]
    target_col = roles["target_col"]
    diag_cols = roles["diag_cols"]
    protected_exclude = list(set((key_cols or []) + (date_cols or []) + (diag_cols or []) + ([target_col] if target_col else [])))

    # 特征候选：除去待插补列本身、保护列
    feats_all = [c for c in df.columns if c not in protected_exclude]

    filled_info = {}

    for col in cols:
        if col not in df.columns:
            continue

        # 记录原始缺失掩码（在之前流程之后，可能已做了兜底；但这里仅覆盖“原始缺失”的位置）
        original_missing_mask = df[col].isna()

        # 如果此前已经完全无缺失，不需要模型插补
        if original_missing_mask.sum() == 0:
            filled_info[col] = 0
            continue

        # 构建特征列表：移除目标列本身
        features = [c for c in feats_all if c != col and c in df.columns]

        # 将日期列剔除（已在 protected_exclude）
        # 进一步剔除纯文本列（无法直接编码；若有 free_text_notes，按上游阈值可能已删除）
        features = [c for c in features if not (df[c].dtype == "object" and c.lower().endswith("notes"))]

        # 拆分特征类型
        num_feats = [c for c in features if pd.api.types.is_numeric_dtype(df[c])]
        cat_feats = [c for c in features if (str(df[c].dtype) == "object" or "category" in str(df[c].dtype))]

        # 训练/预测数据
        not_missing_idx = (~original_missing_mask).values
        missing_idx = original_missing_mask.values

        X_train = df.loc[not_missing_idx, features]
        X_pred = df.loc[missing_idx, features]
        y_train = df.loc[not_missing_idx, col]

        if y_train.isna().sum() > 0:
            # 若训练标签仍有 NaN，跳过该列
            logging.warning(f"列 {col} 的训练标签仍有 NaN，跳过模型插补。")
            filled_info[col] = 0
            continue

        # 构建预处理与模型
        num_transformer = Pipeline(steps=[
            ("imputer", SimpleImputer(strategy="median"))
        ])
        cat_transformer = Pipeline(steps=[
            ("imputer", SimpleImputer(strategy="most_frequent")),
            ("ohe", OneHotEncoder(handle_unknown="ignore", sparse_output=False))
        ])

        preprocessor = ColumnTransformer(
            transformers=[
                ("num", num_transformer, num_feats),
                ("cat", cat_transformer, cat_feats)
            ],
            remainder="drop"
        )

        if pd.api.types.is_numeric_dtype(df[col]):
            estimator = RandomForestRegressor(
                n_estimators=n_estimators,
                random_state=seed,
                n_jobs=-1
            )
        else:
            # 将 y_train 转为类别
            y_train = y_train.astype(str)
            estimator = RandomForestClassifier(
                n_estimators=n_estimators,
                random_state=seed,
                n_jobs=-1,
                class_weight="balanced"
            )

        pipe = Pipeline(steps=[
            ("pre", preprocessor),
            ("model", estimator)
        ])

        try:
            pipe.fit(X_train, y_train)
            preds = pipe.predict(X_pred)
            # 回写预测值
            df.loc[missing_idx, col] = preds
            filled_info[col] = int(missing_idx.sum())
            logging.info(f"模型插补完成：列 {col}，填充 {filled_info[col]} 个缺失值。")
        except Exception as e:
            logging.error(f"模型插补失败：列 {col}，错误: {e}")
            filled_info[col] = 0

    return df, filled_info


def build_data_dictionary(df_before: pd.DataFrame,
                          df_after: pd.DataFrame,
                          roles: dict,
                          strategy_map: dict) -> pd.DataFrame:
    d = []
    before_missing = compute_missing_rates(df_before).to_dict()
    after_missing = compute_missing_rates(df_after).to_dict()

    for c in df_after.columns:
        role = []
        if c in roles["key_cols"]:
            role.append("key")
        if roles["target_col"] and c == roles["target_col"]:
            role.append("target")
        if c in roles["date_cols"]:
            role.append("date")
        if c in roles["diag_cols"]:
            role.append("diagnosis")
        if c in roles["cat_cols"]:
            role.append("categorical")
        if c in roles["num_cols"]:
            role.append("numeric")
        if c in roles["text_cols"]:
            role.append("text")

        entry = {
            "column": c,
            "dtype": str(df_after[c].dtype),
            "role": ",".join(role) if role else "other",
            "n_unique": int(df_after[c].nunique(dropna=True)),
            "missing_rate_before": float(before_missing.get(c, np.nan)),
            "missing_rate_after": float(after_missing.get(c, np.nan)),
            "impute_strategy": strategy_map.get(c, "none"),
            "notes": ""
        }
        d.append(entry)
    return pd.DataFrame(d)


def save_summary(output_dir: str,
                 df_clean: pd.DataFrame,
                 cols_dropped: list,
                 ffill_counts: dict,
                 knn_counts: dict,
                 fallback_counts: dict,
                 model_counts: dict,
                 missing_before: pd.Series,
                 missing_after: pd.Series,
                 config: dict,
                 dictionary_df: pd.DataFrame):
    os.makedirs(output_dir, exist_ok=True)

    # 清洗摘要
    summary = {
        "timestamp": datetime.now().isoformat(),
        "n_rows": int(df_clean.shape[0]),
        "n_cols": int(df_clean.shape[1]),
        "columns_dropped": cols_dropped,
        "ffill_counts": ffill_counts,
        "knn_counts": knn_counts,
        "fallback_counts": fallback_counts,
        "model_counts": model_counts,
        "missing_rate_before": missing_before.to_dict(),
        "missing_rate_after": missing_after.to_dict(),
        "config_used": config
    }

    with open(os.path.join(output_dir, "cleaning_summary.json"), "w", encoding="utf-8") as f:
        json.dump(summary, f, ensure_ascii=False, indent=2)

    # 另存 CSV 形式的缺失率前/后
    missing_df = pd.DataFrame({
        "column": missing_before.index,
        "missing_rate_before": missing_before.values,
        "missing_rate_after": missing_after.reindex(missing_before.index).values
    })
    missing_df.to_csv(os.path.join(output_dir, "missing_rates_before_after.csv"), index=False, encoding="utf-8")

    # 字段字典
    dictionary_df.to_csv(os.path.join(output_dir, "data_dictionary.csv"), index=False, encoding="utf-8")
    dictionary_df.to_json(os.path.join(output_dir, "data_dictionary.json"), orient="records", force_ascii=False, indent=2)

    # 保存配置
    with open(os.path.join(output_dir, "cleaning_config_used.json"), "w", encoding="utf-8") as f:
        json.dump(config, f, ensure_ascii=False, indent=2)

    logging.info(f"清洗摘要与字段字典已保存到 {output_dir}")


# ------------------------------
# 主流程
# ------------------------------

def main():
    parser = argparse.ArgumentParser(description="住院就诊记录缺失值清理脚本")
    parser.add_argument("--input", type=str, required=True, help="输入 Parquet 文件路径（patient_visits.parquet）")
    parser.add_argument("--output-dir", type=str, required=True, help="输出目录")
    parser.add_argument("--seed", type=int, default=42, help="随机种子")
    parser.add_argument("--log-level", type=str, default="INFO", help="日志级别（INFO/DEBUG/WARN/ERROR）")

    # 删除高缺失列
    parser.add_argument("--drop-threshold", type=float, default=0.95, help="按缺失率删除列的阈值，保护列除外")
    parser.add_argument("--disable-drop", action="store_true", help="禁用按阈值删除高缺失列")

    # 前向填充
    parser.add_argument("--ffill-cols", type=str, default="BMI,smoker_status",
                        help="逗号分隔的列名列表，在同一 patient_id 内按 admission_date 前向填充；为空字符串表示不执行")
    parser.add_argument("--disable-ffill", action="store_true", help="禁用前向填充")

    # KNN 插补
    parser.add_argument("--enable-knn", action="store_true", help="启用 KNN 插补数值列")
    parser.add_argument("--n-neighbors", type=int, default=5, help="KNNImputer 的邻居数")

    # 模型插补
    parser.add_argument("--model-impute-cols", type=str, default="",
                        help="逗号分隔的列名列表，对这些列使用随机森林模型预测插补；为空表示不启用")
    parser.add_argument("--n-estimators", type=int, default=200, help="随机森林基学习器数量（模型插补）")

    # 输出控制
    parser.add_argument("--save-csv", type=int, default=0, help="是否另存 CSV（1 保存，0 不保存）")

    args = parser.parse_args()

    os.makedirs(args.output_dir, exist_ok=True)
    setup_logging(args.output_dir, args.log_level)
    set_global_seed(args.seed)

    t0 = time.time()
    logging.info("开始加载数据...")
    df = load_parquet(args.input)
    n_rows, n_cols = df.shape
    logging.info(f"数据加载完成：{n_rows} 行，{n_cols} 列")

    # 复制用于统计
    df_before = df.copy(deep=True)

    # 去重（按 visit_id）
    if "visit_id" in df.columns:
        dup_count = int(df.duplicated(subset=["visit_id"]).sum())
        if dup_count > 0:
            df = df.drop_duplicates(subset=["visit_id"], keep="first")
            logging.info(f"删除重复 visit_id 记录：{dup_count}")

    # 基础类型与规则
    df = enforce_types_and_basic_rules(df)
    df = apply_lab_sentinel_to_nan(df)

    # 角色探测
    roles = detect_column_roles(df)
    logging.info(f"列角色探测结果：{ {k: (v if isinstance(v, str) else len(v)) for k, v in roles.items()} }")

    # 删除高缺失列（保护列除外）
    cols_dropped = []
    if not args.disable_drop:
        df, cols_dropped = drop_high_missing_columns(df, threshold=args.drop_threshold, protected_cols=roles["protected_cols"])

    # 前向填充（按 patient_id + admission_date）
    ffill_counts = {}
    if not args.disable_ffill:
        ffill_cols = [c.strip() for c in args.ffill_cols.split(",") if c.strip()]
        df, ffill_counts = forward_fill_within_patient(
            df=df,
            group_col="patient_id" if "patient_id" in df.columns else None,
            order_col="admission_date" if "admission_date" in df.columns else None,
            cols=ffill_cols
        )

    missing_before = compute_missing_rates(df)

    # KNN 插补（仅数值列；排除键、目标、日期）
    knn_counts = {}
    if args.enable_knn:
        exclude_for_knn = roles["protected_cols"]
        df, knn_counts = knn_impute_numeric(
            df=df,
            num_cols=roles["num_cols"],
            exclude_cols=exclude_for_knn,
            n_neighbors=args.n_neighbors,
            scale=True,
            seed=args.seed
        )

    # 兜底：数值中位数、类别众数
    fallback_counts = {}
    df, fallback_counts = median_mode_fallback(df, roles["num_cols"], roles["cat_cols"])

    # 模型插补（对指定列，覆盖原始缺失位置）
    model_counts = {}
    model_cols = [c.strip() for c in args.model_impute_cols.split(",") if c.strip()]
    if len(model_cols) > 0:
        df, model_counts = model_impute_columns(df, model_cols, roles, seed=args.seed, n_estimators=args.n_estimators)

    missing_after = compute_missing_rates(df)

    # 构建策略映射（字段字典用）
    strategy_map = {}
    # 标记 KNN 使用的列
    for c, cnt in knn_counts.items():
        if cnt > 0:
            strategy_map[c] = "KNN"
    # 标记前向填充列
    for c, cnt in ffill_counts.items():
        if cnt > 0:
            strategy_map[c] = "forward_fill" if c not in strategy_map else strategy_map[c] + "+forward_fill"
    # 标记模型插补列
    for c, cnt in model_counts.items():
        if cnt > 0:
            strategy_map[c] = "model" if c not in strategy_map else strategy_map[c] + "+model"
    # 对未被上述捕获但有兜底的列
    for c, cnt in fallback_counts.items():
        if cnt > 0 and c not in strategy_map:
            if c in roles["num_cols"]:
                strategy_map[c] = "median"
            elif c in roles["cat_cols"]:
                strategy_map[c] = "mode"

    # 输出
    cleaned_parquet = os.path.join(args.output_dir, "patient_visits_cleaned.parquet")
    df.to_parquet(cleaned_parquet, index=False)
    if args.save_csv == 1:
        cleaned_csv = os.path.join(args.output_dir, "patient_visits_cleaned.csv")
        df.to_csv(cleaned_csv, index=False, encoding="utf-8")

    # 字段字典
    dictionary_df = build_data_dictionary(df_before, df, roles, strategy_map)

    # 保存摘要、字典与配置
    effective_config = {
        "input": args.input,
        "output_dir": args.output_dir,
        "seed": args.seed,
        "drop_threshold": args.drop_threshold,
        "disable_drop": bool(args.disable_drop),
        "ffill_cols": [c.strip() for c in args.ffill_cols.split(",") if c.strip()],
        "disable_ffill": bool(args.disable_ffill),
        "enable_knn": bool(args.enable_knn),
        "n_neighbors": args.n_neighbors,
        "model_impute_cols": model_cols,
        "n_estimators": args.n_estimators,
        "save_csv": args.save_csv,
        "protected_cols": roles["protected_cols"]
    }

    save_summary(
        output_dir=args.output_dir,
        df_clean=df,
        cols_dropped=cols_dropped,
        ffill_counts=ffill_counts,
        knn_counts=knn_counts,
        fallback_counts=fallback_counts,
        model_counts=model_counts,
        missing_before=compute_missing_rates(df_before),
        missing_after=missing_after,
        config=effective_config,
        dictionary_df=dictionary_df
    )

    elapsed = time.time() - t0
    logging.info(f"清洗完成，耗时 {elapsed:.2f} 秒。输出文件位于：{args.output_dir}")


if __name__ == "__main__":
    main()

要点说明：

• 避免目标泄露：模型与 KNN 插补的特征集中均排除 readmission_30d、主键、日期、诊断列。
• 时间一致性：BMI、smoker_status 默认在同一 patient_id 按 admission_date 前向填充，可通过 --ffill-cols 配置。
• 哨兵值：对 lab_* 列（含 lab_glucose、lab_creatinine）将 0 和 -1 视为缺失。
• 删除高缺失列：默认阈值 0.95，free_text_notes 若高缺失会被删除；visit_id/patient_id/diagnosis_code/admission_date/discharge_date/readmission_30d 被保护不删除。
• 输出内容：
  • patient_visits_cleaned.parquet（以及可选 CSV）
  • cleaning_summary.json、missing_rates_before_after.csv
  • data_dictionary.json/csv
  • cleaning_config_used.json
  • 日志文件 cleaning_*.log

如需仅使用中位数/众数兜底插补并跳过 KNN 与模型插补，禁用相应选项： python clean_patient_visits.py --input patient_visits.parquet --output-dir ./cleaned --seed 42 --disable-ffill --disable-drop --save-csv 0

如需对 BMI 使用模型预测插补（在兜底后覆盖原始缺失位置）： python clean_patient_visits.py --input patient_visits.parquet --output-dir ./cleaned --seed 42 --model-impute-cols BMI --n-estimators 300 --enable-knn

备注：KNN 插补在 350k×若干数值特征时计算量较大；可通过 --n-neighbors 调整或禁用以加速。

以下为可直接执行的 Python 脚本，按照设备维度并行处理、严格按时间顺序、在限定插补窗口内进行插值，支持时间索引对齐，并生成详细日志和质量报告。脚本含中英双语注释。

使用说明:

• 依赖: pandas, pyarrow, numpy
• 可选: pyarrow.fs（远程文件系统），multiprocessing（标准库）
• 运行示例: python clean_telemetry.py --input telemetry.parquet --output_dir cleaned_out --num_workers 4 --align_to_minutely False

脚本内容:

#!/usr/bin/env python3

-- coding: utf-8 --

""" 设备遥测缺失值清理脚本 / Telemetry Missing-Value Cleaning Script

功能概述 / Overview:

• 按 device_id 分区并行处理 / Parallel per-device processing
• 严格按时间顺序处理（不重排）/ Strict time-order processing (no reordering)
• 可选时间索引对齐到 1 分钟频率 / Optional time index alignment to 1-minute frequency
• 针对不同字段的缺失策略与插补窗口限制 / Field-specific strategies and interpolation window limits
• 长时间缺口不跨越插值 / No interpolation across long gaps
• 生成处理日志与质量报告 / Logging and quality reports

目标字段与缺失特征 / Fields and Missingness:

• temp_c (~3%，连续1–5分钟缺口) -> 线性/时间插值，窗口≤5
• vibration_rms (~1.5%，短缺口) -> 线性插值，窗口≤5
• rpm (~0.8%，短缺口) -> 线性插值，窗口≤3
• battery_level（有线设备大量缺失）-> 固定值填充(-1) + 缺失指示列
• status (~2%，解析失败) -> 类别型字段: 有限的前后向填充，余量填 "UNKNOWN"
• timestamp (~0.01%) -> 无法排序/对齐，直接删除所在行

业务约束 / Business Constraints:

• 严格按时间顺序处理；如序列不单调上升，跳过插值（仅有限 FFill/BFill）
• 不跨越长时间缺口进行插值（通过 limit 限制）
• 同一 device_id 内序列相关性强：优先时间插值或短窗口 FFill/BFill

输出 / Outputs:

• 分区 Parquet（按 device_id）/ Partitioned Parquet by device_id
• 质量报告 JSON / Quality report in JSON
• 详细日志 / Detailed logs

"""

import argparse import json import logging import os from dataclasses import dataclass, asdict from datetime import datetime from typing import Dict, Any, Tuple, List

import numpy as np import pandas as pd

------------------------------

配置 / Configuration

------------------------------

@dataclass class CleaningConfig: # 插值最大缺口（以分钟计）/ Max consecutive interpolation window (minutes) max_gap_temp: int = 5 max_gap_vibration: int = 5 max_gap_rpm: int = 3 max_gap_humidity: int = 5 max_gap_power: int = 2

# 边界有限前后向填充窗口（以分钟计）/ Edge limited ffill/bfill window (minutes)
edge_fill_limit: int = 1

# 电量固定填充值（表示N/A）/ Fixed fill for battery_level (N/A)
battery_level_fill: float = -1.0

# 类别缺失标签 / Categorical unknown label
unknown_label: str = "UNKNOWN"

# 是否对齐到每分钟索引 / Align to 1-minute index
align_to_minutely: bool = False

# 允许设备内排序（违背业务约束，默认禁止）/ Allow sorting within device (default False per constraint)
allow_sort_within_device: bool = False

# 是否在最终阶段丢弃仍含关键字段缺失的行 / Drop rows with remaining NA in critical fields
drop_rows_with_any_na: bool = False

# 关键数值字段 / Critical numeric fields
critical_numeric_fields: Tuple[str, ...] = (
    "temp_c", "vibration_rms", "rpm"
)

# 输出是否按 device_id 分区存储 / Partition parquet by device_id
partition_output: bool = True

------------------------------

工具函数 / Utility Functions

------------------------------

def setup_logging(output_dir: str) -> None: os.makedirs(output_dir, exist_ok=True) log_path = os.path.join(output_dir, "cleaning.log") logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(processName)s - %(message)s", handlers=[ logging.FileHandler(log_path, mode="w", encoding="utf-8"), logging.StreamHandler() ] ) logging.info("Log file initialized at %s", log_path)

def _na_run_lengths(s: pd.Series) -> List[int]: """ 计算连续缺失段长度（用于统计，不写回）/ Compute consecutive NA run lengths (for stats) """ is_na = s.isna().to_numpy() if not is_na.any(): return [] # Find run lengths runs = [] count = 0 for v in is_na: if v: count += 1 else: if count > 0: runs.append(count) count = 0 if count > 0: runs.append(count) return runs

def _interpolate_limited( s: pd.Series, limit: int, method: str = "time", limit_area: str = "inside" ) -> Tuple[pd.Series, int]: """ 限制窗口插值 / Interpolation with window limit 返回：插值后的序列、被填充的数量 Return: interpolated series and count of filled values """ before_na = s.isna().sum() if isinstance(s.index, pd.DatetimeIndex) and method == "time": out = s.interpolate(method="time", limit=limit, limit_area=limit_area) else: out = s.interpolate(method="linear", limit=limit, limit_area=limit_area) after_na = out.isna().sum() filled = int(before_na - after_na) return out, filled

def _ffill_bfill_limited(s: pd.Series, limit: int) -> Tuple[pd.Series, int, int]: """ 有限窗口前向/后向填充 / Limited ffill/bfill 返回：新序列、ffill填充数、bfill填充数 """ na_before = s.isna().sum() s1 = s.ffill(limit=limit) ffilled = na_before - s1.isna().sum() s2 = s1.bfill(limit=limit) bfilled = s1.isna().sum() - s2.isna().sum() return s2, int(ffilled), int(bfilled)

def _normalize_status(s: pd.Series, unknown_label: str) -> Tuple[pd.Series, Dict[str, int]]: """ 规范化状态字段：限制到 {OK, WARN, FAIL, UNKNOWN} Normalize status values to {OK, WARN, FAIL, UNKNOWN} with limited ffill/bfill strategy externally. """ allowed = {"OK", "WARN", "FAIL"} stats = {"invalid_to_unknown": 0} s_clean = s.copy() mask_invalid = ~s_clean.isna() & ~s_clean.astype(str).isin(allowed) stats["invalid_to_unknown"] = int(mask_invalid.sum()) s_clean.loc[mask_invalid] = unknown_label return s_clean, stats

def ensure_time_index(df: pd.DataFrame, allow_sort: bool) -> Tuple[pd.DataFrame, bool]: """ 确保时间索引且序列单调上升 / Ensure datetime index and monotonic increasing 返回：DataFrame，是否单调 """ df = df.copy() if "timestamp" not in df.columns: raise ValueError("timestamp column missing.") # 删除缺失时间戳的行（无法排序与对齐）/ Drop rows with missing timestamps drop_rows = df["timestamp"].isna().sum() if drop_rows > 0: df = df.loc[~df["timestamp"].isna()].copy()

df["timestamp"] = pd.to_datetime(df["timestamp"], errors="coerce", utc=False)
df = df.loc[~df["timestamp"].isna()].copy()

# 不重排：仅在已单调情况下设置索引；如不单调且不允许排序，则返回不单调标记
# No reordering: if not monotonic and sorting not allowed, return monotonic=False
is_monotonic = df["timestamp"].is_monotonic_increasing
if allow_sort:
    # 业务约束默认不允许，只有明确允许时才排序
    df = df.sort_values("timestamp", kind="stable")
    is_monotonic = True
df = df.set_index("timestamp", drop=True)
return df, is_monotonic

def align_minutely(df: pd.DataFrame) -> Tuple[pd.DataFrame, int]: """ 对齐到每分钟索引（闭区间）/ Align to 1-minute frequency (inclusive) 返回：新DF与新增行数 / new df and number of inserted rows """ if df.empty: return df, 0 start, end = df.index.min(), df.index.max() full_idx = pd.date_range(start=start, end=end, freq="T") before = len(df) out = df.reindex(full_idx) inserted = len(out) - before return out, int(inserted)

------------------------------

设备级处理 / Per-device processing

------------------------------

def process_device( device_id: Any, df_device: pd.DataFrame, cfg: CleaningConfig ) -> Tuple[pd.DataFrame, Dict[str, Any]]: """ 处理单个设备的缺失值清理 / Clean missing values for a single device """ metrics: Dict[str, Any] = { "device_id": device_id, "rows_in": int(len(df_device)), "rows_timestamp_missing_dropped": int(df_device["timestamp"].isna().sum()), "inserted_minute_rows": 0, "monotonic_time": True, "interpolate_filled": {}, "ffill_filled": {}, "bfill_filled": {}, "fixed_filled": {}, "categorical_unknown_filled": {}, "long_gaps_skipped": {}, "rows_out": 0, "post_na_counts": {}, "pre_na_counts": df_device.isna().sum().to_dict(), }

# 确保时间索引与单调性 / Ensure time index and monotonicity
df_ts, is_monotonic = ensure_time_index(df_device, allow_sort=cfg.allow_sort_within_device)
metrics["monotonic_time"] = bool(is_monotonic)
if not is_monotonic:
    logging.warning("Device %s has non-monotonic timestamps; interpolation will be skipped; limited ffill/bfill only.", device_id)

# 可选：对齐到每分钟索引 / Optional: align to minutely index
if cfg.align_to_minutely:
    df_ts, inserted = align_minutely(df_ts)
    metrics["inserted_minute_rows"] = int(inserted)

# 回填 device_id / Restore device_id
df_ts["device_id"] = device_id

# 统计缺口分布（插值前）/ Gap runs before interpolation
gap_stats = {}
targets = {
    "temp_c": cfg.max_gap_temp,
    "vibration_rms": cfg.max_gap_vibration,
    "rpm": cfg.max_gap_rpm,
    "humidity": cfg.max_gap_humidity,
    "power_w": cfg.max_gap_power,
}
for col, lim in targets.items():
    if col in df_ts.columns:
        runs = _na_run_lengths(df_ts[col])
        gap_stats[col] = {
            "num_na_runs": int(len(runs)),
            "num_runs_gt_limit": int(sum(1 for r in runs if r > lim)),
        }
    else:
        gap_stats[col] = {"num_na_runs": 0, "num_runs_gt_limit": 0}
metrics["long_gaps_skipped"] = {k: v["num_runs_gt_limit"] for k, v in gap_stats.items()}

# 数值列插值 / Interpolation for numeric fields
interp_filled_counts = {}
ffill_counts = {}
bfill_counts = {}

def do_interpolate(col: str, limit: int) -> None:
    if col not in df_ts.columns:
        return
    s = df_ts[col]
    if s.dtype.kind not in ("i", "u", "f", "b"):  # 非数值跳过
        return
    if not metrics["monotonic_time"]:
        # 非单调：跳过插值，仅做有限FFill/BFill
        s2, ffc, bfc = _ffill_bfill_limited(s, cfg.edge_fill_limit)
        df_ts[col] = s2
        ffill_counts[col] = ffill_counts.get(col, 0) + ffc
        bfill_counts[col] = bfill_counts.get(col, 0) + bfc
        return
    # 先限定窗口的时间/线性插值（禁止跨越长缺口与边界）/ Limited interpolation (no edges)
    s2, filled = _interpolate_limited(
        s, limit=limit, method="time", limit_area="inside"
    )
    interp_filled_counts[col] = interp_filled_counts.get(col, 0) + filled
    # 边界再有限度前后向填充 / Edge-limited ffill/bfill
    s3, ffc, bfc = _ffill_bfill_limited(s2, cfg.edge_fill_limit)
    df_ts[col] = s3
    ffill_counts[col] = ffill_counts.get(col, 0) + ffc
    bfill_counts[col] = bfill_counts.get(col, 0) + bfc

# 针对目标列执行插值 / Apply interpolation to target columns
for col, lim in targets.items():
    do_interpolate(col, lim)

# battery_level：固定值填充 + 指示列 / battery_level: fixed fill + indicator
fixed_counts = {}
if "battery_level" in df_ts.columns:
    mask_batt_na = df_ts["battery_level"].isna()
    fixed_counts["battery_level"] = int(mask_batt_na.sum())
    if fixed_counts["battery_level"] > 0:
        df_ts.loc[mask_batt_na, "battery_level"] = cfg.battery_level_fill
    # 缺失标记 / Missing indicator
    df_ts["battery_level_missing"] = mask_batt_na.astype(np.int8)

# status：有限FFill/BFill + UNKNOWN / status with limited ffill/bfill + UNKNOWN
cat_unknown_counts = {}
if "status" in df_ts.columns:
    # 规范化异常值 / Normalize invalid values
    s, st_stats = _normalize_status(df_ts["status"], cfg.unknown_label)
    # 有限前后向填充（仅1）/ limited ffill/bfill
    s2, ffc, bfc = _ffill_bfill_limited(s, cfg.edge_fill_limit)
    s2 = s2.fillna(cfg.unknown_label)
    df_ts["status"] = s2.astype("category")
    cat_unknown_counts["status_invalid_to_unknown"] = st_stats["invalid_to_unknown"]
    cat_unknown_counts["status_ffill"] = ffc
    cat_unknown_counts["status_bfill"] = bfc
    cat_unknown_counts["status_final_unknown"] = int((s2 == cfg.unknown_label).sum())

# 其他类别字段：site/firmware_ver -> UNKNOWN / Other categoricals to UNKNOWN
for col in ("site", "firmware_ver"):
    if col in df_ts.columns:
        na_cnt = int(df_ts[col].isna().sum())
        if na_cnt > 0:
            df_ts[col] = df_ts[col].astype("string")
            df_ts[col] = df_ts[col].fillna(cfg.unknown_label)
            cat_unknown_counts[col] = na_cnt

# 最终丢弃策略（可选）/ Optional final drop of rows with NA in critical numeric fields
rows_before_drop = len(df_ts)
if cfg.drop_rows_with_any_na:
    mask_any_na = df_ts[list(cfg.critical_numeric_fields)].isna().any(axis=1)
    drop_cnt = int(mask_any_na.sum())
    if drop_cnt > 0:
        df_ts = df_ts.loc[~mask_any_na].copy()
    metrics["rows_dropped_final_na"] = drop_cnt
else:
    metrics["rows_dropped_final_na"] = 0

# 指标汇总 / Metrics aggregation
metrics["interpolate_filled"] = interp_filled_counts
metrics["ffill_filled"] = ffill_counts
metrics["bfill_filled"] = bfill_counts
metrics["fixed_filled"] = fixed_counts
metrics["categorical_unknown_filled"] = cat_unknown_counts
metrics["rows_out"] = int(len(df_ts))
metrics["post_na_counts"] = df_ts.isna().sum().to_dict()

# 确保列顺序稳定（不重排记录，仅输出整理）/ Keep stable column order for output
if "timestamp" not in df_ts.columns:
    # timestamp 现在是索引 / it's the index
    df_ts = df_ts.reset_index().rename(columns={"index": "timestamp"})
return df_ts, metrics

------------------------------

主流程 / Main pipeline

------------------------------

def main(): parser = argparse.ArgumentParser(description="Telemetry missing value cleaning / 设备遥测缺失值清理") parser.add_argument("--input", required=True, help="Input Parquet path / 输入 Parquet 路径") parser.add_argument("--output_dir", required=True, help="Output directory / 输出目录") parser.add_argument("--num_workers", type=int, default=1, help="Parallel workers by device / 并行进程数（按设备分区）") parser.add_argument("--align_to_minutely", action="store_true", help="Align time index to 1-minute / 对齐到每分钟索引") parser.add_argument("--allow_sort_within_device", action="store_true", help="Allow sorting within device (not recommended) / 允许设备内排序（不建议）") parser.add_argument("--drop_rows_with_any_na", action="store_true", help="Drop rows with remaining NA in critical fields / 丢弃仍含关键缺失的行") args = parser.parse_args()

os.makedirs(args.output_dir, exist_ok=True)
setup_logging(args.output_dir)

cfg = CleaningConfig(
    align_to_minutely=args.align_to_minutely,
    allow_sort_within_device=args.allow_sort_within_device,
    drop_rows_with_any_na=args.drop_rows_with_any_na,
)

logging.info("Cleaning configuration / 清理配置: %s", json.dumps(asdict(cfg), ensure_ascii=False))
logging.info("Interpolation window limits / 插补窗口限制: temp<=%d, vibration<=%d, rpm<=%d, humidity<=%d, power<=%d; edge ffill/bfill<=%d",
             cfg.max_gap_temp, cfg.max_gap_vibration, cfg.max_gap_rpm, cfg.max_gap_humidity, cfg.max_gap_power, cfg.edge_fill_limit)
logging.info("Strategies used / 策略: [使用插值法填充, 使用前向填充, 使用后向填充, 类别型字段用‘未知’标记填充, 数值型字段用固定值填充, 删除含缺失值的行(仅限timestamp或可选最终阶段)]")

# 读取数据 / Read parquet
logging.info("Reading input parquet / 读取Parquet: %s", args.input)
df = pd.read_parquet(args.input)
expected_cols = {"device_id", "timestamp"}
missing_essential = expected_cols - set(df.columns)
if missing_essential:
    raise ValueError(f"Missing essential columns: {missing_essential}")

# 确保原始行顺序不变（不进行全局排序）/ Do not reorder globally
total_rows = len(df)
logging.info("Rows total: %d, Columns: %d", total_rows, len(df.columns))

# 按设备分区 / Partition by device
device_ids = df["device_id"].dropna().unique().tolist()
logging.info("Unique devices: %d", len(device_ids))

# 组装每个设备的子表（保持原顺序）/ Build per-device DataFrames (preserve original order)
groups = {dev: df.loc[df["device_id"] == dev].copy() for dev in device_ids}

# 并行处理 / Parallel processing
results: List[pd.DataFrame] = []
metrics_all: List[Dict[str, Any]] = []

if args.num_workers and args.num_workers > 1:
    import multiprocessing as mp
    with mp.get_context("spawn").Pool(processes=args.num_workers) as pool:
        jobs = []
        for dev, gdf in groups.items():
            jobs.append(pool.apply_async(process_device, (dev, gdf, cfg)))
        for j in jobs:
            cleaned_df, met = j.get()
            results.append(cleaned_df)
            metrics_all.append(met)
else:
    for dev, gdf in groups.items():
        cleaned_df, met = process_device(dev, gdf, cfg)
        results.append(cleaned_df)
        metrics_all.append(met)

# 合并结果 / Concatenate results
cleaned = pd.concat(results, ignore_index=True)

# 写出结果 / Write outputs
if cfg.partition_output:
    # 分区写出 / Partition by device_id
    out_root = os.path.join(args.output_dir, "parquet_by_device")
    os.makedirs(out_root, exist_ok=True)
    # 使用 pyarrow 分区写出（pandas>=1.4 支持 partition_cols）/ partitioned write
    cleaned.to_parquet(
        out_root,
        engine="pyarrow",
        index=False,
        partition_cols=["device_id"]
    )
    logging.info("Partitioned parquet written to: %s", out_root)
else:
    out_path = os.path.join(args.output_dir, "telemetry_cleaned.parquet")
    cleaned.to_parquet(out_path, engine="pyarrow", index=False)
    logging.info("Parquet written to: %s", out_path)

# 质量报告 / Quality report
# 聚合统计 / Aggregate metrics
agg = {
    "generated_at": datetime.utcnow().isoformat() + "Z",
    "total_devices": len(device_ids),
    "rows_in_total": int(sum(m["rows_in"] for m in metrics_all)),
    "rows_out_total": int(sum(m["rows_out"] for m in metrics_all)),
    "rows_timestamp_missing_dropped_total": int(sum(m["rows_timestamp_missing_dropped"] for m in metrics_all)),
    "inserted_minute_rows_total": int(sum(m["inserted_minute_rows"] for m in metrics_all)),
    "devices_non_monotonic": int(sum(1 for m in metrics_all if not m["monotonic_time"])),
    "interpolate_filled_total": {},
    "ffill_filled_total": {},
    "bfill_filled_total": {},
    "fixed_filled_total": {},
    "categorical_unknown_filled_total": {},
    "post_na_counts_total": {},
}

# 累加器 / reducers
def add_dict(acc: Dict[str, int], x: Dict[str, int]):
    for k, v in x.items():
        acc[k] = acc.get(k, 0) + int(v)

for m in metrics_all:
    add_dict(agg["interpolate_filled_total"], m.get("interpolate_filled", {}))
    add_dict(agg["ffill_filled_total"], m.get("ffill_filled", {}))
    add_dict(agg["bfill_filled_total"], m.get("bfill_filled", {}))
    add_dict(agg["fixed_filled_total"], m.get("fixed_filled", {}))
    add_dict(agg["categorical_unknown_filled_total"], m.get("categorical_unknown_filled", {}))
    add_dict(agg["post_na_counts_total"], m.get("post_na_counts", {}))

# 保存报告 / Save reports
report_path = os.path.join(args.output_dir, "quality_report.json")
with open(report_path, "w", encoding="utf-8") as f:
    json.dump({"config": asdict(cfg), "aggregate": agg, "per_device": metrics_all}, f, ensure_ascii=False, indent=2)
logging.info("Quality report written to: %s", report_path)

logging.info("Done. Rows in: %d, Rows out: %d", agg["rows_in_total"], agg["rows_out_total"])

if name == "main": main()

实现细节与策略说明:

• 时间顺序与不可重排: 设备内不进行排序，若检测到 timestamp 非单调且未显式允许排序，则跳过时间插值，仅执行有限窗口前后向填充（edge_fill_limit=1）。
• 插补区间限制: temp/vibration 限制为不超过 5 分钟连续缺口；rpm 为 3 分钟；humidity 为 5 分钟；power 为 2 分钟。使用 interpolate(limit=..., limit_area='inside')，避免跨越长缺口与边界。
• 时间索引对齐: 默认关闭；如启用，将对齐到完整的逐分钟索引并记录插入行数 inserted_minute_rows。对齐可能产生大量新行，应评估设备时间跨度与空闲时段。
• battery_level: 使用固定值 -1 表示不可用/不适用，并生成 battery_level_missing 指示列，便于下游模型识别。
• status: 先将非预期值映射为 UNKNOWN，再进行有限 ffill/bfill（各1步），剩余缺失填 UNKNOWN，保证分类可用。
• 删除行: 仅对 timestamp 缺失行执行硬删除（无时间无法对齐与排序）。若设置 --drop_rows_with_any_na，则对指定关键数值字段的剩余 NA 执行最终行级删除。
• 并行: 使用 multiprocessing 按设备分区并行处理，避免跨设备重排。5,000,000 行规模可在单机内存中处理；如设备数量非常多，建议分批或使用 Dask/Polars 重构为惰性计算管线。
• 质量报告: 包含各字段插值/前后向/固定/类别未知填充的计数、长缺口段计数、时间对齐插入行数、时间非单调设备数与处理后的缺失统计。可用于审计与复现实验。

如需根据不同设备类型（有线/电池）动态调整 battery_level 策略，可在 process_device 内加入基于电源特征（如 power_w 稳定>0 且 battery_level 全缺失）来切换为完全置空+指示列或保持 -1 语义。