server.py

#!/usr/bin/env python3
"""
TradingView MCP Server - Professional Multi-Asset Trading Assistant

Provides real-time market data, technical analysis, and trading recommendations
via Model Context Protocol (MCP) for Claude Desktop integration.

Supported Assets:
- Forex (22+ pairs + gold)
- Stocks (US equities)
- Crypto (BTC, ETH, and major cryptocurrencies)

Data Sources:
- Alpha Vantage API for real-time quotes and historical data
"""

from __future__ import annotations

import logging
import sys
from pathlib import Path
from typing import Any, Dict, List

# Setup logging
log_dir = Path(__file__).parent.parent / "logs"
log_dir.mkdir(exist_ok=True)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    handlers=[
        logging.StreamHandler(sys.stderr),
        logging.FileHandler(log_dir / "tradingview_mcp.log"),
    ],
)
logger = logging.getLogger(__name__)

# Add project root to path
PROJECT_ROOT = Path(__file__).parent.parent
sys.path.insert(0, str(PROJECT_ROOT))

from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP

# Load environment variables
load_dotenv(PROJECT_ROOT / ".env")

# Validate environment setup
import os

# Import our refactored modules
from tradingview_mcp.api import AlphaVantageClient
from tradingview_mcp.config import (
    CRYPTO_SYMBOLS,
    DEFAULT_SPREAD,
    FOREX_PAIRS,
    HIGH_RISK_PAIRS,
    KNOWN_CORRELATIONS,
    MAJOR_PAIRS,
    POPULAR_STOCKS,
    TYPICAL_SPREADS,
)
from tradingview_mcp.indicators import (
    calculate_adx,
    calculate_atr,
    calculate_bollinger_bands,
    calculate_cci,
    calculate_fibonacci_levels,
    calculate_ichimoku,
    calculate_macd,
    calculate_market_profile,
    calculate_moving_averages,
    calculate_pivot_points,
    calculate_rsi,
    calculate_stochastic,
    calculate_volume_profile,
    calculate_vwap,
    calculate_williams_r,
    detect_gaps,
    detect_support_resistance,
)

# Import Pine Script modules
from tradingview_mcp.pine_script import (
    ErrorExplainer,
    PineAutocomplete,
    PineDocumentation,
    PineSandbox,
    PineScriptValidator,
    VersionConverter,
    VersionDetector,
)
from tradingview_mcp.utils import (
    detect_asset_type,
    format_error_response,
    format_success_response,
    validate_api_key,
)

api_key = os.getenv("ALPHA_VANTAGE_API_KEY")
if api_key:
    is_valid, error_msg = validate_api_key(api_key)
    if not is_valid:
        logger.warning(f"API key validation warning: {error_msg}")
else:
    logger.error("ALPHA_VANTAGE_API_KEY not set. Server will not function properly.")

# Initialize API client
api_client = AlphaVantageClient()

# Initialize Pine Script components
pine_validator = PineScriptValidator()
pine_docs = PineDocumentation()
pine_sandbox = PineSandbox()
pine_errors = ErrorExplainer()
pine_version_detector = VersionDetector()
pine_version_converter = VersionConverter()
pine_autocomplete = PineAutocomplete()

# Initialize MCP server
mcp = FastMCP(
    name="TradingView Multi-Asset Trading Assistant",
    instructions=(
        "Professional trading assistant providing real-time market data, "
        "technical analysis, and trading recommendations for multiple asset classes. "
        "Supports Forex (22+ pairs), Stocks (US equities), and Crypto (BTC, ETH, etc). "
        "Includes advanced tools: Volume Profile, Market Profile, VWAP, Fibonacci, "
        "Bollinger Bands, MACD, Moving Averages, ATR, Support/Resistance, Pivot Points, "
        "Stochastic, ADX, Ichimoku Cloud, and more. "
        "\n\n"
        "PINE SCRIPT SUPPORT: Comprehensive Pine Script development tools including "
        "real-time syntax validation, function documentation, code testing sandbox, "
        "error explanations, version detection (v1-v6), version conversion, and "
        "intelligent autocomplete. Full support for Pine Script v6 with type, enum, and map!"
    ),
)

logger.info("TradingView MCP Server initialized")


# ===== HELPER FUNCTIONS =====


def get_quote(symbol: str) -> Dict[str, Any]:
    """
    Universal quote fetcher for all asset types.

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC')

    Returns:
        Quote data or error
    """
    asset_type, formatted_symbol = detect_asset_type(symbol)

    logger.info(f"Fetching quote for {symbol} (type: {asset_type})")

    if asset_type == "forex":
        return api_client.get_forex_quote(formatted_symbol)
    elif asset_type == "crypto":
        return api_client.get_crypto_quote(formatted_symbol)
    else:  # stock
        return api_client.get_stock_quote(formatted_symbol)


def get_historical_data(
    symbol: str, timeframe: str = "1h", outputsize: str = "compact"
) -> Dict[str, Any]:
    """
    Get historical data for any asset type.

    Args:
        symbol: Symbol to fetch
        timeframe: Timeframe (5m, 15m, 30m, 1h, 4h, 1d)
        outputsize: 'compact' or 'full'

    Returns:
        Historical OHLCV data or error
    """
    asset_type, formatted_symbol = detect_asset_type(symbol)

    logger.info(f"Fetching historical data for {symbol} ({timeframe})")

    # Route to appropriate API method based on asset type
    if asset_type == "forex":
        return api_client.get_historical_data_forex(
            formatted_symbol, timeframe, outputsize
        )
    elif asset_type == "crypto":
        return api_client.get_historical_data_crypto(
            formatted_symbol, timeframe, outputsize
        )
    else:  # stock
        return api_client.get_historical_data_stock(
            formatted_symbol, timeframe, outputsize
        )


def get_spread(pair: str) -> float:
    """Get typical spread for a pair."""
    return TYPICAL_SPREADS.get(pair, DEFAULT_SPREAD)


# ===== MCP TOOLS: SERVER MANAGEMENT =====


@mcp.tool()
def health_check() -> dict:
    """
    Check server health and API connectivity status.

    Returns:
        Server health information including version, cache stats, and API status
    """
    # Check API key configuration
    api_key_configured = bool(api_client.api_key)

    # Get cache statistics
    cache_stats = api_client.cache.get_stats()

    # Server version (from pyproject.toml)
    version = "3.3.0"

    return {
        "status": "healthy" if api_key_configured else "degraded",
        "version": version,
        "api_key_configured": api_key_configured,
        "cache": {
            "size": cache_stats["size"],
            "max_size": cache_stats["max_size"],
            "hit_rate": cache_stats["hit_rate"],
            "utilization": cache_stats["utilization"],
            "evictions": cache_stats["evictions"],
        },
        "total_api_calls": api_client._total_calls,
        "warnings": (
            []
            if api_key_configured
            else ["API key not configured - server will not function properly"]
        ),
    }


# ===== MCP TOOLS: MARKET DATA =====


@mcp.tool()
def get_price(symbol: str) -> dict:
    """
    Get current price for any asset (Forex, Stock, or Crypto).

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC', 'TSLA')

    Returns:
        Current price, bid/ask, asset type, and additional data
    """
    quote = get_quote(symbol)

    if "error" in quote:
        return quote

    return {
        "symbol": quote.get("symbol", symbol),
        "asset_type": quote.get("asset_type", "unknown"),
        "price": quote.get("price"),
        "bid": quote.get("bid"),
        "ask": quote.get("ask"),
        "timestamp": quote.get("timestamp"),
        **{
            k: v
            for k, v in quote.items()
            if k not in ["symbol", "asset_type", "price", "bid", "ask", "timestamp"]
        },
    }


@mcp.tool()
def get_multiple_prices(symbols: List[str]) -> list:
    """
    Get current prices for multiple symbols at once.

    Args:
        symbols: List of symbols (e.g., ['EURUSD', 'AAPL', 'BTC'])

    Returns:
        List of price quotes for each symbol
    """
    results = []
    for symbol in symbols:
        results.append(get_price(symbol))
    return results


@mcp.tool()
def list_available_pairs() -> dict:
    """
    List all available forex pairs organized by category.

    Returns:
        Dictionary with majors, crosses, exotics, and commodities
    """
    majors = [p for p in MAJOR_PAIRS if p in FOREX_PAIRS]
    exotics = ["USDTRY", "USDZAR", "USDMXN", "USDBRL"]
    commodities = ["XAUUSD"]
    crosses = [
        p
        for p in FOREX_PAIRS
        if p not in majors and p not in exotics and p not in commodities
    ]

    return {
        "total_pairs": len(FOREX_PAIRS),
        "majors": majors,
        "crosses": crosses,
        "exotics": exotics,
        "commodities": commodities,
        "all_pairs": FOREX_PAIRS,
    }


@mcp.tool()
def list_supported_assets() -> dict:
    """
    List all supported assets by category.

    Returns:
        Forex pairs, popular stocks, cryptocurrencies
    """
    return {
        "forex": {
            "count": len(FOREX_PAIRS),
            "majors": FOREX_PAIRS[:7],
            "crosses": FOREX_PAIRS[7:16],
            "exotics": FOREX_PAIRS[16:20],
            "commodities": ["XAUUSD"],
        },
        "stocks": {
            "count": len(POPULAR_STOCKS),
            "popular": POPULAR_STOCKS[:20],
            "etfs": ["SPY", "QQQ", "IWM", "DIA"],
        },
        "crypto": {
            "count": len(CRYPTO_SYMBOLS),
            "major": CRYPTO_SYMBOLS[:8],
            "altcoins": CRYPTO_SYMBOLS[8:],
        },
        "note": "Any valid symbol can be queried - these are just popular examples",
    }


# ===== MCP TOOLS: TECHNICAL ANALYSIS =====


@mcp.tool()
def analyze_pair(symbol: str, timeframe: str = "1h") -> dict:
    """
    Comprehensive technical analysis for any symbol.

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC')
        timeframe: Time interval ('5m', '15m', '1h', '4h', '1d')

    Returns:
        Detailed analysis including price, indicators, and trading recommendation
    """
    # Get current price
    quote = get_quote(symbol)
    if "error" in quote:
        return quote

    asset_type = quote.get("asset_type", "unknown")

    # Get historical data for analysis
    hist_data = get_historical_data(symbol, timeframe)
    if not hist_data.get("success"):
        # Return basic quote if can't get historical data
        return {
            "symbol": symbol,
            "asset_type": asset_type,
            "timeframe": timeframe,
            "current_price": quote["price"],
            "bid": quote["bid"],
            "ask": quote["ask"],
            "timestamp": quote["timestamp"],
            "note": "Limited analysis - historical data unavailable",
        }

    # Calculate indicators
    ma = calculate_moving_averages(hist_data["data"])
    bb = calculate_bollinger_bands(hist_data["data"])

    # Generate signals and recommendation
    signals = []
    recommendation = "NEUTRAL"

    # Analyze Bollinger Bands
    if "percent_b" in bb:
        if bb["percent_b"] > 1:
            signals.append("Price above upper Bollinger Band (overbought)")
            recommendation = "SELL"
        elif bb["percent_b"] < 0:
            signals.append("Price below lower Bollinger Band (oversold)")
            recommendation = "BUY"

    # Risk classification
    if asset_type == "forex" and symbol in HIGH_RISK_PAIRS:
        risk_level = "HIGH"
    elif asset_type == "forex" and symbol in MAJOR_PAIRS:
        risk_level = "LOW"
    else:
        risk_level = "MEDIUM"

    return {
        "symbol": symbol,
        "asset_type": asset_type,
        "timeframe": timeframe,
        "current_price": quote["price"],
        "bid": quote["bid"],
        "ask": quote["ask"],
        "timestamp": quote["timestamp"],
        "moving_averages": ma,
        "bollinger_bands": bb,
        "signals": signals,
        "recommendation": recommendation,
        "risk_level": risk_level,
    }


@mcp.tool()
def calculate_correlation(pair1: str, pair2: str, period: int = 30) -> dict:
    """
    Calculate correlation between two forex pairs.

    Args:
        pair1: First forex pair
        pair2: Second forex pair
        period: Number of days to analyze (default 30)

    Returns:
        Correlation coefficient and relationship strength
    """
    pair1 = pair1.upper().replace("/", "").replace("_", "")
    pair2 = pair2.upper().replace("/", "").replace("_", "")

    # Check known correlations
    correlation = KNOWN_CORRELATIONS.get((pair1, pair2)) or KNOWN_CORRELATIONS.get(
        (pair2, pair1)
    )

    if correlation is None:
        correlation = 0.0  # Unknown correlation

    # Determine strength
    abs_corr = abs(correlation)
    if abs_corr > 0.8:
        strength = "VERY STRONG"
    elif abs_corr > 0.6:
        strength = "STRONG"
    elif abs_corr > 0.4:
        strength = "MODERATE"
    elif abs_corr > 0.2:
        strength = "WEAK"
    else:
        strength = "VERY WEAK"

    relationship = (
        "POSITIVE" if correlation > 0 else "NEGATIVE" if correlation < 0 else "NEUTRAL"
    )

    return {
        "pair1": pair1,
        "pair2": pair2,
        "correlation": round(correlation, 3),
        "strength": strength,
        "relationship": relationship,
        "period_days": period,
        "interpretation": f"{pair1} and {pair2} have a {strength} {relationship} correlation ({correlation:.2f})",
    }


# ===== MCP TOOLS: FIBONACCI & PIVOT POINTS =====


@mcp.tool()
def get_fibonacci_retracement(symbol: str, timeframe: str = "1h") -> dict:
    """
    Calculate Fibonacci retracement levels.

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC')
        timeframe: Time interval ('5m', '15m', '1h', '4h', '1d')

    Returns:
        Fibonacci levels (0%, 23.6%, 38.2%, 50%, 61.8%, 78.6%, 100%)
    """
    hist_data = get_historical_data(symbol, timeframe)

    if not hist_data.get("success"):
        return hist_data

    levels = calculate_fibonacci_levels(hist_data["data"])

    if "error" in levels:
        return levels

    return {
        "symbol": symbol,
        "timeframe": timeframe,
        "levels": levels,
        "interpretation": "Key retracement levels where price may find support/resistance",
    }


@mcp.tool()
def get_pivot_points(symbol: str, timeframe: str = "1h") -> dict:
    """
    Calculate pivot points for intraday trading.

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC')
        timeframe: Time interval ('5m', '15m', '1h', '4h', '1d')

    Returns:
        Pivot point, 3 resistance levels, 3 support levels
    """
    hist_data = get_historical_data(symbol, timeframe)

    if not hist_data.get("success"):
        return hist_data

    pivots = calculate_pivot_points(hist_data["data"])

    if "error" in pivots:
        return pivots

    return {"symbol": symbol, "timeframe": timeframe, **pivots}


# ===== MCP TOOLS: TREND INDICATORS =====


@mcp.tool()
def get_moving_averages(symbol: str, timeframe: str = "1h") -> dict:
    """
    Calculate multiple moving averages (20, 50, 100, 200 period SMA/EMA).

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC')
        timeframe: Time interval ('5m', '15m', '1h', '4h', '1d')

    Returns:
        SMAs and EMAs for 20, 50, 100, 200 periods and current price
    """
    hist_data = get_historical_data(symbol, timeframe)

    if not hist_data.get("success"):
        return hist_data

    mas = calculate_moving_averages(hist_data["data"])

    return {
        "symbol": symbol,
        "timeframe": timeframe,
        **mas,
        "interpretation": "Compare current price to MA levels to identify trend direction",
    }


@mcp.tool()
def get_macd(symbol: str, timeframe: str = "1h") -> dict:
    """
    Calculate MACD (Moving Average Convergence Divergence).

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC')
        timeframe: Time interval ('5m', '15m', '1h', '4h', '1d')

    Returns:
        MACD line, signal line, histogram, and trading signal
    """
    hist_data = get_historical_data(symbol, timeframe)

    if not hist_data.get("success"):
        return hist_data

    macd = calculate_macd(hist_data["data"])

    if "error" in macd:
        return macd

    return {
        "symbol": symbol,
        "timeframe": timeframe,
        **macd,
        "interpretation": f"MACD is {macd['signal']}. Histogram: {macd['histogram']:.5f}",
    }


@mcp.tool()
def get_adx(symbol: str, timeframe: str = "1h", period: int = 14) -> dict:
    """
    Calculate ADX (Average Directional Index) - trend strength indicator.

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC')
        timeframe: Time interval ('5m', '15m', '1h', '4h', '1d')
        period: Period for calculation (default 14)

    Returns:
        ADX value, +DI, -DI, and trend strength classification
    """
    hist_data = get_historical_data(symbol, timeframe)

    if not hist_data.get("success"):
        return hist_data

    adx = calculate_adx(hist_data["data"], period)

    if "error" in adx:
        return adx

    return {
        "symbol": symbol,
        "timeframe": timeframe,
        **adx,
        "interpretation": f"Trend strength is {adx['trend_strength']} (ADX: {adx['adx']:.1f})",
    }


@mcp.tool()
def get_ichimoku_cloud(symbol: str, timeframe: str = "1h") -> dict:
    """
    Calculate Ichimoku Cloud components.

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC')
        timeframe: Time interval ('5m', '15m', '1h', '4h', '1d')

    Returns:
        Tenkan-sen, Kijun-sen, Senkou Span A/B, cloud position
    """
    hist_data = get_historical_data(symbol, timeframe)

    if not hist_data.get("success"):
        return hist_data

    ichimoku = calculate_ichimoku(hist_data["data"])

    if "error" in ichimoku:
        return ichimoku

    return {"symbol": symbol, "timeframe": timeframe, **ichimoku}


# ===== MCP TOOLS: MOMENTUM INDICATORS =====


@mcp.tool()
def get_stochastic(symbol: str, timeframe: str = "1h", period: int = 14) -> dict:
    """
    Calculate Stochastic Oscillator.

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC')
        timeframe: Time interval ('5m', '15m', '1h', '4h', '1d')
        period: Period for calculation (default 14)

    Returns:
        %K and %D values, signal (OVERBOUGHT/OVERSOLD/NEUTRAL)
    """
    hist_data = get_historical_data(symbol, timeframe)

    if not hist_data.get("success"):
        return hist_data

    stoch = calculate_stochastic(hist_data["data"], period)

    if "error" in stoch:
        return stoch

    return {"symbol": symbol, "timeframe": timeframe, **stoch}


@mcp.tool()
def get_rsi(symbol: str, timeframe: str = "1h", period: int = 14) -> dict:
    """
    Calculate RSI (Relative Strength Index).

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC')
        timeframe: Time interval ('5m', '15m', '1h', '4h', '1d')
        period: Period for calculation (default 14)

    Returns:
        RSI value and signal (OVERBOUGHT/OVERSOLD/NEUTRAL)
    """
    hist_data = get_historical_data(symbol, timeframe)

    if not hist_data.get("success"):
        return hist_data

    rsi = calculate_rsi(hist_data["data"], period)

    if "error" in rsi:
        return rsi

    return {"symbol": symbol, "timeframe": timeframe, **rsi}


@mcp.tool()
def get_cci(symbol: str, timeframe: str = "1h", period: int = 20) -> dict:
    """
    Calculate CCI (Commodity Channel Index).

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC')
        timeframe: Time interval ('5m', '15m', '1h', '4h', '1d')
        period: Period for calculation (default 20)

    Returns:
        CCI value and signal (OVERBOUGHT/OVERSOLD/NEUTRAL)
    """
    hist_data = get_historical_data(symbol, timeframe)

    if not hist_data.get("success"):
        return hist_data

    cci = calculate_cci(hist_data["data"], period)

    if "error" in cci:
        return cci

    return {"symbol": symbol, "timeframe": timeframe, **cci}


@mcp.tool()
def get_williams_r(symbol: str, timeframe: str = "1h", period: int = 14) -> dict:
    """
    Calculate Williams %R.

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC')
        timeframe: Time interval ('5m', '15m', '1h', '4h', '1d')
        period: Period for calculation (default 14)

    Returns:
        Williams %R value and signal (OVERBOUGHT/OVERSOLD/NEUTRAL)
    """
    hist_data = get_historical_data(symbol, timeframe)

    if not hist_data.get("success"):
        return hist_data

    williams = calculate_williams_r(hist_data["data"], period)

    if "error" in williams:
        return williams

    return {"symbol": symbol, "timeframe": timeframe, **williams}


# ===== MCP TOOLS: VOLATILITY INDICATORS =====


@mcp.tool()
def get_bollinger_bands(symbol: str, timeframe: str = "1h", period: int = 20) -> dict:
    """
    Calculate Bollinger Bands.

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC')
        timeframe: Time interval ('5m', '15m', '1h', '4h', '1d')
        period: Period for calculation (default 20)

    Returns:
        Upper band, middle band (SMA), lower band, and %B indicator
    """
    hist_data = get_historical_data(symbol, timeframe)

    if not hist_data.get("success"):
        return hist_data

    bb = calculate_bollinger_bands(hist_data["data"], period)

    if "error" in bb:
        return bb

    return {"symbol": symbol, "timeframe": timeframe, **bb}


@mcp.tool()
def get_atr(symbol: str, timeframe: str = "1h", period: int = 14) -> dict:
    """
    Calculate Average True Range (ATR) - volatility indicator.

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC')
        timeframe: Time interval ('5m', '15m', '1h', '4h', '1d')
        period: Period for calculation (default 14)

    Returns:
        ATR value and percentage
    """
    hist_data = get_historical_data(symbol, timeframe)

    if not hist_data.get("success"):
        return hist_data

    atr = calculate_atr(hist_data["data"], period)

    if "error" in atr:
        return atr

    return {"symbol": symbol, "timeframe": timeframe, **atr}


# ===== MCP TOOLS: VOLUME INDICATORS =====


@mcp.tool()
def get_vwap(symbol: str, timeframe: str = "1h") -> dict:
    """
    Calculate VWAP (Volume Weighted Average Price).

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC')
        timeframe: Time interval ('5m', '15m', '1h', '4h', '1d')

    Returns:
        VWAP value and current price comparison
    """
    hist_data = get_historical_data(symbol, timeframe)

    if not hist_data.get("success"):
        return hist_data

    vwap = calculate_vwap(hist_data["data"])

    # Get current price
    current_quote = get_quote(symbol)
    if "error" in current_quote:
        return current_quote

    current_price = current_quote["price"]
    deviation = ((current_price - vwap) / vwap) * 100 if vwap > 0 else 0

    return {
        "symbol": symbol,
        "timeframe": timeframe,
        "vwap": round(vwap, 5),
        "current_price": current_price,
        "deviation_percent": round(deviation, 2),
        "signal": "ABOVE_VWAP" if current_price > vwap else "BELOW_VWAP",
        "interpretation": f"Price is {abs(deviation):.2f}% {'above' if deviation > 0 else 'below'} VWAP. "
        f"{'Bullish' if deviation > 0 else 'Bearish'} pressure indicated.",
    }


@mcp.tool()
def get_volume_profile(
    symbol: str, timeframe: str = "1h", num_levels: int = 20
) -> dict:
    """
    Get volume profile analysis.

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC')
        timeframe: Time interval ('5m', '15m', '1h', '4h', '1d')
        num_levels: Number of price levels to analyze (default 20)

    Returns:
        Volume profile with POC, high/low volume nodes, and price levels
    """
    hist_data = get_historical_data(symbol, timeframe)

    if not hist_data.get("success"):
        return hist_data

    profile = calculate_volume_profile(hist_data["data"], num_levels)

    if "error" in profile:
        return profile

    return {
        "symbol": symbol,
        "timeframe": timeframe,
        **profile,
        "interpretation": f"POC at {profile['poc']:.5f} shows highest traded volume. "
        f"High volume nodes indicate support/resistance zones.",
    }


@mcp.tool()
def get_market_profile(symbol: str, timeframe: str = "1h") -> dict:
    """
    Get market profile analysis with TPO, POC, and value areas.

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC')
        timeframe: Time interval ('5m', '15m', '1h', '4h', '1d')

    Returns:
        Market profile with POC, value area high/low, and TPO data
    """
    hist_data = get_historical_data(symbol, timeframe)

    if not hist_data.get("success"):
        return hist_data

    profile = calculate_market_profile(hist_data["data"])

    if "error" in profile:
        return profile

    return {"symbol": symbol, "timeframe": timeframe, **profile}


# ===== MCP TOOLS: SUPPORT/RESISTANCE =====


@mcp.tool()
def get_support_resistance(symbol: str, timeframe: str = "1h") -> dict:
    """
    Detect support and resistance levels automatically.

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC')
        timeframe: Time interval ('5m', '15m', '1h', '4h', '1d')

    Returns:
        Top 3 support and resistance levels
    """
    hist_data = get_historical_data(symbol, timeframe)

    if not hist_data.get("success"):
        return hist_data

    sr = detect_support_resistance(hist_data["data"])

    return {"symbol": symbol, "timeframe": timeframe, **sr}


@mcp.tool()
def detect_unfilled_gaps(symbol: str, timeframe: str = "1h") -> dict:
    """
    Detect unfilled price gaps (unfinished business areas).

    Args:
        symbol: Symbol (e.g., 'EURUSD', 'AAPL', 'BTC')
        timeframe: Time interval ('5m', '15m', '1h', '4h', '1d')

    Returns:
        List of unfilled gaps with price levels and timestamps
    """
    hist_data = get_historical_data(symbol, timeframe)

    if not hist_data.get("success"):
        return hist_data

    gaps = detect_gaps(hist_data["data"])

    # Get current price to check if gaps are still unfilled
    current_quote = get_quote(symbol)
    if "error" not in current_quote:
        current_price = current_quote["price"]

        for gap in gaps:
            # Check if gap is filled
            if gap["type"] == "gap_up" and current_price < gap["to_price"]:
                gap["filled"] = True
            elif gap["type"] == "gap_down" and current_price > gap["to_price"]:
                gap["filled"] = True

    unfilled_gaps = [g for g in gaps if not g["filled"]]

    return {
        "symbol": symbol,
        "timeframe": timeframe,
        "total_gaps": len(gaps),
        "unfilled_gaps": len(unfilled_gaps),
        "gaps": unfilled_gaps[:10],  # Return top 10
        "interpretation": f"Found {len(unfilled_gaps)} unfilled gaps. These often act as magnets for price action.",
    }


# ===== MCP TOOLS: SYSTEM =====


@mcp.tool()
def get_server_stats() -> dict:
    """
    Get server statistics including API usage and cache performance.

    Returns:
        Server statistics
    """
    return {"status": "online", **api_client.get_stats()}


# ===== MCP RESOURCES =====


@mcp.resource("forex://pairs")
def list_forex_pairs() -> str:
    """List all available forex pairs with categories."""
    pairs_data = list_available_pairs()

    output = "📊 Available Forex Pairs\n\n"
    output += f"**Major Pairs ({len(pairs_data['majors'])}):**\n"
    output += ", ".join(pairs_data["majors"]) + "\n\n"
    output += f"**Cross Pairs ({len(pairs_data['crosses'])}):**\n"
    output += ", ".join(pairs_data["crosses"]) + "\n\n"
    output += f"**Exotic Pairs ({len(pairs_data['exotics'])}):**\n"
    output += ", ".join(pairs_data["exotics"]) + "\n\n"
    output += f"**Commodities ({len(pairs_data['commodities'])}):**\n"
    output += ", ".join(pairs_data["commodities"]) + "\n\n"
    output += f"**Total:** {pairs_data['total_pairs']} pairs available"

    return output


# ===== PINE SCRIPT MCP TOOLS =====


@mcp.tool()
def validate_pine_script(code: str, version: int = None) -> dict:
    """
    Validate Pine Script code for syntax and semantic errors.

    Checks for:
    - Syntax errors (missing brackets, invalid operators, etc.)
    - Function signature errors (wrong parameters, unknown functions)
    - Deprecated function usage
    - Version compatibility issues
    - Type errors

    Args:
        code: Pine Script source code to validate
        version: Target Pine Script version (1-6). Auto-detected if not provided.

    Returns:
        Validation result with errors, warnings, and suggestions

    Example:
        validate_pine_script("//@version=5\\nindicator(\\"Test\\")\\nplot(ta.sma(close, 20))")
    """
    try:
        result = pine_validator.validate(code, version)

        return {
            "valid": result.valid,
            "version": result.version,
            "errors": [
                {
                    "line": err.line,
                    "column": err.column,
                    "severity": err.severity,
                    "message": err.message,
                    "code": err.code,
                    "suggestion": err.suggestion,
                }
                for err in result.errors
            ],
            "warnings": [
                {
                    "line": warn.line,
                    "column": warn.column,
                    "message": warn.message,
                    "suggestion": warn.suggestion,
                }
                for warn in result.warnings
            ],
            "info": [
                {
                    "message": info.message,
                }
                for info in result.info
            ],
            "summary": f"Validation {'passed' if result.valid else 'failed'} - "
            f"{len(result.errors)} errors, {len(result.warnings)} warnings",
        }
    except Exception as e:
        logger.error(f"Pine Script validation error: {e}")
        return format_error_response(str(e), suggestion="Check code syntax")


@mcp.tool()
def get_pine_documentation(function_name: str, version: int = 5) -> dict:
    """
    Get comprehensive documentation for a Pine Script function or topic.

    Provides:
    - Function signature and parameters
    - Return type
    - Description and usage
    - Code examples
    - Related functions
    - Links to official documentation

    Args:
        function_name: Function name (e.g., 'ta.sma', 'plot', 'indicator', 'map.new') or topic
        version: Pine Script version (default: 5, v6 supported)

    Returns:
        Detailed documentation with examples

    Example:
        get_pine_documentation("ta.sma")
    """
    try:
        # Try function documentation first
        docs = pine_docs.get_function_docs(function_name, version)

        if docs:
            return {
                "function": function_name,
                "version": version,
                "documentation": docs,
                "type": "function",
            }

        # Try topic documentation
        topic_docs = pine_docs.get_topic_docs(function_name)
        if topic_docs:
            return {
                "topic": function_name,
                "version": version,
                "documentation": topic_docs,
                "type": "topic",
            }

        # Search for similar functions
        search_results = pine_docs.search(function_name, version)
        if search_results:
            return {
                "query": function_name,
                "message": f"Function '{function_name}' not found. Did you mean:",
                "suggestions": search_results[:5],
                "type": "search_results",
            }

        return format_error_response(
            f"Documentation for '{function_name}' not found",
            suggestion="Check function name spelling or try searching",
        )

    except Exception as e:
        logger.error(f"Pine documentation error: {e}")
        return format_error_response(str(e))


@mcp.tool()
def test_pine_script(
    code: str,
    symbol: str = "BTCUSD",
    timeframe: str = "1D",
    bars: int = 100,
) -> dict:
    """
    Test Pine Script code in a safe sandbox environment.

    Validates syntax, simulates execution, and provides performance metrics.
    Note: Full backtesting requires TradingView platform integration.

    Args:
        code: Pine Script code to test
        symbol: Symbol to test with (default: BTCUSD)
        timeframe: Timeframe (default: 1D)
        bars: Number of bars to simulate (default: 100)

    Returns:
        Test results with validation, metrics, and any errors

    Example:
        test_pine_script("//@version=5\\nindicator(\\"Test\\")\\nplot(close)", "AAPL", "1h")
    """
    try:
        result = pine_sandbox.test(code, symbol, timeframe, bars)

        return {
            "success": result.success,
            "validation_passed": result.validation_passed,
            "output": result.output,
            "errors": result.errors,
            "warnings": result.warnings,
            "execution_time_ms": round(result.execution_time * 1000, 2),
            "metrics": result.metrics,
            "symbol": symbol,
            "timeframe": timeframe,
        }

    except Exception as e:
        logger.error(f"Pine sandbox error: {e}")
        return format_error_response(str(e), suggestion="Check code syntax")


@mcp.tool()
def explain_pine_error(error_code: str, error_message: str = "") -> dict:
    """
    Get detailed explanation for a Pine Script error.

    Provides:
    - Error description
    - Common causes
    - Multiple solutions
    - Code examples showing correct usage
    - Links to documentation

    Args:
        error_code: Error code (e.g., 'E001', 'E101')
        error_message: Original error message (optional)

    Returns:
        Comprehensive error explanation with solutions

    Example:
        explain_pine_error("E101", "Unknown function: sma")
    """
    try:
        explanation = pine_errors.explain(error_code, error_message)

        return {
            "error_code": explanation.error_code,
            "title": explanation.title,
            "description": explanation.description,
            "causes": explanation.causes,
            "solutions": explanation.solutions,
            "examples": explanation.examples,
            "documentation": explanation.related_docs,
            "formatted": pine_errors.format_explanation(explanation),
        }

    except Exception as e:
        logger.error(f"Pine error explanation error: {e}")
        return format_error_response(str(e))


@mcp.tool()
def detect_pine_version(code: str) -> dict:
    """
    Detect Pine Script version from code and analyze compatibility.

    Analyzes:
    - Version directive (//@version=5)
    - Syntax features (v5 namespaces, v4 var keyword, etc.)
    - Function usage patterns
    - Deprecated features
    - Compatibility issues

    Args:
        code: Pine Script code to analyze

    Returns:
        Detected version, confidence, issues, and upgrade suggestions

    Example:
        detect_pine_version("study(\\"Test\\")\\nmyMa = sma(close, 20)")
    """
    try:
        version_info = pine_version_detector.detect_version(code)

        return {
            "detected_version": version_info.version,
            "detection_source": version_info.detected_from,
            "confidence": f"{version_info.confidence:.0%}",
            "compatibility_issues": version_info.compatibility_issues,
            "deprecated_features": version_info.deprecated_features,
            "suggestions": version_info.suggestions,
            "summary": f"Detected Pine Script v{version_info.version} "
            f"(confidence: {version_info.confidence:.0%}, from: {version_info.detected_from})",
        }

    except Exception as e:
        logger.error(f"Pine version detection error: {e}")
        return format_error_response(str(e))


@mcp.tool()
def convert_pine_version(
    code: str, target_version: int, source_version: int = None
) -> dict:
    """
    Convert Pine Script code between versions.

    Automatically converts:
    - Function names (sma -> ta.sma in v5)
    - study() -> indicator()
    - security() -> request.security()
    - Namespace additions (ta., math., str.)

    Supports:
    - v3 -> v4 conversion
    - v4 -> v5 conversion
    - Automatic source version detection

    Args:
        code: Source Pine Script code
        target_version: Target version (1-6)
        source_version: Source version (auto-detected if not provided)

    Returns:
        Converted code, list of changes, and warnings

    Example:
        convert_pine_version("study(\\"Test\\")\\nplot(sma(close, 20))", 5)
    """
    try:
        converted_code, changes, warnings = pine_version_converter.convert(
            code, target_version, source_version
        )

        return {
            "success": True,
            "source_version": source_version or "auto-detected",
            "target_version": target_version,
            "converted_code": converted_code,
            "changes_made": changes,
            "warnings": warnings,
            "summary": f"Converted from v{source_version or '?'} to v{target_version} - "
            f"{len(changes)} changes made",
        }

    except Exception as e:
        logger.error(f"Pine version conversion error: {e}")
        return format_error_response(str(e), suggestion="Check source code syntax")


@mcp.tool()
def autocomplete_pine(code: str, cursor_position: int) -> dict:
    """
    Get intelligent autocomplete suggestions for Pine Script code.

    Provides:
    - Function completions with signatures
    - Parameter hints
    - Keyword completions
    - Built-in variable suggestions
    - Context-aware namespace completions (ta., math., str.)

    Args:
        code: Current Pine Script code
        cursor_position: Character position of cursor in code

    Returns:
        List of autocomplete suggestions with documentation

    Example:
        autocomplete_pine("indicator(\\"Test\\")\\nta.", 25)
    """
    try:
        suggestions = pine_autocomplete.get_completions(code, cursor_position)

        return {
            "suggestions": [
                {
                    "label": item.label,
                    "kind": item.kind,
                    "detail": item.detail,
                    "documentation": item.documentation,
                    "insert_text": item.insert_text,
                }
                for item in suggestions
            ],
            "count": len(suggestions),
            "cursor_position": cursor_position,
        }

    except Exception as e:
        logger.error(f"Pine autocomplete error: {e}")
        return format_error_response(str(e))


@mcp.tool()
def get_pine_template(template_type: str = "simple") -> dict:
    """
    Get Pine Script code templates for common indicator types.

    Available templates:
    - simple: Basic indicator with moving average
    - strategy: Trading strategy with entries/exits
    - overlay: Support/resistance overlay indicator

    Args:
        template_type: Template type ('simple', 'strategy', 'overlay')

    Returns:
        Ready-to-use Pine Script template code

    Example:
        get_pine_template("strategy")
    """
    try:
        template = pine_sandbox.get_test_template(template_type)

        return {
            "template_type": template_type,
            "code": template,
            "description": f"Pine Script {template_type} indicator template",
        }

    except Exception as e:
        logger.error(f"Pine template error: {e}")
        return format_error_response(
            str(e), suggestion="Use 'simple', 'strategy', or 'overlay'"
        )


# ===== MAIN =====


def main():
    """Run the TradingView MCP server."""
    import argparse

    parser = argparse.ArgumentParser(
        description="TradingView Trading Assistant MCP Server"
    )
    parser.add_argument(
        "transport",
        choices=["stdio", "streamable-http"],
        default="stdio",
        nargs="?",
        help="Transport method (default: stdio)",
    )
    args = parser.parse_args()

    logger.info(f"Starting server with transport: {args.transport}")

    if args.transport == "stdio":
        mcp.run()
    else:
        mcp.run(transport="streamable-http")


if __name__ == "__main__":
    main()