#!/usr/bin/env python3 """ PAPER-ГРИД v2 — по результатам грид-лаборатории 2026-07-22 ЧТО ИСПРАВЛЕНО ПРОТИВ v900 (который потерял $5 419): 1. ШАГ 0.8% вместо 0.22%. Старый шаг был втрое ниже порога выживания: комиссии 0.15% за круг съедали 68% дохода. Правило: шаг ≥ 4× комиссий. 2. НЕТ regime_flatten. НЕТ grid_breakout. Контрфакт показал: принудительные закрытия стоили $6 571, при том что цена возвращалась к уровню входа в 88% случаев (80% — за сутки). 3. ЛИМИТ ИНВЕНТАРЯ вместо аварийного закрытия. Вышли из диапазона — перестаём добавлять уровни и ЖДЁМ. Не режем. 4. СОСТОЯНИЕ ПЕРЕЖИВАЕТ РЕСТАРТ. За месяц наша инфра осиротила 9 514 позиций. Бот пишет состояние на диск/в БД и при старте поднимает его обратно. 5. МЕРИТ САМ СЕБЯ: считает фактический fill-ratio и сравнивает с моделью (80%). Расхождение >30% = модель врёт, это видно сразу, а не через месяц. Запуск (локально, состояние в SQLite): python3 paper_grid_bot.py --symbol LINKUSDT --step 0.8 --levels 20 Запуск на арене (состояние в Postgres, видно в Grafana): PGDSN="host=localhost dbname=sovereign user=... password=..." \ python3 paper_grid_bot.py --symbol LINKUSDT --pg Отчёт по накопленному: python3 paper_grid_bot.py --symbol LINKUSDT --report """ import argparse, json, os, sqlite3, sys, time, urllib.request from datetime import datetime, timezone HERE = os.path.dirname(os.path.abspath(__file__)) # ── параметры из лаборатории (walk-forward подтверждён на 48 наблюдениях) ── DEFAULTS = { "step_pct": 0.8, # ≥ 4× комиссий (0.15%) — порог выживания 0.6% "levels": 20, "unit_usd": 100.0, "spot_fee": 0.00075, "poll_sec": 20, "recenter_drift_pct": 25.0, # сетка ушла дальше — пересобрать (только если пусто) } MODEL_FILL_RATIO = 0.80 # что предсказала модель — с этим сверяемся # ─────────────────────────────── хранилище ─────────────────────────────── class Store: """SQLite по умолчанию; Postgres если задан PGDSN и есть psycopg2""" def __init__(self, symbol, use_pg=False): self.symbol = symbol self.pg = None if use_pg: try: import psycopg2 self.pg = psycopg2.connect(os.environ["PGDSN"]) self.pg.autocommit = True except Exception as e: print(f"[store] Postgres недоступен ({str(e)[:60]}), беру SQLite") self.db = None if not self.pg: self.db = sqlite3.connect(os.path.join(HERE, f"paper_grid_{symbol}.db")) self._init() def _x(self, sql, args=()): if self.pg: cur = self.pg.cursor(); cur.execute(sql.replace("?", "%s"), args); return cur cur = self.db.cursor(); cur.execute(sql, args); self.db.commit(); return cur def _init(self): t = f"paper_grid_{self.symbol.lower()}" self.t_pos, self.t_fill, self.t_meta = f"{t}_pos", f"{t}_fills", f"{t}_meta" self._x(f"""create table if not exists {self.t_pos}( level_idx integer primary key, entry_price double precision, opened_at bigint, hedged integer default 1)""") self._x(f"""create table if not exists {self.t_fill}( id integer primary key {'generated by default as identity' if self.pg else 'autoincrement'}, level_idx integer, entry_price double precision, exit_price double precision, pnl_usd double precision, opened_at bigint, closed_at bigint, close_reason text default 'grid_tp')""") self._x(f"""create table if not exists {self.t_meta}( k text primary key, v text)""") # позиции def open_positions(self): c = self._x(f"select level_idx, entry_price, opened_at from {self.t_pos}") return {r[0]: (float(r[1]), int(r[2])) for r in c.fetchall()} def add_position(self, idx, price, ts): self._x(f"insert into {self.t_pos}(level_idx, entry_price, opened_at) values(?,?,?)" + (" on conflict (level_idx) do nothing" if self.pg else ""), (idx, price, ts)) def close_position(self, idx, entry, exit_px, pnl, opened, closed): self._x(f"delete from {self.t_pos} where level_idx=?", (idx,)) self._x(f"""insert into {self.t_fill}(level_idx, entry_price, exit_price, pnl_usd, opened_at, closed_at) values(?,?,?,?,?,?)""", (idx, entry, exit_px, pnl, opened, closed)) # мета def get(self, k, default=None): c = self._x(f"select v from {self.t_meta} where k=?", (k,)) r = c.fetchone(); return json.loads(r[0]) if r else default def set(self, k, v): s = json.dumps(v) if self.pg: self._x(f"insert into {self.t_meta}(k,v) values(?,?) on conflict (k) do update set v=excluded.v", (k, s)) else: self._x(f"insert or replace into {self.t_meta}(k,v) values(?,?)", (k, s)) # статистика def stats(self): c = self._x(f"select count(*), coalesce(sum(pnl_usd),0), coalesce(min(closed_at),0) from {self.t_fill}") n, pnl, first = c.fetchone() return int(n), float(pnl), int(first or 0) # ─────────────────────────────── рынок ─────────────────────────────── # Источники вынесены в price_sources.py — тот же модуль использует # и copytrade_api, и лабораторные скрипты, чтобы цена везде была одна. # Внутри 10 бирж; с арены (США) живы 8, включая data-api.binance.vision — # публичное зеркало Binance, которое НЕ отдаёт 451 американским IP. # Это открывает лоу-капы, ради которых лаба и затевалась. try: from price_sources import last_candle, symbol_available, SOURCES except ImportError: sys.exit("нет price_sources.py рядом с ботом — " "curl -fsSL https://grid-src.vercel.app/price_sources.py -o price_sources.py") # ─────────────────────────────── бот ─────────────────────────────── def run(cfg): sym = cfg["symbol"] st = Store(sym, cfg["use_pg"]) step = cfg["step_pct"] / 100.0 unit = cfg["unit_usd"] center = st.get("center") if center is None: center = last_candle(sym)["c"] st.set("center", center) st.set("started_at", int(time.time() * 1000)) st.set("config", {k: v for k, v in cfg.items() if k != "use_pg"}) print(f"[init] новая сетка, центр {center:.4f}") else: pos = st.open_positions() print(f"[resume] ВОССТАНОВЛЕНО состояние: центр {center:.4f}, " f"открытых уровней {len(pos)} ← рестарт НЕ осиротил позиции") levels = [center * (1 - step * (i + 1)) for i in range(cfg["levels"])] st.set("levels", levels) touches = st.get("touches", 0) # сколько раз цена касалась свободного уровня fills_at_start, _, _ = st.stats() print(f"[cfg] {sym} шаг {cfg['step_pct']}% × {cfg['levels']} уровней × ${unit:.0f} " f"= резерв ${cfg['levels']*unit:,.0f}") print(f"[cfg] БЕЗ regime_flatten, БЕЗ grid_breakout — только лимит инвентаря") print(f"[cfg] диапазон сетки: {levels[-1]:.4f} … {center:.4f} " f"(глубина {cfg['step_pct']*cfg['levels']:.1f}%)") last_ts = st.get("last_ts", 0) while True: try: c = last_candle(sym) if c["ts"] <= last_ts: time.sleep(cfg["poll_sec"]); continue last_ts = c["ts"]; st.set("last_ts", last_ts) pos = st.open_positions() # --- покупки на уровнях (лимит инвентаря = число уровней, не режем!) for i, lvl in enumerate(levels): if i in pos or c["l"] > lvl: continue touches += 1; st.set("touches", touches) st.add_position(i, lvl, c["ts"]) print(f" ↓ BUY ур.{i:>2} @ {lvl:.4f} (инвентарь {len(pos)+1}/{cfg['levels']})") pos[i] = (lvl, c["ts"]) # --- продажи по грид-тейку for i, (entry, opened) in list(pos.items()): target = entry * (1 + step) if c["h"] >= target: gross = unit * step fees = unit * cfg["spot_fee"] + unit * (1 + step) * cfg["spot_fee"] pnl = gross - fees st.close_position(i, entry, target, pnl, opened, c["ts"]) held_h = (c["ts"] - opened) / 3600000 print(f" ↑ SELL ур.{i:>2} @ {target:.4f} +${pnl:.2f} (держали {held_h:.1f}ч)") # --- диагностика раз в час if c["ts"] % 3600000 < 60000: n, pnl, first = st.stats() days = (c["ts"] - st.get("started_at", c["ts"])) / 86400000 or 1e-9 cap = cfg["levels"] * unit fr = (n / touches * 100) if touches else 0 print(f"[stat] филлов {n} | PnL ${pnl:.2f} | {days:.1f}д | " f"APR {pnl/cap*100*365/days:+.1f}% | " f"fill-ratio {fr:.0f}% (модель {MODEL_FILL_RATIO*100:.0f}%) | " f"инвентарь {len(st.open_positions())}/{cfg['levels']}") # --- пересборка сетки, только если инвентарь пуст if (not st.open_positions() and abs(c["c"] - center) / center * 100 > cfg["recenter_drift_pct"]): center = c["c"]; st.set("center", center) levels = [center * (1 - step * (i + 1)) for i in range(cfg["levels"])] st.set("levels", levels) print(f"[recenter] сетка пересобрана вокруг {center:.4f} (инвентарь был пуст)") time.sleep(cfg["poll_sec"]) except KeyboardInterrupt: print("\nостановлено, состояние сохранено"); return except Exception as e: print(f"[err] {type(e).__name__}: {str(e)[:90]} — повтор через 30с") time.sleep(30) def report(cfg): st = Store(cfg["symbol"], cfg["use_pg"]) n, pnl, first = st.stats() started = st.get("started_at", 0) touches = st.get("touches", 0) cap = cfg["levels"] * cfg["unit_usd"] days = (time.time() * 1000 - started) / 86400000 if started else 0 print(f"=== PAPER-ГРИД {cfg['symbol']} ===") if not started: print("не запускался"); return print(f"работает : {days:.1f} дней") print(f"филлов : {n}") print(f"PnL : ${pnl:.2f}") print(f"APR : {pnl/cap*100*365/days:+.1f}%" if days > 0.1 else "APR: рано") print(f"открыто позиций : {len(st.open_positions())}/{cfg['levels']}") if touches: fr = n / touches * 100 print(f"fill-ratio факт : {fr:.0f}% модель: {MODEL_FILL_RATIO*100:.0f}%") d = abs(fr - MODEL_FILL_RATIO*100) / (MODEL_FILL_RATIO*100) * 100 print(f"расхождение : {d:.0f}% " + ("✅ модель подтверждается" if d < 30 else "⚠️ МОДЕЛЬ ВРЁТ — разбираться")) def main(): p = argparse.ArgumentParser() p.add_argument("--symbol", default="LINKUSDT") p.add_argument("--step", type=float, default=DEFAULTS["step_pct"]) p.add_argument("--levels", type=int, default=DEFAULTS["levels"]) p.add_argument("--unit", type=float, default=DEFAULTS["unit_usd"]) p.add_argument("--pg", action="store_true", help="писать в Postgres арены") p.add_argument("--report", action="store_true") a = p.parse_args() cfg = dict(DEFAULTS, symbol=a.symbol, step_pct=a.step, levels=a.levels, unit_usd=a.unit, use_pg=a.pg) if a.step * 100 < 4 * 0.15 * 100 / 100 * 100: pass if a.step < 0.6: print(f"⚠️ ШАГ {a.step}% НИЖЕ ПОРОГА ВЫЖИВАНИЯ (0.6% = 4× комиссии).") print(" Именно так v900 потерял деньги. Продолжаю, но результат предсказуем.") report(cfg) if a.report else run(cfg) if __name__ == "__main__": main()