Python×AI開発入門：機械学習モデルを実際に作って学ぶ実践ガイド2026

# Anacondaダウンロード（公式サイトから）
# インストール後、以下でバージョン確認
conda --version
python --version

# 機械学習専用の仮想環境作成
conda create -n ml-practice python=3.11
conda activate ml-practice

# 基本的な機械学習ライブラリ
pip install pandas numpy scikit-learn matplotlib seaborn jupyter

# または conda を使用
conda install pandas numpy scikit-learn matplotlib seaborn jupyter

# Jupyter Notebookを起動
jupyter notebook

# または Jupyter Lab（より高機能）
pip install jupyterlab
jupyter lab

# ライブラリのインポートテスト
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_squared_error, r2_score

# バージョン確認
print(f"pandas: {pd.__version__}")
print(f"numpy: {np.__version__}")
print(f"scikit-learn: {sklearn.__version__}")

# 正常にインポートできれば OK
print("環境構築完了！")

# データセットの準備
from sklearn.datasets import fetch_california_housing
import pandas as pd

# カリフォルニア住宅データセットを読み込み
housing = fetch_california_housing()
df = pd.DataFrame(housing.data, columns=housing.feature_names)
df['PRICE'] = housing.target

# データの基本情報確認
print("=== データセット基本情報 ===")
print(f"データ数: {len(df)}")
print(f"特徴量数: {len(df.columns) - 1}")
print("\n=== 最初の5行 ===")
print(df.head())

print("\n=== 基本統計量 ===")
print(df.describe())

print("\n=== データ型とnull値 ===")
print(df.info())

import matplotlib.pyplot as plt
import seaborn as sns

# 図のスタイル設定
plt.style.use('seaborn-v0_8')
fig, axes = plt.subplots(2, 2, figsize=(15, 12))

# 価格の分布
axes[0, 0].hist(df['PRICE'], bins=30, alpha=0.7)
axes[0, 0].set_title('住宅価格の分布')
axes[0, 0].set_xlabel('価格（100万円単位）')

# 特徴量と価格の関係（散布図）
axes[0, 1].scatter(df['MedInc'], df['PRICE'], alpha=0.5)
axes[0, 1].set_title('所得と価格の関係')
axes[0, 1].set_xlabel('世帯所得中央値')
axes[0, 1].set_ylabel('価格')

# 築年数と価格の関係
axes[1, 0].scatter(df['HouseAge'], df['PRICE'], alpha=0.5)
axes[1, 0].set_title('築年数と価格の関係')
axes[1, 0].set_xlabel('築年数')
axes[1, 0].set_ylabel('価格')

# 相関行列のヒートマップ
correlation_matrix = df.corr()
sns.heatmap(correlation_matrix, annot=True, cmap='coolwarm',
            center=0, ax=axes[1, 1])
axes[1, 1].set_title('特徴量間の相関')

plt.tight_layout()
plt.show()

# 強い相関を持つ特徴量を特定
print("=== 価格との相関が強い特徴量 ===")
price_correlation = df.corr()['PRICE'].abs().sort_values(ascending=False)
print(price_correlation[1:])  # 価格自体を除く

from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler

# 特徴量と目的変数の分離
X = df.drop('PRICE', axis=1)  # 特徴量
y = df['PRICE']  # 目的変数（価格）

print("特徴量:", X.columns.tolist())
print("目的変数:", 'PRICE')

# 訓練用とテスト用に分割（8:2の比率）
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

print(f"\n訓練データ数: {len(X_train)}")
print(f"テストデータ数: {len(X_test)}")

# 特徴量の標準化
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)

print("\n標準化完了")
print("訓練データの平均:", X_train_scaled.mean(axis=0).round(3))
print("訓練データの標準偏差:", X_train_scaled.std(axis=0).round(3))

from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_squared_error, r2_score, mean_absolute_error

# 線形回帰モデルの作成
model = LinearRegression()

# モデルの訓練
model.fit(X_train_scaled, y_train)

# 予測
y_train_pred = model.predict(X_train_scaled)
y_test_pred = model.predict(X_test_scaled)

print("=== モデル訓練完了 ===")
print("線形回帰モデルのパラメータ:")
for feature, coef in zip(X.columns, model.coef_):
    print(f"  {feature}: {coef:.3f}")
print(f"切片: {model.intercept_:.3f}")

# 評価指標の計算
def evaluate_model(y_true, y_pred, data_type=""):
    """モデルの評価指標を計算・表示する関数"""
    mse = mean_squared_error(y_true, y_pred)
    rmse = np.sqrt(mse)
    mae = mean_absolute_error(y_true, y_pred)
    r2 = r2_score(y_true, y_pred)

    print(f"=== {data_type} データの評価結果 ===")
    print(f"MSE (平均二乗誤差): {mse:.3f}")
    print(f"RMSE (平均二乗誤差平方根): {rmse:.3f}")
    print(f"MAE (平均絶対誤差): {mae:.3f}")
    print(f"R²スコア (決定係数): {r2:.3f}")
    print()

    return {"MSE": mse, "RMSE": rmse, "MAE": mae, "R2": r2}

# 訓練データとテストデータの評価
train_metrics = evaluate_model(y_train, y_train_pred, "訓練")
test_metrics = evaluate_model(y_test, y_test_pred, "テスト")

# 予測結果の可視化
fig, axes = plt.subplots(1, 2, figsize=(15, 6))

# 実際の価格 vs 予測価格（訓練データ）
axes[0].scatter(y_train, y_train_pred, alpha=0.5)
axes[0].plot([y_train.min(), y_train.max()],
             [y_train.min(), y_train.max()], 'r--', lw=2)
axes[0].set_xlabel('実際の価格')
axes[0].set_ylabel('予測価格')
axes[0].set_title(f'訓練データ予測結果 (R² = {train_metrics["R2"]:.3f})')

# 実際の価格 vs 予測価格（テストデータ）
axes[1].scatter(y_test, y_test_pred, alpha=0.5)
axes[1].plot([y_test.min(), y_test.max()],
             [y_test.min(), y_test.max()], 'r--', lw=2)
axes[1].set_xlabel('実際の価格')
axes[1].set_ylabel('予測価格')
axes[1].set_title(f'テストデータ予測結果 (R² = {test_metrics["R2"]:.3f})')

plt.tight_layout()
plt.show()

# サンプルデータの作成
np.random.seed(42)
n_samples = 1000

# 顧客データの生成
data = {
    'age': np.random.normal(35, 12, n_samples),
    'income': np.random.normal(50000, 20000, n_samples),
    'spending_score': np.random.normal(50, 25, n_samples),
    'years_customer': np.random.exponential(3, n_samples),
    'num_purchases': np.random.poisson(12, n_samples)
}

# DataFrameの作成
customer_df = pd.DataFrame(data)

# 年齢は正の値に制限
customer_df['age'] = np.abs(customer_df['age'])
customer_df['income'] = np.abs(customer_df['income'])

# 購買意欲の高低を決定（目的変数）
# 所得、支出スコア、購入回数が高いほど購買意欲が高いとする
purchase_probability = (
    0.3 * (customer_df['income'] / customer_df['income'].max()) +
    0.4 * (customer_df['spending_score'] / customer_df['spending_score'].max()) +
    0.3 * (customer_df['num_purchases'] / customer_df['num_purchases'].max())
)

# ノイズを加えてリアルさを追加
purchase_probability += np.random.normal(0, 0.1, n_samples)
customer_df['high_value_customer'] = (purchase_probability > 0.5).astype(int)

print("=== 顧客分類データセット ===")
print(customer_df.head(10))
print(f"\n高価値顧客の割合: {customer_df['high_value_customer'].mean():.2%}")

# クラス別の統計量
print("=== クラス別統計量 ===")
for class_value in [0, 1]:
    class_name = "一般顧客" if class_value == 0 else "高価値顧客"
    subset = customer_df[customer_df['high_value_customer'] == class_value]
    print(f"\n{class_name} ({len(subset)}人):")
    print(f"  平均年齢: {subset['age'].mean():.1f}歳")
    print(f"  平均所得: {subset['income'].mean():.0f}円")
    print(f"  平均支出スコア: {subset['spending_score'].mean():.1f}")

# 可視化
fig, axes = plt.subplots(2, 2, figsize=(15, 12))

# 年齢分布の比較
for class_value, label, color in [(0, '一般顧客', 'blue'), (1, '高価値顧客', 'red')]:
    subset = customer_df[customer_df['high_value_customer'] == class_value]
    axes[0, 0].hist(subset['age'], alpha=0.6, label=label, color=color, bins=20)
axes[0, 0].set_title('年齢分布の比較')
axes[0, 0].set_xlabel('年齢')
axes[0, 0].legend()

# 所得分布の比較
for class_value, label, color in [(0, '一般顧客', 'blue'), (1, '高価値顧客', 'red')]:
    subset = customer_df[customer_df['high_value_customer'] == class_value]
    axes[0, 1].hist(subset['income'], alpha=0.6, label=label, color=color, bins=20)
axes[0, 1].set_title('所得分布の比較')
axes[0, 1].set_xlabel('所得')
axes[0, 1].legend()

# 所得 vs 支出スコアの散布図
for class_value, label, color in [(0, '一般顧客', 'blue'), (1, '高価値顧客', 'red')]:
    subset = customer_df[customer_df['high_value_customer'] == class_value]
    axes[1, 0].scatter(subset['income'], subset['spending_score'],
                       alpha=0.6, label=label, color=color)
axes[1, 0].set_title('所得 vs 支出スコア')
axes[1, 0].set_xlabel('所得')
axes[1, 0].set_ylabel('支出スコア')
axes[1, 0].legend()

# 特徴量間の相関
correlation_matrix = customer_df.corr()
sns.heatmap(correlation_matrix, annot=True, cmap='coolwarm',
            center=0, ax=axes[1, 1])
axes[1, 1].set_title('特徴量間の相関')

plt.tight_layout()
plt.show()

from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, confusion_matrix, accuracy_score
from sklearn.metrics import roc_auc_score, roc_curve

# 特徴量と目的変数の分離
X = customer_df.drop('high_value_customer', axis=1)
y = customer_df['high_value_customer']

# 訓練・テストデータに分割
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

# 特徴量の標準化
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)

# モデルの作成と訓練
models = {
    "ロジスティック回帰": LogisticRegression(random_state=42),
    "ランダムフォレスト": RandomForestClassifier(n_estimators=100, random_state=42)
}

model_results = {}

for model_name, model in models.items():
    print(f"\n=== {model_name} ===")

    # 訓練（ランダムフォレストは標準化不要）
    if model_name == "ランダムフォレスト":
        model.fit(X_train, y_train)
        y_pred = model.predict(X_test)
        y_pred_proba = model.predict_proba(X_test)[:, 1]
    else:
        model.fit(X_train_scaled, y_train)
        y_pred = model.predict(X_test_scaled)
        y_pred_proba = model.predict_proba(X_test_scaled)[:, 1]

    # 評価
    accuracy = accuracy_score(y_test, y_pred)
    auc_score = roc_auc_score(y_test, y_pred_proba)

    print(f"精度: {accuracy:.3f}")
    print(f"AUCスコア: {auc_score:.3f}")
    print("\n詳細な分類レポート:")
    print(classification_report(y_test, y_pred,
                              target_names=['一般顧客', '高価値顧客']))

    model_results[model_name] = {
        'model': model,
        'accuracy': accuracy,
        'auc': auc_score,
        'y_pred': y_pred,
        'y_pred_proba': y_pred_proba
    }

# ROC曲線の描画
plt.figure(figsize=(12, 5))

# ROC曲線
plt.subplot(1, 2, 1)
for model_name, results in model_results.items():
    fpr, tpr, _ = roc_curve(y_test, results['y_pred_proba'])
    plt.plot(fpr, tpr, label=f"{model_name} (AUC = {results['auc']:.3f})")

plt.plot([0, 1], [0, 1], 'k--', alpha=0.5)
plt.xlabel('偽陽性率')
plt.ylabel('真陽性率')
plt.title('ROC曲線')
plt.legend()

# 特徴量重要度（ランダムフォレスト）
plt.subplot(1, 2, 2)
rf_model = model_results['ランダムフォレスト']['model']
feature_importance = rf_model.feature_importances_
feature_names = X.columns

# 重要度でソート
sorted_indices = np.argsort(feature_importance)[::-1]
plt.bar(range(len(feature_importance)),
        feature_importance[sorted_indices])
plt.xlabel('特徴量')
plt.ylabel('重要度')
plt.title('特徴量重要度（ランダムフォレスト）')
plt.xticks(range(len(feature_names)),
           [feature_names[i] for i in sorted_indices], rotation=45)

plt.tight_layout()
plt.show()

# 混同行列の表示
fig, axes = plt.subplots(1, 2, figsize=(12, 5))
for idx, (model_name, results) in enumerate(model_results.items()):
    cm = confusion_matrix(y_test, results['y_pred'])
    sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', ax=axes[idx])
    axes[idx].set_title(f'{model_name} 混同行列')
    axes[idx].set_xlabel('予測')
    axes[idx].set_ylabel('実際')

plt.tight_layout()
plt.show()

from sklearn.model_selection import GridSearchCV
from sklearn.ensemble import RandomForestClassifier

# ランダムフォレストのハイパーパラメータを最適化
param_grid = {
    'n_estimators': [50, 100, 200],
    'max_depth': [3, 5, 10, None],
    'min_samples_split': [2, 5, 10],
    'min_samples_leaf': [1, 2, 4]
}

print("グリッドサーチ開始...")
rf = RandomForestClassifier(random_state=42)
grid_search = GridSearchCV(rf, param_grid, cv=5, scoring='accuracy', n_jobs=-1)
grid_search.fit(X_train, y_train)

print("=== 最適なハイパーパラメータ ===")
print(grid_search.best_params_)
print(f"最高CV精度: {grid_search.best_score_:.3f}")

# 最適化されたモデルで予測
best_rf = grid_search.best_estimator_
y_pred_optimized = best_rf.predict(X_test)
optimized_accuracy = accuracy_score(y_test, y_pred_optimized)

print(f"\n最適化前精度: {model_results['ランダムフォレスト']['accuracy']:.3f}")
print(f"最適化後精度: {optimized_accuracy:.3f}")
print(f"改善度: {optimized_accuracy - model_results['ランダムフォレスト']['accuracy']:.3f}")

# 新しい特徴量を作成
customer_df_enhanced = customer_df.copy()

# 1. 年齢グループ
customer_df_enhanced['age_group'] = pd.cut(
    customer_df_enhanced['age'],
    bins=[0, 25, 35, 50, 100],
    labels=['young', 'adult', 'middle', 'senior']
)

# 2. 所得レベル
customer_df_enhanced['income_level'] = pd.cut(
    customer_df_enhanced['income'],
    bins=[0, 30000, 50000, 80000, float('inf')],
    labels=['low', 'medium', 'high', 'very_high']
)

# 3. 相互作用項
customer_df_enhanced['income_spending_interaction'] = (
    customer_df_enhanced['income'] * customer_df_enhanced['spending_score']
)

# カテゴリ変数のダミー化
customer_df_encoded = pd.get_dummies(customer_df_enhanced,
                                   columns=['age_group', 'income_level'])

print("=== 特徴量エンジニアリング後 ===")
print(f"元の特徴量数: {len(customer_df.columns) - 1}")
print(f"新しい特徴量数: {len(customer_df_encoded.columns) - 1}")
print("\n新しい特徴量:", customer_df_encoded.columns.tolist())

from sklearn.model_selection import cross_val_score, StratifiedKFold

def evaluate_with_cv(model, X, y, cv_folds=5):
    """クロスバリデーションでモデルを評価"""
    skf = StratifiedKFold(n_splits=cv_folds, shuffle=True, random_state=42)

    cv_scores = cross_val_score(model, X, y, cv=skf, scoring='accuracy')

    print(f"CV精度: {cv_scores}")
    print(f"平均精度: {cv_scores.mean():.3f} (+/- {cv_scores.std() * 2:.3f})")

    return cv_scores

# 複数のモデルをクロスバリデーションで評価
models_cv = {
    "ロジスティック回帰": LogisticRegression(random_state=42),
    "ランダムフォレスト": RandomForestClassifier(n_estimators=100, random_state=42),
    "最適化RF": best_rf
}

X_for_cv = customer_df.drop('high_value_customer', axis=1)
y_for_cv = customer_df['high_value_customer']

print("=== クロスバリデーション結果 ===")
for model_name, model in models_cv.items():
    print(f"\n{model_name}:")
    if model_name == "ロジスティック回帰":
        X_scaled = StandardScaler().fit_transform(X_for_cv)
        evaluate_with_cv(model, X_scaled, y_for_cv)
    else:
        evaluate_with_cv(model, X_for_cv, y_for_cv)

import joblib
import os

# モデル保存用ディレクトリの作成
os.makedirs('models', exist_ok=True)

# 最適化されたモデルの保存
model_filename = 'models/customer_classification_model.pkl'
scaler_filename = 'models/feature_scaler.pkl'

joblib.dump(best_rf, model_filename)
joblib.dump(scaler, scaler_filename)

print(f"モデルを保存しました: {model_filename}")
print(f"スケーラーを保存しました: {scaler_filename}")

# モデルの読み込みテスト
loaded_model = joblib.load(model_filename)
loaded_scaler = joblib.load(scaler_filename)

# 新しいデータで予測テスト
new_customer = np.array([[30, 45000, 60, 2.5, 15]])  # 年齢、所得、支出スコア、顧客年数、購入回数
prediction = loaded_model.predict(new_customer)
prediction_proba = loaded_model.predict_proba(new_customer)[0]

print(f"\n新規顧客の予測:")
print(f"分類結果: {'高価値顧客' if prediction[0] == 1 else '一般顧客'}")
print(f"確率: 一般顧客 {prediction_proba[0]:.2%}, 高価値顧客 {prediction_proba[1]:.2%}")

def monitor_model_performance(model, X_test, y_test, threshold=0.05):
    """モデル性能の監視を行う関数"""

    # 予測
    y_pred = model.predict(X_test)
    y_pred_proba = model.predict_proba(X_test)[:, 1]

    # 基本指標
    accuracy = accuracy_score(y_test, y_pred)
    auc = roc_auc_score(y_test, y_pred_proba)

    # データドリフトの検出（簡易版）
    feature_means = X_test.mean()
    feature_stds = X_test.std()

    monitoring_report = {
        'accuracy': accuracy,
        'auc_score': auc,
        'prediction_count': len(y_pred),
        'positive_ratio': y_pred.mean(),
        'feature_statistics': {
            'means': feature_means.to_dict(),
            'stds': feature_stds.to_dict()
        },
        'timestamp': pd.Timestamp.now().isoformat()
    }

    return monitoring_report

# モニタリング実行
report = monitor_model_performance(best_rf, X_test, y_test)
print("=== モデルパフォーマンス監視レポート ===")
print(f"精度: {report['accuracy']:.3f}")
print(f"AUCスコア: {report['auc_score']:.3f}")
print(f"予測データ数: {report['prediction_count']}")
print(f"陽性予測割合: {report['positive_ratio']:.2%}")

class CustomerClassifier:
    """顧客分類予測システム"""

    def __init__(self, model_path, scaler_path=None):
        """モデルとスケーラーを読み込み"""
        self.model = joblib.load(model_path)
        self.scaler = joblib.load(scaler_path) if scaler_path else None
        self.feature_names = ['age', 'income', 'spending_score',
                             'years_customer', 'num_purchases']

    def predict_single(self, customer_data):
        """単一顧客の予測"""
        # 辞書形式のデータをDataFrameに変換
        if isinstance(customer_data, dict):
            df = pd.DataFrame([customer_data])
        else:
            df = pd.DataFrame([customer_data], columns=self.feature_names)

        # 前処理
        if self.scaler:
            df_scaled = self.scaler.transform(df)
            prediction = self.model.predict(df_scaled)[0]
            probability = self.model.predict_proba(df_scaled)[0]
        else:
            prediction = self.model.predict(df)[0]
            probability = self.model.predict_proba(df)[0]

        return {
            'prediction': '高価値顧客' if prediction == 1 else '一般顧客',
            'confidence': probability[1] if prediction == 1 else probability[0],
            'probability_high_value': probability[1]
        }

    def predict_batch(self, customers_data):
        """複数顧客の一括予測"""
        predictions = []
        for customer in customers_data:
            result = self.predict_single(customer)
            predictions.append(result)
        return predictions

# システムの使用例
classifier = CustomerClassifier('models/customer_classification_model.pkl')

# サンプル顧客データ
sample_customers = [
    {'age': 28, 'income': 35000, 'spending_score': 45, 'years_customer': 1.5, 'num_purchases': 8},
    {'age': 45, 'income': 75000, 'spending_score': 80, 'years_customer': 5.0, 'num_purchases': 25},
    {'age': 35, 'income': 55000, 'spending_score': 65, 'years_customer': 3.0, 'num_purchases': 18}
]

print("=== 顧客分類システム予測結果 ===")
for i, customer in enumerate(sample_customers, 1):
    result = classifier.predict_single(customer)
    print(f"\n顧客{i}: {customer}")
    print(f"予測: {result['prediction']}")
    print(f"信頼度: {result['confidence']:.2%}")
    print(f"高価値顧客確率: {result['probability_high_value']:.2%}")