Skip to content

Z-Test¤

ZTest pydantic-model ¤

Bases: TopWords

Pydantic model for performing z-tests.

Config:

  • arbitrary_types_allowed: True

Fields:

Source code in lexos/topwords/ztest.py 
 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
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
class ZTest(TopWords):
    """Pydantic model for performing z-tests."""

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
    )
    target_docs: str | Doc | list[str | Doc] = Field(
        ...,
        description="List of target documents.",
    )
    comparison_docs: str | Doc | list[str | Doc] = Field(
        ...,
        description="List of comparison documents.",
    )
    topn: int = Field(10, gt=0, description="Number of top words to return.")
    case_sensitive: bool | None = Field(
        True, description="Whether analysis is case sensitive."
    )
    remove_stopwords: bool | None = Field(
        True, description="Whether to remove stopwords."
    )
    remove_punct: bool | None = Field(
        True, description="Whether to remove punctuation."
    )
    remove_digits: bool | None = Field(False, description="Whether to remove digits.")
    ngrams: int | tuple[int, int] = Field(
        default=1,
        description="The ngram range for analysis, e.g., (1, 2) for unigrams and bigrams.",
    )
    model: str = Field(
        default="xx_sent_ud_sm",
        description="spaCy model name to use for tokenization.",
    )
    tokenizer: Tokenizer = Field(default_factory=Tokenizer, exclude=True)
    topwords: list[tuple[str, float]] = Field(
        default_factory=list,
        description="List of top words with their Z-scores.",
    )

    def __init__(self, **data) -> None:
        """Initialize the ZTest class, ensuring a tokenizer is set.

        If a tokenizer is not provided, creates one using the specified spaCy model.
        """
        # If tokenizer is not provided, create one with the specified model
        if "tokenizer" not in data or data["tokenizer"] is None:
            data["tokenizer"] = Tokenizer(model=data.get("model", "xx_sent_ud_sm"))
        super().__init__(**data)

        # Tokenize the docs if they are strings
        self.target_docs = [
            self.tokenizer.make_doc(doc) if isinstance(doc, str) else doc
            for doc in ensure_list(self.target_docs)
        ]
        self.comparison_docs = [
            self.tokenizer.make_doc(doc) if isinstance(doc, str) else doc
            for doc in ensure_list(self.comparison_docs)
        ]

        # Get the term counts for target and comparison
        target_counts: Counter = self._get_doc_term_counts(self.target_docs)
        comparison_counts: Counter = self._get_doc_term_counts(self.comparison_docs)

        # Total number of tokens in target and comparison
        target_total: int = sum(target_counts.values())
        comparison_total: int = sum(comparison_counts.values())

        # Calculate Z-scores with vectorized NumPy operations.
        terms = list(set(target_counts) | set(comparison_counts))
        tc_arr = np.array([target_counts[t] for t in terms], dtype=np.float64)
        cc_arr = np.array([comparison_counts[t] for t in terms], dtype=np.float64)

        n1, n2 = target_total, comparison_total
        total = n1 + n2

        if n1 > 0 and n2 > 0 and total > 0:
            p1 = tc_arr / n1
            p2 = cc_arr / n2
            p = (tc_arr + cc_arr) / total
            se = np.sqrt(p * (1.0 - p) * (1.0 / n1 + 1.0 / n2))
            valid = (se > 0) & (p > 0) & (p < 1)
            z_scores = np.where(valid, (p1 - p2) / np.where(se > 0, se, 1.0), 0.0)
        else:
            z_scores = np.zeros(len(terms), dtype=np.float64)

        nonzero = z_scores != 0.0
        sorted_results = sorted(
            zip(np.array(terms, dtype=object)[nonzero], z_scores[nonzero].tolist()),
            key=lambda item: abs(item[1]),
            reverse=True,
        )

        # Assign the top N results
        self.topwords = sorted_results[: self.topn]

        # Set the topwords attribute on each document
        for doc in self.target_docs:
            doc._.topwords = self.topwords
        for doc in self.comparison_docs:
            doc._.topwords = self.topwords

    def _filter_doc(self, doc: Doc) -> list[str]:
        """Filter tokens from a document based on preprocessing settings.

        Args:
            doc (Doc): A spaCy Doc object.

        Returns:
            list[str]: List of filtered token texts.
        """
        filtered_tokens = []
        for token in doc:
            if self.remove_stopwords and token.is_stop:
                continue
            if self.remove_punct and token.is_punct:
                continue
            if self.remove_digits and token.is_digit:
                continue
            if not self.case_sensitive:
                token_text = token.text.lower()
            else:
                token_text = token.text
            filtered_tokens.append(token_text)
        return filtered_tokens

    def _get_doc_tokens(self, docs: list[Doc]) -> list[str]:
        """Get tokens from a document after applying preprocessing.

        Args:
            docs (list[Doc]): List of spaCy Doc objects.

        Returns:
            list[str]: List of processed tokens from the documents.
        """
        # Get the ngram range
        if isinstance(self.ngrams, int):
            ngram_range = ensure_list(self.ngrams)
        else:
            ngram_range = self.ngrams

        tokens = []
        for doc in ensure_list(docs):
            # Handle single tokens
            if ngram_range == [1] or ngram_range == (1, 1):
                tokens.extend(self._filter_doc(doc))
            # Handle ngrams
            else:
                # Generate token list
                token_list = self._filter_doc(doc)
                # Generate ngrams
                for n in range(ngram_range[0], ngram_range[-1] + 1):
                    ngrams = zip(*[token_list[i:] for i in range(n)])
                    for ngram in ngrams:
                        tokens.append(" ".join(ngram))
        return tokens

    def _get_doc_term_counts(self, docs: list[Doc]) -> Counter:
        """Build a Counter of terms after applying preprocessing settings.

        Args:
            docs (list[Doc]): List of spaCy Doc objects.

        Returns:
            Counter: Term frequency counts across all documents.
        """
        if isinstance(self.ngrams, int):
            ngram_range = [self.ngrams]
        else:
            ngram_range = list(self.ngrams)

        is_unigram = ngram_range == [1] or ngram_range == [1, 1]
        use_fast_doc_count = (
            is_unigram
            and self.case_sensitive
            and not self.remove_stopwords
            and not self.remove_punct
            and not self.remove_digits
        )

        counts: Counter = Counter()
        for doc in ensure_list(docs):
            if use_fast_doc_count:
                counts.update(count_doc_terms(doc))
            elif is_unigram:
                counts.update(self._filter_doc(doc))
            else:
                token_list = self._filter_doc(doc)
                for n in range(ngram_range[0], ngram_range[-1] + 1):
                    counts.update(
                        " ".join(gram)
                        for gram in zip(*[token_list[i:] for i in range(n)])
                    )
        return counts

    def to_dict(self):
        """Return the topwords as a dictionary with terms and Z-scores."""
        return {
            "topwords": [
                {"term": term, "z_score": z_score}
                for term, z_score in getattr(self, "topwords", [])
            ]
        }

    def to_df(self):
        """Return the topwords as a pandas DataFrame."""
        return pd.DataFrame(
            getattr(self, "topwords", []) or [], columns=["term", "z_score"]
        )

    def to_list_of_dicts(self):
        """Return the topwords as a list of dictionaries with 'term' and 'z_score'."""
        return [
            {"term": term, "z_score": z_score}
            for term, z_score in getattr(self, "topwords", [])
        ]

case_sensitive: bool | None = True pydantic-field ¤

Whether analysis is case sensitive.

model: str = 'xx_sent_ud_sm' pydantic-field ¤

spaCy model name to use for tokenization.

ngrams: int | tuple[int, int] = 1 pydantic-field ¤

The ngram range for analysis, e.g., (1, 2) for unigrams and bigrams.

remove_digits: bool | None = False pydantic-field ¤

Whether to remove digits.

remove_punct: bool | None = True pydantic-field ¤

Whether to remove punctuation.

remove_stopwords: bool | None = True pydantic-field ¤

Whether to remove stopwords.

topn: int = 10 pydantic-field ¤

Number of top words to return.

__init__(**data) -> None ¤

Initialize the ZTest class, ensuring a tokenizer is set.

If a tokenizer is not provided, creates one using the specified spaCy model.

Source code in lexos/topwords/ztest.py 
 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
def __init__(self, **data) -> None:
    """Initialize the ZTest class, ensuring a tokenizer is set.

    If a tokenizer is not provided, creates one using the specified spaCy model.
    """
    # If tokenizer is not provided, create one with the specified model
    if "tokenizer" not in data or data["tokenizer"] is None:
        data["tokenizer"] = Tokenizer(model=data.get("model", "xx_sent_ud_sm"))
    super().__init__(**data)

    # Tokenize the docs if they are strings
    self.target_docs = [
        self.tokenizer.make_doc(doc) if isinstance(doc, str) else doc
        for doc in ensure_list(self.target_docs)
    ]
    self.comparison_docs = [
        self.tokenizer.make_doc(doc) if isinstance(doc, str) else doc
        for doc in ensure_list(self.comparison_docs)
    ]

    # Get the term counts for target and comparison
    target_counts: Counter = self._get_doc_term_counts(self.target_docs)
    comparison_counts: Counter = self._get_doc_term_counts(self.comparison_docs)

    # Total number of tokens in target and comparison
    target_total: int = sum(target_counts.values())
    comparison_total: int = sum(comparison_counts.values())

    # Calculate Z-scores with vectorized NumPy operations.
    terms = list(set(target_counts) | set(comparison_counts))
    tc_arr = np.array([target_counts[t] for t in terms], dtype=np.float64)
    cc_arr = np.array([comparison_counts[t] for t in terms], dtype=np.float64)

    n1, n2 = target_total, comparison_total
    total = n1 + n2

    if n1 > 0 and n2 > 0 and total > 0:
        p1 = tc_arr / n1
        p2 = cc_arr / n2
        p = (tc_arr + cc_arr) / total
        se = np.sqrt(p * (1.0 - p) * (1.0 / n1 + 1.0 / n2))
        valid = (se > 0) & (p > 0) & (p < 1)
        z_scores = np.where(valid, (p1 - p2) / np.where(se > 0, se, 1.0), 0.0)
    else:
        z_scores = np.zeros(len(terms), dtype=np.float64)

    nonzero = z_scores != 0.0
    sorted_results = sorted(
        zip(np.array(terms, dtype=object)[nonzero], z_scores[nonzero].tolist()),
        key=lambda item: abs(item[1]),
        reverse=True,
    )

    # Assign the top N results
    self.topwords = sorted_results[: self.topn]

    # Set the topwords attribute on each document
    for doc in self.target_docs:
        doc._.topwords = self.topwords
    for doc in self.comparison_docs:
        doc._.topwords = self.topwords

to_df() ¤

Return the topwords as a pandas DataFrame.

Source code in lexos/topwords/ztest.py 
258
259
260
261
262
def to_df(self):
    """Return the topwords as a pandas DataFrame."""
    return pd.DataFrame(
        getattr(self, "topwords", []) or [], columns=["term", "z_score"]
    )

to_dict() ¤

Return the topwords as a dictionary with terms and Z-scores.

Source code in lexos/topwords/ztest.py 
249
250
251
252
253
254
255
256
def to_dict(self):
    """Return the topwords as a dictionary with terms and Z-scores."""
    return {
        "topwords": [
            {"term": term, "z_score": z_score}
            for term, z_score in getattr(self, "topwords", [])
        ]
    }

to_list_of_dicts() ¤

Return the topwords as a list of dictionaries with 'term' and 'z_score'.

Source code in lexos/topwords/ztest.py 
264
265
266
267
268
269
def to_list_of_dicts(self):
    """Return the topwords as a list of dictionaries with 'term' and 'z_score'."""
    return [
        {"term": term, "z_score": z_score}
        for term, z_score in getattr(self, "topwords", [])
    ]