"""练习 0：手动实现 train/test 划分，理解监督学习基本流程"""
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_squared_error
# 模拟数据：y = 2x + 噪声
rng = np.random.default_rng(42)
X = rng.uniform(0, 10, size=(100, 1))
y = 2 * X.ravel() + rng.normal(0, 1, size=100)
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)
model = LinearRegression()
model.fit(X_train, y_train)
pred = model.predict(X_test)
print(f"测试集 MSE: {mean_squared_error(y_test, pred):.4f}")
print(f"学到的斜率≈2: {model.coef_[0]:.4f}")

"""练习 1：实现欧氏距离、曼哈顿距离、闵可夫斯基距离"""
import numpy as np
def euclidean(a: np.ndarray, b: np.ndarray) -> float:
    return np.sqrt(np.sum((a - b) ** 2))
def manhattan(a: np.ndarray, b: np.ndarray) -> float:
    return np.sum(np.abs(a - b))
def minkowski(a: np.ndarray, b: np.ndarray, p: int = 3) -> float:
    return np.sum(np.abs(a - b) ** p) ** (1 / p)
# 电影分类例子：接吻次数、打斗次数
train = np.array([
    [104, 3], [100, 2], [81, 1],   # 爱情
    [10, 101], [5, 99], [2, 98],   # 动作
])
labels = ["爱情", "爱情", "爱情", "动作", "动作", "动作"]
unknown = np.array([18, 90])
distances = [euclidean(unknown, x) for x in train]
k = 3
idx = np.argsort(distances)[:k]
votes = [labels[i] for i in idx]
from collections import Counter
print("K=3 预测:", Counter(votes).most_common(1)[0][0])

"""练习 2：0-1 归一化（课件 datingTestSet 同款思路）"""
import numpy as np
def min_max_norm(data: np.ndarray) -> np.ndarray:
    min_vals = data.min(axis=0)
    max_vals = data.max(axis=0)
    return (data - min_vals) / (max_vals - min_vals + 1e-8)
raw = np.array([[40920, 8.3, 0.95], [14488, 7.2, 1.67]])
print("归一化后:\n", min_max_norm(raw))

"""练习 3：绘制 K 值-准确率折线图模板"""
import matplotlib.pyplot as plt
plt.rcParams["font.sans-serif"] = ["Microsoft YaHei", "SimHei"]
plt.rcParams["axes.unicode_minus"] = False
ks = [1, 3, 5, 7, 9, 11]
acc = [0.92, 0.96, 0.97, 0.965, 0.96, 0.955]
plt.figure(figsize=(8, 5))
plt.plot(ks, acc, "o-", linewidth=2)
plt.xlabel("K（邻居数）")
plt.ylabel("测试准确率")
plt.title("K 值与分类准确率")
plt.grid(alpha=0.3)
plt.savefig("k_vs_accuracy_demo.png", dpi=150, bbox_inches="tight")
plt.close()

"""练习 4：sklearn KNN 手写数字识别"""
from pathlib import Path
import numpy as np
from sklearn.neighbors import KNeighborsClassifier
from sklearn.metrics import accuracy_score, classification_report
def load_digit_folder(folder: Path):
    xs, ys = [], []
    for path in sorted(folder.glob("*.txt")):
        label = int(path.stem.split("_")[0])
        rows = []
        with path.open(encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if line:
                    rows.append([int(ch) for ch in line])
        xs.append(np.asarray(rows, dtype=float).reshape(-1))
        ys.append(label)
    return np.asarray(xs), np.asarray(ys)
# 修改为你的数据路径
root = Path("机器学习作业/机器学习作业1/数据集")
X_train, y_train = load_digit_folder(root / "trainingDigits")
X_test, y_test = load_digit_folder(root / "testDigits")
# Baseline：sklearn 默认 K=5, 欧氏距离
clf = KNeighborsClassifier()
clf.fit(X_train, y_train)
pred = clf.predict(X_test)
print(f"Baseline 准确率: {accuracy_score(y_test, pred):.4f}")
print(classification_report(y_test, pred))
# 对比不同距离度量
for name, kw in [
    ("欧氏", {"metric": "euclidean"}),
    ("曼哈顿", {"metric": "manhattan"}),
    ("余弦", {"metric": "cosine"}),
]:
    model = KNeighborsClassifier(n_neighbors=5, **kw)
    model.fit(X_train, y_train)
    acc = accuracy_score(y_test, model.predict(X_test))
    print(f"{name}距离 K=5 准确率: {acc:.4f}")

"""练习 5：K 值 × 距离度量对比（对应作业1 grid 实验）"""
import pandas as pd
rows = []
for metric_name, kw in [
    ("euclidean", {"metric": "euclidean"}),
    ("manhattan", {"metric": "manhattan"}),
]:
    for k in range(1, 16, 2):
        m = KNeighborsClassifier(n_neighbors=k, **kw)
        m.fit(X_train, y_train)
        acc = accuracy_score(y_test, m.predict(X_test))
        rows.append({"metric": metric_name, "k": k, "accuracy": acc})
df = pd.DataFrame(rows)
best = df.loc[df["accuracy"].idxmax()]
print("最优组合:\n", best)

"""练习 6：PCA 降维 + KNN 管线"""
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
for n_comp in [10, 20, 50, 100, 200, 512]:
    pipe = Pipeline([
        ("scaler", StandardScaler()),
        ("pca", PCA(n_components=n_comp, random_state=42)),
        ("knn", KNeighborsClassifier(n_neighbors=5, metric="euclidean")),
    ])
    pipe.fit(X_train, y_train)
    acc = accuracy_score(y_test, pipe.predict(X_test))
    print(f"PCA n={n_comp:3d} → 准确率 {acc:.4f}")

"""练习 7：SVM 手写数字 Baseline + 网格搜索"""
from pathlib import Path
import numpy as np
from sklearn.svm import SVC
from sklearn.metrics import accuracy_score, classification_report
from sklearn.model_selection import GridSearchCV
# 复用练习4的 load_digit_folder
root = Path("机器学习作业/机器学习作业2/手写数字数据集")
X_train, y_train = load_digit_folder(root / "trainingDigits")
X_test, y_test = load_digit_folder(root / "testDigits")
# Baseline SVM (RBF)
baseline = SVC(cache_size=100)
baseline.fit(X_train, y_train)
pred = baseline.predict(X_test)
print(f"Baseline SVM 准确率: {accuracy_score(y_test, pred):.4f}")
# 网格搜索 C × gamma
param_grid = {
    "C": [0.1, 1, 10, 100],
    "gamma": ["scale", 0.001, 0.01, 0.1],
    "kernel": ["rbf"],
}
grid = GridSearchCV(SVC(cache_size=100), param_grid, cv=3, n_jobs=1)
grid.fit(X_train, y_train)
print("最优参数:", grid.best_params_)
print(f"最优测试准确率: {accuracy_score(y_test, grid.predict(X_test)):.4f}")

"""练习 8：双螺旋数据集上的 KNN / SVM 决策边界"""
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.neighbors import KNeighborsClassifier
from sklearn.svm import SVC
df = pd.read_csv("机器学习作业/机器学习作业2/双螺旋/spiral.csv")
X = df[["x", "y"]].values
y = df["label"].values
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.25, random_state=42, stratify=y
)
scaler = StandardScaler()
X_train_s = scaler.fit_transform(X_train)
X_test_s = scaler.transform(X_test)
for name, clf in [
    ("KNN k=5", KNeighborsClassifier(n_neighbors=5)),
    ("SVM linear", SVC(kernel="linear")),
    ("SVM RBF", SVC(kernel="rbf", gamma="scale")),
]:
    clf.fit(X_train_s, y_train)
    acc = accuracy_score(y_test, clf.predict(X_test_s))
    print(f"{name}: {acc:.4f}")
# 可视化决策边界（简化版）
xx, yy = np.meshgrid(
    np.linspace(X_train_s[:, 0].min() - 0.5, X_train_s[:, 0].max() + 0.5, 200),
    np.linspace(X_train_s[:, 1].min() - 0.5, X_train_s[:, 1].max() + 0.5, 200),
)
svm = SVC(kernel="rbf", gamma="scale")
svm.fit(X_train_s, y_train)
Z = svm.predict(np.c_[xx.ravel(), yy.ravel()]).reshape(xx.shape)
plt.contourf(xx, yy, Z, alpha=0.3, cmap="coolwarm")
plt.scatter(X_train_s[:, 0], X_train_s[:, 1], c=y_train, cmap="coolwarm", edgecolors="k", s=20)
plt.title("双螺旋：RBF-SVM 决策边界")
plt.savefig("spiral_svm_boundary.png", dpi=150)
plt.close()

"""练习 9：泰坦尼克 Age 缺失值 KNN 回归填补"""
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.neighbors import KNeighborsRegressor
from sklearn.metrics import mean_absolute_error
df = pd.read_csv("机器学习作业/机器学习作业3/泰坦尼克号/train.csv")
known = df[df["Age"].notna()].copy()
feature_cols = ["Pclass", "Sex", "SibSp", "Parch", "Fare", "Embarked"]
X = known[feature_cols]
y = known["Age"].values
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.25, random_state=42)
pre = ColumnTransformer([
    ("num", Pipeline([
        ("imputer", SimpleImputer(strategy="median")),
        ("scaler", StandardScaler()),
    ]), ["Pclass", "SibSp", "Parch", "Fare"]),
    ("cat", Pipeline([
        ("imputer", SimpleImputer(strategy="most_frequent")),
        ("encoder", OneHotEncoder(handle_unknown="ignore", sparse_output=False)),
    ]), ["Sex", "Embarked"]),
])
pipe = Pipeline([
    ("pre", pre),
    ("model", KNeighborsRegressor(n_neighbors=5, weights="distance")),
])
pipe.fit(X_train, y_train)
pred = pipe.predict(X_test)
print(f"Age 预测 MAE: {mean_absolute_error(y_test, pred):.3f} 岁")

"""练习 10：决策树预测泰坦尼克生存"""
from sklearn.tree import DecisionTreeClassifier, plot_tree
from sklearn.metrics import accuracy_score, classification_report
from sklearn.model_selection import cross_val_score
# 用 KNN 填补后的 Age
def impute_age_knn(df, feature_cols):
    data = df.copy()
    known = data[data["Age"].notna()]
    missing = data[data["Age"].isna()]
    if missing.empty:
        return data["Age"]
    pre = ColumnTransformer([
        ("num", Pipeline([
            ("imputer", SimpleImputer(strategy="median")),
            ("scaler", StandardScaler()),
        ]), [c for c in feature_cols if c not in ("Sex", "Embarked")]),
        ("cat", Pipeline([
            ("imputer", SimpleImputer(strategy="most_frequent")),
            ("encoder", OneHotEncoder(handle_unknown="ignore", sparse_output=False)),
        ]), [c for c in feature_cols if c in ("Sex", "Embarked")]),
    ])
    reg_pipe = Pipeline([("pre", pre), ("model", KNeighborsRegressor(n_neighbors=5))])
    reg_pipe.fit(known[feature_cols], known["Age"])
    data.loc[missing.index, "Age"] = reg_pipe.predict(missing[feature_cols])
    return data["Age"]
features = ["Pclass", "Sex", "Age", "SibSp", "Parch", "Fare", "Embarked"]
df = pd.read_csv("机器学习作业/机器学习作业3/泰坦尼克号/train.csv")
df["Age"] = impute_age_knn(df, ["Pclass", "Sex", "SibSp", "Parch", "Fare", "Embarked"])
X = df[features]
y = df["Survived"]
dt_pipe = Pipeline([
    ("pre", ColumnTransformer([
        ("num", Pipeline([
            ("imputer", SimpleImputer(strategy="median")),
            ("scaler", StandardScaler()),
        ]), ["Pclass", "Age", "SibSp", "Parch", "Fare"]),
        ("cat", Pipeline([
            ("imputer", SimpleImputer(strategy="most_frequent")),
            ("encoder", OneHotEncoder(handle_unknown="ignore", sparse_output=False)),
        ]), ["Sex", "Embarked"]),
    ])),
    ("clf", DecisionTreeClassifier(max_depth=5, min_samples_leaf=5, random_state=42)),
])
cv = cross_val_score(dt_pipe, X, y, cv=5, scoring="accuracy")
print(f"5折CV准确率: {cv.mean():.4f} ± {cv.std():.4f}")
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.25, random_state=42, stratify=y)
dt_pipe.fit(X_train, y_train)
print(classification_report(y_test, dt_pipe.predict(X_test), target_names=["未生还", "生还"]))

"""练习 11：连锁餐饮 K-Means 聚类（作业4核心）"""
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from sklearn.preprocessing import StandardScaler
from sklearn.cluster import KMeans
from sklearn.metrics import silhouette_score
from sklearn.decomposition import PCA
df = pd.read_csv("机器学习作业/机器学习作业4/restaurant.csv")
CLUSTER_FEATURES = ["V3", "V4", "V5", "V6", "V7"]  # 门店数、人数、面积、餐位、营业额
X_raw = df[CLUSTER_FEATURES].astype(float).values
X = StandardScaler().fit_transform(X_raw)
# K 值评估：肘部法 + 轮廓系数
sse_list, sil_list = [], []
for k in range(2, 9):
    km = KMeans(n_clusters=k, init="k-means++", n_init=10, random_state=42)
    labels = km.fit_predict(X)
    sse_list.append(km.inertia_)
    sil_list.append(silhouette_score(X, labels))
    print(f"K={k}  SSE={km.inertia_:.2f}  SC={sil_list[-1]:.4f}")
chosen_k = range(2, 9)[int(np.argmax(sil_list))]
km = KMeans(n_clusters=chosen_k, init="k-means++", n_init=10, random_state=42)
labels = km.fit_predict(X)
df["cluster"] = labels
print("\n各地区聚类结果:")
print(df[["V1", "cluster"] + CLUSTER_FEATURES].sort_values("cluster"))
# PCA 二维可视化
pca = PCA(n_components=2, random_state=42)
X2 = pca.fit_transform(X)
plt.figure(figsize=(8, 6))
plt.scatter(X2[:, 0], X2[:, 1], c=labels, cmap="viridis", s=80, edgecolors="k")
for i, name in enumerate(df["V1"]):
    plt.annotate(name, (X2[i, 0], X2[i, 1]), fontsize=7)
plt.xlabel(f"PC1 ({pca.explained_variance_ratio_[0]*100:.1f}%)")
plt.ylabel(f"PC2 ({pca.explained_variance_ratio_[1]*100:.1f}%)")
plt.title(f"连锁餐饮 PCA 聚类可视化 (K={chosen_k})")
plt.savefig("restaurant_clusters.png", dpi=150, bbox_inches="tight")
plt.close()

"""练习 12：PCA 方差解释比与二维可视化"""
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
df = pd.read_csv("机器学习作业/机器学习作业4/restaurant.csv")
cols = ["V3", "V4", "V5", "V6", "V7", "V8", "V9"]
X = StandardScaler().fit_transform(df[cols].astype(float))
pca = PCA()
pca.fit(X)
print("各主成分方差解释比:", np.round(pca.explained_variance_ratio_, 4))
print("累计解释比:", np.round(np.cumsum(pca.explained_variance_ratio_), 4))
# 碎石图
plt.figure(figsize=(8, 4))
plt.bar(range(1, len(cols)+1), pca.explained_variance_ratio_, label="单个")
plt.plot(range(1, len(cols)+1), np.cumsum(pca.explained_variance_ratio_), "ro-", label="累计")
plt.xlabel("主成分编号")
plt.ylabel("方差解释比")
plt.legend()
plt.title("PCA 碎石图")
plt.savefig("pca_scree_plot.png", dpi=150)
plt.close()

"""练习 13：PyTorch 简易 CNN 图像分类框架"""
import torch
import torch.nn as nn
from torch.utils.data import DataLoader
from torchvision import datasets, transforms
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
class SimpleCNN(nn.Module):
    def __init__(self):
        super().__init__()
        self.net = nn.Sequential(
            nn.Conv2d(1, 16, 3, padding=1), nn.ReLU(),
            nn.MaxPool2d(2),
            nn.Conv2d(16, 32, 3, padding=1), nn.ReLU(),
            nn.MaxPool2d(2),
            nn.Flatten(),
            nn.Linear(32 * 7 * 7, 128), nn.ReLU(),
            nn.Linear(128, 10),
        )
    def forward(self, x):
        return self.net(x)
transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.1307,), (0.3081,)),
])
train_set = datasets.MNIST("./data", train=True, download=True, transform=transform)
test_set = datasets.MNIST("./data", train=False, transform=transform)
train_loader = DataLoader(train_set, batch_size=64, shuffle=True)
model = SimpleCNN().to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
criterion = nn.CrossEntropyLoss()
model.train()
for epoch in range(3):  # 演示用 3 epoch，实际需更多
    total_loss = 0.0
    for x, y in train_loader:
        x, y = x.to(device), y.to(device)
        optimizer.zero_grad()
        loss = criterion(model(x), y)
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
    print(f"Epoch {epoch+1} loss: {total_loss/len(train_loader):.4f}")

"""练习 14：PyTorch LSTM 序列分类骨架（many-to-one）"""
import torch
import torch.nn as nn
class LSTMClassifier(nn.Module):
    def __init__(self, vocab_size, embed_dim=64, hidden_dim=128, num_classes=2):
        super().__init__()
        self.embed = nn.Embedding(vocab_size, embed_dim)
        self.lstm = nn.LSTM(embed_dim, hidden_dim, batch_first=True)
        self.fc = nn.Linear(hidden_dim, num_classes)
    def forward(self, x):
        # x: (batch, seq_len)
        emb = self.embed(x)           # (batch, seq_len, embed_dim)
        _, (h_n, _) = self.lstm(emb)  # h_n: (1, batch, hidden)
        return self.fc(h_n.squeeze(0))
# 模拟 batch
model = LSTMClassifier(vocab_size=5000)
dummy_x = torch.randint(0, 5000, (32, 20))  # 32 句，每句 20 token
logits = model(dummy_x)
print("输出形状:", logits.shape)  # (32, 2)

第1章 绪论（概念）
  ↓
第2章 Python/NumPy（工具）
  ↓
作业1：KNN 手写数字（监督分类入门）
  ↓
作业2：SVM + 双螺旋（非线性分类）
  ↓
作业3：决策树 + 缺失值（数据预处理 + 可解释模型）
  ↓
作业4：K-Means + PCA（无监督学习）
  ↓
第8–9章 CNN / RNN（深度学习进阶）