← ロードマップ一覧に戻る

データサイエンティストロードマップ

データサイエンス機械学習深層学習PythonRTensorFlowPyTorchMLOps

技術

データサイエンティストロードマップ

概要

データサイエンティストは、大量のデータから価値ある洞察を導き出し、ビジネスの意思決定を支援する専門家です。2025年において、データサイエンスはAI駆動の自動化と個人化体験の需要急増により、生成AIとMLOpsのスキルが必須となっています。機械学習モデルの開発から本番環境への展開まで、エンドツーエンドのソリューションを構築できる能力が求められています。

詳細

フェーズ1: 基礎固め（3-6ヶ月）

プログラミング基礎

  • Python完全習得

    • データ型と制御構造
    • 関数とクラス
    • デコレーターとジェネレーター
    • 仮想環境管理（venv、conda）
    • Jupyter Notebook/Lab

  • R言語（統計分析）

    • データフレーム操作
    • ggplot2によるビジュアライゼーション
    • dplyrでのデータ操作
    • 統計的検定と推論

数学・統計基礎

  • 線形代数

    • ベクトルと行列演算
    • 固有値と固有ベクトル
    • 特異値分解（SVD）
    • 主成分分析（PCA）

  • 統計学

    • 記述統計と推論統計
    • 確率分布
    • 仮説検定
    • ベイズ統計

  • 微積分

    • 偏微分
    • 勾配降下法
    • 最適化理論

データ操作とSQL

  • SQL完全理解

    • 複雑なJOIN操作
    • ウィンドウ関数
    • CTEとサブクエリ
    • インデックス最適化

  • データ操作ライブラリ

    • pandas（データフレーム操作）
    • NumPy（数値計算）
    • Polars（高速データ処理）

フェーズ2: 機械学習基礎（6-12ヶ月）

機械学習アルゴリズム

  • 教師あり学習

    • 線形回帰・ロジスティック回帰
    • 決定木とランダムフォレスト
    • 勾配ブースティング（XGBoost、LightGBM）
    • サポートベクターマシン（SVM）
    • ニューラルネットワーク基礎

  • 教師なし学習

    • K-means クラスタリング
    • 階層的クラスタリング
    • DBSCAN
    • 次元削減（PCA、t-SNE、UMAP）

  • scikit-learn マスタリー

    • パイプライン構築
    • 交差検証
    • ハイパーパラメータチューニング
    • モデル評価指標

データ可視化

  • ビジュアライゼーションツール

    • Matplotlib（基本プロット）
    • Seaborn（統計的可視化）
    • Plotly（インタラクティブ可視化）
    • Altair（宣言的可視化）

  • ダッシュボード作成

    • Streamlit
    • Dash
    • Gradio
    • Panel

特徴量エンジニアリング

  • 特徴量作成

    • 数値変換
    • カテゴリカル変数のエンコーディング
    • 時系列特徴量
    • テキスト特徴量

  • 特徴量選択

    • 相関分析
    • 相互情報量
    • 順列重要度
    • SHAP値

フェーズ3: 深層学習とMLOps（12-18ヶ月）

深層学習フレームワーク

  • TensorFlow/Keras

    • Sequential API
    • Functional API
    • カスタムレイヤーとモデル
    • TensorFlow Serving

  • PyTorch

    • テンソル操作
    • 自動微分
    • カスタムデータセット
    • PyTorch Lightning

深層学習アーキテクチャ

  • コンピュータビジョン

    • CNN（畳み込みニューラルネットワーク）
    • 転移学習（ResNet、EfficientNet）
    • 物体検出（YOLO、Faster R-CNN）
    • セグメンテーション

  • 自然言語処理

    • RNNとLSTM
    • Transformer
    • BERT、GPT
    • ファインチューニング

  • 生成AI

    • VAE（変分オートエンコーダー）
    • GAN（敵対的生成ネットワーク）
    • Diffusion Models
    • LLMの活用

MLOps実践

  • 実験管理

    • MLflow
    • Weights & Biases
    • Neptune.ai
    • DVC（Data Version Control）

  • モデルデプロイメント

    • Docker化
    • Kubernetes
    • モデルサービング（TensorFlow Serving、TorchServe）
    • API開発（FastAPI、Flask）

  • モニタリング

    • データドリフト検出
    • モデル性能監視
    • Evidently
    • WhyLabs

フェーズ4: 高度なデータサイエンスと専門分野（18-24ヶ月）

ビッグデータ技術

  • 分散処理

    • Apache Spark
    • Dask
    • Ray
    • PySpark

  • データパイプライン

    • Apache Airflow
    • Prefect
    • Dagster
    • Apache Kafka

クラウドプラットフォーム

  • AWS

    • SageMaker
    • EMR
    • Glue
    • Lambda

  • Google Cloud

    • Vertex AI
    • BigQuery
    • Dataflow
    • AI Platform

  • Azure

    • Azure Machine Learning
    • Databricks
    • Synapse Analytics

専門分野

  • 時系列予測

    • ARIMA、SARIMA
    • Prophet
    • LSTM for Time Series
    • 状態空間モデル

  • 推薦システム

    • 協調フィルタリング
    • コンテンツベースフィルタリング
    • ハイブリッド手法
    • 深層学習ベース推薦

  • 因果推論

    • A/Bテスト設計
    • 傾向スコアマッチング
    • 差分の差分法
    • 因果フォレスト

エッジAIとオンデバイスML

  • モデル最適化

    • TensorFlow Lite
    • ONNX Runtime
    • モデル量子化
    • プルーニング

  • エッジデバイス

    • NVIDIA Jetson
    • Google Coral
    • Intel Neural Compute Stick

メリット・デメリット

メリット

  • 高い需要と報酬: データサイエンティストは最も需要の高い職種の一つで、高い報酬が期待できる
  • 幅広い業界での活躍: 金融、医療、小売、製造業など、あらゆる業界でデータサイエンスが必要
  • 問題解決の面白さ: 複雑なビジネス課題をデータで解決する創造的な仕事
  • 最先端技術: AI/MLの最新技術を常に扱える環境
  • 影響力: データに基づく意思決定で組織に大きなインパクトを与えられる

デメリット

  • 継続的な学習: 技術の進歩が速く、常に新しいスキルの習得が必要
  • データ品質の課題: 現実のデータは汚く、前処理に多くの時間を費やす
  • 期待値管理: AIへの過度な期待と現実のギャップに対処する必要
  • 説明責任: モデルの判断根拠を非技術者に説明する難しさ
  • 倫理的課題: バイアスやプライバシーなど、AIの倫理的問題への対処

参考ページ

書き方の例

基本的な機械学習パイプライン

# ml_pipeline.py
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split, cross_val_score, GridSearchCV
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, confusion_matrix, roc_auc_score
from sklearn.pipeline import Pipeline
import joblib
import matplotlib.pyplot as plt
import seaborn as sns

# データ読み込みとEDA
class DataProcessor:
    def __init__(self, filepath):
        self.data = pd.read_csv(filepath)
        self.X = None
        self.y = None
        
    def explore_data(self):
        """データの基本的な探索"""
        print("データ形状:", self.data.shape)
        print("\nデータ型:")
        print(self.data.dtypes)
        print("\n欠損値:")
        print(self.data.isnull().sum())
        print("\n基本統計量:")
        print(self.data.describe())
        
        # 相関行列の可視化
        numeric_cols = self.data.select_dtypes(include=[np.number]).columns
        plt.figure(figsize=(10, 8))
        sns.heatmap(self.data[numeric_cols].corr(), annot=True, cmap='coolwarm')
        plt.title('Feature Correlation Matrix')
        plt.show()
        
    def preprocess(self, target_col):
        """データの前処理"""
        # 特徴量とターゲットの分離
        self.X = self.data.drop(columns=[target_col])
        self.y = self.data[target_col]
        
        # カテゴリカル変数の処理
        categorical_cols = self.X.select_dtypes(include=['object']).columns
        for col in categorical_cols:
            le = LabelEncoder()
            self.X[col] = le.fit_transform(self.X[col].astype(str))
        
        # 欠損値の処理
        self.X = self.X.fillna(self.X.mean())
        
        return self.X, self.y

# モデル学習と評価
class MLPipeline:
    def __init__(self, X, y):
        self.X_train, self.X_test, self.y_train, self.y_test = train_test_split(
            X, y, test_size=0.2, random_state=42, stratify=y
        )
        self.pipeline = None
        self.best_model = None
        
    def create_pipeline(self):
        """パイプラインの作成"""
        self.pipeline = Pipeline([
            ('scaler', StandardScaler()),
            ('classifier', RandomForestClassifier(random_state=42))
        ])
        
    def hyperparameter_tuning(self):
        """ハイパーパラメータチューニング"""
        param_grid = {
            'classifier__n_estimators': [100, 200, 300],
            'classifier__max_depth': [10, 20, None],
            'classifier__min_samples_split': [2, 5, 10],
            'classifier__min_samples_leaf': [1, 2, 4]
        }
        
        grid_search = GridSearchCV(
            self.pipeline,
            param_grid,
            cv=5,
            scoring='roc_auc',
            n_jobs=-1,
            verbose=1
        )
        
        grid_search.fit(self.X_train, self.y_train)
        self.best_model = grid_search.best_estimator_
        
        print("最適パラメータ:", grid_search.best_params_)
        print("最高スコア:", grid_search.best_score_)
        
    def evaluate_model(self):
        """モデルの評価"""
        # 予測
        y_pred = self.best_model.predict(self.X_test)
        y_pred_proba = self.best_model.predict_proba(self.X_test)[:, 1]
        
        # 評価指標
        print("\n分類レポート:")
        print(classification_report(self.y_test, y_pred))
        
        print("\nROC-AUC スコア:", roc_auc_score(self.y_test, y_pred_proba))
        
        # 混同行列
        cm = confusion_matrix(self.y_test, y_pred)
        plt.figure(figsize=(8, 6))
        sns.heatmap(cm, annot=True, fmt='d', cmap='Blues')
        plt.title('Confusion Matrix')
        plt.ylabel('True Label')
        plt.xlabel('Predicted Label')
        plt.show()
        
        # 特徴量重要度
        feature_importance = self.best_model.named_steps['classifier'].feature_importances_
        feature_names = self.X_train.columns
        
        importance_df = pd.DataFrame({
            'feature': feature_names,
            'importance': feature_importance
        }).sort_values('importance', ascending=False)
        
        plt.figure(figsize=(10, 8))
        sns.barplot(data=importance_df.head(15), x='importance', y='feature')
        plt.title('Top 15 Feature Importances')
        plt.show()
        
    def save_model(self, filepath):
        """モデルの保存"""
        joblib.dump(self.best_model, filepath)
        print(f"モデルを {filepath} に保存しました")

# 使用例
if __name__ == "__main__":
    # データ処理
    processor = DataProcessor('data.csv')
    processor.explore_data()
    X, y = processor.preprocess('target')
    
    # モデル学習
    ml_pipeline = MLPipeline(X, y)
    ml_pipeline.create_pipeline()
    ml_pipeline.hyperparameter_tuning()
    ml_pipeline.evaluate_model()
    ml_pipeline.save_model('best_model.pkl')

深層学習による画像分類

# deep_learning_cnn.py
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
import numpy as np
import matplotlib.pyplot as plt
from sklearn.model_selection import train_test_split

class ImageClassifier:
    def __init__(self, input_shape, num_classes):
        self.input_shape = input_shape
        self.num_classes = num_classes
        self.model = None
        self.history = None
        
    def build_cnn_model(self):
        """CNNモデルの構築"""
        self.model = keras.Sequential([
            # 畳み込み層ブロック1
            layers.Conv2D(32, (3, 3), activation='relu', input_shape=self.input_shape),
            layers.BatchNormalization(),
            layers.MaxPooling2D((2, 2)),
            layers.Dropout(0.25),
            
            # 畳み込み層ブロック2
            layers.Conv2D(64, (3, 3), activation='relu'),
            layers.BatchNormalization(),
            layers.MaxPooling2D((2, 2)),
            layers.Dropout(0.25),
            
            # 畳み込み層ブロック3
            layers.Conv2D(128, (3, 3), activation='relu'),
            layers.BatchNormalization(),
            layers.MaxPooling2D((2, 2)),
            layers.Dropout(0.25),
            
            # 全結合層
            layers.Flatten(),
            layers.Dense(256, activation='relu'),
            layers.BatchNormalization(),
            layers.Dropout(0.5),
            layers.Dense(self.num_classes, activation='softmax')
        ])
        
        # モデルのコンパイル
        self.model.compile(
            optimizer=keras.optimizers.Adam(learning_rate=0.001),
            loss='categorical_crossentropy',
            metrics=['accuracy', keras.metrics.TopKCategoricalAccuracy(k=3)]
        )
        
        self.model.summary()
        
    def create_data_augmentation(self):
        """データ拡張の設定"""
        data_augmentation = keras.Sequential([
            layers.RandomFlip("horizontal"),
            layers.RandomRotation(0.1),
            layers.RandomZoom(0.1),
            layers.RandomBrightness(0.1),
            layers.RandomContrast(0.1),
        ])
        return data_augmentation
        
    def train(self, X_train, y_train, X_val, y_val, epochs=50):
        """モデルの学習"""
        # コールバックの設定
        callbacks = [
            keras.callbacks.EarlyStopping(
                monitor='val_loss',
                patience=10,
                restore_best_weights=True
            ),
            keras.callbacks.ReduceLROnPlateau(
                monitor='val_loss',
                factor=0.5,
                patience=5,
                min_lr=1e-7
            ),
            keras.callbacks.ModelCheckpoint(
                'best_model.h5',
                monitor='val_accuracy',
                save_best_only=True
            )
        ]
        
        # データ拡張を適用
        data_augmentation = self.create_data_augmentation()
        
        # 学習
        self.history = self.model.fit(
            X_train, y_train,
            batch_size=32,
            epochs=epochs,
            validation_data=(X_val, y_val),
            callbacks=callbacks,
            verbose=1
        )
        
    def plot_training_history(self):
        """学習履歴の可視化"""
        fig, axes = plt.subplots(1, 2, figsize=(15, 5))
        
        # 損失のプロット
        axes[0].plot(self.history.history['loss'], label='Training Loss')
        axes[0].plot(self.history.history['val_loss'], label='Validation Loss')
        axes[0].set_title('Model Loss')
        axes[0].set_xlabel('Epoch')
        axes[0].set_ylabel('Loss')
        axes[0].legend()
        axes[0].grid(True)
        
        # 精度のプロット
        axes[1].plot(self.history.history['accuracy'], label='Training Accuracy')
        axes[1].plot(self.history.history['val_accuracy'], label='Validation Accuracy')
        axes[1].set_title('Model Accuracy')
        axes[1].set_xlabel('Epoch')
        axes[1].set_ylabel('Accuracy')
        axes[1].legend()
        axes[1].grid(True)
        
        plt.tight_layout()
        plt.show()
        
    def evaluate(self, X_test, y_test):
        """モデルの評価"""
        test_loss, test_accuracy, test_top3_accuracy = self.model.evaluate(
            X_test, y_test, verbose=0
        )
        
        print(f"\nテスト損失: {test_loss:.4f}")
        print(f"テスト精度: {test_accuracy:.4f}")
        print(f"Top-3精度: {test_top3_accuracy:.4f}")
        
        # 予測結果のサンプル表示
        predictions = self.model.predict(X_test[:10])
        
        fig, axes = plt.subplots(2, 5, figsize=(15, 6))
        axes = axes.ravel()
        
        for i in range(10):
            axes[i].imshow(X_test[i])
            axes[i].set_title(f"Pred: {np.argmax(predictions[i])}, True: {np.argmax(y_test[i])}")
            axes[i].axis('off')
            
        plt.tight_layout()
        plt.show()

# 転移学習の実装
class TransferLearningClassifier:
    def __init__(self, input_shape, num_classes):
        self.input_shape = input_shape
        self.num_classes = num_classes
        self.model = None
        
    def build_transfer_model(self, base_model_name='EfficientNetB0'):
        """転移学習モデルの構築"""
        # ベースモデルの読み込み
        if base_model_name == 'EfficientNetB0':
            base_model = keras.applications.EfficientNetB0(
                input_shape=self.input_shape,
                include_top=False,
                weights='imagenet'
            )
        elif base_model_name == 'ResNet50':
            base_model = keras.applications.ResNet50(
                input_shape=self.input_shape,
                include_top=False,
                weights='imagenet'
            )
        
        # ベースモデルの凍結
        base_model.trainable = False
        
        # カスタムヘッドの追加
        inputs = keras.Input(shape=self.input_shape)
        x = base_model(inputs, training=False)
        x = layers.GlobalAveragePooling2D()(x)
        x = layers.Dense(256, activation='relu')(x)
        x = layers.BatchNormalization()(x)
        x = layers.Dropout(0.5)(x)
        outputs = layers.Dense(self.num_classes, activation='softmax')(x)
        
        self.model = keras.Model(inputs, outputs)
        
        # コンパイル
        self.model.compile(
            optimizer=keras.optimizers.Adam(learning_rate=0.001),
            loss='categorical_crossentropy',
            metrics=['accuracy']
        )
        
    def fine_tune(self, base_model_layers_to_unfreeze=20):
        """ファインチューニング"""
        # ベースモデルの一部を解凍
        base_model = self.model.layers[1]
        base_model.trainable = True
        
        # 最後のN層のみを学習可能にする
        for layer in base_model.layers[:-base_model_layers_to_unfreeze]:
            layer.trainable = False
            
        # 低い学習率で再コンパイル
        self.model.compile(
            optimizer=keras.optimizers.Adam(learning_rate=0.0001),
            loss='categorical_crossentropy',
            metrics=['accuracy']
        )

MLOpsパイプライン

# mlops_pipeline.py
import mlflow
import mlflow.sklearn
import mlflow.tensorflow
from mlflow.tracking import MlflowClient
import pandas as pd
import numpy as np
from datetime import datetime
import json
import yaml

class MLOpsWorkflow:
    def __init__(self, experiment_name):
        self.experiment_name = experiment_name
        mlflow.set_experiment(experiment_name)
        self.client = MlflowClient()
        
    def log_dataset_info(self, X_train, X_test, y_train, y_test):
        """データセット情報のログ"""
        dataset_info = {
            'train_samples': len(X_train),
            'test_samples': len(X_test),
            'features': X_train.shape[1],
            'target_distribution_train': dict(pd.Series(y_train).value_counts()),
            'target_distribution_test': dict(pd.Series(y_test).value_counts())
        }
        
        mlflow.log_params(dataset_info)
        
    def train_and_log_model(self, model, X_train, y_train, X_test, y_test, model_name):
        """モデルの学習とログ"""
        with mlflow.start_run(run_name=f"{model_name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"):
            # データセット情報のログ
            self.log_dataset_info(X_train, X_test, y_train, y_test)
            
            # モデルのハイパーパラメータをログ
            mlflow.log_params(model.get_params())
            
            # モデル学習
            model.fit(X_train, y_train)
            
            # 予測と評価
            y_pred = model.predict(X_test)
            y_pred_proba = model.predict_proba(X_test)[:, 1] if hasattr(model, 'predict_proba') else None
            
            # メトリクスの計算とログ
            from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, roc_auc_score
            
            metrics = {
                'accuracy': accuracy_score(y_test, y_pred),
                'precision': precision_score(y_test, y_pred, average='weighted'),
                'recall': recall_score(y_test, y_pred, average='weighted'),
                'f1_score': f1_score(y_test, y_pred, average='weighted')
            }
            
            if y_pred_proba is not None:
                metrics['roc_auc'] = roc_auc_score(y_test, y_pred_proba)
                
            mlflow.log_metrics(metrics)
            
            # モデルの保存
            mlflow.sklearn.log_model(
                model, 
                model_name,
                registered_model_name=model_name,
                input_example=X_train[:5]
            )
            
            # 特徴量重要度の保存（ツリーベースモデルの場合）
            if hasattr(model, 'feature_importances_'):
                feature_importance = pd.DataFrame({
                    'feature': X_train.columns,
                    'importance': model.feature_importances_
                }).sort_values('importance', ascending=False)
                
                # 特徴量重要度の可視化
                import matplotlib.pyplot as plt
                fig, ax = plt.subplots(figsize=(10, 8))
                feature_importance.head(20).plot.barh(x='feature', y='importance', ax=ax)
                plt.title('Feature Importances')
                plt.tight_layout()
                mlflow.log_figure(fig, 'feature_importances.png')
                plt.close()
                
            return model, metrics

# モデルモニタリング
class ModelMonitor:
    def __init__(self, model, reference_data):
        self.model = model
        self.reference_data = reference_data
        self.monitoring_results = []
        
    def detect_data_drift(self, new_data):
        """データドリフトの検出"""
        from scipy import stats
        
        drift_results = {}
        
        for column in self.reference_data.columns:
            if self.reference_data[column].dtype in ['float64', 'int64']:
                # 数値データの場合：KS検定
                statistic, p_value = stats.ks_2samp(
                    self.reference_data[column],
                    new_data[column]
                )
                drift_results[column] = {
                    'test': 'ks_test',
                    'statistic': statistic,
                    'p_value': p_value,
                    'drift_detected': p_value < 0.05
                }
            else:
                # カテゴリカルデータの場合：カイ二乗検定
                ref_dist = self.reference_data[column].value_counts(normalize=True)
                new_dist = new_data[column].value_counts(normalize=True)
                
                # 分布の整合性を確保
                all_categories = set(ref_dist.index) | set(new_dist.index)
                ref_dist = ref_dist.reindex(all_categories, fill_value=0)
                new_dist = new_dist.reindex(all_categories, fill_value=0)
                
                statistic, p_value = stats.chisquare(new_dist, ref_dist)
                drift_results[column] = {
                    'test': 'chi_square',
                    'statistic': statistic,
                    'p_value': p_value,
                    'drift_detected': p_value < 0.05
                }
                
        return drift_results
        
    def monitor_model_performance(self, new_data, new_labels):
        """モデル性能のモニタリング"""
        predictions = self.model.predict(new_data)
        
        from sklearn.metrics import accuracy_score, precision_score, recall_score
        
        performance = {
            'timestamp': datetime.now(),
            'accuracy': accuracy_score(new_labels, predictions),
            'precision': precision_score(new_labels, predictions, average='weighted'),
            'recall': recall_score(new_labels, predictions, average='weighted'),
            'sample_size': len(new_labels)
        }
        
        self.monitoring_results.append(performance)
        
        # パフォーマンス低下の検出
        if len(self.monitoring_results) > 1:
            recent_accuracy = np.mean([r['accuracy'] for r in self.monitoring_results[-5:]])
            baseline_accuracy = self.monitoring_results[0]['accuracy']
            
            if recent_accuracy < baseline_accuracy * 0.95:  # 5%以上の性能低下
                print(f"警告: モデル性能が低下しています。"
                      f"ベースライン: {baseline_accuracy:.4f}, "
                      f"現在: {recent_accuracy:.4f}")
                
        return performance

# 本番環境へのデプロイ
class ModelDeployment:
    def __init__(self, model_name, model_version):
        self.model_name = model_name
        self.model_version = model_version
        self.client = MlflowClient()
        
    def load_production_model(self):
        """本番モデルのロード"""
        model_uri = f"models:/{self.model_name}/{self.model_version}"
        model = mlflow.sklearn.load_model(model_uri)
        return model
        
    def create_prediction_service(self):
        """予測サービスの作成"""
        from fastapi import FastAPI, HTTPException
        from pydantic import BaseModel
        import uvicorn
        
        app = FastAPI()
        model = self.load_production_model()
        
        class PredictionRequest(BaseModel):
            features: list
            
        class PredictionResponse(BaseModel):
            prediction: float
            probability: list
            model_version: str
            
        @app.post("/predict", response_model=PredictionResponse)
        async def predict(request: PredictionRequest):
            try:
                # 入力データの変換
                input_data = np.array(request.features).reshape(1, -1)
                
                # 予測
                prediction = model.predict(input_data)[0]
                probability = model.predict_proba(input_data)[0].tolist()
                
                return PredictionResponse(
                    prediction=float(prediction),
                    probability=probability,
                    model_version=self.model_version
                )
            except Exception as e:
                raise HTTPException(status_code=400, detail=str(e))
                
        return app
        
# 使用例
if __name__ == "__main__":
    # MLOpsワークフローの初期化
    mlops = MLOpsWorkflow("customer_churn_prediction")
    
    # データの準備（仮想データ）
    from sklearn.datasets import make_classification
    X, y = make_classification(n_samples=1000, n_features=20, n_informative=15)
    X = pd.DataFrame(X, columns=[f'feature_{i}' for i in range(20)])
    
    from sklearn.model_selection import train_test_split
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)
    
    # モデルの学習とログ
    from sklearn.ensemble import RandomForestClassifier
    model = RandomForestClassifier(n_estimators=100, random_state=42)
    
    trained_model, metrics = mlops.train_and_log_model(
        model, X_train, y_train, X_test, y_test, "random_forest_v1"
    )
    
    print("モデルメトリクス:", metrics)