from __future__ import annotations

import math
import statistics
from datetime import datetime
from typing import Any

from backend.data.numbers import finite_number as _number


def _optional_number(value: Any) -> float | None:
    if value in (None, ""):
        return None
    try:
        result = float(value)
    except (TypeError, ValueError):
        return None
    return result if math.isfinite(result) else None


def _rounded_optional(value: Any, digits: int = 2) -> float | None:
    parsed = _optional_number(value)
    return round(parsed, digits) if parsed is not None else None


def _limit_threshold(code: str, name: str) -> float:
    if code.startswith(("4", "8")):
        return 29.0
    if code.startswith(("30", "68")):
        return 19.0
    return 9.5


def _ending_streak(flags: list[bool], end_index: int | None = None) -> int:
    if not flags:
        return 0
    index = len(flags) - 1 if end_index is None else min(end_index, len(flags) - 1)
    streak = 0
    while index >= 0 and flags[index]:
        streak += 1
        index -= 1
    return streak


def _max_streak(flags: list[bool]) -> int:
    best = current = 0
    for value in flags:
        current = current + 1 if value else 0
        best = max(best, current)
    return best


def _rsi(values: list[float], period: int = 6) -> float:
    if len(values) <= period:
        return 50.0
    changes = [values[index] - values[index - 1] for index in range(len(values) - period, len(values))]
    gains = sum(max(change, 0.0) for change in changes) / period
    losses = sum(max(-change, 0.0) for change in changes) / period
    if losses == 0:
        return 100.0 if gains > 0 else 50.0
    return 100 - 100 / (1 + gains / losses)


def _ema(values: list[float], period: int) -> list[float]:
    if not values:
        return []
    alpha = 2 / (period + 1)
    result = [values[0]]
    for value in values[1:]:
        result.append(value * alpha + result[-1] * (1 - alpha))
    return result


def _macd_series(values: list[float]) -> tuple[list[float], list[float]]:
    fast = _ema(values, 12)
    slow = _ema(values, 26)
    dif = [left - right for left, right in zip(fast, slow)]
    return dif, _ema(dif, 9)


def _macd_last(values: list[float]) -> tuple[float, float]:
    dif, dea = _macd_series(values)
    return (dif[-1], dea[-1]) if dif and dea else (0.0, 0.0)


def _weekly_series(rows: list[dict[str, Any]]) -> tuple[list[float], list[float]]:
    weeks: dict[str, tuple[float, float]] = {}
    for row in rows:
        trade_date = str(row.get("trade_date") or "")
        try:
            key = datetime.strptime(trade_date, "%Y%m%d").strftime("%G-%V")
        except ValueError:
            continue
        close = _number(row.get("close"))
        amount = _number(row.get("amount"))
        previous = weeks.get(key, (close, 0.0))
        weeks[key] = (close, previous[1] + amount)
    ordered = list(weeks.values())
    return [item[0] for item in ordered], [item[1] for item in ordered]


def _broken_reversal_metrics(
    rows: list[dict[str, Any]], flags: list[bool], code: str, name: str,
) -> dict[str, Any]:
    result = {"signal": 0, "days": 0, "recovered": 0, "volume_ratio": 0.0}
    if not rows or not flags[-1]:
        return result
    current_close = _number(rows[-1].get("close"))
    current_volume = _number(rows[-1].get("vol"))
    for days in range(1, 4):
        index = len(rows) - 1 - days
        if index <= 0 or flags[index] or _ending_streak(flags, index - 1) < 2:
            continue
        broken_high = _number(rows[index].get("high"))
        broken_volume = _number(rows[index].get("vol"))
        recovered = int(current_close >= broken_high > 0)
        volume_ratio = current_volume / broken_volume if broken_volume else 0.0
        return {
            "signal": int(recovered and volume_ratio >= 1),
            "days": days,
            "recovered": recovered,
            "volume_ratio": round(volume_ratio, 3),
        }
    return result


def _is_limit_bar(rows: list[dict[str, Any]], index: int, code: str, name: str) -> bool:
    if index < 0 or index >= len(rows):
        return False
    return _number(rows[index].get("pct_chg")) >= _limit_threshold(code, name)


def _touched_limit_bar(rows: list[dict[str, Any]], index: int, code: str, name: str) -> bool:
    if index <= 0 or index >= len(rows):
        return False
    previous_close = _number(rows[index - 1].get("close"))
    high = _number(rows[index].get("high"))
    if previous_close <= 0 or high <= 0:
        return False
    touched_change = (high / previous_close - 1) * 100
    return touched_change >= _limit_threshold(code, name)


def _matches(actual: Any, operator: str, expected: Any) -> bool:
    if actual is None:
        return False
    try:
        if operator == "between":
            return float(expected[0]) <= float(actual) <= float(expected[1])
        if operator == "in":
            return actual in expected
        if operator == ">":
            return float(actual) > float(expected)
        if operator == ">=":
            return float(actual) >= float(expected)
        if operator == "<":
            return float(actual) < float(expected)
        if operator == "<=":
            return float(actual) <= float(expected)
        if operator == "==":
            return actual == expected or float(actual) == float(expected)
        if operator == "!=":
            return actual != expected
    except (TypeError, ValueError, IndexError):
        return False
    return False


def _percentile_map(rows: list[dict[str, Any]], field: str, direction: str) -> dict[str, float]:
    ordered = sorted(rows, key=lambda item: _number(item.get(field)))
    denominator = max(1, len(ordered) - 1)
    result = {}
    for index, row in enumerate(ordered):
        percentile = index / denominator
        result[row["ts_code"]] = 1 - percentile if direction == "asc" else percentile
    return result


def _available_percentile_map(
    rows: list[dict[str, Any]], field: str, direction: str,
) -> dict[str, float | None]:
    available = [row for row in rows if row.get(field) is not None]
    result: dict[str, float | None] = {
        str(row.get("ts_code") or ""): None for row in rows
    }
    if not available:
        return result
    ordered = sorted(available, key=lambda item: _number(item.get(field)))
    denominator = max(1, len(ordered) - 1)
    for index, row in enumerate(ordered):
        percentile = 0.5 if len(ordered) == 1 else index / denominator
        result[str(row.get("ts_code") or "")] = (
            1 - percentile if direction == "asc" else percentile
        )
    return result


def _pearson(first: list[float], second: list[float]) -> float:
    if len(first) != len(second) or len(first) < 20:
        return 0.0
    first_mean = statistics.fmean(first)
    second_mean = statistics.fmean(second)
    numerator = sum(
        (left - first_mean) * (right - second_mean)
        for left, right in zip(first, second)
    )
    left_sum = sum((value - first_mean) ** 2 for value in first)
    right_sum = sum((value - second_mean) ** 2 for value in second)
    denominator = math.sqrt(left_sum * right_sum)
    return numerator / denominator if denominator else 0.0


def _risk_flags(
    row: dict[str, Any], regime: str, include_regime_risk: bool = True
) -> list[str]:
    flags = []
    if row.get("pct_chg", 0) >= 9.5:
        flags.append("当日接近涨停，次日存在高开与无法成交风险")
    if row.get("return_10d", 0) >= 25:
        flags.append("短期累计涨幅较高")
    if row.get("volatility_10d", 0) >= 7:
        flags.append("波动率偏高")
    if row.get("amount_billion", 0) < 1:
        flags.append("成交承载力偏弱")
    if include_regime_risk and regime == "retreat":
        flags.append("市场处于退潮阶段，策略可能选择空仓")
    return flags


def _regime_reason(regime: str) -> str:
    return {
        "ice": "情绪和赚钱效应处于低位，重点观察率先抗跌与转折信号。",
        "repair": "核心指标从低位改善，适合观察率先修复且有板块共振的方向。",
        "fermentation": "赚钱效应扩散，主线和梯队持续增强。",
        "climax": "情绪处于高位，后排跟风与兑现风险同时上升。",
        "divergence": "指数或核心仍强，但广度、封板质量开始分化。",
        "retreat": "情绪指标继续走弱，应提高筛选门槛并接受无候选结果。",
    }.get(regime, "市场阶段待确认。")