Simple example of full document classification and questions about skweak HOT 2 CLOSED

import spacy
from skweak import heuristics, aggregation
from sklearn.pipeline import Pipeline
from sklearn.feature_extraction.text import CountVectorizer,TfidfTransformer
from sklearn.linear_model import LogisticRegressionCV,PassiveAggressiveClassifier
from sklearn.svm import NuSVC
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import f1_score,accuracy_score
from afinn import Afinn
import random
from tensorflow.keras.datasets import imdb

nlp = spacy.load("en_core_web_sm")
afinn = Afinn(language='en')

# get IMDB sentiment data
(training_data, training_targets), (testing_data, testing_targets) = imdb.load_data(num_words=10000)
ind2text = {x:k for k,x in imdb.get_word_index().items()}
# convert to text
n_train = 500 # train samples
n_test = 200 # test samples
get_text = lambda data: [" ".join([ind2text[x] for x in d]) for d in data]
X_train,Y_train = get_text(training_data[0:n_train]),list(training_targets[0:n_train])
X_test,Y_test = get_text(testing_data[0:n_test]),list(testing_targets[0:n_test])

# create some whole-text classical classifiers
my_classifiers=[]
my_classifiers.append(Pipeline([
     ('vect', CountVectorizer(ngram_range=(1,2),max_features=500)),
     ('tfidf', TfidfTransformer()),
     ('clf', LogisticRegressionCV(penalty="l1",cv=5,solver='liblinear')),
]))
my_classifiers.append(Pipeline([
     ('vect', CountVectorizer(ngram_range=(1,2),max_features=500)),
     ('tfidf', TfidfTransformer()),
     ('clf', NuSVC()),
]))
my_classifiers.append(Pipeline([
     ('vect', CountVectorizer(ngram_range=(1,2),max_features=500)),
     ('tfidf', TfidfTransformer()),
     ('clf', PassiveAggressiveClassifier()),
]))
my_classifiers.append(Pipeline([
     ('vect', CountVectorizer(ngram_range=(1,2),max_features=500)),
     ('tfidf', TfidfTransformer(use_idf=False)),
     ('clf', RandomForestClassifier()),
]))

all_annotators =skweak.base.CombinedAnnotator() # labeling functions

# Annotator based on a sklearn classification pipeline
class SklearnAnnotator(skweak.base.SpanAnnotator):
    def __init__(self, name, trained_model):
        skweak.base.SpanAnnotator.__init__(self, name)
        self.trained_model = trained_model
        
    def find_spans(self, doc):
        yield 0, len(doc), str(self.trained_model.predict([doc.text])[0])
        
# add afinn annotator (span over whole document)
def labeling_fun_afinn(x):
    yield 0, len(x),('0' if afinn.score(x.text)<=0 else '1')
all_annotators.add_annotator(heuristics.FunctionAnnotator("afinn", lambda x: labeling_fun_afinn(x)))

# apply predictor (span over whole document)
def labeling_fun_clf(x,model):
    yield 0, len(x),str(model.predict([x.text])[0])
for i,clf in enumerate(my_classifiers):
    clf.fit(X_train,Y_train)
    all_annotators.add_annotator(SklearnAnnotator("classifier_%i" % i, clf))

# obtain processed docs
train_docs = list(all_annotators.pipe(nlp.pipe(X_train)))
test_docs = list(all_annotators.pipe(nlp.pipe(X_test)))

# create HMM aggregator
hmm = aggregation.HMM("hmm", ['0','1'],sequence_labelling=False)

# fit and annotate train data
hmm.fit(train_docs)
# apply model to test data (works as a "predict" function?)
test_docs = list(hmm.pipe(test_docs))

# get predicted classes
hmm_preds = [int(doc.spans["hmm"][0].label_) for doc in test_docs]
afinn_preds = [int(afinn.score(doc.text) >=0) for doc in test_docs]

print("\nResults")
print(" skweak HMM: F1=%f, accuracy=%f" % (f1_score(Y_test,hmm_preds),accuracy_score(Y_test,hmm_preds)))
print(" afinn: F1=%f, accuracy=%f" % (f1_score(Y_test,afinn_preds), accuracy_score(Y_test,afinn_preds)))
for i,clf in enumerate(my_classifiers):
    y = clf.predict(X_test)
    print(" classifier %i: F1=%f, accuracy=%f" % (i+1,f1_score(Y_test,y),accuracy_score(Y_test,y)))