#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ СЕРВИС АРБИТРАЖА — скан → алерты в Telegram → бумажная книга → публичная витрина. Запуск: python3 arb_service.py --loop 300 Разово: python3 arb_service.py --once --no-publish ЧТО ИМЕННО ЗДЕСЬ ПРОВЕРЯЕТСЯ, А НЕ ДЕКЛАРИРУЕТСЯ. Мы не знаем заранее, окупается ли фандинг-арбитраж после комиссий. Поэтому КАЖДАЯ найденная возможность открывается в бумажной книге и помечается robust/fragile — по тому, выживает ли разрыв без двух крайних бирж. Через неделю данные скажут сами: если у fragile-корзины итог около нуля или в минус, значит метрика живучести работает и алерты надо слать только по robust. Если разницы нет — метрику выкинуть. Это проверяемая гипотеза, а не украшение интерфейса. В ЧАТ УХОДИТ НЕ ВСЁ. Алерт только если: разрыв держится ≥3 скана подряд, живучесть выше порога, оборот на обеих ногах достаточный, и по этой паре не было алерта последние 6 часов. Сканер, который шлёт всё подряд, читать перестают на второй день. СЕКРЕТЫ берутся из окружения (ARB_TG_TOKEN, ARB_TG_CHAT, VERCEL_TOKEN) либо из соседнего файла arb_secrets.json. В коде их нет. """ import argparse, base64, hashlib, json, os, sys, time, urllib.request sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import arb_sources as S import arb_engine as E HERE = os.path.dirname(os.path.abspath(__file__)) DB = os.path.join(HERE, "arb.db") _SEC = {} try: with open(os.path.join(HERE, "arb_secrets.json"), encoding="utf-8") as f: _SEC = json.load(f) except Exception: pass def _from_sibling(module, attr): """ Достаём ключ из уже стоящего рядом модуля, а не из копии в этом файле. Смысл: ключ живёт в ОДНОМ месте. Ротация — правка одного файла, и нигде нет второго экземпляра, о котором потом забудут. """ try: return getattr(__import__(module), attr, "") or "" except Exception: return "" def secret(name, default=""): return os.environ.get(name) or _SEC.get(name) or default TG_TOKEN = secret("ARB_TG_TOKEN") or _from_sibling("tg_bot", "TOKEN") TG_CHAT = secret("ARB_TG_CHAT", "932299051") VERCEL_TOKEN = secret("VERCEL_TOKEN") or _from_sibling("push_to_vercel", "VERCEL_TOKEN") TEAM = "team_oVflHvk6poDyQRHW88LCTcg8" PROJECT = "arb-radar" ALIAS = "sovereign-arb-radar.vercel.app" # arb-radar.vercel.app занят чужим проектом # ── пороги алертов ── ALERT_STREAK = 3 # сколько сканов подряд должен держаться разрыв ALERT_MIN_APR = 0.40 # 40% годовых — ниже не будим человека ALERT_MIN_ROBUST = 0.25 # эдж должен выживать без двух крайних бирж ALERT_MIN_VOL = 1_000_000 ALERT_MAX_BE_H = 48 # окупаемость входа-выхода не дольше двух суток ALERT_COOLDOWN_H = 6 PAPER_NOTIONAL = 1000.0 PAPER_MIN_APR = 0.20 # в бумагу берём ШИРЕ, чем в алерты — ради статистики # ── БЮДЖЕТ ПУБЛИКАЦИЙ ── # Скан и публикация РАЗВЯЗАНЫ. У Vercel free 100 деплоев в сутки НА АККАУНТ, # а не на проект. Публикация на каждом скане (раз в 5 мин) = 288 деплоев в # сутки: этот сервис в одиночку выжирает лимит и блокирует ВСЕ остальные # деплои аккаунта, включая сайт. Проверено больно 29.07 — деплой сайта # отбился 402 payment_required. # Поэтому: не чаще PUBLISH_EVERY_S, не больше PUBLISH_MAX_DAY за сутки и # только если содержимое реально изменилось. PUBLISH_EVERY_S = 1800 # 30 минут → максимум 48 в сутки PUBLISH_MAX_DAY = 40 # жёсткий потолок, оставляем место остальным def log(*a): print(time.strftime("[%H:%M:%S]"), *a, flush=True) def tg(text): if not TG_TOKEN or not TG_CHAT: return try: data = json.dumps({"chat_id": TG_CHAT, "text": text, "parse_mode": "HTML", "disable_web_page_preview": True}).encode() req = urllib.request.Request( f"https://api.telegram.org/bot{TG_TOKEN}/sendMessage", data=data, headers={"Content-Type": "application/json"}) urllib.request.urlopen(req, timeout=20).read() except Exception as e: log(f" ⚠️ telegram: {type(e).__name__}") def fmt_alert(o): if o["kind"] == "spread": return (f"⚡ СПРЕД {o['symbol']}\n" f"купить {o['buy_venue']} по {o['buy_px']:.6g}\n" f"продать {o['sell_venue']} по {o['sell_px']:.6g}\n" f"грязными {o['gross']*100:.2f}%, комиссии {o['cost_round']*100:.2f}% → " f"чистыми {o['net']*100:.2f}%\n" f"оборот ${o['vol_usd']:,.0f} · держится {o.get('age_h',0):.1f}ч\n" f"⚠️ деньги нужны ЗАРАНЕЕ на обеих биржах — на перевод спреда не хватит") head = "🔁 КЭРРИ" if o["kind"] == "carry" else "⚖️ ФАНДИНГ" lines = [f"{head} {o['symbol']}", f"лонг: {o['long']}", f"шорт: {o['short']}", f"доход {o['apr']*100:.0f}% годовых"] if o["kind"] == "funding": lines.append(f"без двух крайних бирж: {o.get('apr_robust',0)*100:.0f}% " f"(бирж в расчёте {o.get('venues_n',0)})") lines.append(f"ставки: {o['long_rate']*100:+.4f}%/{o['long_interval_h']:.0f}ч " f"против {o['short_rate']*100:+.4f}%/{o['short_interval_h']:.0f}ч") else: lines.append(f"ставка {o['rate_h']*o['interval_h']*100:+.4f}%/{o['interval_h']:.0f}ч, " f"базис {o['basis']*100:+.2f}%") lines += [f"вход+выход стоят {o['cost_round']*100:.2f}% → " f"окупается за {o['breakeven_h']:.0f}ч", f"оборот ${o['vol_usd']:,.0f} · держится {o.get('age_h',0):.1f}ч " f"({o.get('streak',0)} сканов)", "⚠️ дельта-нейтрально, но нужен залог на обеих ногах; " "ставка меняется каждый интервал"] return "\n".join(lines) def qualifies_for_alert(o): if o.get("streak", 0) < ALERT_STREAK: return False if o["vol_usd"] < ALERT_MIN_VOL: return False if o["kind"] == "spread": return o["net"] >= 0.003 if o["apr"] < ALERT_MIN_APR or o["breakeven_h"] > ALERT_MAX_BE_H: return False if o["kind"] == "funding" and o.get("apr_robust", 0) < ALERT_MIN_ROBUST: return False return True def deploy(files): def api(url, body=None, method="GET"): req = urllib.request.Request( url, data=json.dumps(body).encode() if body else None, method=method, headers={"Authorization": "Bearer " + VERCEL_TOKEN, "Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=120) as r: return json.load(r) fl = [{"file": k, "data": base64.b64encode(v.encode()).decode(), "encoding": "base64"} for k, v in files.items()] d = api(f"https://api.vercel.com/v13/deployments?teamId={TEAM}&forceNew=1", {"name": PROJECT, "files": fl, "target": "production", "projectSettings": {"framework": None, "buildCommand": "", "outputDirectory": ".", "installCommand": ""}}, "POST") did = d["id"] for _ in range(60): st = api(f"https://api.vercel.com/v13/deployments/{did}?teamId={TEAM}")["readyState"] if st == "READY": break if st in ("ERROR", "CANCELED"): raise RuntimeError(f"деплой упал: {st}") time.sleep(3) api(f"https://api.vercel.com/v2/deployments/{did}/aliases?teamId={TEAM}", {"alias": ALIAS}, "POST") return did def _pub_state(conn): conn.execute("CREATE TABLE IF NOT EXISTS publish_log(ts INTEGER, day TEXT, hash TEXT)") conn.commit() row = conn.execute("SELECT ts,hash FROM publish_log ORDER BY ts DESC LIMIT 1").fetchone() day = time.strftime("%Y-%m-%d", time.gmtime()) n = conn.execute("SELECT COUNT(*) FROM publish_log WHERE day=?", (day,)).fetchone()[0] return (row[0] if row else 0), (row[1] if row else ""), n, day def may_publish(conn, page, now_ms): """ Три ограничителя: частота, суточный бюджет и изменение содержимого. Лимит Vercel — на ВЕСЬ аккаунт, поэтому расточительность здесь ломает деплой сайта и зеркала, а не только этой витрины. """ last_ts, last_hash, n_today, _ = _pub_state(conn) h = hashlib.sha256(page.encode()).hexdigest()[:16] if n_today >= PUBLISH_MAX_DAY: return False, f"суточный бюджет исчерпан ({n_today}/{PUBLISH_MAX_DAY})" if now_ms - last_ts < PUBLISH_EVERY_S * 1000: left = int((PUBLISH_EVERY_S * 1000 - (now_ms - last_ts)) / 60000) return False, f"рано, следующая через ~{left} мин" if h == last_hash: return False, "содержимое не изменилось" return True, "" def mark_published(conn, page, now_ms): h = hashlib.sha256(page.encode()).hexdigest()[:16] conn.execute("INSERT INTO publish_log(ts,day,hash) VALUES(?,?,?)", (now_ms, time.strftime("%Y-%m-%d", time.gmtime()), h)) conn.commit() def build_page(snap): o = snap["opportunities"] b = snap["book"] def rows(kind, n=15): sel = [x for x in o if x["kind"] == kind][:n] if not sel: return "сейчас ничего выше порога" out = [] for x in sel: if kind == "spread": metric = f"{x['net']*100:.2f}%" extra = f"{x['gross']*100:.2f}%" be = "—" else: metric = f"{x['apr']*100:.0f}%" extra = f"{x.get('apr_robust',0)*100:.0f}%" if kind == "funding" else "—" be = f"{x['breakeven_h']:.0f}ч" cls = "good" if (kind != "funding" or x.get("apr_robust", 0) >= 0.25) else "weak" out.append( f"{x['symbol']}{x['long']}" f"{x['short']}{metric}{extra}" f"{be}${x['vol_usd']:,.0f}") return "".join(out) bk = (f"закрыто {b['closed']} · прибыльных {b['winrate_pct']:.1f}% · " f"итог ${b['closed_pnl']:+,.2f} · открыто {b['open']} (${b['open_pnl']:+,.2f})") cmp_html = "" if b.get("robust_vs_fragile"): r = b["robust_vs_fragile"] cmp_html = ("

Проверка метрики живучести

" "" "" + "".join(f"" f"" f"" for k, v in r.items()) + "
корзиназакрытоитогприбыльных
{k}{v['n']}${v['pnl']:+,.2f}{v['wr_pct']:.1f}%
") return f""" Арбитражный радар — фандинг и спред

Арбитражный радар

обновлено {snap['updated']} · биржи: {', '.join(snap['venues'])} · вселенная — {snap['universe_n']} пар, торгуемых на Binance
Это не сигналы «жми и богатей». Каждая строка — расхождение, которое ещё нужно исполнить руками, с залогом на двух площадках. Годовые посчитаны от текущей ставки, а она меняется каждый интервал. Колонка «окупаемость» честнее годовых: столько часов позицию надо продержать, чтобы отбить одни лишь комиссии входа и выхода.

Фандинг: перп против перпа

дельта-нейтрально. Колонка «без крайних» — доход, если выкинуть две крайние биржи. Если там почти ноль, весь разрыв держится на одной площадке и это, скорее, особенность её формулы, а не рынок.
{rows('funding')}
монеталонгшортгодовых без крайнихокупаемостьоборот

Кэрри: спот и перп на одной бирже

купить монету, продать против неё перп, собирать фандинг. Переводов между биржами нет — самый исполнимый вариант.
{rows('carry')}
монетакупитьпродатьгодовых —окупаемостьоборот

Спред: спот дешевле там, дороже тут

показываем только пары с реальным оборотом на ОБЕИХ биржах. Без этого фильтра список занимают мёртвые монеты с зависшей котировкой.
{rows('spread')}
монетакупитьпродатьчистыми грязными—оборот

Бумажная книга

{bk}
{cmp_html}
Как это считается. Ставка фандинга приводится к годовым через её собственный интервал — у альтов он бывает 1 и 4 часа, а не только 8, и без нормализации доход завышается в разы. Комиссии заложены тейкерские, без VIP-скидок, плюс {E.SLIP_PERP*100:.2f}% проскальзывания на ногу. Цена каждой монеты сверяется с медианой по всем биржам: расхождение больше 3% считается сломанной котировкой и выбрасывается.
Данные: {', '.join(snap['venues'])}. Binance и Bybit не отвечают с нашего узла (451 и 403), поэтому их ставки в расчёт не входят — список монет берётся с Binance, а фандинг считается по доступным площадкам. Исходники: grid-src · машинный срез: arb.json
""" def once(publish=True, alert=True): now = int(time.time() * 1000) log("── скан ──") perps, rep, perr = S.load_perps(log=lambda *a: log(" ", *a)) spots, serr = S.load_spots(log=lambda *a: log(" ", *a)) uni = S.binance_universe(log=lambda *a: log(" ", *a)) if not perps: log("нет ни одной перп-биржи — пропуск") return opps = E.scan(perps, spots, universe=uni, min_apr=PAPER_MIN_APR) conn = E.db(DB) E.track_persistence(conn, opps, now) cur = conn.cursor() for o in opps: cur.execute("INSERT INTO opps(ts,kind,symbol,apr,net,breakeven_h,long_leg," "short_leg,price,vol_usd) VALUES(?,?,?,?,?,?,?,?,?,?)", (now, o["kind"], o["symbol"], o.get("apr", 0), o.get("net", 0), o.get("breakeven_h", 0), o["long"], o["short"], o.get("price", 0), o.get("vol_usd", 0))) conn.commit() # ── бумага: закрыть отжившее, открыть новое ── for pid, sym, kind, held in E.settle_paper(conn, perps, now): log(f" ✖ закрыта бумага #{pid} {sym} ({kind}), держали {held:.1f}ч") opened = 0 for o in opps: if o.get("streak", 0) < 2: continue if cur.execute("SELECT 1 FROM positions WHERE state='open' AND symbol=? " "AND long_leg=? AND short_leg=?", (o["symbol"], o["long"], o["short"])).fetchone(): continue o["_tag"] = ("robust" if (o["kind"] != "funding" or o.get("apr_robust", 0) >= 0.25) else "fragile") E.open_paper(conn, o, now, PAPER_NOTIONAL) opened += 1 if opened: log(f" ✚ открыто бумажных позиций: {opened}") # ── алерты ── sent = 0 if alert: for o in opps: if not qualifies_for_alert(o): continue k = E.key_of(o) row = cur.execute("SELECT alerted_ms,best_apr FROM seen WHERE key=?", (k,)).fetchone() metric = o["net"] if o["kind"] == "spread" else o["apr"] if row and row[0] and (now - row[0]) < ALERT_COOLDOWN_H * 3600000: if metric < row[1] * 1.5: continue tg(fmt_alert(o)) cur.execute("UPDATE seen SET alerted_ms=? WHERE key=?", (now, k)) sent += 1 if sent >= 6: break conn.commit() log(f" возможностей {len(opps)} · алертов {sent}") # ── витрина ── b = E.book_summary(conn) b["robust_vs_fragile"] = {} for tag in ("robust", "fragile"): r = cur.execute( "SELECT COUNT(*), COALESCE(SUM(funding_collected+spread_pnl-entry_cost-exit_cost),0)," " SUM(CASE WHEN funding_collected+spread_pnl-entry_cost-exit_cost>0 THEN 1 ELSE 0 END)" " FROM positions WHERE state='closed' AND meta LIKE ?", ('%"' + tag + '"%',)).fetchone() if r[0]: b["robust_vs_fragile"][tag] = {"n": r[0], "pnl": r[1], "wr_pct": (r[2] or 0) / r[0] * 100} snap = {"updated": time.strftime("%Y-%m-%d %H:%M UTC", time.gmtime()), "venues": sorted(perps.keys()), "universe_n": len(uni), "opportunities": opps[:80], "book": b, "unit_calibration": rep, "errors": {**perr, **serr}} with open(os.path.join(HERE, "arb_snapshot.json"), "w", encoding="utf-8") as f: json.dump(snap, f, ensure_ascii=False, default=str) if publish and VERCEL_TOKEN: page = build_page(snap) ok, why = may_publish(conn, page, now) if not ok: log(f" ⏸ публикация пропущена: {why}") else: try: did = deploy({ "index.html": page, "arb.json": json.dumps(snap, ensure_ascii=False, default=str), "vercel.json": json.dumps({"headers": [{"source": "/(.*)", "headers": [ {"key": "Access-Control-Allow-Origin", "value": "*"}, {"key": "Cache-Control", "value": "no-store"}]}]})}) mark_published(conn, page, now) log(f" 🌐 https://{ALIAS} ({did[:18]}…)") except Exception as e: log(f" ⚠️ публикация: {type(e).__name__}: {e}") conn.close() if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--loop", type=int, default=0, help="секунд между сканами") ap.add_argument("--once", action="store_true") ap.add_argument("--no-publish", action="store_true") ap.add_argument("--no-alert", action="store_true") a = ap.parse_args() if a.loop: while True: try: once(publish=not a.no_publish, alert=not a.no_alert) except Exception as e: log(f"❌ {type(e).__name__}: {e}") time.sleep(a.loop) else: once(publish=not a.no_publish, alert=not a.no_alert)