# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from genlayer import * import json class CryptoChatbot(gl.Contract): zerion_base_url: str last_user_address: str last_prompt: str last_analysis: str last_payload_json: str def __init__(self): # Replace this after deployment by calling set_zerion_base_url(). # It should point to your Vercel proxy, not directly to Zerion, so # ZERION_API_KEY stays server-side. self.zerion_base_url = "https://YOUR_VERCEL_DOMAIN.vercel.app/api/genlayer/zerion" self.last_user_address = "" self.last_prompt = "" self.last_analysis = "" self.last_payload_json = "{}" @gl.public.write def set_zerion_base_url(self, zerion_base_url: str) -> None: if not zerion_base_url.startswith("https://"): raise gl.vm.UserError("Zerion proxy URL must start with https://") self.zerion_base_url = zerion_base_url.rstrip("/") @gl.public.write def analyze_portfolio(self, user_address: str, user_prompt: str) -> str: if len(user_address) < 32: raise gl.vm.UserError("Invalid wallet address") if len(user_prompt.strip()) < 3: raise gl.vm.UserError("Prompt is too short") zerion_base_url = self.zerion_base_url if "YOUR_VERCEL_DOMAIN" in zerion_base_url: raise gl.vm.UserError("Call set_zerion_base_url() before analysis") def number(value): if isinstance(value, int) or isinstance(value, float): return float(value) return 0.0 def text(value, fallback): if isinstance(value, str) and len(value.strip()) > 0: return value.strip() return fallback def top_key(values): best_key = "unknown" best_value = -1.0 if not isinstance(values, dict): return best_key for key, value in values.items(): numeric = number(value) if numeric > best_value: best_key = str(key) best_value = numeric return best_key def normalize_position(item): attributes = item.get("attributes", {}) relationships = item.get("relationships", {}) fungible = attributes.get("fungible_info", {}) quantity = attributes.get("quantity", {}) app_metadata = attributes.get("application_metadata", {}) chain = relationships.get("chain", {}).get("data", {}).get("id", "unknown") return { "symbol": text(fungible.get("symbol"), text(attributes.get("name"), "TOKEN")), "name": text(fungible.get("name"), text(attributes.get("name"), "Unknown token")), "chain": text(chain, "unknown"), "value_usd": round(number(attributes.get("value")), 2), "quantity": number(quantity.get("float")), "type": text(attributes.get("position_type"), "wallet"), "protocol": text(app_metadata.get("name"), ""), } def build_snapshot(portfolio_json, positions_json): attrs = portfolio_json.get("data", {}).get("attributes", {}) total = attrs.get("total", {}) changes = attrs.get("changes", {}) by_chain = attrs.get("positions_distribution_by_chain", {}) by_type = attrs.get("positions_distribution_by_type", {}) positions = [] for item in positions_json.get("data", [])[:8]: if isinstance(item, dict): positions.append(normalize_position(item)) return { "address": user_address, "total_usd": round(number(total.get("positions")), 2), "day_change_percent": number(changes.get("percent_1d")), "top_chain": top_key(by_chain), "by_chain": by_chain if isinstance(by_chain, dict) else {}, "by_type": by_type if isinstance(by_type, dict) else {}, "top_positions": positions, } def valid_answer(answer): if not isinstance(answer, dict): return False intent = answer.get("intent") analysis = answer.get("analysis") risk_score = answer.get("risk_score") actions = answer.get("actions") warnings = answer.get("warnings") valid_intents = [ "portfolio_summary", "risk_review", "yield_scan", "trade_planning", "bridge_planning", "market_question", "other", ] return ( intent in valid_intents and isinstance(analysis, str) and 12 <= len(analysis) <= 1400 and isinstance(risk_score, int) and 1 <= risk_score <= 10 and isinstance(actions, list) and 2 <= len(actions) <= 5 and all(isinstance(action, str) and 3 <= len(action) <= 180 for action in actions) and isinstance(warnings, list) and 1 <= len(warnings) <= 4 and all(isinstance(warning, str) and 3 <= len(warning) <= 180 for warning in warnings) ) def valid_payload(payload): if not isinstance(payload, dict): return False portfolio = payload.get("portfolio") answer = payload.get("answer") return ( isinstance(portfolio, dict) and isinstance(portfolio.get("total_usd"), (int, float)) and isinstance(portfolio.get("top_chain"), str) and isinstance(portfolio.get("top_positions"), list) and valid_answer(answer) ) def totals_close(a, b): if a == 0 and b == 0: return True largest = max(abs(a), abs(b), 1.0) return abs(a - b) / largest <= 0.03 def answers_equivalent(a, b): return ( a.get("intent") == b.get("intent") and abs(a.get("risk_score") - b.get("risk_score")) <= 2 and len(a.get("actions")) >= 2 and len(b.get("actions")) >= 2 ) def leader_fn(): portfolio_url = f"{zerion_base_url}/wallets/{user_address}/portfolio" positions_url = f"{zerion_base_url}/wallets/{user_address}/positions" portfolio_raw = gl.nondet.web.get(portfolio_url).body.decode("utf-8") positions_raw = gl.nondet.web.get(positions_url).body.decode("utf-8") snapshot = build_snapshot( json.loads(portfolio_raw), json.loads(positions_raw), ) prompt = f""" You are Elsa, a crypto-native AI copilot. Use this Zerion wallet snapshot as ground truth and answer the user. Return strict JSON with exactly these keys: - intent: one of portfolio_summary, risk_review, yield_scan, trade_planning, bridge_planning, market_question, other - analysis: concise useful answer - risk_score: integer 1 to 10 - actions: array of 2 to 5 concrete next steps - warnings: array of 1 to 4 safety warnings Rules: - Do not claim a transaction was executed. - Do not guarantee future prices or yield. - If the user asks to trade, prepare the trade checklist and risks. - Mention concentration, chain, protocol, liquidity, bridge, and approval risk when relevant. User request: {user_prompt} Zerion wallet snapshot: {json.dumps(snapshot, sort_keys=True)} """ answer = gl.nondet.exec_prompt(prompt, response_format="json") return { "portfolio": snapshot, "answer": answer, } def validator_fn(leader_result): if not isinstance(leader_result, gl.vm.Return): return False leader_payload = leader_result.calldata if not valid_payload(leader_payload): return False validator_payload = leader_fn() if not valid_payload(validator_payload): return False return ( totals_close( leader_payload["portfolio"]["total_usd"], validator_payload["portfolio"]["total_usd"], ) and leader_payload["portfolio"]["top_chain"] == validator_payload["portfolio"]["top_chain"] and answers_equivalent( leader_payload["answer"], validator_payload["answer"], ) ) result = gl.vm.run_nondet_unsafe(leader_fn, validator_fn) self.last_user_address = user_address self.last_prompt = user_prompt self.last_analysis = result["answer"]["analysis"] self.last_payload_json = json.dumps(result, sort_keys=True) return self.last_analysis @gl.public.view def get_last_analysis(self) -> str: return self.last_analysis @gl.public.view def get_last_payload_json(self) -> str: return self.last_payload_json