Corpus linguistic analysis with NLTK and spaCy: tokenization, collocation, keyword analysis, KWIC concordance, distributional semantics with word2vec, and diachronic frequency trends.
Corpus linguistic analysis with NLTK and spaCy: tokenization, collocation, keyword analysis, KWIC concordance, distributional semantics with word2vec, and diachronic frequency trends.
"""
Tokenize, lemmatize, and optionally clean a list of texts using spaCy.
Parameters
----------
texts : list of str
Raw text documents.
lang : str
spaCy model key (``"en"``, ``"de"``, ``"fr"``, ``"es"``).
lowercase : bool
Lowercase all tokens.
remove_punct : bool
Remove punctuation tokens.
remove_stopwords : bool
Remove function words.
lemmatize : bool
Return lemmas instead of surface forms.
min_token_len : int
Minimum token character length.
batch_size : int
spaCy pipe batch size.
n_process : int
Number of processes for spaCy pipe (1 = single-threaded).
Returns
-------
list of list of str
Tokenized (and optionally lemmatized) documents.
"""
# Disable unused pipes for speed
"ner"
"parser"
if
"ner"
in
else
set
if
else
set
with
for
in
for
in
if
and
or
continue
if
else
if
if
and
in
continue
if
len
continue
return
def
get_pos_tagged_corpus
texts: list[str],
lang: str = "en",
list
list
tuple
str
str
"""
Return POS-tagged tokens as (token, POS_tag) pairs using spaCy Universal POS tags.
"""
"""
Extract named entities across a corpus.
Parameters
----------
entity_types : list of str, optional
Filter to specific types: ``["PERSON", "ORG", "GPE", "DATE"]`` etc.
Returns
-------
pd.DataFrame with columns: doc_idx, text, label, start_char, end_char.
"""
"""
Compute unigram and n-gram frequency distributions plus hapax legomena.
Parameters
----------
tokenized_corpus : list of list of str
Output of ``preprocess_corpus``.
n : int
Maximum n-gram order to compute (1 = unigrams only, 2 = + bigrams, etc.).
Returns
-------
dict with keys ``"unigrams"``, ``"bigrams"`` (if n≥2), ``"trigrams"`` (if n≥3),
``"hapax_legomena"``, ``"type_token_ratio"``, ``"total_tokens"``, ``"vocabulary_size"``.
"""
corpus: list[list[str]],
node_word: str | None = None,
window: int = 5,
stat: str = "pmi",
top_n: int = 30,
min_freq: int = 5,
"""
Compute collocations using pointwise mutual information, log-likelihood, or t-score.
Parameters
----------
corpus : list of list of str
Tokenized corpus.
node_word : str, optional
Focus word. If None, compute the top collocating pairs in the whole corpus.
window : int
Context window size (tokens on each side).
stat : str
Association measure: ``"pmi"``, ``"ll"`` (log-likelihood), ``"tscore"``, ``"raw_freq"``.
top_n : int
Number of top collocates to return.
min_freq : int
Minimum joint frequency filter.
Returns
-------
pd.DataFrame with columns: w1, w2, score, joint_freq.
"""
target_corpus: list[list[str]],
reference_corpus: list[list[str]],
top_n: int = 50,
min_freq: int = 5,
effect_size: bool = True,
"""
Identify statistically key words in a target corpus compared to a reference corpus.
Uses the log-likelihood (G²) statistic (Dunning 1993), which is robust for frequency
comparison across corpora of different sizes.
Parameters
----------
target_corpus : list of list of str
Tokenized target corpus.
reference_corpus : list of list of str
Tokenized reference corpus.
top_n : int
Number of keywords to return (positive keywords only).
min_freq : int
Minimum frequency in target corpus.
effect_size : bool
If True, compute %DIFF and log ratio effect sizes.
Returns
-------
pd.DataFrame sorted by G² descending.
"""
"""
Generate a Key Word In Context (KWIC) concordance.
Parameters
----------
corpus : list of list of str
Tokenized corpus.
keyword : str
Search keyword (exact token match after lowercasing if not case_sensitive).
window : int
Number of context tokens on each side.
case_sensitive : bool
If False, match regardless of case.
max_lines : int
Maximum concordance lines to return.
doc_labels : list of str, optional
Document identifiers for the ``doc_id`` column.
Returns
-------
pd.DataFrame with columns: doc_id, position, left_context, node, right_context.
"""
sentences: list[list[str]],
vector_size: int = 200,
window: int = 5,
min_count: int = 5,
workers: int = 4,
sg: int = 1,
epochs: int = 10,
detect_phrases: bool = True,
"""
Train a Word2Vec model on a tokenized corpus.
Parameters
----------
sentences : list of list of str
Tokenized corpus (one document per inner list).
vector_size : int
Dimensionality of word vectors.
window : int
Context window size.
min_count : int
Ignore words with frequency below this threshold.
workers : int
Number of CPU threads.
sg : int
Training algorithm: 1 = Skip-gram, 0 = CBOW.
epochs : int
Training epochs.
detect_phrases : bool
If True, detect and join common bigram phrases (e.g. "New_York").
Returns
-------
gensim.models.Word2Vec
"""
"""Compute cosine similarity for a list of word pairs."""
for
in
if
not
in
or
not
in
float
"nan"
else
"word1"
"word2"
"cosine_similarity"
round
4
return
Example A: Diachronic Lexical Change in a News Corpus
This example tracks how the normalized frequency of selected words changes over years in a
collection of news articles organized by publication year. It also computes keyword profiles
for individual decades.
# ── Example A ─────────────────────────────────────────────────────────────import os
import glob
import json
# --- Expected corpus structure -----------------------------------------------# news_corpus/# 1990/ article_001.txt article_002.txt ...# 2000/ ...# 2010/ ...# 2020/ ...
CORPUS_ROOT = os.environ.get("NEWS_CORPUS_ROOT", "news_corpus")
TARGET_WORDS = ["internet", "climate", "terrorism", "pandemic", "artificial", "algorithm"]
defload_corpus_by_year(root: str) -> dict[int, list[str]]:
"""Load text files organized in year subdirectories."""
corpus_by_year: dict[int, list[str]] = {}
for year_dir insorted(Path(root).iterdir()):
ifnot year_dir.is_dir():
continuetry:
year = int(year_dir.name)
except ValueError:
continue
texts = []
for txt_file in year_dir.glob("*.txt"):
texts.append(txt_file.read_text(encoding="utf-8", errors="ignore"))
if texts:
corpus_by_year[year] = texts
return corpus_by_year
# --- Load and process -------------------------------------------------------print("Loading corpus...")
corpus_by_year = load_corpus_by_year(CORPUS_ROOT)
years_available = sorted(corpus_by_year.keys())
print(f"Years available: {years_available}")
year_freqs: dict[int, dict] = {}
year_tokens: dict[int, list[list[str]]] = {}
for year in years_available:
print(f"Processing {year}...")
tokenized = preprocess_corpus(
corpus_by_year[year],
lang="en",
lowercase=True,
remove_punct=True,
remove_stopwords=False, # Keep stopwords for full frequency profile
lemmatize=True,
)
year_tokens[year] = tokenized
freq_profile = compute_frequency_profile(tokenized, n=1)
year_freqs[year] = freq_profile
# --- Normalized frequency trends (per million words) ------------------------
records = []
for year in years_available:
total = year_freqs[year]["total_tokens"]
unigrams = year_freqs[year]["unigrams"]
for word in TARGET_WORDS:
freq = unigrams.get(word, 0)
ppm = freq / total * 1_000_000if total > 0else0.0
records.append({"year": year, "word": word, "freq": freq, "per_million": ppm})
freq_df = pd.DataFrame(records)
# --- Plot frequency trends --------------------------------------------------
fig, ax = plt.subplots(figsize=(12, 6))
palette = plt.cm.tab10(np.linspace(0, 1, len(TARGET_WORDS)))
for word, color inzip(TARGET_WORDS, palette):
wdata = freq_df[freq_df["word"] == word].sort_values("year")
ax.plot(wdata["year"], wdata["per_million"], marker="o", label=word, color=color, linewidth=2)
ax.set_xlabel("Year", fontsize=12)
ax.set_ylabel("Frequency per Million Words", fontsize=12)
ax.set_title("Lexical Frequency Trends in News Corpus", fontsize=14)
ax.legend(loc="upper left", fontsize=9, ncol=2)
ax.grid(True, alpha=0.3)
ax.yaxis.set_major_formatter(mticker.FuncFormatter(lambda x, _: f"{x:,.0f}"))
fig.tight_layout()
plt.savefig("diachronic_frequency_trends.png", dpi=150)
plt.show()
# --- Keyword analysis: 2020s vs 1990s ----------------------------------------if2020in year_tokens and1990in year_tokens:
target_texts = list(itertools.chain.from_iterable(
year_tokens[y] for y in years_available if y >= 2020
))
reference_texts = list(itertools.chain.from_iterable(
year_tokens[y] for y in years_available if1990 <= y < 2000
))
# Wrap flat token lists as single-document corpora
keywords_df = keyword_analysis(
target_corpus=[target_texts],
reference_corpus=[reference_texts],
top_n=30,
min_freq=10,
)
print("\nTop keywords (2020s vs 1990s reference):")
print(keywords_df[["word", "target_per_mil", "ref_per_mil", "G2", "log_ratio"]].to_string())
# --- Collocation profile of "climate" over time ----------------------------for year in [1990, 2000, 2010, 2020]:
if year notin year_tokens:
continue
coll_df = compute_collocations(
year_tokens[year],
node_word="climate",
window=4,
stat="pmi",
top_n=10,
min_freq=3,
)
print(f"\nTop collocates of 'climate' in {year}:")
print(coll_df[["w1", "w2", "score"]].to_string(index=False))
Example B: Collocation Profile of Academic Vocabulary Across Disciplines
This example loads academic corpora for different disciplines, identifies the top collocates
of shared academic vocabulary (high-frequency multi-disciplinary terms), and builds word2vec
models per discipline to compare semantic neighbourhoods.
# ── Example B ─────────────────────────────────────────────────────────────# Input: one text file per discipline# HUMANITIES_TXT, SCIENCES_TXT, SOCIAL_SCIENCES_TXT (paths via env vars)
DISCIPLINE_FILES = {
"humanities": os.environ.get("HUMANITIES_TXT", "humanities_corpus.txt"),
"sciences": os.environ.get("SCIENCES_TXT", "sciences_corpus.txt"),
"social_sciences": os.environ.get("SOCIAL_SCIENCES_TXT", "social_sciences_corpus.txt"),
}
ACADEMIC_VOCAB = [
"approach", "analysis", "framework", "evidence",
"significant", "model", "theory", "context",
]
discipline_corpora: dict[str, list[list[str]]] = {}
discipline_models: dict[str, Word2Vec] = {}
for discipline, filepath in DISCIPLINE_FILES.items():
ifnot os.path.exists(filepath):
print(f"Skipping {discipline}: file not found.")
continueprint(f"\n--- {discipline.upper()} ---")
raw_text = Path(filepath).read_text(encoding="utf-8", errors="ignore")
# Split into ~sentence-length chunks for Word2Vec training
sentences = [s.strip() for s in re.split(r"[.!?]\s+", raw_text) iflen(s.split()) > 5]
tokenized = preprocess_corpus(
sentences,
lang="en",
lowercase=True,
remove_punct=True,
remove_stopwords=False,
lemmatize=True,
)
discipline_corpora[discipline] = tokenized
# Frequency profile
freq_profile = compute_frequency_profile(tokenized, n=2)
print(f"Total tokens: {freq_profile['total_tokens']:,}")
print(f"Vocabulary: {freq_profile['vocabulary_size']:,}")
# Train Word2Vec
model = train_word2vec(
tokenized,
vector_size=150,
window=5,
min_count=5,
sg=1,
epochs=15,
detect_phrases=True,
)
discipline_models[discipline] = model
# --- Collocation comparison: 'analysis' across disciplines ------------------print("\n=== Collocation profiles: 'analysis' ===")
for discipline, tokenized in discipline_corpora.items():
coll = compute_collocations(
tokenized,
node_word="analysis",
window=3,
stat="pmi",
top_n=10,
min_freq=5,
)
print(f"\n[{discipline}]")
print(coll[["w1", "w2", "score"]].to_string(index=False))
# --- Semantic similarity: cosine similarity between ACADEMIC_VOCAB words ----print("\n=== Semantic Similarity (sciences vs humanities) ===")
word_pairs = list(itertools.combinations(ACADEMIC_VOCAB, 2))
if"sciences"in discipline_models and"humanities"in discipline_models:
sim_sci = compute_cosine_similarity(discipline_models["sciences"], word_pairs)
sim_hum = compute_cosine_similarity(discipline_models["humanities"], word_pairs)
comparison = sim_sci.merge(
sim_hum,
on=["word1", "word2"],
suffixes=("_sci", "_hum"),
)
comparison["delta"] = (comparison["cosine_similarity_sci"] - comparison["cosine_similarity_hum"]).abs()
print(comparison.sort_values("delta", ascending=False).head(15).to_string(index=False))
# --- Nearest neighbors of "theory" in each discipline ----------------------
TARGET_WORD = "theory"print(f"\n=== Nearest Neighbours of '{TARGET_WORD}' ===")
for discipline, model in discipline_models.items():
if TARGET_WORD in model.wv:
neighbors = model.wv.most_similar(TARGET_WORD, topn=8)
words = ", ".join(f"{w}({s:.2f})"for w, s in neighbors)
print(f"[{discipline}]: {words}")
# --- KWIC concordance for 'framework' in social sciences --------------------if"social_sciences"in discipline_corpora:
kwic_df = kwic(
discipline_corpora["social_sciences"],
keyword="framework",
window=4,
max_lines=20,
)
print("\n=== KWIC: 'framework' in Social Sciences ===")
for _, row in kwic_df.iterrows():
print(f"... {row['left_context']} [{row['node']}] {row['right_context']} ...")
# --- Keyword analysis: Sciences vs Humanities --------------------------------if"sciences"in discipline_corpora and"humanities"in discipline_corpora:
kw_df = keyword_analysis(
target_corpus=discipline_corpora["sciences"],
reference_corpus=discipline_corpora["humanities"],
top_n=25,
min_freq=10,
)
print("\n=== Science Keywords (vs Humanities reference) ===")
print(kw_df[["word", "target_per_mil", "ref_per_mil", "G2", "log_ratio"]].head(20).to_string())
N-Gram Language Models
For probabilistic language modeling and text generation:
from nltk.lm import MLE, Laplace, KneserNeyInterpolated
from nltk.lm.preprocessing import padded_everygram_pipeline
deftrain_ngram_lm(
tokenized_corpus: list[list[str]],
n: int = 3,
model_type: str = "kneser_ney",
) -> tuple:
"""
Train an n-gram language model.
Parameters
----------
tokenized_corpus : list of list of str
Training corpus.
n : int
N-gram order (e.g. 3 = trigram).
model_type : str
``"mle"``, ``"laplace"``, or ``"kneser_ney"``.
Returns
-------
tuple: (fitted model, vocabulary)
"""
train_data, padded_sents = padded_everygram_pipeline(n, tokenized_corpus)
if model_type == "mle":
lm = MLE(n)
elif model_type == "laplace":
lm = Laplace(n)
else:
lm = KneserNeyInterpolated(n)
lm.fit(train_data, padded_sents)
print(f"LM trained: {len(lm.vocab)} vocabulary, order={n}, type={model_type}")
return lm, lm.vocab
defcompute_perplexity(
lm,
test_corpus: list[list[str]],
n: int,
) -> float:
"""
Compute perplexity of the LM on a test corpus.
Lower perplexity = better fit.
"""from nltk.lm.preprocessing import padded_everygram_pipeline
test_data, _ = padded_everygram_pipeline(n, test_corpus)
total_log_prob = 0.0
total_count = 0for sent in test_corpus:
for ngram in ngrams(["<s>"] * (n - 1) + sent + ["</s>"], n):
lp = lm.logscore(ngram[-1], context=ngram[:-1])
ifnot math.isinf(lp):
total_log_prob += lp
total_count += 1if total_count == 0:
returnfloat("inf")
avg_log_prob = total_log_prob / total_count
return2 ** (-avg_log_prob)
Notes and Best Practices
spaCy Model Selection
Use Case
Model
Speed-critical pipeline
en_core_web_sm
Better NER / accuracy
en_core_web_lg
Transformer-based
en_core_web_trf (requires spacy-transformers)
Log-Likelihood vs PMI
Log-likelihood (G²): Best for keyword analysis and low-frequency phenomena; not sensitive
to corpus size imbalance.
PMI: Tends to overestimate associations for rare words; use with min_freq >= 5.
T-score: Conservative; best for identifying grammatically stable collocations.
Word2Vec Practical Tips
Preprocess consistently before training: lowercase, remove punctuation, optionally lemmatize.
Use sg=1 (Skip-gram) for rare words; sg=0 (CBOW) is faster and works well for frequent words.
vector_size=200 is a good default; increase to 300 for large corpora (>50 M tokens).
Save/load models with model.save("model.bin") / Word2Vec.load("model.bin").
References
Biber, D., Conrad, S., & Reppen, R. (1998). Corpus Linguistics: Investigating Language
Structure and Use. Cambridge University Press.
Dunning, T. (1993). Accurate methods for the statistics of surprise and coincidence.
Computational Linguistics, 19(1), 61–74.
Mikolov, T. et al. (2013). Efficient estimation of word representations in vector space.
arXiv:1301.3781.
Rayson, P. (2008). From key words to key semantic domains. International Journal of Corpus
Linguistics, 13(4), 519–549.
