File size: 4,434 Bytes
667fe9d
 
2c1f9dd
 
667fe9d
a092d54
85ac990
a092d54
667fe9d
a092d54
667fe9d
 
a092d54
2c1f9dd
85ac990
2c1f9dd
 
 
 
 
 
 
a092d54
667fe9d
85ac990
a092d54
667fe9d
85ac990
a092d54
85ac990
a092d54
667fe9d
 
85ac990
 
 
204391c
667fe9d
85ac990
667fe9d
85ac990
 
 
2c1f9dd
667fe9d
85ac990
 
 
 
667fe9d
204391c
a092d54
 
 
 
 
2c1f9dd
 
a092d54
 
 
204391c
0ca5366
667fe9d
85ac990
204391c
667fe9d
 
 
85ac990
a092d54
2c1f9dd
85ac990
2c1f9dd
85ac990
2c1f9dd
a092d54
85ac990
 
 
 
2c1f9dd
85ac990
2c1f9dd
85ac990
2c1f9dd
85ac990
 
a092d54
85ac990
 
2c1f9dd
85ac990
 
 
667fe9d
 
a092d54
 
2c1f9dd
a092d54
 
 
 
 
 
2c1f9dd
a092d54
 
 
2c1f9dd
a092d54
 
2c1f9dd
 
 
667fe9d
a092d54
 
5a2db0a
 
 
2c1f9dd
 
a092d54
 
2c1f9dd
5a2db0a
 
 
 
 
2c1f9dd
a092d54
 
2c1f9dd
5a2db0a
 
 
 
2c1f9dd
5a2db0a
 
2c1f9dd
a092d54
 
5a2db0a
2c1f9dd
 
5a2db0a
2c1f9dd
5a2db0a
2c1f9dd
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
from __future__ import annotations

import os
from typing import TYPE_CHECKING

import numpy as np
from joblib import Memory
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import RandomizedSearchCV, cross_val_score, train_test_split
from sklearn.pipeline import Pipeline

from app.constants import CACHE_DIR
from app.data import tokenize

if TYPE_CHECKING:
    from sklearn.base import BaseEstimator

__all__ = ["create_model", "train_model", "evaluate_model", "infer_model"]


def _identity(x: list[str]) -> list[str]:
    """Identity function for use in TfidfVectorizer.

    Args:
        x: Input data

    Returns:
        Unchanged input data
    """
    return x


def create_model(
    max_features: int,
    seed: int | None = None,
    verbose: bool = False,
) -> Pipeline:
    """Create a sentiment analysis model.

    Args:
        max_features: Maximum number of features
        seed: Random seed (None for random seed)
        verbose: Whether to output additional information

    Returns:
        Untrained model
    """
    return Pipeline(
        [
            (
                "vectorizer",
                TfidfVectorizer(
                    max_features=max_features,
                    ngram_range=(1, 2),
                    # disable text processing
                    tokenizer=_identity,
                    preprocessor=_identity,
                    lowercase=False,
                    token_pattern=None,
                ),
            ),
            ("classifier", LogisticRegression(max_iter=1000, random_state=seed)),
        ],
        memory=Memory(CACHE_DIR, verbose=0),
        verbose=verbose,
    )


def train_model(
    model: BaseEstimator,
    token_data: list[str],
    label_data: list[int],
    folds: int = 5,
    seed: int = 42,
    verbose: bool = False,
) -> tuple[BaseEstimator, float]:
    """Train the sentiment analysis model.

    Args:
        model: Untrained model
        token_data: Tokenized text data
        label_data: Label data
        folds: Number of cross-validation folds
        seed: Random seed (None for random seed)
        verbose: Whether to output additional information

    Returns:
        Trained model and accuracy
    """
    text_train, text_test, label_train, label_test = train_test_split(
        token_data,
        label_data,
        test_size=0.2,
        random_state=seed,
    )

    param_distributions = {
        "classifier__C": np.logspace(-4, 4, 20),
        "classifier__solver": ["liblinear", "saga"],
    }

    search = RandomizedSearchCV(
        model,
        param_distributions,
        n_iter=10,
        cv=folds,
        scoring="accuracy",
        random_state=seed,
        n_jobs=-1,
        verbose=verbose,
    )

    os.environ["PYTHONWARNINGS"] = "ignore"
    search.fit(text_train, label_train)
    del os.environ["PYTHONWARNINGS"]

    best_model = search.best_estimator_
    return best_model, best_model.score(text_test, label_test)


def evaluate_model(
    model: BaseEstimator,
    token_data: list[str],
    label_data: list[int],
    folds: int = 5,
    verbose: bool = False,
) -> tuple[float, float]:
    """Evaluate the model using cross-validation.

    Args:
        model: Trained model
        token_data: Tokenized text data
        label_data: Label data
        folds: Number of cross-validation folds
        verbose: Whether to output additional information

    Returns:
        Mean accuracy and standard deviation
    """
    os.environ["PYTHONWARNINGS"] = "ignore"
    scores = cross_val_score(
        model,
        token_data,
        label_data,
        cv=folds,
        scoring="accuracy",
        n_jobs=-1,
        verbose=verbose,
    )
    del os.environ["PYTHONWARNINGS"]
    return scores.mean(), scores.std()


def infer_model(
    model: BaseEstimator,
    text_data: list[str],
    batch_size: int = 32,
    n_jobs: int = 4,
) -> list[int]:
    """Predict the sentiment of the provided text documents.

    Args:
        model: Trained model
        text_data: Text data
        batch_size: Batch size for tokenization
        n_jobs: Number of parallel jobs

    Returns:
        Predicted sentiments
    """
    tokens = tokenize(
        text_data,
        batch_size=batch_size,
        n_jobs=n_jobs,
        show_progress=False,
    )
    return model.predict(tokens)