Update src/evaluation.py

src/evaluation.py

@@ -17,6 +17,7 @@ from config import (
     STATISTICAL_CONFIG,
     EVALUATION_TRACKS,
     MODEL_CATEGORIES,
+    SAMPLE_SIZE_RECOMMENDATIONS,
 )
 from src.utils import get_all_language_pairs, get_google_comparable_pairs
 
@@ -25,35 +26,35 @@ warnings.filterwarnings("ignore", category=RuntimeWarning)
 
 def calculate_sentence_metrics(reference: str, prediction: str) -> Dict[str, float]:
     """Calculate all metrics for a single sentence pair with robust error handling."""
-
+    
     # Handle empty predictions
     if not prediction or not isinstance(prediction, str):
         prediction = ""
-
+    
     if not reference or not isinstance(reference, str):
         reference = ""
-
+    
     # Normalize texts
     normalizer = BasicTextNormalizer()
     pred_norm = normalizer(prediction)
     ref_norm = normalizer(reference)
-
+    
     metrics = {}
-
+    
     # BLEU score (0-100 scale)
     try:
         bleu = BLEU(effective_order=True)
         metrics["bleu"] = bleu.sentence_score(pred_norm, [ref_norm]).score
     except:
         metrics["bleu"] = 0.0
-
+    
     # ChrF score (normalize to 0-1)
     try:
         chrf = CHRF()
         metrics["chrf"] = chrf.sentence_score(pred_norm, [ref_norm]).score / 100.0
     except:
         metrics["chrf"] = 0.0
-
+    
     # Character Error Rate (CER)
     try:
         if len(ref_norm) > 0:
@@ -62,20 +63,18 @@ def calculate_sentence_metrics(reference: str, prediction: str) -> Dict[str, flo
             metrics["cer"] = 1.0 if len(pred_norm) > 0 else 0.0
     except:
         metrics["cer"] = 1.0
-
+    
     # Word Error Rate (WER)
     try:
         ref_words = ref_norm.split()
         pred_words = pred_norm.split()
         if len(ref_words) > 0:
-            metrics["wer"] = Levenshtein.distance(ref_words, pred_words) / len(
-                ref_words
-            )
+            metrics["wer"] = Levenshtein.distance(ref_words, pred_words) / len(ref_words)
         else:
             metrics["wer"] = 1.0 if len(pred_words) > 0 else 0.0
     except:
         metrics["wer"] = 1.0
-
+    
     # Length ratio
     try:
         if len(ref_norm) > 0:
@@ -84,14 +83,14 @@ def calculate_sentence_metrics(reference: str, prediction: str) -> Dict[str, flo
             metrics["len_ratio"] = 1.0 if len(pred_norm) == 0 else float("inf")
     except:
         metrics["len_ratio"] = 1.0
-
+    
     # ROUGE scores
     try:
         scorer = rouge_scorer.RougeScorer(
             ["rouge1", "rouge2", "rougeL"], use_stemmer=True
         )
         rouge_scores = scorer.score(ref_norm, pred_norm)
-
+        
         metrics["rouge1"] = rouge_scores["rouge1"].fmeasure
         metrics["rouge2"] = rouge_scores["rouge2"].fmeasure
         metrics["rougeL"] = rouge_scores["rougeL"].fmeasure
@@ -99,7 +98,7 @@ def calculate_sentence_metrics(reference: str, prediction: str) -> Dict[str, flo
         metrics["rouge1"] = 0.0
         metrics["rouge2"] = 0.0
         metrics["rougeL"] = 0.0
-
+    
     # Quality score (composite metric)
     try:
         quality_components = [
@@ -113,13 +112,13 @@ def calculate_sentence_metrics(reference: str, prediction: str) -> Dict[str, flo
         metrics["quality_score"] = np.mean(quality_components)
     except:
         metrics["quality_score"] = 0.0
-
+    
     return metrics
 
 
 def calculate_statistical_metrics(values: List[float]) -> Dict[str, float]:
     """Calculate statistical measures including confidence intervals."""
-
+    
     if not values or len(values) == 0:
         return {
             "mean": 0.0,
@@ -129,10 +128,10 @@ def calculate_statistical_metrics(values: List[float]) -> Dict[str, float]:
             "ci_upper": 0.0,
             "n_samples": 0,
         }
-
+    
     values = np.array(values)
     values = values[~np.isnan(values)]  # Remove NaN values
-
+    
     if len(values) == 0:
         return {
             "mean": 0.0,
@@ -142,23 +141,23 @@ def calculate_statistical_metrics(values: List[float]) -> Dict[str, float]:
             "ci_upper": 0.0,
             "n_samples": 0,
         }
-
+    
     stats_dict = {
         "mean": float(np.mean(values)),
         "std": float(np.std(values, ddof=1)) if len(values) > 1 else 0.0,
         "median": float(np.median(values)),
         "n_samples": len(values),
     }
-
+    
     # Calculate confidence intervals using bootstrap if enough samples
     if len(values) >= STATISTICAL_CONFIG["min_samples_for_ci"]:
         try:
             confidence_level = STATISTICAL_CONFIG["confidence_level"]
-
+            
             # Bootstrap confidence interval
             def mean_func(x):
                 return np.mean(x)
-
+            
             res = bootstrap(
                 (values,),
                 mean_func,
@@ -166,10 +165,10 @@ def calculate_statistical_metrics(values: List[float]) -> Dict[str, float]:
                 confidence_level=confidence_level,
                 random_state=42,
             )
-
+            
             stats_dict["ci_lower"] = float(res.confidence_interval.low)
             stats_dict["ci_upper"] = float(res.confidence_interval.high)
-
+            
         except Exception as e:
             # Fallback to t-distribution CI
             try:
@@ -184,7 +183,7 @@ def calculate_statistical_metrics(values: List[float]) -> Dict[str, float]:
     else:
         stats_dict["ci_lower"] = stats_dict["mean"]
         stats_dict["ci_upper"] = stats_dict["mean"]
-
+    
     return stats_dict
 
 
@@ -192,51 +191,47 @@ def perform_significance_test(
     values1: List[float], values2: List[float], metric_name: str
 ) -> Dict[str, float]:
     """Perform statistical significance test between two groups."""
-
+    
     if len(values1) < 2 or len(values2) < 2:
         return {"p_value": 1.0, "effect_size": 0.0, "significant": False}
-
+    
     values1 = np.array(values1)
     values2 = np.array(values2)
-
+    
     # Remove NaN values
     values1 = values1[~np.isnan(values1)]
     values2 = values2[~np.isnan(values2)]
-
+    
     if len(values1) < 2 or len(values2) < 2:
         return {"p_value": 1.0, "effect_size": 0.0, "significant": False}
-
+    
     try:
         # Perform t-test
         t_stat, p_value = stats.ttest_ind(values1, values2, equal_var=False)
-
+        
         # Calculate effect size (Cohen's d)
         pooled_std = np.sqrt(
-            (
-                (len(values1) - 1) * np.var(values1, ddof=1)
-                + (len(values2) - 1) * np.var(values2, ddof=1)
-            )
-            / (len(values1) + len(values2) - 2)
+            ((len(values1) - 1) * np.var(values1, ddof=1) +
+             (len(values2) - 1) * np.var(values2, ddof=1)) /
+            (len(values1) + len(values2) - 2)
         )
-
+        
         if pooled_std > 0:
             effect_size = abs(np.mean(values1) - np.mean(values2)) / pooled_std
         else:
             effect_size = 0.0
-
+        
         # Determine significance
-        significance_level = EVALUATION_TRACKS["google_comparable"][
-            "significance_level"
-        ]
+        significance_level = EVALUATION_TRACKS["google_comparable"]["significance_level"]
         significant = p_value < significance_level
-
+        
         return {
             "p_value": float(p_value),
             "effect_size": float(effect_size),
             "significant": significant,
             "t_statistic": float(t_stat),
         }
-
+        
     except Exception as e:
         return {"p_value": 1.0, "effect_size": 0.0, "significant": False}
 
@@ -245,32 +240,32 @@ def evaluate_predictions_by_track(
     predictions: pd.DataFrame, test_set: pd.DataFrame, track: str
 ) -> Dict:
     """Evaluate predictions for a specific track with statistical analysis."""
-
+    
     print(f"🔄 Evaluating for {track} track...")
-
+    
     track_config = EVALUATION_TRACKS[track]
     track_languages = track_config["languages"]
-
+    
     # Filter test set and predictions to track languages
     track_test_set = test_set[
-        (test_set["source_language"].isin(track_languages))
-        & (test_set["target_language"].isin(track_languages))
+        (test_set["source_language"].isin(track_languages)) &
+        (test_set["target_language"].isin(track_languages))
     ].copy()
-
+    
     # Merge predictions with test set
     merged = track_test_set.merge(
         predictions, on="sample_id", how="inner", suffixes=("", "_pred")
     )
-
+    
     if len(merged) == 0:
         return {
             "error": f"No matching samples found for {track} track",
             "evaluated_samples": 0,
             "track": track,
         }
-
+    
     print(f"📊 Evaluating {len(merged)} samples for {track} track...")
-
+    
     # Calculate metrics for each sample
     sample_metrics = []
     for idx, row in merged.iterrows():
@@ -279,78 +274,67 @@ def evaluate_predictions_by_track(
         metrics["source_language"] = row["source_language"]
         metrics["target_language"] = row["target_language"]
         sample_metrics.append(metrics)
-
+    
     sample_df = pd.DataFrame(sample_metrics)
-
+    
     # Aggregate by language pairs with statistical analysis
     pair_metrics = {}
     overall_metrics = defaultdict(list)
-
+    
     # Calculate metrics for each language pair
     for src_lang in track_languages:
         for tgt_lang in track_languages:
             if src_lang == tgt_lang:
                 continue
-
+                
             pair_data = sample_df[
-                (sample_df["source_language"] == src_lang)
-                & (sample_df["target_language"] == tgt_lang)
+                (sample_df["source_language"] == src_lang) &
+                (sample_df["target_language"] == tgt_lang)
             ]
-
+            
             if len(pair_data) >= track_config["min_samples_per_pair"]:
                 pair_key = f"{src_lang}_to_{tgt_lang}"
                 pair_metrics[pair_key] = {}
-
+                
                 # Calculate statistical metrics for each measure
                 for metric in (
-                    METRICS_CONFIG["primary_metrics"]
-                    + METRICS_CONFIG["secondary_metrics"]
+                    METRICS_CONFIG["primary_metrics"] + METRICS_CONFIG["secondary_metrics"]
                 ):
                     if metric in pair_data.columns:
-                        values = (
-                            pair_data[metric]
-                            .replace([np.inf, -np.inf], np.nan)
-                            .dropna()
-                        )
-
+                        values = pair_data[metric].replace([np.inf, -np.inf], np.nan).dropna()
+                        
                         if len(values) > 0:
-                            stats_metrics = calculate_statistical_metrics(
-                                values.tolist()
-                            )
+                            stats_metrics = calculate_statistical_metrics(values.tolist())
                             pair_metrics[pair_key][metric] = stats_metrics
-
+                            
                             # Add to overall metrics for track-level statistics
                             overall_metrics[metric].append(stats_metrics["mean"])
-
+                
                 pair_metrics[pair_key]["sample_count"] = len(pair_data)
                 pair_metrics[pair_key]["languages"] = f"{src_lang}-{tgt_lang}"
-
+    
     # Calculate track-level aggregated statistics
     track_averages = {}
     track_statistics = {}
-
+    
     for metric in overall_metrics:
         if overall_metrics[metric]:
             track_stats = calculate_statistical_metrics(overall_metrics[metric])
             track_averages[metric] = track_stats["mean"]
             track_statistics[metric] = track_stats
-
+    
     # Generate evaluation summary
     summary = {
         "track": track,
         "track_name": track_config["name"],
         "total_samples": len(sample_df),
-        "language_pairs_evaluated": len(
-            [k for k in pair_metrics if pair_metrics[k].get("sample_count", 0) > 0]
-        ),
-        "languages_covered": len(
-            set(sample_df["source_language"]) | set(sample_df["target_language"])
-        ),
+        "language_pairs_evaluated": len([k for k in pair_metrics if pair_metrics[k].get("sample_count", 0) > 0]),
+        "languages_covered": len(set(sample_df["source_language"]) | set(sample_df["target_language"])),
         "min_samples_per_pair": track_config["min_samples_per_pair"],
         "statistical_power": track_config["statistical_power"],
         "significance_level": track_config["significance_level"],
     }
-
+    
     return {
         "sample_metrics": sample_df,
         "pair_metrics": pair_metrics,
@@ -367,13 +351,13 @@ def evaluate_predictions_scientific(
     predictions: pd.DataFrame, test_set: pd.DataFrame, model_category: str = "community"
 ) -> Dict:
     """Comprehensive evaluation across all tracks with scientific rigor."""
-
+    
     print("🔬 Starting scientific evaluation...")
-
+    
     # Validate model category
     if model_category not in MODEL_CATEGORIES:
         model_category = "community"
-
+    
     evaluation_results = {
         "model_category": model_category,
         "category_info": MODEL_CATEGORIES[model_category],
@@ -385,66 +369,56 @@ def evaluate_predictions_scientific(
             "total_samples_available": len(test_set),
         },
     }
-
+    
     # Evaluate each track
     for track_name in EVALUATION_TRACKS.keys():
         track_result = evaluate_predictions_by_track(predictions, test_set, track_name)
         evaluation_results["tracks"][track_name] = track_result
-
+    
     # Cross-track consistency analysis
     evaluation_results["cross_track_analysis"] = analyze_cross_track_consistency(
         evaluation_results["tracks"]
     )
-
+    
     return evaluation_results
 
 
 def analyze_cross_track_consistency(track_results: Dict) -> Dict:
     """Analyze consistency of model performance across different tracks."""
-
+    
     consistency_analysis = {
         "track_correlations": {},
         "performance_stability": {},
         "language_coverage_analysis": {},
     }
-
+    
     # Extract quality scores from each track for correlation analysis
     track_scores = {}
     for track_name, track_data in track_results.items():
-        if (
-            track_data.get("track_averages")
-            and "quality_score" in track_data["track_averages"]
-        ):
+        if track_data.get("track_averages") and "quality_score" in track_data["track_averages"]:
             track_scores[track_name] = track_data["track_averages"]["quality_score"]
-
+    
     # Calculate pairwise correlations (would need more data points for meaningful correlation)
     if len(track_scores) >= 2:
         track_names = list(track_scores.keys())
         for i, track1 in enumerate(track_names):
-            for track2 in track_names[i + 1 :]:
+            for track2 in track_names[i + 1:]:
                 # This would be more meaningful with multiple models
                 consistency_analysis["track_correlations"][f"{track1}_vs_{track2}"] = {
-                    "score_difference": abs(
-                        track_scores[track1] - track_scores[track2]
-                    ),
-                    "relative_performance": track_scores[track1]
-                    / max(track_scores[track2], 0.001),
+                    "score_difference": abs(track_scores[track1] - track_scores[track2]),
+                    "relative_performance": track_scores[track1] / max(track_scores[track2], 0.001),
                 }
-
+    
     # Language coverage analysis
     for track_name, track_data in track_results.items():
         if track_data.get("summary"):
             summary = track_data["summary"]
             consistency_analysis["language_coverage_analysis"][track_name] = {
-                "coverage_rate": summary["language_pairs_evaluated"]
-                / max(summary.get("total_possible_pairs", 1), 1),
-                "samples_per_pair": summary["total_samples"]
-                / max(summary["language_pairs_evaluated"], 1),
-                "statistical_adequacy": summary["total_samples"]
-                >= EVALUATION_TRACKS[track_name]["min_samples_per_pair"]
-                * summary["language_pairs_evaluated"],
+                "coverage_rate": summary["language_pairs_evaluated"] / max(summary.get("total_possible_pairs", 1), 1),
+                "samples_per_pair": summary["total_samples"] / max(summary["language_pairs_evaluated"], 1),
+                "statistical_adequacy": summary["total_samples"] >= EVALUATION_TRACKS[track_name]["min_samples_per_pair"] * summary["language_pairs_evaluated"],
             }
-
+    
     return consistency_analysis
 
 
@@ -452,18 +426,16 @@ def compare_models_statistically(
     model1_results: Dict, model2_results: Dict, track: str = "google_comparable"
 ) -> Dict:
     """Perform statistical comparison between two models on a specific track."""
-
-    if track not in model1_results.get("tracks", {}) or track not in model2_results.get(
-        "tracks", {}
-    ):
+    
+    if track not in model1_results.get("tracks", {}) or track not in model2_results.get("tracks", {}):
         return {"error": f"Track {track} not available for both models"}
-
+    
     track1_data = model1_results["tracks"][track]
     track2_data = model2_results["tracks"][track]
-
+    
     if track1_data.get("error") or track2_data.get("error"):
         return {"error": "One or both models have evaluation errors"}
-
+    
     comparison_results = {
         "track": track,
         "model1_category": model1_results.get("model_category", "unknown"),
@@ -472,22 +444,19 @@ def compare_models_statistically(
         "language_pair_comparisons": {},
         "overall_significance": {},
     }
-
+    
     # Compare each metric
-    for metric in (
-        METRICS_CONFIG["primary_metrics"] + METRICS_CONFIG["secondary_metrics"]
-    ):
-        if metric in track1_data.get(
-            "track_statistics", {}
-        ) and metric in track2_data.get("track_statistics", {}):
-
+    for metric in METRICS_CONFIG["primary_metrics"] + METRICS_CONFIG["secondary_metrics"]:
+        if (metric in track1_data.get("track_statistics", {}) and 
+            metric in track2_data.get("track_statistics", {})):
+            
             # Extract sample-level data for this metric from both models
             # This would require access to the original sample metrics
             # For now, we'll use the aggregated statistics
-
+            
             stats1 = track1_data["track_statistics"][metric]
             stats2 = track2_data["track_statistics"][metric]
-
+            
             # Create comparison summary
             comparison_results["metric_comparisons"][metric] = {
                 "model1_mean": stats1["mean"],
@@ -495,12 +464,10 @@ def compare_models_statistically(
                 "model2_mean": stats2["mean"],
                 "model2_ci": [stats2["ci_lower"], stats2["ci_upper"]],
                 "difference": stats1["mean"] - stats2["mean"],
-                "ci_overlap": not (
-                    stats1["ci_upper"] < stats2["ci_lower"]
-                    or stats2["ci_upper"] < stats1["ci_lower"]
-                ),
+                "ci_overlap": not (stats1["ci_upper"] < stats2["ci_lower"] or 
+                                 stats2["ci_upper"] < stats1["ci_lower"]),
             }
-
+    
     return comparison_results
 
 
@@ -508,49 +475,43 @@ def generate_scientific_report(
     results: Dict, model_name: str = "", baseline_results: Dict = None
 ) -> str:
     """Generate a comprehensive scientific evaluation report."""
-
-    if any(
-        track_data.get("error") for track_data in results.get("tracks", {}).values()
-    ):
+    
+    if any(track_data.get("error") for track_data in results.get("tracks", {}).values()):
         return f"❌ **Evaluation Error**: Unable to complete scientific evaluation"
-
+    
     report = []
-
+    
     # Header
     report.append(f"# 🔬 Scientific Evaluation Report: {model_name or 'Model'}")
     report.append("")
-
+    
     # Model categorization
     category_info = results.get("category_info", {})
     report.append(f"**Model Category**: {category_info.get('name', 'Unknown')}")
-    report.append(
-        f"**Category Description**: {category_info.get('description', 'N/A')}"
-    )
+    report.append(f"**Category Description**: {category_info.get('description', 'N/A')}")
     report.append("")
-
+    
     # Track-by-track analysis
     for track_name, track_data in results.get("tracks", {}).items():
         if track_data.get("error"):
             continue
-
+            
         track_config = EVALUATION_TRACKS[track_name]
         summary = track_data.get("summary", {})
         track_stats = track_data.get("track_statistics", {})
-
+        
         report.append(f"## {track_config['name']}")
         report.append(f"*{track_config['description']}*")
         report.append("")
-
+        
         # Summary statistics
         report.append("### 📊 Summary Statistics")
         report.append(f"- **Samples Evaluated**: {summary.get('total_samples', 0):,}")
-        report.append(
-            f"- **Language Pairs**: {summary.get('language_pairs_evaluated', 0)}"
-        )
+        report.append(f"- **Language Pairs**: {summary.get('language_pairs_evaluated', 0)}")
         report.append(f"- **Languages Covered**: {summary.get('languages_covered', 0)}")
         report.append(f"- **Statistical Power**: {track_config['statistical_power']}")
         report.append("")
-
+        
         # Primary metrics with confidence intervals
         report.append("### 🎯 Primary Metrics (95% Confidence Intervals)")
         for metric in METRICS_CONFIG["primary_metrics"]:
@@ -559,68 +520,51 @@ def generate_scientific_report(
                 mean_val = stats["mean"]
                 ci_lower = stats["ci_lower"]
                 ci_upper = stats["ci_upper"]
-
-                report.append(
-                    f"- **{metric.upper()}**: {mean_val:.4f} [{ci_lower:.4f}, {ci_upper:.4f}]"
-                )
+                
+                report.append(f"- **{metric.upper()}**: {mean_val:.4f} [{ci_lower:.4f}, {ci_upper:.4f}]")
         report.append("")
-
+        
         # Statistical adequacy assessment
-        min_required = track_config["min_samples_per_pair"] * summary.get(
-            "language_pairs_evaluated", 0
-        )
-        adequacy = (
-            "✅ Adequate"
-            if summary.get("total_samples", 0) >= min_required
-            else "⚠️ Limited"
-        )
+        min_required = track_config["min_samples_per_pair"] * summary.get("language_pairs_evaluated", 0)
+        adequacy = "✅ Adequate" if summary.get("total_samples", 0) >= min_required else "⚠️ Limited"
         report.append(f"**Statistical Adequacy**: {adequacy}")
         report.append("")
-
+    
     # Cross-track analysis
     cross_track = results.get("cross_track_analysis", {})
     if cross_track:
         report.append("## 🔄 Cross-Track Consistency Analysis")
-
+        
         coverage_analysis = cross_track.get("language_coverage_analysis", {})
         for track_name, coverage_info in coverage_analysis.items():
-            adequacy = (
-                "✅ Statistically adequate"
-                if coverage_info.get("statistical_adequacy")
-                else "⚠️ Limited statistical power"
-            )
+            adequacy = "✅ Statistically adequate" if coverage_info.get("statistical_adequacy") else "⚠️ Limited statistical power"
             report.append(f"- **{track_name}**: {adequacy}")
-
+        
         report.append("")
-
+    
     # Baseline comparison if available
     if baseline_results:
         report.append("## 📈 Baseline Comparison")
         # This would include detailed statistical comparisons
         report.append("*Statistical comparison with baseline models*")
         report.append("")
-
+    
     # Scientific recommendations
     report.append("## 💡 Scientific Recommendations")
-
+    
     total_samples = sum(
         track_data.get("summary", {}).get("total_samples", 0)
         for track_data in results.get("tracks", {}).values()
         if not track_data.get("error")
     )
-
+    
     if total_samples < SAMPLE_SIZE_RECOMMENDATIONS["publication_quality"]:
-        report.append(
-            "- ⚠️ Consider collecting more evaluation samples for publication-quality results"
-        )
-
+        report.append("- ⚠️ Consider collecting more evaluation samples for publication-quality results")
+    
     google_track = results.get("tracks", {}).get("google_comparable", {})
-    if (
-        not google_track.get("error")
-        and google_track.get("summary", {}).get("total_samples", 0) > 100
-    ):
+    if not google_track.get("error") and google_track.get("summary", {}).get("total_samples", 0) > 100:
         report.append("- ✅ Sufficient data for comparison with commercial systems")
-
+    
     report.append("")
-
-    return "\n".join(report)
+    
+    return "\n".join(report)