mirror of
https://github.com/TauricResearch/TradingAgents.git
synced 2026-09-19 11:15:24 +03:00
8d30fee06b
- the labelled rating decides, whatever dash separates it, and a scale the model echoed is not one - prose naming several ratings is reviewed rather than read as the first word in the text - an unreadable decision is tagged REVIEW everywhere instead of a tradeable Hold - unrated decisions are counted apart from the backtest figures
212 lines
8.6 KiB
Python
212 lines
8.6 KiB
Python
"""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]
|