utils.math.statistics

  1"""
  2Mathematical statistics.
  3"""
  4
  5import numpy
  6import pandas
  7
  8from typing import List, Hashable
  9
 10def findOutliers(
 11    srs: pandas.Series,
 12    sigma: int = 3,
 13    joinNeighbouringOutliers: bool = True,
 14    howManyCountAsNeighbours: int = 3
 15) -> pandas.Series:
 16    """
 17    Find outliers in the array and return the mask where outliers
 18    are marked with `True`. NaNs and INFs, if any, do not count
 19    as outliers.
 20
 21    If `joinNeighbouringOutliers` is `True`, then non-outlier values
 22    between neighbouring outliers will be marked as outliers too. The same
 23    goes for starting/ending elements, if they are close enough to an outlier.
 24
 25    Example with `joinNeighbouringOutliers` set to `True`:
 26
 27    ``` py
 28    import pandas
 29    import numpy
 30
 31    from phab.utils.math import statistics
 32
 33    srs = pandas.Series(
 34        [4, 111, 4, 4, 5, 6, numpy.inf, 2, 4, 4, numpy.nan, 1, 1e15, 4, 3, 3, 101, 2, 4, 3]
 35    )
 36
 37    markedOutliers = statistics.findOutliers(
 38        srs,
 39        sigma=3,
 40        joinNeighbouringOutliers=True,
 41        howManyCountAsNeighbours=3
 42    )
 43
 44    for i, v in srs.items():
 45        if markedOutliers[i] == True:
 46            print(f"{v} - outlier")
 47        else:
 48            print(f"{v} - regular")
 49    # 4.0 - outlier
 50    # 111.0 - outlier
 51    # 4.0 - regular
 52    # 4.0 - regular
 53    # 5.0 - regular
 54    # 6.0 - regular
 55    # inf - regular
 56    # 2.0 - regular
 57    # 4.0 - regular
 58    # 4.0 - regular
 59    # nan - regular
 60    # 1.0 - regular
 61    # 1000000000000000.0 - outlier
 62    # 4.0 - outlier
 63    # 3.0 - outlier
 64    # 3.0 - outlier
 65    # 101.0 - outlier
 66    # 2.0 - outlier
 67    # 4.0 - outlier
 68    # 3.0 - outlier
 69    ```
 70
 71    Example with `joinNeighbouringOutliers` set to `False`:
 72
 73    ``` py
 74    # ...
 75
 76    markedOutliers = statistics.findOutliers(
 77        srs,
 78        sigma=3,
 79        joinNeighbouringOutliers=False,
 80        howManyCountAsNeighbours=3
 81    )
 82
 83    for i, v in srs.items():
 84        if markedOutliers[i] == True:
 85            print(f"{v} - outlier")
 86        else:
 87            print(f"{v} - regular")
 88    # 4.0 - regular
 89    # 111.0 - outlier
 90    # 4.0 - regular
 91    # 4.0 - regular
 92    # 5.0 - regular
 93    # 6.0 - regular
 94    # inf - regular
 95    # 2.0 - regular
 96    # 4.0 - regular
 97    # 4.0 - regular
 98    # nan - regular
 99    # 1.0 - regular
100    # 1000000000000000.0 - outlier
101    # 4.0 - regular
102    # 3.0 - regular
103    # 3.0 - regular
104    # 101.0 - outlier
105    # 2.0 - regular
106    # 4.0 - regular
107    # 3.0 - regular
108    ```
109    """
110    # tbl = pandas.DataFrame({"vls": srs})
111    tbl = pandas.DataFrame(srs)
112
113    # find NaNs/INFs
114    tbl["finite"] = numpy.isfinite(srs)
115
116    # every element is an outlier by default
117    tbl["outliers"] = numpy.ones(len(tbl.index), dtype=bool)
118
119    # NaNs and INFs are not outliers
120    # print(tbl["finite"].loc[lambda x: x == False].index)
121    tbl.loc[
122        tbl["finite"] == False,  # noqa: E712
123        "outliers"
124    ] = False
125
126    tableFinite = tbl[
127        tbl["finite"] == True  # noqa: E712
128    ].iloc[:, 0]
129
130    med = numpy.median(tableFinite.values)  # type:ignore[arg-type] # ya hz
131    sig = 1.48 * numpy.median(numpy.abs(tableFinite - med))  # type:ignore[operator] # ya hz
132
133    for i, v in tableFinite.items():
134        if v > med - sigma * sig:
135            tbl.at[i, "outliers"] = False
136
137    for i, v in tableFinite.items():
138        if v >= med + sigma * sig:
139            tbl.at[i, "outliers"] = True
140
141    if not joinNeighbouringOutliers:
142        # the entire table with all columns (the original and two new ones,
143        # "finite" and "ouliers")
144        # return tbl
145        #
146        # or just two columns: first one with the values and "outliers"
147        # return tbl.iloc[:,0:3:2]
148        #
149        # or just the outliers
150        return tbl["outliers"]
151        #
152        # or just the "outliers" that are True, because it's a waste to pass
153        # around all the values, let alone the entire table
154        # return tbl[tbl["outliers"] == True]["outliers"]
155    else:
156        outlrs: pandas.Series = tbl["outliers"].copy()
157        elementsSincePreviousOutlier: int = 0
158        #
159        # later we might want to have the number of neighbours to scale
160        # with the list length, but for now it is passed as a fixed
161        # number in the `howManyCountAsNeighbours` argument of the function
162        # howManyCountAsNeighbours = round(len(outlrs.index) / 10)
163        #
164        # intuitively it should be `False`, but we need to account
165        # for possible neighbours from the very start
166        countingSincePreviousOutlier: bool = True
167        potentialNeighbourOutliers: List[Hashable] = []
168        for i, v in outlrs.items():
169            if v == True:  # noqa: E712
170                if countingSincePreviousOutlier:
171                    # make all the previous elements to be outliers too
172                    for pno in potentialNeighbourOutliers:
173                        outlrs[pno] = True
174                    potentialNeighbourOutliers = []
175                    elementsSincePreviousOutlier = 0
176                else:
177                    countingSincePreviousOutlier = True
178            else:
179                if countingSincePreviousOutlier:
180                    elementsSincePreviousOutlier += 1
181                    if elementsSincePreviousOutlier > howManyCountAsNeighbours:
182                        elementsSincePreviousOutlier = 0
183                        countingSincePreviousOutlier = False
184                        potentialNeighbourOutliers = []
185                    else:
186                        potentialNeighbourOutliers.append(i)
187        # if there are some pending potential neighbour outliers
188        # after we finished iterating the list, make them outliers
189        if len(potentialNeighbourOutliers) > 0:
190            for pno in potentialNeighbourOutliers:
191                outlrs[pno] = True
192            potentialNeighbourOutliers = []
193        return outlrs