Upload 2 files

app.py

@@ -0,0 +1,498 @@
+import pandas as pd
+import numpy as np
+import warnings
+import feedparser
+from datetime import datetime
+
+warnings.filterwarnings('ignore')
+print("✅ Core imports done.")
+
+from datasets import load_dataset
+
+# ── Fallback generators (non-negotiable for reproducibility) ─────────────────
+def generate_fallback_liar():
+    """Synthetic LIAR-style dataset if HuggingFace load fails."""
+    data = [
+        ("The unemployment rate is at a 50-year low.", "half-true"),
+        ("Vaccines contain microchips for government tracking.", "pants-fire"),
+        ("Climate change is causing more frequent hurricanes.", "mostly-true"),
+        ("The stock market had its best year ever last year.", "false"),
+        ("Water covers about 71% of Earth's surface.", "true"),
+        ("The moon landing was filmed in a Hollywood studio.", "pants-fire"),
+        ("Eating carrots improves night vision significantly.", "barely-true"),
+        ("5G towers spread the COVID-19 virus.", "pants-fire"),
+        ("Exercise reduces the risk of type 2 diabetes.", "true"),
+        ("The Eiffel Tower grows taller in summer.", "mostly-true"),
+    ] * 50  # 500 samples
+    df = pd.DataFrame(data, columns=['statement', 'label'])
+    print("⚠️  Using synthetic LIAR fallback (500 samples).")
+    return df
+
+def generate_fallback_hallucination():
+    """Synthetic hallucination dataset if HuggingFace load fails."""
+    data = [
+        ("The Eiffel Tower is located in Berlin.", True),
+        ("Python was created by Guido van Rossum.", False),
+        ("Shakespeare wrote 'War and Peace'.", True),
+        ("The speed of light is approximately 3×10⁸ m/s.", False),
+        ("The Great Wall of China is visible from space with the naked eye.", True),
+    ] * 40
+    df = pd.DataFrame(data, columns=['claim', 'is_hallucination'])
+    print("⚠️  Using synthetic hallucination fallback (200 samples).")
+    return df
+
+# ── Load LIAR dataset ─────────────────────────────────────────────────────────
+try:
+    liar_raw = load_dataset("liar", trust_remote_code=True)
+    liar_df = pd.DataFrame({
+        'statement': liar_raw['train']['statement'],
+        'label': liar_raw['train']['label']
+    })
+    label_names = ['pants-fire','false','barely-true','half-true','mostly-true','true']
+    liar_df['label'] = liar_df['label'].apply(lambda x: label_names[x] if isinstance(x, int) else x)
+    print(f"✅ LIAR dataset loaded: {len(liar_df)} samples")
+except Exception as e:
+    print(f"LIAR load failed ({e}), using fallback.")
+    liar_df = generate_fallback_liar()
+
+# ── Load TruthfulQA ───────────────────────────────────────────────────────────
+try:
+    tqa_raw = load_dataset("truthful_qa", "generation", trust_remote_code=True)
+    tqa_df = pd.DataFrame({
+        'question': tqa_raw['validation']['question'],
+        'best_answer': tqa_raw['validation']['best_answer'],
+    })
+    print(f"✅ TruthfulQA loaded: {len(tqa_df)} samples")
+except Exception as e:
+    print(f"TruthfulQA load failed ({e}), using fallback.")
+    tqa_df = generate_fallback_hallucination()
+
+# ── Load HaluEval ─────────────────────────────────────────────────────────────
+try:
+    halu_raw = load_dataset("pminervini/HaluEval", "general_samples", trust_remote_code=True)
+    halu_df = pd.DataFrame(halu_raw['data'])
+    print(f"✅ HaluEval loaded: {len(halu_df)} samples")
+except Exception as e:
+    print(f"HaluEval load failed ({e}), using fallback.")
+    halu_df = generate_fallback_hallucination()
+
+print("\n📊 Dataset summary:")
+print(f"  LIAR:       {len(liar_df)} rows, columns: {list(liar_df.columns)}")
+print(f"  TruthfulQA: {len(tqa_df)} rows")
+print(f"  HaluEval:   {len(halu_df)} rows")
+
+# ── Live RSS News Feed ────────────────────────────────────────────────────────
+RSS_FEEDS = {
+    'BBC':     'http://feeds.bbci.co.uk/news/world/rss.xml',
+    'Reuters': 'https://feeds.reuters.com/reuters/topNews',
+    'AP':      'https://rsshub.app/apnews/topics/apf-topnews',
+}
+
+headlines = []
+for source, url in RSS_FEEDS.items():
+    try:
+        feed = feedparser.parse(url)
+        for entry in feed.entries[:10]:
+            pub = entry.get('published', str(datetime.now()))
+            headlines.append({
+                'headline': entry.get('title', ''),
+                'summary': entry.get('summary', ''),
+                'source': source,
+                'published_at': pub,
+                'link': entry.get('link', '')
+            })
+    except Exception as e:
+        print(f"  ⚠️  {source} RSS failed: {e}")
+
+if not headlines:
+    # Fallback static headlines for offline environments
+    headlines = [
+        {'headline': 'Global temperatures hit record highs in 2024', 'summary': '', 'source': 'synthetic', 'published_at': '2024-01-01', 'link': ''},
+        {'headline': 'AI models show improved reasoning capabilities', 'summary': '', 'source': 'synthetic', 'published_at': '2024-01-02', 'link': ''},
+        {'headline': 'New vaccine approved for respiratory illness', 'summary': '', 'source': 'synthetic', 'published_at': '2024-01-03', 'link': ''},
+    ] * 5
+    print("⚠️  Using synthetic headlines (no network access).")
+
+news_df = pd.DataFrame(headlines)
+news_df['published_at'] = pd.to_datetime(news_df['published_at'], errors='coerce', utc=True)
+print(f"✅ Live news loaded: {len(news_df)} headlines from {news_df['source'].nunique()} sources")
+news_df.head(3)
+
+from transformers import pipeline
+from sentence_transformers import SentenceTransformer
+import faiss
+import re
+
+# ── Load lightweight models ───────────────────────────────────────────────────
+print("Loading sentiment pipeline...")
+sentiment_pipeline = pipeline(
+    "sentiment-analysis",
+    model="distilbert-base-uncased-finetuned-sst-2-english",
+    truncation=True, max_length=512
+)
+
+print("Loading NLI pipeline (DeBERTa)...")
+nli_pipeline = pipeline(
+    "zero-shot-classification",
+    model="cross-encoder/nli-deberta-v3-small",
+    device=-1  # CPU
+)
+
+print("Loading sentence embedding model...")
+embedder = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
+
+print("✅ All models loaded.")
+
+# ── Build FAISS Vector Index of Trusted Facts ─────────────────────────────────
+TRUSTED_FACTS = [
+    "Water boils at 100 degrees Celsius at sea level.",
+    "The Earth orbits the Sun, not the other way around.",
+    "The speed of light in a vacuum is approximately 299,792 kilometers per second.",
+    "DNA carries genetic information in living organisms.",
+    "The Great Wall of China is not visible from space with the naked eye.",
+    "Humans and chimpanzees share approximately 98.7% of their DNA.",
+    "The moon is approximately 384,400 kilometers from Earth.",
+    "Mount Everest is the highest mountain above sea level at 8,849 meters.",
+    "Vaccines work by stimulating the immune system to recognize pathogens.",
+    "The human brain contains approximately 86 billion neurons.",
+    "Carbon dioxide concentration in the atmosphere has increased since industrialization.",
+    "The Eiffel Tower is located in Paris, France.",
+    "Python was created by Guido van Rossum and first released in 1991.",
+    "Shakespeare wrote Hamlet, Macbeth, and Romeo and Juliet.",
+    "The United States has 50 states.",
+    "Albert Einstein published the special theory of relativity in 1905.",
+    "Antibiotics are not effective against viral infections.",
+    "The Pacific Ocean is the largest ocean on Earth.",
+    "The human body has 206 bones in adulthood.",
+    "Climate change is driven primarily by human greenhouse gas emissions according to scientific consensus.",
+]
+
+# Encode and index
+fact_embeddings = embedder.encode(TRUSTED_FACTS, convert_to_numpy=True)
+dim = fact_embeddings.shape[1]
+faiss_index = faiss.IndexFlatL2(dim)
+faiss_index.add(fact_embeddings.astype(np.float32))
+
+print(f"✅ FAISS index built with {faiss_index.ntotal} trusted facts (dim={dim})")
+
+# ── Feature Extraction Functions ──────────────────────────────────────────────
+
+SOURCE_CREDIBILITY = {
+    'bbc.co.uk': 0.92, 'reuters.com': 0.94, 'apnews.com': 0.93,
+    'nytimes.com': 0.88, 'theguardian.com': 0.87, 'nature.com': 0.98,
+    'who.int': 0.97, 'cdc.gov': 0.97, 'infowars.com': 0.05,
+    'naturalnews.com': 0.08, 'breitbart.com': 0.22, 'synthetic': 0.50,
+    'BBC': 0.92, 'Reuters': 0.94, 'AP': 0.93,
+}
+
+FAKE_DOI_PATTERN = re.compile(
+    r'10\.\d{4,}/[a-zA-Z0-9./_-]+'
+)
+IMPOSSIBLE_YEAR = re.compile(r'\b(19[0-2]\d|2[1-9]\d{2})\b')
+INVENTED_INSTITUTIONS = re.compile(
+    r'(Institute of [A-Z][a-z]+ [A-Z][a-z]+|Foundation for [A-Z][a-z]+ Research)',
+    re.IGNORECASE
+)
+
+def get_sentiment_score(text: str) -> float:
+    """Returns float in [-1, 1]. Negative = negative sentiment."""
+    try:
+        result = sentiment_pipeline(text[:512])[0]
+        score = result['score']
+        return score if result['label'] == 'POSITIVE' else -score
+    except:
+        return 0.0
+
+def get_source_credibility(source: str) -> float:
+    """Lookup against known domain credibility scores."""
+    for domain, score in SOURCE_CREDIBILITY.items():
+        if domain.lower() in source.lower():
+            return score
+    return 0.5  # unknown source → uncertain
+
+def get_citation_anomaly_score(text: str) -> float:
+    """Detects patterns common in hallucinated citations."""
+    score = 0.0
+    # Fake DOI pattern
+    if FAKE_DOI_PATTERN.search(text): score += 0.3
+    # Impossible year references
+    if IMPOSSIBLE_YEAR.search(text): score += 0.3
+    # Suspicious institution names
+    if INVENTED_INSTITUTIONS.search(text): score += 0.4
+    return min(score, 1.0)
+
+def get_semantic_similarity(text: str, k: int = 3) -> float:
+    """Cosine similarity of input against top-k trusted FAISS facts."""
+    try:
+        emb = embedder.encode([text], convert_to_numpy=True).astype(np.float32)
+        distances, _ = faiss_index.search(emb, k)
+        # Convert L2 distance to similarity (lower distance = higher similarity)
+        avg_dist = np.mean(distances[0])
+        similarity = 1.0 / (1.0 + avg_dist)
+        return float(np.clip(similarity, 0, 1))
+    except:
+        return 0.5
+
+def get_nli_contradiction_score(claim: str, references: list) -> float:
+    """DeBERTa NLI: fraction of references that contradict the claim."""
+    try:
+        result = nli_pipeline(
+            claim,
+            candidate_labels=["entailment", "neutral", "contradiction"],
+            hypothesis_template="This claim is related to: {}",
+        )
+        # Get contradiction score
+        scores = dict(zip(result['labels'], result['scores']))
+        return float(scores.get('contradiction', 0.0))
+    except:
+        return 0.5
+
+def retrieve_reference_sentences(claim: str, k: int = 5) -> list:
+    """Retrieve top-k relevant facts from FAISS index."""
+    try:
+        emb = embedder.encode([claim], convert_to_numpy=True).astype(np.float32)
+        _, indices = faiss_index.search(emb, k)
+        return [TRUSTED_FACTS[i] for i in indices[0] if i < len(TRUSTED_FACTS)]
+    except:
+        return TRUSTED_FACTS[:k]
+
+print("✅ Feature extraction functions defined.")
+
+# ── Compute Features on a Sample ──────────────────────────────────────────────
+SAMPLE_TEXTS = [
+    "The moon is made of cheese.",
+    "Water boils at 100°C at sea level.",
+    "Scientists discovered that 5G towers emit mind-control frequencies.",
+    "The Eiffel Tower is 330 meters tall.",
+    "According to a 2031 study from the Institute of Neural Enhancement, humans only use 10% of their brain.",
+]
+
+rows = []
+for text in SAMPLE_TEXTS:
+    refs = retrieve_reference_sentences(text)
+    row = {
+        'text': text[:60] + '...' if len(text) > 60 else text,
+        'sentiment_score': get_sentiment_score(text),
+        'source_credibility': 0.5,  # unknown source for these samples
+        'nli_contradiction_score': get_nli_contradiction_score(text, refs),
+        'citation_anomaly_score': get_citation_anomaly_score(text),
+        'semantic_similarity': get_semantic_similarity(text),
+    }
+    rows.append(row)
+
+features_df = pd.DataFrame(rows)
+print("✅ Feature matrix computed:")
+features_df
+
+# ── A. Fake News Classifier (LIAR → 3-class) ──────────────────────────────────
+from sklearn.linear_model import LogisticRegression
+from sklearn.preprocessing import LabelEncoder
+from sklearn.model_selection import train_test_split
+from sklearn.metrics import classification_report
+import numpy as np
+
+# Collapse LIAR 6-class to 3-class
+LIAR_MAP = {
+    'pants-fire': 'misinformation',
+    'false': 'misinformation',
+    'barely-true': 'uncertain',
+    'half-true': 'uncertain',
+    'mostly-true': 'credible',
+    'true': 'credible',
+}
+
+liar_sample = liar_df.sample(min(500, len(liar_df)), random_state=42).copy()
+liar_sample['label_3'] = liar_sample['label'].map(LIAR_MAP).fillna('uncertain')
+
+# Encode statements → embeddings for classifier
+print("Encoding LIAR statements...")
+X_liar = embedder.encode(liar_sample['statement'].tolist(), show_progress_bar=True)
+y_liar = liar_sample['label_3'].values
+
+X_train, X_test, y_train, y_test = train_test_split(X_liar, y_liar, test_size=0.2, random_state=42)
+
+fake_news_clf = LogisticRegression(max_iter=500, random_state=42)
+fake_news_clf.fit(X_train, y_train)
+
+print("\n📊 Fake News Classifier Report:")
+print(classification_report(y_test, fake_news_clf.predict(X_test)))
+print("✅ Fake news classifier trained.")
+
+# ── B. Hallucination Scorer ───────────────────────────────────────────────────
+
+def score_hallucination(claim: str) -> dict:
+    """
+    Scores a single claim for hallucination risk.
+    Returns dict with hallucination_risk [0-100] and evidence snippets.
+    """
+    try:
+        references = retrieve_reference_sentences(claim, k=5)
+        contradiction_score = get_nli_contradiction_score(claim, references)
+        similarity = get_semantic_similarity(claim)
+        citation_anomaly = get_citation_anomaly_score(claim)
+
+        # Weighted combination
+        raw_risk = (
+            0.50 * contradiction_score +
+            0.30 * (1 - similarity) +      # low similarity to trusted facts = higher risk
+            0.20 * citation_anomaly
+        )
+        hallucination_risk = int(np.clip(raw_risk * 100, 0, 100))
+
+        return {
+            'hallucination_risk': hallucination_risk,
+            'contradiction_score': round(contradiction_score, 3),
+            'semantic_similarity': round(similarity, 3),
+            'evidence_snippets': references[:3]
+        }
+    except Exception as e:
+        return {'hallucination_risk': 50, 'contradiction_score': 0.5,
+                'semantic_similarity': 0.5, 'evidence_snippets': [], 'error': str(e)}
+
+# Test
+test_claims = [
+    "The moon is made of cheese.",
+    "Water boils at 100 degrees Celsius at sea level.",
+]
+for claim in test_claims:
+    result = score_hallucination(claim)
+    print(f"  '{claim[:50]}...' → risk: {result['hallucination_risk']}%")
+print("✅ Hallucination scorer working.")
+
+# ── C. Event Volatility Forecaster ───────────────────────────────────────────
+try:
+    from statsforecast import StatsForecast
+    from statsforecast.models import AutoARIMA
+    HAS_STATSFORECAST = True
+except ImportError:
+    HAS_STATSFORECAST = False
+    print("⚠️  statsforecast not available, using EWMA fallback.")
+
+def compute_volatility_series(df: pd.DataFrame, window: int = 7) -> pd.Series:
+    """Rolling std of sentiment scores over headlines."""
+    df = df.copy().sort_values('published_at')
+    sentiments = df['headline'].apply(get_sentiment_score)
+    volatility = sentiments.rolling(window=min(window, len(df)), min_periods=1).std().fillna(0)
+    return volatility
+
+def forecast_volatility(series: pd.Series, horizon: int = 3) -> dict:
+    """Forecast next `horizon` periods of volatility."""
+    if HAS_STATSFORECAST and len(series) >= 10:
+        try:
+            sf_df = pd.DataFrame({
+                'unique_id': 'news_vol',
+                'ds': pd.date_range(start='2024-01-01', periods=len(series), freq='D'),
+                'y': series.values
+            })
+            sf = StatsForecast(models=[AutoARIMA()], freq='D')
+            forecast = sf.forecast(df=sf_df, h=horizon)
+            forecasted_values = forecast['AutoARIMA'].values.tolist()
+            trend = 'rising' if forecasted_values[-1] > series.mean() else 'stable'
+            return {'method': 'AutoARIMA', 'forecast': forecasted_values, 'trend': trend}
+        except:
+            pass
+
+    # EWMA fallback
+    ewma = series.ewm(span=min(5, len(series))).mean()
+    last = ewma.iloc[-1]
+    forecasted = [last * (1 + 0.02 * i) for i in range(1, horizon + 1)]
+    trend = 'rising' if forecasted[-1] > series.mean() else 'stable'
+    return {'method': 'EWMA', 'forecast': forecasted, 'trend': trend}
+
+volatility_series = compute_volatility_series(news_df)
+forecast_result = forecast_volatility(volatility_series)
+print(f"✅ Volatility forecast: {forecast_result['method']} → trend: {forecast_result['trend']}")
+
+# ── D. Final Risk Score Aggregator ────────────────────────────────────────────
+# Configurable weights (adjust these constants)
+W_HALLUCINATION  = 0.40
+W_FAKE_NEWS      = 0.35
+W_CITATION       = 0.15
+W_SIMILARITY     = 0.10
+
+COLOR_MAP = {
+    'confirmed':     'rgba(52, 199, 89, 0.15)',    # green
+    'uncertain':     'rgba(255, 204, 0, 0.15)',    # yellow
+    'misinformation':'rgba(255, 59, 48, 0.15)',    # red
+    'hallucination': 'rgba(175, 82, 222, 0.15)',   # purple
+}
+
+def get_fake_news_probability(text: str) -> tuple[str, float]:
+    """Returns (label, probability) from fake news classifier."""
+    try:
+        emb = embedder.encode([text])
+        proba = fake_news_clf.predict_proba(emb)[0]
+        classes = fake_news_clf.classes_
+        label = classes[np.argmax(proba)]
+        confidence = float(np.max(proba))
+        return label, confidence
+    except:
+        return 'uncertain', 0.5
+
+def analyze_text(text: str, source: str = 'unknown') -> dict:
+    """
+    Full pipeline: text → JSON risk payload.
+    This is the function the Gradio API exposes.
+    """
+    try:
+        # --- feature extraction ---
+        halu_result  = score_hallucination(text)
+        fake_label, fake_conf = get_fake_news_probability(text)
+        citation_score = get_citation_anomaly_score(text)
+        similarity     = get_semantic_similarity(text)
+        credibility    = get_source_credibility(source)
+
+        # Normalise fake news label to a risk score
+        fake_risk = {'misinformation': 0.9, 'uncertain': 0.5, 'credible': 0.1}.get(fake_label, 0.5)
+
+        # Aggregate
+        combined_risk = (
+            W_HALLUCINATION * (halu_result['hallucination_risk'] / 100) +
+            W_FAKE_NEWS     * fake_risk +
+            W_CITATION      * citation_score +
+            W_SIMILARITY    * (1 - similarity)
+        )
+        combined_risk = float(np.clip(combined_risk, 0, 1))
+
+        # Determine status
+        if combined_risk < 0.25:
+            status = 'confirmed'
+        elif combined_risk < 0.55:
+            status = 'uncertain'
+        elif halu_result['hallucination_risk'] > 60:
+            status = 'hallucination'
+        else:
+            status = 'misinformation'
+
+        confidence = abs(combined_risk - 0.5) * 2  # distance from uncertain midpoint
+
+        tooltip = (
+            f"{status.title()} risk: {int(combined_risk*100)}%. "
+            f"Hallucination: {halu_result['hallucination_risk']}%. "
+            f"Source credibility: {int(credibility*100)}%."
+        )
+
+        return {
+            'text': text,
+            'status': status,
+            'color': COLOR_MAP[status],
+            'hallucination_risk': halu_result['hallucination_risk'],
+            'fake_news_label': fake_label,
+            'combined_risk': round(combined_risk, 3),
+            'confidence': round(confidence, 3),
+            'volatility_index': round(1 - similarity, 3),
+            'tooltip_message': tooltip,
+            'evidence_snippets': halu_result['evidence_snippets']
+        }
+    except Exception as e:
+        return {
+            'text': text, 'status': 'uncertain', 'color': COLOR_MAP['uncertain'],
+            'hallucination_risk': 50, 'fake_news_label': 'uncertain',
+            'combined_risk': 0.5, 'confidence': 0.0, 'volatility_index': 0.5,
+            'tooltip_message': f'Analysis failed gracefully: {str(e)}',
+            'evidence_snippets': []
+        }
+
+# Quick smoke test
+test = analyze_text("The moon is made of cheese.")
+print(f"✅ Aggregator test: status={test['status']}, risk={test['combined_risk']}")

requirements.txt

@@ -0,0 +1,8 @@
+transformers
+datasets
+sentence-transformers
+faiss-cpu
+gradio
+statsforecast
+feedparser
+scikit-learn