Generating Trading Signals

# Signal Generator Configuration
# Copy to settings.yaml and customize

# Data source settings
data:
  provider: yfinance
  cache_dir: ./data
  default_interval: 1d
  default_period: 6m

# Indicator parameters
indicators:
  rsi:
    period: 14
    overbought: 70
    oversold: 30

  macd:
    fast: 12
    slow: 26
    signal: 9

  bollinger:
    period: 20
    std_dev: 2.0

  stochastic:
    k_period: 14
    d_period: 3
    overbought: 80
    oversold: 20

  atr:
    period: 14
    multiplier: 2.0

  adx:
    period: 14
    trend_threshold: 25

# Signal generation
signals:
  # Component weights (adjust to emphasize certain indicators)
  weights:
    rsi: 1.0
    macd: 1.0
    bollinger: 1.0
    trend: 1.0
    volume: 0.5
    stochastic: 0.5
    adx: 0.5

  # Risk management
  risk:
    atr_multiplier: 2.0
    risk_reward_target: 2.0

  # Confidence thresholds
  thresholds:
    high_confidence: 70
    medium_confidence: 50
    low_confidence: 30

# Predefined watchlists
watchlists:
  crypto_top10:
    - BTC-USD
    - ETH-USD
    - BNB-USD
    - SOL-USD
    - XRP-USD
    - ADA-USD
    - AVAX-USD
    - DOGE-USD
    - DOT-USD
    - MATIC-USD

  crypto_defi:
    - UNI-USD
    - AAVE-USD
    - MKR-USD
    - CRV-USD
    - SNX-USD
    - COMP-USD
    - SUSHI-USD

  crypto_layer2:
    - MATIC-USD
    - OP-USD
    - ARB-USD
    - IMX-USD

# Alert settings
alerts:
  min_confidence: 60
  signal_types:
    - STRONG_BUY
    - STRONG_SELL
  notify_neutral: false

#!/usr/bin/env python3
"""
Technical Indicators Library
Calculate RSI, MACD, Bollinger Bands, and other technical indicators.
"""

import pandas as pd
import numpy as np
from typing import Dict, Any, Tuple


def calculate_sma(data: pd.Series, period: int) -> pd.Series:
    """Calculate Simple Moving Average."""
    return data.rolling(window=period).mean()


def calculate_ema(data: pd.Series, period: int) -> pd.Series:
    """Calculate Exponential Moving Average."""
    return data.ewm(span=period, adjust=False).mean()


def calculate_rsi(data: pd.Series, period: int = 14) -> pd.Series:
    """
    Calculate Relative Strength Index.

    RSI = 100 - (100 / (1 + RS))
    RS = Average Gain / Average Loss
    """
    delta = data.diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()

    rs = gain / loss
    rsi = 100 - (100 / (1 + rs))
    return rsi


def calculate_macd(
    data: pd.Series,
    fast: int = 12,
    slow: int = 26,
    signal: int = 9
) -> Tuple[pd.Series, pd.Series, pd.Series]:
    """
    Calculate MACD (Moving Average Convergence Divergence).

    Returns:
        macd_line: MACD line (fast EMA - slow EMA)
        signal_line: Signal line (EMA of MACD)
        histogram: MACD histogram (MACD - Signal)
    """
    ema_fast = calculate_ema(data, fast)
    ema_slow = calculate_ema(data, slow)

    macd_line = ema_fast - ema_slow
    signal_line = calculate_ema(macd_line, signal)
    histogram = macd_line - signal_line

    return macd_line, signal_line, histogram


def calculate_bollinger_bands(
    data: pd.Series,
    period: int = 20,
    std_dev: float = 2.0
) -> Tuple[pd.Series, pd.Series, pd.Series]:
    """
    Calculate Bollinger Bands.

    Returns:
        upper: Upper band
        middle: Middle band (SMA)
        lower: Lower band
    """
    middle = calculate_sma(data, period)
    std = data.rolling(window=period).std()

    upper = middle + (std * std_dev)
    lower = middle - (std * std_dev)

    return upper, middle, lower


def calculate_atr(
    high: pd.Series,
    low: pd.Series,
    close: pd.Series,
    period: int = 14
) -> pd.Series:
    """Calculate Average True Range."""
    high_low = high - low
    high_close = abs(high - close.shift())
    low_close = abs(low - close.shift())

    tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
    atr = tr.rolling(window=period).mean()

    return atr


def calculate_stochastic(
    high: pd.Series,
    low: pd.Series,
    close: pd.Series,
    k_period: int = 14,
    d_period: int = 3
) -> Tuple[pd.Series, pd.Series]:
    """
    Calculate Stochastic Oscillator.

    Returns:
        k: %K line
        d: %D line (SMA of %K)
    """
    lowest_low = low.rolling(window=k_period).min()
    highest_high = high.rolling(window=k_period).max()

    k = 100 * (close - lowest_low) / (highest_high - lowest_low)
    d = k.rolling(window=d_period).mean()

    return k, d


def calculate_obv(close: pd.Series, volume: pd.Series) -> pd.Series:
    """Calculate On-Balance Volume."""
    obv = (np.sign(close.diff()) * volume).fillna(0).cumsum()
    return obv


def calculate_vwap(
    high: pd.Series,
    low: pd.Series,
    close: pd.Series,
    volume: pd.Series
) -> pd.Series:
    """Calculate Volume Weighted Average Price (intraday)."""
    typical_price = (high + low + close) / 3
    vwap = (typical_price * volume).cumsum() / volume.cumsum()
    return vwap


def calculate_adx(
    high: pd.Series,
    low: pd.Series,
    close: pd.Series,
    period: int = 14
) -> Tuple[pd.Series, pd.Series, pd.Series]:
    """
    Calculate Average Directional Index.

    Returns:
        adx: ADX line
        plus_di: +DI line
        minus_di: -DI line
    """
    # Calculate True Range
    tr1 = high - low
    tr2 = abs(high - close.shift())
    tr3 = abs(low - close.shift())
    tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
    atr = tr.rolling(window=period).mean()

    # Calculate directional movement
    plus_dm = high.diff()
    minus_dm = -low.diff()

    plus_dm = plus_dm.where((plus_dm > minus_dm) & (plus_dm > 0), 0)
    minus_dm = minus_dm.where((minus_dm > plus_dm) & (minus_dm > 0), 0)

    # Calculate DI
    plus_di = 100 * (plus_dm.rolling(window=period).mean() / atr)
    minus_di = 100 * (minus_dm.rolling(window=period).mean() / atr)

    # Calculate DX and ADX
    dx = 100 * abs(plus_di - minus_di) / (plus_di + minus_di)
    adx = dx.rolling(window=period).mean()

    return adx, plus_di, minus_di


def calculate_all_indicators(df: pd.DataFrame, params: Dict[str, Any] = None) -> pd.DataFrame:
    """
    Calculate all indicators for a price DataFrame.

    Args:
        df: DataFrame with columns: open, high, low, close, volume
        params: Optional custom parameters for indicators

    Returns:
        DataFrame with all indicators added
    """
    params = params or {}
    result = df.copy()

    # Price data
    close = df['close']
    high = df['high']
    low = df['low']
    volume = df['volume']

    # Moving Averages
    result['sma_20'] = calculate_sma(close, 20)
    result['sma_50'] = calculate_sma(close, 50)
    result['sma_200'] = calculate_sma(close, 200)
    result['ema_12'] = calculate_ema(close, 12)
    result['ema_26'] = calculate_ema(close, 26)

    # RSI
    rsi_period = params.get('rsi_period', 14)
    result['rsi'] = calculate_rsi(close, rsi_period)

    # MACD
    macd_fast = params.get('macd_fast', 12)
    macd_slow = params.get('macd_slow', 26)
    macd_signal = params.get('macd_signal', 9)
    result['macd'], result['macd_signal'], result['macd_hist'] = calculate_macd(
        close, macd_fast, macd_slow, macd_signal
    )

    # Bollinger Bands
    bb_period = params.get('bb_period', 20)
    bb_std = params.get('bb_std', 2.0)
    result['bb_upper'], result['bb_middle'], result['bb_lower'] = calculate_bollinger_bands(
        close, bb_period, bb_std
    )
    result['bb_width'] = (result['bb_upper'] - result['bb_lower']) / result['bb_middle']
    result['bb_pct'] = (close - result['bb_lower']) / (result['bb_upper'] - result['bb_lower'])

    # ATR
    atr_period = params.get('atr_period', 14)
    result['atr'] = calculate_atr(high, low, close, atr_period)
    result['atr_pct'] = result['atr'] / close * 100

    # Stochastic
    stoch_k = params.get('stoch_k', 14)
    stoch_d = params.get('stoch_d', 3)
    result['stoch_k'], result['stoch_d'] = calculate_stochastic(high, low, close, stoch_k, stoch_d)

    # Volume Indicators
    result['obv'] = calculate_obv(close, volume)
    result['volume_sma'] = calculate_sma(volume, 20)
    result['volume_ratio'] = volume / result['volume_sma']

    # ADX
    adx_period = params.get('adx_period', 14)
    result['adx'], result['plus_di'], result['minus_di'] = calculate_adx(high, low, close, adx_period)

    # Price changes
    result['change_1d'] = close.pct_change() * 100
    result['change_7d'] = close.pct_change(7) * 100
    result['change_30d'] = close.pct_change(30) * 100

    return result


if __name__ == '__main__':
    # Quick test
    import yfinance as yf

    ticker = yf.Ticker('BTC-USD')
    df = ticker.history(period='1y', interval='1d')
    df.columns = [c.lower() for c in df.columns]

    result = calculate_all_indicators(df)
    print(f"Calculated {len([c for c in result.columns if c not in df.columns])} indicators")
    print(f"\nLatest values:")
    print(f"  RSI: {result['rsi'].iloc[-1]:.2f}")
    print(f"  MACD: {result['macd'].iloc[-1]:.2f}")
    print(f"  BB %B: {result['bb_pct'].iloc[-1]:.2f}")
    print(f"  ADX: {result['adx'].iloc[-1]:.2f}")

#!/usr/bin/env python3
"""
Crypto Signal Scanner
Scan multiple assets and generate trading signals.

Usage:
    python scanner.py --symbols BTC-USD,ETH-USD --period 6m
    python scanner.py --watchlist crypto_top10 --output signals.json
"""

import argparse
import json
import sys
from pathlib import Path
from datetime import datetime, timedelta
from typing import List, Dict, Any

import pandas as pd

# Add script directory to path
sys.path.insert(0, str(Path(__file__).parent))

from indicators import calculate_all_indicators
from signals import SignalGenerator, TradingSignal, SignalType, format_signal


# Predefined watchlists
WATCHLISTS = {
    'crypto_top10': [
        'BTC-USD', 'ETH-USD', 'BNB-USD', 'SOL-USD', 'XRP-USD',
        'ADA-USD', 'AVAX-USD', 'DOGE-USD', 'DOT-USD', 'MATIC-USD'
    ],
    'crypto_defi': [
        'UNI-USD', 'AAVE-USD', 'MKR-USD', 'CRV-USD', 'SNX-USD',
        'COMP-USD', 'SUSHI-USD', 'YFI-USD', 'LDO-USD', 'RPL-USD'
    ],
    'crypto_layer2': [
        'MATIC-USD', 'OP-USD', 'ARB-USD', 'IMX-USD', 'LRC-USD'
    ],
    'stocks_tech': [
        'AAPL', 'MSFT', 'GOOGL', 'AMZN', 'NVDA',
        'META', 'TSLA', 'AMD', 'INTC', 'CRM'
    ],
    'etfs_major': [
        'SPY', 'QQQ', 'IWM', 'DIA', 'VTI'
    ],
}


def parse_period(period: str) -> timedelta:
    """Parse period string like '1y', '6m', '30d'."""
    unit = period[-1].lower()
    value = int(period[:-1])

    if unit == 'y':
        return timedelta(days=value * 365)
    elif unit == 'm':
        return timedelta(days=value * 30)
    elif unit == 'd':
        return timedelta(days=value)
    elif unit == 'w':
        return timedelta(weeks=value)
    else:
        raise ValueError(f"Unknown period unit: {unit}")


def fetch_data(symbol: str, period: str = '6m', cache_dir: Path = None) -> pd.DataFrame:
    """Fetch price data for a symbol."""
    end = datetime.now()
    start = end - parse_period(period)

    # Check cache first (no yfinance needed)
    if cache_dir:
        cache_file = cache_dir / f"{symbol.replace('/', '_').replace('-', '_')}_1d.csv"
        if cache_file.exists():
            try:
                df = pd.read_csv(cache_file, parse_dates=['date'], index_col='date')
                if df.index.tz is not None:
                    df.index = df.index.tz_localize(None)
                df = df[(df.index >= pd.Timestamp(start)) & (df.index <= pd.Timestamp(end))]
                if len(df) > 50:
                    return df
            except Exception:
                pass  # Fall through to fetch

    # Fetch from yfinance
    try:
        import yfinance as yf
        ticker = yf.Ticker(symbol)
        df = ticker.history(start=start, end=end, interval='1d')
        df.columns = [c.lower() for c in df.columns]
        df.index.name = 'date'

        if df.index.tz is not None:
            df.index = df.index.tz_localize(None)

        # Cache the data
        if cache_dir and len(df) > 0:
            cache_dir.mkdir(parents=True, exist_ok=True)
            cache_file = cache_dir / f"{symbol.replace('/', '_').replace('-', '_')}_1d.csv"
            df.to_csv(cache_file)

        return df

    except Exception as e:
        print(f"Error fetching {symbol}: {e}")
        return pd.DataFrame()


def scan_symbols(
    symbols: List[str],
    period: str = '6m',
    params: Dict[str, Any] = None,
    cache_dir: Path = None,
    quiet: bool = False
) -> List[TradingSignal]:
    """
    Scan multiple symbols and generate signals.

    Args:
        symbols: List of trading symbols
        period: Lookback period
        params: Signal generator parameters
        cache_dir: Directory for data caching
        quiet: Suppress output

    Returns:
        List of TradingSignal objects
    """
    generator = SignalGenerator(params or {})
    signals = []

    for symbol in symbols:
        if not quiet:
            print(f"Scanning {symbol}...", end=' ')

        df = fetch_data(symbol, period, cache_dir)

        if len(df) < 50:
            if not quiet:
                print("SKIP (insufficient data)")
            continue

        # Calculate indicators
        df = calculate_all_indicators(df, params)

        # Generate signal
        signal = generator.generate_signal(df, symbol)
        signals.append(signal)

        if not quiet:
            emoji = {
                SignalType.STRONG_BUY: "STRONG BUY",
                SignalType.BUY: "BUY",
                SignalType.NEUTRAL: "NEUTRAL",
                SignalType.SELL: "SELL",
                SignalType.STRONG_SELL: "STRONG SELL",
            }
            print(f"{emoji[signal.signal]} ({signal.confidence:.0f}%)")

    return signals


def filter_signals(
    signals: List[TradingSignal],
    min_confidence: float = 0,
    signal_types: List[SignalType] = None
) -> List[TradingSignal]:
    """Filter signals by confidence and type."""
    result = signals

    if min_confidence > 0:
        result = [s for s in result if s.confidence >= min_confidence]

    if signal_types:
        result = [s for s in result if s.signal in signal_types]

    return result


def rank_signals(signals: List[TradingSignal], by: str = 'confidence') -> List[TradingSignal]:
    """Rank signals by confidence or strength."""
    if by == 'confidence':
        return sorted(signals, key=lambda s: s.confidence, reverse=True)
    elif by == 'bullish':
        score_map = {
            SignalType.STRONG_BUY: 5,
            SignalType.BUY: 4,
            SignalType.NEUTRAL: 3,
            SignalType.SELL: 2,
            SignalType.STRONG_SELL: 1,
        }
        return sorted(signals, key=lambda s: (score_map[s.signal], s.confidence), reverse=True)
    elif by == 'bearish':
        score_map = {
            SignalType.STRONG_SELL: 5,
            SignalType.SELL: 4,
            SignalType.NEUTRAL: 3,
            SignalType.BUY: 2,
            SignalType.STRONG_BUY: 1,
        }
        return sorted(signals, key=lambda s: (score_map[s.signal], s.confidence), reverse=True)
    else:
        return signals


def print_summary(signals: List[TradingSignal]) -> None:
    """Print a summary table of all signals."""
    print("\n" + "=" * 80)
    print("  SIGNAL SCANNER RESULTS")
    print("=" * 80)
    print(f"\n  {'Symbol':<12} {'Signal':<14} {'Confidence':>10} {'Price':>14} {'Stop Loss':>12}")
    print("-" * 80)

    for signal in signals:
        sl = f"${signal.stop_loss:,.2f}" if signal.stop_loss else "N/A"
        print(f"  {signal.symbol:<12} {signal.signal.value:<14} {signal.confidence:>9.1f}% ${signal.price:>12,.2f} {sl:>12}")

    print("-" * 80)

    # Summary stats
    buy_count = sum(1 for s in signals if s.signal in [SignalType.STRONG_BUY, SignalType.BUY])
    sell_count = sum(1 for s in signals if s.signal in [SignalType.STRONG_SELL, SignalType.SELL])
    neutral_count = sum(1 for s in signals if s.signal == SignalType.NEUTRAL)

    print(f"\n  Summary: {buy_count} Buy | {neutral_count} Neutral | {sell_count} Sell")
    print(f"  Scanned: {len(signals)} assets | {datetime.now().strftime('%Y-%m-%d %H:%M')}")
    print("=" * 80 + "\n")


def save_results(signals: List[TradingSignal], output_path: Path) -> None:
    """Save signals to JSON file."""
    data = {
        'generated_at': datetime.now().isoformat(),
        'count': len(signals),
        'signals': [s.to_dict() for s in signals]
    }

    output_path.parent.mkdir(parents=True, exist_ok=True)
    with open(output_path, 'w') as f:
        json.dump(data, f, indent=2)

    print(f"Results saved to: {output_path}")


def main():
    parser = argparse.ArgumentParser(description='Crypto Signal Scanner')
    parser.add_argument('--symbols', '-s', help='Comma-separated symbols (e.g., BTC-USD,ETH-USD)')
    parser.add_argument('--watchlist', '-w', help='Predefined watchlist name')
    parser.add_argument('--period', '-p', default='6m', help='Lookback period (e.g., 6m, 1y)')
    parser.add_argument('--min-confidence', type=float, default=0, help='Minimum confidence filter')
    parser.add_argument('--filter', choices=['buy', 'sell', 'all'], default='all', help='Filter signals')
    parser.add_argument('--rank', choices=['confidence', 'bullish', 'bearish'], help='Rank signals')
    parser.add_argument('--output', '-o', help='Output JSON file')
    parser.add_argument('--detail', '-d', action='store_true', help='Show detailed signal breakdown')
    parser.add_argument('--list-watchlists', action='store_true', help='List available watchlists')
    parser.add_argument('--quiet', '-q', action='store_true', help='Minimal output')

    args = parser.parse_args()

    # List watchlists
    if args.list_watchlists:
        print("\nAvailable watchlists:")
        for name, symbols in WATCHLISTS.items():
            print(f"  {name}: {', '.join(symbols[:3])}... ({len(symbols)} symbols)")
        return

    # Determine symbols to scan
    if args.symbols:
        symbols = [s.strip() for s in args.symbols.split(',')]
    elif args.watchlist:
        if args.watchlist not in WATCHLISTS:
            print(f"Unknown watchlist: {args.watchlist}")
            print(f"Available: {', '.join(WATCHLISTS.keys())}")
            sys.exit(1)
        symbols = WATCHLISTS[args.watchlist]
    else:
        symbols = WATCHLISTS['crypto_top10']

    # Set up paths
    script_dir = Path(__file__).parent.parent
    cache_dir = script_dir / 'data'

    if not args.quiet:
        print(f"\nScanning {len(symbols)} symbols...")
        print(f"Period: {args.period}\n")

    # Scan symbols
    signals = scan_symbols(symbols, args.period, cache_dir=cache_dir, quiet=args.quiet)

    # Filter
    if args.filter == 'buy':
        signals = filter_signals(signals, signal_types=[SignalType.STRONG_BUY, SignalType.BUY])
    elif args.filter == 'sell':
        signals = filter_signals(signals, signal_types=[SignalType.STRONG_SELL, SignalType.SELL])

    if args.min_confidence > 0:
        signals = filter_signals(signals, min_confidence=args.min_confidence)

    # Rank
    if args.rank:
        signals = rank_signals(signals, args.rank)

    # Output
    if args.detail:
        for signal in signals:
            print(format_signal(signal))
    else:
        print_summary(signals)

    # Save to file
    if args.output:
        save_results(signals, Path(args.output))


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
Signal Generation Engine
Combine multiple indicators to generate trading signals with confidence scores.
"""

from dataclasses import dataclass, field
from enum import Enum
from typing import List, Dict, Any, Optional
import pandas as pd
import numpy as np


class SignalType(Enum):
    STRONG_BUY = "STRONG_BUY"
    BUY = "BUY"
    NEUTRAL = "NEUTRAL"
    SELL = "SELL"
    STRONG_SELL = "STRONG_SELL"


@dataclass
class SignalComponent:
    """Individual indicator signal."""
    name: str
    signal: SignalType
    weight: float
    value: float
    threshold: str
    reasoning: str


@dataclass
class TradingSignal:
    """Composite trading signal from multiple indicators."""
    symbol: str
    timestamp: pd.Timestamp
    signal: SignalType
    confidence: float  # 0-100
    components: List[SignalComponent] = field(default_factory=list)
    price: float = 0.0
    stop_loss: Optional[float] = None
    take_profit: Optional[float] = None
    risk_reward: Optional[float] = None

    def to_dict(self) -> Dict[str, Any]:
        return {
            'symbol': self.symbol,
            'timestamp': str(self.timestamp),
            'signal': self.signal.value,
            'confidence': round(self.confidence, 1),
            'price': self.price,
            'stop_loss': self.stop_loss,
            'take_profit': self.take_profit,
            'risk_reward': self.risk_reward,
            'components': [
                {
                    'name': c.name,
                    'signal': c.signal.value,
                    'value': round(c.value, 2),
                    'reasoning': c.reasoning
                }
                for c in self.components
            ]
        }


class SignalGenerator:
    """Generate trading signals from technical indicators."""

    def __init__(self, params: Dict[str, Any] = None):
        self.params = params or {}
        self.weights = {
            'rsi': self.params.get('rsi_weight', 1.0),
            'macd': self.params.get('macd_weight', 1.0),
            'bollinger': self.params.get('bb_weight', 1.0),
            'trend': self.params.get('trend_weight', 1.0),
            'volume': self.params.get('volume_weight', 0.5),
            'stochastic': self.params.get('stoch_weight', 0.5),
            'adx': self.params.get('adx_weight', 0.5),
        }

    def _signal_to_score(self, signal: SignalType) -> float:
        """Convert signal to numeric score (-2 to +2)."""
        mapping = {
            SignalType.STRONG_BUY: 2,
            SignalType.BUY: 1,
            SignalType.NEUTRAL: 0,
            SignalType.SELL: -1,
            SignalType.STRONG_SELL: -2,
        }
        return mapping[signal]

    def _score_to_signal(self, score: float) -> SignalType:
        """Convert numeric score to signal."""
        if score >= 1.5:
            return SignalType.STRONG_BUY
        elif score >= 0.5:
            return SignalType.BUY
        elif score <= -1.5:
            return SignalType.STRONG_SELL
        elif score <= -0.5:
            return SignalType.SELL
        else:
            return SignalType.NEUTRAL

    def analyze_rsi(self, row: pd.Series) -> SignalComponent:
        """Analyze RSI for overbought/oversold conditions."""
        rsi = row['rsi']
        oversold = self.params.get('rsi_oversold', 30)
        overbought = self.params.get('rsi_overbought', 70)

        if pd.isna(rsi):
            return SignalComponent(
                name='RSI',
                signal=SignalType.NEUTRAL,
                weight=self.weights['rsi'],
                value=50,
                threshold=f'{oversold}/{overbought}',
                reasoning='Insufficient data'
            )

        if rsi < oversold:
            signal = SignalType.STRONG_BUY
            reasoning = f'Oversold at {rsi:.1f} (< {oversold})'
        elif rsi < 40:
            signal = SignalType.BUY
            reasoning = f'Approaching oversold at {rsi:.1f}'
        elif rsi > overbought:
            signal = SignalType.STRONG_SELL
            reasoning = f'Overbought at {rsi:.1f} (> {overbought})'
        elif rsi > 60:
            signal = SignalType.SELL
            reasoning = f'Approaching overbought at {rsi:.1f}'
        else:
            signal = SignalType.NEUTRAL
            reasoning = f'Neutral zone at {rsi:.1f}'

        return SignalComponent(
            name='RSI',
            signal=signal,
            weight=self.weights['rsi'],
            value=rsi,
            threshold=f'{oversold}/{overbought}',
            reasoning=reasoning
        )

    def analyze_macd(self, row: pd.Series, prev_row: pd.Series = None) -> SignalComponent:
        """Analyze MACD for trend and momentum."""
        macd = row['macd']
        signal_line = row['macd_signal']
        histogram = row['macd_hist']

        if pd.isna(macd) or pd.isna(signal_line):
            return SignalComponent(
                name='MACD',
                signal=SignalType.NEUTRAL,
                weight=self.weights['macd'],
                value=0,
                threshold='crossover',
                reasoning='Insufficient data'
            )

        # Check for crossover
        crossover = False
        if prev_row is not None and not pd.isna(prev_row['macd']):
            prev_above = prev_row['macd'] > prev_row['macd_signal']
            curr_above = macd > signal_line
            crossover = prev_above != curr_above

        if macd > signal_line and histogram > 0:
            if crossover:
                signal = SignalType.STRONG_BUY
                reasoning = 'Bullish crossover with positive momentum'
            else:
                signal = SignalType.BUY
                reasoning = 'MACD above signal, positive momentum'
        elif macd < signal_line and histogram < 0:
            if crossover:
                signal = SignalType.STRONG_SELL
                reasoning = 'Bearish crossover with negative momentum'
            else:
                signal = SignalType.SELL
                reasoning = 'MACD below signal, negative momentum'
        else:
            signal = SignalType.NEUTRAL
            reasoning = 'No clear MACD signal'

        return SignalComponent(
            name='MACD',
            signal=signal,
            weight=self.weights['macd'],
            value=histogram,
            threshold='crossover',
            reasoning=reasoning
        )

    def analyze_bollinger(self, row: pd.Series) -> SignalComponent:
        """Analyze Bollinger Band position."""
        bb_pct = row['bb_pct']
        close = row['close']
        bb_upper = row['bb_upper']
        bb_lower = row['bb_lower']

        if pd.isna(bb_pct):
            return SignalComponent(
                name='Bollinger Bands',
                signal=SignalType.NEUTRAL,
                weight=self.weights['bollinger'],
                value=0.5,
                threshold='0.0/1.0',
                reasoning='Insufficient data'
            )

        if bb_pct < 0:
            signal = SignalType.STRONG_BUY
            reasoning = f'Price below lower band ({close:.2f} < {bb_lower:.2f})'
        elif bb_pct < 0.2:
            signal = SignalType.BUY
            reasoning = f'Price near lower band (%B = {bb_pct:.2f})'
        elif bb_pct > 1:
            signal = SignalType.STRONG_SELL
            reasoning = f'Price above upper band ({close:.2f} > {bb_upper:.2f})'
        elif bb_pct > 0.8:
            signal = SignalType.SELL
            reasoning = f'Price near upper band (%B = {bb_pct:.2f})'
        else:
            signal = SignalType.NEUTRAL
            reasoning = f'Price in middle of bands (%B = {bb_pct:.2f})'

        return SignalComponent(
            name='Bollinger Bands',
            signal=signal,
            weight=self.weights['bollinger'],
            value=bb_pct,
            threshold='0.0/1.0',
            reasoning=reasoning
        )

    def analyze_trend(self, row: pd.Series) -> SignalComponent:
        """Analyze trend using moving averages."""
        close = row['close']
        sma_20 = row.get('sma_20', np.nan)
        sma_50 = row.get('sma_50', np.nan)
        sma_200 = row.get('sma_200', np.nan)

        if pd.isna(sma_20) or pd.isna(sma_50):
            return SignalComponent(
                name='Trend',
                signal=SignalType.NEUTRAL,
                weight=self.weights['trend'],
                value=0,
                threshold='SMA crossovers',
                reasoning='Insufficient data'
            )

        # Count bullish conditions
        bullish = 0
        bearish = 0

        if close > sma_20:
            bullish += 1
        else:
            bearish += 1

        if close > sma_50:
            bullish += 1
        else:
            bearish += 1

        if not pd.isna(sma_200):
            if close > sma_200:
                bullish += 1
            else:
                bearish += 1

            if sma_50 > sma_200:
                bullish += 1
            else:
                bearish += 1

        score = bullish - bearish

        if score >= 3:
            signal = SignalType.STRONG_BUY
            reasoning = 'Strong uptrend: price above all MAs, golden cross'
        elif score >= 1:
            signal = SignalType.BUY
            reasoning = 'Uptrend: price above key moving averages'
        elif score <= -3:
            signal = SignalType.STRONG_SELL
            reasoning = 'Strong downtrend: price below all MAs, death cross'
        elif score <= -1:
            signal = SignalType.SELL
            reasoning = 'Downtrend: price below key moving averages'
        else:
            signal = SignalType.NEUTRAL
            reasoning = 'Mixed trend signals'

        return SignalComponent(
            name='Trend',
            signal=signal,
            weight=self.weights['trend'],
            value=score,
            threshold='SMA crossovers',
            reasoning=reasoning
        )

    def analyze_volume(self, row: pd.Series) -> SignalComponent:
        """Analyze volume for confirmation."""
        volume_ratio = row.get('volume_ratio', np.nan)

        if pd.isna(volume_ratio):
            return SignalComponent(
                name='Volume',
                signal=SignalType.NEUTRAL,
                weight=self.weights['volume'],
                value=1.0,
                threshold='1.5x avg',
                reasoning='Insufficient data'
            )

        change = row.get('change_1d', 0)

        if volume_ratio > 2.0:
            if change > 0:
                signal = SignalType.STRONG_BUY
                reasoning = f'High volume ({volume_ratio:.1f}x) on up move'
            else:
                signal = SignalType.STRONG_SELL
                reasoning = f'High volume ({volume_ratio:.1f}x) on down move'
        elif volume_ratio > 1.5:
            if change > 0:
                signal = SignalType.BUY
                reasoning = f'Above-average volume ({volume_ratio:.1f}x) on up move'
            else:
                signal = SignalType.SELL
                reasoning = f'Above-average volume ({volume_ratio:.1f}x) on down move'
        else:
            signal = SignalType.NEUTRAL
            reasoning = f'Normal volume ({volume_ratio:.1f}x average)'

        return SignalComponent(
            name='Volume',
            signal=signal,
            weight=self.weights['volume'],
            value=volume_ratio,
            threshold='1.5x avg',
            reasoning=reasoning
        )

    def analyze_stochastic(self, row: pd.Series) -> SignalComponent:
        """Analyze Stochastic oscillator."""
        k = row.get('stoch_k', np.nan)
        d = row.get('stoch_d', np.nan)

        if pd.isna(k) or pd.isna(d):
            return SignalComponent(
                name='Stochastic',
                signal=SignalType.NEUTRAL,
                weight=self.weights['stochastic'],
                value=50,
                threshold='20/80',
                reasoning='Insufficient data'
            )

        if k < 20 and d < 20:
            signal = SignalType.STRONG_BUY
            reasoning = f'Oversold (%K={k:.1f}, %D={d:.1f})'
        elif k < 30:
            signal = SignalType.BUY
            reasoning = f'Approaching oversold (%K={k:.1f})'
        elif k > 80 and d > 80:
            signal = SignalType.STRONG_SELL
            reasoning = f'Overbought (%K={k:.1f}, %D={d:.1f})'
        elif k > 70:
            signal = SignalType.SELL
            reasoning = f'Approaching overbought (%K={k:.1f})'
        else:
            signal = SignalType.NEUTRAL
            reasoning = f'Neutral zone (%K={k:.1f})'

        return SignalComponent(
            name='Stochastic',
            signal=signal,
            weight=self.weights['stochastic'],
            value=k,
            threshold='20/80',
            reasoning=reasoning
        )

    def analyze_adx(self, row: pd.Series) -> SignalComponent:
        """Analyze ADX for trend strength."""
        adx = row.get('adx', np.nan)
        plus_di = row.get('plus_di', np.nan)
        minus_di = row.get('minus_di', np.nan)

        if pd.isna(adx):
            return SignalComponent(
                name='ADX',
                signal=SignalType.NEUTRAL,
                weight=self.weights['adx'],
                value=20,
                threshold='25 trend threshold',
                reasoning='Insufficient data'
            )

        if adx < 20:
            signal = SignalType.NEUTRAL
            reasoning = f'Weak/no trend (ADX={adx:.1f})'
        elif adx < 25:
            if plus_di > minus_di:
                signal = SignalType.BUY
                reasoning = f'Developing uptrend (ADX={adx:.1f}, +DI>-DI)'
            else:
                signal = SignalType.SELL
                reasoning = f'Developing downtrend (ADX={adx:.1f}, -DI>+DI)'
        else:
            if plus_di > minus_di:
                signal = SignalType.STRONG_BUY if adx > 40 else SignalType.BUY
                reasoning = f'Strong uptrend (ADX={adx:.1f}, +DI={plus_di:.1f})'
            else:
                signal = SignalType.STRONG_SELL if adx > 40 else SignalType.SELL
                reasoning = f'Strong downtrend (ADX={adx:.1f}, -DI={minus_di:.1f})'

        return SignalComponent(
            name='ADX',
            signal=signal,
            weight=self.weights['adx'],
            value=adx,
            threshold='25 trend threshold',
            reasoning=reasoning
        )

    def calculate_risk_levels(
        self,
        price: float,
        signal: SignalType,
        atr: float
    ) -> tuple:
        """Calculate stop-loss and take-profit levels."""
        atr_multiplier = self.params.get('atr_multiplier', 2.0)
        risk_reward_target = self.params.get('risk_reward', 2.0)

        if signal in [SignalType.STRONG_BUY, SignalType.BUY]:
            stop_loss = price - (atr * atr_multiplier)
            take_profit = price + (atr * atr_multiplier * risk_reward_target)
        elif signal in [SignalType.STRONG_SELL, SignalType.SELL]:
            stop_loss = price + (atr * atr_multiplier)
            take_profit = price - (atr * atr_multiplier * risk_reward_target)
        else:
            stop_loss = None
            take_profit = None

        risk_reward = risk_reward_target if stop_loss else None

        return stop_loss, take_profit, risk_reward

    def generate_signal(
        self,
        df: pd.DataFrame,
        symbol: str = 'Unknown'
    ) -> TradingSignal:
        """
        Generate a composite trading signal from all indicators.

        Args:
            df: DataFrame with price data and calculated indicators
            symbol: Trading symbol

        Returns:
            TradingSignal with composite signal and component breakdown
        """
        row = df.iloc[-1]
        prev_row = df.iloc[-2] if len(df) > 1 else None

        # Analyze each indicator
        components = [
            self.analyze_rsi(row),
            self.analyze_macd(row, prev_row),
            self.analyze_bollinger(row),
            self.analyze_trend(row),
            self.analyze_volume(row),
            self.analyze_stochastic(row),
            self.analyze_adx(row),
        ]

        # Calculate weighted score
        total_weight = sum(c.weight for c in components)
        weighted_score = sum(
            self._signal_to_score(c.signal) * c.weight
            for c in components
        ) / total_weight

        # Convert to composite signal
        composite_signal = self._score_to_signal(weighted_score)

        # Calculate confidence (0-100)
        # Higher when components agree, lower when mixed
        scores = [self._signal_to_score(c.signal) for c in components]
        agreement = 1 - (np.std(scores) / 2)  # 0 to 1
        strength = abs(weighted_score) / 2  # 0 to 1
        confidence = min(100, (agreement * 0.5 + strength * 0.5) * 100)

        # Calculate risk levels
        atr = row.get('atr', 0)
        stop_loss, take_profit, risk_reward = self.calculate_risk_levels(
            row['close'], composite_signal, atr
        )

        return TradingSignal(
            symbol=symbol,
            timestamp=df.index[-1],
            signal=composite_signal,
            confidence=confidence,
            components=components,
            price=row['close'],
            stop_loss=stop_loss,
            take_profit=take_profit,
            risk_reward=risk_reward
        )


def format_signal(signal: TradingSignal) -> str:
    """Format trading signal for display."""
    lines = []

    # Signal header with emoji
    emoji = {
        SignalType.STRONG_BUY: "🟢🟢",
        SignalType.BUY: "🟢",
        SignalType.NEUTRAL: "⚪",
        SignalType.SELL: "🔴",
        SignalType.STRONG_SELL: "🔴🔴",
    }

    lines.append("=" * 70)
    lines.append(f"  {emoji[signal.signal]} {signal.symbol} - {signal.signal.value}")
    lines.append(f"  Confidence: {signal.confidence:.1f}% | Price: ${signal.price:,.2f}")
    lines.append("=" * 70)

    # Risk levels
    if signal.stop_loss:
        lines.append(f"\n  Risk Management:")
        lines.append(f"    Stop Loss:   ${signal.stop_loss:,.2f}")
        lines.append(f"    Take Profit: ${signal.take_profit:,.2f}")
        lines.append(f"    Risk/Reward: 1:{signal.risk_reward:.1f}")

    # Component breakdown
    lines.append(f"\n  Signal Components:")
    lines.append("-" * 70)

    for comp in signal.components:
        indicator_emoji = emoji[comp.signal]
        lines.append(f"    {indicator_emoji} {comp.name:15} | {comp.signal.value:12} | {comp.reasoning}")

    lines.append("-" * 70)
    lines.append(f"  Generated: {signal.timestamp}")
    lines.append("")

    return "\n".join(lines)