Spaces:
Sleeping
Sleeping
File size: 4,309 Bytes
88d2f2a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 | """Builder-code registration & resolution.
Polymarket V2 (May 2026) lets builders register a short alphanumeric
code at ``polymarket.com/settings?tab=builder`` and earns 0.4% of every
fill routed through that code. For the hackathon we cannot reliably
reach the production registration UI from a headless backend, so we
keep two modes:
* **real** — operator pastes the code returned by the Polymarket UI
into the environment / DB and we just thread it through.
* **demo** — we deterministically derive a 10-char code from the
translator's wallet address. The same address always maps to the
same code, which lets ``resolve_translator_for_code`` reverse the
lookup without external state.
The mapping ``code -> translator_address`` is what
``BuilderFeeRouter.recordFill`` needs at fill time; we keep an
in-process cache plus a tiny JSON file under ``outputs/`` so the
mapping survives restarts during demos.
"""
from __future__ import annotations
import hashlib
import json
import os
import threading
from pathlib import Path
from typing import Optional
BUILDER_CODE_LENGTH = 10
# Path to the on-disk demo registry. Kept under outputs/ so it sits with
# the other hackathon artifacts and is easy to clean between demos.
_DEFAULT_REGISTRY_PATH = (
Path(__file__).resolve().parents[2] / "outputs" / "builder_codes.json"
)
# Guard concurrent writes from FillListener tasks running in parallel.
_REGISTRY_LOCK = threading.Lock()
def _registry_path() -> Path:
"""Allow tests to override the on-disk path via env var."""
override = os.getenv("POLYGLOT_BUILDER_REGISTRY_PATH")
return Path(override) if override else _DEFAULT_REGISTRY_PATH
def _load_registry() -> dict[str, str]:
path = _registry_path()
if not path.exists():
return {}
try:
with path.open("r", encoding="utf-8") as fh:
data = json.load(fh)
# Normalize: code -> address (lowercased hex). Drop anything malformed.
return {
str(code): str(addr).lower()
for code, addr in data.items()
if isinstance(code, str) and isinstance(addr, str)
}
except (OSError, json.JSONDecodeError):
return {}
def _save_registry(registry: dict[str, str]) -> None:
path = _registry_path()
path.parent.mkdir(parents=True, exist_ok=True)
tmp = path.with_suffix(".tmp")
with tmp.open("w", encoding="utf-8") as fh:
json.dump(registry, fh, indent=2, sort_keys=True)
tmp.replace(path)
def _derive_demo_code(translator_address: str) -> str:
"""Deterministic ``BUILDER_CODE_LENGTH``-char code from an address.
Uses SHA-256 truncated to keep collisions cosmetic for a hackathon
(probability ~ 2^-40 across realistic demo populations).
"""
digest = hashlib.sha256(translator_address.lower().encode("utf-8")).hexdigest()
return digest[:BUILDER_CODE_LENGTH].upper()
def register_builder_code(
translator_address: str,
*,
real_code: Optional[str] = None,
) -> str:
"""Register a builder code for ``translator_address``.
Parameters
----------
translator_address:
EVM address of the translator agent. Required.
real_code:
If provided, the caller already obtained a code from
polymarket.com (real mode). We persist the mapping but do not
derive anything. In demo mode this is ``None`` and we derive
deterministically.
Returns
-------
The builder code (alphanumeric, ``BUILDER_CODE_LENGTH`` chars in
demo mode; arbitrary length in real mode).
"""
if not translator_address or not translator_address.startswith("0x"):
raise ValueError("translator_address must be a 0x-prefixed EVM address")
code = real_code.strip() if real_code else _derive_demo_code(translator_address)
if not code:
raise ValueError("builder code must be non-empty")
with _REGISTRY_LOCK:
registry = _load_registry()
registry[code] = translator_address.lower()
_save_registry(registry)
return code
def resolve_translator_for_code(code: str) -> Optional[str]:
"""Reverse the mapping. Returns ``None`` if the code is unknown."""
if not code:
return None
with _REGISTRY_LOCK:
registry = _load_registry()
return registry.get(code)
|