qihuo/recommend_trend.py

"""品种推荐：近一周日线走势（多头 / 空头 / 震荡 / 转多 / 转空）。"""
from __future__ import annotations

import logging
from typing import Callable, Optional

import requests

from kline_chart import fetch_sina_klines, ths_to_sina_chart_symbol

logger = logging.getLogger(__name__)

DAILY_LOOKBACK = 7
OVERLAP_WINDOW = 3
OVERLAP_RANGE_THRESHOLD = 0.70
KLINE_FETCH_TIMEOUT = 5

TREND_LONG = "long"
TREND_SHORT = "short"
TREND_RANGE = "range"
TREND_BREAK_LONG = "break_long"
TREND_BREAK_SHORT = "break_short"


def _bar_ohlc(bar: dict) -> tuple[float, float, float, float]:
    o = float(bar.get("o") or bar.get("open") or 0)
    h = float(bar.get("h") or bar.get("high") or o)
    l = float(bar.get("l") or bar.get("low") or o)
    c = float(bar.get("c") or bar.get("close") or o)
    return o, h, l, c


def kline_overlap_ratio(bars: list) -> float:
    """三根 K 线高低价区间的重叠度 = 交集 / 并集（0~1）。"""
    if len(bars) < OVERLAP_WINDOW:
        return 0.0
    chunk = bars[-OVERLAP_WINDOW:]
    lows, highs = [], []
    for bar in chunk:
        _, h, l, _ = _bar_ohlc(bar)
        if h <= 0 and l <= 0:
            continue
        lows.append(l)
        highs.append(h)
    if len(lows) < OVERLAP_WINDOW:
        return 0.0
    overlap = max(0.0, min(highs) - max(lows))
    union = max(highs) - min(lows)
    if union <= 0:
        return 1.0 if overlap > 0 else 0.0
    return overlap / union


def _direction_from_closes(bars: list) -> str:
    if len(bars) < 2:
        return TREND_RANGE
    closes = [_bar_ohlc(b)[3] for b in bars if _bar_ohlc(b)[3] > 0]
    if len(closes) < 2:
        return TREND_RANGE
    if closes[-1] > closes[0]:
        return TREND_LONG
    if closes[-1] < closes[0]:
        return TREND_SHORT
    return TREND_RANGE


def analyze_daily_trend(bars: list, *, overlap_threshold: float = OVERLAP_RANGE_THRESHOLD) -> dict:
    """根据近一周日线判断走势；最近三天重叠度≥阈值视为震荡。"""
    empty = {
        "trend": "",
        "trend_label": "—",
        "trend_transition": False,
        "trend_overlap_pct": None,
        "trend_prev_overlap_pct": None,
    }
    if len(bars) < OVERLAP_WINDOW:
        return empty

    recent = bars[-DAILY_LOOKBACK:] if len(bars) > DAILY_LOOKBACK else bars
    curr_overlap = kline_overlap_ratio(recent)
    prev_overlap = kline_overlap_ratio(recent[:-OVERLAP_WINDOW]) if len(recent) >= OVERLAP_WINDOW * 2 else 0.0

    curr_range = curr_overlap >= overlap_threshold
    prev_range = prev_overlap >= overlap_threshold

    if curr_range:
        trend, label = TREND_RANGE, "震荡"
        transition = False
    else:
        direction = _direction_from_closes(recent[-OVERLAP_WINDOW:])
        if direction == TREND_LONG:
            trend, label = TREND_LONG, "多头"
        elif direction == TREND_SHORT:
            trend, label = TREND_SHORT, "空头"
        else:
            trend, label = TREND_RANGE, "震荡"
        transition = prev_range and trend in (TREND_LONG, TREND_SHORT)
        if transition:
            if trend == TREND_LONG:
                trend, label = TREND_BREAK_LONG, "转多"
            else:
                trend, label = TREND_BREAK_SHORT, "转空"

    return {
        "trend": trend,
        "trend_label": label,
        "trend_transition": transition,
        "trend_overlap_pct": round(curr_overlap * 100, 1),
        "trend_prev_overlap_pct": round(prev_overlap * 100, 1) if prev_overlap else None,
    }


def _normalize_daily_bars(raw: list) -> list:
    out = []
    for row in raw:
        if isinstance(row, list) and len(row) >= 5:
            out.append({
                "d": str(row[0]),
                "o": float(row[1]),
                "h": float(row[2]),
                "l": float(row[3]),
                "c": float(row[4]),
            })
        elif isinstance(row, dict) and row.get("d"):
            out.append({
                "d": str(row["d"]),
                "o": float(row.get("o", 0) or 0),
                "h": float(row.get("h", 0) or 0),
                "l": float(row.get("l", 0) or 0),
                "c": float(row.get("c", 0) or 0),
            })
    return out


def _fetch_sina_daily_quick(chart_sym: str) -> list:
    url = (
        "https://stock2.finance.sina.com.cn/futures/api/json.php/"
        f"IndexService.getInnerFuturesDailyKLine?symbol={chart_sym}"
    )
    try:
        resp = requests.get(
            url, timeout=KLINE_FETCH_TIMEOUT,
            headers={"Referer": "https://finance.sina.com.cn"},
        )
        raw = resp.json()
        if raw and isinstance(raw, list):
            bars = _normalize_daily_bars(raw)
            if bars:
                return bars
    except Exception as exc:
        logger.debug("quick daily kline failed %s: %s", chart_sym, exc)
    return []


def fetch_week_daily_bars(
    symbol: str,
    *,
    fetch_fn: Callable[[str, str], list] | None = None,
) -> list:
    sym = (symbol or "").strip()
    if not sym:
        return []
    if fetch_fn:
        try:
            bars = fetch_fn(sym, "d") or []
        except Exception as exc:
            logger.debug("fetch week daily failed %s: %s", sym, exc)
            return []
        return bars[-DAILY_LOOKBACK:] if bars else []

    chart_sym = ths_to_sina_chart_symbol(sym)
    if not chart_sym:
        return []
    bars = _fetch_sina_daily_quick(chart_sym)
    if not bars:
        try:
            bars = fetch_sina_klines(sym, "d") or []
        except Exception as exc:
            logger.debug("fetch week daily fallback failed %s: %s", sym, exc)
            return []
    return bars[-DAILY_LOOKBACK:] if bars else []


def analyze_product_trend(
    symbol: str,
    *,
    fetch_fn: Callable[[str, str], list] | None = None,
) -> dict:
    """拉取主力合约一周日线并分析走势。"""
    sym = (symbol or "").strip()
    if not sym:
        return analyze_daily_trend([])
    bars = fetch_week_daily_bars(sym, fetch_fn=fetch_fn)
    return analyze_daily_trend(bars)


def trend_sort_key(row: dict) -> tuple:
    """转多/转空优先，其次多头/空头，震荡靠后。"""
    trend = (row.get("trend") or "").strip()
    priority = {
        TREND_BREAK_LONG: 0,
        TREND_BREAK_SHORT: 0,
        TREND_LONG: 1,
        TREND_SHORT: 1,
        TREND_RANGE: 2,
    }
    status_order = {"ok": 0, "margin_ok": 1, "blocked": 2, "no_price": 3}
    return (
        priority.get(trend, 3),
        status_order.get(row.get("status") or "", 9),
        -(int(row.get("max_lots") or 0)),
    )


def sort_recommend_by_trend(rows: list[dict]) -> list[dict]:
    return sorted(rows, key=trend_sort_key)