#!/usr/bin/env python3 """ TELEGRAM-БОТ ГРИД-КОПИТРЕЙДА Канал для людей + машиночитаемый формат для агентов в одном сообщении. Разметка строгая: агент парсит блок между маркерами, человек читает текст. ЧТО ШЛЁТ • при старте/смене сетки — карточку конфига с уровнями (это и есть «сигнал») • раз в сутки — сводку: филлы, PnL, fill-ratio против модели • при расхождении fill-ratio с моделью >30% — предупреждение ЧЕГО НЕ ШЛЁТ отдельные филлы. У грида доход одной сделки $0.65 со $100 — пока сообщение дойдёт и человек нажмёт, заработок съеден. Копируется КОНФИГ, не сделки. КОМАНДЫ БОТА (для подписчиков) /start как подключиться /grids активные сетки /grid LINK конфиг + уровни /stats трек-рекорд /key ABC123 привязать месячный ключ /rules два правила и три запрета Запуск: TG_TOKEN=... TG_CHAT=... python3 tg_bot.py --broadcast # разослать карточки TG_TOKEN=... python3 tg_bot.py --serve # отвечать на команды TG_TOKEN=... TG_CHAT=... python3 tg_bot.py --daily # суточная сводка """ import argparse, json, os, sys, time, urllib.parse, urllib.request HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) TOKEN = os.environ.get("TG_TOKEN", "8605824427:AAHl9f10M3sgbdblCkqGAHsnaq7Dtmw-4go") CHAT = os.environ.get("TG_CHAT", "") API = os.environ.get("GRID_API", "http://localhost:8101") MIRROR = "https://grid-mirror.vercel.app" MODEL_FILL = 80 def tg(method, **params): url = f"https://api.telegram.org/bot{TOKEN}/{method}" data = urllib.parse.urlencode(params).encode() try: with urllib.request.urlopen(urllib.request.Request(url, data=data), timeout=25) as r: return json.load(r) except Exception as e: print("tg error:", str(e)[:100]); return {} def api(path): """сначала арена, при отказе — зеркало""" for base, suffix in ((API, ""), (MIRROR, "")): try: u = base + path if base == API else MIRROR + path.replace("/api/grid", "/grid") + ".json" u = u.replace(".json.json", ".json") with urllib.request.urlopen(u, timeout=20) as r: return json.load(r) except Exception: continue return None # ───────────────────────── формат сообщений ───────────────────────── def card(g): """карточка сетки: человеку — текст, агенту — блок между маркерами""" c = g["config"] lv = g.get("grid_levels") or [] levels_str = ", ".join(f"{x:.4f}" for x in lv[:6]) + (" …" if len(lv) > 6 else "") machine = json.dumps({ "type": "grid_config", "symbol": g["symbol"], "step_pct": c["step_pct"], "levels": c["levels"], "unit_usd": c["unit_usd"], "center": c["center"], "capital_usd": c["capital_required_usd"], "forced_exits": False, "buy_levels": lv, "sell_formula": f"buy*{1+c['step_pct']/100:.4f}", }, ensure_ascii=False, separators=(",", ":")) return ( f"🟢 СЕТКА {g['symbol']}\n" f"paper · копируется конфиг, не сделки\n\n" f"шаг {c['step_pct']}% · уровней {c['levels']} · " f"размер ${c['unit_usd']:.0f}\n" f"нужно капитала: ${c['capital_required_usd']:,.0f}\n" f"центр: {c['center']:.4f}\n\n" f"Уровни покупки:\n{levels_str}\n\n" f"Продажа: цена покупки × {1+c['step_pct']/100:.4f}\n\n" f"⚠️ Не ставь стоп на инвентарь. Вышла цена из диапазона — жди.\n" f"⚠️ Не уменьшай шаг. Ниже 0.6% комиссии съедают доход.\n\n" f"полный конфиг: {MIRROR}/grid/{g['symbol']}.json\n\n" f"
#AGENT_DATA\n{machine}\n#END
" ) def daily(perf): rows = perf.get("by_symbol", []) if not rows: return None lines = [] warn = [] for s in rows: apr = f"{s['apr_pct']:+.1f}%" if s.get("apr_pct") is not None else "—" fr = s.get("fill_ratio_pct") frs = f"{fr:.0f}%" if fr is not None else "—" lines.append(f"{s['symbol']:<9} филлов {s['fills']:>3} · " f"${s['pnl_usd']:>7.2f} · APR {apr:>7} · fill {frs}") if fr is not None and abs(fr - MODEL_FILL) / MODEL_FILL > 0.3: warn.append(f"{s['symbol']}: fill-ratio {fr:.0f}% против модельных {MODEL_FILL}%") body = ("📊 Грид-сводка за сутки\npaper\n\n
"
            + "\n".join(lines) + "
\n" + f"\nИтого: ${perf.get('total_pnl_usd', 0):+.2f}\n") if warn: body += ("\n⚠️ Расхождение с моделью:\n" + "\n".join("• " + w for w in warn) + "\n\nЭто сигнал разбираться, а не масштабировать.") return body HELP = ( "🟢 Грид-копитрейд Sovereign Arena\n\n" "Мы отдаём конфигурацию сетки, а не отдельные сделки. " "Ты ставишь такую же сетку у себя — дальше она работает автономно, " "без зависимости от нашей задержки и аптайма.\n\n" "Команды\n" "/grids — активные сетки\n" "/grid LINK — конфиг и уровни\n" "/stats — трек-рекорд\n" "/rules — правила и запреты\n" "/key SA-VIP-… — привязать месячный ключ\n\n" f"API: {MIRROR}/grid/configs.json\n" "paper-режим · not financial advice" ) RULES = ( "📐 Два правила\n\n" "1. Шаг ≥ 4× комиссий за круг.\n" "Комиссии 0.15% туда-обратно. При шаге 0.22% бирже уходит 68% дохода — " "так наш прошлый грид потерял $5 419. Минимум 0.6%, рабочее значение 0.8%.\n\n" "2. Никаких стопов на инвентаре.\n" "Вышла цена из диапазона — ждём. Цена возвращалась к уровню входа " "в 88% случаев, 80% из них за сутки. Принудительные закрытия стоили $6 571.\n\n" "🚫 Три запрета\n" "• стоп-лосс на инвентаре\n" "• закрытие «по смене режима»\n" "• шаг ниже 0.6%\n\n" "Это paper. Живого трека реальными деньгами нет." ) # ───────────────────────── режимы работы ───────────────────────── def broadcast(): if not CHAT: sys.exit("нужен TG_CHAT") snap = api("/api/grid/snapshot") if not snap: sys.exit("не смог получить данные ни с арены, ни с зеркала") n = 0 for g in snap.get("grids", []): r = tg("sendMessage", chat_id=CHAT, text=card(g), parse_mode="HTML", disable_web_page_preview="true") if r.get("ok"): n += 1 time.sleep(1) print(f"отправлено карточек: {n}") def send_daily(): if not CHAT: sys.exit("нужен TG_CHAT") perf = api("/api/grid/performance") body = daily(perf) if perf else None if not body: print("нет данных для сводки"); return r = tg("sendMessage", chat_id=CHAT, text=body, parse_mode="HTML") print("сводка отправлена" if r.get("ok") else f"ошибка: {r}") def serve(): print("слушаю команды…") offset = 0 while True: upd = tg("getUpdates", offset=offset, timeout=30) for u in upd.get("result", []): offset = u["update_id"] + 1 msg = u.get("message") or {} txt = (msg.get("text") or "").strip() chat = (msg.get("chat") or {}).get("id") if not txt or not chat: continue cmd = txt.split()[0].lower().split("@")[0] arg = txt.split()[1] if len(txt.split()) > 1 else "" if cmd in ("/start", "/help"): tg("sendMessage", chat_id=chat, text=HELP, parse_mode="HTML", disable_web_page_preview="true") elif cmd == "/rules": tg("sendMessage", chat_id=chat, text=RULES, parse_mode="HTML") elif cmd == "/grids": cf = api("/api/grid/configs") or {} gs = cf.get("grids", []) if not gs: tg("sendMessage", chat_id=chat, text="сеток сейчас нет") else: t = "🟢 Активные сетки\n\n
"
                    for g in gs:
                        t += (f"{g['symbol']:<9} шаг {g['step_pct']}% · "
                              f"{g['levels']} ур · ${g['capital_required_usd']:,.0f}\n")
                    t += "
\n/grid SYMBOL — уровни и конфиг" tg("sendMessage", chat_id=chat, text=t, parse_mode="HTML") elif cmd == "/grid": sym = (arg or "LINK").upper() if not sym.endswith("USDT"): sym += "USDT" snap = api("/api/grid/snapshot") or {} g = next((x for x in snap.get("grids", []) if x["symbol"] == sym), None) if not g: tg("sendMessage", chat_id=chat, text=f"нет сетки {sym}. /grids — список") else: tg("sendMessage", chat_id=chat, text=card(g), parse_mode="HTML", disable_web_page_preview="true") elif cmd == "/stats": perf = api("/api/grid/performance") b = daily(perf) if perf else None tg("sendMessage", chat_id=chat, text=b or "данных пока нет", parse_mode="HTML") elif cmd == "/key": try: import importlib.util as il sp = il.spec_from_file_location("ak", os.path.join(HERE, "access_keys.py")) ak = il.module_from_spec(sp); sp.loader.exec_module(ak) r = ak.validate(arg) if r.get("valid"): p = r["period"] tg("sendMessage", chat_id=chat, parse_mode="HTML", text=f"✅ ключ принят\nтариф {r['tier']}, " f"действует до конца {p[2:]}.20{p[:2]}") else: tg("sendMessage", chat_id=chat, text=f"❌ ключ не принят: {r.get('reason')}") except Exception as e: tg("sendMessage", chat_id=chat, text=f"проверка недоступна: {str(e)[:60]}") time.sleep(1) if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--broadcast", action="store_true", help="разослать карточки сеток") ap.add_argument("--daily", action="store_true", help="суточная сводка") ap.add_argument("--serve", action="store_true", help="отвечать на команды") ap.add_argument("--preview", action="store_true", help="показать сообщения без отправки") a = ap.parse_args() if a.preview: snap = api("/api/grid/snapshot") or {} for g in snap.get("grids", [])[:1]: print(card(g)) print("\n" + "=" * 60 + "\n") perf = api("/api/grid/performance") print(daily(perf) if perf else "нет данных") elif a.broadcast: broadcast() elif a.daily: send_daily() elif a.serve: serve() else: ap.print_help()