File size: 1,989 Bytes
ed4d993
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
from collections import deque
from typing import TYPE_CHECKING, Dict, List, Union

if TYPE_CHECKING:
    import pandas as pd


class MetricsTrackerAverage:
    """Metrics Tracker Average."""

    def __init__(self, step: int):
        self.history: List[Dict[str, Union[int, float]]] = [{"step": 0, "score": 0}]
        self.step: int = step
        self.i: int = 0
        self.num: float = 0
        self.denom: float = 0

    @property
    def score(self) -> float:
        return self.num / self.denom if self.denom > 0 else 0

    def on_decision(self) -> None:
        self.denom += 1

    def on_feedback(self, score: float) -> None:
        self.num += score or 0
        self.i += 1
        if self.step > 0 and self.i % self.step == 0:
            self.history.append({"step": self.i, "score": self.score})

    def to_pandas(self) -> "pd.DataFrame":
        import pandas as pd

        return pd.DataFrame(self.history)


class MetricsTrackerRollingWindow:
    """Metrics Tracker Rolling Window."""

    def __init__(self, window_size: int, step: int):
        self.history: List[Dict[str, Union[int, float]]] = [{"step": 0, "score": 0}]
        self.step: int = step
        self.i: int = 0
        self.window_size: int = window_size
        self.queue: deque = deque()
        self.sum: float = 0.0

    @property
    def score(self) -> float:
        return self.sum / len(self.queue) if len(self.queue) > 0 else 0

    def on_decision(self) -> None:
        pass

    def on_feedback(self, value: float) -> None:
        self.sum += value
        self.queue.append(value)
        self.i += 1

        if len(self.queue) > self.window_size:
            old_val = self.queue.popleft()
            self.sum -= old_val

        if self.step > 0 and self.i % self.step == 0:
            self.history.append({"step": self.i, "score": self.sum / len(self.queue)})

    def to_pandas(self) -> "pd.DataFrame":
        import pandas as pd

        return pd.DataFrame(self.history)