第 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による欠損値数の確認¶
入力 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))