Simple tool using the Mintscan API to generate a TSV of Osmosis swaps

extract_osmosis_swaps.py

#!/usr/bin/python3
import json
import sys
import requests

def get_transactions(address):
    resp = requests.get(
        'https://api-osmosis.cosmostation.io/v1/account/new_txs/' + address,
        params={'limit':5000, 'from': 0},
        timeout=31,
    )
    return resp.json()

def get_denoms():
    resp = requests.get(
        'https://api-utility.cosmostation.io/v1//ibc/tokens/osmosis-1',
        timeout=31,
    )
    js = resp.json()['ibc_tokens']
    return {x['hash']: x['base_denom'] for x in js}

def extract_swaps(address, txn_js):
    out = []
    for x in txn_js:
        if not any(
            m['@type'] == '/osmosis.gamm.v1beta1.MsgSwapExactAmountIn'
            for m in x['data']['tx']['body']['messages']
        ):
            continue  # not a swap

        ts = x['data']['timestamp']

        log_js = json.loads(x['data']['raw_log'])

        transfer_evts = [
            a
            for l in log_js
            for e in l['events']
            for a in e['attributes']
            if e['type'] == 'transfer'
        ]

        txn = []
        txn_direction = 0
        for attr in transfer_evts:
            key = attr['key']
            val = attr['value']
            if key == 'recipient':
                txn_direction = +1 if val == address else -1
            if key == 'sender':
                txn_direction = -1 if val == address else +1
            if key == 'amount':
                if 'ibc/' in val:
                    amount, denom = val.split('ibc/')
                    amount = int(amount)
                elif 'uosmo' in val:
                    amount, _ = val.split('uosmo')
                    denom = 'uosmo'
                    amount = int(amount)
                txn.append((txn_direction, amount, denom))
                txn_direction = 0
        out.append((ts, txn))

    return out

def main():
    if len(sys.argv) < 2:
        print("Usage: {} <address>".format(sys.argv[0]))
        print("Generates a TSV of osmosis swaps")
        return

    address = sys.argv[1]
    if not address.startswith('osmo'):
        print("Supply an osmosis address")
        return

    denom_map = get_denoms()
    txn_js = get_transactions(address)

    swaps = extract_swaps(address, txn_js)
    print("timestamp\tdirection\tamount\tdenomination")
    for ts, swap_txns in swaps:
        for (direction, amount, denom) in swap_txns:
            mapped_denom = denom_map.get(denom, denom)
            print('\t'.join(str(x) for x in [
                ts,
                'IN' if direction > 0 else 'OUT',
                amount,
                mapped_denom,
            ]))

if __name__ == "__main__":
    main()