2025-09-29

[MtkN1]

Today I Learned ...

[MtkN1]

集中流動性プールからアカウントの供給量を取得する

コードメモ

.env

WEB3_PROVIDER_URI=
ACCOUNT_ADDRESS=
POOL_ADDRESS=

fetch_cl_pool_positions.py

import json
import os
from math import sqrt
from pathlib import Path

from web3 import Web3

w3 = Web3()

ACCOUNT = Web3.to_checksum_address(os.environ["ACCOUNT_ADDRESS"])
POOL = Web3.to_checksum_address(os.environ["POOL_ADDRESS"])

with open(Path(__file__).parent / "CLGauge.json") as fp:
    ABI_CLGAUGE = json.load(fp)

with open(Path(__file__).parent / "NonfungiblePositionManager.json") as fp:
    ABI_NFPM = json.load(fp)

with open(Path(__file__).parent / "CLPool.json") as fp:
    ABI_POOL = json.load(fp)

with open(Path(__file__).parent / "WETH9.json") as fp:
    ABI_ERC20 = json.load(fp)

pool = w3.eth.contract(address=POOL, abi=ABI_POOL)

# プール情報
token0_addr = Web3.to_checksum_address(pool.functions.token0().call())
token1_addr = Web3.to_checksum_address(pool.functions.token1().call())
gauge_addr = Web3.to_checksum_address(
    pool.functions.gauge().call()
)  # プールに紐づく Gauge

gauge = w3.eth.contract(address=gauge_addr, abi=ABI_CLGAUGE)
nfpm_addr = Web3.to_checksum_address(gauge.functions.nft().call())
nfpm = w3.eth.contract(address=nfpm_addr, abi=ABI_NFPM)


# 価格・tick の取得（slot0 or globalState を順に試す）
def read_sqrt_price_and_tick(pool):
    try:
        s = pool.functions.slot0().call()
        # [0]=sqrtPriceX96, [1]=tick
        return int(s[0]), int(s[1])
    except Exception:
        s = pool.functions.globalState().call()
        return int(s[0]), int(s[1])


sqrtPriceX96, current_tick = read_sqrt_price_and_tick(pool)

# ERC20 情報
tok0 = w3.eth.contract(address=token0_addr, abi=ABI_ERC20)
tok1 = w3.eth.contract(address=token1_addr, abi=ABI_ERC20)
dec0 = tok0.functions.decimals().call()
dec1 = tok1.functions.decimals().call()
sym0 = tok0.functions.symbol().call()
sym1 = tok1.functions.symbol().call()

Q96 = 2**96


# Uniswap v3 互換の数量計算
def get_sqrt_ratio_at_tick(tick: int) -> int:
    # 近似ではなく v3 の TickMath と等価の結果が必要ならライブラリを移植してください。
    # ここでは簡易に tick→価格→sqrtRatio を計算します（1.0001^tick）
    price = 1.0001**tick
    return int(sqrt(price) * Q96)


def amounts_from_liquidity(liq: int, tick_lower: int, tick_upper: int, sqrtP_x96: int):
    sqrtPa = get_sqrt_ratio_at_tick(tick_lower)
    sqrtPb = get_sqrt_ratio_at_tick(tick_upper)
    if sqrtPa > sqrtPb:
        sqrtPa, sqrtPb = sqrtPb, sqrtPa

    if sqrtP_x96 <= sqrtPa:
        # 全部 token0
        amount0 = liq * ((sqrtPb - sqrtPa) * Q96) // (sqrtPa * sqrtPb)
        amount1 = 0
    elif sqrtP_x96 < sqrtPb:
        # 両方
        amount0 = liq * ((sqrtPb - sqrtP_x96) * Q96) // (sqrtP_x96 * sqrtPb)
        amount1 = liq * (sqrtP_x96 - sqrtPa) // Q96
    else:
        # 全部 token1
        amount0 = 0
        amount1 = liq * (sqrtPb - sqrtPa) // Q96
    return amount0, amount1


# ウォレット直保有の tokenId を列挙（NFPM は ERC721Enumerable）
def owned_token_ids(owner):
    n = nfpm.functions.balanceOf(owner).call()
    return [nfpm.functions.tokenOfOwnerByIndex(owner, i).call() for i in range(n)]


# Gauge ステーク中の tokenId 一覧
staked_ids = list(map(int, gauge.functions.stakedValues(ACCOUNT).call()))

owned_ids = owned_token_ids(ACCOUNT)


# 指定プールの token0/token1 と一致するポジションだけを対象にする
def is_position_for_pool(pos_token0, pos_token1):
    # Slipstream は同一ペアで複数の tickSpacing（fee）プールがあり得ます。
    # 最低限のフィルタとしてトークンペア一致を確認。必要なら Factory で (token0,token1,tickSpacing) の getPool を引くと厳密です。
    return (
        Web3.to_checksum_address(pos_token0) == token0_addr
        and Web3.to_checksum_address(pos_token1) == token1_addr
    ) or (
        Web3.to_checksum_address(pos_token0) == token1_addr
        and Web3.to_checksum_address(pos_token1) == token0_addr
    )


def read_position(token_id: int):
    # positions の戻り値タプル（v3 準拠）
    p = nfpm.functions.positions(token_id).call()
    # p[2]=token0, p[3]=token1, p[4]=tickLower, p[5]=tickUpper, p[6]=liquidity が SlipstreamのIFと一致
    return {
        "token0": p[2],
        "token1": p[3],
        # "tickSpacing": int(p[4]),
        "tickLower": int(p[5]),
        "tickUpper": int(p[6]),
        "liquidity": int(p[7]),
    }


def sum_amounts(token_ids):
    total0 = 0
    total1 = 0
    for tid in token_ids:
        pos = read_position(tid)
        if not is_position_for_pool(pos["token0"], pos["token1"]):
            continue
        a0, a1 = amounts_from_liquidity(
            pos["liquidity"], pos["tickLower"], pos["tickUpper"], sqrtPriceX96
        )
        total0 += a0
        total1 += a1
    return total0, total1


total0_owned, total1_owned = sum_amounts(owned_ids)
total0_staked, total1_staked = sum_amounts(staked_ids)

total0 = (total0_owned + total0_staked) / (10**dec0)
total1 = (total1_owned + total1_staked) / (10**dec1)

print(f"Pool: {POOL}")
print(f"Account: {ACCOUNT}")
print(f"Current tick: {current_tick}, sqrtPriceX96: {sqrtPriceX96}")
print(f"Total in pool ({sym0}): {total0:.6f}")
print(f"Total in pool ({sym1}): {total1:.6f}")
print("--- breakdown ---")
print(
    f"  Direct  : {sym0} {total0_owned / (10**dec0):.6f}, {sym1} {total1_owned / (10**dec1):.6f}"
)
print(
    f"  Staked  : {sym0} {total0_staked / (10**dec0):.6f}, {sym1} {total1_staked / (10**dec1):.6f}"
)

[MtkN1]

• CLGauge.json
• NonfungiblePositionManager.json
• CLPool.json
• WETH9.json

[MtkN1]

Uniswap の interface 向け GraphQL でトークンのドル建て価格を取得する

https://github.com/Uniswap/interface/tree/main/packages/uniswap/src/data/graphql/uniswap-data-api

import json

import requests

url = "https://interface.gateway.uniswap.org/v1/graphql"
graphql_query = """
query TokenPrices($contracts: [ContractInput!]!) {
  tokens(contracts: $contracts) {
    chain
    address
    standard
    decimals
    name
    symbol
    market(currency: USD) {
      price {
        value
      }
    }
  }
}
"""
request_json = {
    "operationName": "TokenPrices",
    "variables": {
        "contracts": [
            {
                "chain": "ETHEREUM",
            },
            {
                "chain": "BASE",
            },
            {
                "chain": "BASE",
                "address": "0xcbB7C0000aB88B473b1f5aFd9ef808440eed33Bf",
            },
        ]
    },
    "query": graphql_query,
}
headers = {"Origin": "https://app.uniswap.org"}

r = requests.post(url, headers=headers, json=request_json)
print(r)

response_json = r.json()
print(json.dumps(response_json, ensure_ascii=False, indent=2))

{
  "data": {
    "tokens": [
      {
        "chain": "ETHEREUM",
        "address": null,
        "standard": "NATIVE",
        "decimals": 18,
        "name": "Ethereum",
        "symbol": "ETH",
        "market": {
          "price": {
            "value": 4105.555620160244
          }
        }
      },
      {
        "chain": "BASE",
        "address": null,
        "standard": "NATIVE",
        "decimals": 18,
        "name": "Ethereum",
        "symbol": "ETH",
        "market": {
          "price": {
            "value": 4115.407545565193
          }
        }
      },
      {
        "chain": "BASE",
        "address": "0xcbB7C0000aB88B473b1f5aFd9ef808440eed33Bf",
        "standard": "ERC20",
        "decimals": 8,
        "name": "Coinbase Wrapped BTC",
        "symbol": "cbBTC",
        "market": {
          "price": {
            "value": 112051.44649952077
          }
        }
      }
    ]
  }
}

[MtkN1]

単一版:

import json

import requests

url = "https://interface.gateway.uniswap.org/v1/graphql"
graphql_query = """
query TokenPrice($chain: Chain!, $address: String = null) {
  token(chain: $chain, address: $address) {
    chain
    address
    standard
    decimals
    name
    symbol
    market(currency: USD) {
      price {
        value
      }
    }
  }
}
"""
request_json = {
    "operationName": "TokenPrice",
    "variables": {
        "chain": "BASE",
        "address": "0xcbB7C0000aB88B473b1f5aFd9ef808440eed33Bf",
    },
    "query": graphql_query,
}
headers = {"Origin": "https://app.uniswap.org"}

r = requests.post(url, headers=headers, json=request_json)
print(r)

response_json = r.json()
print(json.dumps(response_json, ensure_ascii=False, indent=2))

{
  "data": {
    "token": {
      "chain": "BASE",
      "address": "0xcbB7C0000aB88B473b1f5aFd9ef808440eed33Bf",
      "standard": "ERC20",
      "decimals": 8,
      "name": "Coinbase Wrapped BTC",
      "symbol": "cbBTC",
      "market": {
        "price": {
          "value": 112124.55281179183
        }
      }
    }
  }
}