Spaces:
Paused
Paused
File size: 4,458 Bytes
cf35cf8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | """Token data model for detected memecoins.
Represents a token detected by the snipers (Pump.fun, Raydium, Uniswap)
before and after safety analysis. Uses Pydantic v2 for validation and
JSON (de)serialisation so tokens can flow through asyncio queues, be
persisted to SQLite, or sent over the Telegram bot wire cheaply.
"""
from __future__ import annotations
from datetime import datetime, timezone
from enum import Enum
from typing import Any, Optional
from pydantic import BaseModel, ConfigDict, Field, field_validator
class Chain(str, Enum):
"""Supported trading chains."""
SOLANA = "solana"
BASE = "base"
class Token(BaseModel):
"""A memecoin detected by one of the snipers.
Attributes:
address: On-chain token mint address (Solana) or contract
address (Base).
name: Human-readable token name (e.g. "Dogecoin 2.0").
symbol: Ticker symbol (e.g. "DOGE2").
chain: Chain the token lives on.
launch_time: UTC datetime when the token was first observed /
created on-chain.
initial_liquidity_usd: USD value of liquidity at launch time.
``None`` until the first liquidity snapshot succeeds.
creator_wallet: Wallet address that created / deployed the token.
supply: Total / max supply in base units (human-readable, not
raw lamports). ``None`` until fetched.
safety_score: Final safety score 0-100 assigned by the safety
pipeline. ``None`` until :class:`RugChecker` has run.
metadata: Free-form bag for platform-specific data — pump.fun
bond curve progress, raydium pool id, image URI, socials,
etc.
"""
model_config = ConfigDict(
use_enum_values=True,
populate_by_name=True,
extra="allow",
)
address: str = Field(..., min_length=32, max_length=64, description="Token mint / contract address")
name: str = Field(..., min_length=1, max_length=128, description="Token name")
symbol: str = Field(..., min_length=1, max_length=32, description="Ticker symbol")
chain: Chain = Field(..., description="Chain the token lives on")
launch_time: datetime = Field(default_factory=lambda: datetime.now(timezone.utc), description="UTC launch / detection time")
initial_liquidity_usd: Optional[float] = Field(None, ge=0.0, description="USD liquidity at launch")
creator_wallet: Optional[str] = Field(None, description="Deployer wallet address")
supply: Optional[float] = Field(None, ge=0.0, description="Total supply in base units")
safety_score: Optional[int] = Field(None, ge=0, le=100, description="Safety pipeline score 0-100")
metadata: dict[str, Any] = Field(default_factory=dict, description="Platform-specific extras")
# ---- validators -------------------------------------------------------
@field_validator("chain", mode="before")
@classmethod
def _normalise_chain(cls, v: Any) -> Any:
"""Accept bare strings ("solana") or Chain members."""
if isinstance(v, str) and not isinstance(v, Chain):
try:
return Chain(v.lower())
except ValueError:
raise ValueError(f"Unsupported chain: {v!r}. Expected one of {[c.value for c in Chain]}")
return v
@field_validator("address", "creator_wallet")
@classmethod
def _strip_whitespace(cls, v: Optional[str]) -> Optional[str]:
if v is None:
return v
v = v.strip()
if not v:
raise ValueError("address / creator_wallet cannot be empty")
return v
@field_validator("symbol")
@classmethod
def _upper_symbol(cls, v: str) -> str:
return v.strip().upper()
# ---- convenience ------------------------------------------------------
@property
def is_solana(self) -> bool:
"""True if this token lives on Solana."""
return self.chain == Chain.SOLANA.value or self.chain == Chain.SOLANA
@property
def is_base(self) -> bool:
"""True if this token lives on Base."""
return self.chain == Chain.BASE.value or self.chain == Chain.BASE
def short_address(self, prefix: int = 4, suffix: int = 4) -> str:
"""Return a shortened address like ``So11…Mo3`` for display."""
if len(self.address) <= prefix + suffix + 1:
return self.address
return f"{self.address[:prefix]}…{self.address[-suffix:]}"
|