القائمة
科学データの前処理パイプラインスキル。欠損値補完（KNNImputer/SimpleImputer）、 エンコーディング（LabelEncoder/OneHot/ダミー変数）、スケーリング（Standard/MinMax/Robust/Pareto）、 対数変換、外れ値処理のテンプレートを提供。全 Exp-01〜13 に横断的に適用される基盤スキル。
التثبيت باستخدام Codex أو Claude انسخ هذا Prompt والصقه في Codex أو Claude أو مساعد آخر ليراجع صفحة Skill ويثبّتها لك.
استنادا إلى تصنيف SOC المهني
| name | scientific-data-preprocessing |
| description | 科学データの前処理パイプラインスキル。欠損値補完（KNNImputer/SimpleImputer）、 エンコーディング（LabelEncoder/OneHot/ダミー変数）、スケーリング（Standard/MinMax/Robust/Pareto）、 対数変換、外れ値処理のテンプレートを提供。全 Exp-01〜13 に横断的に適用される基盤スキル。 |
| tu_tools | [{"key":"biotools","name":"bio.tools","description":"データ前処理ツールレジストリ検索"}] |
科学データ解析における前処理の標準パイプライン。データクリーニング、変換、正規化の 一連の手順をトレーサブルかつ再現可能な形で実装するためのテンプレート集。 全 Exp-01〜13 に共通処理として適用される基盤スキル。
Raw Data
├─ Step 1: データ品質チェック
├─ Step 2: 欠損値処理
├─ Step 3: 外れ値処理
├─ Step 4: エンコーディング
├─ Step 5: スケーリング / 変換
└─ Step 6: 品質確認レポート
import pandas as pd
import numpy as np
def data_quality_report(df, name="dataset"):
"""
データ品質レポートを出力する。
前処理方針の決定材料として最初に実行すること。
"""
report = pd.DataFrame({
"dtype": df.dtypes,
"nunique": df.nunique(),
"missing_count": df.isnull().sum(),
"missing_pct": (df.isnull().sum() / len(df) * 100).round(2),
"zeros_count": (df == 0).sum(),
})
# 数値カラムの統計
numeric_cols = df.select_dtypes(include=[np.number]).columns
for col in numeric_cols:
report.loc[col, "mean"] = df[col].mean()
report.loc[col, "std"] = df[col].std()
report.loc[col, "skewness"] = df[col].skew()
report.loc[col, "kurtosis"] = df[col].kurtosis()
print(f"=== Data Quality Report: {name} ===")
print(f"Shape: {df.shape}")
print(f"Duplicated rows: {df.duplicated().sum()}")
print(f"Columns with >50% missing: "
f"{report[report['missing_pct'] > 50].index.tolist()}")
print(report.to_string())
return report
| 手法 | ユースケース | 注意点 |
|---|---|---|
dropna() | 完全ケース解析、欠損 <5% | サンプルサイズ減少 |
fillna(median) | 数値データ、外れ値あり | 分布を歪めうる |
fillna(mean) | 数値データ、正規分布 | 外れ値に敏感 |
SimpleImputer(strategy='most_frequent') | カテゴリカル変数 | |
KNNImputer(n_neighbors=5) | 変数間に相関がある場合 | 計算コスト高い |
IterativeImputer | 多変量欠損パターン (MICE) | scikit-learn experimental |
from sklearn.impute import SimpleImputer, KNNImputer
def handle_missing_values(df, strategy="auto", numeric_cols=None, categorical_cols=None):
"""
欠損値処理パイプライン。
strategy:
"auto" — 欠損率に応じて自動選択
"drop" — 欠損行を除去
"median" — 中央値で補完
"knn" — KNN で補完 (n_neighbors=5)
"""
df = df.copy()
if numeric_cols is None:
numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
if categorical_cols is None:
categorical_cols = df.select_dtypes(include=["object", "category"]).columns.tolist()
if strategy == "auto":
missing_pct = df[numeric_cols].isnull().mean()
# カラムごとに最適戦略を適用
low_missing = missing_pct[missing_pct < 0.05].index.tolist()
mid_missing = missing_pct[(missing_pct >= 0.05) & (missing_pct < 0.30)].index.tolist()
high_missing = missing_pct[missing_pct >= 0.30].index.tolist()
# Low: 中央値補完
if low_missing:
imp = SimpleImputer(strategy="median")
df[low_missing] = imp.fit_transform(df[low_missing])
# Mid: KNN 補完
if mid_missing:
imp = KNNImputer(n_neighbors=5)
df[mid_missing] = imp.fit_transform(df[mid_missing])
# High: 警告表示のみ (ドロップ推奨)
if high_missing:
print(f"WARNING: High missing rate columns ({'>30%'}): {high_missing}")
print("Consider dropping these columns or using domain-specific imputation.")
elif strategy == "drop":
df = df.dropna(subset=numeric_cols)
elif strategy == "median":
imp = SimpleImputer(strategy="median")
df[numeric_cols] = imp.fit_transform(df[numeric_cols])
elif strategy == "knn":
imp = KNNImputer(n_neighbors=5)
df[numeric_cols] = imp.fit_transform(df[numeric_cols])
# カテゴリカル変数 → 最頻値で補完
if categorical_cols:
imp_cat = SimpleImputer(strategy="most_frequent")
df[categorical_cols] = imp_cat.fit_transform(df[categorical_cols])
return df
def detect_outliers(df, columns, method="iqr", threshold=1.5):
"""
外れ値を検出する。
method:
"iqr" — IQR法 (Q1 - threshold*IQR, Q3 + threshold*IQR)
"zscore" — Z-score 法 (|z| > threshold; default threshold=3)
"mad" — MAD (Median Absolute Deviation) 法
"""
outlier_mask = pd.DataFrame(False, index=df.index, columns=columns)
for col in columns:
vals = df[col].dropna()
if method == "iqr":
Q1, Q3 = vals.quantile(0.25), vals.quantile(0.75)
IQR = Q3 - Q1
lower, upper = Q1 - threshold * IQR, Q3 + threshold * IQR
outlier_mask[col] = (df[col] < lower) | (df[col] > upper)
elif method == "zscore":
z = np.abs((df[col] - vals.mean()) / vals.std())
outlier_mask[col] = z > (threshold if threshold != 1.5 else 3)
elif method == "mad":
median = vals.median()
mad = np.median(np.abs(vals - median))
modified_z = 0.6745 * (df[col] - median) / (mad + 1e-10)
outlier_mask[col] = np.abs(modified_z) > (threshold if threshold != 1.5 else 3.5)
summary = pd.DataFrame({
"outlier_count": outlier_mask.sum(),
"outlier_pct": (outlier_mask.sum() / len(df) * 100).round(2),
})
return outlier_mask, summary
def handle_outliers(df, columns, method="iqr", action="clip"):
"""
外れ値を処理する。
action:
"clip" — 境界値にクリッピング
"remove" — 外れ値行を除去
"nan" — NaN に置換（後段で補完）
"""
outlier_mask, summary = detect_outliers(df, columns, method=method)
df = df.copy()
if action == "remove":
any_outlier = outlier_mask.any(axis=1)
df = df[~any_outlier]
elif action == "clip":
for col in columns:
vals = df[col].dropna()
Q1, Q3 = vals.quantile(0.25), vals.quantile(0.75)
IQR = Q3 - Q1
lower, upper = Q1 - 1.5 * IQR, Q3 + 1.5 * IQR
df[col] = df[col].clip(lower, upper)
elif action == "nan":
for col in columns:
df.loc[outlier_mask[col], col] = np.nan
return df, summary
| 手法 | ユースケース | 適用例 |
|---|---|---|
LabelEncoder | 順序カテゴリ、ツリー系モデル | 結晶構造タイプ → 0,1,2 |
pd.get_dummies() | 名義カテゴリ、線形モデル | 材料名 → ダミー変数 |
OrdinalEncoder | 順序のあるカテゴリ（明示的） | 低/中/高 → 0,1,2 |
TargetEncoder | 高カーディナリティ | 化合物 ID |
from sklearn.preprocessing import LabelEncoder, OrdinalEncoder
def encode_categorical(df, columns, method="auto"):
"""
カテゴリカル変数のエンコーディング。
method:
"auto" — nunique に応じて自動選択
"label" — LabelEncoder
"onehot" — pd.get_dummies()
"ordinal" — OrdinalEncoder
"""
df = df.copy()
encoders = {}
for col in columns:
n_unique = df[col].nunique()
if method == "auto":
chosen = "onehot" if n_unique <= 10 else "label"
else:
chosen = method
if chosen == "label":
le = LabelEncoder()
df[col] = le.fit_transform(df[col].astype(str))
encoders[col] = le
elif chosen == "onehot":
dummies = pd.get_dummies(df[col], prefix=col, drop_first=True)
df = pd.concat([df.drop(columns=[col]), dummies], axis=1)
elif chosen == "ordinal":
oe = OrdinalEncoder()
df[col] = oe.fit_transform(df[[col]])
encoders[col] = oe
return df, encoders
| 手法 | 数式 | ユースケース |
|---|---|---|
StandardScaler | $(x - \mu) / \sigma$ | PCA 前、正規分布仮定 |
MinMaxScaler | $(x - x_{min}) / (x_{max} - x_{min})$ | NN 入力、0〜1 に正規化 |
RobustScaler | $(x - Q_2) / (Q_3 - Q_1)$ | 外れ値に頑健 |
| Pareto Scaling | $(x - \bar{x}) / \sqrt{s}$ | メタボロミクス (Exp-07) |
| Log2 Transform | $\log_2(x + 1)$ | カウント・発現量データ |
| Box-Cox | $(x^\lambda - 1)/\lambda$ | 正規性改善（正値のみ） |
from sklearn.preprocessing import StandardScaler, MinMaxScaler, RobustScaler
def scale_features(df, columns, method="standard"):
"""
特徴量スケーリング。
method:
"standard" — StandardScaler (z-score)
"minmax" — MinMaxScaler (0-1)
"robust" — RobustScaler (IQR-based)
"pareto" — Pareto scaling (metabolomics 向き)
"log2" — log2(x + 1) 変換
"""
df = df.copy()
scaler = None
if method == "standard":
scaler = StandardScaler()
df[columns] = scaler.fit_transform(df[columns])
elif method == "minmax":
scaler = MinMaxScaler()
df[columns] = scaler.fit_transform(df[columns])
elif method == "robust":
scaler = RobustScaler()
df[columns] = scaler.fit_transform(df[columns])
elif method == "pareto":
# Pareto scaling: (x - mean) / sqrt(std)
for col in columns:
mean = df[col].mean()
std = df[col].std()
df[col] = (df[col] - mean) / np.sqrt(std + 1e-10)
elif method == "log2":
for col in columns:
df[col] = np.log2(df[col] + 1)
return df, scaler
def preprocessing_pipeline(df, target_col=None, config=None):
"""
全体前処理パイプライン。設定辞書で制御可能。
config 例:
{
"drop_duplicates": True,
"missing_strategy": "auto",
"outlier_method": "iqr",
"outlier_action": "clip",
"encoding_method": "auto",
"scaling_method": "standard",
}
"""
if config is None:
config = {
"drop_duplicates": True,
"missing_strategy": "auto",
"outlier_method": "iqr",
"outlier_action": "clip",
"encoding_method": "auto",
"scaling_method": "standard",
}
n_original = len(df)
# Step 0: 重複除去
if config.get("drop_duplicates", True):
df = df.drop_duplicates()
print(f" Removed {n_original - len(df)} duplicate rows")
# Step 1: カラム分類
numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
categorical_cols = df.select_dtypes(include=["object", "category"]).columns.tolist()
if target_col and target_col in numeric_cols:
numeric_cols.remove(target_col)
if target_col and target_col in categorical_cols:
categorical_cols.remove(target_col)
# Step 2: 欠損値処理
df = handle_missing_values(
df,
strategy=config.get("missing_strategy", "auto"),
numeric_cols=numeric_cols,
categorical_cols=categorical_cols,
)
# Step 3: 外れ値処理
if numeric_cols:
df, outlier_summary = handle_outliers(
df, numeric_cols,
method=config.get("outlier_method", "iqr"),
action=config.get("outlier_action", "clip"),
)
# Step 4: エンコーディング
if categorical_cols:
df, encoders = encode_categorical(
df, categorical_cols,
method=config.get("encoding_method", "auto"),
)
else:
encoders = {}
# Step 5: スケーリング
final_numeric = df.select_dtypes(include=[np.number]).columns.tolist()
if target_col and target_col in final_numeric:
final_numeric.remove(target_col)
if final_numeric:
df, scaler = scale_features(
df, final_numeric,
method=config.get("scaling_method", "standard"),
)
else:
scaler = None
print(f" Preprocessing complete: {df.shape}")
# チェックポイント: 前処理済みデータを永続化（パイプライン連携用）
from pathlib import Path
results_dir = Path("results")
results_dir.mkdir(parents=True, exist_ok=True)
df.to_csv(results_dir / "preprocessed_data.csv", index=False)
print(f" ✔ Preprocessed data saved: results/preprocessed_data.csv ({df.shape})")
# 前処理サマリーをJSONに保存
import json
summary = {
"original_shape": [n_original, len(df.columns)],
"processed_shape": list(df.shape),
"duplicates_removed": n_original - len(df) if config.get("drop_duplicates", True) else 0,
"numeric_columns": len(df.select_dtypes(include=[np.number]).columns),
"categorical_columns": len(df.select_dtypes(include=["object", "category"]).columns),
"scaling_method": config.get("scaling_method", "standard"),
"missing_strategy": config.get("missing_strategy", "auto"),
}
with open(results_dir / "preprocessing_summary.json", "w") as f:
json.dump(summary, f, indent=2, ensure_ascii=False)
print(f" ✔ Preprocessing summary saved: results/preprocessing_summary.json")
return df, {"encoders": encoders, "scaler": scaler}
| TU Key | ツール名 | 連携内容 |
|---|---|---|
biotools | bio.tools | データ前処理ツールレジストリ検索 |
sc.pp.normalize_total(), sc.pp.log1p() — scRNA-seq 固有の正規化LabelEncoder, pd.get_dummies() — 化合物記述子エンコーディングStandardScaler + log2(x+1) — がんデータ前処理StandardScaler, KNNImputer — 毒性予測データMinMaxScaler, RobustScaler — プロセスデータLabelEncoder, ダミー変数 — 材料タイプエンコーディング実験の監査レポート・データ来歴（プロベナンス）生成スキル。 データ変換履歴・使用ツールのバージョン・データ整合性チェックを 含むトレーサビリティレポートを自動生成する。 「監査レポート作成」「データ来歴を記録」「トレーサビリティ」で発火。
派生実験設計スキル。既存の実験をベースに条件を変更した派生実験を 設計する。実験計画法（DOE）に基づくパラメータ探索を支援。 「派生実験を設計して」「条件を変えて実験」「パラメータ探索」で発火。
実験テンプレート生成スキル。研究目的・仮説・手法・実験条件・評価基準・ スケジュールを構造化した実験計画書を自動作成する。 「実験テンプレート作成して」「実験計画を立てて」「実験プロトコルを作成」で発火。
実験結果を論文形式（LaTeX / IMRaD）にエクスポートするスキル。 Introduction・Materials & Methods・Results・Discussion の構造で 出版準備用の原稿を自動生成する。 「論文にして」「LaTeX出力」「出版準備」で発火。
実験結果の査読・レビュースキル。再現性・統計的妥当性・方法論の 健全性を体系的に評価し、構造化されたレビューレポートを生成する。 「レビューして」「査読して」「実験結果を評価して」で発火。
科学技術・学術論文の執筆スキル。IMRaD 標準、Nature/Science 系、ACS 系、IEEE 系、 Elsevier 系のジャーナル形式に対応した論文構成・セクション設計・文章パターンを提供。 「論文を書いて」「Abstract を作成して」「Methods セクションを書いて」で発火。 assets/ に主要ジャーナル形式の Markdown テンプレートを同梱。