run_baselines.py

#!/usr/bin/env python3
"""Baseline classifiers for human vs. LLM (and model attribution) on 200-token chunks.

Splits are done on document numbers (the same document number denotes the
same source prompt in every subcorpus; splitting on chunks would leak
topics between train and test):
  doc_num % 5 == 0 -> test, doc_num % 5 == 1 -> dev, rest -> train.

Tasks:
  binary       human vs. AI. The human class is ~1.6 % of chunks, so the
               default decision threshold is useless; the threshold on the
               SVM decision score is tuned on the dev set to maximize
               balanced accuracy, and threshold-free ROC-AUC is reported.
  attribution  human + one class per model line (temperatures merged;
               completion-mode variants kept separate from chat variants,
               since base-model output differs qualitatively)

Feature sets are TF-IDF vectorizers over different token columns.
Classifier: LinearSVC (liblinear, internal OVR for multiclass).
"""

import sys
import time
from functools import partial
from pathlib import Path

import numpy as np
import pandas as pd
from scipy.sparse import hstack
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import SGDClassifier
from sklearn.metrics import (accuracy_score, balanced_accuracy_score,
                             classification_report, confusion_matrix, f1_score,
                             roc_auc_score, roc_curve)
from sklearn.svm import LinearSVC

DATA_DIR = Path(__file__).parent / "data"
RESULTS_DIR = Path(__file__).parent / "results_v2"

N_FOLDS = 5  # doc_num % 5: 0 -> test, 1 -> dev, 2-4 -> train

# Base/completion-mode models mimic the human distribution too well and are
# not relevant for detecting assistant-style LLM text in real corpora
# (v2 decision, confirmed by Jiří 2026-07-06); they are excluded from both
# tasks.
EXCLUDE_COMPLETE = True


def model_class(row):
    """Collapse subcorpora into model-line classes for the attribution task."""
    if row["model"] == "human":
        return "human"
    name = row["model"]
    if row["mode"] == "complete":
        name += "-complete"
    return name


def skipgram_analyzer(text):
    """Skip-bigrams over pre-tokenized text: pairs with 1-3 tokens skipped.

    Plain bigrams are covered by the word vectorizers; here only the gapped
    pairs are produced, so the set is complementary.
    """
    toks = text.lower().split()
    feats = []
    for gap in (1, 2, 3):
        for i in range(len(toks) - gap - 1):
            feats.append(f"{toks[i]} <{gap}> {toks[i + gap + 1]}")
    return feats


def make_vectorizers(feature_set):
    """Return list of (column, TfidfVectorizer) for a named feature set."""
    # float32 halves matrix memory; the run was OOM-killed with float64
    Tfidf = partial(TfidfVectorizer, sublinear_tf=True, dtype=np.float32)
    word = ("words", Tfidf(ngram_range=(1, 2), min_df=3, lowercase=True))
    char = ("words", Tfidf(analyzer="char_wb", ngram_range=(3, 5),
                           min_df=3, max_features=300_000, lowercase=True))
    lemma = ("lemmata", Tfidf(ngram_range=(1, 2), min_df=3, lowercase=True))
    # long lemma n-grams: LLMs have favourite multi-word phrases (up to ~5
    # lemmata) that humans lack; min_df keeps the vocabulary tractable
    lemma15 = ("lemmata", Tfidf(ngram_range=(1, 5), min_df=5, lowercase=True))
    skip = ("words", Tfidf(analyzer=skipgram_analyzer, min_df=5))
    pos = ("pos", Tfidf(ngram_range=(1, 3), min_df=3, lowercase=False))
    tag = ("TAG", Tfidf(ngram_range=(1, 4), min_df=3, lowercase=False))
    fun = ("FUN", Tfidf(ngram_range=(1, 4), min_df=3, lowercase=False))
    sets = {
        "word12": [word],
        "char35": [char],
        "lemma12": [lemma],
        "lemma15": [lemma15],
        "skipgram": [skip],
        "pos13": [pos],
        "tag14": [tag],
        "fun14": [fun],
        "morphsyn": [pos, tag, fun],          # fully delexicalized
        "word+pos": [word, pos],
        "word+skip": [word, skip],            # surface-only: needs no lemmata,
                                              # usable with a plain tokenizer
        "lex+skip": [word, lemma15, skip],    # lexical phrases + skipgrams
        "skip+fun": [skip, fun],              # skipgrams + deprels
        "best": [word, lemma15, skip, fun],
        "full": [word, lemma15, tag, fun],
    }
    return sets[feature_set]


# liblinear multiclass on multi-million-feature sets was OOM-killed (>52 GB);
# for these attribution runs a single-pass SGD one-vs-all is used instead
SGD_ATTRIBUTION_SETS = {"lex+skip", "skip+fun", "best", "full"}


def make_classifier(task, feature_set):
    if task == "attribution" and feature_set in SGD_ATTRIBUTION_SETS:
        return SGDClassifier(loss="log_loss", alpha=1e-6,
                             class_weight="balanced", max_iter=30,
                             tol=1e-4, random_state=0)
    return LinearSVC(class_weight="balanced", C=1.0)


def featurize(df_train, df_evals, feature_set):
    """Fit vectorizers on train, transform train + each eval frame.

    Returns [X_train, X_eval1, X_eval2, ...].
    """
    specs = make_vectorizers(feature_set)
    parts = [[] for _ in range(1 + len(df_evals))]
    for col, vec in specs:
        parts[0].append(vec.fit_transform(df_train[col]))
        for i, df_eval in enumerate(df_evals, start=1):
            parts[i].append(vec.transform(df_eval[col]))
    return [p[0] if len(p) == 1 else hstack(p).tocsr() for p in parts]


def tune_threshold(y_dev, scores_dev):
    """Return the decision-score threshold maximizing balanced accuracy on dev."""
    fpr, tpr, thresholds = roc_curve(y_dev, scores_dev)
    balanced = (tpr + (1 - fpr)) / 2
    return thresholds[np.argmax(balanced)]


def run_experiment(df, lang, task, feature_set, out_rows):
    fold = df["doc_num"] % N_FOLDS
    train = df[fold >= 2]
    dev = df[fold == 1]
    test = df[fold == 0]

    if task == "binary":
        ytr = (train["model"] != "human").to_numpy()  # True = AI
        ydev = (dev["model"] != "human").to_numpy()
        yte = (test["model"] != "human").to_numpy()
    else:
        ytr = train.apply(model_class, axis=1).to_numpy()
        yte = test.apply(model_class, axis=1).to_numpy()

    t0 = time.time()
    Xtr, Xdev, Xte = featurize(train, [dev, test], feature_set)
    t_feat = time.time() - t0

    clf = make_classifier(task, feature_set)
    t0 = time.time()
    clf.fit(Xtr, ytr)
    t_fit = time.time() - t0

    tag = f"{lang}_{task}_{feature_set}"
    row = {"lang": lang, "task": task, "features": feature_set,
           "n_features": Xtr.shape[1], "n_train": len(train),
           "n_test": len(test)}

    if task == "binary":
        scores_dev = clf.decision_function(Xdev)
        scores_te = clf.decision_function(Xte)
        thr = tune_threshold(ydev, scores_dev)
        pred = scores_te >= thr
        row["roc_auc"] = roc_auc_score(yte, scores_te)
        row["threshold"] = thr
    else:
        pred = clf.predict(Xte)

    row["accuracy"] = accuracy_score(yte, pred)
    row["balanced_accuracy"] = balanced_accuracy_score(yte, pred)
    row["macro_f1"] = f1_score(yte, pred, average="macro")
    auc_str = f" AUC={row['roc_auc']:.4f}" if "roc_auc" in row else ""
    print(f"[{tag}] acc={row['accuracy']:.4f} "
          f"bal_acc={row['balanced_accuracy']:.4f} "
          f"macroF1={row['macro_f1']:.4f}{auc_str} "
          f"(feat {t_feat:.0f}s, fit {t_fit:.0f}s, {Xtr.shape[1]} features)",
          flush=True)
    out_rows.append(row)

    # per-subcorpus breakdown / confusion matrices on the test set
    detail = test[["subcorpus"]].copy()
    if task == "binary":
        detail["pred_ai"] = pred
        tab = (detail.groupby("subcorpus")["pred_ai"].agg(["mean", "count"])
               .rename(columns={"mean": "frac_predicted_ai", "count": "n_chunks"}))
        tab.to_csv(RESULTS_DIR / f"persubcorpus_{tag}.tsv", sep="\t")
    else:
        labels = sorted(set(yte) | set(pred))
        cm = confusion_matrix(yte, pred, labels=labels)
        pd.DataFrame(cm, index=labels, columns=labels).to_csv(
            RESULTS_DIR / f"confusion_{tag}.tsv", sep="\t")
        rep = classification_report(yte, pred, labels=labels,
                                    output_dict=True, zero_division=0)
        pd.DataFrame(rep).T.to_csv(RESULTS_DIR / f"report_{tag}.tsv", sep="\t")


def main():
    RESULTS_DIR.mkdir(exist_ok=True)
    binary_features = ["word12", "lemma15", "skipgram", "tag14", "fun14",
                       "morphsyn", "word+pos", "word+skip", "lex+skip",
                       "skip+fun", "best", "full"]
    attribution_features = ["word12", "morphsyn", "skip+fun", "lex+skip",
                            "best"]
    langs = sys.argv[1:] if len(sys.argv) > 1 else ["brown", "koditex"]
    summary_path = RESULTS_DIR / "summary.tsv"
    out_rows = (pd.read_csv(summary_path, sep="\t").to_dict("records")
                if summary_path.exists() else [])
    done = {(r["lang"], r["task"], r["features"]) for r in out_rows}
    for lang in langs:
        df = pd.read_pickle(DATA_DIR / f"chunks_{lang}.pkl")
        if EXCLUDE_COMPLETE:
            df = df[df["mode"] != "complete"]
        df = df.drop(columns=["UTAG"])  # unused, frees several GB
        print(f"=== {lang}: {len(df)} chunks, "
              f"{(df['model'] == 'human').sum()} human ===", flush=True)
        for task, feature_sets in [("binary", binary_features),
                                   ("attribution", attribution_features)]:
            for feature_set in feature_sets:
                if (lang, task, feature_set) in done:
                    print(f"[{lang}_{task}_{feature_set}] already in summary, "
                          f"skipping", flush=True)
                    continue
                run_experiment(df, lang, task, feature_set, out_rows)
                pd.DataFrame(out_rows).to_csv(summary_path, sep="\t",
                                              index=False)
    print("done")


if __name__ == "__main__":
    main()