From 84972c36112915d478997c1b98ee9ec6ae7797e1 Mon Sep 17 00:00:00 2001 From: TPTBusiness Date: Sun, 3 May 2026 11:24:28 +0200 Subject: [PATCH] test: add 28 tests for LLM components, RL indicators, and model evaluators --- test/qlib/test_llm_components.py | 195 +++++++++++++++++++++++++++++++ test/qlib/test_rl_indicators.py | 116 ++++++++++++++++++ 2 files changed, 311 insertions(+) create mode 100644 test/qlib/test_llm_components.py create mode 100644 test/qlib/test_rl_indicators.py diff --git a/test/qlib/test_llm_components.py b/test/qlib/test_llm_components.py new file mode 100644 index 00000000..c8669670 --- /dev/null +++ b/test/qlib/test_llm_components.py @@ -0,0 +1,195 @@ +"""Tests for LLM-dependent components with mock backends.""" + +from __future__ import annotations + +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import numpy as np +import pytest + +PROJECT_ROOT = Path(__file__).parent.parent.parent +sys.path.insert(0, str(PROJECT_ROOT)) + + +# ============================================================================= +# ModelCoSTEEREvaluator (model_coder/evaluators.py) +# ============================================================================= + + +class TestModelCoSTEEREvaluator: + def test_init(self): + from rdagent.components.coder.model_coder.evaluators import ModelCoSTEEREvaluator + eva = ModelCoSTEEREvaluator(scen=MagicMock()) + assert eva.scen is not None + + def test_returns_cached_feedback(self): + from rdagent.components.coder.model_coder.evaluators import ModelCoSTEEREvaluator + eva = ModelCoSTEEREvaluator(scen=MagicMock()) + qk = MagicMock() + qk.success_task_to_knowledge_dict = { + "info_task": MagicMock(feedback="cached_fb"), + } + t = MagicMock() + t.get_task_information.return_value = "info_task" + fb = eva.evaluate(target_task=t, implementation=None, gt_implementation=None, queried_knowledge=qk) + assert fb == "cached_fb" + + def test_returns_failed_feedback(self): + from rdagent.components.coder.model_coder.evaluators import ModelCoSTEEREvaluator + eva = ModelCoSTEEREvaluator(scen=MagicMock()) + qk = MagicMock() + qk.success_task_to_knowledge_dict = {} + qk.failed_task_info_set = {"info_task"} + t = MagicMock() + t.get_task_information.return_value = "info_task" + fb = eva.evaluate(target_task=t, implementation=None, gt_implementation=None, queried_knowledge=qk) + assert fb.final_decision is False + assert "failed too many times" in fb.execution_feedback + + def test_raises_on_wrong_task_type(self): + from rdagent.components.coder.model_coder.evaluators import ModelCoSTEEREvaluator + eva = ModelCoSTEEREvaluator(scen=MagicMock()) + qk = MagicMock() + qk.success_task_to_knowledge_dict = {} + qk.failed_task_info_set = set() + t = MagicMock() + t.get_task_information.return_value = "new_task" + with pytest.raises(TypeError, match="Expected ModelTask"): + eva.evaluate(target_task=t, implementation=None, gt_implementation=None, queried_knowledge=qk) + + def test_raises_on_wrong_workspace_type(self): + from rdagent.components.coder.model_coder.evaluators import ModelCoSTEEREvaluator + from rdagent.components.coder.model_coder.model import ModelTask + + eva = ModelCoSTEEREvaluator(scen=MagicMock()) + qk = MagicMock() + qk.success_task_to_knowledge_dict = {} + qk.failed_task_info_set = set() + + t = ModelTask( + name="m1", description="d", architecture="LSTM", + hyperparameters={}, training_hyperparameters={}, + ) + t.get_task_information = MagicMock(return_value="new") + + with pytest.raises(TypeError, match="Expected ModelFBWorkspace"): + eva.evaluate(target_task=t, implementation="not_a_workspace", gt_implementation=None, queried_knowledge=qk) + + +# ============================================================================= +# FactorMultiProcessEvolvingStrategy (factor_coder/evolving_strategy.py) +# ============================================================================= + + +class TestFactorEvolvingStrategy: + def test_init_sets_fields(self): + from rdagent.components.coder.factor_coder.evolving_strategy import FactorMultiProcessEvolvingStrategy + strat = FactorMultiProcessEvolvingStrategy(scen=MagicMock(), settings=MagicMock()) + assert strat.num_loop == 0 + assert strat.haveSelected is False + assert strat.improve_mode is False + + def test_assign_code_list_to_evo_str_input(self): + """assign_code_list_to_evo handles string code (not dict).""" + from rdagent.components.coder.factor_coder.evolving_strategy import FactorMultiProcessEvolvingStrategy + from rdagent.components.coder.CoSTEER.evolvable_subjects import EvolvingItem + from rdagent.components.coder.factor_coder.factor import FactorTask, FactorFBWorkspace + + strat = FactorMultiProcessEvolvingStrategy(scen=MagicMock(), settings=MagicMock()) + evo = EvolvingItem(sub_tasks=[FactorTask("f1", "desc", "formula")]) + evo.sub_workspace_list = [None] + + with patch( + "rdagent.components.coder.factor_coder.evolving_strategy.auto_fix_factor_code", + return_value="fixed_code", + ): + strat.assign_code_list_to_evo(["raw_code"], evo) + assert evo.sub_workspace_list[0] is not None + # Should be a FactorFBWorkspace + from rdagent.components.coder.factor_coder.factor import FactorFBWorkspace + assert isinstance(evo.sub_workspace_list[0], FactorFBWorkspace) + + def test_assign_code_list_to_evo_dict_input(self): + """assign_code_list_to_evo handles dict code.""" + from rdagent.components.coder.factor_coder.evolving_strategy import FactorMultiProcessEvolvingStrategy + from rdagent.components.coder.CoSTEER.evolvable_subjects import EvolvingItem + from rdagent.components.coder.factor_coder.factor import FactorTask + + strat = FactorMultiProcessEvolvingStrategy(scen=MagicMock(), settings=MagicMock()) + evo = EvolvingItem(sub_tasks=[FactorTask("f1", "desc", "formula")]) + evo.sub_workspace_list = [None] + + with patch( + "rdagent.components.coder.factor_coder.evolving_strategy.auto_fix_factor_code", + return_value="fixed", + ): + strat.assign_code_list_to_evo([{"factor.py": "code", "utils.py": "util_code"}], evo) + assert evo.sub_workspace_list[0] is not None + + def test_assign_code_list_skips_none(self): + from rdagent.components.coder.factor_coder.evolving_strategy import FactorMultiProcessEvolvingStrategy + from rdagent.components.coder.CoSTEER.evolvable_subjects import EvolvingItem + from rdagent.components.coder.factor_coder.factor import FactorTask + + strat = FactorMultiProcessEvolvingStrategy(scen=MagicMock(), settings=MagicMock()) + evo = EvolvingItem(sub_tasks=[FactorTask("f1", "desc", "formula")]) + evo.sub_workspace_list = [None] + strat.assign_code_list_to_evo([None], evo) + assert evo.sub_workspace_list[0] is None # unchanged + + +# ============================================================================= +# Eurusd_llm prompt class (eurusd_llm.py) +# ============================================================================= + + +class TestEurusdLLM: + def test_eurusd_llm_importable(self): + from rdagent.components.coder.factor_coder import eurusd_llm + assert eurusd_llm is not None + + def test_eurusd_risk_importable(self): + from rdagent.components.coder.factor_coder import eurusd_risk + assert eurusd_risk is not None + + def test_eurusd_regime_importable(self): + from rdagent.components.coder.factor_coder import eurusd_regime + assert eurusd_regime is not None + + def test_eurusd_debate_importable(self): + from rdagent.components.coder.factor_coder import eurusd_debate + assert eurusd_debate is not None + + # eurusd_macro needs yfinance (optional) + # eurusd_memory needs rank_bm25 (optional) + # eurusd_reflection needs eurusd_memory (chain dependency) + + +# ============================================================================= +# model_coder/evolving_strategy.py import +# ============================================================================= + + +class TestModelEvolvingStrategy: + def test_model_evolving_strategy_importable(self): + from rdagent.components.coder.model_coder import evolving_strategy + assert evolving_strategy is not None + + +# ============================================================================= +# model_coder/eva_utils.py ModelCodeEvaluator + ModelFinalEvaluator +# ============================================================================= + + +class TestModelCodeFinalEvaluators: + def test_model_code_evaluator_init(self): + from rdagent.components.coder.model_coder.eva_utils import ModelCodeEvaluator + eva = ModelCodeEvaluator(scen=MagicMock()) + assert eva.scen is not None + + def test_model_final_evaluator_init(self): + from rdagent.components.coder.model_coder.eva_utils import ModelFinalEvaluator + eva = ModelFinalEvaluator(scen=MagicMock()) + assert eva.scen is not None diff --git a/test/qlib/test_rl_indicators.py b/test/qlib/test_rl_indicators.py new file mode 100644 index 00000000..ef95c27a --- /dev/null +++ b/test/qlib/test_rl_indicators.py @@ -0,0 +1,116 @@ +"""Tests for rl/indicators.py — pure technical indicator functions.""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np +import pandas as pd +import pytest + +PROJECT_ROOT = Path(__file__).parent.parent.parent +sys.path.insert(0, str(PROJECT_ROOT)) + + +def _load_indicators(): + import importlib.util + spec = importlib.util.spec_from_file_location( + "indicators", + PROJECT_ROOT / "rdagent/components/coder/rl/indicators.py", + ) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +@pytest.fixture(scope="module") +def indicators(): + return _load_indicators() + + +@pytest.fixture +def prices(): + rng = np.random.default_rng(42) + return pd.Series(100 + rng.normal(0, 1, 200).cumsum()) + + +class TestRSI: + def test_returns_series(self, indicators, prices): + rsi = indicators.calculate_rsi(prices, period=14) + assert isinstance(rsi, pd.Series) + assert len(rsi) == len(prices) + + def test_first_period_is_nan(self, indicators, prices): + rsi = indicators.calculate_rsi(prices, period=14) + assert rsi.iloc[:13].isna().all() + assert not np.isnan(rsi.iloc[14]) + + def test_range_between_0_and_100(self, indicators, prices): + rsi = indicators.calculate_rsi(prices, period=14) + valid = rsi.dropna() + assert (valid >= 0).all() + assert (valid <= 100).all() + + def test_constant_prices_gives_neutral_rsi(self, indicators): + const = pd.Series([100.0] * 50) + rsi = indicators.calculate_rsi(const, period=14) + # With no change, gain=loss=0 → RSI = NaN (division by zero) + valid = rsi.dropna() + assert len(valid) == 0 # all NaN when no movement + + +class TestMACD: + def test_returns_dataframe(self, indicators, prices): + macd = indicators.calculate_macd(prices) + assert isinstance(macd, pd.DataFrame) + assert list(macd.columns) == ["macd", "signal", "histogram"] + + def test_histogram_is_macd_minus_signal(self, indicators, prices): + macd = indicators.calculate_macd(prices) + computed = macd["macd"] - macd["signal"] + pd.testing.assert_series_equal(macd["histogram"], computed, check_names=False) + + +class TestBollinger: + def test_returns_dataframe(self, indicators, prices): + bb = indicators.calculate_bollinger_bands(prices, period=20) + assert isinstance(bb, pd.DataFrame) + assert list(bb.columns) == ["upper", "middle", "lower"] + + def test_upper_above_middle_lower_below(self, indicators, prices): + bb = indicators.calculate_bollinger_bands(prices, period=20) + valid = bb.dropna() + assert (valid["upper"] > valid["middle"]).all() + assert (valid["lower"] < valid["middle"]).all() + + +class TestATR: + def test_returns_series(self, indicators, prices): + high = prices * 1.01 + low = prices * 0.99 + close = prices + atr = indicators.calculate_atr(high, low, close, period=14) + assert isinstance(atr, pd.Series) + assert len(atr) == len(prices) + + def test_non_negative(self, indicators, prices): + high = prices * 1.01 + low = prices * 0.99 + atr = indicators.calculate_atr(high, low, prices, period=14) + valid = atr.dropna() + assert (valid >= 0).all() + + +class TestCCI: + def test_returns_series(self, indicators, prices): + cci = indicators.calculate_cci(prices, prices * 1.01, prices * 0.99, period=20) + assert isinstance(cci, pd.Series) + + +class TestPrepareFeatures: + def test_returns_dataframe_with_columns(self, indicators, prices): + df_input = pd.DataFrame({"close": prices}) + df = indicators.prepare_features(df_input, ["rsi", "macd"]) + assert isinstance(df, pd.DataFrame) + assert len(df.columns) >= 3 # close + at least rsi + macd columns