Allow the pension parameter replacement_rate_adjust to vary over T and J

[jdebacker]

This PR adds flexibility to the replacement rate adjustment parameter, which scales pensions benefits up and down, to allow the parameter to vary by time, $t$, and ability type, $j$.

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 72.67%. Comparing base (759afbf) to head (0dc79a4).

Additional details and impacted files

@@           Coverage Diff           @@
##           master    #1044   +/-   ##
=======================================
  Coverage   72.67%   72.67%           
=======================================
  Files          20       20           
  Lines        5068     5068           
=======================================
  Hits         3683     3683           
  Misses       1385     1385           

Flag	Coverage Δ
unittests	72.67% <100.00%> (ø)

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
ogcore/TPI.py	37.19% <100.00%> (ø)
ogcore/parameters.py	83.12% <ø> (ø)
ogcore/pensions.py	67.03% <100.00%> (ø)

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[rickecon]

@jdebacker. It looks like the tests stopped because 1 test failed for a random reason (MacOS 3.13). I just reset the test to start running. If it has passed in the morning, then we need to just reset all the tests.

[rickecon]

@jdebacker. I am ready to merge this as soon as we get all the tests to pass.

[jdebacker]

This branch has some failures in local tests. Will report back when passing.

[rickecon]

@jdebacker. I got all the GitHub Action CI tests to pass. It was annoying that I had to manually set off each of the six sets of tests. But they've all passed now. Let me know when your local tests pass, and I will merge this.

[jdebacker]

@rickecon I think my last PR provided the fix to the local test failures (the parameters replacement_rate_adjust needed to be specified in testing_params.json since that parameterization sets J=2).

Note that I haven't been able to confirm this on my own as I get an error related to the dask client that is used in pytest:

@pytest.fixture(scope="module")
def dask_client():
    cluster = LocalCluster(n_workers=NUM_WORKERS, threads_per_worker=2)
    client = Client(cluster)
    yield client
    # teardown
    client.close()
    cluster.close()

If I set this to:

@pytest.fixture(scope="module")
def dask_client():
    cluster = LocalCluster(n_workers=NUM_WORKERS, threads_per_worker=2)
    client = None

The tests work, but I don't get any parallel compute. I haven't been able to get to the bottom of why this error now. But I'd be interested in if you are getting the same of if you can run the tests. One test that you can check this with is pytest tests/test_run_ogcore.py::test_run_micro_macro

[rickecon]

@jdebacker. Yes. I get the same error when I run pytest tests/test_run_ogcore.py::test_run_micro_macro. The traceback is the following.

____________________________________ test_run_micro_macro ____________________________________

tmpdir = local('/private/var/folders/d4/trj3dssd6s3g8kxvjmczz11w0000gn/T/pytest-of-richardevans/pytest-0/test_run_micro_macro0')
dask_client = <Client: 'tcp://127.0.0.1:53149' processes=7 threads=14, memory=64.00 GiB>

    @pytest.mark.local
    def test_run_micro_macro(tmpdir, dask_client):
        og_spec = os.path.join(CUR_PATH, "testing_params.json")
        guid = ""
        start_time = time.time()
    
        REFORM_DIR = os.path.join(tmpdir, "OUTPUT_REFORM_" + guid)
        BASELINE_DIR = os.path.join(tmpdir, "OUTPUT_BASELINE" + guid)
    
        with open("log_{}.log".format(guid), "w") as f:
            f.write("guid: {}\n".format(guid))
            f.write("og_spec: {}\n".format(og_spec))
    
        """
        ------------------------------------------------------------------------
            Run baseline
        ------------------------------------------------------------------------
        """
        p = Specifications(
            baseline=True,
            num_workers=NUM_WORKERS,
            baseline_dir=BASELINE_DIR,
            output_base=BASELINE_DIR,
        )
        p.update_specifications(og_spec)
>       runner(p, time_path=True, client=dask_client)

tests/test_run_ogcore.py:53: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
ogcore/execute.py:46: in runner
    ss_outputs = SS.run_SS(p, client=client)
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^
ogcore/SS.py:1319: in run_SS
    sol = opt.root(
/opt/anaconda3/envs/ogcore-dev/lib/python3.13/site-packages/scipy/optimize/_root.py:253: in root
    sol = _root_hybr(_wrapped_fun, x0, args=args, jac=jac, **options)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
/opt/anaconda3/envs/ogcore-dev/lib/python3.13/site-packages/scipy/optimize/_minpack_py.py:250: in _root_hybr
    retval = _minpack._hybrd(func, x0, args, 1, xtol, maxfev,
/opt/anaconda3/envs/ogcore-dev/lib/python3.13/site-packages/scipy/optimize/_root.py:215: in _wrapped_fun
    return fun(*fargs)
           ^^^^^^^^^^^
ogcore/SS.py:1168: in SS_fsolve
    ) = inner_loop(outer_loop_vars, p, client)
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
ogcore/SS.py:300: in inner_loop
    results = client.gather(futures)
              ^^^^^^^^^^^^^^^^^^^^^^
/opt/anaconda3/envs/ogcore-dev/lib/python3.13/site-packages/distributed/client.py:2556: in gather
    return self.sync(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Client: 'tcp://127.0.0.1:53149' processes=7 threads=14, memory=64.00 GiB>
futures = [<Future: cancelled, key: solve_for_j-09394f85-cc22-45c2-83ba-ef40c6208479>, <Future: cancelled, key: solve_for_j-aef15203-6ee3-4cdf-8fac-8c95df3b1fe5>]
errors = 'raise', direct = False, local_worker = None

    async def _gather(self, futures, errors="raise", direct=None, local_worker=None):
        unpacked, future_set = unpack_remotedata(futures, byte_keys=True)
        mismatched_futures = [f for f in future_set if f.client is not self]
        if mismatched_futures:
            raise ValueError(
                "Cannot gather Futures created by another client. "
                f"These are the {len(mismatched_futures)} (out of {len(futures)}) "
                f"mismatched Futures and their client IDs (this client is {self.id}): "
                f"{ {f: f.client.id for f in mismatched_futures} }"  # noqa: E201, E202
            )
        keys = [future.key for future in future_set]
        bad_data = dict()
        data = {}
    
        if direct is None:
            direct = self.direct_to_workers
        if direct is None:
            try:
                w = get_worker()
            except Exception:
                direct = False
            else:
                if w.scheduler.address == self.scheduler.address:
                    direct = True
    
        async def wait(k):
            """Want to stop the All(...) early if we find an error"""
            try:
                st = self.futures[k]
            except KeyError:
                raise AllExit()
            else:
                await st.wait()
            if st.status != "finished" and errors == "raise":
                raise AllExit()
    
        while True:
            logger.debug("Waiting on futures to clear before gather")
    
            with suppress(AllExit):
                await distributed.utils.All(
                    [wait(key) for key in keys if key in self.futures],
                    quiet_exceptions=AllExit,
                )
    
            failed = ("error", "cancelled")
    
            exceptions = set()
            bad_keys = set()
            for future in future_set:
                key = future.key
                if key not in self.futures or future.status in failed:
                    exceptions.add(key)
                    if errors == "raise":
                        st = future._state
                        exception = st.exception
                        traceback = st.traceback
>                       raise exception.with_traceback(traceback)
E                       distributed.client.FutureCancelledError: solve_for_j-aef15203-6ee3-4cdf-8fac-8c95df3b1fe5 cancelled for reason: lost dependencies.

/opt/anaconda3/envs/ogcore-dev/lib/python3.13/site-packages/distributed/client.py:2417: FutureCancelledError

================================== short test summary info ===================================
FAILED tests/test_run_ogcore.py::test_run_micro_macro - distributed.client.FutureCancelledError: solve_for_j-aef15203-6ee3-4cdf-8fac-8c95df3b1fe5...
=============================== 1 failed, 7 warnings in 59.25s ===============================

[rickecon]

@jdebacker. I restarted Linux/Python 3.12 GA CI tests for the most recent merged OG-Core PR , and it successfully ran. This deepens the mystery. The hypothesis is that something changed in pytest and/or dask that made the two stop working together. However, this is not the case in the most recent OG-Core PR.

[rickecon]

@jdebacker. I got the following information from Google Gemini.

Me: Are there issues between python 3.13 and pytest?

Gemini:
Yes, there have been reports of compatibility issues between Python 3.13 and pytest, particularly during the earlier stages of Python 3.13's development (alpha and beta releases). These issues often stem from changes introduced in Python 3.13 that affect how certain libraries or functionalities interact with pytest.

Common types of issues reported include:

Intermittent Test Failures:
Some users have experienced non-reproducible test failures when running pytest with Python 3.13, which may not be tied to specific code errors but rather to environmental or timing-related factors.

Mocking Compatibility Issues:
Changes in Python 3.13, such as new keyword arguments in built-in functions or changes in how certain modules behave, can lead to issues with existing mocking setups in tests.

Broken Input/Output:
Problems with input handling, particularly when using Pdb or manual input() calls within tests, have been reported, especially in environments with specific Python builds (e.g., on macOS or with uv-managed Python binaries).

Incompatible Packages:
Some third-party packages that pytest relies on, or that are used within projects tested with pytest, may not have been fully updated to support Python 3.13, leading to errors or unexpected behavior.

Troubleshooting:
Update pytest and Dependencies:
Ensure that you are using the latest stable version of pytest and all relevant third-party libraries, as updates often include compatibility fixes for newer Python versions.

Review Test Code for 3.13 Changes:
Examine your test code for areas that might be sensitive to changes in Python 3.13, such as mocking implementations or interactions with built-in functions.

Check Pytest Changelog:
Consult the pytest changelog for details on bug fixes and compatibility improvements related to Python 3.13.

Isolate and Debug:
If encountering intermittent failures, try isolating the problematic tests, running them individually, or using debugging tools to identify the root cause.

[rickecon]

@jdebacker. Ruh-roh. Dask documentation says that it currently only supports Python 3.9-3.12 (see here). However, Dask distributed release descriptions say that it supports Python 3.13 (see release notes for version 24.12.0). and Dask dask release descriptions say that it also supports Python 3.13 (see release notes for version 24.12.0). So Dask should support Python 3.13.

[rickecon]

@jdebacker. I re-ran all the GitHub Action CI tests from the last merged PR for Linux and Mac Python 3.12 and 3.13, and they all passed. I recommend we merge this and figure out why the local pytest is failing later.

[jdebacker]

@rickecon I'm good with that.