从基础到高级详解Python实现随机选择功能的完全指南

import random

# 基本随机选择
print(random.random())  # [0.0, 1.0)随机浮点数
print(random.randint(1, 10))  # [1,10]随机整数
print(random.choice(['a', 'b', 'c']))  # 随机选择元素

# 序列操作
items = list(range(10))
random.shuffle(items)  # 随机洗牌
print("洗牌结果:", items)

# 加权选择
weights = [0.1, 0.2, 0.7]  # 权重之和应为1
print("加权选择:", random.choices(['A', 'B', 'C'], weights=weights, k=5))

# 无放回抽样
population = list(range(100))
sample = random.sample(population, k=10)  # 10个不重复样本
print("无放回抽样:", sample)

# 有放回抽样
with_replacement = [random.choice(population) for _ in range(10)]
print("有放回抽样:", with_replacement)

# 分层抽样
strata = {
    'low': list(range(0, 33)),
    'medium': list(range(33, 66)),
    'high': list(range(66, 100))
}

stratified_sample = []
for stratum, values in strata.items():
    sample_size = max(1, int(len(values) * 0.1))  # 每层10%
    stratified_sample.extend(random.sample(values, sample_size))
    
print("分层抽样:", stratified_sample)

# 设置随机种子
random.seed(42)  # 固定种子实现可重现结果

# 测试可重现性
first_run = [random.randint(1, 100) for _ in range(5)]
random.seed(42)
second_run = [random.randint(1, 100) for _ in range(5)]
print("可重现性验证:", first_run == second_run)  # True

# NumPy随机种子
import numpy as np
np.random.seed(42)

def custom_distribution_sampling(dist, size=1):
    """自定义离散分布抽样"""
    # dist: 概率字典 {'A':0.2, 'B':0.5, 'C':0.3}
    items, probs = zip(*dist.items())
    return np.random.choice(items, size=size, p=probs)

# 使用示例
dist = {'success': 0.3, 'failure': 0.5, 'pending': 0.2}
samples = custom_distribution_sampling(dist, 10)
print("自定义分布抽样:", samples)

# 连续分布抽样
def exponential_distribution(lambd, size=1):
    """指数分布抽样"""
    u = np.random.uniform(size=size)
    return -np.log(1 - u) / lambd

# 测试
samples = exponential_distribution(0.5, 1000)
print("指数分布均值:", np.mean(samples))  # 应接近1/λ=2

class LoadBalancer:
    """基于权重的随机负载均衡器"""
    def __init__(self):
        self.servers = {}  # {server: weight}
        self.total_weight = 0
    
    def add_server(self, server, weight):
        self.servers[server] = weight
        self.total_weight += weight
    
    def remove_server(self, server):
        if server in self.servers:
            self.total_weight -= self.servers.pop(server)
    
    def select_server(self):
        """加权随机选择服务器"""
        rand = random.uniform(0, self.total_weight)
        cumulative = 0
        for server, weight in self.servers.items():
            cumulative += weight
            if rand <= cumulative:
                return server
        return None

# 使用示例
lb = LoadBalancer()
lb.add_server('server1', 5)
lb.add_server('server2', 3)
lb.add_server('server3', 2)

# 统计分布
counts = {'server1':0, 'server2':0, 'server3':0}
for _ in range(10000):
    server = lb.select_server()
    counts[server] += 1

print("负载分布:", counts)  # 比例应接近5:3:2

class ABTestAssigner:
    """A/B测试分组系统"""
    def __init__(self, variants, weights=None):
        self.variants = variants
        self.weights = weights or [1/len(variants)]*len(variants)
        self.assignment = {}  # 用户分配记录
    
    def assign_user(self, user_id):
        """分配用户到测试组"""
        if user_id in self.assignment:
            return self.assignment[user_id]
        
        variant = random.choices(self.variants, weights=self.weights, k=1)[0]
        self.assignment[user_id] = variant
        return variant
    
    def get_assignment_stats(self):
        """获取分组统计"""
        from collections import Counter
        return Counter(self.assignment.values())

# 使用示例
ab_test = ABTestAssigner(['A', 'B', 'C'], weights=[0.4, 0.4, 0.2])

# 模拟用户分配
user_ids = [f"user_{i}" for i in range(1000)]
assignments = [ab_test.assign_user(uid) for uid in user_ids]

# 统计分组
stats = ab_test.get_assignment_stats()
print("A/B测试分组统计:", stats)  # 应接近400:400:200

def quickselect(arr, k):
    """快速选择算法 (O(n)时间复杂度)"""
    if len(arr) == 1:
        return arr[0]
    
    # 随机选择枢轴
    pivot_idx = random.randint(0, len(arr)-1)
    pivot = arr[pivot_idx]
    
    # 分区
    left = [x for i, x in enumerate(arr) if x <= pivot and i != pivot_idx]
    right = [x for i, x in enumerate(arr) if x > pivot]
    
    # 递归选择
    if k < len(left):
        return quickselect(left, k)
    elif k == len(left):
        return pivot
    else:
        return quickselect(right, k - len(left) - 1)

# 使用示例
data = [3, 1, 4, 1, 5, 9, 2, 6]
k = 4  # 第5小元素 (0-indexed)
print(f"第{k+1}小元素:", quickselect(data, k))  # 4

def monte_carlo_pi(n_samples=1000000):
    """蒙特卡洛法估算π值"""
    inside = 0
    for _ in range(n_samples):
        x, y = random.random(), random.random()
        if x**2 + y**2 <= 1:  # 单位圆内
            inside += 1
    return 4 * inside / n_samples

# 测试
pi_estimate = monte_carlo_pi()
print(f"π估计值: {pi_estimate} (误差: {abs(pi_estimate - np.pi)/np.pi:.2%})")

# 向量化实现
def vectorized_monte_carlo(n_samples=1000000):
    """向量化蒙特卡洛"""
    points = np.random.random((n_samples, 2))
    inside = np.sum(np.linalg.norm(points, axis=1) <= 1)
    return 4 * inside / n_samples

# 性能对比
%timeit monte_carlo_pi(1000000)  # 约1秒
%timeit vectorized_monte_carlo(1000000)  # 约0.02秒

import secrets

# 生成安全随机数
print("安全随机整数:", secrets.randbelow(100))
print("安全随机字节:", secrets.token_bytes(16))
print("安全随机十六进制:", secrets.token_hex(16))
print("安全随机URL:", secrets.token_urlsafe(16))

# 安全随机选择
safe_choice = secrets.choice(['A', 'B', 'C'])
print("安全随机选择:", safe_choice)

# 安全令牌生成
def generate_api_key(length=32):
    """生成安全API密钥"""
    alphabet = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
    return ''.join(secrets.choice(alphabet) for _ in range(length))

print("API密钥:", generate_api_key())

class PasswordGenerator:
    """安全密码生成器"""
    def __init__(self, length=12, use_upper=True, use_digits=True, use_special=True):
        self.length = length
        self.char_sets = []
        
        # 必须包含小写字母
        self.char_sets.append("abcdefghijklmnopqrstuvwxyz")
        
        if use_upper:
            self.char_sets.append("ABCDEFGHIJKLMNOPQRSTUVWXYZ")
        if use_digits:
            self.char_sets.append("0123456789")
        if use_special:
            self.char_sets.append("!@#$%^&*()_+-=[]{}|;:,.<>?")
    
    def generate(self):
        """生成安全密码"""
        # 确保每个字符集至少有一个字符
        password = []
        for char_set in self.char_sets:
            password.append(secrets.choice(char_set))
        
        # 填充剩余长度
        all_chars = ''.join(self.char_sets)
        password.extend(secrets.choice(all_chars) for _ in range(self.length - len(password)))
        
        # 随机化顺序
        secrets.SystemRandom().shuffle(password)
        return ''.join(password)

# 使用示例
generator = PasswordGenerator(length=16, use_special=True)
print("安全密码:", generator.generate())

def reservoir_sampling(stream, k):
    """水库抽样算法 (流式随机抽样)"""
    reservoir = []
    for i, item in enumerate(stream):
        if i < k:
            reservoir.append(item)
        else:
            j = random.randint(0, i)
            if j < k:
                reservoir[j] = item
    return reservoir

# 使用示例
# 模拟大型数据流
data_stream = (i for i in range(1000000))
sample = reservoir_sampling(data_stream, 100)
print("水库抽样结果:", sample[:10], "...")

# 验证均匀性
counts = [0]*10
for _ in range(1000):
    sample = reservoir_sampling(range(10), 1)
    counts[sample[0]] += 1
print("均匀性验证:", counts)  # 应接近100

class DistributedSampler:
    """分布式随机抽样系统"""
    def __init__(self, k, num_workers=4):
        self.k = k  # 总样本量
        self.num_workers = num_workers
        self.per_worker_k = k // num_workers
        self.reservoirs = [None] * num_workers
    
    def process_chunk(self, worker_id, chunk):
        """处理数据块"""
        reservoir = []
        for i, item in enumerate(chunk):
            if i < self.per_worker_k:
                reservoir.append(item)
            else:
                j = random.randint(0, i)
                if j < self.per_worker_k:
                    reservoir[j] = item
        self.reservoirs[worker_id] = reservoir
    
    def get_final_sample(self):
        """合并最终样本"""
        # 合并所有工作节点的样本
        all_samples = []
        for res in self.reservoirs:
            all_samples.extend(res)
        
        # 二次抽样
        return random.sample(all_samples, self.k)

# 使用示例
sampler = DistributedSampler(k=1000, num_workers=4)

# 模拟分布式处理
for worker_id in range(4):
    # 每个worker处理1/4数据
    chunk = range(worker_id*250000, (worker_id+1)*250000)
    sampler.process_chunk(worker_id, chunk)

final_sample = sampler.get_final_sample()
print(f"分布式抽样结果: {len(final_sample)}样本")

# 传统循环
def slow_random_array(size):
    return [random.random() for _ in range(size)]

# NumPy向量化
def fast_random_array(size):
    return np.random.random(size)

# 性能对比
size = 1000000
%timeit slow_random_array(size)  # 约100ms
%timeit fast_random_array(size)  # 约5ms

# 多分布向量化
def vectorized_random_samples(size):
    """多分布向量化抽样"""
    uniform = np.random.random(size)
    normal = np.random.normal(0, 1, size)
    poisson = np.random.poisson(5, size)
    return uniform, normal, poisson

from concurrent.futures import ThreadPoolExecutor
import multiprocessing

def parallel_random_generation(n, num_processes=None):
    """并行随机数生成"""
    if num_processes is None:
        num_processes = multiprocessing.cpu_count()
    
    # 分块
    chunk_size = n // num_processes
    chunks = [chunk_size] * num_processes
    chunks[-1] += n % num_processes  # 处理余数
    
    # 并行生成
    with ThreadPoolExecutor(max_workers=num_processes) as executor:
        results = list(executor.map(
            lambda size: [random.random() for _ in range(size)], 
            chunks
        ))
    
    # 合并结果
    return [item for sublist in results for item in sublist]

# 使用示例
n = 1000000
random_numbers = parallel_random_generation(n)
print(f"生成{len(random_numbers)}个随机数")

# 密码学场景必须使用secrets
# 错误做法
token = ''.join(random.choices('abc123', k=16))

# 正确做法
token = secrets.token_urlsafe(16)

# 实验环境设置种子
SEED = 42
random.seed(SEED)
np.random.seed(SEED)

# 生产环境不设种子
if ENV == 'production':
    random.seed(None)

# 根据场景选择分布
if scenario == 'normal':
    samples = np.random.normal(0, 1, 1000)
elif scenario == 'poisson':
    samples = np.random.poisson(5, 1000)
elif scenario == 'custom':
    samples = custom_distribution_sampling(dist)

# 避免循环内随机
# 错误做法
results = [func(random.random()) for _ in range(1000000)]

# 正确做法
randoms = np.random.random(1000000)
results = [func(x) for x in randoms]

# 小数据直接抽样
sample = random.sample(population, k)

# 大数据水库抽样
sample = reservoir_sampling(data_stream, k)

class TestRandomUtils(unittest.TestCase):
    def test_reservoir_sampling(self):
        stream = range(100)
        sample = reservoir_sampling(stream, 10)
        self.assertEqual(len(sample), 10)
        self.assertTrue(all(0 <= x < 100 for x in sample))

    def test_weighted_choice(self):
        choices = ['A', 'B', 'C']
        weights = [0.1, 0.1, 0.8]
        counts = {c:0 for c in choices}
        for _ in range(1000):
            choice = weighted_choice(choices, weights)
            counts[choice] += 1
        self.assertAlmostEqual(counts['C']/1000, 0.8, delta=0.05)

def weighted_selection(items, weights):
    """
    加权随机选择

    参数:
        items: 候选列表
        weights: 权重列表 (需与items等长)

    返回:
        随机选择的元素

    注意:
        权重列表会自动归一化
    """
    total = sum(weights)
    norm_weights = [w/total for w in weights]
    return random.choices(items, weights=norm_weights)[0]