import numpy as np
 
# === 回帰の損失関数 ===
 
def mse_loss(y_true, y_pred):
    """Mean Squared Error: 外れ値に敏感、微分が容易"""
    return np.mean((y_true - y_pred) ** 2)
 
def mae_loss(y_true, y_pred):
    """Mean Absolute Error: 外れ値に頑健、微分不連続点あり"""
    return np.mean(np.abs(y_true - y_pred))
 
def huber_loss(y_true, y_pred, delta=1.0):
    """Huber Loss: MSEとMAEのハイブリッド
    |error| <= delta: MSE（滑らか）
    |error| > delta:  MAE（外れ値に頑健）
    """
    error = y_true - y_pred
    is_small = np.abs(error) <= delta
    squared_loss = 0.5 * error ** 2
    linear_loss = delta * (np.abs(error) - 0.5 * delta)
    return np.mean(np.where(is_small, squared_loss, linear_loss))
 
def quantile_loss(y_true, y_pred, quantile=0.5):
    """Quantile Loss: 特定の分位点を予測する場合に使用
    quantile=0.5 → MAEと同等
    quantile=0.9 → 90パーセンタイルの予測
    """
    error = y_true - y_pred
    return np.mean(np.where(error >= 0, quantile * error, (quantile - 1) * error))
 
 
# === 分類の損失関数 ===
 
def binary_cross_entropy(y_true, y_prob, eps=1e-15):
    """Binary Cross-Entropy: 二値分類の標準的な損失関数"""
    y_prob = np.clip(y_prob, eps, 1 - eps)
    return -np.mean(y_true * np.log(y_prob) + (1 - y_true) * np.log(1 - y_prob))
 
def categorical_cross_entropy(y_true_onehot, y_prob, eps=1e-15):
    """Categorical Cross-Entropy: 多クラス分類の標準的な損失関数"""
    y_prob = np.clip(y_prob, eps, 1 - eps)
    return -np.mean(np.sum(y_true_onehot * np.log(y_prob), axis=1))
 
def hinge_loss(y_true, y_score):
    """Hinge Loss: SVMで使用。y_true は {-1, +1}"""
    return np.mean(np.maximum(0, 1 - y_true * y_score))
 
def focal_loss(y_true, y_prob, gamma=2.0, alpha=0.25, eps=1e-15):
    """Focal Loss: クラス不均衡に強い損失関数（RetinaNetで提案）
    容易なサンプルの損失を下げ、困難なサンプルに集中
    """
    y_prob = np.clip(y_prob, eps, 1 - eps)
    pt = np.where(y_true == 1, y_prob, 1 - y_prob)
    at = np.where(y_true == 1, alpha, 1 - alpha)
    loss = -at * (1 - pt) ** gamma * np.log(pt)
    return np.mean(loss)
 
 
# 損失関数の比較実験
y_true = np.array([3.0, 5.0, 2.5, 7.0, 4.5])
y_pred = np.array([2.8, 5.2, 2.0, 10.0, 4.3])  # 7.0→10.0 は外れ値的な予測
 
print("回帰損失関数の比較（外れ値あり）:")
print(f"  MSE:   {mse_loss(y_true, y_pred):.4f}")
print(f"  MAE:   {mae_loss(y_true, y_pred):.4f}")
print(f"  Huber: {huber_loss(y_true, y_pred, delta=1.0):.4f}")

import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, confusion_matrix
 
# 1. データ準備
from sklearn.datasets import load_breast_cancer
data = load_breast_cancer()
X, y = data.data, data.target
 
# 2. 訓練/テスト分割
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)
 
# 3. 前処理
scaler = StandardScaler()
X_train_s = scaler.fit_transform(X_train)
X_test_s = scaler.transform(X_test)
 
# 4. モデル学習
models = {
    "ロジスティック回帰": LogisticRegression(max_iter=1000, random_state=42),
    "ランダムフォレスト": RandomForestClassifier(n_estimators=100, random_state=42),
}
 
for name, model in models.items():
    # 交差検証
    cv_scores = cross_val_score(model, X_train_s, y_train, cv=5, scoring="f1")
    print(f"\n{name}")
    print(f"  CV F1: {cv_scores.mean():.4f} ± {cv_scores.std():.4f}")
 
    # テスト評価
    model.fit(X_train_s, y_train)
    y_pred = model.predict(X_test_s)
    print(classification_report(y_test, y_pred, target_names=data.target_names))

from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.model_selection import cross_val_score
import pandas as pd
import numpy as np
 
# 現実的なデータセットのシミュレーション
np.random.seed(42)
n = 1000
df = pd.DataFrame({
    "age": np.random.normal(40, 15, n),
    "income": np.random.lognormal(10, 1, n),
    "education": np.random.choice(["high_school", "bachelor", "master", "phd"], n),
    "city": np.random.choice(["tokyo", "osaka", "nagoya", "fukuoka"], n),
    "satisfaction": np.random.normal(3.5, 1.0, n),
})
# 欠損値の挿入
for col in ["age", "income", "satisfaction"]:
    mask = np.random.random(n) < 0.05
    df.loc[mask, col] = np.nan
 
y = (df["income"].fillna(df["income"].median()) > df["income"].median()).astype(int)
X = df.drop(columns=[])
 
# カラム種別の定義
numeric_features = ["age", "income", "satisfaction"]
categorical_features = ["education", "city"]
 
# 数値特徴量の前処理パイプライン
numeric_transformer = Pipeline(steps=[
    ("imputer", SimpleImputer(strategy="median")),
    ("scaler", StandardScaler()),
])
 
# カテゴリ特徴量の前処理パイプライン
categorical_transformer = Pipeline(steps=[
    ("imputer", SimpleImputer(strategy="most_frequent")),
    ("onehot", OneHotEncoder(drop="first", sparse_output=False, handle_unknown="ignore")),
])
 
# 前処理の統合
preprocessor = ColumnTransformer(
    transformers=[
        ("num", numeric_transformer, numeric_features),
        ("cat", categorical_transformer, categorical_features),
    ]
)
 
# パイプライン全体
pipeline = Pipeline(steps=[
    ("preprocessor", preprocessor),
    ("classifier", GradientBoostingClassifier(
        n_estimators=100, max_depth=5, random_state=42
    )),
])
 
# 交差検証（データリークなし）
cv_scores = cross_val_score(pipeline, X, y, cv=5, scoring="f1")
print(f"CV F1: {cv_scores.mean():.4f} ± {cv_scores.std():.4f}")
 
# パイプラインの保存
import joblib
pipeline.fit(X, y)
joblib.dump(pipeline, "models/full_pipeline.joblib")
 
# 本番環境での推論
loaded_pipeline = joblib.load("models/full_pipeline.joblib")
new_data = pd.DataFrame({
    "age": [35], "income": [50000], "education": ["master"],
    "city": ["tokyo"], "satisfaction": [4.0]
})
prediction = loaded_pipeline.predict(new_data)
probability = loaded_pipeline.predict_proba(new_data)[:, 1]
print(f"予測: {prediction[0]}, 確率: {probability[0]:.4f}")

import numpy as np
import matplotlib.pyplot as plt
from sklearn.cluster import KMeans, DBSCAN, AgglomerativeClustering
from sklearn.mixture import GaussianMixture
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import silhouette_score, calinski_harabasz_score, davies_bouldin_score
from sklearn.datasets import make_blobs
 
# データ生成
X, y_true = make_blobs(n_samples=500, n_features=2, centers=4,
                        cluster_std=1.0, random_state=42)
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)
 
# 複数のクラスタリングアルゴリズムの比較
algorithms = {
    "K-Means": KMeans(n_clusters=4, random_state=42, n_init=10),
    "DBSCAN": DBSCAN(eps=0.5, min_samples=5),
    "Agglomerative": AgglomerativeClustering(n_clusters=4),
    "GMM": GaussianMixture(n_components=4, random_state=42),
}
 
fig, axes = plt.subplots(2, 2, figsize=(14, 12))
axes = axes.ravel()
 
for idx, (name, algo) in enumerate(algorithms.items()):
    if name == "GMM":
        labels = algo.fit_predict(X_scaled)
    else:
        labels = algo.fit_predict(X_scaled)
 
    n_clusters = len(set(labels)) - (1 if -1 in labels else 0)
 
    # ノイズ以外のサンプルで評価
    mask = labels != -1
    if mask.sum() > 1 and n_clusters > 1:
        sil = silhouette_score(X_scaled[mask], labels[mask])
        ch = calinski_harabasz_score(X_scaled[mask], labels[mask])
        db = davies_bouldin_score(X_scaled[mask], labels[mask])
    else:
        sil, ch, db = 0, 0, 0
 
    axes[idx].scatter(X_scaled[:, 0], X_scaled[:, 1], c=labels, cmap="viridis", s=10)
    axes[idx].set_title(f"{name}\nSil={sil:.3f}, CH={ch:.0f}, DB={db:.3f}")
    print(f"{name}: clusters={n_clusters}, Silhouette={sil:.4f}, "
          f"CH={ch:.1f}, DB={db:.4f}")
 
plt.tight_layout()
plt.savefig("reports/clustering_comparison.png", dpi=150)
plt.close()

import numpy as np
import matplotlib.pyplot as plt
from sklearn.cluster import KMeans
from sklearn.metrics import silhouette_score, silhouette_samples
 
def optimal_k_analysis(X, k_range=range(2, 11)):
    """エルボー法とシルエット分析でKを決定"""
    inertias = []
    silhouettes = []
 
    for k in k_range:
        km = KMeans(n_clusters=k, random_state=42, n_init=10)
        labels = km.fit_predict(X)
        inertias.append(km.inertia_)
        silhouettes.append(silhouette_score(X, labels))
 
    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 5))
 
    # エルボー法
    ax1.plot(list(k_range), inertias, "bo-")
    ax1.set_xlabel("クラスタ数 K")
    ax1.set_ylabel("慣性 (Inertia)")
    ax1.set_title("エルボー法")
    ax1.grid(True, alpha=0.3)
 
    # シルエットスコア
    ax2.plot(list(k_range), silhouettes, "ro-")
    ax2.set_xlabel("クラスタ数 K")
    ax2.set_ylabel("シルエットスコア")
    ax2.set_title("シルエット分析")
    ax2.grid(True, alpha=0.3)
 
    best_k = list(k_range)[np.argmax(silhouettes)]
    ax2.axvline(x=best_k, color="green", linestyle="--", label=f"Best K={best_k}")
    ax2.legend()
 
    plt.tight_layout()
    plt.savefig("reports/optimal_k_analysis.png", dpi=150)
    plt.close()
 
    return best_k
 
best_k = optimal_k_analysis(X_scaled)
print(f"最適なクラスタ数: {best_k}")

from sklearn.semi_supervised import LabelSpreading, SelfTrainingClassifier
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
import numpy as np
 
# データ生成
X, y = make_classification(n_samples=2000, n_features=20, n_informative=10,
                            n_classes=3, random_state=42)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)
 
# ラベルの一部を欠損させる（5%だけラベルあり）
rng = np.random.RandomState(42)
mask = rng.random(len(y_train)) > 0.05  # 95%のラベルを隠す
y_train_semi = y_train.copy()
y_train_semi[mask] = -1  # -1 はラベルなしを表す
 
n_labeled = (y_train_semi != -1).sum()
n_unlabeled = (y_train_semi == -1).sum()
print(f"ラベル付き: {n_labeled}, ラベルなし: {n_unlabeled}")
 
# Self-Training
base_model = RandomForestClassifier(n_estimators=100, random_state=42)
self_training = SelfTrainingClassifier(
    base_estimator=base_model,
    threshold=0.8,  # 信頼度80%以上のサンプルをラベル付きに追加
    max_iter=20,
)
self_training.fit(X_train, y_train_semi)
score_semi = self_training.score(X_test, y_test)
 
# 比較: ラベル付きデータのみで学習
labeled_mask = y_train_semi != -1
base_model_only = RandomForestClassifier(n_estimators=100, random_state=42)
base_model_only.fit(X_train[labeled_mask], y_train[labeled_mask])
score_labeled = base_model_only.score(X_test, y_test)
 
# 比較: 全ラベルで学習（上限）
full_model = RandomForestClassifier(n_estimators=100, random_state=42)
full_model.fit(X_train, y_train)
score_full = full_model.score(X_test, y_test)
 
print(f"\n精度比較:")
print(f"  ラベル付きのみ ({n_labeled}サンプル): {score_labeled:.4f}")
print(f"  半教師あり学習:                       {score_semi:.4f}")
print(f"  全ラベル学習 (上限):                  {score_full:.4f}")

import numpy as np
 
class QLearningAgent:
    """Q-Learning エージェントの基本実装"""
 
    def __init__(self, n_states, n_actions, learning_rate=0.1,
                 discount_factor=0.99, epsilon=1.0, epsilon_decay=0.995,
                 epsilon_min=0.01):
        self.q_table = np.zeros((n_states, n_actions))
        self.lr = learning_rate
        self.gamma = discount_factor
        self.epsilon = epsilon
        self.epsilon_decay = epsilon_decay
        self.epsilon_min = epsilon_min
        self.n_actions = n_actions
 
    def choose_action(self, state):
        """ε-greedy 方策で行動を選択"""
        if np.random.random() < self.epsilon:
            return np.random.randint(self.n_actions)
        return np.argmax(self.q_table[state])
 
    def update(self, state, action, reward, next_state, done):
        """Q値の更新"""
        if done:
            target = reward
        else:
            target = reward + self.gamma * np.max(self.q_table[next_state])
 
        # Q-Learning 更新則
        self.q_table[state, action] += self.lr * (
            target - self.q_table[state, action]
        )
 
    def decay_epsilon(self):
        """探索率を減衰"""
        self.epsilon = max(self.epsilon_min, self.epsilon * self.epsilon_decay)
 
    def train(self, env, n_episodes=1000):
        """学習ループ"""
        rewards_history = []
 
        for episode in range(n_episodes):
            state = env.reset()
            total_reward = 0
            done = False
 
            while not done:
                action = self.choose_action(state)
                next_state, reward, done, _ = env.step(action)
                self.update(state, action, reward, next_state, done)
                state = next_state
                total_reward += reward
 
            self.decay_epsilon()
            rewards_history.append(total_reward)
 
            if (episode + 1) % 100 == 0:
                avg_reward = np.mean(rewards_history[-100:])
                print(f"Episode {episode+1}: "
                      f"Avg Reward={avg_reward:.2f}, "
                      f"Epsilon={self.epsilon:.4f}")
 
        return rewards_history

import numpy as np
import matplotlib.pyplot as plt
from sklearn.model_selection import learning_curve, validation_curve
from sklearn.tree import DecisionTreeClassifier
from sklearn.ensemble import RandomForestClassifier
 
def plot_learning_curve(estimator, X, y, title="学習曲線"):
    """学習曲線をプロットして過学習/過少適合を診断"""
    train_sizes, train_scores, test_scores = learning_curve(
        estimator, X, y, cv=5,
        train_sizes=np.linspace(0.1, 1.0, 10),
        scoring="accuracy", n_jobs=-1
    )
 
    train_mean = train_scores.mean(axis=1)
    train_std = train_scores.std(axis=1)
    test_mean = test_scores.mean(axis=1)
    test_std = test_scores.std(axis=1)
 
    fig, ax = plt.subplots(figsize=(10, 6))
    ax.fill_between(train_sizes, train_mean - train_std,
                    train_mean + train_std, alpha=0.1, color="blue")
    ax.fill_between(train_sizes, test_mean - test_std,
                    test_mean + test_std, alpha=0.1, color="orange")
    ax.plot(train_sizes, train_mean, "o-", color="blue", label="訓練スコア")
    ax.plot(train_sizes, test_mean, "o-", color="orange", label="検証スコア")
    ax.set_xlabel("訓練サンプル数")
    ax.set_ylabel("精度")
    ax.set_title(title)
    ax.legend()
    ax.grid(True, alpha=0.3)
 
    # 診断メッセージ
    gap = train_mean[-1] - test_mean[-1]
    if gap > 0.1:
        ax.text(0.5, 0.05, f"過学習の兆候（Gap={gap:.3f}）",
                transform=ax.transAxes, color="red", fontsize=12)
    elif test_mean[-1] < 0.7:
        ax.text(0.5, 0.05, "過少適合の兆候",
                transform=ax.transAxes, color="red", fontsize=12)
 
    plt.tight_layout()
    plt.savefig("reports/learning_curve.png", dpi=150)
    plt.close()
 
    return train_mean, test_mean
 
# 過学習する深い決定木
deep_tree = DecisionTreeClassifier(max_depth=None, random_state=42)
plot_learning_curve(deep_tree, X_train_s, y_train, "深い決定木（過学習傾向）")
 
# 正則化された浅い決定木
shallow_tree = DecisionTreeClassifier(max_depth=3, random_state=42)
plot_learning_curve(shallow_tree, X_train_s, y_train, "浅い決定木（適切な複雑度）")

from sklearn.model_selection import validation_curve
import matplotlib.pyplot as plt
import numpy as np
 
def plot_validation_curve(estimator, X, y, param_name, param_range, title="検証曲線"):
    """ハイパーパラメータを変化させたときの訓練/検証スコアを可視化"""
    train_scores, test_scores = validation_curve(
        estimator, X, y,
        param_name=param_name,
        param_range=param_range,
        cv=5, scoring="accuracy", n_jobs=-1
    )
 
    train_mean = train_scores.mean(axis=1)
    train_std = train_scores.std(axis=1)
    test_mean = test_scores.mean(axis=1)
    test_std = test_scores.std(axis=1)
 
    fig, ax = plt.subplots(figsize=(10, 6))
    ax.fill_between(param_range, train_mean - train_std,
                    train_mean + train_std, alpha=0.1, color="blue")
    ax.fill_between(param_range, test_mean - test_std,
                    test_mean + test_std, alpha=0.1, color="orange")
    ax.plot(param_range, train_mean, "o-", color="blue", label="訓練スコア")
    ax.plot(param_range, test_mean, "o-", color="orange", label="検証スコア")
    ax.set_xlabel(param_name)
    ax.set_ylabel("精度")
    ax.set_title(title)
    ax.legend()
    ax.grid(True, alpha=0.3)
 
    best_idx = np.argmax(test_mean)
    ax.axvline(x=param_range[best_idx], color="green", linestyle="--",
               label=f"最適値={param_range[best_idx]}")
    ax.legend()
 
    plt.tight_layout()
    plt.savefig("reports/validation_curve.png", dpi=150)
    plt.close()
 
    return param_range[best_idx]
 
# 決定木の深さに対する検証曲線
best_depth = plot_validation_curve(
    DecisionTreeClassifier(random_state=42),
    X_train_s, y_train,
    param_name="max_depth",
    param_range=[1, 2, 3, 5, 7, 10, 15, 20, 30, None],
    title="決定木: max_depth の検証曲線"
)
print(f"最適な max_depth: {best_depth}")

# 正則化の実装例
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_val_score
import numpy as np
 
# L1, L2, ElasticNet の比較
regularizations = {
    "L1 (Lasso)": LogisticRegression(penalty="l1", solver="saga", C=1.0, max_iter=5000),
    "L2 (Ridge)": LogisticRegression(penalty="l2", solver="saga", C=1.0, max_iter=5000),
    "ElasticNet": LogisticRegression(penalty="elasticnet", solver="saga",
                                      C=1.0, l1_ratio=0.5, max_iter=5000),
    "正則化なし": LogisticRegression(penalty=None, solver="saga", max_iter=5000),
}
 
print("正則化手法の比較:")
print("-" * 60)
for name, model in regularizations.items():
    scores = cross_val_score(model, X_train_s, y_train, cv=5, scoring="accuracy")
    model.fit(X_train_s, y_train)
    n_nonzero = np.sum(np.abs(model.coef_) > 1e-6)
    print(f"  {name:15s}: Accuracy={scores.mean():.4f} ± {scores.std():.4f}, "
          f"非ゼロ係数={n_nonzero}")

from sklearn.model_selection import (
    KFold, StratifiedKFold, TimeSeriesSplit, GroupKFold,
    cross_validate, RepeatedStratifiedKFold, LeaveOneOut
)
from sklearn.ensemble import GradientBoostingClassifier
import numpy as np
 
def comprehensive_cv(model, X, y, cv_strategy="stratified", groups=None):
    """包括的な交差検証を実行"""
 
    strategies = {
        "kfold": KFold(n_splits=5, shuffle=True, random_state=42),
        "stratified": StratifiedKFold(n_splits=5, shuffle=True, random_state=42),
        "repeated": RepeatedStratifiedKFold(n_splits=5, n_repeats=3, random_state=42),
        "timeseries": TimeSeriesSplit(n_splits=5),
    }
 
    if cv_strategy == "group" and groups is not None:
        cv = GroupKFold(n_splits=5)
        scoring = ["accuracy", "f1", "precision", "recall", "roc_auc"]
        results = cross_validate(
            model, X, y, cv=cv, scoring=scoring,
            return_train_score=True, n_jobs=-1, groups=groups
        )
    else:
        cv = strategies[cv_strategy]
        scoring = ["accuracy", "f1", "precision", "recall", "roc_auc"]
        results = cross_validate(
            model, X, y, cv=cv, scoring=scoring,
            return_train_score=True, n_jobs=-1
        )
 
    print(f"交差検証戦略: {cv_strategy}")
    print("-" * 70)
    for metric in scoring:
        train_key = f"train_{metric}"
        test_key = f"test_{metric}"
        print(f"  {metric:12s}: "
              f"Train={results[train_key].mean():.4f} ± {results[train_key].std():.4f}  "
              f"Test={results[test_key].mean():.4f} ± {results[test_key].std():.4f}")
 
    # 過学習度の判定
    train_acc = results["train_accuracy"].mean()
    test_acc = results["test_accuracy"].mean()
    gap = train_acc - test_acc
    if gap > 0.1:
        print(f"\n  ⚠ 過学習の兆候あり (Train-Test Gap = {gap:.4f})")
    elif test_acc < 0.6:
        print(f"\n  ⚠ 過少適合の兆候あり (Test Accuracy = {test_acc:.4f})")
    else:
        print(f"\n  ✓ 良好なフィット (Gap = {gap:.4f})")
 
    return results
 
model = GradientBoostingClassifier(n_estimators=100, random_state=42)
results = comprehensive_cv(model, X_train_s, y_train, "stratified")

# ★ 重要: 交差検証でのデータリークを防ぐ正しい実装
 
# BAD: 交差検証の「前に」全データで前処理
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import cross_val_score
from sklearn.svm import SVC
 
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)  # ← テストデータの情報がリーク！
scores = cross_val_score(SVC(), X_scaled, y, cv=5)
print(f"リークあり: {scores.mean():.4f}")  # 楽観的な結果
 
# GOOD: パイプラインで前処理をCV内に含める
from sklearn.pipeline import Pipeline
 
pipe = Pipeline([
    ("scaler", StandardScaler()),
    ("svm", SVC()),
])
scores = cross_val_score(pipe, X, y, cv=5)  # 各Fold内で個別にfit_transform
print(f"リークなし: {scores.mean():.4f}")  # 正確な結果

import numpy as np
from sklearn.metrics import (
    accuracy_score, precision_score, recall_score,
    f1_score, roc_auc_score, confusion_matrix,
    precision_recall_curve, roc_curve, average_precision_score,
    matthews_corrcoef, cohen_kappa_score, log_loss,
    balanced_accuracy_score
)
import matplotlib.pyplot as plt
 
class ClassificationEvaluator:
    """分類モデルの包括的評価"""
 
    def __init__(self, y_true, y_pred, y_prob=None):
        self.y_true = y_true
        self.y_pred = y_pred
        self.y_prob = y_prob
 
    def print_metrics(self):
        """主要指標を一覧表示"""
        print("=" * 60)
        print("分類評価指標")
        print("=" * 60)
        print(f"  Accuracy:          {accuracy_score(self.y_true, self.y_pred):.4f}")
        print(f"  Balanced Accuracy: {balanced_accuracy_score(self.y_true, self.y_pred):.4f}")
        print(f"  Precision:         {precision_score(self.y_true, self.y_pred):.4f}")
        print(f"  Recall:            {recall_score(self.y_true, self.y_pred):.4f}")
        print(f"  F1-Score:          {f1_score(self.y_true, self.y_pred):.4f}")
        print(f"  MCC:               {matthews_corrcoef(self.y_true, self.y_pred):.4f}")
        print(f"  Cohen's Kappa:     {cohen_kappa_score(self.y_true, self.y_pred):.4f}")
        if self.y_prob is not None:
            print(f"  AUC-ROC:           {roc_auc_score(self.y_true, self.y_prob):.4f}")
            print(f"  AP:                {average_precision_score(self.y_true, self.y_prob):.4f}")
            print(f"  Log Loss:          {log_loss(self.y_true, self.y_prob):.4f}")
 
        cm = confusion_matrix(self.y_true, self.y_pred)
        print(f"\n  混同行列:")
        print(f"           予測=0  予測=1")
        print(f"  実際=0   {cm[0,0]:5d}   {cm[0,1]:5d}  (TN, FP)")
        print(f"  実際=1   {cm[1,0]:5d}   {cm[1,1]:5d}  (FN, TP)")
 
        # 追加メトリクス
        tn, fp, fn, tp = cm.ravel()
        specificity = tn / (tn + fp) if (tn + fp) > 0 else 0
        npv = tn / (tn + fn) if (tn + fn) > 0 else 0
        print(f"\n  Specificity (TNR): {specificity:.4f}")
        print(f"  NPV:               {npv:.4f}")
        print(f"  FPR:               {fp / (fp + tn):.4f}")
        print(f"  FNR:               {fn / (fn + tp):.4f}")
 
    def plot_roc_pr(self):
        """ROC曲線とPR曲線を並べて描画"""
        if self.y_prob is None:
            print("確率予測が必要です")
            return
 
        fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 6))
 
        # ROC曲線
        fpr, tpr, thresholds_roc = roc_curve(self.y_true, self.y_prob)
        auc = roc_auc_score(self.y_true, self.y_prob)
        ax1.plot(fpr, tpr, label=f"AUC = {auc:.3f}")
        ax1.plot([0, 1], [0, 1], "k--", alpha=0.3)
        ax1.set_xlabel("偽陽性率 (FPR)")
        ax1.set_ylabel("真陽性率 (TPR)")
        ax1.set_title("ROC曲線")
        ax1.legend()
        ax1.grid(True, alpha=0.3)
 
        # PR曲線
        prec, rec, thresholds_pr = precision_recall_curve(self.y_true, self.y_prob)
        ap = average_precision_score(self.y_true, self.y_prob)
        ax2.plot(rec, prec, label=f"AP = {ap:.3f}")
        ax2.set_xlabel("再現率 (Recall)")
        ax2.set_ylabel("適合率 (Precision)")
        ax2.set_title("PR曲線")
        ax2.legend()
        ax2.grid(True, alpha=0.3)
 
        plt.tight_layout()
        plt.savefig("reports/roc_pr_curves.png", dpi=150)
        plt.close()
 
    def find_optimal_threshold(self, metric="f1"):
        """最適な閾値を探索"""
        if self.y_prob is None:
            raise ValueError("確率予測が必要です")
 
        thresholds = np.arange(0.1, 0.9, 0.01)
        best_score = 0
        best_threshold = 0.5
 
        print(f"\n閾値最適化（指標: {metric}）:")
        print("-" * 50)
 
        for t in thresholds:
            y_pred_t = (self.y_prob >= t).astype(int)
            if metric == "f1":
                score = f1_score(self.y_true, y_pred_t, zero_division=0)
            elif metric == "precision":
                score = precision_score(self.y_true, y_pred_t, zero_division=0)
            elif metric == "recall":
                score = recall_score(self.y_true, y_pred_t, zero_division=0)
 
            if score > best_score:
                best_score = score
                best_threshold = t
 
        print(f"  最適閾値: {best_threshold:.2f}")
        print(f"  最適{metric}: {best_score:.4f}")
        print(f"  デフォルト閾値(0.5)の{metric}: "
              f"{f1_score(self.y_true, (self.y_prob >= 0.5).astype(int)):.4f}")
 
        return best_threshold
 
# 使用例
model = LogisticRegression(max_iter=1000)
model.fit(X_train_s, y_train)
y_pred = model.predict(X_test_s)
y_prob = model.predict_proba(X_test_s)[:, 1]
 
evaluator = ClassificationEvaluator(y_test, y_pred, y_prob)
evaluator.print_metrics()
evaluator.plot_roc_pr()
best_t = evaluator.find_optimal_threshold("f1")

from sklearn.metrics import (
    classification_report, confusion_matrix,
    roc_auc_score, f1_score
)
from sklearn.preprocessing import label_binarize
import matplotlib.pyplot as plt
import seaborn as sns
import numpy as np
 
def multiclass_evaluation(y_true, y_pred, y_prob, class_names):
    """多クラス分類の包括的評価"""
 
    print("=" * 60)
    print("多クラス分類評価")
    print("=" * 60)
 
    # Classification Report
    print(classification_report(y_true, y_pred, target_names=class_names))
 
    # マクロ / ミクロ / 加重平均
    for avg in ["micro", "macro", "weighted"]:
        f1 = f1_score(y_true, y_pred, average=avg)
        print(f"  F1 ({avg:8s}): {f1:.4f}")
 
    # 多クラス AUC-ROC
    if y_prob is not None:
        try:
            auc_ovr = roc_auc_score(y_true, y_prob, multi_class="ovr", average="macro")
            auc_ovo = roc_auc_score(y_true, y_prob, multi_class="ovo", average="macro")
            print(f"\n  AUC-ROC (OVR macro): {auc_ovr:.4f}")
            print(f"  AUC-ROC (OVO macro): {auc_ovo:.4f}")
        except ValueError as e:
            print(f"  AUC計算エラー: {e}")
 
    # 混同行列のヒートマップ
    cm = confusion_matrix(y_true, y_pred)
    fig, ax = plt.subplots(figsize=(8, 6))
    sns.heatmap(cm, annot=True, fmt="d", cmap="Blues",
                xticklabels=class_names, yticklabels=class_names, ax=ax)
    ax.set_xlabel("予測ラベル")
    ax.set_ylabel("実際のラベル")
    ax.set_title("多クラス混同行列")
    plt.tight_layout()
    plt.savefig("reports/multiclass_confusion.png", dpi=150)
    plt.close()
 
    return cm

import numpy as np
from sklearn.metrics import (
    mean_squared_error, mean_absolute_error,
    r2_score, mean_absolute_percentage_error,
    median_absolute_error, max_error,
    explained_variance_score
)
import matplotlib.pyplot as plt
 
class RegressionEvaluator:
    """回帰モデルの包括的評価"""
 
    def __init__(self, y_true, y_pred, n_features=None):
        self.y_true = np.array(y_true)
        self.y_pred = np.array(y_pred)
        self.n_features = n_features
 
    def all_metrics(self) -> dict:
        """全指標を計算"""
        mse = mean_squared_error(self.y_true, self.y_pred)
        metrics = {
            "MSE": mse,
            "RMSE": np.sqrt(mse),
            "MAE": mean_absolute_error(self.y_true, self.y_pred),
            "Median AE": median_absolute_error(self.y_true, self.y_pred),
            "Max Error": max_error(self.y_true, self.y_pred),
            "MAPE(%)": mean_absolute_percentage_error(self.y_true, self.y_pred) * 100,
            "R²": r2_score(self.y_true, self.y_pred),
            "Explained Var": explained_variance_score(self.y_true, self.y_pred),
        }
        if self.n_features is not None:
            metrics["Adjusted R²"] = self._adjusted_r2(self.n_features)
        return metrics
 
    def _adjusted_r2(self, n_features: int) -> float:
        """自由度調整済みR²"""
        n = len(self.y_true)
        r2 = r2_score(self.y_true, self.y_pred)
        return 1 - (1 - r2) * (n - 1) / (n - n_features - 1)
 
    def print_metrics(self):
        metrics = self.all_metrics()
        print("=" * 50)
        print("回帰評価指標")
        print("=" * 50)
        for name, value in metrics.items():
            print(f"  {name:15s}: {value:.4f}")
 
    def plot_diagnostics(self):
        """回帰診断プロットの作成"""
        residuals = self.y_true - self.y_pred
 
        fig, axes = plt.subplots(2, 2, figsize=(14, 12))
 
        # 1. 予測 vs 実測
        axes[0, 0].scatter(self.y_true, self.y_pred, alpha=0.5, s=20)
        min_val = min(self.y_true.min(), self.y_pred.min())
        max_val = max(self.y_true.max(), self.y_pred.max())
        axes[0, 0].plot([min_val, max_val], [min_val, max_val], "r--")
        axes[0, 0].set_xlabel("実測値")
        axes[0, 0].set_ylabel("予測値")
        axes[0, 0].set_title("予測 vs 実測")
 
        # 2. 残差プロット
        axes[0, 1].scatter(self.y_pred, residuals, alpha=0.5, s=20)
        axes[0, 1].axhline(y=0, color="r", linestyle="--")
        axes[0, 1].set_xlabel("予測値")
        axes[0, 1].set_ylabel("残差")
        axes[0, 1].set_title("残差プロット")
 
        # 3. 残差のヒストグラム
        axes[1, 0].hist(residuals, bins=30, edgecolor="black", alpha=0.7)
        axes[1, 0].axvline(x=0, color="r", linestyle="--")
        axes[1, 0].set_xlabel("残差")
        axes[1, 0].set_ylabel("頻度")
        axes[1, 0].set_title("残差の分布")
 
        # 4. Q-Qプロット
        from scipy import stats
        (osm, osr), (slope, intercept, r) = stats.probplot(residuals, dist="norm")
        axes[1, 1].scatter(osm, osr, alpha=0.5, s=20)
        axes[1, 1].plot(osm, slope * np.array(osm) + intercept, "r--")
        axes[1, 1].set_xlabel("理論分位数")
        axes[1, 1].set_ylabel("サンプル分位数")
        axes[1, 1].set_title(f"Q-Qプロット (R={r:.4f})")
 
        plt.tight_layout()
        plt.savefig("reports/regression_diagnostics.png", dpi=150)
        plt.close()
 
# 使用例
# evaluator = RegressionEvaluator(y_test, y_pred, n_features=10)
# evaluator.print_metrics()
# evaluator.plot_diagnostics()

from sklearn.model_selection import GridSearchCV, RandomizedSearchCV
from sklearn.ensemble import RandomForestClassifier
from scipy.stats import randint, uniform
import time
 
# Grid Search
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("Grid Search")
start = time.time()
grid = GridSearchCV(
    RandomForestClassifier(random_state=42),
    param_grid, cv=5, scoring="f1", n_jobs=-1, verbose=0
)
grid.fit(X_train_s, y_train)
elapsed_grid = time.time() - start
print(f"  組合せ数: {len(grid.cv_results_['params'])}")
print(f"  最良スコア: {grid.best_score_:.4f}")
print(f"  最良パラメータ: {grid.best_params_}")
print(f"  計算時間: {elapsed_grid:.1f}秒")
 
# Random Search
param_distributions = {
    "n_estimators": randint(50, 500),
    "max_depth": [3, 5, 10, 15, 20, None],
    "min_samples_split": randint(2, 20),
    "min_samples_leaf": randint(1, 10),
    "max_features": uniform(0.1, 0.9),
}
 
print("\nRandom Search")
start = time.time()
random_search = RandomizedSearchCV(
    RandomForestClassifier(random_state=42),
    param_distributions, n_iter=50, cv=5, scoring="f1",
    n_jobs=-1, random_state=42, verbose=0
)
random_search.fit(X_train_s, y_train)
elapsed_random = time.time() - start
print(f"  試行数: 50")
print(f"  最良スコア: {random_search.best_score_:.4f}")
print(f"  最良パラメータ: {random_search.best_params_}")
print(f"  計算時間: {elapsed_random:.1f}秒")

import optuna
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.model_selection import cross_val_score
import warnings
warnings.filterwarnings("ignore")
 
def objective(trial):
    """Optunaの目的関数"""
    params = {
        "n_estimators": trial.suggest_int("n_estimators", 50, 500),
        "max_depth": trial.suggest_int("max_depth", 2, 15),
        "learning_rate": trial.suggest_float("learning_rate", 0.01, 0.3, log=True),
        "subsample": trial.suggest_float("subsample", 0.5, 1.0),
        "min_samples_split": trial.suggest_int("min_samples_split", 2, 20),
        "min_samples_leaf": trial.suggest_int("min_samples_leaf", 1, 10),
        "max_features": trial.suggest_float("max_features", 0.1, 1.0),
    }
 
    model = GradientBoostingClassifier(random_state=42, **params)
    scores = cross_val_score(model, X_train_s, y_train, cv=5, scoring="f1", n_jobs=-1)
    return scores.mean()
 
# 最適化実行
study = optuna.create_study(direction="maximize", study_name="gbc_optimization")
study.optimize(objective, n_trials=100, show_progress_bar=True)
 
print(f"\n最良スコア: {study.best_value:.4f}")
print(f"最良パラメータ:")
for key, value in study.best_params.items():
    print(f"  {key}: {value}")
 
# 最適パラメータでモデル学習
best_model = GradientBoostingClassifier(random_state=42, **study.best_params)
best_model.fit(X_train_s, y_train)
test_score = best_model.score(X_test_s, y_test)
print(f"\nテストスコア: {test_score:.4f}")
 
# パラメータの重要度
importance = optuna.importance.get_param_importances(study)
print("\nパラメータ重要度:")
for param, imp in importance.items():
    print(f"  {param}: {imp:.4f}")

import lightgbm as lgb
import optuna
from sklearn.model_selection import StratifiedKFold
import numpy as np
 
def lgbm_objective(trial, X, y, cv=5):
    """LightGBMのOptuna目的関数（実務向け）"""
    params = {
        "objective": "binary",
        "metric": "binary_logloss",
        "boosting_type": "gbdt",
        "verbosity": -1,
        "n_estimators": trial.suggest_int("n_estimators", 100, 2000),
        "learning_rate": trial.suggest_float("learning_rate", 0.005, 0.3, log=True),
        "max_depth": trial.suggest_int("max_depth", 3, 12),
        "num_leaves": trial.suggest_int("num_leaves", 8, 256),
        "min_child_samples": trial.suggest_int("min_child_samples", 5, 100),
        "subsample": trial.suggest_float("subsample", 0.5, 1.0),
        "colsample_bytree": trial.suggest_float("colsample_bytree", 0.3, 1.0),
        "reg_alpha": trial.suggest_float("reg_alpha", 1e-8, 10.0, log=True),
        "reg_lambda": trial.suggest_float("reg_lambda", 1e-8, 10.0, log=True),
    }
 
    skf = StratifiedKFold(n_splits=cv, shuffle=True, random_state=42)
    scores = []
 
    for fold, (train_idx, val_idx) in enumerate(skf.split(X, y)):
        X_train_fold, X_val_fold = X[train_idx], X[val_idx]
        y_train_fold, y_val_fold = y[train_idx], y[val_idx]
 
        model = lgb.LGBMClassifier(**params)
        model.fit(
            X_train_fold, y_train_fold,
            eval_set=[(X_val_fold, y_val_fold)],
            callbacks=[
                lgb.early_stopping(50, verbose=False),
                lgb.log_evaluation(period=0),
            ]
        )
 
        from sklearn.metrics import f1_score
        y_pred = model.predict(X_val_fold)
        scores.append(f1_score(y_val_fold, y_pred))
 
        # Optuna pruning
        trial.report(np.mean(scores), fold)
        if trial.should_prune():
            raise optuna.TrialPruned()
 
    return np.mean(scores)
 
# study = optuna.create_study(direction="maximize",
#                              pruner=optuna.pruners.MedianPruner())
# study.optimize(lambda trial: lgbm_objective(trial, X_train, y_train),
#                n_trials=200, show_progress_bar=True)

import numpy as np
import matplotlib.pyplot as plt
from sklearn.ensemble import RandomForestClassifier
from sklearn.inspection import permutation_importance
import pandas as pd
 
def feature_importance_analysis(model, X_train, X_test, y_test, feature_names):
    """複数の方法で特徴量重要度を分析"""
 
    # 1. モデル組み込みの重要度（不純度ベース）
    if hasattr(model, "feature_importances_"):
        imp_builtin = model.feature_importances_
    else:
        imp_builtin = None
 
    # 2. Permutation Importance（推奨）
    perm_result = permutation_importance(
        model, X_test, y_test, n_repeats=10, random_state=42, n_jobs=-1
    )
 
    # 結果の比較
    df = pd.DataFrame({
        "feature": feature_names,
        "perm_importance_mean": perm_result.importances_mean,
        "perm_importance_std": perm_result.importances_std,
    })
    if imp_builtin is not None:
        df["builtin_importance"] = imp_builtin
 
    df = df.sort_values("perm_importance_mean", ascending=False)
 
    print("特徴量重要度 (Permutation Importance):")
    print("-" * 60)
    for _, row in df.head(15).iterrows():
        print(f"  {row['feature']:30s}: "
              f"{row['perm_importance_mean']:.4f} ± {row['perm_importance_std']:.4f}")
 
    # 可視化
    fig, ax = plt.subplots(figsize=(10, 8))
    top_n = min(20, len(df))
    top_df = df.head(top_n)
    ax.barh(range(top_n), top_df["perm_importance_mean"].values, align="center")
    ax.set_yticks(range(top_n))
    ax.set_yticklabels(top_df["feature"].values)
    ax.set_xlabel("Permutation Importance")
    ax.set_title("特徴量重要度 (Top 20)")
    ax.invert_yaxis()
    plt.tight_layout()
    plt.savefig("reports/feature_importance.png", dpi=150)
    plt.close()
 
    return df
 
# SHAP値による解釈（より詳細）
def shap_analysis(model, X_test, feature_names):
    """SHAP値による特徴量の影響分析"""
    import shap
 
    # TreeExplainer（木ベースモデル用、高速）
    explainer = shap.TreeExplainer(model)
    shap_values = explainer.shap_values(X_test)
 
    # Summary Plot
    shap.summary_plot(shap_values, X_test, feature_names=feature_names, show=False)
    plt.tight_layout()
    plt.savefig("reports/shap_summary.png", dpi=150, bbox_inches="tight")
    plt.close()
 
    # 個別サンプルの説明
    shap.plots.waterfall(explainer(X_test)[0], show=False)
    plt.tight_layout()
    plt.savefig("reports/shap_waterfall.png", dpi=150, bbox_inches="tight")
    plt.close()
 
    return shap_values

# BAD: 99%が正常、1%が不正のデータで Accuracy を使う
# "全て正常と予測" → Accuracy 99% だが不正検出能力ゼロ
 
# GOOD: 不均衡データでは適切な指標を選択
from sklearn.metrics import classification_report, f1_score
from sklearn.ensemble import RandomForestClassifier
from imblearn.over_sampling import SMOTE
from imblearn.pipeline import Pipeline as ImbPipeline
 
# 方法1: class_weight でクラス不均衡に対処
model = LogisticRegression(class_weight="balanced", max_iter=1000)
model.fit(X_train, y_train)
y_pred = model.predict(X_test)
 
# F1, Recall, PR-AUC で評価
print(classification_report(y_test, y_pred))
 
# 方法2: SMOTEでオーバーサンプリング
smote_pipe = ImbPipeline([
    ("smote", SMOTE(random_state=42)),
    ("classifier", RandomForestClassifier(n_estimators=100, random_state=42)),
])
smote_pipe.fit(X_train, y_train)
y_pred_smote = smote_pipe.predict(X_test)
print(f"SMOTE + RF F1: {f1_score(y_test, y_pred_smote):.4f}")
 
# 方法3: コスト敏感学習
from sklearn.ensemble import GradientBoostingClassifier
# sample_weight で誤分類コストを反映
sample_weights = np.where(y_train == 1, 10.0, 1.0)  # 少数クラスに10倍の重み
model_weighted = GradientBoostingClassifier(n_estimators=100, random_state=42)
model_weighted.fit(X_train, y_train, sample_weight=sample_weights)

# BAD: テストデータでスコアを見ながら調整 → 情報リーク
for max_depth in [3, 5, 10, 20]:
    model = DecisionTreeClassifier(max_depth=max_depth)
    model.fit(X_train, y_train)
    score = model.score(X_test, y_test)  # テストで評価 → NG
    print(f"depth={max_depth}: {score:.4f}")
 
# GOOD: 検証データ or 交差検証で調整、テストは最後に1回だけ
from sklearn.model_selection import GridSearchCV
 
param_grid = {"max_depth": [3, 5, 10, 20]}
grid = GridSearchCV(
    DecisionTreeClassifier(random_state=42),
    param_grid, cv=5, scoring="f1"
)
grid.fit(X_train, y_train)  # 訓練+検証のみ
print(f"最良パラメータ: {grid.best_params_}")
print(f"テストスコア: {grid.score(X_test, y_test):.4f}")  # 最後に1回

# BAD: テストデータを含めて fit する
scaler = StandardScaler()
X_all_scaled = scaler.fit_transform(np.vstack([X_train, X_test]))  # リーク！
X_train_scaled = X_all_scaled[:len(X_train)]
X_test_scaled = X_all_scaled[len(X_train):]
 
# GOOD: 訓練データのみで fit し、テストには transform のみ
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)  # fit + transform
X_test_scaled = scaler.transform(X_test)  # transform のみ

# BAD: 全データで特徴量選択してからCV
from sklearn.feature_selection import SelectKBest, f_classif
 
selector = SelectKBest(f_classif, k=10)
X_selected = selector.fit_transform(X, y)  # 全データで選択 → リーク
scores = cross_val_score(SVC(), X_selected, y, cv=5)
 
# GOOD: パイプラインに含めてCV内で選択
pipe = Pipeline([
    ("selector", SelectKBest(f_classif, k=10)),
    ("scaler", StandardScaler()),
    ("svm", SVC()),
])
scores = cross_val_score(pipe, X, y, cv=5)  # 各Fold内で選択

# BAD: 時系列データに通常のランダム分割を使う
# 未来の情報で過去を予測する「先読み」が発生
X_train, X_test, y_train, y_test = train_test_split(
    X_timeseries, y_timeseries, test_size=0.2, random_state=42  # NG
)
 
# GOOD: 時系列の順序を維持する分割
from sklearn.model_selection import TimeSeriesSplit
 
tscv = TimeSeriesSplit(n_splits=5)
for train_idx, test_idx in tscv.split(X_timeseries):
    X_train = X_timeseries[train_idx]
    X_test = X_timeseries[test_idx]
    # 常に過去→未来の方向で学習→評価
    assert train_idx.max() < test_idx.min(), "時系列の順序が崩れている"