Feat/auction (#30)

Add gathering events, in which multiple avatars participate
Add auction event

Closes #24

src/classes/auxiliary.py

@@ -26,6 +26,9 @@ class Auxiliary(Item):
     # 特殊属性（用于存储实例特定数据）
     special_data: dict = field(default_factory=dict)
 
+    def __hash__(self):
+        return hash((self.id, self.name))
+
     def get_info(self, detailed: bool = False) -> str:
         """获取信息"""
         if detailed:

src/classes/avatar/action_mixin.py

@@ -95,7 +95,15 @@ class ActionMixin:
                 continue
 
             params_for_can_start = filter_kwargs_for_callable(action.can_start, plan.params)
+            try:
                 can_start, reason = action.can_start(**params_for_can_start)
+            except TypeError as e:
+                get_logger().logger.warning(
+                    "动作启动失败: Avatar(name=%s) 动作 %s 参数校验异常: %s",
+                    self.name, plan.action_name, e
+                )
+                continue
+
             if not can_start:
                 # 记录不合法动作
                 logger = get_logger().logger

src/classes/avatar/inventory_mixin.py

@@ -162,6 +162,9 @@ class InventoryMixin:
         """
         from src.classes.prices import prices
         
+        # 记录流转
+        self.world.circulation.add_elixir(elixir)
+        
         # 使用统一的卖出价格接口
         total = prices.get_selling_price(elixir, self)
         self.magic_stone = self.magic_stone + total

src/classes/circulation.py

@@ -1,10 +1,11 @@
 from __future__ import annotations
-from typing import Dict, List, TYPE_CHECKING
+from typing import Dict, List, TYPE_CHECKING, Any
 import copy
 
 if TYPE_CHECKING:
     from src.classes.weapon import Weapon
     from src.classes.auxiliary import Auxiliary
+    from src.classes.elixir import Elixir
 
 
 class CirculationManager:
@@ -15,10 +16,17 @@ class CirculationManager:
     """
     
     def __init__(self):
-        # 存储被卖出的法宝
+        # 存储被卖出的兵器
         self.sold_weapons: List[Weapon] = []
-        # 存储被卖出的宝物
+        # 存储被卖出的辅助
         self.sold_auxiliaries: List[Auxiliary] = []
+        # 存储被卖出的丹药
+        self.sold_elixirs: List[Elixir] = []
+        
+    @property
+    def sold_item_count(self) -> int:
+        """记录一共有多少个sold items"""
+        return len(self.sold_weapons) + len(self.sold_auxiliaries) + len(self.sold_elixirs)
 
     def add_weapon(self, weapon: "Weapon") -> None:
         """记录一件流出的兵器"""
@@ -34,17 +42,68 @@ class CirculationManager:
             return
         self.sold_auxiliaries.append(auxiliary.instantiate())
 
+    def add_elixir(self, elixir: "Elixir") -> None:
+        """记录一件流出的丹药"""
+        if elixir is None:
+            return
+        # 尝试实例化，如果失败（如未实现）则拷贝，或者直接存储
+        if hasattr(elixir, "instantiate"):
+            self.sold_elixirs.append(elixir.instantiate())
+        else:
+            self.sold_elixirs.append(copy.copy(elixir))
+
+    def add_item(self, item: Any) -> None:
+        """
+        通用入口：记录一件流出的物品。
+        根据物品类型自动分发到对应的存储列表中。
+        设计此方法的目的是为了统一所有物品流出逻辑，避免各个业务模块手动判断类型。
+        """
+        if item is None:
+            return
+            
+        from src.classes.weapon import Weapon
+        from src.classes.auxiliary import Auxiliary
+        from src.classes.elixir import Elixir
+        
+        if isinstance(item, Weapon):
+            self.add_weapon(item)
+        elif isinstance(item, Auxiliary):
+            self.add_auxiliary(item)
+        elif isinstance(item, Elixir):
+            self.add_elixir(item)
+        # 未来扩展其他类型...
+
+    def remove_item(self, item: Any) -> None:
+        """
+        从流通池移除物品
+        """
+        from src.classes.weapon import Weapon
+        from src.classes.auxiliary import Auxiliary
+        from src.classes.elixir import Elixir
+        
+        if isinstance(item, Weapon):
+            if item in self.sold_weapons:
+                self.sold_weapons.remove(item)
+        elif isinstance(item, Auxiliary):
+            if item in self.sold_auxiliaries:
+                self.sold_auxiliaries.remove(item)
+        elif isinstance(item, Elixir):
+            if item in self.sold_elixirs:
+                self.sold_elixirs.remove(item)
+    
     def to_save_dict(self) -> dict:
         """序列化为字典以便存档"""
         return {
             "weapons": [self._item_to_dict(w) for w in self.sold_weapons],
-            "auxiliaries": [self._item_to_dict(a) for a in self.sold_auxiliaries]
+            "auxiliaries": [self._item_to_dict(a) for a in self.sold_auxiliaries],
+            "elixirs": [self._item_to_dict(e) for e in self.sold_elixirs]
         }
     
     def load_from_dict(self, data: dict) -> None:
         """从字典恢复数据"""
         from src.classes.weapon import weapons_by_id
         from src.classes.auxiliary import auxiliaries_by_id
+        from src.classes.elixir import elixirs_by_id
         
         self.sold_weapons = []
         for w_data in data.get("weapons", []):
@@ -62,6 +121,14 @@ class CirculationManager:
                 auxiliary.special_data = a_data.get("special_data", {})
                 self.sold_auxiliaries.append(auxiliary)
 
+        self.sold_elixirs = []
+        for e_data in data.get("elixirs", []):
+            e_id = e_data.get("id")
+            if e_id in elixirs_by_id:
+                elixir = copy.copy(elixirs_by_id[e_id])
+                elixir.special_data = e_data.get("special_data", {})
+                self.sold_elixirs.append(elixir)
+
     def _item_to_dict(self, item) -> dict:
         """将物品对象转换为简略的存储格式"""
         return {

src/classes/elixir.py

@@ -35,6 +35,9 @@ class Elixir(Item):
     effects: Union[dict[str, object], list[dict[str, object]]] = field(default_factory=dict)
     effect_desc: str = ""
 
+    def __hash__(self):
+        return hash((self.id, self.name))
+
     def get_info(self, detailed: bool = False) -> str:
         """获取信息"""
         if detailed:

src/classes/gathering/__init__.py

@@ -0,0 +1,2 @@
+from .gathering import Gathering, GatheringManager
+from .auction import Auction

src/classes/gathering/auction.py

@@ -0,0 +1,482 @@
+from typing import List, Dict, TYPE_CHECKING
+import asyncio
+from src.classes.gathering.gathering import Gathering, register_gathering
+from src.classes.event import Event
+from src.utils.config import CONFIG
+from src.utils.llm.client import call_llm_with_template
+
+if TYPE_CHECKING:
+    from src.classes.world import World
+    from src.classes.avatar import Avatar
+    from src.classes.item import Item
+
+@register_gathering
+class Auction(Gathering):
+    """
+    拍卖会事件
+    """
+    
+    def is_start(self, world: "World") -> bool:
+        """
+        检测拍卖会是否开始
+        条件：后台积攒的 sold_item_count 到达配置阈值
+        """
+        threshold = CONFIG.game.gathering.auction_trigger_count
+        return world.circulation.sold_item_count >= threshold
+
+    def get_related_avatars(self, world: "World") -> List[int]:
+        """
+        所有存活的 avatar 都参与
+        """
+        return [avatar.id for avatar in world.avatar_manager.get_living_avatars()]
+
+    def get_info(self, world: "World") -> str:
+        # TODO: Implement get_info
+        return "拍卖会正在举行..."
+
+    async def get_needs(self, world: "World", avatars: List["Avatar"]) -> Dict["Item", Dict["Avatar", int]]:
+        """
+        获取所有参与角色对拍卖品的需程度
+        返回格式: Dict[Item, Dict[Avatar, int]]
+        """
+        
+        # 1. 准备拍卖品信息
+        items_info = []
+        
+        # 收集流通管理器中的物品
+        circulation = world.circulation
+        all_items = []
+        all_items.extend(circulation.sold_weapons)
+        all_items.extend(circulation.sold_auxiliaries)
+        all_items.extend(circulation.sold_elixirs)
+        
+        for item in all_items:
+            # 假设所有 item 都有 get_info 或类似方法
+            # 这里统一用 get_detailed_info 如果有的话，或者 str(item)
+            info = getattr(item, "get_detailed_info", lambda: str(item))()
+            # 补充 ID 以便 LLM 引用
+            items_info.append(f"ID: {item.id}, Info: {info}")
+            
+        items_str = "\n".join(items_info)
+        
+        # 2. 准备角色信息并分批处理
+        batch_size = 5
+        tasks = []
+        
+        for i in range(0, len(avatars), batch_size):
+            batch_avatars = avatars[i : i + batch_size]
+            
+            # 构建该批次的 avatar_infos 字符串
+            batch_infos = {}
+            for avatar in batch_avatars:
+                # 使用 avatar.get_info(detailed=True)
+                info = avatar.get_info(detailed=True)
+                batch_infos[avatar.name] = str(info)
+            
+            # 构建模板参数
+            template_params = {
+                "avatar_infos": str(batch_infos),
+                "items": items_str
+            }
+            
+            # 创建并发任务
+            task = call_llm_with_template(
+                template_path=CONFIG.paths.templates / "auction_need.txt",
+                infos=template_params
+            )
+            tasks.append(task)
+            
+        # 3. 并发执行并收集结果
+        results = await asyncio.gather(*tasks)
+        
+        # 4. 合并结果
+        final_needs = {}
+        for result in results:
+            if isinstance(result, dict):
+                final_needs.update(result)
+                
+        # 5. 转换结构为 dict[Item, dict[Avatar, int]]
+        # 建立 name -> Avatar 映射
+        name_to_avatar = {av.name: av for av in avatars}
+        # 建立 id -> Item 映射
+        id_to_item = {str(item.id): item for item in all_items}
+        
+        item_based_needs: Dict["Item", Dict["Avatar", int]] = {}
+        
+        # 遍历 final_needs: {avatar_name: {item_id: score}}
+        for av_name, items_score_map in final_needs.items():
+            avatar = name_to_avatar.get(av_name)
+            if not avatar:
+                continue
+                
+            for item_id, score in items_score_map.items():
+                item = id_to_item.get(str(item_id))
+                if not item:
+                    continue
+                    
+                score_val = int(score)
+                # 只有需求 > 1 才记录（可选优化，1表示完全不需要）
+                if score_val <= 1:
+                    continue
+                    
+                if item not in item_based_needs:
+                    item_based_needs[item] = {}
+                item_based_needs[item][avatar] = score_val
+                
+        return item_based_needs
+
+    def _calculate_bid(self, item: "Item", need_level: int, current_balance: int) -> int:
+        """
+        计算单次出价，根据当前余额动态调整
+        """
+        from src.classes.prices import prices
+        
+        if need_level <= 1:
+            return 0
+            
+        # 策略：
+        # Need 2: min(money, base_price * 0.8)  (捡漏)
+        # Need 3: min(money, base_price * 1.5)  (略微溢价)
+        # Need 4: min(money, base_price * 3.0)  (高倍溢价)
+        # Need 5: money                         (梭哈)
+        
+        base_price = prices.get_price(item)
+        multipliers = {
+            2: 0.8,
+            3: 1.5,
+            4: 3.0,
+        }
+        
+        if need_level >= 5:
+            return current_balance
+        
+        multiplier = multipliers.get(need_level, 0.0)
+        calculated_price = int(base_price * multiplier)
+        
+        # 最终出价不能超过当前余额
+        return min(current_balance, calculated_price)
+
+    def resolve_auctions(
+        self,
+        needs: Dict["Item", Dict["Avatar", int]]
+    ) -> tuple[Dict["Item", tuple["Avatar", int]], List["Item"], Dict["Item", Dict["Avatar", int]]]:
+        """
+        结算拍卖结果
+        Returns:
+            deal_results: 成交结果 {item: (winner, price)}
+            unsold_items: 流拍物品列表
+            all_willing_prices: 所有的出价记录 {item: {avatar: price}} (用于生成故事)
+        """
+        from src.classes.prices import prices
+
+        deal_results = {}
+        unsold_items = []
+        all_willing_prices = {} 
+        
+        # 1. 建立角色资金快照
+        all_avatars = set()
+        for av_map in needs.values():
+            all_avatars.update(av_map.keys())
+        current_balances = {av: int(av.magic_stone) for av in all_avatars}
+        
+        # 2. 物品排序：按价值从高到低结算，优先处理贵重物品
+        sorted_items = sorted(needs.keys(), key=lambda x: prices.get_price(x), reverse=True)
+        
+        for item in sorted_items:
+            avatar_needs = needs[item]
+            bids = {}
+            
+            # 计算该物品的所有有效出价
+            for avatar, need_val in avatar_needs.items():
+                balance = current_balances.get(avatar, 0)
+                if balance <= 0:
+                    continue
+                    
+                bid = self._calculate_bid(item, need_val, balance)
+                if bid > 0:
+                    bids[avatar] = bid
+            
+            if bids:
+                all_willing_prices[item] = bids
+            
+            # 判定流拍
+            if not bids:
+                unsold_items.append(item)
+                continue
+            
+            # 判定赢家 (第二价格密封拍卖)
+            sorted_bids = sorted(bids.items(), key=lambda x: x[1], reverse=True)
+            winner, highest_bid = sorted_bids[0]
+            
+            deal_price = 0
+            if len(sorted_bids) >= 2:
+                second_bid = sorted_bids[1][1]
+                deal_price = min(highest_bid, second_bid + 1)
+            else:
+                # 无竞争：底价成交 (60% bid)
+                deal_price = max(1, int(highest_bid * 0.6))
+            
+            # 只有成交价 <= 余额时才有效（理论上 calculate_bid 已经保证了 bid <= balance，
+            # 但为了逻辑闭环，且 bid >= deal_price，所以 deal_price <= balance 必然成立）
+            
+            # 更新状态
+            current_balances[winner] -= deal_price
+            deal_results[item] = (winner, deal_price)
+            
+        return deal_results, unsold_items, all_willing_prices
+
+    def _generate_deal_events(
+        self,
+        world: "World",
+        deal_results: Dict["Item", tuple["Avatar", int]]
+    ) -> List[Event]:
+        """
+        生成成交事件
+        """
+        events = []
+        month_stamp = world.month_stamp
+        
+        for item, (winner, deal_price) in deal_results.items():
+            deal_event = Event(
+                month_stamp=month_stamp,
+                content=f"在拍卖会上，{winner.name}以 {deal_price} 灵石拍下了{item.name}。",
+                related_avatars=[winner.id],
+                is_major=False
+            )
+            events.append(deal_event)
+            # to_sidebar=False: 不加入 _pending_events，防止重复提交给 Simulator
+            # 但仍会执行关系计数的更新逻辑
+            winner.add_event(deal_event, to_sidebar=False)
+            
+        return events
+
+    def _generate_rivalry_events(
+        self,
+        world: "World",
+        willing_prices: Dict["Item", Dict["Avatar", int]],
+        deal_results: Dict["Item", tuple["Avatar", int]]
+    ) -> List[Event]:
+        """
+        生成竞争（压一头）事件
+        """
+        events = []
+        month_stamp = world.month_stamp
+        
+        for item in deal_results.keys():
+            bids = willing_prices.get(item, {})
+            if len(bids) < 2:
+                continue
+                
+            # 获取出价前两名
+            sorted_bids = sorted(bids.items(), key=lambda x: x[1], reverse=True)
+            winner_avatar = sorted_bids[0][0]
+            runner_up_avatar = sorted_bids[1][0]
+            
+            # 生成压一头事件
+            rivalry_content = f"在{item.name}的竞拍中，{winner_avatar.name}力压{runner_up_avatar.name}一头，最终将其收入囊中。"
+            rivalry_event = Event(
+                month_stamp=month_stamp,
+                content=rivalry_content,
+                related_avatars=[winner_avatar.id, runner_up_avatar.id],
+                is_major=False
+            )
+            events.append(rivalry_event)
+            # 给双方都添加这个互动事件 (to_sidebar=False 防止重复提交)
+            winner_avatar.add_event(rivalry_event, to_sidebar=False)
+            runner_up_avatar.add_event(rivalry_event, to_sidebar=False)
+            
+        return events
+
+    async def _generate_story(
+        self,
+        world: "World",
+        deal_results: Dict["Item", tuple["Avatar", int]],
+        willing_prices: Dict["Item", Dict["Avatar", int]]
+    ) -> List[Event]:
+        """
+        生成故事 (StoryTeller)
+        将本次拍卖的所有重要信息（成交、竞争）汇总传给 LLM，
+        让 LLM 自行选取切入点生成故事。
+        """
+        events = []
+        
+        # 1. 收集所有相关事件文本
+        interaction_lines = []
+        
+        # 收集成交信息
+        for item, (winner, deal_price) in deal_results.items():
+            interaction_lines.append(f"成交：{winner.name}以{deal_price}灵石拍下{item.name}。")
+            
+        # 收集竞争信息（压一头）
+        # 这里为了避免重复太琐碎，只记录竞争激烈（参与者>=2）的情况
+        rivalry_avatars = set()
+        for item, bids in willing_prices.items():
+            if len(bids) < 2:
+                continue
+            sorted_bids = sorted(bids.items(), key=lambda x: x[1], reverse=True)
+            winner = sorted_bids[0][0]
+            runner_up = sorted_bids[1][0]
+            interaction_lines.append(f"竞争：在{item.name}的竞拍中，{winner.name}力压{runner_up.name}（出价{sorted_bids[1][1]}）。")
+            rivalry_avatars.add(winner)
+            rivalry_avatars.add(runner_up)
+            
+        if not interaction_lines:
+            return []
+            
+        interaction_result = "\n".join(interaction_lines)
+        
+        # 2. 收集相关 items 信息
+        # 只收集成交了的或者有竞争的物品
+        related_items = set(deal_results.keys())
+        for item in willing_prices:
+            if len(willing_prices[item]) >= 2:
+                related_items.add(item)
+                
+        items_info_list = []
+        for item in related_items:
+            info = getattr(item, "get_detailed_info", lambda: str(item))()
+            items_info_list.append(f"物品：{item.name}，介绍：{info}")
+        items_info_str = "\n".join(items_info_list)
+        
+        # 3. 收集相关 avatars
+        # 主要是成交者和有明显竞争行为的
+        related_avatars = set()
+        for winner, _ in deal_results.values():
+            related_avatars.add(winner)
+        related_avatars.update(rivalry_avatars)
+        
+        if not related_avatars:
+            return []
+
+        # 4. 调用 StoryTeller
+        from src.classes.story_teller import StoryTeller
+        
+        # 准备模板参数
+        gathering_info = (
+            "事件类型：神秘拍卖会\n"
+            "场景设定：拍卖会发生在一处神秘空间，由一位面目模糊、气息深不可测的神秘人主持。"
+        )
+        
+        # 构建 details (物品信息 + 角色信息)
+        # 物品信息
+        details_list = []
+        if items_info_str:
+            details_list.append("【涉及拍品信息】")
+            details_list.append(items_info_str)
+            
+        # 角色信息
+        details_list.append("\n【相关角色信息】")
+        for av in related_avatars:
+            # 获取详细信息
+            info = av.get_info(detailed=True)
+            details_list.append(f"- {av.name}: {info}")
+            
+        details_text = "\n".join(details_list)
+        
+        story = await StoryTeller.tell_gathering_story(
+            gathering_info=gathering_info,
+            events_text=interaction_result,
+            details_text=details_text,
+            related_avatars=list(related_avatars),
+            prompt="选取其中最有趣的一个侧面或一次竞价进行描写，无需面面俱到。"
+        )
+        
+        # 5. 生成并分发事件
+        story_event = Event(
+            month_stamp=world.month_stamp,
+            content=story,
+            related_avatars=[av.id for av in related_avatars],
+            is_major=True 
+        )
+        events.append(story_event)
+        
+        for av in related_avatars:
+            av.add_event(story_event, to_sidebar=False)
+            
+        return events
+
+    async def execute(self, world: "World") -> List[Event]:
+        """
+        执行拍卖会
+        """
+        events = []
+        
+        # 0. 检查是否有物品
+        # 只要 sold_item_count >= threshold 就已经保证有物品了，但为了安全再检查一次
+        if world.circulation.sold_item_count == 0:
+            return []
+            
+        avatars = [world.avatar_manager.get_avatar(aid) for aid in self.get_related_avatars(world)]
+        # 过滤掉 None
+        avatars = [a for a in avatars if a]
+        
+        if not avatars:
+            return []
+
+        # 1. 计算需求
+        needs = await self.get_needs(world, avatars)
+        
+        # 2. 结算拍卖 (动态计算出价，处理资产穿透)
+        deal_results, unsold_items, willing_prices = self.resolve_auctions(needs)
+        
+        # 3. 执行交易 (扣钱、给物品、移除 circulation)
+        from src.classes.weapon import Weapon
+        from src.classes.auxiliary import Auxiliary
+        from src.classes.elixir import Elixir
+        from src.classes.material import Material
+        from src.classes.prices import prices
+        
+        # 处理成交物品
+        for item, (winner, price) in deal_results.items():
+            # 扣钱
+            winner.magic_stone -= price
+            
+            # 移除 circulation (先移除，避免因为交换装备导致的添加逻辑混淆)
+            world.circulation.remove_item(item)
+            
+            # 给物品
+            if isinstance(item, (Weapon, Auxiliary)):
+                # 装备并处理旧装备
+                # 特殊逻辑：拍卖会换下的旧装备直接销毁（折价回收但不再进入流通池），防止物品无限膨胀
+                
+                if isinstance(item, Weapon):
+                    old_equip = winner.weapon
+                    if old_equip:
+                        # 计算回收价
+                        refund = prices.get_selling_price(old_equip, winner)
+                        winner.magic_stone += refund
+                    # 换装
+                    winner.change_weapon(item)
+                    
+                elif isinstance(item, Auxiliary):
+                    old_equip = winner.auxiliary
+                    if old_equip:
+                        refund = prices.get_selling_price(old_equip, winner)
+                        winner.magic_stone += refund
+                    # 换装
+                    winner.change_auxiliary(item)
+                
+            elif isinstance(item, Elixir):
+                # 丹药直接服用
+                winner.consume_elixir(item)
+                
+            elif isinstance(item, Material):
+                # 材料放入背包
+                winner.add_material(item)
+        
+        # 处理流拍物品：直接销毁（移出流通池）
+        for item in unsold_items:
+            world.circulation.remove_item(item)
+            
+        # 5. 生成基础事件
+        deal_events = self._generate_deal_events(world, deal_results)
+        rivalry_events = self._generate_rivalry_events(world, willing_prices, deal_results)
+        
+        events.extend(deal_events)
+        events.extend(rivalry_events)
+        
+        # 6. 生成故事 (StoryTeller)
+        story_events = await self._generate_story(world, deal_results, willing_prices)
+        events.extend(story_events)
+        
+        return events

src/classes/gathering/gathering.py

@@ -0,0 +1,74 @@
+from abc import ABC, abstractmethod
+from typing import List, Type, TYPE_CHECKING
+
+from src.classes.event import Event
+
+if TYPE_CHECKING:
+    from src.classes.world import World
+
+class Gathering(ABC):
+    """
+    多人聚集事件/场景的抽象基类。
+    用于处理如“拍卖会”、“宗门大比”、“秘境开启”等多角色参与的复杂事件。
+    """
+    
+    @abstractmethod
+    def is_start(self, world: "World") -> bool:
+        """
+        检测该 Gathering 是否应该开始。
+        通常基于时间、地点或特定触发条件。
+        """
+        pass
+
+    @abstractmethod
+    def get_related_avatars(self, world: "World") -> List[int]:
+        """
+        获取参与该 Gathering 的角色 ID 列表。
+        """
+        pass
+
+    @abstractmethod
+    def get_info(self, world: "World") -> str:
+        """
+        获取 Gathering 的描述信息。
+        """
+        pass
+
+    @abstractmethod
+    async def execute(self, world: "World") -> List[Event]:
+        """
+        执行 Gathering 的具体逻辑。
+        
+        备注：
+        因为 Gathering 被设计为瞬时结算（在一个 step 内完成），
+        所以不需要 finish 函数。所有的交互和结果都在 execute 中一次性处理完毕。
+        """
+        pass
+
+# Global registry for Gathering classes
+GATHERING_REGISTRY: List[Type[Gathering]] = []
+
+def register_gathering(cls: Type[Gathering]):
+    """
+    装饰器：注册 Gathering 类
+    """
+    GATHERING_REGISTRY.append(cls)
+    return cls
+
+class GatheringManager:
+    def __init__(self):
+        # 实例化所有注册的 Gathering
+        self.gatherings: List[Gathering] = [cls() for cls in GATHERING_REGISTRY]
+
+    async def check_and_run_all(self, world: "World") -> List[Event]:
+        """
+        检查所有 Gathering，若满足条件则执行
+        """
+        events = []
+        for gathering in self.gatherings:
+            if gathering.is_start(world):
+                # 执行 Gathering 逻辑
+                gathering_events = await gathering.execute(world)
+                if gathering_events:
+                    events.extend(gathering_events)
+        return events

src/classes/mutual_action/gift.py

@@ -231,11 +231,13 @@ class Gift(MutualAction):
                     old_item = target.auxiliary
                     target.auxiliary = new_equip
                 
-                # 旧装备简单处理：折价变成灵石加给目标
+                # 旧装备处理：直接调用 sell_X 接口
+                # 这样既能获得灵石，也能自动触发 CirculationManager 记录流出物品
                 if old_item:
-                    refund = int(getattr(old_item, "price", 0) * 0.5)
-                    if refund > 0:
-                        target.magic_stone += refund
+                    if isinstance(old_item, Weapon):
+                        target.sell_weapon(old_item)
+                    elif isinstance(old_item, Auxiliary):
+                        target.sell_auxiliary(old_item)
                     
             else:
                 # 素材：发起者移除 -> 目标添加

src/classes/story_teller.py

@@ -41,6 +41,7 @@ class StoryTeller:
     
     TEMPLATE_SINGLE_PATH = CONFIG.paths.templates / "story_single.txt"
     TEMPLATE_DUAL_PATH = CONFIG.paths.templates / "story_dual.txt"
+    TEMPLATE_GATHERING_PATH = CONFIG.paths.templates / "story_gathering.txt"
 
     @staticmethod
     def _build_avatar_infos(*actors: "Avatar") -> Dict[str, dict]:
@@ -127,5 +128,48 @@ class StoryTeller:
         
         return StoryTeller._make_fallback_story(event, res, infos["style"])
 
+    @staticmethod
+    async def tell_gathering_story(
+        gathering_info: str,
+        events_text: str,
+        details_text: str,
+        related_avatars: list["Avatar"],
+        prompt: str = ""
+    ) -> str:
+        """
+        生成聚会/拍卖会等多人事件的故事。
+        通用接口，适配 story_gathering.txt 模板。
+        
+        Args:
+            gathering_info: 事件本身的设定信息（如地点、背景、规则等）
+            events_text: 发生的具体事件/交互记录
+            details_text: 详细信息（包括角色信息、物品信息等）
+            related_avatars: 参与的角色列表（主要用于获取世界背景信息）
+            prompt: 额外提示词
+        """
+        if not related_avatars:
+            return events_text
+
+        # 使用第一个角色的世界信息
+        world_info = related_avatars[0].world.static_info
+            
+        infos = {
+            "world_info": world_info,
+            "gathering_info": gathering_info,
+            "events": events_text,
+            "details": details_text,
+            "style": random.choice(story_styles),
+            "story_prompt": prompt
+        }
+        
+        # 增加 token 上限以支持长故事
+        data = await call_llm_with_task_name("story_teller", StoryTeller.TEMPLATE_GATHERING_PATH, infos)
+        story = data.get("story", "").strip()
+        
+        if story:
+            return story
+            
+        return events_text
+
 
 __all__ = ["StoryTeller"]

src/classes/weapon.py

@@ -30,6 +30,9 @@ class Weapon(Item):
     # 特殊属性（如万魂幡的吞噬魂魄计数）
     special_data: dict = field(default_factory=dict)
 
+    def __hash__(self):
+        return hash((self.id, self.name))
+
     def get_info(self, detailed: bool = False) -> str:
         """获取信息"""
         if detailed:

src/classes/world.py

@@ -7,6 +7,7 @@ from src.classes.calendar import Year, Month, MonthStamp
 from src.classes.avatar_manager import AvatarManager
 from src.classes.event_manager import EventManager
 from src.classes.circulation import CirculationManager
+from src.classes.gathering.gathering import GatheringManager
 
 if TYPE_CHECKING:
     from src.classes.avatar import Avatar
@@ -26,6 +27,8 @@ class World():
     phenomenon_start_year: int = 0
     # 出世物品流通管理器
     circulation: CirculationManager = field(default_factory=CirculationManager)
+    # Gathering 管理器
+    gathering_manager: GatheringManager = field(default_factory=GatheringManager)
     # 世界历史文本
     history: str = ""
 

src/sim/new_avatar.py

@@ -19,6 +19,7 @@ from src.classes.technique import get_technique_by_sect, attribute_to_root, Tech
 from src.classes.weapon import Weapon, weapons_by_id, weapons_by_name
 from src.classes.auxiliary import Auxiliary, auxiliaries_by_id, auxiliaries_by_name
 from src.classes.persona import Persona, personas_by_id, personas_by_name
+from src.classes.magic_stone import MagicStone
 
 
 # —— 参数常量（便于调参）——
@@ -437,6 +438,7 @@ class AvatarFactory:
             sect=plan.sect,
         )
 
+        avatar.magic_stone = MagicStone(50)
         avatar.tile = world.map.get_tile(avatar.pos_x, avatar.pos_y)
 
         SectRankAssigner.assign_one(avatar, world)
@@ -550,6 +552,7 @@ class AvatarFactory:
                 sect=sect,
             )
 
+            avatar.magic_stone = MagicStone(50)
             avatar.tile = world.map.get_tile(x, y)
 
             if sect is not None:

src/sim/simulator.py

@@ -261,6 +261,13 @@ class Simulator:
         events = [e for e in results if e]
         return events
     
+    async def _phase_process_gatherings(self):
+        """
+        Gathering 结算阶段：
+        检查并执行注册的多人聚集事件（如拍卖会、大比等）。
+        """
+        return await self.world.gathering_manager.check_and_run_all(self.world)
+    
     def _phase_update_celestial_phenomenon(self):
         """
         更新天地灵机：
@@ -393,48 +400,60 @@ class Simulator:
         """
         events = [] # list of Event
 
-        # 0. 感知与认知更新阶段（包括自动占据洞府）
+        # 1. 感知与认知更新阶段（包括自动占据洞府）
         #    在思考和决策之前，先让角色感知世界
         self._phase_update_perception_and_knowledge()
 
-        # 0.5 长期目标思考阶段（在决策之前）
+        # 2. 长期目标思考阶段（在决策之前）
         events.extend(await self._phase_long_term_objective_thinking())
 
-        # 1. 决策阶段
+        # 3. Gathering 结算阶段
+        events.extend(await self._phase_process_gatherings())
+
+        # 4. 决策阶段
         await self._phase_decide_actions()
 
-        # 2. 提交阶段
+        # 5. 提交阶段
         events.extend(self._phase_commit_next_plans())
 
-        # 3. 执行阶段
+        # 6. 执行阶段
         events.extend(await self._phase_execute_actions())
 
-        # 4. 关系演化阶段
+        # 7. 关系演化阶段
         events.extend(await self._phase_evolve_relations())
 
-        # 5. 结算死亡
+        # 8. 结算死亡
         events.extend(self._phase_resolve_death())
 
-        # 6. 年龄与新生
+        # 9. 年龄与新生
         events.extend(self._phase_update_age_and_birth())
 
-        # 7. 被动结算（时间效果+奇遇）
+        # 10. 被动结算（时间效果+奇遇）
         events.extend(await self._phase_passive_effects())
 
-        # 8. 绰号生成
+        # 11. 绰号生成
         events.extend(await self._phase_nickname_generation())
 
-        # 9. 更新天地灵机
+        # 12. 更新天地灵机
         events.extend(self._phase_update_celestial_phenomenon())
 
-        # 10. 日志
+        # 13. 日志
+        # 去重：基于 ID 去重，防止同一个事件对象（或相同ID的事件）被多次添加
+        # 常见情况：Gathering 既返回了事件，又将其加入了 Avatar 的 pending_events
+        unique_events = {}
+        for e in events:
+            if e.id not in unique_events:
+                unique_events[e.id] = e
+        # 保持原有顺序（Python 3.7+ dict 保持插入序）
+        events = list(unique_events.values())
+
         # 统一写入事件管理器
         if hasattr(self.world, "event_manager") and self.world.event_manager is not None:
             for e in events:
                 self.world.event_manager.add_event(e)
         self._phase_log_events(events)
 
-        # 11. 时间推进
+        # 14. 时间推进
         self.world.month_stamp = self.world.month_stamp + 1
 
 

static/config.yml

@@ -27,6 +27,8 @@ game:
   npc_awakening_rate_per_month: 0.01
   fortune_probability: 0.005
   misfortune_probability: 0.005
+  gathering:
+    auction_trigger_count: 10
 
 df:
   ids_separator: ";"

static/templates/auction_need.txt

@@ -0,0 +1,21 @@
+你是一个价值评估者，这是一个仙侠世界，你负责来评估一些物品，对于一些NPC的价值。
+
+你需要进行决策的NPC的dict[AvatarName, info]为：
+{avatar_infos}
+需要评价需求的物品为：
+{items}
+
+注意，只返回json格式的结果。
+格式为：
+{{
+    "{{avatar_name_1}}": {{
+        {{item_id_1}}: int, # 1~5
+        {{item_id_2}}: int,
+        ...
+    }}, ...
+}}
+
+要求；
+1. 所有的角色要在这里评估所有的物品，也就是最后返回的是num_avatar * num_item。都必须有一个估计。
+2. 分析每一个角色的情况，根据其情况返回需求程度。比如快老死的人对延寿丹药就是最需要的
+3. 需求从1到5，1是完全不需要，5是完全需要。

static/templates/story_gathering.txt

@@ -0,0 +1,27 @@
+你是一个小说家，这是一个仙侠世界，你需要这个修仙界中某一个集体事件，选取一个有趣的切入点或侧面，创作一段300~800字的小故事。
+
+世界背景：
+{world_info}
+
+事件设定：
+{gathering_info}
+
+events：
+{events}
+
+事件细节：
+{details}
+
+写作风格提示：{style}
+额外主题提示：{story_prompt}
+
+注意：
+1. 不要试图面面俱到地描写所有事件，请**自行筛选**其中最有趣、最有戏剧性的部分进行扩写。
+2. 侧重描写角色之间的互动、心理博弈、竞争或合作。
+3. 重点描写竞价过程中的紧张气氛。
+
+只返回json格式的结果，格式为：
+{{
+    "thinking": ..., // 简单思考故事剧情
+    "story": "" // 第三人称的故事正文，仙侠语言风格
+}}

tests/test_action_gift.py

@@ -24,7 +24,8 @@ def target_avatar(base_world):
         age=Age(20, Realm.Qi_Refinement),
         gender=Gender.FEMALE,
         pos_x=0,
-        pos_y=0
+        pos_y=0,
+        personas=[],  # 避免随机特质影响价格测试
     )
 
 @pytest.fixture
@@ -159,7 +160,7 @@ class TestGiftAction:
         dummy_avatar.weapon = new_weapon
         
         old_weapon = create_test_weapon("旧铁剑", Realm.Qi_Refinement, weapon_id=999)     
-        old_weapon.price = 100
+        # old_weapon.price = 100 # Prices 系统接管后，价格由 Realm 决定 (练气期=150)，不再手动指定
         target_avatar.weapon = old_weapon
         target_avatar.magic_stone = 0
         
@@ -169,8 +170,8 @@ class TestGiftAction:
         gift_action._settle_feedback(target_avatar, "Accept")
         
         assert target_avatar.weapon == new_weapon
-        # 50% refund
-        assert target_avatar.magic_stone == 50
+        # 练气期武器基准价 150，卖出倍率 1.0 (无特质加成) -> 150
+        assert target_avatar.magic_stone == 150
 
     # --- 4. 上下文与描述 ---
 

tests/test_auction.py

@@ -0,0 +1,425 @@
+import pytest
+from unittest.mock import MagicMock, patch, AsyncMock
+from src.classes.gathering.auction import Auction
+from src.classes.item import Item
+from src.classes.weapon import Weapon
+from src.classes.auxiliary import Auxiliary
+from src.classes.prices import prices
+from src.utils.config import CONFIG
+
+# Monkeypatch removed as Weapon/Auxiliary now have __hash__ implemented
+
+@pytest.mark.asyncio
+async def test_auction_is_start(base_world, mock_item_data):
+    auction = Auction()
+    weapon = mock_item_data["obj_weapon"]
+    
+    # 初始状态，sold_item_count 为 0
+    # 清空 circulation
+    base_world.circulation.sold_weapons = []
+    base_world.circulation.sold_auxiliaries = []
+    base_world.circulation.sold_elixirs = []
+    
+    # 设置阈值
+    CONFIG.game.gathering.auction_trigger_count = 5
+    
+    assert auction.is_start(base_world) is False
+    
+    # 增加物品数量达到阈值
+    for _ in range(5):
+        base_world.circulation.add_weapon(weapon)
+        
+    assert auction.is_start(base_world) is True
+
+def test_calculate_bid(dummy_avatar, mock_item_data):
+    auction = Auction()
+    item = mock_item_data["obj_weapon"]
+    base_price = prices.get_price(item)
+    
+    # Case 1: 需求低 (<=1) -> 出价 0
+    assert auction._calculate_bid(item, 1, 1000) == 0
+    
+    # Case 2: 需求 2 (捡漏 0.8)
+    expected_price = int(base_price * 0.8)
+    assert auction._calculate_bid(item, 2, 100000) == expected_price
+    
+    # Case 3: 余额不足 -> 出价 = 余额
+    avatar_money = 10
+    bid = auction._calculate_bid(item, 3, avatar_money) # need 3 is 1.5x, definitely > 10
+    assert bid == avatar_money
+    
+    # Case 4: 需求 5 (梭哈) -> 出价 = 余额
+    assert auction._calculate_bid(item, 5, 5000) == 5000
+
+def test_resolve_auctions_basic(dummy_avatar, mock_item_data):
+    """测试基本的竞价结算逻辑（单物品）"""
+    auction = Auction()
+    item = mock_item_data["obj_weapon"]
+    
+    avatar1 = dummy_avatar
+    avatar1.magic_stone = 1000
+    avatar1.name = "A1"
+    
+    # 创建第二个角色
+    avatar2 = MagicMock()
+    avatar2.magic_stone = 1000
+    avatar2.name = "A2"
+    avatar2.__hash__ = MagicMock(return_value=123) # Make it hashable for dict keys
+    
+    # 模拟需求字典
+    needs = {
+        item: {
+            avatar1: 4, # High need
+            avatar2: 2  # Low need
+        }
+    }
+    
+    # Mock prices
+    with patch("src.classes.prices.prices.get_price", return_value=100):
+        deal_results, unsold, willing = auction.resolve_auctions(needs)
+        
+    # 验证结果
+    assert item in deal_results
+    winner, price = deal_results[item]
+    
+    # A1 出价: 100 * 3.0 = 300
+    # A2 出价: 100 * 0.8 = 80
+    # 成交价应为第二高价(80) + 1 = 81
+    assert winner == avatar1
+    assert price == 81
+    assert not unsold
+
+def test_resolve_auctions_asset_protection(dummy_avatar, mock_item_data):
+    """测试资产穿透保护：同一个角色竞拍多个物品"""
+    auction = Auction()
+    item1 = mock_item_data["obj_weapon"] # 贵
+    item2 = mock_item_data["obj_material"] # 便宜
+    
+    avatar = dummy_avatar
+    avatar.magic_stone = 100  # 总共只有 100
+    
+    needs = {
+        item1: {avatar: 5}, # 梭哈 item1
+        item2: {avatar: 5}  # 梭哈 item2
+    }
+    
+    # Mock prices: item1=80, item2=50
+    # item1 应该先结算（价值高），因为是梭哈(need=5)，出价100。
+    # 如果只有一人竞拍，成交价 = max(1, 100 * 0.6) = 60。
+    # 剩余余额 = 100 - 60 = 40。
+    # item2 结算时，余额只有 40，虽然 need=5，但出价只能是 40。
+    # item2 成交价 = max(1, 40 * 0.6) = 24。
+    
+    def get_price_side_effect(item):
+        if item == item1: return 80
+        return 50
+        
+    with patch("src.classes.prices.prices.get_price", side_effect=get_price_side_effect):
+        deal_results, unsold, willing = auction.resolve_auctions(needs)
+    
+    # 验证 item1
+    assert item1 in deal_results
+    winner1, price1 = deal_results[item1]
+    assert winner1 == avatar
+    assert price1 == 60 # 100 * 0.6
+    
+    # 验证 item2
+    assert item2 in deal_results
+    winner2, price2 = deal_results[item2]
+    assert winner2 == avatar
+    # 此时余额只剩 40，出价 40，成交价 40 * 0.6 = 24
+    assert price2 == 24
+    
+    # 总花费 84 <= 100，保护成功
+    assert price1 + price2 <= 100
+
+def test_resolve_auctions_unsold(mock_item_data):
+    """测试流拍"""
+    auction = Auction()
+    item = mock_item_data["obj_weapon"]
+    
+    # 空需求或者需求都很低导致不出价
+    needs = {
+        item: {}
+    }
+    
+    with patch("src.classes.prices.prices.get_price", return_value=100):
+        deal_results, unsold, willing = auction.resolve_auctions(needs)
+        
+    assert item not in deal_results
+    assert item in unsold
+
+@pytest.mark.asyncio
+async def test_execute_flow(base_world, dummy_avatar, mock_item_data):
+    """测试完整的 execute 流程，包括物品交易和销毁"""
+    auction = Auction()
+    item_sold = mock_item_data["obj_weapon"]
+    item_unsold = mock_item_data["obj_auxiliary"] # 使用 Auxiliary 代替 Material
+    
+    # 设置环境
+    # 将物品加入 circulation 以便测试移除逻辑
+    base_world.circulation.sold_weapons = [item_sold]
+    base_world.circulation.sold_auxiliaries = [item_unsold]
+    
+    # 设置 Avatar
+    dummy_avatar.magic_stone = 1000
+    dummy_avatar.weapon = None # 确保没有武器
+    dummy_avatar.auxiliary = None 
+    
+    # 确保 avatar 在 avatar_manager 中
+    base_world.avatar_manager.avatars[dummy_avatar.id] = dummy_avatar
+    
+    # Mock methods
+    # 1. get_related_avatars
+    auction.get_related_avatars = MagicMock(return_value=[dummy_avatar.id])
+    
+    # 2. get_needs (Async) -> 让 item_sold 有人买，item_unsold 没人买
+    async def mock_get_needs(*args, **kwargs):
+        return {
+            item_sold: {dummy_avatar: 4}, # High need
+            item_unsold: {dummy_avatar: 1} # No need
+        }
+    auction.get_needs = mock_get_needs
+    
+    # 3. Mock StoryTeller to avoid LLM
+    with patch("src.classes.story_teller.StoryTeller.tell_gathering_story", new_callable=AsyncMock) as mock_story:
+        mock_story.return_value = "拍卖会故事..."
+        
+        # 4. Mock prices
+        with patch("src.classes.prices.prices.get_price", return_value=100):
+            events = await auction.execute(base_world)
+            
+    # 验证结果
+    
+    # 1. 物品去向
+    # item_sold 应该被 dummy_avatar 装备
+    assert dummy_avatar.weapon == item_sold
+    # item_sold 应该不在 circulation 中
+    assert item_sold not in base_world.circulation.sold_weapons
+    
+    # item_unsold 应该被销毁 (不在 circulation 中，也不在 avatar 背包/装备 中)
+    assert item_unsold not in base_world.circulation.sold_auxiliaries
+    assert dummy_avatar.auxiliary != item_unsold
+    
+    # 2. 资金扣除
+    # Base price 100, Need 4 (3.0x) -> Bid 300
+    # Single bidder -> Deal 300 * 0.6 = 180
+    # Balance 1000 - 180 = 820
+    assert dummy_avatar.magic_stone == 820
+    
+    # 3. 事件生成
+    assert len(events) > 0
+    # 应该包含 story event
+    assert any(e.content == "拍卖会故事..." for e in events)
+
+def test_items_are_hashable():
+    """测试物品类是否可哈希（用作字典键）"""
+    from src.classes.weapon import Weapon
+    from src.classes.weapon_type import WeaponType
+    from src.classes.auxiliary import Auxiliary
+    from src.classes.elixir import Elixir, ElixirType
+    from src.classes.cultivation import Realm
+
+    # Weapon
+    w = Weapon(
+        id=1, 
+        name="TestSword", 
+        weapon_type=WeaponType.SWORD, 
+        realm=Realm.Qi_Refinement, 
+        desc="Test",
+        special_data={"a": 1} # mutable field
+    )
+    s = set()
+    s.add(w)
+    assert w in s
+    d = {w: 1}
+    assert d[w] == 1
+
+    # Auxiliary
+    a = Auxiliary(
+        id=2, 
+        name="TestAux", 
+        realm=Realm.Qi_Refinement, 
+        desc="Test",
+        special_data={"b": 2} # mutable field
+    )
+    s = set()
+    s.add(a)
+    assert a in s
+    
+    # Elixir
+    e = Elixir(
+        id=3,
+        name="TestElixir",
+        realm=Realm.Qi_Refinement,
+        type=ElixirType.Heal,
+        desc="Test",
+        price=100
+    )
+    s = set()
+    s.add(e)
+    assert e in s
+
+def test_resolve_auctions_tie_breaking(dummy_avatar, mock_item_data):
+    """测试出价相同时的判定（稳定性）"""
+    auction = Auction()
+    item = mock_item_data["obj_weapon"]
+    
+    # 两个角色，需求相同，资金充足 -> 理论上出价相同
+    avatar1 = dummy_avatar
+    avatar1.magic_stone = 1000
+    avatar1.name = "A1"
+    
+    avatar2 = MagicMock()
+    avatar2.magic_stone = 1000
+    avatar2.name = "A2"
+    avatar2.__hash__ = MagicMock(return_value=12345) 
+    
+    # 手动构建 needs 字典，控制 key 的顺序
+    # 情况1: A1 在前
+    needs1 = {
+        item: {avatar1: 5, avatar2: 5}
+    }
+    
+    with patch("src.classes.prices.prices.get_price", return_value=100):
+        deal_results1, _, _ = auction.resolve_auctions(needs1)
+        
+    winner1, _ = deal_results1[item]
+    # 如果是稳定排序，且 bid 相等，应该保持顺序，winner 是 A1
+    assert winner1 == avatar1
+    
+    # 情况2: A2 在前
+    needs2 = {
+        item: {avatar2: 5, avatar1: 5}
+    }
+    with patch("src.classes.prices.prices.get_price", return_value=100):
+        deal_results2, _, _ = auction.resolve_auctions(needs2)
+    
+    winner2, _ = deal_results2[item]
+    # winner 应该是 A2
+    assert winner2 == avatar2
+
+def test_resolve_auctions_no_refund_consideration(dummy_avatar, mock_item_data):
+    """测试拍卖结算时不考虑后续装备出售的退款（防止透支）"""
+    auction = Auction()
+    item1 = mock_item_data["obj_weapon"] # 贵, 先结算
+    item2 = mock_item_data["obj_elixir"] # 便宜, 后结算
+    
+    avatar = dummy_avatar
+    avatar.magic_stone = 100 
+    
+    # 假设 avatar 身上有装备，卖出可得 50
+    old_weapon = MagicMock()
+    avatar.weapon = old_weapon
+    # 但 resolve_auctions 只看 snapshot，不看装备退款
+    
+    needs = {
+        item1: {avatar: 5}, # 梭哈 item1, cost 100
+        item2: {avatar: 5}  # 梭哈 item2
+    }
+    
+    # Mock prices: item1=80, item2=50
+    # item1 price 80, need 5 -> bid 100 (balance). Deal: 100*0.6 = 60.
+    # item2 price 50. Remaining balance 100-60=40.
+    # need 5 -> bid 40 (balance). Deal: 40*0.6 = 24.
+    # If refund (50) was considered, balance would be 40+50=90.
+    # Bid 90 -> Deal 54.
+    
+    def get_price_side_effect(item):
+        if item == item1: return 80
+        return 50
+        
+    with patch("src.classes.prices.prices.get_price", side_effect=get_price_side_effect):
+        deal_results, _, _ = auction.resolve_auctions(needs)
+        
+    # item1 应该成交，消耗 60 (100 * 0.6)
+    assert deal_results[item1][0] == avatar
+    assert deal_results[item1][1] == 60
+    
+    # item2 应该成交，消耗 24 (40 * 0.6)
+    # 证明使用了 40 的余额，而不是 90 (如果包含退款)
+    assert item2 in deal_results
+    assert deal_results[item2][1] == 24
+    
+    # 总消耗 84 <= 100
+    assert deal_results[item1][1] + deal_results[item2][1] <= 100
+
+@pytest.mark.asyncio
+async def test_execute_item_types(base_world, dummy_avatar, mock_item_data):
+    """测试不同类型物品的执行逻辑 (Elixir)"""
+    auction = Auction()
+    elixir = mock_item_data["obj_elixir"]
+    
+    dummy_avatar.magic_stone = 1000
+    base_world.circulation.sold_elixirs = [elixir]
+    
+    # Register avatar
+    base_world.avatar_manager.avatars[dummy_avatar.id] = dummy_avatar
+    
+    # Mock resolve_auctions
+    auction.resolve_auctions = MagicMock(return_value=(
+        {elixir: (dummy_avatar, 100)}, 
+        [], 
+        {}
+    ))
+    
+    # Mock dependencies
+    auction.get_related_avatars = MagicMock(return_value=[dummy_avatar.id])
+    auction.get_needs = AsyncMock(return_value={}) # ignored by mocked resolve
+    auction._generate_deal_events = MagicMock(return_value=[])
+    auction._generate_rivalry_events = MagicMock(return_value=[])
+    auction._generate_story = AsyncMock(return_value=[])
+    
+    # Mock circulation remove
+    base_world.circulation.remove_item = MagicMock()
+    
+    # Mock consume_elixir
+    dummy_avatar.consume_elixir = MagicMock()
+    
+    # Ensure items are "in" circulation logic (count > 0)
+    # Circulation.sold_item_count is a property, depends on lists.
+    # We set sold_elixirs above, so it should be > 0.
+    
+    await auction.execute(base_world)
+    
+    # Verify consume_elixir called
+    dummy_avatar.consume_elixir.assert_called_once_with(elixir)
+    
+    # Verify remove_item called
+    base_world.circulation.remove_item.assert_called_once_with(elixir)
+
+@pytest.mark.asyncio
+async def test_get_needs_parsing(base_world, dummy_avatar, mock_item_data):
+    """测试 get_needs 的 LLM 结果解析逻辑"""
+    auction = Auction()
+    item = mock_item_data["obj_weapon"]
+    # Mock circulation
+    base_world.circulation.sold_weapons = [item]
+    
+    # Mock LLM response
+    mock_response = {
+        dummy_avatar.name: {
+            str(item.id): 5  # High need
+        }
+    }
+    
+    with patch("src.classes.gathering.auction.call_llm_with_template", new_callable=AsyncMock) as mock_llm:
+        mock_llm.return_value = mock_response
+        
+        needs = await auction.get_needs(base_world, [dummy_avatar])
+        
+    assert item in needs
+    assert needs[item][dummy_avatar] == 5
+    
+    # Test filtering of low needs (<=1)
+    mock_response_low = {
+        dummy_avatar.name: {
+            str(item.id): 1 
+        }
+    }
+    with patch("src.classes.gathering.auction.call_llm_with_template", new_callable=AsyncMock) as mock_llm:
+        mock_llm.return_value = mock_response_low
+        needs = await auction.get_needs(base_world, [dummy_avatar])
+        
+    # Should be empty because score 1 is filtered
+    assert item not in needs or not needs.get(item)