Python

pip install chainstream-sdk

from chainstream import ChainStreamClient

client = ChainStreamClient(access_token='YOUR_ACCESS_TOKEN')

import asyncio
from chainstream.openapi_client import ApiClient, Configuration
from chainstream.openapi_client.api.token_api import TokenApi
from chainstream.openapi_client.models.chain_symbol import ChainSymbol

ACCESS_TOKEN = 'YOUR_ACCESS_TOKEN'
BASE_URL = 'https://api.chainstream.io'

async def main():
    # 設定 API 用戶端
    config = Configuration(
        host=BASE_URL,
        access_token=ACCESS_TOKEN,
    )
    
    async with ApiClient(config) as api_client:
        # 建立 Token API 實例
        token_api = TokenApi(api_client)
        
        chain = ChainSymbol.SOL
        token_address = 'So11111111111111111111111111111111111111112'  # SOL
        
        print(f'查詢代幣: {chain.value}/{token_address}')
        
        # 取得代幣元資料
        result = await token_api.get_metadata(
            chain=chain,
            token_address=token_address,
        )
        
        print('代幣元資料:')
        print(f'  名稱: {result.name}')
        print(f'  符號: {result.symbol}')
        print(f'  精度: {result.decimals}')

if __name__ == '__main__':
    asyncio.run(main())

import asyncio
import signal
from chainstream import ChainStreamClient
from chainstream.stream import Resolution, TokenCandle

ACCESS_TOKEN = 'YOUR_ACCESS_TOKEN'

async def main():
    client = ChainStreamClient(ACCESS_TOKEN)

    chain = 'sol'
    token_address = 'So11111111111111111111111111111111111111112'  # SOL
    resolution = Resolution.S1

    print(f'訂閱代幣 K 線: {chain}/{token_address}')
    print('監聽中... (按 Ctrl+C 停止)\n')

    message_count = 0

    def on_candle(candle: TokenCandle) -> None:
        nonlocal message_count
        message_count += 1
        print(
            f'[{message_count}] open={candle.open}, close={candle.close}, '
            f'high={candle.high}, low={candle.low}, volume={candle.volume}'
        )

    unsub = await client.stream.subscribe_token_candles(
        chain=chain,
        token_address=token_address,
        resolution=resolution,
        callback=on_candle,
    )

    # 建立事件等待 Ctrl+C
    stop_event = asyncio.Event()

    def signal_handler(sig, frame):
        print(f'\n收到 {message_count} 則訊息')
        print('關閉連線...')
        stop_event.set()

    signal.signal(signal.SIGINT, signal_handler)

    await stop_event.wait()

    # 清理
    unsub.unsubscribe()
    await client.close()

if __name__ == '__main__':
    asyncio.run(main())

results = await token_api.search_token(keyword='bonk')
for t in results:
    print(f'{t.symbol}: {t.address}')

from chainstream.openapi_client.api.wallet_api import WalletApi

async with ApiClient(config) as api_client:
    wallet_api = WalletApi(api_client)
    pnl = await wallet_api.get_pnl(
        chain=ChainSymbol.SOL,
        wallet_address='WALLET_ADDRESS',
    )
    print(f'Total PnL: {pnl.total_pnl_usd}')

from chainstream.openapi_client.api.trade_api import TradeApi

async with ApiClient(config) as api_client:
    trade_api = TradeApi(api_client)
    trades = await trade_api.get_trades(
        chain=ChainSymbol.SOL,
        token_address='TOKEN_ADDRESS',
        limit=20,
    )
    for trade in trades:
        print(f'{trade.timestamp}: {trade.side} {trade.amount}')

import pandas as pd

candles = await client.token.get_candles('sol', 'TOKEN_ADDRESS', resolution='1h', limit=100)
df = pd.DataFrame(candles)
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='s')

from chainstream.openapi_client.exceptions import ApiException

try:
    result = await token_api.get_token(chain=ChainSymbol.SOL, token_address='INVALID')
except ApiException as e:
    print(f'API error {e.status}: {e.reason}')
    if e.status == 429:
        import asyncio
        await asyncio.sleep(2)