# { "Depends": "py-genlayer:5jycge4q8k23462jtb0b9fyey1s9qz928sz2nbrd9mg4sxqg2qng" } import json import genlayer as gl from genlayer import u256 PRICE_SCALE = 1_000_000_000_000_000_000 RETURN_SCALE = 1_000_000 U256_MAX = 2**256 - 1 MAX_RESPONSE_BYTES = 65_536 MAX_ROWS = 100 ASSETS = { "US_INDICES": ["SPY", "QQQ", "IWM"], "ASIA_INDICES": ["EWJ", "EWY", "EWT"], "SECTOR_INDICES": ["SMH", "XBI", "XLE"], } def _parse_integer(value): if isinstance(value, bool): return None if isinstance(value, int): return value if 0 <= value <= U256_MAX else None if not isinstance(value, str) or not value or len(value) > 40: return None if not all("0" <= char <= "9" for char in value): return None number = int(value) return number if number <= U256_MAX else None def _parse_price(value): if isinstance(value, bool): return None if isinstance(value, int): text = str(value) elif isinstance(value, str): text = value else: return None if not text or len(text) > 60 or text.startswith("+") or text.startswith("-"): return None pieces = text.split(".") if len(pieces) > 2 or not pieces[0] or len(pieces[0]) > 38: return None if not all("0" <= char <= "9" for char in pieces[0]): return None fraction = pieces[1] if len(pieces) == 2 else "" if fraction and (len(fraction) > 18 or not all("0" <= char <= "9" for char in fraction)): return None scaled = int(pieces[0]) * PRICE_SCALE + int((fraction + "0" * 18)[:18]) return scaled if 0 < scaled <= U256_MAX else None def _return_units(open_scaled: int, close_scaled: int) -> int: numerator = (close_scaled - open_scaled) * 100 * RETURN_SCALE if numerator >= 0: return (numerator + open_scaled // 2) // open_scaled return -((-numerator + open_scaled // 2) // open_scaled) def _compare_return(left_open: int, left_close: int, right_open: int, right_close: int) -> int: left_cross = (left_close - left_open) * right_open right_cross = (right_close - right_open) * left_open if left_cross > right_cross: return 1 if left_cross < right_cross: return -1 return 0 def _safe_prefix(body): if body is None: return "" return body[:300].decode("utf-8", errors="replace").replace("\n", " ").replace("\r", " ") def _classify_exception(exc): text = (type(exc).__name__ + " " + str(exc)).lower() if "timeout" in text: return "TIMEOUT" if "tls" in text or "ssl" in text or "certificate" in text: return "TLS" if "dns" in text or "resolve" in text or "network" in text or "connection" in text: return "DNS/NETWORK" return "OTHER" def _classify_http(status): if status == 403: return "HTTP_403" if status == 429: return "HTTP_429" if status == 451: return "HTTP_451" if status is not None and 500 <= status <= 599: return "HTTP_5XX" return "HTTP_" + str(status) def _request(name: str, url: str): result = { "name": name, "url": url, "status_code": None, "body_length": None, "body_prefix": "", "okx_code": "", "okx_message": "", "exception_type": "", "exception_summary": "", "classification": "OTHER", } try: response = gl.nondet.web.get(url, headers={"Accept": "application/json"}) result["status_code"] = response.status if response.body is None: return result, None result["body_length"] = len(response.body) result["body_prefix"] = _safe_prefix(response.body) if response.status != 200: result["classification"] = _classify_http(response.status) return result, None if len(response.body) > MAX_RESPONSE_BYTES: result["classification"] = "VALID_HTTP_INVALID_DATA" return result, None try: payload = json.loads(response.body.decode("utf-8")) except (ValueError, UnicodeError): result["classification"] = "INVALID_JSON" return result, None if isinstance(payload, dict): result["okx_code"] = str(payload.get("code", "")) result["okx_message"] = str(payload.get("msg", ""))[:200] return result, payload except Exception as exc: result["exception_type"] = type(exc).__name__ result["exception_summary"] = str(exc)[:200].replace("\n", " ").replace("\r", " ") result["classification"] = _classify_exception(exc) return result, None def _instrument(asset: str): inst_id = asset + "-USDT-SWAP" url = "https://www.okx.com/api/v5/public/instruments?instType=SWAP&instId=" + inst_id result, payload = _request("INSTRUMENT_" + asset, url) result["instrument"] = inst_id result["parser"] = "FAIL" if result["status_code"] != 200 or not isinstance(payload, dict) or result["okx_code"] != "0": if result["status_code"] == 200 and result["okx_code"] == "0": result["classification"] = "INSTRUMENT_MISSING" return result, inst_id rows = payload.get("data") if not isinstance(rows, list) or len(rows) != 1 or not isinstance(rows[0], dict): result["classification"] = "INSTRUMENT_MISSING" return result, inst_id row = rows[0] if row.get("instId") != inst_id or row.get("instType") != "SWAP" or row.get("state") != "live" or row.get("settleCcy") != "USDT": result["classification"] = "INSTRUMENT_MISSING" return result, inst_id result["parser"] = "PASS" result["state"] = row.get("state", "") result["settle_ccy"] = row.get("settleCcy", "") result["classification"] = "VALID" return result, inst_id def _candle(asset: str, start: u256, end: u256): inst_id = asset + "-USDT-SWAP" start_ms = start * 1000 end_ms = end * 1000 url = "https://www.okx.com/api/v5/market/candles?instId=" + inst_id + "&bar=1H&after=" + str(end_ms) + "&limit=100" result, payload = _request("CANDLE_" + asset, url) result["instrument"] = inst_id result["parser"] = "FAIL" result["timestamp"] = None result["open"] = "" result["close"] = "" result["confirm"] = "" result["return_units"] = 0 if result["status_code"] != 200 or not isinstance(payload, dict) or result["okx_code"] != "0": return result rows = payload.get("data") if not isinstance(rows, list) or len(rows) > MAX_ROWS: result["classification"] = "INVALID_CANDLE" return result match = None for row in rows: if isinstance(row, list) and len(row) == 9 and _parse_integer(row[0]) == start_ms: match = row break if match is None or match[8] != "1": result["classification"] = "INVALID_CANDLE" return result opening = _parse_price(match[1]) closing = _parse_price(match[4]) high = _parse_price(match[2]) low = _parse_price(match[3]) if opening is None or closing is None or high is None or low is None: result["classification"] = "INVALID_CANDLE" return result result["parser"] = "PASS" result["classification"] = "VALID" result["timestamp"] = match[0] result["open"] = match[1] result["close"] = match[4] result["confirm"] = match[8] result["return_units"] = _return_units(opening, closing) return result def _asset_result(asset: str, start: u256, end: u256) -> dict: instrument, inst_id = _instrument(asset) if instrument["parser"] != "PASS": return {"asset": asset, "instrument": inst_id, "instrument_result": instrument, "candle_result": None} candle = _candle(asset, start, end) return {"asset": asset, "instrument": inst_id, "instrument_result": instrument, "candle_result": candle} def _category_result(category: str, start: u256, end: u256) -> dict: rows = [_asset_result(asset, start, end) for asset in ASSETS[category]] winner = "" winner_row = None complete = True for row in rows: candle = row["candle_result"] if candle is None or candle["parser"] != "PASS": complete = False continue if winner_row is None: winner = row["asset"] winner_row = candle continue comparison = _compare_return( _parse_price(candle["open"]), _parse_price(candle["close"]), _parse_price(winner_row["open"]), _parse_price(winner_row["close"]), ) if comparison > 0: winner = row["asset"] winner_row = candle return {"category": category, "complete": complete, "winner": winner if complete else "", "assets": rows} def _probe_spy(start: u256, end: u256) -> str: if end != start + 3600 or start % 3600 != 0: raise gl.vm.UserError("window must be UTC-aligned and one hour") return json.dumps({"start": start, "end": end, "result": _asset_result("SPY", start, end)}, separators=(",", ":"), sort_keys=True) def _probe_all(start: u256, end: u256) -> str: if end != start + 3600 or start % 3600 != 0: raise gl.vm.UserError("window must be UTC-aligned and one hour") categories = [_category_result(category, start, end) for category in ("US_INDICES", "ASIA_INDICES", "SECTOR_INDICES")] return json.dumps({"start": start, "end": end, "categories": categories}, separators=(",", ":"), sort_keys=True) def _compact_asset(asset: str, start: u256, end: u256) -> dict: instrument, inst_id = _instrument(asset) candle = None if instrument["parser"] == "PASS": full = _candle(asset, start, end) candle = { "status": full["status_code"], "code": full["okx_code"], "classification": full["classification"], "parser": full["parser"], "timestamp": full["timestamp"], "open": full["open"], "close": full["close"], "confirm": full["confirm"], "return_units": full["return_units"], } return { "asset": asset, "instrument": inst_id, "instrument_status": instrument["status_code"], "instrument_code": instrument["okx_code"], "instrument_classification": instrument["classification"], "instrument_parser": instrument["parser"], "state": instrument.get("state", ""), "settle_ccy": instrument.get("settle_ccy", ""), "candle": candle, } def probe_summary(start: u256, end: u256) -> str: if end != start + 3600 or start % 3600 != 0: raise gl.vm.UserError("window must be UTC-aligned and one hour") categories = [] for category in ("US_INDICES", "ASIA_INDICES", "SECTOR_INDICES"): assets = [_compact_asset(asset, start, end) for asset in ASSETS[category]] winner = "" winner_row = None complete = True for row in assets: candle = row["candle"] if candle is None or candle["parser"] != "PASS": complete = False continue if winner_row is None: winner = row["asset"] winner_row = candle continue left_open = _parse_price(candle["open"]) left_close = _parse_price(candle["close"]) right_open = _parse_price(winner_row["open"]) right_close = _parse_price(winner_row["close"]) if _compare_return(left_open, left_close, right_open, right_close) > 0: winner = row["asset"] winner_row = candle categories.append({"category": category, "complete": complete, "winner": winner if complete else "", "assets": assets}) return json.dumps({"start": start, "end": end, "categories": categories}, separators=(",", ":"), sort_keys=True) def _stable_spy(result: str): payload = json.loads(result)["result"] instrument = payload["instrument_result"] candle = payload["candle_result"] return ( instrument["status_code"], instrument["okx_code"], instrument["classification"], instrument["parser"], candle["status_code"] if candle else None, candle["okx_code"] if candle else "", candle["classification"] if candle else "", candle["parser"] if candle else "", candle["timestamp"] if candle else None, candle["open"] if candle else "", candle["close"] if candle else "", candle["confirm"] if candle else "", ) def _stable_all(result: str): payload = json.loads(result) stable = [] for category in payload["categories"]: for row in category["assets"]: instrument = row["instrument_result"] candle = row["candle_result"] stable.append(( category["category"], row["asset"], instrument["status_code"], instrument["okx_code"], instrument["classification"], instrument["parser"], candle["status_code"] if candle else None, candle["okx_code"] if candle else "", candle["classification"] if candle else "", candle["parser"] if candle else "", candle["timestamp"] if candle else None, candle["open"] if candle else "", candle["close"] if candle else "", candle["confirm"] if candle else "", )) return stable class OutrunOkxQualificationProbe(gl.contract.Contract): def __init__(self): pass @gl.public.write def probe_spy(self, start: u256, end: u256) -> str: def leader_fn(): return _probe_spy(start, end) def validator_fn(leaders_result): if not isinstance(leaders_result, gl.vm.Return): return False try: return _stable_spy(leaders_result.calldata) == _stable_spy(_probe_spy(start, end)) except Exception: return False return gl.vm.run_nondet(leader_fn, validator_fn) @gl.public.write def probe_all(self, start: u256, end: u256) -> str: def leader_fn(): return _probe_all(start, end) def validator_fn(leaders_result): if not isinstance(leaders_result, gl.vm.Return): return False try: return _stable_all(leaders_result.calldata) == _stable_all(_probe_all(start, end)) except Exception: return False return gl.vm.run_nondet(leader_fn, validator_fn)