utils.math.statistics

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