Skip to content

Conditional logic

ConditionalLogic

Bases: BaseModel


              flowchart TD
              src.tradingagents.graph.conditional_logic.ConditionalLogic[ConditionalLogic]

              

              click src.tradingagents.graph.conditional_logic.ConditionalLogic href "" "src.tradingagents.graph.conditional_logic.ConditionalLogic"

Methods:

Name Description
should_continue_market

Determine whether to continue market analysis or clear messages.
should_continue_social

Determine whether to continue social media analysis or clear messages.
should_continue_news

Determine whether to continue news analysis or clear messages.
should_continue_fundamentals

Determine whether to continue fundamentals analysis or clear messages.
should_continue_debate

Determine the next step in the investment debate.
should_continue_risk_analysis

Determine the next step in the risk analysis debate.

max_debate_rounds 

max_debate_rounds: int = Field(
    default=1,
    title="Max Debate Rounds",
    description="Maximum number of Bull/Bear investment debate rounds",
)

max_risk_discuss_rounds 

max_risk_discuss_rounds: int = Field(
    default=1,
    title="Max Risk Discussion Rounds",
    description="Maximum number of Risk debate rounds",
)

should_continue_market 

should_continue_market(state: AgentState) -> Literal['tools_market', 'Msg Clear Market']

Determine whether to continue market analysis or clear messages.

Parameters:

Name Type Description Default

state

AgentState

The current state of the agent.

 required

Returns:

Type Description
Literal['tools_market', 'Msg Clear Market']

Literal["tools_market", "Msg Clear Market"]: Next node to execute.
Source code in src/tradingagents/graph/conditional_logic.py 
22
23
24
25
26
27
28
29
30
31
32
33
def should_continue_market(
    self, state: AgentState
) -> Literal["tools_market", "Msg Clear Market"]:
    """Determine whether to continue market analysis or clear messages.

    Args:
        state (AgentState): The current state of the agent.

    Returns:
        Literal["tools_market", "Msg Clear Market"]: Next node to execute.
    """
    return "tools_market" if state.messages[-1].tool_calls else "Msg Clear Market"

should_continue_social 

should_continue_social(state: AgentState) -> Literal['tools_social', 'Msg Clear Social']

Determine whether to continue social media analysis or clear messages.

Parameters:

Name Type Description Default

state

AgentState

The current state of the agent.

 required

Returns:

Type Description
Literal['tools_social', 'Msg Clear Social']

Literal["tools_social", "Msg Clear Social"]: Next node to execute.
Source code in src/tradingagents/graph/conditional_logic.py 
35
36
37
38
39
40
41
42
43
44
45
46
def should_continue_social(
    self, state: AgentState
) -> Literal["tools_social", "Msg Clear Social"]:
    """Determine whether to continue social media analysis or clear messages.

    Args:
        state (AgentState): The current state of the agent.

    Returns:
        Literal["tools_social", "Msg Clear Social"]: Next node to execute.
    """
    return "tools_social" if state.messages[-1].tool_calls else "Msg Clear Social"

should_continue_news 

should_continue_news(state: AgentState) -> Literal['tools_news', 'Msg Clear News']

Determine whether to continue news analysis or clear messages.

Parameters:

Name Type Description Default

state

AgentState

The current state of the agent.

 required

Returns:

Type Description
Literal['tools_news', 'Msg Clear News']

Literal["tools_news", "Msg Clear News"]: Next node to execute.
Source code in src/tradingagents/graph/conditional_logic.py 
48
49
50
51
52
53
54
55
56
57
def should_continue_news(self, state: AgentState) -> Literal["tools_news", "Msg Clear News"]:
    """Determine whether to continue news analysis or clear messages.

    Args:
        state (AgentState): The current state of the agent.

    Returns:
        Literal["tools_news", "Msg Clear News"]: Next node to execute.
    """
    return "tools_news" if state.messages[-1].tool_calls else "Msg Clear News"

should_continue_fundamentals 

should_continue_fundamentals(
    state: AgentState,
) -> Literal["tools_fundamentals", "Msg Clear Fundamentals"]

Determine whether to continue fundamentals analysis or clear messages.

Parameters:

Name Type Description Default

state

AgentState

The current state of the agent.

 required

Returns:

Type Description
Literal['tools_fundamentals', 'Msg Clear Fundamentals']

Literal["tools_fundamentals", "Msg Clear Fundamentals"]: Next node to execute.
Source code in src/tradingagents/graph/conditional_logic.py 
59
60
61
62
63
64
65
66
67
68
69
70
def should_continue_fundamentals(
    self, state: AgentState
) -> Literal["tools_fundamentals", "Msg Clear Fundamentals"]:
    """Determine whether to continue fundamentals analysis or clear messages.

    Args:
        state (AgentState): The current state of the agent.

    Returns:
        Literal["tools_fundamentals", "Msg Clear Fundamentals"]: Next node to execute.
    """
    return "tools_fundamentals" if state.messages[-1].tool_calls else "Msg Clear Fundamentals"

should_continue_debate 

should_continue_debate(
    state: AgentState,
) -> Literal["Bull Researcher", "Bear Researcher", "Research Manager"]

Determine the next step in the investment debate.

Parameters:

Name Type Description Default

state

AgentState

The current state of the agent.

 required

Returns:

Type Description
Literal['Bull Researcher', 'Bear Researcher', 'Research Manager']

Literal["Bull Researcher", "Bear Researcher", "Research Manager"]: Next node to execute.
Source code in src/tradingagents/graph/conditional_logic.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def should_continue_debate(
    self, state: AgentState
) -> Literal["Bull Researcher", "Bear Researcher", "Research Manager"]:
    """Determine the next step in the investment debate.

    Args:
        state (AgentState): The current state of the agent.

    Returns:
        Literal["Bull Researcher", "Bear Researcher", "Research Manager"]: Next node to execute.
    """
    debate = state.investment_debate_state
    if debate.count >= 2 * self.max_debate_rounds:
        return "Research Manager"
    if debate.current_response.startswith("Bull"):
        return "Bear Researcher"
    return "Bull Researcher"

should_continue_risk_analysis 

should_continue_risk_analysis(
    state: AgentState,
) -> Literal["Aggressive Analyst", "Conservative Analyst", "Neutral Analyst", "Risk Judge"]

Determine the next step in the risk analysis debate.

Parameters:

Name Type Description Default

state

AgentState

The current state of the agent.

 required

Returns:

Type Description
Literal['Aggressive Analyst', 'Conservative Analyst', 'Neutral Analyst', 'Risk Judge']

Literal["Aggressive Analyst", "Conservative Analyst", "Neutral Analyst", "Risk Judge"]: Next node to execute.
Source code in src/tradingagents/graph/conditional_logic.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def should_continue_risk_analysis(
    self, state: AgentState
) -> Literal["Aggressive Analyst", "Conservative Analyst", "Neutral Analyst", "Risk Judge"]:
    """Determine the next step in the risk analysis debate.

    Args:
        state (AgentState): The current state of the agent.

    Returns:
        Literal["Aggressive Analyst", "Conservative Analyst", "Neutral Analyst", "Risk Judge"]: Next node to execute.
    """
    risk = state.risk_debate_state
    if risk.count >= 3 * self.max_risk_discuss_rounds:
        return "Risk Judge"
    if risk.latest_speaker.startswith("Aggressive"):
        return "Conservative Analyst"
    if risk.latest_speaker.startswith("Conservative"):
        return "Neutral Analyst"
    return "Aggressive Analyst"