research(5m): прогон 14 методов на 250 парах, сверка движка, издержки, графики

ЧТО: code/ (стенд и методы), results/ (сводки), charts/ (3 графика), reports/ (издержки, плацебо, по годам), docs/REJECTED.md.

ПОЧЕМУ: цель — найти кандидатов на 5м или доказать с цифрами их отсутствие.

КАК ПРОВЕРЕНО: движок сверен с прод-движком TON-бота 1:1; плацебо и рандом-контроль в отчёте; бутстрап по дням.

code/fetch_5m.py

@@ -0,0 +1,256 @@
+#!/usr/bin/env python3
+"""Выгрузка 5-минутных свечей Bybit USDT-перпов (публичный REST, без ключей).
+
+Формат выхода: data/5m/<SYMBOL>_5m.csv — БЕЗ шапки, 6 полей: ts_ms,open,high,low,close,volume,
+время по возрастанию (по контракту гейта screener5m_gate_f1.sh).
+
+Особенности Bybit: максимум 1000 баров за запрос; внутри диапазона [start,end] отдаются
+ПОСЛЕДНИЕ limit баров → пагинация назад через end = oldest_ts - 1. retCode 10016 = нет данных, пропускаем.
+
+Запуск (из /opt/data/work/screener5m):
+  .venv/bin/python tools/fetch_5m.py --pairs 150 --start 2024-10-01
+  .venv/bin/python tools/fetch_5m.py --pairs 2 --start 2026-09-20 --out /tmp/smoke   # дымовой прогон
+"""
+from __future__ import annotations
+
+import argparse
+import concurrent.futures as cf
+import datetime as dt
+import json
+import os
+import sys
+import threading
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+
+BASE = "https://api.bybit.com"
+STEP_MS = 300_000  # 5 минут
+LIMIT = 1000
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+LOG_LOCK = threading.Lock()
+REQ_LOCK = threading.Lock()
+LAST_REQ = [0.0]
+MIN_INTERVAL = 0.125  # ~8 запросов/с суммарно по всем потокам
+LOG_PATH = None
+# токенизированные TradFi-инструменты в linear-категории — не крипта, исключаем из вселенной
+TRADFI = {"XAU", "XAG", "XPT", "XPD", "CL", "BZ", "WTI", "BRENT", "SOXL", "SQQQ", "SPY", "QQQ",
+          "AAPL", "TSLA", "NVDA", "MSFT", "GOOGL", "GOOG", "META", "AMZN", "COIN", "MSTR", "TLT",
+          "GLD", "USO", "EUR", "GBP", "JPY", "CRCL", "HOOD", "PLTR", "NFLX", "AMD", "INTC"}
+
+
+def log(msg: str) -> None:
+    line = f"[{dt.datetime.now().strftime('%H:%M:%S')}] {msg}"
+    with LOG_LOCK:
+        print(line, flush=True)
+        if LOG_PATH:
+            with open(LOG_PATH, "a", encoding="utf-8") as fh:
+                fh.write(line + "\n")
+
+
+def rate_gate() -> None:
+    with REQ_LOCK:
+        wait = MIN_INTERVAL - (time.time() - LAST_REQ[0])
+        if wait > 0:
+            time.sleep(wait)
+        LAST_REQ[0] = time.time()
+
+
+def api(path: str, params: dict, tries: int = 4) -> dict:
+    url = BASE + path + "?" + urllib.parse.urlencode(params)
+    last = None
+    for i in range(tries):
+        rate_gate()
+        try:
+            req = urllib.request.Request(url, headers={"User-Agent": "screener5m/1.0"})
+            with urllib.request.urlopen(req, timeout=30) as r:
+                return json.loads(r.read().decode())
+        except Exception as e:  # noqa: BLE001
+            last = f"{type(e).__name__}: {e}"
+            time.sleep(1.0 + 2.0 * i)
+    return {"retCode": -1, "retMsg": last or "unknown", "result": {}}
+
+
+def universe(min_turnover: float, n_pairs: int) -> list[tuple[str, float]]:
+    tick = api("/v5/market/tickers", {"category": "linear"}).get("result", {}).get("list", [])
+    info = api("/v5/market/instruments-info", {"category": "linear", "limit": 1000}).get("result", {}).get("list", [])
+    tradable = {i["symbol"] for i in info if i.get("status") == "Trading"}
+    rows = []
+    for t in tick:
+        s = t.get("symbol", "")
+        if not s.endswith("USDT") or s not in tradable:
+            continue
+        base = s[:-4]
+        if base in TRADFI:
+            continue
+        try:
+            to = float(t.get("turnover24h") or 0)
+        except ValueError:
+            continue
+        if to >= min_turnover:
+            rows.append((s, to))
+    rows.sort(key=lambda x: -x[1])
+    log(f"вселенная: торгуемых USDT-перпов {len(tradable)}, после фильтра оборота >= ${min_turnover/1e6:.0f}M — {len(rows)}; беру {min(n_pairs, len(rows))}")
+    return rows[:n_pairs]
+
+
+def fetch_symbol(sym: str, start_ms: int, out_dir: str) -> dict:
+    path = os.path.join(out_dir, f"{sym}_5m.csv")
+    rows: dict[int, str] = {}
+    first_existing = last_existing = None
+    if os.path.isfile(path) and os.path.getsize(path) > 0:
+        with open(path) as fh:
+            lines = [l for l in fh if l.strip()]
+        if lines:
+            first_existing = int(lines[0].split(",")[0])
+            last_existing = int(lines[-1].split(",")[0])
+            for l in lines:
+                rows[int(l.split(",")[0])] = l
+    now_ms = int(time.time() * 1000)
+    end = (first_existing - 1) if first_existing else now_ms
+    skipped = 0
+    pages = 0
+    heal = 0
+    soft_fail = 0
+    while end > start_ms:
+        r = api("/v5/market/kline", {"category": "linear", "symbol": sym, "interval": "5",
+                                     "start": start_ms, "end": end, "limit": LIMIT})
+        code = r.get("retCode")
+        if code == 10016 or code != 0:
+            # У Bybit это и «нет данных», и временный отказ (лимит): сначала повторяем ТО ЖЕ окно
+            # с растущей паузой, и только потом сдвигаемся назад. Иначе пара обрывается на 1-2 страницах.
+            soft_fail += 1
+            if soft_fail <= 4:
+                log(f"  {sym}: retCode={code} на end={end}, повтор окна ({soft_fail}/4) через {2*soft_fail}s")
+                time.sleep(2.0 * soft_fail)
+                continue
+            heal += 1
+            if heal > 8:
+                log(f"  {sym}: retCode={code} восемь раз подряд — прекращаю пагинацию на {pages} страницах")
+                break
+            log(f"  {sym}: retCode={code} держится, сдвигаю окно на 1 страницу назад (перескок {heal}/8)")
+            end = end - LIMIT * STEP_MS
+            soft_fail = 0
+            time.sleep(2.0)
+            continue
+        lst = r.get("result", {}).get("list", []) or []
+        if not lst:
+            soft_fail += 1
+            if soft_fail <= 4:
+                log(f"  {sym}: пустой ответ на end={end}, повтор ({soft_fail}/4)")
+                time.sleep(2.0 * soft_fail)
+                continue
+            log(f"  {sym}: пустой ответ 5 раз — прекращаю пагинацию на {pages} страницах")
+            break
+        soft_fail = 0
+        heal = 0
+        pages += 1
+        oldest = int(lst[-1][0])
+        for it in lst:
+            ts = int(it[0])
+            if ts >= start_ms:
+                rows[ts] = f"{ts},{it[1]},{it[2]},{it[3]},{it[4]},{it[5]}"
+        if first_existing is None:
+            first_existing = oldest
+        if pages % 25 == 0:
+            log(f"  {sym}: страниц {pages}, баров {len(rows)}, дошли до "
+                f"{dt.datetime.fromtimestamp(oldest/1000, dt.timezone.utc).strftime('%Y-%m-%d')}")
+        if oldest <= start_ms:
+            break
+        end = oldest - 1
+    ordered = [rows[k] for k in sorted(rows)]
+    if ordered:
+        with open(path, "w") as fh:
+            fh.write("\n".join(ordered) + "\n")
+    return {"symbol": sym, "bars": len(ordered),
+            "first_ts": (min(rows) if rows else None), "last_ts": (max(rows) if rows else None),
+            "skipped_10016": skipped}
+
+
+def main() -> int:
+    global LOG_PATH
+    p = argparse.ArgumentParser()
+    p.add_argument("--pairs", type=int, default=150)
+    p.add_argument("--start", default="2024-10-01")
+    p.add_argument("--min-turnover", type=float, default=5e6)
+    p.add_argument("--out", default=None)
+    p.add_argument("--workers", type=int, default=5)
+    p.add_argument("--interval", type=float, default=None)
+    args = p.parse_args()
+    global MIN_INTERVAL
+    if args.interval:
+        MIN_INTERVAL = args.interval
+
+    out_dir = args.out or os.path.join(ROOT, "data", "5m")
+    reports = os.path.join(ROOT, "reports")
+    if args.out:
+        os.makedirs(out_dir, exist_ok=True)
+    else:
+        os.makedirs(out_dir, exist_ok=True)
+    os.makedirs(reports, exist_ok=True)
+    LOG_PATH = os.path.join(reports, "fetch_5m.log")
+    start_ms = int(dt.datetime.strptime(args.start, "%Y-%m-%d").replace(tzinfo=dt.timezone.utc).timestamp() * 1000)
+
+    t0 = time.time()
+    syms = universe(args.min_turnover, args.pairs)
+    log(f"выгрузка {len(syms)} пар с {args.start}, потоков {args.workers}, интервал {MIN_INTERVAL}s → ~{1/MIN_INTERVAL:.0f} запр/с")
+    done = 0
+    results = []
+    with cf.ThreadPoolExecutor(max_workers=args.workers) as ex:
+        futs = {ex.submit(fetch_symbol, s, start_ms, out_dir): s for s, _ in syms}
+        for fut in cf.as_completed(futs):
+            sym = futs[fut]
+            try:
+                res = fut.result()
+            except Exception as e:  # noqa: BLE001
+                res = {"symbol": sym, "bars": 0, "first_ts": None, "last_ts": None, "error": str(e)}
+            results.append(res)
+            done += 1
+            el = time.time() - t0
+            log(f"{done}/{len(syms)} {sym}: баров {res.get('bars')} "
+                f"({dt.datetime.fromtimestamp((res.get('first_ts') or 0)/1000, dt.timezone.utc).strftime('%Y-%m-%d')} … "
+                f"{dt.datetime.fromtimestamp((res.get('last_ts') or 0)/1000, dt.timezone.utc).strftime('%Y-%m-%d')})"
+                f" | прошло {el/60:.1f} мин")
+
+    manifest = {"pairs": {r["symbol"]: {"symbol": r["symbol"], "bars": r.get("bars", 0),
+                                        "first_ts": r.get("first_ts"), "last_ts": r.get("last_ts")}
+                          for r in results if r.get("bars")},
+                "interval": "5m", "source": "bybit /v5/market/kline", "start": args.start,
+                "generated_at": dt.datetime.now(dt.timezone.utc).isoformat()}
+    with open(os.path.join(ROOT, "data", "manifest_5m.json"), "w") as fh:
+        json.dump(manifest, fh, ensure_ascii=False, indent=1)
+
+    # валидатор: дубли и покрытие календарных 5м-слотов
+    cov = {}
+    dup_total = 0
+    for r in results:
+        sym = r["symbol"]
+        path = os.path.join(out_dir, f"{sym}_5m.csv")
+        if not os.path.isfile(path):
+            continue
+        tss = []
+        with open(path) as fh:
+            for line in fh:
+                line = line.strip()
+                if line:
+                    tss.append(int(line.split(",")[0]))
+        d = len(tss) - len(set(tss))
+        dup_total += d
+        if len(tss) < 2:
+            continue
+        slots = (tss[-1] - tss[0]) // STEP_MS + 1
+        cov[sym] = {"bars": len(tss), "coverage": round(len(set(tss)) / slots, 4), "dup": d}
+    with open(os.path.join(reports, "validate_5m.json"), "w") as fh:
+        json.dump({"dup": dup_total, "pairs": cov, "checked_at": dt.datetime.now(dt.timezone.utc).isoformat()},
+                  fh, ensure_ascii=False, indent=1)
+
+    total_bars = sum(v["bars"] for v in cov.values())
+    log(f"ГОТОВО: пар в валидации {len(cov)}, баров {total_bars:,}, дублей {dup_total}, "
+        f"медиана покрытия {sorted(v['coverage'] for v in cov.values())[len(cov)//2] if cov else 0}, "
+        f"время {(time.time()-t0)/60:.1f} мин".replace(",", " "))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())