Preliminary work for hawk/dove adaptive risk attitudes #20

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
@@ -203,6 +217,18 @@ def step(self):
                 + f"Final rolling average % hawk: {self.rolling_percent_hawk}"
             )
 
+    @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
+        )
+
     @property
     def max_agent_points(self):
         # what is the current largest point total of any agent?

simulatingrisk/hawkdovevar/app.py

@@ -12,6 +12,24 @@
 
 jupyterviz_params_var = jupyterviz_params.copy()
 del jupyterviz_params_var["agent_risk_level"]
+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": {
+            "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,3 +1,5 @@
+import statistics
+
 from simulatingrisk.hawkdove.model import HawkDoveModel, HawkDoveAgent
 
 
@@ -13,8 +15,53 @@ 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
+        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
 
@@ -23,8 +70,27 @@ def __init__(
         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
+        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

@@ -67,7 +67,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():