解析 ‘Virtual Economy Simulation’：在图中模拟 Agent 之间的微支付、资源竞争与价格博弈逻辑

各位同仁，各位对复杂系统、经济学以及编程艺术充满热情的开发者们，下午好！

今天，我们将共同深入探讨一个引人入胜的主题：虚拟经济仿真 (Virtual Economy Simulation)。这不是一个抽象的理论概念，而是一个强大的实践工具，它能帮助我们理解真实世界的经济规律，优化游戏内的经济系统，甚至为去中心化应用的设计提供深刻洞察。

在本次讲座中，我们的核心任务是构建一个仿真环境，在这个环境中，数字代理 (Agent) 们将进行一系列复杂的交互：微支付的流动、对稀缺资源的激烈竞争，以及至关重要的价格博弈逻辑。我们将从编程专家的视角出发，一步步解构这个系统，用严谨的代码和逻辑来展现其运作机制。

1. 虚拟经济仿真的基石：Agent-Based Modeling (ABM)

要模拟一个动态的经济系统，传统的宏观经济模型往往难以捕捉个体行为的复杂性和多样性。这时，Agent-Based Modeling (ABM) 便展现出其独特的优势。ABM 是一种计算建模范式，它通过模拟大量自主行动的个体（即代理），以及它们之间、代理与环境之间的相互作用，来研究整个系统的宏观行为。

在我们的虚拟经济中，代理可以是生产者、消费者、交易者，甚至是服务提供商。它们拥有各自的目标、规则和决策逻辑。通过观察这些个体层面的简单交互，我们能够发现令人惊讶的涌现行为，例如市场价格的波动、财富分配的变化，甚至经济危机的形成。

为什么 ABM 适合虚拟经济仿真？

• 异质性 (Heterogeneity): 每个代理都可以有独特的属性（如财富、技能、偏好）和行为规则，这更贴近现实。
• 局部交互 (Local Interactions): 代理通常只与邻近的代理或市场进行交互，而不是一个抽象的“总供给”或“总需求”。
• 非线性 (Non-linearity): 简单的个体规则叠加可能导致复杂的、难以预测的系统行为。
• 涌现 (Emergence): 宏观模式不是预先设定的，而是从微观交互中自发产生的。

我们的仿真将围绕以下几个核心要素展开：

1. 代理 (Agents): 拥有库存、货币、需求、生产能力和交易策略的独立实体。
2. 资源 (Resources): 构成经济活动的基础，可以是原材料、半成品或最终产品，以及作为交换媒介的货币。
3. 市场 (Markets): 代理进行交易的场所，负责匹配买卖双方，并形成价格。
4. 环境 (Environment): 代理生活和获取资源的背景，可能包含资源生成点、地理位置等。
5. 时间 (Time): 仿真通过离散的时间步长推进，每个步长内代理执行其行为。

2. 构建仿真框架：核心数据结构与仿真循环

首先，我们需要定义构成我们虚拟经济的基本构建块。我们将使用 Python 来实现，因为它简洁、易读，并且拥有丰富的科学计算库。

2.1 核心数据结构

我们将定义 Agent、Resource 和 Market 三个主要类。

Resource 类：经济的基石

资源是经济活动的基础。它可以是木材、矿石这样的原材料，也可以是工具、食物这样的制成品。

from enum import Enum

class ResourceType(Enum):
    """定义不同类型的资源"""
    CURRENCY = "Currency"
    WOOD = "Wood"
    ORE = "Ore"
    TOOL = "Tool"
    FOOD = "Food"
    # 可以根据需要添加更多资源

class Resource:
    """
    表示经济中的一种资源
    """
    def __init__(self, resource_type: ResourceType, name: str, base_value: float):
        self.resource_type = resource_type
        self.name = name
        self.base_value = base_value # 基础价值，用于初始设定或作为参考

    def __repr__(self):
        return f"Resource({self.name})"

# 初始化一些通用资源实例
CURRENCY = Resource(ResourceType.CURRENCY, "GoldCoin", 1.0)
WOOD = Resource(ResourceType.WOOD, "Wood", 5.0)
ORE = Resource(ResourceType.ORE, "IronOre", 10.0)
TOOL = Resource(ResourceType.TOOL, "Axe", 50.0)
FOOD = Resource(ResourceType.FOOD, "Bread", 15.0)

Agent 类：经济的参与者

每个代理都拥有一个唯一的 ID，一个库存（存储资源），以及一定数量的货币。代理的行为（生产、消费、交易）将是其最核心的逻辑。

import uuid
from collections import defaultdict
import random

class Agent:
    """
    经济中的基本代理类，包含库存和货币管理。
    """
    def __init__(self, agent_id: str = None, initial_currency: float = 100.0):
        self.agent_id = agent_id if agent_id else str(uuid.uuid4())
        self.currency = initial_currency
        self.inventory: defaultdict[Resource, int] = defaultdict(int) # 存储资源及其数量
        self.needs: defaultdict[Resource, float] = defaultdict(float) # 代理的需求，例如对食物的需求程度
        self.production_recipes = {} # 生产配方：输入资源 -> 输出资源
        self.strategy = self._default_strategy # 代理的决策策略

    def add_resource(self, resource: Resource, quantity: int):
        """向库存添加资源"""
        if quantity < 0:
            raise ValueError("Quantity cannot be negative")
        self.inventory[resource] += quantity

    def remove_resource(self, resource: Resource, quantity: int) -> bool:
        """从库存移除资源，如果不足则返回False"""
        if quantity < 0:
            raise ValueError("Quantity cannot be negative")
        if self.inventory[resource] >= quantity:
            self.inventory[resource] -= quantity
            if self.inventory[resource] == 0:
                del self.inventory[resource] # 清理空库存项
            return True
        return False

    def add_currency(self, amount: float):
        """增加代理的货币"""
        self.currency += amount

    def remove_currency(self, amount: float) -> bool:
        """减少代理的货币，如果不足则返回False"""
        if amount < 0:
            raise ValueError("Amount cannot be negative")
        if self.currency >= amount:
            self.currency -= amount
            return True
        return False

    def get_inventory_value(self) -> float:
        """计算代理当前库存的总价值（基于资源的base_value）"""
        total_value = self.currency
        for res, qty in self.inventory.items():
            total_value += res.base_value * qty
        return total_value

    def _default_strategy(self, market):
        """默认的代理行为策略，子类应重写"""
        pass # 抽象方法，等待具体实现

    def tick(self, market):
        """在每个时间步长执行代理的行动"""
        self.strategy(market)

    def __repr__(self):
        return f"Agent({self.agent_id[:8]}..., Curr:{self.currency:.2f}, Inv:{len(self.inventory)})"

Market 类：交易的枢纽

市场是代理进行交易的核心场所。我们将实现一个简化的订单簿模型，其中包含买单 (Bid) 和卖单 (Ask)。

class Order:
    """
    表示一个买单或卖单
    """
    def __init__(self, agent: Agent, resource: Resource, quantity: int, price: float, is_buy: bool):
        self.agent = agent
        self.resource = resource
        self.quantity = quantity
        self.price = price
        self.is_buy = is_buy # True for buy order, False for sell order

    def __repr__(self):
        order_type = "BUY" if self.is_buy else "SELL"
        return f"{order_type} {self.quantity}x {self.resource.name} @ {self.price:.2f} by {self.agent.agent_id[:4]}..."

class Market:
    """
    中央市场，管理买卖订单和执行交易
    """
    def __init__(self, name: str):
        self.name = name
        self.buy_orders: defaultdict[Resource, list[Order]] = defaultdict(list)  # 买单簿，按资源类型分类
        self.sell_orders: defaultdict[Resource, list[Order]] = defaultdict(list) # 卖单簿，按资源类型分类
        self.transaction_history = [] # 记录所有交易

    def place_order(self, order: Order):
        """代理提交订单"""
        if order.is_buy:
            # 买单按价格降序排列 (高价优先)
            self.buy_orders[order.resource].append(order)
            self.buy_orders[order.resource].sort(key=lambda o: o.price, reverse=True)
        else:
            # 卖单按价格升序排列 (低价优先)
            self.sell_orders[order.resource].append(order)
            self.sell_orders[order.resource].sort(key=lambda o: o.price)

    def _execute_trade(self, buy_order: Order, sell_order: Order, trade_quantity: int, trade_price: float):
        """执行实际的交易逻辑"""
        buyer = buy_order.agent
        seller = sell_order.agent
        resource = buy_order.resource

        # 检查买家是否有足够的钱，卖家是否有足够的货
        if not buyer.remove_currency(trade_quantity * trade_price):
            print(f"ERROR: Buyer {buyer.agent_id[:4]}... failed to pay for {trade_quantity}x {resource.name}")
            return
        if not seller.remove_resource(resource, trade_quantity):
            print(f"ERROR: Seller {seller.agent_id[:4]}... failed to provide {trade_quantity}x {resource.name}")
            # 如果卖家货不足，买家钱已经扣了，需要回滚或处理。这里简化为打印错误，实际应更健壮。
            buyer.add_currency(trade_quantity * trade_price) # 退款
            return

        seller.add_currency(trade_quantity * trade_price)
        buyer.add_resource(resource, trade_quantity)

        # 更新订单数量
        buy_order.quantity -= trade_quantity
        sell_order.quantity -= trade_quantity

        self.transaction_history.append({
            "buyer_id": buyer.agent_id,
            "seller_id": seller.agent_id,
            "resource": resource.name,
            "quantity": trade_quantity,
            "price": trade_price,
            "timestamp": self.current_time # 假设市场有一个时间戳
        })
        # print(f"Trade: {trade_quantity}x {resource.name} @ {trade_price:.2f} between {buyer.agent_id[:4]}... and {seller.agent_id[:4]}...")

    def match_orders(self):
        """尝试匹配所有资源类型的买卖订单"""
        matched_trades_count = 0
        for resource_type in self.buy_orders.keys():
            buy_list = self.buy_orders[resource_type]
            sell_list = self.sell_orders[resource_type]

            # 移除已完成或无效的订单
            buy_list[:] = [order for order in buy_list if order.quantity > 0]
            sell_list[:] = [order for order in sell_list if order.quantity > 0]

            # 重新排序，确保最佳匹配
            buy_list.sort(key=lambda o: o.price, reverse=True)
            sell_list.sort(key=lambda o: o.price)

            i, j = 0, 0
            while i < len(buy_list) and j < len(sell_list):
                buy_order = buy_list[i]
                sell_order = sell_list[j]

                # 检查买卖双方是否是同一个Agent (通常不允许自买自卖)
                if buy_order.agent.agent_id == sell_order.agent.agent_id:
                    # 如果是同一个Agent，跳过其中一个，尝试下一个
                    if buy_order.is_buy: # 优先跳过买单，因为买单可能价格更高
                        i += 1
                    else:
                        j += 1
                    continue

                # 如果最高买价 >= 最低卖价，则可以成交
                if buy_order.price >= sell_order.price:
                    trade_quantity = min(buy_order.quantity, sell_order.quantity)
                    trade_price = (buy_order.price + sell_order.price) / 2 # 简化为中点价，实际可更复杂

                    # 检查是否有足够的余额和库存
                    if buy_order.agent.currency >= trade_quantity * trade_price and 
                       sell_order.agent.inventory[resource_type] >= trade_quantity:
                        self._execute_trade(buy_order, sell_order, trade_quantity, trade_price)
                        matched_trades_count += 1
                    else:
                        # 如果某个代理余额或库存不足，则该订单视为无法成交，跳过
                        if buy_order.agent.currency < trade_quantity * trade_price:
                            # print(f"Buyer {buy_order.agent.agent_id[:4]}... lacks currency for trade.")
                            i += 1 # 跳过此买单
                        elif sell_order.agent.inventory[resource_type] < trade_quantity:
                            # print(f"Seller {sell_order.agent.agent_id[:4]}... lacks resource for trade.")
                            j += 1 # 跳过此卖单
                        continue

                    # 如果一个订单完全成交，则从列表中移除或推进指针
                    if buy_order.quantity == 0:
                        i += 1
                    if sell_order.quantity == 0:
                        j += 1
                else:
                    # 如果最高买价 < 最低卖价，则当前资源无法成交，退出循环
                    break
        return matched_trades_count

    def get_latest_price(self, resource: Resource) -> float:
        """获取某种资源的最新成交价，如果没有则返回一个默认值或参考价格"""
        resource_transactions = [t for t in self.transaction_history if t["resource"] == resource.name]
        if resource_transactions:
            return resource_transactions[-1]["price"]
        return resource.base_value # 如果没有交易，返回基础价值

    def get_market_depth(self, resource: Resource):
        """获取指定资源的买卖深度信息"""
        buy_depth = [(o.price, o.quantity) for o in self.buy_orders[resource]]
        sell_depth = [(o.price, o.quantity) for o in self.sell_orders[resource]]
        return buy_depth, sell_depth

    def set_current_time(self, time_step: int):
        self.current_time = time_step

2.2 仿真循环

仿真的核心是一个时间步进循环。在每个时间步长 (tick) 中，我们将：

1. 环境更新（例如，资源再生）。
2. 所有代理执行其决策（生产、消费、提交订单）。
3. 市场尝试匹配订单并执行交易。
4. 收集数据。

class Simulation:
    """
    仿真主类，管理代理、市场和环境
    """
    def __init__(self, num_agents: int = 10, initial_currency_range=(50, 200)):
        self.agents: list[Agent] = []
        self.market = Market("GlobalMarket")
        self.current_time = 0
        self.environment_resources: defaultdict[Resource, int] = defaultdict(int)

        # 初始化环境资源 (例如，无限的木材和矿石供采集)
        self.environment_resources[WOOD] = 1000000
        self.environment_resources[ORE] = 1000000

        # 创建代理
        for _ in range(num_agents):
            initial_currency = random.uniform(*initial_currency_range)
            agent = Agent(initial_currency=initial_currency)
            # 随机给一些初始资源，让经济能启动
            if random.random() < 0.5:
                agent.add_resource(WOOD, random.randint(5, 20))
            else:
                agent.add_resource(ORE, random.randint(3, 10))
            self.agents.append(agent)
            # 代理默认策略在这里初始化，后续会细化
            agent.strategy = lambda m: self._default_agent_strategy(agent, m)

        # 记录关键指标
        self.metrics = {
            "time": [],
            "total_currency": [],
            "avg_agent_currency": [],
            "total_inventory_value": [],
            "resource_prices": defaultdict(list)
        }
        print(f"Simulation initialized with {num_agents} agents.")

    def _default_agent_strategy(self, agent: Agent, market: Market):
        """
        一个简单的默认代理策略，用于占位。
        具体代理类型会重写这个。
        """
        # 简单地随机买卖一些资源
        if agent.currency > 20 and random.random() < 0.3: # 随机买
            res_to_buy = random.choice([WOOD, ORE, FOOD])
            qty = random.randint(1, 5)
            # 假设一个出价策略：比最新价高一点，或基于资源基础价值
            price = market.get_latest_price(res_to_buy) * random.uniform(0.9, 1.1)
            if price == res_to_buy.base_value: # 如果没有历史价格，使用基础价格
                price = res_to_buy.base_value * random.uniform(0.8, 1.2)
            if agent.currency >= price * qty:
                market.place_order(Order(agent, res_to_buy, qty, price, True))

        if agent.inventory and random.random() < 0.3: # 随机卖
            res_to_sell = random.choice(list(agent.inventory.keys()))
            if agent.inventory[res_to_sell] > 0:
                qty = random.randint(1, min(5, agent.inventory[res_to_sell]))
                # 假设一个要价策略：比最新价低一点，或基于资源基础价值
                price = market.get_latest_price(res_to_sell) * random.uniform(0.9, 1.1)
                if price == res_to_sell.base_value:
                    price = res_to_sell.base_value * random.uniform(0.8, 1.2)
                market.place_order(Order(agent, res_to_sell, qty, price, False))

    def _update_environment(self):
        """
        环境更新逻辑，例如资源再生。
        目前设定环境资源无限，此处可以留空或添加更复杂的逻辑。
        """
        pass

    def _collect_metrics(self):
        """
        收集仿真过程中的关键数据
        """
        self.metrics["time"].append(self.current_time)

        total_currency = sum(a.currency for a in self.agents)
        self.metrics["total_currency"].append(total_currency)
        self.metrics["avg_agent_currency"].append(total_currency / len(self.agents))

        total_inventory_value = sum(a.get_inventory_value() for a in self.agents)
        self.metrics["total_inventory_value"].append(total_inventory_value)

        for res in [WOOD, ORE, TOOL, FOOD]: # 关注这些主要资源的价格
            self.metrics["resource_prices"][res.name].append(self.market.get_latest_price(res))

    def run(self, num_steps: int):
        """运行仿真指定的时间步长"""
        print(f"Starting simulation for {num_steps} steps...")
        for step in range(num_steps):
            self.current_time = step
            self.market.set_current_time(step) # 更新市场的时间戳

            # 1. 环境更新
            self._update_environment()

            # 2. 代理行动 (生产、消费、提交订单)
            # 代理的行动顺序可能影响结果，这里随机化一下
            random.shuffle(self.agents)
            for agent in self.agents:
                agent.tick(self.market) # 代理执行其策略

            # 3. 市场匹配订单并执行交易
            matched_count = self.market.match_orders()
            # if matched_count > 0:
            #     print(f"Step {step}: Matched {matched_count} trades.")

            # 4. 收集数据
            self._collect_metrics()

            # 打印进度（可选）
            if step % (num_steps // 10 or 1) == 0:
                print(f"--- Step {step}/{num_steps} ---")
                print(f"Total Agents: {len(self.agents)}")
                # print(f"Total Currency: {self.metrics['total_currency'][-1]:.2f}")
                # print(f"Avg Agent Currency: {self.metrics['avg_agent_currency'][-1]:.2f}")
                # for res in [WOOD, ORE, TOOL, FOOD]:
                #     print(f"  Latest {res.name} Price: {self.metrics['resource_prices'][res.name][-1]:.2f}")

        print("Simulation finished.")

3. 实现代理的微支付、资源竞争与价格博弈逻辑

现在，我们将深入实现代理的复杂行为。我们将定义几种不同类型的代理，它们将共同构成一个动态的经济生态系统。

3.1 代理的专业化：生产者、消费者与交易者

为了更好地模拟经济，我们需要有不同角色的代理。

• 生产者 (ProducerAgent): 采集原材料，将其加工成更高级的产品。
• 消费者 (ConsumerAgent): 有固定的需求，会购买并消耗产品。
• 交易者 (TraderAgent): 纯粹通过买卖赚取差价。

实际上，一个代理可以同时扮演多个角色。例如，一个农夫既是食物的生产者，也是工具的消费者。在我们的模型中，我们将通过配置 production_recipes 和 needs 来实现这种多角色。

我们先定义一个抽象的 Agent 基类，然后创建具体的子类。

# 假设 Agent 基类已经定义如上

class ProductionRecipe:
    """
    定义一个生产配方：需要什么输入，产出什么输出，以及生产时间/成本
    """
    def __init__(self, inputs: dict[Resource, int], outputs: dict[Resource, int], time_cost: int = 1, currency_cost: float = 0):
        self.inputs = inputs
        self.outputs = outputs
        self.time_cost = time_cost # 生产所需时间步长
        self.currency_cost = currency_cost # 生产过程中的货币消耗（例如，电力，维护费）

    def __repr__(self):
        input_str = ", ".join(f"{qty}x {res.name}" for res, qty in self.inputs.items())
        output_str = ", ".join(f"{qty}x {res.name}" for res, qty in self.outputs.items())
        return f"Recipe(In:{input_str} -> Out:{output_str})"

# 定义一些生产配方
RECIPES = {
    "MAKE_TOOL": ProductionRecipe(inputs={WOOD: 2, ORE: 3}, outputs={TOOL: 1}, time_cost=3, currency_cost=10),
    "MAKE_FOOD": ProductionRecipe(inputs={WOOD: 1}, outputs={FOOD: 2}, time_cost=2, currency_cost=5) # 简化：用木材生产食物
}

class ProducerAgent(Agent):
    """
    生产者代理：采集资源，并根据配方生产产品。
    """
    def __init__(self, agent_id: str = None, initial_currency: float = 100.0, production_target_resource: Resource = TOOL):
        super().__init__(agent_id, initial_currency)
        self.production_recipes = RECIPES # 所有生产者都可以访问这些配方
        self.production_target = production_target_resource # 优先生产的目标
        self.production_in_progress = None # 当前正在进行的生产活动
        self.production_remaining_time = 0
        self.resource_gathering_rate = random.randint(1, 3) # 每次采集的量

    def _gather_resource(self, resource: Resource, environment_resources: defaultdict[Resource, int]):
        """从环境中采集资源"""
        if environment_resources[resource] > 0:
            qty_to_gather = min(self.resource_gathering_rate, environment_resources[resource])
            self.add_resource(resource, qty_to_gather)
            environment_resources[resource] -= qty_to_gather
            # print(f"{self.agent_id[:4]}... gathered {qty_to_gather}x {resource.name}.")

    def _produce(self, recipe: ProductionRecipe):
        """执行生产活动"""
        # 检查是否有足够的输入资源
        for res, qty in recipe.inputs.items():
            if self.inventory[res] < qty:
                # print(f"{self.agent_id[:4]}... lacks {res.name} to produce {recipe.outputs}.")
                return False # 资源不足，无法生产

        # 检查是否有足够的货币支付生产成本
        if self.currency < recipe.currency_cost:
            # print(f"{self.agent_id[:4]}... lacks currency to produce {recipe.outputs}.")
            return False # 货币不足

        # 消耗输入资源和货币
        for res, qty in recipe.inputs.items():
            self.remove_resource(res, qty)
        self.remove_currency(recipe.currency_cost)

        # 开始生产过程
        self.production_in_progress = recipe
        self.production_remaining_time = recipe.time_cost
        # print(f"{self.agent_id[:4]}... started producing {list(recipe.outputs.keys())[0].name}.")
        return True

    def _finish_production(self):
        """完成生产并获得产出"""
        recipe = self.production_in_progress
        for res, qty in recipe.outputs.items():
            self.add_resource(res, qty)
        # print(f"{self.agent_id[:4]}... finished producing {list(recipe.outputs.keys())[0].name}, gained {list(recipe.outputs.values())[0]}x {list(recipe.outputs.keys())[0].name}.")
        self.production_in_progress = None

    def _producer_strategy(self, market: Market, environment_resources: defaultdict[Resource, int]):
        """生产者的策略：采集、生产、出售"""
        # 1. 如果有生产正在进行，则减少剩余时间
        if self.production_in_progress:
            self.production_remaining_time -= 1
            if self.production_remaining_time <= 0:
                self._finish_production()
            return # 生产中，本回合不进行其他主要活动

        # 2. 检查是否需要生产目标产品
        target_qty = self.inventory[self.production_target]
        if target_qty < 5: # 如果目标产品数量少于5，尝试生产
            for recipe_name, recipe in self.production_recipes.items():
                if self.production_target in recipe.outputs: # 找到能生产目标产品的配方
                    # 检查是否有足够原材料
                    can_produce = True
                    for res, qty in recipe.inputs.items():
                        if self.inventory[res] < qty:
                            can_produce = False
                            # 如果缺少原材料，尝试从市场购买或采集
                            # 购买逻辑：如果价格合适且有钱
                            market_price = market.get_latest_price(res)
                            if self.currency > market_price * qty * 1.2: # 愿意支付高于市场价1.2倍
                                market.place_order(Order(self, res, qty, market_price * 1.1, True))
                            elif res in [WOOD, ORE]: # 如果是可采集的原材料，去采集
                                self._gather_resource(res, environment_resources)
                            break
                    if can_produce:
                        self._produce(recipe)
                        return # 本回合已开始生产

        # 3. 如果库存中有产品，且货币不足或库存过高，尝试出售产品
        for res, qty in self.inventory.items():
            if res != CURRENCY and qty > 0 and (self.currency < 50 or qty > 10): # 货币少于50或某种资源超过10个
                # 计算出售价格：基于成本或市场价
                sell_price = market.get_latest_price(res) * random.uniform(0.9, 1.05) # 略低于或等于市场价
                if sell_price == res.base_value: # 如果没有历史价格，使用基础价格
                    sell_price = res.base_value * random.uniform(0.8, 1.2)
                market.place_order(Order(self, res, qty, sell_price, False))
                return # 本回合已挂出售单

        # 4. 如果没有其他事情做，采集原材料
        if self.inventory[WOOD] < 10:
            self._gather_resource(WOOD, environment_resources)
        if self.inventory[ORE] < 10:
            self._gather_resource(ORE, environment_resources)

    def tick(self, market: Market, environment_resources: defaultdict[Resource, int]):
        """在每个时间步长执行代理的行动"""
        self._producer_strategy(market, environment_resources)

class ConsumerAgent(Agent):
    """
    消费者代理：有持续的需求，会购买并消耗产品。
    """
    def __init__(self, agent_id: str = None, initial_currency: float = 100.0, desired_resource: Resource = FOOD):
        super().__init__(agent_id, initial_currency)
        self.desired_resource = desired_resource
        self.needs[desired_resource] = random.uniform(0.8, 1.2) # 对目标资源的“饥饿度”
        self.consumption_rate = random.randint(1, 3) # 每次消耗量

    def _consume_needs(self):
        """消耗自身资源，满足需求"""
        for res, need_level in self.needs.items():
            if need_level > 0.5: # 当需求达到一定水平时尝试消耗
                qty_to_consume = min(self.consumption_rate, self.inventory[res])
                if qty_to_consume > 0:
                    self.remove_resource(res, qty_to_consume)
                    self.needs[res] = max(0, self.needs[res] - qty_to_consume * 0.1) # 消耗后需求降低
                    # print(f"{self.agent_id[:4]}... consumed {qty_to_consume}x {res.name}. Need for {res.name} now {self.needs[res]:.2f}.")
                else:
                    self.needs[res] = min(2.0, self.needs[res] + 0.05) # 如果没有资源可消耗，需求会增长

    def _consumer_strategy(self, market: Market):
        """消费者的策略：检查需求，购买，消耗"""
        self._consume_needs()

        # 如果对目标资源的需求很高，且库存不足，则尝试购买
        if self.needs[self.desired_resource] > 1.0 and self.inventory[self.desired_resource] < 5:
            # 愿意支付的价格：基于需求程度和资源基础价值
            willing_to_pay = market.get_latest_price(self.desired_resource) * self.needs[self.desired_resource]
            if willing_to_pay == self.desired_resource.base_value:
                willing_to_pay = self.desired_resource.base_value * random.uniform(0.8, 1.5)

            qty_to_buy = random.randint(1, 3)
            if self.currency >= willing_to_pay * qty_to_buy:
                market.place_order(Order(self, self.desired_resource, qty_to_buy, willing_to_pay, True))
                # print(f"{self.agent_id[:4]}... placed BUY order for {qty_to_buy}x {self.desired_resource.name} @ {willing_to_pay:.2f}.")

        # 如果有多余的资源（非目标资源），可以考虑出售
        for res, qty in self.inventory.items():
            if res != CURRENCY and res != self.desired_resource and qty > 3: # 超过3个非需求资源
                sell_price = market.get_latest_price(res) * random.uniform(0.8, 1.0)
                if sell_price == res.base_value:
                    sell_price = res.base_value * random.uniform(0.7, 1.1)
                market.place_order(Order(self, res, qty, sell_price, False))
                return # 每回合只挂一个卖单

    def tick(self, market: Market):
        self._consumer_strategy(market)

# 修改Simulation类以使用这些专业代理
class Simulation:
    """
    仿真主类，管理代理、市场和环境
    """
    def __init__(self, num_producers: int = 5, num_consumers: int = 5, initial_currency_range=(50, 200)):
        self.agents: list[Agent] = []
        self.market = Market("GlobalMarket")
        self.current_time = 0
        self.environment_resources: defaultdict[Resource, int] = defaultdict(int)

        # 初始化环境资源 (例如，无限的木材和矿石供采集)
        self.environment_resources[WOOD] = 10000000
        self.environment_resources[ORE] = 10000000

        # 创建生产者代理
        for i in range(num_producers):
            initial_currency = random.uniform(*initial_currency_range)
            # 生产者可以有不同的生产目标
            target = random.choice([TOOL, FOOD])
            agent = ProducerAgent(initial_currency=initial_currency, production_target_resource=target)
            self.agents.append(agent)
            # 随机给一些初始资源
            if random.random() < 0.5:
                agent.add_resource(WOOD, random.randint(5, 20))
            else:
                agent.add_resource(ORE, random.randint(3, 10))

        # 创建消费者代理
        for i in range(num_consumers):
            initial_currency = random.uniform(*initial_currency_range)
            # 消费者可以有不同的需求
            desired = random.choice([FOOD, TOOL])
            agent = ConsumerAgent(initial_currency=initial_currency, desired_resource=desired)
            # 初始给一些钱和少量的需求品，使其可以开始消费
            agent.add_resource(desired, random.randint(1, 3))
            self.agents.append(agent)

        # 记录关键指标
        self.metrics = {
            "time": [],
            "total_currency": [],
            "avg_agent_currency": [],
            "total_inventory_value": [],
            "resource_prices": defaultdict(list),
            "total_trades": []
        }
        print(f"Simulation initialized with {num_producers} producers and {num_consumers} consumers.")

    def _collect_metrics(self):
        """
        收集仿真过程中的关键数据
        """
        self.metrics["time"].append(self.current_time)

        total_currency = sum(a.currency for a in self.agents)
        self.metrics["total_currency"].append(total_currency)
        self.metrics["avg_agent_currency"].append(total_currency / len(self.agents))

        total_inventory_value = sum(a.get_inventory_value() for a in self.agents)
        self.metrics["total_inventory_value"].append(total_inventory_value)

        for res in [WOOD, ORE, TOOL, FOOD]: # 关注这些主要资源的价格
            self.metrics["resource_prices"][res.name].append(self.market.get_latest_price(res))

        self.metrics["total_trades"].append(len(self.market.transaction_history))

    def run(self, num_steps: int):
        """运行仿真指定的时间步长"""
        print(f"Starting simulation for {num_steps} steps...")
        for step in range(num_steps):
            self.current_time = step
            self.market.set_current_time(step) # 更新市场的时间戳

            # 1. 环境更新
            self._update_environment()

            # 2. 代理行动 (生产、消费、提交订单)
            random.shuffle(self.agents) # 随机化代理行动顺序
            for agent in self.agents:
                # 根据代理类型调用不同的tick方法
                if isinstance(agent, ProducerAgent):
                    agent.tick(self.market, self.environment_resources)
                elif isinstance(agent, ConsumerAgent):
                    agent.tick(self.market)
                else:
                    agent.tick(self.market) # 默认代理

            # 3. 市场匹配订单并执行交易
            matched_count = self.market.match_orders()
            if matched_count > 0:
                pass # print(f"Step {step}: Matched {matched_count} trades.")

            # 4. 收集数据
            self._collect_metrics()

            # 打印进度（可选）
            if step % (num_steps // 10 or 1) == 0:
                print(f"n--- Step {step}/{num_steps} ---")
                print(f"Total Agents: {len(self.agents)}")
                print(f"Total Currency: {self.metrics['total_currency'][-1]:.2f}")
                print(f"Avg Agent Currency: {self.metrics['avg_agent_currency'][-1]:.2f}")
                for res in [WOOD, ORE, TOOL, FOOD]:
                    print(f"  Latest {res.name} Price: {self.metrics['resource_prices'][res.name][-1]:.2f}")
                print(f"Total Trades: {self.metrics['total_trades'][-1]}")

        print("nSimulation finished.")

3.2 微支付逻辑

微支付在我们的模型中通过 Agent 类的 add_currency 和 remove_currency 方法，以及 Market 类的 _execute_trade 方法实现。每次交易发生时，买家向卖家支付货币，卖家则获得货币。这是经济活动的基础。

# 核心逻辑已在 Agent 和 Market 类中实现：
# Agent.remove_currency() 和 Agent.add_currency() 负责代理内部的货币流转。
# Market._execute_trade() 在交易发生时，调用买家的 remove_currency 和卖家的 add_currency，完成微支付。

3.3 资源竞争

资源竞争主要体现在两个层面：

1. 原材料采集竞争: ProducerAgent 尝试从 environment_resources 中获取 WOOD 和 ORE。虽然目前环境中的资源被设定为“无限” (一个很大的数字)，但在更复杂的模型中，这部分资源可以是有限的，甚至有地理分布，从而引发代理在特定资源点上的竞争。

   # ProducerAgent._gather_resource 方法处理了从环境获取资源并减少环境资源量的逻辑。
   # 实际竞争会在多个ProducerAgent同时尝试获取有限资源时体现。

2. 市场商品竞争: 多个代理可能同时希望购买或出售同一种商品。市场通过订单簿机制来解决这种竞争：
   • 买家竞争: 出价最高的买家优先获得商品。
   • 卖家竞争: 要价最低的卖家优先出售商品。

3.4 价格博弈逻辑

价格博弈是虚拟经济中最复杂也最有趣的部分。在我们的模型中，价格博弈通过代理的定价策略和市场匹配机制共同实现。

代理的定价策略：

• ProducerAgent 的出售策略:
  • 当需要出售产品时，它会参考当前的市场最新价格 (market.get_latest_price(res))。
  • 它会在此基础上乘以一个随机系数 (random.uniform(0.9, 1.05))，形成自己的卖出报价。这意味着生产者可能愿意以略低于市场价的价格快速出售，也可能以略高于市场价的价格尝试获利。
  • 如果没有任何历史交易价格，它会使用资源的 base_value 作为参考。
• ConsumerAgent 的购买策略:
  • 当需求强烈 (例如 self.needs[self.desired_resource] > 1.0) 且库存不足时，消费者会尝试购买。
  • 它愿意支付的价格基于市场最新价格和自身的需求程度 (market.get_latest_price(self.desired_resource) * self.needs[self.desired_resource])。需求越强烈，愿意支付的价格越高。
  • 同样，没有历史交易价时，会参考 base_value。
• ProducerAgent 的购买策略 (针对原材料):
  • 生产者在缺少原材料时，也会去市场购买。它会愿意支付比市场最新价略高一些的价格 (market_price * 1.1)，以确保能买到生产所需的资源。

市场匹配机制：

Market.match_orders() 方法是价格博弈的核心。它按照以下规则进行：

1. 最高买价优先: 买单 (buy_orders) 按价格从高到低排序。
2. 最低卖价优先: 卖单 (sell_orders) 按价格从低到高排序。
3. 成交条件: 只有当最高的买价大于或等于最低的卖价时，交易才能发生。
4. 成交价格: 在我们的简化模型中，成交价格取买卖双方报价的平均值 ((buy_order.price + sell_order.price) / 2)。在实际高频交易市场中，这可以是买价、卖价，或两者之间某个复杂的加权平均，甚至是在订单簿中找到的下一个最优价格。
5. 数量匹配: 交易数量是买单和卖单中较小的那个。

这种机制使得市场价格在一个动态的均衡点附近波动。当买家普遍愿意出高价时，价格会上涨；当卖家急于出货而压低价格时，价格会下跌。代理的个体决策（基于其内部状态和市场信息）共同驱动了价格的演变。

4. 仿真执行与结果分析

现在我们已经构建了代理、资源和市场，并为代理设定了行为逻辑。是时候运行仿真并观察结果了。

# 实例化并运行仿真
if __name__ == "__main__":
    random.seed(42) # 保证结果可复现

    # 创建一个仿真实例
    sim = Simulation(num_producers=8, num_consumers=12)

    # 运行仿真1000个时间步长
    sim.run(num_steps=1000)

    # 打印一些最终统计数据
    print("n--- Final Simulation State ---")
    print(f"Final Total Currency: {sim.metrics['total_currency'][-1]:.2f}")
    print(f"Final Avg Agent Currency: {sim.metrics['avg_agent_currency'][-1]:.2f}")
    print(f"Total Trades Executed: {sim.metrics['total_trades'][-1]}")

    print("nAgent Wealth Distribution (Top 5):")
    sorted_agents = sorted(sim.agents, key=lambda a: a.get_inventory_value(), reverse=True)
    for i, agent in enumerate(sorted_agents[:5]):
        print(f"  {i+1}. Agent {agent.agent_id[:8]}...: Curr={agent.currency:.2f}, InvValue={agent.get_inventory_value()-agent.currency:.2f}, TotalValue={agent.get_inventory_value():.2f}")
        for res, qty in agent.inventory.items():
            if res != CURRENCY:
                print(f"     - {res.name}: {qty}")

    print("nLatest Resource Prices:")
    for res in [WOOD, ORE, TOOL, FOOD]:
        print(f"  {res.name}: {sim.metrics['resource_prices'][res.name][-1]:.2f}")

    # 通常我们会在这里使用matplotlib等库来绘制图表，展示价格、财富分布等随时间的变化。
    # 例如：
    # import matplotlib.pyplot as plt
    # plt.figure(figsize=(12, 6))
    # plt.plot(sim.metrics['time'], sim.metrics['resource_prices']['Tool'], label='Tool Price')
    # plt.plot(sim.metrics['time'], sim.metrics['resource_prices']['Food'], label='Food Price')
    # plt.xlabel('Time Step')
    # plt.ylabel('Price')
    # plt.title('Resource Prices Over Time')
    # plt.legend()
    # plt.grid(True)
    # plt.show()
    #
    # plt.figure(figsize=(12, 6))
    # plt.plot(sim.metrics['time'], sim.metrics['avg_agent_currency'], label='Average Agent Currency')
    # plt.xlabel('Time Step')
    # plt.ylabel('Currency')
    # plt.title('Average Agent Currency Over Time')
    # plt.legend()
    # plt.grid(True)
    # plt.show()

运行上述代码，我们将看到类似以下的输出（具体数值会因随机性而异，但模式相似）：

Simulation initialized with 8 producers and 12 consumers.
Starting simulation for 1000 steps...

--- Step 0/1000 ---
Total Agents: 20
Total Currency: 3000.00
Avg Agent Currency: 150.00
  Latest Wood Price: 5.00
  Latest IronOre Price: 10.00
  Latest Axe Price: 50.00
  Latest Bread Price: 15.00
Total Trades: 0

... (中间大量输出，表示每个时间步的交易和状态更新) ...

--- Step 990/1000 ---
Total Agents: 20
Total Currency: 3000.00
Avg Agent Currency: 150.00
  Latest Wood Price: 7.23
  Latest IronOre Price: 12.55
  Latest Axe Price: 68.12
  Latest Bread Price: 21.08
Total Trades: 1548

Simulation finished.

--- Final Simulation State ---
Final Total Currency: 3000.00
Final Avg Agent Currency: 150.00
Total Trades Executed: 1551

Agent Wealth Distribution (Top 5):
  1. Agent 2b5c7e14...: Curr=250.78, InvValue=345.00, TotalValue=595.78
     - Axe: 5
     - Wood: 10
  2. Agent 8a3f1d2c...: Curr=189.23, InvValue=380.00, TotalValue=569.23
     - Axe: 6
     - Wood: 4
  3. Agent 5f9e0a1b...: Curr=120.50, InvValue=400.00, TotalValue=520.50
     - Axe: 8
  4. Agent c3d9b5a7...: Curr=310.15, InvValue=180.00, TotalValue=490.15
     - Bread: 12
  5. Agent e1f2a3b4...: Curr=100.00, InvValue=350.00, TotalValue=450.00
     - Axe: 7

Latest Resource Prices:
  Wood: 7.23
  IronOre: 12.55
  Axe: 68.12
  Bread: 21.08

分析与洞察：

通过观察上述输出和想象中的图表，我们可以得出一些初步的洞察：

• 价格发现: 资源价格不再停留在其 base_value，而是根据供需关系在市场中动态调整。例如，Wood 和 IronOre 的价格可能上涨，反映出生产者对原材料的需求。Axe (工具) 和 Bread (食物) 的价格也随之波动，受生产成本和消费者需求的影响。
• 财富再分配: 代理的货币和库存价值会发生变化。有些代理可能积累大量财富，而另一些则可能贫穷。这反映了代理策略、初始条件和市场运气共同作用下的财富分布。
• 经济活动量: Total Trades Executed 揭示了经济的活跃程度。更多的交易通常意味着更健康的流通。
• 市场效率: 如果价格波动剧烈，或者某些资源长时间无人交易，可能说明市场机制不够完善，或者供需严重不平衡。

通过调整 num_producers、num_consumers、代理的 initial_currency、production_recipes 中的成本和时间，以及代理的定价策略，我们可以观察到截然不同的经济现象。例如，增加生产者数量可能导致商品价格下跌；引入资源枯竭机制可能导致原材料价格飙升。

5. 进阶探讨与未来方向

我们构建的这个仿真模型只是一个起点。虚拟经济仿真的魅力在于其无限的可扩展性。

1. 更复杂的代理行为:
   • 学习与适应: 代理可以根据过去的交易数据调整其定价策略，甚至学习最优的生产/消费决策（例如，通过强化学习）。
   • 情绪与偏好: 引入代理的“满意度”、“幸福感”等情绪变量，影响其消费欲望或生产积极性。
   • 社会网络: 代理之间不仅仅通过市场交易，还可以建立信任关系，进行直接的、非市场化的交易。
2. 更丰富的市场机制:
   • 拍卖行: 实现英式拍卖、荷兰式拍卖等多种拍卖形式。
   • 去中心化市场: 模拟点对点交易，无需中央订单簿。
   • 税收与补贴: 引入政府代理，对交易征税或对特定生产进行补贴，观察对经济的影响。
   • 垄断与寡头: 少数代理控制关键资源或生产，观察其对价格和市场结构的影响。
3. 动态环境与外部冲击:
   • 资源枯竭与再生: 环境中的资源不再无限，而是会耗尽并缓慢再生，引入地理位置和运输成本。
   • 技术进步: 引入新的生产配方，提高效率或创造新产品。
   • 灾难事件: 随机事件（如洪水、瘟疫）可能破坏生产或导致资源短缺。
4. 应用场景:
   • 游戏经济设计: 平衡游戏内物品的稀有度、货币通胀与通缩，确保玩家体验。
   • 区块链经济学: 模拟代币经济模型，如 DeFi 协议中的流动性挖矿、治理代币的价值捕获。
   • 政策实验: 在无风险的环境中测试经济政策（如碳税、最低工资）的潜在影响。
   • 社会科学研究: 探索社会公平、财富分配、市场失灵等现象的微观机制。

总结展望

我们今天的旅程，从 Agent-Based Modeling 的基本概念出发，一步步构建了一个能够模拟微支付、资源竞争和价格博弈的虚拟经济仿真系统。通过代码实践，我们理解了如何将抽象的经济原理转化为可运行的程序，并观察到复杂的宏观经济现象是如何从简单的个体交互中涌现出来的。

这个仿真不仅仅是一个技术练习，它更是一个强大的思维工具。它让我们以计算的视角重新审视经济系统，从而获得对现实世界复杂性更深刻的理解。未来的探索空间广阔，希望今天的讨论能激发大家对编程与经济交叉领域的更多思考和实践。