# source: https://raw.githubusercontent.com/liliang-cn/trade/3284879148c6634e897f00de4185f6c3898edc09/tests/test_llm_service.py """Unit tests for the LLM strategy code generator. The OpenAI client is mocked so these tests run offline. We focus on: - catalog completeness - markdown-fence stripping - static validation (syntax, IStrategy inheritance, required methods) - file write to user_data/strategies/ - failure paths still leave an audit row in llm_runs """ from __future__ import annotations import tempfile import textwrap from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock import pytest from sqlalchemy import select from llm_strategy.client import LLMClientError from llm_strategy.models import LLMRun from llm_strategy.prompts import list_strategies as list_strategy_objects from llm_strategy.schemas import StrategyId from llm_strategy.service import ( StrategyValidationError, _strip_markdown_fences, generate, list_strategies, list_strategy_files, validate_generated_code, ) from websea_copy_app.config import Settings from websea_copy_app.database import create_engine, create_session_factory, init_database # ─── Fixtures ──────────────────────────────────────────────────────────────── @pytest.fixture async def session_factory(): with tempfile.TemporaryDirectory() as tmp: db_url = f"sqlite+aiosqlite:///{Path(tmp) / 'test.db'}" engine = create_engine(db_url) await init_database(engine) factory = create_session_factory(engine) try: yield factory finally: await engine.dispose() @pytest.fixture def freqtrade_root(): with tempfile.TemporaryDirectory() as tmp: yield Path(tmp) @pytest.fixture def settings(freqtrade_root) -> Settings: return Settings( llm_api_key="test-key", llm_model="gpt-4o-mini", manager_freqtrade_root=str(freqtrade_root), ) def _valid_strategy_code(class_name: str = "Github_liliang_cn_trade__test_llm_service__20260506_125337") -> str: """A minimally-valid IStrategy implementation for tests.""" return textwrap.dedent(f"""\ from pandas import DataFrame from freqtrade.strategy import IStrategy class Github_liliang_cn_trade__test_llm_service__20260506_125337(IStrategy): INTERFACE_VERSION = 3 timeframe = "15m" can_short = False startup_candle_count = 60 process_only_new_candles = True minimal_roi = {{"0": 10}} stoploss = -0.02 def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["enter_long"] = 0 dataframe["enter_short"] = 0 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["exit_long"] = 0 dataframe["exit_short"] = 0 return dataframe """) def _mock_client(response_text: str): """Build a fake AsyncOpenAI whose chat.completions.create returns the given text.""" client = MagicMock() client.close = AsyncMock() response = SimpleNamespace( choices=[ SimpleNamespace( message=SimpleNamespace( content=response_text, refusal=None, ), finish_reason="stop", ) ], usage=SimpleNamespace(prompt_tokens=200, completion_tokens=80, total_tokens=280), ) response.model_dump = MagicMock(return_value={"id": "chatcmpl-test"}) client.chat = SimpleNamespace( completions=SimpleNamespace(create=AsyncMock(return_value=response)) ) return client # ─── Catalog ───────────────────────────────────────────────────────────────── def test_list_strategies_returns_all_five(): strategies = list_strategies() assert len(strategies) == 5 ids = {s.id for s in strategies} assert ids == { StrategyId.SHORT_GRID, StrategyId.LONG_GRID, StrategyId.UPTREND_STABLE_GRID, StrategyId.AGGRESSIVE_RANGE, StrategyId.SUPER_RISK_CONTROL, } def test_each_strategy_has_required_metadata(): for s in list_strategy_objects(): assert s.spec.strip(), f"{s.id} has empty spec" assert s.default_class_name.startswith("LLM"), f"{s.id} class_name should start with LLM" assert s.default_class_name.endswith("Strategy"), f"{s.id} class_name should end with Strategy" assert s.default_timeframe in {"1m", "5m", "15m", "1h"} # ─── Markdown stripping ────────────────────────────────────────────────────── def test_strip_markdown_fences_python(): raw = "```python\nfrom pandas import DataFrame\n```" assert _strip_markdown_fences(raw).strip() == "from pandas import DataFrame" def test_strip_markdown_fences_no_lang(): raw = "```\nx = 1\n```" assert _strip_markdown_fences(raw).strip() == "x = 1" def test_strip_markdown_fences_already_clean(): raw = "from pandas import DataFrame\n" assert _strip_markdown_fences(raw) == raw # ─── Static validation ─────────────────────────────────────────────────────── def test_validate_accepts_well_formed_code(): code = _valid_strategy_code("Github_liliang_cn_trade__test_llm_service__20260506_125337") validate_generated_code(code, expected_class_name="Github_liliang_cn_trade__test_llm_service__20260506_125337") def test_validate_rejects_syntax_error(): code = "class Foo(IStrategy:\n pass\n" # missing close paren with pytest.raises(StrategyValidationError, match="not valid Python"): validate_generated_code(code, expected_class_name="Foo") def test_validate_rejects_wrong_class_name(): code = _valid_strategy_code("Wrong") with pytest.raises(StrategyValidationError, match="expected a class named"): validate_generated_code(code, expected_class_name="Github_liliang_cn_trade__test_llm_service__20260506_125337") def test_validate_rejects_non_istrategy_base(): code = textwrap.dedent("""\ class Github_liliang_cn_trade__test_llm_service__20260506_125337: def populate_indicators(self, dataframe, metadata): return dataframe def populate_entry_trend(self, dataframe, metadata): return dataframe def populate_exit_trend(self, dataframe, metadata): return dataframe """) with pytest.raises(StrategyValidationError, match="must inherit from IStrategy"): validate_generated_code(code, expected_class_name="Github_liliang_cn_trade__test_llm_service__20260506_125337") def test_validate_rejects_missing_required_method(): code = textwrap.dedent("""\ from freqtrade.strategy import IStrategy class Github_liliang_cn_trade__test_llm_service__20260506_125337(IStrategy): def populate_indicators(self, dataframe, metadata): return dataframe def populate_entry_trend(self, dataframe, metadata): return dataframe # populate_exit_trend missing """) with pytest.raises(StrategyValidationError, match="missing required methods"): validate_generated_code(code, expected_class_name="Github_liliang_cn_trade__test_llm_service__20260506_125337") # ─── End-to-end generate() ─────────────────────────────────────────────────── async def test_generate_writes_file_and_persists_run(session_factory, settings, freqtrade_root): code = _valid_strategy_code("Github_liliang_cn_trade__test_llm_service__20260506_125337") client = _mock_client(code) async with session_factory() as session: result = await generate( session=session, settings=settings, strategy_id=StrategyId.LONG_GRID, client=client, ) assert result.class_name == "Github_liliang_cn_trade__test_llm_service__20260506_125337" expected_path = (freqtrade_root / "user_data" / "strategies" / "Github_liliang_cn_trade__test_llm_service__20260506_125337.py").resolve() assert Path(result.file_path) == expected_path assert expected_path.exists() contents = expected_path.read_text(encoding="utf-8") assert "# llm-generated:" in contents assert "class Github_liliang_cn_trade__test_llm_service__20260506_125337(IStrategy)" in contents assert result.usage.total_tokens == 280 async with session_factory() as session: rows = (await session.execute(select(LLMRun))).scalars().all() assert len(rows) == 1 assert rows[0].error is None assert rows[0].output_json["class_name"] == "Github_liliang_cn_trade__test_llm_service__20260506_125337" assert Path(rows[0].output_json["file_path"]) == expected_path async def test_generate_strips_markdown_fences(session_factory, settings, freqtrade_root): raw_code = _valid_strategy_code("Github_liliang_cn_trade__test_llm_service__20260506_125337") fenced = "```python\n" + raw_code.rstrip("\n") + "\n```" client = _mock_client(fenced) async with session_factory() as session: result = await generate( session=session, settings=settings, strategy_id=StrategyId.SHORT_GRID, client=client, ) written = Path(result.file_path).read_text(encoding="utf-8") assert "```" not in written async def test_generate_refuses_existing_file_without_overwrite(session_factory, settings, freqtrade_root): target = freqtrade_root / "user_data" / "strategies" / "Github_liliang_cn_trade__test_llm_service__20260506_125337.py" target.parent.mkdir(parents=True, exist_ok=True) target.write_text("# preexisting\n", encoding="utf-8") with pytest.raises(StrategyValidationError, match="already exists"): async with session_factory() as session: await generate( session=session, settings=settings, strategy_id=StrategyId.LONG_GRID, client=_mock_client(_valid_strategy_code("Github_liliang_cn_trade__test_llm_service__20260506_125337")), ) # File untouched assert target.read_text(encoding="utf-8") == "# preexisting\n" async def test_generate_overwrites_when_explicitly_allowed(session_factory, settings, freqtrade_root): target = freqtrade_root / "user_data" / "strategies" / "Github_liliang_cn_trade__test_llm_service__20260506_125337.py" target.parent.mkdir(parents=True, exist_ok=True) target.write_text("# preexisting\n", encoding="utf-8") client = _mock_client(_valid_strategy_code("Github_liliang_cn_trade__test_llm_service__20260506_125337")) async with session_factory() as session: await generate( session=session, settings=settings, strategy_id=StrategyId.LONG_GRID, overwrite=True, client=client, ) assert "preexisting" not in target.read_text(encoding="utf-8") assert "class Github_liliang_cn_trade__test_llm_service__20260506_125337" in target.read_text(encoding="utf-8") async def test_generate_records_validation_failure_in_db(session_factory, settings, freqtrade_root): """If the LLM emits broken code, the request should fail loudly but the run row should still be persisted with error + raw response, so we can debug what the model said.""" bad_code = "this is not python code" client = _mock_client(bad_code) with pytest.raises(StrategyValidationError): async with session_factory() as session: await generate( session=session, settings=settings, strategy_id=StrategyId.LONG_GRID, client=client, ) async with session_factory() as session: rows = (await session.execute(select(LLMRun))).scalars().all() assert len(rows) == 1 assert "validation failed" in rows[0].error # File NOT written assert not (freqtrade_root / "user_data" / "strategies" / "Github_liliang_cn_trade__test_llm_service__20260506_125337.py").exists() async def test_generate_records_llm_error_in_db(session_factory, settings): client = MagicMock() client.close = AsyncMock() client.chat = SimpleNamespace( completions=SimpleNamespace( create=AsyncMock(side_effect=RuntimeError("rate_limit_exceeded")), ) ) with pytest.raises(LLMClientError): async with session_factory() as session: await generate( session=session, settings=settings, strategy_id=StrategyId.SHORT_GRID, client=client, ) async with session_factory() as session: rows = (await session.execute(select(LLMRun))).scalars().all() assert len(rows) == 1 assert "rate_limit_exceeded" in rows[0].error async def test_generate_rejects_invalid_class_name(session_factory, settings): with pytest.raises(StrategyValidationError, match="not a valid Python identifier"): async with session_factory() as session: await generate( session=session, settings=settings, strategy_id=StrategyId.LONG_GRID, class_name="123-not-valid", client=_mock_client("..."), ) # ─── File listing ──────────────────────────────────────────────────────────── async def test_list_strategy_files_marks_llm_generated(session_factory, settings, freqtrade_root): code = _valid_strategy_code("Github_liliang_cn_trade__test_llm_service__20260506_125337") async with session_factory() as session: await generate( session=session, settings=settings, strategy_id=StrategyId.LONG_GRID, client=_mock_client(code), ) # Drop in a file we did NOT generate, to test the marker logic. other = freqtrade_root / "user_data" / "strategies" / "Github_liliang_cn_trade__test_llm_service__20260506_125337.py" other.write_text("from freqtrade.strategy import IStrategy\nclass Github_liliang_cn_trade__test_llm_service__20260506_125337(IStrategy):\n pass\n", encoding="utf-8") files = list_strategy_files(settings) by_name = {f.file_name: f for f in files} assert by_name["Github_liliang_cn_trade__test_llm_service__20260506_125337.py"].is_llm_generated is True assert by_name["Github_liliang_cn_trade__test_llm_service__20260506_125337.py"].class_name == "Github_liliang_cn_trade__test_llm_service__20260506_125337" assert by_name["Github_liliang_cn_trade__test_llm_service__20260506_125337.py"].is_llm_generated is False def test_missing_api_key_raises_when_no_client_provided(freqtrade_root): from llm_strategy.client import _build_client # Explicit override — the user's local .env has LLM_API_KEY set. no_key_settings = Settings(llm_api_key=None, manager_freqtrade_root=str(freqtrade_root)) with pytest.raises(LLMClientError): _build_client(no_key_settings)