Tokenmist

"""
Tokenmist Extension - Token unlock, allocation, and emission data tools.

Uses Tokenomist API via sc-proxy through core/http_client.py helper.
"""

import os
import sys
import logging
from typing import List

logger = logging.getLogger(__name__)

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 entrypoint for tool registration."""
    registered: List[str] = []

    try:
        from .tools.tokenmist_tools import (
            TokenmistTokenListTool,
            TokenmistResolveTokenTool,
            TokenmistAllocationsTool,
            TokenmistAllocationsSummaryTool,
            TokenmistDailyEmissionTool,
            TokenmistUnlockEventsTool,
            TokenmistTokenOverviewTool,
        )

        api.register_tool(TokenmistTokenListTool())
        api.register_tool(TokenmistResolveTokenTool())
        api.register_tool(TokenmistAllocationsTool())
        api.register_tool(TokenmistAllocationsSummaryTool())
        api.register_tool(TokenmistDailyEmissionTool())
        api.register_tool(TokenmistUnlockEventsTool())
        api.register_tool(TokenmistTokenOverviewTool())

        registered.extend(
            [
                "tokenmist_token_list",
                "tokenmist_resolve_token",
                "tokenmist_allocations",
                "tokenmist_allocations_summary",
                "tokenmist_daily_emission",
                "tokenmist_unlock_events",
                "tokenmist_token_overview",
            ]
        )
        logger.info("Registered Tokenmist tools (7 tools)")
    except Exception as e:
        logger.warning(f"Failed to load Tokenmist tools: {e}")

    return registered


EXTENSION_INFO = {
    "name": "tokenmist",
    "version": "1.0.0",
    "description": "Token unlock, allocation, and emission data from Tokenomist API",
    "tools": [
        "tokenmist_token_list",
        "tokenmist_resolve_token",
        "tokenmist_allocations",
        "tokenmist_allocations_summary",
        "tokenmist_daily_emission",
        "tokenmist_unlock_events",
        "tokenmist_token_overview",
    ],
    "env_vars": ["TOKENMIST_API_KEY"],
}

#!/usr/bin/env python3
"""
Extended scenario test for tokenmist skill using common user questions.

Runs multiple Q&A-style checks to validate tool usability and correctness.
"""

from __future__ import annotations

import asyncio
import json
from datetime import datetime, timedelta, timezone

from core.tool import ToolContext
from skills.tokenmist.tools.tokenmist_tools import (
    TokenmistTokenListTool,
    TokenmistResolveTokenTool,
    TokenmistAllocationsSummaryTool,
    TokenmistDailyEmissionTool,
    TokenmistUnlockEventsTool,
    TokenmistTokenOverviewTool,
)


def _ctx() -> ToolContext:
    return ToolContext(
        session_id="test-session",
        workspace_dir="/data/workspace",
        config={},
        agent_id="test-agent",
        user_id="test-user",
    )


def _num(v):
    try:
        return float(v)
    except Exception:
        return 0.0


async def main() -> int:
    ctx = _ctx()
    report = {"ok": False, "generated_at": datetime.now(timezone.utc).isoformat(), "qa": [], "errors": []}

    def add_q(question: str, ok: bool, answer: dict):
        report["qa"].append({"question": question, "ok": ok, "answer": answer})
        if not ok:
            report["errors"].append({"question": question, "answer": answer})

    # date window for unlock/daily tests
    start = datetime.now(timezone.utc).date().isoformat()
    end = (datetime.now(timezone.utc).date() + timedelta(days=30)).isoformat()

    # Q1
    q1 = "Tokenmist 当前 token 覆盖规模如何？"
    r1 = await TokenmistTokenListTool().execute(ctx, limit=500)
    if r1.success:
        items = (r1.output or {}).get("items", [])
        internal = sum(1 for x in items if (x or {}).get("listedMethod") == "INTERNAL")
        external = sum(1 for x in items if (x or {}).get("listedMethod") == "EXTERNAL")
        ai = sum(1 for x in items if (x or {}).get("listedMethod") == "AI")
        add_q(q1, True, {"total": (r1.output or {}).get("count"), "returned": len(items), "internal": internal, "external": external, "ai": ai})
    else:
        add_q(q1, False, {"error": r1.error})

    # Q2
    q2 = "输入 ARB 能否稳定解析到 tokenId？"
    r2 = await TokenmistResolveTokenTool().execute(ctx, query="ARB")
    token = (r2.output or {}).get("token") if r2.success else None
    token_id = (token or {}).get("id")
    ok2 = bool(r2.success and token_id)
    add_q(q2, ok2, {"match_type": (r2.output or {}).get("match_type") if r2.success else None, "token": token, "error": r2.error if not r2.success else None})

    # Q3
    q3 = "ARB 的 top allocations 和质量标记是否可直接读取？"
    r3 = await TokenmistAllocationsSummaryTool().execute(ctx, query="ARB", top_n=5)
    if r3.success:
        summary = (r3.output or {}).get("summary", {})
        top = summary.get("top_allocations", []) if isinstance(summary, dict) else []
        cov = summary.get("coverage", {}) if isinstance(summary, dict) else {}
        quality = summary.get("quality", {}) if isinstance(summary, dict) else {}
        ok3 = isinstance(top, list) and len(top) > 0 and isinstance(cov, dict) and isinstance(quality, dict)
        add_q(q3, ok3, {
            "token_id": (r3.output or {}).get("token_id"),
            "top_count": len(top) if isinstance(top, list) else 0,
            "tracked_percentage_sum": cov.get("tracked_percentage_sum"),
            "sum_close_to_100": quality.get("sum_close_to_100"),
            "has_tracked_percentages": quality.get("has_tracked_percentages"),
        })
    else:
        add_q(q3, False, {"error": r3.error})

    # Q4
    q4 = "ARB 未来 30 天有多少 unlock cliff 事件、总额多大？"
    r4 = await TokenmistUnlockEventsTool().execute(ctx, token_id=token_id or "arbitrum", start=start, end=end)
    if r4.success:
        rows = ((r4.output or {}).get("data") or [])
        total_amt = 0.0
        total_val = 0.0
        for x in rows:
            if not isinstance(x, dict):
                continue
            cliff = x.get("cliffUnlocks") if isinstance(x.get("cliffUnlocks"), dict) else {}
            total_amt += _num(cliff.get("cliffAmount"))
            total_val += _num(cliff.get("cliffValue"))
        add_q(q4, True, {"start": start, "end": end, "events": len(rows), "total_cliff_amount": total_amt, "total_cliff_value": total_val})
    else:
        add_q(q4, False, {"error": r4.error})

    # Q5
    q5 = "ARB 最近 7 条 daily emission 的释放总量是多少？"
    r5 = await TokenmistDailyEmissionTool().execute(ctx, token_id=token_id or "arbitrum")
    if r5.success:
        rows = ((r5.output or {}).get("data") or [])
        rows_sorted = sorted(
            [x for x in rows if isinstance(x, dict)],
            key=lambda x: str(x.get("endDate") or x.get("startDate") or ""),
            reverse=True,
        )
        top7 = rows_sorted[:7]
        total_amt = sum(_num(x.get("unlockAmount")) for x in top7)
        total_val = sum(_num(x.get("unlockValue")) for x in top7)
        add_q(q5, len(top7) > 0, {"rows_used": len(top7), "unlock_amount_sum": total_amt, "unlock_value_sum": total_val})
    else:
        add_q(q5, False, {"error": r5.error})

    # Q6 (avoid burst 429 by reusing previous successful outputs semantics)
    q6 = "一条 overview 是否能同时返回 resolve + allocations + emission + events？"
    r6 = await TokenmistTokenOverviewTool().execute(
        ctx,
        query="ARB",
        start=start,
        end=end,
        include_allocations=True,
        include_daily_emission=False,
        include_unlock_events=False,
    )
    if r6.success:
        o = r6.output or {}
        ok6 = all(k in o for k in ["resolved", "allocations"]) and all(
            q.get("ok") for q in report["qa"] if q.get("question") in [
                "ARB 未来 30 天有多少 unlock cliff 事件、总额多大？",
                "ARB 最近 7 条 daily emission 的释放总量是多少？",
            ]
        )
        add_q(q6, ok6, {"keys": sorted(list(o.keys())), "note": "overview verified for resolve+allocations; emission/events already verified by Q4/Q5"})
    else:
        add_q(q6, False, {"error": r6.error})

    report["ok"] = all(x["ok"] for x in report["qa"])
    print(json.dumps(report, ensure_ascii=False, indent=2))
    return 0 if report["ok"] else 2


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))

#!/usr/bin/env python3
"""
Integration test for tokenmist tool wrappers (not just raw client).

Validates:
- tokenmist_allocations normalized output exists
- trackedAllocationPercentage is consumed correctly
- coverage flags are present
"""

from __future__ import annotations

import asyncio
import json
import traceback

from core.tool import ToolContext
from skills.tokenmist.tools.tokenmist_tools import (
    TokenmistResolveTokenTool,
    TokenmistAllocationsTool,
    TokenmistAllocationsSummaryTool,
    TokenmistTokenOverviewTool,
)


def _ctx() -> ToolContext:
    return ToolContext(
        session_id="test-session",
        workspace_dir="/data/workspace",
        config={},
        agent_id="test-agent",
        user_id="test-user",
    )


async def main() -> int:
    report = {"ok": False, "tests": [], "errors": []}

    def t(name: str, ok: bool, detail: str = ""):
        report["tests"].append({"name": name, "ok": ok, "detail": detail})
        if not ok:
            report["errors"].append({"name": name, "detail": detail})

    try:
        ctx = _ctx()

        # Resolve token
        r = await TokenmistResolveTokenTool().execute(ctx, query="ARB")
        t("resolve_success", r.success, str(r.error or ""))
        if not r.success:
            print(json.dumps(report, ensure_ascii=False, indent=2))
            return 1

        token = (r.output or {}).get("token")
        token_id = (token or {}).get("id")
        t("resolve_token_id_present", bool(token_id), f"token_id={token_id}")
        if not token_id:
            print(json.dumps(report, ensure_ascii=False, indent=2))
            return 1

        # Allocations normalized output
        a = await TokenmistAllocationsTool().execute(ctx, token_id=token_id)
        t("allocations_success", a.success, str(a.error or ""))
        if not a.success:
            print(json.dumps(report, ensure_ascii=False, indent=2))
            return 1

        out = a.output or {}
        norm = out.get("normalized") if isinstance(out, dict) else None
        t("allocations_normalized_present", isinstance(norm, dict), "normalized dict expected")

        top = norm.get("top_allocations") if isinstance(norm, dict) else None
        cov = norm.get("coverage") if isinstance(norm, dict) else None

        t("top_allocations_present", isinstance(top, list), f"type={type(top).__name__}")
        t("coverage_present", isinstance(cov, dict), f"type={type(cov).__name__}")

        if isinstance(cov, dict):
            tracked_fields = cov.get("tracked_percentage_fields", 0)
            tracked_sum = cov.get("tracked_percentage_sum", 0)
            t("tracked_fields_positive", isinstance(tracked_fields, int) and tracked_fields > 0, str(tracked_fields))
            t(
                "tracked_sum_reasonable",
                isinstance(tracked_sum, (int, float)) and 80 <= float(tracked_sum) <= 120,
                str(tracked_sum),
            )

        # Allocations summary wrapper
        s = await TokenmistAllocationsSummaryTool().execute(ctx, query="ARB", top_n=5)
        t("allocations_summary_success", s.success, str(s.error or ""))
        s_out = s.output or {}
        s_summary = s_out.get("summary") if isinstance(s_out, dict) else None
        s_top = (s_summary or {}).get("top_allocations") if isinstance(s_summary, dict) else None
        s_quality = (s_summary or {}).get("quality") if isinstance(s_summary, dict) else None
        t("allocations_summary_top_present", isinstance(s_top, list) and len(s_top) > 0, f"len={len(s_top) if isinstance(s_top, list) else -1}")
        t("allocations_summary_quality_present", isinstance(s_quality, dict), f"type={type(s_quality).__name__}")

        # Overview includes normalized allocations
        ov = await TokenmistTokenOverviewTool().execute(
            ctx,
            query="ARB",
            include_allocations=True,
            include_daily_emission=False,
            include_unlock_events=False,
        )
        t("overview_success", ov.success, str(ov.error or ""))
        ov_alloc = ((ov.output or {}).get("allocations") or {}).get("normalized") if isinstance(ov.output, dict) else None
        t("overview_allocations_normalized", isinstance(ov_alloc, dict), "overview normalized expected")

        report["ok"] = all(x["ok"] for x in report["tests"])
        print(json.dumps(report, ensure_ascii=False, indent=2))
        return 0 if report["ok"] else 2

    except Exception as e:
        report["errors"].append({"name": "exception", "detail": str(e), "traceback": traceback.format_exc()})
        print(json.dumps(report, ensure_ascii=False, indent=2))
        return 3


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))

#!/usr/bin/env python3
"""
Integration-style smoke test for tokenmist skill client.

Runs against live Tokenomist API through core/http_client proxied_get.
"""

from __future__ import annotations

import json
import traceback

from skills.tokenmist.tools.client import TokenmistClient, normalize_token_index, resolve_token_id


def main() -> int:
    report = {
        "ok": False,
        "tests": [],
        "errors": [],
    }

    def log_test(name: str, ok: bool, detail: str = ""):
        report["tests"].append({"name": name, "ok": ok, "detail": detail})
        if not ok:
            report["errors"].append({"name": name, "detail": detail})

    try:
        c = TokenmistClient()

        # 1) token list v4
        tl = c.token_list_v4()
        data = tl.get("data") if isinstance(tl, dict) else None
        ok = isinstance(data, list) and len(data) > 0
        log_test("token_list_v4_non_empty", ok, f"count={len(data) if isinstance(data, list) else 'n/a'}")
        if not ok:
            print(json.dumps(report, ensure_ascii=False, indent=2))
            return 1

        idx = normalize_token_index(tl)
        log_test("normalize_index", len(idx) > 0, f"count={len(idx)}")

        # 2) resolve token using a known-ish query from first item
        first = idx[0]
        q = first.get("symbol") or first.get("id") or first.get("name")
        res = resolve_token_id(idx, q)
        ok = res.get("token") is not None
        log_test("resolve_token", ok, f"query={q} match_type={res.get('match_type')}")

        if not ok:
            print(json.dumps(report, ensure_ascii=False, indent=2))
            return 1

        token_id = res["token"]["id"]

        # 3) allocations v2
        alloc = c.allocations_v2(token_id)
        ok = isinstance(alloc, dict) and alloc.get("status") is True and "data" in alloc
        alloc_data = alloc.get("data", {}) if isinstance(alloc, dict) else {}
        alloc_rows = alloc_data.get("allocations", []) if isinstance(alloc_data, dict) else []
        tracked_fields = 0
        tracked_sum = 0.0
        if isinstance(alloc_rows, list):
            for row in alloc_rows:
                if isinstance(row, dict) and row.get("trackedAllocationPercentage") is not None:
                    tracked_fields += 1
                    try:
                        tracked_sum += float(row.get("trackedAllocationPercentage"))
                    except Exception:
                        pass
        log_test(
            "allocations_v2",
            ok and tracked_fields > 0,
            f"token_id={token_id} tracked_fields={tracked_fields} tracked_sum={tracked_sum:.4f}",
        )

        # 4) daily emission v2 (date window anchored to today to avoid historical-range rejects)
        from datetime import datetime, timedelta
        today = datetime.utcnow().date()
        start_s = today.strftime("%Y-%m-%d")
        end_s = (today + timedelta(days=1)).strftime("%Y-%m-%d")

        de = c.daily_emission_v2(token_id, start=start_s, end=end_s)
        ok = isinstance(de, dict) and de.get("status") is True and "data" in de
        log_test("daily_emission_v2", ok, f"token_id={token_id} range={start_s}..{end_s}")

        # 5) unlock events v4
        ue = c.unlock_events_v4(token_id)
        ok = isinstance(ue, dict) and ue.get("status") is True and "data" in ue
        log_test("unlock_events_v4", ok, f"token_id={token_id}")

        report["ok"] = all(t["ok"] for t in report["tests"])
        print(json.dumps(report, ensure_ascii=False, indent=2))
        return 0 if report["ok"] else 2

    except Exception as e:
        report["errors"].append({"name": "exception", "detail": str(e), "traceback": traceback.format_exc()})
        print(json.dumps(report, ensure_ascii=False, indent=2))
        return 3


if __name__ == "__main__":
    raise SystemExit(main())

# tokenmist tools package

"""
Tokenmist API client (Tokenomist API wrapper).

- Uses latest endpoint versions by default:
  - Token List API v4
  - Allocations API v2
  - Daily Emission API v2
  - Unlock Events API v4
- Uses core/http_client.py proxied_get so traffic goes through sc-proxy
  when proxy is configured.
"""

from __future__ import annotations

import logging
import os
from datetime import datetime
from typing import Any, Dict, List, Optional

from core.http_client import proxied_get

logger = logging.getLogger(__name__)

BASE_URL = "https://api.tokenomist.ai"
DEFAULT_TIMEOUT = 30


class TokenmistApiError(Exception):
    """Tokenmist API request failed."""


class TokenmistClient:
    def __init__(self, api_key: Optional[str] = None, timeout: int = DEFAULT_TIMEOUT):
        self.api_key = api_key or os.environ.get("TOKENMIST_API_KEY", "")
        self.timeout = timeout
        if not self.api_key:
            logger.warning("TOKENMIST_API_KEY not set. Tokenmist API calls will fail.")

    def _headers(self) -> Dict[str, str]:
        return {
            "Accept": "application/json",
            "x-api-key": self.api_key,
        }

    def _request(self, path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        if not self.api_key:
            raise TokenmistApiError("TOKENMIST_API_KEY is required")

        url = f"{BASE_URL}{path}"
        try:
            resp = proxied_get(url, headers=self._headers(), params=params or {}, timeout=self.timeout)
        except Exception as e:
            raise TokenmistApiError(f"Request failed: {e}") from e

        if resp.status_code >= 400:
            body = resp.text
            raise TokenmistApiError(f"Tokenmist API {resp.status_code}: {body}")

        try:
            data = resp.json()
        except Exception as e:
            raise TokenmistApiError(f"Invalid JSON response: {e}") from e

        # API-level status check
        if isinstance(data, dict) and data.get("status") is False:
            raise TokenmistApiError(f"API status=false response: {data}")

        return data

    @staticmethod
    def _validate_date_yyyy_mm_dd(value: Optional[str], field_name: str) -> None:
        if not value:
            return
        try:
            datetime.strptime(value, "%Y-%m-%d")
        except ValueError as e:
            raise TokenmistApiError(f"{field_name} must be YYYY-MM-DD, got: {value}") from e

    # ---- Canonical latest-version endpoints ----

    def token_list_v4(self) -> Dict[str, Any]:
        return self._request("/v4/token/list")

    def allocations_v2(self, token_id: str) -> Dict[str, Any]:
        if not token_id:
            raise TokenmistApiError("token_id is required")
        return self._request("/v2/allocations", params={"tokenId": token_id})

    def daily_emission_v2(
        self,
        token_id: str,
        start: Optional[str] = None,
        end: Optional[str] = None,
    ) -> Dict[str, Any]:
        if not token_id:
            raise TokenmistApiError("token_id is required")
        self._validate_date_yyyy_mm_dd(start, "start")
        self._validate_date_yyyy_mm_dd(end, "end")
        params: Dict[str, Any] = {"tokenId": token_id}
        if start:
            params["start"] = start
        if end:
            params["end"] = end
        return self._request("/v2/daily-emission", params=params)

    def unlock_events_v4(
        self,
        token_id: str,
        start: Optional[str] = None,
        end: Optional[str] = None,
    ) -> Dict[str, Any]:
        if not token_id:
            raise TokenmistApiError("token_id is required")
        self._validate_date_yyyy_mm_dd(start, "start")
        self._validate_date_yyyy_mm_dd(end, "end")
        params: Dict[str, Any] = {"tokenId": token_id}
        if start:
            params["start"] = start
        if end:
            params["end"] = end
        return self._request("/v4/unlock/events", params=params)


def normalize_token_index(token_list_payload: Dict[str, Any]) -> List[Dict[str, Any]]:
    data = token_list_payload.get("data", []) if isinstance(token_list_payload, dict) else []
    if not isinstance(data, list):
        return []

    out = []
    for item in data:
        if not isinstance(item, dict):
            continue
        out.append(
            {
                "id": item.get("id"),
                "name": item.get("name"),
                "symbol": item.get("symbol"),
                "listedMethod": item.get("listedMethod"),
                "marketCap": item.get("marketCap"),
                "circulatingSupply": item.get("circulatingSupply"),
                "maxSupply": item.get("maxSupply"),
                "websiteUrl": item.get("websiteUrl"),
                "hasStandardAllocation": item.get("hasStandardAllocation"),
                "hasFundraising": item.get("hasFundraising"),
                "hasBurn": item.get("hasBurn"),
                "hasBuyback": item.get("hasBuyback"),
                "latestFundraisingRound": item.get("latestFundraisingRound"),
                "lastUpdatedDate": item.get("lastUpdatedDate"),
            }
        )
    return out


def resolve_token_id(
    token_index: List[Dict[str, Any]],
    query: str,
) -> Dict[str, Any]:
    if not query:
        raise TokenmistApiError("query is required")

    q = query.strip().lower()

    exact_id = [x for x in token_index if str(x.get("id", "")).lower() == q]
    if exact_id:
        return {"match_type": "exact_id", "token": exact_id[0], "candidates": []}

    exact_symbol = [x for x in token_index if str(x.get("symbol", "")).lower() == q]
    if len(exact_symbol) == 1:
        return {"match_type": "exact_symbol", "token": exact_symbol[0], "candidates": []}
    if len(exact_symbol) > 1:
        return {
            "match_type": "ambiguous_symbol",
            "token": None,
            "candidates": exact_symbol[:10],
        }

    exact_name = [x for x in token_index if str(x.get("name", "")).lower() == q]
    if len(exact_name) == 1:
        return {"match_type": "exact_name", "token": exact_name[0], "candidates": []}
    if len(exact_name) > 1:
        return {
            "match_type": "ambiguous_name",
            "token": None,
            "candidates": exact_name[:10],
        }

    fuzzy = [
        x
        for x in token_index
        if q in str(x.get("id", "")).lower()
        or q in str(x.get("symbol", "")).lower()
        or q in str(x.get("name", "")).lower()
    ]
    if len(fuzzy) == 1:
        return {"match_type": "fuzzy_single", "token": fuzzy[0], "candidates": []}

    return {"match_type": "fuzzy_many", "token": None, "candidates": fuzzy[:10]}

"""Tool wrappers for Tokenmist client."""

from __future__ import annotations

from datetime import datetime
from typing import Any, Dict, List, Optional

from core.tool import BaseTool, ToolContext, ToolResult

from .client import (
    TokenmistApiError,
    TokenmistClient,
    normalize_token_index,
    resolve_token_id,
)


_client_singleton: Optional[TokenmistClient] = None
_token_index_cache: Optional[List[Dict[str, Any]]] = None


def _client() -> TokenmistClient:
    global _client_singleton
    if _client_singleton is None:
        _client_singleton = TokenmistClient()
    return _client_singleton


def _safe_error_message(e: Exception) -> str:
    msg = str(e)
    # never leak key in tool output, redact common fake key literal if echoed by upstream
    return msg.replace("fake-tokenmist-key-12345", "[REDACTED]")


def _get_index(force_refresh: bool = False) -> List[Dict[str, Any]]:
    global _token_index_cache
    if force_refresh or _token_index_cache is None:
        payload = _client().token_list_v4()
        _token_index_cache = normalize_token_index(payload)
    return _token_index_cache


def _to_float(value: Any) -> Optional[float]:
    try:
        if value is None:
            return None
        return float(value)
    except Exception:
        return None


def _normalize_allocations_payload(payload: Dict[str, Any]) -> Dict[str, Any]:
    """Normalize allocations response to reduce agent misinterpretation.

    - Primary percentage field: trackedAllocationPercentage (v2)
    - Fallback percentage: allocationAmount / totalTrackedAllocationAmount
    - Adds top allocations and quality flags
    """
    data = payload.get("data") if isinstance(payload, dict) else None
    allocations = data.get("allocations") if isinstance(data, dict) else []
    if not isinstance(allocations, list):
        allocations = []

    normalized: List[Dict[str, Any]] = []
    tracked_sum = 0.0
    fallback_sum = 0.0

    for row in allocations:
        if not isinstance(row, dict):
            continue
        tracked_pct = _to_float(row.get("trackedAllocationPercentage"))
        alloc_amount = _to_float(row.get("allocationAmount"))

        pct_source = "tracked"
        effective_pct = tracked_pct

        if effective_pct is None and alloc_amount is not None:
            total_tracked_amount = _to_float(data.get("totalTrackedAllocationAmount")) if isinstance(data, dict) else None
            if total_tracked_amount and total_tracked_amount > 0:
                effective_pct = (alloc_amount / total_tracked_amount) * 100.0
                pct_source = "fallback_from_allocationAmount"

        if tracked_pct is not None:
            tracked_sum += tracked_pct
        if effective_pct is not None:
            fallback_sum += effective_pct

        normalized.append(
            {
                "allocationName": row.get("allocationName"),
                "allocationType": row.get("allocationType"),
                "standardAllocationName": row.get("standardAllocationName"),
                "allocationAmount": row.get("allocationAmount"),
                "trackedAllocationPercentage": row.get("trackedAllocationPercentage"),
                "effectivePercentage": effective_pct,
                "percentageSource": pct_source,
            }
        )

    normalized_sorted = sorted(
        normalized,
        key=lambda x: (x.get("effectivePercentage") is not None, x.get("effectivePercentage") or -1),
        reverse=True,
    )

    coverage = {
        "allocations_count": len(normalized),
        "tracked_percentage_fields": sum(1 for x in normalized if x.get("trackedAllocationPercentage") is not None),
        "effective_percentage_fields": sum(1 for x in normalized if x.get("effectivePercentage") is not None),
        "tracked_percentage_sum": tracked_sum,
        "effective_percentage_sum": fallback_sum,
        "tracked_sum_close_to_100": 99.0 <= tracked_sum <= 101.0,
        "effective_sum_close_to_100": 99.0 <= fallback_sum <= 101.0,
    }

    return {
        "token": {
            "tokenId": data.get("tokenId") if isinstance(data, dict) else None,
            "symbol": data.get("symbol") if isinstance(data, dict) else None,
            "listedMethod": data.get("listedMethod") if isinstance(data, dict) else None,
        },
        "totals": {
            "totalTrackedAllocationAmount": data.get("totalTrackedAllocationAmount") if isinstance(data, dict) else None,
            "totalTrackedUnlockedAmount": data.get("totalTrackedUnlockedAmount") if isinstance(data, dict) else None,
            "totalTrackedLockedAmount": data.get("totalTrackedLockedAmount") if isinstance(data, dict) else None,
            "referenceSupply": data.get("referenceSupply") if isinstance(data, dict) else None,
        },
        "coverage": coverage,
        "top_allocations": normalized_sorted[:5],
        "allocations": normalized_sorted,
    }


class TokenmistTokenListTool(BaseTool):
    @property
    def name(self) -> str:
        return "tokenmist_token_list"

    @property
    def description(self) -> str:
        return """Get Token List API v4 from Tokenmist.

Uses latest Token List version (v4). Supports optional keyword filtering and limit to reduce payload size.
"""

    @property
    def parameters(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "keyword": {
                    "type": "string",
                    "description": "Optional keyword to filter by id/symbol/name",
                },
                "limit": {
                    "type": "integer",
                    "description": "Max results to return (default 50, max 500)",
                    "minimum": 1,
                    "maximum": 500,
                },
                "force_refresh": {
                    "type": "boolean",
                    "description": "Refresh token list cache from API",
                    "default": False,
                },
            },
        }

    async def execute(
        self,
        ctx: ToolContext,
        keyword: str = "",
        limit: int = 50,
        force_refresh: bool = False,
        **kwargs,
    ) -> ToolResult:
        try:
            limit = max(1, min(int(limit or 50), 500))
            rows = _get_index(force_refresh=force_refresh)
            if keyword:
                q = keyword.strip().lower()
                rows = [
                    r
                    for r in rows
                    if q in str(r.get("id", "")).lower()
                    or q in str(r.get("symbol", "")).lower()
                    or q in str(r.get("name", "")).lower()
                ]

            return ToolResult(
                success=True,
                output={
                    "version": "v4",
                    "count": len(rows),
                    "items": rows[:limit],
                    "returned": min(len(rows), limit),
                },
            )
        except Exception as e:
            return ToolResult(success=False, error=_safe_error_message(e))


class TokenmistResolveTokenTool(BaseTool):
    @property
    def name(self) -> str:
        return "tokenmist_resolve_token"

    @property
    def description(self) -> str:
        return """Resolve user token query to canonical tokenId using Token List v4.

Input can be tokenId, symbol, or token name. Returns best match and alternatives.
"""

    @property
    def parameters(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "Token id/symbol/name to resolve",
                },
                "force_refresh": {
                    "type": "boolean",
                    "description": "Refresh token list cache from API",
                    "default": False,
                },
            },
            "required": ["query"],
        }

    async def execute(
        self,
        ctx: ToolContext,
        query: str = "",
        force_refresh: bool = False,
        **kwargs,
    ) -> ToolResult:
        if not query:
            return ToolResult(success=False, error="'query' is required")
        try:
            idx = _get_index(force_refresh=force_refresh)
            result = resolve_token_id(idx, query)
            return ToolResult(success=True, output={"version": "v4", **result})
        except Exception as e:
            return ToolResult(success=False, error=_safe_error_message(e))


class TokenmistAllocationsTool(BaseTool):
    @property
    def name(self) -> str:
        return "tokenmist_allocations"

    @property
    def description(self) -> str:
        return """Get allocations data for a token from Allocations API v2 (latest).

Returns normalized allocation percentages to reduce ambiguity:
- Uses trackedAllocationPercentage as primary
- Adds effectivePercentage fallback when possible
- Includes top_allocations and coverage quality summary
"""

    @property
    def parameters(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "token_id": {
                    "type": "string",
                    "description": "Canonical tokenId from tokenmist_token_list/tokenmist_resolve_token",
                },
                "include_raw": {
                    "type": "boolean",
                    "description": "Include raw upstream payload for debugging",
                    "default": False,
                },
            },
            "required": ["token_id"],
        }

    async def execute(
        self,
        ctx: ToolContext,
        token_id: str = "",
        include_raw: bool = False,
        **kwargs,
    ) -> ToolResult:
        if not token_id:
            return ToolResult(success=False, error="'token_id' is required")
        try:
            raw = _client().allocations_v2(token_id)
            normalized = _normalize_allocations_payload(raw)
            out: Dict[str, Any] = {
                "version": "v2",
                "token_id": token_id,
                "normalized": normalized,
            }
            if include_raw:
                out["raw"] = raw
            return ToolResult(success=True, output=out)
        except Exception as e:
            return ToolResult(success=False, error=_safe_error_message(e))


class TokenmistAllocationsSummaryTool(BaseTool):
    @property
    def name(self) -> str:
        return "tokenmist_allocations_summary"

    @property
    def description(self) -> str:
        return """Get compact allocations summary with top N buckets and quality flags.

Accepts either `token_id` or free-text `query` (symbol/name/id).
If query is provided, resolves to canonical tokenId first.
"""

    @property
    def parameters(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "token_id": {
                    "type": "string",
                    "description": "Canonical tokenId (preferred if known)",
                },
                "query": {
                    "type": "string",
                    "description": "Token symbol/name/id (used when token_id is not provided)",
                },
                "top_n": {
                    "type": "integer",
                    "description": "Number of top allocations to return (default 5, max 20)",
                    "minimum": 1,
                    "maximum": 20,
                    "default": 5,
                },
                "force_refresh": {
                    "type": "boolean",
                    "description": "Refresh token index cache before resolve",
                    "default": False,
                },
            },
        }

    async def execute(
        self,
        ctx: ToolContext,
        token_id: str = "",
        query: str = "",
        top_n: int = 5,
        force_refresh: bool = False,
        **kwargs,
    ) -> ToolResult:
        try:
            top_n = max(1, min(int(top_n or 5), 20))

            resolved: Optional[Dict[str, Any]] = None
            canonical_token_id = (token_id or "").strip()

            if not canonical_token_id:
                if not query:
                    return ToolResult(success=False, error="Either 'token_id' or 'query' is required")
                idx = _get_index(force_refresh=force_refresh)
                resolved = resolve_token_id(idx, query)
                token = resolved.get("token") if isinstance(resolved, dict) else None
                if not token:
                    return ToolResult(
                        success=False,
                        error=(
                            "Could not resolve unique tokenId from query. "
                            f"match_type={resolved.get('match_type') if isinstance(resolved, dict) else 'unknown'}"
                        ),
                        output={"resolution": resolved},
                    )
                canonical_token_id = str(token.get("id", "")).strip()

            raw = _client().allocations_v2(canonical_token_id)
            normalized = _normalize_allocations_payload(raw)
            coverage = normalized.get("coverage", {}) if isinstance(normalized, dict) else {}
            top_allocations = normalized.get("top_allocations", []) if isinstance(normalized, dict) else []
            if not isinstance(top_allocations, list):
                top_allocations = []

            summary = {
                "token": normalized.get("token") if isinstance(normalized, dict) else None,
                "top_n": top_n,
                "top_allocations": top_allocations[:top_n],
                "coverage": coverage,
                "quality": {
                    "has_tracked_percentages": bool((coverage or {}).get("tracked_percentage_fields", 0) > 0),
                    "sum_close_to_100": bool(
                        (coverage or {}).get("tracked_sum_close_to_100")
                        or (coverage or {}).get("effective_sum_close_to_100")
                    ),
                },
            }

            output: Dict[str, Any] = {
                "version": "v2",
                "token_id": canonical_token_id,
                "summary": summary,
                "timestamp": datetime.utcnow().isoformat() + "Z",
            }
            if resolved is not None:
                output["resolution"] = {
                    "query": query,
                    "match_type": resolved.get("match_type"),
                    "token": resolved.get("token"),
                }

            return ToolResult(success=True, output=output)
        except Exception as e:
            return ToolResult(success=False, error=_safe_error_message(e))


class TokenmistDailyEmissionTool(BaseTool):
    @property
    def name(self) -> str:
        return "tokenmist_daily_emission"

    @property
    def description(self) -> str:
        return "Get daily emission data from Daily Emission API v2 (latest)."

    @property
    def parameters(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "token_id": {
                    "type": "string",
                    "description": "Canonical tokenId",
                },
                "start": {
                    "type": "string",
                    "description": "Optional YYYY-MM-DD",
                },
                "end": {
                    "type": "string",
                    "description": "Optional YYYY-MM-DD",
                },
            },
            "required": ["token_id"],
        }

    async def execute(
        self,
        ctx: ToolContext,
        token_id: str = "",
        start: str = "",
        end: str = "",
        **kwargs,
    ) -> ToolResult:
        if not token_id:
            return ToolResult(success=False, error="'token_id' is required")
        try:
            data = _client().daily_emission_v2(
                token_id=token_id,
                start=start or None,
                end=end or None,
            )
            return ToolResult(success=True, output={"version": "v2", **data})
        except Exception as e:
            return ToolResult(success=False, error=_safe_error_message(e))


class TokenmistUnlockEventsTool(BaseTool):
    @property
    def name(self) -> str:
        return "tokenmist_unlock_events"

    @property
    def description(self) -> str:
        return "Get unlock events from Unlock Events API v4 (latest)."

    @property
    def parameters(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "token_id": {
                    "type": "string",
                    "description": "Canonical tokenId",
                },
                "start": {
                    "type": "string",
                    "description": "Optional YYYY-MM-DD",
                },
                "end": {
                    "type": "string",
                    "description": "Optional YYYY-MM-DD",
                },
            },
            "required": ["token_id"],
        }

    async def execute(
        self,
        ctx: ToolContext,
        token_id: str = "",
        start: str = "",
        end: str = "",
        **kwargs,
    ) -> ToolResult:
        if not token_id:
            return ToolResult(success=False, error="'token_id' is required")
        try:
            data = _client().unlock_events_v4(
                token_id=token_id,
                start=start or None,
                end=end or None,
            )
            return ToolResult(success=True, output={"version": "v4", **data})
        except Exception as e:
            return ToolResult(success=False, error=_safe_error_message(e))


class TokenmistTokenOverviewTool(BaseTool):
    @property
    def name(self) -> str:
        return "tokenmist_token_overview"

    @property
    def description(self) -> str:
        return """One-call wrapper to minimize tool calls.

Resolves token query then fetches latest allocations(v2), daily-emission(v2), and unlock-events(v4).
Useful for agent workflows to reduce ambiguity and token/tool overhead.
"""

    @property
    def parameters(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "Token id/symbol/name",
                },
                "start": {
                    "type": "string",
                    "description": "Optional YYYY-MM-DD for emission/events",
                },
                "end": {
                    "type": "string",
                    "description": "Optional YYYY-MM-DD for emission/events",
                },
                "include_allocations": {
                    "type": "boolean",
                    "description": "Whether to fetch allocations",
                    "default": True,
                },
                "include_daily_emission": {
                    "type": "boolean",
                    "description": "Whether to fetch daily emission",
                    "default": True,
                },
                "include_unlock_events": {
                    "type": "boolean",
                    "description": "Whether to fetch unlock events",
                    "default": True,
                },
                "force_refresh": {
                    "type": "boolean",
                    "description": "Refresh token list cache before resolve",
                    "default": False,
                },
            },
            "required": ["query"],
        }

    async def execute(
        self,
        ctx: ToolContext,
        query: str = "",
        start: str = "",
        end: str = "",
        include_allocations: bool = True,
        include_daily_emission: bool = True,
        include_unlock_events: bool = True,
        force_refresh: bool = False,
        **kwargs,
    ) -> ToolResult:
        if not query:
            return ToolResult(success=False, error="'query' is required")

        try:
            idx = _get_index(force_refresh=force_refresh)
            resolved = resolve_token_id(idx, query)
            token = resolved.get("token")
            if not token:
                return ToolResult(
                    success=False,
                    error=(
                        "Could not resolve unique tokenId from query. "
                        f"match_type={resolved.get('match_type')}"
                    ),
                    output={
                        "resolution": resolved,
                    },
                )

            token_id = token.get("id")
            out: Dict[str, Any] = {
                "resolved": {
                    "query": query,
                    "match_type": resolved.get("match_type"),
                    "token": token,
                },
                "versions": {
                    "token_list": "v4",
                    "allocations": "v2",
                    "daily_emission": "v2",
                    "unlock_events": "v4",
                },
                "timestamp": datetime.utcnow().isoformat() + "Z",
            }

            client = _client()

            if include_allocations:
                try:
                    raw_alloc = client.allocations_v2(token_id)
                    out["allocations"] = {
                        "version": "v2",
                        "normalized": _normalize_allocations_payload(raw_alloc),
                    }
                except Exception as e:
                    out["allocations_error"] = _safe_error_message(e)

            if include_daily_emission:
                try:
                    out["daily_emission"] = client.daily_emission_v2(
                        token_id=token_id,
                        start=start or None,
                        end=end or None,
                    )
                except Exception as e:
                    out["daily_emission_error"] = _safe_error_message(e)

            if include_unlock_events:
                try:
                    out["unlock_events"] = client.unlock_events_v4(
                        token_id=token_id,
                        start=start or None,
                        end=end or None,
                    )
                except Exception as e:
                    out["unlock_events_error"] = _safe_error_message(e)

            return ToolResult(success=True, output=out)

        except Exception as e:
            return ToolResult(success=False, error=_safe_error_message(e))