test: add 28 tests for LLM components, RL indicators, and model evaluators

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

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