From: Jérôme Benoit Date: Fri, 2 Oct 2026 09:30:37 +0000 (+0200) Subject: test(reforcexy): cover the strategy's signals, validation and sizing X-Git-Url: https://git.piment-noir.org/?a=commitdiff_plain;h=b98ac38e8e99c2541755ef2941242edfb5ac0f3e;p=freqai-strategies.git test(reforcexy): cover the strategy's signals, validation and sizing RLAgentStrategy is the thinnest layer of the strategy and was the least covered: 47.2%, with every symbol but `leverage` untouched. It is also where a wrong answer costs money directly — an entry flag raised on a row whose prediction the model has not published trades on a value that does not exist yet. The action literals are asserted as a cross-object equality with the enum rather than as two independent literals, so both survive a freqtrade renumbering only if they agree; both would still pass if each were simply retyped. Signal tests read the flags through a NaN-aware helper, because the assignment creates the column and an untouched row arrives as a missing float rather than a zero — asserting `== 0` would have encoded a shape the strategy does not produce. 125 tests green; gate exits 0 at 59.1%, floor raised to 59 with its measurement; green under three seeded method orderings. --- diff --git a/ReforceXY/.coveragerc b/ReforceXY/.coveragerc index 4b12437..e9e85fd 100644 --- a/ReforceXY/.coveragerc +++ b/ReforceXY/.coveragerc @@ -8,5 +8,5 @@ relative_files = True 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 +# measured 2026-10-02 at 59.1% — raise only, never lower +fail_under = 59 \ No newline at end of file diff --git a/ReforceXY/tests/test_coverage_floor.py b/ReforceXY/tests/test_coverage_floor.py index 86b2a17..88ad223 100644 --- a/ReforceXY/tests/test_coverage_floor.py +++ b/ReforceXY/tests/test_coverage_floor.py @@ -24,7 +24,7 @@ INCLUDE = "include" # 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 +MINIMUM_FLOOR = 59.0 EXCLUSION_KEYS = ( OMIT, INCLUDE, diff --git a/ReforceXY/tests/test_strategy_semantics.py b/ReforceXY/tests/test_strategy_semantics.py new file mode 100644 index 0000000..7f01dc6 --- /dev/null +++ b/ReforceXY/tests/test_strategy_semantics.py @@ -0,0 +1,293 @@ +"""Strategy signal, validation and sizing contracts; requires the Freqtrade QA image.""" + +import unittest +from datetime import datetime, timezone +from types import SimpleNamespace +from unittest import mock + +import numpy as np +import pandas as pd +from qa_support import QaTestCase + +from ReforceXY.user_data.freqaimodels.ReforceXY import Actions +from ReforceXY.user_data.strategies.RLAgentStrategy import ( + ACTION_COLUMN, + RLAgentStrategy, + _ensure_datetime_series, +) + + +def _raised(column): + """True where the flag is set. An untouched row is NaN, not 0: the column is created + by the assignment, so "no signal" arrives as a missing float rather than a zero.""" + return [bool(value) if value == value else False for value in column] + + +def _frame(actions, do_predict=1, count=None): + values = list(actions) + count = count if count is not None else len(values) + return pd.DataFrame( + { + "open": np.linspace(100.0, 110.0, count), + "high": np.linspace(101.0, 111.0, count), + "low": np.linspace(99.0, 109.0, count), + "close": np.linspace(100.5, 110.5, count), + "volume": np.full(count, 10.0), + "date": pd.date_range("2024-01-01", periods=count, freq="5min", tz="UTC"), + ACTION_COLUMN: values, + "do_predict": [do_predict] * count, + } + ) + + +class ActionLiteralTest(QaTestCase): + def test_the_strategy_literals_are_the_production_action_codes(self): + # Asserted as a cross-object equality, not as two literals: the strategy's constants + # are correct only because they equal the enum's values, and two independent + # literals would both still pass after a freqtrade renumbering. + self.assertEqual(Actions.Long_enter.value, RLAgentStrategy._ACTION_ENTER_LONG) + self.assertEqual(Actions.Long_exit.value, RLAgentStrategy._ACTION_EXIT_LONG) + self.assertEqual(Actions.Short_enter.value, RLAgentStrategy._ACTION_ENTER_SHORT) + self.assertEqual(Actions.Short_exit.value, RLAgentStrategy._ACTION_EXIT_SHORT) + + +class DateColumnTest(QaTestCase): + def test_a_missing_date_column_is_refused(self): + with self.assertRaises(ValueError) as caught: + _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_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)) + + 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: + _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 + # that has no data to be wrong about. + series = _ensure_datetime_series(pd.Series([None, None], dtype="Int64")) + self.assertTrue(series.isna().all()) + + +class TradingModeTest(QaTestCase): + def _strategy(self, **config): + strategy = RLAgentStrategy.__new__(RLAgentStrategy) + strategy.config = config + return strategy + + def test_the_two_leveraged_modes_allow_shorting(self): + for mode in ("margin", "futures"): + with self.subTest(mode=mode): + self.assertTrue(self._strategy(trading_mode=mode).is_short_allowed()) + + def test_spot_forbids_shorting(self): + 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: + 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) + self.assertFalse(self._strategy(trading_mode="spot").can_short) + + +class ProtectionConfigTest(QaTestCase): + def _strategy(self, custom_protections): + strategy = RLAgentStrategy.__new__(RLAgentStrategy) + 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) + + def test_a_protection_without_a_method_name_is_refused(self): + strategy = self._strategy([{"stop_duration": 2}]) + with self.assertRaises(ValueError): + _ = strategy.protections + + def test_a_non_list_protection_value_is_refused(self): + strategy = self._strategy({"method": "CooldownPeriod"}) + with self.assertRaises(ValueError): + _ = strategy.protections + + +class LeverageTest(QaTestCase): + def _strategy(self, configured): + strategy = RLAgentStrategy.__new__(RLAgentStrategy) + strategy.config = {"leverage": configured} + strategy._leverage_warning_state = None + return strategy + + def _arguments(self, **overrides): + arguments = { + "pair": "BTC/USDT", + "current_time": datetime(2026, 1, 1, tzinfo=timezone.utc), + "current_rate": 100.0, + "proposed_leverage": 2.0, + "max_leverage": 5.0, + "entry_tag": None, + "side": "long", + } + arguments.update(overrides) + return arguments + + def test_an_absent_configuration_yields_the_proposed_leverage(self): + self.assertEqual(2.0, self._strategy(None).leverage(**self._arguments())) + + def test_a_configured_value_is_clamped_into_the_pair_bounds(self): + for configured, expected in ((1.5, 1.5), (0.5, 1.0), (9.0, 5.0)): + with self.subTest(configured=configured): + 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]): + 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): + # bool is an int subclass, so float(True) succeeds; treating 1.0 as a leverage + # configuration would silently disable the clamp. + self.assertEqual(2.0, self._strategy(True).leverage(**self._arguments())) + + def test_a_non_finite_configuration_falls_back_to_the_proposed_leverage(self): + for configured in (float("nan"), float("inf"), float("-inf")): + with self.subTest(configured=configured): + self.assertEqual(2.0, self._strategy(configured).leverage(**self._arguments())) + + def test_the_result_is_always_at_least_one(self): + self.assertEqual(1.0, self._strategy(0.5).leverage(**self._arguments(max_leverage=0.2))) + + +class EntrySignalTest(QaTestCase): + def _strategy(self): + return RLAgentStrategy.__new__(RLAgentStrategy) + + def test_a_long_entry_action_raises_the_long_flag_and_nothing_else(self): + frame = _frame([1, 0, 0]) + result = self._strategy().populate_entry_trend(frame, {"pair": "BTC/USDT"}) + self.assertEqual([True, False, False], _raised(result["enter_long"])) + self.assertEqual([False] * 3, _raised(result["enter_short"])) + + def test_a_short_entry_action_raises_the_short_flag(self): + result = self._strategy().populate_entry_trend(_frame([3, 0]), {"pair": "BTC/USDT"}) + self.assertEqual([True, False], _raised(result["enter_short"])) + + def test_no_action_raises_no_entry(self): + for action in (0, 2, 4): + with self.subTest(action=action): + result = self._strategy().populate_entry_trend( + _frame([action]), {"pair": "BTC/USDT"} + ) + self.assertFalse(_raised(result["enter_long"])[0]) + self.assertFalse(_raised(result["enter_short"])[0]) + + 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. + result = self._strategy().populate_entry_trend( + _frame([1, 1], do_predict=0), {"pair": "BTC/USDT"} + ) + self.assertEqual([False, False], _raised(result["enter_long"])) + + +class ExitSignalTest(QaTestCase): + def _strategy(self): + return RLAgentStrategy.__new__(RLAgentStrategy) + + def test_an_exit_action_raises_only_its_own_side(self): + result = self._strategy().populate_exit_trend(_frame([2, 0, 4]), {"pair": "BTC/USDT"}) + self.assertEqual([True, False, False], _raised(result["exit_long"])) + self.assertEqual([False, False, True], _raised(result["exit_short"])) + + def test_an_entry_action_raises_no_exit(self): + for action in (1, 3): + with self.subTest(action=action): + result = self._strategy().populate_exit_trend( + _frame([action]), {"pair": "BTC/USDT"} + ) + self.assertFalse(_raised(result["exit_long"])[0]) + self.assertFalse(_raised(result["exit_short"])[0]) + + def test_a_rejected_prediction_closes_the_open_position_of_its_own_side(self): + # do_predict == 2 is a rejected prediction: the open trade must be closed, and + # its side decides which flag is set, not the recorded action. + frame = _frame([1], do_predict=2) + for is_short, expected in ((False, "exit_long"), (True, "exit_short")): + with self.subTest(is_short=is_short): + trade = SimpleNamespace(is_short=is_short) + with mock.patch( + "ReforceXY.user_data.strategies.RLAgentStrategy.Trade.get_trades_proxy", + return_value=[trade], + ): + result = self._strategy().populate_exit_trend( + frame.copy(), {"pair": "BTC/USDT"} + ) + self.assertTrue(_raised(result[expected])[0]) + other = "exit_short" if expected == "exit_long" else "exit_long" + self.assertFalse(_raised(result[other])[0]) + + def test_a_rejected_prediction_with_no_open_trade_raises_no_exit(self): + frame = _frame([1], do_predict=2) + with mock.patch( + "ReforceXY.user_data.strategies.RLAgentStrategy.Trade.get_trades_proxy", + return_value=[], + ): + result = self._strategy().populate_exit_trend(frame, {"pair": "BTC/USDT"}) + self.assertFalse(_raised(result["exit_long"])[0]) + self.assertFalse(_raised(result["exit_short"])[0]) + + +class FeatureEngineeringTest(QaTestCase): + def _strategy(self): + return RLAgentStrategy.__new__(RLAgentStrategy) + + def test_the_standard_features_are_bounded_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")): + with self.subTest(feature=produced): + self.assertEqual(list(frame[source]), list(result[produced])) + + def test_the_time_features_are_normalised_into_the_unit_interval(self): + result = self._strategy().feature_engineering_standard(_frame([0] * 48), {}) + for feature in ("%-day_of_week", "%-hour_of_day"): + with self.subTest(feature=feature): + self.assertTrue(((result[feature] > 0.0) & (result[feature] <= 1.0)).all()) + + def test_the_day_of_week_never_reports_a_zero_weekday(self): + frame = pd.DataFrame({"date": pd.Series(pd.to_datetime(["2024-01-01"], utc=True))}) + result = self._strategy().feature_engineering_standard(frame, {}) + self.assertGreater(result["%-day_of_week"].iloc[0], 0.0) + + def test_the_expanding_feature_is_the_log_difference_of_the_close(self): + frame = _frame([0, 0, 0]) + frame["close"] = [100.0, 110.0, 121.0] + result = self._strategy().feature_engineering_expand_basic(frame, {}) + self.assertTrue(pd.isna(result["%-close_log_return"].iloc[0])) + 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]), {}) + self.assertEqual([0, 0], list(result[ACTION_COLUMN])) + + +if __name__ == "__main__": + unittest.main()