#!/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 that won (true, unbiased)
"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": 80, # page cap per endpoint (50 each) — window is the real limit
}
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]
WIN_WINDOW_DAYS = 90 # measure win rate over resolved bets in this window
def _parse_end(end):
"""Parse an endDate ('2026-06-11T00:00:00Z' or '2026-06-09') to epoch."""
if not end:
return 0
end = end.replace("Z", "")
for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%d"):
try:
return time.mktime(time.strptime(end, fmt))
except ValueError:
continue
return 0
def resolved_positions(wallet, max_pages):
"""Every resolved bet — won or lost — in the last WIN_WINDOW_DAYS.
Polymarket only redeems *winning* shares; losing shares are worth $0 and
sit unredeemed in the wallet forever. So /closed-positions (redeemed or
sold) is heavily survivorship-biased toward winners — the losers pile up
in /positions at curPrice 0. A true win rate has to union both, over the
same time window so the ratio isn't skewed by truncation.
"""
cutoff = time.time() - WIN_WINDOW_DAYS * 86400
now = time.time()
out = []
# redeemed / sold winners (and losers sold before resolution), time-sorted
offset = 0
while offset < max_pages * 50:
page = get_json("/closed-positions",
{"user": wallet, "limit": 50, "offset": offset,
"sortBy": "TIMESTAMP", "sortDirection": "DESC"})
if not page:
break
for p in page:
if p.get("timestamp", 0) >= cutoff:
out.append({"won": p.get("curPrice", 0) >= 0.5,
"pnl": p.get("realizedPnl", 0),
"title": p.get("title", "?"),
"outcome": p.get("outcome", "?"),
"avgPrice": p.get("avgPrice", 0),
"ts": p.get("timestamp", 0)})
offset += 50
if len(page) < 50 or page[-1].get("timestamp", 0) < cutoff:
break
# currently-held positions whose market already resolved (the hidden losers)
offset = 0
while offset < max_pages * 50:
page = get_json("/positions",
{"user": wallet, "limit": 50, "offset": offset,
"sizeThreshold": 0.0})
if not page:
break
for p in page:
end = _parse_end(p.get("endDate"))
if cutoff <= end < now: # resolved, in window, unredeemed
out.append({"won": p.get("curPrice", 0) >= 0.5,
"pnl": p.get("cashPnl", 0),
"title": p.get("title", "?"),
"outcome": p.get("outcome", "?"),
"avgPrice": p.get("avgPrice", 0),
"ts": end})
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 = resolved_positions(wallet, max_positions)
if not resolved:
return None
wins = sum(1 for p in resolved if p["won"])
realized_pnl = sum(p["pnl"] for p in resolved)
trades, markets = recent_trade_frequency(wallet)
recent = sorted(resolved, key=lambda p: p["ts"], reverse=True)[:15]
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["title"],
"outcome": p["outcome"],
"avgPrice": p["avgPrice"],
"realizedPnl": round(p["pnl"], 2),
"won": p["won"],
"timestamp": p["ts"],
}
for p in recent
],
}
# ── 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()