fix(rating): surface an unparseable rating as REVIEW, not a silent Hold

- an unrecognizable Portfolio Manager decision was coerced to Hold, emitting a
  tradeable neutral signal that masked a parsing failure; a fullwidth colon
  (Rating:X) defeated the label regex and hit the same path
- add extract_rating() -> str | None with NFKC normalization and whole-word
  matching; the graph signal now yields a REVIEW sentinel (with an is_review
  guard) when no rating is found
- parse_rating keeps its silent default for compat callers (e.g. the memory log) #1170
This commit is contained in:
Yijia-Xiao
2026-08-30 06:38:15 +00:00
parent 0ef56e6a33
commit 43fc275b36
4 changed files with 118 additions and 19 deletions

View File

@@ -10,7 +10,13 @@ to it.
import pytest import pytest
from tradingagents.agents.utils.rating import RATINGS_5_TIER, parse_rating from tradingagents.agents.utils.rating import (
RATING_REVIEW,
RATINGS_5_TIER,
extract_rating,
is_review,
parse_rating,
)
from tradingagents.graph.signal_processing import SignalProcessor from tradingagents.graph.signal_processing import SignalProcessor
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -84,6 +90,51 @@ class TestSignalProcessor:
llm.invoke.assert_not_called() llm.invoke.assert_not_called()
llm.with_structured_output.assert_not_called() llm.with_structured_output.assert_not_called()
def test_default_when_no_rating_present(self): def test_unparseable_signal_is_review_not_silent_hold(self):
# #1170: an unrecognizable decision must surface REVIEW, not a fabricated
# tradeable Hold.
sp = SignalProcessor() sp = SignalProcessor()
assert sp.process_signal("Plain prose without a recommendation.") == "Hold" signal = sp.process_signal("Plain prose without a recommendation.")
assert signal == RATING_REVIEW
assert is_review(signal)
assert signal not in RATINGS_5_TIER
def test_fullwidth_colon_is_parsed_not_reviewed(self):
# #1170: `RatingOverweight` (fullwidth colon) used to defeat the regex
# and silently become Hold; NFKC normalization now parses it.
sp = SignalProcessor()
assert sp.process_signal("RatingOverweight\n理由はこちら。") == "Overweight"
@pytest.mark.unit
class TestExtractRating:
def test_returns_none_when_absent(self):
assert extract_rating("No directional call here.") is None
assert extract_rating("") is None
def test_whole_word_only(self):
# substrings inside larger words must not match
assert extract_rating("The buyer was holding shares.") is None
def test_parse_rating_keeps_silent_default_for_compat(self):
# parse_rating (used by the memory log) intentionally keeps Hold default.
assert parse_rating("No rating here.") == "Hold"
assert parse_rating("No rating here.", default="Underweight") == "Underweight"
@pytest.mark.unit
class TestGraphSignalContract:
"""The graph-facing signal (TradingAgentsGraph.process_signal) honors the
documented "5-tier or REVIEW" contract, not just the parser in isolation."""
def _bare_graph(self):
from tradingagents.graph.trading_graph import TradingAgentsGraph
g = object.__new__(TradingAgentsGraph)
g.signal_processor = SignalProcessor()
return g
def test_graph_surfaces_review(self):
assert self._bare_graph().process_signal("no rating in here") == RATING_REVIEW
def test_graph_returns_rating(self):
assert self._bare_graph().process_signal("**Rating**: Sell") == "Sell"

View File

@@ -7,42 +7,77 @@ The same five-tier scale (Buy, Overweight, Hold, Underweight, Sell) is used by:
- The memory log (rating tag stored alongside each decision entry) - The memory log (rating tag stored alongside each decision entry)
Centralising it here avoids drift between those call sites. Centralising it here avoids drift between those call sites.
``extract_rating`` returns ``None`` when no rating can be found, so the graph can
surface an explicit ``REVIEW`` signal instead of a fabricated ``Hold`` (#1170).
``parse_rating`` keeps the legacy silent-default behaviour for callers (e.g. the
memory log) that need a rating string regardless.
""" """
from __future__ import annotations from __future__ import annotations
import re import re
import unicodedata
# Canonical, ordered 5-tier scale (most bullish to most bearish). # Canonical, ordered 5-tier scale (most bullish to most bearish).
RATINGS_5_TIER: tuple[str, ...] = ( RATINGS_5_TIER: tuple[str, ...] = (
"Buy", "Overweight", "Hold", "Underweight", "Sell", "Buy", "Overweight", "Hold", "Underweight", "Sell",
) )
# Signal emitted when the model's decision has no recognizable rating. It is not
# a tradeable position: it flags output that needs a human/re-run rather than
# silently degrading to Hold. Callers that map the signal onto the 5-tier enum
# (e.g. ``PortfolioRating(signal)``) should guard with ``is_review`` first.
RATING_REVIEW = "REVIEW"
_RATING_SET = {r.lower() for r in RATINGS_5_TIER} _RATING_SET = {r.lower() for r in RATINGS_5_TIER}
# Matches "Rating: X" / "rating - X" / "Rating: **X**" — tolerates markdown # Matches "Rating: X" / "rating - X" / "Rating: **X**" — tolerates markdown
# bold wrappers and either a colon or hyphen separator. # bold wrappers and either a colon or hyphen separator.
_RATING_LABEL_RE = re.compile(r"rating.*?[:\-][\s*]*(\w+)", re.IGNORECASE) _RATING_LABEL_RE = re.compile(r"rating.*?[:\-][\s*]*(\w+)", re.IGNORECASE)
# Standalone 5-tier word anywhere (word boundaries so "Buyer"/"Holding" don't match).
_RATING_WORD_RE = re.compile(
r"\b(" + "|".join(RATINGS_5_TIER) + r")\b", re.IGNORECASE
)
def parse_rating(text: str, default: str = "Hold") -> str:
"""Heuristically extract a 5-tier rating from prose text.
Two-pass strategy: def extract_rating(text: str) -> str | None:
1. Look for an explicit "Rating: X" label (tolerant of markdown bold). """Extract a 5-tier rating from prose, or ``None`` if none is present.
2. Fall back to the first 5-tier rating word found anywhere in the text.
Returns a Title-cased rating string, or ``default`` if no rating word appears. Two-pass strategy on the NFKC-normalized text (so fullwidth punctuation like
``RatingOverweight`` is matched the same as ASCII):
1. An explicit "Rating: X" label (tolerant of markdown bold).
2. The first standalone 5-tier rating word found anywhere.
""" """
for line in text.splitlines(): if not text:
return None
norm = unicodedata.normalize("NFKC", text)
for line in norm.splitlines():
m = _RATING_LABEL_RE.search(line) m = _RATING_LABEL_RE.search(line)
if m and m.group(1).lower() in _RATING_SET: if m and m.group(1).lower() in _RATING_SET:
return m.group(1).capitalize() return m.group(1).capitalize()
for line in text.splitlines(): m = _RATING_WORD_RE.search(norm)
for word in line.lower().split(): if m:
clean = word.strip("*:.,") return m.group(1).capitalize()
if clean in _RATING_SET:
return clean.capitalize()
return default return None
def parse_rating(text: str, default: str = "Hold") -> str:
"""Extract a 5-tier rating, falling back to ``default`` when none is found.
Legacy convenience wrapper: it always returns a rating string, so an
unparseable decision silently becomes ``default`` (``Hold``). Callers that
must distinguish "no rating" from a real Hold should use
:func:`extract_rating` (or the graph's REVIEW-surfacing signal) instead.
"""
rating = extract_rating(text)
return rating if rating is not None else default
def is_review(signal: str) -> bool:
"""Whether a signal is the non-tradeable REVIEW sentinel (#1170)."""
return signal == RATING_REVIEW

View File

@@ -14,7 +14,7 @@ from __future__ import annotations
from typing import Any from typing import Any
from tradingagents.agents.utils.rating import parse_rating from tradingagents.agents.utils.rating import RATING_REVIEW, extract_rating
class SignalProcessor: class SignalProcessor:
@@ -27,5 +27,12 @@ class SignalProcessor:
self.quick_thinking_llm = quick_thinking_llm self.quick_thinking_llm = quick_thinking_llm
def process_signal(self, full_signal: str) -> str: def process_signal(self, full_signal: str) -> str:
"""Return one of Buy / Overweight / Hold / Underweight / Sell.""" """Return one of Buy / Overweight / Hold / Underweight / Sell, or REVIEW.
return parse_rating(full_signal)
An unrecognizable decision yields ``REVIEW`` rather than a fabricated
``Hold``, so a parsing failure is visible instead of masquerading as a
tradeable neutral signal (#1170). Consumers that map the result onto the
5-tier enum should guard with :func:`~tradingagents.agents.utils.rating.is_review`.
"""
rating = extract_rating(full_signal)
return rating if rating is not None else RATING_REVIEW

View File

@@ -388,6 +388,12 @@ class TradingAgentsGraph:
``checkpoint_enabled`` is set in config, the graph is recompiled with ``checkpoint_enabled`` is set in config, the graph is recompiled with
a per-ticker SqliteSaver so a crashed run can resume from the last a per-ticker SqliteSaver so a crashed run can resume from the last
successful node on a subsequent invocation with the same ticker+date. successful node on a subsequent invocation with the same ticker+date.
Returns ``(final_state, signal)`` where ``signal`` is one of the 5-tier
ratings (Buy / Overweight / Hold / Underweight / Sell) or ``"REVIEW"``
when the decision had no parseable rating (#1170); guard with
``tradingagents.agents.utils.rating.is_review`` before mapping it to the
PortfolioRating enum.
""" """
self.ticker = company_name self.ticker = company_name