← back to Polymtrx Skill
skills/polymtrx/scripts/polymtrx.py
498 lines
#!/usr/bin/env python3
"""
polymtrx — Polymarket prediction-market opportunity scanner.
Reproduces the MTRX / @Polymtrx (PolyMatrix) workflow: pull LIVE Polymarket
markets, rank them by tradeable-edge signals, and emit the raw numbers so an
agent can synthesize a grounded read of "where is the best opportunity right
now?".
Signals (each 0..1, weighted into one opportunity score):
volume real-money $ traded → conviction / it matters
liquidity order-book depth + tight → you can actually get filled
spread (via CLOB, best-effort)
timing near resolution → the edge window is concrete
contest price near 50/50 on a → genuinely undecided, high info
high-volume binary
longshot price near 0/1 → potential mispriced tail
movement recent price change → momentum / news just hit
HARD RAIL: read-only. This engine ONLY fetches public market data. It never
places an order, never touches a wallet, private key, or CLOB credential, and
never signs anything. It surfaces opportunities; a human (or a separately-gated
step) decides and acts.
Zero dependencies (Python 3.9+ stdlib only). No API keys. $0 — every call hits
Polymarket's free public endpoints.
Commands:
polymtrx.py scan "<topic>" rank markets matching a topic by opportunity
polymtrx.py trending rank the most active live markets (no topic)
polymtrx.py market <slug|id> deep detail on one market (prices, spread)
polymtrx.py doctor health-check every endpoint
Usage:
polymtrx.py scan "fed rate cut" --limit 15
polymtrx.py trending --emit markdown
polymtrx.py scan "election" --no-clob # skip order-book depth (faster)
polymtrx.py market will-x-happen-2026
Inspired by MTRX (@Polymtrx) on X. Independent, no-key, $0-local reimplementation
of the public-data half of the "find the best opportunities" idea. See
references/api.md for endpoint lineage.
"""
from __future__ import annotations
import argparse
import concurrent.futures
import json
import sys
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
USER_AGENT = "polymtrx/1.0 (+https://polymarket.com public data; read-only)"
HTTP_TIMEOUT = 20 # seconds per request
GAMMA = "https://gamma-api.polymarket.com"
CLOB = "https://clob.polymarket.com"
# Opportunity-score weights (sum need not be 1; score is normalized to 0..100).
WEIGHTS = {
"volume": 0.30,
"liquidity": 0.15,
"timing": 0.20,
"contest": 0.20,
"longshot": 0.05,
"movement": 0.10,
}
# --------------------------------------------------------------------------- #
# HTTP helpers (stdlib only) #
# --------------------------------------------------------------------------- #
def _get(url: str, timeout: int = HTTP_TIMEOUT) -> bytes:
req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
with urllib.request.urlopen(req, timeout=timeout) as resp:
return resp.read()
def _get_json(url: str, timeout: int = HTTP_TIMEOUT):
return json.loads(_get(url, timeout).decode("utf-8", "replace"))
def _now() -> datetime:
return datetime.now(timezone.utc)
def _parse_float(v):
"""Polymarket returns numbers, strings, and JSON-encoded lists. Coerce."""
if v is None:
return None
if isinstance(v, (int, float)):
return float(v)
if isinstance(v, str):
s = v.strip()
if s.startswith("["):
try:
arr = json.loads(s)
return float(arr[0]) if arr else None
except Exception:
return None
try:
return float(s)
except Exception:
return None
if isinstance(v, list) and v:
return _parse_float(v[0])
return None
def _parse_dt(v):
if not v or not isinstance(v, str):
return None
s = v.replace("Z", "+00:00")
try:
dt = datetime.fromisoformat(s)
return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
except Exception:
return None
# --------------------------------------------------------------------------- #
# Scoring #
# --------------------------------------------------------------------------- #
def _clamp01(x: float) -> float:
return 0.0 if x < 0 else 1.0 if x > 1 else x
def _log_scale(value: float, full: float) -> float:
"""0 at value<=0, ~1 as value approaches `full` (log-ish, saturating)."""
if value <= 0:
return 0.0
import math
return _clamp01(math.log10(1 + value) / math.log10(1 + full))
def score_market(m: dict) -> dict:
"""Compute per-signal scores + a blended opportunity score for one market.
`m` is the normalized dict produced by _normalize_market().
"""
vol = m.get("volume") or 0.0
liq = m.get("liquidity") or 0.0
spread = m.get("spread") # 0..1, may be None
price = m.get("yes_price") # 0..1, may be None
end = m.get("end_dt")
change = m.get("price_change_24h") # signed 0..1, may be None
s = {}
# volume: real money is the strongest signal. $1M volume ~= saturated.
s["volume"] = _log_scale(vol, 1_000_000)
# liquidity: depth (log) blended with tightness of spread if we have it.
depth = _log_scale(liq, 200_000)
if spread is not None:
tight = _clamp01(1.0 - (spread / 0.10)) # 10c spread -> 0
s["liquidity"] = 0.6 * depth + 0.4 * tight
else:
s["liquidity"] = depth
# timing: nearer resolution = sharper, concrete edge window. 0..45 days.
if end is not None:
days = (end - _now()).total_seconds() / 86400.0
if days < 0:
s["timing"] = 0.0 # already resolved / past
else:
s["timing"] = _clamp01(1.0 - (days / 45.0))
else:
s["timing"] = 0.0
# contest: a binary sitting near 50/50 is maximally undecided → high info.
if price is not None:
s["contest"] = _clamp01(1.0 - abs(price - 0.5) / 0.5)
# longshot: price hugging 0 or 1 → potential mispriced tail.
s["longshot"] = _clamp01((abs(price - 0.5) - 0.4) / 0.1)
else:
s["contest"] = 0.0
s["longshot"] = 0.0
# movement: |24h change|, 20c move ~= saturated.
if change is not None:
s["movement"] = _clamp01(abs(change) / 0.20)
else:
s["movement"] = 0.0
blended = sum(WEIGHTS[k] * s.get(k, 0.0) for k in WEIGHTS)
m["signals"] = {k: round(v, 3) for k, v in s.items()}
m["opportunity"] = round(100 * blended, 1)
return m
# --------------------------------------------------------------------------- #
# Normalization #
# --------------------------------------------------------------------------- #
def _normalize_market(mkt: dict, event: dict | None = None) -> dict:
"""Flatten a Gamma market (+ optional parent event) into our shape."""
outcomes = mkt.get("outcomes")
if isinstance(outcomes, str):
try:
outcomes = json.loads(outcomes)
except Exception:
outcomes = None
prices = mkt.get("outcomePrices")
if isinstance(prices, str):
try:
prices = json.loads(prices)
except Exception:
prices = None
yes_price = None
if isinstance(prices, list) and prices:
yes_price = _parse_float(prices[0])
if yes_price is None:
yes_price = _parse_float(mkt.get("lastTradePrice"))
slug = mkt.get("slug") or (event.get("slug") if event else "") or ""
ev_slug = (event.get("slug") if event else "") or slug
vol = _parse_float(mkt.get("volume")) or (
_parse_float(event.get("volume")) if event else None
) or 0.0
liq = _parse_float(mkt.get("liquidity")) or (
_parse_float(event.get("liquidity")) if event else None
) or 0.0
return {
"question": mkt.get("question")
or mkt.get("groupItemTitle")
or (event.get("title") if event else "")
or "",
"slug": slug,
"event_slug": ev_slug,
"url": f"https://polymarket.com/event/{ev_slug}" if ev_slug else "",
"yes_price": yes_price,
"outcomes": outcomes,
"outcome_prices": [p for p in (prices or [])] if prices else None,
"volume": vol,
"liquidity": liq,
"spread": _parse_float(mkt.get("spread")),
"price_change_24h": _parse_float(mkt.get("oneDayPriceChange")),
"end": mkt.get("endDate") or (event.get("endDate") if event else None),
"end_dt": _parse_dt(mkt.get("endDate") or (event.get("endDate") if event else None)),
"clob_token_ids": mkt.get("clobTokenIds"),
"active": mkt.get("active"),
"closed": mkt.get("closed"),
}
def _enrich_clob(m: dict) -> dict:
"""Best-effort order-book depth + spread from CLOB for a YES token.
Never raises — CLOB shape shifts and this is a bonus signal, not core.
"""
token_ids = m.get("clob_token_ids")
if isinstance(token_ids, str):
try:
token_ids = json.loads(token_ids)
except Exception:
token_ids = None
if not (isinstance(token_ids, list) and token_ids):
return m
tok = token_ids[0]
try:
book = _get_json(f"{CLOB}/book?token_id={urllib.parse.quote(str(tok))}", timeout=12)
bids = book.get("bids") or []
asks = book.get("asks") or []
best_bid = max((_parse_float(b.get("price")) for b in bids), default=None)
best_ask = min((_parse_float(a.get("price")) for a in asks), default=None)
if best_bid is not None and best_ask is not None:
m["spread"] = round(abs(best_ask - best_bid), 4)
depth = 0.0
for side in (bids, asks):
for lvl in side:
p = _parse_float(lvl.get("price")) or 0.0
sz = _parse_float(lvl.get("size")) or 0.0
depth += p * sz
if depth > 0:
# Blend CLOB depth into liquidity if Gamma gave us nothing.
m["liquidity"] = max(m.get("liquidity") or 0.0, depth)
m["clob_ok"] = True
except Exception as e: # noqa: BLE001
m["clob_ok"] = False
m["clob_error"] = str(e)[:120]
return m
# --------------------------------------------------------------------------- #
# Fetchers #
# --------------------------------------------------------------------------- #
def fetch_scan(topic: str, limit: int) -> list[dict]:
"""Search live markets matching a topic via Gamma public-search."""
q = urllib.parse.quote(topic)
url = (
f"{GAMMA}/public-search?q={q}"
f"&limit_per_type={max(limit * 2, 20)}&events_status=active"
)
data = _get_json(url)
events = data.get("events", []) if isinstance(data, dict) else []
out: list[dict] = []
for ev in events:
for mkt in (ev.get("markets") or [])[:6]:
if mkt.get("closed"):
continue
out.append(_normalize_market(mkt, ev))
return out
def fetch_trending(limit: int) -> list[dict]:
"""Most active live markets by volume via Gamma /markets."""
url = (
f"{GAMMA}/markets?active=true&closed=false&archived=false"
f"&order=volume24hr&ascending=false&limit={max(limit * 2, 30)}"
)
data = _get_json(url)
markets = data if isinstance(data, list) else data.get("markets", [])
return [_normalize_market(m) for m in markets]
def fetch_market(slug_or_id: str) -> list[dict]:
"""Detail for one market by slug (preferred) or numeric id."""
if slug_or_id.isdigit():
url = f"{GAMMA}/markets/{slug_or_id}"
data = _get_json(url)
mk = data if isinstance(data, dict) else None
return [_normalize_market(mk)] if mk else []
url = f"{GAMMA}/markets?slug={urllib.parse.quote(slug_or_id)}"
data = _get_json(url)
markets = data if isinstance(data, list) else data.get("markets", [])
if not markets:
# Fall back to event slug → its markets.
ev = _get_json(f"{GAMMA}/events?slug={urllib.parse.quote(slug_or_id)}")
evs = ev if isinstance(ev, list) else ev.get("events", [])
out = []
for e in evs:
for m in e.get("markets") or []:
out.append(_normalize_market(m, e))
return out
return [_normalize_market(m) for m in markets]
# --------------------------------------------------------------------------- #
# Pipeline #
# --------------------------------------------------------------------------- #
def rank(markets: list[dict], limit: int, use_clob: bool) -> list[dict]:
# De-dupe by slug+question.
seen = set()
uniq = []
for m in markets:
key = (m.get("slug"), m.get("question"))
if key in seen:
continue
seen.add(key)
uniq.append(m)
# Pre-score to pick the top-N worth an (expensive) CLOB round-trip.
for m in uniq:
score_market(m)
uniq.sort(key=lambda x: x.get("opportunity", 0), reverse=True)
top = uniq[: max(limit, 1)]
if use_clob and top:
with concurrent.futures.ThreadPoolExecutor(max_workers=6) as ex:
list(ex.map(_enrich_clob, top))
for m in top:
score_market(m) # re-score with CLOB-enriched liquidity/spread
top.sort(key=lambda x: x.get("opportunity", 0), reverse=True)
return top
# --------------------------------------------------------------------------- #
# Emitters #
# --------------------------------------------------------------------------- #
def _fmt_pct(p):
return "—" if p is None else f"{p * 100:.0f}%"
def emit_json(cmd: str, arg: str, rows: list[dict]) -> str:
return json.dumps(
{
"tool": "polymtrx",
"command": cmd,
"query": arg,
"generated_at": _now().isoformat(),
"cost": "$0 (local; Polymarket public APIs)",
"count": len(rows),
"markets": rows,
},
indent=2,
default=str, # datetimes (end_dt) → ISO string
)
def emit_markdown(cmd: str, arg: str, rows: list[dict]) -> str:
lines = [
f"# polymtrx {cmd}" + (f' — "{arg}"' if arg else ""),
f"_generated {_now():%Y-%m-%d %H:%M UTC} · $0 (local) · {len(rows)} markets · read-only_",
"",
]
if not rows:
lines.append("_No live markets matched._")
return "\n".join(lines)
for i, m in enumerate(rows, 1):
sig = m.get("signals", {})
end = m.get("end_dt")
when = f"{(end - _now()).days}d" if end else "—"
lines.append(f"### {i}. {m['question']} · **{m.get('opportunity', 0)}/100**")
lines.append(
f"- YES **{_fmt_pct(m.get('yes_price'))}** · vol ${m.get('volume', 0):,.0f} · "
f"liq ${m.get('liquidity', 0):,.0f} · "
f"spread {('—' if m.get('spread') is None else f'{m['spread']*100:.1f}c')} · "
f"resolves in {when}"
)
top_sig = sorted(sig.items(), key=lambda kv: kv[1], reverse=True)[:3]
lines.append("- top signals: " + ", ".join(f"{k} {v:.2f}" for k, v in top_sig))
if m.get("url"):
lines.append(f"- {m['url']}")
lines.append("")
return "\n".join(lines)
# --------------------------------------------------------------------------- #
# doctor #
# --------------------------------------------------------------------------- #
def doctor() -> int:
checks = [
("gamma public-search", f"{GAMMA}/public-search?q=test&limit_per_type=1&events_status=active"),
("gamma markets", f"{GAMMA}/markets?active=true&closed=false&limit=1"),
("clob ok", f"{CLOB}/ok"),
]
ok = True
print("polymtrx doctor — endpoint health ($0, read-only)\n")
for name, url in checks:
try:
_get(url, timeout=12)
print(f" ✅ {name}")
except Exception as e: # noqa: BLE001
ok = False
print(f" ❌ {name}: {str(e)[:100]}")
print("\nall good" if ok else "\nsome endpoints failed — see above")
return 0 if ok else 1
# --------------------------------------------------------------------------- #
# CLI #
# --------------------------------------------------------------------------- #
def main(argv: list[str]) -> int:
p = argparse.ArgumentParser(prog="polymtrx", add_help=True, description=__doc__)
p.add_argument("command", choices=["scan", "trending", "market", "doctor"])
p.add_argument("query", nargs="?", default="", help="topic (scan) or slug/id (market)")
p.add_argument("--limit", type=int, default=12, help="max markets to return")
p.add_argument("--emit", choices=["json", "markdown"], default="json")
p.add_argument("--no-clob", action="store_true", help="skip order-book depth (faster)")
args = p.parse_args(argv)
if args.command == "doctor":
return doctor()
use_clob = not args.no_clob
try:
if args.command == "scan":
if not args.query:
print("error: scan needs a topic, e.g. polymtrx scan \"fed rate\"", file=sys.stderr)
return 2
markets = fetch_scan(args.query, args.limit)
rows = rank(markets, args.limit, use_clob)
elif args.command == "trending":
markets = fetch_trending(args.limit)
rows = rank(markets, args.limit, use_clob)
elif args.command == "market":
if not args.query:
print("error: market needs a slug or id", file=sys.stderr)
return 2
markets = fetch_market(args.query)
rows = rank(markets, max(len(markets), 1), use_clob)
else: # pragma: no cover
return 2
except urllib.error.HTTPError as e:
print(f"HTTP {e.code} from Polymarket: {e.reason}", file=sys.stderr)
return 1
except Exception as e: # noqa: BLE001
print(f"error: {e}", file=sys.stderr)
return 1
out = emit_markdown(args.command, args.query, rows) if args.emit == "markdown" else emit_json(
args.command, args.query, rows
)
print(out)
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))