""" Grid Bot — Trading Loop + HTTP REST API. Run: python main.py [--dry-run] [--once] REST endpoints: GET /api/status — status + quotes GET /api/balance — account/demo balance GET /api/grid — current grid state GET GET /api/orderbook — order book (bids/asks) GET /api/logs — trade log GET /api/settings — current settings POST /api/settings — update settings (grid_levels, step_percent, take_profit_percent, demo_mode) POST /api/bot/start — start trading loop POST /api/bot/stop — stop trading loop POST /api/reset — reset demo balance """ import argparse import asyncio import base64 import json import logging import os import signal import socket import ssl import subprocess import sys import time import urllib.parse import urllib.request from datetime import datetime, timezone from collections import deque from pathlib import Path from threading import Thread, Lock from typing import Optional from flask import Flask, jsonify, request, render_template, redirect, send_from_directory # ─── Local modules ───────────────────────────────────────────────────────────── from config import ( TRADERNET_PUBLIC_KEY, TRADERNET_PRIVATE_KEY, TRADERNET_LOGIN, TRADERNET_PASSWORD, TRADERNET_BASE_URL, SYMBOL, QUOTE, BASE, GRID_LEVELS, GRID_STEP_PERCENT, GRID_TAKE_PROFIT_PERCENT, DEMO_MODE, DEMO_START_BALANCE, HOST, PORT, DEBUG, TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, GRID_STALE_DROP_PERCENT, STATE_DIR, STATE_FILE, SERVER_URL, LIVE_TRADING_DRY_RUN, KRONOS_ENABLED, KRONOS_LIVE_ENABLED, KRONOS_MODEL, KRONOS_TF_MIN, KRONOS_LOOKBACK, KRONOS_PRED_LEN, KRONOS_MIN_CONFIDENCE, KRONOS_ADVICE_FILE, WEB_USERNAME, WEB_PASSWORD, INACTIVITY_RESTART_HOURS, INACTIVITY_PRICE_DRIFT_PCT, TREND_ENABLED, TREND_LOOKBACK, TREND_THRESHOLD_PCT, TREND_MIN_CONFIDENCE, TREND_UP_ANCHOR_SHIFT, TREND_DOWN_ANCHOR_SHIFT, RECONCILE_INTERVAL_SEC, RECONCILE_AUTO_CLEAR, FEE_RATE_PERCENT, FEE_SAFETY_MULTIPLIER, GRID_MODE, # SMA-ATR strategy (2026-06-21) STRATEGY_CENTER_MODE, STRATEGY_SMA_TF_MIN, STRATEGY_SMA_PERIOD, STRATEGY_LOCK_PCT, STRATEGY_UNLOCK_PCT, STRATEGY_BIAS_SOURCE, STRATEGY_BIAS_FALLBACK, STRATEGY_SMA_CACHE_SEC, STRATEGY_LOCK_COOLDOWN_SEC, STRATEGY_ASYMMETRY_UP, STRATEGY_ASYMMETRY_DOWN, STRATEGY_ASYMMETRY_RANGE, # Kill switch (2026-06-22) KILL_SWITCH_ENABLED, KILL_DD_PCT, KILL_API_ERRORS, KILL_API_WINDOW_SEC, KILL_TG_POLL_SEC, KILL_TG_LONG_POLL, LIVE_KILL_CLOSE_POSITION, LIVE_KILL_DD_PCT, GRID_STATE_VERSION, GRID_CONFIG_KEYS, ) from api import TradernetAPI from grid import GridEngine, DemoSimulator, GridState from indicators import compute_sma, compute_atr, compute_sma_atr # Добавляем kronos/ в sys.path, чтобы model/ (исходники Kronos) находился import sys from pathlib import Path as _Path _HERE = _Path(__file__).parent.resolve() if str(_HERE / "kronos") not in sys.path: sys.path.insert(0, str(_HERE / "kronos")) # ─── Globals ───────────────────────────────────────────────────────────────── app = Flask("grid-bot", template_folder="templates") # ─── it.kolp.pro consult endpoint (added 2026-06-18) ─────────────────────── # Изолированный Blueprint: не трогает gridbot-логику, импортируется после app. try: from consult_endpoint import consult_bp app.register_blueprint(consult_bp) except Exception as _e: import logging as _l _l.getLogger("grid-bot").error("[consult] FAILED to register blueprint: %s", _e) # Не валим gridbot — endpoint просто не работает # ─── end consult endpoint ────────────────────────────────────────────────── api_client: Optional[TradernetAPI] = None grid_engine = GridEngine( levels=GRID_LEVELS, step_percent=GRID_STEP_PERCENT, take_profit_percent=GRID_TAKE_PROFIT_PERCENT, symbol=SYMBOL, mode="live" if not DEMO_MODE else "demo", grid_mode=GRID_MODE, ) demo = DemoSimulator(start_balance=DEMO_START_BALANCE) demo_orders_placed = False # Live-mode state: mirrors demo orders/position/trades for real exchange trading # Each entry: {"order_id": int, "side": "BUY"|"SELL", "price": float, "qty": float, # "level_id": int, "status": "pending"|"filled"|"cancelled"|"rejected", # "created_at": float, "raw": dict} live_pending_orders: list[dict] = [] # Live position: {"side": "BUY", "entry_price": float, "qty": float, "order_id": int, # "raw": dict} live_position: Optional[dict] = None # Live trade log: closed round-trip trades # Each: {"entry_order_id": int, "exit_order_id": int, "entry_price": float, # "exit_price": float, "qty": float, "pnl_usdc": float, "pnl_pct": float, # "ts": float} live_trade_log: list[dict] = [] live_orders_placed = False # True after we've placed initial grid BUY limits current_price = 0.0 start_price = 0.0 # price when bot was started bot_running = False bot_thread: Optional[Thread] = None last_error = "" last_quote: dict = {} # последний котировочный снимок (bid/ask/chg/vol) _price_source: str = "none" # tradernet | binance | synth | none _last_external_log_ts: float = 0.0 # rate-limit for ANTI-PHANTOM external-pos spam (module-level, persists across loop ticks) # Stale-grid info (for dashboard): did we rebuild because price dropped without fills? grid_stale = False grid_stale_drop_pct = 0.0 grid_stale_rebuilds = 0 # Inactivity-driven soft rebuild: timestamp последней реальной сделки # (BUY-fill или SELL-fill). Инициализируется в trading_loop при старте # на time.time(). Если за INACTIVITY_RESTART_HOURS часов нет сделок и # BTC отошёл от start_price на >= INACTIVITY_PRICE_DRIFT_PCT — # отменяем висящие BUY-ордера и пересобираем сетку. last_trade_ts = 0.0 inactivity_rebuilds = 0 last_inactivity_rebuild_ts = 0.0 # анти-флуд: 1 ребилд в 6 часов макс. # === KillSwitchManager (2026-06-22) =================================== # Daemon thread мониторит 3 риска в фоне: # 1. Drawdown > KILL_DD_PCT% от пикового equity (balance + unrealized position) # 2. > KILL_API_ERRORS ошибок API за KILL_API_WINDOW_SEC секунд # 3. Telegram /stop команда от TELEGRAM_CHAT_ID (long-poll) # При trigger: kill_event.set() → trading_loop замечает на следующей итерации, # cancel всех ордеров, bot_running=False. Ручной /api/bot/start для resume. class KillSwitchManager: def __init__(self): import threading self.enabled = KILL_SWITCH_ENABLED self.kill_event = threading.Event() self.kill_reason = "none" self._reason_lock = threading.Lock() self._peak_equity = 0.0 self._last_equity = 0.0 # PATCH 2026-07-23: expose current equity in get_status() self._peak_lock = threading.Lock() self._api_errors = deque(maxlen=500) self._tg_last_update_id = 0 self._running = True self._thread = None def start(self): if not self.enabled or self._thread is not None: return import threading self._thread = threading.Thread(target=self._run, daemon=True, name="killswitch") self._thread.start() logger.info(f"[KILL-SWITCH] started (DD>{KILL_DD_PCT}%, {KILL_API_ERRORS}err/{KILL_API_WINDOW_SEC}s, TG every {KILL_TG_POLL_SEC}s)") def stop(self): self._running = False if self._thread: self._thread.join(timeout=5) def update_equity(self, equity): if not self.enabled or equity <= 0: return self._last_equity = equity # PATCH 2026-07-23: expose in get_status() with self._peak_lock: if equity > self._peak_equity: self._peak_equity = equity def record_api_error(self, err=""): if not self.enabled: return self._api_errors.append((time.time(), err[:80] if err else "")) cutoff = time.time() - KILL_API_WINDOW_SEC recent = [t for t, _ in self._api_errors if t > cutoff] if len(recent) > KILL_API_ERRORS: self._trigger(f"api_errors:{len(recent)}/{KILL_API_WINDOW_SEC}s") def current_dd_pct(self, equity): with self._peak_lock: peak = self._peak_equity if peak <= 0 or equity <= 0: return 0.0 return (peak - equity) / peak * 100.0 def check_dd(self, equity, is_live: bool = False): if not self.enabled or equity <= 0: return # В live режиме можно переопределить порог через LIVE_KILL_DD_PCT (>0) threshold = LIVE_KILL_DD_PCT if (is_live and LIVE_KILL_DD_PCT > 0) else KILL_DD_PCT dd = self.current_dd_pct(equity) if dd >= threshold and self._peak_equity > 0: mode = "live" if is_live else "demo" self._trigger(f"drawdown[{mode}]:{dd:.1f}%>={threshold}%") def cancel_live(self, reason: str = "kill_switch"): """Отменяет все live-ордера и опционально закрывает позицию. Вызывается из trading_loop при is_live=True.""" global bot_running cancelled = 0 closed_position = False # 1) Отменяем все висящие ордера на бирже try: _live_cancel_all_pending() cancelled = len(live_pending_orders) if 'live_pending_orders' in globals() else 0 except Exception as e: logger.error(f"[KILL-SWITCH] live cancel failed: {e}") # 2) Опционально закрываем открытую позицию (market-sell) if LIVE_KILL_CLOSE_POSITION and 'live_position' in globals() and live_position: try: # _live_sync_position + _live_close_position — реализация может быть, # пока используем общий путь: cancel_all + alert. # Если в момент trigger позиция открыта, оставляем её на усмотрение TP/SL. logger.warning(f"[KILL-SWITCH] live position open: {live_position.get('qty')} BTC @ {live_position.get('entry_price')}, NOT auto-closing (LIVE_KILL_CLOSE_POSITION=true but no market-sell implemented yet)") closed_position = False except Exception as e: logger.error(f"[KILL-SWITCH] live close failed: {e}") bot_running = False logger.critical(f"[KILL-SWITCH] live cancelled={cancelled}, position_closed={closed_position}, reason={reason}") return {"cancelled": cancelled, "position_closed": closed_position} def is_killed(self): return self.kill_event.is_set() def get_status(self): cutoff = time.time() - KILL_API_WINDOW_SEC recent_errors = sum(1 for t, _ in self._api_errors if t > cutoff) with self._peak_lock: peak = self._peak_equity with self._reason_lock: reason = self.kill_reason return { "enabled": self.enabled, "triggered": self.kill_event.is_set(), "reason": reason, "peak_equity": round(peak, 4), "current_equity": round(self._last_equity, 4), "drawdown_pct": round(self.current_dd_pct(self._last_equity), 2), "dd_pct_threshold": KILL_DD_PCT, "dd_pct_threshold_live": LIVE_KILL_DD_PCT if LIVE_KILL_DD_PCT > 0 else KILL_DD_PCT, "live_close_position_on_kill": LIVE_KILL_CLOSE_POSITION, "api_errors_in_window": recent_errors, "api_errors_threshold": KILL_API_ERRORS, "api_window_sec": KILL_API_WINDOW_SEC, "tg_poll_sec": KILL_TG_POLL_SEC, } def reset(self): with self._reason_lock: self.kill_reason = "none" with self._peak_lock: self._peak_equity = 0.0 self._api_errors.clear() self.kill_event.clear() logger.warning("[KILL-SWITCH] RESET - kill event cleared, peak=0") def _trigger(self, reason): with self._reason_lock: self.kill_reason = reason if not self.kill_event.is_set(): self.kill_event.set() logger.critical(f"[KILL-SWITCH] TRIGGERED: {reason}") try: tg_notify(f"🚨 KILL SWITCH TRIGGERED\n{reason}\nБот будет остановлен на следующей итерации") except Exception as e: logger.warning(f"[KILL-SWITCH] tg_notify failed: {e}") def _poll_telegram(self): if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID: return False import urllib.request as _ur url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/getUpdates" params = f"offset={self._tg_last_update_id + 1}&timeout={KILL_TG_LONG_POLL}&allowed_updates=%5B%22message%22%5D" try: req = _ur.Request(f"{url}?{params}") with _ur.urlopen(req, timeout=KILL_TG_LONG_POLL + 10) as resp: data = json.loads(resp.read().decode("utf-8")) for upd in data.get("result", []): self._tg_last_update_id = max(self._tg_last_update_id, upd.get("update_id", 0)) msg = upd.get("message", {}) text = (msg.get("text") or "").strip() chat_id = str(msg.get("chat", {}).get("id", "")) if text.lower() in ("/stop", "/kill", "/shutdown") and chat_id == str(TELEGRAM_CHAT_ID): return True except Exception as e: logger.debug(f"[KILL-SWITCH] TG poll error: {e}") return False def _run(self): logger.info("[KILL-SWITCH] monitor thread running") while self._running and not self.kill_event.is_set(): try: if self._poll_telegram(): self._trigger("telegram:/stop") break except Exception as e: logger.debug(f"[KILL-SWITCH] run loop err: {e}") time.sleep(0.5) logger.info("[KILL-SWITCH] monitor thread stopped") kill_switch = KillSwitchManager() # ─── SMA-ATR strategy state (2026-06-21) ──────────────────────────────────────── # grid_locked: True when |price - sma| > LOCK_PCT% — сетка flat, ждём возврата. # sma_center: последнее вычисленное SMA(STRATEGY_SMA_PERIOD, STRATEGY_SMA_TF_MIN). # last_sma_update_ts: epoch — для кэша (обновляется раз в STRATEGY_SMA_CACHE_SEC). # last_lock_state_change_ts: для анти-флапа (не чаще раза в STRATEGY_LOCK_COOLDOWN_SEC). # lock_reason: человекочитаемое объяснение последнего lock. grid_locked = False sma_center = 0.0 last_sma_update_ts = 0.0 last_lock_state_change_ts = 0.0 lock_reason = "" # ─── Kronos advisor integration ───────────────────────────────────────────────── # kronos_pause_until: epoch time. If > now, advisor asked to pause the grid. # kronos_last_applied: dict последнего применённого совета (для дашборда / лога). # kronos_last_log_ts: чтобы не флудить в лог каждый цикл. kronos_pause_until = 0.0 kronos_last_applied: dict = {} kronos_last_log_ts = 0.0 kronos_advice_history: list[dict] = [] # последние 50 советов для дашборда def effective_grid_params(base_center: float, base_step: float, live: bool) -> tuple[float, float, float, bool, bool]: """Обёртка над apply_kronos_advice — точка интеграции в trading_loop. Возвращает (eff_center, eff_step, paused_by_kronos). - В DEMO: применяется всегда (если KRONOS_ENABLED). - В LIVE: применяется, только если KRONOS_LIVE_ENABLED=True. - Если Kronos-пауза активна и не истекла — возвращаем (base_center, base_step, True). """ global kronos_last_log_ts # Активная пауза от Kronos? if kronos_pause_until and time.time() < kronos_pause_until: return base_center, base_center, base_step, True, False eff_buy_center, eff_sell_center, eff_step, pause_now, bias_changed = apply_kronos_advice(base_center, base_step, live) if pause_now: return base_center, base_center, base_step, True, False # Лог с троттлингом (раз в 60с) — только если что-то реально поменялось now = time.time() if now - kronos_last_log_ts > 60 and abs(eff_step - base_step) > 1e-6: adv = read_kronos_advice() or {} logger.info( f"[KRONOS] applied: step {base_step*100:.3f}%→{eff_step*100:.3f}%, " f"buy_anchor ${base_center:,.2f}→${eff_buy_center:,.2f}, " f"sell_anchor ${base_center:,.2f}→${eff_sell_center:,.2f} " f"(conf={adv.get('confidence', 0):.2f}, source={adv.get('source', 'n/a')})" ) kronos_last_log_ts = now return eff_buy_center, eff_sell_center, eff_step, False, bias_changed def read_kronos_advice() -> Optional[dict]: """Прочитать kronos_advice.json, если файл свежий и валидный. Возвращает dict или None. Файл считается «протухшим» через 2×KRONOS_PRED_LEN (т.е. дольше горизонта прогноза — данные устарели). """ if not KRONOS_ENABLED: return None path = Path(KRONOS_ADVICE_FILE) if not path.exists(): return None try: data = json.loads(path.read_text()) except Exception as e: logger.debug(f"[KRONOS] failed to read advice: {e}") return None # Проверка свежести try: gen = datetime.fromisoformat(data.get("generated_at", "").replace("Z", "+00:00")) age_sec = (datetime.now(timezone.utc) - gen).total_seconds() horizon_sec = KRONOS_PRED_LEN * KRONOS_TF_MIN * 60 if age_sec > 2 * horizon_sec: logger.debug(f"[KRONOS] advice stale ({age_sec:.0f}s > 2*horizon)") return None except Exception: pass # Проверка confidence if data.get("confidence", 0) < KRONOS_MIN_CONFIDENCE: return None return data def apply_kronos_advice(base_center: float, base_step: float, live: bool, bias_following_conf_threshold: float = 0.7, bias_following_sell_multiplier: float = 2.0) -> tuple[float, float, float, bool, bool]: """Применить Kronos-совет к (центр сетки, шаг). Возвращает (effective_center, effective_step, pause_now). Логика: - step_percent: советник может снизить шаг (при низкой волатильности) или поднять. Минимум: 0.7× base_step (защита от слишком мелких сделок). Максимум: 1.5× base_step. - center: сдвигаем в сторону bias, максимум на 0.2% (если base_step = 0.5%). - pause: если Kronos сказал пауза и уверен — возращаем pause_now=True. - В DEMO всегда применяем. В LIVE — только если KRONOS_LIVE_ENABLED=True. """ global kronos_pause_until, kronos_last_applied, kronos_advice_history advice = read_kronos_advice() if advice is None: return base_center, base_center, base_step, False, False # В LIVE без явного флага — только логируем, не применяем if live and not KRONOS_LIVE_ENABLED: return base_center, base_center, base_step, False, False # Если Kronos снял паузу (pause_grid=false или conf<0.6) — # сбрасываем залипший kronos_pause_until, иначе бот сидит «на паузе» # ещё до 15 мин с момента последней паузы (см. issue 2026-06-07). global kronos_pause_until if not advice.get("pause_grid") or advice.get("confidence", 0) < 0.60: kronos_pause_until = 0.0 # pause (capped at 15 minutes by user request 2026-06-06) if advice.get("pause_grid") and advice.get("confidence", 0) >= 0.60: horizon_sec = KRONOS_PRED_LEN * KRONOS_TF_MIN * 60 # Cap the pause at 15 minutes so KRONOS can't lock the bot for a full day max_pause_sec = 5 * 60 # KRONOS-пауза cap 5 мин (был 15, 2026-06-08: conf=0.95 даёт паузу каждые 15 мин → бот не ставит ордера) if horizon_sec > max_pause_sec: horizon_sec = max_pause_sec was_paused = bool(kronos_pause_until and time.time() < kronos_pause_until) kronos_pause_until = time.time() + horizon_sec # TG только при новой паузе if not was_paused: tg_notify( f"⏸️ KRONOS: пауза сетки на {horizon_sec/60:.0f} мин\n" f"Bias: {advice.get('bias', '?')}\n" f"Ожид. диапазон: {advice.get('expected_range_pct', 0)*100:.2f}%\n" f"Conf: {advice.get('confidence', 0):.2f}" ) # Записываем в last_applied, чтобы в дашборде было видно kronos_last_applied = { "ts": time.time(), "advice": advice, "base_center": base_center, "eff_center": base_center, # на паузе центр не двигаем "base_step": base_step, "eff_step": base_step, # на паузе шаг не двигаем "pause_until": kronos_pause_until, } return base_center, base_center, base_step, True, False # step: защитные пределы ("adjusted" режим по результатам backtest 3-config, # 14-мес BTCUSDT 1h: +1450% PnL, MaxDD -0.69% (лучший из 4-х конфигов), # см. deploy/BACKTEST-RESULTS.md) kronos_step = float(advice.get("step_percent", base_step)) eff_step = max(base_step * 0.85, min(base_step * 1.3, kronos_step)) # center offset (split for BUY and SELL anchors) # bias=up -> BUY anchor = current price (aggressive), SELL anchor = forecast price (above) # bias=down -> BUY anchor = forecast price (below), SELL anchor = current price # bias=neutral -> both = current price (legacy symmetric) center_offset_pct = float(advice.get("center_offset_pct", 0.0)) eff_offset = max(-0.0015, min(0.0015, center_offset_pct)) forecast_center = base_center * (1.0 + eff_offset) bias = advice.get("bias") kronos_conf = float(advice.get("confidence", 0)) # ─── BIAS-FOLLOWING v2 (asymmetric, patch 14.06 11:33) ──────────────── # При conf >= bias_following_conf_threshold (default 0.7) сетка # асимметрично смещается в направлении bias: # bias=up: BUY-anchor остаётся на споте (ловит откат), # SELL-anchor уходит выше спота на offset × multiplier # bias=down: SELL-anchor остаётся на споте, # BUY-anchor уходит ниже спота на offset × multiplier # При conf < threshold — старая логика (один якорь сдвигается). if kronos_conf >= bias_following_conf_threshold and bias in ("up", "down"): if bias == "up": eff_buy_center = base_center # на споте eff_sell_center = base_center * (1.0 + eff_offset * bias_following_sell_multiplier) else: # down eff_buy_center = base_center * (1.0 - eff_offset * bias_following_sell_multiplier) eff_sell_center = base_center # на споте else: # Legacy: один якорь сдвигается if bias == "up": eff_buy_center = base_center eff_sell_center = forecast_center elif bias == "down": eff_buy_center = forecast_center eff_sell_center = base_center else: eff_buy_center = base_center eff_sell_center = base_center prev = kronos_last_applied kronos_last_applied = { "ts": time.time(), "advice": advice, "base_center": base_center, "eff_buy_center": eff_buy_center, "eff_sell_center": eff_sell_center, "eff_center": forecast_center, "base_step": base_step, "eff_step": eff_step, "pause_until": kronos_pause_until, } kronos_advice_history.append({ "ts": time.time(), "eff_step": eff_step, "eff_buy_center": eff_buy_center, "eff_sell_center": eff_sell_center, "eff_center": forecast_center, "bias": bias, "conf": advice.get("confidence", 0), "source": advice.get("source"), "mode": "live" if live else "demo", }) if len(kronos_advice_history) > 50: kronos_advice_history = kronos_advice_history[-50:] # bias_changed: True если bias отличается от прошлого (или если первый раз — пусть trading_loop знает). # Нужно для фикса «Kronos меняет прогноз, а BUY-уровни не пересчитываются». bias_changed = bool(advice.get("bias") != (prev or {}).get("advice", {}).get("bias")) # TG-уведомление: только при ЗНАЧИМОМ изменении (или смене bias) if prev: step_change = abs(eff_step - prev.get("eff_step", eff_step)) / max(eff_step, 1e-9) if step_change > 0.10 or bias_changed: tg_notify( f"🧠 KRONOS: обновлён совет\n" f"Шаг: {prev.get('eff_step', base_step)*100:.3f}% → {eff_step*100:.3f}%\n" f"Buy-anchor: ${prev.get('eff_buy_center', base_center):,.0f} → ${eff_buy_center:,.0f}\n" f"Sell-anchor: ${prev.get('eff_sell_center', base_center):,.0f} → ${eff_sell_center:,.0f}\n" f"Bias: {advice.get('bias', '?')} (было {(prev.get('advice') or {}).get('bias', '?')})\n" f"Conf: {advice.get('confidence', 0):.2f}\n" f"Источник: {advice.get('source', '?')}" ) return eff_buy_center, eff_sell_center, eff_step, False, bias_changed # ─── SMA-ATR strategy (2026-06-21) ────────────────────────────────────────────── def _compute_sma_center() -> Optional[float]: """Compute and cache SMA(STRATEGY_SMA_PERIOD) on STRATEGY_SMA_TF_MIN candles. Uses fetch_candles() (which has its own 2-min cache via getHloc). Returns the SMA value or None if data insufficient. Caches the result for STRATEGY_SMA_CACHE_SEC (default 300s = 5 min). """ global sma_center, last_sma_update_ts now = time.time() if sma_center > 0 and (now - last_sma_update_ts) < STRATEGY_SMA_CACHE_SEC: return sma_center if not api_client: return sma_center if sma_center > 0 else None try: cached = fetch_candles(timeframe_min=STRATEGY_SMA_TF_MIN, force=False) candles = cached.get("candles", []) if isinstance(cached, dict) else [] except Exception as e: logger.warning(f"[SMA] fetch_candles error: {e}") return sma_center if sma_center > 0 else None if not candles or len(candles) < STRATEGY_SMA_PERIOD: logger.debug(f"[SMA] only {len(candles) if candles else 0} candles — insufficient") return sma_center if sma_center > 0 else None closes = [c.get("c") for c in candles if c.get("c") is not None] sma = compute_sma(closes, STRATEGY_SMA_PERIOD) if sma is not None and sma > 0: sma_center = sma last_sma_update_ts = now logger.info( f"[SMA] SMA({STRATEGY_SMA_PERIOD}, {STRATEGY_SMA_TF_MIN}m) = ${sma:,.2f} " f"({len(closes)} candles)" ) return sma_center if sma_center > 0 else None def _get_current_bias() -> str: """Return current bias: 'up' | 'down' | 'range'. Source order: STRATEGY_BIAS_SOURCE (default Kronos), fallback STRATEGY_BIAS_FALLBACK. """ primary = (STRATEGY_BIAS_SOURCE or "kronos").lower() fallback = (STRATEGY_BIAS_FALLBACK or "trend").lower() if primary == "kronos": advice = read_kronos_advice() if advice: b = (advice.get("bias") or "range").lower() # Kronos uses 'flat' as synonym for 'range' if b == "flat": b = "range" return b if b in ("up", "down", "range") else "range" # Kronos unavailable → fall through elif primary == "trend": if TREND_ENABLED: try: lbias, _lconf, _lslope = detect_local_trend(TREND_LOOKBACK) if lbias in ("up", "down"): return lbias except Exception: pass if fallback == "trend" and TREND_ENABLED: try: lbias, _lconf, _lslope = detect_local_trend(TREND_LOOKBACK) if lbias in ("up", "down"): return lbias except Exception: pass elif fallback == "kronos": advice = read_kronos_advice() if advice: b = (advice.get("bias") or "range").lower() return b if b in ("up", "down", "range") else "range" return "range" def _parse_asymmetry(spec: str) -> tuple[int, int]: """Parse '3:2' → (3, 2). Falls back to (3, 2) on parse error.""" try: a, b = spec.split(":", 1) return max(0, int(a)), max(0, int(b)) except Exception: return 3, 2 def _get_asymmetry(bias: str) -> tuple[int, int]: """Return (buy_count, sell_count) for given bias.""" if bias == "up": return _parse_asymmetry(STRATEGY_ASYMMETRY_UP) if bias == "down": return _parse_asymmetry(STRATEGY_ASYMMETRY_DOWN) return _parse_asymmetry(STRATEGY_ASYMMETRY_RANGE) def _resolve_grid_center(current_price: float) -> float: """SMA-ATR strategy: choose grid center based on SMA + lock FSM. Returns: - sma_center if not locked (place orders around SMA) - current_price if locked (no new placement; existing orders were cancelled) - current_price if STRATEGY_CENTER_MODE != "sma" (legacy behaviour) Side effects: - May flip grid_locked on/off and trigger TG alerts (anti-flap cooldown 15 min). - Cancels open BUY orders on transition to locked. """ global grid_locked, lock_reason, last_lock_state_change_ts, demo_orders_placed if STRATEGY_CENTER_MODE != "sma": return current_price sma = _compute_sma_center() if sma is None or sma <= 0: # Cold cache / API down → legacy fallback (current_price), but DO NOT lock. return current_price dev_pct = abs((current_price - sma) / sma) * 100 now = time.time() cooldown_ok = (now - last_lock_state_change_ts) > STRATEGY_LOCK_COOLDOWN_SEC if not grid_locked and dev_pct > STRATEGY_LOCK_PCT: if cooldown_ok: grid_locked = True lock_reason = f"|dev|={dev_pct:.2f}% > {STRATEGY_LOCK_PCT}%" last_lock_state_change_ts = now cancelled = 0 if demo is not None: try: cancelled = demo.cancel_open_buys(reason=f"lock: {lock_reason}") except Exception as e: logger.warning(f"[LOCK] cancel_open_buys failed: {e}") demo_orders_placed = False logger.warning( f"[LOCK] dev={dev_pct:.2f}% (price=${current_price:,.2f}, sma=${sma:,.2f}) " f"> {STRATEGY_LOCK_PCT}% → grid flat, cancelled {cancelled} BUY order(s)" ) try: tg_notify( f"🔒 Grid LOCKED\n" f"Цена ${current_price:,.2f} ушла на {dev_pct:.2f}% от SMA({STRATEGY_SMA_PERIOD}, {STRATEGY_SMA_TF_MIN}m) ${sma:,.2f}\n" f"Отменено BUY: {cancelled}\n" f"Жду возврата в ±{STRATEGY_UNLOCK_PCT}%" ) except Exception as e: logger.warning(f"[LOCK] tg_notify failed: {e}") elif grid_locked and dev_pct < STRATEGY_UNLOCK_PCT: if cooldown_ok: grid_locked = False lock_reason = "" last_lock_state_change_ts = now demo_orders_placed = False logger.info( f"[UNLOCK] dev={dev_pct:.2f}% < {STRATEGY_UNLOCK_PCT}% → re-anchor at SMA ${sma:,.2f}" ) try: tg_notify( f"🔓 Grid UNLOCKED\n" f"Цена ${current_price:,.2f} вернулась в {dev_pct:.2f}% от SMA ${sma:,.2f}\n" f"Сетка восстановлена" ) except Exception as e: logger.warning(f"[UNLOCK] tg_notify failed: {e}") return sma if not grid_locked else current_price # Wallet (live-mode) balance cache. # accountGetSummary is called at most once per WALLET_CACHE_TTL seconds. # On error / 403 we keep the last known good response and surface a "stale" flag. WALLET_CACHE_TTL = 30 # seconds _wallet_lock = Lock() _wallet_cache: dict = {"balances": [], "primary_currency": "USDT", "primary_balance": 0.0, "ts": 0.0, "error": None, "stale": True} # ─── Account info cache (from getOPQ) ───────────────────────────────────────── _opq_lock = Lock() _opq_cache: dict = { "rev": None, "brief_nm": None, "main_curr": "USDT", "active": None, "init_margin": None, "reception": None, "f_kval": None, "ts": 0.0, "stale": True, "error": None, } OPQ_CACHE_TTL = 300.0 # 5 minutes # ─── Quote info cache (from getSecurityInfo) ───────────────────────────────── _quote_info_lock = Lock() _quote_info_cache: dict = { "ticker": SYMBOL, "short_name": "", "currency": "USDT", "lot": None, "min_step": None, "mkt_name": "", "mkt_tz": "", "ts": 0.0, "stale": True, "error": None, } QUOTE_INFO_CACHE_TTL = 3600.0 # 1 hour # ─── Candles cache (from getHloc) ───────────────────────────────────────────── _candles_lock = Lock() _candles_cache: dict = { "timeframe": 60, "candles": [], # [{"t": unix_ts, "o": open, "h": high, "l": low, "c": close, "v": vol}, ...] "ts": 0.0, "stale": True, "error": None, } CANDLES_CACHE_TTL = 120.0 # 2 minutes # Price history for the dashboard chart. # Sampled every PRICE_SAMPLE_INTERVAL seconds (not every loop iteration) to keep size sane. # Each entry: {t: epoch_seconds, p: float, type: "tick"|"fill"|"rebuild", side?: "BUY"|"SELL"} PRICE_SAMPLE_INTERVAL = 30 # seconds between tick samples PRICE_HISTORY_MAX = 5000 # ~41 hours at 30s; covers 24h range comfortably price_history: deque = deque(maxlen=PRICE_HISTORY_MAX) price_history_lock = Lock() _last_price_sample_ts = 0.0 logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s", handlers=[ logging.StreamHandler(), ], ) logger = logging.getLogger(__name__) # ─── Helpers ────────────────────────────────────────────────────────────────── def _tg_resolve_v6(host: str) -> str: """Resolve host to an IPv6 literal. Primary path — хостер фильтрует IPv4-исходящие. Falls back to hostname (default resolver) only if no AAAA record exists.""" try: infos = socket.getaddrinfo(host, 443, family=socket.AF_INET6, type=socket.SOCK_STREAM) return infos[0][4][0] except Exception: return host def tg_notify(text: str): """Send Telegram notification. Appends SERVER_URL footer so the source is always visible. Хостер фильтрует исходящий 443/tcp по IPv4. Используем http.client.HTTPSConnection с явным AF_INET6 сокетом → обходим IPv4-фильтр. Host header выставляем в api.telegram.org для корректного SSL SNI. Retries 3 times with backoff 0.1s, 0.3s. """ if not TELEGRAM_BOT_TOKEN or TELEGRAM_BOT_TOKEN == "": return try: import http.client as _http if SERVER_URL: text = f"{text}\n\n🔗 {SERVER_URL}" host = "api.telegram.org" v6 = _tg_resolve_v6(host) if not v6: logger.error(f"Telegram notify failed: no IPv6 for {host}") return body = urllib.parse.urlencode({"chat_id": TELEGRAM_CHAT_ID, "text": text}).encode() last_err = None for attempt in range(3): try: # AF_INET6 socket к IPv6-литералу v6 → SSL на host (для SNI) sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) sock.settimeout(4) sock.connect((v6, 443, 0, 0)) ctx = ssl.create_default_context() sock = ctx.wrap_socket(sock, server_hostname=host) conn = _http.HTTPSConnection(host, timeout=4) conn.sock = sock conn.request("POST", f"/bot{TELEGRAM_BOT_TOKEN}/sendMessage", body=body, headers={"Host": host, "Content-Type": "application/x-www-form-urlencoded"}) resp = conn.getresponse() if resp.status == 200: conn.close() return last_err = f"HTTP {resp.status}: {resp.read()[:200]!r}" conn.close() except Exception as e: last_err = e if attempt < 2: time.sleep(0.1 + 0.2 * attempt) logger.error(f"Telegram notify failed after retries: {last_err}") except Exception as e: logger.error(f"Telegram notify failed: {e}") def _state_path() -> Path: """Primary state file path (grid-state.json).""" STATE_DIR.mkdir(parents=True, exist_ok=True) return STATE_DIR / "grid-state.json" def load_state() -> dict: """Load state.json with config_version auto-merge (2026-06-30). При несовпадении config_version в state.json и GRID_STATE_VERSION в .env — подтягивает свежие grid-параметры (step_percent, take_profit_percent, grid_levels) из .env. Runtime-state (position, balance, trade_log, demo_mode, symbol) сохраняется. Перед merge: TG-уведомление с деталями. """ path = _state_path() state = None if path.exists(): try: state = json.loads(path.read_text()) except Exception: state = None if state is None: # Fresh state file — use .env defaults, stamp current version state = { "grid_levels": GRID_LEVELS, "step_percent": GRID_STEP_PERCENT, "take_profit_percent": GRID_TAKE_PROFIT_PERCENT, "demo_mode": DEMO_MODE, "active_symbol": SYMBOL, "config_version": GRID_STATE_VERSION, } return state # Version-based auto-merge state_version = state.get("config_version", "0.0") env_version = GRID_STATE_VERSION if state_version != env_version: # Сохранить old config для diagnostics old_config = {k: state.get(k) for k in GRID_CONFIG_KEYS if k in state} # Подтянуть свежие grid-параметры из .env new_config = { "step_percent": GRID_STEP_PERCENT, "take_profit_percent": GRID_TAKE_PROFIT_PERCENT, "grid_levels": GRID_LEVELS, } for k, v in new_config.items(): state[k] = v state["config_version"] = env_version state["config_version_previous"] = state_version # Runtime state не трогаем try: save_state(state) logger.warning( f"[STATE] config_version auto-merge: v{state_version} -> v{env_version}. " f"grid_config: {old_config} -> {new_config}. " f"Runtime state (position, balance, symbol, demo_mode) preserved." ) try: tg_notify( f"🔄 State config auto-merged: v{state_version} → v{env_version}\n" f" step: {(old_config.get('step_percent') or 0)*100:.3f}% → {new_config['step_percent']*100:.3f}%\n" f" TP: {old_config.get('take_profit_percent') or 0:.2f}% → {new_config['take_profit_percent']:.2f}%\n" f" levels: {old_config.get('grid_levels') or 0} → {new_config['grid_levels']}\n" f" Runtime state preserved." ) except Exception: pass except Exception as e: logger.error(f"[STATE] Save after merge failed: {e}") return state def save_state(state: dict): path = _state_path() path.write_text(json.dumps(state, indent=2, ensure_ascii=False)) def read_current_settings() -> dict: s = load_state() return { "grid_levels": s.get("grid_levels", GRID_LEVELS), "step_percent": s.get("step_percent", GRID_STEP_PERCENT), "take_profit_percent": s.get("take_profit_percent", GRID_TAKE_PROFIT_PERCENT), "demo_mode": s.get("demo_mode", DEMO_MODE), "active_symbol": SYMBOL, } def record_price_point(price: float, ptype: str = "tick", side: Optional[str] = None, bid: Optional[float] = None, ask: Optional[float] = None) -> None: """Append a point to price_history. Thread-safe (called from trading_loop thread + flask).""" if price <= 0: return pt = {"t": time.time(), "p": round(price, 4), "type": ptype} if side: pt["side"] = side if bid is not None and bid > 0: pt["bid"] = round(bid, 4) if ask is not None and ask > 0: pt["ask"] = round(ask, 4) with price_history_lock: price_history.append(pt) def detect_local_trend(lookback: int = 60) -> tuple: """Linear-regression slope over the last N price points. Returns (bias, confidence, slope_pct): bias — "up" | "down" | "neutral" confidence — 0..1 (R² of the regression) slope_pct — total % change over the lookback window """ try: with price_history_lock: pts = list(price_history)[-lookback:] except Exception: return ("neutral", 0.0, 0.0) n = len(pts) if n < 10: return ("neutral", 0.0, 0.0) # y = prices, x = 0..n-1 y0 = float(pts[0]["p"]) if y0 <= 0: return ("neutral", 0.0, 0.0) # slope (price per index) via least squares sx = (n - 1) * n / 2.0 sy = 0.0 sxy = 0.0 sxx = 0.0 for i, p in enumerate(pts): py = float(p["p"]) sy += py sxy += i * py sxx += i * i denom = n * sxx - sx * sx if denom == 0: return ("neutral", 0.0, 0.0) m = (n * sxy - sx * sy) / denom y_mean = sy / n # Total % change over the window slope_pct = (m * (n - 1)) / y_mean * 100.0 # R² y_pred_mean = m * (sx / n) ss_res = 0.0 ss_tot = 0.0 for i, p in enumerate(pts): py = float(p["p"]) y_pred = m * i + (y_mean - y_pred_mean) ss_res += (py - y_pred) ** 2 ss_tot += (py - y_mean) ** 2 r2 = max(0.0, 1.0 - ss_res / ss_tot) if ss_tot > 0 else 0.0 # Classify if slope_pct > TREND_THRESHOLD_PCT and r2 >= TREND_MIN_CONFIDENCE: return ("up", r2, slope_pct) if slope_pct < -TREND_THRESHOLD_PCT and r2 >= TREND_MIN_CONFIDENCE: return ("down", r2, slope_pct) return ("neutral", r2, slope_pct) def _parse_account_summary(data: dict) -> list: """Normalize Tradernet account/position response into a list of balances. Accepts several response shapes: - getPositionJson: {"result": {"ps": {"acc": [{"curr": "USD", "currval": ..., "s": ...}, ...]}}} - {"money": [{"curr": "USDT", "free": ..., "locked": ...}, ...]} - {"result": {"money": [...]}} - [{"curr": "USDT", ...}, ...] - {"USDT": {"free": ...}, "BTC": {...}, ...} (currency-keyed dict) Each output item: {currency, free, locked, total} """ if not isinstance(data, dict): # raw list if isinstance(data, list): data = {"money": data} else: return [] # getPositionJson shape: {"result": {"ps": {"acc": [...]}}} if "result" in data and isinstance(data["result"], dict): ps = data["result"].get("ps") if isinstance(ps, dict) and "acc" in ps and isinstance(ps["acc"], list): data = {"money": ps["acc"]} money = data.get("money") if money is None and "result" in data and isinstance(data["result"], dict): money = data["result"].get("money") if money is None: # maybe currency-keyed dict at top level sample = next(iter(data.values()), None) if isinstance(sample, dict) and any(k in sample for k in ("free", "locked", "total", "balance", "currval")): money = [{"curr": k, **(v if isinstance(v, dict) else {})} for k, v in data.items()] out: list = [] if isinstance(money, list): for item in money: if not isinstance(item, dict): continue cur = (item.get("curr") or item.get("currency") or item.get("code") or item.get("symbol") or item.get("asset")) if not cur: continue # Tradernet getPositionJson uses: currval (free) and s (blocked/settled) free = float(item.get("currval") or item.get("free") or item.get("available") or item.get("avail") or item.get("free_balance") or 0) locked = float(item.get("s") or item.get("locked") or item.get("reserved") or item.get("blocked") or item.get("locked_balance") or 0) total = float(item.get("total") or item.get("balance") or (free + abs(locked))) out.append({"currency": str(cur), "free": free, "locked": locked, "total": total}) return out def fetch_wallet_balance(force: bool = False) -> dict: """Get wallet balances from Tradernet via getPositionJson with 30s caching. Thread-safe. Returns the cache dict: {balances: [{currency, free, locked, total}, ...], primary_currency, primary_balance, ts, age_seconds, stale, error} """ global _wallet_cache now = time.time() with _wallet_lock: last_ts = _wallet_cache.get("ts", 0) is_fresh = (now - last_ts) < WALLET_CACHE_TTL if is_fresh and not force and not _wallet_cache.get("stale", True): return dict(_wallet_cache, stale=False, age_seconds=int(now - last_ts)) if not api_client: with _wallet_lock: _wallet_cache.update({"error": "API client not initialised", "stale": True, "ts": now}) return dict(_wallet_cache) # Use sync REST call — safe in Flask handlers, no aiohttp event-loop issues. try: raw = api_client.get_position_json_sync() except Exception as e: logger.warning(f"getPositionJson sync error: {e}") with _wallet_lock: existing = _wallet_cache.get("balances") or [] _wallet_cache.update({"error": str(e), "stale": True, "ts": now, "balances": existing, "primary_balance": _wallet_cache.get("primary_balance", 0.0)}) return dict(_wallet_cache) if not raw: # 403, timeout, etc. — _post returns {} on 403 err = "API ключ невалиден или запрос отклонён (403)" with _wallet_lock: existing = _wallet_cache.get("balances") or [] _wallet_cache.update({"error": err, "stale": True, "ts": now, "balances": existing, "primary_balance": _wallet_cache.get("primary_balance", 0.0)}) logger.warning(f"getPositionJson returned empty: {err}") return dict(_wallet_cache) balances = _parse_account_summary(raw) if not balances: logger.warning(f"getPositionJson unparseable: {str(raw)[:300]}") with _wallet_lock: existing = _wallet_cache.get("balances") or [] _wallet_cache.update({ "error": f"Неизвестный формат ответа: {str(raw)[:120]}", "stale": True, "ts": now, "balances": existing, "primary_balance": _wallet_cache.get("primary_balance", 0.0), }) return dict(_wallet_cache) # Pick primary currency: USDT if present, else USD (Tradernet main_curr is often USDT), # else QUOTE config, else first non-zero balance primary = None cur_set = {b["currency"].upper(): b for b in balances} # Try main_curr from getOPQ as a hint, but we don't fetch it separately for candidate in ["USDT", "USD", QUOTE.upper()]: if candidate in cur_set: primary = candidate break if not primary: # pick first with non-zero balance nonzero = [b for b in balances if b["free"] != 0 or b["locked"] != 0] primary = (nonzero or balances)[0]["currency"].upper() primary_balance = cur_set[primary]["total"] if primary in cur_set else 0.0 with _wallet_lock: _wallet_cache.update({ "balances": balances, "primary_currency": primary, "primary_balance": primary_balance, "ts": now, "error": None, "stale": False, }) logger.info(f"Wallet refreshed: {len(balances)} currencies, {primary}={primary_balance}") return dict(_wallet_cache, stale=False, age_seconds=0) def fetch_account_info(force: bool = False) -> dict: """Get account info (getOPQ) with 5-minute cache. Returns: {rev, brief_nm, main_curr, active, init_margin, reception, f_kval, ts, age_seconds, stale, error} """ global _opq_cache now = time.time() with _opq_lock: last_ts = _opq_cache.get("ts", 0) is_fresh = (now - last_ts) < OPQ_CACHE_TTL if is_fresh and not force and not _opq_cache.get("stale", True): return dict(_opq_cache, stale=False, age_seconds=int(now - last_ts)) if not api_client: with _opq_lock: _opq_cache.update({"error": "API client not initialised", "stale": True, "ts": now}) return dict(_opq_cache) try: raw = api_client.get_opq_sync() except Exception as e: logger.warning(f"getOPQ sync error: {e}") with _opq_lock: existing = {k: _opq_cache.get(k) for k in ("rev", "brief_nm", "main_curr", "active", "init_margin", "reception", "f_kval")} existing.update({"error": str(e), "stale": True, "ts": now}) _opq_cache.update(existing) return dict(_opq_cache) if not raw: with _opq_lock: existing = {k: _opq_cache.get(k) for k in ("rev", "brief_nm", "main_curr", "active", "init_margin", "reception", "f_kval")} existing.update({"error": "API ключ невалиден или запрос отклонён (403)", "stale": True, "ts": now}) _opq_cache.update(existing) return dict(_opq_cache) opq = raw.get("OPQ", {}) with _opq_lock: _opq_cache.update({ "rev": opq.get("rev"), "brief_nm": opq.get("brief_nm"), "main_curr": opq.get("main_curr", "USDT"), "active": opq.get("active"), "init_margin": opq.get("init_margin"), "reception": opq.get("reception"), "f_kval": opq.get("f_kval"), "ts": now, "error": None, "stale": False, }) logger.info(f"Account info refreshed: brief_nm={opq.get('brief_nm')}, main_curr={opq.get('main_curr')}") return dict(_opq_cache, stale=False, age_seconds=0) def fetch_quote_info(force: bool = False) -> dict: """Get security info (lot, min_step, currency, market) for SYMBOL. Returns: {ticker, short_name, currency, lot, min_step, mkt_name, mkt_tz, ts, age_seconds, stale, error} """ global _quote_info_cache now = time.time() with _quote_info_lock: last_ts = _quote_info_cache.get("ts", 0) is_fresh = (now - last_ts) < QUOTE_INFO_CACHE_TTL if is_fresh and not force and not _quote_info_cache.get("stale", True): return dict(_quote_info_cache, stale=False, age_seconds=int(now - last_ts)) if not api_client: with _quote_info_lock: _quote_info_cache.update({"error": "API client not initialised", "stale": True, "ts": now}) return dict(_quote_info_cache) try: info = api_client.get_security_info_sync(SYMBOL) except Exception as e: logger.warning(f"getSecurityInfo sync error: {e}") with _quote_info_lock: existing = {k: _quote_info_cache.get(k) for k in ("ticker", "short_name", "currency", "lot", "min_step", "mkt_name", "mkt_tz")} existing.update({"error": str(e), "stale": True, "ts": now}) _quote_info_cache.update(existing) return dict(_quote_info_cache) if not info: with _quote_info_lock: existing = {k: _quote_info_cache.get(k) for k in ("ticker", "short_name", "currency", "lot", "min_step", "mkt_name", "mkt_tz")} existing.update({"error": "API ключ невалиден или запрос отклонён (403)", "stale": True, "ts": now}) _quote_info_cache.update(existing) return dict(_quote_info_cache) mrkt = info.get("mrkt", {}) or {} with _quote_info_lock: _quote_info_cache.update({ "ticker": info.get("id", SYMBOL), "short_name": info.get("short_name", ""), "currency": info.get("currency", "USDT"), "lot": info.get("lot"), "min_step": info.get("min_step"), "mkt_name": info.get("mkt_name", ""), "mkt_tz": mrkt.get("tz", ""), "ts": now, "error": None, "stale": False, }) logger.info(f"Quote info refreshed: lot={info.get('lot')}, min_step={info.get('min_step')}") return dict(_quote_info_cache, stale=False, age_seconds=0) def fetch_candles(timeframe_min: int = 60, force: bool = False) -> dict: """Get OHLCV candlesticks for SYMBOL via getHloc. Args: timeframe_min: 1, 5, 15, 60, 1440 (daily) force: bypass cache Returns: {timeframe, candles: [{t, o, h, l, c, v}, ...], ts, age_seconds, stale, error} Candles cover last 7 days for intraday, last 30 days for daily. """ global _candles_cache now = time.time() with _candles_lock: last_tf = _candles_cache.get("timeframe", 0) last_ts = _candles_cache.get("ts", 0) is_fresh = (now - last_ts) < CANDLES_CACHE_TTL and last_tf == timeframe_min if is_fresh and not force and not _candles_cache.get("stale", True): return dict(_candles_cache, stale=False, age_seconds=int(now - last_ts)) if not api_client: with _candles_lock: _candles_cache.update({ "timeframe": timeframe_min, "candles": _candles_cache.get("candles", []), "error": "API client not initialised", "stale": True, "ts": now, }) return dict(_candles_cache) # Date range: last 7d for intraday (1/5/15/60 min), 60d for daily now_struct = time.gmtime(now) if timeframe_min >= 1440: days_back = 60 else: days_back = 7 from datetime import datetime, timedelta, timezone dt_to = datetime.fromtimestamp(now, tz=timezone.utc) dt_from = dt_to - timedelta(days=days_back) date_from = dt_from.strftime("%d.%m.%Y %H:%M") date_to = dt_to.strftime("%d.%m.%Y %H:%M") try: raw = api_client.get_hloc_sync(SYMBOL, timeframe_min, date_from, date_to, 0) except Exception as e: logger.warning(f"getHloc sync error: {e}") with _candles_lock: _candles_cache.update({ "timeframe": timeframe_min, "candles": _candles_cache.get("candles", []), "error": str(e), "stale": True, "ts": now, }) return dict(_candles_cache) if not raw: with _candles_lock: _candles_cache.update({ "timeframe": timeframe_min, "candles": _candles_cache.get("candles", []), "error": "API ключ невалиден или запрос отклонён (403)", "stale": True, "ts": now, }) return dict(_candles_cache) # Normalize: Tradernet returns {"hloc": {"TICKER": [[o,h,l,c], ...]}, "xSeries": {...}, "vl": {...}} hloc_list = raw.get("hloc", {}).get(SYMBOL, []) ts_list = raw.get("xSeries", {}).get(SYMBOL, []) vol_list = raw.get("vl", {}).get(SYMBOL, []) candles = [] for i, ohlc in enumerate(hloc_list): if not isinstance(ohlc, (list, tuple)) or len(ohlc) < 4: continue t = ts_list[i] if i < len(ts_list) else 0 v = vol_list[i] if i < len(vol_list) else 0 candles.append({ "t": int(t), "o": float(ohlc[0]), "h": float(ohlc[1]), "l": float(ohlc[2]), "c": float(ohlc[3]), "v": float(v) if v is not None else 0.0, }) with _candles_lock: _candles_cache.update({ "timeframe": timeframe_min, "candles": candles, "ts": now, "error": None, "stale": False, }) logger.info(f"Candles refreshed: tf={timeframe_min}min, {len(candles)} candles") return dict(_candles_cache, stale=False, age_seconds=0) def refresh_price() -> float: """Fetch latest price for SYMBOL synchronously. Uses urllib.request with a fresh HMAC signature per call (no aiohttp session reuse) to avoid aiohttp session hangs in the trading loop's event loop. The hot path adds ~1 RTT (3-5ms in same DC) per call, so the 2-second loop cadence is preserved. Returns the last seen price (current_price) on any failure — never raises. The caller treats a stale-but-nonzero price as a valid tick. """ global current_price, last_quote, _price_source # 1) Tradernet — основной источник (sync HMAC POST) try: import hmac as _hmac, hashlib as _hashlib ts = str(int(time.time())) params = {"tickers": SYMBOL} payload = json.dumps(params, separators=(",", ":")) msg = (payload + ts).encode() sig = _hmac.new( TRADERNET_PRIVATE_KEY.encode(), msg, _hashlib.sha256 ).hexdigest() url = f"{TRADERNET_BASE_URL}/api/getStockQuotesJson" req = urllib.request.Request(url, data=payload.encode(), method="POST") req.add_header("Content-Type", "application/json") req.add_header("X-NtApi-PublicKey", TRADERNET_PUBLIC_KEY) req.add_header("X-NtApi-Timestamp", ts) req.add_header("X-NtApi-Sig", sig) with urllib.request.urlopen(req, timeout=2) as r: data = json.loads(r.read().decode()) items = data.get("result", {}).get("q", []) or data.get("q", []) if items: q = items[0] ltp = q.get("ltp") or q.get("last") or 0 if ltp and ltp > 0: last_quote = { "ltp": float(q.get("ltp") or 0), "bap": float(q.get("bap") or 0), "bbp": float(q.get("bbp") or 0), "ltt": q.get("ltt", ""), "chg": float(q.get("chg") or 0), "chg110": float(q.get("chg110") or q.get("chg_pct") or 0), "vol": float(q.get("vol") or 0), "op": float(q.get("op") or 0), "pp": float(q.get("pp") or 0), } _price_source = "tradernet" return float(ltp) except Exception as e: logger.debug(f"Tradernet price error: {e}") # 2) DEMO + нет Tradernet — публичный Binance (без ключей) как запасной if DEMO_MODE: try: binance_sym = f"{BASE}{QUOTE}".upper() # BTC-USDT.IMEX -> BTCUSDT url = f"https://api.binance.com/api/v3/ticker/price?symbol={binance_sym}" req = urllib.request.Request(url, headers={"User-Agent": "grid-bot/demo"}) with urllib.request.urlopen(req, timeout=4) as r: payload = json.loads(r.read().decode()) px = float(payload.get("price") or 0) if px > 0: _price_source = "binance" return px except Exception as e: logger.debug(f"Binance price fallback error: {e}") return current_price # ─── Trading Loop ───────────────────────────────────────────────────────────── # ─── Live-trading helpers ────────────────────────────────────────────────── # All these are SYNC (urllib-based via api_client.*_sync) because the aiohttp # session in api_client is bound to its own event loop, and our trading loop # runs in a different loop. urllib is safe in any context. def _live_cancel_all_pending(): """Cancel every order we believe is still active on the exchange.""" global live_pending_orders if not api_client: return cancelled = 0 failed = 0 for o in list(live_pending_orders): oid = o.get("order_id") if not oid: continue if LIVE_TRADING_DRY_RUN: logger.info(f"[DRY-RUN] would cancel order {oid} ({o['side']} @ {o['price']:.2f})") cancelled += 1 continue try: resp = api_client.cancel_order_sync(oid) if resp and "error" in resp: logger.warning(f"[LIVE] cancel order {oid} rejected: {resp['error']}") # Probably already filled/cancelled — treat as terminal failed += 1 else: cancelled += 1 except Exception as e: logger.warning(f"[LIVE] cancel order {oid} error: {e}") failed += 1 if cancelled or failed: logger.info(f"[LIVE] cancel_all: cancelled={cancelled}, errors={failed}") live_pending_orders = [] def _live_sync_position(): """Pull current position from getPositionJson and reconcile with local state. Side effects: - If a position appears and we don't track one, mark the oldest pending BUY order as filled and create live_position. - If position disappears and we had one, mark the SELL order as filled and append a trade to live_trade_log. """ global live_position, live_pending_orders, live_trade_log, external_position, _last_external_log_ts if not api_client: return None try: data = api_client.get_position_json_sync(timeout=10) except Exception as e: logger.debug(f"[LIVE] sync_position error: {e}") return None if not data: return None ps = data.get("result", {}).get("ps", {}) pos_list = ps.get("pos", []) or [] btc_pos = None for _p in pos_list: if not isinstance(_p, dict): continue # Tradernet API: ключ ticker - "i" (instr name), qty="q", entry_avg="price_a" # Match by ticker (exact) OR short name+base_currency as fallback. _ticker = _p.get("i") or _p.get("instr_id") or _p.get("symbol") _name = _p.get("name") or "" if _ticker == SYMBOL or SYMBOL.split(".")[0] in _name: btc_pos = _p break if btc_pos: # γ-patch fix: real API keys are "q" (qty) and "price_a" (avg entry), not "vol"/"avg_price" vol = float(btc_pos.get("q") or btc_pos.get("vol") or btc_pos.get("open_bal") or 0) avg = float( btc_pos.get("price_a") or btc_pos.get("bal_price_a") or btc_pos.get("avg_price") or btc_pos.get("open_price") or 0 ) if vol > 0: if live_position is None: filled_buy = None for o in live_pending_orders: if o["side"] == "BUY" and o["status"] == "pending": filled_buy = o break if filled_buy: # ANTI-PHANTOM FIX (2026-06-30 16:00 GMT+7): use MIN single BUY qty # (not sum). Single matched BUY = 0.00001 BTC, external = 0.001 BTC. # After one of our 10 BUYs fills, sum drops to 0.00009 but single # is still 0.00001 — checking single keeps external detection working. _our_min_qty = min((o.get("qty", 0) for o in live_pending_orders if o["side"] == "BUY" and o["status"] == "pending"), default=0) # 2026-07-28 08:21 GMT+7 ANTI-PHANTOM relax: # Threshold is now on EXCESS over SUM of own pending, # not ratio vs single order. Coexisting manual position # of ~0.001 BTC no longer blocks own fills; only true # external excess > 0.005 BTC triggers external path. _our_sum_qty = sum( (o.get("qty", 0) for o in live_pending_orders if o["side"] == "BUY" and o["status"] == "pending"), default=0) _is_external = _our_sum_qty > 0 and (vol - _our_sum_qty) > 0.005 if _is_external: # Re-route to external_position path, do NOT mark buy as filled. # External position is stable (manual trade), so rate-limit spam. external_position = { "side": "BUY", "entry_price": avg, "qty": vol, "order_id": None, "avg_price_exchange": avg, "acc_pos_id": btc_pos.get("acc_pos_id"), "raw": btc_pos, } import time as _t _now = _t.time() if _now - _last_external_log_ts > 60.0: _last_external_log_ts = _now logger.info( f"[LIVE] ANTI-PHANTOM: exchange vol={vol} >> our pending " f"{_our_min_qty} → external (qty={vol} avg={avg}), skipping BUY fill" ) else: logger.debug( f"[LIVE] ANTI-PHANTOM (suppressed): vol={vol} avg={avg}" ) # External is NOT a fill — skip record_price_point / tg_notify. # Use global _last_external_log_ts (module-level, persists). else: filled_buy["status"] = "filled" filled_at = filled_buy["price"] live_position = { "side": "BUY", "entry_price": filled_at, "qty": filled_buy["qty"], "order_id": filled_buy["order_id"], "avg_price_exchange": avg, "raw": btc_pos, } live_pending_orders = [o for o in live_pending_orders if o["order_id"] != filled_buy["order_id"]] logger.info(f"[LIVE] BUY filled @ {filled_at:.2f} qty={filled_buy['qty']} " f"(exchange avg={avg:.2f})") record_price_point(filled_at, "fill", "BUY") tg_notify( f"🟢 LIVE BUY заполнен\n" f"Цена: ${filled_at:,.2f}\n" f"Количество: {filled_buy['qty']} BTC\n" f"Средняя биржи: ${avg:,.2f}\n" f"Текущая: ${current_price:,.2f}" ) else: # γ-v2: external position -> external_position, NOT live_position. # This way the bot's grid BUY orders are NOT blocked by leftover position. external_position = { "side": "BUY", "entry_price": avg, "qty": vol, "order_id": None, "avg_price_exchange": avg, "acc_pos_id": btc_pos.get("acc_pos_id"), "raw": btc_pos, } logger.info( f"[LIVE] external position detected: qty={vol} avg={avg} " f"(tracked separately, grid BUY orders continue on grid levels)" ) try: tg_notify( f"♻️ LIVE: external position detected\n" f"Количество: {vol} BTC\n" f"Средняя биржи: ${avg:,.2f}\n" f"Бот НЕ трогает её. Новые BUY будут на уровнях сетки." ) except Exception: pass else: live_position["avg_price_exchange"] = avg live_position["raw"] = btc_pos else: if external_position is not None and live_position is None: _ext = external_position logger.info( f"[LIVE] external position closed externally: " f"qty={_ext.get('qty')} avg={_ext.get('entry_price')} (no synthetic trade)" ) try: tg_notify( f"♻️ LIVE: external position disappeared\n" f"qty={_ext.get('qty')} avg={_ext.get('entry_price')}\n" f"Closed manually or by other system. Bot does not auto-close." ) except Exception: pass external_position = None if live_position is not None: entry = live_position.get("entry_price", 0) qty = live_position.get("qty", 0) entry_oid = live_position.get("order_id") exit_oid = None exit_price = 0 for o in list(live_pending_orders): if o["side"] == "SELL" and o["status"] == "pending": exit_oid = o["order_id"] exit_price = o["price"] o["status"] = "filled" if exit_price == 0 and live_pending_orders: exit_price = current_price pnl = (exit_price - entry) * qty if entry and exit_price else 0 pnl_pct = (pnl / (entry * qty) * 100) if entry and qty else 0 trade = { "entry_order_id": entry_oid, "exit_order_id": exit_oid, "entry_price": entry, "exit_price": exit_price, "qty": qty, "pnl_usdc": round(pnl, 8), "pnl_pct": round(pnl_pct, 4), "ts": time.time(), } live_trade_log.append(trade) logger.info(f"[LIVE] SELL filled @ {exit_price:.2f} entry={entry:.2f} " f"qty={qty} pnl={pnl:+.6f} USDT ({pnl_pct:+.2f}%)") record_price_point(exit_price, "fill", "SELL") tg_notify( f"🔴 LIVE SELL заполнен\n" f"Entry: ${entry:,.2f}\n" f"Exit: ${exit_price:,.2f}\n" f"Количество: {qty} BTC\n" f"PNL: {pnl:+.6f} USDT ({pnl_pct:+.2f}%)" ) live_pending_orders = [o for o in live_pending_orders if o.get("status") != "filled"] live_position = None return btc_pos def _live_stale_check(): """If price dropped more than GRID_STALE_DROP_PERCENT from start_price with no fills yet — cancel all and rebuild from current price. """ global start_price, live_orders_placed, grid_stale, grid_stale_drop_pct, grid_stale_rebuilds, live_pending_orders if GRID_STALE_DROP_PERCENT <= 0 or start_price <= 0 or current_price <= 0: return if live_position is not None: return if not live_pending_orders: return drop_pct = (start_price - current_price) / start_price * 100 if drop_pct >= GRID_STALE_DROP_PERCENT: cancelled = len(live_pending_orders) _live_cancel_all_pending() start_price = current_price live_orders_placed = False grid_stale = True grid_stale_drop_pct = drop_pct grid_stale_rebuilds += 1 logger.warning( f"[LIVE STALE-GRID] Drop {drop_pct:.2f}% >= {GRID_STALE_DROP_PERCENT}% " f"with no fills. Cancelled {cancelled} orders, rebuilding at ${current_price:,.2f}" ) tg_notify( f"♻️ LIVE Stale-grid: цена упала на {drop_pct:.2f}% без сделок\n" f"Отменено ордеров: {cancelled}\n" f"Новая сетка от ${current_price:,.2f}" ) record_price_point(current_price, "rebuild") def _live_inactivity_check(): """LIVE-equivalent of DEMO inactivity-rebuild. If INACTIVITY_RESTART_HOURS have passed without any trade AND BTC has moved from start_price by >= INACTIVITY_PRICE_DRIFT_PCT, cancel the hanging BUY orders and rebuild the grid around the current price. Anti-flood: at most one rebuild per 6 hours. Called from the LIVE branch of trading_loop before _live_place_grid_buys. Before this patch, inactivity-rebuild lived ONLY in the DEMO branch, which is why the bot in LIVE could hang on an empty grid for weeks. """ global start_price, live_orders_placed, grid_stale, grid_stale_drop_pct global grid_stale_rebuilds, inactivity_rebuilds, last_inactivity_rebuild_ts if INACTIVITY_RESTART_HOURS <= 0 or start_price <= 0 or current_price <= 0: return if live_position is not None: return if not live_pending_orders: return _now_ts = time.time() _idle_h = (_now_ts - last_trade_ts) / 3600.0 _drift_pct = abs((current_price - start_price) / start_price * 100.0) _cooldown_ok = (_now_ts - last_inactivity_rebuild_ts > 6 * 3600) if ( _idle_h >= INACTIVITY_RESTART_HOURS and _drift_pct >= INACTIVITY_PRICE_DRIFT_PCT and _cooldown_ok ): cancelled = len(live_pending_orders) _live_cancel_all_pending() old_start = start_price start_price = current_price live_orders_placed = False grid_stale = True grid_stale_drop_pct = _drift_pct grid_stale_rebuilds += 1 inactivity_rebuilds += 1 last_inactivity_rebuild_ts = _now_ts logger.warning( f"[LIVE INACTIVITY-REBUILD] {INACTIVITY_RESTART_HOURS}h threshold " f"(idle {_idle_h:.1f}h), drift {_drift_pct:.2f}% from start " f"${old_start:,.2f} -> ${current_price:,.2f}. " f"Cancelled {cancelled} orders, will rebuild at ${current_price:,.2f}" ) tg_notify( f"\u267c\ufe0f LIVE Inactivity-rebuild: {_idle_h:.1f}h without trades\n" f"BTC moved {_drift_pct:.2f}% from grid center\n" f"Was: ${old_start:,.2f} -> Now: ${current_price:,.2f}\n" f"Cancelled orders: {cancelled}" ) record_price_point(current_price, "rebuild") def _live_place_grid_buys(grid_center: float): """Place LIMIT BUY orders for all grid levels below current price. Live qty is 0.00001 (1 minimum lot for BTC-USDT.IMEX) per level regardless of grid_engine base_qty, because the live account has only 138 USDT available and 0.001 BTC per level = 70 USDT × 5 = 350 USDT (insufficient). """ global live_pending_orders, live_orders_placed if not api_client: return levels = grid_engine.get_grid_levels(grid_center) placed = 0 failed = 0 LIVE_QTY = 0.00001 # 1 minimum lot for lvl in levels: if lvl.side != "BUY": continue if any(o["side"] == "BUY" and o["level_id"] == lvl.level_id and o["status"] == "pending" for o in live_pending_orders): continue if LIVE_TRADING_DRY_RUN: fake_id = int(time.time() * 1000) + lvl.level_id live_pending_orders.append({ "order_id": fake_id, "side": "BUY", "price": lvl.price, "qty": LIVE_QTY, "level_id": lvl.level_id, "status": "pending", "created_at": time.time(), "raw": {"dry_run": True, "client_id": f"gridbot-buy-{lvl.level_id}"}, }) logger.info(f"[DRY-RUN] would place BUY {LIVE_QTY} BTC @ ${lvl.price:.2f} " f"(fake_id={fake_id})") placed += 1 continue try: resp = api_client.put_order_sync( ticker=SYMBOL, price=lvl.price, qty=LIVE_QTY, action="BUY", order_type="LIMIT", client_id=f"gridbot-buy-{lvl.level_id}" ) if not resp: failed += 1 continue if "error" in resp: logger.warning(f"[LIVE] BUY @ {lvl.price:.2f} rejected: {resp['error']}") failed += 1 continue oid = resp.get("order_id") if not oid: logger.warning(f"[LIVE] BUY @ {lvl.price:.2f} no order_id: {resp}") failed += 1 continue live_pending_orders.append({ "order_id": oid, "side": "BUY", "price": lvl.price, "qty": LIVE_QTY, "level_id": lvl.level_id, "status": "pending", "created_at": time.time(), "raw": resp.get("order", resp), }) placed += 1 except Exception as e: logger.warning(f"[LIVE] BUY @ {lvl.price:.2f} error: {e}") failed += 1 if placed or failed: logger.info(f"[LIVE] Placed {placed} BUY orders, {failed} failed (price ~{grid_center:.2f})") if placed > 0: live_orders_placed = True def _live_place_grid_sells(grid_center: float): """Place LIMIT SELL orders above current price when we have a position.""" global live_pending_orders if not api_client or live_position is None: return if live_position.get("side") != "BUY": return levels = grid_engine.get_grid_levels(grid_center) placed = 0 failed = 0 LIVE_QTY = 0.00001 # match the BUY qty for lvl in levels: if lvl.side != "SELL": continue if any(o["side"] == "SELL" and o["level_id"] == lvl.level_id and o["status"] == "pending" for o in live_pending_orders): continue if LIVE_TRADING_DRY_RUN: fake_id = int(time.time() * 1000) + lvl.level_id + 10000 live_pending_orders.append({ "order_id": fake_id, "side": "SELL", "price": lvl.price, "qty": LIVE_QTY, "level_id": lvl.level_id, "status": "pending", "created_at": time.time(), "raw": {"dry_run": True, "client_id": f"gridbot-sell-{lvl.level_id}"}, }) logger.info(f"[DRY-RUN] would place SELL {LIVE_QTY} BTC @ ${lvl.price:.2f} " f"(fake_id={fake_id})") placed += 1 continue try: resp = api_client.put_order_sync( ticker=SYMBOL, price=lvl.price, qty=LIVE_QTY, action="SELL", order_type="LIMIT", client_id=f"gridbot-sell-{lvl.level_id}" ) if not resp: failed += 1 continue if "error" in resp: logger.warning(f"[LIVE] SELL @ {lvl.price:.2f} rejected: {resp['error']}") failed += 1 continue oid = resp.get("order_id") if not oid: failed += 1 continue live_pending_orders.append({ "order_id": oid, "side": "SELL", "price": lvl.price, "qty": LIVE_QTY, "level_id": lvl.level_id, "status": "pending", "created_at": time.time(), "raw": resp.get("order", resp), }) placed += 1 except Exception as e: logger.warning(f"[LIVE] SELL @ {lvl.price:.2f} error: {e}") failed += 1 if placed or failed: logger.info(f"[LIVE] Placed {placed} SELL orders, {failed} failed (price ~{grid_center:.2f})") def _live_take_profit(): """If position PnL >= take_profit_percent, close it with a market order.""" if not api_client or live_position is None: return if live_position.get("side") != "BUY": return entry = live_position.get("entry_price", 0) qty = live_position.get("qty", 0) if entry <= 0 or qty <= 0 or current_price <= 0: return # Override to LIVE_QTY if needed (qty from pending BUY may be wrong if BUY was external) pnl_pct = (current_price - entry) / entry * 100 if pnl_pct >= grid_engine.take_profit_percent: for o in list(live_pending_orders): if o["side"] == "SELL" and o["status"] == "pending": if LIVE_TRADING_DRY_RUN: logger.info(f"[DRY-RUN] would cancel SELL order {o['order_id']} @ {o['price']:.2f}") else: try: api_client.cancel_order_sync(o["order_id"]) except Exception: pass o["status"] = "cancelled" live_pending_orders = [o for o in live_pending_orders if o.get("status") != "cancelled"] if LIVE_TRADING_DRY_RUN: logger.info(f"[DRY-RUN] would place MARKET SELL {qty} BTC (TP, pnl={pnl_pct:+.2f}%)") tg_notify( f"🎯 LIVE TAKE PROFIT (DRY-RUN)\n" f"Entry: ${entry:,.2f}\n" f"Текущая: ${current_price:,.2f}\n" f"PNL: {pnl_pct:+.2f}%\n" f"Бот бы закрыл позицию market-ордером" ) return try: resp = api_client.put_order_sync( ticker=SYMBOL, price=0, qty=qty, action="SELL", order_type="MARKET", client_id=f"gridbot-tp" ) if resp and "error" not in resp and resp.get("order_id"): logger.info(f"[LIVE TP] Market SELL placed: order_id={resp['order_id']}") tg_notify( f"🎯 LIVE TAKE PROFIT\n" f"Entry: ${entry:,.2f}\n" f"Текущая: ${current_price:,.2f}\n" f"PNL: {pnl_pct:+.2f}%\n" f"Закрытие market-ордером" ) else: logger.warning(f"[LIVE TP] market SELL failed: {resp}") except Exception as e: logger.error(f"[LIVE TP] exception: {e}") def validate_grid_fees( step_percent: float, fee_rate_percent: float = 0.2, safety_multiplier: float = 1.25, ) -> tuple[bool, str, dict]: """QuantDinger validator.py pattern: net-profit guard for grid cell. A grid cell must cover both entry and exit commissions. Compares the price distance (step) with estimated round-trip fees plus safety buffer. Returns (ok, message, details_dict). ok=False → шаг слишком узкий, прибыли не будет после fee. """ fee = fee_rate_percent / 100.0 # 0.2% -> 0.002 round_trip_fee = fee * 2 # entry + exit required_step = round_trip_fee * safety_multiplier if step_percent <= 0: return False, f"step_percent={step_percent} must be > 0", { "step_percent": step_percent, "required_step": required_step, } ok = step_percent >= required_step margin_pct = (step_percent - required_step) / required_step * 100 if required_step > 0 else 0 msg = ( f"step={step_percent*100:.3f}% vs required={required_step*100:.3f}% " f"(round-trip fee {round_trip_fee*100:.3f}% × safety {safety_multiplier}) " f"→ margin {margin_pct:+.1f}%" ) return ok, msg, { "step_percent": step_percent, "required_step": required_step, "round_trip_fee": round_trip_fee, "safety_multiplier": safety_multiplier, "margin_pct": margin_pct, } def _reconcile_demo_position() -> dict: """QuantDinger ledger_reconcile pattern: периодическая сверка demo.position с trade_log. Ловит shadow-позиции, открытые ДО последнего TP/SELL-fill (которые возникают, например, при re-fill на старом уровне сетки после TP). Возвращает dict с ключом 'action': - 'no_position' : demo.position is None - 'valid' : position согласован с trade_log - 'shadow_detected' : position открыт ДО последнего closing trade (если RECONCILE_AUTO_CLEAR=True, очищается) - 'parse_error' : не удалось распарсить exit_time """ if demo is None: return {"action": "no_demo"} if demo.position is None: return {"action": "no_position"} pos = demo.position pos_opened_iso = ( datetime.fromtimestamp(pos.opened_at, tz=timezone.utc).isoformat() if pos.opened_at else None ) # Найти самый последний closing trade last_close = None for t in (demo.trade_log or []): if t.get("side") in ("TAKE_PROFIT", "BUY→SELL"): if last_close is None or t.get("exit_time", "") > last_close.get("exit_time", ""): last_close = t if last_close and pos.opened_at: # Defensive: last_close may lack exit_time (e.g. reconstructed_from_botlog entries) last_close_exit_str = last_close.get("exit_time") if not last_close_exit_str: return { "action": "parse_error", "position_entry": pos.entry_price, "position_qty": pos.qty, "error": f"last_close missing exit_time (side={last_close.get('side')!r}, keys={list(last_close.keys())})", } try: last_close_ts = datetime.fromisoformat(last_close_exit_str).timestamp() except (ValueError, TypeError) as e: return { "action": "parse_error", "position_entry": pos.entry_price, "position_qty": pos.qty, "error": f"unparseable exit_time={last_close_exit_str!r}: {e}", } if pos.opened_at < last_close_ts: # Position opened BEFORE last close → shadow return { "action": "shadow_detected", "position_entry": pos.entry_price, "position_qty": pos.qty, "position_opened_at": pos_opened_iso, "position_age_seconds": time.time() - pos.opened_at, "shadow_of_trade": last_close, "last_close_exit_time": last_close.get("exit_time"), "last_close_entry_price": last_close.get("entry"), "last_close_exit_price": last_close.get("exit"), "auto_cleared": False, } return { "action": "valid", "position_entry": pos.entry_price, "position_qty": pos.qty, "position_opened_at": pos_opened_iso, } def _apply_reconcile_clear(reconcile_result: dict) -> None: """Очистить shadow-позицию и записать в trade_log. Вызывается только при RECONCILE_AUTO_CLEAR=True. """ global demo if reconcile_result.get("action") != "shadow_detected": return pos = demo.position shadow_of = reconcile_result.get("shadow_of_trade") or {} # Записать в trade_log "synthetic close" с source=reconcile exit_ts = time.time() exit_price = current_price pnl = (exit_price - pos.entry_price) * pos.qty pnl_pct = (pnl / (pos.entry_price * pos.qty) * 100) if pos.entry_price and pos.qty else 0 demo.trade_log.append({ "side": "RECONCILE_CLEAR", "entry": pos.entry_price, "exit": exit_price, "qty": pos.qty, "pnl_usdc": round(pnl, 8), "pnl_pct": round(pnl_pct, 4), "entry_time": datetime.fromtimestamp(pos.opened_at, tz=timezone.utc).isoformat() if pos.opened_at else None, "exit_time": datetime.fromtimestamp(exit_ts, tz=timezone.utc).isoformat(), "source": "reconcile", "shadow_of_trade_exit": shadow_of.get("exit_time"), "shadow_of_trade_entry": shadow_of.get("entry"), }) # Возврат резерва не нужен: balance уже корректен (не списывался при reconcile) demo.position = None # ВАЖНО: НЕ закрываем существующие SELL-ордера (они валидны для будущих fill) # — это отличается от _cancel_open_sells при normal TP/SELL. async def trading_loop(): global current_price, start_price, last_error, bot_running global grid_engine, demo, demo_orders_placed global grid_stale, grid_stale_drop_pct, grid_stale_rebuilds global last_inactivity_rebuild_ts, inactivity_rebuilds global last_trade_ts global live_pending_orders, live_position, live_trade_log, live_orders_placed global _diag_demo_logged, _price_source, _last_reconcile_ts settings = read_current_settings() grid_engine = GridEngine( levels=settings["grid_levels"], step_percent=settings["step_percent"], take_profit_percent=settings["take_profit_percent"], symbol=SYMBOL, mode="demo" if settings["demo_mode"] else "live", ) demo = DemoSimulator(start_balance=DEMO_START_BALANCE) # Restore trade history from file (survives restarts) trade_hist_path = Path('/root/grid-bot/trade_history.json') if trade_hist_path.exists(): try: hist = json.loads(trade_hist_path.read_text()) demo.trade_log = hist.get('trade_log', []) demo.balance = hist.get('balance', DEMO_START_BALANCE) logger.info(f"Restored {len(demo.trade_log)} trades from trade_history.json, balance={demo.balance:.2f}") except Exception as e: logger.warning(f"Could not restore trade_history: {e}") demo_orders_placed = False # Reset live state on every (re)start live_pending_orders = [] live_position = None live_trade_log = [] live_orders_placed = False # γ-patch (2026-06-30): safety flag - prevents placing orders before first # successful sync with exchange. Without this, API timeout at startup # would cause bot to place BUY orders next to existing external position. _live_position_synced = False # === Fee-coverage validator (QuantDinger validator pattern) === # Run once on startup, log result. Doesn't block: even if not OK, # grid still builds (DEMO mode is forgiving). _fee_ok, _fee_msg, _fee_details = validate_grid_fees( step_percent=settings["step_percent"], fee_rate_percent=FEE_RATE_PERCENT, safety_multiplier=FEE_SAFETY_MULTIPLIER, ) _fee_level = logging.WARNING if not _fee_ok else logging.INFO logger.log(_fee_level, f"[FEE-VALIDATOR] {_fee_msg}") if not _fee_ok: tg_notify( f"⚠️ Fee-validator: шаг сетки слишком узкий\n" f"{_fee_msg}\n" f"Сетка запущена, но прибыли после fee не будет." ) grid_stale = False _diag_demo_logged = False grid_stale_drop_pct = 0.0 with price_history_lock: price_history.clear() _last_price_sample_ts = 0.0 # Reset inactivity timer on every (re)start — бот только запустился # и «ждёт» первой сделки. Анти-флуд: 1 ребилд в 6 часов. last_trade_ts = time.time() last_inactivity_rebuild_ts = 0.0 _last_reconcile_ts = 0.0 # first reconcile runs immediately on first loop iter price = refresh_price() if price > 0: current_price = price start_price = price bid = last_quote.get("bbp") or None ask = last_quote.get("bap") or None record_price_point(price, "tick", bid=bid, ask=ask) _last_price_sample_ts = time.time() elif DEMO_MODE and current_price <= 0: # Синхронный фолбэк на Tradernet (прямой HMAC-вызов, чтобы не зависеть от async) try: import hmac as _hmac, hashlib as _hashlib ts = str(int(time.time() * 1000)) params = {"tickers": [SYMBOL]} body = json.dumps(params).encode() body_b64 = base64.b64encode(body).decode() msg = (TRADERNET_PRIVATE_KEY + TRADERNET_PUBLIC_KEY + ts + body_b64).encode() sig = _hmac.new(TRADERNET_PRIVATE_KEY.encode(), msg, _hashlib.sha256).hexdigest() url = f"{TRADERNET_BASE_URL}/api/v1/cmd" req = urllib.request.Request(url, data=body, method="POST") req.add_header("Content-Type", "application/json") req.add_header("X-NtApi-PublicKey", TRADERNET_PUBLIC_KEY) req.add_header("X-NtApi-Sig", sig) req.add_header("X-NtApi-Timestamp", ts) req.add_header("X-NtApi-Cmd", "getQuotes") with urllib.request.urlopen(req, timeout=2) as r: payload = json.loads(r.read().decode()) items = payload.get("result", {}).get("q", []) or payload.get("q", []) if items: q = items[0] ltp = float(q.get("ltp") or q.get("last") or 0) if ltp > 0: current_price = ltp start_price = ltp _price_source = "tradernet" logger.info(f"DEMO startup: Tradernet price ${ltp:,.2f}") record_price_point(ltp, "tradernet-startup") _last_price_sample_ts = time.time() except Exception as e: logger.warning(f"DEMO startup Tradernet fetch failed: {e}") if current_price <= 0: # Запасной вариант: Binance try: binance_sym = f"{BASE}{QUOTE}".upper() url = f"https://api.binance.com/api/v3/ticker/price?symbol={binance_sym}" req = urllib.request.Request(url, headers={"User-Agent": "grid-bot/demo-startup"}) with urllib.request.urlopen(req, timeout=2) as r: payload = json.loads(r.read().decode()) px = float(payload.get("price") or 0) if px > 0: current_price = px start_price = px _price_source = "binance" logger.info(f"DEMO startup: Binance fallback price ${px:,.2f}") record_price_point(px, "binance-startup") _last_price_sample_ts = time.time() except Exception as e: logger.warning(f"DEMO startup Binance fetch failed: {e}") if current_price <= 0: # Последний фолбэк — синтетика (только если ничего не доступно) current_price = 100000.0 start_price = 100000.0 _price_source = "synth" logger.warning(f"DEMO mode: no live price anywhere, using synth ${current_price:,.2f}") record_price_point(current_price, "synth-fallback") mode_str = "демо" if settings["demo_mode"] else "реальный" balance_str = "" if not settings["demo_mode"]: # Refresh balance once on startup so TG notification shows real USDT try: b = fetch_wallet_balance(force=True) balance_str = f"\nБаланс: {b.get('balance', 0):.2f} {b.get('currency', '')}" except Exception as e: logger.warning(f"startup balance fetch failed: {e}") tg_notify(f"🚀 Бот запущен\nРежим: {mode_str}\nШаг: {settings['step_percent']*100:.2f}%\nTP: {settings['take_profit_percent']:.2f}%{balance_str}" + ("\n⚠️ DRY-RUN: ордера не отправляются на биржу" if LIVE_TRADING_DRY_RUN else "\n✅ РЕАЛЬНАЯ ТОРГОВЛЯ")) import socket socket.setdefaulttimeout(2) # global socket timeout — prevent hangs in sync urllib calls while bot_running: try: import os as _os_b with open('/tmp/loop_beat', 'a') as _f: _f.write(f"{datetime.now().isoformat()} iter\n") # Watchdog: update wall-clock timestamp for stuck detection trading_loop._last_loop_iter_ts = time.time() # DEBUG: лог для отслеживания активности цикла (троттлинг 5 сек) if not hasattr(trading_loop, '_last_loop_log') or time.time() - trading_loop._last_loop_log > 5: trading_loop._last_loop_log = time.time() logger.debug(f"[loop] iter: current_price={current_price}, bot_running={bot_running}") price = await asyncio.to_thread(refresh_price) if price > 0: current_price = price # Throttled price tick for the chart now_ts = time.time() if now_ts - _last_price_sample_ts >= PRICE_SAMPLE_INTERVAL: bid = last_quote.get("bbp") or None ask = last_quote.get("bap") or None record_price_point(price, "tick", bid=bid, ask=ask) _last_price_sample_ts = now_ts elif DEMO_MODE and _price_source == "binance" and current_price > 0: # DEMO Binance уже загрузился, но один тик пришёл пустым — # держим последнюю цену (не дрейфуем от $100k). pass elif DEMO_MODE and current_price > 0 and _price_source == "synth": # Демо без живого источника: лёгкий random-walk (только если не было реальных данных) import random _drift = (random.random() - 0.5) * 0.006 # ±0.3% current_price = current_price * (1.0 + _drift) now_ts = time.time() if now_ts - _last_price_sample_ts >= PRICE_SAMPLE_INTERVAL: record_price_point(current_price, "synth") _last_price_sample_ts = now_ts # Микро-шум на живой DEMO-цене: имитация межинтревального тика if DEMO_MODE and price > 0 and _price_source == "binance": import random _noise = (random.random() - 0.5) * 0.0002 # ±0.01% (микро-шум) current_price = current_price * (1.0 + _noise) # ─── DEMO-ветка / LIVE-ветка (общая для обоих путей) ── state = load_state() demo_mode = state.get("demo_mode", DEMO_MODE) # First-iter debug: log the resolved value to confirm # which branch is taken. if not _diag_demo_logged: _diag_demo_logged = True logger.info(f"[DEBUG] state.demo_mode={state.get('demo_mode')!r} " f"resolved demo_mode={demo_mode!r} " f"branch={'DEMO' if demo_mode else 'LIVE'}") if demo_mode: # ─── SMA-ATR strategy: center + lock FSM ───────── grid_center = _resolve_grid_center(current_price) # ─── Kronos advisor (DEMO) ───────────────────── eff_buy_center, eff_sell_center, eff_step, kronos_paused, bias_changed = effective_grid_params( grid_center, settings["step_percent"], live=False ) if kronos_paused: # Kronos-пауза: пропускаем перестановку ордеров, логируем logger.debug("[KRONOS] DEMO paused, skipping grid rebuild") else: # ─── Local trend (slope-based) bias override ──────── # Если локальный slope уверенно показывает uptrend, # поднимаем eff_buy_center ВЫШЕ текущей цены, чтобы # BUY-уровни сидели ближе и ловили откат в растущем # тренде. На downtrend — наоборот, опускаем. if TREND_ENABLED: lbias, lconf, lslope = detect_local_trend(TREND_LOOKBACK) if lbias == "up" and lconf >= TREND_MIN_CONFIDENCE: shifted = current_price * (1 + TREND_UP_ANCHOR_SHIFT) if shifted > eff_buy_center: eff_buy_center = shifted logger.info( f"[TREND] uptrend slope={lslope:+.3f}% conf={lconf:.2f} " f"→ buy_anchor +{TREND_UP_ANCHOR_SHIFT*100:.2f}% to ${eff_buy_center:,.2f}" ) elif lbias == "down" and lconf >= TREND_MIN_CONFIDENCE: shifted = current_price * (1 - TREND_DOWN_ANCHOR_SHIFT) if shifted < eff_buy_center: eff_buy_center = shifted logger.info( f"[TREND] downtrend slope={lslope:+.3f}% conf={lconf:.2f} " f"→ buy_anchor -{TREND_DOWN_ANCHOR_SHIFT*100:.2f}% to ${eff_buy_center:,.2f}" ) grid_center = eff_buy_center # BUGFIX-C: при смене bias (up↔down↔neutral) пересобираем BUY-уровни. # Старая логика «place once per session» не учитывала, что Kronos-anchor # сдвигается → BUY-ордера висят по устаревшим ценам. if bias_changed: cancelled = demo.cancel_open_buys( reason=f"bias changed → new buy_anchor ${eff_buy_center:,.2f}" ) demo_orders_placed = False if cancelled > 0: logger.info( f"[KRONOS] bias changed → cancelled {cancelled} open BUY " f"order(s), will rebuild around new buy_anchor ${eff_buy_center:,.2f}" ) # Пересобираем grid_engine только если шаг реально изменился if abs(eff_step - settings["step_percent"]) > 1e-6: grid_engine = GridEngine( levels=settings["grid_levels"], step_percent=eff_step, take_profit_percent=settings["take_profit_percent"], symbol=SYMBOL, mode="demo", grid_mode=GRID_MODE, ) # ─── LIVE mode (real exchange orders) ────────────────────── # Runs in parallel with demo-mode logic above. The two never # touch the same state, so demo and live can both be tested # without interference (only one is active at a time per state). if not demo_mode: # Stale-grid: rebuild if price dropped too far without fills _live_stale_check() # Sync position with exchange (detects fills of pending orders) # γ-patch: устанавливаем флаг при успехе (sync не бросил exception) try: _sync_result = _live_sync_position() # Если sync вернул данные (или пустой list) - считаем success _live_position_synced = (_sync_result is not None) except Exception: pass # ─── SMA-ATR strategy: center + lock FSM ───────── grid_center = _resolve_grid_center(current_price) # ─── Kronos advisor (LIVE) ────────────────────── eff_buy_center_l, eff_sell_center_l, eff_step_l, kronos_paused_l, bias_changed_l = effective_grid_params( grid_center, settings["step_percent"], live=True ) if kronos_paused_l: logger.info("[KRONOS] LIVE paused: not placing new grid orders this cycle") # Не ставим новые ордера, но и не отменяем существующие # (на случай, если это короткая пауза) else: grid_center = eff_buy_center_l # BUGFIX-C: при смене bias — отменить висящие BUY и сбросить live_orders_placed, # чтобы на следующей итерации сетка пересобралась вокруг нового anchor. if bias_changed_l: # Ищем BUY-ордера в live_orders (список от _live_sync_position). # Отменяем через API (нужно знать ID, упростим: пока оставим флаг # на следующий rebuild; в LIVE отмена ордеров реализуется отдельно). live_orders_placed = False logger.info( f"[KRONOS] LIVE bias changed → will rebuild BUY grid " f"around new buy_anchor ${eff_buy_center_l:,.2f}" ) if abs(eff_step_l - settings["step_percent"]) > 1e-6: grid_engine = GridEngine( levels=settings["grid_levels"], step_percent=eff_step_l, take_profit_percent=settings["take_profit_percent"], symbol=SYMBOL, mode="live", grid_mode=GRID_MODE, ) # γ-patch: safety - не размещаем ордера пока sync не подтвердил отсутствие позиции _safe_to_place = _live_position_synced or live_position is not None if not _safe_to_place: logger.debug("[LIVE] position not yet synced - skipping order placement this cycle") else: # Inactivity soft rebuild (LIVE) — must run BEFORE placing # new orders so that an idle hanging grid is replaced first. _live_inactivity_check() # Place initial BUY grid (only once per session / per rebuild) if not kronos_paused_l and not live_orders_placed and live_position is None: _live_place_grid_buys(grid_center) # Once we have a position, place SELL limits if not kronos_paused_l and live_position is not None: _live_place_grid_sells(grid_center) # Take-profit check (независимо от паузы — TP всегда важен) _live_take_profit() # ─── DEMO mode (offline simulator, runs only in demo) ─────── # The simulator just runs the same grid logic on a fake balance # so you can see how the strategy behaves without touching the # real account. The two branches (LIVE / DEMO) are mutually # exclusive by `state.demo_mode` flag. if demo_mode: # Honor Kronos pause: SKIP new placements and after-fill SELL placement while paused # (TP and fill checks still run below, but no new orders) if kronos_paused: logger.debug("[KRONOS] DEMO paused, skipping new placements this cycle") # Skip the rest of the demo placement block # (we still want to fall through to TP/fill checks if position exists) # ─── Inactivity soft rebuild ──────────────────────────── # Если за INACTIVITY_RESTART_HOURS часов не было ни одной # сделки И BTC отошёл от start_price на >= # INACTIVITY_PRICE_DRIFT_PCT процентов — отменяем висящие # BUY-ордера и пересобираем сетку от текущей цены. # Анти-флуд: не чаще 1 раза в 6 часов. _now_ts = time.time() _inactivity_ok = ( INACTIVITY_RESTART_HOURS > 0 and start_price > 0 and current_price > 0 and demo.position is None and len(demo.orders) > 0 ) if _inactivity_ok: _idle_h = (_now_ts - last_trade_ts) / 3600.0 _drift_pct = abs((current_price - start_price) / start_price * 100.0) _cooldown_ok = ( _now_ts - last_inactivity_rebuild_ts > 1 * 3600 ) if ( _idle_h >= INACTIVITY_RESTART_HOURS and _drift_pct >= INACTIVITY_PRICE_DRIFT_PCT and _cooldown_ok ): cancelled = demo.cancel_open_buys( reason=f"inactivity {_idle_h:.1f}h, drift {_drift_pct:.2f}%" ) demo_orders_placed = False old_start = start_price start_price = current_price grid_stale = True grid_stale_drop_pct = _drift_pct grid_stale_rebuilds += 1 inactivity_rebuilds += 1 last_inactivity_rebuild_ts = _now_ts logger.warning( f"[INACTIVITY-REBUILD] {INACTIVITY_RESTART_HOURS}h without trades " f"(idle {_idle_h:.1f}h), drift {_drift_pct:.2f}% " f"from start ${old_start:,.2f} → ${current_price:,.2f}. " f"Cancelled {cancelled} orders, will rebuild at ${current_price:,.2f}" ) tg_notify( f"♻️ Inactivity-rebuild: {_idle_h:.1f}ч без сделок\n" f"BTC отошёл на {_drift_pct:.2f}% от центра сетки\n" f"Было: ${old_start:,.2f} → Стало: ${current_price:,.2f}\n" f"Отменено ордеров: {cancelled}" ) record_price_point(current_price, "rebuild") # ─── Stale-grid recovery ──────────────────────────────── # If no trades yet, position is empty, orders are still pending # and price dropped more than GRID_STALE_DROP_PERCENT from start_price # → cancel all orders and rebuild grid from current price. if ( GRID_STALE_DROP_PERCENT > 0 and start_price > 0 and current_price > 0 and len(demo.trade_log) == 0 and demo.position is None and len(demo.orders) > 0 ): drop_pct = (start_price - current_price) / start_price * 100 if drop_pct >= GRID_STALE_DROP_PERCENT: cancelled = demo.cancel_open_buys( reason=f"stale-grid drop {drop_pct:.2f}%" ) demo_orders_placed = False start_price = current_price grid_stale = True grid_stale_drop_pct = drop_pct grid_stale_rebuilds += 1 logger.warning( f"[STALE-GRID] Drop {drop_pct:.2f}% >= {GRID_STALE_DROP_PERCENT}% " f"with no fills. Cancelled {cancelled} orders, rebuilding at ${current_price:,.2f}" ) tg_notify( f"♻️ Stale-grid: цена упала на {drop_pct:.2f}% без сделок\n" f"Отменено ордеров: {cancelled}\n" f"Новая сетка от ${current_price:,.2f}" ) record_price_point(current_price, "rebuild") # rebuild on next iteration (orders will be re-placed below) # Place grid orders once (with dedup by level_id) — SKIP while Kronos paused OR grid locked if not kronos_paused and not grid_locked and (not demo_orders_placed or demo.position is None): # SMA-ATR: asymmetric 3:2 split based on current bias. # When no position, we still place BUY levels (catch dips). # When position exists, we also place SELL levels (close the round-trip). current_bias = _get_current_bias() buy_count, sell_count = _get_asymmetry(current_bias) # OPTION B (22.06.2026 22:39 GMT+7): в uptrend с открытой позицией # не ставить новый SELL — пусть TP отработает, не закрываемся раньше. if current_bias == "up" and demo.position is not None: sell_count = 0 levels = grid_engine.get_asymmetric_grid_levels( grid_center, buy_count=buy_count, sell_count=sell_count, base_qty=0.0001, buy_center=grid_center, sell_center=grid_center, ) placed_buys = 0 placed_sells = 0 for lvl in levels: if lvl.side == "BUY": if any(o["side"] == "BUY" and o["level_id"] == lvl.level_id and not o["filled"] for o in demo.orders): continue if demo.place_buy_order(lvl.price, lvl.qty, lvl.level_id): placed_buys += 1 elif lvl.side == "SELL" and demo.position: if any(o["side"] == "SELL" and o["level_id"] == lvl.level_id and not o["filled"] for o in demo.orders): continue if demo.place_sell_order(lvl.price, lvl.qty, lvl.level_id): placed_sells += 1 if placed_buys > 0 or placed_sells > 0: demo_orders_placed = True logger.info( f"[DEMO] Placed {placed_buys} BUY / {placed_sells} SELL " f"at price ${current_price:,.2f} (bias={current_bias}, " f"asymmetry={buy_count}:{sell_count}, sma=${sma_center:,.2f})" ) elif grid_locked and demo_orders_placed: # Edge: we just transitioned into locked — cancel remaining orders # and reset the flag so the next unlock builds fresh. cancelled = demo.cancel_open_buys(reason="grid_locked=True") demo_orders_placed = False if cancelled > 0: logger.debug(f"[LOCK] skipped placement, also cancelled {cancelled} open BUY order(s)") # Check fills filled_buys = demo.check_fill_buy(current_price) filled_sells = demo.check_fill_sell(current_price) if filled_sells: _save_trade_history(demo) # PERSIST after BUY→SELL for order in filled_buys: record_price_point(order['price'], "fill", "BUY") tg_notify( f"🟢 BUY заполнен\n" f"Цена: ${order['price']:,.2f}\n" f"Количество: {order['qty']} BTC\n" f"BTC: ${current_price:,.2f}" ) logger.info(f"[DEMO] BUY filled @ ${order['price']:,.2f} qty={order['qty']} BTC | spot=${current_price:,.2f}") # Обновляем last_trade_ts на любой fill (BUY или SELL) if filled_buys or filled_sells: last_trade_ts = time.time() for order in filled_sells: record_price_point(order['price'], "fill", "SELL") pnl = 0 for t in demo.trade_log: if t.get("exit") == order["price"]: pnl = t["pnl_usdc"] break tg_notify( f"🔴 SELL заполнен\n" f"Цена: ${order['price']:,.2f}\n" f"Количество: {order['qty']} BTC\n" f"PNL: {pnl:+.6f} USDT" ) logger.info(f"[DEMO] SELL filled @ ${order['price']:,.2f} qty={order['qty']} BTC | pnl={pnl:+.6f} USDT | spot=${current_price:,.2f}") # After BUY fill, place SELL orders. SELL-anchor lifted to entry+TP% # so SELL sits meaningfully above BUY (not just one step away). if not kronos_paused and not grid_locked and demo.position and demo.position.side == "BUY": entry_price = demo.position.entry_price # SELL center: entry + (TP% * 1.0) so first SELL ≈ TP-target # Multi-level sells fan out downward from there toward entry. sell_anchor_for_filled = entry_price * (1.0 + grid_engine.take_profit_percent / 100.0) # SMA-ATR: use sell_count from current bias (e.g., 2 for up, 3 for down) current_bias = _get_current_bias() _bc, sell_count_after_fill = _get_asymmetry(current_bias) placed_sells = 0 for lvl in grid_engine.get_asymmetric_grid_levels( grid_center, buy_count=0, # only SELL after fill sell_count=sell_count_after_fill, base_qty=0.0001, buy_center=grid_center, sell_center=sell_anchor_for_filled, ): if lvl.side != "SELL": continue if any(o["side"] == "SELL" and o["level_id"] == lvl.level_id and not o["filled"] for o in demo.orders): continue if demo.place_sell_order(lvl.price, lvl.qty, lvl.level_id): placed_sells += 1 if placed_sells > 0: logger.info( f"[DEMO] Placed {placed_sells} SELL (after BUY fill, bias={current_bias}) " f"at entry=${entry_price:,.2f} + TP% {grid_engine.take_profit_percent:.2f}% " f"→ SELL anchor ${sell_anchor_for_filled:,.2f} | spot=${current_price:,.2f}" ) tp_triggered = demo.check_take_profit(current_price, grid_engine.take_profit_percent) if tp_triggered: _save_trade_history(demo) # PERSIST after TAKE_PROFIT if tp_triggered: last_trade = demo.trade_log[-1] if demo.trade_log else {} logger.info( f"[DEMO] TAKE_PROFIT filled @ ${last_trade.get('exit', 0):,.2f} " f"entry=${last_trade.get('entry', 0):,.2f} " f"qty={last_trade.get('qty', 0)} BTC | " f"pnl={last_trade.get('pnl_usdc', 0):+.6f} USDT " f"({last_trade.get('pnl_pct', 0):+.2f}%) | " f"spot=${current_price:,.2f}" ) tg_notify( f"🎯 TAKE PROFIT!\n" f"Entry: ${last_trade.get('entry', 0):,.2f}\n" f"Exit: ${last_trade.get('exit', 0):,.2f}\n" f"PNL: {last_trade.get('pnl_usdc', 0):+.6f} USDT" ) # === Reconcile (QuantDinger ledger_reconcile pattern) === # Каждые RECONCILE_INTERVAL_SEC проверяем demo.position на shadow # (открыт ДО последнего closing trade). При RECONCILE_AUTO_CLEAR=True # очищаем автоматически; иначе только лог + TG. if RECONCILE_INTERVAL_SEC > 0 and (time.time() - _last_reconcile_ts) > RECONCILE_INTERVAL_SEC: _last_reconcile_ts = time.time() try: _recon = _reconcile_demo_position() if _recon.get("action") == "shadow_detected": if RECONCILE_AUTO_CLEAR: _apply_reconcile_clear(_recon) _recon["auto_cleared"] = True logger.warning( f"[RECONCILE] shadow detected → AUTO-CLEARED " f"entry=${_recon.get('position_entry'):,.2f} qty={_recon.get('position_qty')} " f"(shadow of trade exit @ {(_recon.get('shadow_of_trade') or {}).get('exit_time')})" ) tg_notify( f"🧹 Reconcile: shadow-позиция очищена\n" f"Entry: ${_recon.get('position_entry'):,.2f}\n" f"Shadow-трейд exit @ {(_recon.get('shadow_of_trade') or {}).get('exit_time')}" ) record_price_point(current_price, "reconcile", "shadow_cleared") # Persist trade_history try: th_path = Path('/root/grid-bot/trade_history.json') if th_path.exists(): hist = json.loads(th_path.read_text()) else: hist = {} hist['trade_log'] = demo.trade_log hist['balance'] = demo.balance th_path.write_text(json.dumps(hist, indent=2, default=str)) except Exception as e: logger.warning(f"reconcile persist failed: {e}") else: logger.warning( f"[RECONCILE] shadow position detected (auto-clear OFF): {_recon}" ) tg_notify( f"⚠️ Reconcile: shadow-позиция обнаружена\n" f"Entry: ${_recon.get('position_entry'):,.2f}\n" f"Открыта: {_recon.get('position_opened_at')}\n" f"Shadow-трейд exit @ {(_recon.get('shadow_of_trade') or {}).get('exit_time')}\n" f"auto-clear выключен (RECONCILE_AUTO_CLEAR=False)" ) except Exception as e: logger.error(f"[RECONCILE] error: {e}") # end if demo_mode (demo simulator branch) # Проверяем, не завершился ли фоновый kronos_advisor процесс _kronos_refresh_check() # === Kill switch hook (2026-06-22, обновлён 22.06 для live) === # Обновляем peak equity (balance + unrealized position) каждый тик. # Demo: equity = demo.balance + (current_price * demo.position.qty if BUY else 0) # Live: equity = USDT free cash + market value live_position (if BUY). Не включает external. _is_live = (not demo_mode) if _is_live: _equity = 0.0 try: _w = fetch_wallet_balance(force=False) # Equity fix (2026-07-08): peak_equity=1.0 — считал только USDT free. # USDT locked (в BUY) и USD free — игнорировались. # Теперь: USDT total (free+locked) + USD total + позиция. if _w: _balances = _w.get("balances") or [] _cash = 0.0 for b in _balances: cur = b.get("currency", "").upper() if cur == "USDT": _cash += b.get("total", b.get("free", 0)) elif cur == "USD": _cash += b.get("total", b.get("free", 0)) _equity = _cash else: _equity = 0.0 except Exception: _equity = 0.0 # kill switch не сработает если wallet недоступен if live_position and live_position.get("side") == "BUY" and current_price > 0: _equity += current_price * live_position.get("qty", 0) # external_position НЕ учитывается: бот отвечает только за свою часть. # Ручные сделки пользователя не идут в kill switch. else: _equity = demo.balance if demo.position and demo.position.side == "BUY" and current_price > 0: _equity += current_price * demo.position.qty kill_switch.update_equity(_equity) kill_switch.check_dd(_equity, is_live=_is_live) if kill_switch.is_killed(): logger.critical(f"[KILL-SWITCH] loop exiting (mode={'LIVE' if _is_live else 'DEMO'}): {kill_switch.kill_reason}") bot_running = False # Demo: cancel all demo BUY orders # Live: cancel all exchange orders + opt close position (через cancel_live) try: if _is_live: kill_switch.cancel_live(reason=kill_switch.kill_reason) else: cancelled = demo.cancel_open_buys(reason=f"kill_switch:{kill_switch.kill_reason}") if cancelled: logger.warning(f"[KILL-SWITCH] cancelled {cancelled} demo BUY orders") except Exception as _kc_err: logger.warning(f"[KILL-SWITCH] cancel failed: {_kc_err}") break await asyncio.sleep(2) except asyncio.CancelledError: logger.info("Trading loop cancelled") break except Exception as e: logger.error(f"Trading loop error: {e}") last_error = str(e) kill_switch.record_api_error(str(e)) await asyncio.sleep(2) def start_bot_async(): """Запуск trading_loop в отдельном event loop. Ловим все исключения (включая BaseException) и пишем полный traceback в /tmp/bot_crash.log, чтобы не терять информацию о падениях. Без этой обёртки исключение из coroutine проглатывалось и thread умирал молча. """ import traceback as _tb_b kill_switch.start() # Запускаем мониторинг (если enabled) loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: try: loop.run_until_complete(trading_loop()) except BaseException as _bexc: try: with open('/tmp/bot_crash.log', 'a', encoding='utf-8') as _cf: _cf.write(f"\\n=== {time.strftime('%Y-%m-%d %H:%M:%S')} ===\\n") _cf.write(''.join(_tb_b.format_exception(type(_bexc), _bexc, _bexc.__traceback__))) except Exception as _werr: pass logger.error(f"[bot-thread] crashed: {_bexc!r}") finally: try: pending = [t for t in asyncio.all_tasks(loop) if not t.done()] for t in pending: t.cancel() if pending: loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) loop.run_until_complete(loop.shutdown_asyncgens()) except Exception: pass loop.close() def watchdog_thread(): """Watchdog: monitor trading_loop heartbeat, auto-restart on hang. Trading_loop can hang silently (DNS hiccup, urllib timeout miss, aiohttp session leak, etc). Each iter writes /tmp/loop_beat and updates trading_loop._last_loop_iter_ts. Watchdog checks mtime every 30 sec; if heartbeat is stale > STUCK_SEC -- triggers the same /api/bot/stop + /api/bot/start cycle used for manual recovery. Stale heartbeat is NOT panic: slow DNS, GC pause, API restart all cause brief gaps. STUCK_SEC = 240s (4 min) is the trigger threshold. """ import base64 as _b64 import time as _wt global bot_running HEARTBEAT_FILE = "/tmp/loop_beat" STUCK_SEC = 240 CHECK_INTERVAL = 30 _auth = _b64.b64encode(f"{WEB_USERNAME}:{WEB_PASSWORD}".encode()).decode() logger.info( f"[WATCHDOG] started -- interval={CHECK_INTERVAL}s, stuck_sec={STUCK_SEC}s, " f"target=http://127.0.0.1:{PORT}" ) # In-memory heartbeat source (preferred): trading_loop._last_loop_iter_ts, # set at top of every iter in the while-loop body. Reliable -- bot is the # only writer. File mtime fallback: /tmp/loop_beat, also written by bot # but vulnerable to external touch. while True: _wt.sleep(CHECK_INTERVAL) if not bot_running: continue # Primary: in-memory timestamp (set by trading_loop at top of every iter) last_iter_ts = getattr(trading_loop, "_last_loop_iter_ts", 0) if last_iter_ts > 0: age = _wt.time() - last_iter_ts source = "in-mem" else: # Fallback: file mtime (covers the window before first iter) try: age = _wt.time() - os.path.getmtime(HEARTBEAT_FILE) source = "file-mtime" except OSError: continue # file missing -- bot still starting if age < STUCK_SEC: continue # Stuck! Trigger auto-restart logger.error( f"[WATCHDOG] trading_loop heartbeat stale for {int(age)}s " f"(> {STUCK_SEC}s, src={source}) -- triggering auto-restart" ) for endpoint in ("/api/bot/stop", "/api/bot/start"): try: url = f"http://127.0.0.1:{PORT}{endpoint}" req = urllib.request.Request(url, method="POST") req.add_header("Authorization", f"Basic {_auth}") resp = urllib.request.urlopen(req, timeout=5).read() logger.info(f"[WATCHDOG] {endpoint} OK: {resp[:80]!r}") if endpoint == "/api/bot/stop": _wt.sleep(5) # let old thread die except Exception as _e: logger.error(f"[WATCHDOG] {endpoint} failed: {_e}") def _run_async(coro): """Run an async coroutine from a sync Flask handler using a fresh event loop. Each call gets its own loop so we never step on the trading loop's loop and never leak half-closed sessions. """ loop = asyncio.new_event_loop() try: return loop.run_until_complete(coro) finally: loop.close() # ─── Auth ───────────────────────────────────────────────────────────────────── # Web-auth теперь из .env (WEB_USERNAME/WEB_PASSWORD). Если дефолты из config — # на старте выдаём предупреждение, но не блокируем запуск. if WEB_USERNAME == "admin" and WEB_PASSWORD == "changeme": logger.warning( "[AUTH] WEB_USERNAME/WEB_PASSWORD в .env пустые — используются deprecation-дефолты. " "Срочно смените в .env: WEB_USERNAME=... WEB_PASSWORD=..." ) USERNAME = WEB_USERNAME PASSWORD = WEB_PASSWORD _session_cache = {} # simple server-side session: token -> True def check_auth(): auth = request.authorization if not auth: return False return auth.username == USERNAME and auth.password == PASSWORD def require_auth(f): def wrapper(*args, **kwargs): if not check_auth(): resp = jsonify({"error": "Authentication required"}) resp.headers["WWW-Authenticate"] = 'Basic realm="Grid Bot"' return resp, 401 return f(*args, **kwargs) wrapper.__name__ = f.__name__ return wrapper @app.before_request def check_all_requests(): # Allow static assets and login page if request.path in ("/", "/login", "/favicon.ico"): return None if request.path.startswith("/static/"): return None # Public endpoint: /api/consult — приём заявок с сайта it.kolp.pro # Защита: rate-limit (1/час/IP) + sanitization + email validation внутри Blueprint if request.path == "/api/consult" or request.path.startswith("/api/consult/"): return None # Require auth for all API calls if request.path.startswith("/api/"): if not check_auth(): resp = jsonify({"error": "Authentication required"}) resp.headers["WWW-Authenticate"] = 'Basic realm="Grid Bot"' return resp, 401 # ─── REST API ───────────────────────────────────────────────────────────────── @app.route("/login") def login_page(): auth = request.authorization if auth and auth.username == USERNAME and auth.password == PASSWORD: return redirect("/") resp = jsonify({"error": "Authentication required"}) resp.headers["WWW-Authenticate"] = 'Basic realm="Grid Bot"' return resp, 401 @app.route("/") def index(): auth = request.authorization if not auth or auth.username != USERNAME or auth.password != PASSWORD: resp = jsonify({"error": "Authentication required"}) resp.headers["WWW-Authenticate"] = 'Basic realm="Grid Bot"' return resp, 401 return render_template("dashboard.html") @app.route("/api/status") def api_status(): global current_price, start_price, bot_running global grid_stale, grid_stale_drop_pct, grid_stale_rebuilds global inactivity_rebuilds, last_trade_ts, last_error global _price_source, last_inactivity_rebuild_ts state = load_state() demo_mode = state.get("demo_mode", DEMO_MODE) if demo_mode: status = demo.get_status(current_price) if current_price else {} price_change_pct = 0.0 price_change_abs = 0.0 if start_price > 0 and current_price > 0: price_change_abs = current_price - start_price price_change_pct = (price_change_abs / start_price) * 100 return jsonify({ "mode": "demo", "symbol": SYMBOL, "current_price": current_price, "start_price": start_price, "price_source": _price_source, "price_change_pct": round(price_change_pct, 4), "price_change_abs": round(price_change_abs, 4), "bot_running": bot_running, "balance": status.get("balance", DEMO_START_BALANCE), "position": status.get("position"), "open_orders": status.get("open_orders", 0), "quote": dict(last_quote) if last_quote else {}, "total_trades": status.get("total_trades", 0), "total_pnl": status.get("total_pnl", 0), "last_trade": status.get("last_trade"), "error": last_error, "grid_stale": grid_stale, "grid_stale_drop_pct": round(grid_stale_drop_pct, 4), "grid_stale_rebuilds": grid_stale_rebuilds, "inactivity_rebuilds": inactivity_rebuilds, "idle_hours": round((time.time() - last_trade_ts) / 3600.0, 2) if last_trade_ts > 0 else 0.0, "inactivity_threshold_h": INACTIVITY_RESTART_HOURS, "inactivity_drift_pct": INACTIVITY_PRICE_DRIFT_PCT, "stale_threshold_pct": GRID_STALE_DROP_PERCENT, # SMA-ATR strategy (2026-06-21) "strategy_center_mode": STRATEGY_CENTER_MODE, "sma_center": round(sma_center, 2) if sma_center > 0 else None, "sma_dev_pct": round(((current_price - sma_center) / sma_center * 100), 4) if sma_center > 0 else None, "grid_locked": grid_locked, "lock_reason": lock_reason, "strategy_bias": _get_current_bias(), }) else: # Live mode: fetch (cached) wallet balance from Tradernet try: wallet = fetch_wallet_balance(force=False) except Exception as e: wallet = {"balances": [], "primary_currency": "USDT", "primary_balance": 0.0, "stale": True, "error": str(e), "age_seconds": 0} # Account info (getOPQ) — non-blocking, uses cache try: account = fetch_account_info(force=False) except Exception: account = {} # Quote info (lot, min_step, currency) — cached try: quote_info = fetch_quote_info(force=False) except Exception: quote_info = {} # Last quote (bid/ask, day change) try: quote = dict(last_quote) if last_quote else {} except Exception: quote = {} return jsonify({ "mode": "live", "symbol": SYMBOL, "current_price": current_price, "start_price": start_price, "price_source": _price_source, "price_change_pct": round(((current_price - start_price) / start_price * 100), 4) if start_price > 0 and current_price > 0 else 0.0, "price_change_abs": round((current_price - start_price), 4) if start_price > 0 else 0.0, "bot_running": bot_running, "error": last_error, "balance": wallet.get("primary_balance", 0.0), "currency": wallet.get("primary_currency", "USDT"), "balances": wallet.get("balances", []), "wallet_stale": wallet.get("stale", True), "wallet_error": wallet.get("error"), "wallet_age_seconds": int(wallet.get("age_seconds", 0)), "account": { "brief_nm": account.get("brief_nm"), "main_curr": account.get("main_curr"), "active": account.get("active"), "rev": account.get("rev"), }, "quote_info": { "ticker": quote_info.get("ticker", SYMBOL), "short_name": quote_info.get("short_name"), "currency": quote_info.get("currency"), "lot": quote_info.get("lot"), "min_step": quote_info.get("min_step"), "mkt_name": quote_info.get("mkt_name"), }, "quote": quote, "live_position": { "side": live_position.get("side"), "entry_price": live_position.get("entry_price"), "qty": live_position.get("qty"), "order_id": live_position.get("order_id"), "avg_price_exchange": live_position.get("avg_price_exchange"), "current_pnl_pct": round(((current_price - live_position.get("entry_price", 0)) / live_position.get("entry_price", 1) * 100), 4) if live_position and live_position.get("entry_price", 0) > 0 else 0.0, } if live_position else None, "live_pending_orders": [ {"order_id": o["order_id"], "side": o["side"], "price": o["price"], "qty": o["qty"], "level_id": o["level_id"], "status": o["status"]} for o in live_pending_orders ], "live_trade_count": len(live_trade_log), "live_total_pnl": round(sum(t["pnl_usdc"] for t in live_trade_log), 8), "live_last_trade": live_trade_log[-1] if live_trade_log else None, "grid_stale": grid_stale, "grid_stale_drop_pct": round(grid_stale_drop_pct, 4), "grid_stale_rebuilds": grid_stale_rebuilds, "stale_threshold_pct": GRID_STALE_DROP_PERCENT, }) @app.route("/api/balance") def api_balance(): state = load_state() demo_mode = state.get("demo_mode", DEMO_MODE) if demo_mode: return jsonify(demo.get_status(current_price)) # Live: ?refresh=1 forces an immediate API call (bypasses cache) force = request.args.get("refresh") == "1" try: wallet = fetch_wallet_balance(force=force) except Exception as e: wallet = {"balances": [], "primary_currency": "USDT", "primary_balance": 0.0, "stale": True, "error": str(e), "age_seconds": 0} return jsonify({ "mode": "live", "balance": wallet.get("primary_balance", 0.0), "currency": wallet.get("primary_currency", "USDT"), "balances": wallet.get("balances", []), "stale": wallet.get("stale", True), "error": wallet.get("error"), "age_seconds": int(wallet.get("age_seconds", 0)), }) @app.route("/api/grid") def api_grid(): global current_price, grid_engine if current_price <= 0: return jsonify({"error": "Price not available"}), 503 state = load_state() # SMA-ATR strategy: show the asymmetric grid (3:2) that the bot actually places, # centered on SMA (not current_price) when STRATEGY_CENTER_MODE == "sma". if STRATEGY_CENTER_MODE == "sma": grid_center = sma_center if sma_center > 0 else current_price try: bias = _get_current_bias() buy_count, sell_count = _get_asymmetry(bias) except Exception: bias, buy_count, sell_count = "range", 3, 2 levels = grid_engine.get_asymmetric_grid_levels( grid_center, buy_count=buy_count, sell_count=sell_count, base_qty=0.0001, buy_center=grid_center, sell_center=grid_center, ) else: grid_center = current_price levels = grid_engine.get_grid_levels(grid_center) # In live mode, mark each level with whether we have a real exchange order live_orders_by_level = {} for o in live_pending_orders: if o.get("status") == "pending": live_orders_by_level[(o["side"], o["level_id"])] = o # Build set of (side, level_id) that are ACTUALLY placed on the simulator/exchange. # Demo: orders in demo.orders (not filled yet). Live: pending in live_pending_orders. placed_keys: set = set() if grid_engine and grid_engine.mode == "demo": for o in (demo.orders if demo else []): if not o.get("filled"): placed_keys.add((o.get("side"), o.get("level_id"))) else: for o in live_pending_orders: if o.get("status") == "pending": placed_keys.add((o.get("side"), o.get("level_id"))) levels_data = [] for l in levels: d = l.to_dict() match = live_orders_by_level.get((l.side, l.level_id)) if match: d["exchange_order_id"] = match["order_id"] d["exchange_status"] = match["status"] d["exchange_placed"] = True else: d["exchange_placed"] = False d["placed"] = (l.side, l.level_id) in placed_keys levels_data.append(d) return jsonify({ "center_price": grid_center, "levels": levels_data, "total_orders": len(levels), "step_percent": grid_engine.step_percent, "take_profit_percent": grid_engine.take_profit_percent, "mode": grid_engine.mode, "live_pending_count": sum(1 for o in live_pending_orders if o.get("status") == "pending"), "live_position": live_position, }) @app.route("/api/account") def api_account(): """Account info from getOPQ (brief_nm, main_curr, active, etc).""" force = request.args.get("refresh") == "1" try: info = fetch_account_info(force=force) except Exception as e: info = {"error": str(e), "stale": True, "ts": 0, "age_seconds": 0} return jsonify({ "brief_nm": info.get("brief_nm"), "main_curr": info.get("main_curr"), "active": info.get("active"), "rev": info.get("rev"), "init_margin": info.get("init_margin"), "reception": info.get("reception"), "f_kval": info.get("f_kval"), "stale": info.get("stale", True), "error": info.get("error"), "age_seconds": int(info.get("age_seconds", 0)), }) @app.route("/api/quote-info") def api_quote_info(): """Security info for SYMBOL (lot, min_step, currency, market).""" force = request.args.get("refresh") == "1" try: info = fetch_quote_info(force=force) except Exception as e: info = {"error": str(e), "stale": True, "ts": 0, "age_seconds": 0} return jsonify({ "ticker": info.get("ticker", SYMBOL), "short_name": info.get("short_name"), "currency": info.get("currency"), "lot": info.get("lot"), "min_step": info.get("min_step"), "mkt_name": info.get("mkt_name"), "mkt_tz": info.get("mkt_tz"), "stale": info.get("stale", True), "error": info.get("error"), "age_seconds": int(info.get("age_seconds", 0)), }) @app.route("/api/quote") def api_quote(): """Last full quote snapshot (ltp, bid, ask, day change, volume).""" global last_quote, current_price if not last_quote: return jsonify({ "ltp": current_price, "bap": 0, "bbp": 0, "ltt": "", "chg": 0, "chg110": 0, "vol": 0, "op": 0, "pp": 0, "stale": True, }) return jsonify({**last_quote, "stale": False}) @app.route("/api/candles") def api_candles(): """OHLCV candles for the active symbol. Query: ?tf=1|5|15|60|1440 (default 60) &refresh=1 (bypass cache) """ try: tf = int(request.args.get("tf", "60")) except ValueError: tf = 60 if tf not in (1, 5, 15, 60, 1440): tf = 60 force = request.args.get("refresh") == "1" try: data = fetch_candles(timeframe_min=tf, force=force) except Exception as e: data = {"timeframe": tf, "candles": [], "error": str(e), "stale": True, "ts": 0, "age_seconds": 0} return jsonify({ "timeframe": data.get("timeframe", tf), "candles": data.get("candles", []), "stale": data.get("stale", True), "error": data.get("error"), "age_seconds": int(data.get("age_seconds", 0)), }) @app.route("/api/orderbook") @require_auth async def api_orderbook(): global api_client if not api_client: return jsonify({"error": "API client not initialized"}), 503 try: data = await api_client.get_orderbook(SYMBOL, depth=20) # Normalize response - Tradernet returns {result: {bid: [...], ask: [...]}} result = data.get("result", data) return jsonify({ "symbol": SYMBOL, "bids": result.get("bid", result.get("bids", [])), "asks": result.get("ask", result.get("asks", [])), }) except Exception as e: return jsonify({"error": str(e)}), 500 @app.route("/api/logs") def api_logs(): state = load_state() demo_mode = state.get("demo_mode", DEMO_MODE) if demo_mode: return jsonify({ "trades": demo.trade_log[-50:], "total_pnl": sum(t["pnl_usdc"] for t in demo.trade_log), "total_trades": len(demo.trade_log), }) # Live mode: return real trade log from live broker return jsonify({ "trades": live_trade_log[-50:], "total_pnl": round(sum(t["pnl_usdc"] for t in live_trade_log), 8), "total_trades": len(live_trade_log), "open_position": live_position, "pending_orders": [ {"order_id": o["order_id"], "side": o["side"], "price": o["price"], "qty": o["qty"], "level_id": o["level_id"], "status": o["status"]} for o in live_pending_orders ], }) # ─── Price history (for chart) ────────────────────────────────────────── RANGE_SECONDS = { "1h": 3600, "6h": 6 * 3600, "24h": 24 * 3600, "all": None, # all retained points } @app.route("/api/price-history") def api_price_history(): """Return price samples + fills + rebuild markers within the requested time range. Query: ?range=1h|6h|24h|all (default 1h) Response: { range: "1h", range_seconds: 3600, sample_interval: 30, points: [ {t, p, type, side?}, ... ] # ticks (sampled) fills: [ {t, p, side, level_id?}, ... ] # BUY/SELL fills as overlay points rebuilds: [ {t, p}, ... ] # stale-grid rebuild markers levels: [ {price, side, level_id}, ... ] # current grid levels for horizontal lines current_price: float, start_price: float } """ rng = (request.args.get("range") or "1h").lower() if rng not in RANGE_SECONDS: rng = "1h" rng_secs = RANGE_SECONDS[rng] now = time.time() cutoff = now - rng_secs if rng_secs is not None else 0.0 with price_history_lock: # Filter once: bucketed into ticks / fills / rebuilds ticks: list = [] fills: list = [] rebuilds: list = [] for pt in price_history: if pt["t"] < cutoff: continue ptype = pt.get("type", "tick") entry = {"t": pt["t"], "p": pt["p"]} if ptype == "fill": entry["side"] = pt.get("side", "BUY") fills.append(entry) elif ptype == "rebuild": rebuilds.append(entry) else: ticks.append(entry) # Down-sample ticks to keep chart responsive on long ranges MAX_TICKS = 800 if len(ticks) > MAX_TICKS: step = max(1, len(ticks) // MAX_TICKS) ticks = ticks[::step] # Current grid levels for horizontal reference lines levels_data: list = [] if grid_engine and current_price > 0: try: for lv in grid_engine.get_grid_levels(current_price): levels_data.append({ "price": lv.price, "side": lv.side, "level_id": lv.level_id, }) except Exception: pass return jsonify({ "range": rng, "range_seconds": rng_secs, "sample_interval": PRICE_SAMPLE_INTERVAL, "points": ticks, "fills": fills, "rebuilds": rebuilds, "levels": levels_data, "current_price": current_price, "start_price": start_price, "bot_running": bot_running, }) @app.route("/api/kronos") def api_kronos(): """Kronos advisor status: current advice, applied params, history. Auth: требуется (глобальный before_request для /api/*). Returns: { "enabled": bool, # KRONOS_ENABLED из .env "live_enabled": bool, # KRONOS_LIVE_ENABLED из .env "model": str, "tf_min": int, ... "current_advice": dict | None, # что лежит в kronos_advice.json "last_applied": dict | None, # что trading_loop реально применил "pause_until": float, # epoch seconds, 0 если не активна "paused_now": bool, "history": [dict, ...], # последние 20 применённых советов "advice_file": str, "advice_file_exists": bool, } """ advice_path = Path(KRONOS_ADVICE_FILE) current_advice = None if advice_path.exists(): try: current_advice = json.loads(advice_path.read_text()) except Exception: pass paused_now = bool(kronos_pause_until and time.time() < kronos_pause_until) return jsonify({ "enabled": KRONOS_ENABLED, "live_enabled": KRONOS_LIVE_ENABLED, "model": KRONOS_MODEL, "tf_min": KRONOS_TF_MIN, "lookback": KRONOS_LOOKBACK, "pred_len": KRONOS_PRED_LEN, "min_confidence": KRONOS_MIN_CONFIDENCE, "advice_file": KRONOS_ADVICE_FILE, "advice_file_exists": advice_path.exists(), "current_advice": current_advice, "last_applied": kronos_last_applied or None, "pause_until": kronos_pause_until, "paused_now": paused_now, "history": kronos_advice_history[-20:], }) # Глобальный лок на запуск refresh (защита от двойного клика) _kronos_refresh_lock = Lock() _kronos_refresh_pid: Optional[int] = None _kronos_refresh_started: float = 0.0 @app.route("/api/kronos/refresh", methods=["POST"]) def api_kronos_refresh(): """Запустить kronos_advisor_main.py и обновить kronos_advice.json. Неблокирующий: запускает subprocess в фоне, сразу возвращает ответ. Если уже идёт refresh — вернёт 409 Conflict. """ global _kronos_refresh_pid, _kronos_refresh_started with _kronos_refresh_lock: # Если уже идёт процесс, но он завис дольше 5 мин — считаем его мёртвым if _kronos_refresh_pid and _kronos_refresh_started: if time.time() - _kronos_refresh_started > 300: logger.warning("[KRONOS REFRESH] previous run > 5min, считаем мёртвым") _kronos_refresh_pid = None if _kronos_refresh_pid: return jsonify({"ok": False, "error": "уже выполняется", "pid": _kronos_refresh_pid}), 409 # Запускаем kronos_advisor_main.py в фоне venv_python = Path(__file__).parent / "kronos-venv" / "bin" / "python" if not venv_python.exists(): return jsonify({"ok": False, "error": f"venv не найден: {venv_python}"}), 500 cmd = [str(venv_python), str(Path(__file__).parent / "kronos_advisor_main.py")] # В проде — --live, в дев-окружении — по умолчанию binance if KRONOS_LIVE_ENABLED: cmd.append("--live") log_path = Path(__file__).parent / "kronos-advisor.log" log_fh = open(log_path, "a") try: proc = subprocess.Popen( cmd, stdout=log_fh, stderr=subprocess.STDOUT, cwd=str(Path(__file__).parent), start_new_session=True, ) _kronos_refresh_pid = proc.pid _kronos_refresh_started = time.time() except Exception as e: log_fh.close() return jsonify({"ok": False, "error": f"Popen failed: {e}"}), 500 return jsonify({ "ok": True, "message": f"запущен kronos advisor (pid={_kronos_refresh_pid}), подождите 5-30 сек", "pid": _kronos_refresh_pid, }) def _kronos_refresh_check(): """Фоновая проверка: завершился ли subprocess refresh-а. Вызывается в trading_loop.""" global _kronos_refresh_pid, _kronos_refresh_started if not _kronos_refresh_pid: return try: os.kill(_kronos_refresh_pid, 0) # не убиваем, просто проверяем except ProcessLookupError: # Процесс завершился logger.info(f"[KRONOS REFRESH] pid={_kronos_refresh_pid} завершился") _kronos_refresh_pid = None _kronos_refresh_started = 0.0 @app.route("/api/kronos/refresh/status") def api_kronos_refresh_status(): """Статус фонового refresh-процесса (для UI).""" global _kronos_refresh_pid, _kronos_refresh_started running = False if _kronos_refresh_pid: try: os.kill(_kronos_refresh_pid, 0) running = True except ProcessLookupError: # Процесс завершился — чистим глобал _kronos_refresh_pid = None _kronos_refresh_started = 0.0 return jsonify({ "running": running, "pid": _kronos_refresh_pid, "started_at": _kronos_refresh_started, }) # Static assets (chart.js, plugins) — served from /root/grid-bot/static/ @app.route("/static/") def static_files(filename): return send_from_directory(Path(__file__).parent / "static", filename) @app.route("/api/settings", methods=["GET"]) def api_settings_get(): return jsonify(read_current_settings()) @app.route("/api/settings", methods=["POST"]) def api_settings_set(): global grid_engine, demo, demo_orders_placed try: state = load_state() data = request.get_json() or {} if "grid_levels" in data: state["grid_levels"] = max(5, min(20, int(data["grid_levels"]))) if "step_percent" in data: # Convention: value is in PERCENT (0.5 = 0.5%, 0.85 = 0.85%). # Earlier versions inverted this with a /100 if val > 1.0. val = float(data["step_percent"]) state["step_percent"] = max(0.0001, min(0.1, val)) if "take_profit_percent" in data: # Convention: value is in PERCENT (3.0 = 3%, 0.5 = 0.5%). # Earlier versions inverted this with a /100 if val > 1.0, which # collapsed "3" → 0.03 → "TP triggers at 0.03%". Removed. val = float(data["take_profit_percent"]) state["take_profit_percent"] = max(0.01, min(20.0, val)) if "demo_mode" in data: state["demo_mode"] = bool(data["demo_mode"]) save_state(state) grid_engine = GridEngine( levels=state["grid_levels"], step_percent=state["step_percent"], take_profit_percent=state["take_profit_percent"], symbol=SYMBOL, mode="demo" if state.get("demo_mode", DEMO_MODE) else "live", ) # Reset demo state when settings change demo.orders = [] demo.position = None demo_orders_placed = False except Exception as e: return jsonify({"error": f"Save/parse error: {e}"}), 500 return jsonify({"ok": True, "settings": read_current_settings()}) @app.route("/api/bot/start", methods=["POST"]) def api_bot_start(): global bot_running, bot_thread, api_client if bot_running: return jsonify({"ok": False, "error": "Already running"}) # После предыдущего Stop aiohttp-session привязан к закрытому loop — # пересоздаём клиента, чтобы trading_loop получил свежую session. if api_client is not None: try: # Лучше создать новый sync-loop для закрытия старой session import threading as _t def _close(): loop = asyncio.new_event_loop() try: loop.run_until_complete(api_client.close()) finally: loop.close() _t.Thread(target=_close, daemon=True).start() except Exception as e: logger.warning(f"close old api_client failed: {e}") api_client = TradernetAPI( TRADERNET_PUBLIC_KEY, TRADERNET_PRIVATE_KEY, TRADERNET_LOGIN, TRADERNET_PASSWORD, TRADERNET_BASE_URL, ) bot_running = True bot_thread = Thread(target=start_bot_async, daemon=True) bot_thread.start() return jsonify({"ok": True}) @app.route("/api/bot/stop", methods=["POST"]) def api_bot_stop(): global bot_running, bot_thread bot_running = False # Не блокируем воркер — пусть thread-daemon сам завершится на следующей итерации. # Условие while bot_running в начале trading_loop гарантирует выход через ~0.5 сек. bot_thread = None tg_notify("🛑 Бот остановлен") return jsonify({"ok": True}) @app.route("/api/kill-switch/status", methods=["GET"]) def api_kill_switch_status(): """Текущее состояние kill switch — для дашборда и мониторинга.""" return jsonify(kill_switch.get_status()) @app.route("/api/kill-switch/reset", methods=["POST"]) def api_kill_switch_reset(): """Сбрасывает kill state. Требует ручного /api/bot/start для возобновления.""" if not kill_switch.is_killed(): return jsonify({"ok": True, "note": "kill switch not triggered"}) kill_switch.reset() logger.warning("[KILL-SWITCH] RESET via /api/kill-switch/reset (manual)") tg_notify("⚠️ Kill switch сброшен вручную. Требуется /api/bot/start для возобновления.") return jsonify({"ok": True, "note": "Use /api/bot/start to resume trading"}) @app.route("/api/kill-switch/test", methods=["POST"]) def api_kill_switch_test(): """DEV-ONLY: принудительно trigger kill switch для тестирования. Защищён auth.""" from flask import request as _req if WEB_USERNAME and WEB_PASSWORD: auth = _req.authorization if not auth or auth.username != WEB_USERNAME or auth.password != WEB_PASSWORD: return jsonify({"error": "auth required"}), 401 reason = "manual_test" if _req.is_json: body = _req.get_json(silent=True) or {} reason = body.get("reason", "manual_test") kill_switch._trigger(f"test:{reason}") return jsonify({"ok": True, "triggered": reason}) @app.route("/api/reset", methods=["POST"]) def api_reset(): global demo, demo_orders_placed, start_price, current_price demo.reset() demo_orders_placed = False start_price = current_price return jsonify({"ok": True, "balance": demo.balance}) @app.route("/api/soft-rebuild", methods=["POST"]) def api_soft_rebuild(): """Ручной soft rebuild сетки от текущей цены. Логика (аналог inactivity-rebuild, но без 12ч/0.5% порогов): - если есть открытая позиция → отказ - отменить все висящие ордера в demo.orders - сдвинуть start_price к current_price - сбросить demo_orders_placed, чтобы trading_loop переставил сетку - инкрементить счётчики, обновить cooldown """ global start_price, demo_orders_placed, grid_stale, grid_stale_drop_pct global grid_stale_rebuilds, inactivity_rebuilds, last_inactivity_rebuild_ts global live_orders_placed, live_pending_orders is_live = (not DEMO_MODE) # Live-ветка: проверяем live_position, не demo.position if is_live and live_position: return jsonify({ "ok": False, "error": "live_position_open", "warning": "Открытая live-позиция — soft rebuild невозможен, сначала закройте позицию", }), 200 if (not is_live) and demo.position is not None: return jsonify({ "ok": False, "error": "position_open", "warning": "Открытая позиция — soft rebuild невозможен, сначала закройте позицию", }), 200 if current_price <= 0: return jsonify({"ok": False, "error": "no_price"}), 500 # Live: отменяем реальные ордера через биржевой API + сбрасываем флаги if is_live: cancelled = len(live_pending_orders or []) _live_cancel_all_pending() live_orders_placed = False else: cancelled = demo.cancel_open_buys(reason="manual /api/soft-rebuild") demo_orders_placed = False old_start = start_price start_price = current_price drift_pct = ( abs((current_price - old_start) / old_start * 100.0) if old_start > 0 else 0.0 ) grid_stale = True grid_stale_rebuilds += 1 inactivity_rebuilds += 1 last_inactivity_rebuild_ts = time.time() logger.warning( f"[MANUAL-REBUILD] BTC ${old_start:,.2f} → ${current_price:,.2f} " f"(drift {drift_pct:.2f}%). Cancelled {cancelled} orders, " f"will rebuild grid at ${current_price:,.2f}" ) tg_notify( f"♻️ Manual soft-rebuild\n" f"Было: ${old_start:,.2f} → Стало: ${current_price:,.2f}\n" f"Drift: {drift_pct:.2f}%\n" f"Отменено ордеров: {cancelled}" ) record_price_point(current_price, "rebuild") return jsonify({ "ok": True, "old_start": old_start, "new_start": start_price, "drift_pct": drift_pct, "cancelled": cancelled, "inactivity_rebuilds": inactivity_rebuilds, "grid_stale_rebuilds": grid_stale_rebuilds, }) @app.route("/api/_debug/demo_orders") def _debug_demo_orders(): """TEMPORARY debug endpoint — list contents of demo.orders.""" out = [] for o in demo.orders: out.append({"side": o.get("side"), "price": o.get("price"), "qty": o.get("qty"), "level_id": o.get("level_id"), "filled": o.get("filled")}) return jsonify({ "count": len(demo.orders), "unique_level_ids": len(set(o["level_id"] for o in demo.orders if not o["filled"])), "orders": out, "position": {"side": demo.position.side, "entry": demo.position.entry_price, "qty": demo.position.qty} if demo.position else None, "demo_orders_placed_flag": demo_orders_placed, }) # ─── Main ───────────────────────────────────────────────────────────────────── def main(): global api_client parser = argparse.ArgumentParser() parser.add_argument("--dry-run", action="store_true") parser.add_argument("--once", action="store_true") args = parser.parse_args() api_client = TradernetAPI( TRADERNET_PUBLIC_KEY, TRADERNET_PRIVATE_KEY, TRADERNET_LOGIN, TRADERNET_PASSWORD, TRADERNET_BASE_URL, ) logger.info(f"Grid Bot starting — mode={'DEMO' if DEMO_MODE else 'LIVE'}") logger.info(f"Tradernet API: {TRADERNET_BASE_URL}") logger.info(f"Trading pair: {SYMBOL}") if args.once: logger.info("Running once (--once mode)") loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) loop.run_until_complete(trading_loop()) loop.run_until_complete(api_client.close()) return logger.info(f"Starting web dashboard on {HOST}:{PORT}") # Watchdog: detects trading_loop hang and auto-restarts via HTTP. # Started BEFORE app.run() so it lives for the full process lifetime. from threading import Thread _wd = Thread(target=watchdog_thread, daemon=True, name="watchdog") _wd.start() logger.info(f"[WATCHDOG] thread started: {_wd.ident}, alive={_wd.is_alive()}") # Автостарт trading_loop, если это не LIVE-прод-режим # (в LIVE оператор запускает через /api/bot/start вручную после проверки настроек) if DEMO_MODE: logger.info("DEMO mode → auto-starting trading loop") global bot_running bot_running = True from threading import Thread def _safe_start(): try: logger.info("[bot-thread] starting trading_loop...") start_bot_async() except Exception as e: logger.error(f"[bot-thread] crashed: {e}", exc_info=True) _t = Thread(target=_safe_start, daemon=True, name="trading-loop") _t.start() logger.info(f"[main] trading_loop thread started: {_t.ident}, alive={_t.is_alive()}") app.run(host=HOST, port=PORT, debug=DEBUG, threaded=True) def _save_trade_history(demo_obj): """Persist demo trade_log + balance to /root/grid-bot/trade_history.json. Безопасно вызывать после каждой сделки (TP / BUY→SELL fill). При ошибке — пишет WARNING, но не падает (trading loop не должен ломаться). """ try: th_path = Path('/root/grid-bot/trade_history.json') if th_path.exists(): hist = json.loads(th_path.read_text()) else: hist = {} hist['trade_log'] = demo_obj.trade_log hist['balance'] = demo_obj.balance tmp_path = th_path.with_suffix('.json.tmp') tmp_path.write_text(json.dumps(hist, indent=2, default=str)) tmp_path.rename(th_path) logger.debug(f"[PERSIST] trade_history saved: {len(demo_obj.trade_log)} trades, balance=${demo_obj.balance:.2f}") except Exception as e: logger.warning(f"[PERSIST] trade_history save failed: {e}") if __name__ == "__main__": main()