-
Notifications
You must be signed in to change notification settings - Fork 12
/
Copy pathbase.py
239 lines (178 loc) · 7.35 KB
/
base.py
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
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
# Copyright (c) 2023, Haruka Kiyohara, Ren Kishimoto, HAKUHODO Technologies Inc., and Hanjuku-kaso Co., Ltd. All rights reserved.
# Licensed under the Apache 2.0 License.
"""Abstract Base Class for Simulation."""
from abc import ABCMeta, abstractmethod
from dataclasses import dataclass
from typing import Tuple, Union
import numpy as np
@dataclass
class BaseSimulator(metaclass=ABCMeta):
"""Base class to calculate the outcome probability and stochastically determine auction result.
Imported as: :class:`rtbgym.envs.simulator.BaseSimulator`
"""
@abstractmethod
def generate_auction(self, search_volume: int) -> Tuple[np.ndarray]:
"""Sample ad and user pair for each auction.
Parameters
-------
search_volume: int, default=None (> 0)
Total number of auctions to generate.
Returns
-------
ad_ids: ndarray of shape (search_volume, )
IDs of the ads.
user_ids: ndarray of shape (search_volume, )
IDs of the users.
"""
raise NotImplementedError
@abstractmethod
def map_idx_to_features(
self, ad_ids: np.ndarray, user_ids: np.ndarray
) -> np.ndarray:
"""Map the ad and the user index into feature vectors.
Parameters
-------
ad_ids: array-like of shape (search_volume, )
IDs of the ads.
(search_volume is determined in RL environment.)
user_ids: array-like of shape (search_volume, )
IDs of the users.
(search_volume is determined in RL environment.)
Returns
-------
ad_feature_vector: ndarray of shape (search_volume/n_samples, ad_feature_dim)
Ad feature vector for each auction.
user_feature_vector: ndarray of shape (search_volume/n_samples, user_feature_dim)
User feature vector for each auction.
"""
raise NotImplementedError
@abstractmethod
def calc_and_sample_outcome(
self,
timestep: int,
ad_ids: np.ndarray,
user_ids: np.ndarray,
bid_prices: np.ndarray,
) -> Tuple[np.ndarray]:
"""Simulate bidding auction for given queries.
Parameters
-------
timestep: int (> 0)
Timestep in the RL environment.
ad_ids: array-like of shape (search_volume, )
IDs of the ads.
user_ids: array-like of shape (search_volume, )
IDs of the users.
bid_prices: array-like of shape(search_volume, )
Bid price for each action.
Returns
-------
costs: ndarray of shape (search_volume, )
Cost raised (i.e., second price) for each auction.
impressions: ndarray of shape (search_volume, )
Binary indicator of whether impression occurred or not for each auction.
clicks: ndarray of shape (search_volume, )
Binary indicator of whether click occurred or not for each auction.
conversions: ndarray of shape (search_volume, )
Binary indicator of whether conversion occurred or not for each auction.
"""
raise NotImplementedError
@dataclass
class BaseWinningPriceDistribution(metaclass=ABCMeta):
"""Base class to sample the winning price (i.e., second price) and compare it with the given bid price.
Imported as: class:`rtbgym.BaseWinningPriceDistribution`
"""
@abstractmethod
def sample_outcome(
self,
bid_prices: np.ndarray,
ad_ids: np.ndarray,
user_ids: np.ndarray,
ad_feature_vector: np.ndarray,
user_feature_vector: np.ndarray,
timestep: Union[int, np.ndarray],
) -> Tuple[np.ndarray]:
"""Stochastically determine impression and second price for each auction.
Parameters
-------
bid_prices: array-like of shape (search_volume, )
Bid price for each auction.
ad_ids: array-like of shape (search_volume/n_samples, )
Ad ids used for each auction.
user_ids: array-like of shape (search_volume/n_samples, )
User ids used for each auction.
ad_feature_vector: array-like of shape (search_volume/n_samples, ad_feature_dim)
Ad feature vector for each auction.
user_feature_vector: array-like of shape (search_volume/n_samples, user_feature_dim)
User feature vector for each auction.
timestep: {int, array-like of shape (n_samples, )}
Timestep in the RL environment.
Returns
-------
impressions: ndarray of shape (search_volume, )
Whether impression occurred for each auction.
winning_prices: ndarray of shape (search_volume, )
Sampled winning price for each auction.
"""
raise NotImplementedError
@dataclass
class BaseClickAndConversionRate(metaclass=ABCMeta):
"""Base class to Class to define ground-truth CTR/CVR.
Imported as: class:`rtbgym.BaseClickAndConversionRate`
"""
@abstractmethod
def calc_prob(
self,
ad_ids: np.ndarray,
user_ids: np.ndarray,
ad_feature_vector: np.ndarray,
user_feature_vector: np.ndarray,
timestep: Union[int, np.ndarray],
) -> np.ndarray:
"""Calculate Click Through Rate (CTR) / Conversion Rate (CVR).
Parameters
-------
ad_ids: array-like of shape (search_volume/n_samples, )
Ad ids used for each auction.
user_ids: array-like of shape (search_volume/n_samples, )
User ids used for each auction.
ad_feature_vector: array-like of shape (search_volume/n_samples, ad_feature_dim)
Ad feature vector for each auction.
user_feature_vector: array-like of shape (search_volume/n_samples, user_feature_dim)
User feature vector for each auction.
timestep: {int, array-like of shape (n_samples, )}
Timestep in the RL environment.
Returns
-------
ctrs/cvrs: ndarray of shape (search_volume/n_samples, )
Ground-truth CTR (i.e., click per impression) or CVR (i.e., conversion per click) for each auction.
"""
raise NotImplementedError
@abstractmethod
def sample_outcome(
self,
ad_ids: np.ndarray,
user_ids: np.ndarray,
ad_feature_vector: np.ndarray,
user_feature_vector: np.ndarray,
timestep: Union[int, np.ndarray],
) -> np.ndarray:
"""Stochastically determine whether click/conversion occurs or not.
Parameters
-------
ad_ids: array-like of shape (search_volume/n_samples, )
Ad ids used for each auction.
user_ids: array-like of shape (search_volume/n_samples, )
User ids used for each auction.
ad_feature_vector: array-like of shape (search_volume/n_samples, ad_feature_dim)
Ad feature vector for each auction.
user_feature_vector: array-like of shape (search_volume/n_samples, user_feature_dim)
User feature vector for each auction.
timestep: {int, array-like of shape (n_samples, )}
Timestep in the RL environment.
Returns
-------
clicks/conversions: array-like of shape (search_volume/n_samples, )
Whether click occurs (when impression=True) or whether conversion occurs (when click=True).
"""
raise NotImplementedError