context: ReforceXY
project: reforcexy
image: freqai-strategies-reforcexy-qa:ci
- coverage: false
+ coverage: true
steps:
- name: Check out repository
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
--entrypoint sh \
"${{ matrix.image }}" \
/workspace/scripts/run-coverage.sh -s "${{ matrix.context }}/tests" -v
+ - name: Run runtime regressions under three orderings
+ if: matrix.context == 'ReforceXY'
+ run: |
+ docker run --rm \
+ --mount "type=bind,src=${GITHUB_WORKSPACE},dst=/workspace,readonly" \
+ --workdir /workspace \
+ --entrypoint sh \
+ "${{ matrix.image }}" \
+ -c 'for seed in 1 2 3; do
+ FREQAI_QA_SHUFFLE_SEED=$seed python -m unittest discover -s ReforceXY/tests || exit 1
+ done'
reward-space-analysis:
name: Reward space analysis
The shared runtime step sets each strategy's `PYTHONPATH`. QuickAdapter needs
this for direct `unittest` discovery because its model imports `LabelTransformer`
and `Utils` by bare names; ReforceXY resolves its imports without it, so the
-setting is optional there. QuickAdapter's regressions additionally run under
+setting is optional there. Both strategies' regressions run under
`coverage.py`; the reward-space analysis suite runs separately with `uv`,
without a Freqtrade image.
### Coverage gate
-Within the strategy QA matrix, QuickAdapter is the only strategy whose runtime
-regressions enforce a coverage gate. The standalone reward-space analysis suite
-has its own gate; see [its testing documentation](ReforceXY/reward_space_analysis/tests/README.md).
-QuickAdapter's configuration is `quickadapter/.coveragerc`, selected explicitly
-because coverage.py looks for `.coveragerc` in the directory it is run from and
-does not search parents:
+Within the strategy QA matrix, QuickAdapter and ReforceXY both enforce a
+coverage gate. The standalone reward-space analysis suite has its own gate; see
+[its testing documentation](ReforceXY/reward_space_analysis/tests/README.md).
+Each strategy's configuration is that strategy's `.coveragerc`, selected
+explicitly because coverage.py looks for `.coveragerc` in the directory it is run
+from and does not search parents:
```shell
export PYTHONPATH=quickadapter/user_data/strategies
the `user_data` trees would be absent from the report and the gate would
measure whatever happened to be imported.
+```shell
+# ReforceXY
+export COVERAGE_RCFILE=ReforceXY/.coveragerc COVERAGE_FILE=/tmp/.coverage
+
+sh scripts/run-coverage.sh -s ReforceXY/tests -v
+```
+
+Some of the measured tree cannot be reached by a test, and the total therefore
+has a ceiling below 100 %. Six sites are recorded here rather than excluded with
+`omit`, which the floor guard bans — an exclusion key would shrink the
+denominator without the trade-off being visible:
+
+- **Unreachable while the runner is CPU-only.** `_configure_gpu_memory:757-767`
+ (`:751-756` returns before the out-of-range arm is evaluated) and
+ `MyRLEnv.close:4616-4617` (`th.cuda.empty_cache()`). A GPU runner would make
+ both testable.
+- **Version-dependent.** `ReforceXY.py:307`'s `_repair_historic_predictions is
+ None` branch, arc `307→316`; `:309-315` itself is covered. A Freqtrade
+ revision adding the hook would make it reachable.
+- **Dead at any revision.** The `except OverflowError` arm of
+ `_potential_transform` (`:3399-3400`); `:249`, the `raise RuntimeError` for an
+ asynchronous drawer method, guarded by `iscoroutinefunction` rather than by the
+ version probe; and `create_sampler`'s `case _: assert_never(sampler)`
+ (`:2244-2245`), since `_Samplers` has exactly two members and `:2218` raises
+ for anything else.
+
`fail_under` is an absolute floor, never per-module. To change it, re-measure
with the shipped configuration already in place — a measurement taken without
`branch` and `include_namespace_packages` reports a different denominator and
is not a valid input — then update the value and the `# measured` annotation
above it in the same commit. Raise the floor only; a drop needs the reason in
-the pull request. `quickadapter/tests/test_coverage_floor.py` refuses a
+the pull request. `quickadapter/tests/test_coverage_floor.py` and
+`ReforceXY/tests/test_coverage_floor.py` refuse a
placeholder, a missing or undated measurement annotation, a measurement below
the floor it justifies, a floor below the current minimum, a floor that is not a
percentage, a `precision` coarse enough to round the total past the floor, a
dependency versions remain exact.
The BasedPyright and type-stub versions are pinned in each project's
-`.devcontainer/requirements-dev.txt`, as is `coverage` in QuickAdapter's.
+`.devcontainer/requirements-dev.txt`, as is `coverage` in both QA images.
The Freqtrade base images intentionally follow their rolling `stable_freqai`
and `stable_freqairl` tags, so record the resolved image digests when a
reproducible audit is required.
--- /dev/null
+[run]
+branch = True
+source =
+ ReforceXY/user_data
+relative_files = True
+
+[report]
+show_missing = True
+precision = 1
+include_namespace_packages = True
+# measured 2026-10-02 at 57.1% — raise only, never lower
+fail_under = 57
\ No newline at end of file
uv
ruff
basedpyright==1.40.1
+coverage==7.16.2
--- /dev/null
+"""Shared builders and process-state resets for the ReforceXY test suite."""
+
+import copy
+import hashlib
+import importlib
+import os
+import random
+import unittest
+from collections.abc import Iterator
+from contextlib import contextmanager
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from typing import Any
+
+import numpy as np
+import pandas as pd
+from freqtrade.enums import RunMode
+
+PAIR = "BTC/USDT"
+REPO_ROOT = Path(__file__).resolve().parents[2]
+PROJECT_ROOT = REPO_ROOT / "ReforceXY"
+COVERAGERC = PROJECT_ROOT / ".coveragerc"
+TESTS_ROOT = PROJECT_ROOT / "tests"
+MEASURED_TREE = PROJECT_ROOT / "user_data"
+PRODUCTION_MODULE = "ReforceXY.user_data.freqaimodels.ReforceXY"
+STRATEGY_MODULE = "ReforceXY.user_data.strategies.RLAgentStrategy"
+
+# Never handed out: every call deep-copies it, so a test that mutates the returned
+# config cannot leak into the next one.
+_BASE_CONFIG: dict[str, Any] = {
+ "timeframe": "5m",
+ "stake_amount": "unlimited",
+ "exchange": {"pair_whitelist": [PAIR]},
+ "freqai": {
+ "enabled": True,
+ "identifier": "reforcexy-runtime-regression",
+ "train_period_days": 1,
+ "backtest_period_days": 1,
+ "conv_width": 1,
+ "activate_tensorboard": False,
+ "feature_parameters": {
+ "include_timeframes": ["5m"],
+ "include_corr_pairlist": [],
+ "label_period_candles": 1,
+ "principal_component_analysis": False,
+ "noise_standard_deviation": 0,
+ "buffer_train_data_candles": 0,
+ "shuffle_after_split": False,
+ },
+ "data_split_parameters": {"test_size": 0.25, "shuffle": False},
+ "model_training_parameters": {
+ "n_steps": 8,
+ "batch_size": 8,
+ "n_epochs": 1,
+ "device": "cpu",
+ "policy_kwargs": {"net_arch": [8]},
+ },
+ "rl_config": {
+ "model_type": "MaskablePPO",
+ "policy_type": "MlpPolicy",
+ "cpu_count": 1,
+ "drop_ohlc_from_features": False,
+ "model_reward_parameters": {"rr": 2.0, "profit_aim": 0.03},
+ "train_cycles": 1,
+ "n_envs": 1,
+ "n_eval_envs": 1,
+ "n_eval_steps": 16,
+ "n_eval_episodes": 1,
+ "check_envs": False,
+ "add_state_info": False,
+ },
+ # `create_sampler` resolves `auto` through optunahub.load_module, which is an
+ # outbound fetch to hub.optuna.org, so the sampler is pinned for every test. The
+ # pin is only half the guarantee: `create_sampler` also defaults to "tpe" and
+ # raises for anything outside `_Samplers`, so no config value reaches the network.
+ # There is no `pruner` key: `create_pruner` is a staticmethod taking three ints.
+ "rl_config_optuna": {"enabled": False, "sampler": "tpe"},
+ },
+}
+
+
+def _merge(base: dict[str, Any], overrides: dict[str, Any]) -> dict[str, Any]:
+ merged = copy.deepcopy(base)
+ for key, value in overrides.items():
+ current = merged.get(key)
+ if isinstance(current, dict) and isinstance(value, dict):
+ merged[key] = _merge(current, value)
+ else:
+ merged[key] = copy.deepcopy(value)
+ return merged
+
+
+def model_config(tmp_path: Path | str, **overrides: Any) -> dict[str, Any]:
+ """Return a private Freqtrade config rooted at `tmp_path`, deep-merged with `overrides`.
+
+ `runmode` is applied only when the caller supplied none, so a test can ask for
+ RunMode.BACKTEST and get it. QuickAdapter assigns it after the merge and is therefore
+ not overridable; copying that order here would silently hand DRY_RUN to the backtest
+ regressions.
+ """
+ config = _merge(_BASE_CONFIG, overrides)
+ config["user_data_dir"] = Path(config.get("user_data_dir", tmp_path))
+ config.setdefault("runmode", RunMode.DRY_RUN)
+ return config
+
+
+@contextmanager
+def temporary_directory() -> Iterator[Path]:
+ """Yield a temporary directory that is removed on exit. The only call shape."""
+ with TemporaryDirectory() as path:
+ yield Path(path)
+
+
+def ohlcv_frame(closes, volume: float = 10.0, spread: float = 0.001) -> pd.DataFrame:
+ """Return an OHLCV frame whose close is `closes`, bracketed by a relative `spread`."""
+ values = np.asarray(closes, dtype=float)
+ count = values.size
+ spreads = np.broadcast_to(np.asarray(spread, dtype=float), (count,))
+ volumes = np.broadcast_to(np.asarray(volume, dtype=float), (count,))
+ return pd.DataFrame(
+ {
+ "open": values,
+ "high": values * (1.0 + spreads),
+ "low": values * (1.0 - spreads),
+ "close": values,
+ "volume": volumes,
+ "date": pd.date_range("2024-01-01", periods=count, freq="5min"),
+ }
+ )
+
+
+class RecordingPolicy:
+ """A stand-in for an SB3 policy that records what it was asked to predict."""
+
+ def __init__(self):
+ self.observations = []
+ self.masks = []
+
+ def predict(self, observation, **kwargs):
+ self.observations.append(observation.copy())
+ self.masks.append(kwargs.get("action_masks"))
+ return np.array([0]), None
+
+
+def rl_env(
+ prices,
+ *,
+ reward_kwargs: dict[str, Any] | None = None,
+ fee: float = 0.0015,
+ can_short: bool = True,
+ config: dict[str, Any] | None = None,
+ window_size: int = 1,
+ index=None,
+):
+ """Return a reset `MyRLEnv` over a one-column price frame.
+
+ `reward_kwargs` and `config` reach `MyRLEnv.__init__` verbatim; callers own every
+ value they pass, because the defaults here are the ones the shipped config uses and a
+ test that needs a different one must say so.
+ """
+ from ReforceXY.user_data.freqaimodels.ReforceXY import MyRLEnv
+
+ frame = pd.DataFrame({"open": list(prices)}, index=index)
+ env_config = {
+ "stake_amount": "unlimited",
+ "freqai": {
+ "rl_config": {
+ "add_state_info": False,
+ "max_training_drawdown_pct": 0.99,
+ "model_reward_parameters": {},
+ }
+ },
+ }
+ if config is not None:
+ env_config = _merge(env_config, config)
+ env = MyRLEnv(
+ df=frame.copy(),
+ prices=frame,
+ df_raw=frame.copy(),
+ window_size=window_size,
+ reward_kwargs=reward_kwargs
+ if reward_kwargs is not None
+ else {"rr": 2.0, "profit_aim": 0.03},
+ fee=fee,
+ can_short=can_short,
+ config=env_config,
+ live=True,
+ )
+ env.reset()
+ return env
+
+
+# --- test-order independence ------------------------------------------------------
+#
+# `unittest.TestLoader.sortTestMethodsUsing` is a staticmethod ON THE LOADER; a
+# TestCase has no such hook. Assigning a plain function would bind it as a method and
+# raise TypeError at discovery, so the descriptor is explicit. The comparator is a pure
+# function of (seed, name): a fresh random sign per comparison is not a transitive
+# order, and one shared RNG would make each class's order depend on how many classes
+# were loaded before it. With no seed it falls through to name order, which reproduces
+# alphabetical discovery exactly, so the patch is unconditional and every invocation is
+# covered.
+_SEED = os.getenv("FREQAI_QA_SHUFFLE_SEED", "")
+
+
+def _shuffle_cmp(a: str, b: str) -> int:
+ """Order test method names by a seeded hash, or by name when no seed is set."""
+ if not _SEED:
+ return (a > b) - (a < b)
+ ka = int.from_bytes(hashlib.blake2b(f"{_SEED}:{a}".encode(), digest_size=8).digest())
+ kb = int.from_bytes(hashlib.blake2b(f"{_SEED}:{b}".encode(), digest_size=8).digest())
+ return (ka > kb) - (ka < kb)
+
+
+unittest.TestLoader.sortTestMethodsUsing = staticmethod(_shuffle_cmp)
+
+
+# --- process globals --------------------------------------------------------------
+#
+# Captured once, after the module's import-time registrations, so a test that mutates one
+# is undone without destroying the real ones.
+_PRODUCTION = importlib.import_module(PRODUCTION_MODULE)
+_ACTION_MASKS_CACHE_AT_IMPORT = dict(_PRODUCTION.ReforceXY._action_masks_cache)
+_RANDOM_STATE_AT_IMPORT = random.getstate()
+_NUMPY_STATE_AT_IMPORT = np.random.get_state()
+_TORCH_STATE_AT_IMPORT = None
+try: # torch is present in the RL QA image; a CPU-only box may still lack it
+ import torch
+
+ _TORCH_STATE_AT_IMPORT = torch.random.get_rng_state()
+except ImportError: # pragma: no cover - the RL image always provides torch
+ torch = None # type: ignore[assignment]
+
+
+class QaTestCase(unittest.TestCase):
+ """Base case restoring the four process globals, in setUp and again in addCleanup.
+
+ `_action_masks_cache` is a pure memoisation: it is written from a value derived
+ deterministically from the `(can_short, position)` key and read only through
+ `ReforceXY._action_masks_cache`, so clearing it cannot change any returned mask.
+ The RNG states matter because `_get_train_and_eval_environments` calls the
+ process-wide `set_random_seed`, so without a restore any stochastic assertion would
+ depend on test order.
+ """
+
+ def setUp(self):
+ self._restore_process_globals()
+ self.addCleanup(self._restore_process_globals)
+
+ @staticmethod
+ def _restore_process_globals() -> None:
+ _PRODUCTION.ReforceXY._action_masks_cache.clear()
+ _PRODUCTION.ReforceXY._action_masks_cache.update(_ACTION_MASKS_CACHE_AT_IMPORT)
+ random.setstate(_RANDOM_STATE_AT_IMPORT)
+ np.random.set_state(_NUMPY_STATE_AT_IMPORT)
+ if torch is not None:
+ torch.random.set_rng_state(_TORCH_STATE_AT_IMPORT)
--- /dev/null
+"""Guards the coverage gate itself; requires the Freqtrade QA image."""
+
+import configparser
+import os
+import re
+import subprocess
+import sys
+import unittest
+from pathlib import Path
+
+from coverage.config import DEFAULT_EXCLUDE, DEFAULT_PARTIAL, DEFAULT_PARTIAL_ALWAYS, CoverageConfig
+from coverage.files import find_python_files
+from coverage.misc import join_regex
+from coverage.parser import PythonParser
+from qa_support import COVERAGERC, MEASURED_TREE, REPO_ROOT, QaTestCase, temporary_directory
+
+FLOOR = "fail_under"
+PLACES = "precision"
+BRANCH = "branch"
+NAMESPACE_PACKAGES = "include_namespace_packages"
+OMIT = "omit"
+INCLUDE = "include"
+# This is the floor ReforceXY has actually reached, not the target. The gate is only as
+# strong as the measurement behind it, so `MINIMUM_FLOOR` starts at the calibrated value
+# and may only rise. Reaching the agreed 70% target is a separate delivery; until then a
+# lower value here would be a control the project believes it has and does not.
+MINIMUM_FLOOR = 57.0
+EXCLUSION_KEYS = (
+ OMIT,
+ INCLUDE,
+ "patch",
+ "exclude_lines",
+ "exclude_also",
+ "partial_branches",
+ "partial_also",
+ "partial_branches_always",
+)
+COVERAGERC_SECTIONS = ["report", "run"]
+COVERAGERC_SECTION_MESSAGE = (
+ "coverage accepts a [coverage:run] section form; the bare parser used by this class sees it "
+ "literally, so every other guard would KeyError or read an empty section"
+)
+COVERAGE_RUNNER = REPO_ROOT / "scripts" / "run-coverage.sh"
+
+
+def _branch_counts(parser: PythonParser) -> dict[int, int]:
+ """Exactly what Analysis._total_branches() sums — coverage/results.py:148-150.
+
+ `arcs()` is built from the AST alone and never consults `excluded`, so an arc-set
+ comparison on it is identically green in every scenario. `exit_counts()` is that set
+ with excluded endpoints dropped, and the raw dict differs at count-1 entries, so the
+ `> 1` filter is both necessary and sufficient.
+ """
+ return {line: n for line, n in parser.exit_counts().items() if n > 1}
+
+
+def _legal_options(section: str) -> set[str]:
+ """Option names coverage binds in `section`, read from its own option table.
+
+ `spec[1]` is `"section:option"`; `spec[0]` is the ATTRIBUTE name and differs for five
+ of the eight exclusion keys (`run_include` vs `include`, `exclude_list` vs
+ `exclude_lines`, …). Reading `spec[0]` would falsely red any legal `omit`.
+ """
+ return {
+ spec[1].split(":")[1]
+ for spec in CoverageConfig.CONFIG_FILE_OPTIONS
+ if spec[1].startswith(f"{section}:") and spec[1] != "run:_crash"
+ }
+
+
+class CoverageFloorTest(QaTestCase):
+ def setUp(self):
+ super().setUp()
+ self.text = COVERAGERC.read_text()
+ self.parser = configparser.ConfigParser()
+ self.parser.read_string(self.text)
+
+ def _raw_floor(self) -> str:
+ """Extract fail_under by whole-line partitioning, not by regex.
+
+ A regex such as ^fail_under\\s*=\\s*(\\S+) pulls the leading digits out of
+ `fail_under = 10 # measured ...` and passes, so it cannot see the
+ malformation that actually breaks the run. Whole-line partitioning yields the
+ comment too, and float() then rejects it, which is the behaviour we want.
+ """
+ for line in self.text.splitlines():
+ key, separator, value = line.partition("=")
+ if separator and key.strip() == FLOOR:
+ return value.strip()
+ raise AssertionError(f"{FLOOR} is absent from {COVERAGERC}")
+
+ def test_a_the_config_declares_exactly_the_run_and_report_sections(self):
+ # Runs first alphabetically among the `test_…` names, because every other method in
+ # this class reaches `self.parser` and a `[coverage:run]`-prefixed file makes that
+ # access raise before any assertion here could report the real cause.
+ self.assertEqual(
+ sorted(self.parser.sections()),
+ COVERAGERC_SECTIONS,
+ COVERAGERC_SECTION_MESSAGE,
+ )
+
+ def test_the_placeholder_is_substituted(self):
+ self.assertNotIn("__FLOOR__", self.text, "the coverage floor is still a placeholder")
+
+ def test_the_floor_is_a_percentage_in_range(self):
+ value = float(self._raw_floor())
+ self.assertGreater(value, 0.0)
+ self.assertLessEqual(value, 100.0)
+
+ def test_the_floor_carries_its_provenance(self):
+ # The measured date is what tells a reader whether the number is current, and
+ # it is the only thing distinguishing a real floor from a provisional one.
+ lines = [line.strip() for line in self.text.splitlines()]
+ index = next(i for i, line in enumerate(lines) if line.startswith(f"{FLOOR} ="))
+ annotation = " ".join(lines[max(0, index - 2) : index])
+ self.assertIn("measured", annotation, f"the line above {FLOOR} must record the measurement")
+ self.assertNotIn("PROVISIONAL", annotation)
+
+ def test_branch_tracing_is_enabled(self):
+ # Without it, every guard — an early return, a raise — counts as covered the
+ # moment its `if` is evaluated, and the gate measures nothing useful.
+ self.assertTrue(self.parser.getboolean("run", BRANCH))
+
+ def test_unexecuted_files_stay_in_the_denominator(self):
+ self.assertTrue(self.parser.getboolean("report", NAMESPACE_PACKAGES))
+
+ def test_precision_is_fine_enough_for_the_gate_to_compare_what_it_prints(self):
+ # coverage's predicate is `round(total, precision) < fail_under`, so at 0 a run
+ # up to half a point BELOW the floor passes, and the same divergence reaches the
+ # CLI exit code.
+ precision = self.parser.getint("report", PLACES)
+ self.assertGreaterEqual(precision, 1)
+
+ def test_the_floor_is_bound_in_the_section_coverage_reads_it_from(self):
+ # coverage binds fail_under as `report:fail_under` ONLY. The same key under [run]
+ # is silently inert, which would leave the gate dead while the value, the
+ # provenance and the placeholder checks all still passed.
+ self.assertIn(FLOOR, self.parser["report"])
+ self.assertNotIn(FLOOR, self.parser["run"])
+
+ def test_no_production_code_is_excluded_by_any_other_key(self):
+ # `omit` was only the first way to shrink the denominator: `exclude_lines`,
+ # `exclude_also` and the partial_* family are the same lever under other names.
+ # An UNRECOGNISED key only warns (`config.py:349-354`), so a misplaced key is
+ # silently inert rather than fatal — the danger is an unnoticed no-op.
+ for section in COVERAGERC_SECTIONS:
+ for key in EXCLUSION_KEYS:
+ with self.subTest(section=section, key=key):
+ self.assertNotIn(key, self.parser[section])
+
+ def test_every_present_option_is_a_legal_name_for_its_section(self):
+ # The eight-key absence assertion cannot see a TYPO — `exclude_line` is not
+ # `exclude_lines` — and a typo is silently inert. Reading the legal names from
+ # coverage's own option table also makes the guard self-update on a rename.
+ for section in COVERAGERC_SECTIONS:
+ legal = _legal_options(section)
+ for key in self.parser[section]:
+ with self.subTest(section=section, key=key):
+ self.assertIn(
+ key,
+ legal,
+ "coverage would warn and ignore this option, so the gate would be "
+ "weaker than it reads",
+ )
+
+ def test_the_floor_is_never_lowered(self):
+ # `.coveragerc` and README.md:545 both say "raise only, never lower", but nothing
+ # enforced it. Raising the floor stays free, which is exactly the direction the
+ # policy allows.
+ self.assertGreaterEqual(float(self._raw_floor()), MINIMUM_FLOOR)
+
+ def test_the_provenance_records_a_date_and_a_measurement_that_clears_the_floor(self):
+ # Recency is deliberately NOT asserted: refusing a date older than N months would
+ # make this suite fail on a repository nobody has touched. What is enforceable is
+ # that the annotation records a real measurement and that the measurement clears
+ # the floor it justifies.
+ lines = [line.strip() for line in self.text.splitlines()]
+ index = next(i for i, line in enumerate(lines) if line.startswith(f"{FLOOR} ="))
+ annotation = " ".join(lines[max(0, index - 2) : index])
+ with self.subTest(requirement="a well-formed measurement date"):
+ self.assertRegex(annotation, r"\d{4}-\d{2}-\d{2}")
+ with self.subTest(requirement="a recorded measurement"):
+ found = re.search(r"measured[^%]*?([0-9]+(?:\.[0-9]+)?)%", annotation)
+ self.assertIsNotNone(found, f"no percentage recorded above {FLOOR}: {annotation!r}")
+ self.assertGreaterEqual(
+ float(found.group(1)),
+ float(self._raw_floor()),
+ "the annotated measurement must clear the floor it justifies",
+ )
+
+ def test_no_production_source_widens_coverage_s_default_exclusions(self):
+ # The SOURCE side of the same lever. coverage applies DEFAULT_EXCLUDE by
+ # regex-searching the raw file text, so a pragma inside a string literal removes a
+ # real statement, and `while True:` / `if True:` remove branch units — 27% of this
+ # denominator — with no config key reachable at all. Parsing twice, once with the
+ # defaults and once without, is exact by construction: coverage applies them only
+ # when `self.exclude` is truthy, so the difference is the lever and nothing else.
+ measured = {Path(p) for p in find_python_files(str(MEASURED_TREE), True)}
+ self.assertEqual(
+ measured,
+ set(MEASURED_TREE.rglob("*.py")),
+ "the guard must audit exactly the files coverage measures; find_python_files also "
+ "admits .pyw and rejects odd-character names, so a future one turns this red with "
+ "no coverage defect",
+ )
+ for path in sorted(measured):
+ baseline = PythonParser(filename=str(path))
+ baseline.parse_source()
+ guarded = PythonParser(filename=str(path), exclude=join_regex(DEFAULT_EXCLUDE))
+ guarded.parse_source()
+ with self.subTest(module=path.name):
+ self.assertEqual(
+ set(baseline.statements),
+ set(guarded.statements),
+ "coverage's default exclusions remove these statements with no config key",
+ )
+ self.assertEqual(
+ _branch_counts(baseline),
+ _branch_counts(guarded),
+ "coverage's default exclusions remove these branch units with no config key",
+ )
+ # Asserted on `baseline`: both parsers read the file identically, so naming
+ # `guarded` would imply a dependency that does not exist. lines_matching is
+ # not token-aware, so a comment mentioning `while True:` fails this too —
+ # stricter is the safe direction, and it is documented rather than discovered.
+ self.assertEqual(
+ set(),
+ baseline.lines_matching(join_regex(DEFAULT_PARTIAL + DEFAULT_PARTIAL_ALWAYS)),
+ "a branch-level default exclusion matches here",
+ )
+
+ def test_the_config_is_the_one_coverage_resolves(self):
+ # coverage looks for `.coveragerc` in the directory it is run from and does not
+ # search parents, so this runs from the strategy directory with the variable
+ # emptied (`config.py:652-653` tests it for truthiness, so "" is unset).
+ result = subprocess.run(
+ [sys.executable, "-m", "coverage", "debug", "config"],
+ cwd=COVERAGERC.parent,
+ env={**os.environ, "COVERAGE_RCFILE": ""},
+ text=True,
+ capture_output=True,
+ check=False,
+ timeout=60,
+ )
+ found = re.search(r"^\s*config_file:[ \t]*(.+?)[ \t]*$", result.stdout, re.MULTILINE)
+ self.assertIsNotNone(found, result.stdout + result.stderr)
+ self.assertEqual(COVERAGERC.resolve(), Path(found.group(1)).resolve())
+
+ def _run_coverage(self, floor, *, tests_pass=True):
+ with temporary_directory() as directory:
+ source = directory / "source"
+ tests = directory / "tests"
+ source.mkdir()
+ tests.mkdir()
+ (source / "sample.py").write_text(
+ "def choose(flag):\n if flag:\n return 1\n return 2\n"
+ )
+ expected = 1 if tests_pass else 2
+ (tests / "test_sample.py").write_text(
+ "import unittest\nfrom sample import choose\n"
+ "class SampleTest(unittest.TestCase):\n"
+ " def test_one_branch(self):\n"
+ f" self.assertEqual(choose(True), {expected})\n"
+ )
+ rcfile = directory / ".coveragerc"
+ rcfile.write_text(
+ f"[run]\nbranch = True\nsource = {source}\n[report]\nfail_under = {floor}\n"
+ )
+ env = {
+ **os.environ,
+ "PYTHONPATH": str(source),
+ "COVERAGE_RCFILE": str(rcfile),
+ "COVERAGE_FILE": str(directory / ".coverage"),
+ }
+ return subprocess.run(
+ ["sh", str(COVERAGE_RUNNER), "-s", str(tests), "-v"],
+ cwd=directory,
+ env=env,
+ text=True,
+ capture_output=True,
+ check=False,
+ timeout=60,
+ )
+
+ def test_the_runner_propagates_the_actual_coverage_report_status(self):
+ # The same passing suite covers only one branch: 100 fails, 50 passes.
+ for floor, expected_status in ((100, 2), (50, 0)):
+ with self.subTest(floor=floor):
+ result = self._run_coverage(floor)
+ self.assertEqual(result.returncode, expected_status, result.stdout + result.stderr)
+
+ def test_the_runner_preserves_a_test_failure_even_above_the_coverage_floor(self):
+ result = self._run_coverage(50, tests_pass=False)
+ self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
+
+ def test_the_source_is_the_measured_tree(self):
+ self.assertEqual(
+ [line.strip() for line in self.parser["run"]["source"].splitlines() if line.strip()],
+ ["ReforceXY/user_data"],
+ )
+
+
+if __name__ == "__main__":
+ unittest.main()
import numpy as np
import pandas as pd
+from qa_support import QaTestCase
from ReforceXY.user_data.freqaimodels.ReforceXY import Actions, MyRLEnv, ReforceXY
-class PortfolioReturnsTest(unittest.TestCase):
+class PortfolioReturnsTest(QaTestCase):
def make_env(self, prices, *, compound=True, fee=0.0015, price_index=None):
frame = pd.DataFrame({"open": prices}, index=price_index)
env = MyRLEnv(
import copy
import math
import tempfile
-import unittest
from datetime import datetime as dt
from datetime import timedelta, timezone
from pathlib import Path
import numpy as np
import pandas as pd
-from freqtrade.enums import RunMode
from freqtrade.freqai.data_drawer import FreqaiDataDrawer
from freqtrade.freqai.data_kitchen import FreqaiDataKitchen
+from qa_support import QaTestCase, RecordingPolicy, model_config
from sb3_contrib import MaskablePPO
from ReforceXY.user_data.freqaimodels.ReforceXY import (
from ReforceXY.user_data.strategies.RLAgentStrategy import RLAgentStrategy
-class RecordingPolicy:
- def __init__(self):
- self.observations = []
- self.masks = []
-
- def predict(self, observation, **kwargs):
- self.observations.append(observation.copy())
- self.masks.append(kwargs.get("action_masks"))
- return np.array([0]), None
-
-
-def model_config(path):
- return {
- "user_data_dir": Path(path),
- "timeframe": "5m",
- "stake_amount": "unlimited",
- "runmode": RunMode.DRY_RUN,
- "exchange": {"pair_whitelist": ["BTC/USDT"]},
- "freqai": {
- "enabled": True,
- "identifier": "contract-test",
- "train_period_days": 1,
- "backtest_period_days": 1,
- "conv_width": 1,
- "activate_tensorboard": False,
- "feature_parameters": {
- "include_timeframes": ["5m"],
- "include_corr_pairlist": [],
- "label_period_candles": 1,
- "principal_component_analysis": False,
- "noise_standard_deviation": 0,
- "buffer_train_data_candles": 0,
- "shuffle_after_split": False,
- },
- "data_split_parameters": {"test_size": 0.25, "shuffle": False},
- "model_training_parameters": {
- "n_steps": 8,
- "batch_size": 8,
- "n_epochs": 1,
- "device": "cpu",
- "policy_kwargs": {"net_arch": [8]},
- },
- "rl_config": {
- "model_type": "MaskablePPO",
- "policy_type": "MlpPolicy",
- "cpu_count": 1,
- "drop_ohlc_from_features": False,
- "model_reward_parameters": {"rr": 2.0, "profit_aim": 0.03},
- "train_cycles": 1,
- "n_envs": 1,
- "n_eval_envs": 1,
- "n_eval_steps": 16,
- "n_eval_episodes": 1,
- "check_envs": False,
- "add_state_info": False,
- },
- },
- }
-
-
-class ReviewContractsTest(unittest.TestCase):
+class ReviewContractsTest(QaTestCase):
def model(self, *, hold=False, hpo=False):
temp = tempfile.TemporaryDirectory()
self.addCleanup(temp.cleanup)
--- /dev/null
+"""Schedule, conversion and network-architecture algebra; requires the Freqtrade QA image."""
+
+import math
+import unittest
+
+import torch as th
+from qa_support import QaTestCase
+
+from ReforceXY.user_data.freqaimodels.ReforceXY import (
+ ReforceXY,
+ SimpleLinearSchedule,
+ compute_gradient_steps,
+ deepmerge,
+ get_activation_fn,
+ get_net_arch,
+ get_optimizer_class,
+ get_schedule,
+ get_schedule_type,
+ hours_to_seconds,
+ steps_to_days,
+)
+
+LINEAR, CONSTANT, UNKNOWN = ReforceXY._SCHEDULE_TYPES
+
+
+class LinearScheduleTest(QaTestCase):
+ def test_it_falls_from_its_initial_value_to_nothing(self):
+ schedule = SimpleLinearSchedule(0.25)
+ self.assertEqual(0.25, schedule(1.0))
+ self.assertEqual(0.0, schedule(0.0))
+
+ def test_a_string_initial_value_is_read_as_a_number(self):
+ self.assertEqual(0.5, SimpleLinearSchedule("0.5")(1.0))
+
+ def test_progress_beyond_one_is_clamped_so_no_negative_rate_is_ever_emitted(self):
+ # SB3 asks for negative progress when a DQN overshoots the aligned budget; a
+ # negative learning rate would silently invert the update.
+ schedule = SimpleLinearSchedule(0.3)
+ self.assertEqual(0.3, schedule(1.5))
+ self.assertEqual(0.0, schedule(-0.5))
+
+ def test_it_reports_the_value_it_was_built_from(self):
+ self.assertIn("0.5", repr(SimpleLinearSchedule(0.5)))
+
+
+class ScheduleClassificationTest(QaTestCase):
+ def test_a_number_is_a_constant_schedule_holding_that_number(self):
+ kind, initial, final = get_schedule_type(0.4)
+ self.assertEqual(CONSTANT, kind)
+ self.assertEqual(0.4, initial)
+ self.assertEqual(0.4, final)
+
+ def test_a_linear_schedule_is_classified_by_its_two_endpoints(self):
+ kind, initial, final = get_schedule_type(SimpleLinearSchedule(0.2))
+ self.assertEqual(LINEAR, kind)
+ self.assertEqual(0.2, initial)
+ self.assertEqual(0.0, final)
+
+ def test_an_unrecognised_schedule_is_reported_as_unknown_rather_than_guessed(self):
+ kind, initial, final = get_schedule_type("not-a-schedule")
+ self.assertEqual(UNKNOWN, kind)
+ self.assertTrue(math.isnan(initial))
+ self.assertTrue(math.isnan(final))
+
+ def test_a_linear_request_builds_the_linear_schedule(self):
+ self.assertIsInstance(get_schedule(LINEAR, 0.1), SimpleLinearSchedule)
+
+ def test_every_other_request_builds_a_constant_schedule(self):
+ for kind in (CONSTANT, UNKNOWN):
+ with self.subTest(kind=kind):
+ self.assertEqual(0.1, get_schedule(kind, 0.1)(1.0))
+ self.assertEqual(0.1, get_schedule(kind, 0.1)(0.0))
+
+
+class ConversionTest(QaTestCase):
+ def test_hours_to_seconds_scales_by_3600(self):
+ self.assertEqual(7200.0, hours_to_seconds(2.0))
+
+ def test_steps_to_days_divides_the_total_minutes_by_a_day_and_rounds(self):
+ self.assertEqual(60.0, steps_to_days(86_400, "1m"))
+ self.assertEqual(0.0, steps_to_days(1, "1m"))
+
+ def test_deepmerge_recurses_and_leaves_both_inputs_untouched(self):
+ base = {"a": {"b": 1, "c": 2}, "d": 3}
+ src = {"a": {"c": 9}}
+ merged = deepmerge(base, src)
+ self.assertEqual({"a": {"b": 1, "c": 9}, "d": 3}, merged)
+ self.assertEqual({"a": {"b": 1, "c": 2}, "d": 3}, base)
+ self.assertEqual({"a": {"c": 9}}, src)
+
+
+class GradientStepTest(QaTestCase):
+ def test_it_is_the_ceiling_of_the_frequency_quotient_capped_by_the_frequency(self):
+ self.assertEqual(4, compute_gradient_steps(8, 2))
+ self.assertEqual(1, compute_gradient_steps(3, 8))
+
+ def test_a_non_positive_frequency_reports_minus_one_rather_than_a_step_count(self):
+ self.assertEqual(-1, compute_gradient_steps(0, 2))
+
+ def test_a_non_positive_subsample_reports_minus_one(self):
+ self.assertEqual(-1, compute_gradient_steps(8, 0))
+
+ def test_a_non_numeric_frequency_reports_minus_one(self):
+ self.assertEqual(-1, compute_gradient_steps("often", 2))
+
+ def test_a_sequence_frequency_reads_its_first_element(self):
+ self.assertEqual(4, compute_gradient_steps((8, "s"), 2))
+
+
+class ArchitectureTest(QaTestCase):
+ def test_the_on_policy_family_gets_a_separate_actor_and_critic_width(self):
+ for size in ReforceXY._NET_ARCH_SIZES:
+ with self.subTest(size=size):
+ arch = get_net_arch("MaskablePPO", size)
+ self.assertEqual({"pi", "vf"}, set(arch))
+
+ def test_the_value_based_family_gets_a_shared_width_list(self):
+ for model_type in ("DQN", "QR-DQN"):
+ with self.subTest(model_type=model_type):
+ self.assertIsInstance(get_net_arch(model_type, "small"), list)
+
+ def test_a_grown_size_is_wider_than_a_small_one(self):
+ self.assertGreater(
+ get_net_arch("DQN", "large")[0],
+ get_net_arch("DQN", "small")[0],
+ )
+
+ def test_an_unknown_size_falls_back_to_small_for_every_family(self):
+ for model_type in ("MaskablePPO", "DQN"):
+ with self.subTest(model_type=model_type):
+ self.assertEqual(
+ get_net_arch(model_type, "small"), get_net_arch(model_type, "huge")
+ )
+
+ def test_activation_and_optimizer_map_to_their_sb3_symbols(self):
+ self.assertIs(th.nn.ReLU, get_activation_fn("relu"))
+ self.assertIs(th.nn.Tanh, get_activation_fn("tanh"))
+ self.assertIs(th.optim.Adam, get_optimizer_class("adam"))
+
+ def test_an_unknown_activation_or_optimizer_falls_back_to_the_default(self):
+ self.assertIs(th.nn.ReLU, get_activation_fn("no-such-activation"))
+ self.assertIs(th.optim.Adam, get_optimizer_class("no-such-optimizer"))
+
+
+class EvalFrequencyTest(QaTestCase):
+ def _model(self, *, model_type="MaskablePPO", n_envs=1, n_eval_steps=16):
+ model = ReforceXY.__new__(ReforceXY)
+ model.n_envs = n_envs
+ model.n_eval_steps = n_eval_steps
+ model.model_type = model_type
+ return model
+
+ def test_a_non_positive_budget_still_evaluates_once(self):
+ self.assertEqual(1, self._model().get_eval_freq(0))
+
+ def test_eval_freq_never_exceeds_the_calls_the_budget_allows(self):
+ model = self._model()
+ for budget in (16, 64, 16_000):
+ with self.subTest(budget=budget):
+ self.assertLessEqual(model.get_eval_freq(budget), budget)
+
+ def test_the_on_policy_family_prefers_a_rollout_that_fits_the_budget(self):
+ model = self._model()
+ # With no model_params the search walks _PPO_N_STEPS downwards, so a budget
+ # larger than every candidate yields the largest one, never the budget itself.
+ self.assertLess(model.get_eval_freq(16_000), 16_000)
+ self.assertEqual(16, model.get_eval_freq(16))
+
+ def test_more_environments_mean_fewer_callback_calls_for_the_same_budget(self):
+ self.assertLess(
+ self._model(n_envs=4).get_eval_freq(16_000),
+ self._model(n_envs=1).get_eval_freq(16_000),
+ )
+
+ def test_the_value_based_family_uses_the_evaluation_step_budget(self):
+ self.assertEqual(16, self._model(model_type="DQN", n_eval_steps=16).get_eval_freq(16_000))
+
+ def test_hyperopt_reduces_the_interval_but_never_below_one(self):
+ self.assertEqual(
+ 1, self._model(model_type="DQN", n_eval_steps=1).get_eval_freq(16_000, hyperopt=True)
+ )
+
+
+if __name__ == "__main__":
+ unittest.main()
"""Regression contracts for learning eligibility, replay and temporal observations."""
import tempfile
-import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest import mock
from freqtrade.freqai.data_drawer import FreqaiDataDrawer
from freqtrade.freqai.data_kitchen import FreqaiDataKitchen
from optuna import TrialPruned, create_study
+from qa_support import QaTestCase, RecordingPolicy, model_config
-from ReforceXY.tests.test_review_contracts import RecordingPolicy, model_config
from ReforceXY.user_data.freqaimodels.ReforceXY import ReforceXY
-class TrainingObservationsTest(unittest.TestCase):
+class TrainingObservationsTest(QaTestCase):
def test_dqn_eligibility_restart_and_chronology(self):
for algorithm in ("DQN", "QRDQN"):
with self.subTest(algorithm=algorithm), tempfile.TemporaryDirectory() as temp:
--- /dev/null
+"""Guards the import identity and coverage denominator of the measured source.; requires the Freqtrade QA image."""
+
+import importlib
+import random
+import sys
+import unittest
+from pathlib import Path
+
+import numpy as np
+import torch
+from qa_support import (
+ _ACTION_MASKS_CACHE_AT_IMPORT,
+ _NUMPY_STATE_AT_IMPORT,
+ _PRODUCTION,
+ _RANDOM_STATE_AT_IMPORT,
+ _TORCH_STATE_AT_IMPORT,
+ MEASURED_TREE,
+ PRODUCTION_MODULE,
+ STRATEGY_MODULE,
+ TESTS_ROOT,
+ QaTestCase,
+ _shuffle_cmp,
+)
+
+PRODUCTION_NAMES = (PRODUCTION_MODULE, STRATEGY_MODULE)
+# `ReforceXY/user_data` holds no `__init__.py`, so `ReforceXY`, `ReforceXY.user_data` and
+# `ReforceXY.user_data.freqaimodels` are PEP 420 namespace packages carrying no `__file__`
+# at all; `Path(None)` would raise instead of asserting, so every walk below filters on the
+# truthiness of `__file__` first.
+_SOURCES = tuple(sorted(MEASURED_TREE.rglob("*.py")))
+
+# This class derives from unittest.TestCase, NOT from QaTestCase, and so runs no restore.
+# That is deliberate: a check of the restore itself must observe what the restore left
+# behind, not the state setUp already put back. It is one of the two exemptions from the
+# inheritance rule below, together with test_coverage_floor.py.
+_EXEMPT_MODULES = {"test_suite_contract", "test_coverage_floor"}
+
+# Imported here, at module scope, so the parity contract holds on its own: unittest discovery
+# imports every test module before running any test, so a lazy import inside a method would
+# make the assertion unconditionally true. Dotted paths only — CI puts `user_data/strategies`
+# on PYTHONPATH, so a bare-stem name would register a second module object for the very file
+# this contract pins.
+for _name in PRODUCTION_NAMES:
+ importlib.import_module(_name)
+
+
+def _source_files_under(root: Path) -> list[tuple[str, Path]]:
+ """Registered modules whose `__file__` is a `.py` under `root`, as (name, resolved path)."""
+ found = []
+ for name, module in list(sys.modules.items()):
+ raw = getattr(module, "__file__", None)
+ if not raw:
+ continue
+ path = Path(raw).resolve()
+ if path.suffix == ".py" and path.is_relative_to(root):
+ found.append((name, path))
+ return sorted(found)
+
+
+class SuiteContractTest(unittest.TestCase):
+ def test_each_measured_source_file_is_registered_under_exactly_one_dotted_name(self):
+ registered = [
+ path
+ for name, path in _source_files_under(MEASURED_TREE.parent)
+ if name.startswith("ReforceXY.user_data")
+ ]
+ for src in _SOURCES:
+ with self.subTest(module=src.name):
+ matches = [path for path in registered if path == src.resolve()]
+ self.assertEqual(
+ 1,
+ len(matches),
+ "a measured file under exactly one dotted name; a second registration is a "
+ "shadow copy carrying its own module state",
+ )
+
+ def test_every_test_module_is_registered_under_its_bare_stem(self):
+ # The half that catches a shadow copy of the test tree itself. A measured-tree walk
+ # cannot observe `ReforceXY/tests`, so a dotted `from ReforceXY.tests.qa_support
+ # import ...` would register a second module object for a file discovery already
+ # registered under its stem — and every restore in qa_support would then clean a
+ # module the code under test never touches.
+ for name, path in _source_files_under(TESTS_ROOT):
+ with self.subTest(module=name):
+ self.assertEqual(
+ path.stem,
+ name,
+ "a test module registered under anything but its bare stem has a second "
+ "copy, so patching one leaves the other untouched",
+ )
+
+ def test_no_package_marker_under_the_test_directory(self):
+ markers = sorted(
+ path.relative_to(TESTS_ROOT).as_posix() for path in TESTS_ROOT.rglob("__init__.py")
+ )
+ self.assertEqual(
+ [],
+ markers,
+ "a package marker makes the start directory a package, so the test modules lose "
+ "top-level status and `from qa_support import ...` fails under discovery",
+ )
+
+ def test_only_the_meta_modules_bypass_the_shared_base_case(self):
+ # `cls.__module__ == module.__name__` is the attribution rule that separates a class
+ # defined here from one merely imported here: without it, `from unittest import
+ # TestCase` or a re-exported base would be reported as a bypass.
+ offenders = []
+ for name, _ in _source_files_under(TESTS_ROOT):
+ if name in _EXEMPT_MODULES:
+ continue
+ for attr, cls in vars(sys.modules[name]).items():
+ if not isinstance(cls, type) or cls.__module__ != name:
+ continue
+ if cls is unittest.TestCase or not issubclass(cls, unittest.TestCase):
+ continue
+ if not issubclass(cls, QaTestCase):
+ offenders.append(f"{name}.{attr}")
+ self.assertEqual(
+ [],
+ offenders,
+ "every TestCase outside the two meta-modules must derive from QaTestCase, or the "
+ "process globals it restores leak between tests",
+ )
+
+ def test_the_shuffle_is_installed_on_the_loader(self):
+ # `sortTestMethodsUsing` is a staticmethod on the LOADER; a TestCase has no such
+ # hook, and a plain function would bind as a method and raise TypeError at discovery.
+ self.assertIs(_shuffle_cmp, unittest.TestLoader.sortTestMethodsUsing)
+
+ def test_the_restore_puts_the_process_globals_back(self):
+ cache = _PRODUCTION.ReforceXY._action_masks_cache
+ # Dirty every global the way a test that trains a model would.
+ cache[("dirty", 0.5)] = None
+ random.seed(1234)
+ np.random.seed(1234)
+ torch.manual_seed(1234)
+
+ self.addCleanup(QaTestCase._restore_process_globals)
+ QaTestCase._restore_process_globals()
+
+ self.assertEqual(
+ dict(_ACTION_MASKS_CACHE_AT_IMPORT),
+ dict(cache),
+ "_action_masks_cache was not restored",
+ )
+ self.assertEqual(_RANDOM_STATE_AT_IMPORT, random.getstate())
+ self.assertEqual(
+ _NUMPY_STATE_AT_IMPORT[1].tolist(),
+ np.random.get_state()[1].tolist(),
+ )
+ if _TORCH_STATE_AT_IMPORT is not None:
+ self.assertTrue(torch.equal(torch.random.get_rng_state(), _TORCH_STATE_AT_IMPORT))
+
+
+if __name__ == "__main__":
+ unittest.main()
repositories / freqai-strategies.git / commitdiff
