import random

# 基本隨機(jī)選擇
print(random.random())  # [0.0, 1.0)隨機(jī)浮點(diǎn)數(shù)
print(random.randint(1, 10))  # [1,10]隨機(jī)整數(shù)
print(random.choice(['a', 'b', 'c']))  # 隨機(jī)選擇元素

# 序列操作
items = list(range(10))
random.shuffle(items)  # 隨機(jī)洗牌
print("洗牌結(jié)果:", items)

# 加權(quán)選擇
weights = [0.1, 0.2, 0.7]  # 權(quán)重之和應(yīng)為1
print("加權(quán)選擇:", random.choices(['A', 'B', 'C'], weights=weights, k=5))

# 無(wú)放回抽樣
population = list(range(100))
sample = random.sample(population, k=10)  # 10個(gè)不重復(fù)樣本
print("無(wú)放回抽樣:", 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)

# 設(shè)置隨機(jī)種子
random.seed(42)  # 固定種子實(shí)現(xiàn)可重現(xiàn)結(jié)果

# 測(cè)試可重現(xiàn)性
first_run = [random.randint(1, 100) for _ in range(5)]
random.seed(42)
second_run = [random.randint(1, 100) for _ in range(5)]
print("可重現(xiàn)性驗(yàn)證:", first_run == second_run)  # True

# NumPy隨機(jī)種子
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)

# 連續(xù)分布抽樣
def exponential_distribution(lambd, size=1):
    """指數(shù)分布抽樣"""
    u = np.random.uniform(size=size)
    return -np.log(1 - u) / lambd

# 測(cè)試
samples = exponential_distribution(0.5, 1000)
print("指數(shù)分布均值:", np.mean(samples))  # 應(yīng)接近1/λ=2

class LoadBalancer:
    """基于權(quán)重的隨機(jī)負(fù)載均衡器"""
    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):
        """加權(quán)隨機(jī)選擇服務(wù)器"""
        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)

# 統(tǒng)計(jì)分布
counts = {'server1':0, 'server2':0, 'server3':0}
for _ in range(10000):
    server = lb.select_server()
    counts[server] += 1

print("負(fù)載分布:", counts)  # 比例應(yīng)接近5:3:2

class ABTestAssigner:
    """A/B測(cè)試分組系統(tǒng)"""
    def __init__(self, variants, weights=None):
        self.variants = variants
        self.weights = weights or [1/len(variants)]*len(variants)
        self.assignment = {}  # 用戶(hù)分配記錄
    
    def assign_user(self, user_id):
        """分配用戶(hù)到測(cè)試組"""
        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):
        """獲取分組統(tǒng)計(jì)"""
        from collections import Counter
        return Counter(self.assignment.values())

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

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

# 統(tǒng)計(jì)分組
stats = ab_test.get_assignment_stats()
print("A/B測(cè)試分組統(tǒng)計(jì):", stats)  # 應(yīng)接近400:400:200

def quickselect(arr, k):
    """快速選擇算法 (O(n)時(shí)間復(fù)雜度)"""
    if len(arr) == 1:
        return arr[0]
    
    # 隨機(jī)選擇樞軸
    pivot_idx = random.randint(0, len(arr)-1)
    pivot = arr[pivot_idx]
    
    # 分區(qū)
    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:  # 單位圓內(nèi)
            inside += 1
    return 4 * inside / n_samples

# 測(cè)試
pi_estimate = monte_carlo_pi()
print(f"π估計(jì)值: {pi_estimate} (誤差: {abs(pi_estimate - np.pi)/np.pi:.2%})")

# 向量化實(shí)現(xiàn)
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

# 性能對(duì)比
%timeit monte_carlo_pi(1000000)  # 約1秒
%timeit vectorized_monte_carlo(1000000)  # 約0.02秒

import secrets

# 生成安全隨機(jī)數(shù)
print("安全隨機(jī)整數(shù):", secrets.randbelow(100))
print("安全隨機(jī)字節(jié):", secrets.token_bytes(16))
print("安全隨機(jī)十六進(jìn)制:", secrets.token_hex(16))
print("安全隨機(jī)URL:", secrets.token_urlsafe(16))

# 安全隨機(jī)選擇
safe_choice = secrets.choice(['A', 'B', 'C'])
print("安全隨機(jī)選擇:", 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 = []
        
        # 必須包含小寫(xiě)字母
        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):
        """生成安全密碼"""
        # 確保每個(gè)字符集至少有一個(gè)字符
        password = []
        for char_set in self.char_sets:
            password.append(secrets.choice(char_set))
        
        # 填充剩余長(zhǎng)度
        all_chars = ''.join(self.char_sets)
        password.extend(secrets.choice(all_chars) for _ in range(self.length - len(password)))
        
        # 隨機(jī)化順序
        secrets.SystemRandom().shuffle(password)
        return ''.join(password)

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

def reservoir_sampling(stream, k):
    """水庫(kù)抽樣算法 (流式隨機(jī)抽樣)"""
    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

# 使用示例
# 模擬大型數(shù)據(jù)流
data_stream = (i for i in range(1000000))
sample = reservoir_sampling(data_stream, 100)
print("水庫(kù)抽樣結(jié)果:", sample[:10], "...")

# 驗(yàn)證均勻性
counts = [0]*10
for _ in range(1000):
    sample = reservoir_sampling(range(10), 1)
    counts[sample[0]] += 1
print("均勻性驗(yàn)證:", counts)  # 應(yīng)接近100

class DistributedSampler:
    """分布式隨機(jī)抽樣系統(tǒng)"""
    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):
        """處理數(shù)據(jù)塊"""
        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):
        """合并最終樣本"""
        # 合并所有工作節(jié)點(diǎn)的樣本
        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):
    # 每個(gè)worker處理1/4數(shù)據(jù)
    chunk = range(worker_id*250000, (worker_id+1)*250000)
    sampler.process_chunk(worker_id, chunk)

final_sample = sampler.get_final_sample()
print(f"分布式抽樣結(jié)果: {len(final_sample)}樣本")

# 傳統(tǒng)循環(huán)
def slow_random_array(size):
    return [random.random() for _ in range(size)]

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

# 性能對(duì)比
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):
    """并行隨機(jī)數(shù)生成"""
    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  # 處理余數(shù)
    
    # 并行生成
    with ThreadPoolExecutor(max_workers=num_processes) as executor:
        results = list(executor.map(
            lambda size: [random.random() for _ in range(size)], 
            chunks
        ))
    
    # 合并結(jié)果
    return [item for sublist in results for item in sublist]

# 使用示例
n = 1000000
random_numbers = parallel_random_generation(n)
print(f"生成{len(random_numbers)}個(gè)隨機(jī)數(shù)")

# 密碼學(xué)場(chǎng)景必須使用secrets
# 錯(cuò)誤做法
token = ''.join(random.choices('abc123', k=16))

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

# 實(shí)驗(yàn)環(huán)境設(shè)置種子
SEED = 42
random.seed(SEED)
np.random.seed(SEED)

# 生產(chǎn)環(huán)境不設(shè)種子
if ENV == 'production':
    random.seed(None)

# 根據(jù)場(chǎng)景選擇分布
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)

# 避免循環(huán)內(nèi)隨機(jī)
# 錯(cuò)誤做法
results = [func(random.random()) for _ in range(1000000)]

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

# 小數(shù)據(jù)直接抽樣
sample = random.sample(population, k)

# 大數(shù)據(jù)水庫(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):
    """
    加權(quán)隨機(jī)選擇

    參數(shù):
        items: 候選列表
        weights: 權(quán)重列表 (需與items等長(zhǎng))

    返回:
        隨機(jī)選擇的元素

    注意:
        權(quán)重列表會(huì)自動(dòng)歸一化
    """
    total = sum(weights)
    norm_weights = [w/total for w in weights]
    return random.choices(items, weights=norm_weights)[0]