AI Agent 代币经济参与指南
⚠️ No-Go Disclaimer: MSGChain 主网裁决为 No-Go。本文件所有内容反映的是开发阶段的技术设计，不代表主网未独立核验上线状态。生产部署状态请以白皮书为准：https://msgchain.org/whitepaper/
目录
1. 概述
1.1 为什么 Agent 需要参与代币经济
AI Agent 在 MSG Chain 上不仅执行链上交互，更需要深度参与代币经济体系以获得自主经济能力。通过质押、流动性提供、治理投票等操作，Agent 可以实现：
- 自我维持：通过质押收益覆盖 Gas 费用和运营成本
- 链上影响力：通过治理投票参与协议方向决策
- 经济自主权：发行自有代币建立独立经济生态
- 收益优化：多策略资产配置实现复利增长
1.2 MSG Chain 经济基础
| 参数 | 值 |
|---|---|
| chain-id | msg-chain-1 |
| Bech32 前缀 | msg |
| 原生代币 | umsg (1 MSG = 1,000,000 umsg) |
| 共识机制 | Tendermint PoS |
| 验证人数量上限 | 150 |
| 区块时间 | ~6 秒 |
| Unbonding 周期 | 21 天 |
| 治理投票周期 | 7 天 |
1.3 Agent 的经济角色
经济角色矩阵:
+------------------------------------------------------------------+
| Agent 经济参与层次 |
+---------------+------------------+-------------------------------+
| 基础层 | 增值层 | 高级层 |
| --------- | --------- | --------- |
| MSG 质押 | 流动性提供 | Agent 代币发行 |
| 委托验证人 | LP Farming | 债券曲线销售 |
| 收益复投 | 集中度管理 | 代币经济设计 |
| 投票治理 | 无常损失对冲 | 跨链资产桥接 |
+------------------------------------------------------------------+
1.4 MSG 代币效用全景
MSG Token Utility:
+-- 网络费用: Gas、合约部署、IBC 转账
+-- 质押安全: 委托给验证人维护网络安全
+-- 治理投票: 每个提案需最低 1 MSG 初始存款
+-- 流动性挖矿: DEX 池配对提供流动性
+-- 跨链费用: IBC 中继和 Ack 费用
+-- 协议收入: 交易手续费、MEV 分配
1.5 前提条件
Agent 要参与代币经济需具备：
# 最小依赖集合
AGENT_PREREQUISITES = {
"chain_id": "msg-chain-1",
"rpc_endpoint": "https://rpc.msg-chain-1.msg.network",
"rest_endpoint": "https://api.msg-chain-1.msg.network",
"bech32_prefix": "msg",
"denom": "umsg",
"gas_prices": {"attoMSG": 1000000000},
"required_balance": 1_000_000,
}
2. MSG 质押与委托
2.1 质押机制概述
MSG Chain 采用 Tendermint PoS 共识机制。代币持有者通过将 MSG 委托给验证人参与网络安全，并获得质押奖励。
质押生命周期:
+---------+ +---------+ +---------+
| 委托 MSG |---> | 赚取奖励 |---> | 提取/复投 |
+---------+ +---------+ +---------+
验证人选择 奖励按 epoch 21天 unbonding
自动累积 周期(全额提取)
2.2 验证人选择策略
Agent 应根据链上指标自动筛选验证人：
import aiohttp
from dataclasses import dataclass
from typing import Optional
@dataclass
class ValidatorInfo:
"""验证人链上信息"""
operator_address: str
moniker: str
tokens: int
commission_rate: float
commission_max_rate: float
commission_max_change_rate: float
unbonding_tokens: int
jailed: bool
status: str
class ValidatorSelector:
"""
验证人智能选择器
基于多因子评分模型筛选最佳委托目标
"""
def __init__(self, rest_endpoint: str = "https://api.msg-chain-1.msg.network"):
self.rest_endpoint = rest_endpoint
self.session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def fetch_validators(self) -> list[ValidatorInfo]:
"""从 LCD API 获取所有验证人列表"""
url = f"{self.rest_endpoint}/cosmos/staking/v1beta1/validators"
params = {"status": "BOND_STATUS_BONDED"}
async with self.session.get(url, params=params) as resp:
data = await resp.json()
validators = []
for v in data.get("validators", []):
commission = v.get("commission", {}).get("commission_rates", {})
validators.append(ValidatorInfo(
operator_address=v["operator_address"],
moniker=v["description"]["moniker"],
tokens=int(v["tokens"]),
commission_rate=float(commission["rate"]),
commission_max_rate=float(commission["max_rate"]),
commission_max_change_rate=float(commission["max_change_rate"]),
unbonding_tokens=int(v.get("unbonding_tokens", 0)),
jailed=v.get("jailed", False),
status=v.get("status", ""),
))
return validators
def score_validator(self, v: ValidatorInfo) -> float:
"""
综合评分 (0-100):
- 佣金率: 越低越好 (权重 30%)
- 质押量排名: 避免中心化 (权重 25%)
- 节点稳定性: 历史最低 jailed (权重 25%)
- 佣金变化率: 越低越好 (权重 20%)
"""
commission_score = (1.0 - v.commission_rate) * 30
if v.jailed:
commission_score = 0.0
max_change_penalty = v.commission_max_change_rate * 20
if v.commission_max_rate > 1.0:
commission_score -= 15
base_score = commission_score - max_change_penalty
if v.status != "BOND_STATUS_BONDED":
base_score *= 0.5
return max(0.0, base_score)
async def select_top_validators(self, top_n: int = 5) -> list[ValidatorInfo]:
"""选出评分最高的 N 个验证人"""
validators = await self.fetch_validators()
scored = [(self.score_validator(v), v) for v in validators]
scored.sort(key=lambda x: x[0], reverse=True)
return [v for _, v in scored[:top_n]]
2.3 委托操作实现
核心的委托、解委托、重新委托操作：
from dataclasses import dataclass
from typing import Optional, Any
import hashlib
import json
@dataclass
class Coin:
amount: int
denom: str
@dataclass
class MsgDelegate:
delegator_address: str
validator_address: str
amount: Coin
@dataclass
class MsgUndelegate:
delegator_address: str
validator_address: str
amount: Coin
@dataclass
class MsgBeginRedelegate:
delegator_address: str
validator_src_address: str
validator_dst_address: str
amount: Coin
class CosmosClient:
"""简化的 Cosmos SDK 链上客户端"""
def __init__(self, rpc_endpoint: str, private_key: str, bech32_prefix: str = "msg"):
self.rpc_endpoint = rpc_endpoint
self.private_key = private_key
self.bech32_prefix = bech32_prefix
def get_address(self) -> str:
return f"msg1...{self.private_key[:8]}"
async def broadcast_tx(self, msg: Any, memo: str = "") -> dict:
tx_bytes = self._encode_tx(msg)
signature = self._sign(tx_bytes)
signed_tx = {
"tx": {"body": {"messages": [self._msg_to_dict(msg)], "memo": memo}},
"signatures": [signature],
}
return {"txhash": hashlib.sha256(json.dumps(signed_tx).encode()).hexdigest(), "code": 0}
def _msg_to_dict(self, msg: Any) -> dict:
if isinstance(msg, MsgDelegate):
return {
"@type": "/cosmos.staking.v1beta1.MsgDelegate",
"delegator_address": msg.delegator_address,
"validator_address": msg.validator_address,
"amount": {"denom": msg.amount.denom, "amount": str(msg.amount.amount)},
}
if isinstance(msg, MsgUndelegate):
return {
"@type": "/cosmos.staking.v1beta1.MsgUndelegate",
"delegator_address": msg.delegator_address,
"validator_address": msg.validator_address,
"amount": {"denom": msg.amount.denom, "amount": str(msg.amount.amount)},
}
if isinstance(msg, MsgBeginRedelegate):
return {
"@type": "/cosmos.staking.v1beta1.MsgBeginRedelegate",
"delegator_address": msg.delegator_address,
"validator_src_address": msg.validator_src_address,
"validator_dst_address": msg.validator_dst_address,
"amount": {"denom": msg.amount.denom, "amount": str(msg.amount.amount)},
}
return {}
def _encode_tx(self, msg: Any) -> bytes:
return json.dumps(self._msg_to_dict(msg)).encode()
def _sign(self, tx_bytes: bytes) -> str:
return "base64_encoded_signature=="
class AgentStaking:
"""Agent 质押与委托操作"""
GAS_LIMIT_DELEGATE = 200_000
GAS_LIMIT_UNDELEGATE = 250_000
GAS_LIMIT_REDELEGATE = 300_000
MIN_COMPOUND = 100_000
def __init__(
self,
agent_addr: str,
client: CosmosClient,
default_validator: str = "",
):
self.agent_addr = agent_addr
self.client = client
self.default_validator = default_validator
self._delegations: dict[str, int] = {}
async def delegate(self, validator_addr: str, amount: int) -> dict:
msg = MsgDelegate(
delegator_address=self.agent_addr,
validator_address=validator_addr,
amount=Coin(amount=amount, denom="umsg"),
)
result = await self.client.broadcast_tx(msg)
if result.get("code") == 0:
self._delegations[validator_addr] = (
self._delegations.get(validator_addr, 0) + amount
)
return result
async def undelegate(self, validator_addr: str, amount: int) -> dict:
msg = MsgUndelegate(
delegator_address=self.agent_addr,
validator_address=validator_addr,
amount=Coin(amount=amount, denom="umsg"),
)
result = await self.client.broadcast_tx(msg)
if result.get("code") == 0:
self._delegations[validator_addr] = max(
0, self._delegations.get(validator_addr, 0) - amount
)
return result
async def redelegate(
self,
from_validator: str,
to_validator: str,
amount: int,
) -> dict:
msg = MsgBeginRedelegate(
delegator_address=self.agent_addr,
validator_src_address=from_validator,
validator_dst_address=to_validator,
amount=Coin(amount=amount, denom="umsg"),
)
result = await self.client.broadcast_tx(msg)
if result.get("code") == 0:
self._delegations[from_validator] = max(
0, self._delegations.get(from_validator, 0) - amount
)
self._delegations[to_validator] = (
self._delegations.get(to_validator, 0) + amount
)
return result
async def query_delegations(self) -> list[dict]:
url = (
f"{self.client.rpc_endpoint.replace('rpc', 'api')}"
f"/cosmos/staking/v1beta1/delegations/{self.agent_addr}"
)
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
data = await resp.json()
return data.get("delegation_responses", [])
async def query_rewards(self) -> int:
url = (
f"{self.client.rpc_endpoint.replace('rpc', 'api')}"
f"/cosmos/distribution/v1beta1/delegators/{self.agent_addr}/rewards"
)
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
data = await resp.json()
total = 0
for reward in data.get("rewards", []):
for coin in reward.get("reward", []):
if coin["denom"] == "umsg":
total += int(coin["amount"])
return total
async def auto_compound(self) -> Optional[dict]:
rewards = await self.query_rewards()
if rewards < self.MIN_COMPOUND:
return None
validator = self.default_validator
if not validator:
async with ValidatorSelector() as selector:
top = await selector.select_top_validators(1)
if not top:
return None
validator = top[0].operator_address
return await self.delegate(validator, rewards)
2.4 奖励查询与提取
class RewardManager:
"""奖励提取和管理"""
GAS_LIMIT_WITHDRAW = 200_000
def __init__(self, agent_addr: str, client: CosmosClient):
self.agent_addr = agent_addr
self.client = client
async def withdraw_all_rewards(self) -> list[dict]:
delegations = await self._query_all_delegations()
results = []
for val_addr in delegations:
msg = {
"@type": "/cosmos.distribution.v1beta1.MsgWithdrawDelegatorReward",
"delegator_address": self.agent_addr,
"validator_address": val_addr,
}
tx_result = await self.client.broadcast_tx(msg)
results.append(tx_result)
return results
async def _query_all_delegations(self) -> list[str]:
url = (
f"{self.client.rpc_endpoint.replace('rpc', 'api')}"
f"/cosmos/staking/v1beta1/delegations/{self.agent_addr}"
)
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
data = await resp.json()
return [d["delegation"]["validator_address"] for d in data.get("delegation_responses", [])]
async def query_annual_apr(self) -> float:
url = f"{self.client.rpc_endpoint.replace('rpc', 'api')}/cosmos/mint/v1beta1/inflation"
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
inflation_data = await resp.json()
inflation = float(inflation_data.get("inflation", "0.13"))
pool_url = f"{self.client.rpc_endpoint.replace('rpc', 'api')}/cosmos/staking/v1beta1/pool"
async with aiohttp.ClientSession() as session:
async with session.get(pool_url) as resp:
pool_data = await resp.json()
bonded = float(pool_data["pool"]["bonded_tokens"])
not_bonded = float(pool_data["pool"]["not_bonded_tokens"])
total_supply = bonded + not_bonded
bonded_ratio = bonded / total_supply if total_supply > 0 else 1.0
apr = inflation / bonded_ratio if bonded_ratio > 0 else 0.0
commission_avg = 0.10
return apr * (1 - commission_avg)
2.5 重新委托策略
class RedelegationStrategy:
"""
重新委托策略引擎
场景:
1. 佣金率调整: 验证人提升佣金，Agent 转向低佣金节点
2. 安全迁移: 检测到验证人即将 jailed，提前迁移
3. 收益优化: 将委托集中到高收益验证人
"""
REDELEGATION_COOLDOWN = 14
def __init__(self, staking: AgentStaking, selector: ValidatorSelector):
self.staking = staking
self.selector = selector
self._redelegation_history: dict[str, int] = {}
async def evaluate_and_redelegate(self) -> list[dict]:
delegations = await self.staking.query_delegations()
validators = await self.selector.fetch_validators()
validator_map = {v.operator_address: v for v in validators}
results = []
for delegation in delegations:
val_addr = delegation["delegation"]["validator_address"]
amount = int(delegation["balance"]["amount"])
validator = validator_map.get(val_addr)
if not validator:
continue
score = self.selector.score_validator(validator)
if score < 50 and amount > 0:
last_redelegation = self._redelegation_history.get(val_addr, 0)
blocks_since = await self._get_current_block_height() - last_redelegation
if blocks_since < self.REDELEGATION_COOLDOWN * 10000:
continue
top = await self.selector.select_top_validators(1)
if top and top[0].operator_address != val_addr:
tx = await self.staking.redelegate(
from_validator=val_addr,
to_validator=top[0].operator_address,
amount=amount,
)
results.append(tx)
self._redelegation_history[val_addr] = await self._get_current_block_height()
return results
async def _get_current_block_height(self) -> int:
url = f"{self.staking.client.rpc_endpoint.replace('rpc', 'api')}/cosmos/base/tendermint/v1beta1/blocks/latest"
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
data = await resp.json()
return int(data["block"]["header"]["height"])
2.6 Unbonding 管理
class UnbondingManager:
"""
解委托队列管理
委托解绑需要 21 天，Agent 需要追踪 unbonding 队列并在到期后自动提取。
"""
UNBONDING_DAYS = 21
BLOCKS_PER_DAY = 14400
UNBONDING_BLOCKS = UNBONDING_DAYS * BLOCKS_PER_DAY
def __init__(self, agent_addr: str, client: CosmosClient):
self.agent_addr = agent_addr
self.client = client
async def query_unbonding_delegations(self) -> list[dict]:
url = (
f"{self.client.rpc_endpoint.replace('rpc', 'api')}"
f"/cosmos/staking/v1beta1/delegators/{self.agent_addr}/unbonding_delegations"
)
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
data = await resp.json()
return data.get("unbonding_responses", [])
async def monitor_unbonding_queue(self) -> list[dict]:
current_height = await self._get_current_height()
unbondings = await self.query_unbonding_delegations()
completed = []
for entry in unbondings:
for entry_inner in entry.get("entries", []):
completion_height = int(entry_inner.get("completion_height", 0))
blocks_remaining = completion_height - current_height
if blocks_remaining <= 0:
completed.append({
"validator": entry["validator_address"],
"amount": entry_inner["balance"],
"status": "completed",
})
elif blocks_remaining < 14400:
completed.append({
"validator": entry["validator_address"],
"amount": entry_inner["balance"],
"completion_height": completion_height,
"blocks_remaining": blocks_remaining,
"eta_hours": blocks_remaining / 600,
"status": "pending_soon",
})
return completed
async def _get_current_height(self) -> int:
url = f"{self.client.rpc_endpoint}/status"
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
data = await resp.json()
return int(data["result"]["sync_info"]["latest_block_height"])
2.7 质押事件监控
class StakingEventMonitor:
"""质押事件实时监控"""
def __init__(self, rpc_endpoint: str, websocket_endpoint: str):
self.rpc = rpc_endpoint
self.ws = websocket_endpoint
async def subscribe_staking_events(self, agent_addr: str):
import json
query = json.dumps({
"jsonrpc": "2.0",
"method": "subscribe",
"params": {"query": f"transfer.recipient='{agent_addr}'"},
"id": 1,
})
import asyncio
while True:
await asyncio.sleep(60)
2.8 质押数据缓存与统计
@dataclass
class StakingStats:
"""质押状态快照"""
total_delegated: int
total_rewards: int
active_validators: int
avg_commission: float
current_apr: float
compounding_count: int
last_compound_height: int
class StakingAnalytics:
"""质押数据缓存与分析"""
def __init__(self, staking: AgentStaking):
self.staking = staking
self.history: list[StakingStats] = []
self._cache: Optional[StakingStats] = None
self._cache_time = 0
async def get_stats(self) -> StakingStats:
from time import time
if self._cache and time() - self._cache_time < 60:
return self._cache
delegations = await self.staking.query_delegations()
rewards = await self.staking.query_rewards()
total_delegated = sum(int(d.get("balance", {}).get("amount", 0)) for d in delegations)
reward_mgr = RewardManager(self.staking.agent_addr, self.staking.client)
apr = await reward_mgr.query_annual_apr()
stats = StakingStats(
total_delegated=total_delegated,
total_rewards=rewards,
active_validators=len(delegations),
avg_commission=0.10,
current_apr=apr,
compounding_count=len(self.history),
last_compound_height=0,
)
self._cache = stats
self._cache_time = time()
self.history.append(stats)
return stats
def summary_report(self, stats: StakingStats) -> str:
return (
"===== 质押报告 =====\n"
f"总委托: {stats.total_delegated / 1_000_000:.2f} MSG\n"
f"当前 APR: {stats.current_apr * 100:.2f}%\n"
f"验证人数: {stats.active_validators}"
)
3. 流动性提供
3.1 DEX 流动性池机制
MSG Chain 支持 CosmWasm 智能合约实现的 DEX。流动性提供者向交易对池注入代币获取 LP Token，并获得交易手续费分成。
AMM 流动性池:
Agent A --MSG--> [ MSG / USDC 池 ] <--LP-- Agent B
MSG: 1,000,000
USDC: 500,000
LP Token: 1,500
Fee: 0.3%
k = MSG * USDC
3.2 流动性提供核心实现
from dataclasses import dataclass
from typing import Optional
import aiohttp
@dataclass
class AssetInfo:
"""CosmWasm 资产信息"""
native: Optional[str] = None
token: Optional[str] = None
@dataclass
class PoolInfo:
"""流动性池信息"""
pool_id: int
pool_addr: str
assets: list[dict]
total_lp_shares: int
swap_fee: float
class ExecuteMsg:
"""CosmWasm 合约执行消息"""
@staticmethod
def provide_liquidity(
assets: list[AssetInfo],
amounts: list[int],
slippage_tolerance: float = 0.01,
receiver: Optional[str] = None,
) -> dict:
msg = {
"provide_liquidity": {
"assets": [
{
"info": (
{"native_token": {"denom": a.native}}
if a.native
else {"token": {"contract_addr": a.token}}
),
"amount": str(amount),
}
for a, amount in zip(assets, amounts)
],
"slippage_tolerance": str(slippage_tolerance),
}
}
if receiver:
msg["provide_liquidity"]["receiver"] = receiver
return msg
@staticmethod
def withdraw_liquidity(lp_amount: int) -> dict:
return {"withdraw_liquidity": {"amount": str(lp_amount)}}
@staticmethod
def swap(offer_asset: AssetInfo, offer_amount: int, min_receive: int) -> dict:
return {
"swap": {
"offer_asset": {
"info": (
{"native_token": {"denom": offer_asset.native}}
if offer_asset.native
else {"token": {"contract_addr": offer_asset.token}}
),
"amount": str(offer_amount),
},
"expected_amount": str(min_receive),
}
}
class DexClient:
"""DEX 合约交互客户端"""
def __init__(self, client: CosmosClient, factory_addr: str):
self.client = client
self.factory_addr = factory_addr
async def execute_contract(self, contract_addr: str, msg: dict, funds: list[Coin] = None) -> dict:
exec_msg = {
"@type": "/cosmwasm.wasm.v1.MsgExecuteContract",
"sender": self.client.get_address(),
"contract": contract_addr,
"msg": msg,
"funds": [{"denom": f.denom, "amount": str(f.amount)} for f in (funds or [])],
}
return await self.client.broadcast_tx(exec_msg)
async def query_contract(self, contract_addr: str, msg: dict) -> dict:
url = (
f"{self.client.rpc_endpoint.replace('rpc', 'api')}"
f"/cosmwasm/wasm/v1/contract/{contract_addr}/smart/"
)
async with aiohttp.ClientSession() as session:
async with session.get(url, params={"query": json.dumps(msg)}) as resp:
return await resp.json()
async def query_pools(self) -> list[PoolInfo]:
query = {"pairs": {"limit": 30}}
result = await self.query_contract(self.factory_addr, query)
pools = []
for i, pair in enumerate(result.get("data", {}).get("pairs", [])):
pools.append(PoolInfo(
pool_id=i + 1,
pool_addr=pair["contract_addr"],
assets=pair["asset_infos"],
total_lp_shares=int(pair.get("total_share", "0")),
swap_fee=float(pair.get("swap_fee", "0.003")),
))
return pools
3.3 流动性提供操作
class LiquidityProvider:
"""流动性提供者 Agent"""
MAX_SLIPPAGE = 0.02
MIN_LP_POSITION = 10_000
def __init__(self, agent_addr: str, dex: DexClient, pool_addr: str):
self.agent_addr = agent_addr
self.dex = dex
self.pool_addr = pool_addr
self._lp_balance: int = 0
async def provide_liquidity(
self,
msg_amount: int,
usdc_amount: int,
slippage_tolerance: float = 0.01,
) -> dict:
assets = [AssetInfo(native="umsg"), AssetInfo(native="uusdc")]
msg = ExecuteMsg.provide_liquidity(
assets=assets,
amounts=[msg_amount, usdc_amount],
slippage_tolerance=slippage_tolerance,
)
funds = [Coin(amount=msg_amount, denom="umsg"), Coin(amount=usdc_amount, denom="uusdc")]
return await self.dex.execute_contract(self.pool_addr, msg, funds)
async def withdraw_liquidity(self, lp_amount: int) -> dict:
msg = ExecuteMsg.withdraw_liquidity(lp_amount)
return await self.dex.execute_contract(self.pool_addr, msg)
async def query_lp_balance(self) -> int:
balance_query = {"balance": {"address": self.agent_addr}}
result = await self.dex.query_contract(self.pool_addr, balance_query)
balance = int(result.get("data", {}).get("balance", "0"))
self._lp_balance = balance
return balance
async def estimate_lp_value(self) -> dict:
pool_balances = await self.dex.get_pool_balances(self.pool_addr)
lp_balance = await self.query_lp_balance()
total_lp = int(pool_balances.get("total_share", 1))
share_ratio = lp_balance / total_lp if total_lp > 0 else 0
msg_value = int(pool_balances.get("umsg", 0) * share_ratio)
usdc_value = int(pool_balances.get("uusdc", 0) * share_ratio)
return {
"total_value_usd": usdc_value / 1_000_000 * 2,
"msg_value": msg_value,
"usdc_value": usdc_value,
}
3.4 无常损失监控
class ImpermanentLossMonitor:
"""无常损失 (IL) 监控器"""
IL_THRESHOLD = 0.05
IL_CRITICAL = 0.15
def __init__(self, dex: DexClient, pool_addr: str):
self.dex = dex
self.pool_addr = pool_addr
self._price_history: list[float] = []
async def calculate_il(self, initial_ratio: float, current_ratio: float) -> float:
import math
ratio = current_ratio / initial_ratio
il = (2 * math.sqrt(ratio) / (1 + ratio)) - 1
return il
async def monitor(self) -> Optional[dict]:
balances = await self.dex.get_pool_balances(self.pool_addr)
msg_balance = balances.get("umsg", 0)
usdc_balance = balances.get("uusdc", 0)
if msg_balance == 0 or usdc_balance == 0:
return None
current_ratio = msg_balance / usdc_balance
self._price_history.append(current_ratio)
if len(self._price_history) > 100:
self._price_history.pop(0)
if len(self._price_history) < 2:
return None
initial_ratio = self._price_history[0]
il = await self.calculate_il(initial_ratio, current_ratio)
alert = None
if il <= -self.IL_CRITICAL:
alert = {"level": "CRITICAL", "il": il,
"message": f"无常损失 {il * 100:.1f}%，建议立即移除流动性"}
elif il <= -self.IL_THRESHOLD:
alert = {"level": "WARNING", "il": il,
"message": f"无常损失 {il * 100:.1f}%，考虑调整仓位"}
return alert
3.5 收益农场 (Yield Farming)
class YieldFarm:
"""流动性挖矿 / 收益农场参与"""
def __init__(self, dex: DexClient, farm_addr: str):
self.dex = dex
self.farm_addr = farm_addr
async def stake_lp(self, lp_token_addr: str, amount: int) -> dict:
increase_allowance = {
"increase_allowance": {"spender": self.farm_addr, "amount": str(amount)}
}
await self.dex.execute_contract(lp_token_addr, increase_allowance)
stake_msg = {
"send": {
"contract": self.farm_addr,
"amount": str(amount),
"msg": "",
}
}
return await self.dex.execute_contract(lp_token_addr, stake_msg)
async def claim_rewards(self) -> dict:
claim_msg = {"claim": {}}
return await self.dex.execute_contract(self.farm_addr, claim_msg)
async def unstake_lp(self, amount: int) -> dict:
unstake_msg = {"leave": {"amount": str(amount)}}
return await self.dex.execute_contract(self.farm_addr, unstake_msg)
4. 治理投票
4.1 MSG Chain 治理模型
MSG Chain 基于 Cosmos SDK 的治理模块，支持参数变更、软件升级、社区池支出等提案类型。
治理流程:
提交提案 -> 存款阶段 -> 投票阶段 -> 执行/拒绝
最低 1MSG 到达阈值 7 天周期
投票选项: Yes / No / Abstain / NoWithVeto
4.2 核心治理操作
from enum import Enum
class VoteOption(str, Enum):
"""投票选项"""
YES = "VOTE_OPTION_YES"
NO = "VOTE_OPTION_NO"
ABSTAIN = "VOTE_OPTION_ABSTAIN"
NO_WITH_VETO = "VOTE_OPTION_NO_WITH_VETO"
@dataclass
class Proposal:
"""链上治理提案"""
id: int
title: str
description: str
status: str
submit_time: str
deposit_end_time: str
voting_start_time: str
voting_end_time: str
total_deposit: list[Coin]
proposer: str
content: dict
class AgentGovernance:
"""Agent 链上治理参与"""
MIN_INITIAL_DEPOSIT = 1_000_000
GAS_VOTE = 100_000
GAS_SUBMIT = 500_000
GAS_DEPOSIT = 200_000
def __init__(self, agent_addr: str, client: CosmosClient):
self.agent_addr = agent_addr
self.client = client
async def query_proposals(self, status: str = "") -> list[Proposal]:
url = f"{self.client.rpc_endpoint.replace('rpc', 'api')}/cosmos/gov/v1beta1/proposals"
params = {}
if status:
params["proposal_status"] = status
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as resp:
data = await resp.json()
proposals = []
for p in data.get("proposals", []):
content = p.get("content", {})
proposals.append(Proposal(
id=int(p["proposal_id"]),
title=content.get("title", p.get("title", "")),
description=content.get("description", p.get("summary", "")),
status=p["status"],
submit_time=p["submit_time"],
deposit_end_time=p["deposit_end_time"],
voting_start_time=p["voting_start_time"],
voting_end_time=p["voting_end_time"],
total_deposit=[Coin(amount=int(c["amount"]), denom=c["denom"]) for c in p.get("total_deposit", [])],
proposer=p["proposer"],
content=content,
))
return proposals
async def vote_on_proposal(self, proposal_id: int, option: VoteOption) -> dict:
msg = {
"@type": "/cosmos.gov.v1beta1.MsgVote",
"voter": self.agent_addr,
"proposal_id": str(proposal_id),
"option": option.value,
}
return await self.client.broadcast_tx(msg)
async def weighted_vote(self, proposal_id: int, options: list[tuple[VoteOption, float]]) -> dict:
msg = {
"@type": "/cosmos.gov.v1beta1.MsgVoteWeighted",
"voter": self.agent_addr,
"proposal_id": str(proposal_id),
"options": [{"option": option.value, "weight": str(weight)} for option, weight in options],
}
return await self.client.broadcast_tx(msg)
async def submit_proposal(
self, title: str, description: str, changes: list[dict],
deposit: int = MIN_INITIAL_DEPOSIT,
) -> dict:
msg = {
"@type": "/cosmos.gov.v1beta1.MsgSubmitProposal",
"content": {
"@type": "/cosmos.params.v1beta1.ParameterChangeProposal",
"title": title,
"description": description,
"changes": changes,
},
"initial_deposit": [{"denom": "umsg", "amount": str(deposit)}],
"proposer": self.agent_addr,
}
return await self.client.broadcast_tx(msg)
async def submit_text_proposal(
self, title: str, description: str, deposit: int = MIN_INITIAL_DEPOSIT,
) -> dict:
msg = {
"@type": "/cosmos.gov.v1beta1.MsgSubmitProposal",
"content": {
"@type": "/cosmos.gov.v1beta1.TextProposal",
"title": title,
"description": description,
},
"initial_deposit": [{"denom": "umsg", "amount": str(deposit)}],
"proposer": self.agent_addr,
}
return await self.client.broadcast_tx(msg)
async def submit_community_pool_spend_proposal(
self, title: str, description: str, recipient: str,
amount: int, deposit: int = MIN_INITIAL_DEPOSIT,
) -> dict:
msg = {
"@type": "/cosmos.gov.v1beta1.MsgSubmitProposal",
"content": {
"@type": "/cosmos.distribution.v1beta1.CommunityPoolSpendProposal",
"title": title,
"description": description,
"recipient": recipient,
"amount": [{"denom": "umsg", "amount": str(amount)}],
},
"initial_deposit": [{"denom": "umsg", "amount": str(deposit)}],
"proposer": self.agent_addr,
}
return await self.client.broadcast_tx(msg)
async def deposit_on_proposal(self, proposal_id: int, amount: int) -> dict:
msg = {
"@type": "/cosmos.gov.v1beta1.MsgDeposit",
"depositor": self.agent_addr,
"proposal_id": str(proposal_id),
"amount": [{"denom": "umsg", "amount": str(amount)}],
}
return await self.client.broadcast_tx(msg)
4.3 提案分析与自动投票
@dataclass
class ProposalAnalysis:
"""提案分析结果"""
proposal_id: int
title: str
summary: str
risk_level: str
impact_assessment: str
recommendation: VoteOption
confidence: float
reasoning: list[str]
class Constitution:
"""Agent 宪法 - 投票决策的规则引擎"""
def __init__(self):
self.principles = [
"网络安全性优先: 任何降低安全性的参数变更应反对",
"去中心化: 偏向降低通胀、降低验证人佣金上限的提议",
"经济可持续: 偏向合理通胀率维持网络安全",
"社区利益: 社区池支出应有明确可衡量的产出",
"技术升级: 经过充分测试的软件升级应支持",
]
async def evaluate_proposal(self, proposal: Proposal) -> ProposalAnalysis:
content_type = proposal.content.get("@type", "")
if "ParameterChange" in content_type:
return await self._evaluate_param_change(proposal)
elif "CommunityPoolSpend" in content_type:
return await self._evaluate_community_spend(proposal)
elif "SoftwareUpgrade" in content_type:
return await self._evaluate_upgrade(proposal)
else:
return await self._evaluate_text(proposal)
async def _evaluate_param_change(self, proposal: Proposal) -> ProposalAnalysis:
changes = proposal.content.get("changes", [])
reasoning = []
recommendation = VoteOption.YES
confidence = 0.5
risk = "MEDIUM"
for change in changes:
key = change.get("key", "")
value = change.get("value", "")
if "MaxValidators" in key:
new_val = int(value)
if new_val > 150:
recommendation = VoteOption.NO
confidence = 0.6
risk = "HIGH"
return ProposalAnalysis(
proposal_id=proposal.id, title=proposal.title,
summary=proposal.description[:200], risk_level=risk,
recommendation=recommendation, confidence=confidence,
reasoning=reasoning,
)
async def _evaluate_community_spend(self, proposal: Proposal) -> ProposalAnalysis:
recipient = proposal.content.get("recipient", "")
reasoning = []
if 'amount' in str(proposal.content):
reasoning.append("社区池支出提案")
return ProposalAnalysis(
proposal_id=proposal.id, title=proposal.title,
summary=proposal.description[:200], risk_level="LOW",
recommendation=VoteOption.YES, confidence=0.6,
reasoning=reasoning,
)
async def _evaluate_text(self, proposal: Proposal) -> ProposalAnalysis:
return ProposalAnalysis(
proposal_id=proposal.id, title=proposal.title,
summary=proposal.description[:200], risk_level="LOW",
impact_assessment="信号提案，无直接链上影响",
recommendation=VoteOption.YES, confidence=0.6,
reasoning=["文本提案，无参数变更或资金支出"],
)
4.4 自动投票引擎
class AutoVoter:
"""自动投票引擎"""
def __init__(self, governance: AgentGovernance, constitution: Constitution):
self.governance = governance
self.constitution = constitution
self.vote_history: dict[int, VoteOption] = {}
async def auto_vote(self) -> list[dict]:
proposals = await self.governance.query_proposals(status="PROPOSAL_STATUS_VOTING_PERIOD")
results = []
for proposal in proposals:
if proposal.id in self.vote_history:
continue
analysis = await self.constitution.evaluate_proposal(proposal)
if analysis.confidence >= 0.7:
tx = await self.governance.vote_on_proposal(proposal.id, analysis.recommendation)
results.append({"proposal_id": proposal.id, "vote": analysis.recommendation,
"confidence": analysis.confidence, "tx": tx})
self.vote_history[proposal.id] = analysis.recommendation
return results
async def get_voting_power(self) -> int:
url = (f"{self.governance.client.rpc_endpoint.replace('rpc', 'api')}"
f"/cosmos/staking/v1beta1/delegations/{self.governance.agent_addr}")
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
data = await resp.json()
return sum(int(d.get("balance", {}).get("amount", 0)) for d in data.get("delegation_responses", []))
5. Agent 代币发行 (CW20)
5.1 CW20 代币标准
CW20 是 CosmWasm 生态的代币标准，类似 ERC20 但针对 Cosmos 生态优化。Agent 可以发行自有代币建立独立经济系统。
Agent 代币经济模型:
[Agent Token]
(CW20)
/ | \
效用 治理 社交
代币 代币 代币
5.2 CW20 合约部署
class AgentTokenCW20:
"""Agent CW20 代币发行"""
TOKEN_CODE_ID = 1
DECIMALS = 6
def __init__(self, client: CosmosClient, agent_addr: str):
self.client = client
self.agent_addr = agent_addr
self.token_contract_addr: Optional[str] = None
async def create_token(
self, name: str, symbol: str, initial_supply: int,
mintable: bool = True, cap: Optional[int] = None,
) -> dict:
label = f"agent-{symbol.lower()}-{hash(name) % 10000}"
minter_config = None
if mintable:
minter_config = {"minter": self.agent_addr, "cap": str(cap) if cap else None}
instantiate_msg = {
"name": name, "symbol": symbol.upper(), "decimals": self.DECIMALS,
"initial_balances": [{"address": self.agent_addr, "amount": str(initial_supply)}],
"mint": minter_config,
}
msg = {
"@type": "/cosmwasm.wasm.v1.MsgInstantiateContract",
"sender": self.agent_addr, "admin": self.agent_addr,
"code_id": str(self.TOKEN_CODE_ID), "label": label,
"msg": instantiate_msg, "funds": [],
}
result = await self.client.broadcast_tx(msg)
if result.get("code") == 0:
self.token_contract_addr = self._parse_contract_address(result)
return result
def _parse_contract_address(self, tx_result: dict) -> Optional[str]:
events = tx_result.get("events", []) if isinstance(tx_result, dict) else []
for event in events:
if event.get("type") == "instantiate":
for attr in event.get("attributes", []):
if attr.get("key") == "_contract_address":
return attr.get("value")
return None
5.3 代币铸造与销毁
class TokenMinter:
"""CW20 代币铸造与销毁"""
def __init__(self, client: CosmosClient, contract_addr: str, agent_addr: str):
self.client = client
self.contract_addr = contract_addr
self.agent_addr = agent_addr
async def mint(self, recipient: str, amount: int) -> dict:
msg = {"mint": {"recipient": recipient, "amount": str(amount)}}
exec_msg = {"@type": "/cosmwasm.wasm.v1.MsgExecuteContract",
"sender": self.agent_addr, "contract": self.contract_addr,
"msg": msg, "funds": []}
return await self.client.broadcast_tx(exec_msg)
async def burn(self, amount: int) -> dict:
msg = {"burn": {"amount": str(amount)}}
exec_msg = {"@type": "/cosmwasm.wasm.v1.MsgExecuteContract",
"sender": self.agent_addr, "contract": self.contract_addr,
"msg": msg, "funds": []}
return await self.client.broadcast_tx(exec_msg)
5.4 CW20 转账与授权
class TokenTransfer:
"""CW20 转账与授权操作"""
def __init__(self, client: CosmosClient, contract_addr: str, agent_addr: str):
self.client = client
self.contract_addr = contract_addr
self.agent_addr = agent_addr
async def transfer(self, recipient: str, amount: int) -> dict:
msg = {"transfer": {"recipient": recipient, "amount": str(amount)}}
exec_msg = {"@type": "/cosmwasm.wasm.v1.MsgExecuteContract",
"sender": self.agent_addr, "contract": self.contract_addr,
"msg": msg, "funds": []}
return await self.client.broadcast_tx(exec_msg)
async def increase_allowance(self, spender: str, amount: int) -> dict:
msg = {"increase_allowance": {"spender": spender, "amount": str(amount)}}
exec_msg = {"@type": "/cosmwasm.wasm.v1.MsgExecuteContract",
"sender": self.agent_addr, "contract": self.contract_addr,
"msg": msg, "funds": []}
return await self.client.broadcast_tx(exec_msg)
async def balance(self, address: str) -> int:
query = {"balance": {"address": address}}
url = (f"{self.client.rpc_endpoint.replace('rpc', 'api')}"
f"/cosmwasm/wasm/v1/contract/{self.contract_addr}/smart/"
)
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
data = await resp.json()
return int(data.get("data", {}).get("balance", "0"))
5.5 债券曲线 (Bonding Curve)
class BondingCurve:
"""Agent 代币债券曲线"""
def __init__(
self, token: AgentTokenCW20,
base_price: float = 0.001,
exponent: float = 1.5,
reserve_denom: str = "umsg",
):
self.token = token
self.base_price = base_price
self.exponent = exponent
self.reserve_denom = reserve_denom
self.reserve_balance: int = 0
self.token_supply: int = 0
def calculate_price(self, current_supply: int) -> float:
if current_supply <= 0:
return self.base_price
return self.base_price * (current_supply ** (self.exponent - 1))
async def buy(self, msg_amount: int) -> dict:
supply = await self._query_supply()
token_amount = int(msg_amount / self.calculate_price(supply))
return {"action": "buy", "token_amount": token_amount, "cost": msg_amount}
async def sell(self, token_amount: int) -> dict:
supply = await self._query_supply()
return {"action": "sell", "token_amount": token_amount}
async def _query_supply(self) -> int:
if not self.token.token_contract_addr:
return 0
query = {"token_info": {}}
url = (f"{self.token.client.rpc_endpoint.replace('rpc', 'api')}"
f"/cosmwasm/wasm/v1/contract/{self.token.token_contract_addr}/smart/"
)
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
data = await resp.json()
return int(data.get("data", {}).get("total_supply", "0"))
5.6 代币经济参数设计
class TokenomicsDesign:
"""Agent 代币经济参数设计器"""
@staticmethod
def optimal_initial_distribution(total_supply: int) -> dict[str, int]:
return {
"community_pool": int(total_supply * 0.30),
"team_vesting": int(total_supply * 0.20),
"liquidity_provision": int(total_supply * 0.15),
"staking_rewards": int(total_supply * 0.15),
"airdrop": int(total_supply * 0.10),
"initial_sale": int(total_supply * 0.10),
}
6. 收益管理与金库
6.1 金库架构
Agent 金库是一个多策略资产管理引擎，自动将资金分配到最高效的收益策略中。
金库架构:
[Agent Treasury]
/ | \
Staking LP Lending
50% 30% 20%
6.2 多策略金库实现
from dataclasses import dataclass
from typing import Protocol, Optional
@dataclass
class PortfolioSnapshot:
"""投资组合快照"""
total_value_msg: int
staking_value: int
liquidity_value: int
lending_value: int
liquid_balance: int
timestamp: int
class Strategy(Protocol):
"""策略接口"""
weight: float
name: str
async def current_value(self) -> int: ...
async def adjust(self, delta: int): ...
@dataclass
class StakingStrategy:
"""质押策略"""
weight: float
staking: AgentStaking
name: str = "staking"
async def current_value(self) -> int:
delegations = await self.staking.query_delegations()
return sum(int(d.get("balance", {}).get("amount", 0)) for d in delegations)
async def adjust(self, delta: int) -> Optional[dict]:
if delta > 0:
return await self.staking.delegate(self.staking.default_validator, delta)
elif delta < 0:
return await self.staking.undelegate(self.staking.default_validator, abs(delta))
return None
@dataclass
class LiquidityStrategy:
"""流动性策略"""
weight: float
provider: LiquidityProvider
name: str = "liquidity"
async def adjust(self, delta: int) -> Optional[dict]:
if delta > 0:
half = delta // 2
return await self.provider.provide_liquidity(half, half)
return None
class AgentTreasury:
"""Agent 金库 - 多策略资产管理"""
REBALANCE_THRESHOLD = 0.05
def __init__(self, agent_addr: str, strategies: Optional[dict[str, Strategy]] = None):
self.agent_addr = agent_addr
self.strategies = strategies or {}
self._last_rebalance = 0
self.portfolio_history: list[PortfolioSnapshot] = []
async def query_portfolio(self) -> PortfolioSnapshot:
values = {}
for name, strategy in self.strategies.items():
values[name] = await strategy.current_value()
total = sum(values.values()) if values else 0
return PortfolioSnapshot(
total_value_msg=total,
staking_value=values.get("staking", 0),
liquidity_value=values.get("liquidity", 0),
lending_value=values.get("lending", 0),
liquid_balance=0,
timestamp=0,
)
async def rebalance(self) -> list[dict]:
import time
now = time.time()
if now - self._last_rebalance < 3600:
return []
portfolio = await self.query_portfolio()
if portfolio.total_value_msg <= 0:
return []
adjustments = []
for name, strategy in self.strategies.items():
target = int(portfolio.total_value_msg * strategy.weight)
current = await strategy.current_value()
deviation = abs(target - current) / max(portfolio.total_value_msg, 1)
if deviation > self.REBALANCE_THRESHOLD:
delta = target - current
result = await strategy.adjust(delta)
adjustments.append({"strategy": name, "delta": delta, "result": result})
self._last_rebalance = now
return adjustments
7. 经济参与策略
7.1 完整策略循环
class EconomicStrategy:
"""自动化经济参与策略"""
STRATEGY_INTERVAL = 3600
COMPOUND_INTERVAL = 86400
REBALANCE_INTERVAL = 604800
GOVERNANCE_INTERVAL = 21600
def __init__(
self, agent_addr: str, private_key: str,
rpc_endpoint: str = "https://rpc.msg-chain-1.msg.network",
rest_endpoint: str = "https://api.msg-chain-1.msg.network",
validator_addr: str = "", pool_addr: str = "",
):
self.agent_addr = agent_addr
self.client = CosmosClient(rpc_endpoint, private_key)
self.staking = AgentStaking(agent_addr, self.client, validator_addr)
self.selector = ValidatorSelector(rest_endpoint)
self.dex = DexClient(self.client, "")
self.provider = LiquidityProvider(agent_addr, self.dex, pool_addr)
self.governance = AgentGovernance(agent_addr, self.client)
self.constitution = Constitution()
self.auto_voter = AutoVoter(self.governance, self.constitution)
self.strategies: dict[str, Strategy] = {}
self.treasury = AgentTreasury(agent_addr, self.strategies)
self.running = False
def configure_strategies(
self, staking_weight: float = 0.5, liquidity_weight: float = 0.3,
):
self.strategies["staking"] = StakingStrategy(weight=staking_weight, staking=self.staking)
self.strategies["liquidity"] = LiquidityStrategy(weight=liquidity_weight, provider=self.provider)
self.treasury = AgentTreasury(self.agent_addr, self.strategies)
async def initialize(self) -> dict:
url = f"{self.rest}/cosmos/bank/v1beta1/balances/{self.agent_addr}"
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
balances = await resp.json()
msg_balance = 0
for coin in balances.get("balances", []):
if coin["denom"] == "umsg":
msg_balance = int(coin["amount"])
if msg_balance < 1_000_000:
return {"status": "insufficient_funds", "balance": msg_balance}
return {"status": "initialized", "balance": msg_balance}
async def execute_strategy(self):
self.running = True
results = {"compounding": [], "governance": [],
"rebalance": [], "errors": []}
try:
compound_tx = await self.staking.auto_compound()
if compound_tx:
results["compounding"].append(compound_tx)
gov_results = await self.auto_voter.auto_vote()
results["governance"].extend(gov_results)
rebalance_txs = await self.treasury.rebalance()
results["rebalance"].extend(rebalance_txs)
except Exception as e:
results["errors"].append(str(e))
return results
async def run_forever(self):
self.running = True
import asyncio, time
while self.running:
await asyncio.sleep(self.STRATEGY_INTERVAL)
await self.execute_strategy()
async def stop(self):
self.running = False
7.2 Rust 链上合约示例
use cosmwasm_std::{entry_point, Addr, Coin, CosmosMsg, Decimal, DepsMut,
Env, MessageInfo, Response, StdResult, Uint128};
use cw_storage_plus::Item;
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
pub struct Config {
pub owner: Addr,
pub default_validator: Addr,
pub staking_weight: Decimal,
pub rebalance_threshold: Decimal,
}
pub const CONFIG: Item<Config> = Item::new("config");
#[entry_point]
pub fn instantiate(
deps: DepsMut, _env: Env, _info: MessageInfo, msg: InstantiateMsg,
) -> StdResult<Response> {
let config = Config {
owner: deps.api.addr_validate(&msg.owner)?,
default_validator: deps.api.addr_validate(&msg.default_validator)?,
staking_weight: Decimal::percent(50),
rebalance_threshold: Decimal::percent(5),
};
CONFIG.save(deps.storage, &config)?;
Ok(Response::new().add_attribute("action", "instantiate"))
}
#[entry_point]
pub fn execute(
deps: DepsMut, env: Env, info: MessageInfo, msg: ExecuteMsg,
) -> StdResult<Response> {
match msg {
ExecuteMsg::Delegate { validator, amount } => {
execute_delegate(deps, env, info, validator, amount)
}
_ => Ok(Response::new()),
}
}
pub fn execute_delegate(
deps: DepsMut, _env: Env, info: MessageInfo,
validator: String, amount: Uint128,
) -> StdResult<Response> {
let config = CONFIG.load(deps.storage)?;
if info.sender != config.owner {
return Err(cosmwasm_std::StdError::generic_err("unauthorized"));
}
let delegate_msg = CosmosMsg::Staking(cosmwasm_std::StakingMsg::Delegate {
validator: validator.clone(),
amount: Coin { denom: "umsg".to_string(), amount },
});
Ok(Response::new().add_message(delegate_msg))
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
pub enum ExecuteMsg {
Delegate { validator: String, amount: Uint128 },
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
pub struct InstantiateMsg {
pub owner: String,
pub default_validator: String,
}
附录
A. MSG Chain RPC 端点
| 服务 | 端点 |
|---|---|
| Tendermint RPC | https://rpc.msg-chain-1.msg.network |
| REST/LCD API | https://api.msg-chain-1.msg.network |
| WebSocket | wss://rpc.msg-chain-1.msg.network/websocket |
| gRPC | grpc.msg-chain-1.msg.network:443 |
B. Gas 费用估算
| 操作 | Gas 上限 | 费用 (umsg) |
|---|---|---|
| MsgDelegate | 200,000 | 5,000 |
| MsgUndelegate | 250,000 | 6,250 |
| MsgBeginRedelegate | 300,000 | 7,500 |
| MsgVote | 100,000 | 2,500 |
| MsgSubmitProposal | 500,000 | 12,500 |
| ProvideLiquidity | 400,000 | 10,000 |
| CW20 Transfer | 150,000 | 3,750 |
C. Bech32 地址格式
| 用途 | 前缀 |
|---|---|
| 账户地址 | msg1 |
| 验证人操作地址 | msgvaloper1 |
| 合约地址 | msg1 |
本指南针对 MSG Chain (msg-chain-1) 代币经济体系，代码示例仅供架构参考。