eth_hedge_sim/backend/app/exchange/candles.py

"""指数/永续 K 线高低点：期期对冲振幅回看。"""

from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import Any

from .okx.parse import safe_float as okx_safe_float
from .binance.parse import safe_float as bn_safe_float

logger = logging.getLogger(__name__)


@dataclass(frozen=True, slots=True)
class AmplitudeHL:
    high: float
    low: float
    mid: float
    hours: float
    bar_count: int

    @property
    def range_pct(self) -> float:
        if self.mid <= 0:
            return 0.0
        return (self.high - self.low) / self.mid * 100.0


def _hl_from_okx_candles(rows: list[Any]) -> tuple[float, float] | None:
    """OKX candle row: [ts, o, h, l, c, ...] newest first."""
    highs: list[float] = []
    lows: list[float] = []
    for row in rows:
        if not isinstance(row, (list, tuple)) or len(row) < 5:
            continue
        h = okx_safe_float(row[2])
        lo = okx_safe_float(row[3])
        if h is None or lo is None or h <= 0 or lo <= 0:
            continue
        highs.append(float(h))
        lows.append(float(lo))
    if not highs or not lows:
        return None
    return max(highs), min(lows)


def _hl_from_binance_klines(rows: list[Any]) -> tuple[float, float] | None:
    """Binance kline: [openTime, o, h, l, c, ...] oldest first."""
    highs: list[float] = []
    lows: list[float] = []
    for row in rows:
        if not isinstance(row, (list, tuple)) or len(row) < 5:
            continue
        h = bn_safe_float(row[2])
        lo = bn_safe_float(row[3])
        if h is None or lo is None or h <= 0 or lo <= 0:
            continue
        highs.append(float(h))
        lows.append(float(lo))
    if not highs or not lows:
        return None
    return max(highs), min(lows)


def fetch_okx_amplitude_hl(
    *,
    inst_id: str,
    hours: float,
    base_url: str = "https://www.okx.com",
    proxy: str | None = None,
) -> AmplitudeHL | None:
    """用 1H K 线回看 hours；inst 可用指数 ETH-USD 或永续 ETH-USDT-SWAP。"""
    import math

    import httpx

    hrs = max(1.0, float(hours))
    limit = int(min(300, max(2, math.ceil(hrs) + 1)))
    try:
        with httpx.Client(
            base_url=base_url.rstrip("/"),
            timeout=15.0,
            proxy=(proxy or "").strip() or None,
            headers={"Accept": "application/json", "User-Agent": "eth-hedge-sim/0.1"},
        ) as client:
            r = client.get(
                "/api/v5/market/candles",
                params={"instId": inst_id, "bar": "1H", "limit": str(limit)},
            )
            r.raise_for_status()
            body = r.json()
            if str(body.get("code")) != "0":
                logger.warning("OKX candles error: %s", body.get("msg"))
                return None
            data = body.get("data") or []
    except Exception as e:
        logger.warning("OKX candles fetch failed: %s", e)
        return None
    hl = _hl_from_okx_candles(data)
    if hl is None:
        return None
    high, low = hl
    mid = (high + low) / 2.0
    return AmplitudeHL(
        high=high, low=low, mid=mid, hours=hrs, bar_count=len(data)
    )


def fetch_binance_amplitude_hl(
    *,
    symbol: str,
    hours: float,
    fapi_base: str = "https://fapi.binance.com",
    proxy: str | None = None,
) -> AmplitudeHL | None:
    """USDT 永续 1h klines。"""
    import math

    import httpx

    hrs = max(1.0, float(hours))
    limit = int(min(500, max(2, math.ceil(hrs) + 1)))
    sym = str(symbol or "ETHUSDT").upper().replace("-", "")
    try:
        with httpx.Client(
            base_url=fapi_base.rstrip("/"),
            timeout=15.0,
            proxy=(proxy or "").strip() or None,
            headers={"Accept": "application/json", "User-Agent": "eth-hedge-sim/0.1"},
            trust_env=False,
        ) as client:
            r = client.get(
                "/fapi/v1/klines",
                params={"symbol": sym, "interval": "1h", "limit": limit},
            )
            r.raise_for_status()
            data = r.json()
            if not isinstance(data, list):
                return None
    except Exception as e:
        logger.warning("Binance klines fetch failed: %s", e)
        return None
    hl = _hl_from_binance_klines(data)
    if hl is None:
        return None
    high, low = hl
    mid = (high + low) / 2.0
    return AmplitudeHL(
        high=high, low=low, mid=mid, hours=hrs, bar_count=len(data)
    )


def fetch_amplitude_hl_for_runtime(hours: float) -> AmplitudeHL | None:
    """按当前交易所 runtime 拉振幅高低点。"""
    from ..config import get_settings
    from .runtime import load_runtime_settings

    s = get_settings()
    rt = load_runtime_settings()
    ex = str(rt.exchange or "okx").strip().lower()
    hrs = float(hours)
    if ex in ("binance", "bn"):
        return fetch_binance_amplitude_hl(
            symbol=str(rt.perp_inst_id or "ETHUSDT"),
            hours=hrs,
            fapi_base=s.binance_fapi_base,
            proxy=s.binance_http_proxy or None,
        )
    # OKX：优先指数，失败再试永续
    idx = str(rt.index_inst_id or "ETH-USD")
    amp = fetch_okx_amplitude_hl(
        inst_id=idx,
        hours=hrs,
        base_url=s.okx_rest_base,
        proxy=s.okx_http_proxy or None,
    )
    if amp is not None:
        return amp
    return fetch_okx_amplitude_hl(
        inst_id=str(rt.perp_inst_id or "ETH-USDT-SWAP"),
        hours=hrs,
        base_url=s.okx_rest_base,
        proxy=s.okx_http_proxy or None,
    )