首頁
編程語言
C語言|JAVA編程
Python編程
網頁編程
ASP編程|PHP編程
JSP編程
數據庫
MYSQL數據庫|SqlServer數據庫
Oracle數據庫|DB2數據庫
您现在的位置: 程式碼之家 >> 編程語言 >  >> 更多程式語言 >> 更多程式設計知識

Python 加權隨機抽樣算法:兩種高效實作與效能對比

     編輯:更多程式設計知識

加權隨機抽樣常用於廣告分配、負載均衡、遊戲掉寶、A/B 測試。本文實作並對比 展開列表法 與 累積權重法,分析時間空間複雜度、適用場景與優化技巧。

問題定義


weights = {'A': 2, 'B': 2, 'C': 4, 'D': 6, 'E': 11}
# 總權重 25,E 被選中機率 11/25 = 44%

方法一:展開列表法(空間換時間)


import random

def weighted_choice_expanded(weights):
    """將每個元素按權重複製展開,隨機選索引"""
    expanded = []
    for item, w in weights.items():
        expanded.extend([item] * w)
    return random.choice(expanded)

# 優化:預計算展開列表
class WeightedSamplerExpanded:
    def __init__(self, weights):
        self.items = []
        for item, w in weights.items():
            self.items.extend([item] * w)
    
    def sample(self):
        return random.choice(self.items)

複雜度:建構 O(∑w),取樣 O(1),空間 O(∑w)。缺點:權重大或總和大時記憶體爆炸。

方法二:累積權重二分搜尋(標準解法)


import random
import bisect
import itertools

def weighted_choice_cumulative(weights):
    """累積權重 + 二分搜尋,O(log n) 取樣"""
    items = list(weights.keys())
    cum_weights = list(itertools.accumulate(weights.values()))
    total = cum_weights[-1]
    
    r = random.random() * total  # [0, total)
    idx = bisect.bisect_left(cum_weights, r)
    return items[idx]

# 類別版(預計算累積權重)
class WeightedSamplerCumulative:
    def __init__(self, weights):
        self.items = list(weights.keys())
        self.cum_weights = list(itertools.accumulate(weights.values()))
        self.total = self.cum_weights[-1]
    
    def sample(self):
        r = random.random() * self.total
        return self.items[bisect.bisect_left(self.cum_weights, r)]
    
    def sample_n(self, n):
        return [self.sample() for _ in range(n)]

複雜度:建構 O(n),取樣 O(log n),空間 O(n)。優勢:支援任意大權重、浮點數權重、記憶體固定。

方法三:Walker's Alias Method(O(1) 取樣,進階)


import random

class AliasMethod:
    """Vose's Alias Method:建構 O(n),取樣 O(1)"""
    def __init__(self, weights):
        self.n = len(weights)
        self.items = list(weights.keys())
        probs = [w / sum(weights.values()) for w in weights.values()]
        
        # 正規化機率至 n
        scaled = [p * self.n for p in probs]
        
        small = [i for i, p in enumerate(scaled) if p < 1.0]
        large = [i for i, p in enumerate(scaled) if p >= 1.0]
        
        self.prob = [0.0] * self.n
        self.alias = [0] * self.n
        
        while small and large:
            s = small.pop()
            l = large.pop()
            self.prob[s] = scaled[s]
            self.alias[s] = l
            scaled[l] = scaled[l] + scaled[s] - 1.0
            if scaled[l] < 1.0:
                small.append(l)
            else:
                large.append(l)
        
        while large:
            self.prob[large.pop()] = 1.0
        while small:
            self.prob[small.pop()] = 1.0
    
    def sample(self):
        i = random.randrange(self.n)
        if random.random() < self.prob[i]:
            return self.items[i]
        else:
            return self.items[self.alias[i]]

效能基準測試


import timeit
import random

weights = {f'item_{i}': random.randint(1, 100) for i in range(1000)}
total_weight = sum(weights.values())

# 預熱
sampler_cum = WeightedSamplerCumulative(weights)
sampler_alias = AliasMethod(weights)

# 100萬次取樣
n = 1_000_000

t_cum = timeit.timeit(sampler_cum.sample, number=n)
t_alias = timeit.timeit(sampler_alias.sample, number=n)

print(f'Cumulative (bisect): {t_cum:.3f}s / {n} samples')
print(f'Alias Method:        {t_alias:.3f}s / {n} samples')
print(f'Speedup: {t_cum/t_alias:.2f}x')

# 典型輸出:
# Cumulative (bisect): 0.452s / 1000000 samples
# Alias Method:        0.198s / 1000000 samples
# Speedup: 2.28x

選型指南

場景推薦方法理由
權重總和小 (<10000)、簡單腳本展開列表法程式碼最短、O(1) 取樣
一般生產環境、權重可變累積權重 + bisect平衡效能/記憶體/簡潔度
高頻取樣 (>10^6/s)、固定權重Alias Method真正 O(1),無分支預測失敗
需動態增減權重累積權重 + 平衡樹 / Segment Tree支援 O(log n) 更新
浮點數權重、精度敏感累積權重 (decimal) / Alias避免浮點誤差累積

實戰:廣告輪播加權隨機


class AdRotator:
    def __init__(self, ads: list[dict]):
        # ads: [{'id': 1, 'weight': 10, 'url': '...'}, ...]
        self.sampler = WeightedSamplerCumulative({ad['id']: ad['weight'] for ad in ads})
        self.ad_map = {ad['id']: ad for ad in ads}
    
    def get_ad(self):
        ad_id = self.sampler.sample()
        return self.ad_map[ad_id]

# 使用
ads = [
    {'id': 1, 'weight': 50, 'url': 'https://ad1.com'},
    {'id': 2, 'weight': 30, 'url': 'https://ad2.com'},
    {'id': 3, 'weight': 20, 'url': 'https://ad3.com'},
]
rotator = AdRotator(ads)
for _ in range(10):
    print(rotator.get_ad()['url'])

Python 3.10+ 內建支援


# Python 3.10+ random.choices 內建 weights 參數
import random

items = ['A', 'B', 'C', 'D', 'E']
weights = [2, 2, 4, 6, 11]

# 單次抽樣
random.choices(items, weights=weights, k=1)[0]

# 多次抽樣(有放回)
random.choices(items, weights=weights, k=100)

注意:random.choices 為有放回抽樣,每次獨立。無放回加權抽樣需自行實作。

總結

加權隨機抽樣首選 累積權重 + bisect,兼顧效能、記憶體、簡潔度。極高頻場景升級 Alias Method。Python 3.10+ 可直接用 random.choices。避免展開列表法用於大權重場景。

  1. 上一篇文章:
  2. 下一篇文章:
Copyright © 程式碼之家 All Rights Reserved