#!/usr/bin/env python3 """ ============================================================================= filter_watchlist.py — Outil de filtrage de watchlist par liquidité ============================================================================= OBJECTIF -------- Éliminer de la watchlist les paires qui ont un risque élevé de slippage en production. Le rapport d'audit a montré que les paires problématiques en prod (TONUSDC 0% WR, NOTUSDC 0% WR, DASHUSDC 20% WR) étaient des paires USDC à faible volume. CRITÈRES APPLIQUÉS (ajustables en haut du fichier) -------------------------------------------------- 1. Quote currency = USDT uniquement (USDC souvent moins liquide) 2. Volume 24h > 5 M$ (sinon spread > 0.3% probable) 3. Pas de stablecoin 4. Pas de leveraged token 5. Spread bid/ask < 0.1% au moment du check UTILISATION ----------- # Mode analyse — propose une watchlist python3 filter_watchlist.py --propose # Mode validation — teste une watchlist existante python3 filter_watchlist.py --check watchlist.txt # Mode export — produit le fichier au format du bot python3 filter_watchlist.py --propose --export bot_watchlist.txt ============================================================================= """ import argparse import json import sys from pathlib import Path try: import requests except ImportError: print("ERREUR : pip install requests") sys.exit(1) # Critères — à ajuster selon expérience MIN_VOLUME_24H_USDT = 5_000_000 # 5 millions USDT par jour MAX_SPREAD_PCT = 0.10 # spread < 0.1% QUOTE_CURRENCY = "USDT" # uniquement USDT (pas USDC) EXCLUDE_PATTERNS = ( "USDC", "BUSD", "TUSD", "FDUSD", "DAIUSDT", "EURUSDT", "GBPUSDT", "USD1USDT", ) LEVERAGED_PATTERNS = ("UPUSDT", "DOWNUSDT", "BULLUSDT", "BEARUSDT") BINANCE_BASE = "https://data-api.binance.vision" HTTP_TIMEOUT = 15 def fetch_all_tickers() -> list: r = requests.get(f"{BINANCE_BASE}/api/v3/ticker/24hr", timeout=HTTP_TIMEOUT) r.raise_for_status() return r.json() def fetch_orderbook_ticker(symbol: str) -> dict: r = requests.get( f"{BINANCE_BASE}/api/v3/ticker/bookTicker", params={"symbol": symbol}, timeout=HTTP_TIMEOUT, ) r.raise_for_status() return r.json() def is_eligible(ticker: dict) -> tuple: """Retourne (bool, reason).""" sym = ticker["symbol"] if not sym.endswith(QUOTE_CURRENCY): return False, f"non-{QUOTE_CURRENCY}" # On retire le suffixe pour vérifier si la base est un stablecoin base = sym[:-len(QUOTE_CURRENCY)] if base in {"USDC", "BUSD", "TUSD", "FDUSD", "DAI", "USD1", "EUR", "GBP", "TRY"}: return False, "stablecoin/fiat" if any(p in sym for p in EXCLUDE_PATTERNS if p != sym): return False, "stablecoin/exclu" if any(sym.endswith(p) for p in LEVERAGED_PATTERNS): return False, "leveraged token" if sym.startswith("1000"): return False, "1000x token" try: vol = float(ticker["quoteVolume"]) except (KeyError, ValueError): return False, "volume invalide" if vol < MIN_VOLUME_24H_USDT: return False, f"volume < {MIN_VOLUME_24H_USDT/1e6:.0f}M$" return True, "ok" def propose_watchlist(verify_spread: bool = True, max_size: int = 50) -> list: """Construit une watchlist propre à partir du marché actuel.""" print(f"Téléchargement des tickers Binance...") tickers = fetch_all_tickers() print(f" {len(tickers)} tickers reçus") eligible = [] rejected_count = {} for t in tickers: ok, reason = is_eligible(t) if ok: eligible.append(t) else: rejected_count[reason] = rejected_count.get(reason, 0) + 1 print(f"\n--- Filtrage initial (volume + format) ---") print(f" Éligibles : {len(eligible)}") print(f" Rejetés :") for reason, count in sorted(rejected_count.items(), key=lambda x: -x[1]): print(f" {reason:<35}: {count}") # Tri par volume décroissant eligible.sort(key=lambda x: float(x["quoteVolume"]), reverse=True) # Si demandé, vérifier le spread bid/ask en direct final = [] if verify_spread: print(f"\n--- Vérification du spread bid/ask sur top {max_size*2} ---") candidates = eligible[: max_size * 2] for t in candidates: sym = t["symbol"] try: book = fetch_orderbook_ticker(sym) bid = float(book["bidPrice"]) ask = float(book["askPrice"]) if bid <= 0: continue spread = (ask - bid) / bid * 100 if spread <= MAX_SPREAD_PCT: t["spread_pct"] = spread final.append(t) if len(final) >= max_size: break except Exception: continue else: final = eligible[:max_size] return final def print_watchlist(watchlist: list) -> None: print(f"\n--- WATCHLIST PROPOSÉE ({len(watchlist)} paires) ---\n") print(f"{'Rang':<6}{'Symbol':<14}{'Vol 24h (M$)':>16}{'Change 24h':>14}{'Spread':>10}") print("-" * 60) for i, t in enumerate(watchlist, 1): vol_m = float(t["quoteVolume"]) / 1e6 change = float(t["priceChangePercent"]) spread = t.get("spread_pct", 0) print(f"{i:<6}{t['symbol']:<14}{vol_m:>15,.1f}{change:>13.2f}%{spread:>9.3f}%") def check_existing(path: Path) -> None: """Vérifie une watchlist existante contre les critères.""" print(f"Lecture de {path}...") if path.suffix == ".json": symbols = json.loads(path.read_text()) if isinstance(symbols, dict): symbols = list(symbols.keys()) else: symbols = [l.strip() for l in path.read_text().splitlines() if l.strip()] print(f" {len(symbols)} paires à vérifier\n") tickers = fetch_all_tickers() by_sym = {t["symbol"]: t for t in tickers} print(f"{'Symbol':<14}{'Vol 24h':>14}{'Verdict':<30}") print("-" * 60) keep = [] drop = [] for s in symbols: t = by_sym.get(s) if not t: print(f"{s:<14}{'?':>14} ✗ inexistante") drop.append(s) continue ok, reason = is_eligible(t) vol_m = float(t["quoteVolume"]) / 1e6 if ok: print(f"{s:<14}{vol_m:>13,.1f}M ✓ ok") keep.append(s) else: print(f"{s:<14}{vol_m:>13,.1f}M ✗ {reason}") drop.append(s) print(f"\nÀ garder : {len(keep)} | À retirer : {len(drop)}") def main(): parser = argparse.ArgumentParser() g = parser.add_mutually_exclusive_group(required=True) g.add_argument("--propose", action="store_true", help="Proposer une nouvelle watchlist") g.add_argument("--check", type=Path, help="Vérifier une watchlist existante (.txt ou .json)") parser.add_argument("--max-size", type=int, default=30, help="Taille max de la watchlist proposée") parser.add_argument("--no-verify-spread", action="store_true", help="Skip la vérif spread (plus rapide)") parser.add_argument("--export", type=Path, help="Exporter la watchlist proposée dans un fichier") args = parser.parse_args() if args.propose: wl = propose_watchlist( verify_spread=not args.no_verify_spread, max_size=args.max_size, ) print_watchlist(wl) if args.export: args.export.write_text("\n".join(t["symbol"] for t in wl)) print(f"\nExporté → {args.export}") else: check_existing(args.check) if __name__ == "__main__": main()