ml_assignment.py

import argparse
import math
from dataclasses import dataclass

import numpy as np
from sklearn import datasets
from sklearn.cluster import AgglomerativeClustering, DBSCAN, KMeans
from sklearn.ensemble import AdaBoostClassifier, RandomForestClassifier, RandomForestRegressor
from sklearn.metrics import (
    accuracy_score,
    mean_squared_error,
    silhouette_score,
)
from sklearn.model_selection import train_test_split
from sklearn.neural_network import MLPClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.semi_supervised import SelfTrainingClassifier
from sklearn.svm import SVC, SVR
from sklearn.tree import DecisionTreeClassifier


def _print_header(title: str) -> None:
    print("\n" + "=" * 80)
    print(title)
    print("=" * 80)


def clustering_kmeans_demo() -> None:
    _print_header("Clustering: K-means")
    X, _ = datasets.make_blobs(n_samples=400, centers=4, random_state=42, cluster_std=1.2)
    scaler = StandardScaler()
    Xs = scaler.fit_transform(X)

    model = KMeans(n_clusters=4, n_init=20, random_state=42)
    labels = model.fit_predict(Xs)
    sil = silhouette_score(Xs, labels)
    print(f"KMeans inertia: {model.inertia_:.4f}")
    print(f"Silhouette score: {sil:.4f}")


def clustering_modified_kmeans_demo() -> None:
    _print_header("Clustering: Modified K-means (auto-k via silhouette + k-means++)")
    X, _ = datasets.make_blobs(n_samples=500, centers=5, random_state=0, cluster_std=1.3)
    Xs = StandardScaler().fit_transform(X)

    best_k = None
    best_score = -1.0
    best_model = None

    for k in range(2, 9):
        km = KMeans(n_clusters=k, init="k-means++", n_init=30, max_iter=500, random_state=42)
        labels = km.fit_predict(Xs)
        score = silhouette_score(Xs, labels)
        if score > best_score:
            best_score = score
            best_k = k
            best_model = km

    print(f"Best k selected: {best_k}")
    print(f"Best silhouette score: {best_score:.4f}")
    print(f"Inertia at best k: {best_model.inertia_:.4f}")


def clustering_hierarchical_demo() -> None:
    _print_header("Clustering: Hierarchical (Agglomerative)")
    X, _ = datasets.make_blobs(n_samples=350, centers=4, random_state=7)
    Xs = StandardScaler().fit_transform(X)

    model = AgglomerativeClustering(n_clusters=4, linkage="ward")
    labels = model.fit_predict(Xs)
    sil = silhouette_score(Xs, labels)
    print(f"Hierarchical clustering silhouette score: {sil:.4f}")


def fuzzy_c_means(
    X: np.ndarray,
    c: int,
    m: float = 2.0,
    max_iter: int = 200,
    error: float = 1e-5,
    random_state: int = 42,
) -> tuple[np.ndarray, np.ndarray]:
    rng = np.random.default_rng(random_state)
    n = X.shape[0]

    U = rng.random((n, c))
    U = U / U.sum(axis=1, keepdims=True)

    for _ in range(max_iter):
        U_old = U.copy()
        Um = U**m
        centers = (Um.T @ X) / Um.sum(axis=0)[:, None]

        dist = np.linalg.norm(X[:, None, :] - centers[None, :, :], axis=2) + 1e-12

        power = 2.0 / (m - 1.0)
        for i in range(n):
            ratios = (dist[i, :, None] / dist[i, None, :]) ** power
            U[i] = 1.0 / ratios.sum(axis=1)

        if np.linalg.norm(U - U_old) < error:
            break

    return centers, U


def clustering_fuzzy_cmeans_demo() -> None:
    _print_header("Clustering: Fuzzy C-means")
    X, _ = datasets.make_blobs(n_samples=300, centers=3, random_state=21, cluster_std=1.1)
    Xs = StandardScaler().fit_transform(X)

    centers, U = fuzzy_c_means(Xs, c=3, m=2.0)
    labels = np.argmax(U, axis=1)
    sil = silhouette_score(Xs, labels)

    print(f"Fuzzy C-means centers shape: {centers.shape}")
    print(f"Silhouette score (hard labels from memberships): {sil:.4f}")


def density_dbscan_demo() -> None:
    _print_header("Density-based learning: DBSCAN")
    X, _ = datasets.make_moons(n_samples=500, noise=0.08, random_state=42)
    Xs = StandardScaler().fit_transform(X)

    model = DBSCAN(eps=0.3, min_samples=7)
    labels = model.fit_predict(Xs)

    clusters = len(set(labels)) - (1 if -1 in labels else 0)
    noise = int(np.sum(labels == -1))
    print(f"DBSCAN clusters found: {clusters}")
    print(f"Noise points: {noise}")


def density_hdbscan_demo() -> None:
    _print_header("Density-based learning: HDBSCAN")
    try:
        import hdbscan
    except ImportError as exc:
        raise ImportError("hdbscan is not installed. Run: pip install hdbscan") from exc

    X, _ = datasets.make_blobs(n_samples=450, centers=4, random_state=42, cluster_std=[0.2, 0.4, 0.9, 0.3])
    Xs = StandardScaler().fit_transform(X)

    model = hdbscan.HDBSCAN(min_cluster_size=12, min_samples=6)
    labels = model.fit_predict(Xs)

    clusters = len(set(labels)) - (1 if -1 in labels else 0)
    noise = int(np.sum(labels == -1))
    print(f"HDBSCAN clusters found: {clusters}")
    print(f"Noise points: {noise}")


def semi_supervised_self_training_demo() -> None:
    _print_header("Semi-supervised learning: Self-training")
    iris = datasets.load_iris()
    X_train, X_test, y_train, y_test = train_test_split(
        iris.data,
        iris.target,
        test_size=0.25,
        random_state=42,
        stratify=iris.target,
    )

    rng = np.random.default_rng(42)
    unlabeled_mask = rng.random(y_train.shape[0]) < 0.6
    y_train_semi = y_train.copy()
    y_train_semi[unlabeled_mask] = -1

    base_clf = SVC(probability=True, gamma="scale", random_state=42)
    model = SelfTrainingClassifier(base_clf, threshold=0.8)
    model.fit(X_train, y_train_semi)

    pred = model.predict(X_test)
    acc = accuracy_score(y_test, pred)
    print(f"Labeled ratio in train set: {(~unlabeled_mask).mean():.2f}")
    print(f"Self-training test accuracy: {acc:.4f}")


def ensemble_learning_demo() -> None:
    _print_header("Ensemble learning: RFR, RFC, XGBoost, AdaBoost, CatBoost")

    diabetes = datasets.load_diabetes()
    Xr_train, Xr_test, yr_train, yr_test = train_test_split(
        diabetes.data, diabetes.target, test_size=0.25, random_state=42
    )
    rfr = RandomForestRegressor(n_estimators=200, random_state=42)
    rfr.fit(Xr_train, yr_train)
    pred_rfr = rfr.predict(Xr_test)
    rmse_rfr = math.sqrt(mean_squared_error(yr_test, pred_rfr))

    breast = datasets.load_breast_cancer()
    Xc_train, Xc_test, yc_train, yc_test = train_test_split(
        breast.data, breast.target, test_size=0.25, random_state=42, stratify=breast.target
    )
    rfc = RandomForestClassifier(n_estimators=250, random_state=42)
    rfc.fit(Xc_train, yc_train)
    acc_rfc = accuracy_score(yc_test, rfc.predict(Xc_test))

    ada = AdaBoostClassifier(estimator=DecisionTreeClassifier(max_depth=2, random_state=42), n_estimators=150, random_state=42)
    ada.fit(Xc_train, yc_train)
    acc_ada = accuracy_score(yc_test, ada.predict(Xc_test))

    print(f"RandomForestRegressor RMSE: {rmse_rfr:.4f}")
    print(f"RandomForestClassifier accuracy: {acc_rfc:.4f}")
    print(f"AdaBoost accuracy: {acc_ada:.4f}")

    try:
        from xgboost import XGBClassifier

        xgb = XGBClassifier(
            n_estimators=200,
            max_depth=4,
            learning_rate=0.08,
            subsample=0.9,
            colsample_bytree=0.9,
            eval_metric="logloss",
            random_state=42,
        )
        xgb.fit(Xc_train, yc_train)
        acc_xgb = accuracy_score(yc_test, xgb.predict(Xc_test))
        print(f"XGBoost accuracy: {acc_xgb:.4f}")
    except ImportError:
        print("XGBoost: skipped (xgboost not installed)")

    try:
        from catboost import CatBoostClassifier

        cat = CatBoostClassifier(iterations=200, depth=5, learning_rate=0.08, verbose=False, random_state=42)
        cat.fit(Xc_train, yc_train)
        acc_cat = accuracy_score(yc_test, cat.predict(Xc_test))
        print(f"CatBoost accuracy: {acc_cat:.4f}")
    except ImportError:
        print("CatBoost: skipped (catboost not installed)")


def mlp_demo() -> None:
    _print_header("Multilayer Perceptron (MLP)")
    digits = datasets.load_digits()
    X_train, X_test, y_train, y_test = train_test_split(
        digits.data, digits.target, test_size=0.25, random_state=42, stratify=digits.target
    )

    model = MLPClassifier(hidden_layer_sizes=(128, 64), max_iter=350, random_state=42)
    model.fit(X_train, y_train)
    pred = model.predict(X_test)
    acc = accuracy_score(y_test, pred)
    print(f"MLP accuracy: {acc:.4f}")


def rnn_demo() -> None:
    _print_header("Recurrent Neural Network (RNN/LSTM)")
    try:
        import torch
        import torch.nn as nn
    except ImportError as exc:
        raise ImportError("torch is not installed. Run: pip install torch") from exc

    torch.manual_seed(42)
    np.random.seed(42)

    t = np.linspace(0, 40, 600)
    series = np.sin(t) + 0.1 * np.random.randn(len(t))

    seq_len = 20
    X, y = [], []
    for i in range(len(series) - seq_len):
        X.append(series[i : i + seq_len])
        y.append(series[i + seq_len])
    X = np.array(X, dtype=np.float32)
    y = np.array(y, dtype=np.float32)

    X_t = torch.tensor(X).unsqueeze(-1)
    y_t = torch.tensor(y).unsqueeze(-1)

    train_size = int(0.8 * len(X_t))
    X_train, X_test = X_t[:train_size], X_t[train_size:]
    y_train, y_test = y_t[:train_size], y_t[train_size:]

    class LSTMRegressor(nn.Module):
        def __init__(self, hidden_size: int = 32):
            super().__init__()
            self.lstm = nn.LSTM(input_size=1, hidden_size=hidden_size, batch_first=True)
            self.fc = nn.Linear(hidden_size, 1)

        def forward(self, x):
            out, _ = self.lstm(x)
            return self.fc(out[:, -1, :])

    model = LSTMRegressor(hidden_size=32)
    criterion = nn.MSELoss()
    optimizer = torch.optim.Adam(model.parameters(), lr=0.01)

    for _ in range(20):
        model.train()
        optimizer.zero_grad()
        pred = model(X_train)
        loss = criterion(pred, y_train)
        loss.backward()
        optimizer.step()

    model.eval()
    with torch.no_grad():
        test_pred = model(X_test)
        test_loss = criterion(test_pred, y_test).item()
    print(f"RNN/LSTM test MSE: {test_loss:.6f}")


def som_demo() -> None:
    _print_header("Self-Organizing Map (SOM)")
    try:
        from minisom import MiniSom
    except ImportError as exc:
        raise ImportError("MiniSom is not installed. Run: pip install MiniSom") from exc

    iris = datasets.load_iris()
    X = iris.data
    Xs = StandardScaler().fit_transform(X)

    som = MiniSom(8, 8, Xs.shape[1], sigma=1.0, learning_rate=0.5, random_seed=42)
    som.random_weights_init(Xs)
    som.train_random(Xs, 1000)

    winners = np.array([som.winner(x) for x in Xs])
    unique_neurons = len({(int(a), int(b)) for a, b in winners})
    print(f"SOM used neurons: {unique_neurons}")


def hmm_demo() -> None:
    _print_header("Hidden Markov Model (HMM)")
    try:
        from hmmlearn.hmm import GaussianHMM
    except ImportError as exc:
        raise ImportError("hmmlearn is not installed. Run: pip install hmmlearn") from exc

    rng = np.random.default_rng(42)
    X1 = rng.normal(loc=-2.0, scale=0.8, size=(200, 1))
    X2 = rng.normal(loc=2.0, scale=0.8, size=(200, 1))
    X = np.vstack([X1, X2])

    model = GaussianHMM(n_components=2, covariance_type="diag", n_iter=200, random_state=42)
    model.fit(X)
    states = model.predict(X)

    print(f"HMM converged score: {model.score(X):.4f}")
    print(f"Unique hidden states found: {len(np.unique(states))}")


def svm_demo() -> None:
    _print_header("Support Vector Machine (SVM)")
    iris = datasets.load_iris()
    X_train, X_test, y_train, y_test = train_test_split(
        iris.data, iris.target, test_size=0.25, random_state=42, stratify=iris.target
    )

    clf = SVC(kernel="rbf", C=5.0, gamma="scale", random_state=42)
    clf.fit(X_train, y_train)
    pred = clf.predict(X_test)
    acc = accuracy_score(y_test, pred)
    print(f"SVM classification accuracy: {acc:.4f}")


def llm_demo() -> None:
    _print_header("Large Language Model (LLM)")
    try:
        from transformers import pipeline
    except ImportError as exc:
        raise ImportError("transformers is not installed. Run: pip install transformers") from exc

    generator = pipeline("text-generation", model="distilgpt2")
    prompt = "Machine learning is useful because"
    output = generator(prompt, max_length=40, num_return_sequences=1, do_sample=True, temperature=0.8)
    text = output[0]["generated_text"]
    print(f"Prompt: {prompt}")
    print(f"Generated: {text}")


@dataclass
class GRNN:
    sigma: float = 0.5
    X_train: np.ndarray | None = None
    y_train: np.ndarray | None = None

    def fit(self, X: np.ndarray, y: np.ndarray) -> "GRNN":
        self.X_train = np.asarray(X, dtype=float)
        self.y_train = np.asarray(y, dtype=float)
        return self

    def predict(self, X: np.ndarray) -> np.ndarray:
        if self.X_train is None or self.y_train is None:
            raise ValueError("GRNN model is not fitted.")

        X = np.asarray(X, dtype=float)
        preds = []
        for x in X:
            d2 = np.sum((self.X_train - x) ** 2, axis=1)
            w = np.exp(-d2 / (2 * self.sigma**2))
            denom = np.sum(w)
            if denom <= 1e-12:
                preds.append(float(np.mean(self.y_train)))
            else:
                preds.append(float(np.sum(w * self.y_train) / denom))
        return np.array(preds)


def grnn_demo() -> None:
    _print_header("Generalized Regression Neural Network (GRNN)")
    X, y = datasets.make_regression(n_samples=350, n_features=6, noise=12.0, random_state=42)
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.25, random_state=42)

    scaler = StandardScaler()
    X_train_s = scaler.fit_transform(X_train)
    X_test_s = scaler.transform(X_test)

    model = GRNN(sigma=0.8).fit(X_train_s, y_train)
    pred = model.predict(X_test_s)
    rmse = math.sqrt(mean_squared_error(y_test, pred))

    baseline = SVR(C=5.0, epsilon=0.1).fit(X_train_s, y_train)
    base_rmse = math.sqrt(mean_squared_error(y_test, baseline.predict(X_test_s)))

    print(f"GRNN RMSE: {rmse:.4f}")
    print(f"SVR baseline RMSE: {base_rmse:.4f}")


ALGO_FUNCS = {
    "kmeans": clustering_kmeans_demo,
    "modified_kmeans": clustering_modified_kmeans_demo,
    "hierarchical": clustering_hierarchical_demo,
    "fuzzy_cmeans": clustering_fuzzy_cmeans_demo,
    "dbscan": density_dbscan_demo,
    "hdbscan": density_hdbscan_demo,
    "self_training": semi_supervised_self_training_demo,
    "ensemble": ensemble_learning_demo,
    "mlp": mlp_demo,
    "rnn": rnn_demo,
    "som": som_demo,
    "hmm": hmm_demo,
    "svm": svm_demo,
    "llm": llm_demo,
    "grnn": grnn_demo,
}


def run_all() -> None:
    for name, fn in ALGO_FUNCS.items():
        try:
            fn()
        except ImportError as exc:
            print(f"[{name}] skipped: {exc}")


def main() -> None:
    parser = argparse.ArgumentParser(description="ML Assignment Algorithms Runner")
    parser.add_argument(
        "--algo",
        type=str,
        default="all",
        choices=["all", *ALGO_FUNCS.keys()],
        help="Algorithm demo to run",
    )
    args = parser.parse_args()

    if args.algo == "all":
        run_all()
    else:
        ALGO_FUNCS[args.algo]()


if __name__ == "__main__":
    main()