import numpy as np from typing import List, Dict, Tuple, Any, Optional from dataclasses import dataclass, field import traceback from scipy import stats @dataclass class CalibrationResult: """校準結果結構""" original_scores: List[float] calibrated_scores: List[float] score_mapping: Dict[str, float] # breed -> calibrated_score calibration_method: str distribution_stats: Dict[str, float] quality_metrics: Dict[str, float] = field(default_factory=dict) @dataclass class ScoreDistribution: """分數分布統計""" mean: float std: float min_score: float max_score: float percentile_5: float percentile_95: float compression_ratio: float # 分數壓縮比率 effective_range: float # 有效分數範圍 class ScoreCalibrator: """ 動態分數校準系統 解決分數壓縮問題並保持相對排名 """ def __init__(self): """初始化校準器""" self.calibration_methods = { 'dynamic_range_mapping': self._dynamic_range_mapping, 'percentile_stretching': self._percentile_stretching, 'gaussian_normalization': self._gaussian_normalization, 'sigmoid_transformation': self._sigmoid_transformation } self.quality_thresholds = { 'min_effective_range': 0.3, # 最小有效分數範圍 'max_compression_ratio': 0.2, # 最大允許壓縮比率 'target_distribution_range': (0.45, 0.95) # 目標分布範圍 } def calibrate_scores(self, breed_scores: List[Tuple[str, float]], method: str = 'auto') -> CalibrationResult: """ 校準品種分數 Args: breed_scores: (breed_name, score) 元組列表 method: 校準方法 ('auto', 'dynamic_range_mapping', 'percentile_stretching', etc.) Returns: CalibrationResult: 校準結果 """ try: if not breed_scores: return CalibrationResult( original_scores=[], calibrated_scores=[], score_mapping={}, calibration_method='none', distribution_stats={} ) # 提取分數和品種名稱 breeds = [item[0] for item in breed_scores] original_scores = [item[1] for item in breed_scores] # 分析原始分數分布 distribution = self._analyze_score_distribution(original_scores) # 選擇校準方法 if method == 'auto': method = self._select_calibration_method(distribution) # 應用校準 calibration_func = self.calibration_methods.get(method, self._dynamic_range_mapping) calibrated_scores = calibration_func(original_scores, distribution) # 保持排名一致性 calibrated_scores = self._preserve_ranking(original_scores, calibrated_scores) # 建立分數映射 score_mapping = dict(zip(breeds, calibrated_scores)) # 計算品質指標 quality_metrics = self._calculate_quality_metrics( original_scores, calibrated_scores, distribution ) return CalibrationResult( original_scores=original_scores, calibrated_scores=calibrated_scores, score_mapping=score_mapping, calibration_method=method, distribution_stats=self._distribution_to_dict(distribution), quality_metrics=quality_metrics ) except Exception as e: print(f"Error calibrating scores: {str(e)}") print(traceback.format_exc()) # 回傳原始分數作為降級方案 breeds = [item[0] for item in breed_scores] original_scores = [item[1] for item in breed_scores] return CalibrationResult( original_scores=original_scores, calibrated_scores=original_scores, score_mapping=dict(zip(breeds, original_scores)), calibration_method='fallback', distribution_stats={} ) def _analyze_score_distribution(self, scores: List[float]) -> ScoreDistribution: """分析分數分布""" try: scores_array = np.array(scores) # 基本統計 mean_score = np.mean(scores_array) std_score = np.std(scores_array) min_score = np.min(scores_array) max_score = np.max(scores_array) # 百分位數 percentile_5 = np.percentile(scores_array, 5) percentile_95 = np.percentile(scores_array, 95) # 壓縮比率和有效範圍 full_range = max_score - min_score effective_range = percentile_95 - percentile_5 compression_ratio = 1.0 - (effective_range / 1.0) if full_range > 0 else 0.0 return ScoreDistribution( mean=mean_score, std=std_score, min_score=min_score, max_score=max_score, percentile_5=percentile_5, percentile_95=percentile_95, compression_ratio=compression_ratio, effective_range=effective_range ) except Exception as e: print(f"Error analyzing score distribution: {str(e)}") # 返回預設分布 return ScoreDistribution( mean=0.5, std=0.1, min_score=0.0, max_score=1.0, percentile_5=0.4, percentile_95=0.6, compression_ratio=0.6, effective_range=0.2 ) def _select_calibration_method(self, distribution: ScoreDistribution) -> str: """根據分布特性選擇校準方法""" # 高度壓縮的分數需要強力展開 if distribution.compression_ratio > 0.8: return 'percentile_stretching' # 中等壓縮使用動態範圍映射 elif distribution.compression_ratio > 0.5: return 'dynamic_range_mapping' # 分數集中在中間使用 sigmoid 轉換 elif 0.4 <= distribution.mean <= 0.6 and distribution.std < 0.1: return 'sigmoid_transformation' # 其他情況使用高斯正規化 else: return 'gaussian_normalization' def _dynamic_range_mapping(self, scores: List[float], distribution: ScoreDistribution) -> List[float]: """動態範圍映射校準""" try: scores_array = np.array(scores) # 使用5%和95%百分位數作為邊界 lower_bound = distribution.percentile_5 upper_bound = distribution.percentile_95 # 避免除零 if upper_bound - lower_bound < 0.001: upper_bound = distribution.max_score lower_bound = distribution.min_score if upper_bound - lower_bound < 0.001: return scores # 所有分數相同，無需校準 # 映射到目標範圍 [0.45, 0.95] target_min, target_max = self.quality_thresholds['target_distribution_range'] # 線性映射 normalized = (scores_array - lower_bound) / (upper_bound - lower_bound) normalized = np.clip(normalized, 0, 1) # 限制在 [0,1] 範圍 calibrated = target_min + normalized * (target_max - target_min) return calibrated.tolist() except Exception as e: print(f"Error in dynamic range mapping: {str(e)}") return scores def _percentile_stretching(self, scores: List[float], distribution: ScoreDistribution) -> List[float]: """百分位數拉伸校準""" try: scores_array = np.array(scores) # 計算百分位數排名 percentile_ranks = stats.rankdata(scores_array, method='average') / len(scores_array) # 使用平方根轉換來增強差異 stretched_ranks = np.sqrt(percentile_ranks) # 映射到目標範圍 target_min, target_max = self.quality_thresholds['target_distribution_range'] calibrated = target_min + stretched_ranks * (target_max - target_min) return calibrated.tolist() except Exception as e: print(f"Error in percentile stretching: {str(e)}") return self._dynamic_range_mapping(scores, distribution) def _gaussian_normalization(self, scores: List[float], distribution: ScoreDistribution) -> List[float]: """高斯正規化校準""" try: scores_array = np.array(scores) # Z-score 正規化 if distribution.std > 0: z_scores = (scores_array - distribution.mean) / distribution.std # 限制 Z-scores 在合理範圍內 z_scores = np.clip(z_scores, -3, 3) else: z_scores = np.zeros_like(scores_array) # 轉換到目標範圍 target_min, target_max = self.quality_thresholds['target_distribution_range'] target_mean = (target_min + target_max) / 2 target_std = (target_max - target_min) / 6 # 3-sigma 範圍 calibrated = target_mean + z_scores * target_std calibrated = np.clip(calibrated, target_min, target_max) return calibrated.tolist() except Exception as e: print(f"Error in gaussian normalization: {str(e)}") return self._dynamic_range_mapping(scores, distribution) def _sigmoid_transformation(self, scores: List[float], distribution: ScoreDistribution) -> List[float]: """Sigmoid 轉換校準""" try: scores_array = np.array(scores) # 中心化分數 centered = scores_array - distribution.mean # Sigmoid 轉換 (增強中等分數的差異) sigmoid_factor = 10.0 # 控制 sigmoid 的陡峭程度 transformed = 1 / (1 + np.exp(-sigmoid_factor * centered)) # 映射到目標範圍 target_min, target_max = self.quality_thresholds['target_distribution_range'] calibrated = target_min + transformed * (target_max - target_min) return calibrated.tolist() except Exception as e: print(f"Error in sigmoid transformation: {str(e)}") return self._dynamic_range_mapping(scores, distribution) def _preserve_ranking(self, original_scores: List[float], calibrated_scores: List[float]) -> List[float]: """確保校準後的分數保持原始排名""" try: # 獲取原始排名 original_ranks = stats.rankdata([-score for score in original_scores], method='ordinal') # 獲取校準後的排名 calibrated_with_ranks = list(zip(calibrated_scores, original_ranks)) # 按原始排名排序校準後的分數 calibrated_with_ranks.sort(key=lambda x: x[1]) # 重新分配分數以保持排名但使用校準後的分布 sorted_calibrated = sorted(calibrated_scores, reverse=True) # 建立新的分數列表 preserved_scores = [0.0] * len(original_scores) for i, (_, original_rank) in enumerate(calibrated_with_ranks): # 找到原始位置 original_index = original_ranks.tolist().index(original_rank) preserved_scores[original_index] = sorted_calibrated[i] return preserved_scores except Exception as e: print(f"Error preserving ranking: {str(e)}") return calibrated_scores def _calculate_quality_metrics(self, original_scores: List[float], calibrated_scores: List[float], distribution: ScoreDistribution) -> Dict[str, float]: """計算校準品質指標""" try: original_array = np.array(original_scores) calibrated_array = np.array(calibrated_scores) # 範圍改善 original_range = np.max(original_array) - np.min(original_array) calibrated_range = np.max(calibrated_array) - np.min(calibrated_array) range_improvement = calibrated_range / max(0.001, original_range) # 分離度改善 (相鄰分數間的平均差異) original_sorted = np.sort(original_array) calibrated_sorted = np.sort(calibrated_array) original_separation = np.mean(np.diff(original_sorted)) if len(original_sorted) > 1 else 0 calibrated_separation = np.mean(np.diff(calibrated_sorted)) if len(calibrated_sorted) > 1 else 0 separation_improvement = (calibrated_separation / max(0.001, original_separation) if original_separation > 0 else 1.0) # 排名保持度 (Spearman 相關係數) if len(original_scores) > 1: rank_correlation, _ = stats.spearmanr(original_scores, calibrated_scores) rank_correlation = abs(rank_correlation) if not np.isnan(rank_correlation) else 1.0 else: rank_correlation = 1.0 # 分布品質 calibrated_std = np.std(calibrated_array) distribution_quality = min(1.0, calibrated_std * 2) # 標準差越大品質越好（在合理範圍內） return { 'range_improvement': range_improvement, 'separation_improvement': separation_improvement, 'rank_preservation': rank_correlation, 'distribution_quality': distribution_quality, 'effective_range_achieved': calibrated_range, 'compression_reduction': max(0, distribution.compression_ratio - (1.0 - calibrated_range)) } except Exception as e: print(f"Error calculating quality metrics: {str(e)}") return {'error': str(e)} def _distribution_to_dict(self, distribution: ScoreDistribution) -> Dict[str, float]: """將分布統計轉換為字典""" return { 'mean': distribution.mean, 'std': distribution.std, 'min_score': distribution.min_score, 'max_score': distribution.max_score, 'percentile_5': distribution.percentile_5, 'percentile_95': distribution.percentile_95, 'compression_ratio': distribution.compression_ratio, 'effective_range': distribution.effective_range } def apply_tie_breaking(self, breed_scores: List[Tuple[str, float]]) -> List[Tuple[str, float]]: """應用確定性的打破平手機制""" try: # 按分數分組 score_groups = {} for breed, score in breed_scores: rounded_score = round(score, 6) # 避免浮點數精度問題 if rounded_score not in score_groups: score_groups[rounded_score] = [] score_groups[rounded_score].append((breed, score)) # 處理每個分數組 result = [] for rounded_score in sorted(score_groups.keys(), reverse=True): group = score_groups[rounded_score] if len(group) == 1: result.extend(group) else: # 按品種名稱字母順序打破平手 sorted_group = sorted(group, key=lambda x: x[0]) # 為平手的品種分配微小的分數差異 for i, (breed, original_score) in enumerate(sorted_group): adjusted_score = original_score - (i * 0.0001) result.append((breed, adjusted_score)) return result except Exception as e: print(f"Error in tie breaking: {str(e)}") return breed_scores def get_calibration_summary(self, result: CalibrationResult) -> Dict[str, Any]: """獲取校準摘要資訊""" try: summary = { 'method_used': result.calibration_method, 'breeds_processed': len(result.original_scores), 'score_range_before': { 'min': min(result.original_scores) if result.original_scores else 0, 'max': max(result.original_scores) if result.original_scores else 0, 'range': (max(result.original_scores) - min(result.original_scores)) if result.original_scores else 0 }, 'score_range_after': { 'min': min(result.calibrated_scores) if result.calibrated_scores else 0, 'max': max(result.calibrated_scores) if result.calibrated_scores else 0, 'range': (max(result.calibrated_scores) - min(result.calibrated_scores)) if result.calibrated_scores else 0 }, 'distribution_stats': result.distribution_stats, 'quality_metrics': result.quality_metrics, 'improvement_summary': { 'range_expanded': result.quality_metrics.get('range_improvement', 1.0) > 1.1, 'separation_improved': result.quality_metrics.get('separation_improvement', 1.0) > 1.1, 'ranking_preserved': result.quality_metrics.get('rank_preservation', 1.0) > 0.95 } } return summary except Exception as e: print(f"Error generating calibration summary: {str(e)}") return {'error': str(e)} def calibrate_breed_scores(breed_scores: List[Tuple[str, float]], method: str = 'auto') -> CalibrationResult: """ 便利函數：校準品種分數 Args: breed_scores: (breed_name, score) 元組列表 method: 校準方法 Returns: CalibrationResult: 校準結果 """ calibrator = ScoreCalibrator() return calibrator.calibrate_scores(breed_scores, method) def get_calibrated_rankings(breed_scores: List[Tuple[str, float]], method: str = 'auto') -> List[Tuple[str, float, int]]: """ 便利函數：獲取校準後的排名 Args: breed_scores: (breed_name, score) 元組列表 method: 校準方法 Returns: List[Tuple[str, float, int]]: (breed_name, calibrated_score, rank) 列表 """ calibrator = ScoreCalibrator() result = calibrator.calibrate_scores(breed_scores, method) # 打破平手機制 calibrated_with_breed = [(breed, result.score_mapping[breed]) for breed in result.score_mapping] calibrated_with_tie_breaking = calibrator.apply_tie_breaking(calibrated_with_breed) # 添加排名 ranked_results = [] for rank, (breed, score) in enumerate(calibrated_with_tie_breaking, 1): ranked_results.append((breed, score, rank)) return ranked_results