Wallet evaluation

This example shows how to do basic wallet inflows analysis. Here we are building an entirely open-source analyzer. You can see exactly how numbers are calculated, change the rules to your liking or add new ones.

For this demo set the address of interest here:

https://github.com/ChainArgos/api-demos/blob/main/looker/wallet_eval/config.py

# which address to check
ADDRESS = "0x346eF244464679b031750f70D750B3FA65165443"

# which looks to use for the underlying information
WALLET_LOOK_ID = 722
INFLOWS_LOOK_ID = 724

TOKENS_SYMBOL = "Tokens Symbol"
TRANSACTIONS_FROM_ADDRESS = "Transactions From Address"
TRANSACTIONS_SUM_OF_TRANSFER_AMOUNTS = "Transactions Sum of Transfer Amounts"
FROM_WALLET_LABELS = "From Wallet Labels"

WALLETS_ADDRESS = "Wallets Address"
WALLETS_CATEGORIES = "Wallets Categories"

# separator in returned lists
JOINER = " ;; "

Note this file also specifies what DashArgos queries to run on top of. You can set up new queries and point your code to them if you wish.

This file specifies our is this inflow source a problem? definition:

https://github.com/ChainArgos/api-demos/blob/main/looker/wallet_eval/is_suspicious.py

from config import TRANSACTIONS_FROM_ADDRESS, FROM_WALLET_LABELS


def is_suspicious_address(address, inflows, categories):
    this_categories = categories[address]
    this_label = inflows.loc[(inflows[TRANSACTIONS_FROM_ADDRESS] == address)][FROM_WALLET_LABELS]
    if 'blacklisted' in this_categories or 'ofac' in this_categories or 'terrorists' in this_categories or 'darknet market' in categories:
        return True
    elif 'cex' in this_categories:
        # exchanges are not suspicious without one of the above tags
        return False
    elif 'suspicious' in this_categories:
        # this catches non-exchanges which don't have a specific suspicious tag
        return True
    return False

Now we can look at the main function. This grabs all the inflows, looks up the categories associated with inflow wallets, computes what we want to compute and then (lightly) formats the output.

https://github.com/ChainArgos/api-demos/blob/main/looker/wallet_eval/wallet_eval.py

import looker_sdk
import pandas as pd

from looker.wallet_eval.config import ADDRESS, TRANSACTIONS_FROM_ADDRESS, TOKENS_SYMBOL, \
    TRANSACTIONS_SUM_OF_TRANSFER_AMOUNTS
from looker.wallet_eval.helpers import lookup_address_categories, lookup_inflows, grab_wallet_categories
from looker.wallet_eval.is_suspicious import is_suspicious_address


# compute total flows by token and number of flows by token
def compute_total_flows(inflows):
    tokens = inflows[TOKENS_SYMBOL].unique()
    total = {}
    num = {}
    for token in tokens:
        total[token] = sum(inflows.loc[(inflows[TOKENS_SYMBOL] == token)][TRANSACTIONS_SUM_OF_TRANSFER_AMOUNTS])
        num[token] = len(inflows.loc[(inflows[TOKENS_SYMBOL] == token)][TRANSACTIONS_SUM_OF_TRANSFER_AMOUNTS])
    return total, num


# work out what fraction of which flows are from suspicious or other sorts of flows
def main():
    looker = looker_sdk.init40("../looker.ini")

    inflows = lookup_inflows(looker, ADDRESS)

    # grab inflow source addresses and labels
    inflow_addresses = inflows[TRANSACTIONS_FROM_ADDRESS].unique()
    tokens = inflows[TOKENS_SYMBOL].unique()

    # now get categories for each inflow wallet
    category_result = lookup_address_categories(looker, inflow_addresses)
    categories = grab_wallet_categories(inflow_addresses, category_result)

    # we are now in a position to determine which inflows are suspicious, sanctioned or otherwise
    sus_addresses = []
    for address in inflow_addresses:
        if is_suspicious_address(address, inflows, categories):
            sus_addresses.append(address)

    total_amounts, total_xfers = compute_total_flows(inflows)
    sus_amounts, sus_xfers = compute_total_flows(inflows[inflows[TRANSACTIONS_FROM_ADDRESS].isin(sus_addresses)])

    header = ["token", "total inflow", "flagged inflow", "fraction flagged",
              "# inflows", "# flagged inflows", "fraction inflows flagged"]
    data = []
    for token in tokens:
        total_amt = total_amounts[token]
        sus_amt = sus_amounts[token]
        total_num = total_xfers[token]
        sus_num = sus_xfers[token]
        frac = sus_amt / total_amt
        frac_num = sus_num / total_num
        res_l = [token, total_amt, sus_amt, frac, total_num, sus_num, frac_num]
        data.append(res_l)

    out_dataframe = pd.DataFrame(data, columns=header)
    print(str(out_dataframe))


main()

A range of helper functions which wrap the underlying API are here:

https://github.com/ChainArgos/api-demos/blob/main/looker/wallet_eval/helpers.py

from io import StringIO

import looker_sdk
import pandas as pd

from looker.wallet_eval.config import WALLET_LOOK_ID, INFLOWS_LOOK_ID, WALLETS_ADDRESS, WALLETS_CATEGORIES, JOINER


def lookup_address_categories(looker, address_list: str):
    look = looker.look(look_id=str(WALLET_LOOK_ID))
    query_filter = {"wallets.address": ",".join(address_list)}
    q = create_query(looker, look.query, query_filter)
    return run_query(looker, q)


def lookup_inflows(looker, address: str):
    look = looker.look(look_id=str(INFLOWS_LOOK_ID))
    query_filter = {"ethereum_txns.to_address": address}
    q = create_query(looker, look.query, query_filter)
    return run_query(looker, q)


def create_query(looker, base_query, query_filter):
    new_query = looker.create_query(body=looker_sdk.models40.WriteQuery(model=base_query.model, view=base_query.view,
                                                                        fields=base_query.fields, filters=query_filter,
                                                                        pivots=base_query.pivots,
                                                                        total=base_query.total,
                                                                        row_total=base_query.row_total))
    return new_query


def run_query(sdk, query):
    return pd.read_csv(StringIO(sdk.run_query(query_id=query.id, result_format='csv')))


def grab_wallet_categories(inflow_addresses, category_result):
    categories = {}
    for address in inflow_addresses:
        addr_rows = category_result.loc[(category_result[WALLETS_ADDRESS] == address)]
        n_rows, _ = addr_rows.shape
        if n_rows > 0:
            res = JOINER.join(addr_rows[WALLETS_CATEGORIES].unique())
            categories[address] = list(set(res.split(" ;; ")))
        else:
            categories[address] = []
    return categories