# v0.2.0 # { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } # # BetMarket ÃƒÆ’Ã†â€™Ãƒâ€ Ã¢â‚¬â„¢ÃƒÆ’Ã¢â‚¬Â ÃƒÂ¢Ã¢â€šÂ¬Ã¢â€žÂ¢ÃƒÆ’Ã†â€™ÃƒÂ¢Ã¢â€šÂ¬Ã…Â¡ÃƒÆ’Ã¢â‚¬Å¡Ãƒâ€šÃ‚Â¢ÃƒÆ’Ã†â€™Ãƒâ€ Ã¢â‚¬â„¢ÃƒÆ’Ã¢â‚¬Å¡Ãƒâ€šÃ‚Â¢ÃƒÆ’Ã†â€™Ãƒâ€šÃ‚Â¢ÃƒÆ’Ã‚Â¢ÃƒÂ¢Ã¢â€šÂ¬Ã…Â¡Ãƒâ€šÃ‚Â¬ÃƒÆ’Ã¢â‚¬Â¦Ãƒâ€šÃ‚Â¡ÃƒÆ’Ã†â€™ÃƒÂ¢Ã¢â€šÂ¬Ã…Â¡ÃƒÆ’Ã¢â‚¬Å¡Ãƒâ€šÃ‚Â¬ÃƒÆ’Ã†â€™Ãƒâ€ Ã¢â‚¬â„¢ÃƒÆ’Ã¢â‚¬Å¡Ãƒâ€šÃ‚Â¢ÃƒÆ’Ã†â€™Ãƒâ€šÃ‚Â¢ÃƒÆ’Ã‚Â¢ÃƒÂ¢Ã¢â‚¬Å¡Ã‚Â¬Ãƒâ€¦Ã‚Â¡ÃƒÆ’Ã¢â‚¬Å¡Ãƒâ€šÃ‚Â¬ÃƒÆ’Ã†â€™ÃƒÂ¢Ã¢â€šÂ¬Ã…Â¡ÃƒÆ’Ã¢â‚¬Å¡Ãƒâ€šÃ‚Â peer-to-peer YES/NO betting on real-world events. # - Anyone can create_bet by staking GEN with a YES/NO claim and a deadline. # - Anyone else can accept_bet by matching the stake (taking the opposite side). # - After the deadline, anyone can resolve(). The contract fetches web evidence # and asks a single on-chain LLM call to decide YES / NO / PENDING / UNCLEAR. # - Winner gets pot - fee. House gets fee. UNCLEAR refunds both sides. from genlayer import * from dataclasses import dataclass from datetime import datetime from urllib.parse import quote, unquote import json ZERO_ADDR = Address("0x0000000000000000000000000000000000000000") PENDING_REFUND_SECONDS = 3600 TRUSTED_SOURCE_PREFIXES = ( "https://www.bbc.com/", "https://bbc.com/", "https://www.espn.com/", "https://www.skysports.com/", "https://www.uefa.com/", "https://www.fifa.com/", "https://www.soccerway.com/", "https://www.flashscore.com/", "https://www.nba.com/", "https://www.nfl.com/", "https://www.mlb.com/", "https://www.nhl.com/", "https://www.cbssports.com/", "https://www.reuters.com/", "https://apnews.com/", "https://www.theguardian.com/", "https://www.nytimes.com/", "https://www.washingtonpost.com/", "https://www.ft.com/", "https://www.cnn.com/", "https://www.nbcnews.com/", "https://www.cbsnews.com/", "https://abcnews.go.com/", "https://www.aljazeera.com/", "https://www.who.int/", "https://www.un.org/", "https://www.nato.int/", "https://www.sec.gov/", "https://www.cftc.gov/", "https://www.justice.gov/", "https://home.treasury.gov/", "https://www.coindesk.com/", "https://cointelegraph.com/", "https://decrypt.co/", "https://www.theblock.co/", "https://www.coinbase.com/", "https://www.binance.com/", "https://www.kraken.com/", "https://api.dexscreener.com/", "https://www.efcc.gov.ng/", "https://www.cbn.gov.ng/", "https://www.channelstv.com/", "https://www.tvcnews.tv/", "https://www.premiumtimesng.com/", "https://punchng.com/", "https://nta.ng/", "https://www.whitehouse.gov/", "https://www.gov.uk/", "https://ec.europa.eu/", "https://www.pib.gov.in/", "https://www.rbi.org.in/", "https://www.gov.za/", "https://www.resbank.co.za/", "https://www.president.go.ke/", "https://www.centralbank.go.ke/", "https://presidency.gov.gh/", "https://www.bog.gov.gh/", "https://www.boxofficemojo.com/", "https://www.the-numbers.com/", "https://www.imdb.com/", "https://www.oscars.org/", "https://goldenglobes.com/", "https://www.bafta.org/", ) @gl.evm.contract_interface class NativeRecipient: class View: pass class Write: pass @allow_storage @dataclass class ExternalBet: external_id: str base_chain_id: u256 base_escrow: str base_bet_id: u256 token_address: str stake_units: str creator: str accepter: str creator_yes: bool question: str deadline: u256 resolution_url: str status: str outcome: str reasoning: str pending_since: u256 @allow_storage @dataclass class Bet: bet_id: u256 creator: Address accepter: Address creator_yes: bool # True if creator bet YES on the claim question: str # the YES/NO claim, e.g. "Lakers beat Bulls on 2026-04-30" resolution_url: str # optional URL the resolver should consult; "" = web search stake: u256 # amount each side staked, in wei deadline: u256 # unix timestamp; resolve() callable after this status: str # "open" | "active" | "resolved" | "cancelled" outcome: str # "" | "YES" | "NO" | "PENDING" | "UNCLEAR" winner: Address # zero addr if UNCLEAR / unresolved created_at: u256 chat_id: str # telegram chat the bet was placed in (display only) creator_handle: str # display handle for nicer UI accepter_handle: str reasoning: str # short explanation from the LLM at resolve time pending_since: u256 # first PENDING timestamp; zero until needed cancel_requested_by: Address # first active-refund requester; zero until needed class BetMarket(gl.Contract): house: Address fee_bps: u256 # 10000 == 100%; default 1000 == 10% next_id: u256 bets: TreeMap[u256, Bet] open_ids: DynArray[u256] active_ids: DynArray[u256] resolved_ids: DynArray[u256] cancel_requests: TreeMap[u256, Address] # Per-address tallies (for /leaderboard) wins: TreeMap[Address, u256] losses: TreeMap[Address, u256] profit_wei: TreeMap[Address, u256] # cumulative net winnings in wei handles: TreeMap[Address, str] # last seen handle per address external_operator: Address external_bets: TreeMap[str, ExternalBet] def __init__(self, house_addr: str, fee_bps: int, external_operator: str): self.house = Address(house_addr) self.fee_bps = u256(fee_bps) self.external_operator = Address(external_operator) self.next_id = u256(1) # ---------------- Mutations ---------------- @gl.public.write.payable def create_bet( self, question: str, deadline: int, creator_yes: bool, resolution_url: str, chat_id: str, creator_handle: str, ) -> int: amount = gl.message.value assert amount > u256(0), "Must stake > 0 GEN" assert u256(deadline) > int(datetime.now().timestamp()), "Deadline must be in the future" assert len(question) > 0 and len(question) < 500, "Question length out of range" assert self._valid_resolution_spec(resolution_url), "Resolver source is not trusted" bet_id = self.next_id self.next_id = self.next_id + u256(1) bet = Bet( bet_id=bet_id, creator=gl.message.sender_address, accepter=ZERO_ADDR, creator_yes=creator_yes, question=question, resolution_url=resolution_url, stake=amount, deadline=u256(deadline), status="open", outcome="", winner=ZERO_ADDR, created_at=int(datetime.now().timestamp()), chat_id=chat_id, creator_handle=creator_handle, accepter_handle="", reasoning="", pending_since=u256(0), cancel_requested_by=ZERO_ADDR, ) self.bets[bet_id] = bet self.open_ids.append(bet_id) self.handles[gl.message.sender_address] = creator_handle return int(bet_id) @gl.public.write.payable def accept_bet(self, bet_id: int, accepter_handle: str) -> None: bid = u256(bet_id) assert bid in self.bets, "No such bet" bet = self.bets[bid] assert bet.status == "open", "Bet is not open" assert gl.message.sender_address != bet.creator, "Cannot accept your own bet" assert gl.message.value == bet.stake, "Must match stake exactly" assert int(datetime.now().timestamp()) < bet.deadline, "Bet expired" bet.accepter = gl.message.sender_address bet.accepter_handle = accepter_handle bet.status = "active" self.bets[bid] = bet self._remove_id(self.open_ids, bid) self.active_ids.append(bid) self.handles[gl.message.sender_address] = accepter_handle @gl.public.write def cancel_bet(self, bet_id: int) -> None: bid = u256(bet_id) assert bid in self.bets, "No such bet" bet = self.bets[bid] assert bet.status == "open", "Already accepted or resolved" assert gl.message.sender_address == bet.creator, "Only creator may cancel" bet.status = "cancelled" self.bets[bid] = bet self._remove_id(self.open_ids, bid) self._send_gen(bet.creator, bet.stake) @gl.public.write def request_cancel_active(self, bet_id: int) -> None: bid = u256(bet_id) assert bid in self.bets, "No such bet" bet = self.bets[bid] assert bet.status == "active", "Bet is not active" sender = gl.message.sender_address assert sender == bet.creator or sender == bet.accepter, "Only bettors may request cancellation" requester = bet.cancel_requested_by if requester == ZERO_ADDR: bet.cancel_requested_by = sender self.bets[bid] = bet return assert requester != sender, "The other bettor must also request the refund" assert requester == bet.creator or requester == bet.accepter, "Invalid stored refund requester" bet.cancel_requested_by = ZERO_ADDR bet.status = "cancelled" bet.outcome = "CANCELLED" bet.winner = ZERO_ADDR bet.reasoning = "Both bettors agreed to cancel and refund before resolution." self.bets[bid] = bet self._remove_id(self.active_ids, bid) self._send_gen(bet.creator, bet.stake) self._send_gen(bet.accepter, bet.stake) @gl.public.write def resolve(self, bet_id: int) -> None: bid = u256(bet_id) assert bid in self.bets, "No such bet" bet = self.bets[bid] assert bet.status == "active", "Bet is not active" assert int(datetime.now().timestamp()) >= bet.deadline, "Too early to resolve" question = bet.question url = bet.resolution_url result = self._resolve_claim(question, url, int(bet.deadline)) outcome = result["verdict"] reasoning = result["reasoning"] bet.outcome = outcome bet.reasoning = reasoning if outcome == "PENDING" or outcome == "UNCLEAR": now_ts = u256(int(datetime.now().timestamp())) if bet.pending_since == u256(0): bet.pending_since = now_ts bet.status = "active" self.bets[bid] = bet return if now_ts < bet.pending_since + u256(PENDING_REFUND_SECONDS): bet.status = "active" self.bets[bid] = bet return outcome = "UNCLEAR" reasoning = "No conclusive result was available for one hour after the first settlement attempt. Both bettors were refunded." bet.outcome = outcome bet.reasoning = reasoning if outcome == "UNCLEAR": # Refund both sides; no fee, no winner. bet.status = "resolved" bet.winner = ZERO_ADDR self.bets[bid] = bet self._remove_id(self.active_ids, bid) self.resolved_ids.append(bid) self._send_gen(bet.creator, bet.stake) self._send_gen(bet.accepter, bet.stake) return creator_won = (outcome == "YES" and bet.creator_yes) or ( outcome == "NO" and not bet.creator_yes ) winner = bet.creator if creator_won else bet.accepter loser = bet.accepter if creator_won else bet.creator bet.status = "resolved" bet.winner = winner self.bets[bid] = bet self._remove_id(self.active_ids, bid) self.resolved_ids.append(bid) pot = bet.stake + bet.stake fee = (pot * self.fee_bps) // u256(10000) winner_amount = pot - fee self._send_gen(winner, winner_amount) if fee > u256(0): self._send_gen(self.house, fee) self.wins[winner] = self.wins.get(winner, u256(0)) + u256(1) self.losses[loser] = self.losses.get(loser, u256(0)) + u256(1) gained = winner_amount - bet.stake self.profit_wei[winner] = self.profit_wei.get(winner, u256(0)) + gained # ---------------- External-chain registry ---------------- @gl.public.write def register_external_bet( self, external_id: str, base_chain_id: int, base_escrow: str, base_bet_id: int, token_address: str, stake_units: str, creator: str, question: str, deadline: int, creator_yes: bool, resolution_url: str, ) -> None: assert gl.message.sender_address == self.external_operator, "Only external operator may register" assert len(external_id) > 0 and external_id not in self.external_bets, "External bet already registered" assert base_chain_id > 0 and base_bet_id > 0, "Invalid external chain identifiers" assert len(base_escrow) > 0 and len(token_address) > 0 and len(stake_units) > 0, "Incomplete external custody record" assert len(creator) > 0 and len(question) > 0 and len(question) < 500, "Invalid external bet" assert deadline > int(datetime.now().timestamp()), "External deadline must be future" assert self._valid_resolution_spec(resolution_url), "External resolver source is not trusted" self.external_bets[external_id] = ExternalBet( external_id=external_id, base_chain_id=u256(base_chain_id), base_escrow=base_escrow, base_bet_id=u256(base_bet_id), token_address=token_address, stake_units=stake_units, creator=creator, accepter="", creator_yes=creator_yes, question=question, deadline=u256(deadline), resolution_url=resolution_url, status="open", outcome="", reasoning="", pending_since=u256(0), ) @gl.public.write def activate_external_bet(self, external_id: str, accepter: str) -> None: assert gl.message.sender_address == self.external_operator, "Only external operator may activate" assert external_id in self.external_bets, "External bet not found" bet = self.external_bets[external_id] assert bet.status == "open", "External bet is not open" assert len(accepter) > 0 and accepter != bet.creator, "Invalid accepter" bet.accepter = accepter bet.status = "active" self.external_bets[external_id] = bet @gl.public.write def cancel_external_bet(self, external_id: str) -> None: assert gl.message.sender_address == self.external_operator, "Only external operator may cancel" assert external_id in self.external_bets, "External bet not found" bet = self.external_bets[external_id] assert bet.status == "open" or bet.status == "active", "External bet cannot be cancelled" bet.status = "cancelled" bet.outcome = "CANCELLED" bet.reasoning = "Base escrow cancelled and refunded the external bet." self.external_bets[external_id] = bet @gl.public.write def resolve_external_bet(self, external_id: str) -> None: assert gl.message.sender_address == self.external_operator, "Only external operator may resolve" assert external_id in self.external_bets, "External bet not found" bet = self.external_bets[external_id] assert bet.status == "active", "External bet is not active" assert int(datetime.now().timestamp()) >= bet.deadline, "Too early to resolve external bet" result = self._resolve_claim(bet.question, bet.resolution_url, int(bet.deadline)) outcome = result["verdict"] bet.outcome = outcome bet.reasoning = result["reasoning"] if outcome == "PENDING" or outcome == "UNCLEAR": now_ts = u256(int(datetime.now().timestamp())) if bet.pending_since == u256(0): bet.pending_since = now_ts self.external_bets[external_id] = bet return if now_ts < bet.pending_since + u256(PENDING_REFUND_SECONDS): self.external_bets[external_id] = bet return bet.outcome = "UNCLEAR" bet.reasoning = "No conclusive external result was available for one hour after the first settlement attempt." bet.status = "resolved" self.external_bets[external_id] = bet @gl.public.view def get_external_bet(self, external_id: str) -> dict: if external_id not in self.external_bets: return {} bet = self.external_bets[external_id] return { "external_id": bet.external_id, "base_chain_id": int(bet.base_chain_id), "base_escrow": bet.base_escrow, "base_bet_id": int(bet.base_bet_id), "token_address": bet.token_address, "stake_units": bet.stake_units, "creator": bet.creator, "accepter": bet.accepter, "creator_yes": bet.creator_yes, "question": bet.question, "deadline": int(bet.deadline), "resolution_url": bet.resolution_url, "status": bet.status, "outcome": bet.outcome, "reasoning": bet.reasoning, "pending_since": int(bet.pending_since), } # ---------------- Views ---------------- @gl.public.view def get_bet(self, bet_id: int) -> dict: bid = u256(bet_id) if bid not in self.bets: return {} return self._bet_to_dict(self.bets[bid]) @gl.public.view def list_bets(self, status: str, limit: int) -> list: if status == "open": ids = self.open_ids elif status == "active": ids = self.active_ids elif status == "resolved": ids = self.resolved_ids else: return [] out = [] n = len(ids) # Most recent first i = n - 1 count = 0 while i >= 0 and count < limit: out.append(self._bet_to_dict(self.bets[ids[i]])) i -= 1 count += 1 return out @gl.public.view def my_bets(self, addr: str, limit: int) -> list: target = Address(addr) out = [] # Walk all bets newest-first up to limit hits i = int(self.next_id) - 1 while i >= 1 and len(out) < limit: bid = u256(i) if bid in self.bets: b = self.bets[bid] if b.creator == target or b.accepter == target: out.append(self._bet_to_dict(b)) i -= 1 return out @gl.public.view def leaderboard(self, limit: int) -> list: rows = [] for addr in self.profit_wei: rows.append({ "address": str(addr), "handle": self.handles.get(addr, ""), "wins": int(self.wins.get(addr, u256(0))), "losses": int(self.losses.get(addr, u256(0))), "profit_wei": str(self.profit_wei.get(addr, u256(0))), }) rows.sort(key=lambda r: int(r["profit_wei"]), reverse=True) return rows[:limit] @gl.public.view def stats(self) -> dict: return { "total_bets": int(self.next_id) - 1, "open": len(self.open_ids), "active": len(self.active_ids), "resolved": len(self.resolved_ids), "house": str(self.house), "fee_bps": int(self.fee_bps), } # ---------------- Internals ---------------- def _remove_id(self, arr: DynArray[u256], value: u256) -> None: n = len(arr) for i in range(n): if arr[i] == value: arr[i] = arr[n - 1] arr.pop() return def _send_gen(self, recipient: Address, amount: u256) -> None: # Native GEN payouts to user wallets must be emitted as EVM-layer # transfers. ContractAt is for Intelligent Contract messages. NativeRecipient(recipient).emit_transfer(value=amount) def _bet_to_dict(self, b: Bet) -> dict: return { "id": int(b.bet_id), "creator": str(b.creator), "creator_handle": b.creator_handle, "accepter": str(b.accepter), "accepter_handle": b.accepter_handle, "creator_yes": b.creator_yes, "question": b.question, "resolution_url": b.resolution_url, "stake": str(b.stake), "deadline": int(b.deadline), "status": b.status, "outcome": b.outcome, "winner": str(b.winner), "created_at": int(b.created_at), "chat_id": b.chat_id, "reasoning": b.reasoning, "pending_since": int(b.pending_since), "cancel_requested_by": str(b.cancel_requested_by), } def _resolve_claim(self, question: str, url: str, deadline: int) -> dict: if url.startswith("cex:"): return self._resolve_cex_price(url) if url.startswith("crypto:"): return self._resolve_crypto_price(url) if url.startswith("token:"): return self._resolve_dex_token(url) if url.startswith("sports:"): return self._resolve_sports_event(url) if url.startswith("news:") or url.startswith("announcement:") or url.startswith("movie:"): return self._resolve_public_claim(question, url, deadline) # Keep validator agreement focused on the verdict only. Free-form # reasoning differs too often across validators and causes false # consensus failures. def nondet_verdict() -> str: evidence = self._fetch_web_evidence(question, url) prompt = self._build_prompt(question, url, evidence) try: res = gl.nondet.exec_prompt(prompt, response_format="json") except Exception: try: res = gl.nondet.exec_prompt(prompt) except Exception: return "UNCLEAR" if isinstance(res, str): try: data = json.loads(res) except Exception: return "UNCLEAR" else: data = res verdict = str(data.get("verdict", "UNCLEAR")).strip().upper() if verdict not in ("YES", "NO", "UNCLEAR"): return "UNCLEAR" confidence_raw = data.get("confidence", 0) try: confidence = float(confidence_raw) except Exception: confidence = 0.0 if verdict != "UNCLEAR" and confidence < 0.55: return "UNCLEAR" return verdict try: verdict = gl.eq_principle.prompt_comparative( nondet_verdict, principle=( "Both answers must indicate the same final verdict. " "YES/TRUE are equivalent, NO/FALSE are equivalent, and " "UNCLEAR/UNKNOWN/INSUFFICIENT_EVIDENCE are equivalent. " "Different final outcomes are not equivalent." ), ) except Exception: verdict = "UNCLEAR" outcome = str(verdict).strip().upper() if outcome not in ("YES", "NO", "UNCLEAR"): outcome = "UNCLEAR" if outcome == "YES": reason = "Validators agreed the claim resolved YES from available web evidence." elif outcome == "NO": reason = "Validators agreed the claim resolved NO from available web evidence." else: reason = "Validators could not reach a confident YES or NO verdict." return {"verdict": outcome, "reasoning": reason} def _resolve_crypto_price(self, spec: str) -> dict: parts = spec.split(":") if len(parts) != 7 or parts[1] != "coinbase": return { "verdict": "UNCLEAR", "reasoning": "Crypto price resolver was not configured correctly.", } product = parts[2] comparator = parts[3] target_text = parts[4] start_text = parts[5] end_text = parts[6] try: target = float(target_text) start_ts = int(start_text) end_ts = int(end_text) except Exception: return { "verdict": "UNCLEAR", "reasoning": "Crypto price resolver inputs were invalid.", } def nondet_crypto_verdict() -> str: try: start_iso = quote(datetime.utcfromtimestamp(start_ts).isoformat() + "Z") end_iso = quote(datetime.utcfromtimestamp(end_ts).isoformat() + "Z") api_url = ( f"https://api.exchange.coinbase.com/products/{product}/candles" f"?granularity=60&start={start_iso}&end={end_iso}" ) raw = gl.nondet.web.render(api_url, mode="text")[:12000] first = raw.find("[") last = raw.rfind("]") if first < 0 or last <= first: return "UNCLEAR" candles = json.loads(raw[first : last + 1]) except Exception: return "UNCLEAR" if not isinstance(candles, list) or len(candles) == 0: return "UNCLEAR" high = 0.0 low = 1000000000000.0 seen = False for candle in candles: try: # Coinbase candles: [time, low, high, open, close, volume] candle_low = float(candle[1]) candle_high = float(candle[2]) except Exception: continue if candle_high > high: high = candle_high if candle_low < low: low = candle_low seen = True if not seen: return "UNCLEAR" if comparator == "GTE": return "YES" if high >= target else "NO" if comparator == "LTE": return "YES" if low <= target else "NO" return "UNCLEAR" try: verdict = gl.eq_principle.prompt_comparative( nondet_crypto_verdict, principle=( "Both answers must indicate the same crypto price result. " "YES/TRUE are equivalent, NO/FALSE are equivalent, and " "UNCLEAR/UNKNOWN/INSUFFICIENT_EVIDENCE are equivalent." ), ) except Exception: verdict = "UNCLEAR" outcome = str(verdict).strip().upper() if outcome not in ("YES", "NO", "UNCLEAR"): outcome = "UNCLEAR" direction = "at or above" if comparator == "GTE" else "at or below" if outcome == "UNCLEAR": reason = f"Could not verify {product} price data from Coinbase for the bet window." else: reason = f"Coinbase candles show {product} was {direction} {target_text}: {outcome}." return {"verdict": outcome, "reasoning": reason} def _resolve_cex_price(self, spec: str) -> dict: parts = spec.split(":") if len(parts) != 9 or parts[1] != "multi" or parts[2] != "v1": return {"verdict": "UNCLEAR", "reasoning": "CEX resolver was not configured correctly."} asset = parts[3].strip().upper() comparator = parts[4].strip().upper() target_text = parts[5].strip() sources = self._decode_part(parts[8]) try: target = float(target_text) except Exception: return {"verdict": "UNCLEAR", "reasoning": "CEX resolver target was invalid."} def check_range(high: float, low: float) -> str: if comparator == "GTE": return "YES" if high >= target else "NO" if comparator == "LTE": return "YES" if low <= target else "NO" return "UNCLEAR" def parse_rows(source: str, data) -> str: rows = data high_index = 2 low_index = 1 if "api.binance.com" in source or "api.mexc.com" in source: low_index = 3 elif "api.bybit.com" in source: rows = data.get("result", {}).get("list", []) if isinstance(data, dict) else [] low_index = 3 elif "www.okx.com" in source: rows = data.get("data", []) if isinstance(data, dict) else [] low_index = 3 elif "api.gateio.ws" in source: high_index = 3 low_index = 4 elif "api.kucoin.com" in source: rows = data.get("data", []) if isinstance(data, dict) else [] high_index = 3 low_index = 4 elif "sg-api.upbit.com" in source: if not isinstance(data, list): return "UNCLEAR" highs = [] lows = [] for candle in data: if isinstance(candle, dict): try: highs.append(float(candle.get("high_price", 0))) lows.append(float(candle.get("low_price", 0))) except Exception: pass return check_range(max(highs), min(lows)) if highs and lows else "UNCLEAR" if not isinstance(rows, list) or not rows: return "UNCLEAR" highs = [] lows = [] for candle in rows: if not isinstance(candle, list): continue try: highs.append(float(candle[high_index])) lows.append(float(candle[low_index])) except Exception: continue return check_range(max(highs), min(lows)) if highs and lows else "UNCLEAR" def nondet_cex_verdict() -> str: verdicts = [] for source in self._source_hints(sources): try: raw = gl.nondet.web.render(source, mode="text")[:24000] starts = [x for x in (raw.find("["), raw.find("{")) if x >= 0] if not starts: continue first = min(starts) last = max(raw.rfind("]"), raw.rfind("}")) if last <= first: continue verdict = parse_rows(source, json.loads(raw[first:last + 1])) if verdict in ("YES", "NO"): verdicts.append(verdict) except Exception: continue if not verdicts or any(verdict != verdicts[0] for verdict in verdicts): return "UNCLEAR" return verdicts[0] try: verdict = gl.eq_principle.prompt_comparative( nondet_cex_verdict, principle=( "Both answers must indicate the same multi-exchange crypto price result. " "YES/TRUE are equivalent, NO/FALSE are equivalent, and " "UNCLEAR/UNKNOWN/INSUFFICIENT_EVIDENCE are equivalent." ), ) except Exception: verdict = "UNCLEAR" outcome = str(verdict).strip().upper() if outcome not in ("YES", "NO", "UNCLEAR"): outcome = "UNCLEAR" direction = "at or above" if comparator == "GTE" else "at or below" reason = ( "Supported CEX markets had no reliable data or disagreed for " + asset + "." if outcome == "UNCLEAR" else "Available CEX candles agreed that " + asset + " was " + direction + " " + target_text + ": " + outcome + "." ) return {"verdict": outcome, "reasoning": reason} def _resolve_dex_token(self, spec: str) -> dict: parts = spec.split(":") if len(parts) < 12 or parts[1] != "dexscreener" or parts[2] != "v1": return {"verdict": "UNCLEAR", "reasoning": "DEX token resolver was not configured correctly."} chain = self._decode_part(parts[4]).strip().lower() token = self._decode_part(parts[5]).strip().lower() pair = self._decode_part(parts[6]).strip().lower() metric = self._decode_part(parts[7]).strip().lower() comparator = parts[8].strip().upper() target_text = parts[9].strip() source = self._decode_part(parts[11]).strip() try: target = float(target_text) except Exception: return {"verdict": "UNCLEAR", "reasoning": "DEX token target was invalid."} def nondet_dex_verdict() -> str: try: raw = gl.nondet.web.render(source, mode="text")[:20000] data = json.loads(raw) except Exception: return "UNCLEAR" rows = data if isinstance(data, list) else data.get("pairs", []) best = None best_liquidity = -1.0 for row in rows: if not isinstance(row, dict): continue if str(row.get("chainId", "")).strip().lower() != chain: continue base = row.get("baseToken", {}) if not isinstance(base, dict) or str(base.get("address", "")).strip().lower() != token: continue row_pair = str(row.get("pairAddress", "")).strip().lower() if len(pair) > 0 and row_pair != pair: continue liquidity = row.get("liquidity", {}) try: liquidity_usd = float(liquidity.get("usd", 0)) if isinstance(liquidity, dict) else 0.0 except Exception: liquidity_usd = 0.0 if liquidity_usd > best_liquidity: best = row best_liquidity = liquidity_usd if best is None or best_liquidity < 1000.0: return "UNCLEAR" field = "priceUsd" if metric == "market_cap": field = "marketCap" elif metric == "fdv": field = "fdv" try: value = float(best.get(field, 0)) except Exception: return "UNCLEAR" if value <= 0: return "UNCLEAR" if comparator == "GTE": return "YES" if value >= target else "NO" if comparator == "LTE": return "YES" if value <= target else "NO" return "UNCLEAR" try: verdict = gl.eq_principle.prompt_comparative( nondet_dex_verdict, principle=( "Both answers must indicate the same DEX token metric result. " "YES/TRUE are equivalent, NO/FALSE are equivalent, and " "UNCLEAR/UNKNOWN/INSUFFICIENT_EVIDENCE are equivalent." ), ) except Exception: verdict = "UNCLEAR" outcome = str(verdict).strip().upper() if outcome not in ("YES", "NO", "UNCLEAR"): outcome = "UNCLEAR" if outcome == "UNCLEAR": reason = "DEX Screener did not provide a reliable matching pair with at least USD 1,000 liquidity." else: reason = "DEX Screener resolved " + metric + " " + comparator + " " + target_text + " as " + outcome + "." return {"verdict": outcome, "reasoning": reason} def _decode_part(self, value: str) -> str: try: return unquote(value) except Exception: return value def _source_hints(self, source_hint: str) -> list: hints = [] normalized = source_hint.replace("\r", "\n").replace(" || ", "\n") for raw in normalized.split("\n"): hint = raw.strip() if len(hint) == 0: continue hints.append(hint) if len(hints) >= 6: break return hints def _is_trusted_source(self, source: str) -> bool: lowered = source.strip().lower() for prefix in TRUSTED_SOURCE_PREFIXES: if lowered.startswith(prefix): return True return False def _trusted_source_count(self, evidence: str) -> int: return evidence.count("SOURCE ") def _valid_source_bundle(self, source_hint: str) -> bool: hints = self._source_hints(source_hint) return len(hints) >= 2 and all(self._is_trusted_source(source) for source in hints) def _valid_resolution_spec(self, spec: str) -> bool: if spec.startswith("cex:multi:v1:"): parts = spec.split(":") if len(parts) != 9: return False asset = parts[3].strip().upper() comparator = parts[4].strip().upper() sources = self._source_hints(self._decode_part(parts[8])) if len(asset) < 2 or len(asset) > 15 or not asset.isalnum(): return False if comparator not in ("GTE", "LTE") or len(sources) < 2: return False allowed = ( "https://api.exchange.coinbase.com/products/", "https://api.binance.com/api/v3/klines", "https://api.bybit.com/v5/market/kline", "https://www.okx.com/api/v5/market/candles", "https://api.gateio.ws/api/v4/spot/candlesticks", "https://api.mexc.com/api/v3/klines", "https://api.kucoin.com/api/v1/market/candles", "https://sg-api.upbit.com/v1/candles/minutes/1", ) return all(any(source.startswith(prefix) for prefix in allowed) for source in sources) if spec.startswith("crypto:coinbase:"): return True if spec.startswith("token:dexscreener:v1:"): parts = spec.split(":") if len(parts) < 12: return False chain = self._decode_part(parts[4]).strip().lower() token = self._decode_part(parts[5]).strip() pair = self._decode_part(parts[6]).strip() metric = self._decode_part(parts[7]).strip().lower() comparator = parts[8].strip().upper() source = self._decode_part(parts[11]).strip().lower() allowed_chains = ("ethereum", "base", "solana", "bsc", "arbitrum", "optimism", "polygon", "avalanche") if chain not in allowed_chains or len(token) < 32: return False if metric not in ("price_usd", "market_cap", "fdv") or comparator not in ("GTE", "LTE"): return False expected_token = "https://api.dexscreener.com/token-pairs/v1/" + chain + "/" + token.lower() expected_pair = "https://api.dexscreener.com/latest/dex/pairs/" + chain + "/" + pair.lower() return source == expected_token or (len(pair) > 0 and source == expected_pair) if spec.startswith("sports:web:v1:"): parts = spec.split(":") return len(parts) >= 9 and self._valid_source_bundle(self._decode_part(parts[8])) if spec.startswith("news:web:v1:") or spec.startswith("announcement:web:v1:") or spec.startswith("movie:web:v1:"): parts = spec.split(":") return len(parts) >= 7 and self._valid_source_bundle(self._decode_part(parts[6])) return False def _fetch_web_evidence(self, query: str, source_hint: str) -> str: chunks = [] source_count = 0 for source in self._source_hints(source_hint): if not self._is_trusted_source(source): continue try: text = gl.nondet.web.render(source, mode="text")[:3200] if len(text) > 0: source_count += 1 chunks.append("SOURCE " + str(source_count) + ": " + source + "\n" + text) except Exception: pass return "\n\n---\n\n".join(chunks)[:16000] def _resolve_sports_event(self, spec: str) -> dict: parts = spec.split(":") if len(parts) < 9 or parts[1] != "web" or parts[2] != "v1": return { "verdict": "UNCLEAR", "reasoning": "Sports resolver was not configured correctly.", } market = self._decode_part(parts[4]) selection = self._decode_part(parts[5]) event_name = self._decode_part(parts[6]) rule = self._decode_part(parts[7]) source_hint = self._decode_part(parts[8]) def nondet_sports_verdict() -> str: evidence = self._fetch_web_evidence( event_name + " final score result " + selection, source_hint, ) if self._trusted_source_count(evidence) < 2: return "PENDING" prompt = ( "You are resolving a GenLayer YES/NO sports bet.\n" "Resolve only from reliable event/result evidence. Prefer official league/team sources and established sports result publishers when sources conflict.\n" "Return PENDING if the event is not started, live, in extra time, in penalties, delayed, postponed, abandoned without a final official result, or the result is not posted yet.\n" "Return YES only if the final result satisfies the selected YES side.\n" "Return NO only if the final result is available and does not satisfy the selected YES side.\n" "Return UNCLEAR if the evidence is contradictory or too weak.\n" "Respond ONLY with JSON in this shape:\n" '{"state":"FINAL|PENDING|UNCLEAR","verdict":"YES|NO|PENDING|UNCLEAR","confidence":0.0,"reasoning":"short reason"}' "\n\nEVENT: " + event_name + "\nMARKET: " + market + "\nYES SELECTION: " + selection + "\nSETTLEMENT RULE: " + rule + "\n\nEVIDENCE:\n" + evidence ) try: res = gl.nondet.exec_prompt(prompt, response_format="json") except Exception: try: res = gl.nondet.exec_prompt(prompt) except Exception: return "PENDING" if isinstance(res, str): try: data = json.loads(res) except Exception: return "UNCLEAR" else: data = res state = str(data.get("state", "")).strip().upper() verdict = str(data.get("verdict", "UNCLEAR")).strip().upper() if state == "PENDING" or verdict == "PENDING": return "PENDING" if state == "UNCLEAR": return "UNCLEAR" confidence_raw = data.get("confidence", 0) try: confidence = float(confidence_raw) except Exception: confidence = 0.0 if verdict in ("YES", "NO") and confidence >= 0.60: return verdict return "UNCLEAR" try: verdict = gl.eq_principle.prompt_comparative( nondet_sports_verdict, principle=( "Both answers must indicate the same sports settlement state. " "YES/TRUE are equivalent, NO/FALSE are equivalent, " "PENDING/LIVE/NOT_FINAL/POSTPONED are equivalent, and " "UNCLEAR/UNKNOWN/INSUFFICIENT_EVIDENCE are equivalent." ), ) except Exception: verdict = "UNCLEAR" outcome = str(verdict).strip().upper() if outcome not in ("YES", "NO", "PENDING", "UNCLEAR"): outcome = "UNCLEAR" if outcome == "PENDING": reason = "Sports result is not final yet. Retry resolve after the official result is posted." elif outcome == "UNCLEAR": reason = "Validators could not verify a reliable final sports result." else: reason = f"Validators resolved the sports result as {outcome} for selection: {selection}." return {"verdict": outcome, "reasoning": reason} def _resolve_public_claim(self, question: str, spec: str, deadline: int) -> dict: parts = spec.split(":") kind = parts[0].strip().lower() if len(parts) > 0 else "news" if len(parts) < 7 or kind not in ("news", "announcement", "movie") or parts[1] != "web" or parts[2] != "v1": return { "verdict": "UNCLEAR", "reasoning": "Public-fact resolver was not configured correctly.", } try: claim_deadline = int(parts[3]) except Exception: claim_deadline = deadline claim = self._decode_part(parts[4]) rule = self._decode_part(parts[5]) source_hint = self._decode_part(parts[6]) buffer_seconds = 300 if kind == "announcement" else 1800 now_ts = int(datetime.now().timestamp()) if now_ts < claim_deadline + buffer_seconds: return { "verdict": "PENDING", "reasoning": "The deadline passed recently. Retry after the source-update buffer.", } def nondet_public_verdict() -> str: evidence = self._fetch_web_evidence( claim + " before " + datetime.utcfromtimestamp(claim_deadline).isoformat() + "Z", source_hint, ) if self._trusted_source_count(evidence) < 2: return "UNCLEAR" announcement_rule = "" if kind == "announcement": announcement_rule = ( "This is an announcement market. After the deadline and buffer, return NO when the rule names canonical " "publication channels and those inspected channels show no qualifying announcement. Do not require a separate denial.\n" ) prompt = ( "You are resolving a GenLayer YES/NO " + kind + " bet from trusted web evidence.\n" "Follow the written settlement rule exactly and prefer official records over secondary reporting.\n" + announcement_rule + "Return YES only when the defined condition was satisfied by the deadline.\n" "Return NO only when the defined observable condition failed by the deadline.\n" "Return PENDING only when an official result is explicitly still developing or not final.\n" "Return UNCLEAR when evidence is inaccessible, contradictory, weak, or the metric is ambiguous.\n" "Respond ONLY with JSON in this shape:\n" '{"state":"FINAL|PENDING|UNCLEAR","verdict":"YES|NO|PENDING|UNCLEAR","confidence":0.0,"reasoning":"short reason"}' "\n\nCLAIM: " + claim + "\nDEADLINE UTC: " + datetime.utcfromtimestamp(claim_deadline).isoformat() + "Z" + "\nSETTLEMENT RULE: " + rule + "\n\nEVIDENCE:\n" + evidence ) try: res = gl.nondet.exec_prompt(prompt, response_format="json") except Exception: try: res = gl.nondet.exec_prompt(prompt) except Exception: return "UNCLEAR" if isinstance(res, str): try: data = json.loads(res) except Exception: return "UNCLEAR" else: data = res state = str(data.get("state", "")).strip().upper() verdict = str(data.get("verdict", "UNCLEAR")).strip().upper() if state == "PENDING" or verdict == "PENDING": return "PENDING" if state == "UNCLEAR": return "UNCLEAR" try: confidence = float(data.get("confidence", 0)) except Exception: confidence = 0.0 if verdict in ("YES", "NO") and confidence >= 0.60: return verdict return "UNCLEAR" try: verdict = gl.eq_principle.prompt_comparative( nondet_public_verdict, principle=( "Both answers must indicate the same public-fact settlement state. " "YES/TRUE are equivalent, NO/FALSE are equivalent, " "PENDING/NOT_READY/STILL_DEVELOPING are equivalent, and " "UNCLEAR/UNKNOWN/INSUFFICIENT_EVIDENCE are equivalent." ), ) except Exception: verdict = "UNCLEAR" outcome = str(verdict).strip().upper() if outcome not in ("YES", "NO", "PENDING", "UNCLEAR"): outcome = "UNCLEAR" if outcome == "PENDING": reason = "Reliable " + kind + " resolution is not ready yet." elif outcome == "UNCLEAR": reason = "Validators could not reach a reliable " + kind + " verdict." else: reason = "Validators resolved the " + kind + " claim as " + outcome + "." return {"verdict": outcome, "reasoning": reason} def _build_prompt(self, question: str, url: str, evidence: str) -> str: hint = "" if len(url) > 0: hint = f"\nThe creator suggested this source: {url}" return ( "You are an impartial bet adjudicator on the GenLayer network.\n" "You will be given a YES/NO claim and web evidence.\n" "Decide whether the claim is TRUE (YES), FALSE (NO), or you don't have enough\n" "reliable evidence to tell (UNCLEAR).\n" "Use a single best verdict. If the evidence is weak, contradictory, or absent, return UNCLEAR.\n" "Return a confidence between 0 and 1.\n" "Respond ONLY with JSON, no markdown, in this exact shape:\n" '{"verdict": "YES" | "NO" | "UNCLEAR", "confidence": 0.0, "reasoning": ""}\n\n' f"CLAIM: {question}{hint}\n\n" f"EVIDENCE:\n{evidence}" )