Report aggregate hours worked (N) in addition to effective labor (L)

ogcore/SS.py

@@ -877,6 +877,7 @@ def SS_solver(
     I_g_ss = fiscal.get_I_g(Yss, Ig_baseline, p, "SS")
     K_g_ss = fiscal.get_K_g(0, I_g_ss, p, "SS")
     Lss = aggr.get_L(nssmat, p, "SS")
+    Nss = aggr.get_N(nssmat, p, "SS")
     Bss = aggr.get_B(bssmat_splus1, p, "SS", False)
     (
         Dss,
@@ -1207,6 +1208,7 @@ def SS_solver(
         "K_f": K_f_ss,
         "K_d": K_d_ss,
         "L": Lss,
+        "N": Nss,
         "C": Css,
         "I": Iss,
         "I_total": Iss_total,

ogcore/TPI.py

@@ -1745,6 +1745,9 @@ def run_TPI(p, client=None):
         f"Max Euler error labor supply: {np.abs(eul_laborleisure).max()}"
     )
 
+    # Aggregate hours worked (not weighted by effective-labor units).
+    N = aggr.get_N(n_mat[: p.T], p, "TPI")
+
     """
     ------------------------------------------------------------------------
     Save variables/values so they can be used in other modules
@@ -1758,6 +1761,7 @@ def run_TPI(p, client=None):
         "K_f": K_f[: p.T],
         "K_d": K_d[: p.T, ...],
         "L": L[: p.T, ...],
+        "N": N[: p.T, ...],
         "C": C[: p.T, ...],
         "I": I[: p.T, ...],
         "I_total": I_total[: p.T, ...],

ogcore/aggregates.py

@@ -45,6 +45,39 @@ def get_L(n, p, method):
     return L
 
 
+def get_N(n, p, method):
+    r"""
+    Calculate aggregate hours worked.
+
+    Unlike aggregate labor supply :math:`L`, this measure does not weight
+    hours by the effective-labor-units profile :math:`e_{j,s,t}`, so it
+    reports raw hours rather than productivity-adjusted labor:
+
+    .. math::
+        N_{t} = \sum_{s=E}^{E+S}\sum_{j=0}^{J}\omega_{s,t}\lambda_{j}n_{j,s,t}
+
+    Args:
+        n (Numpy array): labor supply of households
+        p (OG-Core Specifications object): model parameters
+        method (str): adjusts calculation dimensions based on 'SS' or
+            'TPI'
+
+    Returns:
+        N (array_like): aggregate hours worked
+
+    """
+    if method == "SS":
+        # p.omega_SS is the (S, J) population weight and already carries the
+        # lambda_j type shares (its columns sum to lambdas), so no separate
+        # lambda multiplication is needed, mirroring get_L.
+        N_presum = p.omega_SS * n
+        N = N_presum.sum()
+    elif method == "TPI":
+        N_presum = n * p.omega[: p.T, :, :]
+        N = N_presum.sum(1).sum(1)
+    return N
+
+
 def get_I(b_splus1, K_p1, K, p, method):
     r"""
     Calculate aggregate investment.

tests/test_aggregates.py

@@ -57,6 +57,30 @@ def test_get_L(n, p, method, expected):
     assert np.allclose(L, expected)
 
 
+# Aggregate hours worked is the same population-weighted sum as get_L but
+# without the effective-labor-units (p.e) weighting.
+N_loop = np.ones(p.T * p.S * p.J).reshape(p.T, p.S, p.J)
+for t in range(p.T):
+    for i in range(p.S):
+        for k in range(p.J):
+            N_loop[t, i, k] *= p.omega[t, i, k].item() * n[t, i, k].item()
+expected_N_SS = N_loop[-1, :, :].sum()
+expected_N_TPI = N_loop.sum(1).sum(1)
+test_data_N = [
+    (n[-1, :, :], p, "SS", expected_N_SS),
+    (n, p, "TPI", expected_N_TPI),
+]
+
+
+@pytest.mark.parametrize("n,p,method,expected", test_data_N, ids=["SS", "TPI"])
+def test_get_N(n, p, method, expected):
+    """
+    Test aggregate hours-worked function.
+    """
+    N = aggr.get_N(n, p, method)
+    assert np.allclose(N, expected)
+
+
 def test_get_L_J1_regression():
     """
     Regression test for issue #1143: aggregate labor for J=1 must use a