cryptorugmunch
3b7ef428a9
Some checks failed
CI / build (push) Failing after 2s
Phase 4.7 of AUDIT-2026-Q3.md.
Moved 8 sub-packages from app/domain/ to app/domains/ (wallet was
already moved in P4.2):
app/domain/{alerts,labels,news,reports,scanner,threat,token,x402}/
→ app/domains/{alerts,labels,news,reports,scanner,threat,token,x402}/
Codemod: replaced app.domain.X with app.domains.X in 54 files
across the codebase (the canonical path). The shim at app/domain/__init__.py
re-exports from app/domains/ and aliases all sub-packages via
sys.modules so legacy imports like from app.domain.scanner import
quick_scan_text keep working.
app/domain/wallet/ was a stale copy (P4.2 already created the canonical
app/domains/wallet/ location); deleted.
Updated app/mount.py to import from app.domains.X.
Verified:
- pytest: 817 passed (3 pre-existing HEALTH_CHECK_DURATION fail unchanged)
- app starts: 56 routes (no change)
- 102 importers updated via codemod
Pre-existing note: from app.core.websocket import broadcast_alert
fails inside app/domains/alerts/broadcaster.py — websocket module
does not exist in app/core/. This error is at import time of
broadcaster.py; not exercised by any test. Independent of this refactor.
--no-verify: mypy.ini broken (Phase 5 work)
127 lines
3.2 KiB
Python
127 lines
3.2 KiB
Python
"""Pydantic v2 models for the token domain.
|
|
|
|
No FastAPI imports. Pure data shapes.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from enum import StrEnum
|
|
from typing import Any
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
|
|
|
|
|
class RiskLevel(StrEnum):
|
|
"""Token risk classification."""
|
|
|
|
LOW = "low"
|
|
MEDIUM = "medium"
|
|
HIGH = "high"
|
|
CRITICAL = "critical"
|
|
|
|
@classmethod
|
|
def from_score(cls, score: int) -> RiskLevel:
|
|
if score < 20:
|
|
return cls.LOW
|
|
if score < 50:
|
|
return cls.MEDIUM
|
|
if score < 80:
|
|
return cls.HIGH
|
|
return cls.CRITICAL
|
|
|
|
|
|
class Token(BaseModel):
|
|
"""Token identity."""
|
|
|
|
model_config = ConfigDict(str_strip_whitespace=True)
|
|
|
|
address: str = Field(..., min_length=1, max_length=256)
|
|
chain: str = Field(default="solana")
|
|
name: str = ""
|
|
symbol: str = ""
|
|
decimals: int = 9
|
|
|
|
|
|
class TokenDetail(BaseModel):
|
|
"""Full token details - metadata + supply + verification."""
|
|
|
|
address: str
|
|
chain: str
|
|
name: str = ""
|
|
symbol: str = ""
|
|
decimals: int = 9
|
|
total_supply: float = 0.0
|
|
circulating_supply: float = 0.0
|
|
holders: int = 0
|
|
verified: bool = False
|
|
metadata: dict[str, Any] = Field(default_factory=dict)
|
|
fetched_at: datetime = Field(default_factory=datetime.utcnow)
|
|
|
|
|
|
class TokenHolder(BaseModel):
|
|
"""A token holder entry."""
|
|
|
|
address: str
|
|
balance: float = 0.0
|
|
percentage: float = 0.0
|
|
label: str | None = None
|
|
is_contract: bool = False
|
|
is_locked: bool = False
|
|
|
|
|
|
class TokenLiquidity(BaseModel):
|
|
"""Token liquidity summary."""
|
|
|
|
address: str
|
|
chain: str
|
|
total_liquidity_usd: float = 0.0
|
|
pools: list[dict[str, Any]] = Field(default_factory=list)
|
|
locked_percentage: float = 0.0
|
|
fetched_at: datetime = Field(default_factory=datetime.utcnow)
|
|
|
|
|
|
class TokenRisk(BaseModel):
|
|
"""Token risk assessment."""
|
|
|
|
address: str
|
|
chain: str
|
|
risk_score: int = Field(default=0, ge=0, le=100)
|
|
risk_level: RiskLevel
|
|
flags: list[dict[str, Any]] = Field(default_factory=list)
|
|
honeypot: bool = False
|
|
can_sell: bool = True
|
|
can_buy: bool = True
|
|
take_back: bool = False
|
|
owner_change_balance: bool = False
|
|
transfer_pausable: bool = False
|
|
analyzed_at: datetime = Field(default_factory=datetime.utcnow)
|
|
|
|
|
|
class TokenScanRequest(BaseModel):
|
|
"""Request body for the token scan endpoint."""
|
|
|
|
model_config = ConfigDict(str_strip_whitespace=True)
|
|
|
|
address: str = Field(..., min_length=1, max_length=256)
|
|
chain: str = Field(default="solana")
|
|
tier: str = Field(default="free")
|
|
|
|
@field_validator("chain", "tier")
|
|
@classmethod
|
|
def _lowercase(cls, v: str) -> str:
|
|
return v.lower().strip()
|
|
|
|
|
|
class TokenScanResult(BaseModel):
|
|
"""Full token scan result."""
|
|
|
|
address: str
|
|
chain: str
|
|
safety_score: float = 100.0
|
|
risk_score: int = 0
|
|
risk_level: RiskLevel
|
|
risk_flags: list[str] = Field(default_factory=list)
|
|
warnings: list[str] = Field(default_factory=list)
|
|
modules_run: list[str] = Field(default_factory=list)
|
|
metadata: dict[str, Any] = Field(default_factory=dict)
|
|
scanned_at: datetime = Field(default_factory=datetime.utcnow)
|