#!/usr/bin/env python3 """ Монитор BUY-триггеров для топ-тикеров MOEX. Отслеживает ключевые уровни и условия для входа в BUY: - W1↓ → разворот (SOS, Spring, MACD crossover) - Ключевые уровни пробоя - RSI, ADX, OrderBook bias - Фундаментальные катализаторы (дивиденды, earnings) Запуск: python src/monitor_buy_triggers.py --once # однократная проверка python src/monitor_buy_triggers.py --daemon # фоновый режим (каждые 60 мин) python src/monitor_buy_triggers.py --daemon --interval 1800 # каждые 30 мин python src/monitor_buy_triggers.py --ticker SBER # только один тикер Выходной файл: reports/buy_triggers_monitor.md """ import argparse import json import os import sys import time from datetime import datetime from typing import Any, Optional import pandas as pd import numpy as np sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from src.db.connection import fetch_ohlcv_combined from src.indicators.calculations import calc_all_indicators from src.analysis.tech_analysis import determine_trend, wyckoff_phase from src.api.tbank import ( get_last_price, get_orderbook_with_history, set_tbank_env, ) PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) REPORT_FILE = os.path.join(PROJECT_ROOT, "reports", "buy_triggers_monitor.md") # ── Конфигурация триггеров ───────────────────────────────── # Каждый тикер: ключевые уровни для мониторинга BUY_TRIGGERS = { "SBER": { "name": "Сбербанк", "sector": "Банки", "triggers": [ {"level": "SOS >313", "desc": "Пробой 313 ₽ на объёме >55M — начало разворота W1"}, {"level": "LPS 296-300", "desc": "ST с низким объёмом в зоне 296-300 — точка входа в накопление"}, {"level": "Pre-div rally", "desc": "Рост перед дивидендной отсечкой 20.07 (DY 12.4%)"}, ], "sos_level": 313.0, "sos_volume": 55_000_000, "div_ex_date": "2026-07-20", "div_amount": 37.64, }, "PLZL": { "name": "Полюс Золото", "sector": "Металлы и Добыча", "triggers": [ {"level": "SOS >2,017", "desc": "Пробой 2,017 ₽ на объёме >1M D1 — подтверждение накопления"}, {"level": "Spring 1,920", "desc": "Ложный пробой 1,920 + быстрый возврат — точка входа"}, {"level": "W1 close >2,017", "desc": "Закрытие недели выше 2,017 — снятие W1↓ блока"}, ], "sos_level": 2017.0, "sos_volume": 1_000_000, "div_ex_date": "2026-07-13", "div_amount": 29.05, }, "ASTR": { "name": "Group Astra", "sector": "ИТ", "triggers": [ {"level": "SOS >179", "desc": "Пробой 179 ₽ на объёме >2.5M — завершение накопления"}, {"level": "Spring 165-167", "desc": "Ложный пробой поддержки + возврат — вход в лонг"}, {"level": "W1 close >170", "desc": "Недельное закрытие выше EMA50 — ослабление W1↓"}, ], "sos_level": 179.0, "sos_volume": 2_500_000, }, "PHOR": { "name": "ФосАгро", "sector": "Химия и Удобрения", "triggers": [ {"level": "SOS >5,375", "desc": "Пробой 5,375 на объёме — подтверждение накопления"}, {"level": "RSI D1 >30", "desc": "Выход RSI из oversold — разрешение на BUY"}, ], "sos_level": 5375.0, }, "ROSN": { "name": "Роснефть", "sector": "Нефть и Газ", "triggers": [ {"level": "D1 close >320", "desc": "Закрытие выше 320 на V>12M — снятие D1↓ блока"}, {"level": "Brent >$75", "desc": "Стабилизация нефти выше $75 — улучшение фундамента"}, {"level": "W1 close >327", "desc": "Недельное закрытие выше EMA50 — снятие W1↓ блока"}, ], "sos_level": 320.0, "sos_volume": 12_000_000, }, "LKOH": { "name": "Лукойл", "sector": "Нефть и Газ", "triggers": [ {"level": "W1 close >4,474", "desc": "Закрытие выше TR — снятие W1↓ и отмена SELL"}, {"level": "Brent >$75", "desc": "Стабилизация нефти — улучшение прогнозов"}, ], "sos_level": 4474.0, }, "X5": { "name": "X5 Group", "sector": "Потребительский", "triggers": [ {"level": "W1 ADX <40", "desc": "Ослабление тренда — снижение давления SELL"}, {"level": "MACD D1 crossover", "desc": "Пересечение MACD выше 0 — смена импульса"}, ], }, } def safe_float(val: Any, default: float = 0.0) -> float: try: if val is None or (isinstance(val, float) and np.isnan(val)): return default return float(val) except (TypeError, ValueError): return default def get_last_val(df: pd.DataFrame, col: str) -> Optional[float]: if col in df.columns and len(df) > 0: v = df[col].iloc[-1] return float(v) if pd.notna(v) else None return None def classify_ticker(ticker: str) -> str: MOEX_TICKERS_SET = { 'SBER', 'GAZP', 'LKOH', 'ROSN', 'NVTK', 'PLZL', 'PHOR', 'MTSS', 'VTBR', 'MOEX', 'X5', 'ASTR', 'SNGSP', } t = ticker.upper() if t in MOEX_TICKERS_SET: return 'MOEX' if 'BITCOIN' in t or 'BTC' in t or 'ETH' in t: return 'CRYPTO' if 'USD' in t or 'EUR' in t or 'RUB' in t: return 'FOREX' return 'UNKNOWN' def check_ticker(ticker: str) -> dict: """ Проверить текущее состояние тикера относительно BUY-триггеров. Возвращает словарь с результатами. """ result = { "ticker": ticker, "name": BUY_TRIGGERS.get(ticker, {}).get("name", ticker), "price": None, "error": None, "trend_w1": "—", "trend_d1": "—", "trend_h1": "—", "rsi_d1": None, "rsi_h1": None, "macd_d1": None, "macd_hist_d1": None, "adx_d1": None, "wyckoff_phase": "—", "ob_verdict": "—", "ob_bias": "—", "triggers": [], "buy_readiness": 0, # 0-100% "alerts": [], } # ── Загрузка цены ── try: price = get_last_price(ticker) result["price"] = round(float(price), 2) if price else None except Exception as e: result["error"] = f"Price: {e}" # ── Загрузка OHLCV ── df_h1 = df_d1 = df_w1 = pd.DataFrame() try: df_h1 = fetch_ohlcv_combined(ticker, 'H1', limit=100) df_d1 = fetch_ohlcv_combined(ticker, 'D1', limit=100) df_w1 = fetch_ohlcv_combined(ticker, 'W1', limit=100) except Exception as e: result["error"] = f"DB: {e}" has_h1 = len(df_h1) > 10 has_d1 = len(df_d1) > 10 has_w1 = len(df_w1) > 10 # ── Расчёт индикаторов ── if has_h1: df_h1 = calc_all_indicators(df_h1) if has_d1: df_d1 = calc_all_indicators(df_d1) if has_w1: df_w1 = calc_all_indicators(df_w1) # ── Тренды ── if has_w1: t, a = determine_trend(df_w1) result["trend_w1"] = t result["adx_w1"] = round(safe_float(a), 1) if a else None if has_d1: t, a = determine_trend(df_d1) result["trend_d1"] = t result["adx_d1"] = round(safe_float(a), 1) if a else None if has_h1: t, a = determine_trend(df_h1) result["trend_h1"] = t result["adx_h1"] = round(safe_float(a), 1) if a else None # ── RSI, MACD ── if has_d1: last_d1 = df_d1.iloc[-1] result["rsi_d1"] = round(safe_float(last_d1.get('RSI_14')), 1) result["macd_d1"] = round(safe_float(last_d1.get('MACD_line')), 2) hist = safe_float(last_d1.get('MACD_hist')) result["macd_hist_d1"] = round(hist, 2) if hist is not None else None if has_h1: last_h1 = df_h1.iloc[-1] result["rsi_h1"] = round(safe_float(last_h1.get('RSI_14')), 1) # ── Wyckoff ── if has_w1: wk = wyckoff_phase(df_w1) result["wyckoff_phase"] = wk.get('phase', '—') # ── OrderBook ── if classify_ticker(ticker) == 'MOEX': try: ob = get_orderbook_with_history(ticker, depth=10, max_age_sec=180) snap = ob['live']['snapshot'] result["ob_verdict"] = ob.get('combined_verdict', ob['live'].get('verdict', '—')) result["ob_imbalance"] = round(snap.get('imbalance_ratio', 0), 2) result["ob_spread"] = round(snap.get('spread_pct', 0), 4) if ob.get('history') and ob['history'].get('bias'): hb = ob['history']['bias'] result["ob_bias"] = f"{hb['label']} (🟢{hb['bullish']}🔴{hb['bearish']})" except Exception: pass # ── Проверка триггеров ── price_val = result["price"] or 0 triggers_config = BUY_TRIGGERS.get(ticker, {}).get("triggers", []) sos_level = BUY_TRIGGERS.get(ticker, {}).get("sos_level") sos_volume = BUY_TRIGGERS.get(ticker, {}).get("sos_volume") trigger_results = [] alerts = [] buy_signals = 0 total_triggers = len(triggers_config) # Проверка SOS (Sign of Strength) — пробой ключевого уровня if sos_level and price_val > 0: w1_down = result["trend_w1"] in ("down",) above_sos = price_val >= sos_level # Проверка объёма D1 для SOS volume_ok = False if has_d1 and sos_volume: last_vol = safe_float(df_d1['Volume'].iloc[-1]) if 'Volume' in df_d1.columns else 0 volume_ok = last_vol >= sos_volume if above_sos: status = "🟢 ДА" if volume_ok else "🟡 Почти (объём мал)" buy_signals += 1 if volume_ok else 0 alerts.append(f"🚀 Цена ВЫШЕ {sos_level:.0f}! " + ("Объём подтверждает ✅" if volume_ok else "Ждём объём >{sos_volume:,}")) else: pct_to_sos = (sos_level - price_val) / sos_level * 100 status = f"🔴 НЕТ ({pct_to_sos:+.1f}% до {sos_level:.0f})" trigger_results.append({ "name": f"SOS >{sos_level:.0f}" if sos_level else "SOS", "result": "above" if above_sos else "below", "status": status, "distance_pct": round((price_val - sos_level) / sos_level * 100, 1) if sos_level else None, }) # Проверка W1↓ — главный блок w1_block = result["trend_w1"] == "down" if w1_block: alerts.append("🔴 W1↓ блокирует BUY (X5 Post-Mortem)") else: alerts.append(f"🟢 W1↓ снят! Тренд: {result['trend_w1']}") buy_signals += 2 # Проверка RSI D1 (не должен быть в перекупленности) rsi_d1 = result.get("rsi_d1") if rsi_d1 is not None: if rsi_d1 < 30: alerts.append(f"🟡 RSI D1={rsi_d1} (oversold) — экстремум, возможен отскок") buy_signals += 1 elif rsi_d1 > 70: alerts.append(f"🔴 RSI D1={rsi_d1} (перекуплен) — не время для BUY") else: if w1_block: pass # W1↓ блокирует в любом случае else: buy_signals += 1 # Проверка MACD D1 crossover macd_hist = result.get("macd_hist_d1") if macd_hist is not None: if macd_hist > 0 and macd_hist > result.get("_prev_macd_hist", 0): alerts.append(f"🟢 MACD D1 hist растёт ({macd_hist:.2f}) — импульс улучшается") buy_signals += 1 elif macd_hist > 0: alerts.append(f"🟡 MACD D1 hist положительный ({macd_hist:.2f}) — стабильно") else: alerts.append(f"🔴 MACD D1 hist отрицательный ({macd_hist:.2f}) — импульс медвежий") # Проверка дивидендного катализатора div_date = BUY_TRIGGERS.get(ticker, {}).get("div_ex_date") if div_date: try: div_dt = datetime.strptime(div_date, "%Y-%m-%d") days_to_div = (div_dt - datetime.now()).days if 0 <= days_to_div <= 14: div_amount = BUY_TRIGGERS.get(ticker, {}).get("div_amount", 0) dy = div_amount / price_val * 100 if price_val > 0 else 0 alerts.append(f"🎯 Дивиденды {div_amount}₽ через {days_to_div}д! DY {dy:.1f}%") buy_signals += 1 except ValueError: pass # OrderBook bias ob_bias = result.get("ob_bias", "") ob_v = result.get("ob_verdict", "") if "БЫЧИЙ" in str(ob_v) or "БЫЧИЙ" in str(ob_bias): alerts.append("🟢 OrderBook бычий — поддержка сантимента") buy_signals += 1 elif "МЕДВЕЖИЙ" in str(ob_v): alerts.append("🔴 OrderBook медвежий — осторожно") # Wyckoff фаза wk = result.get("wyckoff_phase", "") if "накопление" in wk.lower() or "accumulation" in wk.lower(): alerts.append(f"🟢 Wyckoff: {wk} — поддержка BUY-тезиса") buy_signals += 1 elif "маркдаун" in wk.lower() or "markdown" in wk.lower(): alerts.append(f"🔴 Wyckoff: {wk} — давление на цену") elif "консолидация" in wk.lower(): alerts.append(f"🟡 Wyckoff: {wk} — неопределённость") result["triggers"] = trigger_results result["alerts"] = alerts # Buy readiness: 0-100% max_score = max(total_triggers * 2 + 4, 1) # triggers + extra conditions result["buy_readiness"] = min(100, int(buy_signals / max_score * 100)) return result def generate_report(results: list[dict]) -> str: """Генерирует markdown-отчёт.""" now = datetime.now().strftime("%Y-%m-%d %H:%M") lines = [] lines.append(f"# 📊 Мониторинг BUY-триггеров\n") lines.append(f"> **Обновлено:** {now} МСК") lines.append(f"> **Режим:** {'daemon' if '--daemon' in sys.argv else 'однократный'}") lines.append(f"> **Команда:** `python src/monitor_buy_triggers.py`\n") lines.append("---\n") # Сводная таблица lines.append("## 📋 Сводная таблица\n") lines.append("| Тикер | Цена | W1 | D1 | H1 | RSI D1 | MACD Hist | Wyckoff | OB | Готовность |") lines.append("|:-----:|:---:|:--:|:--:|:--:|:------:|:---------:|:-------:|:--:|:----------:|") for r in sorted(results, key=lambda x: x.get("buy_readiness", 0), reverse=True): ticker = r["ticker"] price = f'{r["price"]:.2f}' if r["price"] else "—" tw1 = {"down": "🔴↓", "sideways": "🟡→", "up": "🟢↑"}.get(r.get("trend_w1", ""), r.get("trend_w1", "—")) td1 = {"down": "🔴↓", "sideways": "🟡→", "up": "🟢↑"}.get(r.get("trend_d1", ""), r.get("trend_d1", "—")) th1 = {"down": "🔴↓", "sideways": "🟡→", "up": "🟢↑"}.get(r.get("trend_h1", ""), r.get("trend_h1", "—")) rsi = f'{r["rsi_d1"]:.0f}' if r["rsi_d1"] is not None else "—" macd = f'{r["macd_hist_d1"]:.1f}' if r["macd_hist_d1"] is not None else "—" wk = (r.get("wyckoff_phase", "") or "—")[:20] ob = r.get("ob_verdict", "—")[:16] ready = r.get("buy_readiness", 0) # Цвет готовности if ready >= 70: ready_str = f"🟢 {ready}%" elif ready >= 40: ready_str = f"🟡 {ready}%" else: ready_str = f"🔴 {ready}%" lines.append(f"| **{ticker}** | {price} | {tw1} | {td1} | {th1} | {rsi} | {macd} | {wk} | {ob} | {ready_str} |") lines.append("") lines.append("---\n") # Детальные карточки тикеров lines.append("## 🔍 Детальная проверка\n") for r in sorted(results, key=lambda x: x.get("buy_readiness", 0), reverse=True): ticker = r["ticker"] name = r.get("name", ticker) price = f'{r["price"]:.2f} ₽' if r["price"] else "N/A" ready = r.get("buy_readiness", 0) if ready >= 70: badge = "🟢 ГОТОВ К BUY" elif ready >= 40: badge = "🟡 ПРЕДВАРИТЕЛЬНО" else: badge = "🔴 НЕ ГОТОВ" lines.append(f"### {ticker} ({name}) — {badge}\n") lines.append(f"| Параметр | Значение |") lines.append(f"|----------|---------|") lines.append(f"| **Цена** | {price} |") lines.append(f"| **Тренд W1/D1/H1** | {r.get('trend_w1','?')} / {r.get('trend_d1','?')} / {r.get('trend_h1','?')} |") lines.append(f"| **RSI D1/H1** | {r.get('rsi_d1','?')} / {r.get('rsi_h1','?')} |") lines.append(f"| **MACD Hist D1** | {r.get('macd_hist_d1','?')} |") lines.append(f"| **Wyckoff** | {r.get('wyckoff_phase','?')} |") lines.append(f"| **ADX D1** | {r.get('adx_d1','?')} |") lines.append(f"| **OrderBook** | {r.get('ob_verdict','—')} (imb={r.get('ob_imbalance','?')}) |") lines.append(f"| **OB file bias** | {r.get('ob_bias','—')} |") lines.append(f"| **Готовность** | **{ready}%** |") lines.append("") # Алерты if r.get("alerts"): lines.append("**Сигналы:**\n") for a in r["alerts"]: lines.append(f"- {a}") lines.append("") # Триггеры cfg_triggers = BUY_TRIGGERS.get(ticker, {}).get("triggers", []) if cfg_triggers: lines.append("**Отслеживаемые триггеры:**\n") lines.append("| Триггер | Описание | Статус |") lines.append("|---------|----------|:------:|") sos_level = BUY_TRIGGERS.get(ticker, {}).get("sos_level") for t in cfg_triggers: level_name = t["level"] # Определяем статус из результатов проверки trig_status = "⏳" if r.get("triggers") and level_name == r["triggers"][0].get("name", ""): trig_status = r["triggers"][0].get("status", "⏳") elif "W1" in level_name and r.get("trend_w1") != "down": trig_status = "🟢 Выполнен" elif "W1" in level_name and r.get("trend_w1") == "down": trig_status = "🔴 Нет" lines.append(f"| **{level_name}** | {t['desc']} | {trig_status} |") lines.append("") lines.append("---\n") lines.append("\n*Сгенерировано автоматически. Для обновления запустите `python src/monitor_buy_triggers.py --once`*") return "\n".join(lines) def main(): parser = argparse.ArgumentParser(description="Монитор BUY-триггеров MOEX") parser.add_argument("--daemon", action="store_true", help="Фоновый режим") parser.add_argument("--once", action="store_true", help="Однократная проверка") parser.add_argument("--interval", type=int, default=3600, help="Интервал в секундах (по умолч. 3600 = 1ч)") parser.add_argument("--ticker", type=str, help="Только один тикер") args = parser.parse_args() # Переключаемся на real API try: set_tbank_env('real') except Exception as e: print(f"⚠️ Не удалось переключить API: {e}") tickers_to_monitor = list(BUY_TRIGGERS.keys()) if args.ticker: if args.ticker.upper() not in BUY_TRIGGERS: print(f"❌ Тикер {args.ticker} не найден в конфигурации монитора") sys.exit(1) tickers_to_monitor = [args.ticker.upper()] def run_check(): print(f"\n{'='*60}") print(f"📊 МОНИТОРИНГ BUY-ТРИГГЕРОВ") print(f"{'='*60}") print(f"Время: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} МСК\n") results = [] for ticker in tickers_to_monitor: print(f" Проверка {ticker}...", end=" ", flush=True) try: r = check_ticker(ticker) results.append(r) ready = r.get("buy_readiness", 0) price = r.get("price", 0) w1 = r.get("trend_w1", "?") print(f"{price:.2f} ₽ | W1:{w1} | Готовность: {ready}%") if r.get("alerts"): for a in r["alerts"][:3]: # топ-3 алерта print(f" {a}") except Exception as e: print(f"❌ Ошибка: {e}") # Генерация и сохранение отчёта report = generate_report(results) os.makedirs(os.path.dirname(REPORT_FILE), exist_ok=True) with open(REPORT_FILE, "w", encoding="utf-8") as f: f.write(report) print(f"\n✅ Отчёт сохранён: {REPORT_FILE}") print(f" {len(results)} тикеров проверено") # Вывод топ-готовых sorted_results = sorted(results, key=lambda x: x.get("buy_readiness", 0), reverse=True) if sorted_results and sorted_results[0]["buy_readiness"] >= 60: best = sorted_results[0] print(f"\n🎯 Лучший кандидат: {best['ticker']} ({best.get('name','')}) — готовность {best['buy_readiness']}%") if best.get("alerts"): print(f" Сигналы:") for a in best["alerts"]: print(f" • {a}") return results if args.daemon: print(f"🟢 Фоновый мониторинг запущен. Интервал: {args.interval}с ({args.interval//60} мин)") while True: try: run_check() time.sleep(args.interval) except KeyboardInterrupt: print("\n⏹ Монитор остановлен пользователем") break except Exception as e: print(f"❌ Ошибка: {e}") time.sleep(60) else: run_check() if __name__ == "__main__": main()