#!/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())