-
Notifications
You must be signed in to change notification settings - Fork 0
/
opponent_model.py
121 lines (91 loc) · 4.18 KB
/
opponent_model.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
from collections import defaultdict
from geniusweb.issuevalue.Bid import Bid
from geniusweb.issuevalue.DiscreteValueSet import DiscreteValueSet
from geniusweb.issuevalue.Domain import Domain
from geniusweb.issuevalue.Value import Value
class OpponentModel:
def __init__(self, domain: Domain):
self.offers = []
self.domain = domain
self.issue_estimators = {
i: IssueEstimator(v) for i, v in domain.getIssuesValues().items()
}
def update(self, bid: Bid):
# keep track of all bids received
self.offers.append(bid)
# update all issue estimators with the value that is offered for that issue
for issue_id, issue_estimator in self.issue_estimators.items():
issue_estimator.update(bid.getValue(issue_id))
def get_predicted_utility(self, bid: Bid):
if len(self.offers) == 0 or bid is None:
return 0
# initiate
total_issue_weight = 0.0
value_utilities = []
issue_weights = []
for issue_id, issue_estimator in self.issue_estimators.items():
# get the value that is set for this issue in the bid
value: Value = bid.getValue(issue_id)
# collect both the predicted weight for the issue and
# predicted utility of the value within this issue
value_utilities.append(issue_estimator.get_value_utility(value))
issue_weights.append(issue_estimator.weight)
total_issue_weight += issue_estimator.weight
# normalise the issue weights such that the sum is 1.0
if total_issue_weight == 0.0:
issue_weights = [1 / len(issue_weights) for _ in issue_weights]
else:
issue_weights = [iw / total_issue_weight for iw in issue_weights]
# calculate predicted utility by multiplying all value utilities with their issue weight
predicted_utility = sum(
[iw * vu for iw, vu in zip(issue_weights, value_utilities)]
)
return predicted_utility
class IssueEstimator:
def __init__(self, value_set: DiscreteValueSet):
if not isinstance(value_set, DiscreteValueSet):
raise TypeError(
"This issue estimator only supports issues with discrete values"
)
self.bids_received = 0
self.max_value_count = 0
self.num_values = value_set.size()
self.value_trackers = defaultdict(ValueEstimator)
self.weight = 0
def update(self, value: Value):
self.bids_received += 1
# get the value tracker of the value that is offered
value_tracker = self.value_trackers[value]
# register that this value was offered
value_tracker.update()
# update the count of the most common offered value
self.max_value_count = max([value_tracker.count, self.max_value_count])
# update predicted issue weight
# the intuition here is that if the values of the receiverd offers spread out over all
# possible values, then this issue is likely not important to the opponent (weight == 0.0).
# If all received offers proposed the same value for this issue,
# then the predicted issue weight == 1.0
equal_shares = self.bids_received / self.num_values
self.weight = (self.max_value_count - equal_shares) / (
self.bids_received - equal_shares
)
# recalculate all value utilities
for value_tracker in self.value_trackers.values():
value_tracker.recalculate_utility(self.max_value_count, self.weight)
def get_value_utility(self, value: Value):
if value in self.value_trackers:
return self.value_trackers[value].utility
return 0
class ValueEstimator:
def __init__(self):
self.count = 0
self.utility = 0
def update(self):
self.count += 1
def recalculate_utility(self, max_value_count: int, weight: float):
if weight < 1:
mod_value_count = ((self.count + 1) ** (1 - weight)) - 1
mod_max_value_count = ((max_value_count + 1) ** (1 - weight)) - 1
self.utility = mod_value_count / mod_max_value_count
else:
self.utility = 1