mirror of
https://github.com/NicolasBohn/NexQuant.git
synced 2026-07-27 23:47:46 +00:00
TPTBusiness
0d9b0916f2
- RiskMgmt OOS: 88 tests (leverage bounds, DD limits, trade counting, MC p-value, daily breach) - Kronos adapter: 73 tests (OHLCV idempotence, batch/sequential equivalence, forward-fill) - Auto-fixer: 78 tests (fix idempotence, MultiIndex conversion, fuzzing random patterns) - Factor coder: 65 tests (FactorTask roundtrip, evaluator invariants, workspace paths) - QLib pipeline: 61 tests (Metrics, bandit, precision matrices, noise_var) - Integration: 69 tests (portfolio weights, correlation, RiskMgmt limits, JSON roundtrip)
929 lines
37 KiB
Python
929 lines
37 KiB
Python
"""Tests for factor_coder — evaluators, task, workspace."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
PROJECT_ROOT = Path(__file__).parent.parent.parent
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
|
|
# =============================================================================
|
|
# FactorTask
|
|
# =============================================================================
|
|
|
|
|
|
class TestFactorTask:
|
|
def test_construction_fields(self):
|
|
from rdagent.components.coder.factor_coder.factor import FactorTask
|
|
t = FactorTask(
|
|
factor_name="f1",
|
|
factor_description="desc",
|
|
factor_formulation="formula",
|
|
variables={"x": 1},
|
|
resource="r1",
|
|
)
|
|
assert t.factor_name == "f1"
|
|
assert t.factor_description == "desc"
|
|
assert t.factor_formulation == "formula"
|
|
assert t.variables == {"x": 1}
|
|
assert t.factor_resources == "r1"
|
|
assert t.factor_implementation is False
|
|
assert t.base_code is None # from CoSTEERTask
|
|
|
|
def test_get_task_information(self):
|
|
from rdagent.components.coder.factor_coder.factor import FactorTask
|
|
t = FactorTask("f1", "desc", "formula", variables={"x": 1})
|
|
info = t.get_task_information()
|
|
assert "factor_name: f1" in info
|
|
assert "factor_description: desc" in info
|
|
assert "factor_formulation: formula" in info
|
|
assert "variables: {'x': 1}" in info
|
|
|
|
def test_get_task_brief_information(self):
|
|
from rdagent.components.coder.factor_coder.factor import FactorTask
|
|
t = FactorTask("f1", "desc", "formula")
|
|
info = t.get_task_brief_information()
|
|
assert "factor_name: f1" in info
|
|
|
|
def test_get_task_information_and_implementation_result(self):
|
|
from rdagent.components.coder.factor_coder.factor import FactorTask
|
|
t = FactorTask("f1", "desc", "formula")
|
|
result = t.get_task_information_and_implementation_result()
|
|
assert result["factor_name"] == "f1"
|
|
assert result["factor_description"] == "desc"
|
|
assert "factor_implementation" in result
|
|
|
|
def test_from_dict(self):
|
|
from rdagent.components.coder.factor_coder.factor import FactorTask
|
|
d = {
|
|
"factor_name": "f2",
|
|
"factor_description": "d2",
|
|
"factor_formulation": "f2",
|
|
"variables": {},
|
|
"resource": None,
|
|
"factor_implementation": True,
|
|
}
|
|
t = FactorTask.from_dict(d)
|
|
assert t.factor_name == "f2"
|
|
assert t.factor_implementation is True
|
|
|
|
def test_repr(self):
|
|
from rdagent.components.coder.factor_coder.factor import FactorTask
|
|
t = FactorTask("myfactor", "desc", "formula")
|
|
assert "FactorTask" in repr(t)
|
|
assert "myfactor" in repr(t)
|
|
|
|
|
|
# =============================================================================
|
|
# FactorFBWorkspace
|
|
# =============================================================================
|
|
|
|
|
|
class TestFactorFBWorkspace:
|
|
def test_init_sets_workspace_path(self):
|
|
from rdagent.components.coder.factor_coder.factor import FactorFBWorkspace, FactorTask
|
|
t = FactorTask("f1", "desc", "formula")
|
|
ws = FactorFBWorkspace(target_task=t)
|
|
assert ws.workspace_path is not None
|
|
# Directory is created lazily by execute(), not in __init__
|
|
assert isinstance(ws.workspace_path, Path)
|
|
|
|
def test_execute_returns_message_and_dataframe(self):
|
|
from rdagent.components.coder.factor_coder.factor import FactorFBWorkspace, FactorTask
|
|
t = FactorTask("f1", "desc", "formula")
|
|
t.version = 1
|
|
ws = FactorFBWorkspace(target_task=t)
|
|
# Inject valid factor code
|
|
ws.inject_files(**{
|
|
"factor.py": (
|
|
"import pandas as pd\n"
|
|
"import numpy as np\n"
|
|
"data = pd.read_hdf('intraday_pv.h5', key='data')\n"
|
|
"factor_val = data['$close'].pct_change()\n"
|
|
"factor_val = factor_val.to_frame('f1')\n"
|
|
"factor_val.to_hdf('result.h5', key='data', mode='w')\n"
|
|
),
|
|
})
|
|
msg, df = ws.execute()
|
|
assert isinstance(msg, str)
|
|
assert df is not None
|
|
|
|
def test_execute_succeeds_and_returns_data(self):
|
|
from rdagent.components.coder.factor_coder.factor import FactorFBWorkspace, FactorTask
|
|
t = FactorTask("fl1", "desc", "formula")
|
|
ws = FactorFBWorkspace(target_task=t)
|
|
ws.inject_files(**{
|
|
"factor.py": (
|
|
"import pandas as pd\n"
|
|
"data = pd.read_hdf('intraday_pv.h5', key='data')\n"
|
|
"factor_val = data['$close'].pct_change().to_frame('fl1')\n"
|
|
"factor_val.to_hdf('result.h5', key='data', mode='w')\n"
|
|
),
|
|
})
|
|
msg, df = ws.execute()
|
|
assert FactorFBWorkspace.FB_EXEC_SUCCESS in msg
|
|
assert FactorFBWorkspace.FB_OUTPUT_FILE_FOUND in msg
|
|
assert df is not None
|
|
|
|
|
|
# =============================================================================
|
|
# FactorEvaluatorForCoder (partial integration)
|
|
# =============================================================================
|
|
|
|
|
|
class TestFactorEvaluatorForCoder:
|
|
def test_init_creates_sub_evaluators(self):
|
|
from rdagent.components.coder.factor_coder.evaluators import FactorEvaluatorForCoder
|
|
mock_scen = MagicMock()
|
|
eva = FactorEvaluatorForCoder(scen=mock_scen)
|
|
assert eva.value_evaluator is not None
|
|
assert eva.code_evaluator is not None
|
|
assert eva.final_decision_evaluator is not None
|
|
|
|
def test_evaluate_with_none_implementation(self):
|
|
from rdagent.components.coder.factor_coder.evaluators import FactorEvaluatorForCoder
|
|
eva = FactorEvaluatorForCoder(scen=MagicMock())
|
|
assert eva.evaluate(target_task=MagicMock(), implementation=None) is None
|
|
|
|
def test_evaluate_returns_queried_knowledge_if_present(self):
|
|
from rdagent.components.coder.factor_coder.evaluators import FactorEvaluatorForCoder
|
|
from rdagent.components.coder.factor_coder.factor import FactorTask
|
|
|
|
eva = FactorEvaluatorForCoder(scen=MagicMock())
|
|
|
|
t = FactorTask("f1", "desc", "formula")
|
|
qk = MagicMock()
|
|
qk.success_task_to_knowledge_dict = {"info_f1": MagicMock(feedback="cached_fb")}
|
|
t.get_task_information = MagicMock(return_value="info_f1")
|
|
qk.failed_task_info_set = set()
|
|
|
|
fb = eva.evaluate(target_task=t, implementation=MagicMock(), queried_knowledge=qk)
|
|
assert fb == "cached_fb" # returned from cache
|
|
|
|
def test_evaluate_skips_failed_task(self):
|
|
from rdagent.components.coder.factor_coder.evaluators import FactorEvaluatorForCoder
|
|
from rdagent.components.coder.factor_coder.factor import FactorTask
|
|
|
|
eva = FactorEvaluatorForCoder(scen=MagicMock())
|
|
|
|
t = FactorTask("f1", "desc", "formula")
|
|
qk = MagicMock()
|
|
qk.success_task_to_knowledge_dict = {}
|
|
t.get_task_information = MagicMock(return_value="info_f1")
|
|
qk.failed_task_info_set = {"info_f1"}
|
|
|
|
fb = eva.evaluate(target_task=t, implementation=MagicMock(), queried_knowledge=qk)
|
|
assert fb.final_decision is False
|
|
assert "failed too many times" in fb.execution_feedback
|
|
|
|
|
|
# =============================================================================
|
|
# FactorEvaluator (eva_utils) — constructors and identity
|
|
# =============================================================================
|
|
|
|
|
|
class TestFactorEvaluatorsInit:
|
|
def test_factor_inf_evaluator_init(self):
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorInfEvaluator
|
|
eva = FactorInfEvaluator()
|
|
assert str(eva) == "FactorInfEvaluator"
|
|
|
|
def test_factor_single_column_evaluator_init(self):
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorSingleColumnEvaluator
|
|
eva = FactorSingleColumnEvaluator()
|
|
assert str(eva) == "FactorSingleColumnEvaluator"
|
|
|
|
def test_factor_output_format_evaluator_init(self):
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorOutputFormatEvaluator
|
|
eva = FactorOutputFormatEvaluator()
|
|
assert str(eva) == "FactorOutputFormatEvaluator"
|
|
|
|
def test_factor_missing_values_evaluator_init(self):
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorMissingValuesEvaluator
|
|
eva = FactorMissingValuesEvaluator()
|
|
assert str(eva) == "FactorMissingValuesEvaluator"
|
|
|
|
def test_factor_correlation_evaluator_init(self):
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorCorrelationEvaluator
|
|
eva = FactorCorrelationEvaluator(hard_check=True)
|
|
assert eva.hard_check is True
|
|
assert str(eva) == "FactorCorrelationEvaluator"
|
|
|
|
def test_factor_value_evaluator_init(self):
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorValueEvaluator
|
|
mock_scen = MagicMock()
|
|
eva = FactorValueEvaluator(mock_scen)
|
|
assert eva.scen is mock_scen
|
|
|
|
|
|
# ==============================================================================
|
|
# HYPOTHESIS-BASED PROPERTY TESTS — Code Generation Patterns, Variable
|
|
# Extraction, Evaluator Consistency
|
|
# ==============================================================================
|
|
from hypothesis import given, settings, strategies as st
|
|
import numpy as np
|
|
import pandas as pd
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock
|
|
|
|
from rdagent.components.coder.factor_coder.factor import (
|
|
FactorTask,
|
|
FactorFBWorkspace,
|
|
)
|
|
from rdagent.components.coder.factor_coder.evaluators import (
|
|
FactorEvaluatorForCoder,
|
|
)
|
|
from rdagent.components.coder.factor_coder.eva_utils import (
|
|
FactorInfEvaluator,
|
|
FactorSingleColumnEvaluator,
|
|
FactorOutputFormatEvaluator,
|
|
FactorMissingValuesEvaluator,
|
|
FactorCorrelationEvaluator,
|
|
FactorValueEvaluator,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Strategies
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _valid_factor_task_names() -> st.SearchStrategy:
|
|
return st.text(
|
|
alphabet=st.characters(whitelist_categories=("L", "N", "Lu", "Ll"), whitelist_characters="_"),
|
|
min_size=1,
|
|
max_size=50,
|
|
).filter(lambda s: s and s[0].isalpha() and " " not in s)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 1: FactorTask Field Invariants
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorTaskInvariants:
|
|
"""Property: FactorTask fields maintain invariants after construction."""
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=50).filter(lambda s: " " not in s),
|
|
factor_description=st.text(min_size=0, max_size=200),
|
|
factor_formulation=st.text(min_size=0, max_size=200),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_construction_preserves_all_fields(self, factor_name, factor_description, factor_formulation):
|
|
"""Property: all constructor args are stored as instance attributes."""
|
|
t = FactorTask(factor_name, factor_description, factor_formulation)
|
|
assert t.factor_name == factor_name
|
|
assert t.factor_description == factor_description
|
|
assert t.factor_formulation == factor_formulation
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=50).filter(lambda s: " " not in s),
|
|
factor_description=st.text(min_size=0, max_size=200),
|
|
factor_formulation=st.text(min_size=0, max_size=200),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_default_field_values(self, factor_name, factor_description, factor_formulation):
|
|
"""Property: default fields have expected values."""
|
|
t = FactorTask(factor_name, factor_description, factor_formulation)
|
|
assert t.factor_implementation is False
|
|
assert t.factor_resources is None
|
|
assert t.base_code is None
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=50).filter(lambda s: " " not in s),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_get_task_information_contains_name(self, factor_name):
|
|
"""Property: get_task_information returns string containing factor_name."""
|
|
t = FactorTask(factor_name, "desc", "formula")
|
|
info = t.get_task_information()
|
|
assert factor_name in info
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=50).filter(lambda s: " " not in s),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_get_task_brief_information_contains_name(self, factor_name):
|
|
"""Property: get_task_brief_information returns string containing factor_name."""
|
|
t = FactorTask(factor_name, "desc", "formula")
|
|
info = t.get_task_brief_information()
|
|
assert factor_name in info
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=50).filter(lambda s: " " not in s),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_get_task_information_and_implementation_result(self, factor_name):
|
|
"""Property: returned dict contains expected keys."""
|
|
t = FactorTask(factor_name, "desc", "formula")
|
|
result = t.get_task_information_and_implementation_result()
|
|
assert "factor_name" in result
|
|
assert "factor_description" in result
|
|
assert "factor_formulation" in result
|
|
assert "factor_implementation" in result
|
|
assert result["factor_name"] == factor_name
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 2: FactorTask from_dict
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorTaskFromDict:
|
|
"""Property: FactorTask.from_dict round-trip."""
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=30).filter(lambda s: s.isidentifier()),
|
|
factor_description=st.text(min_size=0, max_size=100),
|
|
factor_formulation=st.text(min_size=0, max_size=100),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_from_dict_round_trip(self, factor_name, factor_description, factor_formulation):
|
|
"""Property: constructing from dict of get_task_information_and_implementation_result preserves values."""
|
|
t1 = FactorTask(factor_name, factor_description, factor_formulation)
|
|
info = t1.get_task_information_and_implementation_result()
|
|
t2 = FactorTask.from_dict(info)
|
|
assert t2.factor_name == t1.factor_name
|
|
assert t2.factor_description == t1.factor_description
|
|
assert t2.factor_formulation == t1.factor_formulation
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=30).filter(lambda s: s.isidentifier()),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_from_dict_with_implementation(self, factor_name):
|
|
"""Property: factor_implementation field restored from dict."""
|
|
d = {
|
|
"factor_name": factor_name,
|
|
"factor_description": "desc",
|
|
"factor_formulation": "formula",
|
|
"variables": {},
|
|
"resource": None,
|
|
"factor_implementation": True,
|
|
}
|
|
t = FactorTask.from_dict(d)
|
|
assert t.factor_implementation is True
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=30).filter(lambda s: s.isidentifier()),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_from_dict_with_variables(self, factor_name):
|
|
"""Property: variables dict restored from dict."""
|
|
d = {
|
|
"factor_name": factor_name,
|
|
"factor_description": "desc",
|
|
"factor_formulation": "formula",
|
|
"variables": {"x": 1, "y": 2},
|
|
"resource": "r1",
|
|
"factor_implementation": False,
|
|
}
|
|
t = FactorTask.from_dict(d)
|
|
assert t.variables == {"x": 1, "y": 2}
|
|
assert t.factor_resources == "r1"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 3: FactorTask Repr
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorTaskRepr:
|
|
"""Property: __repr__ invariants."""
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=30).filter(lambda s: s.isidentifier()),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_repr_contains_factor_task_and_name(self, factor_name):
|
|
"""Property: repr contains 'FactorTask' and factor_name."""
|
|
t = FactorTask(factor_name, "desc", "formula")
|
|
r = repr(t)
|
|
assert "FactorTask" in r
|
|
assert factor_name in r
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=30).filter(lambda s: s.isidentifier()),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_repr_is_string(self, factor_name):
|
|
"""Property: repr returns a string."""
|
|
t = FactorTask(factor_name, "desc", "formula")
|
|
r = repr(t)
|
|
assert isinstance(r, str)
|
|
assert len(r) > 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 4: FactorTask Variables
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorTaskVariables:
|
|
"""Property: variables field invariants."""
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=20).filter(lambda s: s.isidentifier()),
|
|
vars_keys=st.lists(
|
|
st.text(min_size=1, max_size=10).filter(lambda s: s.isidentifier()),
|
|
min_size=0, max_size=10, unique=True,
|
|
),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_variables_stored_correctly(self, factor_name, vars_keys):
|
|
"""Property: variables dict stored as provided."""
|
|
vars_dict = {k: i for i, k in enumerate(vars_keys)}
|
|
t = FactorTask(factor_name, "desc", "formula", variables=vars_dict)
|
|
assert t.variables == vars_dict
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=20).filter(lambda s: s.isidentifier()),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_default_variables_is_empty_dict(self, factor_name):
|
|
"""Property: default variables is empty dict."""
|
|
t = FactorTask(factor_name, "desc", "formula")
|
|
assert t.variables == {}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 5: FactorTask Resource
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorTaskResource:
|
|
"""Property: resource field invariants."""
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=20).filter(lambda s: s.isidentifier()),
|
|
resource=st.one_of(st.none(), st.text(min_size=1, max_size=50)),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_resource_stored_correctly(self, factor_name, resource):
|
|
"""Property: resource field stored as provided or default None."""
|
|
t = FactorTask(factor_name, "desc", "formula", resource=resource)
|
|
assert t.factor_resources == resource
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 6: FactorFBWorkspace Path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorFBWorkspacePath:
|
|
"""Property: FactorFBWorkspace workspace path invariants."""
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=20).filter(lambda s: s.isidentifier()),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_workspace_path_is_valid_path(self, factor_name):
|
|
"""Property: workspace_path is a Path instance."""
|
|
t = FactorTask(factor_name, "desc", "formula")
|
|
ws = FactorFBWorkspace(target_task=t)
|
|
assert isinstance(ws.workspace_path, Path)
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=20).filter(lambda s: s.isidentifier()),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_target_task_reference_preserved(self, factor_name):
|
|
"""Property: target_task reference points back to FactorTask."""
|
|
t = FactorTask(factor_name, "desc", "formula")
|
|
ws = FactorFBWorkspace(target_task=t)
|
|
assert ws.target_task is t
|
|
assert ws.target_task.factor_name == factor_name
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 7: FactorEvaluatorForCoder Construction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorEvaluatorForCoderConstruction:
|
|
"""Property: FactorEvaluatorForCoder constructor creates sub-evaluators."""
|
|
|
|
@given(
|
|
seed=st.integers(min_value=0, max_value=100),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_sub_evaluators_are_created(self, seed):
|
|
"""Property: constructor creates value, code, and final_decision evaluators."""
|
|
mock_scen = MagicMock()
|
|
eva = FactorEvaluatorForCoder(scen=mock_scen)
|
|
assert eva.value_evaluator is not None
|
|
assert eva.code_evaluator is not None
|
|
assert eva.final_decision_evaluator is not None
|
|
|
|
@given(
|
|
seed=st.integers(min_value=0, max_value=100),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_evaluate_none_implementation_returns_none(self, seed):
|
|
"""Property: evaluate with implementation=None returns None."""
|
|
eva = FactorEvaluatorForCoder(scen=MagicMock())
|
|
result = eva.evaluate(target_task=MagicMock(), implementation=None)
|
|
assert result is None
|
|
|
|
@given(
|
|
seed=st.integers(min_value=0, max_value=100),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_scenario_reference_accessible(self, seed):
|
|
"""Property: evaluator has access to scenario."""
|
|
mock_scen = MagicMock()
|
|
eva = FactorEvaluatorForCoder(scen=mock_scen)
|
|
assert eva.scen is mock_scen
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 8: FactorEvaluator SubTypes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorEvaluatorSubTypes:
|
|
"""Property: sub-evaluator types are correct."""
|
|
|
|
@given(
|
|
seed=st.integers(min_value=0, max_value=100),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_value_evaluator_is_factor_value_evaluator(self, seed):
|
|
"""Property: value_evaluator is FactorValueEvaluator instance."""
|
|
eva = FactorEvaluatorForCoder(scen=MagicMock())
|
|
assert isinstance(eva.value_evaluator, FactorValueEvaluator)
|
|
|
|
@given(
|
|
seed=st.integers(min_value=0, max_value=100),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_scen_passed_to_value_evaluator(self, seed):
|
|
"""Property: scenario is passed to value_evaluator."""
|
|
mock_scen = MagicMock()
|
|
eva = FactorEvaluatorForCoder(scen=mock_scen)
|
|
assert eva.value_evaluator.scen is mock_scen
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 9: FactorInfEvaluator
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorInfEvaluator:
|
|
"""Property: FactorInfEvaluator invariants."""
|
|
|
|
@given(seed=st.integers(min_value=0, max_value=100))
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_str_is_correct(self, seed):
|
|
"""Property: __str__ returns 'FactorInfEvaluator'."""
|
|
eva = FactorInfEvaluator()
|
|
assert str(eva) == "FactorInfEvaluator"
|
|
|
|
@given(seed=st.integers(min_value=0, max_value=100))
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_constructor_no_args(self, seed):
|
|
"""Property: FactorInfEvaluator can be constructed without arguments."""
|
|
eva = FactorInfEvaluator()
|
|
assert eva is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 10: FactorSingleColumnEvaluator
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorSingleColumnEvaluator:
|
|
"""Property: FactorSingleColumnEvaluator invariants."""
|
|
|
|
@given(seed=st.integers(min_value=0, max_value=100))
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_str_is_correct(self, seed):
|
|
"""Property: __str__ returns 'FactorSingleColumnEvaluator'."""
|
|
eva = FactorSingleColumnEvaluator()
|
|
assert str(eva) == "FactorSingleColumnEvaluator"
|
|
|
|
@given(seed=st.integers(min_value=0, max_value=100))
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_constructor_no_args(self, seed):
|
|
"""Property: FactorSingleColumnEvaluator can be constructed without arguments."""
|
|
eva = FactorSingleColumnEvaluator()
|
|
assert eva is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 11: FactorOutputFormatEvaluator
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorOutputFormatEvaluator:
|
|
"""Property: FactorOutputFormatEvaluator invariants."""
|
|
|
|
@given(seed=st.integers(min_value=0, max_value=100))
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_str_is_correct(self, seed):
|
|
"""Property: __str__ returns 'FactorOutputFormatEvaluator'."""
|
|
eva = FactorOutputFormatEvaluator()
|
|
assert str(eva) == "FactorOutputFormatEvaluator"
|
|
|
|
@given(seed=st.integers(min_value=0, max_value=100))
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_constructor_no_args(self, seed):
|
|
"""Property: FactorOutputFormatEvaluator can be constructed without arguments."""
|
|
eva = FactorOutputFormatEvaluator()
|
|
assert eva is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 12: FactorMissingValuesEvaluator
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorMissingValuesEvaluator:
|
|
"""Property: FactorMissingValuesEvaluator invariants."""
|
|
|
|
@given(seed=st.integers(min_value=0, max_value=100))
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_str_is_correct(self, seed):
|
|
"""Property: __str__ returns 'FactorMissingValuesEvaluator'."""
|
|
eva = FactorMissingValuesEvaluator()
|
|
assert str(eva) == "FactorMissingValuesEvaluator"
|
|
|
|
@given(seed=st.integers(min_value=0, max_value=100))
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_constructor_no_args(self, seed):
|
|
"""Property: FactorMissingValuesEvaluator can be constructed without arguments."""
|
|
eva = FactorMissingValuesEvaluator()
|
|
assert eva is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 13: FactorCorrelationEvaluator
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorCorrelationEvaluator:
|
|
"""Property: FactorCorrelationEvaluator invariants."""
|
|
|
|
@given(
|
|
hard_check=st.booleans(),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_hard_check_stored_correctly(self, hard_check):
|
|
"""Property: hard_check flag stored correctly."""
|
|
eva = FactorCorrelationEvaluator(hard_check=hard_check)
|
|
assert eva.hard_check is hard_check
|
|
|
|
@given(seed=st.integers(min_value=0, max_value=100))
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_str_contains_correct_name(self, seed):
|
|
"""Property: __str__ contains 'FactorCorrelationEvaluator'."""
|
|
eva = FactorCorrelationEvaluator(hard_check=False)
|
|
assert "FactorCorrelationEvaluator" in str(eva)
|
|
|
|
@given(seed=st.integers(min_value=0, max_value=100))
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_default_hard_check_is_false(self, seed):
|
|
"""Property: hard_check parameter works."""
|
|
eva = FactorCorrelationEvaluator(hard_check=False)
|
|
assert eva.hard_check is False
|
|
eva2 = FactorCorrelationEvaluator(hard_check=True)
|
|
assert eva2.hard_check is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 14: FactorValueEvaluator
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorValueEvaluator:
|
|
"""Property: FactorValueEvaluator invariants."""
|
|
|
|
@given(seed=st.integers(min_value=0, max_value=100))
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_scenario_stored_correctly(self, seed):
|
|
"""Property: scenario reference stored."""
|
|
mock_scen = MagicMock()
|
|
eva = FactorValueEvaluator(mock_scen)
|
|
assert eva.scen is mock_scen
|
|
|
|
@given(seed=st.integers(min_value=0, max_value=100))
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_requires_scenario_arg(self, seed):
|
|
"""Property: FactorValueEvaluator requires scenario argument."""
|
|
mock_scen = MagicMock()
|
|
eva = FactorValueEvaluator(mock_scen)
|
|
assert eva is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 15: FactorTask Version
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorTaskVersion:
|
|
"""Property: version field invariants."""
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=20).filter(lambda s: s.isidentifier()),
|
|
version=st.integers(min_value=0, max_value=1000),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_version_default_and_mutable(self, factor_name, version):
|
|
"""Property: version can be set and retrieved."""
|
|
t = FactorTask(factor_name, "desc", "formula")
|
|
t.version = version
|
|
assert t.version == version
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 16: FactorTask Feedback Field
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorTaskFeedback:
|
|
"""Property: feedback-related fields."""
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=20).filter(lambda s: s.isidentifier()),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_default_implementation_is_false(self, factor_name):
|
|
"""Property: factor_implementation defaults to False."""
|
|
t = FactorTask(factor_name, "desc", "formula")
|
|
assert t.factor_implementation is False
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=20).filter(lambda s: s.isidentifier()),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_implementation_can_be_set(self, factor_name):
|
|
"""Property: factor_implementation can be set to True."""
|
|
t = FactorTask(factor_name, "desc", "formula")
|
|
t.factor_implementation = True
|
|
assert t.factor_implementation is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 17: FactorFBWorkspace FB Constants
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorFBWorkspaceConstants:
|
|
"""Property: FactorFBWorkspace class constants."""
|
|
|
|
def test_fb_exec_success_constant(self):
|
|
"""Property: FB_EXEC_SUCCESS is defined as a non-empty string."""
|
|
assert len(str(FactorFBWorkspace.FB_EXEC_SUCCESS)) > 0
|
|
|
|
def test_fb_output_file_found_constant(self):
|
|
"""Property: FB_OUTPUT_FILE_FOUND is defined as a non-empty string."""
|
|
assert len(str(FactorFBWorkspace.FB_OUTPUT_FILE_FOUND)) > 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 18: FactorTask with Variables from_dict Round-trip
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorTaskRoundTrip:
|
|
"""Property: full round-trip through from_dict preserves all data."""
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=20).filter(lambda s: s.isidentifier()),
|
|
factor_description=st.text(min_size=0, max_size=100),
|
|
factor_formulation=st.text(min_size=0, max_size=100),
|
|
n_vars=st.integers(min_value=0, max_value=10),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_to_dict_from_dict_round_trip(self, factor_name, factor_description, factor_formulation, n_vars):
|
|
"""Property: task.to_dict() → FactorTask.from_dict(d) preserves key fields."""
|
|
t1 = FactorTask(factor_name, factor_description, factor_formulation)
|
|
d = t1.get_task_information_and_implementation_result()
|
|
t2 = FactorTask.from_dict(d)
|
|
assert t2.factor_name == factor_name
|
|
assert t2.factor_description == factor_description
|
|
assert t2.factor_formulation == factor_formulation
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 19: FactorTask CoSTEERTask Inheritance
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorTaskCoSTEER:
|
|
"""Property: FactorTask inherits correctly from CoSTEERTask."""
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=20).filter(lambda s: s.isidentifier()),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_base_code_is_none_by_default(self, factor_name):
|
|
"""Property: base_code attribute is None by default (from CoSTEERTask)."""
|
|
t = FactorTask(factor_name, "desc", "formula")
|
|
assert t.base_code is None
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=20).filter(lambda s: s.isidentifier()),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_base_code_can_be_set(self, factor_name):
|
|
"""Property: base_code can be set."""
|
|
t = FactorTask(factor_name, "desc", "formula")
|
|
t.base_code = "print(42)"
|
|
assert t.base_code == "print(42)"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 20: FactorEvaluatorForCoder Caching Behavior
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestEvaluatorCaching:
|
|
"""Property: evaluator caching behavior."""
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=20).filter(lambda s: s.isidentifier()),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_cached_feedback_returned(self, factor_name):
|
|
"""Property: queried_knowledge with cached feedback returns it."""
|
|
from rdagent.components.coder.factor_coder.factor import FactorTask
|
|
|
|
eva = FactorEvaluatorForCoder(scen=MagicMock())
|
|
t = FactorTask(factor_name, "desc", "formula")
|
|
qk = MagicMock()
|
|
qk.success_task_to_knowledge_dict = {"info": MagicMock(feedback="cached")}
|
|
t.get_task_information = MagicMock(return_value="info")
|
|
qk.failed_task_info_set = set()
|
|
|
|
fb = eva.evaluate(target_task=t, implementation=MagicMock(), queried_knowledge=qk)
|
|
assert fb == "cached"
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=20).filter(lambda s: s.isidentifier()),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_failed_task_returns_negative_feedback(self, factor_name):
|
|
"""Property: failed tasks return negative feedback with 'failed too many times'."""
|
|
from rdagent.components.coder.factor_coder.factor import FactorTask
|
|
|
|
eva = FactorEvaluatorForCoder(scen=MagicMock())
|
|
t = FactorTask(factor_name, "desc", "formula")
|
|
qk = MagicMock()
|
|
qk.success_task_to_knowledge_dict = {}
|
|
t.get_task_information = MagicMock(return_value="info")
|
|
qk.failed_task_info_set = {"info"}
|
|
|
|
fb = eva.evaluate(target_task=t, implementation=MagicMock(), queried_knowledge=qk)
|
|
assert fb.final_decision is False
|
|
assert "failed too many times" in fb.execution_feedback
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 21: FactorTask Information Format
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFactorTaskInformation:
|
|
"""Property: task information output format."""
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=30).filter(lambda s: s.isidentifier()),
|
|
factor_description=st.text(min_size=0, max_size=100),
|
|
factor_formulation=st.text(min_size=0, max_size=100),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_get_task_information_format(self, factor_name, factor_description, factor_formulation):
|
|
"""Property: get_task_information has expected format."""
|
|
t = FactorTask(factor_name, factor_description, factor_formulation)
|
|
info = t.get_task_information()
|
|
assert f"factor_name: {factor_name}" in info
|
|
assert f"factor_description: {factor_description}" in info
|
|
assert f"factor_formulation: {factor_formulation}" in info
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=30).filter(lambda s: s.isidentifier()),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_get_task_information_is_string(self, factor_name):
|
|
"""Property: get_task_information returns str."""
|
|
t = FactorTask(factor_name, "desc", "formula")
|
|
info = t.get_task_information()
|
|
assert isinstance(info, str)
|
|
|
|
@given(
|
|
factor_name=st.text(min_size=1, max_size=30).filter(lambda s: s.isidentifier()),
|
|
)
|
|
@settings(max_examples=50, deadline=10000)
|
|
def test_get_task_brief_information_is_string(self, factor_name):
|
|
"""Property: get_task_brief_information returns str."""
|
|
t = FactorTask(factor_name, "desc", "formula")
|
|
info = t.get_task_brief_information()
|
|
assert isinstance(info, str)
|