Hugging Face's logo

Spaces:
klzn
/
sentimentstream-worker
Running

App Files Community
sentimentstream-worker
File size: 4,499 Bytes
8ff1b66
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""
Moduł do obliczania statystycznej wielkości próbki.

Implementuje wzory statystyczne dla próbkowania populacji.
"""

import math
from dataclasses import dataclass

from app.core.config import settings


# Wartości Z dla poziomów ufności
Z_SCORES = {
    0.90: 1.645,
    0.95: 1.96,
    0.99: 2.576,
}


@dataclass
class SamplePlan:
    """
    Plan próbkowania dla gry.

    Attributes:
        top_helpful: Liczba najprzydatniejszych recenzji.
        statistical_sample: Wielkość próbki statystycznej.
        positive_count: Ile pobrać pozytywnych (stratified).
        negative_count: Ile pobrać negatywnych (stratified).
        total: Łączna liczba recenzji do pobrania.
    """

    top_helpful: int
    statistical_sample: int
    positive_count: int
    negative_count: int
    total: int


def calculate_sample_size(
    population: int,
    confidence_level: float | None = None,
    margin_of_error: float | None = None,
) -> int:
    """
    Oblicza minimalną wielkość próbki dla danej populacji.
    Wykorzystuje wzór Cochrana z korektą dla populacji skończonej.
    """
    if confidence_level is None:
        confidence_level = settings.sample_confidence_level
    if margin_of_error is None:
        margin_of_error = settings.sample_margin_of_error

    # 1. Pobieramy Z-score (np. 1.96 dla 95% ufności). 
    # Mówi on, jak bardzo wynik może odbiegać od średniej w jednostkach odchylenia standardowego.
    z = Z_SCORES.get(confidence_level, 1.96)

    # 2. Zakładamy p=0.5 (maksymalna zmienność). 
    # To daje nam najbezpieczniejszą (największą) wielkość próbki.
    p = 0.5

    # 3. Wzór Cochrana dla nieskończonej populacji:
    # n0 = (Z^2 * p * (1-p)) / e^2
    # Wyjaśnienie: Z kwadrat razy zmienność, podzielone przez kwadrat błędu.
    n_0 = (z ** 2 * p * (1 - p)) / (margin_of_error ** 2)

    # 4. Korekta dla populacji skończonej (Steam ma policzalną liczbę recenzji):
    # n = n0 / (1 + (n0 - 1) / N)
    # Wyjaśnienie: Zmniejszamy próbkę, bo wiemy dokładnie, ile osób (recenzji) jest w "całym świecie" tej gry.
    n = n_0 / (1 + (n_0 - 1) / population)

    # Zaokrąglamy w górę do pełnej recenzji
    return math.ceil(n)


def create_sample_plan(
    total_reviews: int,
    positive_reviews: int,
    negative_reviews: int,
) -> SamplePlan:
    """
    Tworzy plan próbkowania, łącząc dwa podejścia.
    """
    top_helpful = settings.sample_top_helpful
    max_reviews = settings.sample_max_reviews

    # Obliczamy, ile recenzji musimy pobrać, żeby wynik był wiarygodny
    statistical_sample = calculate_sample_size(total_reviews)

    # Pilnujemy, żeby nie przekroczyć ustawionego limitu (np. 3000)
    statistical_sample = min(statistical_sample, max_reviews - top_helpful)

    # Obliczamy jaki procent stanowią pozytywy i negatywy w całości
    if total_reviews > 0:
        pos_ratio = positive_reviews / total_reviews
        neg_ratio = negative_reviews / total_reviews
    else:
        pos_ratio = 0.5
        neg_ratio = 0.5

    # Rozdzielamy naszą próbkę proporcjonalnie do tych wyników (Stratified Sampling)
    pos_target = math.ceil(statistical_sample * pos_ratio)
    neg_target = math.ceil(statistical_sample * neg_ratio)

    # Minority protection: boost the smaller group to minority_min if possible
    minority_min = settings.sample_minority_min

    if pos_target < minority_min and positive_reviews > pos_target:
        pos_target = min(minority_min, positive_reviews)
    
    if neg_target < minority_min and negative_reviews > neg_target:
        neg_target = min(minority_min, negative_reviews)

    # Final adjustment to stay within statistical_sample limit
    if pos_target + neg_target > statistical_sample:
        if pos_target > neg_target:
            pos_target = max(pos_target - (pos_target + neg_target - statistical_sample), minority_min)
        else:
            neg_target = max(neg_target - (pos_target + neg_target - statistical_sample), minority_min)

    # Final cap by actual availability
    positive_count = min(pos_target, positive_reviews)
    negative_count = min(neg_target, negative_reviews)

    # Sumujemy wszystko (Top Helpful + Próbka Statystyczna)
    total = top_helpful + positive_count + negative_count

    return SamplePlan(
        top_helpful=top_helpful,
        statistical_sample=statistical_sample,
        positive_count=positive_count,
        negative_count=negative_count,
        total=total,
    )