Crypto_AI_agent/blockchain_metrics.py at main · biserd/Crypto_AI_agent · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
import os
import logging
import requests
from datetime import datetime, timedelta

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

class EtherscanClient:
    def __init__(self):
        self.api_key = os.environ.get('ETHERSCAN_API_KEY')
        self.base_url = "https://api.etherscan.io/api"

    def get_address_balance(self, address):
        """Get ETH balance for an address"""
        try:
            params = {
                'module': 'account',
                'action': 'balance',
                'address': address,
                'tag': 'latest',
                'apikey': self.api_key
            }
            response = requests.get(self.base_url, params=params)
            response.raise_for_status()
            data = response.json()
            if data['status'] == '1':
                return int(data['result']) / 1e18  # Convert from Wei to ETH
            return None
        except Exception as e:
            logger.error(f"Error fetching address balance: {str(e)}")
            return None

    def get_daily_transactions(self, days=7):
        """Get daily transaction count for last n days"""
        try:
            # Calculate the date range
            end_date = datetime.utcnow()
            start_date = end_date - timedelta(days=days)

            params = {
                'module': 'stats',
                'action': 'dailytx',
                'startdate': int(start_date.timestamp()),
                'enddate': int(end_date.timestamp()),
                'sort': 'asc',
                'apikey': self.api_key
            }

            response = requests.get(self.base_url, params=params)
            if response.status_code != 200:
                logger.error(f"Error response from Etherscan: {response.status_code}")
                return []

            data = response.json()

            if data['status'] == '1' and data['result']:
                # Format the data for the chart
                return [
                    {
                        'date': datetime.fromtimestamp(int(tx['unixTimeStamp'])).strftime('%Y-%m-%d'),
                        'value': int(tx['transactionCount'])
                    }
                    for tx in data['result']
                ]
            logger.warning("No transaction data received from Etherscan")
            return []
        except Exception as e:
            logger.error(f"Error fetching daily transactions: {str(e)}")
            return []

    def get_network_hash_rate(self):
        """Get current network hash rate"""
        try:
            params = {
                'module': 'stats',
                'action': 'ethsupply',
                'apikey': self.api_key
            }

            response = requests.get(self.base_url, params=params)
            response.raise_for_status()
            data = response.json()

            if data['status'] == '1':
                return data['result']
            return None
        except Exception as e:
            logger.error(f"Error fetching network hash rate: {str(e)}")
            return None

    def get_gas_oracle(self):
        """Get current gas prices"""
        try:
            params = {
                'module': 'gastracker',
                'action': 'gasoracle',
                'apikey': self.api_key
            }

            response = requests.get(self.base_url, params=params)
            if response.status_code != 200:
                logger.error(f"Error response from Etherscan gas oracle: {response.status_code}")
                return None

            data = response.json()

            if data['status'] == '1' and data['result']:
                return data['result']
            logger.warning("No gas oracle data received from Etherscan")
            return None
        except Exception as e:
            logger.error(f"Error fetching gas oracle: {str(e)}")
            return None