|
|
import copy |
|
|
import pytest |
|
|
import numpy as np |
|
|
import pandas as pd |
|
|
from sklearn.feature_extraction.text import CountVectorizer |
|
|
|
|
|
|
|
|
@pytest.mark.parametrize('model', [('kmeans_pca_topic_model'), |
|
|
('base_topic_model'), |
|
|
('custom_topic_model'), |
|
|
('merged_topic_model'), |
|
|
('reduced_topic_model')]) |
|
|
def test_update_topics(model, documents, request): |
|
|
topic_model = copy.deepcopy(request.getfixturevalue(model)) |
|
|
old_ctfidf = topic_model.c_tf_idf_ |
|
|
old_topics = topic_model.topics_ |
|
|
|
|
|
topic_model.update_topics(documents, n_gram_range=(1, 3)) |
|
|
|
|
|
assert old_ctfidf.shape[1] < topic_model.c_tf_idf_.shape[1] |
|
|
assert old_topics == topic_model.topics_ |
|
|
|
|
|
updated_topics = [topic if topic != 1 else 0 for topic in old_topics] |
|
|
topic_model.update_topics(documents, topics=updated_topics, n_gram_range=(1, 3)) |
|
|
|
|
|
assert len(set(old_topics)) - 1 == len(set(topic_model.topics_)) |
|
|
|
|
|
old_topics = topic_model.topics_ |
|
|
updated_topics = [topic if topic != 2 else 0 for topic in old_topics] |
|
|
topic_model.update_topics(documents, topics=updated_topics, n_gram_range=(1, 3)) |
|
|
|
|
|
assert len(set(old_topics)) - 1 == len(set(topic_model.topics_)) |
|
|
|
|
|
|
|
|
@pytest.mark.parametrize('model', [('kmeans_pca_topic_model'), |
|
|
('base_topic_model'), |
|
|
('custom_topic_model'), |
|
|
('merged_topic_model'), |
|
|
('reduced_topic_model'), |
|
|
('online_topic_model')]) |
|
|
def test_extract_topics(model, documents, request): |
|
|
topic_model = copy.deepcopy(request.getfixturevalue(model)) |
|
|
nr_topics = 5 |
|
|
documents = pd.DataFrame({"Document": documents, |
|
|
"ID": range(len(documents)), |
|
|
"Topic": np.random.randint(-1, nr_topics-1, len(documents))}) |
|
|
topic_model._update_topic_size(documents) |
|
|
topic_model._extract_topics(documents) |
|
|
freq = topic_model.get_topic_freq() |
|
|
|
|
|
assert topic_model.c_tf_idf_.shape[0] == 5 |
|
|
assert topic_model.c_tf_idf_.shape[1] > 100 |
|
|
assert isinstance(freq, pd.DataFrame) |
|
|
assert nr_topics == len(freq.Topic.unique()) |
|
|
assert freq.Count.sum() == len(documents) |
|
|
assert len(freq.Topic.unique()) == len(freq) |
|
|
|
|
|
|
|
|
@pytest.mark.parametrize('model', [('kmeans_pca_topic_model'), |
|
|
('base_topic_model'), |
|
|
('custom_topic_model'), |
|
|
('merged_topic_model'), |
|
|
('reduced_topic_model'), |
|
|
('online_topic_model')]) |
|
|
def test_extract_topics_custom_cv(model, documents, request): |
|
|
topic_model = copy.deepcopy(request.getfixturevalue(model)) |
|
|
nr_topics = 5 |
|
|
documents = pd.DataFrame({"Document": documents, |
|
|
"ID": range(len(documents)), |
|
|
"Topic": np.random.randint(-1, nr_topics-1, len(documents))}) |
|
|
|
|
|
cv = CountVectorizer(ngram_range=(1, 2)) |
|
|
topic_model.vectorizer_model = cv |
|
|
topic_model._update_topic_size(documents) |
|
|
topic_model._extract_topics(documents) |
|
|
freq = topic_model.get_topic_freq() |
|
|
|
|
|
assert topic_model.c_tf_idf_.shape[0] == 5 |
|
|
assert topic_model.c_tf_idf_.shape[1] > 100 |
|
|
assert isinstance(freq, pd.DataFrame) |
|
|
assert nr_topics == len(freq.Topic.unique()) |
|
|
assert freq.Count.sum() == len(documents) |
|
|
assert len(freq.Topic.unique()) == len(freq) |
|
|
|
|
|
|
|
|
@pytest.mark.parametrize('model', [('kmeans_pca_topic_model'), |
|
|
('base_topic_model'), |
|
|
('custom_topic_model'), |
|
|
('merged_topic_model'), |
|
|
('reduced_topic_model'), |
|
|
('online_topic_model')]) |
|
|
@pytest.mark.parametrize("reduced_topics", [2, 4, 10]) |
|
|
def test_topic_reduction(model, reduced_topics, documents, request): |
|
|
topic_model = copy.deepcopy(request.getfixturevalue(model)) |
|
|
old_topics = copy.deepcopy(topic_model.topics_) |
|
|
old_freq = topic_model.get_topic_freq() |
|
|
|
|
|
topic_model.reduce_topics(documents, nr_topics=reduced_topics) |
|
|
|
|
|
new_freq = topic_model.get_topic_freq() |
|
|
|
|
|
if model != "online_topic_model": |
|
|
assert old_freq.Count.sum() == new_freq.Count.sum() |
|
|
assert len(old_freq.Topic.unique()) == len(old_freq) |
|
|
assert len(new_freq.Topic.unique()) == len(new_freq) |
|
|
assert len(topic_model.topics_) == len(old_topics) |
|
|
assert topic_model.topics_ != old_topics |
|
|
|
|
|
|
|
|
@pytest.mark.parametrize('model', [('kmeans_pca_topic_model'), |
|
|
('base_topic_model'), |
|
|
('custom_topic_model'), |
|
|
('merged_topic_model'), |
|
|
('reduced_topic_model'), |
|
|
('online_topic_model')]) |
|
|
def test_topic_reduction_edge_cases(model, documents, request): |
|
|
topic_model = copy.deepcopy(request.getfixturevalue(model)) |
|
|
topic_model.nr_topics = 100 |
|
|
nr_topics = 5 |
|
|
topics = np.random.randint(-1, nr_topics - 1, len(documents)) |
|
|
old_documents = pd.DataFrame({"Document": documents, |
|
|
"ID": range(len(documents)), |
|
|
"Topic": topics}) |
|
|
topic_model._update_topic_size(old_documents) |
|
|
topic_model._extract_topics(old_documents) |
|
|
old_freq = topic_model.get_topic_freq() |
|
|
|
|
|
new_documents = topic_model._reduce_topics(old_documents) |
|
|
new_freq = topic_model.get_topic_freq() |
|
|
|
|
|
assert not set(old_documents.Topic).difference(set(new_documents.Topic)) |
|
|
pd.testing.assert_frame_equal(old_documents, new_documents) |
|
|
pd.testing.assert_frame_equal(old_freq, new_freq) |
|
|
|
|
|
|
|
|
@pytest.mark.parametrize('model', [('kmeans_pca_topic_model'), |
|
|
('base_topic_model'), |
|
|
('custom_topic_model'), |
|
|
('merged_topic_model'), |
|
|
('reduced_topic_model'), |
|
|
('online_topic_model')]) |
|
|
def test_find_topics(model, request): |
|
|
topic_model = copy.deepcopy(request.getfixturevalue(model)) |
|
|
similar_topics, similarity = topic_model.find_topics("car") |
|
|
|
|
|
assert np.mean(similarity) > 0.1 |
|
|
assert len(similar_topics) > 0 |
|
|
|
