tradingagents/tests/test_rating_integrity.py

"""A decision is recorded as the call that was made, or as needing review.

Two readers used to disagree about the same text: the signal said REVIEW while
the memory log wrote a fabricated Hold. Worse, prose that argued against a Buy
before concluding Underweight was read as Buy, because the parser took the first
rating word anywhere in the document. A wrong direction is worse than no
direction, so an unclear decision is REVIEW everywhere.
"""

from __future__ import annotations

import pytest

from tradingagents.agents.utils.rating import RATING_REVIEW, extract_rating, parse_rating

INVERTED = ("The aggressive analyst pushed hard for a Buy on the AI backlog, but the "
            "conservative case on margin compression carried the debate. "
            "Final rating — Underweight. Trim to half weight over the next two weeks.")
REFUSAL = "I'm sorry, I can't provide a rating for this security."


@pytest.mark.unit
@pytest.mark.parametrize("separator", [":", "-", "—", "–", "：", ": **"])
def test_the_labelled_rating_wins_whatever_separates_it(separator):
    text = f"Buy arguments were raised and rejected.\n\nRating{separator}Underweight\n\nTrim."
    assert extract_rating(text) == "Underweight"


@pytest.mark.unit
def test_a_rating_argued_against_is_not_read_as_the_decision():
    assert extract_rating(INVERTED) == "Underweight"


@pytest.mark.unit
def test_prose_naming_several_ratings_without_a_label_needs_review():
    """Nothing in the text says which one is the call, so guessing risks
    reporting the opposite of the decision."""
    text = "The bull wants Buy, the bear wants Sell, and the committee was split."
    assert extract_rating(text) is None


@pytest.mark.unit
def test_prose_naming_one_rating_is_taken_as_the_call():
    assert extract_rating("On balance we stay Underweight until margins recover.") == "Underweight"


@pytest.mark.unit
def test_a_refusal_has_no_rating_and_is_not_defaulted():
    assert extract_rating(REFUSAL) is None
    assert parse_rating(REFUSAL) == RATING_REVIEW


@pytest.mark.unit
def test_the_scale_quoted_in_a_prompt_does_not_become_the_rating():
    """A free-text answer that echoes the rating scale was read as the first
    tier listed in it."""
    text = ("**Rating Scale**: Buy, Overweight, Hold, Underweight, Sell.\n\n"
            "**Rating**: Sell\n\nExit the position.")
    assert extract_rating(text) == "Sell"


# --- the readers agree ------------------------------------------------------

@pytest.mark.unit
def test_the_memory_log_records_review_rather_than_a_tradeable_hold(tmp_path):
    from tradingagents.agents.utils.memory import TradingMemoryLog

    log = TradingMemoryLog({"memory_log_path": str(tmp_path / "m.md")})
    log.store_decision("NVDA", "2026-01-05", REFUSAL)

    entry = log.load_entries()[0]
    assert entry["rating"] == RATING_REVIEW


@pytest.mark.unit
def test_the_signal_and_the_log_agree_on_the_same_decision(tmp_path):
    from tradingagents.agents.utils.memory import TradingMemoryLog
    from tradingagents.graph.signal_processing import SignalProcessor

    log = TradingMemoryLog({"memory_log_path": str(tmp_path / "m.md")})
    for text in (INVERTED, REFUSAL, "**Rating**: Buy\n\nAccumulate."):
        log.store_decision("NVDA", f"2026-01-0{len(log.load_entries()) + 1}", text)

    signals = [SignalProcessor.process_signal(None, text)
               for text in (INVERTED, REFUSAL, "**Rating**: Buy\n\nAccumulate.")]
    assert [e["rating"] for e in log.load_entries()] == signals


@pytest.mark.unit
def test_an_unscored_decision_is_left_out_of_the_backtest_figures(tmp_path):
    """REVIEW has no direction, so it cannot count for or against the system."""
    from tradingagents.agents.utils.memory import TradingMemoryLog
    from tradingagents.backtest import summarize

    log = TradingMemoryLog({"memory_log_path": str(tmp_path / "m.md")})
    log.store_decision("NVDA", "2026-01-05", "**Rating**: Buy\n\nx")
    log.update_with_outcome("NVDA", "2026-01-05", 0.1, 0.04, 5, "note", "2026-02-01")
    log.store_decision("AAPL", "2026-01-05", REFUSAL)
    log.update_with_outcome("AAPL", "2026-01-05", 0.1, 0.04, 5, "note", "2026-02-01")

    summary = summarize(log)
    assert set(summary.by_rating) == {"Buy"}


@pytest.mark.unit
def test_the_cli_says_when_a_run_produced_no_usable_rating(monkeypatch, tmp_path, capsys):
    """The CLI is the primary entry point; an unreadable decision must be
    visible there, not only in the log."""
    import cli.main as m
    from cli.models import AnalystType

    printed = []

    class _Graph:
        graph = propagator = None

        def create_run_state(self, *a, **k):
            return {"messages": []}

        def record_decision(self, *a, **k):
            pass

        def process_signal(self, text):
            from tradingagents.graph.signal_processing import SignalProcessor
            return SignalProcessor.process_signal(None, text)

        def get_graph_args(self, callbacks=None):
            return {}

        def begin_checkpoint(self, *a, **k):
            return None

        def checkpoint_input(self, state):
            return state

        def clear_checkpoint_on_success(self, *a, **k):
            pass

        def end_checkpoint(self):
            pass

        def stream(self, *a, **k):
            yield {"messages": [], "final_trade_decision": REFUSAL}

    fake = _Graph()
    fake.graph = fake
    fake.propagator = fake
    monkeypatch.setattr(m, "TradingAgentsGraph", lambda *a, **k: fake)
    monkeypatch.setattr(m, "create_layout", lambda: None)
    monkeypatch.setattr(m, "update_display", lambda *a, **k: None)
    monkeypatch.setattr(m, "Live", type("L", (), {"__init__": lambda s, *a, **k: None,
                                                  "__enter__": lambda s: s,
                                                  "__exit__": lambda s, *a: False}))
    monkeypatch.setattr(m.console, "print", lambda *a, **k: printed.append(" ".join(str(x) for x in a)))
    monkeypatch.setattr(m, "display_complete_report", lambda *a, **k: None)
    monkeypatch.setattr(m.typer, "prompt", lambda *a, **k: "N")
    monkeypatch.setattr(m, "get_user_selections", lambda: {
        "ticker": "NVDA", "analysis_date": "2026-01-10",
        "analysts": [AnalystType.MARKET], "asset_type": "stock",
    })
    monkeypatch.setattr(m, "_build_run_config", lambda s, c: {
        "data_cache_dir": str(tmp_path / "c"), "results_dir": str(tmp_path / "r")})

    m.run_analysis()

    assert any("review" in line.lower() for line in printed), printed[-5:]


@pytest.mark.unit
@pytest.mark.parametrize("module, factory, must_name", [
    ("tradingagents.agents.managers.portfolio_manager", "create_portfolio_manager", "Rating"),
    ("tradingagents.agents.managers.research_manager", "create_research_manager", "Recommendation"),
    ("tradingagents.agents.trader.trader", "create_trader", "Action"),
])
def test_a_decision_prompt_states_the_shape_of_its_answer(module, factory, must_name):
    """The field descriptions live in the schema, which a provider without
    structured output never sees. Without the format in the prompt body, the
    fallback answer is prose nobody can read a rating from."""
    import importlib

    from langchain_core.messages import AIMessage

    mod = importlib.import_module(module)
    seen = []

    class _LLM:
        def invoke(self, prompt, *a, **k):
            seen.append(prompt if isinstance(prompt, str) else str(prompt))
            return AIMessage("**Rating**: Hold\n\nnothing to do")

        def with_structured_output(self, *a, **k):
            raise NotImplementedError  # force the free-text path

    state = {
        "company_of_interest": "NVDA", "trade_date": "2026-08-14", "asset_type": "stock",
        "instrument_context": "", "market_report": "M", "sentiment_report": "S",
        "news_report": "N", "fundamentals_report": "F", "investment_plan": "P",
        "trader_investment_plan": "T", "past_context": "", "portfolio_context": "",
        "investment_debate_state": {"bull_history": "b", "bear_history": "r", "history": "h",
                                    "current_response": "", "judge_decision": "", "count": 2},
        "risk_debate_state": {"history": "h", "latest_speaker": "", "count": 3,
                              "aggressive_history": "", "conservative_history": "", "neutral_history": "",
                              "current_aggressive_response": "", "current_conservative_response": "",
                              "current_neutral_response": "", "judge_decision": ""},
    }
    getattr(mod, factory)(_LLM())(state)

    prompt = " ".join(seen)
    assert "## Output" in prompt, "no output-format section in the prompt"
    section = prompt.split("## Output", 1)[1]
    assert f"**{must_name}**" in section, section[:300]