trumpsignal-backend/app/services/kol_onchain.py

"""KOL A-tier: daily on-chain holdings snapshot + change detection.

Data sources (all free):

  Hyperliquid public API  — No key. Perp positions + mark prices for any
                            HL-listed asset (BTC/ETH/SOL/HYPE/...).
                            Endpoint: POST /info {"type":"clearinghouseState"}
                            Prices:   POST /info {"type":"allMids"}

  Etherscan API           — Free key (etherscan.io/register, email only).
                            ERC-20 token balances for any Ethereum address.
                            Env: ETHERSCAN_API_KEY. Skip ETH wallets if unset.

  CoinGecko (no key)      — Fallback price for tokens not listed on HL.
                            Free tier: 30 req/min, batch by contract address.

Pricing priority per token:
  1. HL allMids (by symbol, e.g. "ETH" → $2025)
  2. CoinGecko by contract address (Ethereum only)
  3. Skip (price unknown → usd_value=0, excluded from snapshot)

Daily flow (run_onchain_poll at 02:00 UTC):
  For each HL wallet  → fetch HL positions → snapshot → diff
  For each ETH wallet → fetch Etherscan ERC-20 list → price each token → snapshot → diff
  Diffs with usd_after > $50k (new) or ±25% change written to kol_holding_changes.
"""

from __future__ import annotations

import asyncio
import json
import logging
import time
from datetime import datetime, timezone
from typing import Optional

import httpx
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.database import AsyncSessionLocal
from app.models import KolHoldingChange, KolHoldingSnapshot, KolWallet, utcnow

logger = logging.getLogger(__name__)

HL_API_URL        = "https://api.hyperliquid.xyz/info"
ETHERSCAN_API_URL = "https://api.etherscan.io/v2/api"
COINGECKO_URL     = "https://api.coingecko.com/api/v3"

# Stablecoins — ignored in all snapshots
STABLES = {"USDC", "USDT", "DAI", "BUSD", "TUSD", "USDE", "FRAX", "LUSD", "CRVUSD"}

# Change detection thresholds
_NEW_POSITION_MIN_USD  = 50_000   # new token > $50k → alert
_CHANGE_PCT_THRESHOLD  = 25.0     # ±25% USD move → alert
_CLOSED_MIN_USD        = 10_000   # closed position must have been > $10k to report


# ── Price layer ───────────────────────────────────────────────────────────────

_hl_prices: dict[str, float] = {}          # symbol → USD, refreshed each poll
_hl_prices_fetched_at: float = 0.0

async def _get_hl_prices() -> dict[str, float]:
    """Fetch all HL mark prices (free, no auth). Cached for 5 min per poll cycle.
    Returns empty dict on network failure — callers fall back to CoinGecko."""
    global _hl_prices, _hl_prices_fetched_at
    now = time.time()
    if now - _hl_prices_fetched_at < 300 and _hl_prices:
        return _hl_prices
    try:
        async with httpx.AsyncClient(timeout=10.0) as c:
            r = await c.post(HL_API_URL, json={"type": "allMids"})
            r.raise_for_status()
            raw = r.json()  # {"BTC": "74541.0", "ETH": "2025.0", ...}
        _hl_prices = {sym: float(px) for sym, px in raw.items()}
        _hl_prices_fetched_at = now
        logger.info("[kol_onchain] HL prices loaded: %d assets", len(_hl_prices))
    except Exception as e:
        logger.warning("[kol_onchain] HL price fetch failed (%s) — falling back to CoinGecko", e)
        # Don't overwrite a previous good cache on transient failure
        if not _hl_prices:
            _hl_prices = {}
    return _hl_prices


# CoinGecko symbol→id map for common crypto assets (fallback when HL unavailable)
_CG_SYMBOL_IDS = {
    "BTC": "bitcoin", "ETH": "ethereum", "SOL": "solana",
    "BNB": "binancecoin", "AVAX": "avalanche-2", "MATIC": "matic-network",
    "ARB": "arbitrum", "OP": "optimism", "LINK": "chainlink",
    "UNI": "uniswap", "AAVE": "aave", "MKR": "maker",
    "HYPE": "hyperliquid", "ENA": "ethena", "PENDLE": "pendle",
    "WLD": "worldcoin-wld", "JUP": "jupiter-exchange-solana",
    "WBTC": "wrapped-bitcoin", "STETH": "staked-ether",
    "EETH": "ether-fi-staked-eth", "WEETH": "wrapped-eeth",
}

async def _coingecko_prices_by_symbol(symbols: list[str]) -> dict[str, float]:
    """Batch price lookup by CoinGecko coin ID. No key needed."""
    ids_needed = {s: _CG_SYMBOL_IDS[s] for s in symbols if s in _CG_SYMBOL_IDS}
    if not ids_needed:
        return {}
    try:
        async with httpx.AsyncClient(timeout=15.0) as c:
            r = await c.get(f"{COINGECKO_URL}/simple/price", params={
                "ids": ",".join(ids_needed.values()),
                "vs_currencies": "usd",
            })
            if r.status_code != 200:
                return {}
            data = r.json()
        # Invert: coin_id → price, then map back to symbol
        id_to_price = {v: data.get(v, {}).get("usd", 0) for v in ids_needed.values()}
        return {sym: id_to_price[cid] for sym, cid in ids_needed.items() if id_to_price.get(cid)}
    except Exception as e:
        logger.warning("[kol_onchain] CoinGecko symbol lookup failed: %s", e)
        return {}


async def _coingecko_prices_by_contract(
    contract_addresses: list[str],
) -> dict[str, float]:
    """Batch price lookup by Ethereum contract address. No key needed.
    Chunks into batches of 25 (CoinGecko free tier limit).
    Returns {contract_addr_lower: usd_price}."""
    if not contract_addresses:
        return {}
    result: dict[str, float] = {}
    addrs_lower = [a.lower() for a in contract_addresses[:100]]
    # CoinGecko free tier caps at ~25 addresses per request
    chunk_size = 25
    async with httpx.AsyncClient(timeout=15.0) as c:
        for i in range(0, len(addrs_lower), chunk_size):
            chunk = addrs_lower[i:i + chunk_size]
            try:
                r = await c.get(
                    f"{COINGECKO_URL}/simple/token_price/ethereum",
                    params={"contract_addresses": ",".join(chunk), "vs_currencies": "usd"},
                )
                if r.status_code != 200:
                    logger.warning("[kol_onchain] CoinGecko %s: %s", r.status_code, r.text[:80])
                    continue
                data = r.json()
                result.update({addr: info["usd"] for addr, info in data.items() if "usd" in info})
            except Exception as e:
                logger.warning("[kol_onchain] CoinGecko chunk failed: %s", e)
    return result


# ── Hyperliquid: perp positions ───────────────────────────────────────────────

async def _fetch_hl_positions(address: str) -> tuple[list[dict], float]:
    """Query HL clearinghouseState. Returns (holdings, account_value_usd).
    holdings: [{ticker, amount, usd_value, chain, side}]
    """
    async with httpx.AsyncClient(timeout=15.0) as c:
        r = await c.post(HL_API_URL, json={"type": "clearinghouseState", "user": address})
        r.raise_for_status()
        data = r.json()

    prices = await _get_hl_prices()
    holdings = []
    for pos in data.get("assetPositions", []):
        p = pos.get("position", {})
        coin = p.get("coin", "")
        if not coin:
            continue
        size = float(p.get("szi", 0))
        if abs(size) < 1e-8:
            continue
        price = prices.get(coin, float(p.get("entryPx") or 0))
        notional = abs(size) * price
        if notional < 1:
            continue
        holdings.append({
            "ticker": coin,
            "amount": abs(size),
            "usd_value": round(notional, 2),
            "chain": "hl",
            "side": "long" if size > 0 else "short",
        })

    margin = data.get("marginSummary", {})
    account_value = float(margin.get("accountValue", 0))
    return holdings, account_value


# ── Etherscan: ERC-20 balances ────────────────────────────────────────────────

async def _fetch_eth_holdings(address: str) -> tuple[list[dict], float]:
    """Fetch ERC-20 + ETH holdings via Etherscan free tier.

    Free-tier strategy (no Pro needed):
      Step 1: GET account/balance          → native ETH amount
      Step 2: GET account/tokentx (recent) → discover which ERC-20 contracts
                                             the wallet has interacted with
      Step 3: GET account/tokenbalance     → current balance per contract
      Step 4: Price via HL allMids, then CoinGecko by contract address fallback

    Etherscan free limit: 5 req/s, ~100k req/day — well within budget for
    daily snapshots of a handful of KOL wallets.
    """
    if not settings.etherscan_api_key:
        return [], 0.0

    prices = await _get_hl_prices()
    key = settings.etherscan_api_key

    async with httpx.AsyncClient(timeout=20.0) as c:

        # ── Step 1: ETH native balance ───────────────────────────────────
        eth_r = await c.get(ETHERSCAN_API_URL, params={
            "chainid": "1", "module": "account", "action": "balance",
            "address": address, "tag": "latest", "apikey": key,
        })
        eth_data = eth_r.json()

        # ── Step 2: ERC-20 tx history → unique (contract, symbol, decimals) ─
        # offset=200 is enough to cover all tokens a KOL holds. Vitalik-scale
        # addresses with 100k+ txs can still timeout — real KOL wallets are far smaller.
        tx_r = await c.get(ETHERSCAN_API_URL, params={
            "chainid": "1", "module": "account", "action": "tokentx",
            "address": address, "page": "1", "offset": "200",
            "sort": "desc", "apikey": key,
        }, timeout=30.0)
        tx_data = tx_r.json()

    # Collect unique contracts from tx history
    seen: dict[str, dict] = {}   # contract → {symbol, decimals}
    txs = tx_data.get("result") or []
    if isinstance(txs, list):
        for tx in txs:
            contract = (tx.get("contractAddress") or "").lower()
            if not contract or contract in seen:
                continue
            symbol = (tx.get("tokenSymbol") or "").upper()
            if not symbol or symbol in STABLES:
                continue
            try:
                decimals = int(tx.get("tokenDecimal") or 18)
            except ValueError:
                decimals = 18
            seen[contract] = {"symbol": symbol, "decimals": decimals}

    # ── Step 3: current balance per contract ─────────────────────────────
    holdings: list[dict] = []
    unpriced: list[str] = []          # contracts needing CoinGecko
    contract_meta: dict[str, dict] = {}

    # Rate-limit: Etherscan free = 5 req/s. Sleep 0.22s between calls to stay
    # under burst limit even when looping 80 tokens × 4 wallets in one poll.
    async with httpx.AsyncClient(timeout=20.0) as c:
        for contract, meta in list(seen.items())[:80]:   # cap at 80 tokens
            await asyncio.sleep(0.22)
            bal_r = await c.get(ETHERSCAN_API_URL, params={
                "chainid": "1", "module": "account", "action": "tokenbalance",
                "contractaddress": contract, "address": address,
                "tag": "latest", "apikey": key,
            })
            bal_data = bal_r.json()
            if bal_data.get("message") != "OK":
                continue
            raw = int(bal_data.get("result") or 0)
            if raw == 0:
                continue
            balance = raw / (10 ** meta["decimals"])
            symbol = meta["symbol"]

            hl_price = prices.get(symbol)
            if hl_price and hl_price > 0:
                usd_value = round(balance * hl_price, 2)
                if usd_value >= 500:
                    holdings.append({
                        "ticker": symbol, "amount": round(balance, 6),
                        "usd_value": usd_value, "chain": "ethereum",
                        "contract": contract,
                    })
            else:
                unpriced.append(contract)
                contract_meta[contract] = {
                    "ticker": symbol, "amount": round(balance, 6),
                    "chain": "ethereum", "contract": contract,
                }

    # ── Step 4a: ETH native ──────────────────────────────────────────────
    eth_raw = int(eth_data.get("result") or 0)
    eth_balance = eth_raw / 1e18
    if eth_balance > 0.001:
        eth_price = prices.get("ETH", 0.0)
        # Fallback: CoinGecko by symbol if HL unavailable
        if not eth_price:
            cg_sym = await _coingecko_prices_by_symbol(["ETH"])
            eth_price = cg_sym.get("ETH", 0.0)
        eth_usd = round(eth_balance * eth_price, 2)
        if eth_usd >= 100:
            holdings.append({
                "ticker": "ETH", "amount": round(eth_balance, 6),
                "usd_value": eth_usd, "chain": "ethereum",
            })

    # ── Step 4b: CoinGecko fallback for unpriced tokens ──────────────────
    if unpriced:
        try:
            cg_prices = await _coingecko_prices_by_contract(unpriced)
            for contract, price in cg_prices.items():
                meta = contract_meta.get(contract)
                if not meta or price <= 0:
                    continue
                usd_value = round(meta["amount"] * price, 2)
                if usd_value >= 500:
                    holdings.append({**meta, "usd_value": usd_value})
        except Exception as e:
            logger.warning("[kol_onchain] CoinGecko fallback failed: %s", e)

    # Merge duplicate tickers (different contracts, same symbol) by summing USD value
    merged: dict[str, dict] = {}
    for h in holdings:
        t = h["ticker"]
        if t in merged:
            merged[t]["usd_value"] = round(merged[t]["usd_value"] + h["usd_value"], 2)
            merged[t]["amount"]    = round(merged[t]["amount"]    + h["amount"],    6)
        else:
            merged[t] = dict(h)
    holdings = list(merged.values())

    total_usd = round(sum(h["usd_value"] for h in holdings), 2)
    logger.info("[kol_onchain] %s ETH holdings: %d tokens, $%.0f total",
                address[:10], len(holdings), total_usd)
    return holdings, total_usd


# ── Snapshot + diff ───────────────────────────────────────────────────────────

def _diff_holdings(
    prev: list[dict], curr: list[dict], wallet_id: int,
) -> list[KolHoldingChange]:
    # SUM by ticker — earlier snapshots may contain duplicate ticker rows
    # (different contracts, same symbol e.g. two APE tokens). Dict comprehension
    # would silently drop one and falsely flag +1900% diffs the next day.
    def _sum_by_ticker(rows: list[dict]) -> dict[str, float]:
        agg: dict[str, float] = {}
        for h in rows:
            t = h.get("ticker")
            if not t:
                continue
            agg[t] = agg.get(t, 0.0) + float(h.get("usd_value") or 0)
        return agg

    prev_map = _sum_by_ticker(prev)
    curr_map = _sum_by_ticker(curr)
    now = utcnow()
    changes = []

    for ticker in set(prev_map) | set(curr_map):
        before = prev_map.get(ticker, 0.0)
        after  = curr_map.get(ticker, 0.0)

        if before == 0 and after >= _NEW_POSITION_MIN_USD:
            changes.append(KolHoldingChange(
                wallet_id=wallet_id, detected_at=now, ticker=ticker,
                change_type="new_position", usd_before=0, usd_after=after,
            ))
        elif after == 0 and before >= _CLOSED_MIN_USD:
            changes.append(KolHoldingChange(
                wallet_id=wallet_id, detected_at=now, ticker=ticker,
                change_type="closed", usd_before=before, usd_after=0,
                pct_change=-100.0,
            ))
        elif before > 0 and after > 0:
            pct = (after - before) / before * 100
            if abs(pct) >= _CHANGE_PCT_THRESHOLD:
                changes.append(KolHoldingChange(
                    wallet_id=wallet_id, detected_at=now, ticker=ticker,
                    change_type="increased" if pct > 0 else "decreased",
                    usd_before=before, usd_after=after, pct_change=round(pct, 1),
                ))
    return changes


async def _snapshot_wallet(
    session: AsyncSession,
    wallet: KolWallet,
    holdings: list[dict],
    total_usd: float,
    source: str,
) -> dict:
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")

    # Skip if already snapshotted today
    existing = (await session.execute(
        select(KolHoldingSnapshot).where(
            KolHoldingSnapshot.wallet_id == wallet.id,
            KolHoldingSnapshot.snapshot_date == today,
        )
    )).scalar_one_or_none()
    if existing:
        return {"wallet_id": wallet.id, "status": "already_snapshotted"}

    # Previous snapshot for diff
    prev_row = (await session.execute(
        select(KolHoldingSnapshot)
        .where(KolHoldingSnapshot.wallet_id == wallet.id)
        .order_by(KolHoldingSnapshot.snapshot_date.desc())
        .limit(1)
    )).scalar_one_or_none()
    prev_holdings = json.loads(prev_row.holdings_json) if prev_row else []

    # Store snapshot (strip contract addresses to keep JSON lean)
    clean_holdings = [{k: v for k, v in h.items() if k != "contract"} for h in holdings]
    snap = KolHoldingSnapshot(
        wallet_id=wallet.id,
        snapshot_date=today,
        holdings_json=json.dumps(clean_holdings, ensure_ascii=False),
        total_usd=total_usd,
        source=source,
    )
    session.add(snap)

    # Detect and store changes
    changes = _diff_holdings(prev_holdings, holdings, wallet.id)
    for c in changes:
        session.add(c)
        logger.info(
            "[kol_onchain] %s %s %s $%.0f→$%.0f",
            wallet.handle, c.change_type, c.ticker,
            c.usd_before or 0, c.usd_after or 0,
        )

    await session.flush()
    return {
        "handle": wallet.handle,
        "chain": wallet.chain,
        "holdings": len(holdings),
        "total_usd": total_usd,
        "changes": len(changes),
        "source": source,
    }


# ── Poll entry points ─────────────────────────────────────────────────────────

async def run_hl_poll() -> list[dict]:
    """Poll HL perp positions for all chain='hl' wallets."""
    results = []
    async with AsyncSessionLocal() as session:
        wallets = (await session.execute(
            select(KolWallet).where(KolWallet.chain == "hl", KolWallet.active == True)
        )).scalars().all()

        if not wallets:
            logger.info("[kol_onchain] no HL wallets configured")
            return []

        for wallet in wallets:
            try:
                holdings, total = await _fetch_hl_positions(wallet.address)
                stat = await _snapshot_wallet(session, wallet, holdings, total, "hl")
                results.append(stat)
            except Exception as e:
                logger.warning("[kol_onchain] HL fetch failed %s: %s", wallet.handle, e)
                results.append({"handle": wallet.handle, "error": str(e)})

        await session.commit()
    logger.info("[kol_onchain] HL poll done: %s", results)
    return results


async def run_eth_poll() -> list[dict]:
    """Poll Ethereum ERC-20 holdings for all chain='ethereum' wallets.
    No-op if ETHERSCAN_API_KEY not set."""
    if not settings.etherscan_api_key:
        logger.debug("[kol_onchain] ETHERSCAN_API_KEY not set — skipping ETH poll")
        return []

    results = []
    async with AsyncSessionLocal() as session:
        wallets = (await session.execute(
            select(KolWallet).where(KolWallet.chain == "ethereum", KolWallet.active == True)
        )).scalars().all()

        if not wallets:
            logger.info("[kol_onchain] no Ethereum wallets configured")
            return []

        # Pre-warm HL prices once for the whole batch
        await _get_hl_prices()

        for wallet in wallets:
            try:
                holdings, total = await _fetch_eth_holdings(wallet.address)
                stat = await _snapshot_wallet(session, wallet, holdings, total, "etherscan")
                results.append(stat)
            except Exception as e:
                logger.warning("[kol_onchain] ETH fetch failed %s: %s", wallet.handle, e)
                results.append({"handle": wallet.handle, "error": str(e)})

        await session.commit()
    logger.info("[kol_onchain] ETH poll done: %s", results)
    return results


async def run_onchain_poll() -> dict:
    """Combined entry point for APScheduler. Runs HL + ETH polls."""
    hl  = await run_hl_poll()
    eth = await run_eth_poll()
    return {"hl": hl, "eth": eth}


async def seed_wallets(entries: list[tuple]) -> int:
    """Bulk-insert (handle, chain, address, label, source_url) tuples, skip dupes."""
    count = 0
    async with AsyncSessionLocal() as session:
        for handle, chain, address, label, source_url in entries:
            existing = (await session.execute(
                select(KolWallet).where(
                    KolWallet.chain == chain,
                    KolWallet.address == address.lower(),
                )
            )).scalar_one_or_none()
            if existing:
                continue
            session.add(KolWallet(
                handle=handle, chain=chain,
                address=address.lower(),
                label=label, source_url=source_url,
            ))
            count += 1
        await session.commit()
    return count