fix: harden lazy-loading guards and input validation

_lazy_subpackage: add callable() check in __getattr__, _known_names guard,
exclude aliases from __all__ to prevent builtin shadowing

core: skip class-level caching for max/min/sum aliases

rolling_max/min/sum: raise ValueError for length ≤ 0 instead of silently
defaulting to 30

_indicator_loader: move _ALWAYS_FETCH to module scope; raise ValueError
when required column param is None instead of falling through

squeeze_pro: import _pos_float/_pos_int from utils._core directly
This commit is contained in:
xgboosted
2026-06-28 22:38:33 +10:00
parent 821038e1e1
commit 284eb71fcb
9 changed files with 74 additions and 15 deletions
+4 -1
View File
@@ -22,6 +22,8 @@ _SERIES_COLUMN_PARAMS: dict[str, tuple] = {
"series_b": (("b", "series_b"), "close"),
}
_ALWAYS_FETCH: frozenset[str] = frozenset({"close", "high", "low", "volume"})
# Reverse lookup: indicator name → category name (built once at import)
_INDICATOR_TO_CATEGORY: dict[str, str] = {ind: cat for cat, indicators in Category.items() for ind in indicators}
_INDICATOR_TO_CATEGORY.update({alias: "math" for alias in _MATH_ALIASES})
@@ -54,7 +56,6 @@ def _make_ta_wrapper(func: Callable) -> Callable:
# - it is one of the primary OHLCV columns (close, high, low, volume).
# open_ is optional: only fetched when (a) it has no default (required by
# the function) or (b) the caller explicitly passes open=<col> in kwargs.
_ALWAYS_FETCH = {"close", "high", "low", "volume"}
col_params_required = [
p for p in sig.parameters if p in _COLUMN_PARAM_TO_COL_KEY and (p in _ALWAYS_FETCH or sig.parameters[p].default is inspect.Parameter.empty)
]
@@ -90,6 +91,8 @@ def _make_ta_wrapper(func: Callable) -> Callable:
for param_name in col_params_required:
col_key = _COLUMN_PARAM_TO_COL_KEY[param_name]
col_val = kwargs.pop(col_key, col_key)
if col_val is None:
raise ValueError(f"'{col_key}' cannot be None; pass a Series or column name string")
call_kwargs[param_name] = self._get_column(col_val)
# Extract optional column values only when explicitly requested by user
for param_name in col_params_optional:
+6 -2
View File
@@ -75,6 +75,8 @@ def install_lazy_subpackage(
def __getattr__(self, name: str) -> Any:
pkg = object.__getattribute__(self, "__name__")
if name not in _known_names:
raise AttributeError(f"module {pkg!r} has no attribute {name!r}")
if name in special_map:
submod_name, attr = special_map[name]
mod = importlib.import_module(f"{pkg}.{submod_name}")
@@ -85,7 +87,7 @@ def install_lazy_subpackage(
try:
mod = importlib.import_module(f"{pkg}.{canonical}")
func = getattr(mod, canonical, None)
if func is not None:
if func is not None and callable(func):
object.__setattr__(self, name, func)
return func
except ModuleNotFoundError:
@@ -99,7 +101,9 @@ def install_lazy_subpackage(
new_mod = _LazySubpackage(module_name)
# Expose all lazily-loadable names so `from pkg import *` works without prior access.
new_mod.__all__ = sorted(_known_names)
# Exclude aliases (e.g. max/min/sum) from __all__: wildcard import would
# shadow Python builtins. They remain accessible via explicit attribute access.
new_mod.__all__ = sorted(_known_names - aliases_map.keys())
# Copy module attributes from the original so docstring, path, spec, etc. survive.
new_mod.__doc__ = getattr(original, "__doc__", None)
if hasattr(original, "__path__"):
+6 -3
View File
@@ -13,7 +13,7 @@ from numpy import log10 as npLog10
from numpy import ndarray as npNdarray
from pandas.core.base import PandasObject
from pandas_ta_classic._meta import Category, Imports, version
from pandas_ta_classic._meta import Category, Imports, version, _MATH_ALIASES
from pandas_ta_classic._indicator_loader import _find_indicator_func, _make_ta_wrapper
from pandas_ta_classic.utils import *
@@ -935,8 +935,11 @@ class AnalysisIndicators(BasePandasObject):
wrapper = _make_ta_wrapper(func)
wrapper.__name__ = name
wrapper.__qualname__ = f"AnalysisIndicators.{name}"
# Cache on the class so future calls bypass __getattr__
setattr(type(self), name, wrapper)
# Cache on the class so future calls bypass __getattr__.
# Aliases (max/min/sum) are intentionally excluded: caching them on
# the class would permanently shadow Python builtins at the class level.
if name not in _MATH_ALIASES:
setattr(type(self), name, wrapper)
return wrapper.__get__(self, type(self))
# ichimoku is the only explicit wrapper left: it returns a (Series, DataFrame)
+3 -1
View File
@@ -12,7 +12,9 @@ def rolling_max(
**kwargs: Any,
) -> Optional[Series]:
"""Rolling Maximum over *length* periods (TA-Lib: MAX)."""
length = int(length) if length and length > 0 else 30
length = int(length) if length is not None else 30
if length <= 0:
raise ValueError(f"length must be positive, got {length}")
close = verify_series(close, length)
offset = get_offset(offset)
if close is None:
+3 -1
View File
@@ -12,7 +12,9 @@ def rolling_min(
**kwargs: Any,
) -> Optional[Series]:
"""Rolling Minimum over *length* periods (TA-Lib: MIN)."""
length = int(length) if length and length > 0 else 30
length = int(length) if length is not None else 30
if length <= 0:
raise ValueError(f"length must be positive, got {length}")
close = verify_series(close, length)
offset = get_offset(offset)
if close is None:
+3 -1
View File
@@ -12,7 +12,9 @@ def rolling_sum(
**kwargs: Any,
) -> Optional[Series]:
"""Rolling Summation over *length* periods (TA-Lib: SUM)."""
length = int(length) if length and length > 0 else 30
length = int(length) if length is not None else 30
if length <= 0:
raise ValueError(f"length must be positive, got {length}")
close = verify_series(close, length)
offset = get_offset(offset)
if close is None:
+2 -6
View File
@@ -3,12 +3,8 @@ from typing import Any, Optional
from pandas import DataFrame, Series
from pandas_ta_classic.momentum.mom import mom
from pandas_ta_classic.momentum.squeeze import (
_pos_float,
_pos_int,
_squeeze_detailed,
_squeeze_simplify_columns,
)
from pandas_ta_classic.utils._core import _pos_float, _pos_int
from pandas_ta_classic.momentum.squeeze import _squeeze_detailed, _squeeze_simplify_columns
from pandas_ta_classic.overlap.ema import ema
from pandas_ta_classic.overlap.sma import sma
from pandas_ta_classic.volatility.bbands import bbands
+18
View File
@@ -127,6 +127,24 @@ class TestMath(TestCase):
if HAS_TALIB:
assert_talib(self, result, talib.SUM(self.close, 30))
def test_rolling_max_length_invalid_raises(self):
with self.assertRaises(ValueError):
pandas_ta.rolling_max(self.close, length=0)
with self.assertRaises(ValueError):
pandas_ta.rolling_max(self.close, length=-5)
def test_rolling_min_length_invalid_raises(self):
with self.assertRaises(ValueError):
pandas_ta.rolling_min(self.close, length=0)
with self.assertRaises(ValueError):
pandas_ta.rolling_min(self.close, length=-5)
def test_rolling_sum_length_invalid_raises(self):
with self.assertRaises(ValueError):
pandas_ta.rolling_sum(self.close, length=0)
with self.assertRaises(ValueError):
pandas_ta.rolling_sum(self.close, length=-5)
def test_maxindex(self):
assert_indicator_standard(
self,
+29
View File
@@ -78,6 +78,17 @@ class TestLazySubpackage(unittest.TestCase):
for name in ("sma", "ema", "wma"):
self.assertIn(name, names, f"{name!r} must appear in dir(overlap)")
def test_all_excludes_aliases(self):
"""__all__ must exclude aliases so wildcard imports don't shadow builtins."""
import pandas_ta_classic.math
self.assertNotIn("max", pandas_ta_classic.math.__all__)
self.assertNotIn("min", pandas_ta_classic.math.__all__)
self.assertNotIn("sum", pandas_ta_classic.math.__all__)
self.assertIn("rolling_max", pandas_ta_classic.math.__all__)
self.assertIn("rolling_min", pandas_ta_classic.math.__all__)
self.assertIn("rolling_sum", pandas_ta_classic.math.__all__)
# ---------------------------------------------------------------------------
# _indicator_loader tests
@@ -120,6 +131,24 @@ class TestIndicatorLoader(unittest.TestCase):
self.assertIsNotNone(result2)
self.assertEqual(len(result), len(result2))
def test_aliases_not_cached_on_class(self):
"""max/min/sum must NOT be cached as class attributes to prevent builtin shadowing."""
df = get_sample_data()
df.ta.max(length=5)
df.ta.min(length=5)
df.ta.sum(length=5)
cls = type(df.ta)
self.assertNotIn("max", vars(cls), "max must not be cached on AnalysisIndicators")
self.assertNotIn("min", vars(cls), "min must not be cached on AnalysisIndicators")
self.assertNotIn("sum", vars(cls), "sum must not be cached on AnalysisIndicators")
def test_required_column_none_raises(self):
"""Passing close=None to a required column param must raise ValueError."""
df = get_sample_data()
with self.assertRaises(ValueError):
df.ta.rsi(close=None)
# ---------------------------------------------------------------------------
# Module-level __getattr__ tests