当前位置：首页 > news >正文

特征重要性与数据偏移的交叉分析

news 来源：原创 2025/5/10 18:03:32

特征重要性与数据偏移的交叉分析

1. 引言：模型稳定性的双重挑战

在机器学习系统的开发和部署过程中，面临两大核心挑战：确保模型性能和保障模型稳定性。模型性能通常通过选择重要特征来提升，而模型稳定性则要求特征分布在训练和预测环境中保持相对稳定。然而，这两个目标往往相互制约：最重要的特征可能恰恰是最不稳定的。

现实中，这一矛盾尤为突出：

高度信息性的市场指标往往波动剧烈
极具预测价值的用户行为特征可能随社会事件急剧变化
强信号的传感器数据在不同环境条件下表现迥异

Google研究表明，在生产环境中，约60%的机器学习性能问题源于重要特征的分布偏移。阿里巴巴的一项研究发现，在电商推荐系统中，前10%最重要特征的分布变化对模型性能的影响是其他特征的5倍以上。

本文将深入探讨特征重要性与数据偏移的交叉关系，提出系统化的分析方法，并介绍实用策略来平衡模型性能与稳定性，帮助读者构建更可靠的机器学习系统。

2. 重要特征与偏移风险

2.1 重要特征偏移的高风险影响

特征重要性与偏移风险之间存在复杂的关系。一般而言，重要特征偏移会对模型产生更大影响，这可以从以下几个方面解释：

理论视角

从统计学习理论角度，模型对重要特征偏移的敏感性可以用影响函数(Influence Function)表示：

$\Delta L(\theta) \approx \sum_{i} IF(z_i) \cdot \Delta P(z_i)$

其中 $IF(z_i)$ 是特征 $z_i$ 的影响函数， $\Delta P(z_i)$ 是其分布变化。重要特征通常具有较大的影响函数值，因此即使分布变化相同，也会导致更大的损失增量。

实用计算方法

以下代码展示了如何计算特征重要性与偏移风险的组合指标：

def calculate_combined_risk(model, reference_data, current_data, feature_names):
    """计算特征重要性和偏移风险的组合分数"""
    # 计算特征重要性
    if hasattr(model, 'feature_importances_'):
        importances = model.feature_importances_
    else:
        # 对于没有内置特征重要性的模型，使用排列重要性
        importances = calculate_permutation_importance(model, reference_data)
    
    importance_dict = dict(zip(feature_names, importances))
    
    # 计算分布偏移
    shift_scores = {}
    for i, feature in enumerate(feature_names):
        ref_values = reference_data[:, i]
        current_values = current_data[:, i]
        
        # 计算JS散度作为偏移度量
        shift_scores[feature] = calculate_js_divergence(ref_values, current_values)
    
    # 计算组合风险分数
    combined_risk = {}
    for feature in feature_names:
        # 组合得分是重要性和偏移程度的乘积
        combined_risk[feature] = importance_dict[feature] * shift_scores[feature]
    
    return {
        'feature_importance': importance_dict,
        'shift_scores': shift_scores,
        'combined_risk': combined_risk
    }

2.2 重要性-偏移矩阵分析法

为了系统地分析特征重要性与偏移风险的关系，可以构建重要性-偏移矩阵，将特征按两个维度进行分类：

	低偏移	高偏移
高重要性	理想特征	高风险特征
低重要性	候选特征	可排除特征

这个矩阵有助于制定差异化策略：

理想特征：保持原样，优先使用
高风险特征：需要特殊处理（稳定化转换、监控或替代）
候选特征：重要性可能被低估，考虑提升其权重
可排除特征：优先排除，减少维度和计算开销

以下代码实现了重要性-偏移矩阵的构建：

def create_importance_shift_matrix(feature_risks, importance_threshold=0.05, shift_threshold=0.2):
    """构建特征重要性-偏移风险矩阵"""
    matrix = {
        'ideal_features': [],     # 高重要性，低偏移
        'high_risk_features': [], # 高重要性，高偏移
        'candidate_features': [], # 低重要性，低偏移
        'excludable_features': [] # 低重要性，高偏移
    }
    
    for feature, metrics in feature_risks.items():
        importance = metrics['importance']
        shift_score = metrics['shift_score']
        
        if importance >= importance_threshold:
            if shift_score >= shift_threshold:
                matrix['high_risk_features'].append(feature)
            else:
                matrix['ideal_features'].append(feature)
        else:
            if shift_score >= shift_threshold:
                matrix['excludable_features'].append(feature)
            else:
                matrix['candidate_features'].append(feature)
    
    return matrix

2.3 重要特征稳定性评估

评估重要特征稳定性的一种有效方法是计算其在不同时间窗口或不同数据批次中的重要性波动：

def evaluate_feature_importance_stability(model_class, data_batches, feature_names):
    """评估特征重要性的稳定性"""
    importance_series = {feature: [] for feature in feature_names}
    
    # 在每个数据批次上训练模型并记录特征重要性
    for batch in data_batches:
        X_batch, y_batch = batch
        
        model = model_class()
        model.fit(X_batch, y_batch)
        
        if hasattr(model, 'feature_importances_'):
            importances = model.feature_importances_
        else:
            importances = calculate_permutation_importance(model, X_batch, y_batch)
        
        for i, feature in enumerate(feature_names):
            importance_series[feature].append(importances[i])
    
    # 计算每个特征重要性的变异系数 (CV)
    stability_metrics = {}
    for feature in feature_names:
        imp_array = np.array(importance_series[feature])
        cv = np.std(imp_array) / np.mean(imp_array) if np.mean(imp_array) > 0 else float('inf')
        stability_metrics[feature] = {
            'importance_series': importance_series[feature],
            'cv': cv,  # 变异系数，越小越稳定
            'stability_score': 1 / (1 + cv)  # 稳定性分数，0-1之间，越大越稳定
        }
    
    return stability_metrics

变异系数(CV)是衡量特征重要性稳定性的有效指标，其值越小表示特征重要性越稳定。根据不同领域的经验，一般认为CV<0.1的特征具有高稳定性，CV>0.3的特征则可能存在明显的不稳定性。

3. 特征筛选策略

3.1 考虑偏移风险的特征选择方法

传统特征选择方法主要考虑特征与目标的相关性，但忽略了特征的稳定性。以下是几种考虑偏移风险的特征选择策略：

稳定性加权特征选择

这种方法在特征重要性基础上加入稳定性权重：

def stability_weighted_feature_selection(importances, stability_scores, alpha=0.5, top_k=None):
    """结合重要性和稳定性的特征选择
    
    参数:
        importances: 特征重要性字典
        stability_scores: 特征稳定性分数字典
        alpha: 重要性和稳定性的权衡参数，0-1之间
        top_k: 选择的特征数量
    """
    weighted_scores = {}
    
    for feature in importances:
        # 组合分数 = alpha * 重要性 + (1 - alpha) * 稳定性
        if feature in stability_scores:
            weighted_scores[feature] = alpha * importances[feature] + \
                                      (1 - alpha) * stability_scores[feature]
        else:
            weighted_scores[feature] = alpha * importances[feature]
    
    # 按组合分数排序
    sorted_features = sorted(weighted_scores.items(), key=lambda x: x[1], reverse=True)
    
    if top_k is not None:
        selected_features = [f[0] for f in sorted_features[:top_k]]
    else:
        selected_features = [f[0] for f in sorted_features]
    
    return {
        'selected_features': selected_features,
        'weighted_scores': weighted_scores
    }

参数alpha反映了对重要性和稳定性的偏好。在高度动态的环境中，可以设置较小的alpha值优先选择稳定特征；而在数据分布相对稳定的场景中，则可以使用较大的alpha值优先考虑预测能力。

多目标特征选择

更先进的方法是将特征选择视为多目标优化问题，同时优化预测能力和稳定性：

def pareto_optimal_feature_selection(importances, shift_scores, min_features=5, max_features=20):
    """寻找特征重要性和分布稳定性的帕累托最优前沿"""
    features = list(importances.keys())
    # 将偏移分数转换为稳定性分数（1 - 偏移分数）
    stability_scores = {f: 1 - shift_scores[f] for f in shift_scores}
    
    # 生成所有可能的特征子集（限制大小以控制计算量）
    feature_subsets = []
    for k in range(min_features, max_features + 1):
        for subset in itertools.combinations(features, k):
            feature_subsets.append(subset)
    
    # 评估每个子集
    subset_scores = []
    for subset in feature_subsets:
        # 计算重要性得分（所选特征重要性之和）
        importance_score = sum(importances[f] for f in subset)
        # 计算稳定性得分（所选特征稳定性的平均值）
        stability_score = sum(stability_scores[f] for f in subset) / len(subset)
        
        subset_scores.append({
            'subset': subset,
            'importance_score': importance_score,
            'stability_score': stability_score,
        })
    
    # 寻找帕累托最优解
    pareto_optimal = []
    for i, score_i in enumerate(subset_scores):
        dominated = False
        for j, score_j in enumerate(subset_scores):
            if i != j:
                if (score_j['importance_score'] >= score_i['importance_score'] and 
                    score_j['stability_score'] > score_i['stability_score']) or \
                   (score_j['importance_score'] > score_i['importance_score'] and 
                    score_j['stability_score'] >= score_i['stability_score']):
                    dominated = True
                    break
        
        if not dominated:
            pareto_optimal.append(score_i)
    
    return pareto_optimal

3.2 稳定性与性能之间的权衡

实际应用中，通常需要在稳定性和性能之间做出合理权衡。以下是一种基于业务容忍度的权衡方法：

def performance_stability_tradeoff(performance_metrics, stability_metrics, business_tolerance):
    """基于业务容忍度进行性能和稳定性权衡
    
    参数:
        performance_metrics: 不同特征子集的性能指标
        stability_metrics: 不同特征子集的稳定性指标
        business_tolerance: 业务可接受的最大性能下降比例
    """
    # 获取性能最佳的特征子集
    best_performance_subset = max(performance_metrics.items(), key=lambda x: x[1]['score'])
    best_performance = best_performance_subset[1]['score']
    
    # 按稳定性排序
    sorted_by_stability = sorted(
        stability_metrics.items(), 
        key=lambda x: x[1]['stability_score'], 
        reverse=True
    )
    
    # 寻找满足性能要求的最稳定子集
    for subset_name, stability_metric in sorted_by_stability:
        if subset_name in performance_metrics:
            current_performance = performance_metrics[subset_name]['score']
            performance_drop = (best_performance - current_performance) / best_performance
            
            if performance_drop <= business_tolerance:
                return {
                    'selected_subset': subset_name,
                    'performance': current_performance,
                    'performance_drop': performance_drop,
                    'stability_score': stability_metric['stability_score']
                }
    
    # 如果没有找到满足条件的子集，返回性能最佳的
    return {
        'selected_subset': best_performance_subset[0],
        'performance': best_performance,
        'performance_drop': 0,
        'stability_score': stability_metrics[best_performance_subset[0]]['stability_score']
    }

不同业务场景下的权衡策略也不同：

实时推荐：可接受5-10%的性能下降以换取更高稳定性
风险评估：通常对稳定性要求更高，可接受10-15%的性能下降
异常检测：可平衡考虑，一般可接受8-12%的性能下降

3.3 动态特征选择机制

在高度动态的环境中，静态特征选择可能不够灵活。动态特征选择机制可以根据当前数据分布自动调整使用的特征：

class DynamicFeatureSelector:
    """动态特征选择器，根据当前数据分布调整特征集"""
    
    def __init__(self, base_model, feature_pool, reference_distributions):
        self.base_model = base_model
        self.feature_pool = feature_pool  # 所有候选特征
        self.reference_distributions = reference_distributions
        self.current_features = feature_pool.copy()  # 初始使用所有特征
        self.importance_cache = {}
        self.shift_threshold = 0.3
        
    def update_feature_set(self, current_data):
        """根据当前数据更新特征集"""
        # 计算特征偏移
        shift_scores = {}
        for feature in self.feature_pool:
            if feature in current_data.columns:
                shift_scores[feature] = calculate_distribution_shift(
                    self.reference_distributions[feature],
                    current_data[feature]
                )
        
        # 使用最新数据更新特征重要性（如果有足够数据）
        if hasattr(current_data, 'target') and len(current_data) > 1000:
            model_copy = clone(self.base_model)
            X = current_data.drop('target', axis=1)
            y = current_data['target']
            model_copy.fit(X, y)
            
            # 更新重要性缓存
            if hasattr(model_copy, 'feature_importances_'):
                for i, feature in enumerate(X.columns):
                    self.importance_cache[feature] = model_copy.feature_importances_[i]
        
        # 确定要使用的特征集
        selected_features = []
        for feature in self.feature_pool:
            importance = self.importance_cache.get(feature, 0.01)  # 默认低重要性
            
            # 如果特征重要且分布稳定，或者特征极其重要（即使不稳定）
            if (importance > 0.05 and shift_scores.get(feature, 0) < self.shift_threshold) or \
               importance > 0.2:
                selected_features.append(feature)
        
        # 如果选择的特征太少，添加一些次重要但稳定的特征
        if len(selected_features) < len(self.feature_pool) // 3:
            stable_features = [f for f in self.feature_pool 
                              if f not in selected_features and 
                              shift_scores.get(f, 0) < self.shift_threshold / 2]
            
            # 按重要性排序
            stable_features.sort(key=lambda f: self.importance_cache.get(f, 0), reverse=True)
            selected_features.extend(stable_features[:len(self.feature_pool) // 3])
        
        self.current_features = selected_features
        return selected_features
    
    def transform(self, X):
        """只保留当前选定的特征"""
        return X[self.current_features]

这种动态特征选择器能够随着数据分布的变化自适应调整特征集，既保持模型性能，又增强了系统稳定性。

4. 索引映射与特征跟踪

4.1 原始特征与筛选特征之间的映射关系

在特征工程和选择过程中，常常需要维护原始特征与筛选特征之间的映射关系，这对于特征解释、监控和调试至关重要：

def create_feature_mapping(original_feature_names, selected_indices):
    """创建原始特征与筛选后特征的映射
    
    参数:
        original_feature_names: 原始特征名列表
        selected_indices: 选择的特征索引列表
    
    返回:
        映射字典，包含正向和反向映射
    """
    mapping = {
        'original_to_selected': {},  # 原始索引 -> 选择后索引
        'selected_to_original': {},  # 选择后索引 -> 原始索引
        'original_to_name': {},      # 原始索引 -> 特征名
        'selected_to_name': {}       # 选择后索引 -> 特征名
    }
    
    for original_idx, feature_name in enumerate(original_feature_names):
        mapping['original_to_name'][original_idx] = feature_name
    
    for selected_idx, original_idx in enumerate(selected_indices):
        mapping['original_to_selected'][original_idx] = selected_idx
        mapping['selected_to_original'][selected_idx] = original_idx
        
        feature_name = mapping['original_to_name'].get(original_idx, f"unknown_{original_idx}")
        mapping['selected_to_name'][selected_idx] = feature_name
    
    return mapping

这种映射关系在模型解释、特征重要性分析和分布偏移监控中非常有用。例如，当检测到某个筛选后的特征出现偏移时，可以立即定位到对应的原始特征。

4.2 特征生命周期管理

随着模型迭代和数据演化，特征集合会不断变化，维护特征的生命周期信息有助于理解和管理这一过程：

class FeatureLifecycleTracker:
    """特征生命周期跟踪系统"""
    
    def __init__(self, feature_registry=None):
        self.feature_registry = feature_registry or {}
        self.version_history = []
        self.current_version = 0
    
    def register_feature(self, feature_name, metadata):
        """注册新特征或更新现有特征"""
        if feature_name in self.feature_registry:
            # 更新现有特征
            self.feature_registry[feature_name].update(metadata)
            self.feature_registry[feature_name]['last_updated'] = datetime.now()
            self.feature_registry[feature_name]['update_count'] += 1
        else:
            # 注册新特征
            self.feature_registry[feature_name] = {
                'created_at': datetime.now(),
                'last_updated': datetime.now(),
                'update_count': 0,
                'status': 'active',
                'versions_used': []
            }
            self.feature_registry[feature_name].update(metadata)
    
    def deprecate_feature(self, feature_name, reason):
        """标记特征为已弃用"""
        if feature_name in self.feature_registry:
            self.feature_registry[feature_name]['status'] = 'deprecated'
            self.feature_registry[feature_name]['deprecation_reason'] = reason
            self.feature_registry[feature_name]['deprecated_at'] = datetime.now()
    
    def create_version(self, selected_features, metadata=None):
        """创建新版本的特征集"""
        self.current_version += 1
        version_info = {
            'version': self.current_version,
            'created_at': datetime.now(),
            'features': selected_features,
            'metadata': metadata or {}
        }
        
        # 更新特征使用记录
        for feature in selected_features:
            if feature in self.feature_registry:
                if 'versions_used' not in self.feature_registry[feature]:
                    self.feature_registry[feature]['versions_used'] = []
                self.feature_registry[feature]['versions_used'].append(self.current_version)
        
        self.version_history.append(version_info)
        return self.current_version
    
    def get_feature_history(self, feature_name):
        """获取特征的使用历史"""
        if feature_name in self.feature_registry:
            versions_used = self.feature_registry[feature_name].get('versions_used', [])
            history = [v for v in self.version_history if v['version'] in versions_used]
            return {
                'feature_info': self.feature_registry[feature_name],
                'version_history': history
            }
        return None
    
    def compare_versions(self, version1, version2):
        """比较两个版本的特征集差异"""
        if version1 > len(self.version_history) or version2 > len(self.version_history):
            return {'error': 'Version not found'}
        
        v1 = self.version_history[version1 - 1]
        v2 = self.version_history[version2 - 1]
        
        features_added = set(v2['features']) - set(v1['features'])
        features_removed = set(v1['features']) - set(v2['features'])
        features_common = set(v1['features']).intersection(set(v2['features']))
        
        return {
            'features_added': list(features_added),
            'features_removed': list(features_removed),
            'features_common': list(features_common),
            'version1': v1,
            'version2': v2
        }

特征生命周期管理对于理解模型性能变化至关重要。例如，当模型性能下降时，可以检查是否有重要特征被移除或替换。

5. 特征类型与偏移模式

5.1 不同来源特征的偏移特性分析

不同来源的特征展现出不同的偏移模式，理解这些差异有助于更好地处理数据偏移：

用户行为特征

用户行为特征往往高度信息化，但也最容易发生偏移：

def analyze_behavioral_feature_shifts(historical_data, timeframe='month'):
    """分析用户行为特征的历史偏移模式"""
    # 按时间分组
    time_groups = historical_data.groupby(pd.Grouper(key='timestamp', freq=timeframe))
    
    behavioral_features = [f for f in historical_data.columns 
                          if f.startswith(('click_', 'view_', 'purchase_', 'engagement_'))]
    
    shift_patterns = {}
    
    # 计算每个时间窗口相对于上一窗口的分布变化
    previous_group = None
    for time_key, group in time_groups:
        if previous_group is not None:
            period_shifts = {}
            
            for feature in behavioral_features:
                if feature in group.columns and feature in previous_group.columns:
                    # 计算分布偏移
                    shift = calculate_distribution_shift(
                        previous_group[feature].values, 
                        group[feature].values
                    )
                    
                    period_shifts[feature] = shift
            
            shift_patterns[time_key] = {
                'average_shift': np.mean(list(period_shifts.values())),
                'max_shift': max(period_shifts.values()),
                'feature_shifts': period_shifts
            }
            
        previous_group = group
    
    # 识别周期性模式
    if len(shift_patterns) >= 12:  # 至少需要一年的数据
        seasonal_patterns = detect_seasonal_patterns(shift_patterns)
        return {
            'shift_patterns': shift_patterns,
            'seasonal_patterns': seasonal_patterns,
            'average_monthly_shift': np.mean([p['average_shift'] for p in shift_patterns.values()]),
            'most_volatile_features': identify_most_volatile_features(shift_patterns)
        }
    
    return {
        'shift_patterns': shift_patterns,
        'average_monthly_shift': np.mean([p['average_shift'] for p in shift_patterns.values()]),
        'most_volatile_features': identify_most_volatile_features(shift_patterns)
    }

用户行为特征通常呈现明显的时间模式，如每日、每周和季节性变化。了解这些模式有助于区分正常波动和异常偏移。

静态特征vs动态特征

静态特征（如人口统计学特征）和动态特征（如交易行为）的偏移特性差异很大：

def compare_static_dynamic_feature_stability(data, feature_types, time_periods):
    """比较静态和动态特征的稳定性"""
    static_features = feature_types['static']
    dynamic_features = feature_types['dynamic']
    
    stability_metrics = {
        'static': {f: [] for f in static_features},
        'dynamic': {f: [] for f in dynamic_features}
    }
    
    for period_start, period_end in time_periods:
        period_data = data[(data['timestamp'] >= period_start) & 
                           (data['timestamp'] < period_end)]
        
        previous_period_end = period_start - timedelta(days=(period_end - period_start).days)
        previous_period_start = previous_period_end - timedelta(days=(period_end - period_start).days)
        previous_period_data = data[(data['timestamp'] >= previous_period_start) & 
                                    (data['timestamp'] < previous_period_end)]
        
        # 比较静态特征分布
        for feature in static_features:
            if feature in period_data.columns and feature in previous_period_data.columns:
                shift = calculate_distribution_shift(
                    previous_period_data[feature].values,
                    period_data[feature].values
                )
                stability_metrics['static'][feature].append(shift)
        
        # 比较动态特征分布
        for feature in dynamic_features:
            if feature in period_data.columns and feature in previous_period_data.columns:
                shift = calculate_distribution_shift(
                    previous_period_data[feature].values,
                    period_data[feature].values
                )
                stability_metrics['dynamic'][feature].append(shift)
    
    # 计算平均稳定性指标
    avg_stability = {
        'static': {f: np.mean(shifts) for f, shifts in stability_metrics['static'].items()},
        'dynamic': {f: np.mean(shifts) for f, shifts in stability_metrics['dynamic'].items()}
    }
    
    # 计算总体比较
    overall_comparison = {
        'static_avg_shift': np.mean([np.mean(shifts) for shifts in stability_metrics['static'].values()]),
        'dynamic_avg_shift': np.mean([np.mean(shifts) for shifts in stability_metrics['dynamic'].values()]),
        'static_vs_dynamic_ratio': np.mean([np.mean(shifts) for shifts in stability_metrics['static'].values()]) / 
                                  np.mean([np.mean(shifts) for shifts in stability_metrics['dynamic'].values()])
    }
    
    return {
        'stability_metrics': stability_metrics,
        'avg_stability': avg_stability,
        'overall_comparison': overall_comparison
    }

一般而言，静态特征比动态特征更稳定，但当发生突发事件（如新冠疫情）时，静态特征分布也可能出现巨大变化。

5.2 类别型vs数值型特征的偏移表现

类别型和数值型特征在偏移表现上也存在显著差异：

def analyze_numeric_vs_categorical_shifts(reference_data, current_data, feature_types):
    """分析数值型和类别型特征的偏移差异"""
    numeric_features = feature_types['numeric']
    categorical_features = feature_types['categorical']
    
    shift_analysis = {
        'numeric': {},
        'categorical': {}
    }
    
    # 分析数值型特征
    for feature in numeric_features:
        if feature in reference_data.columns and feature in current_data.columns:
            # 基础统计量变化
            ref_stats = calculate_numeric_statistics(reference_data[feature])
            cur_stats = calculate_numeric_statistics(current_data[feature])
            
            # 分布偏移
            ks_stat, p_value = stats.ks_2samp(
                reference_data[feature].dropna(), 
                current_data[feature].dropna()
            )
            
            shift_analysis['numeric'][feature] = {
                'statistic_changes': {
                    'mean_change': (cur_stats['mean'] - ref_stats['mean']) / max(abs(ref_stats['mean']), 1e-10),
                    'std_change': (cur_stats['std'] - ref_stats['std']) / max(ref_stats['std'], 1e-10),
                    'median_change': (cur_stats['median'] - ref_stats['median']) / max(abs(ref_stats['median']), 1e-10),
                },
                'distribution_shift': {
                    'ks_statistic': ks_stat,
                    'p_value': p_value,
                    'significant': p_value < 0.05
                }
            }
    
    # 分析类别型特征
    for feature in categorical_features:
        if feature in reference_data.columns and feature in current_data.columns:
            ref_counts = reference_data[feature].value_counts(normalize=True)
            cur_counts = current_data[feature].value_counts(normalize=True)
            
            # 计算分类分布差异
            all_categories = set(ref_counts.index) | set(cur_counts.index)
            
            category_shifts = {}
            for category in all_categories:
                ref_freq = ref_counts.get(category, 0)
                cur_freq = cur_counts.get(category, 0)
                
                abs_change = cur_freq - ref_freq
                rel_change = abs_change / max(ref_freq, 1e-10) if ref_freq > 0 else float('inf')
                
                category_shifts[category] = {
                    'reference_freq': ref_freq,
                    'current_freq': cur_freq,
                    'absolute_change': abs_change,
                    'relative_change': rel_change
                }
            
            # 计算整体分布差异（JS散度）
            js_div = calculate_js_divergence_categorical(ref_counts, cur_counts)
            
            # 新出现和消失的类别
            new_categories = [c for c in cur_counts.index if c not in ref_counts.index]
            disappeared_categories = [c for c in ref_counts.index if c not in cur_counts.index]
            
            shift_analysis['categorical'][feature] = {
                'js_divergence': js_div,
                'category_shifts': category_shifts,
                'new_categories': new_categories,
                'disappeared_categories': disappeared_categories,
                'top_changed_categories': sorted(
                    category_shifts.items(), 
                    key=lambda x: abs(x[1]['relative_change']), 
                    reverse=True
                )[:5]
            }
    
    # 对比分析
    avg_numeric_shift = np.mean([v['distribution_shift']['ks_statistic'] 
                                for v in shift_analysis['numeric'].values()])
    
    avg_categorical_shift = np.mean([v['js_divergence'] 
                                    for v in shift_analysis['categorical'].values()])
    
    comparison = {
        'avg_numeric_shift': avg_numeric_shift,
        'avg_categorical_shift': avg_categorical_shift,
        'most_shifted_numeric': sorted(
            [(f, v['distribution_shift']['ks_statistic']) 
             for f, v in shift_analysis['numeric'].items()],
            key=lambda x: x[1],
            reverse=True
        )[:5],
        'most_shifted_categorical': sorted(
            [(f, v['js_divergence']) 
             for f, v in shift_analysis['categorical'].items()],
            key=lambda x: x[1],
            reverse=True
        )[:5]
    }
    
    return {
        'shift_analysis': shift_analysis,
        'comparison': comparison
    }

在实际应用中，类别型特征的偏移通常更容易被发现和解释，但数值型特征偏移可能带来更严重的性能影响，特别是当模型对分布形状敏感时。

6. 结论与展望

6.1 主要发现总结

通过本文的探讨，得出以下关键发现：

重要性-偏移关系：重要特征往往更容易偏移，这一矛盾是机器学习系统稳定性的核心挑战。
差异化策略：不同象限的特征需要差异化处理策略，例如：
- 高重要性高稳定性：保持并优先使用
- 高重要性低稳定性：应用稳定化转换或替代
- 低重要性高稳定性：可作为稳定后备特征
- 低重要性低稳定性：考虑排除
特征类型影响：特征来源和类型显著影响其偏移模式，了解这些模式有助于预测和应对偏移。
动态平衡：成功的机器学习系统需要在特征信息量和稳定性之间达成动态平衡。