Divergent Latency Fade: арбитраж стратегии на Python

"""
Divergent Latency Fade — мониторинг расхождения BTC/USDT ↔ ETH/USDT
------------------------------------------------------------------
• Только ccxt (REST) — без ордеров
• Каждые N секунд выводит: z-score, цены BTC и ETH, а также сигналы
• Настройки регулируются в блоке CONFIG
"""

import ccxt
import time
import numpy as np
from datetime import datetime

# ─────────────────────── CONFIG ────────────────────────
SYMBOL_BASE   = "BTC/USDT"   # «опережающий» актив
SYMBOL_TRAD   = "ETH/USDT"   # «отстающий» актив
INTERVAL_SEC  = 5            # период опроса биржи (сек)
WINDOW        = 30           # глубина истории (точек)
Z_ENTRY       = 2.0          # порог входа
Z_EXIT        = 0.3          # порог выхода
# ────────────────────────────────────────────────────────

exchange = ccxt.binance()
history_base, history_trad = [], []


# ──────────── utils ────────────
def ts() -> str:
    """Текущий HH:MM:SS для логов"""
    return datetime.now().strftime("%H:%M:%S")


def log(text: str):
    print(f"{ts()} | {text}")


# ──────────── math ────────────
def beta_estimate(x: np.ndarray, y: np.ndarray) -> float:
    """
    OLS-оценка β, где  y = α + β·x
    Здесь x = log(BTC), y = log(ETH)
    """
    if len(x) < 2:
        return 1.0
    cov = np.cov(y, x)
    return cov[0, 1] / cov[1, 1] if cov[1, 1] else 1.0


def compute_zscore(log_btc: np.ndarray, log_eth: np.ndarray) -> float:
    beta   = beta_estimate(log_btc, log_eth)
    spread = log_eth - beta * log_btc
    mean, std = spread.mean(), spread.std(ddof=1)
    return (spread[-1] - mean) / std if std > 0 else 0.0


# ──────────── main loop ────────────
log("🚀 Запуск стратегии Divergent Latency Fade")

while True:
    try:
        price_btc = float(exchange.fetch_ticker(SYMBOL_BASE)["last"])
        price_eth = float(exchange.fetch_ticker(SYMBOL_TRAD)["last"])
    except Exception as e:
        log(f"⚠ Ошибка получения цен: {e}")
        time.sleep(INTERVAL_SEC)
        continue

    # пополняем буферы
    history_base.append(price_btc)
    history_trad.append(price_eth)

    if len(history_base) > WINDOW:
        history_base.pop(0)
        history_trad.pop(0)

    # считаем сигнал, когда накопили достаточно точек
    if len(history_base) >= WINDOW:
        log_btc = np.log(np.array(history_base))
        log_eth = np.log(np.array(history_trad))

        z = compute_zscore(log_btc, log_eth)

        if abs(z) > Z_ENTRY:
            direction = "BUY ETH (отстаёт от BTC)" if z < 0 else "SELL ETH (опережает BTC)"
            log(f"🔔 SIGNAL: {direction} | z={z:.2f} | BTC={price_btc:,.0f} | ETH={price_eth:,.1f}")
        elif abs(z) < Z_EXIT:
            log(f"✅ EXIT: z={z:.2f} (вернулись к норме)")
        else:
            log(f"📊 z={z:.2f} | BTC={price_btc:,.0f} | ETH={price_eth:,.1f}")

    time.sleep(INTERVAL_SEC)