Demand from Preferences Part 1

This post covers a second ABM model for demand This model uses mesa and shinylive for python.

The second model uses mesa + shinylive + altair-viz.

But more significantly it uses a Polya urn model to model preferences over products. And the products come from a Hoppe urn model which is closely related to the Chinese Restaurant Process.

The outcome is still a preliminary model of demand. In the next installment I wish to add a negative feedback mechanism to the model so that agents can use thier happiness as a measure of thier welfare to adjust thier purchasing decisions to be more in line with thier budget constraint.

Currently the products follow a Hoppe urn model which is closely related to the Chinese Restaurant Process. The Hoppe urn model I implemented below is based on the generlised polya urn model, meaning that we can addjust the rate of innovation.

I point this out since in most runs the number of products grows increasingly slowly tapering our at 8 after 100 steps. This is one of the two things I track in this model. The second is the preferences of each agent over the products. The initial setting only shows a few agents and their preferences are not so different. But when I pick say 40 agents things get dramatically different and we see a much more diverse set of preferences evolving.

This also suggest a second direction for generalization of the model - looking at aggregation from different population segments (sorted by income or by similar preferences?).

Further work on this specific model is to see how demand aggregates as we get more agents.

Further work on other models are broken into

• adding a budget constraint and a negative feedback loop to get agents to adjust thier purchasing decisions based on thier happiness (utility) and thier budget constraint.
• adding segregation to get local markets.
• adding product categories and affinity between products, allowing us to model substitutes and complements.
• Associating products with hierarchy similar to Maslow’s hierarchy of needs. This should help partition the preference into hard and soft preferences. (Needs vs wants) with multiple levels of wants. This should be a smal tree. I think a nested CRP may be a good model for this. Or a Pitman-Yor process to keep the levels of the tree small. Within the model we want to be able to also have categories of products. In each categories are substitutes and sub-categories.

Some questions I will address in future posts is how to translate preferences into buying patterns for individual agents.

1. How many products will an agent buy given a budget constraint? (The ideal and quick approximation
2. Given relative preferences and actual price how much of each product should the agent buy? I recall chicago price theory has the derivations for this in the first two lectures.
3. This leads to a high dimensional convex optimization problem. How can we solve this efficiently?
4. Alternatively we can use importance sampling to generate a large collection of possible baskets. An approach similar to monte carlo tree search can be used to find the best basket. I like this direction as more complex models may not be so easy to solve empirically.

#| '!! shinylive warning !!': |
#|   shinylive does not work in self-contained HTML documents.
#|   Please set `embed-resources: false` in your metadata.
#| standalone: true
#| viewerHeight: 600
#| components: [viewer] #[editor, viewer]
_='''{.markdown} 
from __future__ import annotations
## Tasks:

'''

# app.py — ShinyLive-ready ABM Polya-Urn demo (Mesa-lite core) + Altair 


from dataclasses import dataclass
from typing import Dict, List, Tuple
import random
from collections import Counter

import pandas as pd
import altair as alt

from shiny import App, Inputs, Outputs, Session, reactive, render, ui
from shinywidgets import render_altair, output_widget

import mesa
from mesa import Model, Agent, space

### Two NonParametric Helper Models (Polya Urn and Hoppe Urn)

class PolyaUrn:
    """A Polya urn model for generating reinforcing preferences based on a Dirichlet process.
       
       Note: 
    
       - Since products are generated by a CRP process (Hoppe Urn), we need to support adding new products via a draw_new_product() method.

       - We may consider using Moran steps to model a drift from many products to a few products, i.e. we draw a ball replacing it with a ball of another color based on the urn's diriclet distribution.
       or we may drift according to some exogenous process. e.g. advertising or social influence. We may do this once or until
       we converge to some top-k products.


    Attributes:
        alpha (float): The reinforcement parameter.
        num_products (int): The number of products (colors).
        urn (Dict[int, int]): A dictionary representing the urn with product indices as keys and bead counts as values.

    """

    def __init__(self, alpha: int = 1, num_products: int = 1):
        self.alpha = alpha
        self.num_products = num_products
        self.urn = {i: 1 for i in range(num_products)}  # one bead per product

    def draw(self) -> int:
        """Draw a ball from the urn and reinforce."""
        total = sum(self.urn.values()) 
        probs = [self.urn[i] / total for i in range(self.num_products)]
        drawn = random.choices(list(self.urn.keys()), weights=probs, k=1)[0]
        self.urn[drawn] += 1  # Reinforce
        return drawn

    def draw_new_product(self) -> int:
        """Draw a new product (color) from the urn."""
        new_product_idx = self.num_products
        self.urn[new_product_idx] = 1  # Add new product with one bead
        self.num_products += 1
        return new_product_idx

class HoppeUrn:
    """A Hoppe urn model for generating new products.
       
       Note: 
       The current model assumes all products are independent.
       
       We will later extend this to support the addition of product categories and affinity between products, allowing us to model substitutes and complements.
       We may also want to assign products to Maslow's hierarchy of needs.

       Ideally, though, the structure of the product space should be easily accessible from this Model.
    """

    def __init__(self, initial_products: int = 2, innovation: float = 1.0):
        
        self.urn = {i: 1 for i in range(initial_products)}  # one bead per product
        self.inovation = innovation
        print(f"initial hoppe urn {self.urn}")

    def draw(self) -> int:
        """ Draw a ball from the urn and reinforce."""
        total = sum(self.urn.values())
        probs = [self.urn[i] / total for i in self.urn.keys()]
        drawn = random.choices(list(self.urn.keys()), weights=probs, k=1)[0]
        if drawn == 0:  # "new product" 
            new_product_idx = max(self.urn.keys()) + 1
            self.urn[new_product_idx] = 1  # Add new product with one bead
            #drawn = new_product_idx
        else:
            self.urn[drawn] += 1  # Reinforce
        return drawn

### Mesa Agent code - Dirichlet Process for preferences

class DemandAgent(mesa.Agent):
    """An agent with fixed initial wealth and a Polya urn for its preferences."""

    def __init__(self, model):
        # Pass the parameters to the parent class.
        super().__init__(model)

        # Create the agent's attribute and set the initial values.
        self.endowment = random.uniform(5, 15)  # Random initial wealth 
        self.wealth = self.endowment  
        self.urn = PolyaUrn()

    def say_hi(self):
        # The agent's step will go here.
        # For demonstration purposes we will print the agent's unique_id
        #print(f"Hi, I am an agent, you can call me {self.unique_id!s}.")
        if self.model.draw == 0:
            self.urn.draw_new_product()
        
        drawn = self.urn.draw()
        print(f"Agent {self.unique_id!s:2} drew: {drawn} | prefrences: {self.urn.urn.values()}")


### Server code

class DemandModel(mesa.Model):
    """A model with some number of agents."""

    def __init__(self, n,counter, innovation: float = 1.0, seed=None):
        super().__init__(seed=seed)
        self.num_agents = n
        self.counter=counter
        self.urn=HoppeUrn(innovation=innovation)
        # Create n agents
        DemandAgent.create_agents(model=self, n=n)
        self.ticks=0
        self.draw=None
        self.num_products_history = [len(self.urn.urn)]  # <-- track unique products over time

    def step(self):
        """ Advance the model by one step."""
        # This function pseudo-randomly reorders the list of agent objects and
        # then iterates through calling the function passed in as the parameter
        self.draw  = self.urn.draw()
        self.ticks +=1
        print(f"drew: {self.draw} | products : {self.urn.urn.values()}")
        self.agents.shuffle_do("say_hi")
        self.num_products_history.append(len(self.urn.urn))  # <-- record after each step

    
### Mesa entry points

### Shiny UI

### Step button



### Shiny APP

from shiny.express import input, render, ui


## UI
ui.input_slider("n", "Total Agants", 1, 100, 5)
ui.input_slider("theta", "Innovation", 0, 100, 1)


ui.input_action_button("btnStep", "Step")
ui.input_action_button("btnReset", "Reset")
ui.tags.br()

model_val = reactive.Value(None)
counter_val = reactive.Value(0)

def build_model():
    # Build a fresh model using the current UI and counter
    return DemandModel(n=input.n(), counter=counter_val.get(), innovation=input.theta()/100.0)

@reactive.calc
def current_model():
    return model_val.get()


@reactive.effect
@reactive.event(input.btnReset)
def reset_sim():
    print("resetting")
    # Bump a counter if you want to track resets
    counter_val.set(counter_val.get() + 1)
    # Create and store a fresh model
    model_val.set(build_model())
    starter_model=model_val.get()
    print(f"model version {starter_model.counter}")   
    # Do an initial step if desired

@reactive.effect 
@reactive.event(input.btnStep)
def sim_step(): 
    if model_val.get() is None:
        model_val.set(build_model())
    starter_model=model_val.get()
    print(f"model version {starter_model.counter}")   
    model_val.get().step()

# --- helpers ---------------------------------------------------------------
def prefs_df(model) -> pd.DataFrame:
    """Rows: agent, product, count."""
    if model is None:
        return pd.DataFrame(columns=["agent", "product", "count"])
    rows = []
    for ag in model.agents:
        for prod, cnt in ag.urn.urn.items():
            rows.append({"agent": str(ag.unique_id), "product": str(prod), "count": cnt})
    return pd.DataFrame(rows)

# --- UI slot ---------------------------------------------------------------
ui.h3("Agent preference shares")
#output_widget("pref_chart")

# --- Server render ---------------------------------------------------------

@render_altair
@reactive.event(input.btnStep)
def pref_chart():
    m = current_model()
    df = prefs_df(m)
    if df.empty:
        return alt.Chart(pd.DataFrame({"x": []})).mark_text().encode()  # harmless placeholder

    # normalized, stacked bars like the barley example
    return (
        alt.Chart(df)
        .mark_bar()
        .encode(
            x=alt.X("sum(count):Q", stack="normalize", title="Preference share"),
            y=alt.Y("agent:N", sort="-x", title="Agent"),
            color=alt.Color("product:N", title="Product"),
            tooltip=["agent:N", "product:N", "count:Q"]
        )
        .properties(height=400)
    )


# --- helpers ---------------------------------------------------------------
import numpy as np

def kn_live_df(model) -> pd.DataFrame:
    if model is None or not getattr(model, "num_products_history", None):
        return pd.DataFrame(columns=["t", "K"])
    hist = model.num_products_history
    return pd.DataFrame({"t": np.arange(len(hist)), "K": hist})

def kn_theory_df(nmax: int, alphas=(0.1, 0.3, 1.0, 3.0, 10.0)) -> pd.DataFrame:
    t = np.arange(1, max(2, nmax) + 1)
    parts = []
    for a in alphas:
        parts.append(pd.DataFrame({"t": t, "alpha": str(a), "E[K_t]": a * np.log(t)}))
    return pd.concat(parts, ignore_index=True)

# --- UI slot ---------------------------------------------------------------
ui.h3("Unique products over time: theory vs. live")
#output_widget("kn_chart")

# --- Server render ---------------------------------------------------------
@render_altair
@reactive.event(input.btnStep)
def kn_chart():
    m = current_model()
    nmax = (m.ticks if m else 200) or 200

    df_theory = kn_theory_df(nmax)
    base = (
        alt.Chart(df_theory)
        .mark_line()
        .encode(
            x=alt.X("t:Q", title="Steps (t)"),
            y=alt.Y("E[K_t]:Q", title="Unique products K_t"),
            color=alt.Color("alpha:N", title="α (innovation)")
        )
    )

    df_live = kn_live_df(m)
    live_layer = (
        alt.Chart(df_live)
        .mark_line(point=True)
        .encode(
            x="t:Q",
            y="K:Q",
            tooltip=["t:Q", "K:Q"]
        )
        .properties(title="Live K_t")
    )

    return (base + live_layer).properties(height=350)


## file: requirements.txt
altair
anywidget
jsonschema
mesa