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
115
116
117
118
119
120
121
122
123
124
|
"""Whale Alert model"""
__docformat__ = "numpy"
from typing import Optional
import textwrap
import requests
import pandas as pd
import numpy as np
import gamestonk_terminal.config_terminal as cfg
class ApiKeyException(Exception):
"""Api Key Exception object"""
def __init__(self, message: str):
super().__init__(message)
self.message = message
def __str__(self) -> str:
return "ApiKeyException: %s" % self.message
def make_request(params: Optional[dict] = None) -> dict:
"""Helper methods for requests [Source: https://docs.whale-alert.io/]
Parameters
----------
params: dict
additional param
Returns
-------
dict:
response from api request
"""
api_key = cfg.API_WHALE_ALERT_KEY or ""
url = "https://api.whale-alert.io/v1/transactions?api_key=" + api_key
response = requests.get(url, params=params)
if not 200 <= response.status_code < 300:
raise ApiKeyException("Invalid Authentication: %s" % response.text)
try:
return response.json()
except Exception as e:
raise ValueError("Invalid Response: %s" % response.text) from e
def get_whales_transactions(min_value: int = 800000, limit: int = 100) -> pd.DataFrame:
"""Whale Alert's API allows you to retrieve live and historical transaction data from major blockchains.
Supported blockchain: Bitcoin, Ethereum, Ripple, NEO, EOS, Stellar and Tron. [Source: https://docs.whale-alert.io/]
Parameters
----------
min_value: int
Minimum value of trade to track.
limit: int
Limit of transactions. Max 100
Returns
-------
pd.DataFrame
Crypto wales transactions
"""
min_value = 800000 if min_value > 800000 else min_value
limit = 100 if limit > 100 else limit
params = {"limit": limit, "min_value": min_value}
response = make_request(params)
data = pd.json_normalize(response["transactions"]).sort_values(
"timestamp", ascending=False
)
data["date"] = pd.to_datetime(data["timestamp"], unit="s")
data.columns = [col.replace(".address", "") for col in data.columns]
data["to_address"] = data["to"].apply(
lambda x: "\n".join(textwrap.wrap(x, width=45)) if isinstance(x, str) else x
)
data["from_address"] = data["from"].apply(
lambda x: "\n".join(textwrap.wrap(x, width=45)) if isinstance(x, str) else x
)
data["from"] = data.apply(
lambda x: x["from.owner"]
if x["from.owner"] not in [np.nan, None, np.NaN]
else x["from.owner_type"],
axis=1,
)
data["to"] = data.apply(
lambda x: x["to.owner"]
if x["to.owner"] not in [np.nan, None, np.NaN]
else x["to.owner_type"],
axis=1,
)
data.drop(
[
"id",
"transaction_count",
"from.owner_type",
"to.owner_type",
"to.owner",
"from.owner",
"transaction_type",
"hash",
"timestamp",
],
axis=1,
inplace=True,
)
return data[
[
"date",
"symbol",
"blockchain",
"amount",
"amount_usd",
"from",
"to",
"from_address",
"to_address",
]
]
|
