quant_agent.py

##### Quantitative Agent #####
def quant_agent(state: AgentState):
    """Analyzes technical indicators and generates trading signals."""
    show_reasoning = state["metadata"]["show_reasoning"]

    data = state["data"]
    prices = data["prices"]
    prices_df = prices_to_df(prices)
    
    # Calculate existing indicators
    # 1. MACD (Moving Average Convergence Divergence)
    macd_line, signal_line = calculate_macd(prices_df)
    
    # 2. RSI (Relative Strength Index)
    rsi = calculate_rsi(prices_df)
    
    # 3. Bollinger Bands
    upper_band, lower_band = calculate_bollinger_bands(prices_df)
    
    # 4. OBV (On-Balance Volume)
    obv = calculate_obv(prices_df)
    
    # ***** EXISTING: Simple Moving Average Crossover *****
    short_window = 10
    long_window = 50
    prices_df['short_ma'] = prices_df['close'].rolling(short_window).mean()
    prices_df['long_ma'] = prices_df['close'].rolling(long_window).mean()
    
    short_ma_prev = prices_df['short_ma'].iloc[-2]
    long_ma_prev = prices_df['long_ma'].iloc[-2]
    short_ma_current = prices_df['short_ma'].iloc[-1]
    long_ma_current = prices_df['long_ma'].iloc[-1]
    
    # ***** NEW 1: Donchian Channels *****
    donchian_period = 20
    prices_df['donchian_high'] = prices_df['high'].rolling(donchian_period).max()
    prices_df['donchian_low'] = prices_df['low'].rolling(donchian_period).min()
    current_price = prices_df['close'].iloc[-1]
    
    if current_price > prices_df['donchian_high'].iloc[-1]:
        donchian_signal = 'bullish'
    elif current_price < prices_df['donchian_low'].iloc[-1]:
        donchian_signal = 'bearish'
    else:
        donchian_signal = 'neutral'
    
    # ***** NEW 2: Stochastic Oscillator *****
    stoch_period = 14
    k_period = 3
    d_period = 3
    
    low_min = prices_df['low'].rolling(stoch_period).min()
    high_max = prices_df['high'].rolling(stoch_period).max()
    prices_df['%K'] = 100 * ((prices_df['close'] - low_min) / (high_max - low_min))
    prices_df['%D'] = prices_df['%K'].rolling(d_period).mean()
    
    k_prev = prices_df['%K'].iloc[-2]
    d_prev = prices_df['%D'].iloc[-2]
    k_current = prices_df['%K'].iloc[-1]
    d_current = prices_df['%D'].iloc[-1]
    
    # Define Stochastic signals:
    # Bullish if %K crosses above %D from below 20
    # Bearish if %K crosses below %D from above 80
    if k_prev < d_prev and k_current > d_current and k_current < 20:
        stoch_signal = 'bullish'
    elif k_prev > d_prev and k_current < d_current and k_current > 80:
        stoch_signal = 'bearish'
    else:
        stoch_signal = 'neutral'

    # Generate individual signals
    signals = []
    
    # MACD signal
    if macd_line.iloc[-2] < signal_line.iloc[-2] and macd_line.iloc[-1] > signal_line.iloc[-1]:
        signals.append('bullish')
    elif macd_line.iloc[-2] > signal_line.iloc[-2] and macd_line.iloc[-1] < signal_line.iloc[-1]:
        signals.append('bearish')
    else:
        signals.append('neutral')
    
    # RSI signal
    if rsi.iloc[-1] < 30:
        signals.append('bullish')
    elif rsi.iloc[-1] > 70:
        signals.append('bearish')
    else:
        signals.append('neutral')
    
    # Bollinger Bands signal
    if current_price < lower_band.iloc[-1]:
        signals.append('bullish')
    elif current_price > upper_band.iloc[-1]:
        signals.append('bearish')
    else:
        signals.append('neutral')
    
    # OBV signal
    obv_slope = obv.diff().iloc[-5:].mean()
    if obv_slope > 0:
        signals.append('bullish')
    elif obv_slope < 0:
        signals.append('bearish')
    else:
        signals.append('neutral')
    
    # Moving Average Crossover Signal
    if short_ma_prev < long_ma_prev and short_ma_current > long_ma_current:
        ma_signal = 'bullish'
    elif short_ma_prev > long_ma_prev and short_ma_current < long_ma_current:
        ma_signal = 'bearish'
    else:
        ma_signal = 'neutral'
    signals.append(ma_signal)
    
    # Donchian Channel Signal
    signals.append(donchian_signal)
    
    # Stochastic Signal
    signals.append(stoch_signal)
    
    # Add reasoning collection
    reasoning = {
        "MACD": {
            "signal": signals[0],
            "details": f"MACD Line crossed {'above' if signals[0] == 'bullish' else 'below' if signals[0] == 'bearish' else 'neither'} the Signal Line"
        },
        "RSI": {
            "signal": signals[1],
            "details": f"RSI is {rsi.iloc[-1]:.2f} ({'oversold' if signals[1] == 'bullish' else 'overbought' if signals[1] == 'bearish' else 'neutral'})"
        },
        "Bollinger": {
            "signal": signals[2],
            "details": f"Price is {'below lower band' if signals[2] == 'bullish' else 'above upper band' if signals[2] == 'bearish' else 'within bands'}"
        },
        "OBV": {
            "signal": signals[3],
            "details": f"OBV slope is {obv_slope:.2f} ({signals[3]})"
        },
        "Moving Average Crossover": {
            "signal": ma_signal,
            "details": f"Short MA ({short_window}-day) is {'above' if ma_signal == 'bullish' else 'below' if ma_signal == 'bearish' else 'not definitively crossing'} Long MA ({long_window}-day)"
        },
        "Donchian": {
            "signal": donchian_signal,
            "details": f"Price is {'above recent high' if donchian_signal == 'bullish' else 'below recent low' if donchian_signal == 'bearish' else 'within the Donchian channel'}"
        },
        "Stochastic": {
            "signal": stoch_signal,
            "details": f"Stochastic %K={k_current:.2f}, %D={d_current:.2f}, {'bullish crossover below 20' if stoch_signal=='bullish' else 'bearish crossover above 80' if stoch_signal=='bearish' else 'no clear crossover signal'}"
        }
    }
    
    # Determine overall signal
    bullish_signals = signals.count('bullish')
    bearish_signals = signals.count('bearish')
    
    if bullish_signals > bearish_signals:
        overall_signal = 'bullish'
    elif bearish_signals > bullish_signals:
        overall_signal = 'bearish'
    else:
        overall_signal = 'neutral'
    
    # Calculate confidence level based on the proportion of indicators agreeing
    total_signals = len(signals)
    confidence = max(bullish_signals, bearish_signals) / total_signals
    
    # Generate the message content
    message_content = {
        "signal": overall_signal,
        "confidence": f"{round(confidence * 100)}%",
        "reasoning": reasoning
    }

    # Create the quant message
    message = HumanMessage(
        content=str(message_content),  # Convert dict to string for message content
        name="quant_agent",
    )

    # Print the reasoning if the flag is set
    if show_reasoning:
        show_agent_reasoning(message_content, "Quant Agent")
    
    return {
        "messages": [message],
        "data": data,
    }