Update src/leaderboard.py

src/leaderboard.py

@@ -13,13 +13,14 @@ from config import (
     MODEL_CATEGORIES,
     STATISTICAL_CONFIG,
     METRICS_CONFIG,
+    SAMPLE_SIZE_RECOMMENDATIONS,
 )
 from src.utils import create_submission_id, sanitize_model_name
 
 
 def initialize_scientific_leaderboard() -> pd.DataFrame:
     """Initialize empty scientific leaderboard DataFrame with all required columns."""
-
+    
     columns = {
         # Basic information
         "submission_id": [],
@@ -28,18 +29,22 @@ def initialize_scientific_leaderboard() -> pd.DataFrame:
         "submission_date": [],
         "model_category": [],
         "description": [],
+        
         # Track-specific quality scores
         "google_comparable_quality": [],
         "ug40_complete_quality": [],
         "language_pair_matrix_quality": [],
+        
         # Track-specific BLEU scores
         "google_comparable_bleu": [],
         "ug40_complete_bleu": [],
         "language_pair_matrix_bleu": [],
+        
         # Track-specific ChrF scores
         "google_comparable_chrf": [],
         "ug40_complete_chrf": [],
         "language_pair_matrix_chrf": [],
+        
         # Statistical metadata
         "google_comparable_ci_lower": [],
         "google_comparable_ci_upper": [],
@@ -47,6 +52,7 @@ def initialize_scientific_leaderboard() -> pd.DataFrame:
         "ug40_complete_ci_upper": [],
         "language_pair_matrix_ci_lower": [],
         "language_pair_matrix_ci_upper": [],
+        
         # Coverage information
         "google_comparable_samples": [],
         "ug40_complete_samples": [],
@@ -54,32 +60,35 @@ def initialize_scientific_leaderboard() -> pd.DataFrame:
         "google_comparable_pairs": [],
         "ug40_complete_pairs": [],
         "language_pair_matrix_pairs": [],
+        
         # Statistical adequacy flags
         "google_comparable_adequate": [],
         "ug40_complete_adequate": [],
         "language_pair_matrix_adequate": [],
+        
         # Detailed results (JSON strings)
         "detailed_google_comparable": [],
         "detailed_ug40_complete": [],
         "detailed_language_pair_matrix": [],
         "cross_track_analysis": [],
+        
         # Metadata
         "evaluation_date": [],
         "leaderboard_version": [],
         "scientific_adequacy_score": [],
     }
-
+    
     return pd.DataFrame(columns)
 
 
 def load_scientific_leaderboard() -> pd.DataFrame:
     """Load current scientific leaderboard from HuggingFace dataset."""
-
+    
     try:
         print("📥 Loading scientific leaderboard...")
         dataset = load_dataset(LEADERBOARD_DATASET + "-scientific", split="train")
         df = dataset.to_pandas()
-
+        
         # Ensure all required columns exist
         required_columns = list(initialize_scientific_leaderboard().columns)
         for col in required_columns:
@@ -96,10 +105,10 @@ def load_scientific_leaderboard() -> pd.DataFrame:
                     df[col] = 2  # Scientific version
                 else:
                     df[col] = ""
-
+        
         print(f"✅ Loaded scientific leaderboard with {len(df)} entries")
         return df
-
+        
     except Exception as e:
         print(f"⚠️ Could not load scientific leaderboard: {e}")
         print("🔄 Initializing empty scientific leaderboard...")
@@ -108,51 +117,37 @@ def load_scientific_leaderboard() -> pd.DataFrame:
 
 def save_scientific_leaderboard(df: pd.DataFrame) -> bool:
     """Save scientific leaderboard to HuggingFace dataset."""
-
+    
     try:
         # Clean data before saving
         df_clean = df.copy()
-
+        
         # Ensure numeric columns are proper types
         numeric_columns = [
-            col
-            for col in df_clean.columns
-            if any(
-                x in col
-                for x in [
-                    "quality",
-                    "bleu",
-                    "chrf",
-                    "ci_",
-                    "samples",
-                    "pairs",
-                    "adequacy",
-                ]
-            )
+            col for col in df_clean.columns 
+            if any(x in col for x in ["quality", "bleu", "chrf", "ci_", "samples", "pairs", "adequacy"])
         ]
-
+        
         for col in numeric_columns:
             if col in df_clean.columns:
                 if "adequate" in col:
                     df_clean[col] = df_clean[col].astype(bool)
                 else:
-                    df_clean[col] = pd.to_numeric(
-                        df_clean[col], errors="coerce"
-                    ).fillna(0.0)
-
+                    df_clean[col] = pd.to_numeric(df_clean[col], errors="coerce").fillna(0.0)
+        
         # Convert to dataset
         dataset = Dataset.from_pandas(df_clean)
-
+        
         # Push to hub
         dataset.push_to_hub(
             LEADERBOARD_DATASET + "-scientific",
             token=HF_TOKEN,
             commit_message=f"Update scientific leaderboard - {datetime.datetime.now().isoformat()[:19]}",
         )
-
+        
         print("✅ Scientific leaderboard saved successfully!")
         return True
-
+        
     except Exception as e:
         print(f"❌ Error saving scientific leaderboard: {e}")
         return False
@@ -166,241 +161,231 @@ def add_model_to_scientific_leaderboard(
     description: str = "",
 ) -> pd.DataFrame:
     """Add new model results to scientific leaderboard."""
-
+    
     # Load current leaderboard
     df = load_scientific_leaderboard()
-
+    
     # Remove existing entry if present
     existing_mask = df["model_name"] == model_name
     if existing_mask.any():
         df = df[~existing_mask]
-
+    
     # Extract track results
     tracks = evaluation_results.get("tracks", {})
     cross_track = evaluation_results.get("cross_track_analysis", {})
-
+    
     # Calculate scientific adequacy score
     adequacy_score = calculate_scientific_adequacy_score(evaluation_results)
-
+    
     # Prepare new entry
     new_entry = {
         "submission_id": create_submission_id(),
         "model_name": sanitize_model_name(model_name),
         "author": author[:100] if author else "Anonymous",
         "submission_date": datetime.datetime.now().isoformat(),
-        "model_category": (
-            model_category if model_category in MODEL_CATEGORIES else "community"
-        ),
+        "model_category": model_category if model_category in MODEL_CATEGORIES else "community",
         "description": description[:500] if description else "",
+        
         # Extract track-specific metrics
         **extract_track_metrics(tracks),
+        
         # Statistical metadata
         **extract_statistical_metadata(tracks),
+        
         # Coverage information
         **extract_coverage_information(tracks),
+        
         # Adequacy flags
         **extract_adequacy_flags(tracks),
+        
         # Detailed results (JSON strings)
         **serialize_detailed_results(tracks, cross_track),
+        
         # Metadata
         "evaluation_date": datetime.datetime.now().isoformat(),
         "leaderboard_version": 2,
         "scientific_adequacy_score": adequacy_score,
     }
-
+    
     # Convert to DataFrame and append
     new_row_df = pd.DataFrame([new_entry])
     updated_df = pd.concat([df, new_row_df], ignore_index=True)
-
+    
     # Save to hub
     save_scientific_leaderboard(updated_df)
-
+    
     return updated_df
 
 
 def extract_track_metrics(tracks: Dict) -> Dict:
     """Extract primary metrics from each track."""
-
+    
     metrics = {}
-
+    
     for track_name in EVALUATION_TRACKS.keys():
         track_data = tracks.get(track_name, {})
         track_averages = track_data.get("track_averages", {})
-
+        
         # Quality score
-        metrics[f"{track_name}_quality"] = float(
-            track_averages.get("quality_score", 0.0)
-        )
-
+        metrics[f"{track_name}_quality"] = float(track_averages.get("quality_score", 0.0))
+        
         # BLEU score
         metrics[f"{track_name}_bleu"] = float(track_averages.get("bleu", 0.0))
-
+        
         # ChrF score
         metrics[f"{track_name}_chrf"] = float(track_averages.get("chrf", 0.0))
-
+    
     return metrics
 
 
 def extract_statistical_metadata(tracks: Dict) -> Dict:
     """Extract confidence intervals from each track."""
-
+    
     metadata = {}
-
+    
     for track_name in EVALUATION_TRACKS.keys():
         track_data = tracks.get(track_name, {})
         track_statistics = track_data.get("track_statistics", {})
-
+        
         quality_stats = track_statistics.get("quality_score", {})
         metadata[f"{track_name}_ci_lower"] = float(quality_stats.get("ci_lower", 0.0))
         metadata[f"{track_name}_ci_upper"] = float(quality_stats.get("ci_upper", 0.0))
-
+    
     return metadata
 
 
 def extract_coverage_information(tracks: Dict) -> Dict:
     """Extract coverage information from each track."""
-
+    
     coverage = {}
-
+    
     for track_name in EVALUATION_TRACKS.keys():
         track_data = tracks.get(track_name, {})
         summary = track_data.get("summary", {})
-
+        
         coverage[f"{track_name}_samples"] = int(summary.get("total_samples", 0))
-        coverage[f"{track_name}_pairs"] = int(
-            summary.get("language_pairs_evaluated", 0)
-        )
-
+        coverage[f"{track_name}_pairs"] = int(summary.get("language_pairs_evaluated", 0))
+    
     return coverage
 
 
 def extract_adequacy_flags(tracks: Dict) -> Dict:
     """Extract statistical adequacy flags for each track."""
-
+    
     adequacy = {}
-
+    
     for track_name in EVALUATION_TRACKS.keys():
         track_data = tracks.get(track_name, {})
         summary = track_data.get("summary", {})
-
-        min_required = EVALUATION_TRACKS[track_name][
-            "min_samples_per_pair"
-        ] * summary.get("language_pairs_evaluated", 0)
+        
+        min_required = EVALUATION_TRACKS[track_name]["min_samples_per_pair"] * summary.get("language_pairs_evaluated", 0)
         is_adequate = summary.get("total_samples", 0) >= min_required
-
+        
         adequacy[f"{track_name}_adequate"] = bool(is_adequate)
-
+    
     return adequacy
 
 
 def serialize_detailed_results(tracks: Dict, cross_track: Dict) -> Dict:
     """Serialize detailed results for storage."""
-
+    
     detailed = {}
-
+    
     for track_name in EVALUATION_TRACKS.keys():
         track_data = tracks.get(track_name, {})
-
+        
         # Remove non-serializable data
         safe_track_data = {}
         for key, value in track_data.items():
             if key != "sample_metrics":  # Skip large DataFrames
                 safe_track_data[key] = value
-
+        
         detailed[f"detailed_{track_name}"] = json.dumps(safe_track_data)
-
+    
     detailed["cross_track_analysis"] = json.dumps(cross_track)
-
+    
     return detailed
 
 
 def calculate_scientific_adequacy_score(evaluation_results: Dict) -> float:
     """Calculate overall scientific adequacy score (0-1)."""
-
+    
     tracks = evaluation_results.get("tracks", {})
-
+    
     adequacy_scores = []
-
+    
     for track_name in EVALUATION_TRACKS.keys():
         track_data = tracks.get(track_name, {})
         summary = track_data.get("summary", {})
-
+        
         if track_data.get("error"):
             adequacy_scores.append(0.0)
             continue
-
+        
         # Sample size adequacy
-        min_required = EVALUATION_TRACKS[track_name][
-            "min_samples_per_pair"
-        ] * summary.get("language_pairs_evaluated", 0)
-        sample_adequacy = min(
-            summary.get("total_samples", 0) / max(min_required, 1), 1.0
-        )
-
+        min_required = EVALUATION_TRACKS[track_name]["min_samples_per_pair"] * summary.get("language_pairs_evaluated", 0)
+        sample_adequacy = min(summary.get("total_samples", 0) / max(min_required, 1), 1.0)
+        
         # Coverage adequacy
-        total_possible_pairs = len(EVALUATION_TRACKS[track_name]["languages"]) * (
-            len(EVALUATION_TRACKS[track_name]["languages"]) - 1
-        )
-        coverage_adequacy = summary.get("language_pairs_evaluated", 0) / max(
-            total_possible_pairs, 1
-        )
-
+        total_possible_pairs = len(EVALUATION_TRACKS[track_name]["languages"]) * (len(EVALUATION_TRACKS[track_name]["languages"]) - 1)
+        coverage_adequacy = summary.get("language_pairs_evaluated", 0) / max(total_possible_pairs, 1)
+        
         # Track adequacy
         track_adequacy = (sample_adequacy + coverage_adequacy) / 2
         adequacy_scores.append(track_adequacy)
-
+    
     return float(np.mean(adequacy_scores))
 
 
 def get_track_leaderboard(
-    df: pd.DataFrame,
-    track: str,
+    df: pd.DataFrame, 
+    track: str, 
     metric: str = "quality",
     category_filter: str = "all",
-    min_adequacy: float = 0.0,
+    min_adequacy: float = 0.0
 ) -> pd.DataFrame:
     """Get leaderboard for a specific track with filtering."""
-
+    
     if df.empty:
         return df
-
+    
     track_quality_col = f"{track}_{metric}"
     track_adequate_col = f"{track}_adequate"
-
+    
     # Filter by adequacy
     if min_adequacy > 0:
         adequacy_mask = df["scientific_adequacy_score"] >= min_adequacy
         df = df[adequacy_mask]
-
+    
     # Filter by category
     if category_filter != "all":
         df = df[df["model_category"] == category_filter]
-
+    
     # Filter to models that have this track
     valid_mask = (df[track_quality_col] > 0) & df[track_adequate_col]
     df = df[valid_mask]
-
+    
     if df.empty:
         return df
-
+    
     # Sort by track-specific metric
     df = df.sort_values(track_quality_col, ascending=False).reset_index(drop=True)
-
+    
     return df
 
 
 def prepare_track_leaderboard_display(df: pd.DataFrame, track: str) -> pd.DataFrame:
     """Prepare track-specific leaderboard for display."""
-
+    
     if df.empty:
         return df
-
+    
     # Select relevant columns for this track
     base_columns = ["model_name", "author", "submission_date", "model_category"]
-
+    
     track_columns = [
         f"{track}_quality",
-        f"{track}_bleu",
+        f"{track}_bleu", 
         f"{track}_chrf",
         f"{track}_ci_lower",
         f"{track}_ci_upper",
@@ -408,13 +393,11 @@ def prepare_track_leaderboard_display(df: pd.DataFrame, track: str) -> pd.DataFr
         f"{track}_pairs",
         f"{track}_adequate",
     ]
-
+    
     # Only include columns that exist
-    available_columns = [
-        col for col in base_columns + track_columns if col in df.columns
-    ]
+    available_columns = [col for col in base_columns + track_columns if col in df.columns]
     display_df = df[available_columns].copy()
-
+    
     # Format numeric columns
     numeric_format = {
         f"{track}_quality": "{:.4f}",
@@ -423,34 +406,25 @@ def prepare_track_leaderboard_display(df: pd.DataFrame, track: str) -> pd.DataFr
         f"{track}_ci_lower": "{:.4f}",
         f"{track}_ci_upper": "{:.4f}",
     }
-
+    
     for col, fmt in numeric_format.items():
         if col in display_df.columns:
             display_df[col] = display_df[col].apply(
                 lambda x: fmt.format(float(x)) if pd.notnull(x) else "0.0000"
             )
-
+    
     # Format confidence intervals
-    if (
-        f"{track}_ci_lower" in display_df.columns
-        and f"{track}_ci_upper" in display_df.columns
-    ):
+    if f"{track}_ci_lower" in display_df.columns and f"{track}_ci_upper" in display_df.columns:
         display_df[f"{track}_confidence_interval"] = (
-            "["
-            + display_df[f"{track}_ci_lower"]
-            + ", "
-            + display_df[f"{track}_ci_upper"]
-            + "]"
+            "[" + display_df[f"{track}_ci_lower"] + ", " + display_df[f"{track}_ci_upper"] + "]"
         )
         # Remove individual CI columns for cleaner display
         display_df = display_df.drop(columns=[f"{track}_ci_lower", f"{track}_ci_upper"])
-
+    
     # Format submission date
     if "submission_date" in display_df.columns:
-        display_df["submission_date"] = pd.to_datetime(
-            display_df["submission_date"]
-        ).dt.strftime("%Y-%m-%d")
-
+        display_df["submission_date"] = pd.to_datetime(display_df["submission_date"]).dt.strftime("%Y-%m-%d")
+    
     # Rename columns for better display
     track_name = EVALUATION_TRACKS[track]["name"].split()[0]  # First word
     column_renames = {
@@ -466,15 +440,15 @@ def prepare_track_leaderboard_display(df: pd.DataFrame, track: str) -> pd.DataFr
         f"{track}_pairs": "Pairs",
         f"{track}_adequate": "Adequate",
     }
-
+    
     display_df = display_df.rename(columns=column_renames)
-
+    
     return display_df
 
 
 def get_scientific_leaderboard_stats(df: pd.DataFrame, track: str = None) -> Dict:
     """Get comprehensive statistics for the scientific leaderboard."""
-
+    
     if df.empty:
         return {
             "total_models": 0,
@@ -483,7 +457,7 @@ def get_scientific_leaderboard_stats(df: pd.DataFrame, track: str = None) -> Dic
             "adequacy_distribution": {},
             "best_models_by_track": {},
         }
-
+    
     stats = {
         "total_models": len(df),
         "models_by_category": df["model_category"].value_counts().to_dict(),
@@ -491,23 +465,23 @@ def get_scientific_leaderboard_stats(df: pd.DataFrame, track: str = None) -> Dic
         "track_statistics": {},
         "best_models_by_track": {},
     }
-
+    
     # Adequacy distribution
     adequacy_bins = pd.cut(
-        df["scientific_adequacy_score"],
+        df["scientific_adequacy_score"], 
         bins=[0, 0.3, 0.6, 0.8, 1.0],
-        labels=["Poor", "Fair", "Good", "Excellent"],
+        labels=["Poor", "Fair", "Good", "Excellent"]
     )
     stats["adequacy_distribution"] = adequacy_bins.value_counts().to_dict()
-
+    
     # Track-specific statistics
     for track_name in EVALUATION_TRACKS.keys():
         quality_col = f"{track_name}_quality"
         adequate_col = f"{track_name}_adequate"
-
+        
         if quality_col in df.columns and adequate_col in df.columns:
             track_models = df[df[adequate_col] & (df[quality_col] > 0)]
-
+            
             if len(track_models) > 0:
                 stats["track_statistics"][track_name] = {
                     "participating_models": len(track_models),
@@ -515,7 +489,7 @@ def get_scientific_leaderboard_stats(df: pd.DataFrame, track: str = None) -> Dic
                     "std_quality": float(track_models[quality_col].std()),
                     "best_quality": float(track_models[quality_col].max()),
                 }
-
+                
                 # Best model for this track
                 best_model = track_models.loc[track_models[quality_col].idxmax()]
                 stats["best_models_by_track"][track_name] = {
@@ -523,20 +497,22 @@ def get_scientific_leaderboard_stats(df: pd.DataFrame, track: str = None) -> Dic
                     "category": best_model["model_category"],
                     "quality": float(best_model[quality_col]),
                 }
-
+    
     return stats
 
 
 def perform_fair_comparison(
-    df: pd.DataFrame, model_names: List[str], shared_pairs_only: bool = True
+    df: pd.DataFrame, 
+    model_names: List[str], 
+    shared_pairs_only: bool = True
 ) -> Dict:
     """Perform fair comparison between models using only shared language pairs."""
-
+    
     models = df[df["model_name"].isin(model_names)]
-
+    
     if len(models) == 0:
         return {"error": "No models found"}
-
+    
     comparison = {
         "models": list(models["model_name"]),
         "fair_comparison_possible": True,
@@ -544,51 +520,47 @@ def perform_fair_comparison(
         "statistical_significance": {},
         "recommendations": [],
     }
-
+    
     # Check if fair comparison is possible
     categories = models["model_category"].unique()
     if len(categories) > 1:
         comparison["recommendations"].append(
             "⚠️ Comparing models from different categories - interpret results carefully"
         )
-
+    
     # For each track, compare models
     for track_name in EVALUATION_TRACKS.keys():
         quality_col = f"{track_name}_quality"
         adequate_col = f"{track_name}_adequate"
-
+        
         track_models = models[models[adequate_col] & (models[quality_col] > 0)]
-
+        
         if len(track_models) >= 2:
             comparison["track_comparisons"][track_name] = {
                 "participating_models": len(track_models),
-                "quality_scores": dict(
-                    zip(track_models["model_name"], track_models[quality_col])
-                ),
+                "quality_scores": dict(zip(track_models["model_name"], track_models[quality_col])),
                 "confidence_intervals": {},
             }
-
+            
             # Extract confidence intervals
             for _, model in track_models.iterrows():
                 ci_lower = model.get(f"{track_name}_ci_lower", 0)
                 ci_upper = model.get(f"{track_name}_ci_upper", 0)
-                comparison["track_comparisons"][track_name]["confidence_intervals"][
-                    model["model_name"]
-                ] = [ci_lower, ci_upper]
-
+                comparison["track_comparisons"][track_name]["confidence_intervals"][model["model_name"]] = [ci_lower, ci_upper]
+    
     return comparison
 
 
 def export_scientific_leaderboard(
-    df: pd.DataFrame,
-    track: str = "all",
-    format: str = "csv",
-    include_detailed: bool = False,
+    df: pd.DataFrame, 
+    track: str = "all", 
+    format: str = "csv", 
+    include_detailed: bool = False
 ) -> str:
     """Export scientific leaderboard in specified format."""
-
+    
     timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
-
+    
     if track != "all":
         # Export specific track
         export_df = prepare_track_leaderboard_display(df, track)
@@ -600,27 +572,22 @@ def export_scientific_leaderboard(
         else:
             # Select essential columns
             essential_columns = [
-                "model_name",
-                "author",
-                "submission_date",
-                "model_category",
-                "scientific_adequacy_score",
+                "model_name", "author", "submission_date", "model_category",
+                "scientific_adequacy_score"
             ]
-
+            
             # Add track-specific quality scores
             for track_name in EVALUATION_TRACKS.keys():
-                essential_columns.extend(
-                    [
-                        f"{track_name}_quality",
-                        f"{track_name}_adequate",
-                    ]
-                )
-
+                essential_columns.extend([
+                    f"{track_name}_quality",
+                    f"{track_name}_adequate",
+                ])
+            
             available_columns = [col for col in essential_columns if col in df.columns]
             export_df = df[available_columns].copy()
-
+        
         filename_prefix = f"salt_leaderboard_scientific_{timestamp}"
-
+    
     # Export in specified format
     if format == "csv":
         filename = f"{filename_prefix}.csv"
@@ -633,5 +600,5 @@ def export_scientific_leaderboard(
         export_df.to_excel(filename, index=False)
     else:
         raise ValueError(f"Unsupported format: {format}")
-
-    return filename
+    
+    return filename