每个Agent 2个动作的多智能体仿真

Question

这是一个模拟代码，运行一个多智能体系统，开发来模拟价格如何波动的动态过程中，代理人作出他们的行动。每个代理可以根据其价值和价格做出“获取”或“删除”所选选项的决定。由于代码的速度太慢，我无法实现许多增强。运行小规模模拟需要花费大量时间(例如选项= 20，代理= 40，持续时间= 10)。

我读到了map和lambda如何优化嵌套循环。在这种情况下，如何实施这些原则？函数get和remove有自己的几个步骤。

使用以下方法运行：

pile, index = init_pile()
run_pile(pile, index)

初始化功能：

def init_pile(agents= 40, options=20):

    # initialise pile
    colnames = [ 'wallet']
    rownames = [ 'counts','value','price']
    for i in range(options):
        colnames.append('o_' + str(i+1))
    for o in range(agents):
        rownames.append('agent_' + str(o+1))
    pile = pd.DataFrame(0, index=rownames,columns=colnames)

    # initialise budgets for agents
    mu, sigma = 500, 20  
    budgets = np.random.normal(mu, sigma, agents)
    pile.loc[3:,'wallet'] = budgets

    # initialise option counts
    values = np.random.randint(10, 50, options)
    pile.loc['counts',1:] = values

    # initialise option values
    mu, sigma = 30, 5  
    values = np.random.normal(mu, sigma, options)
    pile.loc['value',1:] = values

    # initialise option prices
    mu, sigma = 29, 4  
    values = np.random.normal(mu, sigma, options)
    pile.loc['price',1:] = values

    # initializing index
    index = []
    index.append(1000)

    return pile,index  

主桩建造功能：

def run_pile(pile,index,duration=20,agents= 40, options=20):
    agents = list(range(agents))
    for time in range(duration):
        for option in range(options):
            for agent in agents:
                # agent will buy if option perceived value > current value
                price = pile.loc['price', str('o_' + str(option + 1))]
                value = pile.loc['value', str('o_' + str(option + 1))]

                if price < value:
                    get(pile, agent, option)

                if price > value:
                    remove(pile, agent, option)

                #decay(pile,options)

            agents = shuffle(agents)
        calculate_index(pile, index)

    print(pile)

get函数：

def get(pile, agent, option):
    # if the agent can get the option
    budget = pile.loc[str('agent_' + str(agent + 1)), 'wallet']
    current_price = pile.loc['price', str('o_' + str(option + 1))]
    counts = pile.loc['counts', str('o_' + str(option + 1))]

    # if agent has money and there are options remaining
    if budget > current_price and counts > 0:
        # buy the option
        pile.loc[str('agent_' + str(agent + 1)), str('o_' + str(option + 1))] +=1

        # subtract from wallet
        pile.loc[str('agent_' + str(agent + 1)), 'wallet'] -= current_price

        # subtract from option count 
        pile.loc['counts', str('o_' + str(option + 1))] -= 1

        # inflate price of option
        pile.loc['price', str('o_' + str(option + 1))] *= 1.1

remove函数：

def remove(pile, agent, option):
    current_price = pile.loc['price', str('o_' + str(option + 1))]
    #if the agent owns the option 
    if pile.loc[str('agent_' + str(agent + 1)), str('o_' + str(option + 1))] >0:
        # agent loses the option
        pile.loc[str('agent_' + str(agent + 1)),str('o_' + str(option + 1))] -= 1

        # agent gains money placed in wallet
        pile.loc[str('agent_' + str(agent + 1)), 'wallet'] += current_price

        # add option count 
        pile.loc['counts', str('o_' + str(option + 1))] += 1

        # deflate the option price
        pile.loc['price', str('o_' + str(option + 1))] *= 0.99

calculate_index函数：

def calculate_index(pile, index, method = 'average'):
    if method == 'average':
        ind = ((pile.iloc[2,1:].sum()/20) / index[-1] ) * 1000
        index.append(ind)

decay函数：

def decay(pile, options):
    # if option price is more than perceived value
    # deflate the price
    for option in range(options):
        pile.loc['price', str('o_' + str(option + 1))] *= 0.999999

decay函数包含类似于get和remove的操作，但是我对其进行了注释，因为它已经太慢了，我需要解决这个问题。

Answer 1

将所有的数据都放在一个单一的结构中，很难准确地理解事物是如何相互作用的，因为您需要做额外的工作来构建相关的索引。相反，您至少应该将pile分解为3种结构：

• 形状wallets (1，number_of_agents)；
• 形状options (number_of_options，3)；
• stocks的形状(number_of_options，number_of_agents)。

也就是说，如果您打算对pandas提供的数据文件使用一些操作。

从外表上看，你不知道。

你似乎想要的是，对于每一种选择，每个代理人都有自己的决定；没有特定的顺序，而是一个接一个。因此，您将无法利用pandas提供的矢量化操作，您也可以放弃这种依赖。

相反，我将创建一个Option类来保存一个选项的属性(例如价格、值、金额和可能的id)，并创建一个Agent类来保存一个钱包和一个期权组合。Option类将对buy、sell和decay负责。Agent类将负责get或remove从他们的投资组合中选择一个选项。

然后使用线程组织所有这些代理：threading模块提供了运行线程 (不按特定顺序执行)、代码锁段 (使单个代理一次检查一个选项)和同步剂的方法，以便每个代理都等待每个人都有机会检查一个选项，然后再转到下一个选项。

完成后，您只需根据您的持续时间重复所需的时间：

import random
import threading


class InvalidOperation(Exception):
    pass


class Option:
    def __init__(self, id, price, value, count):
        self.id = id
        self.price = price
        self.value = value
        self.count = count

    def buy(self):
        if not self.count:
            raise InvalidOperation

        self.count -= 1
        self.price *= 1.1

    def sell(self):
        self.count += 1
        self.price *= 0.99

    def decay(self):
        self.price *= 0.999999

    def __repr__(self):
        return f'{self.__class__.__name__}({self.id}, {self.price}, {self.value}, {self.count})'


class Agent:
    def __init__(self, wallet, options):
        self.wallet = wallet
        self.portfolio = {option.id: 0 for option in options}


    def run(self, option):
        if option.price < option.value:
            self.get(option)

        if option.price > option.value:
            self.remove(option)

        option.decay()

    def get(self, option):
        if self.wallet > option.price and option.count:
            print(id(self), 'buying', option)
            self.wallet -= option.price
            self.portfolio[option.id] += 1
            option.buy()

    def remove(self, option):
        if self.portfolio[option.id] > 0:
            print(id(self), 'selling', option)
            self.wallet += option.price
            self.portfolio[option.id] -= 1
            option.sell()


def run_agent(agent, options, barrier, mutex):
    for option in options:
        with mutex:
            # Ensure that a single  agent take a decision at a time
            agent.run(option)
        # Everyone will get a saying at this
        # option before going to the next
        barrier.wait()


def run_pile(options, agents=40, duration=20):
    mutex = threading.Lock()
    barrier = threading.Barrier(agents)

    index = []
    agents = [
            Agent(random.normalvariate(500, 20), options)
            for _ in range(agents)
    ]

    for time in range(duration):
        print('time', time, 'starting')
        agents_threads = [
                threading.Thread(
                    target=run_agent,
                    args=(agent, options, barrier, mutex),
                ) for agent in agents
        ]
        for thread in agents_threads:
            thread.start()
        for thread in agents_threads:
            thread.join()

        print('time', time, 'ending')
        calculate_index(options, agents, index)

    return index


def build_options(options=20):
    return [
            Option(
                i,
                random.normalvariate(29, 4),
                random.normalvariate(30, 5),
                random.randint(10, 50),
            ) for i in range(options)
    ]


def calculate_index(options, agents, index):
    # Keeping empty as I don't really get it
    pass


if __name__ == '__main__':
    options = build_options()
    run_pile(options)

我留下了一些print调用，这样您就可以看到事情是“有序的”，但是您应该在最后的代码中删除它们。