RugMunchMedia/rmi-backend
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

chore(arch): delete dead code — 4 ai_pipeline variants and stale app/domain/

Browse source 
- Remove ai_pipeline.py, ai_pipeline2.py, ai_pipeline_v2.py, ai_pipeline_v3.py (661 lines of dead code, 0 imports)
- Remove stale app/domain/ directory (empty, only __pycache__ remains)
This commit is contained in:
Crypto Rug Munch 2026-07-09 14:27:11 +02:00
parent 52a5b45e09
commit 21c5a564fa
4 changed files with 0 additions and 661 deletions
Download patch file Download diff file Expand all files Collapse all files

119
app/ai_pipeline.py
View file

@ -1,119 +0,0 @@
#!/usr/bin/env python3
"""
RMI AI Pipeline - Batch Ollama Cloud Modules
=============================================
Wallet Profiling | RAG Enrichment | Alert Ranking | Market Briefing | Post-Mortem
All use Ollama Cloud deepseek-v4-flash. ~$0.001 per operation.
"""
import json
import logging
import os
import httpx
logger = logging.getLogger("rmi.ai_pipeline")
OLLAMA_KEY = os.getenv("OLLAMA_API_KEY", os.getenv("DEEPSEEK_API_KEY", ""))
OLLAMA_URL = "https://ollama.com/v1/chat/completions"
MODEL = "deepseek-v4-flash"
def _call_ai(system: str, prompt: str, max_tokens: int = 200, temp: float = 0.3) -> str:
try:
body = json.dumps(
{
"model": MODEL,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": prompt},
],
"max_tokens": max_tokens,
"temperature": temp,
}
).encode()
with httpx.Client(timeout=15) as client:
resp = client.post(
OLLAMA_URL,
content=body,
headers={
"Authorization": f"Bearer {OLLAMA_KEY}",
"Content-Type": "application/json",
},
)
resp.raise_for_status()
return resp.json()["choices"][0]["message"]["content"].strip()
except Exception as e:
logger.error(f"AI call failed: {e}")
return ""
# ── 7. WALLET BEHAVIORAL PROFILING ──
WALLET_SYSTEM = """Classify a crypto wallet into a persona based on transaction patterns.
Reply with ONLY: persona_name|confidence_0-100
Personas:
- Day Trader: frequent buys/sells, short holds, high volume
- Whale Accumulator: large buys, holds long, rare sells
- Bot Farm: identical transaction patterns, same gas, rapid-fire
- Insider: buys before pumps, sells before dumps, too perfect timing
- Honeypot Victim: bought tokens that can't be sold
- Scam Deployer: creates tokens, drains liquidity, repeats
- Airdrop Hunter: tiny transactions, hundreds of tokens, zero holds
- Diamond Hands: bought once, never sold, regardless of price
- Degen Gambler: buys meme coins, holds minutes, high risk tolerance
- Unknown: insufficient data"""
def profile_wallet(tx_data: dict) -> str:
summary = json.dumps(tx_data)[:1000]
result = _call_ai(WALLET_SYSTEM, f"Transactions:\n{summary}", max_tokens=30)
return result if "|" in result else "Unknown|0"
# ── 9. RAG QUERY ENRICHMENT ──
RAG_SYSTEM = """You reformat raw RAG search results into a coherent, readable answer.
Keep it under 150 words. Preserve key facts. Add a 1-line summary at the end."""
def enrich_rag_results(query: str, raw_docs: str) -> str:
return _call_ai(RAG_SYSTEM, f"Query: {query}\n\nRaw results:\n{raw_docs[:2000]}")
# ── 12. ALERT PRIORITIZATION ──
ALERT_SYSTEM = """Rank these crypto security alerts by urgency. Reply ONLY with the alert IDs in priority order, comma-separated.
Priority rules: CRITICAL (immediate rug/hack) > HIGH (likely scam) > MEDIUM (suspicious) > LOW (noise)."""
def rank_alerts(alerts: list) -> list:
summary = "\n".join(
f"ID:{a.get('id', '?')} | {a.get('severity', '?')} | {a.get('title', '?')[:100]}"
for a in alerts[:20]
)
result = _call_ai(ALERT_SYSTEM, summary, max_tokens=50)
return [x.strip() for x in result.split(",") if x.strip()]
# ── 6. DAILY MARKET BRIEFING ──
MARKET_SYSTEM = """Write a 3-paragraph daily crypto market briefing from scanner data.
Para 1: Market overview (most scanned chains, scan volume)
Para 2: Top risks (worst tokens found today, emerging patterns)
Para 3: What to watch (trending scam types, new threat vectors)
Use Telegram HTML formatting. Keep it under 250 words. Professional but direct tone."""
def generate_market_briefing(scan_summary: dict) -> str:
return _call_ai(MARKET_SYSTEM, json.dumps(scan_summary)[:2000], max_tokens=350, temp=0.5)
# ── 15. INCIDENT POST-MORTEM ──
AUTOPSY_SYSTEM = """Write a forensic post-mortem of a crypto scam incident.
Structure:
1. What happened (1 sentence)
2. How it worked (the mechanics, 2-3 sentences)
3. Red flags that were visible beforehand
4. How to protect against similar scams
Keep it under 200 words. Use <b>bold</b> for key findings. Professional forensic tone."""
def write_post_mortem(incident: dict) -> str:
return _call_ai(AUTOPSY_SYSTEM, json.dumps(incident)[:1500], max_tokens=300, temp=0.4)

118
app/ai_pipeline2.py
View file

@ -1,118 +0,0 @@
#!/usr/bin/env python3
"""
RMI AI Pipeline Part 2 - Remaining 7 Modules
=============================================
Community Forensics | Cross-Chain Entity | Ghost Blog | Social Media | Token Compare
All Ollama Cloud deepseek-v4-flash. ~$0.001/operation.
"""
import json
import logging
import os
import httpx
logger = logging.getLogger("rmi.ai_pipeline2")
OLLAMA_KEY = os.getenv("OLLAMA_API_KEY", os.getenv("DEEPSEEK_API_KEY", ""))
OLLAMA_URL = "https://ollama.com/v1/chat/completions"
MODEL = "deepseek-v4-flash"
def _call_ai(system: str, prompt: str, max_tokens: int = 250, temp: float = 0.3) -> str:
try:
body = json.dumps(
{
"model": MODEL,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": prompt},
],
"max_tokens": max_tokens,
"temperature": temp,
}
).encode()
with httpx.Client(timeout=15) as client:
resp = client.post(
OLLAMA_URL,
content=body,
headers={
"Authorization": f"Bearer {OLLAMA_KEY}",
"Content-Type": "application/json",
},
)
resp.raise_for_status()
return resp.json()["choices"][0]["message"]["content"].strip()
except Exception as e:
logger.error(f"AI call failed: {e}")
return ""
# ── 8. COMMUNITY FORENSICS AUTO-ANALYSIS ──
FORENSICS_SYSTEM = """You are a crypto forensics investigator. A community member submitted a suspicious token for review.
Analyze the information and provide:
1. Initial verdict (LIKELY SCAM / SUSPICIOUS / NEEDS MORE INFO)
2. Key concerns (2-3 bullet points)
3. Recommended next steps for the investigator
Keep it under 150 words."""
def analyze_community_submission(submission: dict) -> str:
return _call_ai(FORENSICS_SYSTEM, json.dumps(submission)[:1500], max_tokens=250)
# ── 10. CROSS-CHAIN ENTITY DETECTION ──
CROSSCHAIN_SYSTEM = """You identify crypto entities operating across multiple blockchains.
Given wallet data from different chains, determine if they're the same entity.
Reply format: MATCH|confidence_0-100|reason OR NO_MATCH|reason"""
def detect_cross_chain(wallets: dict) -> str:
return _call_ai(CROSSCHAIN_SYSTEM, json.dumps(wallets)[:1500], max_tokens=100)
# ── 11. GHOST BLOG AUTO-DRAFT ──
GHOST_SYSTEM = """You are a crypto security blogger for Rug Munch Intelligence (rugmunch.io).
Write a blog post draft from scanner data and incident reports.
Structure:
- Title (catchy, SEO-friendly, under 80 chars)
- Hook (1 sentence that grabs attention)
- Body (3-4 paragraphs explaining the threat)
- Key takeaways (2-3 bullet points)
- Call to action (check your tokens, use our scanner)
Use markdown formatting. Professional but engaging tone."""
def draft_blog_post(topic: str, data: dict) -> str:
prompt = f"Topic: {topic}\n\nData:\n{json.dumps(data)[:2000]}"
return _call_ai(GHOST_SYSTEM, prompt, max_tokens=500, temp=0.6)
# ── 13. SOCIAL MEDIA POST GENERATOR ──
SOCIAL_SYSTEM = """You are the social media manager for Rug Munch Intelligence (@CryptoRugMunch).
Write a tweet/telegram post about a crypto security finding.
Rules:
- Under 280 chars for Twitter, under 500 for Telegram
- Start with a hook (stat, warning, or question)
- Include $TICKER if relevant
- End with a call to action or link
- Use emojis sparingly (1-2 max)
- No hashtag spam (2-3 max)
Reply format: TWITTER: <tweet> | TELEGRAM: <post>"""
def generate_social_post(incident: dict, platform: str = "both") -> str:
return _call_ai(SOCIAL_SYSTEM, json.dumps(incident)[:1000], max_tokens=200, temp=0.7)
# ── 14. TOKEN COMPARISON ENGINE ──
COMPARE_SYSTEM = """Compare two crypto tokens for safety. Given their scanner results, determine which is safer and why.
Reply format:
SAFER: <token_name>
REASON: <2-3 sentence comparison>
SCORE_DIFF: <token1_score> vs <token2_score>
KEY_DIFFERENCES: <bullet points>"""
def compare_tokens(token_a: dict, token_b: dict) -> str:
prompt = f"Token A:\n{json.dumps(token_a)[:800]}\n\nToken B:\n{json.dumps(token_b)[:800]}"
return _call_ai(COMPARE_SYSTEM, prompt, max_tokens=200)

170
app/ai_pipeline_v2.py
View file

@ -1,170 +0,0 @@
"""
RMI AI Pipeline v2 - Production Grade
======================================
Caching, fallbacks, rate limiting, smart prompts.
All 12 modules battle-tested against Ollama Cloud.
"""
import hashlib
import json
import logging
import os
import time
import httpx
logger = logging.getLogger("rmi.ai")
OLLAMA_KEY = os.getenv("OLLAMA_API_KEY", "")
OLLAMA_URL = "https://ollama.com/v1/chat/completions"
MODEL = "deepseek-v4-flash"
CACHE_TTL = 300 # 5 min cache for identical calls
# Simple TTL cache
_cache = {}
def _cached_call(system: str, prompt: str, max_tokens: int = 250, temp: float = 0.3) -> str:
key = hashlib.sha256(f"{system[:50]}|{prompt[:100]}".encode()).hexdigest()
now = time.time()
if key in _cache and now - _cache[key][0] < CACHE_TTL:
return _cache[key][1]
try:
body = json.dumps(
{
"model": MODEL,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": prompt},
],
"max_tokens": max_tokens,
"temperature": temp,
}
).encode()
with httpx.Client(timeout=12) as client:
resp = client.post(
OLLAMA_URL,
content=body,
headers={
"Authorization": f"Bearer {OLLAMA_KEY}",
"Content-Type": "application/json",
},
)
resp.raise_for_status()
result = resp.json()["choices"][0]["message"]["content"].strip()
_cache[key] = (now, result)
return result
except Exception as e:
logger.error(f"Ollama AI call failed: {e}")
return ""
# ── 1. TOKEN RISK EXPLAINER (improved) ──
def explain_risks(scan: dict) -> str:
if not scan or scan.get("safety_score") is None:
return "<b>Unable to analyze</b> - no scanner data."
score = scan.get("safety_score", 50)
flags = scan.get("risk_flags", [])
green = scan.get("green_flags", [])
name = scan.get("name", scan.get("symbol", "token"))
mods = len(scan.get("modules_run", []))
prompt = f"Token:{name} Score:{score}/100 Risks:{', '.join(flags[:5]) or 'none'} Green:{', '.join(green[:3]) or 'none'} Modules:{mods}"
system = """You explain token risk to non-technical users. 3-4 sentences. Start with safety score. Mention top risks in plain English. End with "Always DYOR." Use <b>bold</b> for key terms. Never give financial advice."""
result = _cached_call(system, prompt, max_tokens=150, temp=0.2)
return result or f"<b>Safety: {score}/100</b>. Risk flags: {', '.join(flags[:3])}. Always DYOR."
# ── 2. NEWS CLASSIFIER (improved) ──
def classify_news(title: str, content: str = "") -> str:
text = f"{title} {content[:200]}"
system = """Classify crypto news into ONE word: SCAM MARKET REGULATION SECURITY DEFI MEMECOIN GENERAL"""
result = _cached_call(system, text, max_tokens=8, temp=0.1)
if result:
for cat in ["SCAM", "MARKET", "REGULATION", "SECURITY", "DEFI", "MEMECOIN", "GENERAL"]:
if cat in result.upper():
return cat
# Fast fallback
t = text.lower()
if any(w in t for w in ["hack", "exploit", "rug", "scam", "phish", "drain"]):
return "SCAM"
if any(w in t for w in ["price", "btc", "eth", "bull", "bear", "market"]):
return "MARKET"
return "GENERAL"
# ── 3. WALLET PROFILER ──
def profile_wallet(tx: dict) -> str:
system = """Classify wallet persona from tx data. Reply: PERSONA|confidence. Options: DayTrader Whale BotFarm Insider ScamDeployer AirdropHunter DiamondHands DegenGambler Unknown"""
return _cached_call(system, json.dumps(tx)[:1000], max_tokens=25) or "Unknown|0"
# ── 4. RAG ENRICHER ──
def enrich_rag(query: str, docs: str) -> str:
system = """Reformat RAG chunks into 2-3 sentence coherent answer. Preserve key facts."""
return _cached_call(system, f"Q:{query}\nD:{docs[:2000]}", max_tokens=200) or docs[:400]
# ── 5. ALERT RANKER ──
def rank_alerts(alerts: list) -> list:
summary = "\n".join(
f"{a.get('id', '?')}|{a.get('severity', '?')}|{(a.get('title', '') or '')[:80]}"
for a in alerts[:10]
)
result = _cached_call("Rank these by urgency. Reply: id1,id2,id3...", summary, max_tokens=50)
return [x.strip() for x in (result or "").split(",") if x.strip()]
# ── 6. MARKET BRIEFING ──
def briefing(data: dict) -> str:
system = """3-paragraph crypto market briefing. P1:volume+chains P2:top risks P3:what to watch. <b>bold</b> key findings. Under 250 words."""
return (
_cached_call(system, json.dumps(data)[:2000], max_tokens=350, temp=0.5)
or "Briefing unavailable."
)
# ── 7. INCIDENT AUTOPSY ──
def post_mortem(incident: dict) -> str:
system = """Crypto scam forensic post-mortem. What happened→How→Red flags→Protection. <b>bold</b> findings. Under 200 words."""
return (
_cached_call(system, json.dumps(incident)[:1500], max_tokens=300, temp=0.4)
or "Autopsy unavailable."
)
# ── 8. COMMUNITY FORENSICS ──
def analyze_submission(sub: dict) -> str:
system = """Analyze suspicious token submission. Verdict:LIKELY SCAM/SUSPICIOUS/MORE INFO + 2-3 concerns."""
return _cached_call(system, json.dumps(sub)[:1500], max_tokens=200) or "Analysis unavailable."
# ── 9. CROSS-CHAIN DETECTION ──
def cross_chain(wallets: dict) -> str:
system = """Same entity across chains? Reply: MATCH|conf|reason or NO_MATCH|reason"""
return _cached_call(system, json.dumps(wallets)[:1500], max_tokens=80) or "Unknown"
# ── 10. BLOG DRAFT ──
def blog_draft(topic: str, data: dict) -> str:
system = """Crypto security blog post draft. Title|Hook|Body(3-4para)|KeyTakeaways|CTA. Markdown. Professional."""
return (
_cached_call(
system, f"Topic:{topic}\nData:{json.dumps(data)[:2000]}", max_tokens=500, temp=0.6
)
or f"# {topic}\n\nDraft unavailable."
)
# ── 11. SOCIAL POSTS ──
def social_post(incident: dict) -> str:
system = """Tweet+Telegram post about crypto security finding. Twitter:<280 chars> | Telegram:<500 chars>. Hook first."""
return (
_cached_call(system, json.dumps(incident)[:1000], max_tokens=200, temp=0.7)
or "Post unavailable."
)
# ── 12. TOKEN COMPARE ──
def compare_tokens(a: dict, b: dict) -> str:
system = """Compare 2 tokens for safety. SAFER:<name> REASON:<2sentences> SCORE_DIFF:<a vs b> KEY_DIFFERENCES:<bullets>"""
prompt = f"A:{json.dumps(a)[:800]}\nB:{json.dumps(b)[:800]}"
return _cached_call(system, prompt, max_tokens=200) or "Comparison unavailable."

254
app/ai_pipeline_v3.py
View file

@ -1,254 +0,0 @@
"""
RMI AI Pipeline v3 - Full Production
=====================================
Redis caching, FastAPI endpoints, usage tracking, retry logic.
"""
import contextlib
import hashlib
import json
import logging
import os
import time
from datetime import UTC, datetime
import httpx
logger = logging.getLogger("rmi.ai_v3")
OLLAMA_KEY = os.getenv("OLLAMA_API_KEY", "")
OLLAMA_URL = "https://ollama.com/v1/chat/completions"
MODEL = "deepseek-v4-flash"
# ── Redis Cache (survives restarts) ──
REDIS_AVAILABLE = False
try:
import redis
_redis = redis.Redis(
host=os.getenv("REDIS_HOST", "rmi-redis"),
port=int(os.getenv("REDIS_PORT", "6379")),
password=os.getenv("REDIS_PASSWORD", ""),
db=1,
socket_connect_timeout=2,
)
_redis.ping()
REDIS_AVAILABLE = True
except Exception:
logging.getLogger(__name__).warning("swallowed exception", exc_info=True)
def _cache_get(key: str) -> str | None:
if REDIS_AVAILABLE:
try:
return _redis.get(f"rmi:ai:{key}")
except Exception:
logging.getLogger(__name__).warning("swallowed exception", exc_info=True)
return None
def _cache_set(key: str, value: str, ttl: int = 300):
if REDIS_AVAILABLE:
with contextlib.suppress(BaseException):
_redis.setex(f"rmi:ai:{key}", ttl, value)
# ── Usage Tracking ──
_usage = {"total_calls": 0, "total_tokens": 0, "total_cost": 0.0}
def _track(prompt_tokens: int, completion_tokens: int, cost: float):
_usage["total_calls"] += 1
_usage["total_tokens"] += prompt_tokens + completion_tokens
_usage["total_cost"] += cost
def usage_stats() -> dict:
return {**_usage, "timestamp": datetime.now(UTC).isoformat()}
# ── Retry with Exponential Backoff ──
def _call_ollama(
system: str, prompt: str, max_tokens: int = 250, temp: float = 0.3, cache_ttl: int = 300
) -> str:
cache_key = hashlib.sha256(f"{system[:60]}|{prompt[:120]}".encode()).hexdigest()
cached = _cache_get(cache_key)
if cached:
val = cached.decode() if isinstance(cached, bytes) else cached
if isinstance(val, str):
return val
for attempt in range(3):
try:
body = json.dumps(
{
"model": MODEL,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": prompt},
],
"max_tokens": max_tokens,
"temperature": temp,
}
).encode()
req_headers = {
"Authorization": f"Bearer {OLLAMA_KEY}",
"Content-Type": "application/json",
}
with httpx.Client(timeout=12) as client:
resp = client.post(OLLAMA_URL, content=body, headers=req_headers)
resp.raise_for_status()
data = resp.json()
result = data["choices"][0]["message"]["content"].strip()
usage = data.get("usage", {})
_track(usage.get("prompt_tokens", 0), usage.get("completion_tokens", 0), 0.000001)
_cache_set(cache_key, result, cache_ttl)
return result
except Exception as e:
if attempt < 2:
time.sleep(2**attempt)
else:
logger.warning(f"Ollama failed after 3 retries: {e}")
return ""
# ── ALL 12 MODULES (Unified) ──
def explain_risks(scan: dict) -> str:
s = scan.get("safety_score", 50)
f = scan.get("risk_flags", [])
g = scan.get("green_flags", [])
n = scan.get("name", scan.get("symbol", "token"))
r = _call_ollama(
"Explain token risk to non-technical user. 3-4 sentences. Start with safety score. Use <b>bold</b>. End with DYOR.",
f"Token:{n} Score:{s}/100 Risks:{', '.join(f[:5]) or 'none'} Green:{', '.join(g[:3]) or 'none'}",
150,
0.2,
600,
)
return r or f"<b>Safety: {s}/100</b>. Risk flags: {', '.join(f[:3])}. Always DYOR."
def classify_news(title: str, content: str = "") -> str:
r = _call_ollama(
"Classify crypto news: SCAM MARKET REGULATION SECURITY DEFI MEMECOIN GENERAL. Reply ONE word.",
f"{title} {content[:200]}",
8,
0.1,
3600,
)
for cat in ["SCAM", "MARKET", "REGULATION", "SECURITY", "DEFI", "MEMECOIN"]:
if cat in r.upper():
return cat
t = (title + content).lower()
if any(w in t for w in ["hack", "exploit", "rug", "scam", "drain"]):
return "SCAM"
if any(w in t for w in ["price", "btc", "eth", "bull", "bear"]):
return "MARKET"
return "GENERAL"
def profile_wallet(tx: dict) -> str:
return (
_call_ollama(
"Classify wallet persona: PERSONA|conf. DayTrader Whale BotFarm Insider ScamDeployer AirdropHunter DiamondHands DegenGambler Unknown",
json.dumps(tx)[:1000],
25,
)
or "Unknown|0"
)
def enrich_rag(query: str, docs: str) -> str:
return (
_call_ollama(
"Reformat RAG chunks into 2-3 sentence answer.", f"Q:{query}\nD:{docs[:2000]}", 200
)
or docs[:400]
)
def rank_alerts(alerts: list) -> list:
s = "\n".join(
f"{a.get('id', '?')}|{a.get('severity', '?')}|{str(a.get('title', ''))[:80]}"
for a in alerts[:10]
)
r = _call_ollama("Rank by urgency. Reply: id1,id2,id3...", s, 50)
return [x.strip() for x in r.split(",") if x.strip()] if r else []
def briefing(data: dict) -> str:
return (
_call_ollama(
"3-para crypto market briefing. P1:volume P2:risks P3:watch. <b>bold</b>. 250 words.",
json.dumps(data)[:2000],
350,
0.5,
1800,
)
or "Briefing unavailable."
)
def post_mortem(incident: dict) -> str:
return (
_call_ollama(
"Forensic post-mortem: What→How→RedFlags→Protection. <b>bold</b>. 200 words.",
json.dumps(incident)[:1500],
300,
0.4,
3600,
)
or "Autopsy unavailable."
)
def analyze_submission(sub: dict) -> str:
return (
_call_ollama("Analyze suspicious token. Verdict+2-3 concerns.", json.dumps(sub)[:1500], 200)
or "Analysis unavailable."
)
def cross_chain(wallets: dict) -> str:
return (
_call_ollama(
"Same entity across chains? MATCH|conf|reason or NO_MATCH|reason",
json.dumps(wallets)[:1500],
80,
)
or "Unknown"
)
def blog_draft(topic: str, data: dict) -> str:
return (
_call_ollama(
"Blog post: Title|Hook|Body|Takeaways|CTA. Markdown.",
f"Topic:{topic}\n{json.dumps(data)[:2000]}",
500,
0.6,
3600,
)
or f"# {topic}\n\nDraft unavailable."
)
def social_post(incident: dict) -> str:
return (
_call_ollama(
"Tweet(<280)+Telegram(<500). Hook first.", json.dumps(incident)[:1000], 200, 0.7
)
or "Post unavailable."
)
def compare_tokens(a: dict, b: dict) -> str:
return (
_call_ollama(
"Compare 2 tokens: SAFER name REASON SCORE_DIFF KEY_DIFFERENCES",
f"A:{json.dumps(a)[:800]}\nB:{json.dumps(b)[:800]}",
200,
)
or "Comparison unavailable."
)