Coinglass

"""
Coinglass Extension - Crypto Derivatives Data Tools

Provides crypto derivatives market data including:
- Funding rates (V2 API)
- Open interest (V2 API)
- Long/Short ratios (V2 API)
- Liquidations (V2/V4 API)
- Futures market data (V4 API)
  - Supported coins and exchanges
  - Comprehensive market data for all coins
  - Pair-specific market metrics
  - OHLC price history

Environment Variables Required:
- COINGLASS_API_KEY: Coinglass API key

Usage:
    This extension is auto-loaded by the ExtensionLoader.
    Tools are available to agents configured with these tools in agents.yaml.
"""
import os
import sys
import logging
from typing import List

try:
    from core.tool import ToolRegistry
except Exception:
    ToolRegistry = None  # Standalone script usage

logger = logging.getLogger(__name__)

# Add local tools directory to path for imports
TOOLS_DIR = os.path.join(os.path.dirname(__file__), 'tools')
if TOOLS_DIR not in sys.path:
    sys.path.insert(0, TOOLS_DIR)


def register(api) -> List[str]:
    """
    Extension entry point - register all Coinglass tools.

    Args:
        api: ExtensionApi instance with registry and config

    Returns:
        List of registered tool names
    """
    registered = []

    try:
        from .coinglass import (
            FundingRateTool,
            LongShortRatioTool,
            GlobalAccountRatioTool,
            TopAccountRatioTool,
            TopPositionRatioTool,
            TakerBuySellExchangesTool,
            NetPositionTool,
            OpenInterestTool,
            LiquidationsTool,
            LiquidationAnalysisTool,
            CoinLiquidationHistoryTool,
            PairLiquidationHistoryTool,
            LiquidationCoinListTool,
            LiquidationOrdersTool,
            HyperliquidWhaleAlertsTool,
            HyperliquidWhalePositionsTool,
            HyperliquidPositionsByCoinTool,
            HyperliquidPositionDistributionTool,
            SupportedCoinsTool,
            SupportedExchangesTool,
            CoinsMarketDataTool,
            PairMarketDataTool,
            OHLCHistoryTool,
            TakerVolumeHistoryTool,
            AggregatedTakerVolumeTool,
            CumulativeVolumeDeltaTool,
            CoinNetflowTool,
            WhaleTransferTool,
            BTCETFFlowsTool,
            BTCETFPremiumDiscountTool,
            BTCETFHistoryTool,
            BTCETFListTool,
            HKBTCETFFlowsTool,
            ETHETFFlowsTool,
            ETHETFListTool,
            SOLETFFlowsTool,
            XRPETFFlowsTool,
        )
        # Register existing V2 tools
        api.register_tool(FundingRateTool())
        api.register_tool(LongShortRatioTool())

        # Register advanced long/short ratio tools
        api.register_tool(GlobalAccountRatioTool())
        api.register_tool(TopAccountRatioTool())
        api.register_tool(TopPositionRatioTool())
        api.register_tool(TakerBuySellExchangesTool())
        api.register_tool(NetPositionTool())

        api.register_tool(OpenInterestTool())
        api.register_tool(LiquidationsTool())
        api.register_tool(LiquidationAnalysisTool())

        # Register advanced liquidation tools
        api.register_tool(CoinLiquidationHistoryTool())
        api.register_tool(PairLiquidationHistoryTool())
        api.register_tool(LiquidationCoinListTool())
        api.register_tool(LiquidationOrdersTool())

        # Register Hyperliquid tools
        api.register_tool(HyperliquidWhaleAlertsTool())
        api.register_tool(HyperliquidWhalePositionsTool())
        api.register_tool(HyperliquidPositionsByCoinTool())
        api.register_tool(HyperliquidPositionDistributionTool())

        # Register new V4 futures market tools
        api.register_tool(SupportedCoinsTool())
        api.register_tool(SupportedExchangesTool())
        api.register_tool(CoinsMarketDataTool())
        api.register_tool(PairMarketDataTool())
        api.register_tool(OHLCHistoryTool())

        # Register Volume & Flow tools
        api.register_tool(TakerVolumeHistoryTool())
        api.register_tool(AggregatedTakerVolumeTool())
        api.register_tool(CumulativeVolumeDeltaTool())
        api.register_tool(CoinNetflowTool())

        # Register Whale Transfer tool
        api.register_tool(WhaleTransferTool())

        # Register Bitcoin ETF tools
        api.register_tool(BTCETFFlowsTool())
        api.register_tool(BTCETFPremiumDiscountTool())
        api.register_tool(BTCETFHistoryTool())
        api.register_tool(BTCETFListTool())
        api.register_tool(HKBTCETFFlowsTool())

        # Register Ethereum & Other ETF tools
        api.register_tool(ETHETFFlowsTool())
        api.register_tool(ETHETFListTool())
        api.register_tool(SOLETFFlowsTool())
        api.register_tool(XRPETFFlowsTool())

        registered.extend([
            "funding_rate",
            "long_short_ratio",
            "cg_global_account_ratio",
            "cg_top_account_ratio",
            "cg_top_position_ratio",
            "cg_taker_exchanges",
            "cg_net_position",
            "cg_open_interest",
            "cg_liquidations",
            "cg_liquidation_analysis",
            "cg_coin_liquidation_history",
            "cg_pair_liquidation_history",
            "cg_liquidation_coin_list",
            "cg_liquidation_orders",
            "cg_hyperliquid_whale_alerts",
            "cg_hyperliquid_whale_positions",
            "cg_hyperliquid_positions_by_coin",
            "cg_hyperliquid_position_distribution",
            "cg_supported_coins",
            "cg_supported_exchanges",
            "cg_coins_market_data",
            "cg_pair_market_data",
            "cg_ohlc_history",
            "cg_taker_volume_history",
            "cg_aggregated_taker_volume",
            "cg_cumulative_volume_delta",
            "cg_coin_netflow",
            "cg_whale_transfers",
            "cg_btc_etf_flows",
            "cg_btc_etf_premium_discount",
            "cg_btc_etf_history",
            "cg_btc_etf_list",
            "cg_hk_btc_etf_flows",
            "cg_eth_etf_flows",
            "cg_eth_etf_list",
            "cg_sol_etf_flows",
            "cg_xrp_etf_flows",
        ])
        logger.info("Registered Coinglass tools (37 tools)")
    except Exception as e:
        logger.warning(f"Failed to load Coinglass tools: {e}")

    return registered


# Extension metadata
EXTENSION_INFO = {
    "name": "coinglass",
    "version": "3.0.0",
    "description": "Coinglass crypto derivatives data tools - V4 API with advanced long/short ratios, liquidations, Hyperliquid whale tracking, volume & flow analysis, on-chain whale transfers, comprehensive ETF data (Bitcoin, Ethereum, Solana, XRP, Hong Kong), and futures market data (37 tools)",
    "tools": [
        "funding_rate",
        "long_short_ratio",
        "cg_global_account_ratio",
        "cg_top_account_ratio",
        "cg_top_position_ratio",
        "cg_taker_exchanges",
        "cg_net_position",
        "cg_open_interest",
        "cg_liquidations",
        "cg_liquidation_analysis",
        "cg_coin_liquidation_history",
        "cg_pair_liquidation_history",
        "cg_liquidation_coin_list",
        "cg_liquidation_orders",
        "cg_hyperliquid_whale_alerts",
        "cg_hyperliquid_whale_positions",
        "cg_hyperliquid_positions_by_coin",
        "cg_hyperliquid_position_distribution",
        "cg_supported_coins",
        "cg_supported_exchanges",
        "cg_coins_market_data",
        "cg_pair_market_data",
        "cg_ohlc_history",
        "cg_taker_volume_history",
        "cg_aggregated_taker_volume",
        "cg_cumulative_volume_delta",
        "cg_coin_netflow",
        "cg_whale_transfers",
        "cg_btc_etf_flows",
        "cg_btc_etf_premium_discount",
        "cg_btc_etf_history",
        "cg_btc_etf_list",
        "cg_hk_btc_etf_flows",
        "cg_eth_etf_flows",
        "cg_eth_etf_list",
        "cg_sol_etf_flows",
        "cg_xrp_etf_flows",
    ],
    "env_vars": [
        "COINGLASS_API_KEY",
    ],
}

"""
Coinglass Tools Module

Provides access to Coinglass cryptocurrency derivatives data including
funding rates and long/short account ratios across major exchanges.

Usage:
    from tools.coinglass import get_funding_rates, get_long_short_ratio

    # Get BTC funding rates
    rates = get_funding_rates("BTC")

    # Get BTC long/short ratio
    ratio = get_long_short_ratio("BTC", "h1")

Environment Variables Required:
- COINGLASS_API_KEY: Your Coinglass API key
"""

from .funding_rate import (
    get_funding_rates,
    get_symbol_funding_rate,
    get_funding_rate_by_exchange,
    analyze_funding_opportunity
)

from .long_short_ratio import (
    get_long_short_ratio,
    get_exchange_ratio,
    get_sentiment,
    compare_exchanges
)

__all__ = [
    # Funding Rate functions
    "get_funding_rates",
    "get_symbol_funding_rate",
    "get_funding_rate_by_exchange",
    "analyze_funding_opportunity",
    # Long/Short Ratio functions
    "get_long_short_ratio",
    "get_exchange_ratio",
    "get_sentiment",
    "compare_exchanges"
]

"""
Shared Coinglass API helper — eliminates return-None error pattern.

Every tools/*.py file repeated the same boilerplate:
  1. _get_api_key() → return None if missing
  2. proxied_get(url, headers, timeout) → return None on any error
  3. check response code → return None if not "0"

This module centralizes that into one `cg_request()` function that:
  - Raises typed exceptions instead of returning None
  - Provides actionable error messages for each failure mode
  - Distinguishes API key errors, rate limits, server errors, parse errors
"""

import os
import json
import requests

try:
    from core.http_client import proxied_get
except ImportError:
    # Fallback for testing outside platform — use plain requests
    def proxied_get(url, params=None, headers=None, timeout=30,
                    **kwargs):
        return requests.get(
            url, params=params, headers=headers, timeout=timeout,
            **kwargs
        )


# ── Coinglass-specific exceptions ───────────────────────────

class CoinglassError(Exception):
    """Base exception for all Coinglass API errors."""

    def __init__(self, message, code="UNKNOWN", suggestion=""):
        self.code = code
        self.suggestion = suggestion
        super().__init__(message)


class CoinglassAPIKeyError(CoinglassError):
    """API key missing or invalid."""
    pass


class CoinglassRateLimitError(CoinglassError):
    """Rate limited by Coinglass."""
    pass


class CoinglassServerError(CoinglassError):
    """Coinglass server returned 5xx."""
    pass


# ── API configuration ──────────────────────────────────────

BASE_URL_V2 = "https://open-api.coinglass.com/public/v2"
BASE_URL_V4 = "https://open-api-v4.coinglass.com"

HEADER_KEY_V2 = "coinglassSecret"
HEADER_KEY_V4 = "CG-API-KEY"


def _get_api_key():
    """Get API key from environment."""
    return os.getenv("COINGLASS_API_KEY")


def _suggestion_for_status(status):
    """Map HTTP status to actionable suggestion."""
    suggestions = {
        401: "API key invalid or expired. Check COINGLASS_API_KEY.",
        403: "Access denied. This endpoint may require a paid plan.",
        404: "Endpoint not found. The API version may have changed.",
        429: "Rate limited. Wait 60 seconds before retrying.",
        500: "Coinglass server error. Retry in 1-2 minutes.",
        502: "Coinglass gateway error. Retry in 1-2 minutes.",
        503: "Coinglass service unavailable. Retry in 1-2 minutes.",
    }
    return suggestions.get(status, f"HTTP {status} error.")


def cg_request(endpoint, params=None, version="v4", timeout=30):
    """
    Make a Coinglass API request with structured error handling.

    Args:
        endpoint: API path (e.g. "api/futures/supported-coins"
                  or "funding" for v2)
        params: Query parameters dict
        version: "v2" or "v4" (default "v4")
        timeout: Request timeout in seconds

    Returns:
        Parsed response data (the "data" field from Coinglass response).

    Raises:
        CoinglassAPIKeyError: API key missing or rejected
        CoinglassRateLimitError: 429 rate limit
        CoinglassServerError: 5xx server error
        CoinglassError: Any other API error
    """
    api_key = _get_api_key()
    if not api_key:
        raise CoinglassAPIKeyError(
            "COINGLASS_API_KEY not set in environment",
            code="NO_API_KEY",
            suggestion="Set COINGLASS_API_KEY in your .env file."
        )

    if version == "v2":
        base_url = BASE_URL_V2
        headers = {"accept": "application/json", HEADER_KEY_V2: api_key}
    else:
        base_url = BASE_URL_V4
        headers = {"accept": "application/json", HEADER_KEY_V4: api_key}

    url = f"{base_url}/{endpoint}"

    try:
        response = proxied_get(
            url, params=params, headers=headers, timeout=timeout
        )
        response.raise_for_status()
    except requests.exceptions.HTTPError as e:
        status = getattr(e.response, "status_code", None)
        suggestion = _suggestion_for_status(status)
        if status == 401:
            raise CoinglassAPIKeyError(
                f"HTTP 401 from Coinglass: {e}", code="HTTP_401",
                suggestion=suggestion
            ) from e
        if status == 429:
            raise CoinglassRateLimitError(
                f"Rate limited by Coinglass: {e}", code="HTTP_429",
                suggestion=suggestion
            ) from e
        if status and status >= 500:
            raise CoinglassServerError(
                f"Coinglass server error: {e}", code=f"HTTP_{status}",
                suggestion=suggestion
            ) from e
        raise CoinglassError(
            f"HTTP {status} from Coinglass: {e}",
            code=f"HTTP_{status}", suggestion=suggestion
        ) from e
    except requests.exceptions.ConnectionError as e:
        raise CoinglassError(
            f"Cannot connect to Coinglass: {e}",
            code="CONNECTION_ERROR",
            suggestion="Check network. Retry in 30 seconds."
        ) from e
    except requests.exceptions.Timeout as e:
        raise CoinglassError(
            f"Coinglass request timed out after {timeout}s",
            code="TIMEOUT",
            suggestion="Retry with a longer timeout or simpler query."
        ) from e
    except requests.exceptions.RequestException as e:
        raise CoinglassError(
            f"Request failed: {type(e).__name__}: {e}",
            code="REQUEST_ERROR",
            suggestion="Unexpected network error. Retry."
        ) from e

    # Parse JSON
    try:
        data = response.json()
    except (json.JSONDecodeError, ValueError) as e:
        raise CoinglassError(
            f"Invalid JSON from Coinglass: {e}",
            code="PARSE_ERROR",
            suggestion="API may be returning an error page. Try again later."
        ) from e

    # Check Coinglass response code
    if isinstance(data, dict):
        code = data.get("code")
        if code is not None and str(code) != "0":
            msg = data.get("msg", "Unknown API error")
            raise CoinglassError(
                f"Coinglass API error [{code}]: {msg}",
                code=f"API_{code}",
                suggestion="Check parameters. Some endpoints "
                           "require specific symbols or exchanges."
            )
        # Unwrap standard response envelope
        if "data" in data:
            return data["data"]

    return data

#!/usr/bin/env python3
"""
Coinglass Bitcoin ETF Module

Provides Bitcoin ETF data including flows, net assets,
premium/discount, history, and Hong Kong ETF data.
"""

import sys
import json
import argparse
from typing import Dict, Any, Optional, List

from ._api import cg_request


def get_btc_etf_flows() -> Optional[List[Dict[str, Any]]]:
    """Get Bitcoin ETF flow history (inflows/outflows by fund)."""
    return cg_request("api/etf/bitcoin/flow-history")


def get_btc_etf_net_assets() -> Optional[List[Dict[str, Any]]]:
    """Get Bitcoin ETF net assets history."""
    return cg_request("api/reference/bitcoin-etf-netassets-history")


def get_btc_etf_premium_discount() -> Optional[List[Dict[str, Any]]]:
    """Get Bitcoin ETF premium/discount rate history."""
    return cg_request("api/etf/bitcoin/premium-discount/history")


def get_btc_etf_history(
    etf_ticker: Optional[str] = None
) -> Optional[List[Dict[str, Any]]]:
    """
    Get comprehensive Bitcoin ETF history.

    Args:
        etf_ticker: Filter by specific ETF ticker (e.g. "GBTC", "IBIT").
    """
    params = {}
    if etf_ticker:
        params["ticker"] = etf_ticker
    return cg_request("api/etf/bitcoin/history", params=params or None)


def get_btc_etf_list() -> Optional[List[Dict[str, Any]]]:
    """Get list of all Bitcoin ETFs with details."""
    return cg_request("api/etf/bitcoin/list")


def get_hk_btc_etf_flows() -> Optional[List[Dict[str, Any]]]:
    """Get Hong Kong Bitcoin ETF flow history."""
    return cg_request("api/hk-etf/bitcoin/flow-history")


def main():
    """CLI entry point."""
    parser = argparse.ArgumentParser(description="Coinglass BTC ETF Tools")
    parser.add_argument("action", choices=[
        "flows", "assets", "premium", "history", "list", "hk-flows"
    ])
    parser.add_argument("--ticker", help="ETF ticker filter")
    parser.add_argument("--json", "-j", action="store_true")
    args = parser.parse_args()

    actions = {
        "flows": get_btc_etf_flows,
        "assets": get_btc_etf_net_assets,
        "premium": get_btc_etf_premium_discount,
        "history": lambda: get_btc_etf_history(args.ticker),
        "list": get_btc_etf_list,
        "hk-flows": get_hk_btc_etf_flows,
    }

    try:
        result = actions[args.action]()
        print(json.dumps(result, indent=2))
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Coinglass Funding Rate Module

Fetch funding rates across major cryptocurrency exchanges including
Binance, OKX, Bybit, KuCoin, MEXC, Bitfinex, Kraken, and more.
"""

import sys
import json
import argparse
from typing import Dict, Any, Optional, List

from ._api import cg_request

# Supported exchanges
EXCHANGES = [
    "Binance", "OKX", "Bybit", "KuCoin", "MEXC", "CoinEx",
    "Bitfinex", "Kraken", "dYdX", "Gate", "Bitmex"
]

SYMBOLS = [
    "BTC", "ETH", "SOL", "BNB", "XRP",
    "DOGE", "ADA", "AVAX", "LINK", "MATIC"
]


def get_funding_rates(
    symbol: Optional[str] = None
) -> Optional[Dict[str, Any]]:
    """
    Fetch funding rates across all exchanges.

    Args:
        symbol: Optional symbol filter (BTC, ETH, etc.).

    Returns:
        Dict with funding rate data. If symbol given, filtered to
        that symbol only.

    Raises:
        CoinglassError: On API failure.
    """
    data = cg_request("funding", version="v2")

    if symbol and isinstance(data, list):
        filtered = [
            d for d in data
            if d.get("symbol", "").upper() == symbol.upper()
        ]
        return filtered

    return data


def get_symbol_funding_rate(
    symbol: str,
    exchange: Optional[str] = None
) -> Optional[Dict[str, Any]]:
    """
    Get funding rate for a specific symbol and optionally a specific
    exchange.

    Args:
        symbol: Symbol to query (BTC, ETH, etc.)
        exchange: Optional exchange name (Binance, OKX, etc.)

    Returns:
        Dict with rate, rate_percent, next_funding_time, etc.
        None if no data found for the given symbol/exchange.
    """
    data = get_funding_rates(symbol)
    if not data:
        return None

    symbol_data = data[0] if isinstance(data, list) and data else None
    if not symbol_data:
        return None

    if exchange:
        for rate_info in symbol_data.get("uMarginList", []):
            if (rate_info.get("exchangeName", "").lower()
                    == exchange.lower()):
                rate = rate_info.get("rate", 0)
                predicted = rate_info.get("predictedRate")
                return {
                    "symbol": symbol.upper(),
                    "exchange": rate_info.get("exchangeName"),
                    "rate": rate,
                    "rate_percent": rate * 100,
                    "next_funding_time": rate_info.get(
                        "nextFundingTime"
                    ),
                    "funding_interval_hours": rate_info.get(
                        "fundingIntervalHours"
                    ),
                    "predicted_rate": predicted,
                    "predicted_rate_percent": (
                        predicted * 100 if predicted else None
                    ),
                }
        return None

    # Average across all exchanges
    rates = [
        r.get("rate", 0)
        for r in symbol_data.get("uMarginList", [])
        if r.get("rate") is not None
    ]
    if not rates:
        return None

    avg_rate = sum(rates) / len(rates)
    return {
        "symbol": symbol.upper(),
        "exchange": "average",
        "rate": avg_rate,
        "rate_percent": avg_rate * 100,
        "num_exchanges": len(rates),
        "exchanges_data": symbol_data.get("uMarginList", []),
    }


def get_funding_rate_by_exchange(
    exchange: str
) -> Optional[List[Dict[str, Any]]]:
    """
    Get funding rates for all symbols on a specific exchange.

    Args:
        exchange: Exchange name (Binance, OKX, etc.)

    Returns:
        List of {symbol, rate, rate_percent} dicts.
    """
    data = get_funding_rates()
    if not data:
        return None

    items = data if isinstance(data, list) else []
    results = []
    for symbol_data in items:
        sym = symbol_data.get("symbol", "")
        for rate_info in symbol_data.get("uMarginList", []):
            if (rate_info.get("exchangeName", "").lower()
                    == exchange.lower()):
                rate = rate_info.get("rate", 0)
                results.append({
                    "symbol": sym,
                    "rate": rate,
                    "rate_percent": rate * 100,
                })
    return results if results else None


def analyze_funding_opportunity(
    symbol: str,
    threshold: float = 0.01
) -> Optional[Dict[str, Any]]:
    """
    Analyze funding rate arbitrage opportunities.

    Finds exchanges with the highest and lowest rates and
    calculates the spread.

    Args:
        symbol: Symbol to analyze.
        threshold: Minimum absolute rate to flag (default 0.01 = 1%).

    Returns:
        Dict with highest, lowest, spread, and opportunity flag.
    """
    data = get_funding_rates(symbol)
    if not data:
        return None

    symbol_data = data[0] if isinstance(data, list) and data else None
    if not symbol_data:
        return None

    rates = []
    for r in symbol_data.get("uMarginList", []):
        rate = r.get("rate")
        if rate is not None:
            rates.append({
                "exchange": r.get("exchangeName"),
                "rate": rate,
                "rate_percent": rate * 100,
            })

    if not rates:
        return None

    rates.sort(key=lambda x: x["rate"])
    highest = rates[-1]
    lowest = rates[0]
    spread = highest["rate"] - lowest["rate"]

    return {
        "symbol": symbol.upper(),
        "highest": highest,
        "lowest": lowest,
        "spread": spread,
        "spread_percent": spread * 100,
        "opportunity": abs(spread) >= threshold,
        "num_exchanges": len(rates),
    }


def main():
    """CLI entry point."""
    parser = argparse.ArgumentParser(
        description="Coinglass Funding Rate Tools"
    )
    parser.add_argument("--symbol", "-s", help="Symbol (BTC, ETH)")
    parser.add_argument("--exchange", "-e", help="Exchange name")
    parser.add_argument("--analyze", "-a", action="store_true",
                        help="Analyze arbitrage opportunity")
    parser.add_argument("--all", action="store_true",
                        help="Show all funding rates")
    parser.add_argument("--json", "-j", action="store_true")
    args = parser.parse_args()

    try:
        if args.analyze and args.symbol:
            result = analyze_funding_opportunity(args.symbol)
        elif args.exchange:
            result = get_funding_rate_by_exchange(args.exchange)
        elif args.symbol:
            result = get_symbol_funding_rate(
                args.symbol, args.exchange
            )
        elif args.all:
            result = get_funding_rates()
        else:
            parser.print_help()
            return

        if result:
            print(json.dumps(result, indent=2))
        else:
            print("No data found", file=sys.stderr)
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Coinglass Futures Market Module

Provides futures market data: supported coins, exchanges, pairs,
coin-level market data, pair-level data, and OHLC price history.
"""

import sys
import json
import argparse
from typing import Dict, Any, Optional, List

from ._api import cg_request


def get_supported_coins() -> Optional[List[str]]:
    """Get list of coins supported by Coinglass futures data."""
    return cg_request("api/futures/supported-coins")


def get_supported_exchanges() -> Optional[List[Dict[str, Any]]]:
    """Get list of exchanges with supported trading pairs."""
    return cg_request("api/futures/supported-exchange-pairs")


def get_supported_pairs(
    exchange: str = "Binance"
) -> Optional[List[Dict[str, Any]]]:
    """
    Get supported trading pairs for a specific exchange.

    Args:
        exchange: Exchange name (Binance, OKX, Bybit, etc.)
    """
    return cg_request(
        "api/futures/supported-exchange-pairs",
        params={"exchange": exchange}
    )


def get_coins_data(
    symbol: Optional[str] = None
) -> Optional[List[Dict[str, Any]]]:
    """
    Get futures market data aggregated by coin.

    Args:
        symbol: Optional coin filter (BTC, ETH, etc.)
    """
    params = {}
    if symbol:
        params["symbol"] = symbol
    return cg_request("api/futures/coins-markets", params=params or None)


def get_pair_data(
    symbol: str = "BTC",
    exchange: Optional[str] = None
) -> Optional[List[Dict[str, Any]]]:
    """
    Get futures market data by trading pair.

    Args:
        symbol: Coin symbol (BTC, ETH, etc.)
        exchange: Optional exchange filter.
    """
    params = {"symbol": symbol}
    if exchange:
        params["exchange"] = exchange
    return cg_request("api/futures/pairs-markets", params=params)


def get_ohlc_history(
    symbol: str = "BTC",
    interval: str = "h4",
    exchange: Optional[str] = None
) -> Optional[List[Dict[str, Any]]]:
    """
    Get OHLC price history for a futures pair.

    Args:
        symbol: Coin symbol.
        interval: Time interval (m1, m5, m15, h1, h4, h12, h24).
        exchange: Optional exchange filter.
    """
    params = {"symbol": symbol, "interval": interval}
    if exchange:
        params["exchange"] = exchange
    return cg_request("api/futures/price/history", params=params)


def main():
    """CLI entry point."""
    parser = argparse.ArgumentParser(
        description="Coinglass Futures Market Tools"
    )
    parser.add_argument("action", choices=[
        "coins", "exchanges", "pairs", "market", "ohlc"
    ])
    parser.add_argument("--symbol", "-s", default="BTC")
    parser.add_argument("--exchange", "-e", default=None)
    parser.add_argument("--interval", "-i", default="h4")
    args = parser.parse_args()

    actions = {
        "coins": get_supported_coins,
        "exchanges": get_supported_exchanges,
        "pairs": lambda: get_supported_pairs(args.exchange or "Binance"),
        "market": lambda: get_coins_data(args.symbol),
        "ohlc": lambda: get_ohlc_history(
            args.symbol, args.interval, args.exchange
        ),
    }

    try:
        result = actions[args.action]()
        print(json.dumps(result, indent=2))
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Coinglass Hyperliquid Module

Provides Hyperliquid-specific data: whale alerts, whale positions,
positions by coin, user positions, and position distribution.
"""

import sys
import json
import argparse
from typing import Dict, Any, Optional, List

from ._api import cg_request


def get_whale_alerts() -> Optional[List[Dict[str, Any]]]:
    """Get Hyperliquid whale position alerts (large position changes)."""
    return cg_request("api/hyperliquid/whale-alert")


def get_whale_positions() -> Optional[List[Dict[str, Any]]]:
    """Get current Hyperliquid whale positions."""
    return cg_request("api/hyperliquid/whale-position")


def get_positions_by_coin(
    symbol: str = "BTC"
) -> Optional[List[Dict[str, Any]]]:
    """
    Get Hyperliquid positions aggregated by coin.

    Args:
        symbol: Coin symbol (BTC, ETH, SOL, etc.)
    """
    return cg_request(
        "api/hyperliquid/position",
        params={"symbol": symbol}
    )


def get_user_positions(
    address: str = ""
) -> Optional[List[Dict[str, Any]]]:
    """
    Get positions for a specific Hyperliquid wallet.

    Args:
        address: Wallet address to query.
    """
    params = {}
    if address:
        params["address"] = address
    return cg_request(
        "api/hyperliquid/user-position",
        params=params or None
    )


def get_position_distribution(
    symbol: str = "BTC"
) -> Optional[Dict[str, Any]]:
    """
    Get wallet position distribution for a coin on Hyperliquid.

    Args:
        symbol: Coin symbol (BTC, ETH, SOL, etc.)
    """
    return cg_request(
        "api/hyperliquid/wallet/position-distribution",
        params={"symbol": symbol}
    )


def main():
    """CLI entry point."""
    parser = argparse.ArgumentParser(
        description="Coinglass Hyperliquid Tools"
    )
    parser.add_argument("action", choices=[
        "alerts", "whales", "coin", "user", "distribution"
    ])
    parser.add_argument("--symbol", "-s", default="BTC")
    parser.add_argument("--address", "-a", default="")
    args = parser.parse_args()

    actions = {
        "alerts": get_whale_alerts,
        "whales": get_whale_positions,
        "coin": lambda: get_positions_by_coin(args.symbol),
        "user": lambda: get_user_positions(args.address),
        "distribution": lambda: get_position_distribution(args.symbol),
    }

    try:
        result = actions[args.action]()
        print(json.dumps(result, indent=2))
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Coinglass Liquidations Advanced Module

Provides detailed liquidation data: coin/pair history,
coin list aggregation, liquidation orders, and heatmap data.
"""

import sys
import json
import argparse
from typing import Dict, Any, Optional, List

from ._api import cg_request


def get_coin_liquidation_history(
    symbol: str = "BTC",
    interval: str = "h4"
) -> Optional[List[Dict[str, Any]]]:
    """
    Get aggregated liquidation history for a coin.

    Args:
        symbol: Coin symbol (BTC, ETH, SOL, etc.)
        interval: Time interval (h1, h4, h12, h24).
    """
    return cg_request(
        "api/futures/liquidation/aggregated-history",
        params={"symbol": symbol, "interval": interval}
    )


def get_pair_liquidation_history(
    symbol: str = "BTC",
    exchange: str = "Binance",
    interval: str = "h4"
) -> Optional[List[Dict[str, Any]]]:
    """
    Get liquidation history for a specific trading pair.

    Args:
        symbol: Coin symbol.
        exchange: Exchange name.
        interval: Time interval (h1, h4, h12, h24).
    """
    return cg_request(
        "api/futures/liquidation/history",
        params={
            "symbol": symbol,
            "exchange": exchange,
            "interval": interval,
        }
    )


def get_liquidation_coin_list(
    symbol: Optional[str] = None
) -> Optional[List[Dict[str, Any]]]:
    """
    Get liquidation summary aggregated by coin.

    Args:
        symbol: Optional coin filter.
    """
    params = {}
    if symbol:
        params["symbol"] = symbol
    return cg_request(
        "api/futures/liquidation/coin-list",
        params=params or None
    )


def get_liquidation_orders(
    symbol: str = "BTC",
    exchange: Optional[str] = None
) -> Optional[List[Dict[str, Any]]]:
    """
    Get recent large liquidation orders.

    Args:
        symbol: Coin symbol.
        exchange: Optional exchange filter.
    """
    params = {"symbol": symbol}
    if exchange:
        params["exchange"] = exchange
    return cg_request("api/futures/liquidation/order", params=params)


def get_liquidation_heatmap(
    symbol: str = "BTC",
    exchange: str = "Binance"
) -> Optional[Dict[str, Any]]:
    """
    Get liquidation heatmap data (price levels with liquidation density).

    Args:
        symbol: Coin symbol.
        exchange: Exchange name.
    """
    return cg_request(
        "api/futures/liquidation/heatmap/model1",
        params={"symbol": symbol, "exchange": exchange}
    )


def main():
    """CLI entry point."""
    parser = argparse.ArgumentParser(
        description="Coinglass Liquidation Advanced Tools"
    )
    parser.add_argument("action", choices=[
        "coin-history", "pair-history", "coin-list",
        "orders", "heatmap"
    ])
    parser.add_argument("--symbol", "-s", default="BTC")
    parser.add_argument("--exchange", "-e", default="Binance")
    parser.add_argument("--interval", "-i", default="h4")
    args = parser.parse_args()

    actions = {
        "coin-history": lambda: get_coin_liquidation_history(
            args.symbol, args.interval
        ),
        "pair-history": lambda: get_pair_liquidation_history(
            args.symbol, args.exchange, args.interval
        ),
        "coin-list": lambda: get_liquidation_coin_list(args.symbol),
        "orders": lambda: get_liquidation_orders(
            args.symbol, args.exchange
        ),
        "heatmap": lambda: get_liquidation_heatmap(
            args.symbol, args.exchange
        ),
    }

    try:
        result = actions[args.action]()
        print(json.dumps(result, indent=2))
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Coinglass Liquidations Module

Provides liquidation data across exchanges: individual liquidations
and aggregated (long/short) liquidation summaries.
"""

import sys
import json
import argparse
from typing import Dict, Any, Optional

from ._api import cg_request


def get_liquidations(
    symbol: str = "BTC",
    time_type: str = "h24"
) -> Optional[Dict[str, Any]]:
    """
    Get liquidation data across exchanges.

    Args:
        symbol: Coin symbol (BTC, ETH, SOL, etc.)
        time_type: Time window (h1, h4, h12, h24).

    Returns:
        Dict with liquidation data: long/short amounts, counts.

    Raises:
        CoinglassError: On API failure.
    """
    # Map h-format to API format
    range_map = {"h1": "1h", "h4": "4h", "h12": "12h", "h24": "24h"}
    v4_range = range_map.get(time_type, time_type)
    params = {"symbol": symbol.upper(), "range": v4_range}
    return cg_request(
        "api/futures/liquidation/exchange-list", params=params
    )


def get_liquidation_aggregated(
    symbol: str = "BTC",
    time_type: str = "h24"
) -> Optional[Dict[str, Any]]:
    """
    Get aggregated liquidation summary (total longs vs shorts).

    Args:
        symbol: Coin symbol.
        interval: Time interval.

    Returns:
        Dict with total long/short liquidations and ratio.
    """
    data = get_liquidations(symbol, time_type)
    if not data:
        return None

    # Aggregate across exchanges
    entries = data if isinstance(data, list) else [data]
    total_long = 0
    total_short = 0
    for entry in entries:
        total_long += entry.get("longLiquidationUsd", 0) or 0
        total_short += entry.get("shortLiquidationUsd", 0) or 0

    total = total_long + total_short
    return {
        "symbol": symbol.upper(),
        "interval": time_type,
        "total_long_usd": total_long,
        "total_short_usd": total_short,
        "total_usd": total,
        "long_ratio": total_long / total if total else 0,
        "short_ratio": total_short / total if total else 0,
        "dominant": "long" if total_long > total_short else "short",
    }


def main():
    """CLI entry point."""
    parser = argparse.ArgumentParser(
        description="Coinglass Liquidation Tools"
    )
    parser.add_argument("--symbol", "-s", default="BTC")
    parser.add_argument("--interval", "-i", default="h4")
    parser.add_argument("--exchange", "-e", default=None)
    parser.add_argument("--aggregated", "-a", action="store_true",
                        help="Show aggregated summary")
    args = parser.parse_args()

    try:
        if args.aggregated:
            result = get_liquidation_aggregated(
                args.symbol, args.interval
            )
        else:
            result = get_liquidations(
                args.symbol, args.interval
            )

        if result:
            print(json.dumps(result, indent=2))
        else:
            print("No data found", file=sys.stderr)
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Coinglass Long/Short Advanced Module

Provides advanced long/short data: global account ratio,
top trader ratios, taker buy/sell by exchange, net positions.
"""

import sys
import json
import argparse
from typing import Dict, Any, Optional, List

from ._api import cg_request


def _to_pair(symbol: str) -> str:
    """Convert symbol to trading pair (BTC → BTCUSDT)."""
    s = symbol.upper()
    return s if s.endswith("USDT") else f"{s}USDT"


# Exchange name normalization for consistent API calls
_EXCHANGE_ALIASES = {
    "binance": "Binance", "okx": "OKX", "bybit": "Bybit",
    "bitget": "Bitget", "gate": "Gate", "bitmex": "Bitmex",
    "dydx": "dYdX", "kraken": "Kraken", "coinex": "CoinEx",
}


def _normalize_exchange(name: str) -> str:
    """Normalize exchange name to Coinglass format."""
    return _EXCHANGE_ALIASES.get(name.lower(), name)


def get_global_account_ratio(
    symbol: str = "BTC",
    exchange: str = "Binance",
    interval: str = "1h"
) -> Optional[List[Dict[str, Any]]]:
    """
    Get global long/short account ratio history.

    Args:
        symbol: Coin symbol.
        interval: Time interval (h1, h4, h12, h24).
        exchange: Optional exchange filter.
    """
    params = {
        "symbol": _to_pair(symbol),
        "exchange": _normalize_exchange(exchange),
        "interval": interval,
    }
    return cg_request(
        "api/futures/global-long-short-account-ratio/history",
        params=params
    )


def get_top_account_ratio(
    symbol: str = "BTC",
    exchange: str = "Binance",
    interval: str = "1h"
) -> Optional[List[Dict[str, Any]]]:
    """
    Get top trader long/short account ratio history.

    Args:
        symbol: Coin symbol.
        interval: Time interval.
        exchange: Optional exchange filter.
    """
    params = {
        "symbol": _to_pair(symbol),
        "exchange": _normalize_exchange(exchange),
        "interval": interval,
    }
    return cg_request(
        "api/futures/top-long-short-account-ratio/history",
        params=params
    )


def get_top_position_ratio(
    symbol: str = "BTC",
    exchange: str = "Binance",
    interval: str = "1h"
) -> Optional[List[Dict[str, Any]]]:
    """
    Get top trader long/short position ratio history.

    Args:
        symbol: Coin symbol.
        interval: Time interval.
        exchange: Optional exchange filter.
    """
    params = {
        "symbol": _to_pair(symbol),
        "exchange": _normalize_exchange(exchange),
        "interval": interval,
    }
    return cg_request(
        "api/futures/top-long-short-position-ratio/history",
        params=params
    )


def get_taker_buysell_exchanges(
    symbol: str = "BTC",
    range_type: str = "4h"
) -> Optional[List[Dict[str, Any]]]:
    """
    Get taker buy/sell volume by exchange.

    Args:
        symbol: Coin symbol.
    """
    return cg_request(
        "api/futures/taker-buy-sell-volume/exchange-list",
        params={"symbol": symbol, "range": range_type}
    )


def get_net_position(
    symbol: str = "BTC",
    exchange: str = "Binance",
    interval: str = "1h"
) -> Optional[List[Dict[str, Any]]]:
    """
    Get net position history (v1).

    Args:
        symbol: Coin symbol.
        interval: Time interval.
        exchange: Optional exchange filter.
    """
    params = {
        "symbol": _to_pair(symbol),
        "exchange": _normalize_exchange(exchange),
        "interval": interval,
    }
    return cg_request(
        "api/futures/net-position/history", params=params
    )


def get_net_position_v2(
    symbol: str = "BTC",
    exchange: str = "Binance",
    interval: str = "1h"
) -> Optional[List[Dict[str, Any]]]:
    """
    Get net position history (v2 — more exchanges).

    Args:
        symbol: Coin symbol.
        interval: Time interval.
        exchange: Optional exchange filter.
    """
    params = {
        "symbol": _to_pair(symbol),
        "exchange": _normalize_exchange(exchange),
        "interval": interval,
    }
    return cg_request(
        "api/futures/v2/net-position/history", params=params
    )


def main():
    """CLI entry point."""
    parser = argparse.ArgumentParser(
        description="Coinglass Long/Short Advanced Tools"
    )
    parser.add_argument("action", choices=[
        "global", "top-account", "top-position",
        "taker", "net", "net-v2"
    ])
    parser.add_argument("--symbol", "-s", default="BTC")
    parser.add_argument("--exchange", "-e", default=None)
    parser.add_argument("--interval", "-i", default="h4")
    args = parser.parse_args()

    actions = {
        "global": lambda: get_global_account_ratio(
            args.symbol, args.exchange or "Binance", args.interval
        ),
        "top-account": lambda: get_top_account_ratio(
            args.symbol, args.exchange or "Binance", args.interval
        ),
        "top-position": lambda: get_top_position_ratio(
            args.symbol, args.exchange or "Binance", args.interval
        ),
        "taker": lambda: get_taker_buysell_exchanges(args.symbol),
        "net": lambda: get_net_position(
            args.symbol, args.exchange or "Binance", args.interval
        ),
        "net-v2": lambda: get_net_position_v2(
            args.symbol, args.exchange or "Binance", args.interval
        ),
    }

    try:
        result = actions[args.action]()
        print(json.dumps(result, indent=2))
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Coinglass Long/Short Ratio Module

Provides aggregate long/short position ratio data across exchanges.
Ratio > 1 means more longs; < 1 means more shorts.
"""

import sys
import json
import argparse
from typing import Dict, Any, Optional, List

from ._api import cg_request


def get_long_short_ratio(
    symbol: str = "BTC",
    interval: str = "h4"
) -> Optional[Dict[str, Any]]:
    """
    Get long/short ratio data across exchanges.

    Args:
        symbol: Coin symbol (BTC, ETH, etc.)
        interval: Time interval (h1, h4, h12, h24).

    Returns:
        Dict with exchange-level long/short data.

    Raises:
        CoinglassError: On API failure.
    """
    data = cg_request(
        "long_short", params={"symbol": symbol, "time_type": interval},
        version="v2"
    )

    # v2 returns wrapped data — may need to handle both list and dict
    return data


def get_exchange_ratio(
    symbol: str = "BTC",
    exchange: str = "Binance",
    interval: str = "h4"
) -> Optional[Dict[str, Any]]:
    """
    Get long/short ratio for a specific exchange.

    Args:
        symbol: Coin symbol.
        exchange: Exchange name.
        interval: Time interval.

    Returns:
        Dict with ratio, longPercent, shortPercent for the exchange.
        None if exchange not found in data.
    """
    data = get_long_short_ratio(symbol, interval)
    if not data:
        return None

    # Data may be a list of exchange entries
    entries = data if isinstance(data, list) else [data]
    for entry in entries:
        if entry.get("exchangeName", "").lower() == exchange.lower():
            long_pct = entry.get("longRate", 0)
            short_pct = entry.get("shortRate", 0)
            ratio = long_pct / short_pct if short_pct else 0
            return {
                "symbol": symbol.upper(),
                "exchange": entry.get("exchangeName"),
                "long_percent": long_pct * 100,
                "short_percent": short_pct * 100,
                "ratio": ratio,
            }
    return None


def get_sentiment(
    symbol: str = "BTC",
    interval: str = "h4"
) -> Optional[Dict[str, Any]]:
    """
    Get aggregated market sentiment from long/short ratios.

    Args:
        symbol: Coin symbol.
        interval: Time interval.

    Returns:
        Dict with avg ratio, sentiment label, and per-exchange breakdown.
    """
    data = get_long_short_ratio(symbol, interval)
    if not data:
        return None

    entries = data if isinstance(data, list) else [data]
    ratios = []
    for entry in entries:
        long_r = entry.get("longRate", 0)
        short_r = entry.get("shortRate", 0)
        if short_r:
            ratios.append({
                "exchange": entry.get("exchangeName", ""),
                "ratio": long_r / short_r,
            })

    if not ratios:
        return None

    avg_ratio = sum(r["ratio"] for r in ratios) / len(ratios)
    if avg_ratio > 1.2:
        sentiment = "Very Bullish"
    elif avg_ratio > 1.0:
        sentiment = "Bullish"
    elif avg_ratio > 0.8:
        sentiment = "Bearish"
    else:
        sentiment = "Very Bearish"

    return {
        "symbol": symbol.upper(),
        "avg_ratio": avg_ratio,
        "sentiment": sentiment,
        "exchanges": ratios,
    }


def compare_exchanges(
    symbol: str = "BTC",
    interval: str = "h4"
) -> Optional[List[Dict[str, Any]]]:
    """
    Compare long/short ratios across all exchanges.

    Returns:
        Sorted list of exchanges by ratio (most bullish first).
    """
    data = get_long_short_ratio(symbol, interval)
    if not data:
        return None

    entries = data if isinstance(data, list) else [data]
    results = []
    for entry in entries:
        long_r = entry.get("longRate", 0)
        short_r = entry.get("shortRate", 0)
        ratio = long_r / short_r if short_r else 0
        results.append({
            "exchange": entry.get("exchangeName", ""),
            "ratio": round(ratio, 4),
            "long_percent": round(long_r * 100, 2),
            "short_percent": round(short_r * 100, 2),
        })

    results.sort(key=lambda x: x["ratio"], reverse=True)
    return results if results else None


def main():
    """CLI entry point."""
    parser = argparse.ArgumentParser(
        description="Coinglass Long/Short Ratio Tools"
    )
    parser.add_argument("--symbol", "-s", default="BTC")
    parser.add_argument("--exchange", "-e", default=None)
    parser.add_argument("--interval", "-i", default="h4")
    parser.add_argument("--sentiment", action="store_true",
                        help="Show sentiment analysis")
    parser.add_argument("--compare", action="store_true",
                        help="Compare across exchanges")
    args = parser.parse_args()

    try:
        if args.sentiment:
            result = get_sentiment(args.symbol, args.interval)
        elif args.compare:
            result = compare_exchanges(args.symbol, args.interval)
        elif args.exchange:
            result = get_exchange_ratio(
                args.symbol, args.exchange, args.interval
            )
        else:
            result = get_long_short_ratio(args.symbol, args.interval)

        if result:
            print(json.dumps(result, indent=2))
        else:
            print("No data found", file=sys.stderr)
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Coinglass Open Interest Module

Fetch aggregate open interest data across major cryptocurrency exchanges.
"""

import sys
import json
import argparse
from typing import Dict, Any, Optional, List

from ._api import cg_request

# MCP Tool Schema
MCP_OPEN_INTEREST_SCHEMA = {
    "name": "cg_open_interest",
    "title": "Coinglass Open Interest",
    "description": (
        "Get aggregate open interest across exchanges for a symbol."
    ),
    "inputSchema": {
        "type": "object",
        "properties": {
            "symbol": {
                "type": "string",
                "description": "Symbol (BTC, ETH, SOL, etc.)"
            },
            "interval": {
                "type": "string",
                "description": "Time interval: 0 (all), h1, h4, h12, h24",
                "default": "0",
                "enum": ["0", "h1", "h4", "h12", "h24"]
            }
        },
        "required": ["symbol"],
        "additionalProperties": False
    }
}


def get_open_interest(
    symbol: str = "BTC",
    interval: str = "0"
) -> Optional[Dict[str, Any]]:
    """
    Get aggregate open interest across exchanges for a symbol.

    Uses v2 API for basic OI data.

    Args:
        symbol: Coin symbol (BTC, ETH, SOL, etc.)
        interval: Time interval (0=all, h1, h4, h12, h24).

    Returns:
        Dict with open interest data by exchange.

    Raises:
        CoinglassError: On API failure.
    """
    return cg_request(
        "open_interest",
        params={"symbol": symbol, "interval": interval},
        version="v2"
    )


def get_open_interest_history(
    symbol: str = "BTC",
    interval: str = "h4",
    exchange: Optional[str] = None
) -> Optional[List[Dict[str, Any]]]:
    """
    Get open interest history (aggregated across exchanges).

    Uses v4 API for historical data.

    Args:
        symbol: Coin symbol.
        interval: Time interval (h1, h4, h12, h24).
        exchange: Optional exchange filter.

    Returns:
        List of historical OI data points.
    """
    params = {"symbol": symbol, "interval": interval}
    if exchange:
        params["exchange"] = exchange
    return cg_request(
        "api/futures/open-interest/aggregated-history",
        params=params
    )


def main():
    """CLI entry point."""
    parser = argparse.ArgumentParser(
        description="Coinglass Open Interest Tools"
    )
    parser.add_argument("--symbol", "-s", default="BTC")
    parser.add_argument("--interval", "-i", default="h4")
    parser.add_argument("--exchange", "-e", default=None)
    parser.add_argument("--history", action="store_true",
                        help="Show OI history")
    parser.add_argument("--json", "-j", action="store_true")
    args = parser.parse_args()

    try:
        if args.history:
            result = get_open_interest_history(
                args.symbol, args.interval, args.exchange
            )
        else:
            result = get_open_interest(args.symbol, args.interval)

        if result:
            print(json.dumps(result, indent=2))
        else:
            print("No data found", file=sys.stderr)
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Coinglass ETF Module (Non-Bitcoin)

Provides ETF data for Ethereum, Solana, XRP, and HK listings.
"""

import sys
import json
import argparse
from typing import Dict, Any, Optional, List

from ._api import cg_request


def get_eth_etf_flows() -> Optional[List[Dict[str, Any]]]:
    """Get Ethereum ETF flow history (inflows/outflows)."""
    return cg_request("api/etf/ethereum/flow-history")


def get_eth_etf_list() -> Optional[List[Dict[str, Any]]]:
    """Get list of all Ethereum ETFs with details."""
    return cg_request("api/etf/ethereum/list")


def get_eth_etf_premium_discount() -> Optional[List[Dict[str, Any]]]:
    """Get Ethereum ETF premium/discount history."""
    return cg_request("api/etf/ethereum/premium-discount/history")


def get_sol_etf_flows() -> Optional[List[Dict[str, Any]]]:
    """Get Solana ETF flow history."""
    return cg_request("api/etf/solana/flow-history")


def get_sol_etf_list() -> Optional[List[Dict[str, Any]]]:
    """Get list of all Solana ETFs."""
    return cg_request("api/etf/solana/list")


def get_xrp_etf_flows() -> Optional[List[Dict[str, Any]]]:
    """Get XRP ETF flow history."""
    return cg_request("api/etf/xrp/flow-history")


def get_xrp_etf_list() -> Optional[List[Dict[str, Any]]]:
    """Get list of all XRP ETFs."""
    return cg_request("api/etf/xrp/list")


def get_hk_eth_etf_flows() -> Optional[List[Dict[str, Any]]]:
    """Get Hong Kong Ethereum ETF flows."""
    return cg_request("api/hk-etf/ethereum/flow-history")


def main():
    """CLI entry point."""
    parser = argparse.ArgumentParser(description="Coinglass ETF Tools")
    parser.add_argument("action", choices=[
        "eth-flows", "eth-list", "sol-flows", "xrp-flows"
    ])
    parser.add_argument("--json", "-j", action="store_true")
    args = parser.parse_args()

    actions = {
        "eth-flows": get_eth_etf_flows,
        "eth-list": get_eth_etf_list,
        "sol-flows": get_sol_etf_flows,
        "xrp-flows": get_xrp_etf_flows,
    }

    try:
        result = actions[args.action]()
        print(json.dumps(result, indent=2))
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Coinglass Volume & Flow Module

Provides taker volume history, aggregated taker volume,
cumulative volume delta (CVD), coin netflow, and vol-weighted OHLC.
"""

import sys
import json
import argparse
from typing import Dict, Any, Optional, List

from ._api import cg_request


def _to_pair(symbol: str) -> str:
    """Convert symbol to trading pair (BTC → BTCUSDT)."""
    s = symbol.upper()
    return s if s.endswith("USDT") else f"{s}USDT"


def get_taker_volume_history(
    symbol: str = "BTC",
    exchange: str = "Binance",
    interval: str = "1h"
) -> Optional[List[Dict[str, Any]]]:
    """
    Get taker buy/sell volume history for a coin.

    Args:
        symbol: Coin symbol.
        interval: Time interval (h1, h4, h12, h24).
        exchange: Optional exchange filter.
    """
    return cg_request(
        "api/futures/v2/taker-buy-sell-volume/history",
        params={
            "symbol": _to_pair(symbol),
            "exchange": exchange,
            "interval": interval,
        }
    )


def get_aggregated_taker_volume(
    symbol: str = "BTC",
    interval: str = "h4"
) -> Optional[List[Dict[str, Any]]]:
    """
    Get aggregated taker buy/sell volume history across exchanges.

    Args:
        symbol: Coin symbol.
        interval: Time interval.
    """
    return cg_request(
        "api/futures/aggregated-taker-buy-sell-volume/history",
        params={"symbol": _to_pair(symbol), "interval": interval}
    )


def get_cumulative_volume_delta(
    symbol: str = "BTC",
    exchange: str = "Binance",
    interval: str = "1h"
) -> Optional[List[Dict[str, Any]]]:
    """
    Get cumulative volume delta (CVD) history.

    CVD tracks net buying vs selling pressure over time.

    Args:
        symbol: Coin symbol.
        interval: Time interval.
        exchange: Optional exchange filter.
    """
    return cg_request(
        "api/futures/cvd/history",
        params={
            "symbol": _to_pair(symbol),
            "exchange": exchange,
            "interval": interval,
        }
    )


def get_coin_netflow(
    symbol: Optional[str] = None
) -> Optional[List[Dict[str, Any]]]:
    """
    Get exchange netflow data by coin.

    Shows net inflow/outflow of coins across exchanges.

    Args:
        symbol: Optional coin filter.
    """
    params = {}
    if symbol:
        params["symbol"] = symbol
    return cg_request("api/futures/netflow-list", params=params or None)


def get_volume_ohlc_history(
    symbol: str = "BTC",
    exchange: str = "Binance",
    interval: str = "1h"
) -> Optional[List[Dict[str, Any]]]:
    """
    Get volume-weighted OHLC history.

    Args:
        symbol: Coin symbol.
        interval: Time interval.
        exchange: Optional exchange filter.
    """
    return cg_request(
        "api/futures/vol-weight-ohlc-history",
        params={
            "symbol": _to_pair(symbol),
            "exchange": exchange,
            "interval": interval,
        }
    )


def main():
    """CLI entry point."""
    parser = argparse.ArgumentParser(
        description="Coinglass Volume & Flow Tools"
    )
    parser.add_argument("action", choices=[
        "taker", "aggregated", "cvd", "netflow", "ohlc"
    ])
    parser.add_argument("--symbol", "-s", default="BTC")
    parser.add_argument("--exchange", "-e", default=None)
    parser.add_argument("--interval", "-i", default="h4")
    args = parser.parse_args()

    actions = {
        "taker": lambda: get_taker_volume_history(
            args.symbol, args.exchange or "Binance", args.interval
        ),
        "aggregated": lambda: get_aggregated_taker_volume(
            args.symbol, args.interval
        ),
        "cvd": lambda: get_cumulative_volume_delta(
            args.symbol, args.exchange or "Binance", args.interval
        ),
        "netflow": lambda: get_coin_netflow(args.symbol),
        "ohlc": lambda: get_volume_ohlc_history(
            args.symbol, args.exchange or "Binance", args.interval
        ),
    }

    try:
        result = actions[args.action]()
        print(json.dumps(result, indent=2))
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Coinglass Whale Transfer Module

Provides on-chain whale transfer data including:
- Large transfers (minimum $10M) across major blockchains
- Bitcoin, Ethereum, Tron, Ripple, Dogecoin, Litecoin, Polygon,
  Algorand, Bitcoin Cash, Solana
- Past 6 months of data
"""

import sys
import json
import argparse
from typing import Dict, Any, Optional

from ._api import cg_request


def get_whale_transfers() -> Optional[Dict[str, Any]]:
    """
    Get large on-chain transfers (minimum $10M) across major blockchains.

    Returns:
        List of whale transfers with transaction hash, asset, amount,
        exchange, transfer type (1=inflow, 2=outflow, 3=internal),
        addresses, and timestamp.

    Raises:
        CoinglassError: On API failure with actionable error message.
    """
    return cg_request("api/chain/v2/whale-transfer")


def main():
    """CLI entry point."""
    parser = argparse.ArgumentParser(
        description="Coinglass Whale Transfer Tool"
    )
    parser.add_argument("--json", "-j", action="store_true",
                        help="Output as JSON")
    parser.parse_args()

    try:
        result = get_whale_transfers()
        print(json.dumps(result, indent=2))
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()