Reduce rapid test demand from vaccinated and when incidences are low

src/create_initial_states/task_build_full_params.py

@@ -103,6 +103,15 @@ def task_create_full_params(depends_on, produces):
     params = _add_educ_rapid_test_fade_in_params(params)
     params = _add_private_rapid_test_demand_fade_in_params(params)
     params = _add_rapid_test_reaction_params(params)
+
+    # these are arbitrary and will have to be estimated.
+    params.loc[
+        ("rapid_test_demand", "low_incidence_factor", "other_demand"), "value"
+    ] = 0.25
+    params.loc[
+        ("rapid_test_demand", "low_incidence_factor", "worker_demand"), "value"
+    ] = 0.25
+
     params = _add_event_params(params)
 
     # seasonality parameter

src/testing/rapid_tests.py

@@ -32,6 +32,8 @@ def rapid_test_demand(
     If randomize is True the calculated demand is distributed randomly in the entire
     population (excluding a share of refusers).
 
+    After Easter vaccinated individuals do not perform rapid tests.
+
     """
     date = get_date(states)
 
@@ -49,6 +51,12 @@ def rapid_test_demand(
         educ_workers_params = params.loc[("rapid_test_demand", "educ_worker_shares")]
         students_params = params.loc[("rapid_test_demand", "student_shares")]
         private_demand_params = params.loc[("rapid_test_demand", "private_demand")]
+        other_low_incidence_factor = params.loc[
+            ("rapid_test_demand", "low_incidence_factor", "other_demand"), "value"
+        ]
+        work_low_incidence_factor = params.loc[
+            ("rapid_test_demand", "low_incidence_factor", "worker_demand"), "value"
+        ]
 
     # get work demand inputs
     share_of_workers_with_offer = get_piecewise_linear_interpolation_for_one_day(
@@ -83,6 +91,7 @@ def rapid_test_demand(
         states=states,
         contacts=contacts,
         compliance_multiplier=work_compliance_multiplier,
+        low_incidence_factor=work_low_incidence_factor,
     )
 
     educ_demand = _calculate_educ_rapid_test_demand(
@@ -102,11 +111,25 @@ def rapid_test_demand(
     )
 
     other_contact_demand = _calculate_other_meeting_rapid_test_demand(
-        states=states, contacts=contacts, demand_share=private_demand_share
+        states=states,
+        contacts=contacts,
+        demand_share=private_demand_share,
+        low_incidence_factor=other_low_incidence_factor,
     )
 
     private_demand = hh_demand | sym_without_pcr_demand | other_contact_demand
     rapid_test_demand = work_demand | educ_demand | private_demand
+    preemptive_rapid_test_demand = work_demand | educ_demand | other_contact_demand
+
+    # vaccinated individuals do not test themselves for work, educ or leisure contacts
+    if date > pd.Timestamp("2021-04-05"):
+        preemptive_rapid_test_demand = _only_not_fully_vaccinated_test_themselves(
+            preemptive_rapid_test_demand, states
+        )
+
+    rapid_test_demand = (
+        preemptive_rapid_test_demand | hh_demand | sym_without_pcr_demand
+    )
 
     if randomize and date > pd.Timestamp("2021-04-05"):  # only randomize after Easter
         assert (
@@ -204,8 +227,12 @@ def _get_student_demand(eligible, states, student_multiplier):
     return student_demand
 
 
-def _calculate_work_rapid_test_demand(states, contacts, compliance_multiplier):
+def _calculate_work_rapid_test_demand(
+    states, contacts, compliance_multiplier, low_incidence_factor
+):
     date = get_date(states)
+    weekly_cases_per_100_000 = 7 * 100_000 * states["new_known_case"].mean()
+
     work_cols = [col for col in contacts if col.startswith("work_")]
     has_work_contacts = (contacts[work_cols] > 0).any(axis=1)
 
@@ -229,8 +256,17 @@ def _calculate_work_rapid_test_demand(states, contacts, compliance_multiplier):
 
     should_get_test = has_work_contacts & too_long_since_last_test
     complier = states["rapid_test_compliance"] >= (1 - compliance_multiplier)
-    receives_offer_and_accepts = should_get_test & complier
-    work_rapid_test_demand = should_get_test & receives_offer_and_accepts
+    work_rapid_test_demand = should_get_test & complier
+
+    # assume that people become more negligent when the incidences are low
+    if date > pd.Timestamp("2021-06-01") and weekly_cases_per_100_000 < 35:
+        non_negligent = np.random.choice(
+            [True, False],
+            len(work_rapid_test_demand),
+            p=[low_incidence_factor, 1 - low_incidence_factor],
+        )
+        work_rapid_test_demand = work_rapid_test_demand & non_negligent
+
     return work_rapid_test_demand
 
 
@@ -271,10 +307,11 @@ def _calculate_own_symptom_rapid_test_demand(states, demand_share):
     return own_symptom_demand
 
 
-def _calculate_other_meeting_rapid_test_demand(states, contacts, demand_share):
-    scaling_factor = 1.0
-    demand_share = scaling_factor * demand_share
-
+def _calculate_other_meeting_rapid_test_demand(
+    states, contacts, demand_share, low_incidence_factor
+):
+    weekly_cases_per_100_000 = 7 * 100_000 * states["new_known_case"].mean()
+    date = get_date(states)
     complier = states["quarantine_compliance"] >= (1 - demand_share)
     not_tested_recently = states["cd_received_rapid_test"] < -3
 
@@ -284,6 +321,16 @@ def _calculate_other_meeting_rapid_test_demand(states, contacts, demand_share):
     with_relevant_contact = (contacts[weekly_other_cols] > 0).any(axis=1)
 
     to_be_tested = complier & not_tested_recently & with_relevant_contact
+
+    # when incidences are low there are less testing requirements and
+    # testing is perceived to have a lower benefit
+    if date < pd.Timestamp("2021-06-01") or weekly_cases_per_100_000 > 35:
+        tests_despite_low_incidence = np.random.choice(
+            [True, False],
+            len(to_be_tested),
+            p=[low_incidence_factor, 1 - low_incidence_factor],
+        )
+        to_be_tested = to_be_tested & tests_despite_low_incidence
     return to_be_tested
 
 
@@ -325,3 +372,21 @@ def _randomize_rapid_tests(states, target_share_to_be_tested, share_refuser, see
     rapid_test_demand = pd.Series(False, index=states.index)
     rapid_test_demand[to_test_indices] = True
     return rapid_test_demand
+
+
+def _only_not_fully_vaccinated_test_themselves(rapid_test_demand, states):
+    """Exclude fully vaccinated individuals from being tested.
+
+    The immunity countdown is initialized at -9999. The vaccine countdown is set
+    between -1 and 21. This is only the 1st vaccine. Assuming 30 days between
+    shots and 14 days of wait period after the 2nd shot, individuals are freed
+    from testing obligations ~45 days after their first shot. This translates
+    into countdown values between -45 and -20. For simplicity we simply assume
+    that individuals stop testing themselves 40 days after their first shot.
+
+    """
+    more_than_14d_since_vaccination = states["cd_is_immune_by_vaccine"].between(
+        -9990, -40
+    )
+    lowered_test_demand = rapid_test_demand & ~more_than_14d_since_vaccination
+    return lowered_test_demand

src/testing/task_get_and_plot_share_of_tests_for_symptomatics.py

@@ -55,7 +55,7 @@
     }
 )
 def task_prepare_characteristics_of_the_tested(depends_on, produces):
-    df = pd.read_excel(depends_on["data"], sheet_name="Klinische_Aspekte", header=1)
+    df = pd.read_excel(depends_on["data"], sheet_name="Klinische_Aspekte", header=2)
 
     df = _clean_data(df)
     df = convert_weekly_to_daily(df.reset_index(), divide_by_7_cols=[])

tests/test_rapid_tests.py

@@ -8,6 +8,7 @@
 from src.testing.rapid_tests import _calculate_work_rapid_test_demand
 from src.testing.rapid_tests import _determine_if_hh_had_event
 from src.testing.rapid_tests import _get_eligible_educ_participants
+from src.testing.rapid_tests import _only_not_fully_vaccinated_test_themselves
 from src.testing.rapid_tests import _randomize_rapid_tests
 
 
@@ -25,8 +26,9 @@ def educ_states():
         "school",  # 5: student to be tested
     ]
     states["cd_received_rapid_test"] = [-2, -5, -20, -5, -20, -5]
+    states["cd_is_immune_by_vaccine"] = -10_000
     states["rapid_test_compliance"] = 1.0
-    states["date"] = pd.Timestamp("2021-05-05")
+    states["date"] = pd.Timestamp("2021-07-05")
     return states
 
 
@@ -80,6 +82,7 @@ def work_states():
     # 6: yes if compliance_multiplier >= 0.3
     states["cd_received_rapid_test"] = [-1, -10, -10, -10, -5, -10]
     states["rapid_test_compliance"] = [0.8, 0.8, 0.8, 0.8, 0.8, 0.3]
+    states["new_known_case"] = False
     return states
 
 
@@ -112,7 +115,10 @@ def test_calculate_work_rapid_test_demand_early(work_states, work_contacts):
     work_states["date"] = pd.Timestamp("2021-04-01")
     expected = pd.Series([False, False, True, True, False, True])
     res = _calculate_work_rapid_test_demand(
-        work_states, work_contacts, compliance_multiplier=1.0  # perfect compliance
+        work_states,
+        work_contacts,
+        compliance_multiplier=1.0,  # perfect compliance
+        low_incidence_factor=1.0,
     )
     pd.testing.assert_series_equal(res, expected)
 
@@ -121,7 +127,10 @@ def test_calculate_work_rapid_test_demand_late(work_states, work_contacts):
     work_states["date"] = pd.Timestamp("2021-05-05")
     expected = pd.Series([False, False, True, True, True, False])
     res = _calculate_work_rapid_test_demand(
-        work_states, work_contacts, compliance_multiplier=0.3
+        work_states,
+        work_contacts,
+        compliance_multiplier=0.3,
+        low_incidence_factor=1.0,
     )
     pd.testing.assert_series_equal(res, expected)
 
@@ -130,7 +139,10 @@ def test_calculate_work_rapid_test_demand_no_compliance(work_states, work_contac
     work_states["date"] = pd.Timestamp("2021-05-05")
     expected = pd.Series([False, False, False, False, False, False])
     res = _calculate_work_rapid_test_demand(
-        work_states, work_contacts, compliance_multiplier=0.0
+        work_states,
+        work_contacts,
+        compliance_multiplier=0.0,
+        low_incidence_factor=1.0,
     )
     pd.testing.assert_series_equal(res, expected)
 
@@ -141,7 +153,10 @@ def test_calculate_work_rapid_test_demand_imperfect_compliance(
     work_states["date"] = pd.Timestamp("2021-05-05")
     expected = pd.Series([False, False, True, True, True, False])
     res = _calculate_work_rapid_test_demand(
-        work_states, work_contacts, compliance_multiplier=0.5
+        work_states,
+        work_contacts,
+        compliance_multiplier=0.5,
+        low_incidence_factor=1.0,
     )
     pd.testing.assert_series_equal(res, expected)
 
@@ -233,14 +248,19 @@ def test_calculate_other_meeting_rapid_test_demand():
     states = pd.DataFrame()
     states["quarantine_compliance"] = [0.2, 0.8, 0.8, 0.8]
     states["cd_received_rapid_test"] = [-99, -2, -99, -99]
+    states["new_known_case"] = [False, False, False, False]
+    states["date"] = pd.Timestamp("2021-04-15")
 
     contacts = pd.DataFrame()
     contacts["other_recurrent_weekly_1"] = [3, 3, 0, 3]
     contacts["other_non_recurrent"] = 2
     demand_share = 0.3
 
     res = _calculate_other_meeting_rapid_test_demand(
-        states=states, contacts=contacts, demand_share=demand_share
+        states=states,
+        contacts=contacts,
+        demand_share=demand_share,
+        low_incidence_factor=1.0,
     )
 
     # 0: non-complier, 1: recently tested, 2: no relevant contact, 3: test
@@ -276,3 +296,21 @@ def test_random_rapid_test_demand_lln():
     )
     assert not res[states["rapid_test_compliance"] < 0.15].any()
     assert res.mean() == pytest.approx(0.6, abs=0.001)
+
+
+def test_exclude_vaccinated_from_being_tested():
+    states = pd.Series(
+        [-10_030, -50, -3, 5] * 2, name="cd_is_immune_by_vaccine"
+    ).to_frame()
+    rapid_test_demand = pd.Series([True] * 4 + [False] * 4)
+    res = _only_not_fully_vaccinated_test_themselves(rapid_test_demand, states)
+    expected = pd.Series(
+        [
+            True,  # never vaccinated
+            False,  # long enough since vaccination
+            True,  # not long enough since vaccination
+            True,  # not long enough since vaccination
+        ]
+        + [False] * 4
+    )
+    pd.testing.assert_series_equal(res, expected, check_names=False)