""" starter_8_whale_tracker.py — Whale-tracker / alerter. NOT a trading bot. Watches a configurable set of high-PnL wallets (or large absolute trade sizes) and pushes a Discord/Telegram alert on every new fill. This is the cheapest way to get an information edge without exposing capital. Customize: WATCH_WALLETS, MIN_TRADE_USD, alert sinks (Discord/Telegram). Run: python main.py Defaults to DRY_RUN=true (alerts disabled, prints to stdout instead). """ from __future__ import annotations import asyncio import os import signal import time from typing import Optional from dotenv import load_dotenv from common import ( JsonlLogger, close_http, evaluate_wallet, fetch_leaderboard, fetch_user_activity, get_http, ) # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- # Comma-separated wallets to track. Leave empty to auto-pull from leaderboard. WATCH_WALLETS = [w.strip() for w in os.getenv("WATCH_WALLETS", "").split(",") if w.strip()] # How many top leaderboard wallets to track if WATCH_WALLETS is empty. AUTO_TOP_N = int(os.getenv("AUTO_TOP_N", "20")) # Period for leaderboard auto-pull: day | week | month | all LEADERBOARD_PERIOD = os.getenv("LEADERBOARD_PERIOD", "week") # Min PnL filter when auto-pulling leaderboard. MIN_PNL_USD = float(os.getenv("MIN_PNL_USD", "20000")) # Don't alert on trades smaller than this. MIN_TRADE_USD = float(os.getenv("MIN_TRADE_USD", "1000")) # Polling interval (seconds). Free tier limits to ~50 req/sec. POLL_INTERVAL_SEC = float(os.getenv("POLL_INTERVAL_SEC", "30")) DISCORD_WEBHOOK = os.getenv("DISCORD_WEBHOOK", "").strip() TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN", "").strip() TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID", "").strip() DRY_RUN = os.getenv("DRY_RUN", "true").lower() != "false" # --------------------------------------------------------------------------- # Alert sinks # --------------------------------------------------------------------------- async def send_discord(text: str): if not DISCORD_WEBHOOK or DRY_RUN: return http = await get_http() try: await http.post(DISCORD_WEBHOOK, json={"content": text}, timeout=5.0) except Exception as e: print(f"discord error: {e}") async def send_telegram(text: str): if not (TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID) or DRY_RUN: return http = await get_http() url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage" try: await http.post(url, json={ "chat_id": TELEGRAM_CHAT_ID, "text": text, "parse_mode": "Markdown", "disable_web_page_preview": True, }, timeout=5.0) except Exception as e: print(f"telegram error: {e}") def format_alert(wallet: str, label: Optional[str], trade: dict) -> str: side = trade.get("side") or trade.get("action") or "?" asset = trade.get("title") or trade.get("question") or trade.get("asset") or trade.get("slug", "?") size = float(trade.get("size", 0) or 0) price = float(trade.get("price", 0) or 0) notional = size * price short_wallet = wallet[:8] + "…" + wallet[-4:] name = label or short_wallet return ( f"🐋 *{name}* `{short_wallet}`\n" f" {side.upper()} `{asset[:60]}`\n" f" size={size:.2f} @ {price:.4f} = *${notional:,.0f}*\n" f" poly: https://polymarket.com/profile/{wallet}" ) # --------------------------------------------------------------------------- # Main loop # --------------------------------------------------------------------------- async def init_watch_list() -> list[tuple[str, Optional[str]]]: """Returns list of (wallet, label) tuples.""" if WATCH_WALLETS: return [(w, None) for w in WATCH_WALLETS] print(f"Auto-pulling top {AUTO_TOP_N} from leaderboard ({LEADERBOARD_PERIOD})…") rows = await fetch_leaderboard(period=LEADERBOARD_PERIOD, limit=200) qualified = [] for row in rows: wallet = row.get("proxyWallet") or row.get("address") or row.get("user") pnl = float(row.get("pnl", 0) or 0) if not wallet: continue if pnl < MIN_PNL_USD: continue name = row.get("displayUsername") or row.get("name") qualified.append((wallet.lower(), name)) if len(qualified) >= AUTO_TOP_N: break return qualified async def main(): load_dotenv() log = JsonlLogger("logs/whale_v1") print("=" * 60) print("Polymarket Whale Tracker") print("=" * 60) print(f"Mode: {'DRY-RUN (stdout only)' if DRY_RUN else '🔴 LIVE alerts'}") print(f"Min trade USD: ${MIN_TRADE_USD}") print(f"Discord: {'configured' if DISCORD_WEBHOOK else 'off'}") print(f"Telegram: {'configured' if TELEGRAM_BOT_TOKEN else 'off'}") print("=" * 60) watchlist = await init_watch_list() if not watchlist: print("⚠️ No wallets to watch. Set WATCH_WALLETS in .env.") return print(f"Watching {len(watchlist)} wallets:") for w, label in watchlist[:10]: print(f" - {label or '(anon)'} {w[:10]}…") if len(watchlist) > 10: print(f" …and {len(watchlist) - 10} more") # Track last-seen timestamp per wallet to detect new trades only. last_seen: dict[str, float] = {} for w, _ in watchlist: try: acts = await fetch_user_activity(w, limit=1) if acts: last_seen[w] = float(acts[0].get("timestamp", time.time())) except Exception: last_seen[w] = time.time() stop = asyncio.Event() for sig in (signal.SIGINT, signal.SIGTERM): try: asyncio.get_running_loop().add_signal_handler(sig, stop.set) except NotImplementedError: signal.signal(sig, lambda *_: stop.set()) while not stop.is_set(): for wallet, label in watchlist: try: acts = await fetch_user_activity(wallet, limit=20) except Exception as e: log.log("activity_error", wallet=wallet, error=str(e)) continue since = last_seen.get(wallet, 0) new = [] max_ts = since for a in acts: ts = float(a.get("timestamp", 0) or 0) if ts > since: new.append(a) if ts > max_ts: max_ts = ts last_seen[wallet] = max_ts for trade in reversed(new): # oldest first size = float(trade.get("size", 0) or 0) price = float(trade.get("price", 0) or 0) if size * price < MIN_TRADE_USD: continue msg = format_alert(wallet, label, trade) print("\n" + msg) log.log("whale_trade", wallet=wallet, label=label, **{ k: trade.get(k) for k in ("side", "size", "price", "title", "slug") }) await asyncio.gather(send_discord(msg), send_telegram(msg)) try: await asyncio.wait_for(stop.wait(), timeout=POLL_INTERVAL_SEC) except asyncio.TimeoutError: pass await close_http() print("Done.") if __name__ == "__main__": asyncio.run(main())