# Copyright (c) Yuta Saito, Yusuke Narita, and ZOZO Technologies, Inc. All rights reserved.
# Licensed under the Apache 2.0 License.
"""Offline Bandit Algorithms."""
from dataclasses import dataclass
from typing import Tuple, Optional, Union
import numpy as np
from scipy.special import softmax
from sklearn.base import clone, ClassifierMixin, is_classifier
from sklearn.linear_model import LogisticRegression
from sklearn.utils import check_random_state, check_scalar
from tqdm import tqdm
from .base import BaseOfflinePolicyLearner
from ..utils import check_bandit_feedback_inputs
[docs]@dataclass
class IPWLearner(BaseOfflinePolicyLearner):
"""Off-policy learner with Inverse Probability Weighting.
Parameters
-----------
n_actions: int
Number of actions.
len_list: int, default=1
Length of a list of actions recommended in each impression.
When Open Bandit Dataset is used, 3 should be set.
base_classifier: ClassifierMixin
Machine learning classifier used to train an offline decision making policy.
References
------------
Miroslav DudÃk, Dumitru Erhan, John Langford, and Lihong Li.
"Doubly Robust Policy Evaluation and Optimization.", 2014.
Damien Lefortier, Adith Swaminathan, Xiaotao Gu, Thorsten Joachims, and Maarten de Rijke.
"Large-scale Validation of Counterfactual Learning Methods: A Test-Bed.", 2016.
"""
base_classifier: Optional[ClassifierMixin] = None
def __post_init__(self) -> None:
"""Initialize class."""
super().__post_init__()
if self.base_classifier is None:
self.base_classifier = LogisticRegression(random_state=12345)
else:
assert is_classifier(
self.base_classifier
), "base_classifier must be a classifier"
self.base_classifier_list = [
clone(self.base_classifier) for _ in np.arange(self.len_list)
]
def _create_train_data_for_opl(
self,
context: np.ndarray,
action: np.ndarray,
reward: np.ndarray,
pscore: np.ndarray,
**kwargs,
) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
"""Create training data for off-policy learning.
Parameters
-----------
context: array-like, shape (n_rounds, dim_context)
Context vectors in each round, i.e., :math:`x_t`.
action: array-like, shape (n_rounds,)
Action sampled by a behavior policy in each round of the logged bandit feedback, i.e., :math:`a_t`.
reward: array-like, shape (n_rounds,)
Observed rewards (or outcome) in each round, i.e., :math:`r_t`.
pscore: array-like, shape (n_rounds,), default=None
Propensity scores, the probability of selecting each action by behavior policy,
in the given logged bandit feedback.
Returns
--------
(X, sample_weight, y): Tuple[np.ndarray, np.ndarray, np.ndarray]
Feature vectors, sample weights, and outcome for training the base machine learning model.
"""
return context, (reward / pscore), action
[docs] def fit(
self,
context: np.ndarray,
action: np.ndarray,
reward: np.ndarray,
pscore: Optional[np.ndarray] = None,
position: Optional[np.ndarray] = None,
) -> None:
"""Fits an offline bandit policy using the given logged bandit feedback data.
Note
--------
This `fit` method trains a deterministic policy :math:`\\pi: \\mathcal{X} \\rightarrow \\mathcal{A}`
via a cost-sensitive classification reduction as follows:
.. math::
\\hat{\\pi}
& \\in \\arg \\max_{\\pi \\in \\Pi} \\hat{V}_{\\mathrm{IPW}} (\\pi ; \\mathcal{D}) \\\\
& = \\arg \\max_{\\pi \\in \\Pi} \\mathbb{E}_{\\mathcal{D}} \\left[\\frac{\\mathbb{I} \\{\\pi (x_{i})=a_{i} \\}}{\\pi_{b}(a_{i} | x_{i})} r_{i} \\right] \\\\
& = \\arg \\min_{\\pi \\in \\Pi} \\mathbb{E}_{\\mathcal{D}} \\left[\\frac{r_i}{\\pi_{b}(a_{i} | x_{i})} \\mathbb{I} \\{\\pi (x_{i}) \\neq a_{i} \\} \\right],
where :math:`\\mathbb{E}_{\\mathcal{D}} [\cdot]` is the empirical average over observations in :math:`\\mathcal{D}`.
See the reference for the details.
Parameters
-----------
context: array-like, shape (n_rounds, dim_context)
Context vectors in each round, i.e., :math:`x_t`.
action: array-like, shape (n_rounds,)
Action sampled by a behavior policy in each round of the logged bandit feedback, i.e., :math:`a_t`.
reward: array-like, shape (n_rounds,)
Observed rewards (or outcome) in each round, i.e., :math:`r_t`.
pscore: array-like, shape (n_rounds,), default=None
Action choice probabilities by a behavior policy (propensity scores), i.e., :math:`\\pi_b(a_t|x_t)`.
position: array-like, shape (n_rounds,), default=None
Positions of each round in the given logged bandit feedback.
If None is given, a learner assumes that there is only one position.
When `len_list` > 1, position has to be set.
"""
check_bandit_feedback_inputs(
context=context,
action=action,
reward=reward,
pscore=pscore,
position=position,
)
if pscore is None:
n_actions = np.int(action.max() + 1)
pscore = np.ones_like(action) / n_actions
if self.len_list == 1:
position = np.zeros_like(action, dtype=int)
else:
assert (
isinstance(position, np.ndarray) and position.ndim == 1
), f"when len_list > 1, position must be a 1-dimensional ndarray"
for position_ in np.arange(self.len_list):
X, sample_weight, y = self._create_train_data_for_opl(
context=context[position == position_],
action=action[position == position_],
reward=reward[position == position_],
pscore=pscore[position == position_],
)
self.base_classifier_list[position_].fit(
X=X, y=y, sample_weight=sample_weight
)
[docs] def predict(self, context: np.ndarray) -> np.ndarray:
"""Predict best actions for new data.
Note
--------
Action set predicted by this `predict` method can contain duplicate items.
If you want a non-repetitive action set, then please use the `sample_action` method.
Parameters
-----------
context: array-like, shape (n_rounds_of_new_data, dim_context)
Context vectors for new data.
Returns
-----------
action_dist: array-like, shape (n_rounds_of_new_data, n_actions, len_list)
Action choices by a classifier, which can contain duplicate items.
If you want a non-repetitive action set, please use the `sample_action` method.
"""
assert (
isinstance(context, np.ndarray) and context.ndim == 2
), "context must be 2-dimensional ndarray"
n_rounds = context.shape[0]
action_dist = np.zeros((n_rounds, self.n_actions, self.len_list))
for position_ in np.arange(self.len_list):
predicted_actions_at_position = self.base_classifier_list[
position_
].predict(context)
action_dist[
np.arange(n_rounds),
predicted_actions_at_position,
np.ones(n_rounds, dtype=int) * position_,
] += 1
return action_dist
[docs] def predict_score(self, context: np.ndarray) -> np.ndarray:
"""Predict non-negative scores for all possible products of action and position.
Parameters
-----------
context: array-like, shape (n_rounds_of_new_data, dim_context)
Context vectors for new data.
Returns
-----------
score_predicted: array-like, shape (n_rounds_of_new_data, n_actions, len_list)
Scores for all possible pairs of action and position predicted by a classifier.
"""
assert (
isinstance(context, np.ndarray) and context.ndim == 2
), "context must be 2-dimensional ndarray"
n_rounds = context.shape[0]
score_predicted = np.zeros((n_rounds, self.n_actions, self.len_list))
for position_ in np.arange(self.len_list):
score_predicteds_at_position = self.base_classifier_list[
position_
].predict_proba(context)
score_predicted[:, :, position_] = score_predicteds_at_position
return score_predicted
[docs] def sample_action(
self,
context: np.ndarray,
tau: Union[int, float] = 1.0,
random_state: Optional[int] = None,
) -> np.ndarray:
"""Sample (non-repetitive) actions based on scores predicted by a classifier.
Note
--------
This `sample_action` method samples a **non-repetitive** set of actions for new data :math:`x \\in \\mathcal{X}`
by first computing non-negative scores for all possible candidate products of action and position
:math:`(a, k) \\in \\mathcal{A} \\times \\mathcal{K}` (where :math:`\\mathcal{A}` is an action set and
:math:`\\mathcal{K}` is a position set), and using softmax function as follows:
.. math::
& P (A_1 = a_1 | x) = \\frac{\\mathrm{exp}(f(x,a_1,1) / \\tau)}{\\sum_{a^{\\prime} \\in \\mathcal{A}} \\mathrm{exp}( f(x,a^{\\prime},1) / \\tau)} , \\\\
& P (A_2 = a_2 | A_1 = a_1, x) = \\frac{\\mathrm{exp}(f(x,a_2,2) / \\tau)}{\\sum_{a^{\\prime} \\in \\mathcal{A} \\backslash \\{a_1\\}} \\mathrm{exp}(f(x,a^{\\prime},2) / \\tau )} ,
\\ldots
where :math:`A_k` is a random variable representing an action at a position :math:`k`.
:math:`\\tau` is a temperature hyperparameter.
:math:`f: \\mathcal{X} \\times \\mathcal{A} \\times \\mathcal{K} \\rightarrow \\mathbb{R}_{+}`
is a scoring function which is now implemented in the `predict_score` method.
Parameters
----------------
context: array-like, shape (n_rounds_of_new_data, dim_context)
Context vectors for new data.
tau: int or float, default=1.0
A temperature parameter, controlling the randomness of the action choice.
As :math:`\\tau \\rightarrow \\infty`, the algorithm will select arms uniformly at random.
random_state: int, default=None
Controls the random seed in sampling actions.
Returns
-----------
action: array-like, shape (n_rounds_of_new_data, n_actions, len_list)
Action sampled by a trained classifier.
"""
assert (
isinstance(context, np.ndarray) and context.ndim == 2
), "context must be 2-dimensional ndarray"
check_scalar(tau, name="tau", target_type=(int, float), min_val=0)
n_rounds = context.shape[0]
random_ = check_random_state(random_state)
action = np.zeros((n_rounds, self.n_actions, self.len_list))
score_predicted = self.predict_score(context=context)
for i in tqdm(np.arange(n_rounds), desc="[sample_action]", total=n_rounds):
action_set = np.arange(self.n_actions)
for position_ in np.arange(self.len_list):
score_ = softmax(score_predicted[i, action_set, position_] / tau)
action_sampled = random_.choice(action_set, p=score_, replace=False)
action[i, action_sampled, position_] = 1
action_set = np.delete(action_set, action_set == action_sampled)
return action
[docs] def predict_proba(
self,
context: np.ndarray,
tau: Union[int, float] = 1.0,
) -> np.ndarray:
"""Obtains action choice probabilities for new data based on scores predicted by a classifier.
Note
--------
This `predict_proba` method obtains action choice probabilities for new data :math:`x \\in \\mathcal{X}`
by first computing non-negative scores for all possible candidate actions
:math:`a \\in \\mathcal{A}` (where :math:`\\mathcal{A}` is an action set),
and using a Plackett-Luce ranking model as follows:
.. math::
P (A = a | x) = \\frac{\\mathrm{exp}(f(x,a) / \\tau)}{\\sum_{a^{\\prime} \\in \\mathcal{A}} \\mathrm{exp}(f(x,a^{\\prime}) / \\tau)},
where :math:`A` is a random variable representing an action, and :math:`\\tau` is a temperature hyperparameter.
:math:`f: \\mathcal{X} \\times \\mathcal{A} \\rightarrow \\mathbb{R}_{+}`
is a scoring function which is now implemented in the `predict_score` method.
**Note that this method can be used only when `len_list=1`, please use the `sample_action` method otherwise.**
Parameters
----------------
context: array-like, shape (n_rounds_of_new_data, dim_context)
Context vectors for new data.
tau: int or float, default=1.0
A temperature parameter, controlling the randomness of the action choice.
As :math:`\\tau \\rightarrow \\infty`, the algorithm will select arms uniformly at random.
Returns
-----------
choice_prob: array-like, shape (n_rounds_of_new_data, n_actions, len_list)
Action choice probabilities obtained by a trained classifier.
"""
assert (
self.len_list == 1
), f"predict_proba method can be used only when len_list = 1"
assert (
isinstance(context, np.ndarray) and context.ndim == 2
), "context must be 2-dimensional ndarray"
check_scalar(tau, name="tau", target_type=(int, float), min_val=0)
score_predicted = self.predict_score(context=context)
choice_prob = softmax(score_predicted / tau, axis=1)
return choice_prob