Replace namespace traversal with bounded Python source-tree hashing to cover dynamic helpers without scanning installed libraries. Preserve resolver snapshots and source provenance.
Track ft-pandas-ta metadata and verify native archive and prediction cache invalidation. Unused Python source changes intentionally invalidate reuse.
from scripts.backtest_cache_contract import (
assert_native_implementation_cache,
assert_native_startup_predictions,
+ assert_source_tree_provenance,
)
class NativeResultCacheTest(QaTestCase):
+ def test_source_tree_provenance_preserves_snapshots_and_boundaries(self):
+ with temporary_directory() as directory:
+ assert_source_tree_provenance(self, directory)
+
def test_native_startup_and_predictions_do_not_require_test_pythonpath(self):
with temporary_directory() as directory:
assert_native_startup_predictions(
source metadata are private SDK contracts; recheck them on Freqtrade upgrades.
"""
-import ast
import json
+import os
import platform
+import site
import sys
+import sysconfig
from collections.abc import Iterator, Mapping
from hashlib import sha256
from importlib.metadata import PackageNotFoundError, version
-from importlib.util import resolve_name
-from inspect import getfile, isfunction, unwrap
+from inspect import getfile
from pathlib import Path
-from types import ModuleType
from typing import Any, Final
import freqtrade
"cloudpickle",
"TA-Lib",
"technical",
- "pandas-ta",
+ "ft-pandas-ta",
)
_REGRESSION_DEPENDENCIES: Final = ("xgboost", "lightgbm", "catboost", "ngboost", "scikit-image")
_RL_DEPENDENCIES: Final = ("gymnasium", "stable-baselines3", "sb3-contrib")
-def _import_namespaces(source: bytes, package: str) -> Iterator[dict[str, Any]]:
- """Find loaded imports even when a native-resolved leaf owns no methods."""
- for node in ast.walk(ast.parse(source)):
- if isinstance(node, ast.Import):
- names = [alias.name for alias in node.names]
- elif isinstance(node, ast.ImportFrom):
- name = node.module or ""
- if node.level:
- if not package:
- continue
- try:
- name = resolve_name("." * node.level + name, package)
- except ImportError:
- continue
- names = [name, *(f"{name}.{alias.name}" for alias in node.names)]
- else:
- continue
- for name in names:
- module = sys.modules.get(name)
- if module is not None:
- yield vars(module)
-
-
-def _class_namespaces(cls: type) -> Iterator[dict[str, Any]]:
- """Use actual function globals: native-resolved modules may be unregistered."""
- for member in vars(cls).values():
- if isinstance(member, (staticmethod, classmethod)):
- member = member.__func__
- elif isinstance(member, property):
- member = member.fget
- if callable(member):
- member = unwrap(member)
- if isfunction(member):
- yield member.__globals__
- module = sys.modules.get(cls.__module__)
- # Data-only imported classes have no function globals. Reject name collisions.
- if module is not None and vars(module).get(cls.__name__) is cls:
- yield vars(module)
+def _installed_roots() -> set[Path]:
+ """Locate interpreter libraries without excluding sibling project directories."""
+ paths = sysconfig.get_paths()
+ roots = {
+ Path(value).resolve()
+ for key in ("stdlib", "platstdlib", "purelib", "platlib")
+ if (value := paths.get(key))
+ }
+ roots.update(
+ Path(path).resolve() for path in site.getsitepackages([sys.prefix, sys.base_prefix])
+ )
+ roots.add(Path(site.getusersitepackages()).resolve())
+ roots.add(Path(freqtrade.__file__).resolve().parent)
+ return roots
+
+
+def _python_sources(root: Path, excluded: set[Path]) -> Iterator[Path]:
+ """Walk source files only; directory links cannot expand or cycle the boundary."""
+
+ def fail(error: OSError) -> None:
+ raise error
+
+ for directory, children, files in os.walk(root, onerror=fail):
+ parent = Path(directory)
+ children[:] = [
+ name
+ for name in children
+ if name not in ("__pycache__", ".git")
+ and not (child := parent / name).is_symlink()
+ and not (child / "pyvenv.cfg").is_file()
+ and not any(child.resolve().is_relative_to(path) for path in excluded)
+ ]
+ for name in files:
+ path = parent / name
+ if (
+ path.suffix == ".py"
+ and not any(path.resolve().is_relative_to(directory) for directory in excluded)
+ and path.is_file()
+ ):
+ yield path
def implementation_fingerprint(implementations: Mapping[str, type]) -> str:
- """Hash selected classes, reachable local helpers and release dependencies.
+ """Hash selected classes, bounded local Python trees and release dependencies.
- Paths locate sources but are not part of identity. Resolver snapshots take
- precedence over disk source. Local helper traversal stays inside directories
- containing selected non-SDK MRO implementations; it never scans user data or
- installed packages. Version metadata does not detect same-version SDK/binary
- modifications. File hashes are not memoized across strategy constructions.
+ Every Python file inside local implementation/MRO directories contributes,
+ including unused or dynamically imported helpers. Installed libraries never
+ create scan roots. Resolver snapshots override disk source, even through
+ file aliases. Absolute paths locate files but are not part of the identity.
"""
classes = {role: list(leaf.__mro__[:-1]) for role, leaf in implementations.items()}
- local_classes = [
- cls
- for lineage in classes.values()
- for cls in lineage
- if cls.__module__ != "abc" and not cls.__module__.startswith("freqtrade.")
- ]
- sources: dict[tuple[str, bytes], None] = {}
- source_cache: dict[tuple[str, str | Path], bytes] = {}
- snapshots: dict[Path, str] = {}
- imports: dict[tuple[bytes, str], tuple[dict[str, Any], ...]] = {}
- directories: set[Path] = set()
- namespaces: list[dict[str, Any]] = []
-
- def add_source(path: str | Path, snapshot: str | None = None) -> bytes:
- key = ("snapshot", snapshot) if snapshot is not None else ("file", Path(path).resolve())
- if key not in source_cache:
- source_cache[key] = (
- snapshot.encode() if snapshot is not None else Path(path).read_bytes()
- )
- sources[(Path(path).name, source_cache[key])] = None
- return source_cache[key]
-
- def imported_namespaces(payload: bytes, package: str) -> tuple[dict[str, Any], ...]:
- key = (payload, package)
- if key not in imports:
- imports[key] = tuple(_import_namespaces(payload, package))
- return imports[key]
-
- for lineage in classes.values():
+ excluded = _installed_roots()
+ snapshots: dict[Path, set[bytes]] = {}
+ class_sources: dict[tuple[str, int], tuple[Path, bytes | None]] = {}
+ roots: dict[Path, list[tuple[str, int, str, str]]] = {}
+ for role, lineage in classes.items():
leaf = lineage[0]
- for cls in lineage:
- if cls not in local_classes:
+ for index, cls in enumerate(lineage):
+ if cls.__module__ in ("abc", "builtins") or cls.__module__.startswith("freqtrade."):
continue
snapshot = cls.__dict__.get("__source__")
- path = cls.__dict__.get("__file__")
- if snapshot is None and path is None and cls.__module__ == leaf.__module__:
+ filename = cls.__dict__.get("__file__")
+ if snapshot is None and filename is None and cls.__module__ == leaf.__module__:
snapshot = leaf.__dict__.get("__source__")
- path = leaf.__dict__.get("__file__")
- if snapshot is None and path is None:
- path = getfile(cls)
- payload = add_source(path or cls.__module__, snapshot)
- if path is not None:
- resolved = Path(path).resolve()
- directories.add(resolved.parent)
- if snapshot is not None:
- snapshots[resolved] = snapshot
- namespaces.extend(_class_namespaces(cls))
- namespaces.extend(imported_namespaces(payload, cls.__module__.rpartition(".")[0]))
-
- visited: set[int] = set()
- while namespaces:
- namespace = namespaces.pop()
- if id(namespace) in visited:
- continue
- visited.add(id(namespace))
- filename = namespace.get("__file__")
- if filename is None:
- continue
- path = Path(filename).resolve()
- if not any(path.is_relative_to(directory) for directory in directories):
- continue
- payload = add_source(path, snapshots.get(path))
- namespaces.extend(imported_namespaces(payload, namespace.get("__package__") or ""))
- for value in tuple(namespace.values()):
- if callable(value):
- value = unwrap(value)
- if isfunction(value):
- namespaces.append(value.__globals__)
- elif isinstance(value, ModuleType):
- namespaces.append(vars(value))
- elif isinstance(value, type):
- namespaces.extend(_class_namespaces(value))
-
- add_source(__file__)
+ filename = leaf.__dict__.get("__file__")
+ if filename is None:
+ try:
+ filename = getfile(cls)
+ except TypeError as error:
+ if cls.__module__ in sys.builtin_module_names:
+ continue
+ raise OperationalException(
+ f"Backtest cache: cannot locate Python source for {cls.__qualname__}"
+ ) from error
+ path = Path(filename).resolve()
+ if any(path.is_relative_to(directory) for directory in excluded):
+ continue
+ if path.suffix != ".py":
+ raise OperationalException(
+ f"Backtest cache: Python source is required for {cls.__qualname__}"
+ )
+ payload = snapshot.encode() if snapshot is not None else None
+ class_sources[(role, index)] = (path, payload)
+ roots.setdefault(path.parent, []).append(
+ (role, index, cls.__module__, cls.__qualname__)
+ )
+ if payload is not None:
+ snapshots.setdefault(path, set()).add(payload)
+
+ source_cache: dict[Path, tuple[bytes, ...]] = {}
+ sources: set[tuple[str, bytes]] = set()
+
+ def payloads(path: Path) -> tuple[bytes, ...]:
+ path = path.resolve()
+ if path not in source_cache:
+ source_cache[path] = (
+ tuple(sorted(snapshots[path])) if path in snapshots else (path.read_bytes(),)
+ )
+ return source_cache[path]
+
+ for (role, index), (path, snapshot) in class_sources.items():
+ label = json.dumps(("class", role, index))
+ for payload in (snapshot,) if snapshot is not None else payloads(path):
+ sources.add((label, payload))
+ for root, anchors in roots.items():
+ anchor = min(anchors)
+ paths = set(_python_sources(root, excluded))
+ paths.update(path for path in snapshots if path.is_relative_to(root))
+ for path in paths:
+ label = json.dumps(("tree", anchor, path.relative_to(root).as_posix()))
+ for payload in payloads(path):
+ sources.add((label, payload))
+ sources.add(("identity-helper", Path(__file__).read_bytes()))
rl = any(
cls.__module__.startswith("freqtrade.freqai.RL.")
for lineage in classes.values()
### Backtest result caching
Both strategies bind native result-cache reuse to the selected strategy/model
-implementations, inherited code, loaded local Python helpers in implementation
-directories, Python/Freqtrade versions and relevant dependency versions.
+implementations, their local ancestors, Python/Freqtrade versions and relevant
+dependency versions. Source coverage includes every `.py` file recursively below
+each local implementation's directory, including dynamically imported and unused
+files. Adding, editing, deleting or renaming any such file invalidates reuse.
+Keep implementation code in dedicated directories, separate from generated data.
+
+Installed libraries, bytecode, virtual environments and non-Python artifacts do
+not expand this source boundary. Directory symlinks are not followed; Python
+file symlinks contribute their target contents unless the target is an installed
+library. Code and resources outside these boundaries, and same-version changes
+to installed dependencies, require separate identifiers for cold comparisons.
+
Unchanged inputs retain native parameter handling, cache-age rules and storage
-identifiers. Existing artifacts are not deleted. Version metadata does not
-detect arbitrary same-version edits to installed dependencies. Restart the bot
-or webserver after editing local code; cache identity does not reload modules.
+identifiers. Existing artifacts are not deleted. Restart the bot or webserver
+after editing local code; cache identity does not reload modules.
This outer cache is separate from FreqAI models and predictions. Invalidation
does not erase their state or reconstruct a historical environment. Native result
)
)
+ def test_dynamic_model_helper_isolates_prediction_cache(self):
+ with temporary_directory() as directory:
+ implementations = directory / "implementations"
+ implementations.mkdir()
+ helper_name = f"_qa_dynamic_settings_{directory.name.replace('-', '_')}"
+ parent_name = helper_name + "_parent"
+ self.addCleanup(sys.modules.pop, helper_name, None)
+ self.addCleanup(sys.modules.pop, parent_name, None)
+ helper = implementations / f"{helper_name}.py"
+ helper.write_text("SCALE = 1.0\n")
+ (implementations / f"{parent_name}.py").write_text(
+ "from quickadapter.user_data.freqaimodels.QuickAdapterRegressorV3 import QuickAdapterRegressorV3\n"
+ "class DynamicBaseRegressor(QuickAdapterRegressorV3):\n"
+ " def predict(self, unfiltered_df, dk, **kwargs):\n"
+ " result, status = super().predict(unfiltered_df, dk, **kwargs)\n"
+ " result['&s-extrema'] *= self.prediction_scale\n"
+ " return result, status\n"
+ )
+ (implementations / "dynamic_regressor.py").write_text(
+ "import importlib\n"
+ f"from {parent_name} import DynamicBaseRegressor\n"
+ f"settings = importlib.import_module({helper_name!r})\n"
+ "class DynamicRegressor(DynamicBaseRegressor):\n"
+ " prediction_scale = settings.SCALE\n"
+ )
+ frame = ohlcv_frame(100.0 + np.sin(np.arange(720) / 9.0) * 4.0)
+ options = {"model_name": "DynamicRegressor", "model_path": implementations}
+ first = self._run_backtest(directory / "shared", frame, **options)
+ unchanged = self._run_backtest(directory / "shared", frame, **options)
+ self.assertEqual(first.data_path.parent, unchanged.data_path.parent)
+ pd.testing.assert_frame_equal(first.return_dataframe, unchanged.return_dataframe)
+ helper.write_text("SCALE = -1.0\n")
+ sys.modules.pop(helper_name)
+ importlib.invalidate_caches()
+ changed = self._run_backtest(directory / "shared", frame, **options)
+ fresh = self._run_backtest(directory / "fresh", frame, **options)
+ repeated = self._run_backtest(directory / "shared", frame, **options)
+ self.assertNotEqual(first.data_path.parent, changed.data_path.parent)
+ self.assertEqual(changed.data_path.parent, repeated.data_path.parent)
+ pd.testing.assert_frame_equal(changed.return_dataframe, fresh.return_dataframe)
+ pd.testing.assert_frame_equal(repeated.return_dataframe, fresh.return_dataframe)
+ np.testing.assert_allclose(
+ changed.return_dataframe[LABEL], -first.return_dataframe[LABEL], equal_nan=True
+ )
+ self.assertFalse(
+ np.allclose(
+ changed.return_dataframe[LABEL],
+ first.return_dataframe[LABEL],
+ equal_nan=True,
+ )
+ )
+
def test_shifted_cache_is_recomputed_and_changed_inputs_isolate_models(self):
with temporary_directory() as directory:
position = np.arange(720)
from scripts.backtest_cache_contract import (
assert_native_implementation_cache,
assert_native_startup_predictions,
+ assert_source_tree_provenance,
)
class NativeResultCacheTest(QaTestCase):
+ def test_source_tree_provenance_preserves_snapshots_and_boundaries(self):
+ with temporary_directory() as directory:
+ assert_source_tree_provenance(self, directory)
+
def test_native_startup_and_predictions_do_not_require_test_pythonpath(self):
with temporary_directory() as directory:
assert_native_startup_predictions(
source metadata are private SDK contracts; recheck them on Freqtrade upgrades.
"""
-import ast
import json
+import os
import platform
+import site
import sys
+import sysconfig
from collections.abc import Iterator, Mapping
from hashlib import sha256
from importlib.metadata import PackageNotFoundError, version
-from importlib.util import resolve_name
-from inspect import getfile, isfunction, unwrap
+from inspect import getfile
from pathlib import Path
-from types import ModuleType
from typing import Any, Final
import freqtrade
"cloudpickle",
"TA-Lib",
"technical",
- "pandas-ta",
+ "ft-pandas-ta",
)
_REGRESSION_DEPENDENCIES: Final = ("xgboost", "lightgbm", "catboost", "ngboost", "scikit-image")
_RL_DEPENDENCIES: Final = ("gymnasium", "stable-baselines3", "sb3-contrib")
-def _import_namespaces(source: bytes, package: str) -> Iterator[dict[str, Any]]:
- """Find loaded imports even when a native-resolved leaf owns no methods."""
- for node in ast.walk(ast.parse(source)):
- if isinstance(node, ast.Import):
- names = [alias.name for alias in node.names]
- elif isinstance(node, ast.ImportFrom):
- name = node.module or ""
- if node.level:
- if not package:
- continue
- try:
- name = resolve_name("." * node.level + name, package)
- except ImportError:
- continue
- names = [name, *(f"{name}.{alias.name}" for alias in node.names)]
- else:
- continue
- for name in names:
- module = sys.modules.get(name)
- if module is not None:
- yield vars(module)
-
-
-def _class_namespaces(cls: type) -> Iterator[dict[str, Any]]:
- """Use actual function globals: native-resolved modules may be unregistered."""
- for member in vars(cls).values():
- if isinstance(member, (staticmethod, classmethod)):
- member = member.__func__
- elif isinstance(member, property):
- member = member.fget
- if callable(member):
- member = unwrap(member)
- if isfunction(member):
- yield member.__globals__
- module = sys.modules.get(cls.__module__)
- # Data-only imported classes have no function globals. Reject name collisions.
- if module is not None and vars(module).get(cls.__name__) is cls:
- yield vars(module)
+def _installed_roots() -> set[Path]:
+ """Locate interpreter libraries without excluding sibling project directories."""
+ paths = sysconfig.get_paths()
+ roots = {
+ Path(value).resolve()
+ for key in ("stdlib", "platstdlib", "purelib", "platlib")
+ if (value := paths.get(key))
+ }
+ roots.update(
+ Path(path).resolve() for path in site.getsitepackages([sys.prefix, sys.base_prefix])
+ )
+ roots.add(Path(site.getusersitepackages()).resolve())
+ roots.add(Path(freqtrade.__file__).resolve().parent)
+ return roots
+
+
+def _python_sources(root: Path, excluded: set[Path]) -> Iterator[Path]:
+ """Walk source files only; directory links cannot expand or cycle the boundary."""
+
+ def fail(error: OSError) -> None:
+ raise error
+
+ for directory, children, files in os.walk(root, onerror=fail):
+ parent = Path(directory)
+ children[:] = [
+ name
+ for name in children
+ if name not in ("__pycache__", ".git")
+ and not (child := parent / name).is_symlink()
+ and not (child / "pyvenv.cfg").is_file()
+ and not any(child.resolve().is_relative_to(path) for path in excluded)
+ ]
+ for name in files:
+ path = parent / name
+ if (
+ path.suffix == ".py"
+ and not any(path.resolve().is_relative_to(directory) for directory in excluded)
+ and path.is_file()
+ ):
+ yield path
def implementation_fingerprint(implementations: Mapping[str, type]) -> str:
- """Hash selected classes, reachable local helpers and release dependencies.
+ """Hash selected classes, bounded local Python trees and release dependencies.
- Paths locate sources but are not part of identity. Resolver snapshots take
- precedence over disk source. Local helper traversal stays inside directories
- containing selected non-SDK MRO implementations; it never scans user data or
- installed packages. Version metadata does not detect same-version SDK/binary
- modifications. File hashes are not memoized across strategy constructions.
+ Every Python file inside local implementation/MRO directories contributes,
+ including unused or dynamically imported helpers. Installed libraries never
+ create scan roots. Resolver snapshots override disk source, even through
+ file aliases. Absolute paths locate files but are not part of the identity.
"""
classes = {role: list(leaf.__mro__[:-1]) for role, leaf in implementations.items()}
- local_classes = [
- cls
- for lineage in classes.values()
- for cls in lineage
- if cls.__module__ != "abc" and not cls.__module__.startswith("freqtrade.")
- ]
- sources: dict[tuple[str, bytes], None] = {}
- source_cache: dict[tuple[str, str | Path], bytes] = {}
- snapshots: dict[Path, str] = {}
- imports: dict[tuple[bytes, str], tuple[dict[str, Any], ...]] = {}
- directories: set[Path] = set()
- namespaces: list[dict[str, Any]] = []
-
- def add_source(path: str | Path, snapshot: str | None = None) -> bytes:
- key = ("snapshot", snapshot) if snapshot is not None else ("file", Path(path).resolve())
- if key not in source_cache:
- source_cache[key] = (
- snapshot.encode() if snapshot is not None else Path(path).read_bytes()
- )
- sources[(Path(path).name, source_cache[key])] = None
- return source_cache[key]
-
- def imported_namespaces(payload: bytes, package: str) -> tuple[dict[str, Any], ...]:
- key = (payload, package)
- if key not in imports:
- imports[key] = tuple(_import_namespaces(payload, package))
- return imports[key]
-
- for lineage in classes.values():
+ excluded = _installed_roots()
+ snapshots: dict[Path, set[bytes]] = {}
+ class_sources: dict[tuple[str, int], tuple[Path, bytes | None]] = {}
+ roots: dict[Path, list[tuple[str, int, str, str]]] = {}
+ for role, lineage in classes.items():
leaf = lineage[0]
- for cls in lineage:
- if cls not in local_classes:
+ for index, cls in enumerate(lineage):
+ if cls.__module__ in ("abc", "builtins") or cls.__module__.startswith("freqtrade."):
continue
snapshot = cls.__dict__.get("__source__")
- path = cls.__dict__.get("__file__")
- if snapshot is None and path is None and cls.__module__ == leaf.__module__:
+ filename = cls.__dict__.get("__file__")
+ if snapshot is None and filename is None and cls.__module__ == leaf.__module__:
snapshot = leaf.__dict__.get("__source__")
- path = leaf.__dict__.get("__file__")
- if snapshot is None and path is None:
- path = getfile(cls)
- payload = add_source(path or cls.__module__, snapshot)
- if path is not None:
- resolved = Path(path).resolve()
- directories.add(resolved.parent)
- if snapshot is not None:
- snapshots[resolved] = snapshot
- namespaces.extend(_class_namespaces(cls))
- namespaces.extend(imported_namespaces(payload, cls.__module__.rpartition(".")[0]))
-
- visited: set[int] = set()
- while namespaces:
- namespace = namespaces.pop()
- if id(namespace) in visited:
- continue
- visited.add(id(namespace))
- filename = namespace.get("__file__")
- if filename is None:
- continue
- path = Path(filename).resolve()
- if not any(path.is_relative_to(directory) for directory in directories):
- continue
- payload = add_source(path, snapshots.get(path))
- namespaces.extend(imported_namespaces(payload, namespace.get("__package__") or ""))
- for value in tuple(namespace.values()):
- if callable(value):
- value = unwrap(value)
- if isfunction(value):
- namespaces.append(value.__globals__)
- elif isinstance(value, ModuleType):
- namespaces.append(vars(value))
- elif isinstance(value, type):
- namespaces.extend(_class_namespaces(value))
-
- add_source(__file__)
+ filename = leaf.__dict__.get("__file__")
+ if filename is None:
+ try:
+ filename = getfile(cls)
+ except TypeError as error:
+ if cls.__module__ in sys.builtin_module_names:
+ continue
+ raise OperationalException(
+ f"Backtest cache: cannot locate Python source for {cls.__qualname__}"
+ ) from error
+ path = Path(filename).resolve()
+ if any(path.is_relative_to(directory) for directory in excluded):
+ continue
+ if path.suffix != ".py":
+ raise OperationalException(
+ f"Backtest cache: Python source is required for {cls.__qualname__}"
+ )
+ payload = snapshot.encode() if snapshot is not None else None
+ class_sources[(role, index)] = (path, payload)
+ roots.setdefault(path.parent, []).append(
+ (role, index, cls.__module__, cls.__qualname__)
+ )
+ if payload is not None:
+ snapshots.setdefault(path, set()).add(payload)
+
+ source_cache: dict[Path, tuple[bytes, ...]] = {}
+ sources: set[tuple[str, bytes]] = set()
+
+ def payloads(path: Path) -> tuple[bytes, ...]:
+ path = path.resolve()
+ if path not in source_cache:
+ source_cache[path] = (
+ tuple(sorted(snapshots[path])) if path in snapshots else (path.read_bytes(),)
+ )
+ return source_cache[path]
+
+ for (role, index), (path, snapshot) in class_sources.items():
+ label = json.dumps(("class", role, index))
+ for payload in (snapshot,) if snapshot is not None else payloads(path):
+ sources.add((label, payload))
+ for root, anchors in roots.items():
+ anchor = min(anchors)
+ paths = set(_python_sources(root, excluded))
+ paths.update(path for path in snapshots if path.is_relative_to(root))
+ for path in paths:
+ label = json.dumps(("tree", anchor, path.relative_to(root).as_posix()))
+ for payload in payloads(path):
+ sources.add((label, payload))
+ sources.add(("identity-helper", Path(__file__).read_bytes()))
rl = any(
cls.__module__.startswith("freqtrade.freqai.RL.")
for lineage in classes.values()
from unittest import mock
import freqtrade
+from BacktestIdentity import implementation_fingerprint
from freqtrade.configuration import TimeRange
from freqtrade.enums import RunMode
from freqtrade.optimize.backtest_caching import get_strategy_run_id
snapshot = deepcopy(nested)
config["api_server"] = {"enabled": False, "username": "unrelated-cache-setting"}
config["telegram"] = {"enabled": False}
- (implementations / "unrelated.py").write_text("UNUSED = 123\n")
test.assertEqual(cached(load())[strategy_name]["profit_total"], 0.125)
test.assertIs(config["freqai"], nested)
test.assertEqual(nested, snapshot)
test.assertNotIn(strategy_name, cached(load()))
test.assertEqual(cached(load())[strategy_name]["profit_total"], 0.125)
- def upgraded_dependency(name):
- return "999.0+cache-test" if name == "torch" else distribution_version(name)
+ for dependency in ("torch", "ft-pandas-ta"):
+ with test.subTest(dependency=dependency):
- with mock.patch("BacktestIdentity.version", side_effect=upgraded_dependency):
- test.assertNotIn(strategy_name, cached(load()))
- test.assertEqual(cached(load())[strategy_name]["profit_total"], 0.125)
+ def upgraded_dependency(name, selected_dependency=dependency):
+ return (
+ "999.0+cache-test"
+ if name == selected_dependency
+ else distribution_version(name)
+ )
+
+ with mock.patch("BacktestIdentity.version", side_effect=upgraded_dependency):
+ test.assertNotIn(strategy_name, cached(load()))
+ test.assertEqual(cached(load())[strategy_name]["profit_total"], 0.125)
helper.write_text("VALUE = -1\ndef coefficient():\n return VALUE\n")
importlib.invalidate_caches()
test.assertNotIn(strategy_name, cached(load()))
"VALUE = 0.5\n@lru_cache(None)\ndef coefficient():\n return VALUE\n"
)
test.assertNotIn(strategy_name, cached(load()))
+ for imports, ancestor in (
+ ("from sklearn.base import BaseEstimator", "BaseEstimator"),
+ ("from typing import Generic, TypeVar\nT = TypeVar('T')", "Generic[T]"),
+ ):
+ with test.subTest(external_ancestor=ancestor):
+ model_path.write_text(
+ f"{model_import}\n{imports}\n"
+ f"class CacheModel(CacheBase, {ancestor}):\n"
+ " def __init__(self, *args, **kwargs):\n"
+ " raise AssertionError('Identity must not construct a model')\n"
+ )
+ store(load())
+ test.assertEqual(cached(load())[strategy_name]["profit_total"], 0.125)
+ with test.subTest(helper_import="dynamic", leaf_methods=False):
+ helper.write_text("VALUE = 1\n")
+ sys.modules.pop(helper_name, None)
+ model_path.write_text(
+ "import importlib\n"
+ f"from {base_name} import CacheParent\n"
+ f"settings = importlib.import_module({helper_name!r})\n"
+ "class CacheModel(CacheParent):\n"
+ " cache_scale = settings.VALUE\n"
+ )
+ store(load())
+ test.assertEqual(cached(load())[strategy_name]["profit_total"], 0.125)
+ helper.write_text("VALUE = 2\n")
+ sys.modules.pop(helper_name, None)
+ importlib.invalidate_caches()
+ test.assertNotIn(strategy_name, cached(load()))
+ with test.subTest(source_boundary="conservative"):
+ store(load())
+ extra = implementations / "extra.py"
+ extra.write_text("unused Python need not parse!\n")
+ test.assertNotIn(strategy_name, cached(load()))
+ store(load())
+ extra.write_text("UNUSED = 42\n")
+ test.assertNotIn(strategy_name, cached(load()))
+ store(load())
+ renamed = implementations / "renamed.py"
+ extra.rename(renamed)
+ test.assertNotIn(strategy_name, cached(load()))
+ renamed.unlink()
+ marker = implementations / "marker.py"
+ marker.write_text("UNUSED_MARKER = True\n")
+ extra.write_text("UNUSED = 42\n")
+ store(load())
+ extra.unlink()
+ test.assertNotIn(strategy_name, cached(load()))
+ marker.unlink()
+ store(load())
+ (implementations / "model.joblib").write_bytes(b"generated model")
+ (implementations / "metadata.json").write_text('{"generated": true}\n')
+ test.assertEqual(cached(load())[strategy_name]["profit_total"], 0.125)
+
+
+def assert_source_tree_provenance(test: Any, directory: Path) -> None:
+ """Source location, aliases and resolver snapshots must not corrupt cache identity."""
+ source = "class Selected:\n pass\n"
+ implementations = []
+ for name in ("first", "relocated"):
+ root = directory / name
+ root.mkdir()
+ path = root / "selected.py"
+ path.write_text(source)
+ implementations.append(
+ type(
+ "Selected",
+ (),
+ {"__module__": "_cache_provenance", "__file__": str(path), "__source__": source},
+ )
+ )
+ selected, relocated = implementations
+ fingerprint = implementation_fingerprint({"model": selected})
+ test.assertEqual(fingerprint, implementation_fingerprint({"model": relocated}))
+ root = Path(selected.__file__).parent
+ path = root / "selected.py"
+ path.write_text("class Selected:\n changed_on_disk = True\n")
+ test.assertEqual(fingerprint, implementation_fingerprint({"model": selected}))
+ path.unlink()
+ test.assertEqual(fingerprint, implementation_fingerprint({"model": selected}))
+ path.write_text(source)
+ changed = type(
+ "Selected",
+ (),
+ {
+ "__module__": "_cache_provenance",
+ "__file__": str(path),
+ "__source__": source.replace("pass", "changed_snapshot = True"),
+ },
+ )
+ ordered = implementation_fingerprint({"model": selected, "strategy": changed})
+ test.assertEqual(ordered, implementation_fingerprint({"strategy": changed, "model": selected}))
+ test.assertNotEqual(
+ ordered, implementation_fingerprint({"model": changed, "strategy": selected})
+ )
+ packages = [root / name for name in ("one", "two")]
+ for package, value in zip(packages, (1, 2), strict=True):
+ package.mkdir()
+ (package / "settings.py").write_text(f"VALUE = {value}\n")
+ before_swap = implementation_fingerprint({"model": selected})
+ for package, value in zip(packages, (2, 1), strict=True):
+ (package / "settings.py").write_text(f"VALUE = {value}\n")
+ test.assertNotEqual(before_swap, implementation_fingerprint({"model": selected}))
+ external = directory / "external"
+ external.mkdir()
+ helper = external / "helper.py"
+ helper.write_text("VALUE = 1\n")
+ (root / "linked.py").symlink_to(helper)
+ (root / "cycle").symlink_to(root, target_is_directory=True)
+ (root / "outside").symlink_to(external, target_is_directory=True)
+ before_link_change = implementation_fingerprint({"model": selected})
+ helper.write_text("VALUE = 2\n")
+ after_link_change = implementation_fingerprint({"model": selected})
+ test.assertNotEqual(before_link_change, after_link_change)
+ (external / "unrelated.py").write_text("VALUE = 3\n")
+ test.assertEqual(after_link_change, implementation_fingerprint({"model": selected}))
def assert_native_startup_predictions(
repositories / freqai-strategies.git / commitdiff
