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,14 +1,17 @@
+import os.path
+
 import solara
 
 from simulatingrisk.hawkdove.app import page as hawkdove_page
 from simulatingrisk.hawkdovevar.app import page as hawkdove_var_page
 from simulatingrisk.risky_bet.app import page as riskybet_page
 from simulatingrisk.risky_food.app import page as riskyfood_page
 
 
 @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()))
 
 
@@ -17,9 +20,9 @@
     return hawkdove_page
 
 
 @solara.component
 def hawkdove_var():
     return hawkdove_var_page
 
 
 @solara.component

simulatingrisk/charts/histogram.py

@@ -73,4 +73,4 @@ def plot_risk_histogram(model):
     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/hawkdovevar/app.py

@@ -10,8 +10,30 @@
 )
 from simulatingrisk.hawkdove.app import plot_hawks
 
+# adjust single-risk params for variable risk
 jupyterviz_params_var = jupyterviz_params.copy()
+# remove parameter for agent risk level;
+# add parameters for adaptive risk strategies
 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": {
+            "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))

tests/test_hawkdovevar.py

@@ -0,0 +1,150 @@
+import statistics
+from unittest.mock import patch, Mock
+
+import pytest
+
+from simulatingrisk.hawkdovevar.model import (
+    HawkDoveVariableRiskModel,
+    HawkDoveVariableRiskAgent,
+)
+
+
+def test_init():
+    model = HawkDoveVariableRiskModel(5)
+    # defaults
+    assert model.risk_adjustment is None
+    assert model.hawk_odds == 0.5
+    assert model.include_diagonals is True
+    # unused but should be set to default
+    assert model.adjust_round_n == 10
+
+    # init with risk adjustment
+    model = HawkDoveVariableRiskModel(
+        5,
+        include_diagonals=False,
+        hawk_odds=0.2,
+        risk_adjustment="adopt",
+        adjust_every=5,
+    )
+
+    assert model.risk_adjustment == "adopt"
+    assert model.adjust_round_n == 5
+    assert model.hawk_odds == 0.2
+    assert model.include_diagonals is False
+
+    # handle string none for solara app parameters
+    model = HawkDoveVariableRiskModel(5, risk_adjustment="none")
+    assert model.risk_adjustment is None
+
+    # complain about invalid adjustment type
+    with pytest.raises(ValueError, match="Unsupported risk adjustment 'bogus'"):
+        HawkDoveVariableRiskModel(3, risk_adjustment="bogus")
+
+
+def test_init_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
+
+
+adjustment_testdata = [
+    # init parameters, expected adjustment round
+    ({"risk_adjustment": None}, None),
+    ({"risk_adjustment": "adopt"}, 10),
+    ({"risk_adjustment": "average"}, 10),
+    ({"risk_adjustment": "average", "adjust_every": 3}, 3),
+]
+
+
+@pytest.mark.parametrize("params,expect_adjust_step", adjustment_testdata)
+def test_adjustment_round(params, expect_adjust_step):
+    model = HawkDoveVariableRiskModel(3, **params)
+
+    run_for = (expect_adjust_step or 10) + 1
+
+    # step through the model enough rounds to encounter one adjustment rounds
+    # if adjustment is enabled; start at 1 (step count starts at 1)
+    for i in range(1, run_for):
+        model.step()
+        if i == expect_adjust_step:
+            assert model.adjustment_round
+        else:
+            assert not model.adjustment_round
+
+
+def test_most_successful_neighbor():
+    # initialize an agent with a mock model
+    agent = HawkDoveVariableRiskAgent(1, Mock(), 1000)
+    mock_neighbors = [
+        Mock(points=2),
+        Mock(points=4),
+        Mock(points=23),
+        Mock(points=31),
+    ]
+
+    with patch.object(HawkDoveVariableRiskAgent, "neighbors", mock_neighbors):
+        assert agent.most_successful_neighbor.points == 31
+
+
+def test_agent_play_adjust():
+    mock_model = Mock(risk_adjustment="adopt")
+    agent = HawkDoveVariableRiskAgent(1, mock_model)
+    # simulate no neighbors to skip payoff calculation
+    with patch.object(
+        HawkDoveVariableRiskAgent, "neighbors", new=[]
+    ) as mock_adjust_risk:
+        with patch.object(HawkDoveVariableRiskAgent, "adjust_risk") as mock_adjust_risk:
+            # when it is not an adjustment round, should not call adjust risk
+            mock_model.adjustment_round = False
+            agent.play()
+            assert mock_adjust_risk.call_count == 0
+
+            # should call adjust risk when the model indicates
+            mock_model.adjustment_round = True
+            agent.play()
+            assert mock_adjust_risk.call_count == 1
+
+
+def test_adjust_risk_adopt():
+    # initialize an agent with a mock model
+    agent = HawkDoveVariableRiskAgent(1, Mock(risk_adjustment="adopt"))
+    # set a known risk level
+    agent.risk_level = 2
+    # adjust wealth as if the model had run
+    agent.points = 20
+    # set a mock neighbor with more points than current agent
+    neighbor = Mock(points=1500, risk_level=3)
+    with patch.object(HawkDoveVariableRiskAgent, "most_successful_neighbor", neighbor):
+        agent.adjust_risk()
+        # default behavior is to adopt successful risk level
+        assert agent.risk_level == neighbor.risk_level
+
+        # now simulate a wealthiest neighbor with fewer points than current agent
+        neighbor.points = 12
+        neighbor.risk_level = 3
+        prev_risk_level = agent.risk_level
+        agent.adjust_risk()
+        # risk level should not be changed
+        assert agent.risk_level == prev_risk_level
+
+
+def test_adjust_risk_average():
+    # same as previous test, but with average risk adjustment strategy
+    agent = HawkDoveVariableRiskAgent(1, Mock(risk_adjustment="average"))
+    # set a known risk level
+    agent.risk_level = 2
+    # adjust points  as if the model had run
+    agent.points = 300
+    # set a neighbor with more points than current agent
+    neighbor = Mock(points=350, risk_level=3)
+    with patch.object(HawkDoveVariableRiskAgent, "most_successful_neighbor", neighbor):
+        prev_risk_level = agent.risk_level
+        agent.adjust_risk()
+        # new risk level should be average of previous and most successful
+        assert agent.risk_level == round(
+            statistics.mean([neighbor.risk_level, prev_risk_level])
+        )