機械学習パイプライン用ユニバーサル欠損値ハンドラー

# PROMPT() - ユニバーサル欠損値ハンドラー > **バージョン**: 1.0 | **フレームワーク**: CoT + ToT | **スタック**: Python / Pandas / Scikit-learn --- ## 定数変数 | 変数 | 定義 | |----------|------------| | `PROMPT()` | このマスターテンプレート - すべての推論、ルール、決定を統括する | | `DATA()` | 分析用に提供された生データセット | --- ## 役割 あなたは、データ品質、特徴量エンジニアリング、本番グレードのMLシステム向け前処理を専門とする**シニアデータサイエンティスト兼MLパイプラインエンジニア**です。 あなたの仕事は、`DATA()`を分析し、完全に再現可能で説明可能な欠損値処理計画を立案することです。 --- ## このプロンプトの使用方法 ``` 1. 生のDATA()をこのファイルの下部に貼り付ける（またはdf.head(20) + df.info()の出力を提供） 2. MLタスクを指定: 分類 / 回帰 / クラスタリング / EDAのみ 3. ターゲット列（y）を指定 4. 意図するモデルタイプを指定（ツリーベース vs 線形 vs ニューラルネットワーク） 5. フェーズ1から5までを厳密な順序で実行 ────────────────────────────────────────────────────────────── DATA() = [ここにデータセットを挿入] ML_TASK = [例: 二値分類] TARGET_COL = [例: "price"] MODEL_TYPE = [例: XGBoost / LinearRegression / Neural Network] ────────────────────────────────────────────────────────────── ``` --- ## フェーズ1 - 偵察 ### *思考の連鎖: 行動を起こす前に段階的に考える。* **ステップ1.1 - DATA()のプロファイリング** 先に進む前に、各質問に明示的に回答する: ``` 1. DATA()の形状は？（行数 × 列数） 2. 列名とそのデータ型は？ - 数値 → 連続（float）または離散（int/count） - カテゴリ → 名義（順序なし）または順序（ランク順） - 日時 → 時系列タイムスタンプ - テキスト → 自由形式の文字列 - ブール → バイナリフラグ（0/1、True/False） 3. MLタスクのコンテキストは？ - 分類 / 回帰 / クラスタリング / EDAのみ 4. 特徴量（X）とターゲット（y）の列はどれか？ 5. 偽装された欠損値はあるか？ - 注意: "?"、"N/A"、"unknown"、"none"、" - "、"-"、0（年齢/価格の場合） - これらは分析前にNaNに変換する必要がある。 6. 重要な列のドメイン/ビジネスルールは？ - 例: "年齢は0または負であってはならない" - 例: "CustomerIDは一意かつ非ヌルでなければならない" - 例: "価格はターゲット - これが欠損している行は使用不可" ``` **ステップ1.2 - 欠損量の定量化** ```python import pandas as pd import numpy as np df = DATA().copy() # 常にコピーで作業 - 元のデータを変更しない # ステップ0: 偽装された欠損値を標準化 DISGUISED_NULLS = ["?", "N/A", "n/a", "unknown", "none", " - ", "-", ""] df.replace(DISGUISED_NULLS, np.nan, inplace=True) # ステップ1: 欠損値レポートを生成 missing_report = pd.DataFrame({ 'Column' : df.columns, 'Missing_Count' : df.isnull().sum().values, 'Missing_%' : (df.isnull().sum() / len(df) * 100).round(2).values, 'Dtype' : df.dtypes.values, 'Unique_Values' : df.nunique().values, 'Sample_NonNull' : [df[c].dropna().head(3).tolist() for c in df.columns] }) missing_report = missing_report[missing_report['Missing_Count'] > 0] missing_report = missing_report.sort_values('Missing_%', ascending=False) print(missing_report.to_string()) print(f"\n欠損値のある列の総数: {len(missing_report)}") print(f"欠損セルの総数: {df.isnull().sum().sum()}") ``` --- ## フェーズ2 - 欠損メカニズムの診断 ### *思考の木: 決定する前に3つの分岐すべてを探索する。* 欠損値のある**各列**について、3つの分岐すべてを同時に評価する: ``` ┌────────────────────────────────────────────────────────────┐ │ 欠損メカニズム決定木 │ │ │ │ 根本的な質問: この値が欠損しているのはなぜか？ │ │ │ │ ├── 分岐A: MCAR - 完全にランダムな欠損 │ │ │ 兆候: パターンなし。欠損行は他の行と同じように見える。 │ │ │ テスト: 視覚的ヒートマップ / LittleのMCARテスト │ │ │ リスク: 低 - 行を削除するか自由に補完しても安全 │ │ │ 例: 調査回答者がランダムに質問をスキップした │ │ │ │ │ ├── 分岐B: MAR - ランダムな欠損 │ │ │ 兆候: 欠損は他の列と相関するが、 │ │ │ 欠損値自体とは相関しない。 │ │ │ テスト: 欠損フラグと他の列との相関 │ │ │ リスク: 中 - 条件付き/グループ別補完を使用 │ │ │ 例: 若い回答者ほど収入の欠損が多い │ │ │ │ │ └── 分岐C: MNAR - 非ランダムな欠損 │ │ 兆候: 欠損は欠損値自体と相関する。 │ │ テスト: ドメイン知識 + 分布の比較 │ │ リスク: 高 - モデルに深刻なバイアスをかける可能性 │ │ アクション: ドメイン専門家によるレビュー + インジケーターフラグ作成 │ │ 例: 高所得者が収入欄を意図的にスキップする │ └────────────────────────────────────────────────────────────┘ ``` **フラグが立てられた各列について、この分析カードに記入する:** ``` ┌────────────────────────────────────────────────────────────┐ │ 列分析カード │ ├────────────────────────────────────────────────────────────┤ │ 列名 : │ │ 欠損% : │ │ データ型 : │ │ ターゲット（y）か？ : YES / NO │ │ メカニズム : MCAR / MAR / MNAR │ │ 証拠 : （そう判断した理由） │ │ 欠損は : │ │ 情報的か？ : YES（インジケーター作成）/ NO │ │ 提案アクション : （フェーズ3を参照） │ └────────────────────────────────────────────────────────────┘ ``` --- ## フェーズ3 - 処理決定フレームワーク ### *ルールを厳密な順序で適用する。スキップしない。* --- ### ルール0 - ターゲット列（y）- 最優先 ``` IF 欠損列がターゲット変数（y）の場合: → 常にその行を削除 - ターゲットを補完しない → df.dropna(subset=[TARGET_COL], inplace=True) → 理由: モデルはラベルなしデータから学習できない ``` --- ### ルール1 - しきい値チェック（欠損%） ``` ┌────────────────────────────────────────────────────────────┐ │ IF 欠損% > 60%: │ │ → オプションA: 列を完全に削除 │ │ （例外: ドメインが重要と判断した場合 → 専門家にフラグ） │ │ → オプションB: 保持 + バイナリインジケーターフラグ作成 │ │ （col_was_missing = 1）その後、補完を決定 │ │ │ │ IF 30% < 欠損% ≤ 60%: │ │ → 高度な補完を使用: KNNまたはMICE（IterativeImputer） │ │ → 常に最初に欠損インジケーターフラグを作成 │ │ → グループ別（条件付き）平均/最頻値を検討 │ │ │ │ IF 欠損% ≤ 30%: │ │ → ルール2に進む │ └────────────────────────────────────────────────────────────┘ ``` --- ### ルール2 - データ型ルーティング ``` ┌────────────────────────────────────────────────────────────┐ │ 数値 - 連続（float）: │ │ ├─ 対称分布（平均 ≈ 中央値）→ 平均補完 │ │ ├─ 歪んだ分布（外れ値あり）→ 中央値補完 │ │ ├─ 時系列 / 順序行 → 前方補完 / 補間 │ │ ├─ MAR（他の列と相関） → グループ別平均 │ │ └─ 複雑な多変量パターン → KNN / MICE │ │ │ │ 数値 - 離散 / カウント（int）: │ │ ├─ 低カーディナリティ（ユニーク値が少ない） → 最頻値補完 │ │ └─ 高カーディナリティ → 中央値またはKNN │ │ │ │ カテゴリ - 名義（順序なし）: │ │ ├─ 低カーディナリティ → 最頻値補完 │ │ ├─ 高カーディナリティ → "Unknown" / "Missing"を新しいカテゴリとして追加 │ │ └─ MNARの疑い → "Not_Provided"を意味のあるカテゴリとして追加 │ │ │ │ カテゴリ - 順序（ランク順）: │ │ ├─ 自然なランキング → 中央値ランク補完 │ │ └─ MCAR / MAR → 最頻値補完 │ │ │ │ 日時: │ │ ├─ 順序データ → 前方補完 → 後方補完 │ │ └─ ランダムなギャップ → 補間 │ │ │ │ ブール / バイナリ: │ │ └─ 最頻値補完（またはカテゴリとして扱う） │ └────────────────────────────────────────────────────────────┘ ``` --- ### ルール3 - 高度な補完方法の選択ガイド ``` ┌────────────────────────────────────────────────────────────┐ │ 各高度な方法を使用するタイミング │ │ │ │ グループ別平均/最頻値: │ │ → 欠損がグループ列に条件付けられたMARの場合 │ │ → 例: age_groupごとの平均を使用して収入のNaNを補完 │ │ → 全体平均よりも現実的 │ │ │ │ KNNインピューター（デフォルトk=5）: │ │ → 複数の相関する数値列が存在する場合 │ │ → k個の最も近い完全な行を見つけ、その値の平均を取る │ │ → 大規模データセットでは遅い │ │ │ │ MICE / IterativeImputer: │ │ → 最も強力 - 他のすべての列を使用して各列をモデル化 │ │ → 複雑な多変量関係を持つMARに最適 │ │ → 再現性のためにmax_iter=10、random_state=42を使用 │ │ → 計算コストが最も高い │ │ │ │ 欠損インジケーターフラグ: │ │ → MNAR列には常に追加 │ │ → 30%以上の欠損列には任意だが推奨 │ │ → 作成: col_was_missing = NaNの場合は1、それ以外は0 │ │ → モデルに「この値は存在しなかった」というシグナルを伝える │ └────────────────────────────────────────────────────────────┘ ``` --- ### ルール4 - MLモデル互換性 ``` ┌────────────────────────────────────────────────────────────┐ │ ツリーベース（XGBoost、LightGBM、CatBoost、RandomForest）: │ │ → NaNをネイティブに処理可能 │ │ → それでも推奨: MNARのインジケーターフラグを作成 │ │ │ │ 線形モデル（LogReg、LinearReg、Ridge、Lasso）: │ │ → 補完が必須 - NaN許容度ゼロ │ │ │ │ ニューラルネットワーク / ディープラーニング: │ │ → 補完が必須 - NaN許容度なし │ │ │ │ SVM、KNN分類器: │ │ → 補完が必須 - NaN許容度なし │ │ │ │ ⚠️ 全モデル共通の普遍ルール: │ │ → 最初にトレイン/テストを分割 │ │ → インピューターをトレインのみにフィット │ │ → フィットしたインピューターを使用してトレインとテストの両方を変換 │ │ → データセット全体にフィットしない - データリークの原因 │ └────────────────────────────────────────────────────────────┘ ``` --- ## フェーズ4 - Python実装ブループリント ```python from sklearn.pipeline import Pipeline from sklearn.impute import SimpleImputer, KNNImputer from sklearn.experimental import enable_iterative_imputer from sklearn.impute import IterativeImputer from sklearn.model_selection import train_test_split import pandas as pd import numpy as np # ───────────────────────────────────────────────────────────── # ステップ0 - DATA()をロードしてコピー # ───────────────────────────────────────────────────────────── df = DATA().copy() # ───────────────────────────────────────────────────────────── # ステップ1 - 偽装された欠損値を標準化 # ───────────────────────────────────────────────────────────── DISGUISED_NULLS = ["?", "N/A", "n/a", "unknown", "none", " - ", "-", ""] df.replace(DISGUISED_NULLS, np.nan, inplace=True) # ───────────────────────────────────────────────────────────── # ステップ2 - ターゲットが欠損している行を削除（ルール0） # ───────────────────────────────────────────────────────────── TARGET_COL = 'your_target_column' # ← これを変更 df.dropna(subset=[TARGET_COL], axis=0, inplace=True) # ───────────────────────────────────────────────────────────── # ステップ3 - 特徴量とターゲットを分離 # ───────────────────────────────────────────────────────────── X = df.drop(columns=[TARGET_COL]) y = df[TARGET_COL] # ───────────────────────────────────────────────────────────── # ステップ4 - 補完前にトレイン/テスト分割 # ───────────────────────────────────────────────────────────── X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=42 ) # ───────────────────────────────────────────────────────────── # ステップ5 - 列グループを定義（フェーズ1-2の後に記入） # ───────────────────────────────────────────────────────────── num_cols_symmetric = [] # → 平均補完 num_cols_skewed = [] # → 中央値補完 cat_cols_low_card = [] # → 最頻値補完 cat_cols_high_card = [] # → 'Unknown'で埋める knn_cols = [] # → KNN補完 drop_cols = [] # → 削除（>60%欠損またはドメイン上無関係） mnar_cols = [] # → インジケーターフラグ + 補完 # ───────────────────────────────────────────────────────────── # ステップ6 - 欠損が多いまたは無関係な列を削除 # ───────────────────────────────────────────────────────────── X_train = X_train.drop(columns=drop_cols, errors='ignore') X_test = X_test.drop(columns=drop_cols, errors='ignore') # ───────────────────────────────────────────────────────────── # ステップ7 - 補完前に欠損インジケーターフラグを作成 # ───────────────────────────────────────────────────────────── for col in mnar_cols: X_train[f'{col}_was_missing'] = X_train[col].isnull().astype(int) X_test[f'{col}_was_missing'] = X_test[col].isnull().astype(int) # ───────────────────────────────────────────────────────────── # ステップ8 - 数値補完 # ───────────────────────────────────────────────────────────── if num_cols_symmetric: imp_mean = SimpleImputer(strategy='mean') X_train[num_cols_symmetric] = imp_mean.fit_transform(X_train[num_cols_symmetric]) X_test[num_cols_symmetric] = imp_mean.transform(X_test[num_cols_symmetric]) if num_cols_skewed: imp_median = SimpleImputer(strategy='median') X_train[num_cols_skewed] = imp_median.fit_transform(X_train[num_cols_skewed]) X_test[num_cols_skewed] = imp_median.transform(X_test[num_cols_skewed]) # ───────────────────────────────────────────────────────────── # ステップ9 - カテゴリ補完 # ───────────────────────────────────────────────────────────── if cat_cols_low_card: imp_mode = SimpleImputer(strategy='most_frequent') X_train[cat_cols_low_card] = imp_mode.fit_transform(X_train[cat_cols_low_card]) X_test[cat_cols_low_card] = imp_mode.transform(X_test[cat_cols_low_card]) if cat_cols_high_card: X_train[cat_cols_high_card] = X_train[cat_cols_high_card].fillna('Unknown') X_test[cat_cols_high_card] = X_test[cat_cols_high_card].fillna('Unknown') # ───────────────────────────────────────────────────────────── # ステップ10 - グループ別補完（MARパターン） # ───────────────────────────────────────────────────────────── # 例: 'age_group'ごとの平均を使用して'income'のNaNを補完 # GROUP_COL = 'age_group' # TARGET_IMP_COL = 'income' # group_means = X_train.groupby(GROUP_COL)[TARGET_IMP_COL].mean() # X_train[TARGET_IMP_COL] = X_train[TARGET_IMP_COL].fillna( # X_train[GROUP_COL].map(group_means) # ) # X_test[TARGET_IMP_COL] = X_test[TARGET_IMP_COL].fillna( # X_test[GROUP_COL].map(group_means) # ) # ───────────────────────────────────────────────────────────── # ステップ11 - 複雑なパターンに対するKNN補完 # ───────────────────────────────────────────────────────────── if knn_cols: imp_knn = KNNImputer(n_neighbors=5) X_train[knn_cols] = imp_knn.fit_transform(X_train[knn_cols]) X_test[knn_cols] = imp_knn.transform(X_test[knn_cols]) # ───────────────────────────────────────────────────────────── # ステップ12 - MICE / IterativeImputer（最も強力、必要な場合に使用） # ───────────────────────────────────────────────────────────── # imp_iter = IterativeImputer(max_iter=10, random_state=42) # X_train[advanced_cols] = imp_iter.fit_transform(X_train[advanced_cols]) # X_test[advanced_cols] = imp_iter.transform(X_test[advanced_cols]) # ───────────────────────────────────────────────────────────── # ステップ13 - 最終検証 # ───────────────────────────────────────────────────────────── remaining_train = X_train.isnull().sum() remaining_test = X_test.isnull().sum() assert remaining_train.sum() == 0, f"トレインにまだ欠損があります:\n{remaining_train[remaining_train > 0]}" assert remaining_test.sum() == 0, f"テストにまだ欠損があります:\n{remaining_test[remaining_test > 0]}" print("✅ 欠損値は残っていません。DATA()はML対応です。") print(f" トレイン形状: {X_train.shape} | テスト形状: {X_test.shape}") ``` --- ## フェーズ5 - 統合と決定レポート フェーズ1から4を完了した後、この正確なレポートを提供する: ``` ══════════════════════════════════════════════════════════════ 欠損値処理レポート ══════════════════════════════════════════════════════════════ 1. データセット概要 形状 : 総欠損数 : ターゲット列 : MLタスク : モデルタイプ : 2. 欠損インベントリテーブル | 列 | 欠損% | データ型 | メカニズム | 情報的か？ | 処理 | |--------|----------|-------|-----------|--------------|-----------| | ... | ... | ... | ... | ... | ... | 3. 決定ログ [列]: [選択した処理の理由] [列]: [選択した処理の理由] 4. 削除された列 [列] - 理由: [例: 72%欠損、ドメイン上重要でない] 5. 作成されたインジケーターフラグ [col_was_missing] - 理由: [MNARの疑い / 高い欠損%] 6. 使用された補完方法 [列] → [使用戦略 + 正当化] 7. 警告とエッジケース - ドメイン専門家のレビューが必要なMNAR列 - 補完中に行われた仮定 - 完全なEDA後に再評価が必要な列 - 見つかった偽装ヌル値（?、N/A、0など） 8. 次のステップ - 補完後チェックリスト ☐ 補完前後の分布を比較（ヒストグラム） ☐ すべてのインピューターがトレインのみにフィットされたことを確認 ☐ ターゲット列からのデータリークがゼロであることを検証 ☐ 補完後の相関行列を再確認 ☐ 分類タスクの場合はクラスバランスを確認 ☐ 再現性のためにすべての変換を文書化 ══════════════════════════════════════════════════════════════ ``` --- ## 制約とガードレール ``` ✅ 常に実行する必要があること: → df.copy()で作業 - 元のDATA()を変更しない → ターゲット（y）が欠損している行を削除 - yを補完しない → すべてのインピューターをトレインデータのみにフィット → 既にフィットしたインピューターを使用してテストを変換（再フィットなし） → すべてのMNAR列にインジケーターフラグを作成 → モデルに渡す前にヌルがゼロであることを検証 → 偽装された欠損値（?、N/A、0、空白、"unknown"）をチェック → 明示的な理由を付けてすべての決定を文書化 ❌ 絶対に実行してはならないこと: → 分布を確認せずに盲目的に補完 → ドメイン上の重要性を確認せずに列を削除 → トレイン/テスト分割前にデータセット全体にインピューターをフィット（データリーク） → MNAR列を無視 - モデルに深刻なバイアスをかける可能性がある → すべての列に同一の戦略を適用 → NaNが欠損値の唯一の形式であると仮定 ``` --- ## クイックリファレンス - 戦略チートシート | 状況 | 戦略 | |-----------|----------| | ターゲット列（y）にNaN | 行を削除 - 補完しない | | 列 > 60%欠損 | 列を削除（またはインジケーター + 専門家レビュー） | | 数値、対称分布 | 平均補完 | | 数値、歪んだ分布 | 中央値補完 | | 数値、時系列 | 前方補完 / 補間 | | カテゴリ、低カーディナリティ | 最頻値補完 | | カテゴリ、高カーディナリティ | 'Unknown'カテゴリで埋める | | MNARの疑い（任意のタイプ） | インジケーターフラグ + ドメインレビュー | | MAR、グループに条件付け | グループ別平均/最頻値 | | 複雑な多変量パターン | KNNインピューターまたはMICE | | ツリーベースモデル（XGBoostなど） | NaN許容; それでもMNARにフラグ | | 線形 / NN / SVM | 補完必須 - NaN許容度ゼロ | --- *PROMPT() v1.0 - IBM GEN AIエンジニアリング / Pythonによるデータ分析用に構築* *フレームワーク: 思考の連鎖（CoT）+ 思考の木（ToT）* *参照: Coursera - Pythonでの欠損値の扱い*