Machine Learning: KNNとクラスタリング — 最近傍分類と教師なしセグメンテーション

from sklearn.cluster import KMeans

# RFM clustering: group users by Recency, Frequency, Monetary
rfm = df[["recency", "frequency", "monetary"]]
kmeans = KMeans(n_clusters=4, random_state=42)
df["segment"] = kmeans.fit_predict(rfm)

for i in range(4):
    segment = df[df["segment"] == i]
    print(f"Segment {i}: {len(segment)} users, "
          f"Avg Monetary={segment['monetary'].mean():.0f} USD")

from sklearn.neighbors import KNeighborsClassifier
from sklearn.datasets import load_iris
from sklearn.model_selection import cross_val_score
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
import numpy as np

X, y = load_iris(return_X_y=True)

# Compare different K values
for k in [1, 3, 5, 7, 11, 21]:
    pipe = Pipeline([
        ("scaler", StandardScaler()),
        ("knn", KNeighborsClassifier(n_neighbors=k)),
    ])
    scores = cross_val_score(pipe, X, y, cv=5, scoring="accuracy")
    print(f"K={k:2d}: Accuracy={scores.mean():.3f} +/- {scores.std():.3f}")

# Executed successfully

from sklearn.neighbors import KNeighborsClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report
import numpy as np

rng = np.random.default_rng(42)
n = 1000
X = np.column_stack([
    rng.uniform(0, 100, n),   # browsing_time
    rng.uniform(0, 50, n),    # cart_value_usd
    rng.integers(1, 30, n),   # pages_viewed
    rng.integers(0, 10, n),   # previous_purchases
])
y = (X[:, 1] * 0.05 + X[:, 3] * 0.3 + rng.normal(0, 0.5, n) > 2).astype(int)

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

pipe = Pipeline([
    ("scaler", StandardScaler()),
    ("knn", KNeighborsClassifier(n_neighbors=7, weights="distance")),
])
pipe.fit(X_train, y_train)
print(classification_report(y_test, pipe.predict(X_test)))

# Executed successfully

from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler
import matplotlib.pyplot as plt
import numpy as np

rng = np.random.default_rng(42)
# Generate 4 natural clusters
X = np.vstack([
    rng.normal([20, 500], [3, 50], (200, 2)),   # High value
    rng.normal([50, 200], [5, 30], (300, 2)),   # Medium
    rng.normal([80, 50], [8, 20], (350, 2)),    # Low
    rng.normal([10, 800], [2, 40], (150, 2)),   # VIP
])

scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

# Elbow method
inertias = []
K_range = range(2, 10)
for k in K_range:
    km = KMeans(n_clusters=k, random_state=42, n_init=10)
    km.fit(X_scaled)
    inertias.append(km.inertia_)

fig, ax = plt.subplots(figsize=(8, 5))
ax.plot(K_range, inertias, "bo-", linewidth=2)
ax.set_xlabel("Number of Clusters (K)")
ax.set_ylabel("Inertia (Within-Cluster Sum of Squares)")
ax.set_title("Elbow Method for Optimal K")
ax.axvline(x=4, color="red", linestyle="--", label="Elbow at K=4")
ax.legend()
plt.tight_layout()
plt.savefig("elbow_method.png", dpi=150)

# Executed successfully

from sklearn.cluster import KMeans
from sklearn.metrics import silhouette_score
from sklearn.preprocessing import StandardScaler
import numpy as np

# Using same X_scaled from previous example
for k in range(2, 8):
    km = KMeans(n_clusters=k, random_state=42, n_init=10)
    labels = km.fit_predict(X_scaled)
    score = silhouette_score(X_scaled, labels)
    print(f"K={k}: Silhouette Score={score:.3f}")

# Executed successfully

from sklearn.cluster import DBSCAN
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import silhouette_score
import numpy as np

rng = np.random.default_rng(42)
# Create non-spherical clusters (two circles)
from sklearn.datasets import make_circles
X, _ = make_circles(n_samples=500, factor=0.5, noise=0.05, random_state=42)

X_scaled = StandardScaler().fit_transform(X)

# K-Means vs DBSCAN
km = KMeans(n_clusters=2, random_state=42, n_init=10)
km_labels = km.fit_predict(X_scaled)

db = DBSCAN(eps=0.3, min_samples=10)
db_labels = db.fit_predict(X_scaled)

n_clusters = len(set(db_labels)) - (1 if -1 in db_labels else 0)
n_noise = list(db_labels).count(-1)

print(f"K-Means Silhouette: {silhouette_score(X_scaled, km_labels):.3f}")
print(f"DBSCAN Clusters: {n_clusters}, Noise points: {n_noise}")
if n_clusters > 1:
    db_valid = db_labels != -1
    print(f"DBSCAN Silhouette: {silhouette_score(X_scaled[db_valid], db_labels[db_valid]):.3f}")

# Executed successfully

from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler
import pandas as pd
import numpy as np

rng = np.random.default_rng(42)
n = 5000
df = pd.DataFrame({
    "user_id": range(10001, 10001 + n),
    "recency_days": rng.integers(1, 365, n),
    "frequency": rng.integers(1, 50, n),
    "monetary_usd": rng.exponential(500, n),
})

# RFM features
rfm = df[["recency_days", "frequency", "monetary_usd"]]

# Log transform for skewed features
rfm_log = rfm.copy()
rfm_log["frequency"] = np.log1p(rfm_log["frequency"])
rfm_log["monetary_usd"] = np.log1p(rfm_log["monetary_usd"])

# Scale
scaler = StandardScaler()
rfm_scaled = scaler.fit_transform(rfm_log)

# K-Means with K=4
km = KMeans(n_clusters=4, random_state=42, n_init=10)
df["segment"] = km.fit_predict(rfm_scaled)

# Analyze segments
segment_summary = df.groupby("segment").agg({
    "recency_days": "mean",
    "frequency": "mean",
    "monetary_usd": "mean",
    "user_id": "count",
}).round(1)
segment_summary.columns = ["avg_recency", "avg_frequency", "avg_monetary", "count"]
segment_summary = segment_summary.sort_values("avg_monetary", ascending=False)

# Label segments
labels = ["Champions", "Loyal", "At Risk", "Lost"]
for idx, (_, row) in enumerate(segment_summary.iterrows()):
    print(f"{labels[idx]:10s}: {int(row['count']):5d} users, "
          f"Recency={row['avg_recency']:.0f}d, "
          f"Freq={row['avg_frequency']:.1f}, "
          f"Monetary={row['avg_monetary']:.0f} USD")

# Executed successfully