nhra_gt.subgames.queuing

Patient Queuing Game and Wardrop Equilibrium Solver.

This module models the choice patients make between attending an Emergency Department (ED) or a General Practitioner (GP), based on expected wait times and out-of-pocket costs. It uses fixed-point iteration to find the equilibrium demand levels.

Classes

PatientUtilityParams dataclass

Parameters defining patient choice utility.

Attributes:

Name	Type	Description
gp_out_of_pocket	float	Financial cost of GP visit ($).
gp_wait_time_min	float	Expected wait time for GP (minutes).
patient_time_value_hour	float	Shadow price of patient time ($/hr).
ed_base_utility	float	Intrinsic utility of ED (e.g. equipment access).
logit_sensitivity	float	Rationality parameter for logit choice.
ed_outside_utility	float	Utility of not seeking care.

Source code in src/nhra_gt/subgames/queuing.py

@dataclass(frozen=True)
class PatientUtilityParams:
    """
    Parameters defining patient choice utility.

    Attributes:
        gp_out_of_pocket: Financial cost of GP visit ($).
        gp_wait_time_min: Expected wait time for GP (minutes).
        patient_time_value_hour: Shadow price of patient time ($/hr).
        ed_base_utility: Intrinsic utility of ED (e.g. equipment access).
        logit_sensitivity: Rationality parameter for logit choice.
        ed_outside_utility: Utility of not seeking care.
    """

    gp_out_of_pocket: float = 40.0
    gp_wait_time_min: float = 15.0
    patient_time_value_hour: float = 25.0
    ed_base_utility: float = 0.0  # ED is "free" but has other disutilities (travel, etc.)
    logit_sensitivity: float = 0.1
    ed_outside_utility: float = -100.0
    queuing_init_prob: float = 0.5

Functions

beartype(fn)

Conditional beartype decorator.

Source code in src/nhra_gt/subgames/queuing.py

def beartype(fn):  # type: ignore[no-untyped-def]
    """Conditional beartype decorator."""
    if _beartype is None:
        return fn
    return _beartype(fn)

calculate_patient_utilities(ed_wait_min, p, p_global)

Calculates utilities for choosing ED vs GP.

Returns:

Type	Description
tuple[Any, Any]	Tuple of (utility_ed, utility_gp).

Source code in src/nhra_gt/subgames/queuing.py

@beartype
def calculate_patient_utilities(
    ed_wait_min: Any, p: PatientUtilityParams, p_global: Any
) -> tuple[Any, Any]:
    """
    Calculates utilities for choosing ED vs GP.

    Returns:
        Tuple of (utility_ed, utility_gp).
    """
    u_ed = p.ed_base_utility - (
        ed_wait_min / p_global.ops.minutes_per_hour * p.patient_time_value_hour
    )
    u_gp = jnp.array(
        -(p.gp_wait_time_min / p_global.ops.minutes_per_hour * p.patient_time_value_hour)
        - p.gp_out_of_pocket
    )
    return u_ed, u_gp

solve_queuing_equilibrium_jax(total_base_demand, capacity, discharge_delay, params, p_global, max_iter=5)

Finds the Wardrop Equilibrium for patient demand using fixed-point iteration. Returns (demand_ed, prob_ed).

Source code in src/nhra_gt/subgames/queuing.py

@beartype
def solve_queuing_equilibrium_jax(
    total_base_demand: Any,
    capacity: Any,
    discharge_delay: Any,
    params: PatientUtilityParams,
    p_global: Any,
    max_iter: int = 5,
) -> tuple[Any, Any]:
    """
    Finds the Wardrop Equilibrium for patient demand using fixed-point iteration.
    Returns (demand_ed, prob_ed).
    """
    if jax is None:  # pragma: no cover
        raise ImportError("`solve_queuing_equilibrium_jax` requires `jax` to be installed.")
    from jax import lax

    from nhra_gt.engine import mm_s_queue_wait_jax

    def body_fun(i, state):
        d_curr, _ = state
        # 1. Calculate resulting wait time at current demand
        wait_min = mm_s_queue_wait_jax(
            d_curr,
            1.0 / jnp.maximum(1e-9, discharge_delay),
            jnp.array(capacity * p_global.ops.capacity_scalar),
            p_global,
        )

        # 2. Calculate utilities
        u_ed, u_gp = calculate_patient_utilities(wait_min, params, p_global)

        # 3. Logit choice (ED vs GP vs Outside Option)
        u_outside = params.ed_outside_utility
        logits = jnp.array([u_ed, u_gp, u_outside])
        prob_ed = jax.nn.softmax(logits * params.logit_sensitivity)[0]

        # 4. Resulting demand
        return total_base_demand * prob_ed, prob_ed

    # JIT-friendly loop
    d_final, p_final = lax.fori_loop(
        0, max_iter, body_fun, (jnp.array(total_base_demand), jnp.array(params.queuing_init_prob))
    )

    return d_final, p_final

solve_queuing_equilibrium_legacy(total_base_demand, capacity, discharge_delay, params, p_global, max_iter=5)

Legacy version of the queuing solver. Returns (demand_ed, prob_ed).

Source code in src/nhra_gt/subgames/queuing.py

def solve_queuing_equilibrium_legacy(
    total_base_demand: float,
    capacity: float,
    discharge_delay: float,
    params: PatientUtilityParams,
    p_global: Any,
    max_iter: int = 5,
) -> tuple[float, float]:
    """Legacy version of the queuing solver. Returns (demand_ed, prob_ed)."""
    from nhra_gt.engine import mm_s_queue_wait

    p_final = params.queuing_init_prob
    d_final = total_base_demand
    for _ in range(max_iter):
        wait_min = mm_s_queue_wait(
            d_final,
            1.0 / max(1e-9, discharge_delay),
            capacity * p_global.ops.capacity_scalar,
            p_global,
        )

        # Utilities
        u_ed, u_gp = calculate_patient_utilities(wait_min, params, p_global)
        u_outside = params.ed_outside_utility

        # Logit
        e_ed = np.exp(u_ed * params.logit_sensitivity)
        e_gp = np.exp(u_gp * params.logit_sensitivity)
        e_out = np.exp(u_outside * params.logit_sensitivity)

        prob_ed = e_ed / (e_ed + e_gp + e_out)
        d_final = total_base_demand * float(prob_ed)
        p_final = float(prob_ed)

    return d_final, p_final