Skip to content

第 5 章 ソースコード一覧

書籍第 5 章「前処理と特徴量選択」に掲載しているソースコードの一覧です。番号は書籍の「入力」番号と対応しています。上のバッジから、この章のコードをまとめた Google Colab ノートブックを開けます。

入力 5.1: Titanicデータセットの読み込みと確認

import seaborn as sns

# Titanicデータセットの読み込み
titanic = sns.load_dataset("titanic")
print(titanic.shape)
print(titanic.head())

入力 5.2: isnullによる欠損値数の確認

# 各列の欠損値の数を確認
print(titanic.isnull().sum())

入力 5.3: 列ごとの欠損値の割合の確認

# 欠損値の割合を確認
missing_ratio = titanic.isnull().sum() / len(titanic) * 100
print(missing_ratio[missing_ratio > 0].round(1))

入力 5.4: dropnaによる欠損行の削除

# 欠損値を含む行を削除
df_dropped_rows = titanic.dropna()
print(f"元のデータ: {len(titanic)} 行")
print(f"削除後:     {len(df_dropped_rows)} 行")

入力 5.5: 欠損値を含む行・列の削除

# 特定の列の欠損値を含む行のみ削除
df_age = titanic.dropna(subset=["age"])
print(f"age の欠損行を削除: {len(df_age)} 行")

# 欠損値の多い列を削除
df_no_deck = titanic.drop(columns=["deck"])
print(f"deck 列を削除: {df_no_deck.shape}")

入力 5.6: fillnaによる欠損値の補完

import seaborn as sns

titanic = sns.load_dataset("titanic")

# 平均値で補完
age_mean = titanic["age"].fillna(titanic["age"].mean())
print(f"平均値で補完: 欠損値 {age_mean.isnull().sum()} 件")

# 中央値で補完
age_median = titanic["age"].fillna(titanic["age"].median())
print(f"中央値で補完: 欠損値 {age_median.isnull().sum()} 件")

# 最頻値で補完(カテゴリ変数に適している)
embarked_mode = titanic["embarked"].fillna(
    titanic["embarked"].mode()[0]
)
print(f"最頻値で補完: 欠損値 {embarked_mode.isnull().sum()} 件")
print(f"embarked の最頻値: {titanic['embarked'].mode()[0]}")

入力 5.7: SimpleImputerによる欠損値補完

import pandas as pd
import numpy as np
from sklearn.impute import SimpleImputer

# サンプルデータの作成
df = pd.DataFrame({
    "age": [25, np.nan, 30, 35, np.nan, 28],
    "salary": [300, 450, np.nan, 500, 350, np.nan]
})
print("補完前:")
print(df)

# 平均値で補完する SimpleImputer
imputer = SimpleImputer(strategy="mean")
df_imputed = pd.DataFrame(
    imputer.fit_transform(df),
    columns=df.columns
)
print("\n補完後:")
print(df_imputed)

入力 5.8: StandardScalerによる標準化

import pandas as pd
from sklearn.preprocessing import StandardScaler

# サンプルデータ
df = pd.DataFrame({
    "age": [25, 30, 35, 40, 45],
    "salary": [300, 400, 500, 600, 700]
})
print("変換前:")
print(df)

# 標準化
scaler = StandardScaler()
df_scaled = pd.DataFrame(
    scaler.fit_transform(df),
    columns=df.columns
)
print("\n標準化後:")
print(df_scaled.round(2))
print(f"\n平均: {df_scaled.mean().values.round(2)}")
print(f"標準偏差: {df_scaled.std().values.round(2)}")

入力 5.9: MinMaxScalerによる正規化

import pandas as pd
from sklearn.preprocessing import MinMaxScaler

df = pd.DataFrame({
    "age": [25, 30, 35, 40, 45],
    "salary": [300, 400, 500, 600, 700]
})

# 正規化
scaler = MinMaxScaler()
df_normalized = pd.DataFrame(
    scaler.fit_transform(df),
    columns=df.columns
)
print("正規化後:")
print(df_normalized)

入力 5.10: スケーリング有無によるk-NNの精度比較

import seaborn as sns
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.neighbors import KNeighborsClassifier

# ペンギンデータの読み込みと前処理
penguins = sns.load_dataset("penguins").dropna()
X = penguins[["bill_length_mm", "bill_depth_mm",
              "flipper_length_mm", "body_mass_g"]]
y = penguins["species"]

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

# スケーリングなし
knn = KNeighborsClassifier()
knn.fit(X_train, y_train)
print(f"スケーリングなし: {knn.score(X_test, y_test):.4f}")

# スケーリングあり
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)

knn_scaled = KNeighborsClassifier()
knn_scaled.fit(X_train_scaled, y_train)
print(f"スケーリングあり: {knn_scaled.score(X_test_scaled, y_test):.4f}")

入力 5.11: LabelEncoderによるラベルエンコーディング

import pandas as pd
from sklearn.preprocessing import LabelEncoder

# サンプルデータ
df = pd.DataFrame({
    "color": ["赤", "青", "緑", "赤", "青", "緑", "赤"]
})

# ラベルエンコーディング
le = LabelEncoder()
df["color_encoded"] = le.fit_transform(df["color"])
print(df)
print(f"\nクラス一覧: {le.classes_}")

入力 5.12: get_dummiesによるワンホットエンコーディング

import pandas as pd

df = pd.DataFrame({
    "color": ["赤", "青", "緑", "赤", "青"],
    "size": ["S", "M", "L", "M", "S"],
    "price": [100, 200, 150, 120, 180]
})

# ワンホットエンコーディング
df_encoded = pd.get_dummies(df, columns=["color", "size"])
print(df_encoded)

入力 5.13: drop_firstによる冗長列の除外

import pandas as pd

# drop_first=True で最初のカテゴリを除外
df_encoded = pd.get_dummies(
    df, columns=["color"], drop_first=True
)
print(df_encoded)

入力 5.14: OneHotEncoderによるワンホットエンコーディング

import pandas as pd
from sklearn.preprocessing import OneHotEncoder

df = pd.DataFrame({
    "color": ["赤", "青", "緑", "赤", "青"]
})

# OneHotEncoder の使用
encoder = OneHotEncoder(sparse_output=False)
encoded = encoder.fit_transform(df[["color"]])

# 結果を DataFrame に変換
df_encoded = pd.DataFrame(
    encoded,
    columns=encoder.get_feature_names_out()
)
print(df_encoded)

入力 5.15: OrdinalEncoderによる順序エンコーディング

import pandas as pd
from sklearn.preprocessing import OrdinalEncoder

df = pd.DataFrame({
    "education": ["高校", "大学", "大学院", "高校", "大学"]
})

# 順序を指定してエンコーディング
encoder = OrdinalEncoder(
    categories=[["高校", "大学", "大学院"]]
)
df["education_encoded"] = encoder.fit_transform(
    df[["education"]]
)
print(df)

入力 5.16: 特徴量間の相関行列の計算

import seaborn as sns

# ペンギンデータの読み込み
penguins = sns.load_dataset("penguins").dropna()

# 数値列のみを選択
numeric_cols = penguins.select_dtypes(include="number")

# 相関行列の計算
corr_matrix = numeric_cols.corr()

# 表示用に、値を小数第2位に丸め、長い列名を短縮する
corr_disp = corr_matrix.round(2)
# 行名(index)と列名(columns)に同じ短縮名をまとめて代入
corr_disp.index = corr_disp.columns = [
    "bill_len", "bill_dep", "flip_len", "mass"
]
print(corr_disp)

入力 5.17: 相関行列のヒートマップによる可視化

import matplotlib.pyplot as plt
import seaborn as sns

plt.figure(figsize=(6.2, 4.7))
sns.heatmap(corr_matrix, annot=True, fmt=".2f",
            cmap="coolwarm", center=0)
plt.title("相関行列")
plt.tight_layout()
plt.show()

入力 5.18: VarianceThresholdによる特徴量選択

import pandas as pd
from sklearn.feature_selection import VarianceThreshold

# サンプルデータ(分散が 0 の列を含む)
df = pd.DataFrame({
    "feature_A": [1, 2, 3, 4, 5],
    "feature_B": [10, 10, 10, 10, 10],  # 分散が 0
    "feature_C": [5, 3, 8, 2, 9]
})

# 分散が 0 より大きい特徴量のみを選択
selector = VarianceThreshold(threshold=0.0)
X_selected = selector.fit_transform(df)

# 選択された特徴量を確認
selected_features = df.columns[selector.get_support()]
removed_features = df.columns[~selector.get_support()]
print(f"選択された特徴量: {list(selected_features)}")
print(f"削除された特徴量: {list(removed_features)}")

入力 5.19: ランダムフォレストによる特徴量重要度の確認

import seaborn as sns
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import LabelEncoder

# ペンギンデータの準備
penguins = sns.load_dataset("penguins").dropna()
X = penguins[["bill_length_mm", "bill_depth_mm",
              "flipper_length_mm", "body_mass_g"]]
le = LabelEncoder()
y = le.fit_transform(penguins["species"])

# ランダムフォレストで特徴量重要度を計算
rf = RandomForestClassifier(n_estimators=100, random_state=42)
rf.fit(X, y)

# 特徴量重要度の表示
importance = pd.Series(
    rf.feature_importances_, index=X.columns
).sort_values(ascending=False)
print(importance.round(4))

# 棒グラフで可視化
importance.plot(kind="barh")
plt.xlabel("特徴量重要度")
plt.title("ランダムフォレストの特徴量重要度")
plt.tight_layout()
plt.show()

入力 5.20: SelectKBestによる特徴量選択

import seaborn as sns
import pandas as pd
from sklearn.feature_selection import SelectKBest, f_classif
from sklearn.preprocessing import LabelEncoder

# データの準備
penguins = sns.load_dataset("penguins").dropna()
X = penguins[["bill_length_mm", "bill_depth_mm",
              "flipper_length_mm", "body_mass_g"]]
le = LabelEncoder()
y = le.fit_transform(penguins["species"])

# 上位2つの特徴量を選択
selector = SelectKBest(score_func=f_classif, k=2)
X_selected = selector.fit_transform(X, y)

# 選択された特徴量を確認
selected = X.columns[selector.get_support()]
print(f"選択された特徴量: {list(selected)}")

# 各特徴量のスコア
scores = pd.Series(
    selector.scores_, index=X.columns
).sort_values(ascending=False)
print(f"\n特徴量スコア:")
print(scores.round(2))

入力 5.21: 基本的なパイプラインの構築

import seaborn as sns
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler
from sklearn.neighbors import KNeighborsClassifier
from sklearn.model_selection import train_test_split

# データの準備(欠損値はパイプラインの中で補完する)
penguins = sns.load_dataset("penguins")
X = penguins[["bill_length_mm", "bill_depth_mm",
              "flipper_length_mm", "body_mass_g"]]
y = penguins["species"]

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

# パイプラインの構築
pipeline = Pipeline([
    ("imputer", SimpleImputer(strategy="mean")),  # 1. 欠損値を平均値で補完
    ("scaler", StandardScaler()),                 # 2. 標準化
    ("classifier", KNeighborsClassifier())        # 3. k近傍法で分類
])

# 学習と予測(前処理とモデルが一括で実行される)
pipeline.fit(X_train, y_train)
score = pipeline.score(X_test, y_test)
print(f"パイプラインの精度: {score:.4f}")

入力 5.22: ColumnTransformerによる列の種類ごとの前処理

import seaborn as sns
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.neighbors import KNeighborsClassifier
from sklearn.model_selection import train_test_split

# Titanicデータの準備
titanic = sns.load_dataset("titanic")
features = ["age", "fare", "pclass", "sex", "embarked"]
X = titanic[features]
y = titanic["survived"]

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

# 数値列とカテゴリ列を定義
numeric_features = ["age", "fare"]
categorical_features = ["pclass", "sex", "embarked"]

# 数値列の前処理パイプライン
numeric_transformer = Pipeline([
    ("imputer", SimpleImputer(strategy="median")),
    ("scaler", StandardScaler())
])

# カテゴリ列の前処理パイプライン
categorical_transformer = Pipeline([
    ("imputer", SimpleImputer(strategy="most_frequent")),
    ("encoder", OneHotEncoder(handle_unknown="ignore"))
])

# ColumnTransformer で統合
preprocessor = ColumnTransformer([
    ("num", numeric_transformer, numeric_features),
    ("cat", categorical_transformer, categorical_features)
])

# 前処理 + モデルのパイプライン
full_pipeline = Pipeline([
    ("preprocessor", preprocessor),
    ("classifier", KNeighborsClassifier())
])

# 学習と評価
full_pipeline.fit(X_train, y_train)
score = full_pipeline.score(X_test, y_test)
print(f"Titanic 生存予測の精度: {score:.4f}")

入力 5.23: cross_val_scoreによる交差検証

import seaborn as sns
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.neighbors import KNeighborsClassifier
from sklearn.model_selection import cross_val_score

# データの準備
penguins = sns.load_dataset("penguins").dropna()
X = penguins[["bill_length_mm", "bill_depth_mm",
              "flipper_length_mm", "body_mass_g"]]
y = penguins["species"]

# パイプラインの構築
pipeline = Pipeline([
    ("scaler", StandardScaler()),
    ("classifier", KNeighborsClassifier())
])

# 5分割交差検証
scores = cross_val_score(pipeline, X, y, cv=5)
print(f"各フォールドのスコア: {scores.round(4)}")
print(f"平均スコア: {scores.mean():.4f}")
print(f"標準偏差:   {scores.std():.4f}")

入力 5.24: テストデータの分離と交差検証の併用

from sklearn.model_selection import train_test_split
from sklearn.model_selection import cross_val_score

# まずテストデータを分離(最終評価用)
X_trainval, X_test, y_trainval, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

# 訓練+検証データで交差検証
scores = cross_val_score(pipeline, X_trainval, y_trainval, cv=5)
print(f"交差検証スコア: {scores.mean():.4f}")

# 最終評価はテストデータで 1 回だけ行う
pipeline.fit(X_trainval, y_trainval)
test_score = pipeline.score(X_test, y_test)
print(f"テストスコア:   {test_score:.4f}")

入力 5.25: GridSearchCVによるハイパーパラメータ探索

import seaborn as sns
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.neighbors import KNeighborsClassifier
from sklearn.model_selection import GridSearchCV
from sklearn.model_selection import train_test_split

# データの準備
penguins = sns.load_dataset("penguins").dropna()
X = penguins[["bill_length_mm", "bill_depth_mm",
              "flipper_length_mm", "body_mass_g"]]
y = penguins["species"]

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

# パイプラインの構築
pipeline = Pipeline([
    ("scaler", StandardScaler()),
    ("classifier", KNeighborsClassifier())
])

# 探索するパラメータの候補(「ステップ名__パラメータ名」で指定)
param_grid = {
    "classifier__n_neighbors": [3, 5, 7, 9, 11],       # k の候補
    "classifier__weights": ["uniform", "distance"]     # 近傍の重み付け方法
}

# GridSearchCV の実行
grid_search = GridSearchCV(
    pipeline,
    param_grid,
    cv=5,
    scoring="accuracy",
    n_jobs=-1
)
grid_search.fit(X_train, y_train)

# 結果の表示
print(f"最適なパラメータ: {grid_search.best_params_}")
print(f"最高スコア(交差検証): {grid_search.best_score_:.4f}")
print(f"テストスコア: {grid_search.score(X_test, y_test):.4f}")

入力 5.26: 交差検証による複数モデルの比較

import seaborn as sns
import pandas as pd
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import cross_val_score
from sklearn.neighbors import KNeighborsClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.tree import DecisionTreeClassifier
from sklearn.ensemble import RandomForestClassifier

# データの準備
penguins = sns.load_dataset("penguins").dropna()
X = penguins[["bill_length_mm", "bill_depth_mm",
              "flipper_length_mm", "body_mass_g"]]
y = penguins["species"]

# 比較するモデルの定義
models = {
    "k-NN": KNeighborsClassifier(),
    "Logistic": LogisticRegression(max_iter=1000),
    "Decision Tree": DecisionTreeClassifier(random_state=42),
    "Random Forest": RandomForestClassifier(random_state=42)
}

# 各モデルの交差検証スコアを計算
results = []
for name, model in models.items():
    pipeline = Pipeline([
        ("scaler", StandardScaler()),
        ("classifier", model)
    ])
    scores = cross_val_score(pipeline, X, y, cv=5)
    results.append({
        "Model": name,
        "Mean Score": scores.mean(),
        "Std": scores.std()
    })

# 結果を DataFrame で表示
df_results = pd.DataFrame(results)
df_results = df_results.sort_values(
    "Mean Score", ascending=False
)
print(df_results.round(4).to_string(index=False))