#!/usr/bin/env python3 """ ПУШ ГРИД-КАРТОЧЕК В INNER CIRCLE Токен и chat_id НЕ хранятся здесь — берутся из уже работающего /opt/arena/vip_brief_push.py. Секрет не переносится и не светится. Что шлёт: --cards карточки сеток (конфиг + уровни). Это и есть «сигнал» для грида: подписчик ставит такую же сетку у себя, дальше она автономна. --daily суточную сводку: филлы, PnL, fill-ratio против модельных 80%. Каждое сообщение читаемо человеком СВЕРХУ и машиной СНИЗУ (блок #AGENT_DATA), поэтому один канал обслуживает и людей, и агентов. Запуск: python3 tg_grid_push.py --cards python3 tg_grid_push.py --daily python3 tg_grid_push.py --cards --dry # показать, не отправляя """ import argparse, json, os, re, sys, time, urllib.parse, urllib.request HERE = os.path.dirname(os.path.abspath(__file__)) SOURCE = os.environ.get("VIP_PUSH_SRC", "/opt/arena/vip_brief_push.py") API = os.environ.get("GRID_API", "http://localhost:8101") MIRROR = "https://grid-mirror.vercel.app" CABINET = MIRROR + "/cabinet.html" MODEL_FILL = 80 def creds(): """достать токен и chat_id из работающего vip_brief_push.py""" tok = os.environ.get("IC_TOKEN") chat = os.environ.get("IC_CHAT") if tok and chat: return tok, chat try: src = open(SOURCE, encoding="utf-8", errors="ignore").read() except Exception as e: sys.exit(f"не смог прочитать {SOURCE}: {e}\n" f"либо задай IC_TOKEN и IC_CHAT переменными окружения") t = re.search(r"\b(\d{9,10}:[A-Za-z0-9_-]{30,})", src) c = re.search(r"(-100\d{9,})", src) if not t: sys.exit(f"в {SOURCE} не нашёл токен бота") if not c: sys.exit(f"в {SOURCE} не нашёл chat_id канала") return t.group(1), c.group(1) def tg(method, token, **params): url = f"https://api.telegram.org/bot{token}/{method}" data = urllib.parse.urlencode(params).encode() with urllib.request.urlopen(urllib.request.Request(url, data=data), timeout=25) as r: return json.load(r) def api(path): """арена, при отказе — зеркало""" try: with urllib.request.urlopen(API + path, timeout=20) as r: return json.load(r) except Exception: pass try: m = MIRROR + path.replace("/api/grid", "/grid").rstrip("/") + ".json" with urllib.request.urlopen(m.replace(".json.json", ".json"), timeout=20) as r: return json.load(r) except Exception: return None def card(g): c, lv = g["config"], g.get("grid_levels") or [] 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=(",", ":")) head = ", ".join(f"{x:.4f}" for x in lv[:6]) + (" …" if len(lv) > 6 else "") per_fill = c["unit_usd"] * c["step_pct"] / 100 * 0.81 return (f"🟢 СЕТКА {g['symbol']} paper\n\n" f"Копируется настройка сетки, а не отдельные сделки: ставишь один раз — " f"дальше работает сама, без зависимости от нашей задержки.\n\n" f"шаг {c['step_pct']}% · заявок {c['levels']} × ${c['unit_usd']:.0f}\n" f"нужно капитала: ${c['capital_required_usd']:,.0f}\n" f"с одного срабатывания: ~${per_fill:.2f}\n\n" f"Цены покупки:\n{head}\n" f"Продажа: цена покупки × {1+c['step_pct']/100:.4f}\n\n" f"⛔️ Не ставь стоп на купленное. Ушла цена ниже — жди. " f"Возвращалась в 88 случаях из 100, 8 из 10 — в тот же день.\n" f"⛔️ Не уменьшай шаг. Ниже 0.6% комиссии съедают больше половины дохода.\n\n" f"👉 Все цены и кнопка «скопировать»: {CABINET}\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") lines.append(f"{s['symbol']:<9} сделок {s['fills']:>3} · ${s['pnl_usd']:>7.2f} · " f"{apr:>7} · исполн {f'{fr:.0f}%' if fr is not None else '—'}") if fr is not None and abs(fr - MODEL_FILL) / MODEL_FILL > 0.3: warn.append(f"{s['symbol']}: исполнение {fr:.0f}% против ожидаемых {MODEL_FILL}%") txt = ("📊 Грид — сводка за сутки paper\n\n
" + "\n".join(lines) +
           f"
\n\nИтого: ${perf.get('total_pnl_usd',0):+.2f}\n") if warn: txt += ("\n⚠️ Расходится с моделью:\n" + "\n".join("• " + w for w in warn) + "\n\nЭто повод разбираться, а не масштабировать.") txt += f"\n\n👉 {CABINET}" return txt def main(): ap = argparse.ArgumentParser() ap.add_argument("--cards", action="store_true") ap.add_argument("--daily", action="store_true") ap.add_argument("--dry", action="store_true", help="показать, не отправляя") a = ap.parse_args() if not (a.cards or a.daily): ap.print_help(); return msgs = [] if a.cards: snap = api("/api/grid/snapshot") if not snap: sys.exit("нет данных ни с арены, ни с зеркала") msgs = [card(g) for g in snap.get("grids", [])] if a.daily: perf = api("/api/grid/performance") d = daily(perf) if perf else None if d: msgs.append(d) if not msgs: print("нечего отправлять"); return if a.dry: for m in msgs: print(m); print("\n" + "=" * 60 + "\n") print(f"(режим --dry, {len(msgs)} сообщений НЕ отправлено)") return token, chat = creds() print(f"канал {chat}, бот …{token[-6:]}") ok = 0 for m in msgs: r = tg("sendMessage", token, chat_id=chat, text=m, parse_mode="HTML", disable_web_page_preview="true") if r.get("ok"): ok += 1 print(f" отправлено, message_id={r['result']['message_id']}") else: print(" ошибка:", str(r)[:120]) time.sleep(1.2) print(f"итого: {ok}/{len(msgs)}") if __name__ == "__main__": main()