# v0.2.16 # { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } import json from genlayer import * @gl.evm.contract_interface class _Recipient: class View: pass class Write: pass # Insurable stablecoins -> CoinGecko ids (also used as DefiLlama keys "coingecko:") COINS = {"usdc": "usd-coin", "usdt": "tether", "dai": "dai"} # Premium = ceil(payout / multiple). Riskier coin = lower multiple = higher premium. BAND_MULTIPLE = {"LOW": 15, "MODERATE": 8, "HIGH": 3} MIN_PREMIUM = 1000 # wei, avoids dust premiums POOL_CAP_DIVISOR = 5 # one policy can never exceed 20% of the pool # Insured condition limits. Prices travel as integers: 970000 means $0.970000. MIN_THRESHOLD_MICRO = 500000 MAX_THRESHOLD_MICRO = 990000 MIN_SUSTAIN_MINUTES = 60 MAX_SUSTAIN_MINUTES = 7 * 1440 MIN_SAMPLES_TO_PRICE = 500 # 90 days of hourly prices is about 2160 MIN_SAMPLES_TO_SETTLE = 6 # Timing rules, enforced here in the contract (not only in the frontend). MIN_LEAD_SECONDS = 1800 # coverage must start at least 30 minutes after the quote request QUOTE_TTL_SECONDS = 3600 # a quote can be bought for 60 minutes after it is issued MIN_WINDOW_SECONDS = 6 * 3600 # shortest coverage window: 6 hours (enough price samples) MAX_WINDOW_SECONDS = 14 * 86400 # longest coverage window: 14 days def _clean(text: str) -> str: return text.replace("```json", "").replace("```", "").strip() # ----------------------------------------------------------------- time _DIGITS = "0123456789" _TIME_FORMAT = "time must be canonical UTC: YYYY-MM-DDTHH:MM:SSZ" def _num(s: str, a: int, b: int) -> int: part = s[a:b] if len(part) != b - a or any(c not in _DIGITS for c in part): raise gl.vm.UserError(_TIME_FORMAT) return int(part) def _days_in_month(y: int, m: int) -> int: if m == 2: return 29 if (y % 4 == 0 and (y % 100 != 0 or y % 400 == 0)) else 28 return 30 if m in (4, 6, 9, 11) else 31 def _to_epoch(s: str) -> int: """Strict canonical UTC only (YYYY-MM-DDTHH:MM:SSZ) to unix seconds. Pure integer math: no datetime module, no offsets, no fractions, no local time.""" if ( len(s) != 20 or s[4] != "-" or s[7] != "-" or s[10] != "T" or s[13] != ":" or s[16] != ":" or s[19] != "Z" ): raise gl.vm.UserError(_TIME_FORMAT) y, mo, d = _num(s, 0, 4), _num(s, 5, 7), _num(s, 8, 10) h, mi, se = _num(s, 11, 13), _num(s, 14, 16), _num(s, 17, 19) if y < 1970 or y > 2200 or mo < 1 or mo > 12: raise gl.vm.UserError("not a real UTC time") if d < 1 or d > _days_in_month(y, mo) or h > 23 or mi > 59 or se > 59: raise gl.vm.UserError("not a real UTC time") yy = y - 1 if mo <= 2 else y era = yy // 400 yoe = yy - era * 400 mp = mo + 9 if mo <= 2 else mo - 3 doy = (153 * mp + 2) // 5 + d - 1 doe = yoe * 365 + yoe // 4 - yoe // 100 + doy days = era * 146097 + doe - 719468 return days * 86400 + h * 3600 + mi * 60 + se def _now() -> int: """Trusted current time: the transaction timestamp, which is part of the transaction itself and therefore identical on every validator. It is never taken from the caller, a web page or the model. Runtimes expose it differently, so three routes are tried: gl.message_raw (older), gl.message.raw (newer), then the Python clock, which GenVM pins to the transaction timestamp.""" try: raw = str(gl.message_raw["datetime"]) except (AttributeError, KeyError, TypeError): try: raw = str(gl.message.raw["datetime"]) except (AttributeError, KeyError, TypeError): import datetime raw = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%S") return _to_epoch(raw[:19] + "Z") # ------------------------------------------------------------- price evidence # Plain functions, so the arithmetic on fetched prices is the same on every validator and # can be tested without a network. The model only reconciles the results (see check_claim). def _json_body(raw: str): a, b = raw.find("{"), raw.rfind("}") return json.loads(raw[a : b + 1]) def _points_coingecko(raw: str) -> list: """CoinGecko market_chart JSON -> sorted [(unix_seconds, usd_price)].""" pts = [(int(t) // 1000, float(p)) for t, p in _json_body(raw)["prices"]] pts.sort() return pts def _points_llama(raw: str, key: str) -> list: """DefiLlama coins chart JSON -> sorted [(unix_seconds, usd_price)].""" rows = _json_body(raw)["coins"][key]["prices"] pts = [(int(r["timestamp"]), float(r["price"])) for r in rows] pts.sort() return pts def _longest_below(points: list, threshold: float, start_ts: int, end_ts: int) -> dict: """Longest run of consecutive samples at or below `threshold` inside [start_ts, end_ts]. A run is measured from its first low sample to the next sample (or the window end), because the price is only known at sample times.""" inside = [(t, p) for t, p in points if start_ts <= t <= end_ts] best = 0 run_start = None for i, (t, p) in enumerate(inside): if p <= threshold: if run_start is None: run_start = t nxt = inside[i + 1][0] if i + 1 < len(inside) else end_ts best = max(best, (nxt - run_start) // 60) else: run_start = None return { "samples": len(inside), "min_price": min(p for _, p in inside) if inside else None, "minutes_below": int(best), } def _history_stats(points: list, threshold: float) -> dict: dips = 0 below = 0 prev = False for _, p in points: cur = p <= threshold below += 1 if cur else 0 if cur and not prev: dips += 1 prev = cur return { "samples": len(points), "span_days": (points[-1][0] - points[0][0]) // 86400 if points else 0, "min_price": min(p for _, p in points) if points else None, "samples_below": below, "dips_below": dips, } class PegShield(gl.Contract): admin: Address pool_balance: u256 outstanding_liability: u256 quote_count: u256 policy_count: u256 quotes: TreeMap[str, str] # id -> JSON text policies: TreeMap[str, str] # id -> JSON text def __init__(self): self.admin = gl.message.sender_address self.pool_balance = u256(0) self.outstanding_liability = u256(0) self.quote_count = u256(0) self.policy_count = u256(0) # ------------------------------------------------------------------ pool @gl.public.write.payable def fund_pool(self) -> None: v = gl.message.value if v == u256(0): raise gl.vm.UserError("send some value") self.pool_balance = u256(int(self.pool_balance) + int(v)) # ----------------------------------------------------------------- quote @gl.public.write def request_quote( self, coin: str, threshold_micro: int, sustain_minutes: int, coverage_start: str, coverage_end: str, requested_payout: int, ) -> None: if coin not in COINS: raise gl.vm.UserError("unknown coin") if threshold_micro < MIN_THRESHOLD_MICRO or threshold_micro > MAX_THRESHOLD_MICRO: raise gl.vm.UserError("depeg price must be between $0.50 and $0.99") if sustain_minutes < MIN_SUSTAIN_MINUTES or sustain_minutes > MAX_SUSTAIN_MINUTES: raise gl.vm.UserError("minutes below must be between 60 and 10080") if requested_payout <= 0: raise gl.vm.UserError("payout must be positive") start_ts = _to_epoch(coverage_start) # rejects anything not canonical UTC end_ts = _to_epoch(coverage_end) now = _now() if start_ts < now + MIN_LEAD_SECONDS: raise gl.vm.UserError("coverage must start at least 30 minutes in the future") if end_ts - start_ts < MIN_WINDOW_SECONDS or end_ts - start_ts > MAX_WINDOW_SECONDS: raise gl.vm.UserError("coverage window must be between 6 hours and 14 days") coingecko_id = COINS[coin] threshold = threshold_micro / 1000000 url = ( "https://api.coingecko.com/api/v3/coins/" + coingecko_id + "/market_chart?vs_currency=usd&days=90" ) def price() -> str: try: stats = _history_stats(_points_coingecko(gl.nondet.web.render(url, mode="text")), threshold) except Exception: stats = None task = f""" You are an underwriter for stablecoin depeg insurance. Coin: {coin.upper()}. Insured condition: the USD price is at or below ${threshold:.4f} for at least {sustain_minutes} minutes, at any time in the coverage window. Statistics from the last 90 days of hourly prices (null means the data was unavailable): {json.dumps(stats)} Choose a risk band: - "LOW": the price almost never reached the insured level - "MODERATE": it reached it a few times - "HIGH": it reached it often, or the lowest price is far below the level - "UNPRICEABLE": statistics are null or have fewer than {MIN_SAMPLES_TO_PRICE} samples Respond ONLY with this JSON, nothing else, no formatting prefix or suffix: {{ "risk_band": str, "rationale": str }} """ try: out = json.loads(_clean(gl.nondet.exec_prompt(task))) band = str(out.get("risk_band", "UNPRICEABLE")).upper() rationale = str(out.get("rationale", ""))[:600] except Exception: band, rationale = "UNPRICEABLE", "The model's answer could not be read." return json.dumps({"risk_band": band, "rationale": rationale, "stats": stats}) result = gl.eq_principle.prompt_comparative(price, "The value of risk_band has to match") data = json.loads(result) band = str(data.get("risk_band", "UNPRICEABLE")).upper() stats = data.get("stats") or {} if band not in BAND_MULTIPLE or int(stats.get("samples", 0)) < MIN_SAMPLES_TO_PRICE: band = "UNPRICEABLE" required_premium = 0 if band != "UNPRICEABLE": m = BAND_MULTIPLE[band] required_premium = -(-requested_payout // m) # ceiling division if required_premium < MIN_PREMIUM: required_premium = MIN_PREMIUM n = int(self.quote_count) + 1 self.quote_count = u256(n) qid = "PGQ-" + str(n) self.quotes[qid] = json.dumps( { "id": qid, "requester": gl.message.sender_address.as_hex, "coin": coin, "threshold_micro": threshold_micro, "sustain_minutes": sustain_minutes, "coverage_start": coverage_start, "coverage_end": coverage_end, "start_ts": start_ts, "end_ts": end_ts, "quoted_at": now, "valid_until": now + QUOTE_TTL_SECONDS, "requested_payout": requested_payout, "risk_band": band, "required_premium": required_premium, "history_samples": int(stats.get("samples", 0)), "history_dips": int(stats.get("dips_below", 0)), "history_min_price": stats.get("min_price"), "rationale": str(data.get("rationale", ""))[:600], "consumed": False, } ) # ------------------------------------------------------------------- buy @gl.public.write.payable def buy_policy_from_quote(self, quote_id: str) -> None: if quote_id not in self.quotes: raise gl.vm.UserError("quote not found") q = json.loads(self.quotes[quote_id]) if q["consumed"]: raise gl.vm.UserError("quote already used") if q["risk_band"] == "UNPRICEABLE": raise gl.vm.UserError("quote is unpriceable") if gl.message.sender_address.as_hex != q["requester"]: raise gl.vm.UserError("buyer must be the quote requester") now = _now() if now >= int(q["start_ts"]): raise gl.vm.UserError("coverage has already started or ended") if now > int(q["valid_until"]): raise gl.vm.UserError("quote expired") if int(gl.message.value) != int(q["required_premium"]): raise gl.vm.UserError("send exactly the required premium") payout = int(q["requested_payout"]) pool_after = int(self.pool_balance) + int(gl.message.value) if payout > pool_after // POOL_CAP_DIVISOR: raise gl.vm.UserError("payout exceeds 20% of pool; fund the pool first") if int(self.outstanding_liability) + payout > pool_after: raise gl.vm.UserError("pool cannot back this policy") self.pool_balance = u256(pool_after) self.outstanding_liability = u256(int(self.outstanding_liability) + payout) q["consumed"] = True self.quotes[quote_id] = json.dumps(q) n = int(self.policy_count) + 1 self.policy_count = u256(n) pid = "PGN-" + str(n) self.policies[pid] = json.dumps( { "id": pid, "quote_id": quote_id, "holder": gl.message.sender_address.as_hex, "coin": q["coin"], "threshold_micro": q["threshold_micro"], "sustain_minutes": q["sustain_minutes"], "coverage_start": q["coverage_start"], "coverage_end": q["coverage_end"], "start_ts": q["start_ts"], "end_ts": q["end_ts"], "premium": q["required_premium"], "payout_amount": payout, "status": "ACTIVE", "resolved_minutes_below": 0, "evidence_note": "", } ) # ----------------------------------------------------------------- claim @gl.public.write def check_claim(self, policy_id: str) -> None: if policy_id not in self.policies: raise gl.vm.UserError("policy not found") p = json.loads(self.policies[policy_id]) if p["status"] != "ACTIVE": raise gl.vm.UserError("policy is not active") # Settlement is only possible once the insured window has closed. This is # checked against the trusted transaction time before any evidence is fetched. if _now() < int(p["end_ts"]): raise gl.vm.UserError("coverage window has not ended") coin = p["coin"] coingecko_id = COINS[coin] threshold = int(p["threshold_micro"]) / 1000000 sustain = int(p["sustain_minutes"]) start_ts = int(p["start_ts"]) end_ts = int(p["end_ts"]) start = p["coverage_start"] end = p["coverage_end"] url_a = ( "https://api.coingecko.com/api/v3/coins/" + coingecko_id + "/market_chart/range?vs_currency=usd&from=" + str(start_ts) + "&to=" + str(end_ts) ) span = (end_ts - start_ts) // 3600 + 1 key_b = "coingecko:" + coingecko_id url_b = ( "https://coins.llama.fi/chart/" + key_b + "?start=" + str(start_ts) + "&span=" + str(span) + "&period=1h" ) def judge() -> str: try: a = _longest_below(_points_coingecko(gl.nondet.web.render(url_a, mode="text")), threshold, start_ts, end_ts) except Exception: a = None try: b = _longest_below(_points_llama(gl.nondet.web.render(url_b, mode="text"), key_b), threshold, start_ts, end_ts) except Exception: b = None task = f""" You settle stablecoin depeg insurance claims. Coin: {coin.upper()}. Insured level: USD price at or below ${threshold:.4f}. Required duration: {sustain} minutes. Coverage window (UTC): {start} to {end}. The window has already ended. The contract already measured each public price source over the window. Each result gives the number of price samples inside the window, the lowest price seen, and minutes_below: the longest run of samples at or below the insured level, measured from its first low sample to the next sample. null means the source was unavailable. SOURCE A (CoinGecko): {json.dumps(a)} SOURCE B (DefiLlama): {json.dumps(b)} Rules: 1. If both are available and their minutes_below are within 30 minutes of each other, use the LOWER one. 2. If they differ by more, use the source with more samples, unless its lowest price is implausible (below $0.10 or above $5.00). Explain the choice in evidence_note. 3. If only one is available and it has at least {MIN_SAMPLES_TO_SETTLE} samples, use it. 4. If neither is usable, the verdict is INSUFFICIENT_EVIDENCE. Respond ONLY with this JSON, nothing else, no formatting prefix or suffix: {{ "resolved_minutes_below": int, "verdict": str, "evidence_note": str }} verdict is one of BELOW_THRESHOLD, NOT_BELOW, INSUFFICIENT_EVIDENCE. """ return _clean(gl.nondet.exec_prompt(task)) result = gl.eq_principle.prompt_comparative( judge, "resolved_minutes_below must be within 30 minutes of each other", ) data = json.loads(result) verdict = str(data.get("verdict", "INSUFFICIENT_EVIDENCE")) note = str(data.get("evidence_note", ""))[:600] p["evidence_note"] = note if verdict == "INSUFFICIENT_EVIDENCE": self.policies[policy_id] = json.dumps(p) return minutes = int(data.get("resolved_minutes_below", 0)) p["resolved_minutes_below"] = minutes payout = int(p["payout_amount"]) self.outstanding_liability = u256(int(self.outstanding_liability) - payout) # Only this arithmetic moves money, never the model's wording. if minutes >= sustain: pay = min(payout, int(self.pool_balance)) self.pool_balance = u256(int(self.pool_balance) - pay) p["status"] = "PAID_OUT" self.policies[policy_id] = json.dumps(p) if pay > 0: _Recipient(Address(p["holder"])).emit_transfer(value=u256(pay)) else: p["status"] = "DECLINED" self.policies[policy_id] = json.dumps(p) # ----------------------------------------------------------------- views @gl.public.view def get_summary(self) -> dict[str, int]: return { "pool_balance": int(self.pool_balance), "outstanding_liability": int(self.outstanding_liability), "quote_count": int(self.quote_count), "policy_count": int(self.policy_count), } @gl.public.view def get_quote(self, quote_id: str) -> str: return self.quotes.get(quote_id, "") @gl.public.view def get_policy(self, policy_id: str) -> str: return self.policies.get(policy_id, "") @gl.public.view def get_coins(self) -> list[str]: return list(COINS.keys())