ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

公平抽签算法实现:Fisher-Yates与蓄水池抽样详解

公平抽签算法实现:Fisher-Yates与蓄水池抽样详解 1. 问题背景与需求分析公平抽签是一个在各类竞赛、活动组织中常见的实际问题。假设有N个人参与抽签需要从中选出M个人如何设计一个算法确保每个人被选中的概率完全相同这个问题看似简单但在实际编程实现中需要考虑诸多细节。我在组织校园编程竞赛时曾遇到过类似需求从120名报名者中随机抽取30名参赛者。最初用Excel随机排序发现边缘位置出现概率偏差后来改用算法实现才解决公平性问题。这正是这类问题的典型应用场景。2. 核心算法解析2.1 洗牌算法Fisher-Yates Shuffle最经典的解决方案是Fisher-Yates洗牌算法其核心思想是通过倒序交换实现均匀排列import random def fair_draw(all_candidates, select_num): n len(all_candidates) for i in range(n-1, n-select_num-1, -1): j random.randint(0, i) all_candidates[i], all_candidates[j] all_candidates[j], all_candidates[i] return all_candidates[-select_num:]算法时间复杂度为O(M)空间复杂度O(1)。关键点在于从后向前遍历随机范围逐步缩小原地交换元素2.2 蓄水池抽样算法当N很大如百万级而M较小时更优的方案是蓄水池抽样import random def reservoir_sampling(data_stream, k): reservoir data_stream[:k] for i in range(k, len(data_stream)): j random.randint(0, i) if j k: reservoir[j] data_stream[i] return reservoir该算法特点只需单次遍历数据不需要预先知道总数据量每个元素被选中的概率严格相等3. 实现细节与边界处理3.1 随机数生成的质量常见误区是使用低质量随机源# 不推荐写法伪随机性不足 random.seed(time.time()) % 1000推荐做法# 使用系统级随机源Linux with open(/dev/urandom, rb) as f: seed int.from_bytes(f.read(4), big) random.seed(seed)3.2 重复抽签问题处理当需要多次执行抽签时需确保各次结果独立# 错误示例随机状态污染 results [fair_draw(data, 10) for _ in range(5)] # 可能产生关联 # 正确做法 def batch_draw(data, m, times): return [fair_draw(data.copy(), m) for _ in range(times)]4. 概率验证与测试方法4.1 蒙特卡洛测试通过大量重复实验验证概率分布from collections import defaultdict def probability_test(n, m, trials10000): counts defaultdict(int) for _ in range(trials): sample fair_draw(list(range(n)), m) for x in sample: counts[x] 1 return {k: v/trials for k, v in counts.items()}4.2 卡方检验统计检验各元素出现频率是否均匀from scipy.stats import chisquare def chi2_test(n, m, trials10000): probs probability_test(n, m, trials) observed list(probs.values()) expected [m/n]*n return chisquare(observed, f_expexpected)5. 性能优化实践5.1 大规模数据优化当N1e6时内存优化方案def streaming_draw(iterator, m): reservoir [] for i, item in enumerate(iterator): if i m: reservoir.append(item) else: r random.randint(0, i) if r m: reservoir[r] item return reservoir5.2 并行化实现多线程加速方案from multiprocessing import Pool def parallel_draw(data, m, workers4): chunk_size len(data) // workers with Pool(workers) as p: chunks [data[i*chunk_size:(i1)*chunk_size] for i in range(workers)] samples p.starmap(fair_draw, [(chunk, m//workers) for chunk in chunks]) return [item for sublist in samples for item in sublist]6. 实际应用案例6.1 在线考试系统组卷在某在线教育平台的组卷系统中我们实现了这样的抽题逻辑def select_questions(question_bank, counts): result {} for type_, (total, need) in counts.items(): pool question_bank[type_] selected fair_draw(pool, need) result[type_] selected return result6.2 会议抽奖系统大型年会抽奖系统的关键实现class LotterySystem: def __init__(self, participants): self.pool participants self.backup participants.copy() def draw(self, n): if n len(self.pool): self.pool self.backup.copy() winners fair_draw(self.pool, n) for winner in winners: self.pool.remove(winner) return winners7. 常见问题与解决方案7.1 随机性不足问题现象在小样本量时出现明显聚集解决方案使用密码学安全随机源增加熵池混合import secrets def secure_shuffle(items): n len(items) for i in range(n-1, 0, -1): j secrets.randbelow(i1) items[i], items[j] items[j], items[i] return items7.2 重复元素处理当列表中存在重复元素时需要特殊处理def dedup_draw(items, m): unique list(set(items)) if len(unique) m: raise ValueError(Not enough distinct items) return fair_draw(unique, m)8. 算法扩展与变种8.1 加权随机抽样某些场景需要按权重抽样import bisect import random def weighted_sample(items, weights, m): cumulative [] total 0 for w in weights: total w cumulative.append(total) result [] for _ in range(m): r random.uniform(0, total) idx bisect.bisect_left(cumulative, r) result.append(items[idx]) return result8.2 分层抽样当数据具有明显分层结构时def stratified_sample(data, strata, m): strata_counts {k: len(v) for k, v in data.items()} total sum(strata_counts.values()) samples [] for stratum, items in data.items(): stratum_m max(1, round(m * strata_counts[stratum] / total)) samples.extend(fair_draw(items, stratum_m)) return fair_draw(samples, m)
返回列表