tools/feature_engineering.py

import pandas as pd
import pandas_ta_classic as ta
import yfinance as yf
from typing import List
import logging

logger = logging.getLogger(__name__)

def compute_indicators(symbol: str, indicators: List[str] = ["RSI", "MACD"]) -> str:
    """Calculate technical indicators. Example: compute_indicators("AAPL", ["RSI", "MACD"])"""
    df = yf.download(symbol, period="1y", progress=False)
    if df.empty:
        return f"No data for {symbol}"
    
    # Handle MultiIndex
    if isinstance(df.columns, pd.MultiIndex):
        df.columns = df.columns.get_level_values(0)
        
    result = df[['Close']].copy()
    
    for ind in indicators:
        try:
            if ind.upper() == "RSI":
                result['RSI'] = ta.rsi(df['Close'])
            elif ind.upper() == "MACD":
                macd = ta.macd(df['Close'])
                result = pd.concat([result, macd], axis=1)
            elif ind.upper() == "BBANDS":
                bb = ta.bbands(df['Close'])
                result = pd.concat([result, bb], axis=1)
            # Add more as needed
        except Exception as e:
            return f"Error computing {ind}: {str(e)}"
            
    return result.tail(10).to_json(orient="index")

def rolling_stats(symbol: str, window: int = 20) -> str:
    """Calculate rolling mean and volatility. Example: rolling_stats("MSFT", window=20)"""
    df = yf.download(symbol, period="1y", progress=False)
    if df.empty: return "No data"
    
    close = df['Close']
    if isinstance(close, pd.DataFrame): close = close.iloc[:, 0]
    
    stats = pd.DataFrame()
    stats['Mean'] = close.rolling(window=window).mean()
    stats['Std'] = close.rolling(window=window).std()
    
    return stats.tail(10).to_json(orient="index")

def get_technical_summary(symbol: str) -> str:
    """Get technical analysis summary. Example: get_technical_summary("NVDA")"""
    try:
        # Need enough data for 200 SMA
        df = yf.download(symbol, period="2y", progress=False)
        if df.empty:
            return f"No data found for {symbol}"
            
        # Handle MultiIndex if present
        # yfinance structure: columns are (Price, Ticker)
        if isinstance(df.columns, pd.MultiIndex):
            # If we have a MultiIndex, we want the 'Close' level, and then the specific ticker
            # But usually df['Close'] gives us the Close column(s).
            pass 
            
        close = df['Close']
        if isinstance(close, pd.DataFrame):
            # If multiple tickers or just weird structure, take the first column
            close = close.iloc[:, 0]
            
        # 1. RSI
        rsi = ta.rsi(close, length=14)
        if rsi is None or rsi.empty:
            return f"Not enough data to calculate RSI for {symbol}"
        current_rsi = rsi.iloc[-1]
        
        # 2. MACD
        macd = ta.macd(close)
        if macd is None or macd.empty:
            return f"Not enough data to calculate MACD for {symbol}"
        
        # MACD_12_26_9, MACDh_12_26_9 (hist), MACDs_12_26_9 (signal)
        # Column names depend on pandas_ta version/settings, but usually standard
        macd_line = macd.iloc[-1, 0] # First column is usually MACD line
        macd_signal = macd.iloc[-1, 2] # Third column is usually Signal line
        
        # 3. Moving Averages
        sma_50 = ta.sma(close, length=50)
        sma_200 = ta.sma(close, length=200)
        
        if sma_50 is None or sma_50.empty:
            return "Not enough data for SMA 50"
        if sma_200 is None or sma_200.empty:
            # Fallback if 200 SMA not available (e.g. IPO recently)
            sma_200_val = 0
            has_200 = False
        else:
            sma_200_val = sma_200.iloc[-1]
            has_200 = True
            
        sma_50_val = sma_50.iloc[-1]
        current_price = close.iloc[-1]
        
        # Scoring Logic
        score = 0
        reasons = []
        
        # RSI Logic
        if current_rsi < 30:
            score += 1
            reasons.append(f"RSI is Oversold ({current_rsi:.2f})")
        elif current_rsi > 70:
            score -= 1
            reasons.append(f"RSI is Overbought ({current_rsi:.2f})")
        else:
            reasons.append(f"RSI is Neutral ({current_rsi:.2f})")
            
        # MACD Logic
        if macd_line > macd_signal:
            score += 1
            reasons.append("MACD Bullish Crossover")
        else:
            score -= 1
            reasons.append("MACD Bearish Crossover")
            
        # Trend Logic
        if current_price > sma_50_val:
            score += 1
            reasons.append("Price above 50 SMA (Bullish Trend)")
        else:
            score -= 1
            reasons.append("Price below 50 SMA (Bearish Trend)")
            
        if has_200:
            if current_price > sma_200_val:
                score += 1
                reasons.append("Price above 200 SMA (Long-term Bullish)")
            else:
                score -= 1
                reasons.append("Price below 200 SMA (Long-term Bearish)")
        else:
            reasons.append("200 SMA not available (Insufficient history)")
            
        # Final Verdict
        if score >= 2:
            signal = "STRONG BUY"
        elif score == 1:
            signal = "BUY"
        elif score == 0:
            signal = "NEUTRAL"
        elif score == -1:
            signal = "SELL"
        else:
            signal = "STRONG SELL"
            
        return (f"Technical Summary for {symbol}:\n"
                f"Signal: {signal} (Score: {score})\n"
                f"Price: ${current_price:.2f}\n"
                f"--------------------------------\n"
                f"Analysis:\n" + "\n".join([f"- {r}" for r in reasons]))
                
    except Exception as e:
        return f"Error performing technical analysis: {str(e)}"