#!/usr/bin/env python3 # -*- coding: utf-8 -*- import os import time import json import logging import re from typing import Dict, Any, List, Tuple, Optional, Set import pymysql import requests # ================= НАСТРОЙКИ ================= DB_HOST = os.getenv("DB_HOST", "103.102.228.54") DB_USER = os.getenv("DB_USER", "phpmyadmin") DB_PASSWORD = os.getenv("DB_PASSWORD", "OE2hxy8iNUvB") DB_NAME = os.getenv("DB_NAME", "pumpfun") WHITE_DIAP_PATH = os.getenv("WHITE_DIAP_PATH", "/var/www/html/vera/whiteDiapazon.json") # Один и тот же бот можно использовать для всех групп TG_BOT_TOKEN = os.getenv("TG_BOT_TOKEN", "8442183787:AAF8o1PmZ72CQB9chVji91_roE18aVA9nlo") # Чат, куда шлём mint при col_82 -> 1 (и только если col_78 == 1) TG_WHITE_OK_CHAT_ID = int(os.getenv("TG_WHITE_OK_CHAT_ID", "-1002615752660")) # Чат, куда шлём mint, когда одновременно col_78=1 и col_81=1 и col_82=1 TG_ALL_ONES_CHAT_ID = int(os.getenv("TG_ALL_ONES_CHAT_ID", "-1002615752660")) LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper() BATCH_SIZE = int(os.getenv("BATCH_SIZE", "200")) logging.basicConfig( level=getattr(logging, LOG_LEVEL, logging.INFO), format="[%(asctime)s] %(levelname)s - %(message)s" ) COLNAME_RE = re.compile(r"^[A-Za-z0-9_]+$") # --- dip-файлы для 82: whiteDiapazondip*.json --- BASE_WDIAP_DIR = os.path.dirname(WHITE_DIAP_PATH) or "/var/www/html/vera" def _dip_suffix_for_82(usd_mc: float) -> str: try: v = int(float(usd_mc)) except Exception: v = 0 if 0 <= v <= 30000: return "030" if 30001 <= v <= 60000:return "3060" if 60001 <= v <= 90000:return "6090" return "901000000" def _white_diap_dip_path(usd_mc: float) -> str: # обратите внимание: в названии «Diapazon**dip**» return os.path.join(BASE_WDIAP_DIR, f"whiteDiapazondip{_dip_suffix_for_82(usd_mc)}.json") _WDIAP_DIP_CACHE: Dict[str, 'WhiteDiapazon'] = {} def _get_white_diap_for_mc(usd_mc: float) -> 'WhiteDiapazon': p = _white_diap_dip_path(usd_mc) inst = _WDIAP_DIP_CACHE.get(p) if inst is None: inst = WhiteDiapazon(p) _WDIAP_DIP_CACHE[p] = inst inst.load_if_changed() return inst # === [вершина файла, рядом с существующими импортами и константами] === # ### ADD: новые TG-группы для 97..106 TG_97_OK_CHAT_ID = int(os.getenv("TG_97_OK_CHAT_ID", "-1002615752660")) TG_98_OK_CHAT_ID = int(os.getenv("TG_98_OK_CHAT_ID", "-1002615752660")) TG_99_OK_CHAT_ID = int(os.getenv("TG_99_OK_CHAT_ID", "-1002615752660")) TG_100_OK_CHAT_ID = int(os.getenv("TG_100_OK_CHAT_ID", "-1002615752660")) TG_101_OK_CHAT_ID = int(os.getenv("TG_101_OK_CHAT_ID", "-1002615752660")) TG_102_OK_CHAT_ID = int(os.getenv("TG_102_OK_CHAT_ID", "-1002615752660")) TG_103_OK_CHAT_ID = int(os.getenv("TG_103_OK_CHAT_ID", "-1002615752660")) TG_104_OK_CHAT_ID = int(os.getenv("TG_104_OK_CHAT_ID", "-1002615752660")) TG_105_OK_CHAT_ID = int(os.getenv("TG_105_OK_CHAT_ID", "-1002615752660")) TG_106_OK_CHAT_ID = int(os.getenv("TG_106_OK_CHAT_ID", "-1002615752660")) TG_107_OK_CHAT_ID = int(os.getenv("TG_107_OK_CHAT_ID", "-1002615752660")) # Список колонок кошельков топ-10 (как в vera_4_.py) WALLET_COLS = ["col_16","col_21","col_26","col_31","col_36","col_41","col_46","col_51","col_56","col_61"] # === [чуть ниже, рядом с WhiteDiapazon или перед «ВСПОМОГАТЕЛЬНЫЕ»] === # ### ADD: простой белый список адресов для колоночного фильтра (как Whitelist) class SimpleWhitelist: def __init__(self, path: str): self.path = path self.mtime = 0.0 self.addrs: Set[str] = set() def load_if_changed(self): try: mtime = os.path.getmtime(self.path) if mtime != self.mtime: with open(self.path, "r", encoding="utf-8") as f: data = json.load(f) arr = data.get("wallets") if isinstance(data, dict) and "wallets" in data else data self.addrs = {str(x).strip() for x in (arr or []) if str(x).strip()} self.mtime = mtime logging.info("Whitelist reloaded (%s): %d", self.path, len(self.addrs)) except FileNotFoundError: self.addrs = set(); self.mtime = 0.0 except Exception as e: logging.error("Error reading %s: %s", self.path, e) # === [рядом с BASE_WDIAP_DIR — используем ту же директорию] === # ### ADD: файловая адресация для 97..106 (как в 95/96 — по bucket от usd_market_cap) def _bucket_suffix_for_93_0(usd_mc: Optional[float]) -> str: try: v = int(float(usd_mc or 0.0)) except Exception: v = 0 if 0 <= v <= 10000: return "010" if 10001 <= v <= 15000: return "1015" if 15001 <= v <= 20000: return "1520" if 20001 <= v <= 25000: return "2025" if 25001 <= v <= 30000: return "2530" return "3080" # Кэшируем объекты по путям _WL_MULTI_CACHE: Dict[str, SimpleWhitelist] = {} _WD_MULTI_CACHE: Dict[str, 'WhiteDiapazon'] = {} def _path_white_for(col_no: int, usd_mc: Optional[float]) -> str: suf = _bucket_suffix_for_93_0(usd_mc) # файлы вида: 97white010.json, 98white1015.json, ... return os.path.join(BASE_WDIAP_DIR, f"{col_no}white{suf}.json") def _path_white_diap_for(col_no: int, usd_mc: Optional[float]) -> str: suf = _bucket_suffix_for_93_0(usd_mc) # файлы вида: 97whiteDiapazon010.json, ... return os.path.join(BASE_WDIAP_DIR, f"{col_no}whiteDiapazon{suf}.json") def _get_multi_wl(col_no: int, usd_mc: Optional[float]) -> SimpleWhitelist: p = _path_white_for(col_no, usd_mc) inst = _WL_MULTI_CACHE.get(p) if inst is None: inst = SimpleWhitelist(p); _WL_MULTI_CACHE[p] = inst inst.load_if_changed() return inst def _get_multi_wd(col_no: int, usd_mc: Optional[float]) -> 'WhiteDiapazon': p = _path_white_diap_for(col_no, usd_mc) inst = _WD_MULTI_CACHE.get(p) if inst is None: inst = WhiteDiapazon(p); _WD_MULTI_CACHE[p] = inst inst.load_if_changed() return inst # === [SQL helpers секция — рядом с select_mints_with_col82_null и update_col82_atomic] === # ### ADD: кандидаты и апдейтеры для 97..106 def fetch_rows_for_colN_candidates(conn, col_no: int, limit: int = 150) -> List[Dict[str, Any]]: col_name = f"col_{col_no}" cols = ", ".join(["mint","usd_market_cap","col_94"] + WALLET_COLS + [col_name]) sql = f""" SELECT {cols} FROM tokens WHERE col_93=0 AND {col_name} IS NULL AND COALESCE(usd_market_cap,0) >= 8000 AND col_94 = 1 -- было: (col_94 IS NULL OR col_94=1) ORDER BY mint ASC LIMIT %s """ with conn.cursor() as cur: cur.execute(sql, (limit,)) return cur.fetchall() def update_colN_atomic(conn, mint: str, col_no: int, val01: int) -> bool: col_name = f"col_{col_no}" sql = f""" UPDATE tokens SET {col_name}=%s WHERE LOWER(mint)=LOWER(%s) AND col_93=0 AND {col_name} IS NULL AND COALESCE(usd_market_cap,0) >= 8000 AND col_94 = 1 -- было: (col_94 IS NULL OR col_94=1) """ with conn.cursor() as cur: return cur.execute(sql, (val01, mint)) > 0 # ===== Публикация в index.html на 5 минут (как для 94=0 в vera_4_.py) ===== FAIL_HTML_PATH = os.getenv("FAIL_HTML_PATH", "/var/www/html/vera/index.html") FAIL_HTML_STATE = os.getenv("FAIL_HTML_STATE", "/var/www/html/vera/index_state.json") FAIL_TTL_SECONDS = int(os.getenv("FAIL_TTL_SECONDS", "300")) # 5 минут def _now_ms() -> int: return int(time.time() * 1000) def _safe_mkdirs(path: str) -> None: os.makedirs(os.path.dirname(path) or ".", exist_ok=True) def _load_state(path: str) -> Dict[str, int]: try: with open(path, "r", encoding="utf-8") as f: data = json.load(f) or {} cleaned = {} for k, v in (data.items() if isinstance(data, dict) else []): try: ts = int(v) if k and ts > 0: cleaned[k] = ts except Exception: continue return cleaned except FileNotFoundError: return {} except Exception: return {} def _write_state_atomic(path: str, state: Dict[str, int]) -> None: _safe_mkdirs(path) tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(state, f, ensure_ascii=False, separators=(",", ":")) os.replace(tmp, path) def _render_html(path: str, mints: List[str]) -> None: _safe_mkdirs(path) html = [ "", '', "", ' ', ' ', " Отправленные токены (последние 5 минут)", " ", "", "", "

Отправленные в Telegram токены (живут 5 минут)

", f'
Сейчас в списке: {len(mints)}
', '
' ] if mints: for m in mints: html.append(f'
{m}
') else: html.append('
Пока пусто
') html.extend([ "
", "", "", "" ]) tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: f.write("\n".join(html)) os.replace(tmp, path) def _purge_expired_and_render(state: Dict[str, int]) -> Dict[str, int]: now = _now_ms() ttl_ms = FAIL_TTL_SECONDS * 1000 alive = {m: ts for m, ts in state.items() if now - ts <= ttl_ms} if len(alive) != len(state): _render_html(FAIL_HTML_PATH, sorted(alive.keys())) _write_state_atomic(FAIL_HTML_STATE, alive) return alive return state def publish_fail_to_html(mint: str) -> None: """Добавляет mint в HTML-страницу (с TTL), если его ещё нет.""" try: state = _load_state(FAIL_HTML_STATE) state = _purge_expired_and_render(state) if mint not in state: state[mint] = _now_ms() _write_state_atomic(FAIL_HTML_STATE, state) _render_html(FAIL_HTML_PATH, sorted(state.keys())) except Exception as e: logging.warning("Не удалось опубликовать в index.html для %s: %s", mint, e) # === [ЛОГИКА — рядом с evaluate_white_for_rows или ниже] === # ### ADD: объединённый расчёт (аналог col_95 & col_96 из vera_4_.py) def _decide_for_colN( top10: Set[str], wl_addrs: Set[str], diap: 'WhiteDiapazon', vals_row: Dict[str, Any], min_wallet_hits: int = 1, min_range_hits: int = 3 ) -> bool: """ Истина, если одновременно выполняются: - совпадений адресов из топ-10 с whitelist >= min_wallet_hits - попаданий значений в белые диапазоны >= min_range_hits """ if not top10 or not wl_addrs: return False wallet_hits = len(top10 & wl_addrs) if wallet_hits < max(1, int(min_wallet_hits)): return False hits = 0 for c in (diap.columns() or []): if diap.value_hits(c, vals_row.get(c)): hits += 1 return hits >= max(1, int(min_range_hits)) def _fetch_needed_cols_for_diap(conn, mint: str, cols: List[str]) -> Dict[str, Any]: if not cols: return {} safe = [c for c in cols if COLNAME_RE.match(c)] if not safe: return {} fields = ", ".join(["mint"] + safe) sql = f"SELECT {fields} FROM tokens WHERE LOWER(mint)=LOWER(%s)" with conn.cursor() as cur: cur.execute(sql, (mint,)) r = cur.fetchone() or {} r.pop("mint", None) return r # соответствие: номер столбца -> TG чат _COL_TO_TG = { 97: TG_97_OK_CHAT_ID, 98: TG_98_OK_CHAT_ID, 99: TG_99_OK_CHAT_ID, 100: TG_100_OK_CHAT_ID, 101: TG_101_OK_CHAT_ID, 102: TG_102_OK_CHAT_ID, 103: TG_103_OK_CHAT_ID, 104: TG_104_OK_CHAT_ID, 105: TG_105_OK_CHAT_ID, 106: TG_106_OK_CHAT_ID, 107: TG_107_OK_CHAT_ID, } def process_cols_97_106(conn) -> None: for col_no in range(97, 107): try: rows = fetch_rows_for_colN_candidates(conn, col_no, limit=120) if not rows: continue updated = 0 for r in rows: # Доп. страховка: считаем только при 94=1 if r.get("col_94") != 1: continue mint = r["mint"] umc = r.get("usd_market_cap") # Топ-10 кошельков из строки-кандидата top10 = {(r.get(c) or "").strip() for c in WALLET_COLS if (r.get(c) or "").strip()} if not top10: continue # ждём, пока подтянутся адреса # Файлы whitelist/whiteDiapazon выбираем по bucket от usd_market_cap и номеру колонки wl = _get_multi_wl(col_no, umc).addrs wd = _get_multi_wd(col_no, umc) cols_needed = wd.columns() or [] vals = _fetch_needed_cols_for_diap(conn, mint, cols_needed) # --- Новое требование по количеству совпадений кошельков --- wallet_hits = len(top10 & wl) min_wallet_hits = 2 if col_no in (97, 98, 99) else 1 # Подсчёт попаданий в белые диапазоны (порог >=3 — без изменений) range_hits = 0 for c in cols_needed: if wd.value_hits(c, vals.get(c)): range_hits += 1 ok = (wallet_hits >= min_wallet_hits) and (range_hits >= 3) v = 1 if ok else 0 if update_colN_atomic(conn, mint, col_no, v): updated += 1 if v == 1: # RECHECK перед отправкой в TG (защита от «мигания») with conn.cursor() as cur: cur.execute(f""" SELECT col_94, {f"col_{col_no}"} AS coln FROM tokens WHERE LOWER(mint)=LOWER(%s) LIMIT 1 """, (mint,)) chk = cur.fetchone() or {} if chk.get("col_94") == 1 and chk.get("coln") == 1: chat_id = _COL_TO_TG.get(col_no) if chat_id: send_tg_message(TG_BOT_TOKEN, chat_id, mint) if updated: logging.info("Обновлено col_%d: %d", col_no, updated) except Exception as e: logging.error("Ошибка обработки col_%d: %s", col_no, e) # --- Пороговые проценты для 107 (рост market_cap относительно usd_market_cap) # числа заданы в долях: 1.00 = +100%, 1.20 = +120%, 0.35 = +35%, и т.д. _COL_107_THRESHOLDS = { 97: 1.50, 98: 1.50, 99: 1.50, 100: 1.50, 101: 1.50, 102: 1.50, 103: 0.40, 104: 1.50, 105: 1.50, 106: 0.40, } def _fetch_candidates_for_107(conn, limit: int = 250): """ Кандидаты для 107: - ветка 93=0, - 94=1, - col_107 IS NULL, - все 97..106 уже посчитаны (NOT NULL), - usd_market_cap >= 8000. """ cols = ", ".join([f"col_{i}" for i in range(97, 107)]) sql = f""" SELECT mint, usd_market_cap, market_cap, {cols}, col_94 FROM tokens WHERE col_93=0 AND col_94=1 AND col_107 IS NULL AND COALESCE(usd_market_cap,0) >= 8000 AND {" AND ".join([f"col_{i} IS NOT NULL" for i in range(97,107)])} ORDER BY mint ASC LIMIT %s """ with conn.cursor() as cur: cur.execute(sql, (limit,)) return cur.fetchall() def _update_col107_atomic(conn, mint: str, val01: int) -> bool: sql = """ UPDATE tokens SET col_107=%s WHERE LOWER(mint)=LOWER(%s) AND col_93=0 AND col_94=1 AND col_107 IS NULL AND COALESCE(usd_market_cap,0) >= 8000 """ with conn.cursor() as cur: return cur.execute(sql, (val01, mint)) > 0 def process_col_107(conn) -> None: """ После расчётов 97..106 начинает слежение за ростом market_cap. Берём минимальный порог из _COL_107_THRESHOLDS по тем ячейкам 97..106, где значение =1. Если market_cap >= usd_market_cap * (1 + минимальный_порог) -> col_107=1 + отправка в TG. Если среди 97..106 нет ни одной ячейки равной 1 -> col_107=0 (и больше не следим). """ rows = _fetch_candidates_for_107(conn, 250) if not rows: return for r in rows: mint = r["mint"] try: umc = float(r.get("usd_market_cap") or 0.0) mc = float(r.get("market_cap") or 0.0) except Exception: umc = 0.0; mc = 0.0 # соберём пороги только по «единичным» ячейкам 97..106 on_thresholds = [] for col_no, thr in _COL_107_THRESHOLDS.items(): v = r.get(f"col_{col_no}") if v == 1: on_thresholds.append(thr) if not on_thresholds: # ничего не было «1» в 97..106 — фиксируем 107=0 и не следим дальше if _update_col107_atomic(conn, mint, 0): pass continue min_thr = min(on_thresholds) # доля: 0.35 -> +35% target = umc * (1.0 + min_thr) if umc > 0.0 and mc >= target: if _update_col107_atomic(conn, mint, 1): # повторно перечитаем и убедимся, что 94=1 и 107=1 — как в 97..106 перед отправкой with conn.cursor() as cur: cur.execute(""" SELECT col_94, col_107 FROM tokens WHERE LOWER(mint)=LOWER(%s) LIMIT 1 """, (mint,)) chk = cur.fetchone() or {} if chk.get("col_94") == 1 and chk.get("col_107") == 1: send_tg_message(TG_BOT_TOKEN, TG_107_OK_CHAT_ID, mint) def nullify_cols_97_106_when_94_zero(conn) -> int: """ Если col_94=0 в ветке 93=0 — держим 97..106 в NULL. Возвращает количество затронутых строк (для логов). """ sql = """ UPDATE tokens SET col_97=NULL, col_98=NULL, col_99=NULL, col_100=NULL, col_101=NULL, col_102=NULL, col_103=NULL, col_104=NULL, col_105=NULL, col_106=NULL, col_107=NULL WHERE col_93=0 AND col_94=0 AND ( col_97 IS NOT NULL OR col_98 IS NOT NULL OR col_99 IS NOT NULL OR col_100 IS NOT NULL OR col_101 IS NOT NULL OR col_102 IS NOT NULL OR col_103 IS NOT NULL OR col_104 IS NOT NULL OR col_105 IS NOT NULL OR col_106 IS NOT NULL OR col_107 IS NOT NULL ) """ with conn.cursor() as cur: cur.execute(sql) return cur.rowcount or 0 # ================= ВСПОМОГАТЕЛЬНЫЕ ================= def get_connection(): return pymysql.connect( host=DB_HOST, user=DB_USER, password=DB_PASSWORD, database=DB_NAME, charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, autocommit=True ) TG_SENT_CACHE = os.getenv("TG_SENT_CACHE", "tg_sent_cache.txt") _SENT_KEYS = None def _load_sent_keys() -> Set[str]: global _SENT_KEYS if _SENT_KEYS is None: _SENT_KEYS = set() try: with open(TG_SENT_CACHE, "r", encoding="utf-8") as f: for line in f: _SENT_KEYS.add(line.strip()) except FileNotFoundError: pass return _SENT_KEYS def _mark_sent(key: str) -> None: s = _load_sent_keys() if key in s: return s.add(key) try: os.makedirs(os.path.dirname(TG_SENT_CACHE) or ".", exist_ok=True) with open(TG_SENT_CACHE, "a", encoding="utf-8") as f: f.write(key + "\n") except Exception: pass def send_tg_message(bot_token: str, chat_id: int, text: str) -> None: """ Антидубль на уровне чата и текста. """ if not bot_token or not chat_id: return key = f"{chat_id}:{text}" if key in _load_sent_keys(): return try: url = f"https://api.telegram.org/bot{bot_token}/sendMessage" r = requests.post(url, timeout=15, json={ "chat_id": chat_id, "text": text, "disable_web_page_preview": True }) if r.status_code != 200: logging.warning("Telegram send non-200: %s %s", r.status_code, r.text[:200]) return _mark_sent(key) except Exception as e: logging.warning("Ошибка отправки в Telegram (%s): %s", text, e) # ---- Белые диапазоны ---- class WhiteDiapazon: def __init__(self, path: str): self.path = path self.mtime = 0.0 self.ranges: Dict[str, List[Tuple[float, float]]] = {} def load_if_changed(self) -> bool: try: mtime = os.path.getmtime(self.path) except FileNotFoundError: if self.ranges: logging.warning("whiteDiapazon.json исчез — очистил диапазоны.") self.ranges = {} self.mtime = 0.0 return False except Exception as e: logging.error("Ошибка доступа к %s: %s", self.path, e) return False if mtime == self.mtime: return False try: with open(self.path, "r", encoding="utf-8") as f: data = json.load(f) or {} rngs = data.get("ranges") or {} parsed: Dict[str, List[Tuple[float, float]]] = {} for col, arr in rngs.items(): if not isinstance(col, str) or not COLNAME_RE.match(col): continue pairs: List[Tuple[float, float]] = [] for it in (arr or []): if not isinstance(it, (list, tuple)) or len(it) != 2: continue try: lo = float(it[0]); hi = float(it[1]) if lo > hi: lo, hi = hi, lo pairs.append((lo, hi)) except Exception: continue if pairs: parsed[col] = pairs self.ranges = parsed self.mtime = mtime logging.info("whiteDiapazon перезагружен: %d столбцов.", len(self.ranges)) return True except Exception as e: logging.error("Ошибка чтения %s: %s", self.path, e) return False def columns(self) -> List[str]: return list(self.ranges.keys()) def value_hits(self, col: str, value: Optional[float]) -> bool: if value is None: return False try: v = float(value) except Exception: return False for lo, hi in self.ranges.get(col, []): if lo <= v <= hi: return True return False # ---- SQL помощники ---- def select_mints_for_full_scan(conn, last_mint: Optional[str], limit: int) -> List[str]: if last_mint: sql = """ SELECT mint FROM tokens WHERE mint > %s AND col_93=1 AND COALESCE(usd_market_cap,0) >= 8000 ORDER BY mint ASC LIMIT %s """ params = (last_mint, limit) else: sql = """ SELECT mint FROM tokens WHERE col_93=1 AND COALESCE(usd_market_cap,0) >= 8000 ORDER BY mint ASC LIMIT %s """ params = (limit,) with conn.cursor() as cur: cur.execute(sql, params) return [r["mint"] for r in cur.fetchall()] def select_mints_with_col82_null(conn, limit: int) -> List[str]: sql = """ SELECT mint FROM tokens WHERE col_93=1 AND col_78=1 AND col_82 IS NULL AND COALESCE(usd_market_cap,0) >= 8000 ORDER BY mint ASC LIMIT %s """ with conn.cursor() as cur: cur.execute(sql, (limit,)) return [r["mint"] for r in cur.fetchall()] def read_cols_78_81_82_93(conn, mint: str) -> Tuple[Optional[int], Optional[int], Optional[int], Optional[int]]: """ Читаем флаги, отдавая приоритет строке col_93=1 (если есть). """ sql = """ SELECT col_78, col_81, col_82, col_93 FROM tokens WHERE LOWER(mint)=LOWER(%s) ORDER BY (col_93=1) DESC, col_93 DESC LIMIT 1 """ with conn.cursor() as cur: cur.execute(sql, (mint,)) r = cur.fetchone() or {} def to_int(x): try: return int(x) if x is not None else None except: return None return (to_int(r.get("col_78")), to_int(r.get("col_81")), to_int(r.get("col_82")), to_int(r.get("col_93"))) def fetch_columns_for_mints(conn, mints: List[str], cols: List[str]) -> Dict[str, Dict[str, Any]]: """ Возвращает {mint: {col: value}} строго из строки ветки col_93=1. Это устраняет смешивание значений между копиями одного mint. """ if not mints: return {} safe_cols = [c for c in cols if COLNAME_RE.match(c)] # если нет нужных колонок — всё равно вернём пустые записи по всем mint fields = ", ".join(["mint"] + (safe_cols if safe_cols else [])) placeholders = ",".join(["%s"] * len(mints)) lowers = [m.lower() for m in mints] sql = f""" SELECT {fields} FROM tokens WHERE LOWER(mint) IN ({placeholders}) AND col_93=1 """ out: Dict[str, Dict[str, Any]] = {} with conn.cursor() as cur: cur.execute(sql, lowers) for row in cur.fetchall(): m = row.pop("mint") out[m] = {k: row.get(k) for k in (safe_cols or [])} # гарантируем ключи для всех mint for m in mints: out.setdefault(m, {}) return out def backfill_col82_zero_if_no_ranges(conn, batch: int = 200) -> int: """ Если whiteDiapazon пуст или не загрузился — считаем, что попаданий нет, и переводим застрявшие NULL в 0 (только ветка 93=1, MC >= 8000). Возвращает число обновлённых строк. """ mints = select_mints_with_col82_null(conn, batch) changed = 0 for mint in mints: if update_col82_atomic(conn, mint, 0): changed += 1 return changed def dogon_post_82_and_allones(conn, sent_whiteok_cache: Set[str], sent_allones_cache: Set[str]) -> None: # 82-OK: 78=1 & 82=1 в ветке 93=1 with conn.cursor() as cur: cur.execute(""" SELECT mint FROM tokens WHERE col_93=1 AND col_78=1 AND col_82=1 ORDER BY mint ASC LIMIT 200 """) for r in cur.fetchall(): m = r["mint"] if m not in sent_whiteok_cache: send_tg_message(TG_BOT_TOKEN, TG_WHITE_OK_CHAT_ID, m) sent_whiteok_cache.add(m) # все три: 78=1 & 81=1 & 82=1 cur.execute(""" SELECT mint FROM tokens WHERE col_93=1 AND col_78=1 AND col_81=1 AND col_82=1 ORDER BY mint ASC LIMIT 200 """) for r in cur.fetchall(): m = r["mint"] if m not in sent_allones_cache: send_tg_message(TG_BOT_TOKEN, TG_ALL_ONES_CHAT_ID, m) sent_allones_cache.add(m) def update_col82_atomic(conn, mint: str, val1_or_0: int) -> bool: """ 82: один раз, только ветка 93=1, только при достаточном MC и при col_78=1. """ sql = """ UPDATE tokens SET col_82=%s WHERE LOWER(mint)=LOWER(%s) AND col_93=1 AND col_78=1 AND col_82 IS NULL AND COALESCE(usd_market_cap,0) >= 8000 """ with conn.cursor() as cur: return cur.execute(sql, (val1_or_0, mint)) > 0 def read_cols_78_81_82(conn, mint: str) -> Tuple[Optional[int], Optional[int], Optional[int]]: sql = "SELECT col_78, col_81, col_82 FROM tokens WHERE LOWER(mint)=LOWER(%s)" with conn.cursor() as cur: cur.execute(sql, (mint,)) r = cur.fetchone() or {} def to_int(x): try: return int(x) if x is not None else None except: return None return (to_int(r.get("col_78")), to_int(r.get("col_81")), to_int(r.get("col_82"))) # ================= ЛОГИКА WHITE (82) ================= def evaluate_white_for_rows(rows: Dict[str, Dict[str, Any]], columns: List[str], diap) -> Dict[str, int]: """ Для каждой строки (mint) считает попадания в диапазоны и возвращает {mint: 0|1} для col_82. 0 — если попало 0/1/2 значений; 1 — если попало >=3. """ res: Dict[str, int] = {} for mint, vals in rows.items(): hits = 0 for c in columns: v = vals.get(c) if v is None: continue if diap.value_hits(c, v): hits += 1 res[mint] = 1 if hits >= 3 else 0 return res def loop(): conn = get_connection() logging.info("Подключено к MySQL: %s/%s", DB_HOST, DB_NAME) last_full_scan_mint: Optional[str] = None sent_allones_cache: Set[str] = set() sent_whiteok_cache: Set[str] = set() while True: try: try: conn.ping(reconnect=True) except Exception: conn = get_connection() mints = select_mints_with_col82_null(conn, BATCH_SIZE) if not mints: # нет кандидатов на 82 — догоним рассылки 82 dogon_post_82_and_allones(conn, sent_whiteok_cache, sent_allones_cache) # сначала гарантированно зануляем 97..106 там, где 94=0 try: affected = nullify_cols_97_106_when_94_zero(conn) if affected: logging.info("col_97..106 -> NULL (94=0): %d строк", affected) except Exception as e: logging.error("Ошибка nullify 97..106 при 94=0: %s", e) # затем один проход по расчёту 97..106 try: process_cols_97_106(conn) except Exception as e: logging.error("Ошибка блока 97..106: %s", e) try: process_col_107(conn) except Exception as e: logging.error("Ошибка блока 107 (market_cap рост): %s", e) time.sleep(5) continue # есть кандидаты на 82 — считаем их батчем for mint in mints: with conn.cursor() as cur: cur.execute(""" SELECT usd_market_cap, col_78, col_81, col_82, col_93 FROM tokens WHERE LOWER(mint)=LOWER(%s) ORDER BY (col_93=1) DESC LIMIT 1 """, (mint,)) r = cur.fetchone() or {} try: usd_mc = float(r.get("usd_market_cap") or 0.0) except Exception: usd_mc = 0.0 c78 = int(r["col_78"]) if r.get("col_78") is not None else None c81 = int(r["col_81"]) if r.get("col_81") is not None else None c82 = int(r["col_82"]) if r.get("col_82") is not None else None c93 = int(r["col_93"]) if r.get("col_93") is not None else None # строго считаем только при 93=1, MC>=8000 и 78=1 if c93 != 1 or usd_mc < 8000 or c78 != 1: continue diap = _get_white_diap_for_mc(usd_mc) cols_needed = diap.columns() # если диапазонов нет — не висим в NULL: ставим 0 if not cols_needed: update_col82_atomic(conn, mint, 0) continue data = fetch_columns_for_mints(conn, [mint], cols_needed) vals = data.get(mint, {}) hits = 0 for c in cols_needed: if diap.value_hits(c, vals.get(c)): hits += 1 v = 1 if hits >= 3 else 0 changed82 = update_col82_atomic(conn, mint, v) # рассылки по 82 if c78 == 1: if changed82 and v == 1 and mint not in sent_whiteok_cache: send_tg_message(TG_BOT_TOKEN, TG_WHITE_OK_CHAT_ID, mint) sent_whiteok_cache.add(mint) if c81 == 1 and (v == 1 or c82 == 1) and mint not in sent_allones_cache: cur82 = 1 if v == 1 else c82 if cur82 == 1: send_tg_message(TG_BOT_TOKEN, TG_ALL_ONES_CHAT_ID, mint) sent_allones_cache.add(mint) # догоним рассылки 82 dogon_post_82_and_allones(conn, sent_whiteok_cache, sent_allones_cache) # перед расчётом 97..106 сначала занулим их там, где 94=0 try: affected = nullify_cols_97_106_when_94_zero(conn) if affected: logging.info("col_97..106 -> NULL (94=0): %d строк", affected) except Exception as e: logging.error("Ошибка nullify 97..106 при 94=0: %s", e) # затем — один проход по 97..106 try: process_cols_97_106(conn) except Exception as e: logging.error("Ошибка блока 97..106: %s", e) try: process_col_107(conn) except Exception as e: logging.error("Ошибка блока 107 (market_cap рост): %s", e) time.sleep(5) except KeyboardInterrupt: logging.info("Остановка по Ctrl+C") break except Exception as e: logging.error("Критическая ошибка цикла: %s", e) time.sleep(5) if __name__ == "__main__": loop()