1 | from geniusweb.issuevalue.Value import Value
|
---|
2 | from geniusweb.issuevalue.Bid import Bid
|
---|
3 | from geniusweb.issuevalue.Domain import Domain
|
---|
4 |
|
---|
5 | class FrequencyAnalyzer:
|
---|
6 | def __init__(self) -> None:
|
---|
7 | self.number_bids: int = 0
|
---|
8 | self.last_bid = None
|
---|
9 | self.domain: Domain
|
---|
10 |
|
---|
11 | self.frequency_table: dict[str, tuple[float, dict[Value, float], int]] = {}
|
---|
12 |
|
---|
13 | def set_domain(self, domain: Domain):
|
---|
14 | self.domain = domain
|
---|
15 |
|
---|
16 | def _init_table(self) -> None:
|
---|
17 | if self.last_bid is None:
|
---|
18 | raise MissingHistoryException()
|
---|
19 |
|
---|
20 | issues = self.domain.getIssues()
|
---|
21 |
|
---|
22 | # init frequency table
|
---|
23 | for issue in issues:
|
---|
24 | values = self.domain.getValues(issue)
|
---|
25 | value_freqs = { value : 0.0 for value in values }
|
---|
26 | self.frequency_table[issue] = (1.0/float(len(issues)), value_freqs, 0)
|
---|
27 |
|
---|
28 | issues_in_bid = self.last_bid.getIssues()
|
---|
29 |
|
---|
30 | # init with first bid
|
---|
31 | for issue in issues_in_bid:
|
---|
32 | freq, value_freqs, value_max_occurence = self.frequency_table[issue]
|
---|
33 |
|
---|
34 | freq: float = 1/len(issues_in_bid)
|
---|
35 | issue_value = self.last_bid.getValue(issue)
|
---|
36 |
|
---|
37 | if issue_value is None:
|
---|
38 | raise ValueIsNoneException()
|
---|
39 |
|
---|
40 | value_freqs[issue_value] = 1.0
|
---|
41 | value_max_occurence: int = 1
|
---|
42 |
|
---|
43 | self.frequency_table[issue] = (freq, value_freqs, value_max_occurence)
|
---|
44 |
|
---|
45 |
|
---|
46 | def _update_issue_frequency(self, bid: Bid, issue: str, n) -> None:
|
---|
47 | if self.last_bid is None:
|
---|
48 | raise MissingHistoryException()
|
---|
49 |
|
---|
50 | issues = self.domain.getIssues()
|
---|
51 |
|
---|
52 | # if an issue has the same value
|
---|
53 | if self.last_bid.getValue(issue) == bid.getValue(issue):
|
---|
54 | # update frequency of current bid
|
---|
55 | freq, value_freqs, value_max_occurence = self.frequency_table[issue]
|
---|
56 | self.frequency_table[issue] = ((freq * self.number_bids + 1)/float(self.number_bids + 1), value_freqs, value_max_occurence)
|
---|
57 |
|
---|
58 | for other_issue in issues:
|
---|
59 | # and 'compensate' this frequency change with others
|
---|
60 | if issue != other_issue:
|
---|
61 | other_freq, other_value_freqs, other_value_max_occurence = self.frequency_table[other_issue]
|
---|
62 | self.frequency_table[other_issue] = ((other_freq * self.number_bids)/float(self.number_bids + 1), other_value_freqs, other_value_max_occurence)
|
---|
63 |
|
---|
64 | def _update_issue_value_frequency(self, current_value, issue: str) -> None:
|
---|
65 | if current_value is None:
|
---|
66 | raise ValueIsNoneException()
|
---|
67 |
|
---|
68 | freq, value_freqs, value_max_occurence = self.frequency_table[issue]
|
---|
69 | current_freq = value_freqs[current_value]
|
---|
70 |
|
---|
71 | if current_freq == 1.0:
|
---|
72 | max_repeat = 1
|
---|
73 | else:
|
---|
74 | max_repeat = 0
|
---|
75 |
|
---|
76 | for value in self.domain.getIssuesValues()[issue]:
|
---|
77 | if value == current_value:
|
---|
78 | occurence = 1
|
---|
79 | else:
|
---|
80 | occurence = 0
|
---|
81 |
|
---|
82 | value_freqs[value] = ((value_freqs[value] * value_max_occurence) + occurence) / (value_max_occurence + max_repeat)
|
---|
83 |
|
---|
84 | self.frequency_table[issue] = (freq, value_freqs, value_max_occurence + max_repeat)
|
---|
85 |
|
---|
86 | def add_bid(self, bid: Bid, n: float =.1) -> None:
|
---|
87 | if bid is None:
|
---|
88 | return
|
---|
89 |
|
---|
90 | if self.last_bid is None:
|
---|
91 | self.last_bid = bid
|
---|
92 | self._init_table()
|
---|
93 | return
|
---|
94 |
|
---|
95 | for issue in self.domain.getIssues():
|
---|
96 | self._update_issue_frequency(bid, issue, n)
|
---|
97 | self._update_issue_value_frequency(bid.getValue(issue), issue)
|
---|
98 |
|
---|
99 | self.number_bids += 1
|
---|
100 |
|
---|
101 | def _get_max_value(self, issue: str) -> Value:
|
---|
102 | _, value_frequencies, _ = self.frequency_table[issue]
|
---|
103 | max_freq: float = -1.0
|
---|
104 | max_key = None
|
---|
105 |
|
---|
106 | for key, freq in value_frequencies.items():
|
---|
107 | if max_freq < freq:
|
---|
108 | max_freq = freq
|
---|
109 | max_key = key
|
---|
110 |
|
---|
111 | assert max_key is not None
|
---|
112 |
|
---|
113 | return max_key
|
---|
114 |
|
---|
115 | """
|
---|
116 | Returns an approximation of the opponents utility for the given bid
|
---|
117 | """
|
---|
118 | def get_utility(self, bid: Bid):
|
---|
119 | utility = 0.0
|
---|
120 |
|
---|
121 | for issue in self.domain.getIssues():
|
---|
122 | freq, value_freqs, _ = self.frequency_table[issue]
|
---|
123 | issue_value = bid.getValue(issue)
|
---|
124 | if issue_value is not None:
|
---|
125 | # Take the 'importance' of the current issue, and multiply it by the utility with the associated value
|
---|
126 | utility += freq * value_freqs[issue_value]
|
---|
127 | # sum of all importances is 1.0
|
---|
128 | # best values of each issue is always 1.0
|
---|
129 | # => max utility is 1.0, thus admissable
|
---|
130 | else:
|
---|
131 | utility += 0
|
---|
132 |
|
---|
133 | return utility
|
---|
134 |
|
---|
135 | """
|
---|
136 | Return a list of issues and the difference in their importance [0.0, 1.0]
|
---|
137 | The higher the number, the better the compatibility
|
---|
138 | """
|
---|
139 | def utility_compatibility(self, other_importance: dict[str, float]) -> dict[str, float]:
|
---|
140 | compatibility: dict[str, float] = dict()
|
---|
141 |
|
---|
142 | for issue in self.domain.getIssues():
|
---|
143 | freq, _, _ = self.frequency_table[issue]
|
---|
144 | compatibility[issue] = abs(other_importance[issue] - freq)
|
---|
145 |
|
---|
146 | return compatibility
|
---|
147 |
|
---|
148 | """
|
---|
149 | Return next predicted bid based on frequency analysis
|
---|
150 | """
|
---|
151 | def predict(self) -> Bid:
|
---|
152 | if len(self.frequency_table) == 0:
|
---|
153 | raise MissingHistoryException()
|
---|
154 |
|
---|
155 | prediction: dict[str, Value] = {}
|
---|
156 |
|
---|
157 | for issue in self.frequency_table:
|
---|
158 | prediction[issue] = self._get_max_value(issue)
|
---|
159 |
|
---|
160 | return Bid(prediction)
|
---|
161 |
|
---|
162 |
|
---|
163 | class MissingHistoryException(Exception):
|
---|
164 | def __init__(self, *args: object) -> None:
|
---|
165 | super().__init__(*args)
|
---|
166 |
|
---|
167 | class ValueIsNoneException(Exception):
|
---|
168 | def __init__(self, *args: object) -> None:
|
---|
169 | super().__init__(*args)
|
---|
170 |
|
---|
171 | class BidIsNoneException(Exception):
|
---|
172 | def __init__(self, *args: object) -> None:
|
---|
173 | super().__init__(*args)
|
---|