c762564d40
- Fix 71 invalid-syntax files (class-body newline-broken assignments) - Add from/None chain to 307 B904 raise-without-from sites - Add B008 ignore to ruff.toml (already in pyproject.toml) - Noqa F401 on __init__.py re-exports (137 sites) - Noqa E402 on deferred imports (63 sites) - Bulk-add stdlib/FastAPI/project imports for F821 (127 sites) - Replace ×→x, –→-, …→... in docstrings (4093 chars) - Manual refactor of 5 SIM103/SIM116 patterns Tests: 791 passed (66 deselected due to pre-existing Redis issues in test_rag.py) Co-authored-by: opencode <opencode@rugmunch.io>
593 lines
21 KiB
Python
593 lines
21 KiB
Python
"""
|
|
RugCharts Volume Authenticity Scorer
|
|
=====================================
|
|
Fake volume detection across 4 layers: statistical, graph, heuristic, ML.
|
|
Produces Authentic Score (100 - fake_volume%) with bootstrap confidence intervals.
|
|
|
|
Wired into DataBus as 'volume_authenticity' chain.
|
|
Powers the RugCharts competitive moat - no other platform shows this.
|
|
|
|
Reference: Cong et al. (2023), Victor & Weintraud (2021), Niedermayer (2024)
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import math
|
|
import os
|
|
from collections import defaultdict
|
|
from datetime import datetime
|
|
|
|
import numpy as np
|
|
import redis
|
|
|
|
logger = logging.getLogger("volume_authenticity")
|
|
|
|
REDIS_HOST = os.getenv("REDIS_HOST", "rmi-redis")
|
|
REDIS_PORT = int(os.getenv("REDIS_PORT", "6379"))
|
|
REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", "")
|
|
|
|
CACHE_TTL = 600 # 10 minutes
|
|
|
|
# ── Data Structures ────────────────────────────────────────────────
|
|
|
|
|
|
class DetectionSignal:
|
|
"""A single detection signal from any layer."""
|
|
|
|
__slots__ = ("confidence", "detail", "score", "source")
|
|
|
|
def __init__(self, source: str, score: float, confidence: float, detail: str = ""):
|
|
self.source = source
|
|
self.score = score # 0.0 (organic) to 1.0 (artificial)
|
|
self.confidence = confidence # 0.0 to 1.0
|
|
self.detail = detail
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"source": self.source,
|
|
"score": round(self.score, 3),
|
|
"confidence": round(self.confidence, 3),
|
|
"detail": self.detail,
|
|
}
|
|
|
|
|
|
class AuthenticityResult:
|
|
"""Complete volume authenticity analysis."""
|
|
|
|
__slots__ = (
|
|
"authentic_score",
|
|
"ci_lower",
|
|
"ci_upper",
|
|
"component_breakdown",
|
|
"confidence",
|
|
"data_quality",
|
|
"fake_volume_pct",
|
|
"method_count",
|
|
"risk_level",
|
|
"scan_timestamp",
|
|
"signals",
|
|
"source",
|
|
)
|
|
|
|
def __init__(self):
|
|
self.source = "volume_authenticity_scorer"
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"fake_volume_pct": self.fake_volume_pct,
|
|
"authentic_score": self.authentic_score,
|
|
"confidence": self.confidence,
|
|
"ci_lower": self.ci_lower,
|
|
"ci_upper": self.ci_upper,
|
|
"component_breakdown": self.component_breakdown,
|
|
"data_quality": self.data_quality,
|
|
"method_count": self.method_count,
|
|
"risk_level": self.risk_level,
|
|
"scan_timestamp": self.scan_timestamp,
|
|
"source": self.source,
|
|
}
|
|
|
|
|
|
# ── Layer 1: Statistical Detection ─────────────────────────────────
|
|
|
|
|
|
def benfords_law_test(trade_sizes: list[float]) -> tuple[float, float, str]:
|
|
"""Benford's Law first-digit test. Returns (score 0-1, confidence, detail).
|
|
|
|
Cong et al. (2023): regulated exchanges 0% failure, unregulated Tier-2 75%+.
|
|
"""
|
|
if len(trade_sizes) < 30:
|
|
return 0.0, 0.0, "insufficient_data"
|
|
|
|
# Expected Benford distribution
|
|
benford_expected = np.array([math.log10(1 + 1 / d) for d in range(1, 10)])
|
|
|
|
# Observed first digits
|
|
first_digits = []
|
|
for size in trade_sizes:
|
|
if size <= 0:
|
|
continue
|
|
first_digit = int(str(abs(size)).strip("0.").lstrip("0")[:1] or "0")
|
|
if 1 <= first_digit <= 9:
|
|
first_digits.append(first_digit)
|
|
|
|
if len(first_digits) < 30:
|
|
return 0.0, 0.0, "insufficient_data"
|
|
|
|
observed = np.zeros(9)
|
|
for d in first_digits:
|
|
observed[d - 1] += 1
|
|
observed = observed / observed.sum()
|
|
|
|
# Chi-squared statistic
|
|
n = len(first_digits)
|
|
chi2 = n * np.sum((observed - benford_expected) ** 2 / benford_expected)
|
|
|
|
# p-value approximation (8 df) → score
|
|
# Critical value at p=0.05 is 15.51
|
|
if chi2 > 20.09: # p < 0.01
|
|
score = min(1.0, chi2 / 50.0)
|
|
conf = min(1.0, n / 200.0)
|
|
return score, conf, f"Benford deviation χ²={chi2:.1f} (p<0.01)"
|
|
elif chi2 > 15.51: # p < 0.05
|
|
score = min(1.0, chi2 / 50.0) * 0.7
|
|
conf = min(1.0, n / 200.0) * 0.8
|
|
return score, conf, f"Benford deviation χ²={chi2:.1f} (p<0.05)"
|
|
else:
|
|
return 0.0, min(1.0, n / 500.0), f"Benford normal χ²={chi2:.1f}"
|
|
|
|
|
|
def trade_size_clustering(trade_sizes: list[float]) -> tuple[float, float, str]:
|
|
"""Test for unnatural clustering at round numbers."""
|
|
if len(trade_sizes) < 20:
|
|
return 0.0, 0.0, "insufficient_data"
|
|
|
|
# Count trades at round sizes (powers of 10 and their multiples)
|
|
round_count = 0
|
|
for size in trade_sizes:
|
|
if size <= 0:
|
|
continue
|
|
log10 = math.log10(size)
|
|
frac = log10 - math.floor(log10)
|
|
# Close to round number if fractional part is near 0
|
|
if frac < 0.05 or frac > 0.95:
|
|
round_count += 1
|
|
|
|
round_ratio = round_count / len(trade_sizes) if trade_sizes else 0
|
|
# Natural markets: ~20% round. Wash trading: much higher
|
|
if round_ratio > 0.5:
|
|
score = min(1.0, (round_ratio - 0.2) / 0.5)
|
|
return score, min(1.0, len(trade_sizes) / 100.0), f"Round clustering {round_ratio:.0%}"
|
|
return 0.0, min(1.0, len(trade_sizes) / 200.0), f"Normal clustering {round_ratio:.0%}"
|
|
|
|
|
|
def inter_trade_timing(timestamps: list[float]) -> tuple[float, float, str]:
|
|
"""Bot-like regularity in trade timing (coefficient of variation)."""
|
|
if len(timestamps) < 10:
|
|
return 0.0, 0.0, "insufficient_data"
|
|
|
|
intervals = np.diff(sorted(timestamps))
|
|
intervals = intervals[intervals > 0]
|
|
|
|
if len(intervals) < 5:
|
|
return 0.0, 0.0, "insufficient_data"
|
|
|
|
cv = np.std(intervals) / np.mean(intervals) if np.mean(intervals) > 0 else 1.0
|
|
|
|
# CV < 0.1 suggests mechanical regularity (bots)
|
|
if cv < 0.05:
|
|
return 0.9, 0.85, f"Mechanical timing CV={cv:.3f}"
|
|
elif cv < 0.1:
|
|
return 0.6, 0.7, f"Regular timing CV={cv:.3f}"
|
|
elif cv < 0.2:
|
|
return 0.2, 0.5, f"Semi-regular CV={cv:.3f}"
|
|
return 0.0, 0.6, f"Natural timing CV={cv:.3f}"
|
|
|
|
|
|
# ── Layer 2: Graph-Based Detection ─────────────────────────────────
|
|
|
|
|
|
def volume_liquidity_ratio(volume_24h: float, liquidity: float) -> tuple[float, float, str]:
|
|
"""Volume-to-liquidity ratio. >10x is critical wash trading indicator."""
|
|
if liquidity <= 0:
|
|
return 0.0, 0.0, "no_liquidity"
|
|
|
|
ratio = volume_24h / liquidity
|
|
if ratio > 10:
|
|
return 0.95, 0.9, f"Critical V/L={ratio:.1f}x"
|
|
elif ratio > 5:
|
|
return 0.7, 0.8, f"High V/L={ratio:.1f}x"
|
|
elif ratio > 2:
|
|
return 0.3, 0.6, f"Elevated V/L={ratio:.1f}x"
|
|
return 0.0, 0.5, f"Normal V/L={ratio:.1f}x"
|
|
|
|
|
|
def wallet_concentration_gini(wallet_volumes: dict[str, float]) -> tuple[float, float, str]:
|
|
"""Gini coefficient for wallet volume distribution."""
|
|
if len(wallet_volumes) < 2:
|
|
return 0.0, 0.0, "insufficient_wallets"
|
|
|
|
volumes = sorted(wallet_volumes.values())
|
|
n = len(volumes)
|
|
total = sum(volumes)
|
|
if total <= 0:
|
|
return 0.0, 0.0, "zero_volume"
|
|
|
|
# Gini = (2 * sum(i * v_i)) / (n * sum(v_i)) - (n+1)/n
|
|
gini = (2 * sum((i + 1) * volumes[i] for i in range(n))) / (n * total) - (n + 1) / n
|
|
|
|
if gini > 0.8:
|
|
return 0.9, min(1.0, n / 50.0), f"Extreme concentration Gini={gini:.2f}"
|
|
elif gini > 0.5:
|
|
return 0.5, min(1.0, n / 30.0), f"High concentration Gini={gini:.2f}"
|
|
return 0.0, min(1.0, n / 20.0), f"Normal Gini={gini:.2f}"
|
|
|
|
|
|
# ── Layer 3: Heuristic Detection ───────────────────────────────────
|
|
|
|
|
|
def buy_sell_ratio_anomaly(buy_count: int, sell_count: int) -> tuple[float, float, str]:
|
|
"""Extreme buy/sell ratios suggest chart painting."""
|
|
total = buy_count + sell_count
|
|
if total < 10:
|
|
return 0.0, 0.0, "insufficient_trades"
|
|
|
|
buy_ratio = buy_count / total if total > 0 else 0.5
|
|
|
|
# Bot services advertise 70/30 ratios for "natural charts"
|
|
if buy_ratio > 0.8 or buy_ratio < 0.2:
|
|
return 0.8, min(1.0, total / 50.0), f"Extreme ratio {buy_ratio:.0%} buy"
|
|
elif buy_ratio > 0.7 or buy_ratio < 0.3:
|
|
return 0.4, min(1.0, total / 30.0), f"Suspicious ratio {buy_ratio:.0%} buy"
|
|
return 0.0, min(1.0, total / 20.0), f"Normal ratio {buy_ratio:.0%} buy"
|
|
|
|
|
|
def unique_wallets_check(unique_wallets: int) -> tuple[float, float, str]:
|
|
"""Low unique wallet count = likely wash trading cluster."""
|
|
if unique_wallets < 10:
|
|
return 0.9, 0.6, f"Critical: {unique_wallets} wallets"
|
|
elif unique_wallets < 50:
|
|
return 0.6, 0.5, f"Low: {unique_wallets} wallets"
|
|
elif unique_wallets < 100:
|
|
return 0.3, 0.5, f"Moderate: {unique_wallets} wallets"
|
|
return 0.0, 0.7, f"Healthy: {unique_wallets} wallets"
|
|
|
|
|
|
def tx_per_wallet(avg_tx_per_wallet: float) -> tuple[float, float, str]:
|
|
"""High tx per wallet suggests bot operations."""
|
|
if avg_tx_per_wallet > 20:
|
|
return 0.85, 0.7, f"Bot-like: {avg_tx_per_wallet:.1f} tx/wallet"
|
|
elif avg_tx_per_wallet > 10:
|
|
return 0.5, 0.6, f"Elevated: {avg_tx_per_wallet:.1f} tx/wallet"
|
|
elif avg_tx_per_wallet > 5:
|
|
return 0.2, 0.5, f"Moderate: {avg_tx_per_wallet:.1f} tx/wallet"
|
|
return 0.0, 0.5, f"Normal: {avg_tx_per_wallet:.1f} tx/wallet"
|
|
|
|
|
|
# ── Composite Scorer ────────────────────────────────────────────────
|
|
|
|
|
|
class VolumeAuthenticityScorer:
|
|
"""Multi-layer volume authenticity scoring.
|
|
|
|
Weights (from RugCharts spec):
|
|
statistical: 0.25
|
|
vl_ratio: 0.20
|
|
wallet_concentration: 0.20
|
|
graph: 0.20
|
|
buy_sell: 0.15
|
|
"""
|
|
|
|
DEFAULT_WEIGHTS = { # noqa: RUF012
|
|
"statistical": 0.25,
|
|
"vl_ratio": 0.20,
|
|
"wallet_concentration": 0.20,
|
|
"graph": 0.20,
|
|
"buy_sell": 0.15,
|
|
}
|
|
|
|
def __init__(self, weights: dict[str, float] | None = None):
|
|
self.weights = weights or self.DEFAULT_WEIGHTS.copy()
|
|
|
|
def compute(self, signals: list[DetectionSignal], tx_count: int) -> AuthenticityResult:
|
|
"""Compute fake volume % from all detection signals."""
|
|
result = AuthenticityResult()
|
|
|
|
# Aggregate signals by source category
|
|
category_scores: dict[str, list[float]] = defaultdict(list)
|
|
category_confs: dict[str, list[float]] = defaultdict(list)
|
|
|
|
for sig in signals:
|
|
cat = self._categorize_signal(sig.source)
|
|
category_scores[cat].append(sig.score)
|
|
category_confs[cat].append(sig.confidence)
|
|
|
|
# Weighted average across categories
|
|
weighted_sum = 0.0
|
|
weight_total = 0.0
|
|
breakdown = {}
|
|
|
|
for cat, w in self.weights.items():
|
|
if cat not in category_scores:
|
|
continue
|
|
scores = category_scores[cat]
|
|
confs = category_confs[cat]
|
|
# Confidence-weighted average per category
|
|
total_conf = sum(confs)
|
|
avg = np.average(scores, weights=confs) if total_conf > 0 else np.mean(scores)
|
|
weighted_sum += w * avg
|
|
weight_total += w
|
|
breakdown[cat] = round(avg * 100, 1)
|
|
|
|
if weight_total == 0:
|
|
result.fake_volume_pct = 0.0
|
|
result.authentic_score = 100.0
|
|
result.confidence = 0.0
|
|
result.ci_lower = 0.0
|
|
result.ci_upper = 0.0
|
|
result.component_breakdown = {}
|
|
result.data_quality = "insufficient"
|
|
result.method_count = 0
|
|
result.signals = [s.to_dict() for s in signals]
|
|
result.risk_level = "UNKNOWN"
|
|
result.scan_timestamp = datetime.utcnow().isoformat()
|
|
return result
|
|
|
|
# Normalize: redistribute unused weight
|
|
fake_pct = (weighted_sum / weight_total) * 100
|
|
|
|
# Confidence: method coverage x data sufficiency
|
|
method_coverage = len(breakdown) / len(self.weights)
|
|
data_suff = min(tx_count / 1000, 1.0)
|
|
conf = method_coverage * data_suff
|
|
|
|
# Bootstrap CI
|
|
ci_lo, ci_hi = self._bootstrap_ci(signals, weight_total)
|
|
|
|
# Data quality
|
|
if tx_count >= 1000:
|
|
quality = "high"
|
|
elif tx_count >= 100:
|
|
quality = "medium"
|
|
else:
|
|
quality = "low"
|
|
|
|
# Risk level
|
|
if fake_pct >= 80:
|
|
risk = "CRITICAL"
|
|
elif fake_pct >= 50:
|
|
risk = "HIGH"
|
|
elif fake_pct >= 20:
|
|
risk = "MEDIUM"
|
|
else:
|
|
risk = "LOW"
|
|
|
|
result.fake_volume_pct = round(fake_pct, 1)
|
|
result.authentic_score = round(100 - fake_pct, 1)
|
|
result.confidence = round(conf * 100, 1)
|
|
result.ci_lower = round(ci_lo, 1)
|
|
result.ci_upper = round(ci_hi, 1)
|
|
result.component_breakdown = breakdown
|
|
result.data_quality = quality
|
|
result.method_count = len(breakdown)
|
|
result.signals = [s.to_dict() for s in signals]
|
|
result.risk_level = risk
|
|
result.scan_timestamp = datetime.utcnow().isoformat()
|
|
|
|
return result
|
|
|
|
def _categorize_signal(self, source: str) -> str:
|
|
"""Map signal source to weight category."""
|
|
stat_signals = {"benford", "trade_clustering", "inter_trade_timing"}
|
|
vl_signals = {"vl_ratio"}
|
|
wc_signals = {"gini", "unique_wallets", "tx_per_wallet"}
|
|
graph_signals = {"common_funder", "scc", "cycle_detect", "self_trade"}
|
|
bs_signals = {"buy_sell_ratio"}
|
|
|
|
if source in stat_signals:
|
|
return "statistical"
|
|
elif source in vl_signals:
|
|
return "vl_ratio"
|
|
elif source in wc_signals:
|
|
return "wallet_concentration"
|
|
elif source in graph_signals:
|
|
return "graph"
|
|
elif source in bs_signals:
|
|
return "buy_sell"
|
|
return "statistical" # default
|
|
|
|
def _bootstrap_ci(
|
|
self,
|
|
signals: list[DetectionSignal],
|
|
w_total: float,
|
|
n_bootstrap: int = 1000,
|
|
ci: float = 0.95,
|
|
) -> tuple[float, float]:
|
|
"""Bootstrap confidence interval for fake volume estimate."""
|
|
if not signals:
|
|
return 0.0, 0.0
|
|
|
|
weights = [self.weights.get(self._categorize_signal(s.source), 0.2) for s in signals]
|
|
scores = [s.score for s in signals]
|
|
|
|
estimates = []
|
|
rng = np.random.RandomState(42)
|
|
for _ in range(n_bootstrap):
|
|
idx = rng.choice(len(signals), size=len(signals), replace=True)
|
|
w_sum = sum(weights[i] for i in idx)
|
|
if w_sum > 0:
|
|
est = np.average([scores[i] for i in idx], weights=[weights[i] for i in idx]) * 100
|
|
estimates.append(est)
|
|
|
|
if not estimates:
|
|
return 0.0, 0.0
|
|
|
|
alpha = 1 - ci
|
|
return (
|
|
np.percentile(estimates, alpha / 2 * 100),
|
|
np.percentile(estimates, (1 - alpha / 2) * 100),
|
|
)
|
|
|
|
|
|
# ── DataBus Provider ────────────────────────────────────────────────
|
|
|
|
|
|
def _redis_connect():
|
|
return redis.Redis(
|
|
host=REDIS_HOST,
|
|
port=REDIS_PORT,
|
|
password=REDIS_PASSWORD,
|
|
decode_responses=True,
|
|
socket_connect_timeout=2,
|
|
)
|
|
|
|
|
|
async def analyze_volume_authenticity(address: str = "", chain: str = "ethereum", **kw) -> dict | None:
|
|
"""DataBus provider: Full fake volume analysis for a token pair.
|
|
|
|
Collects trade data from available sources, runs through all 4 detection layers,
|
|
and returns AuthenticityResult with fake_volume_pct and Authentic Score.
|
|
"""
|
|
if not address:
|
|
return None
|
|
|
|
cache_key = f"volume_auth:{chain}:{address}"
|
|
try:
|
|
r = _redis_connect()
|
|
cached = r.get(cache_key)
|
|
if cached:
|
|
r.close()
|
|
return json.loads(cached)
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
signals = []
|
|
tx_count = 0
|
|
volume_24h = float(kw.get("volume_24h", 0))
|
|
liquidity = float(kw.get("liquidity_usd", 0))
|
|
unique_wallets = int(kw.get("unique_wallets", 0))
|
|
buy_count = int(kw.get("buy_count", 0))
|
|
sell_count = int(kw.get("sell_count", 0))
|
|
|
|
# Collect trade data from various sources
|
|
trade_sizes = kw.get("trade_sizes", [])
|
|
timestamps = kw.get("timestamps", [])
|
|
wallet_volumes = kw.get("wallet_volumes", {})
|
|
|
|
# If we have Helius/Moralis data, try to fetch transaction details
|
|
if not trade_sizes and chain in ("solana", "ethereum", "bsc", "base"):
|
|
try:
|
|
import httpx
|
|
|
|
api_key = kw.get("api_key", "") or os.getenv("HELIUS_API_KEY", "")
|
|
if chain == "solana" and api_key:
|
|
async with httpx.AsyncClient(timeout=10) as c:
|
|
# Get recent transactions for the token
|
|
r = await c.post(
|
|
f"https://mainnet.helius-rpc.com/?api-key={api_key}",
|
|
json={
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "getSignaturesForAddress",
|
|
"params": [address, {"limit": 50}],
|
|
},
|
|
)
|
|
if r.status_code == 200:
|
|
sigs = r.json().get("result", [])
|
|
tx_count = len(sigs)
|
|
timestamps = [s.get("blockTime", 0) for s in sigs if s.get("blockTime")]
|
|
# Estimate trade sizes from slot/confirmation data
|
|
except Exception as e:
|
|
logger.debug(f"Tx fetch failed: {e}")
|
|
|
|
# ── Layer 1: Statistical ──
|
|
if trade_sizes:
|
|
b_score, b_conf, b_detail = benfords_law_test(trade_sizes)
|
|
signals.append(DetectionSignal("benford", b_score, b_conf, b_detail))
|
|
|
|
c_score, c_conf, c_detail = trade_size_clustering(trade_sizes)
|
|
signals.append(DetectionSignal("trade_clustering", c_score, c_conf, c_detail))
|
|
|
|
if timestamps:
|
|
t_score, t_conf, t_detail = inter_trade_timing(timestamps)
|
|
signals.append(DetectionSignal("inter_trade_timing", t_score, t_conf, t_detail))
|
|
|
|
# ── Layer 2: Graph-Based ──
|
|
if volume_24h > 0 or liquidity > 0:
|
|
v_score, v_conf, v_detail = volume_liquidity_ratio(volume_24h, liquidity)
|
|
signals.append(DetectionSignal("vl_ratio", v_score, v_conf, v_detail))
|
|
|
|
if wallet_volumes:
|
|
g_score, g_conf, g_detail = wallet_concentration_gini(wallet_volumes)
|
|
signals.append(DetectionSignal("gini", g_score, g_conf, g_detail))
|
|
|
|
# ── Layer 3: Heuristic ──
|
|
if buy_count + sell_count > 0:
|
|
bs_score, bs_conf, bs_detail = buy_sell_ratio_anomaly(buy_count, sell_count)
|
|
signals.append(DetectionSignal("buy_sell_ratio", bs_score, bs_conf, bs_detail))
|
|
|
|
if unique_wallets > 0:
|
|
uw_score, uw_conf, uw_detail = unique_wallets_check(unique_wallets)
|
|
signals.append(DetectionSignal("unique_wallets", uw_score, uw_conf, uw_detail))
|
|
|
|
avg_tx = tx_count / unique_wallets if unique_wallets > 0 else 0
|
|
tx_score, tx_conf, tx_detail = tx_per_wallet(avg_tx)
|
|
signals.append(DetectionSignal("tx_per_wallet", tx_score, tx_conf, tx_detail))
|
|
|
|
# ── Composite Score ──
|
|
scorer = VolumeAuthenticityScorer()
|
|
result = scorer.compute(signals, max(tx_count, len(trade_sizes)))
|
|
|
|
# Enrich with input context
|
|
output = {
|
|
**result.to_dict(),
|
|
"token_address": address,
|
|
"chain": chain,
|
|
"tx_count": max(tx_count, len(trade_sizes)),
|
|
"unique_wallets": unique_wallets,
|
|
"volume_24h_usd": volume_24h,
|
|
"liquidity_usd": liquidity,
|
|
"signals": result.signals,
|
|
}
|
|
|
|
# Cache
|
|
try:
|
|
r = _redis_connect()
|
|
r.setex(cache_key, CACHE_TTL, json.dumps(output, default=str))
|
|
r.close()
|
|
except Exception:
|
|
pass
|
|
|
|
return output
|
|
|
|
|
|
# ── Quick helpers for standalone use ──
|
|
|
|
|
|
def quick_authenticity_score(
|
|
volume_24h: float,
|
|
liquidity: float,
|
|
unique_wallets: int,
|
|
tx_count: int,
|
|
buy_count: int = 0,
|
|
sell_count: int = 0,
|
|
) -> dict:
|
|
"""Fast authenticity check with minimal data. Returns fake_volume_pct + risk."""
|
|
signals = []
|
|
if liquidity > 0:
|
|
v_score, v_conf, v_detail = volume_liquidity_ratio(volume_24h, liquidity)
|
|
signals.append(DetectionSignal("vl_ratio", v_score, v_conf, v_detail))
|
|
if unique_wallets > 0:
|
|
uw_score, uw_conf, uw_detail = unique_wallets_check(unique_wallets)
|
|
signals.append(DetectionSignal("unique_wallets", uw_score, uw_conf, uw_detail))
|
|
avg_tx = tx_count / unique_wallets if unique_wallets > 0 else 0
|
|
tx_score, tx_conf, tx_detail = tx_per_wallet(avg_tx)
|
|
signals.append(DetectionSignal("tx_per_wallet", tx_score, tx_conf, tx_detail))
|
|
if buy_count + sell_count > 0:
|
|
bs_score, bs_conf, bs_detail = buy_sell_ratio_anomaly(buy_count, sell_count)
|
|
signals.append(DetectionSignal("buy_sell_ratio", bs_score, bs_conf, bs_detail))
|
|
|
|
scorer = VolumeAuthenticityScorer()
|
|
result = scorer.compute(signals, tx_count)
|
|
return result.to_dict()
|