German Sentiment Analysis with BERT: nlptown vs oliverguhr on 20 Real Sentences

{
  "text": "Nicht schlecht, aber auch nicht besonders gut.",
  "expected": "neutral",
  "category": "negation"
},
{
  "text": "Der Support war ja wirklich der Hammer, drei Wochen keine Antwort auf meine Anfrage.",
  "expected": "negative",
  "category": "sarcasm"
},
{
  "text": "Also ehrlich, des Ding is a Schrott, funktioniert von Anfang an net richtig.",
  "expected": "negative",
  "category": "colloquial"
}

[
  {
    "id": 1,
    "text": "Der Kundenservice war schnell und hat mein Problem sofort gelöst.",
    "expected": "positive",
    "category": "clear_positive"
  },
  {
    "id": 2,
    "text": "Ich bin sehr zufrieden mit der Lieferung, alles kam pünktlich und unbeschädigt an.",
    "expected": "positive",
    "category": "clear_positive"
  },
  {
    "id": 3,
    "text": "Endlich mal ein Produkt, das genauso funktioniert wie beschrieben. Top!",
    "expected": "positive",
    "category": "clear_positive"
  },
  {
    "id": 4,
    "text": "Das Gerät ist nach zwei Wochen kaputt gegangen und der Support antwortet nicht.",
    "expected": "negative",
    "category": "clear_negative"
  },
  {
    "id": 5,
    "text": "Absolute Zeitverschwendung, ich fordere eine vollständige Rückerstattung.",
    "expected": "negative",
    "category": "clear_negative"
  },
  {
    "id": 6,
    "text": "Die Verpackung war beschädigt und der Inhalt ist zerbrochen angekommen.",
    "expected": "negative",
    "category": "clear_negative"
  },
  {
    "id": 7,
    "text": "Die Bestellung wurde am Dienstag aufgegeben und am Donnerstag versendet.",
    "expected": "neutral",
    "category": "clear_neutral"
  },
  {
    "id": 8,
    "text": "Ich habe eine Frage zur Rechnungsstellung für meinen letzten Einkauf.",
    "expected": "neutral",
    "category": "clear_neutral"
  },
  {
    "id": 9,
    "text": "Können Sie mir bitte mitteilen, wann das Produkt wieder verfügbar ist?",
    "expected": "neutral",
    "category": "clear_neutral"
  },
  {
    "id": 10,
    "text": "Nicht schlecht, aber auch nicht besonders gut.",
    "expected": "neutral",
    "category": "negation"
  },
  {
    "id": 11,
    "text": "Das Produkt ist nicht so schlecht, wie ich erwartet hatte.",
    "expected": "positive",
    "category": "negation"
  },
  {
    "id": 12,
    "text": "Ich kann nicht behaupten, dass ich begeistert bin.",
    "expected": "negative",
    "category": "negation"
  },
  {
    "id": 13,
    "text": "Der Support war ja wirklich der Hammer, drei Wochen keine Antwort auf meine Anfrage.",
    "expected": "negative",
    "category": "sarcasm"
  },
  {
    "id": 14,
    "text": "Toll, das dritte Ersatzgerät und wieder ein Defekt. Genau das, was ich wollte.",
    "expected": "negative",
    "category": "sarcasm"
  },
  {
    "id": 15,
    "text": "Wow, eine Wartezeit von 45 Minuten in der Hotline, das nenne ich mal Kundenservice.",
    "expected": "negative",
    "category": "sarcasm"
  },
  {
    "id": 16,
    "text": "Die Lieferung war super schnell, aber die Qualität hat mich leider enttäuscht.",
    "expected": "mixed",
    "category": "mixed_sentiment"
  },
  {
    "id": 17,
    "text": "Der Mitarbeiter war sehr freundlich, konnte mein Problem aber nicht lösen.",
    "expected": "mixed",
    "category": "mixed_sentiment"
  },
  {
    "id": 18,
    "text": "Guter Preis, aber die Verarbeitung wirkt ziemlich billig.",
    "expected": "mixed",
    "category": "mixed_sentiment"
  },
  {
    "id": 19,
    "text": "Also ehrlich, des Ding is a Schrott, funktioniert von Anfang an net richtig.",
    "expected": "negative",
    "category": "colloquial"
  },
  {
    "id": 20,
    "text": "Läuft bei euch, echt top gemacht, weiter so!",
    "expected": "positive",
    "category": "colloquial"
  }
]

from transformers import pipeline

NLPTOWN_MODEL = "nlptown/bert-base-multilingual-uncased-sentiment"
OLIVERGUHR_MODEL = "oliverguhr/german-sentiment-bert"

def map_nlptown_stars_to_label(raw_label: str) -> str:
    stars = int(raw_label[0])
    if stars <= 2:
        return "negative"
    if stars == 3:
        return "neutral"
    return "positive"

nlptown_pipe = pipeline("sentiment-analysis", model=NLPTOWN_MODEL)
oliverguhr_pipe = pipeline("sentiment-analysis", model=OLIVERGUHR_MODEL)

for item in eval_set:
    nt_raw = nlptown_pipe(item["text"])[0]
    og_raw = oliverguhr_pipe(item["text"])[0]
    nt_label = map_nlptown_stars_to_label(nt_raw["label"])
    og_label = og_raw["label"].lower()
    # compare nt_label, og_label against item["expected"]

"""
German sentiment analysis: head-to-head comparison of two HuggingFace models.

Models compared:
  - nlptown/bert-base-multilingual-uncased-sentiment
    Multilingual, outputs 1-5 stars. Mapped to labels as:
    1-2 stars -> negative, 3 -> neutral, 4-5 -> positive.
  - oliverguhr/german-sentiment-bert
    German-specific, outputs positive/negative/neutral natively.

Scoring notes:
  - Results are reported as raw counts ("9/13 correct"), not percentages,
    since the eval set is small (~20 sentences) and percentages would
    imply more precision than the sample size supports.
  - Neutral-labeled sentences are scored separately from positive/negative,
    because oliverguhr predicts "neutral" natively while nlptown only
    reaches it via the star mapping above.
  - "Mixed" sentiment sentences are reported qualitatively only (not
    scored correct/incorrect), since neither model has a "mixed" label.

Usage:
    python sentiment_comparison.py
"""

import json
import sys
import time
from pathlib import Path

from transformers import pipeline

if sys.platform == "win32":
    sys.stdout.reconfigure(encoding="utf-8")

EVAL_SET_PATH = Path(__file__).parent / "german_sentiment_eval_set.json"
RESULTS_PATH = Path(__file__).parent / "results_table.md"

NLPTOWN_MODEL = "nlptown/bert-base-multilingual-uncased-sentiment"
OLIVERGUHR_MODEL = "oliverguhr/german-sentiment-bert"

NLPTOWN_LABEL = "nlptown (multilingual)"
OLIVERGUHR_LABEL = "oliverguhr (German-specific)"


def map_nlptown_stars_to_label(raw_label: str) -> str:
    """Fixed mapping, decided before any run and never adjusted afterward."""
    stars = int(raw_label[0])
    if stars <= 2:
        return "negative"
    if stars == 3:
        return "neutral"
    return "positive"


def load_eval_set() -> list[dict]:
    with open(EVAL_SET_PATH, encoding="utf-8") as f:
        return json.load(f)


def run_model(pipe, sentences: list[dict]) -> list[dict]:
    """Run a pipeline over all sentences, returning per-sentence raw output + latency."""
    results = []
    for item in sentences:
        start = time.perf_counter()
        raw = pipe(item["text"])[0]
        elapsed = time.perf_counter() - start
        results.append({"raw_label": raw["label"], "score": raw["score"], "latency": elapsed})
    return results


def build_rows(eval_set: list[dict], nlptown_raw: list[dict], oliverguhr_raw: list[dict]) -> list[dict]:
    rows = []
    for item, nt, og in zip(eval_set, nlptown_raw, oliverguhr_raw):
        nt_mapped = map_nlptown_stars_to_label(nt["raw_label"])
        og_label = og["raw_label"].lower()
        rows.append({
            "id": item["id"],
            "text": item["text"],
            "category": item["category"],
            "expected": item["expected"],
            "nlptown_mapped": nt_mapped,
            "nlptown_latency": nt["latency"],
            "oliverguhr_label": og_label,
            "oliverguhr_latency": og["latency"],
            "models_agree": nt_mapped == og_label,
        })
    return rows


def print_legend():
    print("Legend:")
    print(f"  nlptown    = {NLPTOWN_MODEL} (multilingual, stars mapped to labels)")
    print(f"  oliverguhr = {OLIVERGUHR_MODEL} (German-specific, native labels)")
    print()


def print_results_table(rows: list[dict]):
    print(f"{'ID':<3} {'Category':<16} {'Expected':<9} {NLPTOWN_LABEL:<22} {OLIVERGUHR_LABEL:<28} {'Agree?':<7}")
    print("-" * 95)
    for r in rows:
        agree = "yes" if r["models_agree"] else "no"
        print(f"{r['id']:<3} {r['category']:<16} {r['expected']:<9} "
              f"{r['nlptown_mapped']:<22} {r['oliverguhr_label']:<28} {agree:<7}")


def compute_summary(rows: list[dict]) -> dict:
    posneg_rows = [r for r in rows if r["expected"] in ("positive", "negative")]
    neutral_rows = [r for r in rows if r["expected"] == "neutral"]
    mixed_rows = [r for r in rows if r["expected"] == "mixed"]

    return {
        "posneg_total": len(posneg_rows),
        "nlptown_posneg_correct": sum(1 for r in posneg_rows if r["nlptown_mapped"] == r["expected"]),
        "oliverguhr_posneg_correct": sum(1 for r in posneg_rows if r["oliverguhr_label"] == r["expected"]),
        "neutral_total": len(neutral_rows),
        "nlptown_neutral_correct": sum(1 for r in neutral_rows if r["nlptown_mapped"] == r["expected"]),
        "oliverguhr_neutral_correct": sum(1 for r in neutral_rows if r["oliverguhr_label"] == r["expected"]),
        "mixed_rows": mixed_rows,
        "agreement_count": sum(1 for r in rows if r["models_agree"]),
        "nlptown_avg_latency_ms": sum(r["nlptown_latency"] for r in rows) / len(rows) * 1000,
        "oliverguhr_avg_latency_ms": sum(r["oliverguhr_latency"] for r in rows) / len(rows) * 1000,
    }


def print_summary(rows: list[dict], summary: dict):
    print("\n--- Summary ---")

    print(f"Positive/negative accuracy ({summary['posneg_total']} sentences):")
    print(f"  {NLPTOWN_LABEL}: {summary['nlptown_posneg_correct']}/{summary['posneg_total']} correct")
    print(f"  {OLIVERGUHR_LABEL}: {summary['oliverguhr_posneg_correct']}/{summary['posneg_total']} correct")

    print(f"\nNeutral accuracy ({summary['neutral_total']} sentences, reported separately):")
    print(f"  {NLPTOWN_LABEL}: {summary['nlptown_neutral_correct']}/{summary['neutral_total']} correct")
    print(f"  {OLIVERGUHR_LABEL}: {summary['oliverguhr_neutral_correct']}/{summary['neutral_total']} correct")

    print(f"\nAgreement rate: {summary['agreement_count']}/{len(rows)} sentences")

    print("\nAvg latency:")
    print(f"  {NLPTOWN_LABEL}: {summary['nlptown_avg_latency_ms']:.1f} ms")
    print(f"  {OLIVERGUHR_LABEL}: {summary['oliverguhr_avg_latency_ms']:.1f} ms")

    mixed_rows = summary["mixed_rows"]
    print(f"\n--- Mixed-sentiment cases ({len(mixed_rows)}, qualitative only — no model has a 'mixed' label) ---")
    for r in mixed_rows:
        print(f'  [{r["id"]}] "{r["text"]}"')
        print(f"      {NLPTOWN_LABEL}: {r['nlptown_mapped']}  |  {OLIVERGUHR_LABEL}: {r['oliverguhr_label']}")

    disagreements = [r for r in rows if not r["models_agree"]]
    print(f"\n--- Disagreement cases ({len(disagreements)}) ---")
    for r in disagreements:
        print(f'  [{r["id"]}] "{r["text"]}"')
        print(f"      expected: {r['expected']}  |  {NLPTOWN_LABEL}: {r['nlptown_mapped']}  |  "
              f"{OLIVERGUHR_LABEL}: {r['oliverguhr_label']}")


def write_markdown_table(rows: list[dict]):
    lines = [
        f"**nlptown** = `{NLPTOWN_MODEL}` (multilingual, stars mapped to labels)  ",
        f"**oliverguhr** = `{OLIVERGUHR_MODEL}` (German-specific, native labels)",
        "",
        "| ID | Category | Expected | nlptown (multilingual) | oliverguhr (German-specific) | Agree? |",
        "|---|---|---|---|---|---|",
    ]
    for r in rows:
        agree_icon = "✅" if r["models_agree"] else "❌"
        lines.append(
            f"| {r['id']} | {r['category']} | {r['expected']} | {r['nlptown_mapped']} | "
            f"{r['oliverguhr_label']} | {agree_icon} |"
        )

    RESULTS_PATH.write_text("\n".join(lines), encoding="utf-8")
    print(f"\nMarkdown results table written to {RESULTS_PATH}")


def main():
    eval_set = load_eval_set()

    print("Loading models...")
    nlptown_pipe = pipeline("sentiment-analysis", model=NLPTOWN_MODEL)
    oliverguhr_pipe = pipeline("sentiment-analysis", model=OLIVERGUHR_MODEL)

    print(f"Running {len(eval_set)} sentences through both models...\n")
    nlptown_raw = run_model(nlptown_pipe, eval_set)
    oliverguhr_raw = run_model(oliverguhr_pipe, eval_set)

    rows = build_rows(eval_set, nlptown_raw, oliverguhr_raw)

    print_legend()
    print_results_table(rows)

    summary = compute_summary(rows)
    print_summary(rows, summary)

    write_markdown_table(rows)


if __name__ == "__main__":
    main()