Trading graph

def propagate(
    self,
    company_name: str,
    trade_date: str,
    on_message: Callable[[AnyMessage], None] | None = None,
    on_state: Callable[[AgentState], None] | None = None,
) -> tuple[AgentState, str]:
    """Run the trading agents graph for a company on a specific date.

    Args:
        company_name (str): Company name or ticker symbol.
        trade_date (str): Trading date in YYYY-MM-DD format.
        on_message (Callable[[AnyMessage], None] | None, optional):
            Callback invoked once per newly-produced message during the
            stream. When provided, takes precedence over the default
            debug print path so callers (CLI, TUI) can route output
            through Rich panels instead of message.pretty_print().
            Defaults to None.
        on_state (Callable[[AgentState], None] | None, optional):
            Callback invoked once per stream chunk with the full
            AgentState snapshot. Used by the Textual TUI to update the
            phase progress sidebar from analyst-report / debate
            fields; the CLI does not need this. Defaults to None.

    Returns:
        tuple[AgentState, str]: The final agent state and the extracted signal decision.

    Raises:
        RuntimeError: If the graph execution produces no output.
    """
    self.ticker = company_name

    init_agent_state = self.propagator.create_initial_state(company_name, trade_date)
    args = self.propagator.get_graph_args(callbacks=self.callbacks or None)

    # Stream in "values" mode (set by Propagator.get_graph_args) so each chunk
    # is the full state snapshot after a node runs. The graph clears
    # state.messages between analysts via Msg Clear nodes, so the only way
    # to capture every round of LLM dialogue is to collect messages as they
    # appear in stream chunks (deduped by id).
    raw_state = None
    last_emitted_id = None
    collected: dict[str, AnyMessage] = {}
    for chunk in self.graph.stream(init_agent_state, **args):
        last_emitted_id = self._dispatch_messages(
            chunk, collected, last_emitted_id, on_message
        )
        if on_state is not None:
            self._dispatch_state(chunk, on_state)
        raw_state = chunk

    if raw_state is None:
        raise RuntimeError("Graph produced no output")

    final_state = (
        AgentState.model_validate(raw_state) if isinstance(raw_state, dict) else raw_state
    )

    self.curr_state = final_state
    self._log_state(trade_date, final_state, list(collected.values()))
    return final_state, self.process_signal(final_state.final_trade_decision)

def reflect_and_remember(self, returns_losses: float) -> None:
    """Reflect on decisions and update memory based on returns.

    Args:
        returns_losses (float): Actual returns or losses from the trade.

    Raises:
        RuntimeError: If there is no current state to reflect on.
    """
    if self.curr_state is None:
        raise RuntimeError("No state available to reflect on. Run propagate() first.")
    self.reflector.reflect_bull_researcher(self.curr_state, returns_losses, self.bull_memory)
    self.reflector.reflect_bear_researcher(self.curr_state, returns_losses, self.bear_memory)
    self.reflector.reflect_trader(self.curr_state, returns_losses, self.trader_memory)
    self.reflector.reflect_invest_judge(
        self.curr_state, returns_losses, self.invest_judge_memory
    )
    self.reflector.reflect_risk_manager(
        self.curr_state, returns_losses, self.risk_manager_memory
    )

def process_signal(self, full_signal: str) -> str:
    """Process a signal to extract the core decision.

    Args:
        full_signal (str): The raw text signal.

    Returns:
        str: The extracted decision (BUY/SELL/HOLD).
    """
    return self.signal_processor.process_signal(full_signal)