Plotting the Network of Secondary Liquidity of Tokens

secondary_liquidity_of_tokens.py

"""
Analyze and visualize Ethereum contract transactions.
It connects to the Ethereum mainnet, retrieves transaction logs, and creates an interactive graph showing
transaction history between various addresses. The script supports command-line arguments for customization such
as specifying contract addresses, enabling circular layouts, and excluding certain types of addresses.

Author: Alex McFarlane <[email protected]>
License: MIT

Can be called like as follows (example is USDY):
```sh
python secondary_liquidity_of_tokens.py -a 0xea0f7eebdc2ae40edfe33bf03d332f8a7f617528 -p 0x96F6eF951840721AdBF46Ac996b59E0235CB985C -m 0x0000000000000000000000000000000000000000 --show_delegate --show_internal_team --circular_layout
```
"""
import matplotlib.pyplot as plt
import numpy as np
import networkx as nx
import requests
import json
import brownie
from typing import List, Optional

import matplotlib.patches as mpatches

try:
    brownie.network.connect("mainnet")
except ConnectionError as err:
    if 'Already connected to network' in str(err):
        pass
    else:
        raise err


def network_plot(
        token_abi_address: str,
        token_proxy_address: str,
        mint_from_address: str = "0x0000000000000000000000000000000000000000",
        show_delegate: bool = True,
        show_internal_team: bool = False,
        circular_layout: bool = True,
        internal_team: List[str] = [],
        output_file: Optional[str] = None
    ):
    """
    Produces a network plot of token transfers.

    Args:
        token_abi_address: REAL abi address, usually located behind proxy.
        token_proxy_address: Token address that people interact with.
        mint_from_address: The address tokens are minted from. Must be included in Transfer event logs.
        show_delegate: If True, tokens delegated to other addresses will be shown.
        show_internal_team: If True, internal team trades will be shown. Helps filter out non-essential transfers.
        circular_layout: If True, layout will be set to circular. Recommended for initial identification of delegates.
        output_file: Save plot to this file.

    Example:
        >>> network_plot(
        ...     token_abi_address='0xea0f7eebdc2ae40edfe33bf03d332f8a7f617528',
        ...     token_proxy_address='0x96F6eF951840721AdBF46Ac996b59E0235CB985C',
        ...     mint_from_address='0x0000000000000000000000000000000000000000',
        ...     show_delegate=True,
        ...     show_internal_team=False,
        ...     circular_layout=True)
    """

    # Note you can only call this every 5 secs
    abi_url = f"https://api.etherscan.io/api?module=contract&action=getabi&address={token_abi_address}"
    response = requests.get(abi_url)
    abi = json.loads(response.json()['result'])

    contract = brownie.Contract.from_abi("", token_proxy_address, abi)
    logs = contract.events.Transfer.getLogs(fromBlock=0)

    # Initiate Graph
    G = nx.MultiDiGraph()

    for entry in logs:
        from_address = entry['args']['from']
        to_address = entry['args']['to']
        
        if G.has_edge(from_address, to_address):
            G[from_address][to_address][0]['summed_value'] += entry['args']['value']
            G[from_address][to_address][0]['count'] += 1
        else:
            G.add_edge(from_address, to_address, summed_value=entry['args']['value'], count=1)

    # Ignore delegate
    rename_dict = {n:n[:6]+'...'+n[-3:]for n in G.nodes()}

    # Relabel
    G = nx.relabel_nodes(G, rename_dict)

    target_node = mint_from_address
    target_node = rename_dict[target_node]
    void = "0x0000000000000000000000000000000000000000"
    void = rename_dict[void]
    # Replace internal_team = {} with below line:
    internal_team = set(internal_team)

    # Lists to hold color and line width values
    color_dict = {'mint': 'lightgreen', 'both': 'orange', 'redeem': 'salmon', 'none': 'grey'}
    node_color_dict = {'mint': 'green', 'both': 'orange', 'redeem': 'red', 'none': 'skyblue'}
    node_directions = {n: 'none' for n in G.nodes()}
    widths = []

    width_on = 'count'

    # determine mind/redeemers on original graph
    for (node1, node2, edge_data) in G.edges(data=True):
        if target_node in [node1, node2]:
            if target_node == node1:
                if node_directions[node2] == 'none':
                    node_directions[node2] = "mint"
                elif node_directions[node2] == 'redeem':
                    node_directions[node2] = "both"
            
            if target_node == node2:
                if node_directions[node1] == 'none':
                    node_directions[node1] = "redeem"
                elif node_directions[node1] == 'mint':
                    node_directions[node1] = "both"

    graph = G.copy()
    if not show_delegate:
        graph.remove_node(target_node)
    if not show_internal_team:
        for node in internal_team - set([target_node]):
            graph.remove_node(node)
        graph.remove_node(void)

    line_colors = []
    for (node1, node2, edge_data) in graph.edges(data=True):
        if target_node in [node1, node2]:
            if target_node == node1:
                line_colors.append(color_dict[node_directions[node2]])
            else:
                line_colors.append(color_dict[node_directions[node1]])
        else:
            line_colors.append('grey')
        widths.append(np.log(float(1 + edge_data[width_on])))

    node_colors = []
    for node in graph.nodes():
        if node == void:
            c = 'black'
        elif node == target_node:
            c = 'blue'
        elif node in internal_team:
            c = 'purple'
        else:
            c = node_color_dict[node_directions[node]]
        node_colors.append(c)


    # Normalize widths to reasonable range for visualization
    max_width = max(widths)
    edges_with_target = [(node1, node2, edge_data) for (node1, node2, edge_data) in graph.edges(data=True) if target_node in [node1, node2]]
    edges_without_target = [(node1, node2, edge_data) for (node1, node2, edge_data) in graph.edges(data=True) if target_node not in [node1, node2]]
    line_colors_with_target = [color for color, (node1, node2, edge_data) in zip(line_colors, graph.edges(data=True)) if target_node in [node1, node2]]
    line_colors_without_target = [color for color, (node1, node2, edge_data) in zip(line_colors, graph.edges(data=True)) if target_node not in [node1, node2]]

    n_widths_with_target = [0.5 + 2 * np.log(float(1 + edge_data[width_on])) / max_width for (node1, node2, edge_data) in edges_with_target]
    n_widths_without_target = [0.5 + 2 * np.log(float(1 + edge_data[width_on])) / max_width for (node1, node2, edge_data) in edges_without_target]

    if circular_layout:
        pos = nx.circular_layout(graph)
    else:
        pos = nx.spring_layout(graph)

    fig, ax = plt.subplots(figsize=(10,10))

    nx.draw_networkx_nodes(graph, pos, node_color=node_colors, node_size=50, ax=ax)  # Decreased node_size to make nodes smaller
    nx.draw_networkx_labels(graph, pos, font_size=5, ax=ax)  # Decrease label size.

    # Drawing the edges
    nx.draw_networkx_edges(graph, pos,
                        edgelist=edges_with_target,
                        width=n_widths_with_target,
                        edge_color=line_colors_with_target, 
                        arrowstyle='-|>',
                        ax=ax)

    nx.draw_networkx_edges(graph, pos,
                        edgelist=edges_without_target,
                        width=n_widths_without_target,
                        edge_color=line_colors_without_target, 
                        arrowstyle='-|>',
                        connectionstyle='arc3,rad=0.3',
                        ax=ax)

    fig.suptitle(f'{contract.name()} Transaction History', fontsize=20)
    ax.set_title('Lines are transactions made, thickness by log(txn count)')
    ax.axis('off')

    # Define colors for the legend
    p_redeem = mpatches.Patch(color='red', label='Redeem only')
    p_mint = mpatches.Patch(color='green', label='Mint only')
    p_both = mpatches.Patch(color='orange', label='Redeem / mint')
    p_team = mpatches.Patch(color='purple', label='Internal')
    p_delegate = mpatches.Patch(color='blue', label='Delegate')
    p_void = mpatches.Patch(color='black', label='Void')

    ax.legend(handles=[p_redeem, p_mint, p_both, p_team, p_delegate, p_void], title="Node Key")

    fig.tight_layout()
    if output_file:
        plt.savefig(output_file)
    else:
        plt.show()

if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="Analyze Ethereum Contracts")
    parser.add_argument("-a", "--token_abi_address", required=True,
                        help="Address with REAL abi (usually is behind proxy)")
    parser.add_argument("-p", "--token_proxy_address", required=True,
                        help="This is the Token address that people interact with")
    parser.add_argument("-m", "--mint_from_address", required=True,
                        help="This is the address tokens are minted from; this may be a delegate address but it needs to be in the Transfer event logs")
    parser.add_argument("--show_delegate", action='store_true',
                        help="Some RWA protocols delegate to other addresses to distribute assets - currently I only support one address in this code",
                        default=False)
    parser.add_argument("--show_internal_team", action='store_true',
                        help="Some RWA protocols have clear internal team trades from testing so if we have logic for capturing that we can filter out transfers to their mum, dad and wife.",
                        default=True)
    parser.add_argument("--circular_layout", action='store_true',
                        help="Would  advise using circular layout at first because it helps identify any delegates",
                        default=True)
    parser.add_argument("-i", "--internal_team", required=False,
                        help="Comma separated list of internal team addresses",
                        default="")
    parser.add_argument("-o", "--output_file", required=False,
                        help="Path to output file. If provided, saves the plot to this file.",
                        default="")
 
    args = parser.parse_args()

    internal_team_list = args.internal_team.split(',') if args.internal_team else []

    network_plot(
        args.token_abi_address,
        args.token_proxy_address,
        args.mint_from_address,
        args.show_delegate,
        args.show_internal_team, 
        args.circular_layout,
        internal_team_list,
        args.output_file
    )