从 ML 到 DL：什么时候该升级

import numpy as np
from dataclasses import dataclass

@dataclass
class UpgradeEvaluation:
    """升级时机评估类"""
    data_size: int  # 样本数量
    feature_dim: int  # 特征维度
    task_complexity: int  # 任务复杂度（1-5）
    current_performance: float  # 当前ML模型性能
    required_performance: float  # 业务要求的性能
    has_gpu: bool  # 是否有GPU资源
    has_dl_expert: bool  # 是否有DL专家
    development_cycle: int  # 允许的开发周期（月）
    maintenance_budget: float  # 维护预算（万元/年）
    
    def __post_init__(self):
        # 初始化评估维度权重
        self.weights = {
            'data_size': 0.2,
            'feature_complexity': 0.2,
            'task_complexity': 0.2,
            'performance_gap': 0.2,
            'resource_availability': 0.1,
            'expertise_availability': 0.1
        }
    
    def evaluate_data_size(self):
        """评估数据规模"""
        if self.data_size < 50000:
            return 0.0
        elif self.data_size < 200000:
            return 0.5
        else:
            return 1.0
    
    def evaluate_feature_complexity(self):
        """评估特征复杂度"""
        if self.feature_dim < 500:
            return 0.0
        elif self.feature_dim < 2000:
            return 0.5
        else:
            return 1.0
    
    def evaluate_task_complexity(self):
        """评估任务复杂度"""
        return min(1.0, self.task_complexity / 5.0)
    
    def evaluate_performance_gap(self):
        """评估性能差距"""
        if self.current_performance >= self.required_performance:
            return 0.0
        gap = (self.required_performance - self.current_performance) / self.required_performance
        return min(1.0, gap)
    
    def evaluate_resource_availability(self):
        """评估资源可用性"""
        return 1.0 if self.has_gpu else 0.5
    
    def evaluate_expertise_availability(self):
        """评估专业知识可用性"""
        return 1.0 if self.has_dl_expert else 0.3
    
    def calculate_upgrade_score(self):
        """计算升级分数"""
        scores = {
            'data_size': self.evaluate_data_size(),
            'feature_complexity': self.evaluate_feature_complexity(),
            'task_complexity': self.evaluate_task_complexity(),
            'performance_gap': self.evaluate_performance_gap(),
            'resource_availability': self.evaluate_resource_availability(),
            'expertise_availability': self.evaluate_expertise_availability()
        }
        
        total_score = sum(scores[key] * self.weights[key] for key in scores)
        return total_score
    
    def should_upgrade(self):
        """判断是否应该升级"""
        score = self.calculate_upgrade_score()
        return score >= 0.6

# 使用示例
if __name__ == '__main__':
    # 案例1：小规模数据，简单任务
    evaluation1 = UpgradeEvaluation(
        data_size=10000,
        feature_dim=100,
        task_complexity=2,
        current_performance=0.85,
        required_performance=0.9,
        has_gpu=False,
        has_dl_expert=False,
        development_cycle=1,
        maintenance_budget=5
    )
    
    print("案例1评估结果：")
    print(f"升级分数：{evaluation1.calculate_upgrade_score():.4f}")
    print(f"是否应该升级：{evaluation1.should_upgrade()}")
    
    # 案例2：大规模数据，复杂任务
    evaluation2 = UpgradeEvaluation(
        data_size=500000,
        feature_dim=5000,
        task_complexity=5,
        current_performance=0.8,
        required_performance=0.95,
        has_gpu=True,
        has_dl_expert=True,
        development_cycle=6,
        maintenance_budget=20
    )
    
    print("\n案例2评估结果：")
    print(f"升级分数：{evaluation2.calculate_upgrade_score():.4f}")
    print(f"是否应该升级：{evaluation2.should_upgrade()}")

import torch
from transformers import BertTokenizer, BertForSequenceClassification, AdamW
from torch.utils.data import Dataset, DataLoader
import pandas as pd

# 自定义数据集类
class TextClassificationDataset(Dataset):
    def __init__(self, texts, labels, tokenizer, max_length=128):
        self.texts = texts
        self.labels = labels
        self.tokenizer = tokenizer
        self.max_length = max_length
    
    def __len__(self):
        return len(self.texts)
    
    def __getitem__(self, idx):
        text = self.texts[idx]
        label = self.labels[idx]
        
        # 编码文本
        encoding = self.tokenizer.encode_plus(
            text,
            add_special_tokens=True,
            max_length=self.max_length,
            return_token_type_ids=False,
            padding='max_length',
            truncation=True,
            return_attention_mask=True,
            return_tensors='pt'
        )
        
        return {
            'input_ids': encoding['input_ids'].flatten(),
            'attention_mask': encoding['attention_mask'].flatten(),
            'labels': torch.tensor(label, dtype=torch.long)
        }

# 迁移学习微调函数
def fine_tune_bert(train_data, val_data, num_labels, epochs=3, batch_size=16, learning_rate=2e-5):
    """微调预训练BERT模型"""
    # 加载预训练模型和分词器
    tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
    model = BertForSequenceClassification.from_pretrained('bert-base-uncased', num_labels=num_labels)
    
    # 创建数据集
    train_dataset = TextClassificationDataset(
        train_data['text'].tolist(),
        train_data['label'].tolist(),
        tokenizer
    )
    
    val_dataset = TextClassificationDataset(
        val_data['text'].tolist(),
        val_data['label'].tolist(),
        tokenizer
    )
    
    # 创建数据加载器
    train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
    val_loader = DataLoader(val_dataset, batch_size=batch_size)
    
    # 设置优化器
    optimizer = AdamW(model.parameters(), lr=learning_rate)
    
    # 设备设置
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model.to(device)
    
    # 训练循环
    for epoch in range(epochs):
        model.train()
        train_loss = 0
        
        for batch in train_loader:
            # 将数据移到设备上
            input_ids = batch['input_ids'].to(device)
            attention_mask = batch['attention_mask'].to(device)
            labels = batch['labels'].to(device)
            
            # 前向传播
            outputs = model(
                input_ids=input_ids,
                attention_mask=attention_mask,
                labels=labels
            )
            
            loss = outputs.loss
            train_loss += loss.item()
            
            # 反向传播和优化
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
        
        # 计算平均训练损失
        avg_train_loss = train_loss / len(train_loader)
        
        # 验证模型
        model.eval()
        val_loss = 0
        correct_predictions = 0
        
        with torch.no_grad():
            for batch in val_loader:
                # 将数据移到设备上
                input_ids = batch['input_ids'].to(device)
                attention_mask = batch['attention_mask'].to(device)
                labels = batch['labels'].to(device)
                
                # 前向传播
                outputs = model(
                    input_ids=input_ids,
                    attention_mask=attention_mask,
                    labels=labels
                )
                
                loss = outputs.loss
                val_loss += loss.item()
                
                # 计算准确率
                logits = outputs.logits
                predictions = torch.argmax(logits, dim=1)
                correct_predictions += torch.sum(predictions == labels)
        
        # 计算平均验证损失和准确率
        avg_val_loss = val_loss / len(val_loader)
        val_accuracy = correct_predictions.double() / len(val_dataset)
        
        print(f"Epoch {epoch+1}/{epochs}:")
        print(f"Train Loss: {avg_train_loss:.4f}")
        print(f"Val Loss: {avg_val_loss:.4f}")
        print(f"Val Accuracy: {val_accuracy:.4f}")
        print("=" * 50)
    
    return model, tokenizer

# 使用示例
if __name__ == '__main__':
    # 创建模拟数据
    train_data = pd.DataFrame({
        'text': [f'This is a positive example {i}' for i in range(1000)] + [f'This is a negative example {i}' for i in range(1000)],
        'label': [1] * 1000 + [0] * 1000
    })
    
    val_data = pd.DataFrame({
        'text': [f'This is a positive validation {i}' for i in range(200)] + [f'This is a negative validation {i}' for i in range(200)],
        'label': [1] * 200 + [0] * 200
    })
    
    # 微调BERT模型
    model, tokenizer = fine_tune_bert(train_data, val_data, num_labels=2, epochs=2, batch_size=8)
    
    # 保存微调后的模型
    model.save_pretrained('./fine_tuned_bert')
    tokenizer.save_pretrained('./fine_tuned_bert')
    print("微调后的模型已保存到 ./fine_tuned_bert")

import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import accuracy_score, f1_score
import tensorflow as tf
from tensorflow.keras.layers import Input, Dense, Dropout
from tensorflow.keras.models import Model
from tensorflow.keras.optimizers import Adam

class HybridModel:
    def __init__(self, ml_model, dl_model, ml_features, dl_features):
        self.ml_model = ml_model
        self.dl_model = dl_model
        self.ml_features = ml_features
        self.dl_features = dl_features
        self.scaler = StandardScaler()
    
    def preprocess_ml_features(self, X):
        """预处理机器学习特征"""
        ml_X = X[self.ml_features]
        return self.scaler.transform(ml_X)
    
    def preprocess_dl_features(self, X):
        """预处理深度学习特征"""
        return X[self.dl_features].values
    
    def fit(self, X, y, epochs=10, batch_size=32, validation_split=0.2):
        """训练混合模型"""
        # 预处理ML特征
        ml_X = self.preprocess_ml_features(X)
        
        # 预处理DL特征
        dl_X = self.preprocess_dl_features(X)
        
        # 训练ML模型
        self.ml_model.fit(ml_X, y)
        
        # 训练DL模型
        self.dl_model.fit(dl_X, y, epochs=epochs, batch_size=batch_size, validation_split=validation_split)
    
    def predict(self, X):
        """预测"""
        # 预处理ML特征
        ml_X = self.preprocess_ml_features(X)
        
        # 预处理DL特征
        dl_X = self.preprocess_dl_features(X)
        
        # ML模型预测
        ml_pred = self.ml_model.predict_proba(ml_X)[:, 1]
        
        # DL模型预测
        dl_pred = self.dl_model.predict(dl_X).flatten()
        
        # 融合预测结果（简单平均，可根据实际情况调整融合策略）
        hybrid_pred = (ml_pred + dl_pred) / 2
        
        # 转换为类别标签
        return np.where(hybrid_pred > 0.5, 1, 0)
    
    def evaluate(self, X, y):
        """评估模型"""
        predictions = self.predict(X)
        accuracy = accuracy_score(y, predictions)
        f1 = f1_score(y, predictions)
        return {'accuracy': accuracy, 'f1': f1}

# 创建深度学习模型
def create_dl_model(input_dim, hidden_dims=[128, 64], dropout_rate=0.3):
    """创建深度学习模型"""
    inputs = Input(shape=(input_dim,))
    x = Dense(hidden_dims[0], activation='relu')(inputs)
    x = Dropout(dropout_rate)(x)
    
    for dim in hidden_dims[1:]:
        x = Dense(dim, activation='relu')(x)
        x = Dropout(dropout_rate)(x)
    
    outputs = Dense(1, activation='sigmoid')(x)
    
    model = Model(inputs=inputs, outputs=outputs)
    model.compile(optimizer=Adam(learning_rate=1e-3), loss='binary_crossentropy', metrics=['accuracy'])
    
    return model

# 使用示例
if __name__ == '__main__':
    # 生成模拟数据
    np.random.seed(42)
    n_samples = 2000
    
    # 结构化特征（适合传统ML）
    structured_features = ['feature1', 'feature2', 'feature3', 'feature4', 'feature5']
    
    # 高维特征（适合DL）
    high_dim_features = [f'high_dim_feature{i}' for i in range(1, 101)]
    
    # 创建数据
    structured_data = np.random.randn(n_samples, len(structured_features))
    high_dim_data = np.random.randn(n_samples, len(high_dim_features))
    
    # 生成标签
    weights_structured = np.random.randn(len(structured_features))
    weights_high_dim = np.random.randn(len(high_dim_features))
    
    structured_scores = structured_data @ weights_structured
    high_dim_scores = high_dim_data @ weights_high_dim
    
    scores = structured_scores + high_dim_scores + np.random.randn(n_samples)
    y = np.where(scores > 0, 1, 0)
    
    # 创建DataFrame
    structured_df = pd.DataFrame(structured_data, columns=structured_features)
    high_dim_df = pd.DataFrame(high_dim_data, columns=high_dim_features)
    df = pd.concat([structured_df, high_dim_df], axis=1)
    df['label'] = y
    
    # 划分训练集和测试集
    train_size = int(0.8 * n_samples)
    train_df = df.iloc[:train_size]
    test_df = df.iloc[train_size:]
    
    # 创建ML模型
    rf_model = RandomForestClassifier(n_estimators=100, random_state=42)
    
    # 创建DL模型
    dl_model = create_dl_model(input_dim=len(high_dim_features), hidden_dims=[128, 64], dropout_rate=0.3)
    
    # 创建混合模型
    hybrid_model = HybridModel(
        ml_model=rf_model,
        dl_model=dl_model,
        ml_features=structured_features,
        dl_features=high_dim_features
    )
    
    # 训练混合模型
    print("训练混合模型...")
    hybrid_model.fit(train_df.drop('label', axis=1), train_df['label'], epochs=5, batch_size=32)
    
    # 评估混合模型
    test_results = hybrid_model.evaluate(test_df.drop('label', axis=1), test_df['label'])
    print(f"混合模型测试结果: {test_results}")
    
    # 单独评估ML模型
    ml_test_X = hybrid_model.preprocess_ml_features(test_df.drop('label', axis=1))
    ml_predictions = rf_model.predict(ml_test_X)
    ml_accuracy = accuracy_score(test_df['label'], ml_predictions)
    ml_f1 = f1_score(test_df['label'], ml_predictions)
    print(f"ML模型测试结果: accuracy={ml_accuracy:.4f}, f1={ml_f1:.4f}")
    
    # 单独评估DL模型
    dl_test_X = hybrid_model.preprocess_dl_features(test_df.drop('label', axis=1))
    dl_predictions = (dl_model.predict(dl_test_X).flatten() > 0.5).astype(int)
    dl_accuracy = accuracy_score(test_df['label'], dl_predictions)
    dl_f1 = f1_score(test_df['label'], dl_predictions)
    print(f"DL模型测试结果: accuracy={dl_accuracy:.4f}, f1={dl_f1:.4f}")

import numpy as np
from dataclasses import dataclass

@dataclass
class CostBenefitAnalysis:
    # 成本参数
    hardware_cost: float  # 硬件成本（万元）
    cloud_cost_per_month: float  # 云服务成本（万元/月）
    data_annotation_cost: float  # 数据标注成本（万元）
    development_hours: float  # 开发工时（小时）
    developer_hourly_rate: float  # 开发人员小时费率（元/小时）
    maintenance_hours_per_month: float  # 每月维护工时（小时）
    
    # 收益参数
    performance_improvement: float  # 性能提升（如准确率提升百分比）
    revenue_per_percent_improvement: float  # 每提升1%性能带来的收入增长（万元/月）
    efficiency_improvement: float  # 效率提升（如处理速度提升百分比）
    cost_reduction_per_percent_efficiency: float  # 每提升1%效率带来的成本降低（万元/月）
    new_business_opportunities: float  # 新业务机会带来的收入增长（万元/月）
    
    # 时间参数
    evaluation_period: int = 24  # 评估周期（月）
    
    def calculate_development_cost(self):
        """计算开发成本"""
        return (self.development_hours * self.developer_hourly_rate) / 10000
    
    def calculate_maintenance_cost(self):
        """计算维护成本"""
        return (self.maintenance_hours_per_month * self.developer_hourly_rate * self.evaluation_period) / 10000
    
    def calculate_total_cost(self):
        """计算总成本"""
        # 硬件成本
        hardware = self.hardware_cost
        
        # 云服务成本
        cloud = self.cloud_cost_per_month * self.evaluation_period
        
        # 数据标注成本
        data_annotation = self.data_annotation_cost
        
        # 开发成本
        development = self.calculate_development_cost()
        
        # 维护成本
        maintenance = self.calculate_maintenance_cost()
        
        total_cost = hardware + cloud + data_annotation + development + maintenance
        return total_cost
    
    def calculate_performance_benefit(self):
        """计算性能提升带来的收益"""
        return self.performance_improvement * self.revenue_per_percent_improvement * self.evaluation_period
    
    def calculate_efficiency_benefit(self):
        """计算效率提升带来的收益"""
        return self.efficiency_improvement * self.cost_reduction_per_percent_efficiency * self.evaluation_period
    
    def calculate_new_business_benefit(self):
        """计算新业务机会带来的收益"""
        return self.new_business_opportunities * self.evaluation_period
    
    def calculate_total_benefit(self):
        """计算总收益"""
        performance_benefit = self.calculate_performance_benefit()
        efficiency_benefit = self.calculate_efficiency_benefit()
        new_business_benefit = self.calculate_new_business_benefit()
        
        total_benefit = performance_benefit + efficiency_benefit + new_business_benefit
        return total_benefit
    
    def calculate_roi(self):
        """计算投资回报率"""
        total_cost = self.calculate_total_cost()
        total_benefit = self.calculate_total_benefit()
        
        if total_cost == 0:
            return float('inf')
        
        roi = (total_benefit - total_cost) / total_cost * 100
        return roi
    
    def calculate_payback_period(self):
        """计算投资回收期（月）"""
        total_cost = self.calculate_total_cost()
        monthly_benefit = (self.calculate_performance_benefit() + 
                          self.calculate_efficiency_benefit() + 
                          self.calculate_new_business_benefit()) / self.evaluation_period
        
        if monthly_benefit <= 0:
            return float('inf')
        
        payback_period = total_cost / monthly_benefit
        return payback_period
    
    def generate_report(self):
        """生成成本收益分析报告"""
        total_cost = self.calculate_total_cost()
        total_benefit = self.calculate_total_benefit()
        roi = self.calculate_roi()
        payback_period = self.calculate_payback_period()
        
        report = {
            '评估周期': f'{self.evaluation_period}个月',
            '总成本': f'{total_cost:.2f}万元',
            '总收益': f'{total_benefit:.2f}万元',
            '投资回报率(ROI)': f'{roi:.2f}%',
            '投资回收期': f'{payback_period:.2f}个月',
            '成本明细': {
                '硬件成本': f'{self.hardware_cost:.2f}万元',
                '云服务成本': f'{self.cloud_cost_per_month * self.evaluation_period:.2f}万元',
                '数据标注成本': f'{self.data_annotation_cost:.2f}万元',
                '开发成本': f'{self.calculate_development_cost():.2f}万元',
                '维护成本': f'{self.calculate_maintenance_cost():.2f}万元'
            },
            '收益明细': {
                '性能提升收益': f'{self.calculate_performance_benefit():.2f}万元',
                '效率提升收益': f'{self.calculate_efficiency_benefit():.2f}万元',
                '新业务机会收益': f'{self.calculate_new_business_benefit():.2f}万元'
            }
        }
        
        return report

# 使用示例
if __name__ == '__main__':
    # 案例1：图像识别系统升级
    cba1 = CostBenefitAnalysis(
        hardware_cost=50,  # 购买GPU服务器等硬件成本
        cloud_cost_per_month=10,  # 云服务成本
        data_annotation_cost=30,  # 数据标注成本
        development_hours=2000,  # 开发工时
        developer_hourly_rate=500,  # 开发人员小时费率
        maintenance_hours_per_month=50,  # 每月维护工时
        
        performance_improvement=15,  # 性能提升15%
        revenue_per_percent_improvement=2,  # 每提升1%性能带来2万元/月收入增长
        efficiency_improvement=30,  # 效率提升30%
        cost_reduction_per_percent_efficiency=0.5,  # 每提升1%效率带来0.5万元/月成本降低
        new_business_opportunities=5,  # 新业务机会带来5万元/月收入增长
        
        evaluation_period=24  # 评估24个月
    )
    
    report1 = cba1.generate_report()
    print("案例1：图像识别系统升级成本收益分析")
    print("=" * 60)
    for key, value in report1.items():
        if isinstance(value, dict):
            print(f"{key}:")
            for sub_key, sub_value in value.items():
                print(f"  {sub_key}: {sub_value}")
        else:
            print(f"{key}: {value}")
    
    print("\n" + "=" * 60)
    
    # 案例2：文本分类系统升级
    cba2 = CostBenefitAnalysis(
        hardware_cost=0,  # 使用云服务，无需购买硬件
        cloud_cost_per_month=5,  # 云服务成本
        data_annotation_cost=10,  # 数据标注成本
        development_hours=1000,  # 开发工时
        developer_hourly_rate=500,  # 开发人员小时费率
        maintenance_hours_per_month=30,  # 每月维护工时
        
        performance_improvement=10,  # 性能提升10%
        revenue_per_percent_improvement=1.5,  # 每提升1%性能带来1.5万元/月收入增长
        efficiency_improvement=20,  # 效率提升20%
        cost_reduction_per_percent_efficiency=0.3,  # 每提升1%效率带来0.3万元/月成本降低
        new_business_opportunities=3,  # 新业务机会带来3万元/月收入增长
        
        evaluation_period=24  # 评估24个月
    )
    
    report2 = cba2.generate_report()
    print("案例2：文本分类系统升级成本收益分析")
    print("=" * 60)
    for key, value in report2.items():
        if isinstance(value, dict):
            print(f"{key}:")
            for sub_key, sub_value in value.items():
                print(f"  {sub_key}: {sub_value}")
        else:
            print(f"{key}: {value}")