Skip to content

Risk manager

risk_manager

Functions:

Name Description
create_risk_manager

Creates a risk manager node for the trading graph.

create_risk_manager 

create_risk_manager(
    llm: ChatModel, memory: FinancialSituationMemory
) -> Callable[[AgentState], dict[str, Any]]

Creates a risk manager node for the trading graph.

Parameters:

Name Type Description Default

llm

ChatModel

The language model to use for generating responses.

 required

memory

FinancialSituationMemory

The memory module for retrieving past financial situations.

 required

Returns:

Type Description
Callable[[AgentState], dict[str, Any]]

Callable[[AgentState], dict[str, Any]]: A function representing the risk manager node.
Source code in src/tradingagents/agents/managers/risk_manager.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def create_risk_manager(
    llm: ChatModel, memory: FinancialSituationMemory
) -> Callable[[AgentState], dict[str, Any]]:
    """Creates a risk manager node for the trading graph.

    Args:
        llm (ChatModel): The language model to use for generating responses.
        memory (FinancialSituationMemory): The memory module for retrieving past financial situations.

    Returns:
        Callable[[AgentState], dict[str, Any]]: A function representing the risk manager node.
    """

    def risk_manager_node(state: AgentState) -> dict[str, Any]:
        """Executes the risk manager logic to evaluate risk and make a final trade decision.

        Args:
            state (AgentState): The current state of the agent, including reports, investment plan, and risk state.

        Returns:
            dict[str, Any]: A dictionary containing the updated risk_debate_state and the final_trade_decision.
        """
        risk = state.risk_debate_state

        curr_situation = (
            f"{state.market_report}\n\n"
            f"{state.sentiment_report}\n\n"
            f"{state.news_report}\n\n"
            f"{state.fundamentals_report}"
        )
        past_memories = memory.get_memories(curr_situation, n_matches=2)
        past_memory_str = "".join(rec["recommendation"] + "\n\n" for rec in past_memories)

        prompt = load_prompt("risk_manager").format(
            trader_plan=state.trader_investment_plan,
            past_memory_str=past_memory_str,
            history=risk.history,
        )

        response = llm.invoke(prompt)

        new_risk_state = RiskDebateState(
            judge_decision=response.content,
            history=risk.history,
            aggressive_history=risk.aggressive_history,
            conservative_history=risk.conservative_history,
            neutral_history=risk.neutral_history,
            latest_speaker="Judge",
            current_aggressive_response=risk.current_aggressive_response,
            current_conservative_response=risk.current_conservative_response,
            current_neutral_response=risk.current_neutral_response,
            count=risk.count,
        )

        return {"risk_debate_state": new_risk_state, "final_trade_decision": response.content}

    return risk_manager_node