From 5ed55009b5ac38c7d494013f696eccc2f7236306 Mon Sep 17 00:00:00 2001 From: jaxperro Date: Fri, 12 Jun 2026 22:03:00 -0400 Subject: [PATCH] Polymarket smart money tracker: find >75% win-rate wallets betting weekly Dashboard + terminal tool that pulls the 7d/30d/all leaderboards, measures each wallet's win rate over its most recent resolved bets, and its distinct markets traded per week. Zero dependencies (Python 3 stdlib only). Co-Authored-By: Claude Fable 5 --- .gitignore | 3 + README.md | 49 ++++++ smart_money.py | 423 +++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 475 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 smart_money.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 00000000..b908d4cb --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +__pycache__/ +*.pyc +.DS_Store diff --git a/README.md b/README.md new file mode 100644 index 00000000..26e9a313 --- /dev/null +++ b/README.md @@ -0,0 +1,49 @@ +# Polymarket Smart Money + +Finds Polymarket wallets that **win more than 75% of their resolved bets** and +**bet multiple times per week** — the "smart money" worth watching. + +Zero dependencies. One file, Python 3 stdlib only. + +## Run the dashboard + +```bash +python3 smart_money.py +``` + +Open **http://localhost:8899**, hit **Scan**, and wait a minute or two. +Adjust the filters (win rate, bets/week, minimum resolved bets, candidate +pool size) and the table updates live while the scan runs. Click any trader +to see their recent resolved bets and a link to their Polymarket profile. + +## Run in the terminal + +```bash +python3 smart_money.py --scan # default 150-wallet pool +python3 smart_money.py --scan --pool 300 # broader sweep +``` + +## How it works + +1. **Candidates** — pulls the 7d, 30d, and all-time leaderboards from + `data-api.polymarket.com/v1/leaderboard` and dedupes into a candidate pool + (default 150 wallets). +2. **Win rate** — for each wallet, pages through `/closed-positions` (up to + 300 most recent resolved positions). A *win* is a resolved position with + `realizedPnl > 0`. +3. **Frequency** — counts trades from `/activity` over the last 4 weeks; + *bets/week* is the number of **distinct markets** traded per week, so 50 + fills on one order don't count as 50 bets. +4. **Filter** — keeps wallets with win rate ≥ 75%, ≥ 2 bets/week, and ≥ 10 + resolved bets (so a 3-for-3 fluke doesn't rank as a 100% winner). + +## Caveats + +- Candidates come from the leaderboards, so this surfaces *profitable* sharps. + A high-win-rate wallet that has never cracked any leaderboard window won't + appear — scanning every wallet on the platform isn't feasible via the + public API. +- Win rate is measured over each wallet's most recent ~300 resolved + positions, not their entire history. +- High win rate ≠ high EV: someone selling early for +$1 on every position + counts as winning. Check the realized PnL column alongside the win rate. diff --git a/smart_money.py b/smart_money.py new file mode 100644 index 00000000..321bd530 --- /dev/null +++ b/smart_money.py @@ -0,0 +1,423 @@ +#!/usr/bin/env python3 +"""Polymarket Smart Money Tracker. + +Finds wallets that win more than 75% of their resolved bets and bet +multiple times per week. Zero dependencies — Python stdlib only. + +Run the dashboard: python3 smart_money.py (http://localhost:8899) +Run a terminal scan: python3 smart_money.py --scan +""" + +import argparse +import json +import ssl +import sys +import threading +import time +import urllib.error +import urllib.parse +import urllib.request +from concurrent.futures import ThreadPoolExecutor, as_completed +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +DATA_API = "https://data-api.polymarket.com" +PORT = 8899 + +# Scan defaults — adjustable in the UI +DEFAULTS = { + "pool": 150, # candidate wallets pulled from the leaderboard + "min_win_rate": 75.0, # percent of resolved bets with realizedPnl > 0 + "min_bets_week": 2.0, # distinct markets traded per week, recent 4 weeks + "min_resolved": 10, # resolved bets required (filters 3-for-3 flukes) + "max_positions": 300, # most recent resolved positions sampled per wallet +} +FREQ_WEEKS = 4 # window for the bets-per-week measurement + + +def _ssl_context(): + ctx = ssl.create_default_context() + try: + import certifi + ctx.load_verify_locations(certifi.where()) + return ctx + except ImportError: + pass + # Some Python installs (notably python.org builds on macOS) ship without + # usable CA certs. Probe once; fall back to unverified for this + # read-only public API rather than failing every request. + try: + req = urllib.request.Request(DATA_API + "/v1/leaderboard?limit=1", + headers={"User-Agent": "Mozilla/5.0"}) + urllib.request.urlopen(req, timeout=10, context=ctx).read() + return ctx + except urllib.error.URLError as e: + if isinstance(getattr(e, "reason", None), ssl.SSLCertVerificationError): + print("warning: no usable CA certificates found; SSL verification " + "disabled (pip3 install certifi to fix)", file=sys.stderr) + return ssl._create_unverified_context() + return ctx + + +SSL_CTX = _ssl_context() + + +def get_json(path, params=None, retries=2): + url = DATA_API + path + if params: + url += "?" + urllib.parse.urlencode(params) + for attempt in range(retries + 1): + try: + req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) + with urllib.request.urlopen(req, timeout=15, context=SSL_CTX) as r: + return json.loads(r.read().decode()) + except (urllib.error.URLError, TimeoutError, json.JSONDecodeError): + if attempt == retries: + return None + time.sleep(1 + attempt) + + +def leaderboard_candidates(pool): + """Unique wallets from the 7d/30d/all leaderboards, best PnL first.""" + seen = {} + for window in ("7d", "30d", "all"): + offset = 0 + while offset < pool: + page = get_json("/v1/leaderboard", + {"window": window, "limit": 50, "offset": offset}) + if not page: + break + for u in page: + w = u.get("proxyWallet") + if w and w not in seen: + seen[w] = { + "wallet": w, + "username": u.get("userName") or w[:10] + "...", + "leaderboard_pnl": u.get("pnl", 0), + "volume": u.get("vol", 0), + } + offset += 50 + if len(page) < 50: + break + ranked = sorted(seen.values(), key=lambda u: u["leaderboard_pnl"], reverse=True) + return ranked[:pool] + + +def closed_positions(wallet, max_positions): + """Most recent resolved positions, newest first. + + The API defaults to sorting by realizedPnl descending — without an + explicit TIMESTAMP sort you get a wallet's biggest *wins* first, which + inflates every win rate toward 100%. Sort by time so we sample the + actual recent record. + """ + out = [] + offset = 0 + while offset < max_positions: + page = get_json("/closed-positions", + {"user": wallet, "limit": 50, "offset": offset, + "sortBy": "TIMESTAMP", "sortDirection": "DESC"}) + if not page: + break + out.extend(page) + offset += 50 + if len(page) < 50: + break + return out + + +def recent_trade_frequency(wallet, weeks=FREQ_WEEKS): + """(trades, distinct markets) over the last `weeks` weeks.""" + cutoff = time.time() - weeks * 7 * 86400 + trades = 0 + markets = set() + offset = 0 + while offset < 1000: + page = get_json("/activity", + {"user": wallet, "type": "TRADE", "limit": 500, "offset": offset}) + if not page: + break + for t in page: + if t.get("timestamp", 0) >= cutoff: + trades += 1 + if t.get("conditionId"): + markets.add(t["conditionId"]) + offset += 500 + if len(page) < 500 or page[-1].get("timestamp", 0) < cutoff: + break + return trades, len(markets) + + +def analyze_wallet(candidate, max_positions): + wallet = candidate["wallet"] + resolved = closed_positions(wallet, max_positions) + if not resolved: + return None + # A bet won if the outcome it held resolved YES. For resolved positions + # curPrice is binary (1 = won, 0 = lost), so it's a cleaner signal than + # the sign of realizedPnl — a hedged position can win yet net $0 PnL. + def won(p): + return p.get("curPrice", 0) >= 0.5 + wins = sum(1 for p in resolved if won(p)) + realized_pnl = sum(p.get("realizedPnl", 0) for p in resolved) + trades, markets = recent_trade_frequency(wallet) + return { + **candidate, + "resolved": len(resolved), + "wins": wins, + "win_rate": round(wins / len(resolved) * 100, 1), + "realized_pnl": round(realized_pnl, 2), + "trades_4w": trades, + "markets_4w": markets, + "bets_per_week": round(markets / FREQ_WEEKS, 1), + "recent": [ + { + "title": p.get("title", "?"), + "outcome": p.get("outcome", "?"), + "avgPrice": p.get("avgPrice", 0), + "realizedPnl": round(p.get("realizedPnl", 0), 2), + "won": won(p), + "timestamp": p.get("timestamp", 0), + } + for p in resolved[:15] + ], + } + + +# ── scan state shared with the web UI ────────────────────────────────────── + +scan_lock = threading.Lock() +scan_state = {"state": "idle", "done": 0, "total": 0, "results": [], "params": {}} + + +def run_scan(params): + with scan_lock: + scan_state.update(state="scanning", done=0, total=0, results=[], params=params) + candidates = leaderboard_candidates(params["pool"]) + with scan_lock: + scan_state["total"] = len(candidates) + results = [] + with ThreadPoolExecutor(max_workers=12) as pool: + futures = {pool.submit(analyze_wallet, c, params["max_positions"]): c + for c in candidates} + for f in as_completed(futures): + try: + r = f.result() + except Exception: + r = None + with scan_lock: + scan_state["done"] += 1 + if r: + results.append(r) + scan_state["results"] = sorted( + results, key=lambda x: (x["win_rate"], x["realized_pnl"]), + reverse=True) + with scan_lock: + scan_state["state"] = "done" + + +def filter_results(results, params): + return [r for r in results + if r["win_rate"] >= params["min_win_rate"] + and r["bets_per_week"] >= params["min_bets_week"] + and r["resolved"] >= params["min_resolved"]] + + +# ── web UI ────────────────────────────────────────────────────────────────── + +PAGE = """ +Polymarket Smart Money + +

Polymarket Smart Money

+wallets winning >75% of resolved bets, betting multiple times per week
+
+ + + + + +
+
+
+
Hit Scan to pull the leaderboards and +analyze each wallet's resolved bets. A full scan takes a minute or two.
+
+
+""" + + +class Handler(BaseHTTPRequestHandler): + def _send(self, body, ctype="application/json", code=200): + data = body.encode() if isinstance(body, str) else json.dumps(body).encode() + self.send_response(code) + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def do_GET(self): + url = urllib.parse.urlparse(self.path) + if url.path == "/": + return self._send(PAGE, "text/html") + if url.path == "/api/status": + q = urllib.parse.parse_qs(url.query) + params = { + "min_win_rate": float(q.get("min_win_rate", [DEFAULTS["min_win_rate"]])[0]), + "min_bets_week": float(q.get("min_bets_week", [DEFAULTS["min_bets_week"]])[0]), + "min_resolved": int(q.get("min_resolved", [DEFAULTS["min_resolved"]])[0]), + } + with scan_lock: + snapshot = dict(scan_state) + results = list(scan_state["results"]) + return self._send({ + "state": snapshot["state"], + "done": snapshot["done"], + "total": snapshot["total"], + "analyzed": len(results), + "filtered": filter_results(results, params), + }) + self._send({"error": "not found"}, code=404) + + def do_POST(self): + if self.path == "/api/scan": + length = int(self.headers.get("Content-Length", 0)) + body = json.loads(self.rfile.read(length) or b"{}") + params = {**DEFAULTS, **{k: body[k] for k in body if k in DEFAULTS}} + params["pool"] = int(params["pool"]) + with scan_lock: + busy = scan_state["state"] == "scanning" + if not busy: + threading.Thread(target=run_scan, args=(params,), daemon=True).start() + return self._send({"started": not busy}) + self._send({"error": "not found"}, code=404) + + def log_message(self, *args): + pass + + +def terminal_scan(args): + params = {**DEFAULTS, "pool": args.pool} + print(f"Pulling leaderboards (pool={args.pool})...") + run_scan(params) + matches = filter_results(scan_state["results"], params) + print(f"\n{'─' * 78}") + print(f" Smart money: win rate ≥ {params['min_win_rate']}%, " + f"≥ {params['min_bets_week']} bets/wk, ≥ {params['min_resolved']} resolved") + print(f"{'─' * 78}") + print(f"{'Trader':<22} {'Win%':>6} {'Record':>9} {'Bets/wk':>8} {'Realized PnL':>15}") + for r in matches: + rec = f"{r['wins']}-{r['resolved'] - r['wins']}" + print(f"{r['username']:<22} {r['win_rate']:>5.1f}% {rec:>9} " + f"{r['bets_per_week']:>8.1f} ${r['realized_pnl']:>14,.2f}") + print(f"{'─' * 78}") + print(f"{len(matches)} of {len(scan_state['results'])} analyzed wallets match.\n") + + +def main(): + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--scan", action="store_true", help="run once in the terminal") + ap.add_argument("--pool", type=int, default=DEFAULTS["pool"], + help="candidate wallets to analyze (default 150)") + ap.add_argument("--port", type=int, default=PORT) + args = ap.parse_args() + if args.scan: + return terminal_scan(args) + print(f"Polymarket Smart Money dashboard → http://localhost:{args.port}") + ThreadingHTTPServer(("127.0.0.1", args.port), Handler).serve_forever() + + +if __name__ == "__main__": + main()