diff --git a/README.md b/README.md index 23e84da6..0f5b9763 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,10 @@ ### Smart Money Follower +Smart Money Follower OG Developer: https://github.com/yllvar
+OG Repo: https://github.com/yllvar/Smart_Money_Follower

+GMGN.ai API Wrapper OG Developer: https://github.com/1f1n
+OG Repo: https://github.com/1f1n/gmgnai-wrapper + #### Overview The **Smart Money Follower** is a Python-based tool designed to analyze and follow top-performing wallets in the cryptocurrency space using the GMGN.ai API. It provides insights into wallet activities, evaluates traded tokens, and presents data in a structured format for analysis. @@ -12,24 +17,60 @@ The **Smart Money Follower** is a Python-based tool designed to analyze and foll #### Requirements - Python 3.7+ - Dependencies: - - `httpx` + - `fake-useragent` - `tabulate` - - `gmgn` (GMGN.ai API wrapper) - - `logging` + - `tls_client` + - `PyYAML` -#### Setup -1. **Installation**: - ```bash - pip install httpx tabulate gmgn +## Setup +#### 1. **Clone Git** + + ``` + git clone https://github.com/LetsStartWithPurple/Smart_Money_Follower.git ``` +#### 2. **Start Virtual Environment** -2. **Configuration**: - - Ensure you have valid API credentials for GMGN.ai. Update credentials in the `gmgn` initialization within `SmartMoneyFollower`. +Navigate to the project directory: + ```bash + cd Smart_Money_Follower + ``` +Create Venv + ```bash + python3 -m venv venv + ``` +Start Virtual Environment + ```bash + source venv/bin/activate + ``` -3. **Execution**: +#### 4. **Install Requirements**: + ```bash + pip install -r requirements.txt + ``` + +#### 5. **Execution**: ```bash python smart_money_follower.py ``` + ```bash +usage: smart_money_follower.py [-h] [--config CONFIG] [--path PATH] [--verbose VERBOSE] + [--export-format {csv,txt}] [--timeframe {1d,7d,30d}] + [--winrate WINRATE] + +Smart Money Follower Configuration + +options: + -h, --help show this help message and exit + --config CONFIG Path to the config file + --path PATH Path to export files + --verbose VERBOSE Verbose script logs + --export-format {csv,txt} + Export format (csv or txt) + --timeframe {1d,7d,30d} + Select timeframe of wallet scan + --winrate WINRATE Set winrate between 0 and 100 +``` +or you can adjust these settings in the config/config.yaml file #### Usage - Upon execution, the script fetches top wallets, analyzes their recent activities, evaluates tokens they've traded, and prints out a summarized analysis including realized profits, transaction volumes, and last activity timestamps. diff --git a/config/ConfigManager.py b/config/ConfigManager.py new file mode 100644 index 00000000..5b224f2e --- /dev/null +++ b/config/ConfigManager.py @@ -0,0 +1,131 @@ +import argparse +import yaml +import os +from yaml import YAMLError +from config.config_validators import * + + +class ConfigManager: + def __init__(self, args=None): + self._config_path = os.path.abspath(args.config) + self._config_data = self._load_config() + self._args = args + self._final_config = self._merge_config_and_args() + + def _load_config(self): + """Load the configuration file.""" + try: + with open(self._config_path, 'r') as f: + return yaml.safe_load(f) + except FileNotFoundError: + print(f"Warning: Config file '{self._config_path}' not found. Using defaults.") + return {} + except YAMLError as e: + print(f"Error: Failed to parse YAML in config file '{self._config_path}': {e}") + return {} + + def _merge_config_and_args(self): + """Merge the config file and command-line arguments, with args taking precedence.""" + wallet_settings = self._config_data.get("wallet_settings", {}) + return { + "path": validate_path( + self._args.path if self._args and self._args.path else self._config_data.get("path", "data") + ), + "verbose": validate_verbose( + self._args.verbose if self._args and self._args.verbose else self._config_data.get("verbose", True) + ), + "export_format": validate_export_format( + self._args.export_format if self._args and self._args.export_format else self._config_data.get("export_format", "csv") + ), + "timeframe": validate_timeframe( + self._args.timeframe if self._args and self._args.timeframe else wallet_settings.get("timeframe", "7d") + ), + "wallet_tag": validate_wallet_tag( + wallet_settings.get("wallet_tag", "smart_degen") + ), + "win_rate": validate_win_rate( + self._args.winrate if self._args and self._args.winrate else wallet_settings.get("win_rate", 60) + ) + } + + @property + def path(self): + return self._final_config["path"] + + @path.setter + def path(self, new_path): + self._final_config["path"] = validate_path(new_path) + + @property + def verbose(self): + return self._final_config["verbose"] + + @verbose.setter + def verbose(self, verbose): + self._final_config["verbose"] = validate_verbose(verbose) + + @property + def export_format(self): + return self._final_config["export_format"] + + @export_format.setter + def export_format(self, export_format): + self._final_config["export_format"] = validate_export_format(export_format) + + @property + def timeframe(self): + return self._final_config["timeframe"] + + @timeframe.setter + def timeframe(self, timeframe): + self._final_config["timeframe"] = validate_timeframe(timeframe) + + @property + def wallet_tag(self): + return self._final_config["wallet_tag"] + + @wallet_tag.setter + def wallet_tag(self, wallet_tag): + self._final_config["wallet_tag"] = validate_wallet_tag(wallet_tag) + + @property + def win_rate(self): + return self._final_config["win_rate"] + + @win_rate.setter + def win_rate(self, win_rate): + self._final_config["win_rate"] = validate_win_rate(win_rate) + + @property + def config(self): + return self._final_config + + +def parse_args(): + """Parse command-line arguments.""" + parser = argparse.ArgumentParser(description="Smart Money Follower Configuration") + parser.add_argument("--config", type=str, default="config/config.yaml", help="Path to the config file") + parser.add_argument("--path", type=str, help="Path to export files") + parser.add_argument("--verbose", type=bool, help="Verbose script logs") + parser.add_argument("--export-format", type=str, choices=["csv", "txt"], help="Export format (csv or txt)") + parser.add_argument("--timeframe", type=str, choices=["1d", "7d", "30d"], help="Select timeframe of wallet scan") + parser.add_argument("--winrate", type=int, help="Set winrate between 0 and 100") + return parser.parse_args() + + +if __name__ == "__main__": + # Parse arguments + args = parse_args() + #args.config = "config.yaml" # for testing purposes + + # Create ConfigManager instance + config_manager = ConfigManager(args=args) + + # Access properties + print("Final Configuration:") + print(f"Path: {config_manager.path}") + print(f"Verbose: {config_manager.verbose}") + print(f"Export Format: {config_manager.export_format}") + print(f"Timeframe: {config_manager.timeframe}") + print(f"Wallet Tag: {config_manager.wallet_tag}") + print(f"Win Rate: {config_manager.win_rate}") \ No newline at end of file diff --git a/config/__init__.py b/config/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/config/config.yaml b/config/config.yaml new file mode 100644 index 00000000..d984414f --- /dev/null +++ b/config/config.yaml @@ -0,0 +1,26 @@ +#### Path to export wallet data +path: "data" + +#### Export settings +# csv or txt +export_format: "csv" + +#### Verbose Settings +verbose: False + +#### Wallet Search settings +# timeframe options - 1d, 7d, 30d +# +# wallet tags options - +# all = all tags +# pump_smart = Pump.Fun Smart Money +# smart_degen = Smart Money (default) +# reowned = KOL/VC/Influencer +# snipe_bot = Snipe Bot +# +# winrate - set 0 to 100 (default is 60) + +wallet_settings: + timeframe: "7d" + wallet_tag: "smart_degen" + win_rate: 60 diff --git a/config/config_validators.py b/config/config_validators.py new file mode 100644 index 00000000..41a20df7 --- /dev/null +++ b/config/config_validators.py @@ -0,0 +1,34 @@ +def validate_path(path): + if not isinstance(path, str): + raise ValueError("Path must be a string") + return path + +def validate_verbose(verbose): + if not isinstance(verbose, bool): + raise ValueError("Verbose must be a boolean") + return verbose + +def validate_export_format(export_format): + valid_formats = ["csv", "txt"] + if export_format not in valid_formats: + raise ValueError("Export format must be 'csv' or 'txt'") + return export_format + +def validate_timeframe(timeframe): + valid_timeframes = ["1d", "7d", "30d"] + if timeframe not in valid_timeframes: + raise ValueError("Timeframe must be '1d', '7d', or '30d'.") + return timeframe + +def validate_wallet_tag(wallet_tag): + valid_tags = ["all", "pump_smart", "smart_degen", "reowned", "snipe_bot"] + if wallet_tag not in valid_tags: + raise ValueError(f"Wallet tag must be one of {valid_tags}") + return wallet_tag + +def validate_win_rate(win_rate): + if not isinstance(win_rate, int): + raise ValueError("Win Rate must be an integer") + elif not (0 <= win_rate <= 100): + raise ValueError("Win Rate must be between 0 and 100") + return win_rate / 100 \ No newline at end of file diff --git a/gmgn/__init__.py b/gmgn/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/gmgn/client.py b/gmgn/client.py new file mode 100644 index 00000000..4240bb4e --- /dev/null +++ b/gmgn/client.py @@ -0,0 +1,272 @@ +import random +import platform +import tls_client +from fake_useragent import UserAgent + +# author - 1f1n +# date - 05/06/2024 + +class gmgn: + BASE_URL = "https://gmgn.ai/defi/quotation" + + def __init__(self): + pass + + def randomiseRequest(self): + # Randomly choose a valid browser identifier + self.identifier = random.choice( + [browser for browser in tls_client.settings.ClientIdentifiers.__args__ if + browser.startswith(('chrome', 'safari', 'firefox', 'opera'))] + ) + self.sendRequest = tls_client.Session(random_tls_extension_order=True, client_identifier=self.identifier) + + # Extract parts of the identifier + parts = self.identifier.split('_') + identifier, version, *rest = parts + other = rest[0] if rest else None + + # Detect OS dynamically + current_os = platform.system().lower() + if current_os == 'darwin': # macOS + os_type = 'mac' + elif current_os == 'linux': + os_type = 'linux' + else: + os_type = 'windows' + + # Adjust identifier and os_type for specific cases + if identifier == 'opera': + identifier = 'chrome' + elif version == 'ios': + os_type = 'ios' + + self.user_agent = UserAgent(browsers=[identifier], os=[os_type]).random + + self.headers = { + 'Host': 'gmgn.ai', + 'accept': 'application/json, text/plain, */*', + 'accept-language': 'fr-FR,fr;q=0.9,en-US;q=0.8,en;q=0.7', + 'dnt': '1', + 'priority': 'u=1, i', + 'referer': 'https://gmgn.ai/?chain=sol', + 'user-agent': self.user_agent + } + + + def getTokenInfo(self, contractAddress: str) -> dict: + """ + Gets info on a token. + """ + self.randomiseRequest() + if not contractAddress: + return "You must input a contract address." + url = f"{self.BASE_URL}/v1/tokens/sol/{contractAddress}" + + request = self.sendRequest.get(url, headers=self.headers) + + jsonResponse = request.json() + + return jsonResponse + + def getNewPairs(self, limit: int = None) -> dict: + """ + Limit - Limits how many tokens are in the response. + """ + self.randomiseRequest() + if not limit: + limit = 50 + elif limit > 50: + return "You cannot have more than check more than 50 pairs." + + url = f"{self.BASE_URL}/v1/pairs/sol/new_pairs?limit={limit}&orderby=open_timestamp&direction=desc&filters[]=not_honeypot" + + request = self.sendRequest.get(url, headers=self.headers) + + jsonResponse = request.json()['data'] + + return jsonResponse + + def getTrendingWallets(self, timeframe: str = None, walletTag: str = None) -> dict: + """ + Gets a list of trending wallets based on a timeframe and a wallet tag. + + Timeframes\n + 1d = 1 Day\n + 7d = 7 Days\n + 30d = 30 days\n + + ---------------- + + Wallet Tags\n + pump_smart = Pump.Fun Smart Money\n + smart_degen = Smart Money\n + reowned = KOL/VC/Influencer\n + snipe_bot = Snipe Bot\n + + """ + self.randomiseRequest() + if not timeframe: + timeframe = "7d" + if not walletTag: + walletTag = "smart_degen" + + url = f"{self.BASE_URL}/v1/rank/sol/wallets/{timeframe}?tag={walletTag}&orderby=pnl_{timeframe}&direction=desc" + + request = self.sendRequest.get(url, headers=self.headers) + + jsonResponse = request.json()['data'] + + return jsonResponse + + def getTrendingTokens(self, timeframe: str = None) -> dict: + """ + Gets a list of trending tokens based on a timeframe. + + Timeframes\n + 1m = 1 Minute\n + 5m = 5 Minutes\n + 1h = 1 Hour\n + 6h = 6 Hours\n + 24h = 24 Hours\n + """ + timeframes = ["1m", "5m", "1h", "6h", "24h"] + self.randomiseRequest() + if timeframe not in timeframes: + return "Not a valid timeframe." + + if not timeframe: + timeframe = "1h" + + if timeframe == "1m": + url = f"{self.BASE_URL}/v1/rank/sol/swaps/{timeframe}?orderby=swaps&direction=desc&limit=20" + else: + url = f"{self.BASE_URL}/v1/rank/sol/swaps/{timeframe}?orderby=swaps&direction=desc" + + request = self.sendRequest.get(url, headers=self.headers) + + jsonResponse = request.json()['data'] + + return jsonResponse + + def getTokensByCompletion(self, limit: int = None) -> dict: + """ + Gets tokens by their bonding curve completion progress.\n + + Limit - Limits how many tokens in the response. + """ + self.randomiseRequest() + if not limit: + limit = 50 + elif limit > 50: + return "Limit cannot be above 50." + + url = f"{self.BASE_URL}/v1/rank/sol/pump?limit={limit}&orderby=progress&direction=desc&pump=true" + + request = self.sendRequest.get(url, headers=self.headers) + + jsonResponse = request.json()['data'] + + return jsonResponse + + def findSnipedTokens(self, size: int = None) -> dict: + """ + Gets a list of tokens that have been sniped.\n + + Size - The amount of tokens in the response + """ + self.randomiseRequest() + if not size: + size = 10 + elif size > 39: + return "Size cannot be more than 39" + + url = f"{self.BASE_URL}/v1/signals/sol/snipe_new?size={size}&is_show_alert=false&featured=false" + + request = self.sendRequest.get(url, headers=self.headers) + + jsonResponse = request.json()['data'] + + return jsonResponse + + def getGasFee(self): + """ + Get the current gas fee price. + """ + self.randomiseRequest() + url = f"{self.BASE_URL}/v1/chains/sol/gas_price" + + request = self.sendRequest.get(url, headers=self.headers) + + jsonResponse = request.json()['data'] + + return jsonResponse + + def getTokenUsdPrice(self, contractAddress: str = None) -> dict: + """ + Get the realtime USD price of the token. + """ + self.randomiseRequest() + if not contractAddress: + return "You must input a contract address." + + url = f"{self.BASE_URL}/v1/sol/tokens/realtime_token_price?address={contractAddress}" + + request = self.sendRequest.get(url, headers=self.headers) + + jsonResponse = request.json()['data'] + + return jsonResponse + + def getTopBuyers(self, contractAddress: str = None) -> dict: + """ + Get the top buyers of a token. + """ + self.randomiseRequest() + if not contractAddress: + return "You must input a contract address." + + url = f"{self.BASE_URL}/v1/tokens/top_buyers/sol/{contractAddress}" + + request = self.sendRequest.get(url, headers=self.headers) + + jsonResponse = request.json()['data'] + + return jsonResponse + + def getSecurityInfo(self, contractAddress: str = None) -> dict: + """ + Gets security info about the token. + """ + self.randomiseRequest() + if not contractAddress: + return "You must input a contract address." + + url = f"{self.BASE_URL}/v1/tokens/security/sol/{contractAddress}" + + request = self.sendRequest.get(url, headers=self.headers) + + jsonResponse = request.json()['data'] + + return jsonResponse + + def getWalletInfo(self, walletAddress: str = None, period: str = None) -> dict: + """ + Gets various information about a wallet address. + + Period - 7d, 30d - The timeframe of the wallet you're checking. + """ + self.randomiseRequest() + periods = ["7d", "30d"] + + if not walletAddress: + return "You must input a wallet address." + if not period or period not in periods: + period = "7d" + + url = f"{self.BASE_URL}/v1/smartmoney/sol/walletNew/{walletAddress}?period={period}" + + request = self.sendRequest.get(url, headers=self.headers) + + jsonResponse = request.json()['data'] + + return jsonResponse diff --git a/gmgn/scripts/__init__.py b/gmgn/scripts/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/gmgn/scripts/getNewPairs.py b/gmgn/scripts/getNewPairs.py new file mode 100644 index 00000000..38f1dd70 --- /dev/null +++ b/gmgn/scripts/getNewPairs.py @@ -0,0 +1,7 @@ +from ..client import gmgn + +gmgn = gmgn() + +getNewPairs = gmgn.getNewPairs(limit=1) + +print(getNewPairs) \ No newline at end of file diff --git a/gmgn/scripts/getTokenInfo.py b/gmgn/scripts/getTokenInfo.py new file mode 100644 index 00000000..f99e3328 --- /dev/null +++ b/gmgn/scripts/getTokenInfo.py @@ -0,0 +1,7 @@ +from ..client import gmgn + +gmgn = gmgn() + +getTokenInfo = gmgn.getTokenInfo(contractAddress="9eLRcHw2G4Ugrnp1p5165PuZsQ2YSc9GnBpGZS7Cpump") + +print(getTokenInfo) \ No newline at end of file diff --git a/gmgn/scripts/getTrendingWallets.py b/gmgn/scripts/getTrendingWallets.py new file mode 100644 index 00000000..fadc3802 --- /dev/null +++ b/gmgn/scripts/getTrendingWallets.py @@ -0,0 +1,7 @@ +from ..client import gmgn + +gmgn = gmgn() + +getTrendingWallets = gmgn.getTrendingWallets(timeframe="7d", walletTag="smart_degen") + +print(getTrendingWallets) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 00000000..d7f3fab1 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +tls_client +httpx +tabulate +fake-useragent +PyYAML \ No newline at end of file diff --git a/src/fetch_wallet.py b/smart_money_follower.py similarity index 66% rename from src/fetch_wallet.py rename to smart_money_follower.py index 0950abc3..8f5d12fe 100644 --- a/src/fetch_wallet.py +++ b/smart_money_follower.py @@ -1,27 +1,32 @@ -import httpx import logging import time from datetime import datetime -from tabulate import tabulate -from gmgn import gmgn +from tabulate import tabulate +from gmgn.client import gmgn +import csv +import os +from config.ConfigManager import ConfigManager, parse_args class SmartMoneyFollower: - def __init__(self): + def __init__(self, config: ConfigManager): self.gmgn = gmgn() + self.config = config self.logger = logging.getLogger("SmartMoneyFollower") - logging.basicConfig(level=logging.INFO) + logging.basicConfig(level=logging.INFO if self.config.verbose else logging.WARNING) + self.export_path = self.config.path + self.export_format = self.config.export_format - def get_top_wallets(self, timeframe="7d", walletTag="smart_degen"): + def get_top_wallets(self, timeframe="7d", wallet_tag="smart_degen"): """ Fetch top performing wallets using the getTrendingWallets endpoint. :param timeframe: Time period for trending wallets (default "7d"). - :param walletTag: Tag to filter wallets (default "smart_degen"). + :param wallet_tag: Tag to filter wallets (default "smart_degen"). :return: List of top performing wallets. """ try: - response = self.gmgn.getTrendingWallets(timeframe, walletTag) + response = self.gmgn.getTrendingWallets(timeframe, wallet_tag) return response['rank'] except Exception as e: self.logger.error(f"Error fetching top wallets: {e}") @@ -86,7 +91,7 @@ def run_strategy(self): """ try: # Step 1: Get top wallets - top_wallets = self.get_top_wallets() + top_wallets = self.get_top_wallets(timeframe=self.config.timeframe, wallet_tag=self.config.wallet_tag) if not top_wallets: self.logger.warning("No top wallets found.") return @@ -96,7 +101,7 @@ def run_strategy(self): # Step 2: Analyze each wallet's activity for wallet in top_wallets: wallet_address = wallet.get('wallet_address') - wallet_activity = self.analyze_wallet_activity(wallet_address) + wallet_activity = self.analyze_wallet_activity(wallet_address, period=self.config.timeframe) # Log wallet activity data vertically self.logger.info(f"Wallet Activity for {wallet_address}:") @@ -105,7 +110,7 @@ def run_strategy(self): # Filter wallets with a win rate higher than 0.6 winrate = wallet_activity.get('winrate', 0) - if winrate is not None and winrate > 0.6: + if winrate is not None and winrate >= self.config.win_rate: wallet_info = { 'wallet_address': wallet_address, 'realized_profit': wallet_activity.get('realized_profit', 'N/A'), @@ -125,12 +130,56 @@ def run_strategy(self): time.sleep(1) # Rate limiting - # Step 4: Print the analysis output + # Step 4: Export to file + self.export_data(wallet_data) + + # Step 5: Print the analysis output self.print_analysis_output(wallet_data) + except Exception as e: self.logger.error(f"Error running strategy: {e}") + def export_data(self, data): + """ + Export the wallet analysis data to the specified format. + + :param data: List of wallet data dictionaries. + """ + file_path = "" + if not data: + self.logger.warning("No data to export") + return + + os.makedirs(self.export_path, exist_ok=True) + + if self.export_format == "csv": + file_path = os.path.join(self.export_path, f"wallet_list_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv") + with open(file_path, mode="w", newline="", encoding="utf-8") as file: + writer = csv.DictWriter(file, fieldnames=data[0].keys()) + writer.writeheader() + writer.writerows(data) + elif self.export_format == "txt": + file_path = os.path.join(self.export_path, f"wallet_list_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt") + with open(file_path, mode="w") as file: + for wallet in data: + for key, value in wallet.items(): + file.write(f"{key}: {value}\n") + file.write("\n") + + print(f"Data exported to {file_path if file_path else self.export_path}") if __name__ == "__main__": - follower = SmartMoneyFollower() - follower.run_strategy() + try: + # Parse command-line arguments and initiate ConfigManager + args = parse_args() + manager = ConfigManager(args) + + # Follower instance + follower = SmartMoneyFollower(manager) + + # Run + follower.run_strategy() + + except ValueError as e: + print(f"Configuration Error: {e}") + exit(1) \ No newline at end of file