"""RAPTOR — supervised external scanner entrypoint (whale copy-follow). Port of senpi-skills/raptor (producer v4.0.1, config v3.1) to the v2 `scan(inputs, ctx)` contract. The daemon loop is gone — the runtime supervisor calls scan() every interval_seconds (180s). THE STRATEGY — quality-first whale copy-follow pipeline (ported VERBATIM; see scoring.py for the pure gating/scoring): 1. Quality hot traders — discovery_get_top_traders(WEEKLY, sort PROFIT_AND_LOSS_UNREALIZED, consistency ELITE/RELIABLE, open_position_filter, limit 20); keep traders with unrealized delta-pnl >= $500k. Sort by delta-pnl desc; scan the top `positionsFetchLimit` (10). 2. Strongest position — per trader, leaderboard_get_trader_positions; pick the position with the largest |delta_pnl| (>= $100k). 3. Concentration — best |delta_pnl| / total |delta_pnl| across the trader's book must be >= 0.35 (conviction, not a diversified book). 4. Smart-money alignment — leaderboard_get_markets; the asset must show SM pct >= 2.0, SM traders >= 10, and (when required) direction match. 5. Whale entry-discipline — market_get_prices; skip if the price has already run > 5% from the whale's entry in the whale's direction (don't buy the top). 6. Additive score (max ~16) — TCS + trader-$ tier + ROI + concentration + SM strength + 4h/1h confirm + 15m contrib + better-than-whale edge bonus. 7. Emit the SINGLE best candidate, and only if its score >= minScore (6). DESIGN CHOICE — emit best-only vs runtime-owns-slots. Turbine/spider emit ALL gated candidates and let the runtime apply the slot ceiling. RAPTOR is deliberately different: it emits AT MOST ONE signal per tick (the single highest-score whale-aligned setup). This is the strategy, not an oversight — raptor's whole thesis is "follow the strongest conviction signal right now," and the source producer pushed exactly one signal per tick after sorting candidates by score and taking the top. Keeping `slots: 2` in the recipe lets a prior position still be open while a new tick emits the next-best; the runtime owns whether a second slot gets filled across ticks. We preserve the per-tick <=1 emission ceiling on purpose. Sizing — carried on each signal's data{} so the OPEN_POSITION rule action sizes identically to the source: - marginPct = percent of withdrawable (0–100), 25.0 (35.0 if score >= 10). Dual-DEX equity is collapsed via max() (NOT sum()) — one cross-margined wallet, two sub-DEX views; summing double-counts free balance. - leverage = conviction tier (>=10 -> 10x, >=8 -> 8x, >=6 -> 7x), clamped to the per-asset Hyperliquid venue max via strategy_get_asset_trading_limits. FLAGS — behaviors that changed under the v2 contract (NOT silent drops): 1. Daemon loop / producer_daemon — DROPPED. The runtime supervises this module and calls scan() each interval_seconds. scan() is single-pass and SYNC. 2. push_signal / ingest POST — DROPPED. scan() returns a plain list[dict]; the scaffold owns delivery, dedup (signal_id), and the wire envelope. The source's [0,1] normalized wire score (min(score/16, 1.0)) is NOT recomputed here — the raw additive score rides on data{} and the scaffold/runtime own wire scoring. (SCORE_NORMALIZATION_DIVISOR is kept in scoring.py for parity.) 3. seen-events.json 4h dedup — MOVED to ctx.state. The per-(trader, asset) 4h event dedup that the source kept in a JSON file lives in ctx.state now: each tick reads the latest {key: ts} map, prunes by the 4h window, applies it, stamps the chosen candidate, and appends the updated map for next tick. 4. xyz banned — preserved (config.smAlignment.xyzBanned). xyz-dex SM markets and xyz: positions are dropped in scoring.py. 5. ONLY read-only MCP calls. scan() never opens/closes/cancels — it produces signals. (A test asserts no create/close/cancel tool is ever called.) MCP tools used (all read-only): discovery_get_top_traders, leaderboard_get_trader_positions, leaderboard_get_markets, market_get_prices, strategy_get_asset_trading_limits, strategy_get_clearinghouse_state. State (in ctx.state, newest record wins): {"seen": {dedup_key: epoch_seconds}}. """ import sys import time import scoring # ── MCP data fetchers (route the producer's calls through ctx.senpi_mcp) ── def _fetch_quality_traders(ctx, limit, min_delta_usd): """discovery_get_top_traders(WEEKLY, ELITE/RELIABLE, delta gate) -> quality list.""" try: raw = ctx.senpi_mcp.call_tool("discovery_get_top_traders", { "time_frame": "WEEKLY", "sort_by": "PROFIT_AND_LOSS_UNREALIZED", "consistency": ["ELITE", "RELIABLE"], "open_position_filter": True, "limit": limit, }) except Exception: return [] return scoring.parse_quality_traders(raw, min_delta_usd=min_delta_usd) def _fetch_trader_positions(ctx, trader_address): """leaderboard_get_trader_positions -> list of position dicts (nested parser).""" try: raw = ctx.senpi_mcp.call_tool("leaderboard_get_trader_positions", {"trader_id": trader_address}) except Exception: return [] return scoring.parse_trader_positions(raw) def _fetch_sm_map(ctx): """leaderboard_get_markets -> {TOKEN: sm_record} (xyz dropped).""" try: raw = ctx.senpi_mcp.call_tool("leaderboard_get_markets", {"limit": 100}) except Exception: return {} return scoring.parse_sm_map(raw) def _fetch_current_px(ctx, asset): """market_get_prices -> current price for one asset, or 0.0. Verbatim parse.""" try: raw = ctx.senpi_mcp.call_tool("market_get_prices", {"assets": [asset]}) except Exception: return 0.0 if not raw: return 0.0 data = raw.get("data", raw) if isinstance(raw, dict) else raw if isinstance(data, dict): return scoring.safe_float(data.get(asset, 0)) return 0.0 def _get_safe_leverage(ctx, asset, requested_leverage): """Clamp requested leverage to the per-asset HL venue max via strategy_get_asset_trading_limits. Verbatim from producer.get_safe_leverage: reads data.leverage.value (or a scalar leverage), min()s with requested.""" try: limits = ctx.senpi_mcp.call_tool("strategy_get_asset_trading_limits", { "strategy_wallet": ctx.wallet, "coin": asset, }) except Exception: limits = None if limits: data = limits.get("data", limits) if isinstance(limits, dict) else limits if isinstance(data, dict): lev = data.get("leverage", {}) if isinstance(lev, dict): max_lev = int(scoring.safe_float(lev.get("value", 20))) return scoring.clamp_leverage(requested_leverage, max_lev) if isinstance(lev, (int, float)): return scoring.clamp_leverage(requested_leverage, int(lev)) return requested_leverage def _get_account(ctx): """(account_value, held_assets) from strategy_get_clearinghouse_state. Dual-DEX equity collapse: account_value via max() across main/xyz sections (two views of ONE cross-margined wallet — summing double-counts free balance -> 2x sizing). Verbatim from raptor_config.get_positions. """ try: ch = ctx.senpi_mcp.call_tool("strategy_get_clearinghouse_state", {"strategy_wallet": ctx.wallet}) except Exception: return 0.0, [] if not ch: return 0.0, [] data = ch.get("data", ch) if isinstance(ch, dict) else ch if not isinstance(data, dict): return 0.0, [] account_value = 0.0 held = [] for section in ("main", "xyz"): s = data.get(section, {}) if not isinstance(s, dict): continue ms = s.get("marginSummary", {}) account_value = max(account_value, scoring.safe_float(ms.get("accountValue", 0))) for ap in s.get("assetPositions", []) or []: pos = ap.get("position", ap) if scoring.safe_float(pos.get("szi", 0)) == 0: continue coin = pos.get("coin", "") if coin: held.append(coin) return account_value, held # ── ctx.state I/O (per-(trader, asset) 4h event dedup) ── def _load_seen(ctx): if ctx.state is None or len(ctx.state) == 0: return {} last = ctx.state.last() or {} seen = last.get("seen", {}) return dict(seen) if isinstance(seen, dict) else {} # ── Candidate evaluation (verbatim pipeline from build_signal) ── def _evaluate_trader(ctx, trader, sm_map, inputs): """Run one quality trader through the full pipeline. Returns a candidate dict (asset/direction/score/reasons/...) or None on any gate miss.""" positions = _fetch_trader_positions(ctx, trader["address"]) if not positions: return None best_pos, best_abs_pnl, _total, concentration = scoring.pick_strongest_position(positions) if not best_pos or best_abs_pnl < float(inputs.get("minPositionPnl", scoring.MIN_POSITION_PNL)): return None if concentration < float(inputs.get("minConcentration", scoring.MIN_CONCENTRATION)): return None sm = sm_map.get(best_pos["asset"]) if not scoring.sm_alignment_ok( sm, best_pos["direction"], min_sm_pct=float(inputs.get("minSmPct", scoring.MIN_SM_PCT)), min_sm_traders=int(inputs.get("minSmTraders", scoring.MIN_SM_TRADERS)), require_direction_match=bool(inputs.get("requireDirectionMatch", scoring.REQUIRE_DIRECTION_MATCH)), ): return None # v3.3 entry-discipline: fetch current price only when we have a whale entry. current_px = 0.0 whale_entry_px = scoring.safe_float(best_pos.get("whale_entry_px", 0)) if whale_entry_px > 0: current_px = _fetch_current_px(ctx, best_pos["asset"]) if not scoring.entry_discipline_ok(best_pos["direction"], whale_entry_px, current_px): return None scored = scoring.score_position(trader, best_pos, concentration, sm, current_px) if scored is None: return None score, reasons = scored return { "asset": best_pos["asset"], "direction": best_pos["direction"], "score": score, "reasons": reasons, "traderId": trader["address"][:10] + "...", "fullTraderId": trader["address"], "tcs": trader["tcs_label"], "traderDeltaPnl": trader["unrealized_pnl"], "positionDeltaPnl": best_pos["delta_pnl"], "concentration": concentration, "smPct": sm["pct"], "smTraders": sm["traders"], "priceChg4h": sm["price_chg_4h"], "priceChg1h": sm["price_chg_1h"], "whaleEntryPx": whale_entry_px if whale_entry_px > 0 else None, "currentPx": current_px if (whale_entry_px > 0 and current_px > 0) else None, } def scan(inputs, ctx): now = time.time() quality_pool_size = int(inputs.get("qualityPoolSize", 20)) positions_fetch_limit = int(inputs.get("positionsFetchLimit", 10)) min_delta_usd = float(inputs.get("minDeltaPnl", scoring.MIN_TRADER_DELTA_PNL)) min_score = int(inputs.get("minScore", scoring.MIN_SCORE)) margin_pct_base = float(inputs.get("marginPctBase", scoring.MARGIN_PCT_BASE)) margin_pct_high = float(inputs.get("marginPctHighConv", scoring.MARGIN_PCT_HIGH_CONV)) dedupe_hours = float(inputs.get("eventDedupeHours", scoring.EVENT_DEDUPE_HOURS)) default_leverage = int(inputs.get("defaultLeverage", scoring.DEFAULT_LEVERAGE)) account_value, held_assets = _get_account(ctx) if account_value <= 0: return [] held_upper = {h.upper() for h in held_assets} quality_traders = _fetch_quality_traders(ctx, quality_pool_size, min_delta_usd) if not quality_traders: return [] sm_map = _fetch_sm_map(ctx) if not sm_map: return [] seen = scoring.prune_seen_events(_load_seen(ctx), dedupe_hours, now=now) scan_count = min(len(quality_traders), positions_fetch_limit) candidates = [] for trader in quality_traders[:scan_count]: cand = _evaluate_trader(ctx, trader, sm_map, inputs) if not cand: continue if scoring.is_event_seen(seen, trader["address"], cand["asset"], dedupe_hours, now=now): continue if cand["asset"] in held_upper: continue candidates.append(cand) out = [] if candidates: candidates.sort(key=lambda c: c["score"], reverse=True) best = candidates[0] if best["score"] >= min_score: # Conviction-tier sizing: emit percent of withdrawable (0–100). margin_pct = margin_pct_high if best["score"] >= 10 else margin_pct_base requested = scoring.get_leverage_for_score( best["score"], scoring.LEVERAGE_TIERS, default_leverage) leverage = _get_safe_leverage(ctx, best["asset"], requested) # Stamp the 4h event dedup for (trader, asset) so we don't re-emit. seen[scoring.dedup_key(best["fullTraderId"], best["asset"])] = now out.append({ "asset": best["asset"], "direction": best["direction"], "marginPct": round(margin_pct * 100, 2), "leverage": float(leverage), "data": { "score": best["score"], "tcs": best["tcs"], "traderId": best["traderId"], "traderDeltaPnl": float(best["traderDeltaPnl"]), "positionDeltaPnl": float(best["positionDeltaPnl"]), "concentration": round(best["concentration"], 4), "smPct": float(best["smPct"]), "smTraders": int(best["smTraders"]), "priceChg4h": float(best["priceChg4h"]), "priceChg1h": float(best["priceChg1h"]), "whaleEntryPx": best["whaleEntryPx"], "currentPx": best["currentPx"], "reasons": best["reasons"], "heldAssets": held_assets, }, }) # Persist the pruned + updated dedup map for next tick. # If this append fails (e.g. state_history_max_count is 0/unset so append # raises, or a transient persist error), the next tick reloads stale dedup # state and may re-emit already-suppressed signals. Log-and-continue: don't # crash the tick, but make the failure visible on stderr (the runtime # supervisor captures the scaffold child's stderr) instead of swallowing it. if ctx.state is not None: try: ctx.state.append({"seen": seen}) except Exception as exc: print( f"[raptor.scan] WARNING: dedup-state append failed; next tick " f"may re-emit suppressed signals: {exc!r}", file=sys.stderr, ) return out