--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
+ FREQAI_QA_SHUFFLE_SEED=$seed python -m unittest discover -s "${{ matrix.context }}/tests" || exit 1
done'
reward-space-analysis:
`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` and
-`ReforceXY/tests/test_coverage_floor.py` refuse a
+the pull request. Both floor guards 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
disabled branch trace, a disabled namespace walk, a source tree that is not the
-measured one, a `fail_under` that has drifted into the inert `[run]` section, the
-three of coverage's default source exclusions, a coverage pragma in production
-source, and any `omit`, `include`, `exclude_lines`, `exclude_also` or
-`partial_*` of production code. The shared `scripts/run-coverage.sh` runner is
+measured one, a `fail_under` that has drifted into the inert `[run]` section, any
+`omit`, `include`, `exclude_lines`, `exclude_also`, `patch` or `partial_*` under
+either section, and a misspelled option. `ReforceXY`'s guard is the stricter of
+the two: it also rejects a second measurement root (`source_pkgs`, `source_dirs`,
+`plugins`) and pins that no statement or branch leaves the denominator through
+coverage's default exclusions — the statement-level three and the branch-level
+`# pragma: no branch` / `while True:` family — which it detects by parsing each
+measured file twice, with and without `DEFAULT_EXCLUDE`. The shared `scripts/run-coverage.sh` runner is
exercised with real passing and failing coverage reports and a failing test suite;
the checks assert process exit status, not workflow token spelling.
from typing import Any
import numpy as np
-import pandas as pd
from freqtrade.enums import RunMode
PAIR = "BTC/USDT"
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."""
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
_SEED = os.getenv("FREQAI_QA_SHUFFLE_SEED", "")
+def reseed(value: str) -> str:
+ """Set the loader's shuffle seed and return its previous value.
+
+ Exists so the suite contract can exercise the comparator the loader really
+ holds, rather than a copy of it. Call it only while nothing else is
+ discovering tests, and restore the returned value afterwards.
+ """
+ global _SEED
+ previous = _SEED
+ _SEED = value
+ return previous
+
+
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:
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)
["ReforceXY/user_data"],
)
+ def test_no_other_option_declares_a_measurement_root(self):
+ # `source`, `source_pkgs` and `source_dirs` are three ways of telling coverage
+ # what to measure, and `plugins` can replace the file reporter that decides it.
+ # Any of them alongside `source` is either a silent second root or a silent
+ # change of instrument, and no other guard can see it: the spelling check
+ # accepts them because coverage binds them legally.
+ for key in ("source_pkgs", "source_dirs", "plugins"):
+ with self.subTest(option=key):
+ self.assertNotIn(
+ key,
+ self.parser["run"],
+ f"run:{key} adds a measurement surface the gate does not account for",
+ )
+
if __name__ == "__main__":
unittest.main()
class ReviewContractsTest(QaTestCase):
- def model(self, *, hold=False, hpo=False):
+ def model(self, *, hold=False):
temp = tempfile.TemporaryDirectory()
self.addCleanup(temp.cleanup)
config = model_config(temp.name)
config["freqai"]["rl_config"]["model_reward_parameters"]["hold_potential_enabled"] = hold
- config["freqai"]["continual_learning"] = hpo
- config["freqai"]["rl_config_optuna"] = {"enabled": hpo}
+ config["freqai"]["continual_learning"] = False
+ config["freqai"]["rl_config_optuna"] = {"enabled": False}
model = ReforceXY(config=config)
model.live = True
model.can_short = True
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.assertEqual("SimpleLinearSchedule(initial_value=0.5)", repr(SimpleLinearSchedule(0.5)))
-
class ScheduleClassificationTest(QaTestCase):
def test_a_number_is_a_constant_schedule_holding_that_number(self):
arch = get_net_arch("MaskablePPO", size)
self.assertEqual({"pi", "vf"}, set(arch))
self.assertEqual(arch["pi"], arch["vf"])
- self.assertEqual([128, 128], arch["pi"]) if size == "small" else None
+ if size == "small":
+ self.assertEqual([128, 128], arch["pi"])
def test_the_value_based_family_gets_a_shared_width_list(self):
for model_type in ("DQN", "QR-DQN"):
return model
def test_a_non_positive_budget_still_evaluates_once(self):
+ # A boundary worth pinning as such. Deleting the early return would NOT turn
+ # this red — max_n_calls floors at 1 anyway — so it is not claimed to cover it.
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_eval_freq_is_capped_by_the_calls_the_budget_allows(self):
+ # n_eval_steps above the budget is the case where the final min() bites; with
+ # the value-based branch the uncapped result is strictly greater.
+ model = self._model(model_type="DQN", n_eval_steps=100)
+ self.assertEqual(64, model.get_eval_freq(64))
+ self.assertEqual(8, model.get_eval_freq(8))
def test_the_on_policy_family_prefers_a_rollout_that_fits_the_budget(self):
model = self._model()
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_by_the_configured_factor(self):
- model = self._model(model_type="DQN", n_eval_steps=16)
- self.assertEqual(16, model.get_eval_freq(16_000))
- self.assertEqual(
- round(16 / ReforceXY._HYPEROPT_EVAL_FREQ_REDUCTION_FACTOR),
- model.get_eval_freq(16_000, hyperopt=True),
- )
+ def test_hyperopt_reduces_the_interval(self):
+ # Two explicit outcomes rather than the production formula restated: changing
+ # _HYPEROPT_EVAL_FREQ_REDUCTION_FACTOR moves both of them.
+ model = self._model(model_type="DQN")
+ for n_eval_steps, plain, reduced in ((16, 16, 4), (64, 64, 16)):
+ with self.subTest(n_eval_steps=n_eval_steps):
+ model.n_eval_steps = n_eval_steps
+ self.assertEqual(plain, model.get_eval_freq(16_000))
+ self.assertEqual(reduced, model.get_eval_freq(16_000, hyperopt=True))
def test_the_reduced_interval_never_falls_below_one(self):
model = self._model(model_type="DQN", n_eval_steps=1)
import numpy as np
import pandas as pd
+from freqtrade.persistence import LocalTrade, Trade
from qa_support import QaTestCase
from ReforceXY.user_data.freqaimodels.ReforceXY import Actions
class DateColumnTest(QaTestCase):
def test_a_missing_date_column_is_refused(self):
- with self.assertRaises(ValueError) as caught:
+ with self.assertRaises(ValueError):
_ensure_datetime_series(None)
- self.assertIn("date", str(caught.exception))
- def test_a_non_null_date_column_keeps_its_values(self):
- series = pd.Series(pd.date_range("2024-01-01", periods=3, freq="5min", tz="UTC"))
- self.assertEqual(3, len(_ensure_datetime_series(series)))
+ def test_datetime_conversion_preserves_instants_and_index(self):
+ series = pd.Series(
+ pd.date_range("2024-01-01", periods=3, freq="5min", tz="UTC"),
+ index=[7, 11, 19],
+ )
+ expected = pd.Series(
+ [
+ pd.Timestamp("2024-01-01T00:00:00Z"),
+ pd.Timestamp("2024-01-01T00:05:00Z"),
+ pd.Timestamp("2024-01-01T00:10:00Z"),
+ ],
+ index=[7, 11, 19],
+ dtype="datetime64[ms, UTC]",
+ )
+ pd.testing.assert_series_equal(expected, _ensure_datetime_series(series))
def test_an_integer_column_inside_the_epoch_range_is_read_as_milliseconds(self):
millis = int(pd.Timestamp("2024-01-01", tz="UTC").value // 1_000_000)
series = _ensure_datetime_series(pd.Series([millis, millis + 300_000]))
- self.assertEqual(2, len(series))
+ pd.testing.assert_series_equal(
+ pd.Series(
+ [
+ pd.Timestamp("2024-01-01T00:00:00Z"),
+ pd.Timestamp("2024-01-01T00:05:00Z"),
+ ],
+ dtype="datetime64[ms, UTC]",
+ ),
+ series,
+ )
def test_an_integer_epoch_in_seconds_is_refused_as_a_corrupted_unit(self):
millis = int(pd.Timestamp("2024-01-01", tz="UTC").value // 1_000_000)
- with self.assertRaises(ValueError) as caught:
+ with self.assertRaises(ValueError):
_ensure_datetime_series(pd.Series([millis // 1000]))
- self.assertIn("epoch-ms", str(caught.exception))
def test_an_all_null_integer_column_is_read_without_a_range_probe(self):
# Nothing to probe means nothing to reject; refusing here would reject a column
self.assertFalse(self._strategy(trading_mode="spot").is_short_allowed())
def test_an_undeclared_trading_mode_is_refused_rather_than_defaulted(self):
- with self.assertRaises(ValueError) as caught:
+ with self.assertRaises(ValueError):
self._strategy(trading_mode="margin-ish").is_short_allowed()
- self.assertIn("trading_mode", str(caught.exception))
def test_can_short_follows_the_mode(self):
self.assertTrue(self._strategy(trading_mode="futures").can_short)
strategy.config = {"custom_protections": custom_protections}
return strategy
- def test_a_well_formed_protection_list_is_passed_through_untouched(self):
- given = [{"method": "CooldownPeriod", "stop_duration": 2}]
- self.assertIs(given, self._strategy(given).protections)
-
def test_an_empty_list_is_accepted(self):
self.assertEqual([], self._strategy([]).protections)
self.assertEqual(expected, self._strategy(configured).leverage(**self._arguments()))
def test_a_non_numeric_configuration_falls_back_to_the_proposed_leverage(self):
- for configured in ("two", None, [1]):
+ for configured in ("two", [1], {"a": 1}):
with self.subTest(configured=configured):
- if configured is None:
- continue
self.assertEqual(2.0, self._strategy(configured).leverage(**self._arguments()))
def test_a_boolean_is_not_read_as_a_number(self):
def test_a_row_without_a_prediction_raises_no_entry(self):
# do_predict != 1 means the prediction is not yet confirmed; acting on it would
- # trade on a value the model has not published.
+ # trade on a value the model has not published. Both sides are exercised, because
+ # the two conditions are independent and either one can be dropped on its own.
result = self._strategy().populate_entry_trend(
- _frame([1, 1], do_predict=0), {"pair": "BTC/USDT"}
+ _frame(
+ [Actions.Long_enter.value, Actions.Short_enter.value],
+ do_predict=0,
+ ),
+ {"pair": "BTC/USDT"},
)
self.assertEqual([False, False], _raised(result["enter_long"]))
+ self.assertEqual([False, False], _raised(result["enter_short"]))
class ExitSignalTest(QaTestCase):
other = "exit_short" if expected == "exit_long" else "exit_long"
self.assertFalse(_raised(result[other])[0])
+ def test_a_rejected_prediction_ignores_closed_positions(self):
+ # Reached through freqtrade's own backtest collections rather than by asserting
+ # the call arguments, so a strategy that filtered differently would still pass.
+ # With `is_open` dropped the proxy returns closed trades too, and this goes red.
+ closed = SimpleNamespace(pair="BTC/USDT", is_open=False, is_short=False)
+ opened = SimpleNamespace(pair="BTC/USDT", is_open=True, is_short=True)
+ for open_trades, expected_short in (([], False), ([opened], True)):
+ with self.subTest(open_short=expected_short):
+ with (
+ mock.patch.object(Trade, "use_db", False),
+ mock.patch.object(LocalTrade, "bt_trades", [closed]),
+ mock.patch.object(LocalTrade, "bt_trades_open", open_trades),
+ ):
+ result = self._strategy().populate_exit_trend(
+ _frame([Actions.Neutral.value], do_predict=2),
+ {"pair": "BTC/USDT"},
+ )
+ self.assertEqual([False], _raised(result["exit_long"]))
+ self.assertEqual([expected_short], _raised(result["exit_short"]))
+
+ def test_a_row_without_a_prediction_raises_no_exit(self):
+ result = self._strategy().populate_exit_trend(
+ _frame(
+ [Actions.Long_exit.value, Actions.Short_exit.value],
+ do_predict=0,
+ ),
+ {"pair": "BTC/USDT"},
+ )
+ self.assertEqual([False, False], _raised(result["exit_long"]))
+ self.assertEqual([False, False], _raised(result["exit_short"]))
+
def test_a_rejected_prediction_with_no_open_trade_raises_no_exit(self):
frame = _frame([1], do_predict=2)
with mock.patch(
def _strategy(self):
return RLAgentStrategy.__new__(RLAgentStrategy)
- def test_the_standard_features_are_bounded_copies_of_the_ohlc(self):
+ def test_the_standard_features_are_exact_copies_of_the_ohlc(self):
frame = _frame([0, 0, 0])
result = self._strategy().feature_engineering_standard(frame, {})
for source, produced in (("close", "%-raw_close"), ("open", "%-raw_open")):
self.assertAlmostEqual(np.log(1.1), result["%-close_log_return"].iloc[1], places=12)
def test_the_target_column_is_the_neutral_action(self):
- result = self._strategy().set_freqai_targets(_frame([1, 3]), {})
+ result = self._strategy().set_freqai_targets(
+ _frame([Actions.Long_enter.value, Actions.Short_enter.value]), {}
+ )
self.assertEqual([0, 0], list(result[ACTION_COLUMN]))
"""Guards the import identity and coverage denominator of the measured source.; requires the Freqtrade QA image."""
-import hashlib
import importlib
-import os
import random
import sys
import unittest
-from functools import cmp_to_key
from pathlib import Path
import numpy as np
STRATEGY_MODULE,
TESTS_ROOT,
QaTestCase,
- _shuffle_cmp,
+ reseed,
)
PRODUCTION_NAMES = (PRODUCTION_MODULE, STRATEGY_MODULE)
# 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"}
+# SuiteContractTest 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 the ONLY exemption from the
+# inheritance rule below — not its module, which is why the guard names this class.
# 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
importlib.import_module(_name)
-def _seeded_cmp(seed: str):
- """The comparator qa_support builds for a given seed, rebuilt without touching _SEED."""
-
- def compare(a: str, b: str) -> int:
- 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)
-
- return compare
-
-
def _source_files_under(root: Path) -> list[tuple[str, Path]]:
"""Registered modules whose `__file__` is a `.py` under `root`, as (name, resolved path)."""
found = []
"top-level status and `from qa_support import ...` fails under discovery",
)
- def test_only_the_meta_modules_bypass_the_shared_base_case(self):
+ def test_only_the_restore_observer_bypasses_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 = []
+ offenders = set()
for name, _ in _source_files_under(TESTS_ROOT):
- if name in _EXEMPT_MODULES:
- continue
- for attr, cls in vars(sys.modules[name]).items():
+ for cls in vars(sys.modules[name]).values():
if not isinstance(cls, type) or cls.__module__ != name:
continue
- if cls is unittest.TestCase or not issubclass(cls, unittest.TestCase):
+ if not issubclass(cls, unittest.TestCase) or cls is SuiteContractTest:
continue
if not issubclass(cls, QaTestCase):
- offenders.append(f"{name}.{attr}")
+ offenders.add(f"{name}.{cls.__qualname__}")
self.assertEqual(
[],
- offenders,
- "every TestCase outside the two meta-modules must derive from QaTestCase, or the "
+ sorted(offenders),
+ "every TestCase except SuiteContractTest must derive from QaTestCase, or the "
"process globals it restores leak between tests",
)
- def test_the_order_is_alphabetical_exactly_when_no_seed_is_set(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.
- # What is asserted here is the ORDER the loader produces, not the identity of the
- # function it holds — a constant comparator installed by the same line would satisfy
- # an identity check and defeat the shuffle.
- names = ["test_c", "test_a", "test_b", "test_d"]
- ordered = sorted(names, key=cmp_to_key(_shuffle_cmp))
- if os.environ.get("FREQAI_QA_SHUFFLE_SEED", ""):
- # A seeded run must NOT be alphabetical; the other guard in this class
- # proves it is deterministic for a given seed.
- self.assertNotEqual(sorted(names), ordered)
- else:
- self.assertEqual(sorted(names), ordered)
- self.assertIs(_shuffle_cmp, unittest.TestLoader.sortTestMethodsUsing)
-
- def test_a_seeded_order_differs_from_alphabetical(self):
- names = ["test_c", "test_a", "test_b", "test_d"]
- alphabetical = sorted(names)
- for seed in ("1", "2", "3"):
- with self.subTest(seed=seed):
- shuffled = sorted(names, key=cmp_to_key(_seeded_cmp(seed)))
- self.assertNotEqual(
- alphabetical,
- shuffled,
- f"seed {seed} left the method order alphabetical, so the shuffle is inert",
- )
+ def test_the_loader_order_is_seeded_and_repeatable(self):
+ # Asserted through the loader that consumes the comparator, on a probe class that
+ # only supplies names: a copy of the comparator would leave every one of these
+ # green, so it is what makes this guard able to fail at all.
+ names = [f"test_{index:02d}" for index in range(32)]
+ probe = type("OrderProbe", (), {name: (lambda self: None) for name in reversed(names)})
+ loader = unittest.TestLoader()
+ previous = reseed("")
+ try:
+ self.assertEqual(
+ names, loader.getTestCaseNames(probe), "unseeded order is not alphabetical"
+ )
+ orders = []
+ for seed in ("1", "2", "3"):
+ with self.subTest(seed=seed):
+ reseed(seed)
+ ordered = loader.getTestCaseNames(probe)
+ self.assertNotEqual(names, ordered, f"seed {seed} left the order alphabetical")
+ self.assertEqual(
+ ordered, loader.getTestCaseNames(probe), "order is not repeatable"
+ )
+ orders.append(tuple(ordered))
+ self.assertGreater(
+ len(set(orders)),
+ 1,
+ "changing the seed must change the loader's method order",
+ )
+ finally:
+ reseed(previous)
def test_the_restore_puts_the_process_globals_back(self):
cache = _PRODUCTION.ReforceXY._action_masks_cache
repositories / freqai-strategies.git / commitdiff
