charieswei
/
TradingAgents
mirror of https://github.com/TauricResearch/TradingAgents.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat: Fast-Reject [CRITICAL ABORT] short-circuit mechanism (#128) 

* chore: initial commit for feat/fast-reject-short-circuits branch

* test: fix fast-reject unit tests (25/25 passing)

- Fix test_normal_flow_without_abort: expected 'Bull Researcher' not
  'Bear Researcher' — when current_response is empty the conditional
  logic routes to Bull first (conditional_logic.py line 74)
- Analyst abort-instruction tests: replace closure-attribute hacks
  (market_analyst.llm = ...) with patch() on prefetch_tools_parallel
  and run_tool_loop so the tests run without network access
- Fix co_consts substring check: system_message is compiled into one
  large string constant, so use any(...in str(c)...) instead of direct
  tuple membership
- PM tests: create mock_llm before create_portfolio_manager so the
  closure captures it and mock_llm.invoke.called assertions work; add
  missing company_of_interest key to all state dicts
- Integration tests: merge analyst/PM node outputs back into state
  ({**state, **result}) instead of replacing it, preserving keys like
  fundamentals_report and investment_debate_state; patch network calls;
  fix normal-flow routing assertions to != 'Portfolio Manager' since
  the exact next node depends on transient debate state
This commit is contained in:
ahmet guzererler 2026-03-27 00:19:43 +01:00 committed by GitHub
parent 02cbaecf62
commit c1194b7f77
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 762 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
.kilocodemodes Normal file
View File

@ -0,0 +1,24 @@
customModes:
- slug: code-reviewer
name: Code Reviewer
roleDefinition: |
You are a senior software engineer conducting thorough code reviews. You focus on code quality, security, performance, and maintainability.
groups:
- read
- browser
customInstructions: |
Provide constructive feedback on code patterns, potential bugs, security issues, and improvement opportunities. Be specific and actionable in suggestions.
source: project
- slug: frontend-specialist
name: Frontend Specialist
roleDefinition: |
You are a frontend developer expert in React, TypeScript, and modern CSS. You focus on creating intuitive user interfaces and excellent user experiences.
groups:
- read
- browser
- - edit
- fileRegex: \.(tsx?|jsx?|css|scss|less)$
description: Frontend files only
customInstructions: |
Prioritize accessibility, responsive design, and performance. Use semantic HTML and follow React best practices.
source: project

53
docs/agent/decisions/019-fast-reject-short-circuits.md Normal file
View File

@ -0,0 +1,53 @@
---
type: decision
status: active
date: 2026-03-26
agent_author: "kilocode"
tags: [performance, optimization, short-circuit, critical-abort]
related_files: [tradingagents/graph/conditional_logic.py, tradingagents/agents/analysts/fundamentals_analyst.py, tradingagents/agents/analysts/market_analyst.py, tradingagents/agents/managers/portfolio_manager.py, tradingagents/agents/utils/agent_states.py]
---
## Context
The TradingAgentsGraph currently forces every stock through the entire analysis pipeline: Market Analyst → Social → News → Fundamentals → Bull/Bear Debate → Risk Debate → Portfolio Manager.
If a stock is fundamentally bankrupt or facing catastrophic SEC delisting, we waste API tokens and time running 5 LLMs through debate rounds. This is inefficient and costly, especially when early analysis clearly indicates the stock should be avoided.
## The Decision
Implement a [CRITICAL ABORT] trigger mechanism that allows early analysts (Fundamentals_Analyst, Market_Analyst) to short-circuit the pipeline and route directly to Portfolio_Manager for an immediate AVOID/SELL decision.
The trigger will be a special string "[CRITICAL ABORT]" embedded in analyst reports that signals the system to bypass remaining analysis phases and proceed directly to portfolio decision-making.
## Constraints
- Must maintain backward compatibility with existing workflows
- Should not interfere with normal analysis flow for healthy stocks
- Must be detectable by conditional logic without breaking existing state transitions
- Should work with both Market Analyst and Fundamentals Analyst
- The Portfolio Manager must recognize and properly handle critical abort scenarios
## Actionable Rules
1. **Trigger Format**: Analysts should include the exact string "[CRITICAL ABORT]" in their report when they detect catastrophic conditions (bankruptcy, SEC delisting, etc.)
2. **Eligible Analysts**: Only Market_Analyst and Fundamentals_Analyst can trigger critical aborts, as they are the earliest in the pipeline and can identify fundamental issues
3. **Routing Logic**:
- Modify conditional_logic.py to check for "[CRITICAL ABORT]" in market_report or fundamentals_report
- When detected, bypass Social, News, and Debate phases and route directly to Portfolio Manager
- The Portfolio Manager should receive a special context indicating this is a critical abort scenario
4. **Portfolio Manager Handling**:
- When receiving a critical abort signal, the Portfolio Manager should automatically recommend "SELL" or "AVOID"
- The decision should include reasoning based on the aborting analyst's report
- No further debate or risk analysis should be performed
5. **State Preservation**:
- The aborting analyst's report should be preserved in the state
- Other report fields can be left empty or marked as "SKIPPED DUE TO CRITICAL ABORT"
- Investment debate state and risk debate state should reflect that these phases were skipped
6. **Logging and Monitoring**:
- Critical abort events should be logged for audit purposes
- Metrics should track the number of aborts vs full pipeline executions

565
tests/unit/test_fast_reject.py Normal file
View File

@ -0,0 +1,565 @@
"""Comprehensive unit tests for Fast-Reject [CRITICAL ABORT] feature.
This module tests the critical abort mechanism that short-circuits the trading agent
workflow when catastrophic conditions are detected in market or fundamentals reports.
"""
import pytest
from unittest.mock import MagicMock, patch
from langchain_core.messages import AIMessage
from tradingagents.agents.analysts.fundamentals_analyst import create_fundamentals_analyst
from tradingagents.agents.analysts.market_analyst import create_market_analyst
from tradingagents.agents.managers.portfolio_manager import create_portfolio_manager
from tradingagents.graph.conditional_logic import ConditionalLogic
# ---------------------------------------------------------------------------
# Mock Data
# ---------------------------------------------------------------------------
# Market report with abort
market_report_abort = "[CRITICAL ABORT] Reason: Trading halted pending SEC investigation"
# Fundamentals report with abort
fundamentals_report_abort = "[CRITICAL ABORT] Reason: Negative gross margin with bankruptcy filing"
# Normal market report
normal_market_report = "Market analysis shows strong bullish trend with positive momentum..."
# Normal fundamentals report
normal_fundamentals_report = "Company fundamentals are strong with healthy margins and growth prospects..."
# Macro regime report
macro_regime_report = "Current macro environment shows stable interest rates and moderate inflation."
# ---------------------------------------------------------------------------
# ConditionalLogic Tests
# ---------------------------------------------------------------------------
class TestConditionalLogicAbortDetection:
"""Tests for critical abort detection in ConditionalLogic."""
def test_check_critical_abort_detected_in_market_report(self):
"""Verify abort is detected in market_report."""
cl = ConditionalLogic()
state = {
"market_report": market_report_abort,
"fundamentals_report": normal_fundamentals_report,
}
result = cl._check_critical_abort(state, "market_report")
assert result is True
def test_check_critical_abort_detected_in_fundamentals_report(self):
"""Verify abort is detected in fundamentals_report."""
cl = ConditionalLogic()
state = {
"market_report": normal_market_report,
"fundamentals_report": fundamentals_report_abort,
}
result = cl._check_critical_abort(state, "fundamentals_report")
assert result is True
def test_check_critical_abort_not_detected(self):
"""Verify normal reports pass through without abort detection."""
cl = ConditionalLogic()
state = {
"market_report": normal_market_report,
"fundamentals_report": normal_fundamentals_report,
}
result = cl._check_critical_abort(state, "market_report")
assert result is False
def test_check_critical_abort_empty_report(self):
"""Verify abort is not detected when report field is empty."""
cl = ConditionalLogic()
state = {
"market_report": "",
"fundamentals_report": normal_fundamentals_report,
}
result = cl._check_critical_abort(state, "market_report")
assert result is False
def test_check_critical_abort_missing_report_field(self):
"""Verify abort is not detected when report field is missing."""
cl = ConditionalLogic()
state = {
"fundamentals_report": normal_fundamentals_report,
}
result = cl._check_critical_abort(state, "market_report")
assert result is False
def test_check_critical_abort_partial_match(self):
"""Verify abort is detected even with partial match."""
cl = ConditionalLogic()
state = {
"market_report": "Some text [CRITICAL ABORT] Reason: Test",
"fundamentals_report": normal_fundamentals_report,
}
result = cl._check_critical_abort(state, "market_report")
assert result is True
class TestConditionalLogicFlowControl:
"""Tests for flow control when abort is detected."""
def test_should_continue_debate_with_abort(self):
"""Verify debate is bypassed when abort detected."""
cl = ConditionalLogic()
state = {
"market_report": market_report_abort,
"fundamentals_report": normal_fundamentals_report,
"investment_debate_state": {
"history": [],
"bull_history": [],
"bear_history": [],
"current_response": "",
"judge_decision": "",
"count": 0,
},
}
result = cl.should_continue_debate(state)
assert result == "Portfolio Manager"
def test_should_continue_risk_analysis_with_abort(self):
"""Verify risk analysis is bypassed when abort detected."""
cl = ConditionalLogic()
state = {
"market_report": market_report_abort,
"fundamentals_report": normal_fundamentals_report,
"risk_debate_state": {
"history": [],
"aggressive_history": [],
"conservative_history": [],
"neutral_history": [],
"latest_speaker": "Aggressive",
"current_aggressive_response": "",
"current_conservative_response": "",
"current_neutral_response": "",
"judge_decision": "",
"count": 0,
},
}
result = cl.should_continue_risk_analysis(state)
assert result == "Portfolio Manager"
def test_normal_flow_without_abort(self):
"""Verify normal flow continues when no abort detected."""
cl = ConditionalLogic()
state = {
"market_report": normal_market_report,
"fundamentals_report": normal_fundamentals_report,
"investment_debate_state": {
"history": [],
"bull_history": [],
"bear_history": [],
"current_response": "",
"judge_decision": "",
"count": 0,
},
}
result = cl.should_continue_debate(state)
assert result == "Bull Researcher" # Bull speaks first when current_response is empty
def test_normal_flow_without_abort_risk_analysis(self):
"""Verify normal risk analysis flow continues when no abort detected."""
cl = ConditionalLogic()
state = {
"market_report": normal_market_report,
"fundamentals_report": normal_fundamentals_report,
"risk_debate_state": {
"history": [],
"aggressive_history": [],
"conservative_history": [],
"neutral_history": [],
"latest_speaker": "Aggressive",
"count": 0,
},
}
result = cl.should_continue_risk_analysis(state)
assert result == "Conservative Analyst"
def test_abort_in_fundamentals_bypasses_debate(self):
"""Verify debate is bypassed when fundamentals report contains abort."""
cl = ConditionalLogic()
state = {
"market_report": normal_market_report,
"fundamentals_report": fundamentals_report_abort,
"investment_debate_state": {
"history": [],
"bull_history": [],
"bear_history": [],
"current_response": "",
"judge_decision": "",
"count": 0,
},
}
result = cl.should_continue_debate(state)
assert result == "Portfolio Manager"
def test_abort_in_fundamentals_bypasses_risk_analysis(self):
"""Verify risk analysis is bypassed when fundamentals report contains abort."""
cl = ConditionalLogic()
state = {
"market_report": normal_market_report,
"fundamentals_report": fundamentals_report_abort,
"risk_debate_state": {
"history": [],
"aggressive_history": [],
"conservative_history": [],
"neutral_history": [],
"latest_speaker": "Aggressive",
"current_aggressive_response": "",
"current_conservative_response": "",
"current_neutral_response": "",
"judge_decision": "",
"count": 0,
},
}
result = cl.should_continue_risk_analysis(state)
assert result == "Portfolio Manager"
def test_abort_in_market_bypasses_risk_analysis(self):
"""Verify market abort bypasses risk analysis."""
cl = ConditionalLogic()
state = {
"market_report": market_report_abort,
"fundamentals_report": normal_fundamentals_report,
"risk_debate_state": {
"history": [],
"aggressive_history": [],
"conservative_history": [],
"neutral_history": [],
"latest_speaker": "Aggressive",
"current_aggressive_response": "",
"current_conservative_response": "",
"current_neutral_response": "",
"judge_decision": "",
"count": 0,
},
}
result = cl.should_continue_risk_analysis(state)
assert result == "Portfolio Manager"
# ---------------------------------------------------------------------------
# Analyst Report Tests
# ---------------------------------------------------------------------------
class TestMarketAnalystAbortInstructions:
"""Tests for market analyst abort instructions in system prompt."""
def test_market_analyst_includes_abort_instructions(self):
"""Verify market analyst produces abort report when LLM signals critical abort."""
# run_tool_loop is the injectable boundary — patch it to return the abort message
# without making any network calls.
mock_result = MagicMock()
mock_result.content = market_report_abort
mock_result.tool_calls = []
with patch("tradingagents.agents.analysts.market_analyst.prefetch_tools_parallel", return_value={}), \
patch("tradingagents.agents.analysts.market_analyst.run_tool_loop", return_value=mock_result):
market_analyst = create_market_analyst(MagicMock())
state = {
"trade_date": "2024-01-01",
"company_of_interest": "AAPL",
"messages": [],
}
result = market_analyst(state)
# Verify the report contains abort
assert "[CRITICAL ABORT]" in result.get("market_report", "")
def test_market_analyst_abort_conditions(self):
"""Verify market analyst abort conditions are documented in the system prompt constant."""
market_analyst = create_market_analyst(MagicMock())
# The system_message is built from adjacent string literals that the compiler
# concatenates into one big string constant stored in co_consts.
# Check that at least one constant contains the trigger phrase as a substring.
assert any(
"CRITICAL ABORT TRIGGER" in str(c)
for c in market_analyst.__code__.co_consts
)
class TestFundamentalsAnalystAbortInstructions:
"""Tests for fundamentals analyst abort instructions in system prompt."""
def test_fundamentals_analyst_includes_abort_instructions(self):
"""Verify fundamentals analyst produces abort report when LLM signals critical abort."""
mock_result = MagicMock()
mock_result.content = fundamentals_report_abort
mock_result.tool_calls = []
with patch("tradingagents.agents.analysts.fundamentals_analyst.prefetch_tools_parallel", return_value={}), \
patch("tradingagents.agents.analysts.fundamentals_analyst.run_tool_loop", return_value=mock_result):
fundamentals_analyst = create_fundamentals_analyst(MagicMock())
state = {
"trade_date": "2024-01-01",
"company_of_interest": "AAPL",
"messages": [],
}
result = fundamentals_analyst(state)
# Verify the report contains abort
assert "[CRITICAL ABORT]" in result.get("fundamentals_report", "")
def test_fundamentals_analyst_abort_conditions(self):
"""Verify fundamentals analyst abort conditions are documented in the system prompt constant."""
fundamentals_analyst = create_fundamentals_analyst(MagicMock())
# The system_message is built from adjacent string literals compiled into one constant.
assert any(
"CRITICAL ABORT TRIGGER" in str(c)
for c in fundamentals_analyst.__code__.co_consts
)
# ---------------------------------------------------------------------------
# Portfolio Manager Tests
# ---------------------------------------------------------------------------
class TestPortfolioManagerAbortDetection:
"""Tests for portfolio manager abort detection and response."""
def _make_abort_state(self, market_report, fundamentals_report):
"""Build a minimal state dict suitable for portfolio_manager_node."""
return {
"company_of_interest": "AAPL",
"market_report": market_report,
"fundamentals_report": fundamentals_report,
"macro_regime_report": macro_regime_report,
"risk_debate_state": {
"history": [],
"aggressive_history": [],
"conservative_history": [],
"neutral_history": [],
"current_aggressive_response": "",
"current_conservative_response": "",
"current_neutral_response": "",
"count": 0,
},
"news_report": "",
"sentiment_report": "",
"investment_plan": "BUY AAPL",
}
def test_portfolio_manager_detects_abort(self):
"""Verify PM detects abort and recommends SELL/AVOID."""
# Create mock LLM *before* the closure so the closure captures it.
mock_llm = MagicMock()
mock_llm.invoke.return_value = MagicMock(
content="RECOMMENDATION: SELL - Trading halted pending SEC investigation"
)
portfolio_manager = create_portfolio_manager(mock_llm, MagicMock())
state = self._make_abort_state(market_report_abort, normal_fundamentals_report)
result = portfolio_manager(state)
# Verify the closure's LLM was actually called
assert mock_llm.invoke.called
# Verify the result contains SELL recommendation
assert "SELL" in result.get("final_trade_decision", "").upper()
def test_portfolio_manager_uses_aborting_analyst_report(self):
"""Verify PM decision text reflects the abort reason from the analyst report."""
mock_llm = MagicMock()
mock_llm.invoke.return_value = MagicMock(
content="RECOMMENDATION: SELL - Trading halted pending SEC investigation"
)
portfolio_manager = create_portfolio_manager(mock_llm, MagicMock())
state = self._make_abort_state(market_report_abort, normal_fundamentals_report)
result = portfolio_manager(state)
recommendation = result.get("final_trade_decision", "")
assert "SEC investigation" in recommendation
def test_portfolio_manager_normal_flow(self):
"""Verify PM works normally without abort."""
mock_llm = MagicMock()
mock_llm.invoke.return_value = MagicMock(
content="RECOMMENDATION: BUY - Strong bullish trend with positive momentum"
)
portfolio_manager = create_portfolio_manager(mock_llm, MagicMock())
state = self._make_abort_state(normal_market_report, normal_fundamentals_report)
result = portfolio_manager(state)
# Verify the closure's LLM was actually called
assert mock_llm.invoke.called
# Verify the result contains BUY recommendation
assert "BUY" in result.get("final_trade_decision", "").upper()
def test_portfolio_manager_uses_fundamentals_abort_report(self):
"""Verify PM uses fundamentals report when it contains abort."""
mock_llm = MagicMock()
mock_llm.invoke.return_value = MagicMock(
content="RECOMMENDATION: AVOID - Negative gross margin with bankruptcy filing"
)
portfolio_manager = create_portfolio_manager(mock_llm, MagicMock())
state = self._make_abort_state(normal_market_report, fundamentals_report_abort)
result = portfolio_manager(state)
recommendation = result.get("final_trade_decision", "")
assert "bankruptcy" in recommendation.lower()
def test_portfolio_manager_avoids_recommendation(self):
"""Verify PM recommends AVOID when fundamentals report has abort."""
mock_llm = MagicMock()
mock_llm.invoke.return_value = MagicMock(
content="RECOMMENDATION: AVOID - Negative gross margin with bankruptcy filing"
)
portfolio_manager = create_portfolio_manager(mock_llm, MagicMock())
state = self._make_abort_state(normal_market_report, fundamentals_report_abort)
result = portfolio_manager(state)
assert "AVOID" in result.get("final_trade_decision", "").upper()
# ---------------------------------------------------------------------------
# Integration Tests
# ---------------------------------------------------------------------------
class TestFastRejectFullFlow:
"""Integration tests for the complete fast-reject short-circuit flow."""
# Shared initial state template for integration tests
_base_state = {
"ticker": "AAPL",
"trade_date": "2024-01-01",
"company_of_interest": "AAPL",
"macro_regime_report": macro_regime_report,
"risk_debate_state": {
"history": [],
"aggressive_history": [],
"conservative_history": [],
"neutral_history": [],
"current_aggressive_response": "",
"current_conservative_response": "",
"current_neutral_response": "",
"count": 0,
},
"investment_debate_state": {
"history": [],
"bull_history": [],
"bear_history": [],
"current_response": "",
"judge_decision": "",
"count": 0,
},
"news_report": "",
"sentiment_report": "",
"investment_plan": "BUY AAPL",
"messages": [],
}
def _make_state(self, market_report, fundamentals_report):
return {**self._base_state, "market_report": market_report, "fundamentals_report": fundamentals_report}
def test_fast_reject_full_flow(self):
"""Test the complete short-circuit flow from analyst to portfolio manager."""
mock_market_ai = MagicMock()
mock_market_ai.content = market_report_abort
mock_market_ai.tool_calls = []
state = self._make_state(market_report_abort, normal_fundamentals_report)
# Patch network-calling helpers; control analyst output via run_tool_loop mock
with patch("tradingagents.agents.analysts.market_analyst.prefetch_tools_parallel", return_value={}), \
patch("tradingagents.agents.analysts.market_analyst.run_tool_loop", return_value=mock_market_ai):
market_analyst = create_market_analyst(MagicMock())
analyst_result = market_analyst(state)
state = {**state, **analyst_result} # merge so all keys are preserved
# Verify market report contains abort
assert "[CRITICAL ABORT]" in state.get("market_report", "")
# Run portfolio manager (mock LLM captured by closure)
mock_llm = MagicMock()
mock_llm.invoke.return_value = MagicMock(
content="RECOMMENDATION: SELL - Trading halted pending SEC investigation"
)
portfolio_manager = create_portfolio_manager(mock_llm, MagicMock())
pm_result = portfolio_manager(state)
state = {**state, **pm_result} # merge so market_report is still accessible
# Verify portfolio manager detected abort
assert "SELL" in state.get("final_trade_decision", "").upper()
# Verify conditional logic would bypass debate and risk analysis
cl = ConditionalLogic()
assert cl.should_continue_debate(state) == "Portfolio Manager"
assert cl.should_continue_risk_analysis(state) == "Portfolio Manager"
def test_fast_reject_fundamentals_flow(self):
"""Test the complete short-circuit flow with fundamentals abort."""
mock_market_ai = MagicMock()
mock_market_ai.content = normal_market_report
mock_market_ai.tool_calls = []
state = self._make_state(normal_market_report, fundamentals_report_abort)
with patch("tradingagents.agents.analysts.market_analyst.prefetch_tools_parallel", return_value={}), \
patch("tradingagents.agents.analysts.market_analyst.run_tool_loop", return_value=mock_market_ai):
market_analyst = create_market_analyst(MagicMock())
analyst_result = market_analyst(state)
state = {**state, **analyst_result}
# Market report should be normal (abort is in fundamentals)
assert "[CRITICAL ABORT]" not in state.get("market_report", "")
# Fundamentals abort must survive the merge
assert "[CRITICAL ABORT]" in state.get("fundamentals_report", "")
mock_llm = MagicMock()
mock_llm.invoke.return_value = MagicMock(
content="RECOMMENDATION: AVOID - Negative gross margin with bankruptcy filing"
)
portfolio_manager = create_portfolio_manager(mock_llm, MagicMock())
pm_result = portfolio_manager(state)
state = {**state, **pm_result}
assert "AVOID" in state.get("final_trade_decision", "").upper()
cl = ConditionalLogic()
assert cl.should_continue_debate(state) == "Portfolio Manager"
assert cl.should_continue_risk_analysis(state) == "Portfolio Manager"
def test_fast_reject_normal_flow(self):
"""Test the complete flow without abort."""
mock_market_ai = MagicMock()
mock_market_ai.content = normal_market_report
mock_market_ai.tool_calls = []
state = self._make_state(normal_market_report, normal_fundamentals_report)
with patch("tradingagents.agents.analysts.market_analyst.prefetch_tools_parallel", return_value={}), \
patch("tradingagents.agents.analysts.market_analyst.run_tool_loop", return_value=mock_market_ai):
market_analyst = create_market_analyst(MagicMock())
analyst_result = market_analyst(state)
state = {**state, **analyst_result}
assert "[CRITICAL ABORT]" not in state.get("market_report", "")
mock_llm = MagicMock()
mock_llm.invoke.return_value = MagicMock(
content="RECOMMENDATION: BUY - Strong bullish trend with positive momentum"
)
portfolio_manager = create_portfolio_manager(mock_llm, MagicMock())
pm_result = portfolio_manager(state)
state = {**state, **pm_result}
assert "BUY" in state.get("final_trade_decision", "").upper()
# Normal flow: conditional logic must NOT route directly to Portfolio Manager
cl = ConditionalLogic()
debate_result = cl.should_continue_debate(state)
risk_result = cl.should_continue_risk_analysis(state)
assert debate_result != "Portfolio Manager"
assert risk_result != "Portfolio Manager"

31
tradingagents/agents/analysts/fundamentals_analyst.py
View File

@ -87,9 +87,34 @@ def create_fundamentals_analyst(llm):
"driver, an FCF deviation from net income, or an unusual balance-sheet move — you "
"may call `get_balance_sheet`, `get_cashflow`, or `get_income_statement` to examine "
"the raw quarterly data directly.\n\n"
"Write a comprehensive report covering: multi-quarter revenue and margin trends, "
"TTM metrics, relative valuation vs peers, sector outperformance or underperformance, "
"and a clear medium-term fundamental thesis. "
"## CRITICAL ABORT TRIGGER\n\n"
"If you detect any of the following CATASTROPHIC conditions, you MUST immediately "
"prepend `[CRITICAL ABORT]` to your report and provide specific reasoning:\n\n"
"### Bankruptcy and Financial Distress:\n"
"- Bankruptcy filing or Chapter 11/7 proceedings\n"
"- Negative gross margins (gross margin < 0%)\n"
"- Negative operating margins (operating margin < 0%)\n"
"- Negative net income with no path to recovery\n"
"- Negative book value or negative equity\n"
"- Cash flow from operations < 0 with no turnaround plan\n\n"
"### SEC and Regulatory Issues:\n"
"- SEC enforcement action or investigation for material fraud\n"
"- Impending SEC delisting (notice of non-compliance)\n"
"- Going concern warning from auditor\n"
"- Regulatory shutdown or cease-and-desist order\n\n"
"### Material Fraud and Accounting Issues:\n"
"- Evidence of accounting manipulation or earnings management\n"
"- Revenue recognition violations\n"
"- Material restatement of financial statements\n"
"- Insider trading violations or SEC violations\n\n"
"### Format Requirements:\n"
"When triggering a critical abort, your report MUST start with:\n"
"`[CRITICAL ABORT] Reason: <specific reason for abort>`\n\n"
"Example: `[CRITICAL ABORT] Reason: Bankruptcy filing detected - negative gross margin of -15% with no path to recovery`\n\n"
"## Normal Operation\n\n"
"If no catastrophic conditions are detected, write a comprehensive report covering: "
"multi-quarter revenue and margin trends, TTM metrics, relative valuation vs peers, "
"sector outperformance or underperformance, and a clear medium-term fundamental thesis. "
"Do not simply state trends are mixed — provide detailed, fine-grained analysis that "
"identifies inflection points, acceleration or deceleration in growth, and specific "
"risks and opportunities. "

27
tradingagents/agents/analysts/market_analyst.py
View File

@ -66,6 +66,33 @@ def create_market_analyst(llm):
"choices before calling `get_indicators`. For example, in risk-off environments "
"favour ATR, Bollinger Bands, and long-term SMAs; in risk-on environments favour "
"momentum indicators like MACD and short EMAs.\n\n"
"## CRITICAL ABORT TRIGGER\n\n"
"If you detect any of the following CATASTROPHIC market conditions, you MUST immediately "
"prepend `[CRITICAL ABORT]` to your report and provide specific reasoning:\n\n"
"### Trading and Market Issues:\n"
"- Trading halted pending delisting or investigation\n"
"- Delisting announcement from exchange or regulatory body\n"
"- Trading halted due to catastrophic news or material information\n"
"- Market cap collapse (e.g., < $50M or > 90% decline in 24h)\n"
"- Extreme volatility (e.g., > 200% daily move)\n\n"
"### Regulatory and Legal Issues:\n"
"- SEC enforcement action or investigation\n"
"- Regulatory shutdown or cease-and-desist order\n"
"- Bankruptcy or insolvency filing\n"
"- Material fraud or accounting scandal\n"
"- Going concern warning from auditor\n\n"
"### Catastrophic News and Events:\n"
"- Earnings miss with -90% or worse guidance\n"
"- Major product recall or safety issue\n"
"- CEO resignation or major leadership scandal\n"
"- Lawsuit with > $1B damages or regulatory fine\n"
"- Natural disaster or catastrophic event impacting operations\n\n"
"### Format Requirements:\n"
"When triggering a critical abort, your report MUST start with:\n"
"`[CRITICAL ABORT] Reason: <specific reason for abort>`\n\n"
"Example: `[CRITICAL ABORT] Reason: Trading halted pending delisting - SEC notice of non-compliance`\n\n"
"## Normal Operation\n\n"
"If no catastrophic conditions are detected, continue with your analysis:\n\n"
"2. Select the **most relevant indicators** for the given market condition from "
"the list below. Choose up to **8 indicators** that provide complementary insights "
"without redundancy.\n\n"

43
tradingagents/agents/managers/portfolio_manager.py
View File

@ -15,6 +15,13 @@ def create_portfolio_manager(llm, memory):
trader_plan = state["investment_plan"]
macro_regime_report = state.get("macro_regime_report", "")
# Check for critical abort in market_report or fundamentals_report
is_critical_abort = (
"[CRITICAL ABORT]" in market_research_report or
"[CRITICAL ABORT]" in fundamentals_report
)
# Build current situation with all reports
macro_section = f"\n\nMacro Regime:\n{macro_regime_report}" if macro_regime_report else ""
curr_situation = f"{market_research_report}\n\n{sentiment_report}\n\n{news_report}\n\n{fundamentals_report}{macro_section}"
past_memories = memory.get_memories(curr_situation, n_matches=2)
@ -25,7 +32,39 @@ def create_portfolio_manager(llm, memory):
macro_context = f"\n\nCurrent Macro Regime:\n{macro_regime_report}\nEnsure your risk assessment reflects the macro environment — in risk-off regimes, apply higher standards for position entry and tighter risk controls.\n" if macro_regime_report else ""
prompt = f"""As the Portfolio Manager, synthesize the risk analysts' debate and deliver the final trading decision.
# Build prompt based on whether this is a critical abort scenario
if is_critical_abort:
# Critical abort: Use the aborting analyst's report and recommend SELL/AVOID
abort_report = market_research_report if "[CRITICAL ABORT]" in market_research_report else fundamentals_report
prompt = f"""As the Portfolio Manager, you have received a critical abort signal from an early analyst. This indicates catastrophic conditions (bankruptcy, SEC delisting, etc.) that require immediate action.
{instrument_context}
---
**CRITICAL ABORT DETECTED**
**Aborting Analyst's Report:**
{abort_report}
**Context:**
- Trader's proposed plan: **{trader_plan}**
- Lessons from past decisions: **{past_memory_str}**
**Required Output Structure:**
1. **Rating**: State one of Buy / Overweight / Hold / Underweight / Sell.
2. **Executive Summary**: A concise action plan covering entry strategy, position sizing, key risk levels, and time horizon.
3. **Investment Thesis**: Detailed reasoning based on the critical abort signal and the aborting analyst's report.
---
**IMPORTANT**: Based on the critical abort signal, you should recommend SELL or AVOID. Do not proceed with any other analysis. The aborting analyst has identified fundamental issues that make this investment unacceptable."""
response = llm.invoke(prompt)
else:
# Normal flow: Synthesize all reports and make decision
prompt = f"""As the Portfolio Manager, synthesize the risk analysts' debate and deliver the final trading decision.
{macro_context}
{instrument_context}
@ -57,7 +96,7 @@ def create_portfolio_manager(llm, memory):
Be decisive and ground every conclusion in specific evidence from the analysts."""
response = llm.invoke(prompt)
response = llm.invoke(prompt)
new_risk_debate_state = {
"judge_decision": response.content,

24
tradingagents/graph/conditional_logic.py
View File

@ -11,6 +11,21 @@ class ConditionalLogic:
self.max_debate_rounds = max_debate_rounds
self.max_risk_discuss_rounds = max_risk_discuss_rounds
def _check_critical_abort(self, state: AgentState, report_field: str) -> bool:
"""Check if a report contains [CRITICAL ABORT] trigger.
Args:
state: The current agent state
report_field: The field name to check (e.g., 'market_report', 'fundamentals_report')
Returns:
True if [CRITICAL ABORT] is found in the report, False otherwise
"""
report = state.get(report_field, "")
if not report:
return False
return "[CRITICAL ABORT]" in report
def should_continue_market(self, state: AgentState):
"""Determine if market analysis should continue."""
messages = state["messages"]
@ -46,6 +61,10 @@ class ConditionalLogic:
def should_continue_debate(self, state: AgentState) -> str:
"""Determine if debate should continue."""
# Check for critical abort in market_report or fundamentals_report
if self._check_critical_abort(state, "market_report") or self._check_critical_abort(state, "fundamentals_report"):
return "Portfolio Manager"
if (
state["investment_debate_state"]["count"] >= 2 * self.max_debate_rounds
): # 3 rounds of back-and-forth between 2 agents
@ -56,6 +75,11 @@ class ConditionalLogic:
def should_continue_risk_analysis(self, state: AgentState) -> str:
"""Determine if risk analysis should continue."""
# Check for critical abort in any report
if self._check_critical_abort(state, "market_report") or self._check_critical_abort(state, "fundamentals_report"):
return "Portfolio Manager"
if (
state["risk_debate_state"]["count"] >= 3 * self.max_risk_discuss_rounds
): # 3 rounds of back-and-forth between 3 agents