from __future__ import annotations import re import threading import time from datetime import date, datetime, timezone from decimal import Decimal, InvalidOperation from html import unescape from typing import Any import requests class CoinResearchError(RuntimeError): pass def _decimal(value: object) -> Decimal: try: number = Decimal(str(value)) except (InvalidOperation, TypeError, ValueError): return Decimal("0") return number if number.is_finite() else Decimal("0") def _plain_text(value: object, limit: int = 700) -> str: text = re.sub(r"<[^>]+>", " ", unescape(str(value or ""))) text = re.sub(r"\s+", " ", text).strip() return text[:limit] def _usd(value: object) -> str | None: number = _decimal(value) return format(number, ".2f") if number > 0 else None def _percent(value: object) -> str | None: number = _decimal(value) return format(number, ".2f") if number != 0 else None def evaluate_market_profile( details: dict[str, Any], *, symbol: str, binance_ticker: dict[str, Any], today: date | None = None, ) -> dict[str, Any]: market = details.get("market_data") if isinstance(details.get("market_data"), dict) else {} market_cap = _decimal((market.get("market_cap") or {}).get("usd")) global_volume = _decimal((market.get("total_volume") or {}).get("usd")) binance_volume = _decimal(binance_ticker.get("quoteVolume")) change_24h = _decimal(binance_ticker.get("priceChangePercent")) change_7d = _decimal((market.get("price_change_percentage_7d_in_currency") or {}).get("usd")) change_30d = _decimal((market.get("price_change_percentage_30d_in_currency") or {}).get("usd")) genesis = str(details.get("genesis_date") or "").strip() or None age_days: int | None = None if genesis: try: age_days = max(0, ((today or datetime.now(timezone.utc).date()) - date.fromisoformat(genesis)).days) except ValueError: genesis = None score = 45 strengths: list[str] = [] risks: list[str] = [] if market_cap >= Decimal("1000000000"): score += 20 strengths.append("large_market_cap") elif market_cap >= Decimal("100000000"): score += 12 strengths.append("established_market_cap") elif market_cap and market_cap < Decimal("20000000"): score -= 15 risks.append("small_market_cap") effective_volume = max(global_volume, binance_volume) if effective_volume >= Decimal("50000000"): score += 16 strengths.append("strong_volume") elif effective_volume >= Decimal("10000000"): score += 9 strengths.append("adequate_volume") elif effective_volume < Decimal("1000000"): score -= 18 risks.append("weak_volume") liquidity_ratio = global_volume / market_cap if market_cap > 0 else Decimal("0") if liquidity_ratio >= Decimal("0.03"): score += 8 strengths.append("healthy_turnover") elif market_cap > 0 and liquidity_ratio < Decimal("0.005"): score -= 8 risks.append("low_turnover") if age_days is None: risks.append("unknown_age") elif age_days >= 730: score += 8 strengths.append("established_history") elif age_days >= 180: score += 4 elif age_days < 60: score -= 12 risks.append("very_young") if abs(change_24h) > Decimal("15") or abs(change_7d) > Decimal("45"): score -= 14 risks.append("high_volatility") stability = "HIGH_VOLATILITY" elif abs(change_24h) <= Decimal("5") and abs(change_7d) <= Decimal("18"): score += 5 strengths.append("moderate_price_behavior") stability = "RELATIVELY_STABLE" else: stability = "MEDIUM_VOLATILITY" if market_cap <= 0 or effective_volume <= 0: score -= 12 risks.append("incomplete_market_data") score = max(0, min(100, score)) verdict = "ACCEPTED" if score >= 72 and stability != "HIGH_VOLATILITY" else ( "CAUTION" if score >= 52 else "AVOID" ) links = details.get("links") if isinstance(details.get("links"), dict) else {} homepages = links.get("homepage") if isinstance(links.get("homepage"), list) else [] homepage = next((str(item) for item in homepages if str(item).startswith("http")), None) categories = [str(item) for item in details.get("categories", []) if str(item).strip()][:6] description = details.get("description") if isinstance(details.get("description"), dict) else {} return { "symbol": symbol, "name": str(details.get("name") or symbol[:-4]), "provider_id": str(details.get("id") or ""), "source": "CoinGecko and Binance public market data", "source_url": f"https://www.coingecko.com/en/coins/{details.get('id')}", "homepage": homepage, "description": _plain_text(description.get("en")), "categories": categories, "genesis_date": genesis, "age_days": age_days, "market_cap_rank": details.get("market_cap_rank"), "market_cap_usd": _usd(market_cap), "global_volume_24h_usd": _usd(global_volume), "binance_volume_24h_usd": _usd(binance_volume), "circulating_supply": _usd(market.get("circulating_supply")), "total_supply": _usd(market.get("total_supply")), "max_supply": _usd(market.get("max_supply")), "change_24h_percent": _percent(change_24h), "change_7d_percent": _percent(change_7d), "change_30d_percent": _percent(change_30d), "liquidity_ratio_percent": format(liquidity_ratio * Decimal("100"), ".2f") if liquidity_ratio > 0 else None, "score": score, "verdict": verdict, "stability": stability, "strengths": strengths, "risks": list(dict.fromkeys(risks)), "updated_at": datetime.now(timezone.utc).isoformat(), } class CoinResearchService: endpoint = "https://api.coingecko.com/api/v3" def __init__(self, session: requests.Session | None = None, cache_seconds: int = 3600) -> None: self._session = session or requests.Session() self._cache_seconds = max(300, int(cache_seconds)) self._cache: dict[str, tuple[float, dict[str, Any]]] = {} self._lock = threading.Lock() def _get(self, path: str, params: dict[str, str]) -> Any: try: response = self._session.get( f"{self.endpoint}{path}", params=params, timeout=12, headers={"Accept": "application/json", "User-Agent": "BinanceSpotAssistant/1.0 read-only"}, ) response.raise_for_status() return response.json() except (requests.RequestException, ValueError) as exc: raise CoinResearchError("Public project information is unavailable right now.") from exc def profile(self, symbol: str, binance_ticker: dict[str, Any]) -> dict[str, Any]: clean_symbol = str(symbol).strip().upper() base_asset = clean_symbol[:-4] cache_key = clean_symbol with self._lock: cached = self._cache.get(cache_key) if cached and time.time() - cached[0] < self._cache_seconds: return dict(cached[1]) search = self._get("/search", {"query": base_asset}) coins = search.get("coins", []) if isinstance(search, dict) else [] exact = [item for item in coins if str(item.get("symbol", "")).upper() == base_asset] if not exact: raise CoinResearchError("The project could not be matched safely to a public profile.") exact.sort(key=lambda item: item.get("market_cap_rank") or 10**9) coin_id = str(exact[0].get("id") or "") if not coin_id: raise CoinResearchError("The project profile is incomplete.") details = self._get( f"/coins/{coin_id}", { "localization": "false", "tickers": "false", "market_data": "true", "community_data": "false", "developer_data": "false", "sparkline": "false", }, ) if not isinstance(details, dict): raise CoinResearchError("The project profile is invalid.") result = evaluate_market_profile(details, symbol=clean_symbol, binance_ticker=binance_ticker) with self._lock: self._cache[cache_key] = (time.time(), dict(result)) return result