Tracking Crypto Prices

# Tracking Crypto Prices - Configuration
# Version: 2.0.0
# Author: Jeremy Longshore <jeremy@intentsolutions.io>

# =============================================================================
# API Configuration
# =============================================================================
api:
  coingecko:
    # API key (optional, for higher rate limits)
    # Get free key from: https://www.coingecko.com/en/api
    # Can also be set via COINGECKO_API_KEY environment variable
    api_key: ${COINGECKO_API_KEY}

    # Use Pro API endpoint (requires paid key)
    use_pro: false

    # Request timeout in seconds
    timeout: 10

    # Maximum retries on failure
    max_retries: 3

  yfinance:
    # Enable yfinance as fallback data source
    enabled: true

# =============================================================================
# Cache Configuration
# =============================================================================
cache:
  # Enable caching to reduce API calls
  enabled: true

  # Time-to-live for spot prices in seconds
  # Lower values = fresher data, higher API usage
  # Recommended: 30-60 seconds
  spot_ttl: 30

  # Time-to-live for historical data in seconds
  # Historical data changes less frequently
  # Recommended: 3600 (1 hour)
  historical_ttl: 3600

  # Cache storage directory (relative to skill root)
  directory: ./data

# =============================================================================
# Currency Configuration
# =============================================================================
currency:
  # Default fiat currency for price display
  default: usd

  # Supported currencies (CoinGecko supports 30+)
  supported:
    - usd    # US Dollar
    - eur    # Euro
    - gbp    # British Pound
    - jpy    # Japanese Yen
    - cad    # Canadian Dollar
    - aud    # Australian Dollar
    - chf    # Swiss Franc
    - cny    # Chinese Yuan
    - inr    # Indian Rupee
    - krw    # Korean Won
    - btc    # Bitcoin (for crypto-to-crypto)
    - eth    # Ethereum (for crypto-to-crypto)

# =============================================================================
# Predefined Watchlists
# =============================================================================
# Use CoinGecko IDs (not ticker symbols)
# Find IDs at: https://www.coingecko.com/en/api/documentation
watchlists:
  # Top 10 by market cap
  top10:
    - bitcoin
    - ethereum
    - tether
    - binancecoin
    - solana
    - ripple
    - cardano
    - avalanche-2
    - dogecoin
    - polkadot

  # DeFi protocols
  defi:
    - uniswap
    - aave
    - chainlink
    - maker
    - compound-governance-token
    - curve-dao-token
    - sushi
    - 1inch
    - pancakeswap-token

  # Layer 2 solutions
  layer2:
    - matic-network
    - arbitrum
    - optimism
    - immutable-x
    - starknet
    - mantle

  # Stablecoins
  stablecoins:
    - tether
    - usd-coin
    - dai
    - frax
    - true-usd
    - paxos-standard

  # Memecoins
  memecoins:
    - dogecoin
    - shiba-inu
    - pepe
    - floki
    - bonk
    - dogwifcoin

  # Custom watchlist (user-configurable)
  custom: []
    # Add your favorite coins here:
    # - bitcoin
    # - ethereum
    # - your-favorite-coin

# =============================================================================
# Display Configuration
# =============================================================================
display:
  # Number of decimal places for prices
  price_decimals:
    high_value: 2      # Prices >= $1
    low_value: 6       # Prices < $1
    very_low: 8        # Prices < $0.0001

  # Show percentage change colors (ANSI)
  # Note: Only works in terminal, not in file output
  color_output: true

  # Table width for aligned output
  table_width: 80

# =============================================================================
# Rate Limiting
# =============================================================================
rate_limit:
  # Minimum seconds between API requests
  # CoinGecko free tier: ~10-50 calls/minute
  min_interval: 1.5

  # Exponential backoff on rate limit
  backoff_factor: 2
  max_backoff: 60

#!/usr/bin/env python3
"""
Crypto API Client - Multi-Source Price Data Fetcher

Provides unified interface to CoinGecko (primary) and yfinance (fallback)
for cryptocurrency price data retrieval.

Author: Jeremy Longshore <jeremy@intentsolutions.io>
Version: 2.0.0
License: MIT
"""

import os
import time
from datetime import datetime, timedelta
from typing import Optional, Dict, List, Any
from dataclasses import dataclass
from enum import Enum

try:
    import requests
except ImportError:
    raise ImportError("Required: pip install requests")


class APIError(Exception):
    """Base exception for API errors."""
    pass


class RateLimitError(APIError):
    """Rate limit exceeded."""
    pass


class NetworkError(APIError):
    """Network connectivity error."""
    pass


class SymbolNotFoundError(APIError):
    """Cryptocurrency symbol not found."""
    pass


class DataSource(Enum):
    """Available data sources."""
    COINGECKO = "coingecko"
    YFINANCE = "yfinance"


@dataclass
class PriceData:
    """Standardized price data structure."""
    symbol: str
    name: str
    price: float
    currency: str
    change_24h: Optional[float] = None
    change_7d: Optional[float] = None
    volume_24h: Optional[float] = None
    market_cap: Optional[float] = None
    timestamp: Optional[str] = None
    source: str = "coingecko"

    def to_dict(self) -> dict:
        """Convert to dictionary."""
        return {
            "symbol": self.symbol,
            "name": self.name,
            "price": self.price,
            "currency": self.currency,
            "change_24h": self.change_24h,
            "change_7d": self.change_7d,
            "volume_24h": self.volume_24h,
            "market_cap": self.market_cap,
            "timestamp": self.timestamp or datetime.utcnow().isoformat(),
            "source": self.source
        }


class CryptoAPIClient:
    """
    Unified cryptocurrency API client with multi-source support.

    Primary: CoinGecko API (free tier or Pro)
    Fallback: yfinance (Yahoo Finance)
    """

    COINGECKO_BASE = "https://api.coingecko.com/api/v3"
    COINGECKO_PRO_BASE = "https://pro-api.coingecko.com/api/v3"

    # Common symbol to CoinGecko ID mapping
    SYMBOL_MAP = {
        "BTC": "bitcoin",
        "ETH": "ethereum",
        "SOL": "solana",
        "XRP": "ripple",
        "ADA": "cardano",
        "DOGE": "dogecoin",
        "DOT": "polkadot",
        "AVAX": "avalanche-2",
        "MATIC": "matic-network",
        "LINK": "chainlink",
        "UNI": "uniswap",
        "AAVE": "aave",
        "MKR": "maker",
        "COMP": "compound-governance-token",
        "CRV": "curve-dao-token",
        "SUSHI": "sushi",
        "ARB": "arbitrum",
        "OP": "optimism",
        "IMX": "immutable-x",
        "USDT": "tether",
        "USDC": "usd-coin",
        "DAI": "dai",
        "BNB": "binancecoin",
        "SHIB": "shiba-inu",
        "PEPE": "pepe",
        "BONK": "bonk"
    }

    def __init__(self, config: Optional[dict] = None):
        """
        Initialize the API client.

        Args:
            config: Optional configuration dictionary with API keys
        """
        self.config = config or {}

        # CoinGecko API setup
        self.api_key = (
            os.environ.get("COINGECKO_API_KEY") or
            self.config.get("api", {}).get("coingecko", {}).get("api_key")
        )
        self.use_pro = (
            self.api_key and
            self.config.get("api", {}).get("coingecko", {}).get("use_pro", False)
        )

        self.base_url = (
            self.COINGECKO_PRO_BASE if self.use_pro else self.COINGECKO_BASE
        )

        # Rate limiting
        self._last_request_time = 0
        self._min_request_interval = 1.5  # seconds (CoinGecko free tier)
        self._retry_count = 0
        self._max_retries = 3

        # yfinance availability
        self._yfinance_available = None

        # Coin list cache
        self._coin_list_cache = None
        self._coin_list_timestamp = None

    def _check_yfinance(self) -> bool:
        """Check if yfinance is available."""
        if self._yfinance_available is None:
            try:
                import yfinance
                self._yfinance_available = True
            except ImportError:
                self._yfinance_available = False
        return self._yfinance_available

    def _rate_limit(self) -> None:
        """Apply rate limiting between requests."""
        elapsed = time.time() - self._last_request_time
        if elapsed < self._min_request_interval:
            time.sleep(self._min_request_interval - elapsed)
        self._last_request_time = time.time()

    def _make_request(
        self,
        endpoint: str,
        params: Optional[dict] = None
    ) -> dict:
        """
        Make HTTP request to CoinGecko API with error handling.

        Args:
            endpoint: API endpoint path
            params: Query parameters

        Returns:
            JSON response data

        Raises:
            RateLimitError: Rate limit exceeded
            NetworkError: Connection failed
            APIError: Other API errors
        """
        self._rate_limit()

        url = f"{self.base_url}{endpoint}"
        headers = {}

        if self.api_key:
            if self.use_pro:
                headers["x-cg-pro-api-key"] = self.api_key
            else:
                headers["x-cg-demo-api-key"] = self.api_key

        try:
            response = requests.get(
                url,
                params=params,
                headers=headers,
                timeout=10
            )

            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 60))
                raise RateLimitError(
                    f"Rate limit exceeded. Retry after {retry_after}s"
                )

            if response.status_code == 404:
                raise SymbolNotFoundError("Resource not found")

            response.raise_for_status()
            return response.json()

        except requests.exceptions.ConnectionError as e:
            raise NetworkError(f"Connection failed: {e}")
        except requests.exceptions.Timeout:
            raise NetworkError("Request timed out")
        except requests.exceptions.RequestException as e:
            raise APIError(f"Request failed: {e}")

    def _symbol_to_id(self, symbol: str) -> str:
        """
        Convert ticker symbol to CoinGecko ID.

        Args:
            symbol: Cryptocurrency ticker (e.g., "BTC")

        Returns:
            CoinGecko ID (e.g., "bitcoin")
        """
        symbol_upper = symbol.upper()

        # Check direct mapping
        if symbol_upper in self.SYMBOL_MAP:
            return self.SYMBOL_MAP[symbol_upper]

        # Assume lowercase symbol is the ID
        return symbol.lower()

    def list_coins(
        self,
        query: Optional[str] = None,
        limit: int = 100
    ) -> List[dict]:
        """
        List available cryptocurrencies.

        Args:
            query: Optional search query
            limit: Maximum results to return

        Returns:
            List of coin info dictionaries
        """
        # Refresh coin list if stale (older than 1 hour)
        now = time.time()
        if (self._coin_list_cache is None or
            self._coin_list_timestamp is None or
            now - self._coin_list_timestamp > 3600):

            try:
                self._coin_list_cache = self._make_request("/coins/list")
                self._coin_list_timestamp = now
            except APIError:
                # Return empty if we can't fetch
                return []

        coins = self._coin_list_cache

        # Filter by query
        if query:
            query_lower = query.lower()
            coins = [
                c for c in coins
                if (query_lower in c.get("id", "").lower() or
                    query_lower in c.get("name", "").lower() or
                    query_lower in c.get("symbol", "").lower())
            ]

        # Limit results
        return coins[:limit]

    def get_current_price(
        self,
        symbol: str,
        currency: str = "usd"
    ) -> dict:
        """
        Get current price for a cryptocurrency.

        Args:
            symbol: Cryptocurrency symbol (e.g., "BTC") or CoinGecko ID
            currency: Fiat currency code (e.g., "usd")

        Returns:
            Price data dictionary
        """
        coin_id = self._symbol_to_id(symbol)

        try:
            data = self._make_request(
                f"/coins/{coin_id}",
                params={
                    "localization": "false",
                    "tickers": "false",
                    "community_data": "false",
                    "developer_data": "false",
                    "sparkline": "false"
                }
            )

            market_data = data.get("market_data", {})
            current_price = market_data.get("current_price", {})
            price = current_price.get(currency.lower())

            if price is None:
                raise APIError(f"Price not available in {currency}")

            return PriceData(
                symbol=data.get("symbol", symbol).upper(),
                name=data.get("name", symbol),
                price=price,
                currency=currency.upper(),
                change_24h=market_data.get("price_change_percentage_24h"),
                change_7d=market_data.get("price_change_percentage_7d"),
                volume_24h=market_data.get("total_volume", {}).get(currency.lower()),
                market_cap=market_data.get("market_cap", {}).get(currency.lower()),
                source="coingecko"
            ).to_dict()

        except SymbolNotFoundError:
            # Try yfinance fallback
            if self._check_yfinance():
                return self._get_price_yfinance(symbol, currency)
            raise SymbolNotFoundError(f"Unknown symbol: {symbol}")

        except RateLimitError:
            # Try yfinance fallback
            if self._check_yfinance():
                return self._get_price_yfinance(symbol, currency)
            raise

    def _get_price_yfinance(
        self,
        symbol: str,
        currency: str = "usd"
    ) -> dict:
        """
        Get price using yfinance as fallback.

        Args:
            symbol: Cryptocurrency symbol
            currency: Fiat currency

        Returns:
            Price data dictionary
        """
        try:
            import yfinance as yf
        except ImportError:
            raise APIError("yfinance not installed")

        # yfinance uses format like "BTC-USD"
        ticker_symbol = f"{symbol.upper()}-{currency.upper()}"

        try:
            ticker = yf.Ticker(ticker_symbol)
            info = ticker.info

            if not info or "regularMarketPrice" not in info:
                raise SymbolNotFoundError(f"Unknown symbol: {symbol}")

            return PriceData(
                symbol=symbol.upper(),
                name=info.get("shortName", symbol),
                price=info.get("regularMarketPrice", 0),
                currency=currency.upper(),
                change_24h=info.get("regularMarketChangePercent"),
                volume_24h=info.get("regularMarketVolume"),
                market_cap=info.get("marketCap"),
                source="yfinance"
            ).to_dict()

        except Exception as e:
            raise APIError(f"yfinance error: {e}")

    def get_historical_prices(
        self,
        symbol: str,
        currency: str = "usd",
        period: Optional[str] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None
    ) -> List[dict]:
        """
        Get historical OHLCV data.

        Args:
            symbol: Cryptocurrency symbol
            currency: Fiat currency
            period: Period string (e.g., "7d", "30d", "90d", "1y", "max")
            start_date: Custom start date
            end_date: Custom end date

        Returns:
            List of OHLCV data points
        """
        coin_id = self._symbol_to_id(symbol)

        # Convert period to days
        if period:
            period_map = {
                "1d": 1, "7d": 7, "14d": 14, "30d": 30,
                "60d": 60, "90d": 90, "180d": 180,
                "1y": 365, "2y": 730, "max": "max"
            }
            days = period_map.get(period.lower(), 30)
        elif start_date and end_date:
            days = (end_date - start_date).days
        else:
            days = 30

        try:
            # Use market_chart for period-based queries
            if isinstance(days, int):
                data = self._make_request(
                    f"/coins/{coin_id}/market_chart",
                    params={
                        "vs_currency": currency.lower(),
                        "days": days,
                        "interval": "daily" if days > 1 else "hourly"
                    }
                )

                prices = data.get("prices", [])
                volumes = data.get("total_volumes", [])

                results = []
                for i, (timestamp, price) in enumerate(prices):
                    dt = datetime.fromtimestamp(timestamp / 1000)
                    vol = volumes[i][1] if i < len(volumes) else None

                    results.append({
                        "date": dt.strftime("%Y-%m-%d"),
                        "timestamp": timestamp,
                        "price": price,
                        "volume": vol
                    })

                return results

            else:  # "max" period
                data = self._make_request(
                    f"/coins/{coin_id}/market_chart",
                    params={
                        "vs_currency": currency.lower(),
                        "days": "max"
                    }
                )

                prices = data.get("prices", [])
                results = []

                for timestamp, price in prices:
                    dt = datetime.fromtimestamp(timestamp / 1000)
                    results.append({
                        "date": dt.strftime("%Y-%m-%d"),
                        "timestamp": timestamp,
                        "price": price
                    })

                return results

        except (RateLimitError, SymbolNotFoundError):
            # Try yfinance fallback
            if self._check_yfinance():
                return self._get_historical_yfinance(
                    symbol, currency, period, start_date, end_date
                )
            raise

    def _get_historical_yfinance(
        self,
        symbol: str,
        currency: str = "usd",
        period: Optional[str] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None
    ) -> List[dict]:
        """
        Get historical data using yfinance.

        Args:
            symbol: Cryptocurrency symbol
            currency: Fiat currency
            period: Period string
            start_date: Start date
            end_date: End date

        Returns:
            List of OHLCV data points
        """
        try:
            import yfinance as yf
        except ImportError:
            raise APIError("yfinance not installed")

        ticker_symbol = f"{symbol.upper()}-{currency.upper()}"
        ticker = yf.Ticker(ticker_symbol)

        # Map period to yfinance format
        if period:
            yf_period_map = {
                "1d": "1d", "7d": "7d", "14d": "14d", "30d": "1mo",
                "60d": "2mo", "90d": "3mo", "180d": "6mo",
                "1y": "1y", "2y": "2y", "max": "max"
            }
            yf_period = yf_period_map.get(period.lower(), "1mo")
            df = ticker.history(period=yf_period)
        elif start_date and end_date:
            df = ticker.history(start=start_date, end=end_date)
        else:
            df = ticker.history(period="1mo")

        results = []
        for date, row in df.iterrows():
            results.append({
                "date": date.strftime("%Y-%m-%d"),
                "open": row.get("Open"),
                "high": row.get("High"),
                "low": row.get("Low"),
                "close": row.get("Close"),
                "volume": row.get("Volume")
            })

        return results

#!/usr/bin/env python3
"""
Cache Manager - TTL-Based Caching for Price Data

Provides intelligent caching with configurable TTL for spot prices
and historical data to minimize API calls and improve response times.

Author: Jeremy Longshore <jeremy@intentsolutions.io>
Version: 2.0.0
License: MIT
"""

import json
import time
import hashlib
from datetime import datetime
from pathlib import Path
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, asdict
import threading


@dataclass
class CacheEntry:
    """Represents a cached item with metadata."""
    data: Any
    timestamp: float
    ttl: int
    key: str

    @property
    def is_expired(self) -> bool:
        """Check if cache entry has expired."""
        return time.time() - self.timestamp > self.ttl

    @property
    def age(self) -> float:
        """Get age of cache entry in seconds."""
        return time.time() - self.timestamp

    def to_dict(self) -> dict:
        """Convert to dictionary for JSON serialization."""
        return {
            "data": self.data,
            "timestamp": self.timestamp,
            "ttl": self.ttl,
            "key": self.key
        }

    @classmethod
    def from_dict(cls, d: dict) -> "CacheEntry":
        """Create CacheEntry from dictionary."""
        return cls(
            data=d["data"],
            timestamp=d["timestamp"],
            ttl=d["ttl"],
            key=d["key"]
        )


class CacheManager:
    """
    TTL-based cache manager for cryptocurrency price data.

    Features:
    - Separate TTLs for spot prices (short) and historical data (long)
    - File-based persistence
    - Thread-safe operations
    - Automatic cleanup of expired entries
    - Stale data access for fallback scenarios
    """

    def __init__(
        self,
        cache_dir: Optional[Path] = None,
        spot_ttl: int = 30,
        historical_ttl: int = 3600
    ):
        """
        Initialize the cache manager.

        Args:
            cache_dir: Directory for cache files (default: ./data)
            spot_ttl: TTL for spot prices in seconds (default: 30)
            historical_ttl: TTL for historical data in seconds (default: 3600)
        """
        self.cache_dir = Path(cache_dir) if cache_dir else Path("./data")
        self.spot_ttl = spot_ttl
        self.historical_ttl = historical_ttl

        # Ensure cache directory exists
        self.cache_dir.mkdir(parents=True, exist_ok=True)

        # In-memory cache
        self._memory_cache: Dict[str, CacheEntry] = {}
        self._lock = threading.RLock()

        # File paths
        self._spot_cache_file = self.cache_dir / "spot_cache.json"
        self._historical_cache_file = self.cache_dir / "historical_cache.json"

        # Load existing cache
        self._load_cache()

    def _load_cache(self) -> None:
        """Load cache from disk."""
        with self._lock:
            # Load spot cache
            if self._spot_cache_file.exists():
                try:
                    with open(self._spot_cache_file, "r") as f:
                        data = json.load(f)
                        for key, entry_data in data.items():
                            entry = CacheEntry.from_dict(entry_data)
                            if not entry.is_expired:
                                self._memory_cache[key] = entry
                except (json.JSONDecodeError, KeyError):
                    pass

            # Load historical cache
            if self._historical_cache_file.exists():
                try:
                    with open(self._historical_cache_file, "r") as f:
                        data = json.load(f)
                        for key, entry_data in data.items():
                            entry = CacheEntry.from_dict(entry_data)
                            if not entry.is_expired:
                                self._memory_cache[key] = entry
                except (json.JSONDecodeError, KeyError):
                    pass

    def _save_cache(self) -> None:
        """Persist cache to disk."""
        with self._lock:
            spot_data = {}
            historical_data = {}

            for key, entry in self._memory_cache.items():
                if key.startswith("spot:"):
                    spot_data[key] = entry.to_dict()
                elif key.startswith("hist:"):
                    historical_data[key] = entry.to_dict()

            # Write spot cache
            try:
                with open(self._spot_cache_file, "w") as f:
                    json.dump(spot_data, f, indent=2)
            except IOError:
                pass

            # Write historical cache
            try:
                with open(self._historical_cache_file, "w") as f:
                    json.dump(historical_data, f, indent=2)
            except IOError:
                pass

    def _make_key(self, prefix: str, *parts: str) -> str:
        """
        Create a cache key from parts.

        Args:
            prefix: Key prefix (e.g., "spot", "hist")
            *parts: Key components

        Returns:
            Cache key string
        """
        key_str = ":".join([prefix] + [str(p).lower() for p in parts])
        return key_str

    def get_spot_price(
        self,
        symbol: str,
        currency: str,
        allow_stale: bool = False
    ) -> Optional[dict]:
        """
        Get cached spot price.

        Args:
            symbol: Cryptocurrency symbol
            currency: Fiat currency
            allow_stale: Return stale data if no fresh data available

        Returns:
            Cached price data or None
        """
        key = self._make_key("spot", symbol, currency)

        with self._lock:
            entry = self._memory_cache.get(key)

            if entry is None:
                return None

            if entry.is_expired:
                if allow_stale:
                    # Return stale data with warning flag
                    data = entry.data.copy() if isinstance(entry.data, dict) else entry.data
                    if isinstance(data, dict):
                        data["_cache_stale"] = True
                        data["_cache_age"] = entry.age
                    return data
                return None

            data = entry.data.copy() if isinstance(entry.data, dict) else entry.data
            if isinstance(data, dict):
                data["_cached"] = True
                data["_cache_age"] = entry.age
            return data

    def set_spot_price(
        self,
        symbol: str,
        currency: str,
        data: dict
    ) -> None:
        """
        Cache spot price data.

        Args:
            symbol: Cryptocurrency symbol
            currency: Fiat currency
            data: Price data to cache
        """
        key = self._make_key("spot", symbol, currency)

        with self._lock:
            self._memory_cache[key] = CacheEntry(
                data=data,
                timestamp=time.time(),
                ttl=self.spot_ttl,
                key=key
            )
            self._save_cache()

    def get_historical(
        self,
        cache_key: str,
        allow_stale: bool = False
    ) -> Optional[List[dict]]:
        """
        Get cached historical data.

        Args:
            cache_key: Unique key for the historical query
            allow_stale: Return stale data if no fresh data available

        Returns:
            Cached historical data or None
        """
        key = self._make_key("hist", cache_key)

        with self._lock:
            entry = self._memory_cache.get(key)

            if entry is None:
                return None

            if entry.is_expired:
                if allow_stale:
                    return entry.data
                return None

            return entry.data

    def set_historical(
        self,
        cache_key: str,
        data: List[dict]
    ) -> None:
        """
        Cache historical data.

        Args:
            cache_key: Unique key for the historical query
            data: Historical data to cache
        """
        key = self._make_key("hist", cache_key)

        with self._lock:
            self._memory_cache[key] = CacheEntry(
                data=data,
                timestamp=time.time(),
                ttl=self.historical_ttl,
                key=key
            )
            self._save_cache()

    def invalidate(self, pattern: Optional[str] = None) -> int:
        """
        Invalidate cache entries matching a pattern.

        Args:
            pattern: Key pattern to match (None = all)

        Returns:
            Number of entries invalidated
        """
        with self._lock:
            if pattern is None:
                count = len(self._memory_cache)
                self._memory_cache.clear()
            else:
                pattern_lower = pattern.lower()
                keys_to_remove = [
                    k for k in self._memory_cache
                    if pattern_lower in k.lower()
                ]
                count = len(keys_to_remove)
                for key in keys_to_remove:
                    del self._memory_cache[key]

            self._save_cache()
            return count

    def clear(self) -> None:
        """Clear all cached data."""
        with self._lock:
            self._memory_cache.clear()

            # Remove cache files
            if self._spot_cache_file.exists():
                self._spot_cache_file.unlink()
            if self._historical_cache_file.exists():
                self._historical_cache_file.unlink()

    def cleanup(self) -> int:
        """
        Remove all expired entries.

        Returns:
            Number of entries removed
        """
        with self._lock:
            expired_keys = [
                k for k, v in self._memory_cache.items()
                if v.is_expired
            ]
            for key in expired_keys:
                del self._memory_cache[key]

            if expired_keys:
                self._save_cache()

            return len(expired_keys)

    def get_stats(self) -> dict:
        """
        Get cache statistics.

        Returns:
            Dictionary with cache stats
        """
        with self._lock:
            spot_entries = [
                e for k, e in self._memory_cache.items()
                if k.startswith("spot:")
            ]
            hist_entries = [
                e for k, e in self._memory_cache.items()
                if k.startswith("hist:")
            ]

            spot_expired = sum(1 for e in spot_entries if e.is_expired)
            hist_expired = sum(1 for e in hist_entries if e.is_expired)

            return {
                "total_entries": len(self._memory_cache),
                "spot_entries": len(spot_entries),
                "spot_expired": spot_expired,
                "historical_entries": len(hist_entries),
                "historical_expired": hist_expired,
                "cache_dir": str(self.cache_dir),
                "spot_ttl": self.spot_ttl,
                "historical_ttl": self.historical_ttl
            }

    def get_all_symbols(self) -> List[str]:
        """
        Get all symbols currently in cache.

        Returns:
            List of cached symbols
        """
        with self._lock:
            symbols = set()
            for key in self._memory_cache.keys():
                if key.startswith("spot:"):
                    parts = key.split(":")
                    if len(parts) >= 2:
                        symbols.add(parts[1].upper())
            return sorted(list(symbols))

#!/usr/bin/env python3
"""
Price Formatters - Output Formatting for Price Data

Provides human-readable table output, JSON formatting, CSV export,
and minimal output modes for cryptocurrency price data.

Author: Jeremy Longshore <jeremy@intentsolutions.io>
Version: 2.0.0
License: MIT
"""

import json
import csv
import io
from datetime import datetime
from typing import List, Optional, Dict, Any


class PriceFormatter:
    """
    Formats cryptocurrency price data for various output modes.

    Supports:
    - Table: Human-readable aligned tables
    - JSON: Machine-readable JSON
    - CSV: Spreadsheet-compatible export
    - Minimal: Single-line output for scripting
    """

    # Currency symbols for formatting
    CURRENCY_SYMBOLS = {
        "USD": "$",
        "EUR": "€",
        "GBP": "£",
        "JPY": "¥",
        "CAD": "C$",
        "AUD": "A$",
        "CHF": "CHF ",
        "CNY": "¥",
        "INR": "₹",
        "KRW": "₩"
    }

    def __init__(self, currency: str = "USD"):
        """
        Initialize formatter.

        Args:
            currency: Default currency code
        """
        self.currency = currency.upper()
        self.currency_symbol = self.CURRENCY_SYMBOLS.get(self.currency, "")

    def _format_price(self, price: float) -> str:
        """
        Format a price value with appropriate precision.

        Args:
            price: Price value

        Returns:
            Formatted price string
        """
        if price >= 1000:
            return f"{self.currency_symbol}{price:,.2f}"
        elif price >= 1:
            return f"{self.currency_symbol}{price:.2f}"
        elif price >= 0.0001:
            return f"{self.currency_symbol}{price:.6f}"
        else:
            return f"{self.currency_symbol}{price:.8f}"

    def _format_large_number(self, value: Optional[float]) -> str:
        """
        Format large numbers with K/M/B suffixes.

        Args:
            value: Number to format

        Returns:
            Formatted string
        """
        if value is None:
            return "N/A"

        if value >= 1_000_000_000_000:
            return f"${value / 1_000_000_000_000:.2f}T"
        elif value >= 1_000_000_000:
            return f"${value / 1_000_000_000:.2f}B"
        elif value >= 1_000_000:
            return f"${value / 1_000_000:.2f}M"
        elif value >= 1_000:
            return f"${value / 1_000:.2f}K"
        else:
            return f"${value:.2f}"

    def _format_change(self, change: Optional[float]) -> str:
        """
        Format percentage change with color indicators.

        Args:
            change: Percentage change

        Returns:
            Formatted change string
        """
        if change is None:
            return "N/A"

        if change >= 0:
            return f"+{change:.2f}%"
        else:
            return f"{change:.2f}%"

    def format_prices(
        self,
        prices: List[dict],
        format_type: str = "table",
        verbose: bool = False
    ) -> str:
        """
        Format a list of price data.

        Args:
            prices: List of price dictionaries
            format_type: Output format (table, json, csv, minimal)
            verbose: Include extra details

        Returns:
            Formatted output string
        """
        if format_type == "json":
            return self._format_json(prices)
        elif format_type == "csv":
            return self._format_csv_prices(prices)
        elif format_type == "minimal":
            return self._format_minimal(prices)
        else:
            return self._format_table(prices, verbose)

    def _format_table(self, prices: List[dict], verbose: bool = False) -> str:
        """Format prices as aligned table."""
        if not prices:
            return "No price data available"

        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        lines = []

        # Header
        lines.append("=" * 80)
        lines.append(f"  CRYPTO PRICES{' ' * 45}Updated: {timestamp}")
        lines.append("=" * 80)
        lines.append("")

        # Column headers
        if verbose:
            header = f"  {'Symbol':<10} {'Price':>14} {'24h':>10} {'7d':>10} {'Volume':>12} {'Market Cap':>12}"
        else:
            header = f"  {'Symbol':<10} {'Price':>14} {'24h Change':>12} {'Volume (24h)':>14} {'Market Cap':>12}"
        lines.append(header)
        lines.append("-" * 80)

        # Data rows
        total_change = 0
        total_weight = 0

        for p in prices:
            symbol = p.get("symbol", "???").upper()
            price = p.get("price", 0)
            change_24h = p.get("change_24h")
            change_7d = p.get("change_7d")
            volume = p.get("volume_24h")
            market_cap = p.get("market_cap")

            price_str = self._format_price(price)
            change_24h_str = self._format_change(change_24h)
            change_7d_str = self._format_change(change_7d) if verbose else ""
            volume_str = self._format_large_number(volume)
            mcap_str = self._format_large_number(market_cap)

            # Track weighted average change
            if change_24h is not None and market_cap is not None:
                total_change += change_24h * market_cap
                total_weight += market_cap

            # Stale/cached indicators
            suffix = ""
            if p.get("_cache_stale"):
                suffix = " (stale)"
            elif p.get("_cached"):
                suffix = " (cached)"

            if verbose:
                row = f"  {symbol:<10} {price_str:>14} {change_24h_str:>10} {change_7d_str:>10} {volume_str:>12} {mcap_str:>12}{suffix}"
            else:
                row = f"  {symbol:<10} {price_str:>14} {change_24h_str:>12} {volume_str:>14} {mcap_str:>12}{suffix}"

            lines.append(row)

        lines.append("-" * 80)

        # Summary
        if total_weight > 0:
            weighted_change = total_change / total_weight
            lines.append(f"  Total 24h Change: {self._format_change(weighted_change)} (weighted)")
        lines.append("")
        lines.append("=" * 80)

        return "\n".join(lines)

    def _format_json(self, prices: List[dict]) -> str:
        """Format prices as JSON."""
        # Clean internal fields
        cleaned = []
        for p in prices:
            clean_p = {k: v for k, v in p.items() if not k.startswith("_")}
            cleaned.append(clean_p)

        output = {
            "prices": cleaned,
            "meta": {
                "count": len(cleaned),
                "currency": self.currency,
                "timestamp": datetime.utcnow().isoformat()
            }
        }

        return json.dumps(output, indent=2)

    def _format_csv_prices(self, prices: List[dict]) -> str:
        """Format prices as CSV."""
        if not prices:
            return ""

        output = io.StringIO()
        fieldnames = ["symbol", "name", "price", "currency", "change_24h",
                      "change_7d", "volume_24h", "market_cap", "timestamp", "source"]
        writer = csv.DictWriter(output, fieldnames=fieldnames, extrasaction="ignore")
        writer.writeheader()

        for p in prices:
            # Filter out internal fields
            clean_p = {k: v for k, v in p.items() if not k.startswith("_")}
            writer.writerow(clean_p)

        return output.getvalue()

    def _format_minimal(self, prices: List[dict]) -> str:
        """Format prices as minimal one-line output."""
        parts = []
        for p in prices:
            symbol = p.get("symbol", "???").upper()
            price = p.get("price", 0)
            change = p.get("change_24h")

            if change is not None:
                change_str = f"({self._format_change(change)})"
            else:
                change_str = ""

            parts.append(f"{symbol}:{self._format_price(price)}{change_str}")

        return " | ".join(parts)

    def format_historical(
        self,
        symbol: str,
        data: List[dict],
        format_type: str = "table"
    ) -> str:
        """
        Format historical price data.

        Args:
            symbol: Cryptocurrency symbol
            data: List of historical data points
            format_type: Output format

        Returns:
            Formatted output string
        """
        if format_type == "json":
            return self._format_historical_json(symbol, data)
        elif format_type == "csv":
            return self._format_historical_csv(data)
        else:
            return self._format_historical_table(symbol, data)

    def _format_historical_table(self, symbol: str, data: List[dict]) -> str:
        """Format historical data as table."""
        if not data:
            return "No historical data available"

        lines = []
        lines.append("=" * 70)
        lines.append(f"  HISTORICAL PRICES: {symbol.upper()}")
        lines.append(f"  Period: {data[0].get('date', 'N/A')} to {data[-1].get('date', 'N/A')}")
        lines.append("=" * 70)
        lines.append("")

        # Determine columns based on data
        has_ohlc = "open" in data[0]

        if has_ohlc:
            lines.append(f"  {'Date':<12} {'Open':>12} {'High':>12} {'Low':>12} {'Close':>12}")
        else:
            lines.append(f"  {'Date':<12} {'Price':>14} {'Volume':>16}")

        lines.append("-" * 70)

        for d in data[-30:]:  # Show last 30 entries
            date = d.get("date", "N/A")

            if has_ohlc:
                open_p = self._format_price(d.get("open", 0))
                high = self._format_price(d.get("high", 0))
                low = self._format_price(d.get("low", 0))
                close = self._format_price(d.get("close", 0))
                lines.append(f"  {date:<12} {open_p:>12} {high:>12} {low:>12} {close:>12}")
            else:
                price = self._format_price(d.get("price", 0))
                volume = self._format_large_number(d.get("volume"))
                lines.append(f"  {date:<12} {price:>14} {volume:>16}")

        if len(data) > 30:
            lines.append(f"  ... ({len(data) - 30} more entries)")

        lines.append("-" * 70)
        lines.append(f"  Total data points: {len(data)}")
        lines.append("=" * 70)

        return "\n".join(lines)

    def _format_historical_json(self, symbol: str, data: List[dict]) -> str:
        """Format historical data as JSON."""
        output = {
            "symbol": symbol.upper(),
            "data": data,
            "meta": {
                "count": len(data),
                "start_date": data[0].get("date") if data else None,
                "end_date": data[-1].get("date") if data else None,
                "timestamp": datetime.utcnow().isoformat()
            }
        }
        return json.dumps(output, indent=2)

    def _format_historical_csv(self, data: List[dict]) -> str:
        """Format historical data as CSV."""
        if not data:
            return ""

        output = io.StringIO()

        # Determine columns from first row
        if "open" in data[0]:
            fieldnames = ["date", "open", "high", "low", "close", "volume"]
        else:
            fieldnames = ["date", "price", "volume"]

        writer = csv.DictWriter(output, fieldnames=fieldnames, extrasaction="ignore")
        writer.writeheader()

        for d in data:
            writer.writerow(d)

        return output.getvalue()

    def print_coin_list(
        self,
        coins: List[dict],
        query: Optional[str] = None
    ) -> None:
        """
        Print a list of available coins.

        Args:
            coins: List of coin info dictionaries
            query: Search query that was used
        """
        if not coins:
            if query:
                print(f"No coins found matching '{query}'")
            else:
                print("No coins available")
            return

        print("=" * 60)
        if query:
            print(f"  SEARCH RESULTS: '{query}'")
        else:
            print("  AVAILABLE CRYPTOCURRENCIES")
        print("=" * 60)
        print(f"  {'Symbol':<10} {'ID':<25} {'Name':<20}")
        print("-" * 60)

        for coin in coins[:50]:  # Limit to 50 results
            symbol = coin.get("symbol", "").upper()
            coin_id = coin.get("id", "")
            name = coin.get("name", "")

            # Truncate long names
            if len(name) > 18:
                name = name[:17] + "..."

            print(f"  {symbol:<10} {coin_id:<25} {name:<20}")

        if len(coins) > 50:
            print(f"  ... and {len(coins) - 50} more")

        print("-" * 60)
        print(f"  Total: {len(coins)} coins")
        print("=" * 60)

#!/usr/bin/env python3
"""
Crypto Price Tracker - Main CLI Entry Point

Track real-time and historical cryptocurrency prices across exchanges.
Foundation skill for the crypto plugin ecosystem.

Author: Jeremy Longshore <jeremy@intentsolutions.io>
Version: 2.0.0
License: MIT
"""

import argparse
import json
import sys
from datetime import datetime
from pathlib import Path
from typing import Optional

# Add scripts directory to path for local imports
SCRIPT_DIR = Path(__file__).parent
sys.path.insert(0, str(SCRIPT_DIR))

from api_client import CryptoAPIClient, APIError
from cache_manager import CacheManager
from formatters import PriceFormatter


def parse_args() -> argparse.Namespace:
    """Parse command-line arguments."""
    parser = argparse.ArgumentParser(
        description="Track real-time and historical cryptocurrency prices",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s --symbol BTC                    # Get Bitcoin price
  %(prog)s --symbols BTC,ETH,SOL           # Get multiple prices
  %(prog)s --watchlist top10               # Scan top 10 by market cap
  %(prog)s --symbol BTC --period 30d       # 30-day history
  %(prog)s --symbol ETH --output csv       # Export to CSV
  %(prog)s --list                          # Search available coins
        """
    )

    # Symbol selection (mutually exclusive)
    symbol_group = parser.add_mutually_exclusive_group()
    symbol_group.add_argument(
        "--symbol", "-s",
        type=str,
        help="Single cryptocurrency symbol (e.g., BTC, ETH)"
    )
    symbol_group.add_argument(
        "--symbols",
        type=str,
        help="Comma-separated list of symbols (e.g., BTC,ETH,SOL)"
    )
    symbol_group.add_argument(
        "--watchlist", "-w",
        type=str,
        choices=["top10", "defi", "layer2", "stablecoins", "memecoins", "custom"],
        help="Use predefined watchlist"
    )
    symbol_group.add_argument(
        "--list", "-l",
        action="store_true",
        help="List available cryptocurrencies (search with --query)"
    )

    # Search options
    parser.add_argument(
        "--query", "-q",
        type=str,
        help="Search query for --list mode"
    )

    # Historical data options
    parser.add_argument(
        "--period", "-p",
        type=str,
        help="Historical period (e.g., 7d, 30d, 90d, 1y, max)"
    )
    parser.add_argument(
        "--start",
        type=str,
        help="Start date for custom range (YYYY-MM-DD)"
    )
    parser.add_argument(
        "--end",
        type=str,
        help="End date for custom range (YYYY-MM-DD)"
    )

    # Output options
    parser.add_argument(
        "--format", "-f",
        type=str,
        choices=["table", "json", "csv", "minimal"],
        default="table",
        help="Output format (default: table)"
    )
    parser.add_argument(
        "--output", "-o",
        type=str,
        help="Output file path (default: stdout)"
    )

    # Currency options
    parser.add_argument(
        "--currency", "-c",
        type=str,
        default="usd",
        help="Fiat currency for prices (default: usd)"
    )

    # Cache control
    parser.add_argument(
        "--no-cache",
        action="store_true",
        help="Bypass cache and fetch fresh data"
    )
    parser.add_argument(
        "--clear-cache",
        action="store_true",
        help="Clear all cached data"
    )

    # Verbosity
    parser.add_argument(
        "--verbose", "-v",
        action="store_true",
        help="Enable verbose output"
    )
    parser.add_argument(
        "--quiet",
        action="store_true",
        help="Suppress non-essential output"
    )

    # Version
    parser.add_argument(
        "--version",
        action="version",
        version="%(prog)s 2.0.0"
    )

    return parser.parse_args()


def load_config() -> dict:
    """Load configuration from settings.yaml."""
    config_path = SCRIPT_DIR.parent / "config" / "settings.yaml"

    if config_path.exists():
        try:
            import yaml
            with open(config_path, "r") as f:
                return yaml.safe_load(f) or {}
        except ImportError:
            # Fallback if PyYAML not installed
            pass
        except Exception:
            pass

    # Default configuration
    return {
        "cache": {
            "enabled": True,
            "spot_ttl": 30,
            "historical_ttl": 3600,
            "directory": str(SCRIPT_DIR.parent / "data")
        },
        "currency": {
            "default": "usd"
        },
        "watchlists": {
            "top10": ["bitcoin", "ethereum", "tether", "binancecoin", "solana",
                      "ripple", "cardano", "avalanche-2", "dogecoin", "polkadot"],
            "defi": ["uniswap", "aave", "chainlink", "maker",
                     "compound-governance-token", "curve-dao-token", "sushi"],
            "layer2": ["matic-network", "arbitrum", "optimism", "immutable-x"],
            "stablecoins": ["tether", "usd-coin", "dai", "frax", "true-usd"],
            "memecoins": ["dogecoin", "shiba-inu", "pepe", "floki", "bonk"]
        }
    }


def get_watchlist_symbols(watchlist_name: str, config: dict) -> list:
    """Get symbols for a watchlist."""
    watchlists = config.get("watchlists", {})

    if watchlist_name == "custom":
        custom = watchlists.get("custom", [])
        if not custom:
            print("Error: Custom watchlist is empty. Configure in settings.yaml",
                  file=sys.stderr)
            sys.exit(1)
        return custom

    symbols = watchlists.get(watchlist_name)
    if not symbols:
        print(f"Error: Unknown watchlist '{watchlist_name}'", file=sys.stderr)
        sys.exit(1)

    return symbols


def handle_list_command(
    client: CryptoAPIClient,
    query: Optional[str],
    formatter: PriceFormatter,
    args: argparse.Namespace
) -> None:
    """Handle --list command to search available coins."""
    try:
        coins = client.list_coins(query=query)

        if args.format == "json":
            print(json.dumps(coins, indent=2))
        else:
            formatter.print_coin_list(coins, query)

    except APIError as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


def handle_price_command(
    symbols: list,
    client: CryptoAPIClient,
    cache: CacheManager,
    formatter: PriceFormatter,
    args: argparse.Namespace,
    config: dict
) -> None:
    """Handle price lookup for one or more symbols."""
    currency = args.currency.lower()
    use_cache = not args.no_cache and config.get("cache", {}).get("enabled", True)

    results = []
    cache_hits = 0

    for symbol in symbols:
        # Check cache first
        if use_cache:
            cached = cache.get_spot_price(symbol, currency)
            if cached:
                results.append(cached)
                cache_hits += 1
                continue

        # Fetch from API
        try:
            price_data = client.get_current_price(symbol, currency)
            results.append(price_data)

            # Cache the result
            if use_cache:
                cache.set_spot_price(symbol, currency, price_data)

        except APIError as e:
            if args.verbose:
                print(f"Warning: Failed to fetch {symbol}: {e}", file=sys.stderr)

            # Try cache as fallback
            if use_cache:
                stale = cache.get_spot_price(symbol, currency, allow_stale=True)
                if stale:
                    stale["_stale"] = True
                    results.append(stale)
                    if not args.quiet:
                        print(f"Warning: Using stale cache for {symbol}",
                              file=sys.stderr)

    if not results:
        print("Error: No price data available", file=sys.stderr)
        sys.exit(1)

    # Output results
    output_text = formatter.format_prices(
        results,
        format_type=args.format,
        verbose=args.verbose
    )

    if args.output:
        output_path = Path(args.output)
        output_path.parent.mkdir(parents=True, exist_ok=True)
        with open(output_path, "w") as f:
            f.write(output_text)
        if not args.quiet:
            print(f"Output written to {output_path}")
    else:
        print(output_text)

    # Cache stats
    if args.verbose and use_cache and len(symbols) > 1:
        print(f"\nCache: {cache_hits}/{len(symbols)} hits", file=sys.stderr)


def handle_historical_command(
    symbol: str,
    client: CryptoAPIClient,
    cache: CacheManager,
    formatter: PriceFormatter,
    args: argparse.Namespace,
    config: dict
) -> None:
    """Handle historical price lookup."""
    currency = args.currency.lower()
    use_cache = not args.no_cache and config.get("cache", {}).get("enabled", True)

    # Parse period or date range
    if args.start and args.end:
        start_date = datetime.strptime(args.start, "%Y-%m-%d")
        end_date = datetime.strptime(args.end, "%Y-%m-%d")
        period = None
    elif args.period:
        period = args.period
        start_date = None
        end_date = None
    else:
        period = "30d"  # Default
        start_date = None
        end_date = None

    # Check cache
    cache_key = f"{symbol}_{period or f'{args.start}_{args.end}'}_{currency}"
    if use_cache:
        cached = cache.get_historical(cache_key)
        if cached:
            results = cached
        else:
            results = None
    else:
        results = None

    # Fetch if not cached
    if results is None:
        try:
            results = client.get_historical_prices(
                symbol,
                currency=currency,
                period=period,
                start_date=start_date,
                end_date=end_date
            )

            if use_cache:
                cache.set_historical(cache_key, results)

        except APIError as e:
            print(f"Error: {e}", file=sys.stderr)
            sys.exit(1)

    # Output results
    output_text = formatter.format_historical(
        symbol,
        results,
        format_type=args.format
    )

    if args.output:
        output_path = Path(args.output)
        if args.format == "csv" and not output_path.suffix:
            output_path = output_path.with_suffix(".csv")
        output_path.parent.mkdir(parents=True, exist_ok=True)
        with open(output_path, "w") as f:
            f.write(output_text)
        if not args.quiet:
            print(f"Output written to {output_path}")
    else:
        print(output_text)


def main() -> None:
    """Main entry point."""
    args = parse_args()
    config = load_config()

    # Initialize components
    cache_dir = Path(config.get("cache", {}).get("directory",
                     str(SCRIPT_DIR.parent / "data")))
    cache = CacheManager(
        cache_dir=cache_dir,
        spot_ttl=config.get("cache", {}).get("spot_ttl", 30),
        historical_ttl=config.get("cache", {}).get("historical_ttl", 3600)
    )

    client = CryptoAPIClient(config=config)
    formatter = PriceFormatter(currency=args.currency.upper())

    # Handle cache operations
    if args.clear_cache:
        cache.clear()
        if not args.quiet:
            print("Cache cleared")
        return

    # Handle list command
    if args.list:
        handle_list_command(client, args.query, formatter, args)
        return

    # Determine symbols to fetch
    symbols = []
    if args.symbol:
        symbols = [args.symbol]
    elif args.symbols:
        symbols = [s.strip() for s in args.symbols.split(",")]
    elif args.watchlist:
        symbols = get_watchlist_symbols(args.watchlist, config)
    else:
        # Default: show help
        print("Error: Specify --symbol, --symbols, --watchlist, or --list",
              file=sys.stderr)
        print("Use --help for usage information", file=sys.stderr)
        sys.exit(1)

    # Handle historical vs current prices
    if args.period or args.start:
        if len(symbols) > 1:
            print("Error: Historical data supports single symbol only",
                  file=sys.stderr)
            sys.exit(1)
        handle_historical_command(
            symbols[0], client, cache, formatter, args, config
        )
    else:
        handle_price_command(
            symbols, client, cache, formatter, args, config
        )


if __name__ == "__main__":
    main()