crypto_monitor_user/lib/hub/hub_divergence_scan_lib.py

"""行情区:Top20 内 MACD 背离扫描(档 A)+ 4h/日线/周线共振."""
from __future__ import annotations

import json
from datetime import datetime
from pathlib import Path
from typing import Any, Callable, Mapping, Sequence

from lib.hub.hub_volume_rank_lib import TOP_N_DEFAULT, get_cached_rank, volume_rank_timezone

SCAN_CACHE_VERSION = 1
SCAN_TIMEFRAMES: tuple[str, ...] = ("4h", "1d", "1w")
SWING_LOOKBACK = 4
SWING_ALIGN_BARS = 30
RECENCY_BARS = 60
MACD_FAST = 12
MACD_SLOW = 26
MACD_SIGNAL = 9

TAB_LABELS: dict[str, str] = {
    "4h": "4h背离",
    "1d": "日线背离",
    "1w": "周线背离",
}

TF_SHORT: dict[str, str] = {"4h": "4h", "1d": "日线", "1w": "周线"}


def default_cache_path() -> Path:
    from lib.paths import hub_data_dir

    return hub_data_dir() / "hub_divergence_scan.json"


def ema_array(values: Sequence[float | None], period: int) -> list[float | None]:
    out: list[float | None] = [None] * len(values)
    if period <= 0 or len(values) < period:
        return out
    k = 2.0 / (period + 1)
    sma = sum(v for v in values[:period] if v is not None) / period
    out[period - 1] = sma
    prev = sma
    for i in range(period, len(values)):
        v = values[i]
        if v is None:
            continue
        prev = v * k + prev * (1 - k)
        out[i] = prev
    return out


def find_swings(values: Sequence[float | None], lookback: int) -> tuple[list[dict], list[dict]]:
    lows: list[dict] = []
    highs: list[dict] = []
    lb = max(1, int(lookback))
    n = len(values)
    for i in range(lb, n - lb):
        v = values[i]
        if v is None:
            continue
        is_low = True
        is_high = True
        for j in range(1, lb + 1):
            lv = values[i - j]
            rv = values[i + j]
            if lv is None or rv is None or v > lv or v > rv:
                is_low = False
            if lv is None or rv is None or v < lv or v < rv:
                is_high = False
        if is_low:
            lows.append({"i": i, "v": float(v)})
        if is_high:
            highs.append({"i": i, "v": float(v)})
    return lows, highs


def build_macd_by_index(closes: Sequence[float]) -> list[float | None]:
    ema12 = ema_array(closes, MACD_FAST)
    ema26 = ema_array(closes, MACD_SLOW)
    macd: list[float | None] = [None] * len(closes)
    for i in range(len(closes)):
        if ema12[i] is not None and ema26[i] is not None:
            macd[i] = ema12[i] - ema26[i]
    return macd


def detect_latest_macd_divergence(
    closes: Sequence[float],
    *,
    swing_lookback: int = SWING_LOOKBACK,
    align_bars: int = SWING_ALIGN_BARS,
    recency_bars: int = RECENCY_BARS,
) -> dict[str, Any]:
    """档 A:最近一对摆动 MACD 顶/底背离(与 chart.js detectDivergences 同类)."""
    if len(closes) < swing_lookback * 2 + 10:
        return {"direction": None}
    macd = build_macd_by_index(closes)
    p_lows, p_highs = find_swings(closes, swing_lookback)
    i_lows, i_highs = find_swings(macd, swing_lookback)

    def recent_enough(idx: int) -> bool:
        return idx >= max(0, len(closes) - recency_bars)

    if len(p_lows) >= 2 and len(i_lows) >= 2:
        p1, p2 = p_lows[-2], p_lows[-1]
        i1, i2 = i_lows[-2], i_lows[-1]
        if (
            abs(p1["i"] - i1["i"]) < align_bars
            and abs(p2["i"] - i2["i"]) < align_bars
            and p2["v"] < p1["v"]
            and i2["v"] > i1["v"]
            and recent_enough(p2["i"])
        ):
            return {"direction": "bull", "bar_index": p2["i"]}

    if len(p_highs) >= 2 and len(i_highs) >= 2:
        p1, p2 = p_highs[-2], p_highs[-1]
        i1, i2 = i_highs[-2], i_highs[-1]
        if (
            abs(p1["i"] - i1["i"]) < align_bars
            and abs(p2["i"] - i2["i"]) < align_bars
            and p2["v"] > p1["v"]
            and i2["v"] < i1["v"]
            and recent_enough(p2["i"])
        ):
            return {"direction": "bear", "bar_index": p2["i"]}

    return {"direction": None}


def chart_candles_to_bars(candles: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]:
    out: list[dict[str, Any]] = []
    for c in candles:
        try:
            t = c.get("time")
            if t is None:
                continue
            ms = int(t) * 1000 if int(t) < 10_000_000_000 else int(t)
            out.append(
                {
                    "open_time_ms": ms,
                    "open": float(c["open"]),
                    "high": float(c["high"]),
                    "low": float(c["low"]),
                    "close": float(c["close"]),
                    "volume": float(c.get("volume") or 0),
                }
            )
        except (KeyError, TypeError, ValueError):
            continue
    return out


def normalize_ohlcv_rows(rows: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]:
    if not rows:
        return []
    first = rows[0]
    if first.get("open_time_ms") is not None:
        return [dict(r) for r in rows]
    return chart_candles_to_bars(rows)


def bars_to_closes(bars: Sequence[Mapping[str, Any]], *, exclude_open: bool = True) -> list[float]:
    rows = list(bars)
    if exclude_open and len(rows) > 1:
        rows = rows[:-1]
    out: list[float] = []
    for b in rows:
        try:
            out.append(float(b["close"]))
        except (KeyError, TypeError, ValueError):
            continue
    return out


def bar_time_at(bars: Sequence[Mapping[str, Any]], index: int) -> int | None:
    if index < 0 or index >= len(bars):
        return None
    try:
        return int(bars[index]["open_time_ms"])
    except (KeyError, TypeError, ValueError):
        return None


def analyze_ohlcv_bars(bars: Sequence[Mapping[str, Any]]) -> dict[str, Any]:
    closed = list(bars)
    if len(closed) > 1:
        closed = closed[:-1]
    closes = bars_to_closes(bars, exclude_open=True)
    hit = detect_latest_macd_divergence(closes)
    direction = hit.get("direction")
    bar_index = hit.get("bar_index")
    open_time_ms = None
    if direction and bar_index is not None:
        open_time_ms = bar_time_at(closed, int(bar_index))
    bars_ago = None
    if direction and bar_index is not None:
        bars_ago = max(0, len(closed) - 1 - int(bar_index))
    return {
        "direction": direction,
        "bar_index": bar_index,
        "open_time_ms": open_time_ms,
        "bars_ago": bars_ago,
    }


def compute_confluence(tf_hits: Mapping[str, Mapping[str, Any]]) -> dict[str, Any]:
    dirs: dict[str, str] = {}
    for tf in SCAN_TIMEFRAMES:
        d = (tf_hits.get(tf) or {}).get("direction")
        if d in ("bull", "bear"):
            dirs[tf] = d

    if not dirs:
        return {
            "confluence": 0,
            "confluence_kind": "none",
            "confluence_css": "none",
            "is_split": False,
            "split_detail": "",
            "primary_direction": None,
            "direction_label": "",
            "timeframes_hit": [],
        }

    unique = set(dirs.values())
    if len(unique) > 1:
        parts = []
        for tf in SCAN_TIMEFRAMES:
            if tf in dirs:
                label = "底" if dirs[tf] == "bull" else "顶"
                parts.append(f"{TF_SHORT.get(tf, tf)}{label}")
        return {
            "confluence": 0,
            "confluence_kind": "分歧",
            "confluence_css": "split",
            "is_split": True,
            "split_detail": " · ".join(parts),
            "primary_direction": _latest_direction(tf_hits),
            "direction_label": "分歧",
            "timeframes_hit": list(dirs.keys()),
        }

    direction = next(iter(unique))
    count = len(dirs)
    return {
        "confluence": count,
        "confluence_kind": f"{count}周期",
        "confluence_css": f"c{count}",
        "is_split": False,
        "split_detail": "",
        "primary_direction": direction,
        "direction_label": "底背离" if direction == "bull" else "顶背离",
        "timeframes_hit": list(dirs.keys()),
    }


def _latest_direction(tf_hits: Mapping[str, Mapping[str, Any]]) -> str | None:
    best_tf = None
    best_ms = -1
    for tf in SCAN_TIMEFRAMES:
        row = tf_hits.get(tf) or {}
        d = row.get("direction")
        ms = row.get("open_time_ms")
        if d not in ("bull", "bear") or ms is None:
            continue
        if int(ms) > best_ms:
            best_ms = int(ms)
            best_tf = tf
    if best_tf is None:
        return None
    return (tf_hits.get(best_tf) or {}).get("direction")


def freshness_label(timeframe: str, bars_ago: int | None) -> str:
    if bars_ago is None:
        return ""
    n = int(bars_ago)
    if timeframe == "1w":
        return f"{n}周前" if n else "本周"
    if n <= 0:
        return "当根"
    return f"{n}根K前"


def build_symbol_scan_row(
    *,
    rank: int,
    symbol: str,
    volume_label: str,
    tf_hits: Mapping[str, Mapping[str, Any]],
) -> dict[str, Any]:
    conf = compute_confluence(tf_hits)
    tf_map = {tf: (tf_hits.get(tf) or {}).get("direction") for tf in SCAN_TIMEFRAMES}
    return {
        "rank": rank,
        "symbol": symbol,
        "volume_label": volume_label,
        "direction": conf.get("primary_direction"),
        "direction_label": conf.get("direction_label") or "",
        "confluence": conf.get("confluence") or 0,
        "confluence_kind": conf.get("confluence_kind") or "none",
        "confluence_css": conf.get("confluence_css") or "none",
        "is_split": bool(conf.get("is_split")),
        "split_detail": conf.get("split_detail") or "",
        "timeframes": tf_map,
        "tf_detail": {
            tf: {
                "direction": (tf_hits.get(tf) or {}).get("direction"),
                "open_time_ms": (tf_hits.get(tf) or {}).get("open_time_ms"),
                "bars_ago": (tf_hits.get(tf) or {}).get("bars_ago"),
                "freshness": freshness_label(tf, (tf_hits.get(tf) or {}).get("bars_ago")),
            }
            for tf in SCAN_TIMEFRAMES
        },
    }


def filter_tab_items(items: Sequence[Mapping[str, Any]], tab: str) -> list[dict[str, Any]]:
    tab = (tab or "").strip().lower()
    if tab not in SCAN_TIMEFRAMES:
        return [dict(x) for x in items]
    out: list[dict[str, Any]] = []
    for row in items:
        tf = (row.get("tf_detail") or {}).get(tab) or {}
        if tf.get("direction") not in ("bull", "bear"):
            continue
        item = dict(row)
        item["tab_timeframe"] = tab
        item["tab_direction"] = tf.get("direction")
        item["tab_direction_label"] = "底背离" if tf.get("direction") == "bull" else "顶背离"
        item["tab_freshness"] = tf.get("freshness") or ""
        item["tab_open_time_ms"] = tf.get("open_time_ms")
        out.append(item)
    out.sort(
        key=lambda x: (
            -1 if x.get("is_split") else int(x.get("confluence") or 0),
            int(x.get("rank") or 999),
        ),
        reverse=True,
    )
    return out


def load_scan_cache(path: Path | None = None) -> dict[str, Any]:
    p = path or default_cache_path()
    if not p.is_file():
        return {"version": SCAN_CACHE_VERSION, "exchanges": {}}
    try:
        data = json.loads(p.read_text(encoding="utf-8"))
        if not isinstance(data, dict):
            return {"version": SCAN_CACHE_VERSION, "exchanges": {}}
        if int(data.get("version") or 0) < SCAN_CACHE_VERSION:
            return {"version": SCAN_CACHE_VERSION, "exchanges": {}}
        data.setdefault("version", SCAN_CACHE_VERSION)
        data.setdefault("exchanges", {})
        return data
    except Exception:
        return {"version": SCAN_CACHE_VERSION, "exchanges": {}}


def save_scan_cache(data: dict[str, Any], path: Path | None = None) -> None:
    p = path or default_cache_path()
    p.parent.mkdir(parents=True, exist_ok=True)
    payload = dict(data)
    payload["version"] = SCAN_CACHE_VERSION
    payload["updated_at"] = datetime.now(volume_rank_timezone()).isoformat(timespec="seconds")
    p.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")


def merge_exchange_scan(
    cache: dict[str, Any],
    exchange_key: str,
    *,
    rank_date: str | None,
    items: list[dict[str, Any]],
    error: str | None = None,
) -> dict[str, Any]:
    ex_k = str(exchange_key or "").strip().lower()
    exchanges = dict(cache.get("exchanges") or {})
    exchanges[ex_k] = {
        "rank_date": rank_date,
        "items": items,
        "error": error,
        "scanned_at": datetime.now(volume_rank_timezone()).isoformat(timespec="seconds"),
    }
    out = dict(cache)
    out["exchanges"] = exchanges
    return out


def get_cached_scan(
    cache: dict[str, Any],
    exchange_key: str,
    *,
    tab: str = "4h",
) -> dict[str, Any]:
    ex_k = str(exchange_key or "").strip().lower()
    ex_data = (cache.get("exchanges") or {}).get(ex_k) or {}
    all_items = list(ex_data.get("items") or [])
    tab_key = (tab or "4h").strip().lower()
    items = filter_tab_items(all_items, tab_key) if tab_key in SCAN_TIMEFRAMES else all_items
    return {
        "ok": True,
        "exchange_key": ex_k,
        "tab": tab_key,
        "rank_date": ex_data.get("rank_date"),
        "updated_at": cache.get("updated_at"),
        "scanned_at": ex_data.get("scanned_at"),
        "items": items,
        "item_count": len(items),
        "error": ex_data.get("error"),
    }


def scan_top_symbols(
    rank_items: Sequence[Mapping[str, Any]],
    fetch_bars: Callable[[str, str], Sequence[Mapping[str, Any]]],
) -> list[dict[str, Any]]:
    """对 Top N 币种扫描三周期背离.fetch_bars(symbol, timeframe) -> OHLCV rows."""
    out: list[dict[str, Any]] = []
    for row in rank_items:
        symbol = str(row.get("symbol") or "").strip().upper()
        if not symbol:
            continue
        tf_hits: dict[str, dict[str, Any]] = {}
        for tf in SCAN_TIMEFRAMES:
            try:
                bars = fetch_bars(symbol, tf)
                tf_hits[tf] = analyze_ohlcv_bars(bars)
            except Exception:
                tf_hits[tf] = {"direction": None}
        out.append(
            build_symbol_scan_row(
                rank=int(row.get("rank") or 0),
                symbol=symbol,
                volume_label=str(row.get("volume_label") or row.get("volume_quote") or ""),
                tf_hits=tf_hits,
            )
        )
    return out


def cache_is_stale(
    cache: dict[str, Any],
    exchange_key: str,
    *,
    rank_date: str | None,
    max_age_sec: float = 3600.0,
) -> bool:
    ex_k = str(exchange_key or "").strip().lower()
    ex_data = (cache.get("exchanges") or {}).get(ex_k) or {}
    if not ex_data.get("items") and not ex_data.get("error"):
        return True
    if rank_date and ex_data.get("rank_date") != rank_date:
        return True
    updated = cache.get("updated_at") or ex_data.get("scanned_at")
    if not updated:
        return True
    try:
        dt = datetime.fromisoformat(str(updated))
        age = (datetime.now(dt.tzinfo) - dt).total_seconds()
        return age > max_age_sec
    except Exception:
        return True