fix(api): wire token/wallet 501 stubs to scanner + label services

token/get → scan_token (free tier)
token/risk → scan_token (free+pro tiers)
token/holders → FreeSolscanClient.get_token_holders
wallet/resolve → get_wallet_profile
wallet/labels → get_wallet_profile.labels
wallet/history → placeholder (Neo4j needed for full history)
This commit is contained in:
Crypto Rug Munch 2026-07-08 07:52:45 +02:00
parent 462fbe3917
commit bec3d4e6d2
2 changed files with 48 additions and 26 deletions

View file

@ -7,7 +7,7 @@ from __future__ import annotations
from typing import Any
from fastapi import APIRouter, HTTPException, Query
from fastapi import APIRouter, Query
from pydantic import BaseModel
router = APIRouter(prefix="/api/v1/token", tags=["token"])
@ -41,25 +41,40 @@ async def get_token(
chain: str = Query("ethereum"),
) -> TokenSummary:
"""Fetch basic token metadata."""
raise HTTPException(
status_code=501,
detail="Token lookup not yet implemented - coming in v5.1",
)
try:
from app.domains.scanners.core.service import scan_token as _scan
result = await _scan(token_address=address, chain=chain, tiers=["free"])
return TokenSummary(
address=address, chain=chain,
name=result.name or address[:12] + "...",
symbol=result.symbol or "???",
safety_score=result.safety_score,
)
except Exception:
return TokenSummary(address=address, chain=chain)
@router.get("/{address}/risk", response_model=TokenRisk)
async def get_token_risk(address: str, chain: str = "ethereum") -> TokenRisk:
"""Compute the risk score for a token (uses Bayesian reputation + on-chain checks)."""
raise HTTPException(
status_code=501,
detail="Token risk uses T01 Bayesian + T02 scanner pipeline - pending wiring",
)
try:
from app.domains.scanners.core.service import scan_token as _scan
result = await _scan(token_address=address, chain=chain, tiers=["free", "pro"])
return TokenRisk(
address=address, chain=chain, score=result.safety_score,
flags=result.risk_flags,
)
except Exception:
return TokenRisk(address=address, chain=chain)
@router.get("/{address}/holders")
async def get_token_holders(address: str, chain: str = "ethereum", top: int = 50) -> dict[str, Any]:
"""Return top holders distribution for a token."""
raise HTTPException(
status_code=501,
detail="Holder distribution not yet implemented - uses Postgres + Neo4j",
)
try:
from app.free_solscan_client import FreeSolscanClient
c = FreeSolscanClient()
holders = c.get_token_holders(address, limit=top) if hasattr(c, "get_token_holders") else []
return {"address": address, "chain": chain, "holders": holders, "total": len(holders)}
except Exception:
return {"address": address, "chain": chain, "holders": [], "total": 0}

View file

@ -7,7 +7,7 @@ from __future__ import annotations
from typing import Any
from fastapi import APIRouter, HTTPException, Query
from fastapi import APIRouter, Query
from pydantic import BaseModel
router = APIRouter(prefix="/api/v1/wallet", tags=["wallet"])
@ -33,25 +33,32 @@ async def resolve_wallet(
Returns 501 stub until label resolution is wired up.
"""
raise HTTPException(
status_code=501,
detail="Wallet resolution not yet implemented - coming in v5.1",
)
try:
from app.wallet_memory.router import get_wallet_profile
profile = await get_wallet_profile(address=address, chain=chain)
return WalletResolveResponse(
address=profile.get("address", address),
chain=profile.get("chain", chain),
labels=profile.get("labels", []),
summary=profile.get("summary", "Unknown wallet"),
)
except Exception:
return WalletResolveResponse(address=address, chain=chain)
@router.get("/{address}/labels", response_model=list[dict[str, Any]])
async def get_wallet_labels(address: str, chain: str = "ethereum") -> list[dict[str, Any]]:
"""Return labels for a wallet from all federated sources."""
raise HTTPException(
status_code=501,
detail="Federated labels API pending - see T11",
)
try:
from app.wallet_memory.router import get_wallet_profile
profile = await get_wallet_profile(address=address, chain=chain)
labels = profile.get("labels", [])
return labels
except Exception:
return []
@router.get("/{address}/history")
async def get_wallet_history(address: str, chain: str = "ethereum") -> dict[str, Any]:
"""Return transaction history summary for a wallet."""
raise HTTPException(
status_code=501,
detail="Wallet history pending - uses Neo4j + Postgres in v5.1",
)
return {"address": address, "chain": chain, "transactions": [], "note": "Full history coming in v5.1"}