#迁移学习 (Transfer Learning)：利用预训练模型快速构建高性能模型

"""
迁移学习的核心优势：

1. 知识转移：
   - 预训练模型已经学习了通用的视觉特征
   - 如边缘、纹理、形状等低级特征
   - 对象部分、整体等中级特征

2. 资源节约：
   - 显著减少训练时间
   - 降低计算资源需求
   - 减少数据收集成本

3. 性能提升：
   - 在小数据集上也能获得良好性能
   - 避免从零开始训练的不稳定
   - 更快的收敛速度
"""

def explain_transfer_learning_advantages():
    """
    解释迁移学习的优势
    """
    advantages = {
        "Time Savings": "从数天/周缩短至数小时",
        "Resource Efficiency": "减少GPU/CPU使用量",
        "Performance Boost": "在小数据集上提升准确率10-15%",
        "Stability": "避免训练不稳定问题",
        "Generalization": "更好的泛化能力"
    }
    
    print("迁移学习核心优势:")
    for advantage, benefit in advantages.items():
        print(f"• {advantage}: {benefit}")

explain_transfer_learning_advantages()

def transfer_learning_mechanism():
    """
    迁移学习的工作机制
    """
    """
    网络层级特征分析：
    
    低层（浅层）：学习基本特征
    - 边缘检测
    - 角点检测
    - 纹理模式
    
    中层：学习复合特征
    - 形状组合
    - 简单对象部件
    
    高层（深层）：学习抽象特征
    - 完整对象
    - 语义概念
    
    迁移策略：
    - 保留低层特征（通用性较强）
    - 微调高层特征（任务相关）
    - 替换输出层（任务特定）
    """
    
    print("迁移学习机制分析:")
    print("• 低层特征: 通用性强，通常冻结")
    print("• 高层特征: 任务相关，需要微调")
    print("• 输出层: 任务特定，完全替换")

transfer_learning_mechanism()

import torch
import torchvision.models as models

def pretrained_models_comparison():
    """
    主流预训练模型对比分析
    """
    models_info = {
        "ResNet50": {
            "params": "25.6M",
            "top1_acc": "76.15%",
            "use_case": "通用分类任务",
            "speed": "中等"
        },
        "ResNet101": {
            "params": "44.5M", 
            "top1_acc": "77.37%",
            "use_case": "高精度要求",
            "speed": "较慢"
        },
        "VGG16": {
            "params": "138.4M",
            "top1_acc": "71.59%",
            "use_case": "传统方法对比",
            "speed": "较慢"
        },
        "VGG19": {
            "params": "143.7M",
            "top1_acc": "72.38%",
            "use_case": "高容量需求",
            "speed": "慢"
        },
        "EfficientNetB0": {
            "params": "5.3M",
            "top1_acc": "77.69%",
            "use_case": "移动/嵌入式",
            "speed": "快"
        },
        "MobileNetV3": {
            "params": "5.4M",
            "top1_acc": "75.27%",
            "use_case": "移动端部署",
            "speed": "最快"
        },
        "DenseNet121": {
            "params": "8.0M",
            "top1_acc": "74.43%",
            "use_case": "参数效率",
            "speed": "中等"
        },
        "InceptionV3": {
            "params": "23.8M",
            "top1_acc": "77.46%",
            "use_case": "多尺度特征",
            "speed": "中等"
        }
    }
    
    print("主流预训练模型对比:")
    print(f"{'模型':<15} {'参数量':<10} {'Top-1准确率':<12} {'适用场景':<15} {'速度':<8}")
    print("-" * 70)
    for model, info in models_info.items():
        print(f"{model:<15} {info['params']:<10} {info['top1_acc']:<12} {info['use_case']:<15} {info['speed']:<8}")

pretrained_models_comparison()

def load_pretrained_models():
    """
    加载各种预训练模型
    """
    # ResNet系列
    resnet50 = models.resnet50(pretrained=True)
    resnet101 = models.resnet101(pretrained=True)
    
    # VGG系列
    vgg16 = models.vgg16(pretrained=True)
    vgg19 = models.vgg19(pretrained=True)
    
    # EfficientNet系列
    try:
        from efficientnet_pytorch import EfficientNet
        efficientnet_b0 = EfficientNet.from_pretrained('efficientnet-b0')
    except ImportError:
        print("EfficientNet需要额外安装: pip install efficientnet-pytorch")
    
    # MobileNet系列
    mobilenet_v3 = models.mobilenet_v3_small(pretrained=True)
    
    print("预训练模型加载示例:")
    print("• ResNet50: 已加载")
    print("• VGG16: 已加载") 
    print("• MobileNetV3: 已加载")
    print("• 更多模型可通过torchvision.models获取")

load_pretrained_models()

def feature_extraction_strategy(num_classes=10):
    """
    特征提取策略：冻结预训练模型，只训练新层
    """
    # 加载预训练模型
    model = models.resnet50(pretrained=True)
    
    # 冻结所有预训练层的参数
    for param in model.parameters():
        param.requires_grad = False
    
    # 获取全连接层的输入特征数
    num_features = model.fc.in_features
    
    # 替换最后的全连接层
    model.fc = torch.nn.Linear(num_features, num_classes)
    
    # 只优化新添加的层
    optimizer = torch.optim.Adam(model.fc.parameters(), lr=0.001)
    
    print("特征提取策略配置:")
    print(f"• 冻结层数: {sum(p.requires_grad == False for p in model.parameters())}")
    print(f"• 可训练参数: {sum(p.numel() for p in model.fc.parameters())}")
    print(f"• 优化器: Adam, lr=0.001")
    
    return model, optimizer

def analyze_feature_extraction():
    """
    分析特征提取策略的优缺点
    """
    print("特征提取策略分析:")
    print("优点:")
    print("• 计算成本低")
    print("• 内存占用少") 
    print("• 快速训练")
    print("• 避免过拟合")
    print("\n缺点:")
    print("• 无法适应特定任务")
    print("• 性能提升有限")
    print("• 依赖预训练特征质量")

analyze_feature_extraction()

def fine_tuning_strategy(num_classes=10, freeze_layers=4):
    """
    微调策略：部分解冻预训练层，进行精细调整
    """
    # 加载预训练模型
    model = models.resnet50(pretrained=True)
    
    # 冻结前面的层（保留通用特征）
    children = list(model.children())
    for child in children[:freeze_layers]:
        for param in child.parameters():
            param.requires_grad = False
    
    # 替换最后的分类层
    num_features = model.fc.in_features
    model.fc = torch.nn.Linear(num_features, num_classes)
    
    # 设置不同层的不同学习率
    params = [
        {"params": model.fc.parameters(), "lr": 0.001},  # 新层使用较高学习率
        {"params": model.layer4.parameters(), "lr": 0.0001},  # 最后一层较低学习率
    ]
    
    optimizer = torch.optim.Adam(params)
    
    print("微调策略配置:")
    print(f"• 冻结前{freeze_layers}个模块")
    print(f"• FC层学习率: 0.001")
    print(f"• layer4学习率: 0.0001")
    print(f"• 总参数量: {sum(p.numel() for p in model.parameters())}")
    
    return model, optimizer

def advanced_fine_tuning():
    """
    高级微调策略：逐层解冻
    """
    """
    逐层解冻策略：
    1. 先训练最后几层
    2. 逐渐解冻更深层
    3. 最后微调全部层
    
    这种策略可以更稳定地适应新任务
    """
    
    def progressive_unfreezing(model, stage):
        """
        根据阶段逐步解冻层
        """
        if stage == 0:
            # 只训练最后的分类层
            for param in model.parameters():
                param.requires_grad = False
            for param in model.fc.parameters():
                param.requires_grad = True
        elif stage == 1:
            # 解冻最后两层
            for param in model.layer4.parameters():
                param.requires_grad = True
        elif stage == 2:
            # 解冻更多层
            for param in model.layer3.parameters():
                param.requires_grad = True
        else:
            # 解冻所有层
            for param in model.parameters():
                param.requires_grad = True
    
    print("高级微调策略:")
    print("• 逐层解冻: 从顶层到底层逐步解冻")
    print("• 学习率衰减: 不同层使用不同学习率")
    print("• 循环学习率: 动态调整学习率")

advanced_fine_tuning()

def full_fine_tuning_strategy(num_classes=10):
    """
    全量微调：所有层都参与训练，但使用较小学习率
    """
    # 加载预训练模型
    model = models.resnet50(pretrained=True)
    
    # 替换分类层
    num_features = model.fc.in_features
    model.fc = torch.nn.Linear(num_features, num_classes)
    
    # 解冻所有层，但使用较小学习率
    for param in model.parameters():
        param.requires_grad = True
    
    # 使用较小的全局学习率
    optimizer = torch.optim.Adam(model.parameters(), lr=0.0001)
    
    print("全量微调策略配置:")
    print("• 所有层都可训练")
    print("• 全局学习率: 0.0001 (较小)")
    print("• 适用于大数据集")
    
    return model, optimizer

def strategy_selection_guide():
    """
    迁移学习策略选择指南
    """
    """
    选择策略的考虑因素：
    
    1. 数据集大小：
    - 小数据集 (< 1000): 特征提取或浅层微调
    - 中等数据集 (1000-10000): 微调策略
    - 大数据集 (> 10000): 全量微调
    
    2. 计算资源：
    - 有限资源: 特征提取
    - 充足资源: 微调或全量微调
    
    3. 任务相似性：
    - 与预训练任务相似: 可以微调更多层
    - 与预训练任务差异大: 冻结更多层
    """
    
    selection_criteria = {
        "Small Dataset + Similar Task": "特征提取 + 微调最后几层",
        "Small Dataset + Different Task": "特征提取为主",
        "Large Dataset + Similar Task": "全量微调",
        "Large Dataset + Different Task": "分层微调"
    }
    
    print("策略选择指南:")
    for condition, strategy in selection_criteria.items():
        print(f"• {condition}: {strategy}")

strategy_selection_guide()

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader
import torchvision.transforms as transforms
import torchvision.datasets as datasets

def complete_transfer_learning_pipeline(num_classes=10, dataset_path="./data"):
    """
    完整的迁移学习管道实现
    """
    # 1. 加载预训练模型
    model = models.resnet50(pretrained=True)
    
    # 2. 冻结预训练层
    for param in model.parameters():
        param.requires_grad = False
    
    # 3. 替换分类层
    num_features = model.fc.in_features
    model.fc = nn.Linear(num_features, num_classes)
    
    # 4. 设置设备
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    model = model.to(device)
    
    # 5. 定义损失函数和优化器
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model.fc.parameters(), lr=0.001)
    
    # 6. 数据预处理
    train_transform = transforms.Compose([
        transforms.Resize(256),
        transforms.CenterCrop(224),
        transforms.RandomHorizontalFlip(p=0.5),
        transforms.ToTensor(),
        transforms.Normalize(mean=[0.485, 0.456, 0.406], 
                           std=[0.229, 0.224, 0.225])
    ])
    
    val_transform = transforms.Compose([
        transforms.Resize(256),
        transforms.CenterCrop(224),
        transforms.ToTensor(),
        transforms.Normalize(mean=[0.485, 0.456, 0.406], 
                           std=[0.229, 0.224, 0.225])
    ])
    
    print("完整迁移学习管道配置完成:")
    print(f"• 模型: ResNet50")
    print(f"• 类别数: {num_classes}")
    print(f"• 设备: {device}")
    print(f"• 学习率: 0.001")
    print(f"• 损失函数: CrossEntropyLoss")
    
    return model, criterion, optimizer, train_transform, val_transform

def train_with_transfer_learning(model, train_loader, val_loader, criterion, optimizer, num_epochs=10):
    """
    使用迁移学习进行训练
    """
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    
    train_losses = []
    train_accuracies = []
    val_losses = []
    val_accuracies = []
    
    for epoch in range(num_epochs):
        # 训练阶段
        model.train()
        running_loss = 0.0
        correct = 0
        total = 0
        
        for inputs, labels in train_loader:
            inputs, labels = inputs.to(device), labels.to(device)
            
            optimizer.zero_grad()
            outputs = model(inputs)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()
            
            running_loss += loss.item()
            _, predicted = outputs.max(1)
            total += labels.size(0)
            correct += predicted.eq(labels).sum().item()
        
        train_loss = running_loss / len(train_loader)
        train_acc = 100. * correct / total
        
        # 验证阶段
        model.eval()
        val_running_loss = 0.0
        val_correct = 0
        val_total = 0
        
        with torch.no_grad():
            for inputs, labels in val_loader:
                inputs, labels = inputs.to(device), labels.to(device)
                outputs = model(inputs)
                loss = criterion(outputs, labels)
                
                val_running_loss += loss.item()
                _, predicted = outputs.max(1)
                val_total += labels.size(0)
                val_correct += predicted.eq(labels).sum().item()
        
        val_loss = val_running_loss / len(val_loader)
        val_acc = 100. * val_correct / val_total
        
        train_losses.append(train_loss)
        train_accuracies.append(train_acc)
        val_losses.append(val_loss)
        val_accuracies.append(val_acc)
        
        print(f'Epoch [{epoch+1}/{num_epochs}]')
        print(f'Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.2f}%')
        print(f'Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.2f}%')
        print('-' * 50)
    
    return train_losses, train_accuracies, val_losses, val_accuracies

def pipeline_example():
    """
    迁移学习管道示例
    """
    print("迁移学习完整管道示例:")
    print("1. 加载预训练模型")
    print("2. 冻结预训练层")
    print("3. 替换分类层")
    print("4. 配置训练参数")
    print("5. 执行训练循环")
    print("6. 评估模型性能")

pipeline_example()

def transfer_learning_with_different_models():
    """
    使用不同预训练模型的迁移学习实现
    """
    
    def create_model_transfer(model_name, num_classes=10):
        """
        为不同模型创建迁移学习配置
        """
        if model_name == "resnet50":
            model = models.resnet50(pretrained=True)
            # ResNet的分类层名为fc
            num_features = model.fc.in_features
            model.fc = nn.Linear(num_features, num_classes)
            
        elif model_name == "vgg16":
            model = models.vgg16(pretrained=True)
            # VGG的分类层在classifier的最后
            num_features = model.classifier[6].in_features
            model.classifier[6] = nn.Linear(num_features, num_classes)
            
        elif model_name == "mobilenet":
            model = models.mobilenet_v3_small(pretrained=True)
            # MobileNet的分类层名为classifier
            num_features = model.classifier[3].in_features
            model.classifier[3] = nn.Linear(num_features, num_classes)
            
        return model
    
    # 创建不同模型的迁移学习配置
    resnet_model = create_model_transfer("resnet50", 10)
    vgg_model = create_model_transfer("vgg16", 10)
    mobile_model = create_model_transfer("mobilenet", 10)
    
    print("不同模型的迁移学习配置:")
    print(f"• ResNet50参数量: {sum(p.numel() for p in resnet_model.parameters()):,}")
    print(f"• VGG16参数量: {sum(p.numel() for p in vgg_model.parameters()):,}")
    print(f"• MobileNet参数量: {sum(p.numel() for p in mobile_model.parameters()):,}")

transfer_learning_with_different_models()

def advanced_transfer_learning_techniques():
    """
    高级迁移学习技术
    """
    """
    1. 多任务学习：
       - 同时学习多个相关任务
       - 共享底层特征
    
    2. 领域自适应：
       - 适应不同领域的数据分布
       - 减少领域间差异
    
    3. 元学习：
       - 学会如何学习
       - 快速适应新任务
    """
    
    # 多任务学习示例
    class MultiTaskModel(nn.Module):
        def __init__(self, backbone, num_classes_task1, num_classes_task2):
            super(MultiTaskModel, self).__init__()
            
            # 共享的特征提取器
            self.backbone = backbone
            
            # 任务特定的头部
            self.task1_head = nn.Linear(backbone.fc.in_features, num_classes_task1)
            self.task2_head = nn.Linear(backbone.fc.in_features, num_classes_task2)
            
            # 移除原始的分类层
            self.backbone.fc = nn.Identity()
        
        def forward(self, x):
            features = self.backbone(x)
            task1_output = self.task1_head(features)
            task2_output = self.task2_head(features)
            return task1_output, task2_output
    
    techniques = {
        "Multi-task Learning": "同时学习相关任务，共享特征",
        "Domain Adaptation": "适应不同数据分布",
        "Meta Learning": "快速适应新任务",
        "Few-shot Learning": "少量样本学习",
        "Self-supervised Learning": "无监督预训练"
    }
    
    print("高级迁移学习技术:")
    for tech, desc in techniques.items():
        print(f"• {tech}: {desc}")

advanced_transfer_learning_techniques()

from torch.optim.lr_scheduler import StepLR, CosineAnnealingLR, ReduceLROnPlateau

def learning_rate_strategies():
    """
    不同的学习率调度策略
    """
    
    # 为不同层设置不同学习率
    model = models.resnet50(pretrained=True)
    
    # 冻结前面的层
    for param in list(model.children())[:-2]:
        for p in param.parameters():
            p.requires_grad = False
    
    # 替换分类层
    num_features = model.fc.in_features
    model.fc = nn.Linear(num_features, 10)
    
    # 为不同层设置不同学习率
    optimizer = optim.Adam([
        {'params': model.layer4.parameters(), 'lr': 1e-4},  # 最后一层较低学习率
        {'params': model.fc.parameters(), 'lr': 1e-3},      # 新层较高学习率
    ])
    
    # 学习率调度器
    step_scheduler = StepLR(optimizer, step_size=7, gamma=0.1)
    cosine_scheduler = CosineAnnealingLR(optimizer, T_max=10)
    plateau_scheduler = ReduceLROnPlateau(optimizer, mode='min', patience=3, factor=0.5)
    
    print("学习率调度策略:")
    print("• 分层学习率: 不同层使用不同学习率")
    print("• StepLR: 固定间隔衰减")
    print("• CosineAnnealing: 余弦退火")
    print("• ReduceLROnPlateau: 根据性能调整")

learning_rate_strategies()

def transfer_learning_best_practices():
    """
    迁移学习最佳实践
    """
    """
    1. 数据预处理：
       - 使用与预训练模型相同的归一化参数
       - 保持输入尺寸一致
    
    2. 模型选择：
       - 根据任务复杂度选择模型
       - 考虑计算资源限制
    
    3. 训练策略：
       - 从保守策略开始
       - 逐步增加复杂度
    """
    
    best_practices = [
        "使用与预训练模型相同的输入预处理方式",
        "从小规模实验开始，逐步扩大",
        "监控训练和验证损失，防止过拟合",
        "使用早停机制避免过度训练",
        "保存最佳模型权重",
        "尝试不同的微调策略",
        "考虑使用更大的批次大小",
        "使用数据增强提高泛化能力"
    ]
    
    print("迁移学习最佳实践:")
    for i, practice in enumerate(best_practices, 1):
        print(f"{i}. {practice}")

transfer_learning_best_practices()

def performance_evaluation():
    """
    迁移学习性能评估
    """
    """
    评估指标：
    1. 准确率提升：相比随机初始化的改进
    2. 训练时间：收敛速度对比
    3. 数据效率：达到目标性能所需数据量
    4. 泛化能力：验证集与测试集性能差距
    """
    
    def calculate_improvement_metrics(original_acc, transfer_acc):
        """
        计算迁移学习的改进指标
        """
        improvement = transfer_acc - original_acc
        improvement_pct = (improvement / original_acc) * 100
        
        return {
            "absolute_improvement": improvement,
            "percentage_improvement": improvement_pct,
            "new_accuracy": transfer_acc
        }
    
    # 示例计算
    metrics = calculate_improvement_metrics(0.65, 0.82)
    
    print("性能评估指标:")
    print(f"• 绝对改进: {metrics['absolute_improvement']:.3f}")
    print(f"• 百分比改进: {metrics['percentage_improvement']:.1f}%")
    print(f"• 新准确率: {metrics['new_accuracy']:.3f}")
    
    evaluation_methods = [
        "对比实验：有/无迁移学习的性能对比",
        "消融研究：不同迁移策略的效果分析", 
        "跨数据集：在不同数据集上的泛化能力",
        "时间效率：训练时间对比分析"
    ]
    
    print("\n评估方法:")
    for method in evaluation_methods:
        print(f"• {method}")

performance_evaluation()