from collections import defaultdict from clickhouse_driver import Client import sys import os sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) from API.coingeckoAPI import CoingeckoAPI from API.jupAPI import JupAPI from utils.config import * from utils.ClickHouseManager import clickhouse_manager from typing import Optional, Dict, List, Any #============================================================================================================================================= #DATABASE AND REQUESTS #============================================================================================================================================= def get_wallet_transactions_from_DB(wallet): return clickhouse_manager.get_wallet_transactions(wallet) def add_transactions_to_DB(transactions): if not transactions: print("⚠️ Нет транзакций для сохранения") return False try: result = clickhouse_manager.insert_transactions(transactions) if result: print(f"✅ Сохранено {len(transactions)} транзакций в базу данных") return result except Exception as e: print(f"❌ Ошибка при сохранении транзакций: {e}") return False def get_tokens_migration_time_and_creator_from_DB(tokens): if not tokens: return [] client = Client('localhost', database='tokens_info') result = [] try: for token in tokens: query = """ SELECT token_address, creator, migrated FROM tokens WHERE token_address = %(token)s """ rows = client.execute(query, {'token': token}) if rows: row = rows[0] result.append({ "token": row[0], "creator": row[1], "migtime": row[2] or 0 }) else: result.append({ "token": token, "creator": None, "migtime": 0 }) finally: client.disconnect() return result def get_token_migration_time_and_creator_from_JUP(token): pool_info = JupAPI.get_token_pool_info(token) migration_time = pool_info["migration_started"] creator = pool_info["creator_hash"] return migration_time, creator def get_tokens_migration_time_and_creator(tokens): token_data = get_tokens_migration_time_and_creator_from_DB(tokens) token_data_dict = {entry["token"]: entry for entry in token_data} result = [] for token in tokens: entry = token_data_dict.get(token) if entry is None or entry["creator"] is None or entry["migtime"] == 0: migration_time, creator = get_token_migration_time_and_creator_from_JUP(token) if creator: if entry is None: new_migtime = migration_time if migration_time else 1 result.append({"token": token, "creator": creator, "migtime": new_migtime}) else: if migration_time: entry["creator"] = creator entry["migtime"] = migration_time else: entry["creator"] = creator entry["migtime"] = 1 result.append(entry) else: if entry: result.append(entry) else: result.append(entry) return result def get_ath_of_token_with_JUP(token): all_candles = JupAPI().get_all_candles(token, interval="1_DAY") ath_candle = max(all_candles, key=lambda c: c["high"]) ath_info = { "ath_price": int(ath_candle["high"]), "ath_time" : int(ath_candle["time"]) } return ath_info #============================================================================================================================================= #CALCULATIONS #============================================================================================================================================= def get_market_cap(transaction): delta_sol = transaction.get("delta_sol", 0) delta_token = transaction.get("delta_token", 1) timestamp = transaction.get("blocktime") if delta_token == 0: return 0 sol_price = CoingeckoAPI.get_sol_price_on_timestamp(timestamp) mc = abs(delta_sol / delta_token) * sol_price * 1_000_000_000 return mc def get_profit_percent_to_transactions(transactions): fifo_queue = [] for tx in sorted(transactions, key=lambda x: x.get("blocktime", 0)): delta_token = tx.get("delta_token", 0) delta_sol = tx.get("delta_sol", 0) if tx.get("type") == "buy" and delta_token > 0: amount = delta_token price = abs(delta_sol) / amount if amount != 0 else 0 fifo_queue.append({"amount": amount, "price": price}) tx["profit"] = None elif tx.get("type") == "sell" and delta_token < 0: sold_amount = abs(delta_token) total_cost = 0.0 while sold_amount > 0 and fifo_queue: lot = fifo_queue[0] used_amount = min(sold_amount, lot["amount"]) total_cost += used_amount * lot["price"] lot["amount"] -= used_amount sold_amount -= used_amount if lot["amount"] == 0: fifo_queue.pop(0) revenue = abs(delta_sol) tx["profit"] = (revenue - total_cost) / total_cost * 100 if total_cost != 0 else None return transactions def get_token_buys_in_valid_mc(transactions): return any( tx.get("type") == "buy" and MIN_MARKET_CAP <= tx.get("MC", 0) <= MAX_MARKET_CAP for tx in transactions ) def count_all_buys_in_valid_mc(wallet_result): """ Считает общее количество покупок в валидном MC по всем токенам кошелька Args: wallet_result: результат анализа кошелька с транзакциями Returns: int: общее количество покупок в валидном MC """ if not wallet_result or not wallet_result.get("tokens"): return 0 total_buys = 0 for token_entry in wallet_result.get("tokens", []): for tx in token_entry.get("transactions", []): if (tx.get("type") == "buy" and MIN_MARKET_CAP <= tx.get("MC", 0) <= MAX_MARKET_CAP): total_buys += 1 return total_buys def get_first_hold_time(transactions): """ Рассчитывает время удержания токена от первой покупки до первой полной распродажи (обнуления баланса). Игнорирует все транзакции после первой полной распродажи. Логика идентична mig_db.py Возвращает: int: время удержания в секундах 'hold': если продаж не было и баланс не обнулился None: если покупок не было """ transactions = sorted(transactions, key=lambda x: x.get('blocktime', 0)) balance = 0 first_buy_time = None first_full_sell_time = None for tx in transactions: if tx.get('type') == 'buy': if first_buy_time is None: first_buy_time = tx.get('blocktime') balance += tx.get('delta_token', 0) elif tx.get('type') == 'sell': balance -= abs(tx.get('delta_token', 0)) if balance <= 0 and first_buy_time is not None: first_full_sell_time = tx.get('blocktime') break # прекращаем подсчет после первой полной распродажи if first_buy_time is None: return None if first_full_sell_time is None: return 'hold' return first_full_sell_time - first_buy_time def get_tokens_hold_befor_migrations(wallet_result, tokens_info): if not wallet_result or not tokens_info: return 0.0 tokens_info_dict = {item["token"]: item for item in tokens_info} held_before_migration = 0 bought_before_migration = 0 for token_entry in wallet_result.get("tokens", []): if not token_entry.get("buy_in_valid_mc"): continue token = token_entry["token"] migration_time = tokens_info_dict.get(token, {}).get("migtime", 0) if migration_time <= 1: continue balance = 0 has_valid_buy = False for tx in sorted(token_entry["transactions"], key=lambda x: x["blocktime"]): if tx["blocktime"] > migration_time: break if tx["type"] == "buy" and MIN_MARKET_CAP <= tx["MC"] <= MAX_MARKET_CAP: balance += tx["delta_token"] has_valid_buy = True elif tx["type"] == "sell": balance -= abs(tx["delta_token"]) if has_valid_buy: bought_before_migration += 1 if balance > 0: held_before_migration += 1 return (held_before_migration / bought_before_migration) * 100 if bought_before_migration > 0 else 0.0 #============================================================================================================================================= #MAIN FUNCTIONS FOR ANALYZE TOKENS ON WALLETS #============================================================================================================================================= def process_wallet_transactions_result(wallet): wallet_transactions_result = get_wallet_transactions_from_DB(wallet) if not wallet_transactions_result: return {"wallet": wallet, "tokens": []} wallet = wallet_transactions_result[0].get('wallet') # Группируем транзакции по токенам token_groups = defaultdict(list) for tx in wallet_transactions_result: if isinstance(tx, dict): token_key = tx.get('mint_address') token_groups[token_key].append(tx) analyzed_tokens = [] # Обрабатываем каждую группу токенов for token, transactions in token_groups.items(): # Транзакции уже проанализированы при загрузке в базу данных analyzed_transactions = transactions buy_in_valid_mc = get_token_buys_in_valid_mc(analyzed_transactions) first_hold_time = get_first_hold_time(analyzed_transactions) token_result = { "token": token, "buy_in_valid_mc": buy_in_valid_mc, "first_hold_time": first_hold_time, "transactions": analyzed_transactions } analyzed_tokens.append(token_result) result = { "wallet": wallet, "tokens": analyzed_tokens } return result #============================================================================================================================================= # MAIN FILTERS #============================================================================================================================================= def get_max_tokens_per_hour(wallet_result): if not wallet_result or not wallet_result.get("tokens"): return 0 analyzed_tokens = wallet_result.get("tokens", []) purchase_times = [] for token_entry in analyzed_tokens: if not token_entry.get("buy_in_valid_mc"): continue transactions = token_entry.get("transactions", []) if not transactions: continue buy_transactions = [tx for tx in transactions if tx.get("type") == "buy"] if not buy_transactions: continue first_buy = min(buy_transactions, key=lambda x: x.get("blocktime", 0)) block_time = first_buy.get("blocktime", 0) if block_time: purchase_times.append(block_time) if not purchase_times: return 0 purchase_times.sort() max_count = 0 n = len(purchase_times) left = 0 for right in range(n): while purchase_times[right] - purchase_times[left] >= 3600: left += 1 count = right - left + 1 max_count = max(max_count, count) return max_count def get_hold_time_on_wallet_percent(wallet_result): """ Рассчитывает процент токенов с достаточным временем удержания. Логика аналогична filters.py: токен валиден если hold_time >= MIN_HOLD_TIME_SECONDS или если это большое значение (2^63-1, означающее что токены все еще держатся). """ if not wallet_result or not wallet_result.get("tokens"): return 100.0 # Если нет токенов, считаем что фильтр пройден analyzed_tokens = wallet_result.get("tokens", []) tokens_with_short_hold = 0 total_tokens = 0 for token_entry in analyzed_tokens: if not token_entry.get("buy_in_valid_mc"): continue first_hold_time = token_entry.get("first_hold_time") total_tokens += 1 # Проверяем время удержания аналогично filters.py if isinstance(first_hold_time, (int, float)): # Большое значение (2^63-1) означает что токены до сих пор держатся - это валидно if first_hold_time < MIN_HOLD_TIME_SECONDS and first_hold_time != (2 ** 63 - 1): tokens_with_short_hold += 1 elif first_hold_time is None: # None означает проблему с данными - считаем невалидным tokens_with_short_hold += 1 # Другие значения (например 'hold') считаем валидными if total_tokens == 0: return 100.0 # Возвращаем процент токенов с достаточным удержанием (100% - процент коротких) short_hold_percent = (tokens_with_short_hold / total_tokens) * 100 return 100.0 - short_hold_percent def get_min_buy_percent(wallet_result): if not wallet_result or not wallet_result.get("tokens"): return 0.0 analyzed_tokens = wallet_result.get("tokens", []) total_buys = 0 big_buys = 0 for token_entry in analyzed_tokens: if not token_entry.get("buy_in_valid_mc"): continue for tx in token_entry.get("transactions", []): if tx.get("type") != "buy": continue # Проверяем MC транзакции - считаем только покупки в валидном MC mc = tx.get("MC", 0) if not (MIN_MARKET_CAP <= mc <= MAX_MARKET_CAP): continue # Пропускаем покупки в невалидном MC delta_sol = abs(tx.get("delta_sol", 0)) total_buys += 1 if delta_sol >= MIN_BUY: big_buys += 1 if total_buys == 0: return 0.0 # Нет покупок — возвращаем 0 big_buys_percent = (big_buys / total_buys) * 100 return big_buys_percent def get_migration_percent(wallet_result, tokens_info): if not wallet_result or not wallet_result.get("tokens"): return 0.0 analyzed_tokens = wallet_result.get("tokens", []) tokens_info_dict = {item["token"]: item for item in tokens_info} migration_tokens = 0 total_tokens = 0 for token_entry in analyzed_tokens: if not token_entry.get("buy_in_valid_mc"): continue total_tokens += 1 token = token_entry.get("token") token_info = tokens_info_dict.get(token, {}) migration_time = token_info.get("migtime", 0) if migration_time and migration_time > 0: transactions = token_entry.get("transactions", []) buy_until_migration = any( tx.get("type") == "buy" and tx.get("blocktime", 0) <= migration_time for tx in transactions ) if buy_until_migration: migration_tokens += 1 if total_tokens == 0: return 0.0 migration_percent = (migration_tokens / total_tokens) * 100 return migration_percent def get_created_tokens_percent(wallet_result, tokens_info): if not wallet_result or not wallet_result.get("tokens"): return 0.0 analyzed_tokens = wallet_result.get("tokens", []) wallet = wallet_result.get("wallet") if not wallet: return 0.0 tokens_info_dict = {item["token"]: item for item in tokens_info} created_tokens = 0 total_tokens = 0 for token_entry in analyzed_tokens: if not token_entry.get("buy_in_valid_mc"): continue total_tokens += 1 token = token_entry.get("token") token_info = tokens_info_dict.get(token, {}) creator = token_info.get("creator") if creator == wallet: created_tokens += 1 if total_tokens == 0: return 0.0 created_percent = (created_tokens / total_tokens) * 100 return created_percent def filter_normal_plus(wallet_result, tokens_info): admission_percent=ADMISSION_PERCENT profit_percentage=PROFIT_PERCENT_FOR_NORMALS if not wallet_result or not wallet_result.get("tokens"): return False analyzed_tokens = wallet_result.get("tokens", []) tokens_info_dict = {item["token"]: item for item in tokens_info} valid_tokens_count = 0 profitable_tokens_count = 0 for token_entry in analyzed_tokens: if not token_entry.get("buy_in_valid_mc"): continue token = token_entry.get("token") token_info = tokens_info_dict.get(token, {}) migration_time = token_info.get("migtime", 0) # Для немигрированных токенов - анализируем как раньше if migration_time in [0, 1]: valid_tokens_count += 1 try: ath_info = get_ath_of_token_with_JUP(token) ath_price = ath_info["ath_price"] ath_time = ath_info["ath_time"] valid_buys = [ tx.get("MC", 0) for tx in token_entry.get("transactions", []) if (tx.get("type") == "buy" and tx.get("blocktime", 0) <= ath_time and MIN_MARKET_CAP <= tx.get("MC", 0) <= MAX_MARKET_CAP) ] if valid_buys and max((ath_price - mc) / mc * 100 for mc in valid_buys) >= profit_percentage: profitable_tokens_count += 1 except: continue # Для мигрированных токенов - учитываем только токены, которые НЕ покупались ДО миграции, но покупались ПОСЛЕ elif migration_time > 1: # Проверяем были ли покупки ДО миграции pre_migration_buys = [ tx for tx in token_entry.get("transactions", []) if (tx.get("type") == "buy" and tx.get("blocktime", 0) <= migration_time and MIN_MARKET_CAP <= tx.get("MC", 0) <= MAX_MARKET_CAP) ] # Проверяем есть ли покупки ПОСЛЕ миграции post_migration_buys = [ tx for tx in token_entry.get("transactions", []) if (tx.get("type") == "buy" and tx.get("blocktime", 0) > migration_time and MIN_MARKET_CAP <= tx.get("MC", 0) <= MAX_MARKET_CAP) ] # Учитываем только токены, которые НЕ покупались до миграции, но покупались после if not pre_migration_buys and post_migration_buys: valid_tokens_count += 1 try: ath_info = get_ath_of_token_with_JUP(token) ath_price = ath_info["ath_price"] ath_time = ath_info["ath_time"] # Анализируем только покупки после миграции и до ATH valid_buys = [ tx.get("MC", 0) for tx in post_migration_buys if tx.get("blocktime", 0) <= ath_time ] if valid_buys and max((ath_price - mc) / mc * 100 for mc in valid_buys) >= profit_percentage: profitable_tokens_count += 1 except: continue if valid_tokens_count == 0: return False profit_percent = (profitable_tokens_count / valid_tokens_count) * 100 return profit_percent >= admission_percent def classification(len_tokens, len_tokens_with_buy_in_valid_mc, is_tokens_per_hour_valid, is_hold_time_valid, is_min_buy_valid, is_migration_valid, is_created_tokens_valid, wallet_result, tokens_info): try: # === Пройдены базовые фильтры === if is_hold_time_valid and is_tokens_per_hour_valid and is_min_buy_valid: if is_migration_valid and is_created_tokens_valid: if len_tokens == 1: return "GOLD FRESH" else: if len_tokens_with_buy_in_valid_mc >= 2: return "GOLD" elif len_tokens_with_buy_in_valid_mc == 1: return "SILVER" else: if filter_normal_plus(wallet_result, tokens_info): return "NORMALPLUS" return "NORMAL" # === Не пройдены базовые фильтры === return "TRASH" except Exception as e: print(f"ERROR in classification: {e}") return "ERROR" #============================================================================================================================================= #MAIN FUNCTIONS FOR FILTER WALLET #============================================================================================================================================= def analyze_wallet_transactions(wallet_address, raw_transactions): """ Принимает сырые транзакции кошелька, анализирует их и сохраняет в базу данных. Args: wallet_address: адрес кошелька raw_transactions: список транзакций для анализа Returns: bool: True если анализ и сохранение успешны, False иначе """ if not raw_transactions: print("⚠️ Нет транзакций для анализа") return False # Группируем транзакции по токенам token_groups = defaultdict(list) for tx in raw_transactions: if isinstance(tx, dict): token_key = tx.get('mint_address') or tx.get('token_extracted') if token_key: token_groups[token_key].append(tx) all_db_transactions = [] # Обрабатываем каждую группу токенов for token, transactions in token_groups.items(): updated_transactions = [] sorted_transactions = sorted(transactions, key=lambda x: x.get("blocktime", 0)) # Анализируем каждую транзакцию for tx in sorted_transactions: delta_token = tx.get("delta_token", 0) delta_sol = tx.get("delta_sol", 0) tx_type = tx.get("type") mc = get_market_cap(tx) updated_tx = { "signature": tx.get("signature"), "type": tx_type, "profit": None, "blocktime": tx.get("blocktime"), "delta_sol": delta_sol, "delta_token": delta_token, "fee": tx.get("fee"), "MC": mc } updated_transactions.append(updated_tx) # Рассчитываем прибыль по FIFO updated_transactions = get_profit_percent_to_transactions(updated_transactions) # Подготавливаем транзакции для сохранения в базу данных for tx in updated_transactions: db_tx = { 'wallet_address': wallet_address, 'token_extracted': token, 'signature': tx.get('signature', ''), 'type': tx.get('type', ''), 'profit': tx.get('profit'), 'blocktime': tx.get('blocktime', 0), 'delta_sol': tx.get('delta_sol', 0), 'delta_token': tx.get('delta_token', 0), 'fee': tx.get('fee', 0), 'MC': tx.get('MC', 0) } all_db_transactions.append(db_tx) # Сохраняем все проанализированные транзакции в базу данных if all_db_transactions: success = add_transactions_to_DB(all_db_transactions) if success: print(f"✅ Анализ кошелька {wallet_address} завершен, обработано {len(token_groups)} токенов") return True else: print(f"❌ Ошибка при сохранении данных анализа кошелька {wallet_address}") return False else: print("⚠️ Нет данных для сохранения после анализа") return False def wallet_metrics(wallet): # Подготовка к фильтрам wallet_result = process_wallet_transactions_result(wallet) token_list = [token_entry["token"] for token_entry in wallet_result["tokens"]] tokens_info = get_tokens_migration_time_and_creator(token_list) # Вызов фильтров len_tokens = len(wallet_result["tokens"]) len_tokens_with_buy_in_valid_mc = len([token for token in wallet_result["tokens"] if token.get("buy_in_valid_mc")]) len_all_buys_in_valid_mc = count_all_buys_in_valid_mc(wallet_result) max_tokens_per_hour = get_max_tokens_per_hour(wallet_result) hold_time_percent = get_hold_time_on_wallet_percent(wallet_result) min_buy_percent = get_min_buy_percent(wallet_result) migration_percent = get_migration_percent(wallet_result, tokens_info) created_tokens_percent = get_created_tokens_percent(wallet_result, tokens_info) metrics = { "len_tokens": len_tokens, "len_tokens_with_buy_in_valid_mc": len_tokens_with_buy_in_valid_mc, "len_all_buys_in_valid_mc": len_all_buys_in_valid_mc, "max_tokens_per_hour": max_tokens_per_hour, "hold_time_percent": hold_time_percent, "min_buy_percent": min_buy_percent, "migration_percent": migration_percent, "created_tokens_percent": created_tokens_percent, "wallet_result": wallet_result, "tokens_info": tokens_info } return metrics def apply_filters(metrics) -> Optional[Dict[str, Any]]: len_tokens = metrics["len_tokens"] len_tokens_with_buy_in_valid_mc = metrics["len_tokens_with_buy_in_valid_mc"] len_all_buys_in_valid_mc = metrics["len_all_buys_in_valid_mc"] max_tokens_per_hour = metrics["max_tokens_per_hour"] hold_time_percent = metrics["hold_time_percent"] min_buy_percent = metrics["min_buy_percent"] migration_percent = metrics["migration_percent"] created_tokens_percent = metrics["created_tokens_percent"] wallet_result = metrics["wallet_result"] tokens_info = metrics["tokens_info"] passed_max_tokens_per_hour = MAX_TOKENS_PER_HOUR passed_migration_percent = MIN_TOKENS_BEFORE_MIGRATION_PERCENT if len_all_buys_in_valid_mc <= 2: passed_min_buy_percent = 50 else: passed_min_buy_percent = MIN_NORMAL_BUYS_PERCENT if len_tokens_with_buy_in_valid_mc <= 2: passed_hold_time_percent = 50 passed_created_tokens_percent = 50 else: passed_hold_time_percent = MIN_HOLD_TIME_PERCENT passed_created_tokens_percent = MAX_CREATED_PERCENT_PERCENT is_tokens_per_hour_valid = max_tokens_per_hour <= passed_max_tokens_per_hour is_hold_time_valid = hold_time_percent >= passed_hold_time_percent is_min_buy_valid = min_buy_percent >= passed_min_buy_percent is_migration_valid = migration_percent >= passed_migration_percent is_created_tokens_valid = created_tokens_percent <= passed_created_tokens_percent wallet_class = classification( len_tokens, len_tokens_with_buy_in_valid_mc, is_tokens_per_hour_valid, is_hold_time_valid, is_min_buy_valid, is_migration_valid, is_created_tokens_valid, wallet_result, tokens_info ) return { "token_amount": len_tokens, "tokens_in_MC": len_tokens_with_buy_in_valid_mc, "f_tokens_ph": max_tokens_per_hour, "f_hold_time": hold_time_percent, "f_minbuy": min_buy_percent, "f_before_mig": migration_percent, "f_created_tokens": created_tokens_percent, "wallet_type": wallet_class } def analyze_wallet(wallet_address): """ Основная функция для анализа кошелька. Получает метрики, применяет фильтры и возвращает результат классификации. Args: wallet_address: адрес кошелька для анализа Returns: dict: результат анализа с типом кошелька и метриками """ try: # Получаем метрики кошелька metrics = wallet_metrics(wallet_address) # Применяем фильтры result = apply_filters(metrics) if result: # Добавляем информацию о кошельке result["wallet"] = wallet_address result["type"] = result["wallet_type"] result["len_tokens"] = result["token_amount"] result["len_tokens_with_buy_in_valid_mc"] = result["tokens_in_MC"] return result except Exception as e: print(f"Ошибка при анализе кошелька {wallet_address}: {e}") return None