优化器：SGD, Momentum, Adam, AdamW

import numpy as np

# 三种梯度下降的实现对比
def batch_gd(X, y, theta, lr=0.01, epochs=100):
    """批量梯度下降：用全部数据计算梯度"""
    m = X.shape[0]
    for epoch in range(epochs):
        grad = (1/m) * X.T @ (X @ theta - y)  # 全部样本
        theta = theta - lr * grad
    return theta

def sgd(X, y, theta, lr=0.01, epochs=100):
    """随机梯度下降：每次只用一个样本"""
    m = X.shape[0]
    for epoch in range(epochs):
        for i in range(m):
            xi = X[i:i+1]
            yi = y[i:i+1]
            grad = xi.T @ (xi @ theta - yi)  # 单样本
            theta = theta - lr * grad
    return theta

def minibatch_gd(X, y, theta, lr=0.01, epochs=100, batch_size=32):
    """Mini-batch 梯度下降：每次用一批样本"""
    m = X.shape[0]
    for epoch in range(epochs):
        indices = np.random.permutation(m)
        for start in range(0, m, batch_size):
            batch_idx = indices[start:start + batch_size]
            X_b = X[batch_idx]
            y_b = y[batch_idx]
            grad = (1/len(batch_idx)) * X_b.T @ (X_b @ theta - y_b)
            theta = theta - lr * grad
    return theta

import torch
import torch.nn as nn

# PyTorch 中的 Mini-batch 梯度下降
class LinearModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.linear = nn.Linear(10, 1)

    def forward(self, x):
        return self.linear(x)

model = LinearModel()
criterion = nn.MSELoss()
optimizer = torch.optim.SGD(model.parameters(), lr=0.01)

# 模拟 mini-batch 训练
X = torch.randn(1000, 10)
y = torch.randn(1000, 1)
batch_size = 32

for epoch in range(50):
    perm = torch.randperm(X.size(0))
    X_shuffled = X[perm]
    y_shuffled = y[perm]

    for i in range(0, X.size(0), batch_size):
        X_batch = X_shuffled[i:i+batch_size]
        y_batch = y_shuffled[i:i+batch_size]

        optimizer.zero_grad()
        pred = model(X_batch)
        loss = criterion(pred, y_batch)
        loss.backward()
        optimizer.step()

import numpy as np

def sgd_with_momentum(grads, lr=0.01, beta=0.9):
    """带 Momentum 的 SGD"""
    velocity = np.zeros_like(grads[0])
    thetas = [np.zeros_like(grads[0])]

    for g in grads:
        velocity = beta * velocity + lr * g
        theta = thetas[-1] - velocity
        thetas.append(theta)
    return thetas

def nesterov_accelerated_gradient(grads, lr=0.01, beta=0.9):
    """Nesterov 加速梯度下降"""
    velocity = np.zeros_like(grads[0])
    thetas = [np.zeros_like(grads[0])]

    for g in grads:
        # 先看一步：θ_look_ahead = θ - β·v
        look_ahead = thetas[-1] - beta * velocity
        # 在看的位置计算梯度（此处用 g 近似）
        velocity = beta * velocity + lr * g
        theta = thetas[-1] - velocity
        thetas.append(theta)
    return thetas

# 对比两种方法的收敛路径
grads = np.sin(np.linspace(0, 4*np.pi, 200)).reshape(-1, 1)
momentum_path = sgd_with_momentum(grads)
nag_path = nesterov_accelerated_gradient(grads)
print(f"Momentum 最终位置: {momentum_path[-1][0]:.4f}")
print(f"NAG 最终位置: {nag_path[-1][0]:.4f}")

import torch
import torch.nn as nn

# PyTorch 中的 SGD + Momentum
model = nn.Sequential(
    nn.Linear(784, 256),
    nn.ReLU(),
    nn.Linear(256, 10)
)

# 标准 SGD（无动量）
sgd = torch.optim.SGD(model.parameters(), lr=0.01)

# SGD + Momentum
sgd_momentum = torch.optim.SGD(
    model.parameters(),
    lr=0.01,
    momentum=0.9,           # 动量系数
    weight_decay=1e-4,      # L2 正则化
    nesterov=False          # 设为 True 则启用 NAG
)

# SGD + Nesterov
sgd_nesterov = torch.optim.SGD(
    model.parameters(),
    lr=0.01,
    momentum=0.9,
    nesterov=True           # 启用 Nesterov 加速
)

print(f"SGD 状态: {sgd.state_dict()}")
print(f"Momentum 缓冲数量: {len(sgd_momentum.state)}")

import numpy as np

def adagrad(grads, lr=0.01, eps=1e-8):
    """AdaGrad 优化器"""
    theta = np.zeros_like(grads[0])
    G = np.zeros_like(grads[0])  # 累积梯度平方

    for g in grads:
        G += g ** 2  # 单调递增，永不停止
        update = lr * g / (np.sqrt(G) + eps)
        theta = theta - update
    return theta

def rmsprop(grads, lr=0.001, beta=0.9, eps=1e-8):
    """RMSProp 优化器：用 EMA 替代累积"""
    theta = np.zeros_like(grads[0])
    E_g2 = np.zeros_like(grads[0])  # 梯度平方的 EMA

    for g in grads:
        E_g2 = beta * E_g2 + (1 - beta) * g ** 2
        update = lr * g / (np.sqrt(E_g2) + eps)
        theta = theta - update
    return theta

# 模拟：某些维度梯度大，某些梯度小
np.random.seed(42)
large_grads = np.random.randn(100, 5) * 10   # 频繁大幅更新
small_grads = np.random.randn(100, 5) * 0.1  # 稀疏小幅更新

import torch
import torch.nn as nn

# RMSProp 在 PyTorch 中的使用
model = nn.Sequential(
    nn.Linear(784, 128),
    nn.ReLU(),
    nn.Linear(128, 64),
    nn.ReLU(),
    nn.Linear(64, 10)
)

# RMSProp 优化器
optimizer = torch.optim.RMSprop(
    model.parameters(),
    lr=0.001,              # RMSProp 推荐学习率较小
    alpha=0.99,            # EMA 衰减系数 (即 β)
    eps=1e-8,              # 数值稳定性
    weight_decay=0,        # L2 正则化（可选）
    momentum=0,            # 可选：RMSProp + Momentum
    centered=False         # 是否使用中心化 RMSProp
)

# 带 Momentum 的 RMSProp（相当于 Adam 的前身）
optimizer_with_mom = torch.optim.RMSprop(
    model.parameters(),
    lr=0.001,
    alpha=0.99,
    momentum=0.9,          # 加入动量
    centered=True          # 中心化 RMSProp
)

import numpy as np

def adam(grads, lr=0.001, beta1=0.9, beta2=0.999, eps=1e-8):
    """Adam 优化器从零实现"""
    theta = np.zeros_like(grads[0])
    m = np.zeros_like(grads[0])  # 一阶矩（均值）
    v = np.zeros_like(grads[0])  # 二阶矩（方差）

    for t, g in enumerate(grads, 1):
        m = beta1 * m + (1 - beta1) * g
        v = beta2 * v + (1 - beta2) * g ** 2

        # 偏差校正（关键！）
        m_hat = m / (1 - beta1 ** t)
        v_hat = v / (1 - beta2 ** t)

        # 参数更新
        update = lr * m_hat / (np.sqrt(v_hat) + eps)
        theta = theta - update

        if t in [1, 2, 10, 100]:
            correction = 1 / (1 - beta1 ** t)
            print(f"t={t}: β₁ᵗ={beta1**t:.6f}, 校正因子={correction:.6f}")

    return theta

# 验证偏差校正：t=1 时校正因子 = 1/(1-0.9) = 10
print(f"t=1: 校正因子 = {1/(1-0.9**1):.2f}")  # 10.00
print(f"t=10: 校正因子 = {1/(1-0.9**10):.4f}")  # 1.5377
print(f"t=100: 校正因子 = {1/(1-0.9**100):.6f}")  # 1.0000

import torch
import torch.nn as nn
import torchvision.models as models

# Adam 在 PyTorch 中的标准使用
model = models.resnet18(num_classes=10)

optimizer = torch.optim.Adam(
    model.parameters(),
    lr=0.001,              # 推荐学习率
    betas=(0.9, 0.999),    # (β₁, β₂)
    eps=1e-8,              # 数值稳定项
    weight_decay=0,        # L2 正则（注意：这不是 AdamW）
    amsgrad=False          # 是否使用 AMSGrad 变体
)

# Adam 状态检查
print(f"默认超参数: lr={optimizer.defaults['lr']}")
print(f"β₁={optimizer.defaults['betas'][0]}, β₂={optimizer.defaults['betas'][1]}")
print(f"ε={optimizer.defaults['eps']}")
print(f"状态字典中有多少组参数: {len(optimizer.state)}")
# 每个参数组都有 m (exp_avg) 和 v (exp_avg_sq) 两个状态

import numpy as np

def adam_with_weight_decay(grads, theta_init, lr=0.001,
                            beta1=0.9, beta2=0.999, eps=1e-8, weight_decay=0.01):
    """Adam + weight_decay（有问题的做法）"""
    theta = theta_init.copy()
    m, v = np.zeros_like(theta), np.zeros_like(theta)

    for t, g in enumerate(grads, 1):
        g_reg = g + weight_decay * theta  # weight_decay 混入梯度
        m = beta1 * m + (1 - beta1) * g_reg
        v = beta2 * v + (1 - beta2) * g_reg ** 2
        m_hat = m / (1 - beta1 ** t)
        v_hat = v / (1 - beta2 ** t)
        theta = theta - lr * m_hat / (np.sqrt(v_hat) + eps)
    return theta

def adamw(grads, theta_init, lr=0.001,
          beta1=0.9, beta2=0.999, eps=1e-8, weight_decay=0.01):
    """AdamW：权重衰减与梯度更新解耦"""
    theta = theta_init.copy()
    m, v = np.zeros_like(theta), np.zeros_like(theta)

    for t, g in enumerate(grads, 1):
        m = beta1 * m + (1 - beta1) * g  # 注意：g 不加 weight_decay
        v = beta2 * v + (1 - beta2) * g ** 2
        m_hat = m / (1 - beta1 ** t)
        v_hat = v / (1 - beta2 ** t)
        # 解耦的权重衰减
        theta = theta - lr * m_hat / (np.sqrt(v_hat) + eps) - lr * weight_decay * theta
    return theta

import torch
import torch.nn as nn
import torchvision.models as models

model = models.resnet50(num_classes=100)

# ❌ 错误做法：Adam + weight_decay（不等价于 L2 正则）
bad_optimizer = torch.optim.Adam(
    model.parameters(),
    lr=0.001,
    weight_decay=0.01  # 在 Adam 中，这不是真正的 L2 正则
)

# ✅ 正确做法：AdamW（解耦权重衰减）
good_optimizer = torch.optim.AdamW(
    model.parameters(),
    lr=0.001,
    weight_decay=0.01,     # 真正的解耦权重衰减
    betas=(0.9, 0.999),
    eps=1e-8
)

# 学习率预热 + 余弦退火（AdamW 最佳实践）
from torch.optim.lr_scheduler import OneCycleLR
scheduler = OneCycleLR(
    good_optimizer,
    max_lr=0.003,
    epochs=100,
    steps_per_epoch=len(train_loader),
    pct_start=0.1,          # 前 10% warmup
    anneal_strategy='cos'   # 余弦退火
)

import torch
import torch.nn as nn
from torchvision.models import resnet18

def setup_optimizers(model, lr=0.001):
    """配置六种优化器用于公平对比"""
    optimizers = {
        "SGD": torch.optim.SGD(
            model.parameters(), lr=0.1, momentum=0.9, weight_decay=5e-4),
        "SGD+NAG": torch.optim.SGD(
            model.parameters(), lr=0.1, momentum=0.9,
            weight_decay=5e-4, nesterov=True),
        "Adam": torch.optim.Adam(
            model.parameters(), lr=0.001, betas=(0.9, 0.999)),
        "AdamW": torch.optim.AdamW(
            model.parameters(), lr=0.001, weight_decay=0.01),
        "RMSProp": torch.optim.RMSprop(
            model.parameters(), lr=0.001, alpha=0.99),
        "AdaGrad": torch.optim.Adagrad(
            model.parameters(), lr=0.01),
    }
    return optimizers

# 统一训练框架
def benchmark_optimizer(name, optimizer, model, trainloader, testloader,
                        epochs=50, device='cuda'):
    model_copy = deepcopy(model).to(device)
    criterion = nn.CrossEntropyLoss()

    for epoch in range(epochs):
        model_copy.train()
        for X, y in trainloader:
            X, y = X.to(device), y.to(device)
            optimizer.zero_grad()
            loss = criterion(model_copy(X), y)
            loss.backward()
            optimizer.step()

    # 测试精度
    model_copy.eval()
    correct = total = 0
    with torch.no_grad():
        for X, y in testloader:
            correct += (model_copy(X.to(device)).argmax(1) == y.to(device)).sum()
            total += y.size(0)
    return (correct / total).item()

import matplotlib.pyplot as plt

# 可视化训练曲线对比
results = {
    "SGD":        {"train_acc": [0.35, 0.55, 0.68, 0.78, 0.85, 0.89],
                   "test_acc":  [0.32, 0.52, 0.65, 0.75, 0.82, 0.87]},
    "Adam":       {"train_acc": [0.55, 0.75, 0.85, 0.91, 0.94, 0.96],
                   "test_acc":  [0.50, 0.70, 0.79, 0.83, 0.85, 0.86]},
    "AdamW":      {"train_acc": [0.55, 0.75, 0.85, 0.91, 0.94, 0.95],
                   "test_acc":  [0.50, 0.71, 0.81, 0.86, 0.88, 0.89]},
}

epochs = [10, 20, 30, 40, 50, 60]
for name, data in results.items():
    plt.plot(epochs, data["train_acc"], '--', label=f"{name} train")
    plt.plot(epochs, data["test_acc"], '-', label=f"{name} test")

plt.xlabel("Epochs")
plt.ylabel("Accuracy")
plt.title("Optimizer Comparison on CIFAR-10 (ResNet-18)")
plt.legend()
plt.grid(True, alpha=0.3)
plt.savefig("optimizer_comparison.png", dpi=150)

import torch
import torch.nn as nn
from torchvision.models import resnet50

def get_optimizer_and_scheduler(name, model, steps_per_epoch, epochs=100):
    """统一的优化器+调度器配置工厂"""
    if name == "adamw_cosine":
        optimizer = torch.optim.AdamW(
            model.parameters(), lr=0.001, weight_decay=0.01)
        scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(
            optimizer, T_max=epochs, eta_min=1e-6)
    elif name == "sgd_cosine_warmup":
        optimizer = torch.optim.SGD(
            model.parameters(), lr=0.1, momentum=0.9, weight_decay=5e-4)
        scheduler = torch.optim.lr_scheduler.OneCycleLR(
            optimizer, max_lr=0.1, epochs=epochs,
            steps_per_epoch=steps_per_epoch, pct_start=0.1)
    elif name == "adamw_warmup_cosine":
        optimizer = torch.optim.AdamW(
            model.parameters(), lr=0.001, weight_decay=0.01)
        # 手动 warmup + cosine
        def lr_lambda(current_step):
            warmup_steps = 500
            if current_step < warmup_steps:
                return current_step / warmup_steps  # 线性预热
            else:
                # 余弦退火
                import math
                progress = (current_step - warmup_steps) / (epochs * steps_per_epoch - warmup_steps)
                return 0.5 * (1 + math.cos(math.pi * progress))
        scheduler = torch.optim.lr_scheduler.LambdaLR(optimizer, lr_lambda)
    else:
        raise ValueError(f"Unknown config: {name}")

    return optimizer, scheduler

import torch
import torch.nn as nn
from torch.utils.data import DataLoader

# 完整训练循环（AdamW + Cosine + 梯度裁剪）
def train_model(model, trainloader, testloader, epochs=100, device='cuda'):
    model = model.to(device)
    criterion = nn.CrossEntropyLoss()
    optimizer = torch.optim.AdamW(
        model.parameters(), lr=0.001, weight_decay=0.01)
    scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(
        optimizer, T_max=epochs, eta_min=1e-6)

    best_acc = 0
    for epoch in range(epochs):
        # 训练
        model.train()
        total_loss = correct = total = 0
        for X, y in trainloader:
            X, y = X.to(device), y.to(device)

            optimizer.zero_grad()
            outputs = model(X)
            loss = criterion(outputs, y)
            loss.backward()

            # 梯度裁剪（防止梯度爆炸）
            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)

            optimizer.step()
            total_loss += loss.item() * X.size(0)
            correct += outputs.argmax(1).eq(y).sum().item()
            total += X.size(0)

        scheduler.step()
        train_acc = correct / total

        # 验证
        model.eval()
        val_correct = val_total = 0
        with torch.no_grad():
            for X, y in testloader:
                pred = model(X.to(device)).argmax(1)
                val_correct += pred.eq(y.to(device)).sum().item()
                val_total += y.size(0)
        val_acc = val_correct / val_total

        if val_acc > best_acc:
            best_acc = val_acc
            torch.save(model.state_dict(), 'best_model.pth')

        if (epoch + 1) % 10 == 0:
            print(f"Epoch {epoch+1}: lr={scheduler.get_last_lr()[0]:.6f}, "
                  f"train={train_acc:.4f}, val={val_acc:.4f}, best={best_acc:.4f}")