Skip to content

Utils

The utils module contains the Discretizer and PowerTransformer classes for preparing continuous data for training, as well as a collection of miscellaneous functions for reproducible random number generation, evaluation, learning rate schedules, and serialization. See the Utilities guide for an introduction to preparing data with the Discretizer and PowerTransformer.

ColumnParams

Bases: TypedDict

Per-column discretization parameters.

Source code in src/qbm/utils/discretization.py 
 8
 9
10
11
12
13
class ColumnParams(TypedDict):
    """Per-column discretization parameters."""

    n_bits: int
    x_min: NotRequired[float]
    x_max: NotRequired[float]

Discretizer

Discretizes dataframe columns into bit representations and converts them back. Columns whose names end in "_bit" are treated as single bits and are not scaled.

Source code in src/qbm/utils/discretization.py 
 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
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
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
class Discretizer:
    """
    Discretizes dataframe columns into bit representations and converts them back.
    Columns whose names end in "_bit" are treated as single bits and are not scaled.
    """

    def __init__(
        self,
        df: pd.DataFrame,
        n_bits: int,
        epsilon: Mapping[str, Mapping[str, float]] = {},
    ) -> None:
        """
        Initializes the discretizer.

        Args:
            df: Dataframe of numerical values.
            n_bits: Number of bits to discretize to.
            epsilon: Optional dictionary of min/max offset values.

        Raises:
            ValueError: If n_bits is not positive, or if any non-bit column has
                zero range (x_max <= x_min).
        """
        if n_bits <= 0:
            raise ValueError(f"n_bits must be positive (got {n_bits})")
        self.columns = df.columns
        self.n_bits = n_bits
        self.epsilon = epsilon
        self.params: dict[str, ColumnParams] = {}
        self.split_indices = []
        self.n_bits_total = 0

        for i, column in enumerate(self.columns):
            if column.endswith("_bit"):
                self.params[column] = {"n_bits": 1}
            else:
                x_min = df[column].min()
                x_max = df[column].max()
                if column in self.epsilon:
                    x_min -= self.epsilon[column]["min"]
                    x_max += self.epsilon[column]["max"]
                if x_max <= x_min:
                    raise ValueError(
                        f"Column '{column}' has zero range "
                        f"(x_min = {x_min}, x_max = {x_max})"
                    )
                self.params[column] = {
                    "n_bits": self.n_bits,
                    "x_min": x_min,
                    "x_max": x_max,
                }

            # Update the split indices
            if i < len(self.columns) - 1:
                self.split_indices.append(
                    self.params[column]["n_bits"]
                    if i == 0
                    else self.split_indices[i - 1] + self.params[column]["n_bits"]
                )

            self.n_bits_total += self.params[column]["n_bits"]

    @staticmethod
    def bit_vector_to_int(bit_vector: Sequence[int] | np.ndarray) -> int:
        """
        Converts a bit vector to its integer representation.

        Args:
            bit_vector: Input bit vector.

        Returns:
            Integer representation of the input bit vector.

        Raises:
            ValueError: If any element of bit_vector is not 0 or 1.
        """
        if any(x not in (0, 1) for x in bit_vector):
            raise ValueError(
                f"bit_vector elements must be 0 or 1 (got {list(bit_vector)})"
            )
        return int("".join(str(x) for x in bit_vector), 2)

    @staticmethod
    def bit_vector_to_string(bit_vector: Sequence[int]) -> str:
        """
        Converts a bit vector to a bit string.

        Args:
            bit_vector: Input bit vector.

        Returns:
            Bit string of the input bit vector.
        """
        return "".join(str(x) for x in bit_vector)

    @staticmethod
    def int_to_bit_vector(x: int, n_bits: int) -> list[int]:
        """
        Converts the integer x to an n_bits-bit bit vector.

        Args:
            x: Integer value which to convert.
            n_bits: Length of the bit vector.

        Returns:
            Bit vector of length n_bits.

        Raises:
            ValueError: If x is negative, if n_bits is not positive, or if x does
                not fit in n_bits bits.
        """
        if x < 0:
            raise ValueError(f"x must be non-negative (got {x})")
        if n_bits <= 0:
            raise ValueError(f"n_bits must be positive (got {n_bits})")
        if x >= 2**n_bits:
            raise ValueError(
                f"x = {x} does not fit in {n_bits} bits (max is {2**n_bits - 1})"
            )
        return [1 if i == "1" else 0 for i in bin(x)[2:].zfill(n_bits)]

    @staticmethod
    @np.vectorize
    def discretize(x: float, n_bits: int, x_min: float, x_max: float) -> int:
        """
        Convert the value x into its n_bits-bit integer representation.

        Args:
            x: Float value to convert.
            n_bits: Number of bits to discretize to.
            x_min: Minimum value for scaling.
            x_max: Maximum value for scaling.

        Returns:
            An integer representation of x.

        Raises:
            ValueError: If x is out of the range [0, 2**n_bits - 1].
            TypeError: If the discretized value is not an integer.
        """
        scaling_factor = (2**n_bits - 1) / (x_max - x_min)

        x = round((x - x_min) * scaling_factor)
        if x < 0 or x > 2**n_bits - 1:
            raise ValueError(
                f"Discretized value {x} is out of range [0, {2**n_bits - 1}]"
            )
        if not isinstance(x, int):
            raise TypeError(f"Discretized value {x!r} is not an integer")
        return x

    @staticmethod
    @np.vectorize
    def undiscretize(x: float, n_bits: int, x_min: float, x_max: float) -> float:
        """
        Convert the value x into a float from its n_bits-bit integer representation.

        Args:
            x: Int value to convert.
            n_bits: Number of bits to discretize to.
            x_min: Minimum value for scaling.
            x_max: Maximum value for scaling.

        Returns:
            A float representation of x.

        Raises:
            ValueError: If x >= 2**n_bits.
        """
        scaling_factor = (2**n_bits - 1) / (x_max - x_min)

        if x >= 2**n_bits:
            raise ValueError(f"Value {x} is out of range [0, {2**n_bits - 1}]")
        return x / scaling_factor + x_min

    def discretize_df(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Convert all columns of a dataframe to bit representation.

        Args:
            df: Dataframe which to convert.

        Returns:
            A discretized version of df.
        """
        df_discretized = df.copy()
        for column in df.columns:
            if column.endswith("_bit"):
                df_discretized[column] = df[column].astype(np.int8)
            else:
                df_discretized[column] = self.discretize(
                    df[column], **self.params[column]
                )

        return df_discretized

    def undiscretize_df(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Convert all columns of a dataframe to floats from bit representation.

        Args:
            df: Dataframe which to convert.

        Returns:
            An undiscretized version of df.
        """
        df_undiscretized = df.copy()
        for column in df.columns:
            if column.endswith("_bit"):
                df_undiscretized[column] = df[column].astype(np.int8)
            else:
                df_undiscretized[column] = self.undiscretize(
                    df[column], **self.params[column]
                )

        return df_undiscretized

    def df_to_bit_array(self, df: pd.DataFrame) -> np.ndarray:
        """
        Converts a dataframe of floats to a bit array.

        Args:
            df: Dataframe which to convert.

        Returns:
            Array of bits of shape (df.shape[0], self.n_bits_total).

        Raises:
            ValueError: If the columns of df do not match the discretizer's columns.
        """
        if set(self.columns) != set(df.columns):
            raise ValueError(
                f"df columns {list(df.columns)} do not match the discretizer's "
                f"columns {list(self.columns)}"
            )

        df = self.discretize_df(df)
        bit_array = np.hstack(
            [
                np.vstack(
                    [
                        self.int_to_bit_vector(x, self.params[column]["n_bits"])
                        for x in df[column]
                    ]
                )
                for column in self.columns
            ]
        )

        return bit_array

    def bit_array_to_df(self, bit_array: np.ndarray) -> pd.DataFrame:
        """
        Converts a bit array to a dataframe of floats.

        Args:
            bit_array: Bit array which to convert.

        Returns:
            Dataframe of shape (bit_array.shape[0], len(self.columns)).

        Raises:
            ValueError: If the width of bit_array does not match n_bits_total.
        """
        if len(bit_array[0]) != self.n_bits_total:
            raise ValueError(
                f"bit_array width {len(bit_array[0])} does not match "
                f"n_bits_total {self.n_bits_total}"
            )

        rows = [
            [
                self.bit_vector_to_int(x)
                for x in np.array_split(bit_vector, self.split_indices)
            ]
            for bit_vector in bit_array
        ]
        df = self.undiscretize_df(pd.DataFrame(rows, columns=self.columns))

        return df

__init__(df, n_bits, epsilon={})

Initializes the discretizer.

Parameters:

Name Type Description Default
df DataFrame

Dataframe of numerical values.

 required
n_bits int

Number of bits to discretize to.

 required
epsilon Mapping[str, Mapping[str, float]]

Optional dictionary of min/max offset values.

 {}

Raises:

Type Description
ValueError

If n_bits is not positive, or if any non-bit column has zero range (x_max <= x_min).
Source code in src/qbm/utils/discretization.py 
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
def __init__(
    self,
    df: pd.DataFrame,
    n_bits: int,
    epsilon: Mapping[str, Mapping[str, float]] = {},
) -> None:
    """
    Initializes the discretizer.

    Args:
        df: Dataframe of numerical values.
        n_bits: Number of bits to discretize to.
        epsilon: Optional dictionary of min/max offset values.

    Raises:
        ValueError: If n_bits is not positive, or if any non-bit column has
            zero range (x_max <= x_min).
    """
    if n_bits <= 0:
        raise ValueError(f"n_bits must be positive (got {n_bits})")
    self.columns = df.columns
    self.n_bits = n_bits
    self.epsilon = epsilon
    self.params: dict[str, ColumnParams] = {}
    self.split_indices = []
    self.n_bits_total = 0

    for i, column in enumerate(self.columns):
        if column.endswith("_bit"):
            self.params[column] = {"n_bits": 1}
        else:
            x_min = df[column].min()
            x_max = df[column].max()
            if column in self.epsilon:
                x_min -= self.epsilon[column]["min"]
                x_max += self.epsilon[column]["max"]
            if x_max <= x_min:
                raise ValueError(
                    f"Column '{column}' has zero range "
                    f"(x_min = {x_min}, x_max = {x_max})"
                )
            self.params[column] = {
                "n_bits": self.n_bits,
                "x_min": x_min,
                "x_max": x_max,
            }

        # Update the split indices
        if i < len(self.columns) - 1:
            self.split_indices.append(
                self.params[column]["n_bits"]
                if i == 0
                else self.split_indices[i - 1] + self.params[column]["n_bits"]
            )

        self.n_bits_total += self.params[column]["n_bits"]

bit_array_to_df(bit_array)

Converts a bit array to a dataframe of floats.

Parameters:

Name Type Description Default
bit_array ndarray

Bit array which to convert.

 required

Returns:

Type Description
DataFrame

Dataframe of shape (bit_array.shape[0], len(self.columns)).

Raises:

Type Description
ValueError

If the width of bit_array does not match n_bits_total.
Source code in src/qbm/utils/discretization.py 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
def bit_array_to_df(self, bit_array: np.ndarray) -> pd.DataFrame:
    """
    Converts a bit array to a dataframe of floats.

    Args:
        bit_array: Bit array which to convert.

    Returns:
        Dataframe of shape (bit_array.shape[0], len(self.columns)).

    Raises:
        ValueError: If the width of bit_array does not match n_bits_total.
    """
    if len(bit_array[0]) != self.n_bits_total:
        raise ValueError(
            f"bit_array width {len(bit_array[0])} does not match "
            f"n_bits_total {self.n_bits_total}"
        )

    rows = [
        [
            self.bit_vector_to_int(x)
            for x in np.array_split(bit_vector, self.split_indices)
        ]
        for bit_vector in bit_array
    ]
    df = self.undiscretize_df(pd.DataFrame(rows, columns=self.columns))

    return df

bit_vector_to_int(bit_vector) staticmethod

Converts a bit vector to its integer representation.

Parameters:

Name Type Description Default
bit_vector Sequence[int] | ndarray

Input bit vector.

 required

Returns:

Type Description
int

Integer representation of the input bit vector.

Raises:

Type Description
ValueError

If any element of bit_vector is not 0 or 1.
Source code in src/qbm/utils/discretization.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
@staticmethod
def bit_vector_to_int(bit_vector: Sequence[int] | np.ndarray) -> int:
    """
    Converts a bit vector to its integer representation.

    Args:
        bit_vector: Input bit vector.

    Returns:
        Integer representation of the input bit vector.

    Raises:
        ValueError: If any element of bit_vector is not 0 or 1.
    """
    if any(x not in (0, 1) for x in bit_vector):
        raise ValueError(
            f"bit_vector elements must be 0 or 1 (got {list(bit_vector)})"
        )
    return int("".join(str(x) for x in bit_vector), 2)

bit_vector_to_string(bit_vector) staticmethod

Converts a bit vector to a bit string.

Parameters:

Name Type Description Default
bit_vector Sequence[int]

Input bit vector.

 required

Returns:

Type Description
str

Bit string of the input bit vector.
Source code in src/qbm/utils/discretization.py 
 99
100
101
102
103
104
105
106
107
108
109
110
@staticmethod
def bit_vector_to_string(bit_vector: Sequence[int]) -> str:
    """
    Converts a bit vector to a bit string.

    Args:
        bit_vector: Input bit vector.

    Returns:
        Bit string of the input bit vector.
    """
    return "".join(str(x) for x in bit_vector)

df_to_bit_array(df)

Converts a dataframe of floats to a bit array.

Parameters:

Name Type Description Default
df DataFrame

Dataframe which to convert.

 required

Returns:

Type Description
ndarray

Array of bits of shape (df.shape[0], self.n_bits_total).

Raises:

Type Description
ValueError

If the columns of df do not match the discretizer's columns.
Source code in src/qbm/utils/discretization.py 
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
def df_to_bit_array(self, df: pd.DataFrame) -> np.ndarray:
    """
    Converts a dataframe of floats to a bit array.

    Args:
        df: Dataframe which to convert.

    Returns:
        Array of bits of shape (df.shape[0], self.n_bits_total).

    Raises:
        ValueError: If the columns of df do not match the discretizer's columns.
    """
    if set(self.columns) != set(df.columns):
        raise ValueError(
            f"df columns {list(df.columns)} do not match the discretizer's "
            f"columns {list(self.columns)}"
        )

    df = self.discretize_df(df)
    bit_array = np.hstack(
        [
            np.vstack(
                [
                    self.int_to_bit_vector(x, self.params[column]["n_bits"])
                    for x in df[column]
                ]
            )
            for column in self.columns
        ]
    )

    return bit_array

discretize(x, n_bits, x_min, x_max) staticmethod

Convert the value x into its n_bits-bit integer representation.

Parameters:

Name Type Description Default
x float

Float value to convert.

 required
n_bits int

Number of bits to discretize to.

 required
x_min float

Minimum value for scaling.

 required
x_max float

Maximum value for scaling.

 required

Returns:

Type Description
int

An integer representation of x.

Raises:

Type Description
ValueError

If x is out of the range [0, 2**n_bits - 1].
TypeError

If the discretized value is not an integer.
Source code in src/qbm/utils/discretization.py 
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
@staticmethod
@np.vectorize
def discretize(x: float, n_bits: int, x_min: float, x_max: float) -> int:
    """
    Convert the value x into its n_bits-bit integer representation.

    Args:
        x: Float value to convert.
        n_bits: Number of bits to discretize to.
        x_min: Minimum value for scaling.
        x_max: Maximum value for scaling.

    Returns:
        An integer representation of x.

    Raises:
        ValueError: If x is out of the range [0, 2**n_bits - 1].
        TypeError: If the discretized value is not an integer.
    """
    scaling_factor = (2**n_bits - 1) / (x_max - x_min)

    x = round((x - x_min) * scaling_factor)
    if x < 0 or x > 2**n_bits - 1:
        raise ValueError(
            f"Discretized value {x} is out of range [0, {2**n_bits - 1}]"
        )
    if not isinstance(x, int):
        raise TypeError(f"Discretized value {x!r} is not an integer")
    return x

discretize_df(df)

Convert all columns of a dataframe to bit representation.

Parameters:

Name Type Description Default
df DataFrame

Dataframe which to convert.

 required

Returns:

Type Description
DataFrame

A discretized version of df.
Source code in src/qbm/utils/discretization.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def discretize_df(self, df: pd.DataFrame) -> pd.DataFrame:
    """
    Convert all columns of a dataframe to bit representation.

    Args:
        df: Dataframe which to convert.

    Returns:
        A discretized version of df.
    """
    df_discretized = df.copy()
    for column in df.columns:
        if column.endswith("_bit"):
            df_discretized[column] = df[column].astype(np.int8)
        else:
            df_discretized[column] = self.discretize(
                df[column], **self.params[column]
            )

    return df_discretized

int_to_bit_vector(x, n_bits) staticmethod

Converts the integer x to an n_bits-bit bit vector.

Parameters:

Name Type Description Default
x int

Integer value which to convert.

 required
n_bits int

Length of the bit vector.

 required

Returns:

Type Description
list[int]

Bit vector of length n_bits.

Raises:

Type Description
ValueError

If x is negative, if n_bits is not positive, or if x does not fit in n_bits bits.
Source code in src/qbm/utils/discretization.py 
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
@staticmethod
def int_to_bit_vector(x: int, n_bits: int) -> list[int]:
    """
    Converts the integer x to an n_bits-bit bit vector.

    Args:
        x: Integer value which to convert.
        n_bits: Length of the bit vector.

    Returns:
        Bit vector of length n_bits.

    Raises:
        ValueError: If x is negative, if n_bits is not positive, or if x does
            not fit in n_bits bits.
    """
    if x < 0:
        raise ValueError(f"x must be non-negative (got {x})")
    if n_bits <= 0:
        raise ValueError(f"n_bits must be positive (got {n_bits})")
    if x >= 2**n_bits:
        raise ValueError(
            f"x = {x} does not fit in {n_bits} bits (max is {2**n_bits - 1})"
        )
    return [1 if i == "1" else 0 for i in bin(x)[2:].zfill(n_bits)]

undiscretize(x, n_bits, x_min, x_max) staticmethod

Convert the value x into a float from its n_bits-bit integer representation.

Parameters:

Name Type Description Default
x float

Int value to convert.

 required
n_bits int

Number of bits to discretize to.

 required
x_min float

Minimum value for scaling.

 required
x_max float

Maximum value for scaling.

 required

Returns:

Type Description
float

A float representation of x.

Raises:

Type Description
ValueError

If x >= 2**n_bits.
Source code in src/qbm/utils/discretization.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
@staticmethod
@np.vectorize
def undiscretize(x: float, n_bits: int, x_min: float, x_max: float) -> float:
    """
    Convert the value x into a float from its n_bits-bit integer representation.

    Args:
        x: Int value to convert.
        n_bits: Number of bits to discretize to.
        x_min: Minimum value for scaling.
        x_max: Maximum value for scaling.

    Returns:
        A float representation of x.

    Raises:
        ValueError: If x >= 2**n_bits.
    """
    scaling_factor = (2**n_bits - 1) / (x_max - x_min)

    if x >= 2**n_bits:
        raise ValueError(f"Value {x} is out of range [0, {2**n_bits - 1}]")
    return x / scaling_factor + x_min

undiscretize_df(df)

Convert all columns of a dataframe to floats from bit representation.

Parameters:

Name Type Description Default
df DataFrame

Dataframe which to convert.

 required

Returns:

Type Description
DataFrame

An undiscretized version of df.
Source code in src/qbm/utils/discretization.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def undiscretize_df(self, df: pd.DataFrame) -> pd.DataFrame:
    """
    Convert all columns of a dataframe to floats from bit representation.

    Args:
        df: Dataframe which to convert.

    Returns:
        An undiscretized version of df.
    """
    df_undiscretized = df.copy()
    for column in df.columns:
        if column.endswith("_bit"):
            df_undiscretized[column] = df[column].astype(np.int8)
        else:
            df_undiscretized[column] = self.undiscretize(
                df[column], **self.params[column]
            )

    return df_undiscretized

PowerTransformer

Transforms data points that lie beyond the provided threshold by taking their power (<1) to scale them closer to the mean.

Source code in src/qbm/utils/transformations.py 
  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
class PowerTransformer:
    """
    Transforms data points that lie beyond the provided threshold by taking their
    power (<1) to scale them closer to the mean.
    """

    def __init__(
        self,
        df: pd.DataFrame,
        threshold: float = 1.0,
        power: float = 0.5,
        columns: Sequence[Hashable] | None = None,
    ) -> None:
        """
        Initializes the transformer.

        Args:
            df: Dataframe of data to scale.
            threshold: Number of standard deviations from the mean beyond which to
                begin the scaling (applied to both tails).
            power: Power at which to scale the outlier.
            columns: Optional list of columns to apply the transformation to. If no
                columns are provided, then all columns are transformed.
        Raises:
            ValueError: If power >= 1, if threshold < 1, if power <= 0, or if
                columns is not a subset of df.columns.
        """
        if power >= 1:
            raise ValueError(f"power must be < 1 (got {power})")
        if power <= 0:
            raise ValueError(f"power must be > 0 (got {power})")
        if threshold < 1:
            raise ValueError(f"threshold must be >= 1 (got {threshold})")

        if columns is None:
            self.columns = df.columns
        else:
            if not set(columns).issubset(df.columns):
                raise ValueError(
                    f"columns {list(columns)} are not a subset of "
                    f"df.columns {list(df.columns)}"
                )
            self.columns = columns
        self.power = power
        self.threshold = threshold
        self.offset = threshold - threshold**power

        self.μ = {}
        self.σ = {}
        for column in df.columns:
            self.μ[column] = df[column].mean()
            self.σ[column] = df[column].std()

    def transform(self, df: pd.DataFrame, inplace: bool = False) -> pd.DataFrame:
        """
        Transforms the data to the scaled space.

        Args:
            df: Dataframe to scale.
            inplace: If True then it operates on the same dataframe, if False then
                it creates a copy.

        Returns:
            Dataframe of transformed data.

        Raises:
            ValueError: If a configured column is missing from df.
        """
        if not set(self.columns).issubset(df.columns):
            raise ValueError(
                f"df is missing configured columns "
                f"{list(set(self.columns) - set(df.columns))}"
            )
        if not inplace:
            df = df.copy()

        for column in self.columns:
            μ = self.μ[column]
            σ = self.σ[column]
            x = (df[column] - μ) / σ
            mask = np.abs(x) > self.threshold
            x[mask] = ((np.abs(x) ** self.power + self.offset) * np.sign(x))[mask]
            df[column] = x * σ + μ

        return df

    def inverse_transform(
        self, df: pd.DataFrame, inplace: bool = False
    ) -> pd.DataFrame:
        """
        Transforms the data back from the scaled space.

        Args:
            df: Dataframe to scale.
            inplace: If True then it operates on the same dataframe, if False then
                it creates a copy.

        Returns:
            Dataframe of untransformed data.

        Raises:
            ValueError: If a configured column is missing from df.
        """
        if not set(self.columns).issubset(df.columns):
            raise ValueError(
                f"df is missing configured columns "
                f"{list(set(self.columns) - set(df.columns))}"
            )
        if not inplace:
            df = df.copy()

        for column in self.columns:
            μ = self.μ[column]
            σ = self.σ[column]
            x = (df[column] - μ) / σ
            mask = np.abs(x) > self.threshold
            x[mask] = ((np.abs(x) - self.offset) ** (1 / self.power) * np.sign(x))[mask]
            df[column] = x * σ + μ

        return df

__init__(df, threshold=1.0, power=0.5, columns=None)

Initializes the transformer.

Parameters:

Name Type Description Default
df DataFrame

Dataframe of data to scale.

 required
threshold float

Number of standard deviations from the mean beyond which to begin the scaling (applied to both tails).

 1.0
power float

Power at which to scale the outlier.

 0.5
columns Sequence[Hashable] | None

Optional list of columns to apply the transformation to. If no columns are provided, then all columns are transformed.

 None

Raises: ValueError: If power >= 1, if threshold < 1, if power <= 0, or if columns is not a subset of df.columns.

Source code in src/qbm/utils/transformations.py 
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
def __init__(
    self,
    df: pd.DataFrame,
    threshold: float = 1.0,
    power: float = 0.5,
    columns: Sequence[Hashable] | None = None,
) -> None:
    """
    Initializes the transformer.

    Args:
        df: Dataframe of data to scale.
        threshold: Number of standard deviations from the mean beyond which to
            begin the scaling (applied to both tails).
        power: Power at which to scale the outlier.
        columns: Optional list of columns to apply the transformation to. If no
            columns are provided, then all columns are transformed.
    Raises:
        ValueError: If power >= 1, if threshold < 1, if power <= 0, or if
            columns is not a subset of df.columns.
    """
    if power >= 1:
        raise ValueError(f"power must be < 1 (got {power})")
    if power <= 0:
        raise ValueError(f"power must be > 0 (got {power})")
    if threshold < 1:
        raise ValueError(f"threshold must be >= 1 (got {threshold})")

    if columns is None:
        self.columns = df.columns
    else:
        if not set(columns).issubset(df.columns):
            raise ValueError(
                f"columns {list(columns)} are not a subset of "
                f"df.columns {list(df.columns)}"
            )
        self.columns = columns
    self.power = power
    self.threshold = threshold
    self.offset = threshold - threshold**power

    self.μ = {}
    self.σ = {}
    for column in df.columns:
        self.μ[column] = df[column].mean()
        self.σ[column] = df[column].std()

inverse_transform(df, inplace=False)

Transforms the data back from the scaled space.

Parameters:

Name Type Description Default
df DataFrame

Dataframe to scale.

 required
inplace bool

If True then it operates on the same dataframe, if False then it creates a copy.

 False

Returns:

Type Description
DataFrame

Dataframe of untransformed data.

Raises:

Type Description
ValueError

If a configured column is missing from df.
Source code in src/qbm/utils/transformations.py 
 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
def inverse_transform(
    self, df: pd.DataFrame, inplace: bool = False
) -> pd.DataFrame:
    """
    Transforms the data back from the scaled space.

    Args:
        df: Dataframe to scale.
        inplace: If True then it operates on the same dataframe, if False then
            it creates a copy.

    Returns:
        Dataframe of untransformed data.

    Raises:
        ValueError: If a configured column is missing from df.
    """
    if not set(self.columns).issubset(df.columns):
        raise ValueError(
            f"df is missing configured columns "
            f"{list(set(self.columns) - set(df.columns))}"
        )
    if not inplace:
        df = df.copy()

    for column in self.columns:
        μ = self.μ[column]
        σ = self.σ[column]
        x = (df[column] - μ) / σ
        mask = np.abs(x) > self.threshold
        x[mask] = ((np.abs(x) - self.offset) ** (1 / self.power) * np.sign(x))[mask]
        df[column] = x * σ + μ

    return df

transform(df, inplace=False)

Transforms the data to the scaled space.

Parameters:

Name Type Description Default
df DataFrame

Dataframe to scale.

 required
inplace bool

If True then it operates on the same dataframe, if False then it creates a copy.

 False

Returns:

Type Description
DataFrame

Dataframe of transformed data.

Raises:

Type Description
ValueError

If a configured column is missing from df.
Source code in src/qbm/utils/transformations.py 
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
def transform(self, df: pd.DataFrame, inplace: bool = False) -> pd.DataFrame:
    """
    Transforms the data to the scaled space.

    Args:
        df: Dataframe to scale.
        inplace: If True then it operates on the same dataframe, if False then
            it creates a copy.

    Returns:
        Dataframe of transformed data.

    Raises:
        ValueError: If a configured column is missing from df.
    """
    if not set(self.columns).issubset(df.columns):
        raise ValueError(
            f"df is missing configured columns "
            f"{list(set(self.columns) - set(df.columns))}"
        )
    if not inplace:
        df = df.copy()

    for column in self.columns:
        μ = self.μ[column]
        σ = self.σ[column]
        x = (df[column] - μ) / σ
        mask = np.abs(x) > self.threshold
        x[mask] = ((np.abs(x) ** self.power + self.offset) * np.sign(x))[mask]
        df[column] = x * σ + μ

    return df

compute_df_ensemble_stats(dfs)

Computes the means, medians, and standard deviations column/row-wise over the input list of dataframes.

Parameters:

Name Type Description Default
dfs Sequence[DataFrame]

List of dataframes with identical row/column names.

 required

Returns:

Type Description
dict[str, DataFrame]

Dictionary of dataframes with the means, medians, and standard deviations.

Raises:

Type Description
ValueError

If dfs is empty.
TypeError

If any of the computed statistics is not a DataFrame.
Source code in src/qbm/utils/misc.py 
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
def compute_df_ensemble_stats(
    dfs: Sequence[pd.DataFrame],
) -> dict[str, pd.DataFrame]:
    """
    Computes the means, medians, and standard deviations column/row-wise over the input
    list of dataframes.

    Args:
        dfs: List of dataframes with identical row/column names.

    Returns:
        Dictionary of dataframes with the means, medians, and standard deviations.

    Raises:
        ValueError: If dfs is empty.
        TypeError: If any of the computed statistics is not a DataFrame.
    """
    if len(dfs) == 0:
        raise ValueError("dfs must not be empty")
    df = pd.concat(dfs)
    means = df.groupby(df.index).mean()
    medians = df.groupby(df.index).median()
    stds = df.groupby(df.index).std()
    if not isinstance(means, pd.DataFrame):
        raise TypeError("Grouped means is not a DataFrame")
    if not isinstance(medians, pd.DataFrame):
        raise TypeError("Grouped medians is not a DataFrame")
    if not isinstance(stds, pd.DataFrame):
        raise TypeError("Grouped stds is not a DataFrame")

    return {"means": means, "medians": medians, "stds": stds}

compute_df_stats(df)

Compute the min, max, mean, median, and standard deviation of the columns in the dataframe.

Parameters:

Name Type Description Default
df DataFrame

Dataframe.

 required

Returns:

Type Description
DataFrame

Dataframe of the statistics.
Source code in src/qbm/utils/misc.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def compute_df_stats(df: pd.DataFrame) -> pd.DataFrame:
    """
    Compute the min, max, mean, median, and standard deviation of the columns in the
    dataframe.

    Args:
        df: Dataframe.

    Returns:
        Dataframe of the statistics.
    """
    return pd.DataFrame.from_dict(
        {
            "min": df.min(),
            "max": df.max(),
            "mean": df.mean(),
            "median": df.median(),
            "std": df.std(),
        },
        orient="index",
    )

compute_kl_divergence(p_data, q_data, n_bins=32, epsilon_smooth=None, relative_smooth=False)

Computes the D_KL(p_data || q_data).

Note

this is a crude approximation of the KL divergence.

Parameters:

Name Type Description Default
p_data ndarray

Array of data values to compute the p distribution from.

 required
q_data ndarray

Array of data values to compute the q distribution from.

 required
n_bins int

Number of bins to use in histograms.

 32
epsilon_smooth float | None

Value to use with q distribution smoothing.

 None
relative_smooth bool

Whether or not the smoothed values are relative to the p distribution.

 False

Returns:

Type Description
float

D_KL(p || q).

Raises:

Type Description
ValueError

If p_data or q_data is empty, if n_bins is not positive, or if either distribution does not sum to 1.
Source code in src/qbm/utils/misc.py 
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
def compute_kl_divergence(
    p_data: np.ndarray,
    q_data: np.ndarray,
    n_bins: int = 32,
    epsilon_smooth: float | None = None,
    relative_smooth: bool = False,
) -> float:
    """
    Computes the D_KL(p_data || q_data).

    Note:
        this is a crude approximation of the KL divergence.

    Args:
        p_data: Array of data values to compute the p distribution from.
        q_data: Array of data values to compute the q distribution from.
        n_bins: Number of bins to use in histograms.
        epsilon_smooth: Value to use with q distribution smoothing.
        relative_smooth: Whether or not the smoothed values are relative to the p
            distribution.

    Returns:
        D_KL(p || q).

    Raises:
        ValueError: If p_data or q_data is empty, if n_bins is not positive, or if
            either distribution does not sum to 1.
    """
    if p_data.shape[0] == 0:
        raise ValueError("p_data must not be empty")
    if q_data.shape[0] == 0:
        raise ValueError("q_data must not be empty")
    if n_bins <= 0:
        raise ValueError(f"n_bins must be positive (got {n_bins})")
    # Bin over the combined range so that mass in either distribution
    # outside the other's support is not silently dropped
    lo = min(p_data.min(), q_data.min())
    hi = max(p_data.max(), q_data.max())
    hist_data, _ = np.histogram(p_data, bins=n_bins, range=(lo, hi))
    hist_samples, _ = np.histogram(q_data, bins=n_bins, range=(lo, hi))

    p = hist_data / p_data.shape[0]
    q = hist_samples / q_data.shape[0]

    if epsilon_smooth is not None:
        smooth_mask = np.logical_and(p > 0, q == 0)
        not_smooth_mask = np.logical_not(smooth_mask)
        q[smooth_mask] = epsilon_smooth

        if relative_smooth:
            q[smooth_mask] *= p[smooth_mask]

        q[not_smooth_mask] -= q[smooth_mask].sum() / not_smooth_mask.sum()

    if not np.isclose(p.sum(), 1, atol=1e-3):
        raise ValueError(f"p distribution does not sum to 1 (sums to {p.sum()})")
    if not np.isclose(q.sum(), 1, atol=1e-3):
        raise ValueError(f"q distribution does not sum to 1 (sums to {q.sum()})")

    support = np.logical_and(p > 0, q > 0)
    p = p[support]
    q = q[support]

    return (p * np.log(p / q)).sum()

compute_lower_tail_concentration(z, U, V)

Lower tail concentration function defined as: L(z) = P(U <= z | V <= z) = P(U <= z, V <= z) / P(U <= z) References: - https://freakonometrics.hypotheses.org/2435 - https://openacttexts.github.io/Loss-Data-Analytics/C-DependenceModel (section 14.5.4.3) - https://www.casact.org/sites/default/files/old/studynotes_venter_tails_of_copulas.pdf (section 3)

Parameters:

Name Type Description Default
z float | ndarray

Tail dependence parameter (scalar or array of parameters).

 required
U ndarray

Input array for first variable (e.g. X.rank() / (len(X) + 1)).

 required
V ndarray

Input array for second variable (e.g. Y.rank() / (len(Y) + 1)).

 required

Returns:

Type Description
float | ndarray

Lower tail concentration function (scalar or array, one value per z).
Source code in src/qbm/utils/misc.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
def compute_lower_tail_concentration(
    z: float | np.ndarray, U: np.ndarray, V: np.ndarray
) -> float | np.ndarray:
    """
    Lower tail concentration function defined as:
    L(z) = P(U <= z | V <= z) = P(U <= z, V <= z) / P(U <= z)
    References:
        - https://freakonometrics.hypotheses.org/2435
        - https://openacttexts.github.io/Loss-Data-Analytics/C-DependenceModel
            (section 14.5.4.3)
        - https://www.casact.org/sites/default/files/old/studynotes_venter_tails_of_copulas.pdf
            (section 3)

    Args:
        z: Tail dependence parameter (scalar or array of parameters).
        U: Input array for first variable (e.g. X.rank() / (len(X) + 1)).
        V: Input array for second variable (e.g. Y.rank() / (len(Y) + 1)).

    Returns:
        Lower tail concentration function (scalar or array, one value per z).
    """
    z_expanded = np.asarray(z)[..., np.newaxis]
    return np.sum(np.logical_and(z_expanded >= U, z_expanded >= V), axis=-1) / np.sum(
        z_expanded >= U, axis=-1
    )

compute_lr_exp_decay(epoch, decay_epoch, period, base=2.0)

Exponential decay function for use in learning rate scheduling. It is relative, so one must multiply the base learning rate by the output of this function.

Parameters:

Name Type Description Default
epoch float | Sequence[float] | ndarray

Current epoch (scalar or array of epochs).

 required
decay_epoch float

Epoch at which to begin the decay.

 required
period float

Decay period.

 required
base float

Base number of the exponential decay.

 2.0

Returns:

Type Description
float | ndarray

The learning rate scaling factor (scalar or array, matching the input).
Source code in src/qbm/utils/misc.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def compute_lr_exp_decay(
    epoch: float | Sequence[float] | np.ndarray,
    decay_epoch: float,
    period: float,
    base: float = 2.0,
) -> float | np.ndarray:
    """
    Exponential decay function for use in learning rate scheduling. It is relative, so
    one must multiply the base learning rate by the output of this function.

    Args:
        epoch: Current epoch (scalar or array of epochs).
        decay_epoch: Epoch at which to begin the decay.
        period: Decay period.
        base: Base number of the exponential decay.

    Returns:
        The learning rate scaling factor (scalar or array, matching the input).
    """
    epoch_array = np.asarray(epoch)
    return base ** (np.minimum(decay_epoch - epoch_array, 0) / period)

compute_upper_tail_concentration(z, U, V)

Upper tail concentration function defined as: R(z) = P(U > z | V > z) = P(U > z, V > z) / P(U > z) References: - https://freakonometrics.hypotheses.org/2435 - https://openacttexts.github.io/Loss-Data-Analytics/C-DependenceModel (section 14.5.4.3) - https://www.casact.org/sites/default/files/old/studynotes_venter_tails_of_copulas.pdf (section 3)

Parameters:

Name Type Description Default
z float | ndarray

Tail dependence parameter (scalar or array of parameters).

 required
U ndarray

Input array for first variable (e.g. X.rank() / (len(X) + 1)).

 required
V ndarray

Input array for second variable (e.g. Y.rank() / (len(Y) + 1)).

 required

Returns:

Type Description
float | ndarray

Upper tail concentration function (scalar or array, one value per z).
Source code in src/qbm/utils/misc.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
def compute_upper_tail_concentration(
    z: float | np.ndarray, U: np.ndarray, V: np.ndarray
) -> float | np.ndarray:
    """
    Upper tail concentration function defined as:
    R(z) = P(U > z | V > z) = P(U > z, V > z) / P(U > z)
    References:
        - https://freakonometrics.hypotheses.org/2435
        - https://openacttexts.github.io/Loss-Data-Analytics/C-DependenceModel
            (section 14.5.4.3)
        - https://www.casact.org/sites/default/files/old/studynotes_venter_tails_of_copulas.pdf
            (section 3)

    Args:
        z: Tail dependence parameter (scalar or array of parameters).
        U: Input array for first variable (e.g. X.rank() / (len(X) + 1)).
        V: Input array for second variable (e.g. Y.rank() / (len(Y) + 1)).

    Returns:
        Upper tail concentration function (scalar or array, one value per z).
    """
    z_expanded = np.asarray(z)[..., np.newaxis]
    return np.sum(np.logical_and(z_expanded < U, z_expanded < V), axis=-1) / np.sum(
        z_expanded < U, axis=-1
    )

filter_df_on_values(df, column_values, drop_filter_columns=True)

Return a copy of the dataframe filtered conditionally on provided column values.

Parameters:

Name Type Description Default
df DataFrame

Dataframe to filter.

 required
column_values Mapping[Any, Any]

Dictionary where the keys are column names, and the values are values on which to filter the dataframe.

 required
drop_filter_columns bool

If True returns a copy of the dataframe with the filtered columns dropped.

 True

Returns:

Type Description
DataFrame

A dataframe filtered conditionally on the provided column values.
Source code in src/qbm/utils/misc.py 
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
def filter_df_on_values(
    df: pd.DataFrame,
    column_values: Mapping[Any, Any],
    drop_filter_columns: bool = True,
) -> pd.DataFrame:
    """
    Return a copy of the dataframe filtered conditionally on provided
    column values.

    Args:
        df: Dataframe to filter.
        column_values: Dictionary where the keys are column names, and the
            values are values on which to filter the dataframe.
        drop_filter_columns: If True returns a copy of the dataframe with
            the filtered columns dropped.

    Returns:
        A dataframe filtered conditionally on the provided column values.
    """
    df = df.copy()
    for column, value in column_values.items():
        df = df.loc[df[column] == value]

    if drop_filter_columns:
        df.drop(column_values.keys(), axis=1, inplace=True)

    return df

get_rng(seed=None)

Creates a random number generator with the specified seed value.

Parameters:

Name Type Description Default
seed int | None

Seed value for the rng.

 None

Returns:

Type Description
RandomState

Numpy RandomState object.
Source code in src/qbm/utils/misc.py 
 97
 98
 99
100
101
102
103
104
105
106
107
def get_rng(seed: int | None = None) -> RandomState:
    """
    Creates a random number generator with the specified seed value.

    Args:
        seed: Seed value for the rng.

    Returns:
        Numpy RandomState object.
    """
    return RandomState(MT19937(SeedSequence(seed)))

load_artifact(file_path)

Loads a pickle or json artifact (depending on the file extension).

Parameters:

Name Type Description Default
file_path str | Path

Path of the file to load.

 required

Returns:

Type Description
Any

Loaded python object.

Raises:

Type Description
FileNotFoundError

If the file does not exist.
ValueError

If the file has an unsupported file extension.
Source code in src/qbm/utils/misc.py 
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
def load_artifact(file_path: str | Path) -> Any:
    """
    Loads a pickle or json artifact (depending on the file extension).

    Args:
        file_path: Path of the file to load.

    Returns:
        Loaded python object.

    Raises:
        FileNotFoundError: If the file does not exist.
        ValueError: If the file has an unsupported file extension.
    """
    if isinstance(file_path, str):
        file_path = Path(file_path)

    if not file_path.exists():
        raise FileNotFoundError(f"File {file_path} does not exist")
    if file_path.suffix not in (".json", ".pkl"):
        raise ValueError(
            f"File {file_path} has an unsupported extension "
            f"'{file_path.suffix}' (must be '.json' or '.pkl')"
        )

    if file_path.suffix == ".json":
        with open(file_path) as f:
            return json.load(f)
    elif file_path.suffix == ".pkl":
        with open(file_path, "rb") as f:
            return pickle.load(f)

save_artifact(artifact, file_path)

Saves a pickle or json artifact (depending on the file extension).

Parameters:

Name Type Description Default
artifact Any

Python object to save.

 required
file_path str | Path

Path of the file to save.

 required

Raises:

Type Description
ValueError

If the file has an unsupported file extension.
Source code in src/qbm/utils/misc.py 
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
def save_artifact(artifact: Any, file_path: str | Path) -> None:
    """
    Saves a pickle or json artifact (depending on the file extension).

    Args:
        artifact: Python object to save.
        file_path: Path of the file to save.

    Raises:
        ValueError: If the file has an unsupported file extension.
    """
    if isinstance(file_path, str):
        file_path = Path(file_path)

    if not file_path.parent.exists():
        file_path.parent.mkdir(parents=True)

    if file_path.suffix not in (".json", ".pkl"):
        raise ValueError(
            f"File {file_path} has an unsupported extension "
            f"'{file_path.suffix}' (must be '.json' or '.pkl')"
        )

    if file_path.suffix == ".json":
        with open(file_path, "w") as f:
            json.dump(artifact, f, indent=4)
    elif file_path.suffix == ".pkl":
        with open(file_path, "wb") as f:
            pickle.dump(artifact, f)