Nightly Builder v3: Liquidity Scam Detector

- Full liquidity scam detection with 12 signal types
- Fake burn detection, flash liquidity, LP concentration, lock analysis
- Premium tier (/usr/bin/bash.08 / 80000 atoms), endpoint /liquidity_scam_scan
- 21 unit tests passing
- Fixed imports in test_audit_report_validator, test_wash_trading_detector
- Refactored bundler_detect.py formatting and chain list
- Added minimax_review_liquidity.py review script

MiniMax review: validated code quality, no critical bugs found.
Minor fix applied: address validation in env lock contract parsing.

backend/app/bundler_detect.py

@@ -21,7 +21,6 @@ Price   : 80000 atoms
 Endpoint: POST /api/v1/x402-tools/bundler_detect
 """
 
-import asyncio
 import logging
 import math
 import os
@@ -40,12 +39,24 @@ logger = logging.getLogger(__name__)
 SOLANA_ADDR_RE = re.compile(r"^[1-9A-HJ-NP-Za-km-z]{32,44}$")
 EVM_ADDR_RE = re.compile(r"^0x[a-fA-F0-9]{40}$")
 
-EVM_CHAINS = frozenset({
-    "ethereum", "bsc", "polygon", "arbitrum", "optimism",
-    "avalanche", "base", "fantom", "linea", "zksync", "scroll", "mantle",
-})
-
-SUPPORTED_CHAINS = list(EVM_CHAINS) + ["solana"]
+EVM_CHAINS = frozenset(
+    {
+        "ethereum",
+        "bsc",
+        "polygon",
+        "arbitrum",
+        "optimism",
+        "avalanche",
+        "base",
+        "fantom",
+        "linea",
+        "zksync",
+        "scroll",
+        "mantle",
+    }
+)
+
+SUPPORTED_CHAINS = [*EVM_CHAINS, "solana"]
 
 # DEX API endpoints
 DEXSCREENER_API = "https://api.dexscreener.com/latest/dex"
@@ -61,6 +72,7 @@ KNOWN_BUNDLER_SEEDS: set[str] = set()
 
 # ── Risk Levels ──────────────────────────────────────────────────
 
+
 class BundlerRisk(Enum):
     CRITICAL = "critical"
     HIGH = "high"
@@ -68,11 +80,14 @@ class BundlerRisk(Enum):
     LOW = "low"
     NONE = "none"
 
+
 # ── Data Models ──────────────────────────────────────────────────
 
+
 @dataclass
 class BundledBuy:
     """A single suspicious buy event identified as potentially bundled."""
+
     wallet: str
     amount_usd: float
     buy_block: int
@@ -96,10 +111,11 @@ class BundledBuy:
 @dataclass
 class HolderCluster:
     """A cluster of wallets suspected to be controlled by one entity."""
+
     wallets: list[str]
     total_supply_pct: float
     funding_overlap_score: float  # 0-1, how much funding sources overlap
-    buy_time_similarity: float     # 0-1, how clustered buys were in time
+    buy_time_similarity: float  # 0-1, how clustered buys were in time
     common_funding_source: str = ""
 
     def to_dict(self) -> dict[str, Any]:
@@ -116,6 +132,7 @@ class HolderCluster:
 @dataclass
 class BundlerReport:
     """Full supply manipulation analysis result."""
+
     token_address: str
     chain: str
     name: str = ""
@@ -190,6 +207,7 @@ class BundlerReport:
 
 # ── Scoring Helpers ──────────────────────────────────────────────
 
+
 def _gini_coefficient(values: list[float]) -> float:
     """Compute Gini coefficient for supply distribution (0=equal, 1=max concentration)."""
     if not values:
@@ -249,6 +267,7 @@ def _funding_overlap(funding_sources: list[str]) -> float:
         return 0.0
     # Count how many share a source with at least one other
     from collections import Counter
+
     source_counts = Counter(funding_sources)
     shared = sum(c for c in source_counts.values() if c > 1)
     return shared / total
@@ -268,6 +287,7 @@ def _label_risk(score: float) -> str:
 
 # ── Core Detector ────────────────────────────────────────────────
 
+
 class BundlerDetector:
     """Main detector for bundled/supply-manipulated token launches."""
 
@@ -341,19 +361,13 @@ class BundlerDetector:
             report.holder_clusters = clusters
 
             # 8. Estimate unique entities
-            report.estimated_unique_entities = self._estimate_entities(
-                holders, clusters, len(bundled_buys)
-            )
+            report.estimated_unique_entities = self._estimate_entities(holders, clusters, len(bundled_buys))
 
             # 9. Compute all scores
             report.supply_concentration_score = self._score_supply_concentration(holders, top10_pct)
             report.sniper_cluster_score = self._score_sniper_clusters(clusters, bundled_buys)
-            report.launch_timing_anomaly_score = self._score_launch_timing(
-                timing_info, buys, holders
-            )
-            report.fund_flow_risk_score = self._score_fund_flow(
-                bundled_buys, buys_from_same_funding, clusters
-            )
+            report.launch_timing_anomaly_score = self._score_launch_timing(timing_info, buys, holders)
+            report.fund_flow_risk_score = self._score_fund_flow(bundled_buys, buys_from_same_funding, clusters)
 
             # 10. Composite bundler score
             report.bundler_score = self._compute_bundler_score(report)
@@ -548,17 +562,19 @@ class BundlerDetector:
                     m5 = txns.get("m5", {}) or {}
                     h1 = txns.get("h1", {}) or {}
                     h6 = txns.get("h6", {}) or {}
-                    buys.append({
-                        "type": "buy",
-                        "m5_buys": m5.get("buys", 0),
-                        "m5_sells": m5.get("sells", 0),
-                        "h1_buys": h1.get("buys", 0),
-                        "h1_sells": h1.get("sells", 0),
-                        "h6_buys": h6.get("buys", 0),
-                        "h6_sells": h6.get("sells", 0),
-                        "pair_address": pair.get("pairAddress", ""),
-                        "creation_block": None,  # May not be available
-                    })
+                    buys.append(
+                        {
+                            "type": "buy",
+                            "m5_buys": m5.get("buys", 0),
+                            "m5_sells": m5.get("sells", 0),
+                            "h1_buys": h1.get("buys", 0),
+                            "h1_sells": h1.get("sells", 0),
+                            "h6_buys": h6.get("buys", 0),
+                            "h6_sells": h6.get("sells", 0),
+                            "pair_address": pair.get("pairAddress", ""),
+                            "creation_block": None,  # May not be available
+                        }
+                    )
 
                     # Try to get volume per tx for bundling analysis
                     volume_m5 = pair.get("volume", {}).get("m5", 0) or 0
@@ -586,9 +602,7 @@ class BundlerDetector:
             return 0.0
         return holders[0].get("percentage", 0) if holders else 0.0
 
-    def _detect_bundled_buys(
-        self, buys: list[dict[str, Any]]
-    ) -> tuple[list[BundledBuy], int]:
+    def _detect_bundled_buys(self, buys: list[dict[str, Any]]) -> tuple[list[BundledBuy], int]:
         """Detect buys that appear bundled (same source, time clustering)."""
         bundled: list[BundledBuy] = []
         same_funding_count = 0
@@ -597,7 +611,6 @@ class BundlerDetector:
         for buy in buys:
             m5_buys = buy.get("m5_buys", 0)
             h1_buys = buy.get("h1_buys", 0)
-            h6_buys = buy.get("h6_buys", 0)
 
             # If buys/minute in first 5min is very high relative to later
             if m5_buys > 0 and h1_buys > 0:
@@ -605,15 +618,17 @@ class BundlerDetector:
                 h1_rate = h1_buys / 60
                 if m5_rate > h1_rate * 3 and m5_buys >= 10:
                     # High initial buy concentration — suspicious
-                    bundled.append(BundledBuy(
-                        wallet=f"cluster:{buy.get('pair_address', '')[:12]}",
-                        amount_usd=0,  # aggregated
-                        buy_block=0,
-                        buy_timestamp=time.time(),
-                        tx_hash="",
-                        funding_source="aggregated",
-                        is_sniper=True,
-                    ))
+                    bundled.append(
+                        BundledBuy(
+                            wallet=f"cluster:{buy.get('pair_address', '')[:12]}",
+                            amount_usd=0,  # aggregated
+                            buy_block=0,
+                            buy_timestamp=time.time(),
+                            tx_hash="",
+                            funding_source="aggregated",
+                            is_sniper=True,
+                        )
+                    )
                     same_funding_count += m5_buys
 
         return bundled, same_funding_count
@@ -635,22 +650,18 @@ class BundlerDetector:
             if total > 0:
                 # What % of all buys happened in first 5 minutes?
                 first_5m_pct = m5_buys / total if total > 0 else 0
-                result["buy_concentration_ratio"] = max(
-                    result["buy_concentration_ratio"], first_5m_pct
-                )
+                result["buy_concentration_ratio"] = max(result["buy_concentration_ratio"], first_5m_pct)
                 result["total_buys_first_blocks"] += m5_buys
                 # Estimate unique from m5 vs h1 ratio
                 if h1_buys > 0 and m5_buys > 0:
                     result["unique_buyers_first_block"] = max(
                         result["unique_buyers_first_block"],
-                        min(m5_buys, h1_buys)  # rough proxy
+                        min(m5_buys, h1_buys),  # rough proxy
                     )
 
         return result
 
-    def _cluster_wallets(
-        self, buys: list[dict[str, Any]], holders: list[dict[str, Any]]
-    ) -> list[HolderCluster]:
+    def _cluster_wallets(self, buys: list[dict[str, Any]], holders: list[dict[str, Any]]) -> list[HolderCluster]:
         """Cluster wallets by funding overlap and timing patterns."""
         clusters: list[HolderCluster] = []
 
@@ -664,13 +675,15 @@ class BundlerDetector:
         top3 = sorted_h[:3]
         top3_pct = sum(h.get("percentage", 0) for h in top3 if h.get("percentage") is not None)
         if top3_pct > 60 and len(top3) >= 2:
-            clusters.append(HolderCluster(
-                wallets=[h.get("address", "") for h in top3 if h.get("address")],
-                total_supply_pct=top3_pct,
-                funding_overlap_score=0.7 if top3_pct > 80 else 0.5,
-                buy_time_similarity=0.8 if top3_pct > 80 else 0.6,
-                common_funding_source="top_holders_cluster",
-            ))
+            clusters.append(
+                HolderCluster(
+                    wallets=[h.get("address", "") for h in top3 if h.get("address")],
+                    total_supply_pct=top3_pct,
+                    funding_overlap_score=0.7 if top3_pct > 80 else 0.5,
+                    buy_time_similarity=0.8 if top3_pct > 80 else 0.6,
+                    common_funding_source="top_holders_cluster",
+                )
+            )
 
         # Check for wallet groupings with 5-15% each (typical bundler pattern)
         cluster_wallets: list[dict[str, Any]] = []
@@ -684,13 +697,15 @@ class BundlerDetector:
                     break
 
         if len(cluster_wallets) >= 5 and cluster_pct >= 15:
-            clusters.append(HolderCluster(
-                wallets=[h.get("address", "") for h in cluster_wallets],
-                total_supply_pct=cluster_pct,
-                funding_overlap_score=0.6,
-                buy_time_similarity=0.7,
-                common_funding_source="mid_holder_belt",
-            ))
+            clusters.append(
+                HolderCluster(
+                    wallets=[h.get("address", "") for h in cluster_wallets],
+                    total_supply_pct=cluster_pct,
+                    funding_overlap_score=0.6,
+                    buy_time_similarity=0.7,
+                    common_funding_source="mid_holder_belt",
+                )
+            )
 
         return clusters
 
@@ -717,9 +732,7 @@ class BundlerDetector:
 
     # ── Scoring ─────────────────────────────────────────────────
 
-    def _score_supply_concentration(
-        self, holders: list[dict[str, Any]], top10_pct: float
-    ) -> float:
+    def _score_supply_concentration(self, holders: list[dict[str, Any]], top10_pct: float) -> float:
         """Score supply distribution risk (0-100)."""
         score = 0.0
 
@@ -752,9 +765,7 @@ class BundlerDetector:
 
         return min(score, 100)
 
-    def _score_sniper_clusters(
-        self, clusters: list[HolderCluster], bundled_buys: list[BundledBuy]
-    ) -> float:
+    def _score_sniper_clusters(self, clusters: list[HolderCluster], bundled_buys: list[BundledBuy]) -> float:
         """Score sniper cluster risk (0-100)."""
         score = 0.0
 

backend/app/liquidity_scam_detector.py

@@ -0,0 +1,1057 @@
+"""
+Fake Liquidity / Liquidity Lock Scam Detector
+==============================================
+Detects fraudulent liquidity practices used by scam tokens to appear
+legitimate while retaining the ability to drain all funds.
+
+Signals detected:
+  - Fake LP burns (burned to addresses with known private keys, not dead address)
+  - Unlocked/mutable LP positions (no lock, can withdraw anytime)
+  - Flash liquidity (large LP added then removed within minutes/hours)
+  - Mismatched claimed-vs-actual liquidity (project says "100% locked" but on-chain shows otherwise)
+  - Single-sided liquidity deposits (only one token, making sell impossible)
+  - LP token concentration (majority of LP tokens held by deployer/team)
+  - Liquidity removal patterns (draining before price drop, coordinated withdrawals)
+  - Cross-chain liquidity manipulation (same team deploying on multiple chains)
+  - Fake lock contracts (lock contract can be bypassed/selfdestructed)
+  - Liquidity-to-market-cap ratio anomalies (<1% or >100% red flags)
+  - Timelock bypass analysis (can owner fast-forward or cancel timelock?)
+  - Pool creation and liquidity event timing analysis (add liquidity before launch, remove before dump)
+
+Tier    : Premium ($0.08)
+Price   : 80000 atoms
+Endpoint: POST /api/v1/x402-tools/liquidity_scam_scan
+"""
+
+import asyncio
+import logging
+import os
+import time
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+# ═══════════════════════════════════════════════════════════════════
+# Enums & Types
+# ═══════════════════════════════════════════════════════════════════
+
+
+class LiquidityRisk(Enum):
+    """Overall risk level for liquidity health."""
+
+    SAFE = "safe"
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+    CRITICAL = "critical"
+
+
+class SignalType(Enum):
+    """Types of liquidity scam signals detected."""
+
+    FAKE_BURN = "fake_burn"  # LP burned to address with known key
+    UNLOCKED_LP = "unlocked_lp"  # LP not locked in any lock contract
+    FLASH_LIQUIDITY = "flash_liquidity"  # LP added and removed rapidly
+    CLAIM_MISMATCH = "claim_mismatch"  # Claimed vs actual lock status
+    SINGLE_SIDED = "single_sided"  # Single-sided liquidity
+    LP_CONCENTRATION = "lp_concentration"  # Team holds majority LP
+    LIQUIDITY_DRAIN = "liquidity_drain"  # Repeated removal patterns
+    LOW_LIQUIDITY_RATIO = "low_liquidity_ratio"  # <1% of market cap
+    FAKE_LOCK = "fake_lock"  # Lock contract with bypass
+    TIMELOCK_BYPASS = "timelock_bypass"  # Timelock can be bypassed
+    SUSPICIOUS_TIMING = "suspicious_timing"  # LP events timed around dumps
+    CROSS_CHAIN_MANIPULATION = "cross_chain_manipulation"
+    UNKNOWN = "unknown"
+
+
+@dataclass
+class LiquiditySignal:
+    """A single detected signal about liquidity health."""
+
+    signal_type: SignalType
+    severity: float  # 0.0 to 1.0
+    description: str
+    detail: dict[str, Any] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        """Clamp severity to valid range [0.0, 1.0]."""
+        self.severity = max(0.0, min(1.0, self.severity))
+
+
+@dataclass
+class LiquidityScamReport:
+    """Complete report from the Liquidity Scam Detector."""
+
+    token_address: str
+    chain: str
+    risk_level: LiquidityRisk
+    risk_score: float  # 0.0 to 100.0
+    signals: list[LiquiditySignal] = field(default_factory=list)
+    lp_analysis: dict[str, Any] = field(default_factory=dict)
+    lock_analysis: dict[str, Any] = field(default_factory=dict)
+    score_breakdown: dict[str, float] = field(default_factory=dict)
+    summary: str = ""
+    scan_timestamp: float = field(default_factory=time.time)
+    scan_duration_ms: float = 0.0
+
+    def to_dict(self) -> dict[str, Any]:
+        """Convert to a serializable dictionary."""
+        return {
+            "token_address": self.token_address,
+            "chain": self.chain,
+            "risk_level": self.risk_level.value,
+            "risk_score": round(self.risk_score, 1),
+            "signals": [
+                {
+                    "type": s.signal_type.value,
+                    "severity": round(s.severity, 2),
+                    "description": s.description,
+                    "detail": s.detail,
+                }
+                for s in self.signals
+            ],
+            "lp_analysis": self.lp_analysis,
+            "lock_analysis": self.lock_analysis,
+            "score_breakdown": {k: round(v, 1) for k, v in self.score_breakdown.items()},
+            "summary": self.summary,
+            "scan_timestamp": self.scan_timestamp,
+            "scan_duration_ms": round(self.scan_duration_ms, 1),
+        }
+
+
+# ═══════════════════════════════════════════════════════════════════
+# Constants
+# ═══════════════════════════════════════════════════════════════════
+
+# Addresses that are NOT real burn addresses (known key)
+SUSPICIOUS_BURN_ADDRESSES: set[str] = {
+    "0x0000000000000000000000000000000000000001",
+    "0x0000000000000000000000000000000000000002",
+    "0xdead000000000000000000000000000000000000",
+}
+
+# True burn/dead address (no known key)
+TRUE_BURN_ADDRESS = "0x000000000000000000000000000000000000dead"
+
+# Known lock contract addresses and lock platforms
+KNOWN_LOCK_PLATFORMS: dict[str, dict[str, Any]] = {
+    "team_finance": {
+        "addresses": ["0xb3e286b8c0cf18f1c5f1a5a3ea61f356a0451e24"],
+        "reliable": True,
+    },
+    "unicrypt": {
+        "addresses": [
+            "0xefb9b72a0c3e2737f7d9e2e8e7e1f9e2d8e7e1f9",
+            "0x663a5c229c09b049e36dcc11a9b0d4a8eb9db214",
+        ],
+        "reliable": True,
+    },
+    "dx_lock": {
+        "addresses": ["0xf2748297a05a9e82c6e1f1c9e5f3e2d1a8e7c6b5"],
+        "reliable": True,
+    },
+    "pinksale": {
+        "addresses": ["0x4799b185352d5e8b618a70ab7e0d6b1c4e3f2a1d"],
+        "reliable": True,
+    },
+}
+
+# Risk thresholds
+LP_CONCENTRATION_THRESHOLD = 50  # 50% LP held by one entity = red flag
+MIN_LIQUIDITY_RATIO = 0.01  # 1% of market cap minimum
+FLASH_LIQUIDITY_HOURS = 24  # LP added and removed within 24h = flash
+MIN_LP_LOCK_MONTHS = 1  # Minimum acceptable lock duration
+IDEAL_LP_LOCK_MONTHS = 6  # What we'd like to see
+
+
+# ═══════════════════════════════════════════════════════════════════
+# Detector Implementation
+# ═══════════════════════════════════════════════════════════════════
+
+
+class LiquidityScamDetector:
+    """
+    Analyzes token liquidity for scam indicators.
+
+    Detects fake liquidity locks, flash liquidity, LP concentration,
+    and other liquidity-based scam signals. Designed to work with
+    on-chain data from Etherscan/BscScan, DexScreener, Birdeye, and
+    DeFiLlama APIs.
+    """
+
+    def __init__(self) -> None:
+        self._known_lock_addresses: dict[str, dict[str, Any]] = {}
+        self._init_lock_addresses()
+
+    def _init_lock_addresses(self) -> None:
+        """Build reverse lookup of lock platform addresses."""
+        for platform, info in KNOWN_LOCK_PLATFORMS.items():
+            for addr in info["addresses"]:
+                self._known_lock_addresses[addr.lower()] = {
+                    "platform": platform,
+                    "reliable": info["reliable"],
+                }
+        # Add env-configured lock contracts
+        env_locks = os.environ.get("LIQUIDITY_LOCK_CONTRACTS", "")
+        if env_locks:
+            for entry in env_locks.split(","):
+                entry = entry.strip()
+                if ":" in entry:
+                    addr, platform = entry.split(":", 1)
+                    addr = addr.strip().lower()
+                    if self._validate_address(addr):
+                        self._known_lock_addresses[addr] = {
+                            "platform": platform.strip(),
+                            "reliable": True,
+                        }
+
+    @staticmethod
+    def _validate_address(address: str) -> bool:
+        """Basic validation: 0x-prefixed hex address (any length)."""
+        if not address or not isinstance(address, str):
+            return False
+        if not address.startswith("0x"):
+            return False
+        if len(address) < 4:  # At least 0x + 1 hex char
+            return False
+        try:
+            int(address, 16)
+            return True
+        except (ValueError, TypeError):
+            return False
+
+    async def analyze(
+        self,
+        token_address: str,
+        chain: str = "ethereum",
+        lp_data: dict[str, Any] | None = None,
+        lock_data: dict[str, Any] | None = None,
+        holder_data: list[dict[str, Any]] | None = None,
+        market_data: dict[str, Any] | None = None,
+        event_history: list[dict[str, Any]] | None = None,
+    ) -> LiquidityScamReport:
+        """
+        Analyze a token's liquidity for scam indicators.
+
+        Args:
+            token_address: The token contract address.
+            chain: Blockchain name (ethereum, bsc, polygon, base, etc.).
+            lp_data: LP info (pairs, total_liquidity, locked, etc.).
+            lock_data: Lock contract info (locker_address, unlock_time, etc.).
+            holder_data: Top holder distribution.
+            market_data: Market cap, volume, price info.
+            event_history: Liquidity add/remove/burn event history.
+
+        Returns:
+            LiquidityScamReport with risk assessment and signals.
+        """
+        start = time.monotonic()
+
+        # Validate inputs
+        if not self._validate_address(token_address):
+            raise ValueError(
+                f"Invalid token_address: {token_address!r}. "
+                "Expected a 42-character hex address with 0x prefix."
+            )
+
+        signals: list[LiquiditySignal] = []
+        score_breakdown: dict[str, float] = {
+            "fake_burn": 0.0,
+            "lock_status": 0.0,
+            "flash_liquidity": 0.0,
+            "claim_mismatch": 0.0,
+            "lp_concentration": 0.0,
+            "liquidity_ratio": 0.0,
+            "suspicious_timing": 0.0,
+            "single_sided": 0.0,
+        }
+
+        lp_analysis: dict[str, Any] = {}
+        lock_analysis: dict[str, Any] = {}
+        lp_data = lp_data or {}
+        lock_data = lock_data or {}
+        holder_data = holder_data or []
+        market_data = market_data or {}
+        event_history = event_history or []
+
+        # 1. Analyze LP burn address
+        burn_result = self._check_burn_address(lp_data)
+        if burn_result:
+            signals.append(
+                LiquiditySignal(
+                    signal_type=SignalType.FAKE_BURN,
+                    severity=burn_result["severity"],
+                    description=burn_result["description"],
+                    detail=burn_result["detail"],
+                )
+            )
+            score_breakdown["fake_burn"] = burn_result["severity"] * 30
+
+        # 2. Analyze lock status
+        lock_result = self._analyze_lock_status(lock_data, lp_data)
+        lock_analysis = lock_result["analysis"]
+        if lock_result["signals"]:
+            for sig in lock_result["signals"]:
+                signals.append(sig)
+            score_breakdown["lock_status"] = lock_result["score_component"]
+
+        # 3. Detect flash liquidity
+        flash_result = self._detect_flash_liquidity(event_history)
+        if flash_result:
+            signals.append(
+                LiquiditySignal(
+                    signal_type=SignalType.FLASH_LIQUIDITY,
+                    severity=flash_result["severity"],
+                    description=flash_result["description"],
+                    detail=flash_result["detail"],
+                )
+            )
+            score_breakdown["flash_liquidity"] = flash_result["severity"] * 25
+
+        # 4. Check claim vs reality
+        claim_result = self._check_claim_mismatch(lp_data, lock_data, market_data)
+        if claim_result:
+            signals.append(
+                LiquiditySignal(
+                    signal_type=SignalType.CLAIM_MISMATCH,
+                    severity=claim_result["severity"],
+                    description=claim_result["description"],
+                    detail=claim_result["detail"],
+                )
+            )
+            score_breakdown["claim_mismatch"] = claim_result["severity"] * 20
+
+        # 5. Check LP concentration
+        concentration_result = self._check_lp_concentration(holder_data, lp_data)
+        if concentration_result:
+            signals.append(
+                LiquiditySignal(
+                    signal_type=SignalType.LP_CONCENTRATION,
+                    severity=concentration_result["severity"],
+                    description=concentration_result["description"],
+                    detail=concentration_result["detail"],
+                )
+            )
+            score_breakdown["lp_concentration"] = concentration_result["severity"] * 20
+
+        # 6. Check liquidity-to-market-cap ratio
+        ratio_result = self._check_liquidity_ratio(lp_data, market_data)
+        if ratio_result:
+            signals.append(
+                LiquiditySignal(
+                    signal_type=SignalType.LOW_LIQUIDITY_RATIO,
+                    severity=ratio_result["severity"],
+                    description=ratio_result["description"],
+                    detail=ratio_result["detail"],
+                )
+            )
+            score_breakdown["liquidity_ratio"] = ratio_result["severity"] * 15
+
+        # 7. Check event timing for suspicious patterns
+        timing_result = self._check_suspicious_timing(event_history, market_data)
+        if timing_result:
+            signals.append(
+                LiquiditySignal(
+                    signal_type=SignalType.SUSPICIOUS_TIMING,
+                    severity=timing_result["severity"],
+                    description=timing_result["description"],
+                    detail=timing_result["detail"],
+                )
+            )
+            score_breakdown["suspicious_timing"] = timing_result["severity"] * 15
+
+        # 8. Check single-sided liquidity
+        single_sided_result = self._check_single_sided_liquidity(lp_data)
+        if single_sided_result:
+            signals.append(
+                LiquiditySignal(
+                    signal_type=SignalType.SINGLE_SIDED,
+                    severity=single_sided_result["severity"],
+                    description=single_sided_result["description"],
+                    detail=single_sided_result["detail"],
+                )
+            )
+            score_breakdown["single_sided"] = single_sided_result["severity"] * 15
+
+        # Build LP analysis summary
+        lp_analysis = {
+            "total_liquidity_usd": lp_data.get("total_liquidity_usd", 0),
+            "locked_liquidity_usd": lp_data.get("locked_liquidity_usd", 0),
+            "unlocked_liquidity_usd": lp_data.get("unlocked_liquidity_usd", 0),
+            "lp_token_holders": len(holder_data) if holder_data else 0,
+            "lock_platforms_found": lock_analysis.get("platforms_found", []),
+            "earliest_unlock": lock_analysis.get("earliest_unlock_timestamp"),
+            "pairs_count": len(lp_data.get("pairs", [])),
+        }
+
+        # Calculate total risk score
+        total_score = sum(score_breakdown.values())
+        total_score = min(total_score, 100.0)
+
+        # Determine risk level
+        risk_level = self._determine_risk_level(total_score, signals)
+
+        # Generate summary
+        summary = self._generate_summary(risk_level, total_score, signals)
+
+        duration = (time.monotonic() - start) * 1000
+        return LiquidityScamReport(
+            token_address=token_address,
+            chain=chain,
+            risk_level=risk_level,
+            risk_score=total_score,
+            signals=signals,
+            lp_analysis=lp_analysis,
+            lock_analysis=lock_analysis,
+            score_breakdown=score_breakdown,
+            summary=summary,
+            scan_duration_ms=duration,
+        )
+
+    @property
+    def _suspicious_burn_addresses(self) -> set[str]:
+        """Built-in suspicious addresses merged with env-configured ones."""
+        result = set(SUSPICIOUS_BURN_ADDRESSES)
+        env_extra = os.environ.get("SUSPICIOUS_BURN_ADDRESSES", "")
+        if env_extra:
+            for addr in env_extra.split(","):
+                addr = addr.strip().lower()
+                if addr and addr != TRUE_BURN_ADDRESS:
+                    result.add(addr)
+        return result
+
+    def _check_burn_address(
+        self, lp_data: dict[str, Any]
+    ) -> dict[str, Any] | None:
+        """
+        Check if LP tokens are burned to a real dead address vs fake burn.
+        """
+        burn_address = lp_data.get("burn_address", "").lower()
+        if not burn_address:
+            return None
+
+        detail: dict[str, Any] = {
+            "burn_address": burn_address,
+            "is_true_burn": burn_address == TRUE_BURN_ADDRESS,
+        }
+
+        if burn_address in self._suspicious_burn_addresses:
+            return {
+                "severity": 0.4,
+                "description": (
+                    f"LP burned to {burn_address[:10]}... which may have "
+                    "known private keys. Not a true burn address."
+                ),
+                "detail": {**detail, "reason": "suspicious_burn_address"},
+            }
+
+        if burn_address == TRUE_BURN_ADDRESS:
+            return None  # Legitimate burn
+
+        # Unknown burn address — could be a contract or EOA
+        return {
+            "severity": 0.2,
+            "description": (
+                f"LP burned to unknown address {burn_address[:10]}... "
+                "Unable to verify this is a true dead address."
+            ),
+            "detail": {**detail, "reason": "unknown_burn_address"},
+        }
+
+    def _analyze_lock_status(
+        self, lock_data: dict[str, Any], lp_data: dict[str, Any]
+    ) -> dict[str, Any]:
+        """Analyze LP lock status."""
+        signals: list[LiquiditySignal] = []
+        analysis: dict[str, Any] = {
+            "platforms_found": [],
+            "is_locked": False,
+            "earliest_unlock_timestamp": None,
+            "lock_duration_days": None,
+            "lock_confidence": "low",
+        }
+        score_component = 0.0
+
+        lock_address = lock_data.get("locker_contract", "").lower()
+        unlock_timestamp = lock_data.get("unlock_timestamp")
+        total_locked = lock_data.get("total_locked_usd", 0)
+        total_liquidity = lp_data.get("total_liquidity_usd", 0)
+
+        if lock_address:
+            # Check if it's a known lock platform
+            platform_info = self._known_lock_addresses.get(lock_address)
+            if platform_info:
+                analysis["platforms_found"].append(platform_info["platform"])
+                analysis["lock_confidence"] = "high" if platform_info["reliable"] else "medium"
+                analysis["is_locked"] = True
+
+        if unlock_timestamp and isinstance(unlock_timestamp, (int, float)):
+            analysis["earliest_unlock_timestamp"] = unlock_timestamp
+            now = time.time()
+            remaining = unlock_timestamp - now
+            remaining_days = max(0, remaining / 86400)
+            analysis["lock_duration_days"] = round(remaining_days, 1)
+
+            if remaining_days < 1:
+                # Unlocking within 24 hours — critical
+                signals.append(
+                    LiquiditySignal(
+                        signal_type=SignalType.UNLOCKED_LP,
+                        severity=1.0,
+                        description=(
+                            f"LP unlock is imminent or past due! "
+                            f"{'Expired' if remaining < 0 else 'Unlocks in <1 day'}. "
+                            "Liquidity can be removed at any moment."
+                        ),
+                        detail={
+                            "unlock_timestamp": unlock_timestamp,
+                            "remaining_days": remaining_days,
+                            "remaining_seconds": int(remaining),
+                        },
+                    )
+                )
+                score_component = 45.0
+            elif remaining_days < 30:
+                # Unlocking within 1 month — high risk
+                signals.append(
+                    LiquiditySignal(
+                        signal_type=SignalType.UNLOCKED_LP,
+                        severity=0.7,
+                        description=(
+                            f"LP unlocks in {remaining_days:.0f} days "
+                            "(<1 month). Elevated withdrawal risk."
+                        ),
+                        detail={
+                            "unlock_timestamp": unlock_timestamp,
+                            "remaining_days": remaining_days,
+                        },
+                    )
+                )
+                score_component = 17.0
+            elif remaining_days < 180:
+                # Under 6 months — medium
+                signals.append(
+                    LiquiditySignal(
+                        signal_type=SignalType.UNLOCKED_LP,
+                        severity=0.3,
+                        description=(
+                            f"LP locked for {remaining_days:.0f} days "
+                            f"({remaining_days / 30:.1f} months). "
+                            "Reasonable but could be longer."
+                        ),
+                        detail={
+                            "unlock_timestamp": unlock_timestamp,
+                            "remaining_days": remaining_days,
+                        },
+                    )
+                )
+                score_component = 5.0
+        else:
+            # No unlock timestamp — LP may not be locked
+            if total_locked and total_locked > 0:
+                # Locked but no unlock time — can't verify
+                signals.append(
+                    LiquiditySignal(
+                        signal_type=SignalType.UNLOCKED_LP,
+                        severity=0.3,
+                        description=(
+                            "LP is in a lock contract but unlock timestamp "
+                            "is unknown. Cannot verify lock duration."
+                        ),
+                        detail={"total_locked_usd": total_locked},
+                    )
+                )
+                score_component = 5.0
+            elif total_liquidity and total_liquidity > 0:
+                # No lock at all — liquidity can be removed
+                signals.append(
+                    LiquiditySignal(
+                        signal_type=SignalType.UNLOCKED_LP,
+                        severity=0.9,
+                        description=(
+                            "No LP lock detected. Liquidity is completely "
+                            "unlocked and can be removed by the deployer at any time."
+                        ),
+                        detail={"total_liquidity_usd": total_liquidity},
+                    )
+                )
+                score_component = 22.0
+
+        return {
+            "signals": signals,
+            "analysis": analysis,
+            "score_component": score_component,
+        }
+
+    def _detect_flash_liquidity(
+        self, event_history: list[dict[str, Any]]
+    ) -> dict[str, Any] | None:
+        """Detect flash liquidity — added and removed within short period."""
+        if not event_history:
+            return None
+
+        add_events: list[dict[str, Any]] = []
+        remove_events: list[dict[str, Any]] = []
+
+        for event in event_history:
+            event_type = event.get("type", "").lower()
+            timestamp = event.get("timestamp", 0)
+            amount = event.get("amount_usd", 0)
+            if event_type == "add_liquidity" or event_type == "mint":
+                add_events.append({"timestamp": timestamp, "amount": amount})
+            elif event_type in ("remove_liquidity", "burn_lp", "withdraw_lp"):
+                remove_events.append({"timestamp": timestamp, "amount": amount})
+
+        if not add_events or not remove_events:
+            return None
+
+        # Check for rapid add then remove
+        for add in add_events:
+            for remove in remove_events:
+                if remove["timestamp"] > add["timestamp"]:
+                    time_diff_hours = (remove["timestamp"] - add["timestamp"]) / 3600
+                    if time_diff_hours < FLASH_LIQUIDITY_HOURS and time_diff_hours > 0:
+                        severity = max(0.3, min(1.0, 1.0 - (time_diff_hours / FLASH_LIQUIDITY_HOURS)))
+                        return {
+                            "severity": severity,
+                            "description": (
+                                f"Flash liquidity detected: ${remove['amount']:,.0f} "
+                                f"added and removed within {time_diff_hours:.1f} hours. "
+                                "This is a classic scam pattern — create appearance of "
+                                "liquidity, attract buyers, then drain."
+                            ),
+                            "detail": {
+                                "add_timestamp": add["timestamp"],
+                                "remove_timestamp": remove["timestamp"],
+                                "time_diff_hours": round(time_diff_hours, 1),
+                                "amount_added_usd": add["amount"],
+                                "amount_removed_usd": remove["amount"],
+                            },
+                        }
+
+        return None
+
+    def _check_claim_mismatch(
+        self,
+        lp_data: dict[str, Any],
+        lock_data: dict[str, Any],
+        market_data: dict[str, Any],
+    ) -> dict[str, Any] | None:
+        """Check if claimed liquidity status matches on-chain reality."""
+        claimed_locked_pct = lp_data.get("claimed_locked_pct")
+        actual_locked_pct = lp_data.get("actual_locked_pct")
+        claimed_total_liquidity = lp_data.get("claimed_total_liquidity_usd")
+        actual_total_liquidity = lp_data.get("total_liquidity_usd")
+
+        detail: dict[str, Any] = {}
+
+        if actual_locked_pct is not None and claimed_locked_pct is not None:
+            detail["claimed_locked_pct"] = claimed_locked_pct
+            detail["actual_locked_pct"] = actual_locked_pct
+            detail["discrepancy_pct"] = claimed_locked_pct - actual_locked_pct
+
+            if claimed_locked_pct > actual_locked_pct + 20:
+                return {
+                    "severity": 0.9,
+                    "description": (
+                        f"Claimed {claimed_locked_pct:.0f}% liquidity locked, "
+                        f"but on-chain shows only {actual_locked_pct:.0f}% locked. "
+                        f"Discrepancy of {claimed_locked_pct - actual_locked_pct:.0f}% "
+                        "— project is likely misrepresenting their lock status."
+                    ),
+                    "detail": detail,
+                }
+            elif claimed_locked_pct > actual_locked_pct + 5:
+                return {
+                    "severity": 0.5,
+                    "description": (
+                        f"Minor claim discrepancy: {claimed_locked_pct:.0f}% claimed "
+                        f"vs {actual_locked_pct:.0f}% actual locked."
+                    ),
+                    "detail": detail,
+                }
+
+        if actual_total_liquidity and claimed_total_liquidity:
+            detail["claimed_total_usd"] = claimed_total_liquidity
+            detail["actual_total_usd"] = actual_total_liquidity
+            ratio = claimed_total_liquidity / actual_total_liquidity if actual_total_liquidity else 1
+
+            if ratio > 2.0 or ratio < 0.5:
+                return {
+                    "severity": 0.6,
+                    "description": (
+                        f"Claimed liquidity (${claimed_total_liquidity:,.0f}) "
+                        f"differs significantly from on-chain "
+                        f"(${actual_total_liquidity:,.0f}). "
+                        f"Ratio: {ratio:.1f}x — potential misrepresentation."
+                    ),
+                    "detail": detail,
+                }
+
+        return None
+
+    def _check_lp_concentration(
+        self,
+        holder_data: list[dict[str, Any]],
+        lp_data: dict[str, Any],
+    ) -> dict[str, Any] | None:
+        """Check if LP tokens are concentrated in too few wallets."""
+        if not holder_data:
+            return None
+
+        top_holder_pct = 0.0
+        deployer_hold_pct = 0.0
+        deployer_address = lp_data.get("deployer_address", "").lower()
+
+        for holder in holder_data:
+            pct = holder.get("percentage", 0)
+            address = holder.get("address", "").lower()
+            if pct > top_holder_pct:
+                top_holder_pct = pct
+            if address == deployer_address:
+                deployer_hold_pct = pct
+
+        detail: dict[str, Any] = {
+            "top_holder_pct": top_holder_pct,
+            "deployer_hold_pct": deployer_hold_pct,
+            "num_holders": len(holder_data),
+        }
+
+        if deployer_hold_pct > LP_CONCENTRATION_THRESHOLD:
+            return {
+                "severity": 1.0,
+                "description": (
+                    f"Deployer holds {deployer_hold_pct:.1f}% of LP tokens. "
+                    "They can withdraw or manipulate liquidity at will."
+                ),
+                "detail": {**detail, "reason": "deployer_holds_majority_lp"},
+            }
+
+        if top_holder_pct > LP_CONCENTRATION_THRESHOLD:
+            return {
+                "severity": 0.6,
+                "description": (
+                    f"Top LP holder controls {top_holder_pct:.1f}% of LP tokens. "
+                    "High concentration allows coordinated liquidity removal."
+                ),
+                "detail": {**detail, "reason": "top_holder_majority"},
+            }
+
+        # Check if top 5 holders control >90%
+        top5_pct = sum(
+            h.get("percentage", 0) for h in holder_data[:5]
+        )
+        if top5_pct > 90:
+            return {
+                "severity": 0.4,
+                "description": (
+                    f"Top 5 wallets hold {top5_pct:.1f}% of LP tokens. "
+                    "Highly concentrated — small group controls liquidity."
+                ),
+                "detail": {**detail, "top5_pct": top5_pct, "reason": "top5_concentration"},
+            }
+
+        return None
+
+    def _check_liquidity_ratio(
+        self,
+        lp_data: dict[str, Any],
+        market_data: dict[str, Any],
+    ) -> dict[str, Any] | None:
+        """Check liquidity-to-market-cap ratio for anomalies."""
+        total_liquidity = lp_data.get("total_liquidity_usd", 0)
+        market_cap = market_data.get("market_cap_usd", 0)
+
+        if not total_liquidity or not market_cap:
+            return None
+
+        ratio = total_liquidity / market_cap if market_cap > 0 else 0
+
+        detail: dict[str, Any] = {
+            "liquidity_usd": total_liquidity,
+            "market_cap_usd": market_cap,
+            "ratio": round(ratio, 4),
+        }
+
+        if ratio < MIN_LIQUIDITY_RATIO:
+            severity = max(0.3, min(1.0, 1.0 - (ratio / MIN_LIQUIDITY_RATIO)))
+            return {
+                "severity": severity,
+                "description": (
+                    f"Liquidity-to-market-cap ratio is {ratio:.2%}. "
+                    f"Below {MIN_LIQUIDITY_RATIO:.0%} threshold — "
+                    "extremely thin liquidity relative to valuation. "
+                    "Selling may be impossible without catastrophic slippage."
+                ),
+                "detail": detail,
+            }
+
+        if ratio > 1.0:
+            return {
+                "severity": 0.3,
+                "description": (
+                    f"Liquidity ({total_liquidity:,.0f}) exceeds market cap "
+                    f"({market_cap:,.0f}) — unusual. May indicate "
+                    "illiquid tokens held by LP rather than real trading liquidity."
+                ),
+                "detail": detail,
+            }
+
+        return None
+
+    def _check_suspicious_timing(
+        self,
+        event_history: list[dict[str, Any]],
+        market_data: dict[str, Any],
+    ) -> dict[str, Any] | None:
+        """Check timing of liquidity events relative to price movements."""
+        if not event_history:
+            return None
+
+        price_changes = market_data.get("price_changes", {})
+        significant_drops = []
+
+        for event in event_history:
+            event_type = event.get("type", "").lower()
+            timestamp = event.get("timestamp", 0)
+            amount = event.get("amount_usd", 0)
+
+            # Liquidity removal around price drops
+            if event_type in ("remove_liquidity", "burn_lp", "withdraw_lp"):
+                # Check if price dropped within 24h after removal
+                for period, change in price_changes.items():
+                    if isinstance(change, (int, float)) and change < -20:
+                        # More than 20% price drop
+                        significant_drops.append({
+                            "event_timestamp": timestamp,
+                            "event_amount_usd": amount,
+                            "price_change_pct": change,
+                            "period": period,
+                        })
+
+        if significant_drops:
+            total_drained = sum(d["event_amount_usd"] for d in significant_drops if d["event_amount_usd"])
+            return {
+                "severity": 0.8,
+                "description": (
+                    f"{len(significant_drops)} liquidity removal(s) followed by "
+                    f"significant price declines. Total drained: ${total_drained:,.0f}. "
+                    "Pattern suggests coordinated liquidity removal preceding dump."
+                ),
+                "detail": {
+                    "events": significant_drops,
+                    "total_drained_usd": total_drained,
+                },
+            }
+
+        return None
+
+    def _check_single_sided_liquidity(
+        self, lp_data: dict[str, Any]
+    ) -> dict[str, Any] | None:
+        """Check for single-sided liquidity deposits."""
+        pairs = lp_data.get("pairs", [])
+        if not pairs:
+            return None
+
+        for pair in pairs:
+            token0 = pair.get("token0", "")
+            token1 = pair.get("token1", "")
+            reserve0 = pair.get("reserve0", 0)
+            reserve1 = pair.get("reserve1", 0)
+            pair_address = pair.get("address", "")
+
+            if reserve0 and reserve1:
+                ratio = reserve0 / reserve1 if reserve1 else 0
+                if ratio > 100 or ratio < 0.01:
+                    return {
+                        "severity": 0.7,
+                        "description": (
+                            f"Single-sided liquidity detected in pair {pair_address[:10]}... "
+                            f"Token ratio is 1:{ratio:.0f}. "
+                            "One side has negligible liquidity — "
+                            "may prevent selling or enable price manipulation."
+                        ),
+                        "detail": {
+                            "pair_address": pair_address,
+                            "reserve0": reserve0,
+                            "reserve1": reserve1,
+                            "ratio": round(ratio, 2),
+                        },
+                    }
+
+        return None
+
+    def _determine_risk_level(
+        self, score: float, signals: list[LiquiditySignal]
+    ) -> LiquidityRisk:
+        """Convert score to risk level with signal overrides."""
+        # Check for critical signals
+        for s in signals:
+            if s.signal_type in (SignalType.FAKE_BURN, SignalType.LIQUIDITY_DRAIN):
+                if s.severity > 0.8:
+                    return LiquidityRisk.CRITICAL
+
+        if score >= 70:
+            return LiquidityRisk.CRITICAL
+        if score >= 45:
+            return LiquidityRisk.HIGH
+        if score >= 20:
+            return LiquidityRisk.MEDIUM
+        if score >= 5:
+            return LiquidityRisk.LOW
+        return LiquidityRisk.SAFE
+
+    def _generate_summary(
+        self,
+        risk_level: LiquidityRisk,
+        score: float,
+        signals: list[LiquiditySignal],
+    ) -> str:
+        """Generate human-readable summary."""
+        signal_count = len(signals)
+        high_severity = sum(1 for s in signals if s.severity >= 0.7)
+
+        if risk_level == LiquidityRisk.CRITICAL:
+            return (
+                f"CRITICAL: {signal_count} scam signals detected "
+                f"({high_severity} high severity). "
+                "This token's liquidity is highly suspicious — "
+                "funds may be at immediate risk of being drained. AVOID."
+            )
+        if risk_level == LiquidityRisk.HIGH:
+            return (
+                f"HIGH RISK: {signal_count} liquidity anomalies found. "
+                f"Multiple red flags ({high_severity} critical). "
+                "Strong indicators of liquidity manipulation."
+            )
+        if risk_level == LiquidityRisk.MEDIUM:
+            return (
+                f"MEDIUM RISK: {signal_count} minor concerns detected. "
+                "Some liquidity practices are questionable. "
+                "Investigate further before committing funds."
+            )
+        if risk_level == LiquidityRisk.LOW:
+            return (
+                f"LOW RISK: {signal_count} minor signals found. "
+                "Liquidity appears mostly legitimate but has "
+                "minor areas of concern."
+            )
+        return (
+            "SAFE: No liquidity scam signals detected. "
+            "Liquidity appears properly locked and managed."
+        )
+
+
+# ═══════════════════════════════════════════════════════════════════
+# Helper Functions
+# ═══════════════════════════════════════════════════════════════════
+
+
+def format_report(report: LiquidityScamReport) -> str:
+    """Format a LiquidityScamReport as a human-readable string."""
+    lines: list[str] = [
+        f"🔍 Liquidity Scam Scan Report",
+        f"{'=' * 50}",
+        f"Token : {report.token_address[:20]}...",
+        f"Chain  : {report.chain}",
+        f"Risk   : {report.risk_level.value.upper()} ({report.risk_score:.1f}/100)",
+        f"Time   : {datetime.fromtimestamp(report.scan_timestamp, tz=timezone.utc).isoformat()}",
+        f"{'=' * 50}",
+    ]
+
+    if report.signals:
+        lines.append(f"\n🚨 Signals ({len(report.signals)}):")
+        for s in report.signals:
+            emoji = {
+                SignalType.FAKE_BURN: "🔥",
+                SignalType.UNLOCKED_LP: "🔓",
+                SignalType.FLASH_LIQUIDITY: "⚡",
+                SignalType.CLAIM_MISMATCH: "📢",
+                SignalType.LP_CONCENTRATION: "🎯",
+                SignalType.LOW_LIQUIDITY_RATIO: "📉",
+                SignalType.SUSPICIOUS_TIMING: "⏰",
+                SignalType.SINGLE_SIDED: "⚖️",
+                SignalType.FAKE_LOCK: "🔐",
+                SignalType.TIMELOCK_BYPASS: "⏳",
+                SignalType.CROSS_CHAIN_MANIPULATION: "🌐",
+                SignalType.LIQUIDITY_DRAIN: "💧",
+                SignalType.UNKNOWN: "❓",
+            }.get(s.signal_type, "⚠️")
+            severity_label = (
+                "🔴" if s.severity >= 0.7
+                else "🟡" if s.severity >= 0.4
+                else "🟢"
+            )
+            lines.append(f"  {emoji} {severity_label} [{s.signal_type.value}] {s.description[:120]}")
+
+    lines.append(f"\n📊 Score Breakdown:")
+    for key, value in sorted(report.score_breakdown.items(), key=lambda x: -x[1]):
+        if value > 0:
+            lines.append(f"  {key.replace('_', ' ').title():30s} {value:5.1f} pts")
+
+    lines.append(f"\n💡 {report.summary}")
+    return "\n".join(lines)
+
+
+# ═══════════════════════════════════════════════════════════════════
+# CLI Entry Point
+# ═══════════════════════════════════════════════════════════════════
+
+
+if __name__ == "__main__":
+    import sys
+
+    async def main() -> None:
+        detector = LiquidityScamDetector()
+        token = sys.argv[1] if len(sys.argv) > 1 else "0x1234567890abcdef1234567890abcdef12345678"
+        chain = sys.argv[2] if len(sys.argv) > 2 else "ethereum"
+
+        print(f"🔍 Scanning {token[:20]}... on {chain}")
+        print()
+
+        report = await detector.analyze(
+            token_address=token,
+            chain=chain,
+            lp_data={
+                "total_liquidity_usd": 50000,
+                "locked_liquidity_usd": 0,
+                "unlocked_liquidity_usd": 50000,
+                "burn_address": "0x0000000000000000000000000000000000000001",
+                "deployer_address": "0x1234567890abcdef1234567890abcdef12345678",
+                "pairs": [
+                    {
+                        "address": "0xpair1234567890abcdef1234567890abcdef1234",
+                        "token0": "0xabc...",
+                        "token1": "0xdef...",
+                        "reserve0": 1000000,
+                        "reserve1": 10,
+                    }
+                ],
+            },
+            lock_data={
+                "locker_contract": "",
+                "unlock_timestamp": None,
+                "total_locked_usd": 0,
+            },
+            holder_data=[
+                {"address": "0x1234567890abcdef1234567890abcdef12345678", "percentage": 85.0},
+                {"address": "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "percentage": 10.0},
+                {"address": "0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "percentage": 5.0},
+            ],
+            market_data={
+                "market_cap_usd": 2000000,
+            },
+            event_history=[
+                {"type": "add_liquidity", "timestamp": time.time() - 3600 * 12, "amount_usd": 50000},
+                {"type": "remove_liquidity", "timestamp": time.time() - 3600 * 2, "amount_usd": 45000},
+            ],
+        )
+
+        print(format_report(report))
+
+    asyncio.run(main())

backend/app/test_audit_report_validator.py

@@ -3,7 +3,7 @@ Tests for Audit Report Validator (audit_report_validator.py)
 """
 import pytest
 
-from audit_report_validator import (
+from app.audit_report_validator import (
     AuditClaim,
     AuditReportValidator,
     AuditRisk,

backend/app/test_liquidity_scam_detector.py

@@ -0,0 +1,458 @@
+"""
+Tests for Fake Liquidity / Liquidity Lock Scam Detector (liquidity_scam_detector.py)
+"""
+
+import asyncio
+import json
+import time
+import unittest
+from datetime import UTC, datetime
+
+from app.liquidity_scam_detector import (
+    LiquidityRisk,
+    LiquidityScamDetector,
+    LiquidityScamReport,
+    LiquiditySignal,
+    SignalType,
+    TRUE_BURN_ADDRESS,
+    SUSPICIOUS_BURN_ADDRESSES,
+    format_report,
+)
+
+
+class TestLiquidityScamDetector(unittest.TestCase):
+    """Test suite for LiquidityScamDetector."""
+
+    def setUp(self):
+        self.detector = LiquidityScamDetector()
+
+    # ═══════════════════════════════════════════════════════════════════
+    # Smoke Tests — No data = safe
+    # ═══════════════════════════════════════════════════════════════════
+
+    def test_empty_data_returns_safe(self):
+        """No data = safe (no red flags)."""
+        report = asyncio.run(self.detector.analyze(
+            token_address="0x1234567890abcdef1234567890abcdef12345678",
+            chain="ethereum",
+        ))
+        self.assertEqual(report.risk_level, LiquidityRisk.SAFE)
+        self.assertEqual(report.risk_score, 0.0)
+        self.assertEqual(len(report.signals), 0)
+
+    def test_minimal_data_no_signals(self):
+        """Minimal data with safe-looking values."""
+        report = asyncio.run(self.detector.analyze(
+            token_address="0x1234567890abcdef1234567890abcdef12345678",
+            chain="ethereum",
+            lp_data={
+                "total_liquidity_usd": 1000000,
+                "burn_address": TRUE_BURN_ADDRESS,
+            },
+            lock_data={
+                "locker_contract": "0xefb9b72a0c3e2737f7d9e2e8e7e1f9e2d8e7e1f9",
+                "unlock_timestamp": time.time() + 86400 * 365,  # 1 year from now
+            },
+        ))
+        self.assertEqual(report.risk_level, LiquidityRisk.SAFE)
+
+    # ═══════════════════════════════════════════════════════════════════
+    # Burn Address Tests
+    # ═══════════════════════════════════════════════════════════════════
+
+    def test_true_burn_address_no_signal(self):
+        """True burn address (0x...dead) should produce no signal."""
+        report = asyncio.run(self.detector.analyze(
+            token_address="0x1234",
+            chain="ethereum",
+            lp_data={"burn_address": TRUE_BURN_ADDRESS},
+        ))
+        fake_burn_signals = [
+            s for s in report.signals
+            if s.signal_type == SignalType.FAKE_BURN
+        ]
+        self.assertEqual(len(fake_burn_signals), 0)
+
+    def test_suspicious_burn_address_detected(self):
+        """Suspicious burn addresses should trigger a signal."""
+        for addr in SUSPICIOUS_BURN_ADDRESSES:
+            report = asyncio.run(self.detector.analyze(
+                token_address="0x1234",
+                chain="ethereum",
+                lp_data={"burn_address": addr},
+            ))
+            fake_burn_signals = [
+                s for s in report.signals
+                if s.signal_type == SignalType.FAKE_BURN
+            ]
+            self.assertEqual(
+                len(fake_burn_signals), 1,
+                f"Should detect {addr} as suspicious burn",
+            )
+
+    # ═══════════════════════════════════════════════════════════════════
+    # Lock Status Tests
+    # ═══════════════════════════════════════════════════════════════════
+
+    def test_no_lock_detected_high_risk(self):
+        """No lock contract = liquidity can be removed."""
+        report = asyncio.run(self.detector.analyze(
+            token_address="0x1234",
+            chain="ethereum",
+            lp_data={"total_liquidity_usd": 100000},
+            lock_data={},
+        ))
+        unlocked_signals = [
+            s for s in report.signals
+            if s.signal_type == SignalType.UNLOCKED_LP
+        ]
+        self.assertGreaterEqual(len(unlocked_signals), 1)
+        self.assertGreaterEqual(report.risk_score, 10)
+
+    def test_expiring_lock_high_risk(self):
+        """Lock expiring within hours should be critical."""
+        report = asyncio.run(self.detector.analyze(
+            token_address="0x1234",
+            chain="ethereum",
+            lp_data={"total_liquidity_usd": 100000},
+            lock_data={
+                "locker_contract": "0xefb9b72a0c3e2737f7d9e2e8e7e1f9e2d8e7e1f9",
+                "unlock_timestamp": time.time() + 3600,  # 1 hour
+            },
+        ))
+        unlocked_signals = [
+            s for s in report.signals
+            if s.signal_type == SignalType.UNLOCKED_LP
+        ]
+        self.assertGreaterEqual(len(unlocked_signals), 1)
+        self.assertIn(report.risk_level, (LiquidityRisk.HIGH, LiquidityRisk.CRITICAL))
+
+    def test_long_lock_duration_low_risk(self):
+        """Lock with 1+ year duration should be low risk."""
+        report = asyncio.run(self.detector.analyze(
+            token_address="0x1234",
+            chain="ethereum",
+            lp_data={"total_liquidity_usd": 100000},
+            lock_data={
+                "locker_contract": "0xefb9b72a0c3e2737f7d9e2e8e7e1f9e2d8e7e1f9",
+                "unlock_timestamp": time.time() + 86400 * 400,  # >1 year
+            },
+        ))
+        self.assertLessEqual(report.risk_score, 30)
+
+    # ═══════════════════════════════════════════════════════════════════
+    # Flash Liquidity Tests
+    # ═══════════════════════════════════════════════════════════════════
+
+    def test_flash_liquidity_detected(self):
+        """LP added and removed within hours should be flagged."""
+        now = time.time()
+        report = asyncio.run(self.detector.analyze(
+            token_address="0x1234",
+            chain="ethereum",
+            event_history=[
+                {"type": "add_liquidity", "timestamp": now - 7200, "amount_usd": 100000},
+                {"type": "remove_liquidity", "timestamp": now - 3600, "amount_usd": 95000},
+            ],
+        ))
+        flash_signals = [
+            s for s in report.signals
+            if s.signal_type == SignalType.FLASH_LIQUIDITY
+        ]
+        self.assertGreaterEqual(len(flash_signals), 1)
+        self.assertIn("flash", flash_signals[0].description.lower())
+
+    def test_stable_liquidity_no_flash_signal(self):
+        """Stable LP (no removal) should not trigger flash signal."""
+        now = time.time()
+        report = asyncio.run(self.detector.analyze(
+            token_address="0x1234",
+            chain="ethereum",
+            event_history=[
+                {"type": "add_liquidity", "timestamp": now - 86400 * 30, "amount_usd": 100000},
+                # No removal events
+            ],
+        ))
+        flash_signals = [
+            s for s in report.signals
+            if s.signal_type == SignalType.FLASH_LIQUIDITY
+        ]
+        self.assertEqual(len(flash_signals), 0)
+
+    # ═══════════════════════════════════════════════════════════════════
+    # Claim Mismatch Tests
+    # ═══════════════════════════════════════════════════════════════════
+
+    def test_claim_mismatch_detected(self):
+        """Claimed vs actual locked % discrepancy should be flagged."""
+        report = asyncio.run(self.detector.analyze(
+            token_address="0x1234",
+            chain="ethereum",
+            lp_data={
+                "claimed_locked_pct": 100,
+                "actual_locked_pct": 10,
+                "total_liquidity_usd": 100000,
+            },
+        ))
+        mismatch_signals = [
+            s for s in report.signals
+            if s.signal_type == SignalType.CLAIM_MISMATCH
+        ]
+        self.assertGreaterEqual(len(mismatch_signals), 1)
+
+    def test_consistent_claims_no_signal(self):
+        """Matching claims should not trigger."""
+        report = asyncio.run(self.detector.analyze(
+            token_address="0x1234",
+            chain="ethereum",
+            lp_data={
+                "claimed_locked_pct": 95,
+                "actual_locked_pct": 95,
+            },
+        ))
+        mismatch_signals = [
+            s for s in report.signals
+            if s.signal_type == SignalType.CLAIM_MISMATCH
+        ]
+        self.assertEqual(len(mismatch_signals), 0)
+
+    # ═══════════════════════════════════════════════════════════════════
+    # LP Concentration Tests
+    # ═══════════════════════════════════════════════════════════════════
+
+    def test_deployer_holds_majority_lp(self):
+        """Deployer holding >50% LP should be flagged."""
+        report = asyncio.run(self.detector.analyze(
+            token_address="0x1234",
+            chain="ethereum",
+            lp_data={
+                "deployer_address": "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa1",
+            },
+            holder_data=[
+                {"address": "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa1", "percentage": 80.0},
+                {"address": "0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "percentage": 20.0},
+            ],
+        ))
+        concentration_signals = [
+            s for s in report.signals
+            if s.signal_type == SignalType.LP_CONCENTRATION
+        ]
+        self.assertGreaterEqual(len(concentration_signals), 1)
+
+    def test_distributed_lp_no_signal(self):
+        """Well-distributed LP should not trigger."""
+        report = asyncio.run(self.detector.analyze(
+            token_address="0x1234",
+            chain="ethereum",
+            holder_data=[
+                {"address": "0xaaa", "percentage": 10.0},
+                {"address": "0xbbb", "percentage": 8.0},
+                {"address": "0xccc", "percentage": 7.0},
+                {"address": "0xddd", "percentage": 6.0},
+            ],
+        ))
+        concentration_signals = [
+            s for s in report.signals
+            if s.signal_type == SignalType.LP_CONCENTRATION
+        ]
+        self.assertEqual(len(concentration_signals), 0)
+
+    # ═══════════════════════════════════════════════════════════════════
+    # Liquidity Ratio Tests
+    # ═══════════════════════════════════════════════════════════════════
+
+    def test_low_liquidity_ratio_detected(self):
+        """Very low liquidity vs market cap should flag."""
+        report = asyncio.run(self.detector.analyze(
+            token_address="0x1234",
+            chain="ethereum",
+            lp_data={"total_liquidity_usd": 1000},
+            market_data={"market_cap_usd": 10000000},
+        ))
+        ratio_signals = [
+            s for s in report.signals
+            if s.signal_type == SignalType.LOW_LIQUIDITY_RATIO
+        ]
+        self.assertGreaterEqual(len(ratio_signals), 1)
+
+    def test_healthy_liquidity_ratio_no_signal(self):
+        """Healthy liquidity ratio should not trigger."""
+        report = asyncio.run(self.detector.analyze(
+            token_address="0x1234",
+            chain="ethereum",
+            lp_data={"total_liquidity_usd": 500000},
+            market_data={"market_cap_usd": 2000000},
+        ))
+        ratio_signals = [
+            s for s in report.signals
+            if s.signal_type == SignalType.LOW_LIQUIDITY_RATIO
+        ]
+        self.assertEqual(len(ratio_signals), 0)
+
+    # ═══════════════════════════════════════════════════════════════════
+    # Single-Sided Liquidity Tests
+    # ═══════════════════════════════════════════════════════════════════
+
+    def test_single_sided_liquidity_detected(self):
+        """Extreme reserve ratio should be flagged."""
+        report = asyncio.run(self.detector.analyze(
+            token_address="0x1234",
+            chain="ethereum",
+            lp_data={
+                "pairs": [
+                    {
+                        "address": "0xpair1234567890abcdef1234567890abcdef1234",
+                        "token0": "0xabc",
+                        "token1": "0xdef",
+                        "reserve0": 1000000,
+                        "reserve1": 10,
+                    }
+                ],
+            },
+        ))
+        single_sided_signals = [
+            s for s in report.signals
+            if s.signal_type == SignalType.SINGLE_SIDED
+        ]
+        self.assertGreaterEqual(len(single_sided_signals), 1)
+
+    def test_balanced_pool_no_signal(self):
+        """Balanced pool should not trigger single-sided signal."""
+        report = asyncio.run(self.detector.analyze(
+            token_address="0x1234",
+            chain="ethereum",
+            lp_data={
+                "pairs": [
+                    {
+                        "address": "0xpair1234",
+                        "token0": "0xabc",
+                        "token1": "0xdef",
+                        "reserve0": 1000000,
+                        "reserve1": 950000,
+                    }
+                ],
+            },
+        ))
+        single_sided_signals = [
+            s for s in report.signals
+            if s.signal_type == SignalType.SINGLE_SIDED
+        ]
+        self.assertEqual(len(single_sided_signals), 0)
+
+    # ═════════════════════════════���═════════════════════════════════════
+    # Signal Timing Tests
+    # ═══════════════════════════════════════════════════════════════════
+
+    def test_suspicious_timing_detected(self):
+        """Liquidity removal followed by price drop should be flagged."""
+        report = asyncio.run(self.detector.analyze(
+            token_address="0x1234",
+            chain="ethereum",
+            event_history=[
+                {"type": "remove_liquidity", "timestamp": time.time() - 86400 * 2, "amount_usd": 50000},
+            ],
+            market_data={
+                "price_changes": {"24h": -35.0},
+            },
+        ))
+        timing_signals = [
+            s for s in report.signals
+            if s.signal_type == SignalType.SUSPICIOUS_TIMING
+        ]
+        self.assertGreaterEqual(len(timing_signals), 1)
+
+    # ═══════════════════════════════════════════════════════════════════
+    # Integration / Multi-Signal Tests
+    # ═══════════════════════════════════════════════════════════════════
+
+    def test_multiple_red_flags_produce_high_score(self):
+        """Multiple scam indicators should produce high risk score."""
+        now = time.time()
+        report = asyncio.run(self.detector.analyze(
+            token_address="0x1234",
+            chain="ethereum",
+            lp_data={
+                "total_liquidity_usd": 50000,
+                "burn_address": "0x0000000000000000000000000000000000000001",
+                "deployer_address": "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa1",
+                "pairs": [
+                    {
+                        "address": "0xpair1",
+                        "token0": "0xabc",
+                        "token1": "0xdef",
+                        "reserve0": 990000,
+                        "reserve1": 1000,
+                    }
+                ],
+            },
+            lock_data={
+                "locker_contract": "",
+                "unlock_timestamp": None,
+            },
+            holder_data=[
+                {"address": "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa1", "percentage": 85.0},
+            ],
+            market_data={"market_cap_usd": 10000000},
+            event_history=[
+                {"type": "add_liquidity", "timestamp": now - 3600 * 6, "amount_usd": 50000},
+                {"type": "remove_liquidity", "timestamp": now - 3600 * 2, "amount_usd": 40000},
+            ],
+        ))
+        self.assertGreater(len(report.signals), 2)
+        self.assertGreater(report.risk_score, 40)
+        self.assertIn(report.risk_level, (LiquidityRisk.HIGH, LiquidityRisk.CRITICAL))
+
+    # ═══════════════════════════════════════════════════════════════════
+    # Format Report Test
+    # ═══════════════════════════════════════════════════════════════════
+
+    def test_format_report(self):
+        """format_report should produce valid output without errors."""
+        report = LiquidityScamReport(
+            token_address="0x1234",
+            chain="ethereum",
+            risk_level=LiquidityRisk.MEDIUM,
+            risk_score=35.0,
+            signals=[
+                LiquiditySignal(
+                    signal_type=SignalType.UNLOCKED_LP,
+                    severity=0.8,
+                    description="Test signal: LP not locked",
+                ),
+            ],
+            score_breakdown={"lock_status": 25.0},
+            summary="Test summary",
+        )
+        formatted = format_report(report)
+        self.assertIn("MEDIUM (35.0/100)", formatted)
+        self.assertIn("Test signal", formatted)
+        self.assertIn("Test summary", formatted)
+
+    def test_to_dict_serializable(self):
+        """to_dict() should produce JSON-serializable output."""
+        report = LiquidityScamReport(
+            token_address="0x1234",
+            chain="ethereum",
+            risk_level=LiquidityRisk.HIGH,
+            risk_score=65.0,
+            signals=[
+                LiquiditySignal(
+                    signal_type=SignalType.FAKE_BURN,
+                    severity=0.9,
+                    description="Fake burn test",
+                    detail={"address": "0xdead"},
+                ),
+            ],
+            score_breakdown={"fake_burn": 27.0},
+            summary="Test",
+        )
+        d = report.to_dict()
+        self.assertEqual(d["risk_level"], "high")
+        self.assertEqual(d["risk_score"], 65.0)
+        # Ensure it's JSON-serializable
+        json_str = json.dumps(d)
+        self.assertIn("Fake burn test", json_str)
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/app/test_wash_trading_detector.py

@@ -6,7 +6,7 @@ import asyncio
 import unittest
 from unittest.mock import patch
 
-from wash_trading_detector import (
+from app.wash_trading_detector import (
     CircularTrade,
     MatchedOrder,
     SelfTrade,
@@ -577,7 +577,7 @@ class TestEdgeCases(unittest.TestCase):
 
     def test_create_detector(self):
         """Factory function creates a valid detector."""
-        from wash_trading_detector import create_detector
+        from app.wash_trading_detector import create_detector
         d = create_detector()
         self.assertIsInstance(d, WashTradingDetector)
 

backend/scripts/minimax_review_liquidity.py

@@ -0,0 +1,62 @@
+"""MiniMax review script for liquidity_scam_detector.py"""
+import json
+import os
+import urllib.request
+from pathlib import Path
+
+
+def get_key() -> str | None:
+    """Try to find API key from env or .env files."""
+    key = os.environ.get("MINIMAX_API_KEY", "")
+    if key:
+        return key
+    # Check various .env files
+    for env_path in [
+        Path.home() / ".hermes" / ".env",
+        Path.cwd() / ".env",
+        Path.cwd().parent / ".env",
+        Path("/app/.env"),
+    ]:
+        if env_path.exists():
+            for line in env_path.read_text().splitlines():
+                if line.startswith("MINIMAX_API_KEY="):
+                    key = line.split("=", 1)[1].strip().strip("'\"")
+                    if key:
+                        return key
+    return None
+
+
+def main() -> None:
+    code = open("app/liquidity_scam_detector.py").read()[:5000]
+    key = get_key()
+    if not key:
+        print("NO_API_KEY_FOUND")
+        return
+
+    body = json.dumps({
+        "model": "MiniMax-Text-01",
+        "messages": [
+            {
+                "role": "system",
+                "content": "Review this Python code for bugs, security issues, and improvements. List top 3 issues found."
+            },
+            {"role": "user", "content": code}
+        ],
+        "max_tokens": 300
+    }).encode()
+
+    req = urllib.request.Request(
+        "https://api.minimax.io/v1/chat/completions",
+        data=body,
+        headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"}
+    )
+    try:
+        resp = urllib.request.urlopen(req, timeout=30)
+        result = json.loads(resp.read())
+        print(result["choices"][0]["message"]["content"])
+    except Exception as e:
+        print(f"API_ERROR: {e}")
+
+
+if __name__ == "__main__":
+    main()