FM / window

/usr/local/lib64/python3.6/site-packages/pandas/tests/window/

/usr/local/lib64/python3.6/site-packages/pandas/tests/window

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
common.py	6139	0644	edit dl rm
conftest.py	7409	0644	edit dl rm
test_api.py	10429	0644	edit dl rm
test_apply.py	5270	0644	edit dl rm
test_base_indexer.py	8161	0644	edit dl rm
test_dtypes.py	7339	0644	edit dl rm
test_ewm.py	4087	0644	edit dl rm
test_expanding.py	7118	0644	edit dl rm
test_grouper.py	19314	0644	edit dl rm
test_numba.py	2942	0644	edit dl rm
test_pairwise.py	8405	0644	edit dl rm
test_rolling.py	21693	0644	edit dl rm
test_timeseries_window.py	24905	0644	edit dl rm
test_window.py	2033	0644	edit dl rm
__init__.py	0	0644	edit dl rm

Edit: /usr/local/lib64/python3.6/site-packages/pandas/tests/window/test_numba.py (2942B)

import numpy as np
import pytest

import pandas.util._test_decorators as td

from pandas import Series, option_context
import pandas._testing as tm
from pandas.core.util.numba_ import NUMBA_FUNC_CACHE

@td.skip_if_no("numba", "0.46.0")
@pytest.mark.filterwarnings("ignore:\\nThe keyword argument")
# Filter warnings when parallel=True and the function can't be parallelized by Numba
class TestApply:
    @pytest.mark.parametrize("jit", [True, False])
    def test_numba_vs_cython(self, jit, nogil, parallel, nopython, center):
        def f(x, *args):
            arg_sum = 0
            for arg in args:
                arg_sum += arg
            return np.mean(x) + arg_sum

if jit:
            import numba

f = numba.jit(f)

engine_kwargs = {"nogil": nogil, "parallel": parallel, "nopython": nopython}
        args = (2,)

s = Series(range(10))
        result = s.rolling(2, center=center).apply(
            f, args=args, engine="numba", engine_kwargs=engine_kwargs, raw=True
        )
        expected = s.rolling(2, center=center).apply(
            f, engine="cython", args=args, raw=True
        )
        tm.assert_series_equal(result, expected)

@pytest.mark.parametrize("jit", [True, False])
    def test_cache(self, jit, nogil, parallel, nopython):
        # Test that the functions are cached correctly if we switch functions
        def func_1(x):
            return np.mean(x) + 4

def func_2(x):
            return np.std(x) * 5

if jit:
            import numba

func_1 = numba.jit(func_1)
            func_2 = numba.jit(func_2)

engine_kwargs = {"nogil": nogil, "parallel": parallel, "nopython": nopython}

roll = Series(range(10)).rolling(2)
        result = roll.apply(
            func_1, engine="numba", engine_kwargs=engine_kwargs, raw=True
        )
        expected = roll.apply(func_1, engine="cython", raw=True)
        tm.assert_series_equal(result, expected)

# func_1 should be in the cache now
        assert (func_1, "rolling_apply") in NUMBA_FUNC_CACHE

result = roll.apply(
            func_2, engine="numba", engine_kwargs=engine_kwargs, raw=True
        )
        expected = roll.apply(func_2, engine="cython", raw=True)
        tm.assert_series_equal(result, expected)
        # This run should use the cached func_1
        result = roll.apply(
            func_1, engine="numba", engine_kwargs=engine_kwargs, raw=True
        )
        expected = roll.apply(func_1, engine="cython", raw=True)
        tm.assert_series_equal(result, expected)

@td.skip_if_no("numba", "0.46.0")
def test_use_global_config():
    def f(x):
        return np.mean(x) + 2

s = Series(range(10))
    with option_context("compute.use_numba", True):
        result = s.rolling(2).apply(f, engine=None, raw=True)
    expected = s.rolling(2).apply(f, engine="numba", raw=True)
    tm.assert_series_equal(expected, result)

import numpy as np import pytest import pandas.util._test_decorators as td from pandas import Series, option_context import pandas._testing as tm from pandas.core.util.numba_ import NUMBA_FUNC_CACHE @td.skip_if_no("numba", "0.46.0") @pytest.mark.filterwarnings("ignore:\\nThe keyword argument") # Filter warnings when parallel=True and the function can't be parallelized by Numba class TestApply: @pytest.mark.parametrize("jit", [True, False]) def test_numba_vs_cython(self, jit, nogil, parallel, nopython, center): def f(x, *args): arg_sum = 0 for arg in args: arg_sum += arg return np.mean(x) + arg_sum if jit: import numba f = numba.jit(f) engine_kwargs = {"nogil": nogil, "parallel": parallel, "nopython": nopython} args = (2,) s = Series(range(10)) result = s.rolling(2, center=center).apply( f, args=args, engine="numba", engine_kwargs=engine_kwargs, raw=True ) expected = s.rolling(2, center=center).apply( f, engine="cython", args=args, raw=True ) tm.assert_series_equal(result, expected) @pytest.mark.parametrize("jit", [True, False]) def test_cache(self, jit, nogil, parallel, nopython): # Test that the functions are cached correctly if we switch functions def func_1(x): return np.mean(x) + 4 def func_2(x): return np.std(x) * 5 if jit: import numba func_1 = numba.jit(func_1) func_2 = numba.jit(func_2) engine_kwargs = {"nogil": nogil, "parallel": parallel, "nopython": nopython} roll = Series(range(10)).rolling(2) result = roll.apply( func_1, engine="numba", engine_kwargs=engine_kwargs, raw=True ) expected = roll.apply(func_1, engine="cython", raw=True) tm.assert_series_equal(result, expected) # func_1 should be in the cache now assert (func_1, "rolling_apply") in NUMBA_FUNC_CACHE result = roll.apply( func_2, engine="numba", engine_kwargs=engine_kwargs, raw=True ) expected = roll.apply(func_2, engine="cython", raw=True) tm.assert_series_equal(result, expected) # This run should use the cached func_1 result = roll.apply( func_1, engine="numba", engine_kwargs=engine_kwargs, raw=True ) expected = roll.apply(func_1, engine="cython", raw=True) tm.assert_series_equal(result, expected) @td.skip_if_no("numba", "0.46.0") def test_use_global_config(): def f(x): return np.mean(x) + 2 s = Series(range(10)) with option_context("compute.use_numba", True): result = s.rolling(2).apply(f, engine=None, raw=True) expected = s.rolling(2).apply(f, engine="numba", raw=True) tm.assert_series_equal(expected, result)