REF: Restructure

Move kernel info closer to kernels

Author: bashtage

arch/covariance/__init__.py

@@ -0,0 +1,12 @@
+from typing import Dict, Type
+
+from . import kernel
+
+KERNEL_ESTIMATORS: Dict[str, Type[kernel.CovarianceEstimator]] = {
+    est_name.lower(): getattr(kernel, est_name) for est_name in kernel.KERNELS
+}
+KERNEL_ESTIMATORS.update(
+    {est_name: getattr(kernel, est_name) for est_name in kernel.KERNELS}
+)
+KNOWN_KERNELS = "\n".join(sorted([k for k in KERNEL_ESTIMATORS]))
+KERNEL_ERR = f"kernel is not a known estimator. Must be one of:\n {KNOWN_KERNELS}"

arch/covariance/kernel.py

@@ -24,6 +24,7 @@
     "Andrews",
     "Gallant",
     "NeweyWest",
+    "normalize_kernel_name",
 ]
 
 KERNELS = [
@@ -42,6 +43,22 @@
 ]
 
 
+def normalize_kernel_name(name: str) -> str:
+    """
+    Normalize a Kernel name using standard replacements
+
+    Removes - and _ and converts to lower case.
+
+    Returns
+    -------
+    str
+        The normalized kernel name.
+    """
+    name = name.replace("-", "").replace("_", "")
+    name = name.lower()
+    return name
+
+
 class CovarianceEstimate(object):
     r"""
     Covariance estimate using a long-run covariance estimator

arch/covariance/var.py

@@ -1,4 +1,4 @@
-from typing import Dict, NamedTuple, Optional, Tuple, Type
+from typing import Dict, NamedTuple, Optional, Tuple
 
 import numpy as np
 from numpy import zeros
@@ -7,8 +7,12 @@
 from statsmodels.tools import add_constant
 from statsmodels.tsa.tsatools import lagmat
 
-from arch.covariance import kernel
-from arch.covariance.kernel import CovarianceEstimate, CovarianceEstimator
+from arch.covariance import KERNEL_ERR, KERNEL_ESTIMATORS
+from arch.covariance.kernel import (
+    CovarianceEstimate,
+    CovarianceEstimator,
+    normalize_kernel_name,
+)
 from arch.typing import ArrayLike, NDArray
 
 
@@ -19,21 +23,6 @@ class VARModel(NamedTuple):
     intercept: bool
 
 
-def _normalize_name(name: str) -> str:
-    name = name.replace("-", "").replace("_", "")
-    name = name.lower()
-    return name
-
-
-KERNELS: Dict[str, Type[CovarianceEstimator]] = {}
-for name in kernel.__all__:
-    estimator = getattr(kernel, name)
-    if issubclass(estimator, kernel.CovarianceEstimator):
-        KERNELS[_normalize_name(name)] = estimator
-        KERNELS[name] = estimator
-print(KERNELS)
-
-
 class PreWhitenRecoloredCovariance(CovarianceEstimator):
     """
     Parameters
@@ -86,20 +75,11 @@ def __init__(
         self._auto_lag_selection = True
         self._format_lags(lags)
         self._sample_autocov = sample_autocov
-        original_kernel = kernel
-        kernel = _normalize_name(kernel)
-        if kernel not in KERNELS:
-            import string
-
-            available = [key for key in KERNELS if key[0] in string.ascii_uppercase]
-            available_val = "\n ".join(
-                [f"{knl} {_normalize_name(knl)}" for knl in available]
-            )
-            raise ValueError(
-                f"kernel {original_kernel} was not found. The available kernels "
-                f"are:\n\n{available_val}"
-            )
-        self._kernel = KERNELS[kernel]
+        kernel = normalize_kernel_name(kernel)
+        if kernel not in KERNEL_ESTIMATORS:
+            raise ValueError(KERNEL_ERR)
+
+        self._kernel = KERNEL_ESTIMATORS[kernel]
         self._kernel_instance: Optional[CovarianceEstimator] = None
 
         # Attach for testing only

arch/tests/covariance/test_var.py

@@ -108,7 +108,7 @@ def direct_var(
         else:
             # Branch is a workaround of NumPy 1.15
             # TODO: Remove after NumPy 1.15 dropped
-            resids[:, i: i + 1] = lhs[:, i: i + 1]
+            resids[:, i : i + 1] = lhs[:, i : i + 1]
     return params, resids
 
 

arch/unitroot/_phillips_ouliaris.py

@@ -7,14 +7,10 @@
 from statsmodels.iolib.table import SimpleTable
 from statsmodels.regression.linear_model import RegressionResults
 
-import arch.covariance.kernel as lrcov
+from arch.covariance import KERNEL_ERR, KERNEL_ESTIMATORS
+from arch.covariance.kernel import CovarianceEstimator
 from arch.typing import ArrayLike1D, ArrayLike2D
-from arch.unitroot._shared import (
-    KERNEL_ERR,
-    KERNEL_ESTIMATORS,
-    ResidualCointegrationTestResult,
-    _cross_section,
-)
+from arch.unitroot._shared import ResidualCointegrationTestResult, _cross_section
 from arch.unitroot.critical_values.phillips_ouliaris import (
     CV_PARAMETERS,
     CV_TAU_MIN,
@@ -314,7 +310,7 @@ def __init__(
         order: int = 2,
         xsection: Optional[RegressionResults] = None,
         test_type: str = "Za",
-        kernel_est: Optional[lrcov.CovarianceEstimator] = None,
+        kernel_est: Optional[CovarianceEstimator] = None,
         rho: float = 0.0,
     ) -> None:
         super().__init__(

arch/unitroot/_shared.py

@@ -1,10 +1,10 @@
-from typing import Any, Dict, NamedTuple, Optional, Tuple, Type
+from typing import Any, Dict, NamedTuple, Optional, Tuple
 
 import pandas as pd
 from statsmodels.iolib.summary import Summary
 from statsmodels.regression.linear_model import OLS, RegressionResults
 
-import arch.covariance.kernel as lrcov
+from arch.covariance import KERNEL_ESTIMATORS
 from arch.typing import ArrayLike1D, ArrayLike2D
 from arch.utility.array import ensure1d, ensure2d
 from arch.utility.timeseries import add_trend
@@ -14,13 +14,6 @@
 except ImportError:
     pass
 
-KERNEL_ESTIMATORS: Dict[str, Type[lrcov.CovarianceEstimator]] = {
-    kernel.lower(): getattr(lrcov, kernel) for kernel in lrcov.KERNELS
-}
-KERNEL_ESTIMATORS.update({kernel: getattr(lrcov, kernel) for kernel in lrcov.KERNELS})
-KNOWN_KERNELS = "\n".join(sorted([k for k in KERNEL_ESTIMATORS]))
-KERNEL_ERR = f"kernel is not a known estimator. Must be one of:\n {KNOWN_KERNELS}"
-
 
 class CointegrationSetup(NamedTuple):
     y: pd.Series

arch/unitroot/cointegration.py

@@ -8,7 +8,8 @@
 from statsmodels.iolib.table import SimpleTable
 from statsmodels.regression.linear_model import OLS, RegressionResults
 
-import arch.covariance.kernel as lrcov
+from arch.covariance import KERNEL_ERR, KERNEL_ESTIMATORS
+from arch.covariance.kernel import CovarianceEstimate, CovarianceEstimator
 from arch.typing import ArrayLike1D, ArrayLike2D, NDArray
 from arch.unitroot._engle_granger import EngleGrangerTestResults, engle_granger
 from arch.unitroot._phillips_ouliaris import (
@@ -17,8 +18,6 @@
     phillips_ouliaris,
 )
 from arch.unitroot._shared import (
-    KERNEL_ERR,
-    KERNEL_ESTIMATORS,
     _check_cointegrating_regression,
     _check_kernel,
     _cross_section,
@@ -46,7 +45,7 @@ def __init__(
         params: pd.Series,
         cov: pd.DataFrame,
         resid: pd.Series,
-        kernel_est: lrcov.CovarianceEstimator,
+        kernel_est: CovarianceEstimator,
         num_x: int,
         trend: str,
         df_adjust: bool,
@@ -126,7 +125,7 @@ def rsquared_adj(self) -> float:
         return self._rsquared_adj
 
     @cached_property
-    def _cov_est(self) -> lrcov.CovarianceEstimate:
+    def _cov_est(self) -> CovarianceEstimate:
         r = np.asarray(self._resid)
         kern_class = self._kernel_est.__class__
         bw = self._bandwidth
@@ -329,7 +328,7 @@ def __init__(
         lags: int,
         leads: int,
         cov_type: str,
-        kernel_est: lrcov.CovarianceEstimator,
+        kernel_est: CovarianceEstimator,
         num_x: int,
         trend: str,
         reg_results: RegressionResults,
@@ -776,7 +775,7 @@ def _cov(
         df_adjust: bool,
         rhs: pd.DataFrame,
         resids: pd.Series,
-    ) -> Tuple[pd.DataFrame, lrcov.CovarianceEstimator]:
+    ) -> Tuple[pd.DataFrame, CovarianceEstimator]:
         """Estimate the covariance"""
         kernel = kernel.lower().replace("-", "").replace("_", "")
         if kernel not in KERNEL_ESTIMATORS:
@@ -810,7 +809,7 @@ def __init__(
         cov: pd.DataFrame,
         resid: pd.Series,
         omega_112: float,
-        kernel_est: lrcov.CovarianceEstimator,
+        kernel_est: CovarianceEstimator,
         num_x: int,
         trend: str,
         df_adjust: bool,
@@ -981,7 +980,7 @@ def __init__(
 
     def _common_fit(
         self, kernel: str, bandwidth: Optional[float], force_int: bool, diff: bool
-    ) -> Tuple[lrcov.CovarianceEstimator, NDArray, NDArray]:
+    ) -> Tuple[CovarianceEstimator, NDArray, NDArray]:
         kernel = _check_kernel(kernel)
         res = _cross_section(self._y, self._x, self._trend)
         x = np.asarray(self._x)