-
Notifications
You must be signed in to change notification settings - Fork 1
/
minerva_s.py
197 lines (156 loc) · 8.28 KB
/
minerva_s.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
"""
This file is from r2b2 and has been modified to allow for
a margin other than zero under the null hypothesis.
"""
"""Minerva audit module."""
import math
from typing import List
import click
from audit_s import Audit_S
from contest import Contest
class Minerva_S(Audit_S):
"""Minerva audit implementation.
A Minerva audit is a type of risk-limiting audit that accounts for round-by-round auditor
decisions. For a given sample size (in the context of a round schedule), the audit software
calculates a minimum number of votes for the reported winner that must be found in the sample
to stop the audit and confirm the reported outcome.
Attributes:
alpha (float): Risk limit. Alpha represents the chance that, given an incorrectly called
election, the audit will fail to force a full recount.
max_fraction_to_draw (float): The maximum number of ballots the auditors are willing to draw
as a fraction of the ballots in the contest.
rounds (List[int]): Cumulative round schedule.
min_winner_ballots (List[int]): Stopping sizes (or kmins) respective to the round schedule.
contest (Contest): Contest to be audited.
"""
def __init__(self, alpha: float, max_fraction_to_draw: float, contest: Contest, null_margin: int = 0):
"""Initialize a Minerva audit."""
super().__init__(alpha, 0.0, max_fraction_to_draw, True, contest, null_margin)
self.min_sample_size = self.get_min_sample_size()
self.rounds = []
self.min_winner_ballots = []
self.realized_risks = []
def get_min_sample_size(self):
"""Computes the minimum sample size that has a stopping size (kmin).
Returns:
int: The minimum sample size of the audit.
"""
# TODO: Implement "meaningful minimum" sample size, perhaps with tolerance 10^-18.
# p0 is not .5 for contests with odd total ballots.
p0 = (self.contest.contest_ballots // 2) / self.contest.contest_ballots
num = math.log(1 / self.alpha)
denom = math.log(self.contest.winner_prop / p0)
return math.ceil(num / denom)
def next_sample_size(self, *args, **kwargs):
pass
def stopping_condition(self, votes_for_winner: int, verbose: bool = False) -> bool:
"""Check, without finding the kmin, whether the audit is complete."""
if len(self.rounds) < 1:
raise Exception('Attempted to call stopping condition without any rounds.')
tail_null = sum(self.distribution_null[votes_for_winner:])
#print("tail_null (minerva_s): "+str(tail_null))
tail_reported = sum(self.distribution_reported_tally[votes_for_winner:])
#print("tail_reported: "+str(tail_reported))
if verbose:
click.echo('\np-value: {}'.format(tail_null / tail_reported))
self.realized_risks.append(tail_null / tail_reported)
stop = self.alpha * tail_reported > tail_null
pvalue = tail_null / tail_reported
return {
"stop": stop,
"pvalue": pvalue
}
def next_min_winner_ballots(self, sample_size) -> int:
"""Compute kmin in interactive context."""
return self.find_kmin(False)
def compute_min_winner_ballots(self, rounds: List[int], *args, **kwargs):
"""Compute the minimum number of winner ballots for a round schedule.
Extend the audit's round schedule with the passed (partial) round schedule, and then extend
the audit's minimum number of winner ballots schedule with the corresponding minimums to
meet the stopping condition.
Args:
rounds (List[int]): A (partial) round schedule of the audit.
"""
if len(rounds) < 1:
raise ValueError('Round schedule must contain at least 1 round.')
if len(self.rounds) > 0 and rounds[0] <= self.rounds[-1]:
raise ValueError('Sample sizes must exceed past sample sizes.')
for i in range(0, len(rounds)):
if rounds[i] < self.min_sample_size:
raise ValueError('Sample size must be >= minimum sample size.')
if rounds[i] > self.contest.contest_ballots * self.max_fraction_to_draw:
raise ValueError(
'Sample size cannot exceed the maximum fraction of contest ballots to draw.')
if i >= 1 and rounds[i] <= rounds[i - 1]:
raise ValueError('Round schedule is cumulative and so must strictly increase.')
for sample_size in rounds:
self.rounds.append(sample_size)
self.current_dist_null()
self.current_dist_reported()
self.find_kmin(True)
self.truncate_dist_null()
self.truncate_dist_reported()
def find_kmin(self, append: bool):
"""Search for a kmin (minimum number of winner ballots) satisfying all stopping criteria.
Args:
append (bool): Optionally append the kmins to the min_winner_ballots list. This may
not always be desirable here because, for example, appending happens automatically
outside this method during an interactive audit.
"""
for possible_kmin in range(self.rounds[-1] // 2 + 1, len(self.distribution_null)):
tail_null = sum(self.distribution_null[possible_kmin:])
tail_reported = sum(self.distribution_reported_tally[possible_kmin:])
# Minerva's stopping criterion: tail_reported / tail_null > 1 / alpha.
if self.alpha * tail_reported > tail_null:
if append:
self.min_winner_ballots.append(possible_kmin)
return possible_kmin
# Sentinel of None plays nice with truncation.
if append:
self.min_winner_ballots.append(None)
return None
def compute_all_min_winner_ballots(self, max_sample_size: int = None, *args, **kwargs):
"""Compute the minimum number of winner ballots for the complete (that is, ballot-by-ballot)
round schedule. Note that Minerva ballot-by-ballot is equivalent to the BRAVO audit.
Note: Due to limited convolutional precision, results may be off somewhat after the
stopping probability very nearly equals 1.
Args:
max_sample_size (int): Optionally set the maximum sample size to generate stopping sizes
(kmins) up to. If not provided the maximum sample size is determined by max_frac_to_draw
and the total contest ballots.
Returns:
None, kmins are appended to the min_winner_ballots list.
"""
if len(self.rounds) > 0:
raise Exception("This audit already has an (at least partial) round schedule.")
if max_sample_size is None:
max_sample_size = math.ceil(self.contest.contest_ballots * self.max_fraction_to_draw)
if max_sample_size < self.min_sample_size:
raise ValueError("Maximum sample size must be greater than or equal to minimum size.")
if max_sample_size > self.contest.contest_ballots:
raise ValueError("Maximum sample size cannot exceed total contest ballots.")
for sample_size in range(self.min_sample_size, max_sample_size + 1):
self.rounds.append(sample_size)
self.current_dist_null()
self.current_dist_reported()
# First kmin computed directly.
if sample_size == self.min_sample_size:
current_kmin = self.find_kmin(True)
else:
tail_null = sum(self.distribution_null[current_kmin:])
tail_reported = sum(self.distribution_reported_tally[current_kmin:])
if self.alpha * tail_reported > tail_null:
self.min_winner_ballots.append(current_kmin)
else:
current_kmin += 1
self.min_winner_ballots.append(current_kmin)
self.truncate_dist_null()
self.truncate_dist_reported()
def compute_risk(self, *args, **kwargs):
"""Return the current risk level of an interactive Minerva audit. Non-interactive and bulk
Minerva audits are not considered here since the sampled number of reported winner ballots
is not available.
"""
if len(self.realized_risks) < 1:
return None
return min(self.realized_risks)