Pump.fun 上的五个 Grok 智能体：永不休息的 Meme 币流水线

Pump.fun 上的五个 Grok 智能体：永不停歇的 Memecoin 流水线 2026年8月26日 · 23分钟阅读 · 查看原文 ↗ Grok Bot Crypto AI Marketing 2026 年 8 月 11 日，xAI 推出了 Grok Bot 的测试版。这不是一个拥有扩展上下文窗口的聊天机器人。它是一个持久的云端智能体，拥有自己的虚拟电脑：浏览器、文件系统、终端、你各种工具的登录。它在笔记本合上时依然工作，能接着上次中断的地方继续任务，并记住你喜欢事情怎么做。xAI 称这些为"永远在线的 AI 队友"，而这是头一次，这种级别的智能体栈以订阅形式提供，而不需要你自己搭基础设施。 与此同时，pump.fun 依然是 Solana 上最大的 memecoin 工厂：任何人在两分钟内、花大约两美元就能发行一个代币，一条联合曲线（bonding curve）机械地把价格推上去，当市值达到六万九千美元时，代币自动迁移到一个 DEX。每天有成千上万个代币发行。其中百分之一到百分之二能"毕业"。其余的归零。 这篇文章讲的是把这两样东西连成一条流水线：一个 Grok 智能体全天候监控 pump.fun 的发行，过滤噪音，按一组信号给代币打分，然后在早期曲线上买入或放弃。附带架构、代码、一个 GitHub 骨架，以及关于为什么大多数人会亏钱的诚实统计。 为什么用 Grok Bot，而不是一个脚本 在 pump.fun 上做 memecoin 交易，是一个用一次性提示词无法解决、用脚本也很难解决的任务。代币全天候发行，入场窗口以分钟计，而买入决策需要同时评估好几个参数：联合曲线、社交信号、钱包行为、叙事。 一个脚本可以监控发行并按指标过滤。但一个脚本无法评估一个代币的叙事：为什么这个 meme 可能爆火而那个不会。一个脚本无法读懂创建者的 Twitter 帖子，判断代币背后是有社区还是只是一次性砸盘。一个脚本无法根据当前的市场状态调整策略。 Grok Bot 正好填补这个空白。它的智能体架构包含四个专门化的智能体：一个协调者、一个研究员、一个逻辑学家和一个唱反调者。它们并行工作，在给出回答前交叉验证彼此的结论。对一条 memecoin 流水线来说，这意味着一个智能体拉取曲线数据，另一个分析社交信号，第三个检查入场逻辑，第四个扮演魔鬼代言人、寻找不买的理由。 再加上持久性：Grok Bot 运行在一台不关机的云端电脑上。它可以整夜监控 pump.fun，在凌晨三点发现一个很有潜力的发行，按每个信号评估它，到早上交付一份带建议的报告。或者，如果你配置了自动执行，它会在早期曲线上买入并设置止损。 流水线架构 流水线由五个阶段组成，每个阶段都能自主运行，并把结果传给下一个。 > 发行监控器。一条到 pump.fun API 的 WebSocket 连接，按基本指标过滤新代币流：年龄、曲线上的成交量、独立买家数量、是否有元数据。代币分析器。对每个通过过滤的代币，智能体收集扩展数据：创建者的资料、社交链接、第一批钱包（狙击手、内幕者、捆绑者）的行为、来自数据提供方的风险评分。叙事打分器。智能体评估 meme 潜力：叙事的贴合度、传播性、是否有社区、是否踩中当前趋势。这是脚本做不到、而 LLM 能做到的部分。决策。所有信号汇入一个打分矩阵。代币要么超过阈值进入执行，要么被记为带着理由的"跳过"而记录下来。执行与监控。通过联合曲线上的一笔 Solana 交易（或毕业后通过 Jupiter）进行买入，带设定的仓位大小和止损。 class PumpFunPipeline: """Pipeline skeleton: monitor -> filter -> analyze -> decide -> execute.""" def __init__(self, config: dict): self.ws = WebSocketMonitor(config["ws_url"]) self.analyzer = TokenAnalyzer(config["data_api_key"]) self.scorer = NarrativeScorer(config["grok_api_key"]) self.executor = TradeExecutor(config["wallet_key"], config["rpc_url"]) self.risk = RiskManager(config["max_position_sol"], config["daily_loss_limit"]) self.log = TradeLog(config["log_path"]) async def run(self): async for token in self.ws.stream_launches(): # FILTER: basic metrics on the stream if not self.passes_basic_filter(token): continue # ANALYZE: extended data + risk score analysis = await self.analyzer.analyze(token) if analysis.risk_score > 7: # 1-10, higher = more dangerous self.log.skip(token, "high_risk", analysis.risk_score) continue # NARRATIVE: LLM evaluates meme potential narrative = await self.scorer.evaluate(token, analysis) # DECIDE: scoring matrix score = self.compute_score(analysis, narrative) if score < self.config["min_score"]: self.log.skip(token, "low_score", score) continue # BRAKE: check limits before execution if not self.risk.can_trade(): self.log.skip(token, "risk_limit") break # EXECUTE tx = await self.executor.buy(token, self.risk.position_size(score)) self.log.entry(token, tx, score, analysis, narrative) def passes_basic_filter(self, token) -> bool: return ( token.unique_buyers >= 5 and # not an empty curve token.bonding_curve_pct < 40 and # still early, not near graduation token.has_metadata and # has a description and image token.age_minutes > 2 # survived the first 2 minutes ) def compute_score(self, analysis, narrative) -> float: weights = { "risk_inverse": 0.25, # lower risk = better "social_signal": 0.20, # creator and community activity "narrative_fit": 0.25, # match with the current trend "curve_health": 0.15, # healthy curve without snipers "wallet_diversity": 0.15, # buyer diversity } return sum( weights[k] * getattr(analysis, k, 0) * getattr(narrative, k, 1) for k in weights ) 发行监控器：WebSocket 与流过滤 pump.fun 每天产出成千上万个代币。大多数一出生就死了：没有元数据、零买家，创建者一分钟后就把它们弃置了。流水线的第一项工作，就是在流上把死掉的过滤掉，而不用对每个代币都花算力做深度分析。 来自数据提供方（Solana Tracker、PumpPortal、NoLimitNodes）的 WebSocket 流，实时投递带着基本指标的发行：代币地址、元数据、联合曲线状态、交易数。这一层的过滤又快又便宜：我们只放过有元数据（描述、图片、链接）、独立买家超过五个、且曲线填充不到百分之四十（这样毕业前还有增长潜力）的代币。 import asyncio import websockets import json class WebSocketMonitor: def __init__(self, url: str): self.url = url async def stream_launches(self): async with websockets.connect(self.url) as ws: # subscribe to new tokens await ws.send(json.dumps({ "method": "subscribeNewToken", "params": {"launchpad": "pumpfun"} })) async for msg in ws: data = json.loads(msg) if data.get("type") == "newToken": yield Token.from_ws(data) class Token: @classmethod def from_ws(cls, data: dict): return cls( address=data["mint"], name=data.get("name", ""), symbol=data.get("symbol", ""), description=data.get("description", ""), image_url=data.get("image", ""), twitter=data.get("twitter", ""), website=data.get("website", ""), bonding_curve_pct=data.get("progress", 0), unique_buyers=data.get("uniqueHolders", 0), volume_sol=data.get("volumeSOL", 0), age_minutes=data.get("ageMinutes", 0), has_metadata=bool(data.get("name") and data.get("image")), risk_score=data.get("riskScore", 5), ) 这一层有两个要点。第一：年龄过滤（age_minutes > 2）会切掉那些一创建就在第一秒被机器人扫货的狙击代币。如果一个代币存活了两分钟并且有自然的买家，它就已经超过了流上百分之九十的代币。第二：来自数据提供方的风险评分（Solana Tracker 根据狙击手、内幕者、捆绑者、流动性深度和合约权限给出 1 到 10 的评级）是第一个自动过滤器，在花 LLM 调用去分析之前先切掉有毒代币。 代币分析器：扩展数据 对于一个通过基本过滤的代币，智能体通过 REST API 收集扩展数据。 import httpx class TokenAnalyzer: def __init__(self, api_key: str): self.client = httpx.AsyncClient( base_url="https://data.solanatracker.io", headers={"x-api-key": api_key}, ) async def analyze(self, token: Token) -> dict: # pull data from several endpoints in parallel token_data, holders, trades = await asyncio.gather( self.client.get(f"/tokens/{token.address}"), self.client.get(f"/tokens/{token.address}/holders/top"), self.client.get(f"/tokens/{token.address}/trades", params={"limit": 50}), ) td = token_data.json() top_holders = holders.json() recent_trades = trades.json() # WALLET ANALYSIS: look for sniper and insider patterns sniper_count = sum(1 for h in top_holders if h.get("is_sniper", False)) insider_pct = sum(h["percentage"] for h in top_holders[:5]) / 100 # CURVE HEALTH: even distribution is better than concentration curve_health = 1.0 - insider_pct # 0 = all with top 5, 1 = distributed # SOCIAL SIGNALS has_twitter = bool(td.get("twitter")) has_website = bool(td.get("website")) has_telegram = bool(td.get("telegram")) social_signal = (has_twitter + has_website + has_telegram) / 3 return { "risk_score": td.get("risk", {}).get("score", 5), "sniper_count": sniper_count, "insider_pct": insider_pct, "curve_health": curve_health, "social_signal": social_signal, "wallet_diversity": len(set(t["wallet"] for t in recent_trades)) / max(len(recent_trades), 1), "volume_sol": td.get("pools", [{}])[0].get("liquidity", {}).get("quote", 0), "market_cap": td.get("pools", [{}])[0].get("marketCap", {}).get("usd", 0), } 这里的关键是，分析同时在好几个层面上运行。风险评分是一个聚合值，但它背后站着具体的指标：头部持有者中有多少狙击手、前五个钱包持有多少份额的代币（越高，砸盘概率越大）、创建者有没有社交渠道。一个既没有 Twitter 也没有网站、且百分之四十集中在一个钱包里的代币，不是投资，是陷阱，而代码分析器会在花 LLM 调用之前就切掉这类情况。 但指标只能捕捉已知的模式。新颖的分配方案、不寻常的钱包行为、十个钱包同属一个操作者并按顺序买入的复杂捆绑链，数值过滤器是看不出来的。为此你需要第二层分析，而第一个 LLM 智能体就在这里登场。 审计智能体：Grok 读取钱包行为 审计者接收原始交易和持有者数据，寻找指标覆盖不到的模式。一个地址买入了，紧接着第二个地址在三秒后出现、买入完全相同的数量，然后是第三个。对脚本来说，这是三个独立的买家。对能看到时间间隔和金额的 LLM 来说，这是一次协同购买，它会这么说。 class WalletAuditor: def __init__(self, api_key: str, model: str = "grok-4-fast"): self.client = httpx.AsyncClient( base_url="https://api.x.ai/v1", headers={"Authorization": f"Bearer {api_key}"}, timeout=30, ) self.model = model async def audit(self, token: Token, trades: list[dict], holders: list[dict]) -> dict: trades_summary = "\n".join( f" {t['wallet'][:8]}... | {t['side']} | {t['amount_sol']:.3f} SOL | " f"+{t['seconds_after_launch']}s" for t in trades[:30] ) holders_summary = "\n".join( f" {h['address'][:8]}... | {h['percentage']:.1f}% | " f"{'sniper' if h.get('is_sniper') else 'organic'}" for h in holders[:10] ) prompt = f"""You are a wallet auditor for a new memecoin on pump.fun. Transactions (first 30): {trades_summary} Top holders: {holders_summary} Find and rate: 1. coordinated_buys: are there coordinated purchases (same amounts, intervals under 5s, linked addresses) 2. wash_trading: signs of wash trading (one wallet buying and selling repeatedly) 3. creator_dump_risk: does it look like the creator or an insider is preparing a dump 4. organic_score: what fraction of buyers look organic Reply ONLY JSON: {{"coordinated_buys": false, "wash_trading": false, "creator_dump_risk": 0.0, "organic_score": 0.0}}""" resp = await self.client.post("/chat/completions", json={ "model": self.model, "messages": [{"role": "user", "content": prompt}], "temperature": 0, }) text = resp.json()["choices"][0]["message"]["content"] try: return json.loads(text.strip().strip("`").replace("json", "").strip()) except json.JSONDecodeError: return {"coordinated_buys": True, "wash_trading": True, "creator_dump_risk": 1.0, "organic_score": 0.0} 注意解析错误时的回退：它不会默默放行，而是返回最悲观的数值。如果 LLM 崩了或返回垃圾，流水线就跳过而不是买入。这条规则对流水线里的每个智能体都成立：错误等于拒绝，而不是静默跳过检查。 叙事打分器：LLM 给出脚本给不了的东西的地方 这就是 Grok 登场的地方。一个脚本能计算指标，但无法评估为什么"CATBALD"这个 meme 这周可能爆火而"RANDOMDOG"不会。叙事是那个唯一的变量，能把毕业的那百分之一到百分之二的代币，变成毕业后带来数倍回报的代币。 Grok 智能体接收上下文：代币的描述、图片、创建者的 Twitter、当前的加密 Twitter 趋势，并交付一份结构化的结论。 class NarrativeScorer: def __init__(self, api_key: str): self.client = httpx.AsyncClient( base_url="https://api.x.ai/v1", headers={"Authorization": f"Bearer {api_key}"}, ) async def evaluate(self, token: Token, analysis: dict) -> dict: prompt = f"""You are analyzing a new memecoin on pump.fun. Token: {token.name} ({token.symbol}) Description: {token.description} Creator Twitter: {token.twitter or 'none'} Website: {token.website or 'none'} Age: {token.age_minutes} minutes Bonding curve: {token.bonding_curve_pct}% Unique buyers: {token.unique_buyers} Risk score: {analysis['risk_score']}/10 Top-5 wallets hold: {analysis['insider_pct']:.0%} Rate 0.0-1.0: 1. narrative_fit: does the meme match a current trend (events, viral topics) 2. virality: viral potential (humor, recognition, shock value) 3. community: signs of a real community behind the token 4. timing: is the launch timing right for market mood Reply ONLY JSON, no explanation: {{"narrative_fit": 0.0, "virality": 0.0, "community": 0.0, "timing": 0.0}}""" response = await self.client.post("/chat/completions", json={ "model": "grok-4-fast", "messages": [{"role": "user", "content": prompt}], "temperature": 0, }) text = response.json()["choices"][0]["message"]["content"] return json.loads(text.strip().strip("`").strip("json").strip()) 一个重要的细节：模型以 temperature=0 调用以保证可复现性，响应是严格的 JSON、不带解释，这样解析就不会坏。提示词里包含具体的代币指标，这样 LLM 不会幻觉出数据，而是评估已经收集好的东西。而且关键的是，LLM 不做买入决策，它提供四个分数，连同指标信号一起进入打分矩阵。是代码做决定，不是模型。 时机智能体：Grok 评估市场时刻 同一个 meme，可能在看涨的周一带来十倍回报，也可能在周日的一次砸盘中一分钟内归零。叙事智能体评估的是代币，而不是它周围的市场。为此你需要一个单独的智能体来看大局：加密 Twitter 的情绪、pump.fun 过去几小时的成交量、BTC 主导地位、SOL 趋势。 class MarketTimingAgent: def __init__(self, api_key: str, model: str = "grok-4-fast"): self.client = httpx.AsyncClient( base_url="https://api.x.ai/v1", headers={"Authorization": f"Bearer {api_key}"}, timeout=30, ) self.model = model async def evaluate_timing(self, market_context: dict) -> dict: prompt = f"""You are evaluating the MOMENT for buying memecoins on Solana. Current market context: - SOL 24h: {market_context['sol_24h_change']}% - BTC dominance: {market_context['btc_dominance']}% - pump.fun volume 4h: {market_context['pf_volume_4h']} SOL - Graduations 4h: {market_context['graduations_4h']} - Avg volume per launch: {market_context['avg_volume_per_launch']} SOL - Hour (UTC): {market_context['hour_utc']} Rate 0.0-1.0: 1. market_mood: overall market sentiment (0 = panic, 1 = greed) 2. meme_season: is this a memecoin season or a lull 3. volume_signal: is pump.fun volume normal or anomalous 4. timing_score: overall timing score for entry Reply ONLY JSON: {{"market_mood": 0.0, "meme_season": 0.0, "volume_signal": 0.0, "timing_score": 0.0}}""" resp = await self.client.post("/chat/completions", json={ "model": self.model, "messages": [{"role": "user", "content": prompt}], "temperature": 0, }) text = resp.json()["choices"][0]["message"]["content"] try: return json.loads(text.strip().strip("`").replace("json", "").strip()) except json.JSONDecodeError: return {"market_mood": 0.3, "meme_season": 0.3, "volume_signal": 0.3, "timing_score": 0.3} 需要一个单独的时机智能体，不是为了好看，而是因为叙事智能体和时机智能体回答的是不同的问题。叙事："这个代币好不好？"时机："现在是买任何代币的好时刻吗？"如果 meme 季节为零且 SOL 在下跌，再好的 meme 也救不了你，而时机智能体会在流水线把预算花在一个糟糕的日子里之前拦住它。 这个智能体不是每个代币都调用，而是周期性地、每十五到三十分钟调用一次，并缓存它的结论。这节省了模型调用：市场上下文的变化比发行流慢。 检查智能体：买入前的对抗式验证 流水线里的最后一个智能体，也是对安全最重要的一个。前面所有智能体都朝着"批准"努力：监控器放行、审计者没发现问题、叙事打分不错、时机良好。检查者朝着"拒绝"努力。它的工作就是找到一个不买的理由。 这是 maker-checker（制造者-检查者）模式：一些智能体生成打分（制造者），另一个试图推翻它（检查者）。角色分离能防止一个智能体成为自己工作的裁判。生成者对自己的输出过于宽容，检查者没有这个问题，因为它被给了相反的指令。 class AdversarialChecker: def __init__(self, api_key: str, model: str = "grok-4"): # checker on a stronger model than the generators self.client = httpx.AsyncClient( base_url="https://api.x.ai/v1", headers={"Authorization": f"Bearer {api_key}"}, timeout=30, ) self.model = model async def check(self, token: Token, audit: dict, narrative: dict, timing: dict, score: float) -> dict: prompt = f"""You are an adversarial checker. Your job is to find reasons NOT to buy this token. Previous agents approved the purchase. Token: {token.name} ({token.symbol}), score: {score:.2f} Audit: coordinated_buys={audit.get('coordinated_buys')}, wash_trading={audit.get('wash_trading')}, creator_dump_risk={audit.get('creator_dump_risk'):.1f}, organic={audit.get('organic_score'):.1f} Narrative: fit={narrative.get('narrative_fit'):.1f}, virality={narrative.get('virality'):.1f}, community={narrative.get('community'):.1f} Timing: moment={timing.get('timing_score'):.1f}, meme_season={timing.get('meme_season'):.1f} Look for AGAINST. What can go wrong? Is there a contradiction between signals? Are there red flags the previous agents missed? Reply ONLY JSON: {{"approve": true, "confidence": 0.0, "risk_flags": [], "reason": ""}} approve=false if you found a serious red flag. confidence from 0 to 1.""" resp = await self.client.post("/chat/completions", json={ "model": self.model, "messages": [{"role": "user", "content": prompt}], "temperature": 0, }) text = resp.json()["choices"][0]["message"]["content"] try: result = json.loads(text.strip().strip("`").replace("json", "").strip()) except json.JSONDecodeError: result = {"approve": False, "confidence": 0, "risk_flags": ["parse_error"], "reason": "checker failed to respond"} return result 两个关键决策。第一：检查者跑在一个更强的模型上（grok-4 而不是 grok-4-fast）。生成者跑在快速便宜的模型上，因为它们数量多、速度重要。检查者只有一个，而它的质量决定安全。你为一通更贵的调用付费，但那通调用能防止你买入其他智能体漏掉的代币。第二：在解析错误时，检查者返回 approve: false。和审计者一样，一次检查错误等于拒绝，而不是静默跳过。 五个智能体如何协同工作 WebSocket 流 | v [ MONITOR ] --- 代码过滤（指标、年龄、元数据） | v [ AUDITOR ] --- Grok 分析钱包和模式 | v [ NARRATIVE ] --- Grok 为 meme 潜力和传播性打分 | v （缓存 15 分钟） [ TIMING ] --- Grok 评估市场时刻 ----------------+ | | v | 打分矩阵（代码聚合所有智能体的分数） | | | v | [ CHECKER ] --- Grok 寻找不买的理由（对抗式） | v 执行（如果检查者批准）--- 通过 Jito 发送 Solana TX 监控器和执行是代码。审计、叙事、时机和检查是四个独立的 Grok 调用，带不同的提示词和不同的角色。时机被缓存，不是每个代币都调用。检查者只对通过打分阈值的代币调用，也就是每天几单而不是几千单。 每个获批代币的 LLM 总成本：审计者（fast）+ 叙事（fast）+ 检查者（full）= 三次调用。时机被缓存摊销了。对在审计者处被切掉的代币，成本是一次调用。对在代码过滤处被切掉的代币，成本是零次调用。 风险管理：为什么没有刹车这就是个赌场 这里开始诚实的部分。pump.fun 是一个统计上残酷的环境：百分之一到百分之二的代币能毕业，大多数在几分钟内归零，最后一个在曲线上买入的人，会成为第一个卖出的人的退出流动性。迁移后几分钟内回撤百分之九十及以上是常态，不是例外。 在这个环境里，一个没有刹车的智能体就是一台高速烧钱的机器。所以风险管理不是"好习惯"，而是这条流水线存在的前提。 class RiskManager: def __init__(self, max_position_sol: float, daily_loss_limit_sol: float): self.max_pos = max_position_sol # max SOL per trade self.daily_limit = daily_loss_limit_sol self.today_loss = 0.0 self.today_trades = 0 self.max_daily_trades = 10 # no more than 10 trades per day self.open_positions = 0 self.max_open = 3 # no more than 3 open at once def can_trade(self) -> bool: if self.today_loss >= self.daily_limit: return False # daily loss limit exhausted if self.today_trades >= self.max_daily_trades: return False # daily trade count limit if self.open_positions >= self.max_open: return False # too many open positions return True def position_size(self, score: float) -> float: # position size proportional to score, capped base = self.max_pos * min(score, 1.0) # shrink as daily limit approaches remaining = self.daily_limit - self.today_loss return min(base, remaining * 0.3) # no more than 30% of remaining limit def record_result(self, pnl_sol: float): if pnl_sol < 0: self.today_loss += abs(pnl_sol) self.today_trades += 1 五个刹车，每一个都是强制的。仓位上限：无论分数多高，单个代币都不值得超过一个设定的 SOL 数。每日亏损限额：如果流水线今天已经亏过阈值，就停下来直到明天。交易次数上限：每天不超过十次，因为每一笔交易都是一次风险，一天二十笔交易是二十次风险，而不是二十个机会。持仓上限：同时不超过三个，以免分散注意力和资金。接近限额时缩减仓位：一天的最后几笔交易要小，因为每日预算几乎用完了。 执行：在联合曲线上买入 在 pump.fun 上买入是一笔调用联合曲线程序的 Solana 交易。毕业后，交易通过 PumpSwap 或 Raydium 进行，经由 Jupiter 路由。 from solders.keypair import Keypair from solana.rpc.async_api import AsyncClient class TradeExecutor: def __init__(self, wallet_key: str, rpc_url: str): self.wallet = Keypair.from_base58_string(wallet_key) self.rpc = AsyncClient(rpc_url) self.jito_url = "https://mainnet.block-engine.jito.wtf" # for priority TXs async def buy(self, token: Token, amount_sol: float) -> dict: # build a buy transaction on the bonding curve # in production: via the pump.fun SDK or a direct program call tx_data = await self.build_buy_tx(token.address, amount_sol) # send via Jito for priority block inclusion result = await self.send_with_priority(tx_data, tip_lamports=10_000) return { "tx_hash": result["signature"], "token": token.address, "amount_sol": amount_sol, "price_at_buy": token.bonding_curve_pct, } async def sell(self, token_address: str, pct: float = 100) -> dict: """Full or partial sell.""" # analogous to buy, but calling sell on the curve or a swap via Jupiter pass async def set_stop_loss(self, token_address: str, stop_pct: float = 50): """Price monitoring and automatic sell on drawdown.""" while True: price = await self.get_current_price(token_address) if price.change_pct <= -stop_pct: await self.sell(token_address, pct=100) return {"stopped": True, "loss_pct": price.change_pct} await asyncio.sleep(5) # check every 5 seconds 两个技术要点。Jito 是 Solana 的 MEV 层，它允许通过付小费把一笔交易优先打包进区块。对 memecoin 来说，入场窗口以秒计，这就是"在早期曲线上上车"和"在顶部买入"的区别。而止损不是通过订单实现的（联合曲线上没有订单），而是通过价格监控和在设定回撤时自动卖出。这是 Grok Bot 能在它的持久化电脑上全天候执行的行为。 Grok Bot 作为整条流水线的编排者 下面是所有部分在 Grok Bot 里如何合到一起。你不是在笔记本上跑一个 Python 脚本。你把任务描述给 Grok Bot，它在自己的云端电脑上启动流水线并 7×24 小时工作。 Grok Bot 获得你工具的访问权（钱包、API 密钥、监控），并把流水线作为一个循环来跑：监控流、过滤、分析、决策、执行、记录。当有什么事需要你决策时（比如，一个代币分数很高但风险评分临界），它给你发通知并等待。当流水线遇到错误时，它告诉你哪里出错了，你在聊天里修。 持久性意味着跑流水线的 VM 在会话之间不关机。机器人记得它已经看过哪些代币、买了哪些、跳过了哪些以及为什么。这是长期工作记忆，而不是每次从零重启。 可能出什么问题，以及为什么通常都会出问题 我会说出那些"加密交易机器人"文章通常省略的东西。 pump.fun 的统计是残酷的。百分之一到百分之二的代币能毕业。在那些毕业的代币里，大多数给早期买家带来数倍回报，然后在迁移后第一个小时内归零。曲线上的最后一个买家几乎总是成为创建者和狙击手的退出流动性。这不是系统故障，这是架构本身：联合曲线在结构上奖励先来者，牺牲后来者。 智能体不改变这个数学。它让过滤更快、更广，但不会把一个百分之一的任务变成百分之五十的任务。如果你的流水线每天买十个代币、其中一个毕业，那你九个亏、一个赚。唯一的问题是收益能否覆盖亏损，而这由仓位大小、止损速度和你在曲线上入场多早决定。 智能体消除不了的主要风险： > MEV 和抢跑：拥有直接 mempool 访问权的机器人能看到你的交易并抢在你前面买。Jito 小费能部分保护，但不是完全保护。Rug 和创建者砸盘：一个代币可能在每个指标上都完美，却因为创建者抽走流动性而在五分钟内归零。衰减速度：一个在前二十分钟内没有建立起成交量的 memecoin 通常会死。智能体必须快速止损，而不是等待反弹。LLM 调用成本：每一次叙事评估都是一次模型调用。在每天成千上万个发行的情况下，如果流过滤不够紧，调用成本可能超过利润。 日志：唯一通往改进的路 每一笔交易、每一次跳过、每一次止损都带着完整上下文记录：分数、指标、叙事评估、入场和出场价、盈亏（PnL）。没有日志，流水线就不会改进，因为你不知道哪些信号有效、哪些是噪音。 import json from datetime import datetime class TradeLog: def __init__(self, path: str): self.path = path def entry(self, token, tx, score, analysis, narrative): record = { "timestamp": datetime.utcnow().isoformat(), "action": "buy", "token": token.address, "name": token.name, "score": round(score, 3), "risk_score": analysis["risk_score"], "narrative_fit": narrative.get("narrative_fit", 0), "virality": narrative.get("virality", 0), "curve_pct_at_buy": token.bonding_curve_pct, "amount_sol": tx["amount_sol"], "tx_hash": tx["tx_hash"], } self._write(record) def skip(self, token, reason, detail=None): self._write({ "timestamp": datetime.utcnow().isoformat(), "action": "skip", "token": token.address, "name": token.name, "reason": reason, "detail": detail, }) def close(self, token_address, pnl_sol, hold_minutes): self._write({ "timestamp": datetime.utcnow().isoformat(), "action": "close", "token": token_address, "pnl_sol": round(pnl_sol, 4), "hold_minutes": hold_minutes, }) def _write(self, record): with open(self.path, "a") as f: f.write(json.dumps(record) + "\n") 日志是 JSONL 格式（每条记录一行 JSON），这样就能一行一行地读、过滤和分析，而不用加载整个文件。跑一周流水线之后，你就会有足够的数据看出：哪个分数水平真的能预测毕业、哪些叙事信号有效哪些无效，以及每一次 LLM 调用相对于所找到代币的利润成本是多少。 构建顺序 在每个阶段都最小化风险的技术顺序。先组装监控器和过滤器，在不执行的情况下跑一周，看看有多少代币通过、它们的质量如何。然后接上分析器和打分，再收集一周日志但不买入，把你的跳过与那些代币的真实命运对比。然后以最小资金（每笔 0.01-0.05 SOL）打开执行，带硬刹车，收集真实的 PnL。只有在有了两周真实数据之后，才开始增加仓位大小，并且要渐进地加，而不是跳跃式地加。 每个阶段都捕获它那一类错误：纸面过滤显示过滤器到底有没有用，纸面打分显示它能不能预测出什么，微资金显示真实成本（滑点、手续费、MEV），只有到这时才放大资金。跳过任何阶段，就等于凭希望交易。 我的仓库： http://github.com/zostaff/grokbot-pumpfun 我的 tg 频道： https://t.me/zostaffsmartarc 标签：# X # Grok Bot # Crypto # AI # Marketing # Growth # Side Hustle # Side Hustle # Thread 相关文章 Grok Bot 交易大厅：如何用 10 步运行一个 8 智能体的加密交易台（完整教程） Elon Musk 转发了视频指南，我决定把智能体写成文字版指南 Grok Bot Crypto AI Marketing