"""
Squeeze-VWAP Bot — Market Screener
=====================================
3-фазное сканирование:
1. Volume + Trades фильтр (1 API call)
2. ATR волатильность фильтр
3. Squeeze + Z-VWAP combo signal → watchlist
Ищет монеты где:
- Squeeze сжат ИЛИ только что released
- Z-VWAP отклонился от fair value
- Waddah показывает силу
"""
import time
import json
import logging
from datetime import datetime, timezone
from src.config import (
BLACKLIST, MIN_VOLUME_24H, MIN_TRADES_24H,
ATR_MIN_PCT, ATR_PERIOD, NATR_5M_MIN,
TIMEFRAME, WATCHLIST_FILE, COOLDOWNS_FILE, COOLDOWN_MIN,
ZVWAP_ENTRY_THRESHOLD,
)
from src.indicators import (
calc_atr_pct, calc_squeeze_momentum, calc_z_vwap,
calc_waddah_attar, calc_adx, calc_combo_signal,
)
logger = logging.getLogger("screener")
class Screener:
def __init__(self, exchange, notifier=None):
self.exchange = exchange
self.notifier = notifier
self.watchlist = []
self.get_open_positions = None # callback
self._load_watchlist()
# ============================================================
# PHASE 1: Quick filter (1 API call)
# ============================================================
def get_candidates(self):
"""Фильтр первого уровня — vol + trades + blacklist."""
tickers = self.exchange.get_all_tickers_24h()
candidates = []
for t in tickers:
symbol = t["symbol"]
if not symbol.endswith("USDT"):
continue
if symbol in BLACKLIST:
continue
volume_24h = float(t.get("quoteVolume", 0))
trades_24h = int(t.get("count", 0))
if volume_24h < MIN_VOLUME_24H:
continue
if trades_24h < MIN_TRADES_24H:
continue
candidates.append({
"symbol": symbol,
"volume_24h": volume_24h,
"trades_24h": trades_24h,
"price": float(t.get("lastPrice", 0)),
"change_pct": float(t.get("priceChangePercent", 0)),
})
logger.info(
f"Phase 1: {len(candidates)} candidates "
f"(vol>${MIN_VOLUME_24H / 1e6:.0f}M, trades>{MIN_TRADES_24H / 1e3:.0f}K)"
)
return candidates
# ============================================================
# PHASE 2: ATR volatility filter
# ============================================================
def filter_by_atr(self, candidates):
"""Фильтр второго уровня — ATR(14)/1h >= порога + NATR(14)/5m >= 1.2%."""
passed = []
for c in candidates:
symbol = c["symbol"]
try:
# ATR on 1h
klines = self.exchange.get_klines(symbol, "1h", limit=ATR_PERIOD + 5)
if len(klines) < ATR_PERIOD + 1:
continue
atr_pct = calc_atr_pct(klines, ATR_PERIOD)
if atr_pct is None:
continue
if atr_pct < ATR_MIN_PCT:
continue
# NATR on 5m — skip sleepy coins
klines_5m = self.exchange.get_klines(symbol, "5m", limit=ATR_PERIOD + 5)
if len(klines_5m) >= ATR_PERIOD + 1:
natr_5m = calc_atr_pct(klines_5m, ATR_PERIOD)
if natr_5m is not None and natr_5m < NATR_5M_MIN:
logger.debug(f"NATR/5m skip {symbol}: {natr_5m:.2f}% < {NATR_5M_MIN}%")
continue
c["natr_5m"] = round(natr_5m, 2) if natr_5m else None
c["atr_pct"] = round(atr_pct, 2)
passed.append(c)
time.sleep(0.05)
except Exception as e:
logger.debug(f"ATR error {symbol}: {e}")
continue
logger.info(f"Phase 2: {len(passed)} passed ATR filter (>={ATR_MIN_PCT}%) + NATR/5m>={NATR_5M_MIN}%")
return passed
# ============================================================
# PHASE 3: Squeeze + Z-VWAP combo → watchlist
# ============================================================
def check_combo_signals(self, candidates):
"""
Фильтр третьего уровня — полный combo signal.
Добавляет в watchlist если:
- Z-VWAP отклонился (|Z| > threshold) → direction
- Squeeze: в сжатии ИЛИ только что released
- Любой дополнительный confluence (Waddah, histogram, ADX)
Минимальный score для watchlist: 2 (direction + хотя бы 1 confluence)
Score >= 3 нужен для ВХОДА (проверяется в manager)
"""
new_entries = []
cooldowns = self._load_cooldowns()
for c in candidates:
symbol = c["symbol"]
# Cooldown check
if symbol in cooldowns and time.time() < cooldowns[symbol]:
continue
try:
# 300 свечей 5m для всех индикаторов
klines = self.exchange.get_klines(symbol, TIMEFRAME, limit=300)
if len(klines) < 100:
continue
combo = calc_combo_signal(klines)
if combo is None:
continue
# Минимум score 2 для watchlist (direction + 1 confluence)
if combo['score'] < 2:
continue
entry = {
"symbol": symbol,
"direction": combo['direction'], # 1=long, -1=short
"score": combo['score'],
"z_score": round(combo['zvwap']['z_score'], 2),
"vwap": round(combo['zvwap']['vwap'], 6),
"is_squeeze": bool(combo['squeeze']['is_squeeze']),
"squeeze_released": bool(combo['squeeze']['squeeze_released']),
"histogram": round(combo['squeeze']['histogram'], 6),
"waddah_strong": bool(combo['waddah']['is_strong']),
"waddah_dir": combo['waddah']['direction'],
"adx": round(combo['adx']['adx'], 1),
"is_trending": bool(combo['adx']['is_trending']),
"reasons": combo['reasons'],
"price": float(klines[-2][4]),
"atr_pct": c.get("atr_pct", 0),
"volume_24h": c["volume_24h"],
"added_at": datetime.now(timezone.utc).isoformat(),
}
new_entries.append(entry)
dir_str = "🟢 LONG" if combo['direction'] == 1 else "🔴 SHORT"
logger.info(
f"→ WATCHLIST: {symbol} {dir_str} score={combo['score']}/5 "
f"Z={combo['zvwap']['z_score']:.2f} "
f"sqz={'SQZ' if combo['squeeze']['is_squeeze'] else 'REL' if combo['squeeze']['squeeze_released'] else '---'} "
f"wad={'STR' if combo['waddah']['is_strong'] else 'low'}"
)
time.sleep(0.05)
except Exception as e:
logger.debug(f"Combo error {symbol}: {e}")
continue
# Update watchlist — add new, skip duplicates & open positions
existing_symbols = {e["symbol"] for e in self.watchlist}
open_positions = set()
if self.get_open_positions:
try:
open_positions = set(self.get_open_positions().keys())
except Exception:
pass
actually_added = []
for entry in new_entries:
if entry["symbol"] in open_positions:
continue
if entry["symbol"] not in existing_symbols:
self.watchlist.append(entry)
actually_added.append(entry)
else:
# Update existing entry with fresh data
for i, existing in enumerate(self.watchlist):
if existing["symbol"] == entry["symbol"]:
self.watchlist[i] = entry
break
self._save_watchlist()
logger.info(
f"Phase 3: {len(actually_added)} new, "
f"{len(self.watchlist)} total in watchlist"
)
# Notify new entries
for entry in actually_added:
dir_str = "🟢 LONG" if entry["direction"] == 1 else "🔴 SHORT"
sqz_str = "🔴 In Squeeze" if entry["is_squeeze"] else ("💥 Released!" if entry["squeeze_released"] else "⚪ No squeeze")
reasons_str = "\n".join(f" • {r}" for r in entry["reasons"])
msg = (
f"📋 *WATCHLIST: {entry['symbol']}*\n"
f"{dir_str} | Score: {entry['score']}/5\n"
f"Z-VWAP: {entry['z_score']:+.2f}\n"
f"Squeeze: {sqz_str}\n"
f"Waddah: {'💪 Strong' if entry['waddah_strong'] else '🤏 Weak'}\n"
f"ADX: {entry['adx']:.0f} ({'trending ⚠️' if entry['is_trending'] else 'ranging ✓'})\n"
f"{reasons_str}\n"
f"⏳ Ждём score ≥ 3 для входа..."
)
self._notify(msg)
return actually_added
# ============================================================
# FULL SCAN
# ============================================================
def _cleanup_stale_watchlist(self):
"""Убрать: 1) open positions, 2) старше 2ч, 3) Z вернулся к neutral."""
if not self.watchlist:
return
open_positions = set()
if self.get_open_positions:
try:
open_positions = set(self.get_open_positions().keys())
except Exception:
pass
now = datetime.now(timezone.utc)
before = len(self.watchlist)
fresh = []
for entry in self.watchlist:
symbol = entry["symbol"]
# Skip if has open position
if symbol in open_positions:
continue
# TTL: 2 hours max
added = entry.get("added_at", "")
if added:
try:
age_min = (now - datetime.fromisoformat(added)).total_seconds() / 60
if age_min > 120:
logger.info(f"Watchlist expired: {symbol} (age={age_min:.0f}min)")
continue
except Exception:
pass
# Z-VWAP check: if Z returned to neutral, remove
# (only after 15 min grace period)
try:
age_min_check = 0
if added:
try:
age_min_check = (now - datetime.fromisoformat(added)).total_seconds() / 60
except Exception:
pass
if age_min_check >= 15:
klines = self.exchange.get_klines(symbol, TIMEFRAME, limit=100)
if len(klines) >= 60:
zvwap = calc_z_vwap(klines)
if zvwap and abs(zvwap['z_score']) < 0.8:
logger.info(
f"Watchlist removed: {symbol} Z returned to neutral "
f"(Z={zvwap['z_score']:.2f})"
)
continue
time.sleep(0.05)
except Exception:
pass
fresh.append(entry)
self.watchlist = fresh
if len(fresh) < before:
self._save_watchlist()
def run_scan(self):
"""Полный цикл сканирования."""
logger.info("=" * 40)
logger.info("SCAN START")
start = time.time()
self._cleanup_stale_watchlist()
candidates = self.get_candidates()
if not candidates:
logger.warning("No candidates found")
return []
atr_passed = self.filter_by_atr(candidates)
if not atr_passed:
logger.info("No candidates passed ATR filter")
return []
new_entries = self.check_combo_signals(atr_passed)
elapsed = round(time.time() - start, 1)
logger.info(
f"SCAN DONE in {elapsed}s | "
f"candidates={len(candidates)} → ATR={len(atr_passed)} → "
f"new WL={len(new_entries)}"
)
return new_entries
# ============================================================
# WATCHLIST MANAGEMENT
# ============================================================
def remove_from_watchlist(self, symbol):
"""Убрать символ из watchlist."""
self.watchlist = [e for e in self.watchlist if e["symbol"] != symbol]
self._save_watchlist()
def get_watchlist(self):
"""Текущий watchlist."""
return self.watchlist.copy()
def add_cooldown(self, symbol):
"""Добавить cooldown после стопа."""
cooldowns = self._load_cooldowns()
cooldowns[symbol] = time.time() + (COOLDOWN_MIN * 60)
self._save_cooldowns(cooldowns)
logger.info(f"Cooldown {COOLDOWN_MIN}min set for {symbol}")
# ============================================================
# PERSISTENCE
# ============================================================
def _load_watchlist(self):
try:
with open(WATCHLIST_FILE, "r") as f:
data = json.load(f)
self.watchlist = data.get("watchlist", []) if isinstance(data, dict) else data
except Exception:
self.watchlist = []
def _save_watchlist(self):
try:
with open(WATCHLIST_FILE, "w") as f:
json.dump({
"watchlist": self.watchlist,
"updated_at": datetime.now(timezone.utc).isoformat(),
}, f, indent=2)
except Exception as e:
logger.error(f"Error saving watchlist: {e}")
def _load_cooldowns(self):
try:
with open(COOLDOWNS_FILE, "r") as f:
cooldowns = json.load(f)
now = time.time()
return {k: v for k, v in cooldowns.items() if v > now}
except Exception:
return {}
def _save_cooldowns(self, cooldowns):
try:
with open(COOLDOWNS_FILE, "w") as f:
json.dump(cooldowns, f, indent=2)
except Exception as e:
logger.error(f"Error saving cooldowns: {e}")
def _notify(self, msg):
if self.notifier:
try:
self.notifier(msg)
except Exception as e:
logger.debug(f"Notify error: {e}")
📜 Git History
c6f6bd5chore: initial commit — version control setup5 weeks ago
Show last diff
Loading...