布洛托博弈模拟系统：用Streamlit构建布洛托博弈模拟系统的全链路解析

一、应用架构概览

该应用是一个基于 Streamlit 构建的交互式 Web 应用，核心功能包括：

• 参数设置：用户通过侧边栏设置兵力、战场数量、策略类型等参数。
• 博弈模拟：根据策略生成双方兵力分配和战场争夺结果。
• 动态可视化：通过动画展示兵力部署和战场控制过程。
• 结果分析：以图表和数据展示胜负分布、价值控制等关键指标。

二、核心模块详解

1. 动画生成模块：animate_blotto_battlefield

功能：生成战场动态演示 GIF。
实现细节：

• 使用 Matplotlib 和 Streamlit 的动画功能（FuncAnimation）。
• 通过 FuncAnimation 动态更新战场状态图（散点图表示兵力分布）。
• 限制条件：当战场数量 > 15 时禁用动画（性能优化）。

代码亮点：

• 使用 st.image 展示生成的 GIF。
• 动画逻辑清晰，通过 update 函数逐帧更新数据。

2. 模拟核心模块：blotto_game

功能：执行布洛托博弈模拟，返回结果和动画路径。
参数：

• total_troops：总兵力。
• n_targets：战场数量。
• strategy_type：策略类型（随机/指数分布/混合/纳什均衡）。
• battlefield_value：用户自定义战场价值权重。
• animate：是否生成动画。

关键逻辑：

• 兵力分配：根据策略类型调用不同分配函数（如 random, exponential, nash_equilibrium）。
• 胜负判定：比较双方控制的战场价值总和，确定最终胜负。
• 结果结构：

• {
    "player1": 数组（兵力分配）,
    "player2": 数组（兵力分配）,
    "result": 数组（1/2/0 表示胜负）,
    "p1_value": 玩家1控制价值总和,
    "p2_value": 玩家2控制价值总和,
    "winner": 1/2/0（最终胜者）
  }

策略类型说明：

• 随机分配：均匀随机分布兵力。
• 指数分布：模仿真实博弈中的重点防御策略（少数战场集中兵力）。
• 混合策略：组合多种基础策略。
• 纳什均衡：理论上最优策略，双方无法单方面改变策略获得更好结果。

3. 用户界面模块：main

功能：构建 Streamlit 交互界面。
布局设计：

• 侧边栏：参数控制面板，包含兵力滑块、战场数量、策略选择、自定义价值输入等。
• 主界面：显示模拟结果、图表、动画及原始数据。

关键组件：

• st.slider：设置兵力和战场数量。
• st.selectbox：选择策略类型（随机/指数分布/混合/纳什均衡）。
• st.checkbox：启用动画、显示原始数据。
• st.button：触发模拟运行。
• st.metric：展示关键指标（如控制价值、胜负率）。
• st.pyplot：显示柱状图（胜负分布、兵力与价值对比）。
• st.image：展示动态 GIF。

交互流程：

1. 用户在侧边栏设置参数。
2. 点击 "运行战场模拟" 按钮触发 blotto_game。
3. 模拟完成后，主界面展示结果图表、动画及原始数据。
4. 用户可点击 "重新运行相同配置" 按钮重复模拟。

三、性能与用户体验优化

1. 动画性能限制：

   • 当战场数量 > 15 时禁用动画，避免高负载。
   • 使用 st.spinner 显示加载状态，提升用户体验。

2. 数据验证：

   • 自定义战场价值输入时，验证数量是否与战场数量一致，避免错误。

3. 策略说明：

   • 使用 st.info 显示不同策略的说明，增强用户理解。

4. 错误处理：

   • 输入验证失败时显示错误提示（如战场价值数量不匹配）。

import numpy as np
import matplotlib.pyplot as plt
import matplotlib.animation as animation
import matplotlib.cm as cm
from matplotlib.font_manager import FontProperties
import matplotlib.patches as patches
import streamlit as st
import time
import os
from io import BytesIO
from PIL import Image
from scipy.spatial import distance
import sys
import matplotlib as mpl

# 解决中文显示问题
def set_chinese_font():
    """设置中文字体支持"""
    # 尝试使用系统字体
    try:
        # Windows系统
        if sys.platform.startswith('win'):
            plt.rcParams['font.sans-serif'] = ['SimHei']  # 黑体
            plt.rcParams['axes.unicode_minus'] = False
        # macOS系统
        elif sys.platform.startswith('darwin'):
            plt.rcParams['font.sans-serif'] = ['Arial Unicode MS']  # macOS自带中文字体
        # Linux系统
        else:
            plt.rcParams['font.sans-serif'] = ['WenQuanYi Micro Hei']  # 文泉驿微米黑
        
        # 验证字体是否可用
        test_text = "中文测试"
        fig, ax = plt.subplots()
        ax.text(0.5, 0.5, test_text, ha='center', va='center')
        plt.close(fig)
    except:
        # 如果系统字体不可用，尝试使用内置字体
        try:
            # 使用DejaVu Sans作为后备方案
            plt.rcParams['font.sans-serif'] = ['DejaVu Sans', 'Arial', 'sans-serif']
            plt.rcParams['axes.unicode_minus'] = False
        except:
            # 最后尝试使用matplotlib内置字体
            plt.rcParams['font.sans-serif'] = ['DejaVu Sans']
            plt.rcParams['axes.unicode_minus'] = False

# 设置中文字体
set_chinese_font()

# 添加临时文件管理函数
def clear_temp_files():
    """清除临时文件"""
    for file in os.listdir():
        if file.startswith("blotto_temp") and (file.endswith(".gif") or file.endswith(".png")):
            try:
                os.remove(file)
            except:
                pass

def generate_battlefield_positions(n_targets, size=100, min_dist=15):
    """生成不重叠的战场位置（使用最小距离约束）"""
    positions = []
    attempts = 0
    
    while len(positions) < n_targets and attempts < 1000:
        attempt = np.random.rand(2) * size
        valid = True
        
        # 检查最小距离
        for pos in positions:
            if distance.euclidean(attempt, pos) < min_dist:
                valid = False
                break
        
        if valid:
            positions.append(attempt)
        
        attempts += 1
    
    # 如果无法生成所有点，使用次优方案
    if len(positions) < n_targets:
        st.warning(f"警告：无法为所有战场生成足够间距的位置（生成了 {len(positions)}/{n_targets}）")
        # 补充剩余位置
        while len(positions) < n_targets:
            positions.append(np.random.rand(2) * size)
    
    return np.array(positions)

def random_partition(total, n_parts):
    """生成 n_parts 个非负整数，和为 total（随机分配）"""
    if n_parts == 1:
        return np.array([total])
    cuts = np.sort(np.random.choice(range(1, total), n_parts - 1, replace=False))
    parts = np.diff([0] + cuts.tolist() + [total])
    return np.maximum(parts, 0)  # 确保非负

def exponential_distribution(total, n_targets, scale=0.2):
    """指数分布策略（更接近博弈均衡）"""
    weights = np.random.exponential(scale, n_targets)
    weights /= weights.sum()
    troops = (weights * total).astype(int)

    # 调整余数
    remainder = total - troops.sum()
    if remainder != 0:
        # 安全调整
        if remainder > 0:
            indices = np.random.choice(n_targets, remainder, replace=False)
            troops[indices] += 1
        else:
            for _ in range(-remainder):
                valid_idx = np.where(troops > 0)[0]
                if len(valid_idx) > 0:
                    idx = np.random.choice(valid_idx)
                    troops[idx] -= 1
    
    return troops

def mixed_strategy(total, n_targets, base_strategies=3):
    """混合策略（多个基础策略的线性组合）"""
    strategies = []
    for i in range(base_strategies):
        scale = 0.1 + i * 0.15
        strategies.append(exponential_distribution(total, n_targets, scale))
    
    weights = np.random.dirichlet(np.ones(base_strategies))
    troops = np.rint(np.dot(weights, strategies)).astype(int)
    
    # 安全调整余数
    remainder = total - troops.sum()
    if remainder != 0:
        if remainder > 0:
            indices = np.random.choice(n_targets, remainder, replace=False)
            troops[indices] += 1
        else:
            for _ in range(-remainder):
                valid_idx = np.where(troops > 0)[0]
                if len(valid_idx) > 0:
                    idx = np.random.choice(valid_idx)
                    troops[idx] -= 1
    
    return troops

def nash_equilibrium_pure(total, n_targets, values):
    """改进的纳什均衡计算（考虑战场价值）"""
    # 生成候选策略
    num_strategies = min(15, total // 3)  # 限制策略数量
    strategies = [exponential_distribution(total, n_targets, scale=0.2 + i*0.1) 
                  for i in range(num_strategies)]
    
    # 构建收益矩阵（使用战场价值）
    payoff_matrix = np.zeros((num_strategies, num_strategies))
    
    for i, s1 in enumerate(strategies):
        for j, s2 in enumerate(strategies):
            # 改进的收益计算：价值加权
            p1_wins = np.sum((s1 > s2) * values)
            p2_wins = np.sum((s2 > s1) * values)
            payoff_matrix[i, j] = p1_wins - p2_wins  # P1的净收益
    
    # 寻找纳什均衡（纯策略）
    row_min = np.min(payoff_matrix, axis=1)
    col_max = np.max(payoff_matrix, axis=0)
    
    equilibria = []
    for i in range(num_strategies):
        for j in range(num_strategies):
            if payoff_matrix[i, j] >= row_min[i] and payoff_matrix[i, j] <= col_max[j]:
                equilibria.append((strategies[i], strategies[j]))
    
    return equilibria[0] if equilibria else None

def animate_blotto_battlefield(player1, player2, result, values, animation_speed=500):
    """动态动画展示战场攻防"""
    n_targets = len(player1)
    battlefield = generate_battlefield_positions(n_targets)
    
    fig, ax = plt.subplots(figsize=(10, 6))
    ax.set_xlim(0, 100)
    ax.set_ylim(0, 100)
    ax.set_title("布洛托战场 - 动态演示")
    ax.set_xlabel("X 坐标")
    ax.set_ylabel("Y 坐标")
    ax.grid(True, linestyle='--', alpha=0.4)
    
    # 添加玩家基地
    base1 = patches.Rectangle((5, 5), 10, 10, linewidth=2, edgecolor='blue', facecolor='#b0e2ff', alpha=0.7)
    base2 = patches.Rectangle((85, 85), 10, 10, linewidth=2, edgecolor='orange', facecolor='#ffec8b', alpha=0.7)
    ax.add_patch(base1)
    ax.add_patch(base2)
    ax.text(10, 10, "P1 基地", ha='center', va='center', fontsize=10, color='blue')
    ax.text(90, 90, "P2 基地", ha='center', va='center', fontsize=10, color='orange')
    
    # 初始化战场元素
    markers = []
    texts = []
    lines = []
    
    # 创建初始空元素
    for i in range(n_targets):
        marker = ax.scatter([], [], s=0, color='gray', alpha=0, zorder=3)
        markers.append(marker)
        
        text = ax.text(0, 0, "", ha='center', va='center', fontsize=9, alpha=0)
        texts.append(text)
        
        lines.append([])
    
    # 添加结果摘要文本
    result_text = ax.text(50, -8, "", ha='center', va='top', fontsize=12, 
                          color='black', weight='bold', backgroundcolor='#f0f0f0')
    
    # 价值文本
    value_text = ax.text(50, 102, "", ha='center', va='bottom', fontsize=10, 
                         color='black', weight='bold')
    
    # 进度条
    progress = st.progress(0)
    
    # 动画更新函数
    def update(frame):
        # 确定当前显示的战场数量
        current_targets = min(frame // 2 + 1, n_targets)
        
        # 更新进度条
        progress.progress(min(100, int(frame / (n_targets * 2 + 5) * 100)))
        
        # 更新价值显示
        if current_targets > 0:
            current_values = "战场价值: " + " | ".join([f"#{i+1}:{values[i]}" for i in range(current_targets)])
            value_text.set_text(current_values)
        
        # 更新结果摘要
        if frame > n_targets * 2:
            p1_wins = np.sum(result[:current_targets] == 1)
            p2_wins = np.sum(result[:current_targets] == 2)
            draws = np.sum(result[:current_targets] == 0)
            result_text.set_text(f"当前结果: P1胜: {p1_wins} | P2胜: {p2_wins} | 平局: {draws}")
        
        # 移除上一帧的移动线条
        for line in lines:
            if line:
                line[0].remove()
                line.clear()
        
        # 更新战场元素
        for i in range(n_targets):
            x, y = battlefield[i]
            
            if i < current_targets:
                # 设置标记样式
                marker_size = 30 + 15 * max(player1[i], player2[i]) / max(player1.max(), player2.max())
                
                if result[i] == 1:
                    color = '#32CD32'  # 亮绿色
                elif result[i] == 2:
                    color = '#FF6347'  # 番茄红色
                else:
                    color = '#A9A9A9'  # 灰色
                
                # 更新标记
                markers[i].set_offsets([x, y])
                markers[i].set_sizes([marker_size])
                markers[i].set_color(color)
                markers[i].set_alpha(0.8)
                
                # 更新文本
                text_str = f"战场 #{i+1}\nP1: {player1[i]}\nP2: {player2[i]}"
                texts[i].set_position((x, y))
                texts[i].set_text(text_str)
                texts[i].set_alpha(1.0)
                texts[i].set_backgroundcolor('#f0f0f0')
                
                # 添加兵力移动动画（只在新战场上添加）
                if i == current_targets - 1 and frame % 2 == 0:
                    line1, = ax.plot([10, x], [10, y], 'b-', alpha=0.2, linewidth=player1[i]/10)
                    line2, = ax.plot([90, x], [90, y], 'orange', alpha=0.2, linewidth=player2[i]/10)
                    lines[i].extend([line1, line2])
            else:
                # 隐藏未显示的战场
                markers[i].set_alpha(0)
                texts[i].set_alpha(0)
        
        return markers + texts + [result_text, value_text]
    
    # 创建动画
    frames = n_targets * 2 + 10  # 额外帧用于显示最终结果
    ani = animation.FuncAnimation(fig, update, frames=frames,
                                  interval=animation_speed, blit=True)
    
    # 保存为GIF用于Streamlit
    gif_path = f"blotto_temp_{time.time()}.gif"
    ani.save(gif_path, writer='pillow', fps=1000/animation_speed)
    plt.close(fig)
    
    progress.progress(100)
    time.sleep(0.5)
    progress.empty()
    
    return gif_path

def blotto_game(total_troops=100, n_targets=5, player1_strategy=None,
                player2_strategy=None, strategy_type="random", battlefield_value=None,
                animate=True, nash_mode=False):
    """增强型布洛托博弈模拟"""
    # 战场价值分配
    if battlefield_value is None:
        battlefield_value = np.random.randint(1, 5, n_targets)
    else:
        battlefield_value = np.array(battlefield_value)
        # 如果提供的值不足，用随机值补充
        if len(battlefield_value) < n_targets:
            battlefield_value = np.append(battlefield_value, 
                                         np.random.randint(1, 5, n_targets - len(battlefield_value)))
    
    # 策略选择
    strategies = {
        "random": random_partition,
        "exponential": exponential_distribution,
        "mixed": mixed_strategy
    }
    
    # 纳什均衡模式
    if nash_mode:
        nash_strat = nash_equilibrium_pure(total_troops, n_targets, battlefield_value)
        if nash_strat:
            player1, player2 = nash_strat
        else:
            player1 = random_partition(total_troops, n_targets)
            player2 = random_partition(total_troops, n_targets)
    else:
        # 玩家策略分配
        if player1_strategy is None:
            player1 = strategies[strategy_type](total_troops, n_targets)
        else:
            player1 = np.array(player1_strategy)
            
        if player2_strategy is None:
            player2 = strategies[strategy_type](total_troops, n_targets)
        else:
            player2 = np.array(player2_strategy)
    
    # 检查资源分配并确保为正
    if player1.sum() != total_troops:
        player1 = (player1 / player1.sum() * total_troops).astype(int)
        player1 = np.maximum(player1, 0)
        player1[0] += total_troops - player1.sum()
        
    if player2.sum() != total_troops:
        player2 = (player2 / player2.sum() * total_troops).astype(int)
        player2 = np.maximum(player2, 0)
        player2[0] += total_troops - player2.sum()
    
    # 确保兵力非负
    player1 = np.maximum(player1, 0)
    player2 = np.maximum(player2, 0)
    
    # 胜负判定
    p1_wins = (player1 > player2)
    p2_wins = (player2 > player1)
    result = np.where(p1_wins, 1, np.where(p2_wins, 2, 0))
    
    # 价值加权胜负统计
    p1_value = np.sum(battlefield_value[p1_wins])
    p2_value = np.sum(battlefield_value[p2_wins])
    tie_value = np.sum(battlefield_value[result == 0])
    total_value = battlefield_value.sum()
    
    # 结果汇总
    result_details = {
        "player1": player1,
        "player2": player2,
        "result": result,
        "p1_value": p1_value,
        "p2_value": p2_value,
        "tie_value": tie_value,
        "battlefield_value": battlefield_value,
        "total_value": total_value,
        "winner": 1 if p1_value > p2_value else 2 if p2_value > p1_value else 0
    }
    
    # 生成动画
    gif_path = None
    if animate and n_targets <= 15:  # 限制战场数量以保持性能
        with st.spinner('生成战场动画中...'):
            gif_path = animate_blotto_battlefield(player1, player2, result, battlefield_value)
    
    return result_details, gif_path

# Streamlit 主界面
def main():
    # 设置页面配置
    st.set_page_config(
        page_title="布洛托博弈模拟器",
        page_icon="⚔️",
        layout="wide",
        initial_sidebar_state="expanded"
    )
    
    # 清除临时文件
    clear_temp_files()
    
    # 应用标题和说明
    st.title("⚔️ 布洛托博弈战场模拟器")
    st.markdown("""
    **布洛托博弈**是经典的博弈论问题：双方指挥官在多个战场上分配有限兵力争夺控制权。
    本模拟器可视化展示了不同策略下兵力分配和战场争夺的动态过程。
    """)
    
    # 侧边栏控制面板
    with st.sidebar:
        st.header("⚙️ 模拟参数")
        
        total_troops = st.slider("总兵力", 10, 500, 100, 10)
        n_targets = st.slider("战场数量", 2, 20, 7, 1)
        
        strategy_options = ["random", "exponential", "mixed", "nash_equilibrium"]
        strategy_names = ["随机分配", "指数分布", "混合策略", "纳什均衡"]
        strategy_type = st.selectbox("策略类型", strategy_names, index=1)
        
        # 自定义战场价值
        st.subheader("战场价值设定")
        if st.checkbox("自定义价值权重"):
            values = []
            cols = st.columns(3)
            for i in range(n_targets):
                with cols[i % 3]:
                    values.append(st.number_input(f"战场 #{i+1} 价值", 1, 10, (i % 3) + 1))
            battlefield_value = values
        else:
            battlefield_value = None
        
        # 高级选项
        with st.expander("高级选项"):
            animation_speed = st.slider("动画速度", 100, 2000, 500, 100)
            show_raw = st.checkbox("显示原始数据")
            enable_animation = st.checkbox("启用动画", True)
            
        # 运行模拟按钮
        run_simulation = st.button("运行战场模拟", type="primary")
    
    # 策略类型映射
    strategy_type = strategy_options[strategy_names.index(strategy_type)]
    
    # 当按下运行按钮时执行模拟
    if run_simulation:
        # 验证自定义战场价值
        if battlefield_value is not None and len(battlefield_value) != n_targets:
            st.error(f"错误：输入的战场价值数量({len(battlefield_value)})必须与战场数量({n_targets})一致")
            return
        
        # 执行模拟
        with st.spinner('战场模拟中...'):
            results, gif_path = blotto_game(
                total_troops=total_troops,
                n_targets=n_targets,
                strategy_type=strategy_type if strategy_type != "nash_equilibrium" else "random",
                battlefield_value=battlefield_value,
                animate=enable_animation,
                nash_mode=(strategy_type == "nash_equilibrium")
            )
        
        # 显示整体结果
        st.subheader("战斗结果总结")
        winner_text = {
            1: "🏆 **玩家1获胜**",
            2: "🏆 **玩家2获胜**",
            0: "⚖️ **战局平手**"
        }[results["winner"]]
        
        cols = st.columns(3)
        cols[0].metric("玩家1控制价值", f"{results['p1_value']:.1f}/{results['total_value']:.1f}")
        cols[1].metric("玩家2控制价值", f"{results['p2_value']:.1f}/{results['total_value']:.1f}")
        cols[2].metric("最终结果", winner_text)
        
        # 显示胜利分布
        p1_wins = np.sum(results["result"] == 1)
        p2_wins = np.sum(results["result"] == 2)
        draws = np.sum(results["result"] == 0)
        
        fig, ax = plt.subplots(figsize=(6, 3))
        bars = ax.bar(["玩家1胜", "平局", "玩家2胜"], [p1_wins, draws, p2_wins], 
                     color=['#1f77b4', '#7f7f7f', '#ff7f0e'])
        ax.set_title("战场胜负分布")
        ax.bar_label(bars)
        st.pyplot(fig)
        
        # 显示价值分布
        fig, ax = plt.subplots(figsize=(8, 4))
        width = 0.4
        indices = np.arange(n_targets)
        
        bars1 = ax.bar(indices - width/2, results["battlefield_value"], width, 
                      label="战场价值", color='#9467bd', alpha=0.7)
        
        ax2 = ax.twinx()
        bars2 = ax2.bar(indices + width/2, results["player1"], width/3, 
                        label="玩家1兵力", color='#1f77b4', alpha=0.8)
        bars3 = ax2.bar(indices + width/2 + width/3, results["player2"], width/3, 
                        label="玩家2兵力", color='#ff7f0e', alpha=0.8)
        
        ax.set_xticks(indices)
        ax.set_xticklabels([f"战场 {i+1}" for i in range(n_targets)])
        ax.set_title("战场价值与兵力分布")
        ax.legend(loc="upper left")
        ax2.legend(loc="upper right")
        st.pyplot(fig)
        
        # 显示动画
        if gif_path:
            st.subheader("战场动态演示")
            st.image(gif_path)
        elif enable_animation and n_targets > 15:
            st.warning("动画功能在超过15个战场时已禁用以保持性能")
        
        # 显示原始数据
        if show_raw:
            with st.expander("原始模拟数据"):
                st.subheader("玩家1兵力分配")
                st.write(results["player1"])
                
                st.subheader("玩家2兵力分配")
                st.write(results["player2"])
                
                st.subheader("战场价值权重")
                st.write(results["battlefield_value"])
                
                st.subheader("战场结果")
                st.write(np.array([1 if x == 1 else 2 if x == 2 else 0 for x in results["result"]]))
        
        # 显示策略说明
        st.subheader("策略分析")
        if strategy_type == "random":
            st.info("**随机策略**: 兵力随机分配到战场，可能产生不均衡分布")
        elif strategy_type == "exponential":
            st.info("**指数分布策略**: 模仿真实博弈中的纳什均衡分配，少数战场获得主要兵力")
        elif strategy_type == "mixed":
            st.info("**混合策略**: 组合多种基础策略的兵力分配方案")
        elif strategy_type == "nash_equilibrium":
            st.info("**纳什均衡策略**: 理论上最优策略，双方无法单方面改变策略获得更好结果")
        
        # 添加重播按钮
        if st.button("重新运行相同配置"):
            st.experimental_rerun()
    else:
        # 初始显示区域
        st.subheader("模拟说明")
        st.markdown("""
        1. 在左侧边栏设置战场参数和策略
        2. 点击"运行战场模拟"开始战斗
        3. 结果将展示兵力分配、战场控制和动态动画
        
        **策略类型说明**:
        - ⚄️ **随机分配**: 均匀随机分布兵力
        - 📉 **指数分布**: 类似真实世界的重点防御策略
        - 🧩 **混合策略**: 多种策略的组合方案
        - ⚖️ **纳什均衡**: 理论上的最优解
        """)
        
        # 示例图片
        st.image("https://upload.wikimedia.org/wikipedia/commons/thumb/e/e7/Colonel_Blotto.svg/800px-Colonel_Blotto.svg.png",
                 caption="布洛托博弈示意图 (来源: Wikipedia)")
    
    # 页脚
    st.markdown("---")
    st.caption("© 2025 布洛托博弈 | 博弈论应用 | 使用Python、Streamlit和Matplotlib构建")

if __name__ == "__main__":
    main()