Pricing

Bases: Module, IMixedEffect

Source code in wt_ml/layers/pricing.py
 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
class Pricing(Module, IMixedEffect):
    def __init__(
        self,
        encodings: dict[str, Any],
        hierarchy_categories: list[str | list[str]] | None = None,
        hyperparameters: Hyperparams | None = None,
        name: str | None = None,
    ):
        """Multiplicative price elasticity factor affecting baseline sales that also scales ROI of investments.

        Args:
            hierarchy (pd.DataFrame): The hierarchy used to build features learnt by the model to generate impacts.
            hyperparameters (Hyperparams | None, optional): An instance of `FileHyperparameterConfig` class
                                                            that stores all the hyperparameters of Pricing layer.
                                                            `Module` parent class sets these hyperparameters if None.
            name (str | None, optional): Name of the Pricing Layer.
                                         `Module` parent class sets name of the class if None.
        """
        super().__init__(hyperparameters=hyperparameters, name=name)
        self.encodings = encodings
        self.hierarchy_categories = hierarchy_categories

    def build(self, input_shapes: InputShapes):  # noqa: U100
        """Builds the `price_params_emb_layer` hierarchical variable
        for generating price elasticity curve for each granularity.
        Shape of the variable: (num_granular, 2). 2 denotes offset and exponent.

        Args:
            input_shapes (InputShapes): A tuple of tensor shapes of `price` and `hierarchy` passed to `__call__`.
        """
        n_instances = input_shapes.prices[2]
        self.pricing_params_emb_layer = self.hyperparameters.get_submodule(
            "pricing_params_emb_layer",
            module_type=HierchicalEmbedding,
            kwargs=dict(
                encodings=self.encodings,
                columns=self.hierarchy_categories,
                # 2 here denotes offset and exponent
                shape=[n_instances, 2],
                # This initializes to a state where any change in price reduces revenue while still making learning
                # other distributions easy. You can roughly think of this as price * (mult / (price + 1) ** 2)
                # Where mult is a specially calculated value so that the result with price = 1 is 1.
                # Order is offset (softplus), exponent (softplus + 1)
                bias_initializer=tf.constant_initializer(np.tile([[0.6, 0.6]], (n_instances, 1)).reshape(-1)),
                feature_names=[
                    [f"{signal}_offset", f"{signal}_exponent"] for signal in get_lookups(self.encodings["price_dev"])
                ],
            ),
            help="The embedding for the parameters for the pricing elasticity curve.",
        )

    def __call__(
        self, batch: PricingInput, training: bool = False, debug: bool = False, skip_metrics: bool = False
    ) -> PricingIntermediaries:
        """Pricing Layer Forward Propagation.
        We take in the mean normalized $price$ signal of shape `(num_time, num_granular, n_sim)`.
        Then we take the $offset$ and $exponent$ learnt by the model, each of shape `(num_granular,)`.
        The impact is calculated as follows:

        $volume = \\frac{normalization\\_mult} {(price + offset) ^ {exponent}}$

        $normalization\\_{mult} = (1 + offset) ^ {exponent}$

        $impact = volume * price$

        This $impact$ is of shape `(num_time, num_granular, n_sim)`

        > NOTE: normalization\\_mult is a factor to neglect the impact of prices which equal the average price.

        Args:
            price (TensorLike): mean normalized price_per_hl for each granularity each week.
                                Shape: (num_time, num_granular, n_sim)
            hierarchy (dict[str, TensorLike]): Hierarchical Placeholder for creating hierarchical variable.
            training (bool, optional): Whether this is a training or inference run. Defaults to False.

        Returns:
            PricingIntermediaries: Intermediate calculations like offset, asymptote, exponent, etc., and final impact.
        """
        params_emb = self.pricing_params_emb_layer(batch.hierarchy, training=training, skip_metrics=skip_metrics)
        offset_emb, exponent_emb = tf.unstack(params_emb, axis=2)
        offset = softplus(offset_emb * 10) + 0.01
        exponent = monotonic_sigmoid(exponent_emb / 4) * 4 + 1
        # Here we introduce a mult which is a normalization parameter
        # the equation for volume by price is: volume = mult / ((price + offset) ** exponent)
        # We want volume = 1 when price = 1 so we need to solve
        # 1 = mult / ((1 + offset) ** exponent)
        # mult = (1 + offset) ** exponent
        normalization_mult = (1 + offset) ** exponent
        volume = normalization_mult[:, None] / ((batch.prices + offset[:, None]) ** exponent[:, None])
        # Impact is the revenue of this volume at that price, so just the product.
        revenue = tf.math.multiply(
            volume, tf.where(batch.prices > 0, batch.prices, tf.math.reciprocal_no_nan(volume)), name="impact"
        )
        impact_by_signal = revenue
        # Reduce over signal axis
        impact = tf.math.reduce_prod(impact_by_signal, axis=2, name="impact")
        return PricingIntermediaries(
            offset_emb=offset_emb if debug else None,
            exponent_emb=exponent_emb if debug else None,
            offset=offset if debug else None,
            exponent=exponent if debug else None,
            volume=volume if debug else None,
            revenue=revenue if debug else None,
            impact_by_signal=impact_by_signal,
            impact=impact,
            signal_names=tf.gather(tf.convert_to_tensor(get_lookups(self.encodings["price_dev"])), batch.signal_index),
        )

__call__(batch, training=False, debug=False, skip_metrics=False)

Pricing Layer Forward Propagation. We take in the mean normalized \(price\) signal of shape (num_time, num_granular, n_sim). Then we take the \(offset\) and \(exponent\) learnt by the model, each of shape (num_granular,). The impact is calculated as follows:

\(volume = \frac{normalization\_mult} {(price + offset) ^ {exponent}}\)

\(normalization\_{mult} = (1 + offset) ^ {exponent}\)

\(impact = volume * price\)

This \(impact\) is of shape (num_time, num_granular, n_sim)

NOTE: normalization_mult is a factor to neglect the impact of prices which equal the average price.

Parameters:

Name Type Description Default
price TensorLike

mean normalized price_per_hl for each granularity each week. Shape: (num_time, num_granular, n_sim)

required
hierarchy dict[str, TensorLike]

Hierarchical Placeholder for creating hierarchical variable.

required
training bool

Whether this is a training or inference run. Defaults to False.

False

Returns:

Name Type Description
PricingIntermediaries PricingIntermediaries

Intermediate calculations like offset, asymptote, exponent, etc., and final impact.

Source code in wt_ml/layers/pricing.py
 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
def __call__(
    self, batch: PricingInput, training: bool = False, debug: bool = False, skip_metrics: bool = False
) -> PricingIntermediaries:
    """Pricing Layer Forward Propagation.
    We take in the mean normalized $price$ signal of shape `(num_time, num_granular, n_sim)`.
    Then we take the $offset$ and $exponent$ learnt by the model, each of shape `(num_granular,)`.
    The impact is calculated as follows:

    $volume = \\frac{normalization\\_mult} {(price + offset) ^ {exponent}}$

    $normalization\\_{mult} = (1 + offset) ^ {exponent}$

    $impact = volume * price$

    This $impact$ is of shape `(num_time, num_granular, n_sim)`

    > NOTE: normalization\\_mult is a factor to neglect the impact of prices which equal the average price.

    Args:
        price (TensorLike): mean normalized price_per_hl for each granularity each week.
                            Shape: (num_time, num_granular, n_sim)
        hierarchy (dict[str, TensorLike]): Hierarchical Placeholder for creating hierarchical variable.
        training (bool, optional): Whether this is a training or inference run. Defaults to False.

    Returns:
        PricingIntermediaries: Intermediate calculations like offset, asymptote, exponent, etc., and final impact.
    """
    params_emb = self.pricing_params_emb_layer(batch.hierarchy, training=training, skip_metrics=skip_metrics)
    offset_emb, exponent_emb = tf.unstack(params_emb, axis=2)
    offset = softplus(offset_emb * 10) + 0.01
    exponent = monotonic_sigmoid(exponent_emb / 4) * 4 + 1
    # Here we introduce a mult which is a normalization parameter
    # the equation for volume by price is: volume = mult / ((price + offset) ** exponent)
    # We want volume = 1 when price = 1 so we need to solve
    # 1 = mult / ((1 + offset) ** exponent)
    # mult = (1 + offset) ** exponent
    normalization_mult = (1 + offset) ** exponent
    volume = normalization_mult[:, None] / ((batch.prices + offset[:, None]) ** exponent[:, None])
    # Impact is the revenue of this volume at that price, so just the product.
    revenue = tf.math.multiply(
        volume, tf.where(batch.prices > 0, batch.prices, tf.math.reciprocal_no_nan(volume)), name="impact"
    )
    impact_by_signal = revenue
    # Reduce over signal axis
    impact = tf.math.reduce_prod(impact_by_signal, axis=2, name="impact")
    return PricingIntermediaries(
        offset_emb=offset_emb if debug else None,
        exponent_emb=exponent_emb if debug else None,
        offset=offset if debug else None,
        exponent=exponent if debug else None,
        volume=volume if debug else None,
        revenue=revenue if debug else None,
        impact_by_signal=impact_by_signal,
        impact=impact,
        signal_names=tf.gather(tf.convert_to_tensor(get_lookups(self.encodings["price_dev"])), batch.signal_index),
    )

__init__(encodings, hierarchy_categories=None, hyperparameters=None, name=None)

Multiplicative price elasticity factor affecting baseline sales that also scales ROI of investments.

Parameters:

Name Type Description Default
hierarchy DataFrame

The hierarchy used to build features learnt by the model to generate impacts.

required
hyperparameters Hyperparams | None

An instance of FileHyperparameterConfig class that stores all the hyperparameters of Pricing layer. Module parent class sets these hyperparameters if None.

None
name str | None

Name of the Pricing Layer. Module parent class sets name of the class if None.

None
Source code in wt_ml/layers/pricing.py
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def __init__(
    self,
    encodings: dict[str, Any],
    hierarchy_categories: list[str | list[str]] | None = None,
    hyperparameters: Hyperparams | None = None,
    name: str | None = None,
):
    """Multiplicative price elasticity factor affecting baseline sales that also scales ROI of investments.

    Args:
        hierarchy (pd.DataFrame): The hierarchy used to build features learnt by the model to generate impacts.
        hyperparameters (Hyperparams | None, optional): An instance of `FileHyperparameterConfig` class
                                                        that stores all the hyperparameters of Pricing layer.
                                                        `Module` parent class sets these hyperparameters if None.
        name (str | None, optional): Name of the Pricing Layer.
                                     `Module` parent class sets name of the class if None.
    """
    super().__init__(hyperparameters=hyperparameters, name=name)
    self.encodings = encodings
    self.hierarchy_categories = hierarchy_categories

build(input_shapes)

Builds the price_params_emb_layer hierarchical variable for generating price elasticity curve for each granularity. Shape of the variable: (num_granular, 2). 2 denotes offset and exponent.

Parameters:

Name Type Description Default
input_shapes InputShapes

A tuple of tensor shapes of price and hierarchy passed to __call__.

required
Source code in wt_ml/layers/pricing.py
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
def build(self, input_shapes: InputShapes):  # noqa: U100
    """Builds the `price_params_emb_layer` hierarchical variable
    for generating price elasticity curve for each granularity.
    Shape of the variable: (num_granular, 2). 2 denotes offset and exponent.

    Args:
        input_shapes (InputShapes): A tuple of tensor shapes of `price` and `hierarchy` passed to `__call__`.
    """
    n_instances = input_shapes.prices[2]
    self.pricing_params_emb_layer = self.hyperparameters.get_submodule(
        "pricing_params_emb_layer",
        module_type=HierchicalEmbedding,
        kwargs=dict(
            encodings=self.encodings,
            columns=self.hierarchy_categories,
            # 2 here denotes offset and exponent
            shape=[n_instances, 2],
            # This initializes to a state where any change in price reduces revenue while still making learning
            # other distributions easy. You can roughly think of this as price * (mult / (price + 1) ** 2)
            # Where mult is a specially calculated value so that the result with price = 1 is 1.
            # Order is offset (softplus), exponent (softplus + 1)
            bias_initializer=tf.constant_initializer(np.tile([[0.6, 0.6]], (n_instances, 1)).reshape(-1)),
            feature_names=[
                [f"{signal}_offset", f"{signal}_exponent"] for signal in get_lookups(self.encodings["price_dev"])
            ],
        ),
        help="The embedding for the parameters for the pricing elasticity curve.",
    )