Hawk/dove updating risk attitudes

simulatingrisk/about_app.md

@@ -1,2 +1,2 @@
 
-These simulatiions are associated with the CDH project [Simulating risk, risking simulations](https://cdh.princeton.edu/projects/simulating-risk/).
+These simulations are associated with the CDH project [Simulating risk, risking simulations](https://cdh.princeton.edu/projects/simulating-risk/).

simulatingrisk/app.py

@@ -1,3 +1,5 @@
+import os.path
+
 import solara
 
 from simulatingrisk.hawkdove.app import page as hawkdove_page
@@ -8,7 +10,8 @@
 
 @solara.component
 def Home():
-    with open("simulatingrisk/about_app.md") as readmefile:
+    # load about markdown file in the same directory
+    with open(os.path.join(os.path.dirname(__file__), "about_app.md")) as readmefile:
         return solara.Markdown("\n".join(readmefile.readlines()))
 
 

simulatingrisk/charts/histogram.py

@@ -61,16 +61,12 @@ def plot_risk_histogram(model):
     # adapted from mesa visualiation tutorial
     # https://mesa.readthedocs.io/en/stable/tutorials/visualization_tutorial.html#Building-your-own-visualization-component
 
-    # Note: you must initialize a figure using this method instead of
+    # Note: per Mesa docs, has to be initialized using this method instead of
     # plt.figure(), for thread safety purpose
     fig = Figure()
     ax = fig.subplots()
     # generate a histogram of current risk levels
     risk_levels = [agent.risk_level for agent in model.schedule.agents]
-    # Note: you have to use Matplotlib's OOP API instead of plt.hist
-    # because plt.hist is not thread-safe.
     ax.hist(risk_levels, bins=risk_bins)
     ax.set_title("risk levels")
-    # You have to specify the dependencies as follows, so that the figure
-    # auto-updates when viz.model or viz.df is changed.
-    solara.FigureMatplotlib(fig, dependencies=[model])
+    solara.FigureMatplotlib(fig)

simulatingrisk/hawkdove/model.py

@@ -50,7 +50,10 @@ def set_risk_level(self):
         raise NotImplementedError
 
     def __repr__(self):
-        return f"<{self.__class__.__name__} id={self.unique_id} r={self.risk_level}>"
+        return (
+            f"<{self.__class__.__name__} id={self.unique_id} "
+            + f"r={self.risk_level} points={self.points}>"
+        )
 
     def initial_choice(self, hawk_odds=None):
         # first round : choose what to play randomly or based on initial hawk odds
@@ -129,7 +132,18 @@ def points_rank(self):
 
 
 class HawkDoveModel(mesa.Model):
-    """ """
+    """
+    Model for hawk/dove game with risk attitudes.
+
+    :param grid_size: number for square grid size (creates n*n agents)
+    :param include_diagonals: whether agents should include diagonals
+        or not when considering neighbors (default: True)
+    :param hawk_odds: odds for playing hawk on the first round (default: 0.5)
+    :param risk_adjustment: strategy agents should use for adjusting risk;
+        None (default), adopt, or average
+    :param adjust_every: when risk adjustment is enabled, adjust every
+        N rounds (default: 10)
+    """
 
     #: whether the simulation is running
     running = True  # required for batch run

simulatingrisk/hawkdove/server.py

@@ -57,8 +57,8 @@ def agent_portrayal(agent):
     "grid_size": grid_size,
 }
 
-
-jupyterviz_params = {
+# parameters common to both hawk/dove variants
+common_jupyterviz_params = {
     "grid_size": {
         "type": "SliderInt",
         "value": grid_size,
@@ -72,13 +72,6 @@ def agent_portrayal(agent):
         "value": True,
         "label": "Include diagonal neighbors",
     },
-    "agent_risk_level": {
-        "type": "SliderInt",
-        "min": 0,
-        "max": 8,
-        "step": 1,
-        "value": 2,
-    },
     "hawk_odds": {
         "type": "SliderFloat",
         "value": 0.5,
@@ -89,6 +82,16 @@ def agent_portrayal(agent):
     },
 }
 
+# in single-risk variant, risk level is set for all agents at init time
+jupyterviz_params = common_jupyterviz_params.copy()
+jupyterviz_params["agent_risk_level"] = {
+    "type": "SliderInt",
+    "min": 0,
+    "max": 8,
+    "step": 1,
+    "value": 2,
+}
+
 
 def draw_hawkdove_agent_space(model, agent_portrayal):
     # custom agent space chart, modeled on default

simulatingrisk/hawkdovevar/app.py

@@ -5,13 +5,32 @@
 from simulatingrisk.hawkdovevar.model import HawkDoveVariableRiskModel
 from simulatingrisk.hawkdove.server import (
     agent_portrayal,
-    jupyterviz_params,
+    common_jupyterviz_params,
     draw_hawkdove_agent_space,
 )
 from simulatingrisk.hawkdove.app import plot_hawks
 
-jupyterviz_params_var = jupyterviz_params.copy()
-del jupyterviz_params_var["agent_risk_level"]
+# start with common hawk/dove params, then add params for variable risk
+jupyterviz_params_var = common_jupyterviz_params.copy()
+jupyterviz_params_var.update(
+    {
+        "risk_adjustment": {
+            "type": "Select",
+            "value": "adopt",
+            "values": ["none", "adopt", "average"],
+            "description": "If and how agents update their risk level",
+        },
+        "adjust_every": {
+            "label": "Adjustment frequency (# rounds)",
+            "type": "SliderInt",
+            "min": 1,
+            "max": 30,
+            "step": 1,
+            "value": 10,
+            "description": "How many rounds between risk adjustment",
+        },
+    }
+)
 
 page = JupyterViz(
     HawkDoveVariableRiskModel,

simulatingrisk/hawkdovevar/model.py

@@ -1,9 +1,13 @@
+import statistics
+
 from simulatingrisk.hawkdove.model import HawkDoveModel, HawkDoveAgent
 
 
 class HawkDoveVariableRiskAgent(HawkDoveAgent):
     """
-    An agent with random risk attitude playing Hawk or Dove
+    An agent with random risk attitude playing Hawk or Dove. Optionally
+    adjusts risks based on most successful neighbor, depending on model
+    configuration.
     """
 
     def set_risk_level(self):
@@ -13,18 +17,94 @@ def set_risk_level(self):
         # generate a random risk level
         self.risk_level = self.random.randint(0, num_neighbors)
 
+    def play(self):
+        super().play()
+        # when enabled by the model, periodically adjust risk level
+        if self.model.adjustment_round:
+            self.adjust_risk()
+
+    @property
+    def most_successful_neighbor(self):
+        """identify and return the neighbor with the most points"""
+        # sort neighbors by points, highest points first
+        # adapted from risky bet wealthiest neighbor
+        return sorted(self.neighbors, key=lambda x: x.points, reverse=True)[0]
+
+    def adjust_risk(self):
+        # look at neighbors
+        # if anyone has more points
+        # either adopt their risk attitude or average theirs with yours
+
+        best = self.most_successful_neighbor
+        # if most successful neighbor has more points and a different
+        # risk attitude, adjust
+        if best.points > self.points and best.risk_level != self.risk_level:
+            # adjust risk based on model configuration
+            if self.model.risk_adjustment == "adopt":
+                # adopt neighbor's risk level
+                self.risk_level = best.risk_level
+            elif self.model.risk_adjustment == "average":
+                # average theirs with mine, then round to a whole number
+                # since this model uses discrete risk levels
+                self.risk_level = round(
+                    statistics.mean([self.risk_level, best.risk_level])
+                )
+
 
 class HawkDoveVariableRiskModel(HawkDoveModel):
+    """
+    Model for hawk/dove game with variable risk attitudes.
+
+    :param grid_size: number for square grid size (creates n*n agents)
+    :param include_diagonals: whether agents should include diagonals
+        or not when considering neighbors (default: True)
+    :param hawk_odds: odds for playing hawk on the first round (default: 0.5)
+    :param risk_adjustment: strategy agents should use for adjusting risk;
+        None (default), adopt, or average
+    :param adjust_every: when risk adjustment is enabled, adjust every
+        N rounds (default: 10)
+    """
+
     risk_attitudes = "variable"
     agent_class = HawkDoveVariableRiskAgent
 
+    supported_risk_adjustments = (None, "adopt", "average")
+
     def __init__(
         self,
         grid_size,
         include_diagonals=True,
         hawk_odds=0.5,
+        risk_adjustment=None,
+        adjust_every=10,
     ):
         super().__init__(
             grid_size, include_diagonals=include_diagonals, hawk_odds=hawk_odds
         )
-        # no custom logic or params yet, but will be adding risk updating logic
+        # convert string input from solara app parameters to None
+        if risk_adjustment == "none":
+            risk_adjustment = None
+        # make sure risk adjustment is valid
+        if risk_adjustment not in self.supported_risk_adjustments:
+            risk_adjust_opts = ", ".join(
+                [opt or "none" for opt in self.supported_risk_adjustments]
+            )
+            raise ValueError(
+                f"Unsupported risk adjustment '{risk_adjustment}'; "
+                + f"must be one of {risk_adjust_opts}"
+            )
+
+        self.risk_adjustment = risk_adjustment
+        self.adjust_round_n = adjust_every
+
+    @property
+    def adjustment_round(self) -> bool:
+        """is the current round an adjustment round?"""
+        # check if the current step is an adjustment round
+        # when risk adjustment is enabled, agents should adjust their risk
+        # strategy every N rounds;
+        return (
+            self.risk_adjustment
+            and self.schedule.steps > 0
+            and self.schedule.steps % self.adjust_round_n == 0
+        )

tests/test_hawkdove.py

@@ -10,7 +10,6 @@
     HawkDoveSingleRiskModel,
     HawkDoveSingleRiskAgent,
 )
-from simulatingrisk.hawkdovevar.model import HawkDoveVariableRiskModel
 
 
 def test_agent_neighbors():
@@ -67,7 +66,10 @@ def test_agent_repr():
     agent_id = 1
     risk_level = 3
     agent = HawkDoveSingleRiskAgent(agent_id, Mock(agent_risk_level=risk_level))
-    assert repr(agent) == f"<HawkDoveSingleRiskAgent id={agent_id} r={risk_level}>"
+    assert (
+        repr(agent)
+        == f"<HawkDoveSingleRiskAgent id={agent_id} r={risk_level} points=0>"
+    )
 
 
 def test_model_single_risk_level():
@@ -87,16 +89,6 @@ def test_model_single_risk_level():
         assert agent.risk_level == risk_level
 
 
-def test_variable_risk_level():
-    model = HawkDoveVariableRiskModel(
-        5,
-        include_diagonals=True,
-    )
-    # when risk level is variable/random, agents should have different risk levels
-    risk_levels = set([agent.risk_level for agent in model.schedule.agents])
-    assert len(risk_levels) > 1
-
-
 def test_num_dove_neighbors():
     # initialize an agent with a mock model
     agent = HawkDoveSingleRiskAgent(1, Mock(agent_risk_level=2))