Spaces:
Paused
Paused
| """ | |
| Dataset Evaluation Script - Ultimate Edition | |
| Evaluates system performance with detailed metrics | |
| """ | |
| import pandas as pd | |
| import time | |
| import sys | |
| from datetime import datetime | |
| from app import load_embedding_model, verify_titles, embedding_model | |
| def evaluate_dataset(dataset_path='data/large_dataset.csv', sample_size=100): | |
| """ | |
| Comprehensive evaluation on dataset | |
| Args: | |
| dataset_path: Path to CSV file | |
| sample_size: Number of samples to test (None for all) | |
| """ | |
| print("\n" + "="*70) | |
| print("📊 Cross-Lingual Title Verification - Ultimate Evaluation") | |
| print("="*70) | |
| # Load dataset | |
| print(f"\n📂 Loading dataset: {dataset_path}") | |
| try: | |
| df = pd.read_csv(dataset_path) | |
| print(f"✅ Loaded {len(df):,} rows") | |
| except Exception as e: | |
| print(f"❌ Error: {e}") | |
| return None | |
| # Validate columns | |
| required_columns = ['title_a', 'lang_a', 'title_b', 'lang_b', 'domain', 'label'] | |
| missing = [col for col in required_columns if col not in df.columns] | |
| if missing: | |
| print(f"❌ Missing columns: {missing}") | |
| return None | |
| # Sample if needed | |
| if sample_size and sample_size < len(df): | |
| df = df.sample(sample_size, random_state=42) | |
| print(f"📊 Using sample of {sample_size:,} rows") | |
| # Dataset info | |
| print(f"\n📋 Dataset Information:") | |
| print(f" Total pairs: {len(df):,}") | |
| print(f" Unique languages: {df['lang_a'].nunique() + df['lang_b'].nunique()}") | |
| print(f" Domains: {list(df['domain'].unique())}") | |
| print(f" Labels: {df['label'].value_counts().to_dict()}") | |
| # Load model | |
| print(f"\n🔧 Initializing system...") | |
| if embedding_model is None: | |
| load_embedding_model() | |
| print(f"✅ System ready") | |
| # Evaluate | |
| print(f"\n🔍 Evaluating {len(df):,} title pairs...") | |
| print("="*70) | |
| results = [] | |
| correct = 0 | |
| total = 0 | |
| tp, fp, tn, fn = 0, 0, 0, 0 | |
| # Language detection accuracy tracking | |
| lang_detection_correct = 0 | |
| lang_detection_total = 0 | |
| start_time = time.time() | |
| for idx, row in df.iterrows(): | |
| try: | |
| result = verify_titles( | |
| title_a=row['title_a'], | |
| title_b=row['title_b'], | |
| lang_a=row['lang_a'], | |
| lang_b=row['lang_b'], | |
| domain=row['domain'] | |
| ) | |
| predicted = result['label'] | |
| actual = row['label'] | |
| # Accuracy | |
| if predicted == actual: | |
| correct += 1 | |
| # Confusion matrix | |
| if actual == 'EQUIVALENT' and predicted == 'EQUIVALENT': | |
| tp += 1 | |
| elif actual == 'NOT_EQUIVALENT' and predicted == 'EQUIVALENT': | |
| fp += 1 | |
| elif actual == 'NOT_EQUIVALENT' and predicted == 'NOT_EQUIVALENT': | |
| tn += 1 | |
| elif actual == 'EQUIVALENT' and predicted == 'NOT_EQUIVALENT': | |
| fn += 1 | |
| # Check language detection accuracy | |
| detected_a = result['detected_languages']['title_a'] | |
| detected_b = result['detected_languages']['title_b'] | |
| if detected_a == row['lang_a']: | |
| lang_detection_correct += 1 | |
| if detected_b == row['lang_b']: | |
| lang_detection_correct += 1 | |
| lang_detection_total += 2 | |
| results.append({ | |
| 'title_a': row['title_a'][:50], | |
| 'title_b': row['title_b'][:50], | |
| 'actual': actual, | |
| 'predicted': predicted, | |
| 'correct': predicted == actual, | |
| 'score': result['final_score'] | |
| }) | |
| total += 1 | |
| # Progress | |
| if total % 10 == 0: | |
| progress = (total / len(df)) * 100 | |
| elapsed = time.time() - start_time | |
| speed = total / elapsed if elapsed > 0 else 0 | |
| print(f"Progress: {progress:5.1f}% | {total}/{len(df)} | " | |
| f"Accuracy: {(correct/total)*100:.1f}% | " | |
| f"Speed: {speed:.1f} pairs/sec", end='\r') | |
| except Exception as e: | |
| print(f"\n⚠️ Error at row {idx}: {e}") | |
| continue | |
| elapsed = time.time() - start_time | |
| # Compute metrics | |
| accuracy = correct / total if total > 0 else 0 | |
| precision = tp / (tp + fp) if (tp + fp) > 0 else 0 | |
| recall = tp / (tp + fn) if (tp + fn) > 0 else 0 | |
| f1 = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0 | |
| lang_accuracy = lang_detection_correct / lang_detection_total if lang_detection_total > 0 else 0 | |
| # Print results | |
| print("\n" + "="*70) | |
| print("📊 EVALUATION RESULTS") | |
| print("="*70) | |
| print(f"\n📈 Classification Performance:") | |
| print(f" {'Accuracy:':<20} {accuracy:.3f} ({correct:,}/{total:,})") | |
| print(f" {'Precision:':<20} {precision:.3f}") | |
| print(f" {'Recall:':<20} {recall:.3f}") | |
| print(f" {'F1-Score:':<20} {f1:.3f}") | |
| print(f"\n🌍 Language Detection:") | |
| print(f" {'Accuracy:':<20} {lang_accuracy:.3f} ({lang_detection_correct:,}/{lang_detection_total:,})") | |
| print(f"\n🔢 Confusion Matrix:") | |
| print(f" True Positives (TP): {tp:,}") | |
| print(f" False Positives (FP): {fp:,}") | |
| print(f" True Negatives (TN): {tn:,}") | |
| print(f" False Negatives (FN): {fn:,}") | |
| print(f"\n⏱️ Performance:") | |
| print(f" Total time: {elapsed:.1f} seconds") | |
| print(f" Average time: {elapsed/total:.3f} seconds/pair") | |
| print(f" Throughput: {total/elapsed:.1f} pairs/second") | |
| # Show errors if any | |
| if fp + fn > 0: | |
| print(f"\n❌ Sample Errors (first 5):") | |
| print("="*70) | |
| errors = [r for r in results if not r['correct']][:5] | |
| for i, err in enumerate(errors, 1): | |
| print(f"\n{i}. {err['actual']} → {err['predicted']} (score: {err['score']:.3f})") | |
| print(f" A: {err['title_a']}...") | |
| print(f" B: {err['title_b']}...") | |
| print("\n" + "="*70 + "\n") | |
| return { | |
| 'accuracy': accuracy, | |
| 'precision': precision, | |
| 'recall': recall, | |
| 'f1_score': f1, | |
| 'language_detection_accuracy': lang_accuracy, | |
| 'confusion_matrix': {'tp': tp, 'fp': fp, 'tn': tn, 'fn': fn}, | |
| 'performance': { | |
| 'total_time': elapsed, | |
| 'avg_time': elapsed / total, | |
| 'throughput': total / elapsed | |
| } | |
| } | |
| if __name__ == '__main__': | |
| import argparse | |
| parser = argparse.ArgumentParser(description='Evaluate cross-lingual verification') | |
| parser.add_argument('--dataset', default='data/large_dataset.csv', help='Dataset path') | |
| parser.add_argument('--sample', type=int, default=100, help='Sample size (0 for all)') | |
| args = parser.parse_args() | |
| sample_size = None if args.sample == 0 else args.sample | |
| metrics = evaluate_dataset(args.dataset, sample_size) | |
| if metrics: | |
| print("✅ Evaluation completed!") | |
| else: | |
| print("❌ Evaluation failed!") | |
| sys.exit(1) | |