"""入场信号分类 + 供需分析 + 量价形态识别

五种入场信号类型：
- 突破型 (breakout): 放量突破 20 日阻力位
- 确认型 (breakout_confirm): 突破后 1-2 日放量确认
- 回踩型 (pullback): 上升趋势中缩量回踩均线支撑
- 启动型 (launch): 缩量整理后首日放量
- 反转型 (reversal): 连续下跌后放量长阳反转
"""

import logging
from enum import Enum

import numpy as np
import pandas as pd

from app.config import settings

logger = logging.getLogger(__name__)


class EntrySignal(str, Enum):
    BREAKOUT = "breakout"
    BREAKOUT_CONFIRM = "breakout_confirm"
    PULLBACK = "pullback"
    LAUNCH = "launch"
    REVERSAL = "reversal"
    NONE = "none"


# ── 入场信号检测 ──


def classify_entry_signal(df: pd.DataFrame) -> dict:
    """分类入场信号类型

    优先级: breakout > breakout_confirm > pullback > launch > reversal

    Returns:
        {
            "signal_type": EntrySignal,
            "signal_score": float (0-100),
            "details": dict,
        }
    """
    if df is None or len(df) < 30:
        return {"signal_type": EntrySignal.NONE, "signal_score": 0, "details": {}}

    # 突破型
    result = detect_breakout(df)
    if result:
        result["signal_score"] = _score_breakout(df, result)
        return result

    # 突破确认型（突破后 1-3 日放量确认）
    result = detect_breakout_confirm(df)
    if result:
        result["signal_score"] = _score_breakout_confirm(df, result)
        return result

    # 回踩型
    result = detect_pullback(df)
    if result:
        result["signal_score"] = _score_pullback(df, result)
        return result

    # 启动型（缩量横盘后首日放量）
    result = detect_launch(df)
    if result:
        result["signal_score"] = _score_launch(df, result)
        return result

    # 反转型（连续下跌后放量长阳）
    result = detect_reversal(df)
    if result:
        result["signal_score"] = _score_reversal(df, result)
        return result

    return {"signal_type": EntrySignal.NONE, "signal_score": 0, "details": {}}


def detect_breakout(df: pd.DataFrame) -> dict | None:
    """突破型检测

    条件：
    1. 价格突破或贴近 20 日阻力位（close >= resist * 0.98）
    2. 量能放大 (vol > vol_ma5 * 1.2)
    3. MA5 > MA20（短期偏多）
    4. 收盘偏强 (close 在当日上半部)
    """
    last = df.iloc[-1]

    required = ["close", "high", "low", "vol", "ma5", "ma10", "ma20", "vol_ma5"]
    if any(c not in df.columns for c in required):
        return None
    if pd.isna(last["vol_ma5"]) or last["vol_ma5"] == 0:
        return None

    if len(df) < 22:
        return None
    resist_20d = df["high"].iloc[-21:-1].max()

    # 放宽：贴近阻力位 2% 即视为突破前兆
    if last["close"] < resist_20d * 0.98:
        return None

    # 量能放大
    min_vol_ratio = getattr(settings, "breakout_min_volume_ratio", 1.2)
    if last["vol"] <= last["vol_ma5"] * min_vol_ratio:
        return None

    # MA5 > MA20
    if pd.isna(last["ma20"]) or not (last["ma5"] > last["ma20"]):
        return None

    # 收盘偏强
    day_range = last["high"] - last["low"]
    if day_range > 0 and (last["close"] - last["low"]) / day_range < 0.45:
        return None

    breakout_pct = (last["close"] - resist_20d) / resist_20d * 100

    return {
        "signal_type": EntrySignal.BREAKOUT,
        "details": {
            "resist_level": round(resist_20d, 2),
            "breakout_pct": round(breakout_pct, 2),
            "volume_ratio": round(last["vol"] / last["vol_ma5"], 2),
        },
    }


def detect_breakout_confirm(df: pd.DataFrame) -> dict | None:
    """突破确认型检测

    条件：近 1-3 日内曾突破 20 日高点，今日放量确认（不要求再创新高）
    这捕捉了"突破→缩量回踩确认→再放量"的经典模式。

    1. 近 3 日内有 1 日 close > 20日前的高点
    2. 今日 vol > vol_ma5（放量确认）
    3. 今日 close > 昨日 close * 0.99（没有大幅回落）
    4. MA5 > MA20
    """
    if len(df) < 24:
        return None

    last = df.iloc[-1]
    required = ["close", "high", "low", "vol", "pct_chg", "ma5", "ma20", "vol_ma5"]
    if any(c not in df.columns for c in required):
        return None
    if pd.isna(last["vol_ma5"]) or last["vol_ma5"] == 0:
        return None
    if pd.isna(last["ma20"]):
        return None

    # 20 日阻力（不含近 3 日）
    resist_20d = df["high"].iloc[-23:-3].max()

    # 近 3 日内有突破
    recent_3 = df.tail(3)
    had_breakout = any(
        row["close"] > resist_20d
        for _, row in recent_3.iterrows()
        if not pd.isna(row["close"])
    )
    if not had_breakout:
        return None

    # 今日放量
    if last["vol"] <= last["vol_ma5"] * 1.0:
        return None

    # 今日没有大幅回落
    if last["close"] < df.iloc[-2]["close"] * 0.99:
        return None

    # MA5 > MA20
    if not (last["ma5"] > last["ma20"]):
        return None

    return {
        "signal_type": EntrySignal.BREAKOUT_CONFIRM,
        "details": {
            "resist_level": round(resist_20d, 2),
            "volume_ratio": round(last["vol"] / last["vol_ma5"], 2),
            "confirm_pct": round(last["pct_chg"], 2),
        },
    }


def detect_pullback(df: pd.DataFrame) -> dict | None:
    """回踩型检测

    条件：
    1. MA20 > MA60 — 中长期上升趋势
    2. 近 5 日内有 MA5 > MA20 — 近期处于上升趋势
    3. 价格靠近 MA10 或 MA20 (<3%)
    4. 近 3 日缩量（相对前 5 日）或 当日开始放量反弹
    5. RSI 30-70（放宽）
    """
    last = df.iloc[-1]

    required = ["close", "vol", "ma5", "ma10", "ma20", "ma60", "rsi14"]
    if any(c not in df.columns for c in required):
        return None
    if pd.isna(last["ma60"]) or pd.isna(last["rsi14"]):
        return None

    # 中长期上升趋势
    if not (last["ma20"] > last["ma60"]):
        return None

    # 近 5 日内有 MA5 > MA20
    recent_5 = df.tail(5)
    has_uptrend = any(
        row["ma5"] > row["ma20"]
        for _, row in recent_5.iterrows()
        if not pd.isna(row["ma5"]) and not pd.isna(row["ma20"])
    )
    if not has_uptrend:
        return None

    # 价格靠近 MA10 或 MA20（放宽至 3%）
    near_ma10 = (
        not pd.isna(last["ma10"])
        and last["ma10"] > 0
        and abs(last["close"] - last["ma10"]) / last["ma10"] < 0.03
    )
    near_ma20 = (
        not pd.isna(last["ma20"])
        and last["ma20"] > 0
        and abs(last["close"] - last["ma20"]) / last["ma20"] < 0.03
    )
    if not (near_ma10 or near_ma20):
        return None

    # 缩量回调 OR 当日放量反弹（二选一）
    if len(df) < 8:
        return None
    recent_3_vol = df["vol"].tail(3).mean()
    prev_5_vol = df["vol"].iloc[-8:-3].mean()

    shrinking = prev_5_vol > 0 and recent_3_vol / prev_5_vol < 0.9
    bouncing_today = (
        last["pct_chg"] > 0
        and not pd.isna(last["vol_ma5"])
        and last["vol_ma5"] > 0
        and last["vol"] > last["vol_ma5"] * 1.1
    )

    if not (shrinking or bouncing_today):
        return None

    # RSI（放宽至 30-70）
    if not (30 <= last["rsi14"] <= 70):
        return None

    support_ma = "MA10" if near_ma10 else "MA20"
    support_price = last["ma10"] if near_ma10 else last["ma20"]
    shrink_ratio = recent_3_vol / prev_5_vol if prev_5_vol > 0 else 1

    return {
        "signal_type": EntrySignal.PULLBACK,
        "details": {
            "support_ma": support_ma,
            "support_price": round(support_price, 2),
            "volume_shrink_ratio": round(shrink_ratio, 2),
            "bouncing": bouncing_today,
        },
    }


def detect_launch(df: pd.DataFrame) -> dict | None:
    """启动型检测（缩量横盘后首日放量）

    条件：
    1. 近 5-10 日缩量至均量的 60% 以下
    2. 今日放量（vol > vol_ma5 * 1.2）
    3. close > MA60（中长线偏多）
    4. 今日涨跌幅 > 0（方向向上）
    """
    last = df.iloc[-1]

    required = ["close", "high", "vol", "pct_chg", "ma5", "ma20", "ma60", "vol_ma5"]
    if any(c not in df.columns for c in required):
        return None
    if pd.isna(last["ma60"]) or pd.isna(last["vol_ma5"]) or last["vol_ma5"] == 0:
        return None

    # close > MA60
    if last["close"] < last["ma60"]:
        return None

    # 今日放量
    vol_ratio_today = last["vol"] / last["vol_ma5"]
    if vol_ratio_today < 1.2:
        return None

    # 今日上涨
    if last["pct_chg"] <= 0:
        return None

    # 近 5-10 日有缩量期（存在 vol < vol_ma5 * 0.6 的日子）
    if len(df) < 12:
        return None
    recent_10 = df.iloc[-11:-1]  # 不含今日
    vol_ma_col = "vol_ma10" if "vol_ma10" in df.columns else "vol_ma5"
    had_dry_up = False
    for _, row in recent_10.iterrows():
        if pd.isna(row.get(vol_ma_col)) or row[vol_ma_col] == 0:
            continue
        if row["vol"] < row[vol_ma_col] * 0.6:
            had_dry_up = True
            break

    if not had_dry_up:
        return None

    # 近 10 日振幅不大（横盘整理）
    recent_10_close = df["close"].iloc[-11:-1]
    if len(recent_10_close) < 5:
        return None
    price_range = (recent_10_close.max() - recent_10_close.min()) / recent_10_close.min()
    if price_range > 0.12:  # 12% 以内算横盘
        return None

    high_20d = df["high"].tail(20).max()

    return {
        "signal_type": EntrySignal.LAUNCH,
        "details": {
            "volume_ratio_today": round(vol_ratio_today, 2),
            "price_range_pct": round(price_range * 100, 2),
            "resist_level": round(high_20d, 2),
        },
    }


def detect_reversal(df: pd.DataFrame) -> dict | None:
    """反转型检测（连续下跌后放量长阳）

    条件：
    1. 近 5 日有 3 日以上下跌
    2. 今日放量长阳（pct_chg > 3%, vol > vol_ma5 * 1.5）
    3. 收盘价高于近 5 日最高价（强势反转）
    4. RSI 从低位回升（近 5 日最低 RSI < 40）
    """
    if len(df) < 10:
        return None

    last = df.iloc[-1]
    required = ["close", "high", "low", "vol", "pct_chg", "rsi14", "vol_ma5"]
    if any(c not in df.columns for c in required):
        return None
    if pd.isna(last["vol_ma5"]) or last["vol_ma5"] == 0 or pd.isna(last["rsi14"]):
        return None

    # 近 5 日有 3 日以上下跌
    recent_5 = df.tail(5)
    down_count = sum(1 for _, r in recent_5.iterrows() if r["pct_chg"] < 0)
    if down_count < 3:
        return None

    # 今日放量长阳
    if last["pct_chg"] < 3:
        return None
    if last["vol"] < last["vol_ma5"] * 1.5:
        return None

    # 收盘价高于近 5 日最高价（强势反转）
    high_5d = df["high"].iloc[-6:-1].max()
    if last["close"] < high_5d:
        return None

    # RSI 曾在低位
    rsi_5d = df["rsi14"].tail(5)
    min_rsi = rsi_5d.min()
    if not pd.isna(min_rsi) and min_rsi > 40:
        return None

    return {
        "signal_type": EntrySignal.REVERSAL,
        "details": {
            "down_days": down_count,
            "reversal_pct": round(last["pct_chg"], 2),
            "volume_ratio": round(last["vol"] / last["vol_ma5"], 2),
            "min_rsi": round(float(min_rsi), 1),
        },
    }


# ── 信号质量评分 ──


def _score_breakout(df: pd.DataFrame, signal: dict) -> float:
    """突破型评分 (0-100)"""
    last = df.iloc[-1]
    score = 0

    breakout_pct = signal["details"].get("breakout_pct", 0)
    if breakout_pct > 3:
        score += 25
    elif breakout_pct > 1.5:
        score += 18
    elif breakout_pct > 0:
        score += 12
    else:
        score += 6  # 贴近未突破

    vol_ratio = signal["details"].get("volume_ratio", 1)
    if vol_ratio > 2.5:
        score += 25
    elif vol_ratio > 2.0:
        score += 20
    elif vol_ratio > 1.5:
        score += 14
    elif vol_ratio > 1.2:
        score += 8

    if all(c in df.columns for c in ["ma60"]):
        if not pd.isna(last["ma60"]) and last["ma5"] > last["ma10"] > last["ma20"] > last["ma60"]:
            score += 25
        elif last["ma5"] > last["ma10"] > last["ma20"]:
            score += 15
    else:
        score += 10

    if all(c in df.columns for c in ["dif", "dea"]):
        if not pd.isna(last["dif"]) and last["dif"] > last["dea"] and last["dif"] > 0:
            score += 15
        elif not pd.isna(last["dif"]) and last["dif"] > last["dea"]:
            score += 8

    if "rsi14" in df.columns and not pd.isna(last["rsi14"]):
        if 50 <= last["rsi14"] <= 75:
            score += 10
        elif 45 <= last["rsi14"] <= 80:
            score += 5

    return min(score, 100)


def _score_breakout_confirm(df: pd.DataFrame, signal: dict) -> float:
    """突破确认型评分 (0-100)"""
    last = df.iloc[-1]
    score = 0

    vol_ratio = signal["details"].get("volume_ratio", 1)
    if vol_ratio > 2.0:
        score += 25
    elif vol_ratio > 1.5:
        score += 18
    elif vol_ratio > 1.0:
        score += 10

    confirm_pct = signal["details"].get("confirm_pct", 0)
    if confirm_pct > 3:
        score += 20
    elif confirm_pct > 1:
        score += 14
    elif confirm_pct > 0:
        score += 8

    if all(c in df.columns for c in ["ma60"]):
        if not pd.isna(last["ma60"]) and last["ma5"] > last["ma10"] > last["ma20"] > last["ma60"]:
            score += 25
        elif last["ma5"] > last["ma10"] > last["ma20"]:
            score += 15

    if all(c in df.columns for c in ["dif", "dea"]):
        if not pd.isna(last["dif"]) and last["dif"] > last["dea"] and last["dif"] > 0:
            score += 15
        elif not pd.isna(last["dif"]) and last["dif"] > last["dea"]:
            score += 8

    if "rsi14" in df.columns and not pd.isna(last["rsi14"]):
        if 50 <= last["rsi14"] <= 70:
            score += 15
        elif 45 <= last["rsi14"] <= 75:
            score += 8

    return min(score, 100)


def _score_pullback(df: pd.DataFrame, signal: dict) -> float:
    """回踩型评分 (0-100)"""
    last = df.iloc[-1]
    score = 0

    shrink = signal["details"].get("volume_shrink_ratio", 1)
    bouncing = signal["details"].get("bouncing", False)
    if bouncing:
        score += 20  # 已开始反弹加分
    elif shrink < 0.5:
        score += 30
    elif shrink < 0.6:
        score += 22
    elif shrink < 0.7:
        score += 15
    elif shrink < 0.9:
        score += 8

    support_ma = signal["details"].get("support_ma", "")
    if support_ma == "MA20":
        score += 18
    elif support_ma == "MA10":
        score += 12

    if all(c in df.columns for c in ["ma60"]):
        if not pd.isna(last["ma60"]) and last["ma5"] > last["ma10"] > last["ma20"] > last["ma60"]:
            score += 25
        elif last["ma5"] > last["ma10"] > last["ma20"]:
            score += 15

    if all(c in df.columns for c in ["dif", "dea"]):
        if not pd.isna(last["dif"]) and last["dif"] > 0:
            score += 10
        elif not pd.isna(last["dif"]) and last["dif"] > last["dea"]:
            score += 5

    if "rsi14" in df.columns and not pd.isna(last["rsi14"]):
        if 40 <= last["rsi14"] <= 55:
            score += 15
        elif 35 <= last["rsi14"] <= 65:
            score += 8

    return min(score, 100)


def _score_launch(df: pd.DataFrame, signal: dict) -> float:
    """启动型评分 (0-100)"""
    last = df.iloc[-1]
    score = 0
    details = signal.get("details", {})

    range_pct = details.get("price_range_pct", 10)
    if range_pct < 4:
        score += 20
    elif range_pct < 8:
        score += 14
    elif range_pct < 12:
        score += 8

    vol_ratio = details.get("volume_ratio_today", 1)
    if vol_ratio > 2.5:
        score += 25
    elif vol_ratio > 1.8:
        score += 18
    elif vol_ratio > 1.2:
        score += 10

    if all(c in df.columns for c in ["ma60"]):
        if not pd.isna(last["ma60"]) and last["ma5"] > last["ma20"] > last["ma60"]:
            score += 20
        elif last["ma5"] > last["ma20"]:
            score += 10

    if all(c in df.columns for c in ["dif", "dea"]):
        if not pd.isna(last["dif"]) and last["dif"] > 0:
            score += 15
        elif not pd.isna(last["dif"]):
            score += 5

    # MACD 金叉或 DIF 上翘加分
    if "dif" in df.columns and "dea" in df.columns and len(df) >= 3:
        if not pd.isna(last["dif"]) and not pd.isna(df.iloc[-3]["dif"]):
            if last["dif"] > df.iloc[-3]["dif"]:  # DIF 上翘
                score += 10

    return min(score, 100)


def _score_reversal(df: pd.DataFrame, signal: dict) -> float:
    """反转型评分 (0-100)"""
    last = df.iloc[-1]
    score = 0

    reversal_pct = signal["details"].get("reversal_pct", 0)
    if reversal_pct > 5:
        score += 25
    elif reversal_pct > 3:
        score += 18

    vol_ratio = signal["details"].get("volume_ratio", 1)
    if vol_ratio > 3:
        score += 25
    elif vol_ratio > 2:
        score += 18
    elif vol_ratio > 1.5:
        score += 12

    min_rsi = signal["details"].get("min_rsi", 50)
    if min_rsi < 25:
        score += 20  # 极度超卖反转
    elif min_rsi < 30:
        score += 15
    elif min_rsi < 35:
        score += 10

    # 站上 MA5 加分
    if "ma5" in df.columns and not pd.isna(last["ma5"]):
        if last["close"] > last["ma5"]:
            score += 10

    # 量能超过近 20 日最大量
    if len(df) >= 20:
        vol_20d_max = df["vol"].iloc[-21:-1].max()
        if last["vol"] > vol_20d_max:
            score += 10

    return min(score, 100)


# ── 供需分析评分 ──


def score_supply_demand(df: pd.DataFrame) -> float:
    """供需评分 (0-100)，窗口从 10 日扩展到 20 日

    维度：
    - 上涨日 vs 下跌日量比 (40): 需求主导度
    - 量能收缩/扩张 (25): 供需变化趋势
    - 量价配合 (20): 上涨放量+下跌缩量
    - 经典量价形态 (15): 缩量横盘→放量、底部量堆
    """
    score = 0

    if len(df) < 10:
        return 50.0

    # 使用 20 日窗口（如果数据够）
    window = min(20, len(df))
    recent = df.tail(window)
    last = df.iloc[-1]

    # 上涨日 vs 下跌日量比 (40)
    up_days = recent[recent["pct_chg"] > 0]
    down_days = recent[recent["pct_chg"] <= 0]

    if len(up_days) > 0 and len(down_days) > 0:
        avg_up_vol = up_days["vol"].mean()
        avg_down_vol = down_days["vol"].mean()
        if avg_down_vol > 0:
            ratio = avg_up_vol / avg_down_vol
            if ratio > 2.0:
                score += 40
            elif ratio > 1.5:
                score += 30
            elif ratio > 1.2:
                score += 20
            elif ratio > 1.0:
                score += 10
    elif len(up_days) > len(down_days):
        score += 30

    # 量能收缩/扩张 (25)
    vol_ma_col = "vol_ma10" if "vol_ma10" in df.columns else "vol_ma5"
    if vol_ma_col in df.columns and not pd.isna(last[vol_ma_col]) and last[vol_ma_col] > 0:
        recent_3_vol = recent["vol"].tail(3).mean()
        shrink_ratio = recent_3_vol / last[vol_ma_col]

        if shrink_ratio < 0.4:
            score += 25
        elif shrink_ratio < 0.6:
            score += 18
        elif shrink_ratio < 0.8:
            score += 10
        elif shrink_ratio > 1.5:
            score += 20
        elif shrink_ratio > 1.2:
            score += 12

    # 量价配合 (20)
    if _check_bullish_volume_divergence(df):
        score += 20

    # 经典量价形态 (15)
    if _detect_volume_pattern(df) > 0:
        score += 15

    return min(score, 100)


def _detect_volume_pattern(df: pd.DataFrame) -> int:
    """检测经典量价形态

    返回形态强度 (0-2):
    1 = 缩量横盘后首日放量
    2 = 底部量堆（近 20 日出现量堆+价格企稳）
    """
    if len(df) < 20:
        return 0

    strength = 0

    # 形态 1：缩量横盘后首日放量
    vol_ma_col = "vol_ma10" if "vol_ma10" in df.columns else "vol_ma5"
    if vol_ma_col in df.columns:
        recent_10 = df.iloc[-11:-1]
        had_dry_up = any(
            row["vol"] < row[vol_ma_col] * 0.5
            for _, row in recent_10.iterrows()
            if not pd.isna(row[vol_ma_col]) and row[vol_ma_col] > 0
        )
        last = df.iloc[-1]
        vol_expanding = (
            not pd.isna(last[vol_ma_col])
            and last[vol_ma_col] > 0
            and last["vol"] > last[vol_ma_col] * 1.3
            and last["pct_chg"] > 0
        )
        if had_dry_up and vol_expanding:
            strength = max(strength, 1)

    # 形态 2：底部量堆
    if len(df) >= 20:
        vol_20d = df["vol"].tail(20)
        vol_mean = vol_20d.mean()
        if vol_mean > 0:
            vol_max = vol_20d.max()
            vol_min = vol_20d.min()
            # 量堆：存在量能是均量 2 倍以上的日子，且价格没有大幅下跌
            high_vol_days = sum(1 for v in vol_20d if v > vol_mean * 2)
            price_change = (df["close"].iloc[-1] - df["close"].iloc[-20]) / df["close"].iloc[-20] * 100
            if high_vol_days >= 2 and price_change > -3:
                strength = max(strength, 2)

    return strength


def analyze_volume_pattern(df: pd.DataFrame) -> dict:
    """量价模式分析"""
    default = {"demand_supply_ratio": 1.0, "volume_trend": "stable", "accumulation_score": 0}

    if len(df) < 10:
        return default

    window = min(20, len(df))
    recent = df.tail(window)

    up_days = recent[recent["pct_chg"] > 0]
    down_days = recent[recent["pct_chg"] <= 0]
    avg_up_vol = up_days["vol"].mean() if len(up_days) > 0 else 0
    avg_down_vol = down_days["vol"].mean() if len(down_days) > 0 else 1
    ratio = round(avg_up_vol / max(avg_down_vol, 1), 2)

    if len(df) >= 10:
        half = window // 2
        vol_first = df["vol"].iloc[-window:-half].mean()
        vol_recent = df["vol"].tail(half).mean()
        if vol_first > 0:
            trend_ratio = vol_recent / vol_first
            trend = "expanding" if trend_ratio > 1.2 else ("contracting" if trend_ratio < 0.8 else "stable")
        else:
            trend = "stable"
    else:
        trend = "stable"

    accumulation = 0
    if ratio > 1.5:
        accumulation += 40
    if trend == "expanding" and df.iloc[-1]["pct_chg"] > 0:
        accumulation += 30
    if _check_volume_dry_up_before_rally(df):
        accumulation += 30

    return {
        "demand_supply_ratio": ratio,
        "volume_trend": trend,
        "accumulation_score": accumulation,
    }


# ── 辅助函数 ──


def _check_higher_highs(df: pd.DataFrame) -> bool:
    if len(df) < 20:
        return False
    recent = df.tail(20)
    highs = recent["high"].values
    first_half_max = highs[:10].max()
    second_half_max = highs[10:].max()
    return second_half_max >= first_half_max


def _check_bullish_volume_divergence(df: pd.DataFrame) -> bool:
    if len(df) < 6:
        return False
    recent = df.tail(5)
    up_days = recent[recent["pct_chg"] > 0]
    down_days = recent[recent["pct_chg"] <= 0]
    if len(up_days) == 0 or len(down_days) == 0:
        return len(up_days) > len(down_days)
    avg_up_vol = up_days["vol"].mean()
    avg_down_vol = down_days["vol"].mean()
    return avg_up_vol > avg_down_vol


def _check_volume_dry_up_before_rally(df: pd.DataFrame) -> bool:
    if len(df) < 15:
        return False
    last = df.iloc[-1]
    vol_ma_col = "vol_ma10" if "vol_ma10" in df.columns else "vol_ma5"
    if vol_ma_col not in df.columns:
        return False
    recent_10 = df.tail(10)
    had_dry_up = any(
        row["vol"] < row[vol_ma_col] * 0.5
        for _, row in recent_10.iterrows()
        if not pd.isna(row[vol_ma_col]) and row[vol_ma_col] > 0
    )
    vol_ma_col_5 = "vol_ma5" if "vol_ma5" in df.columns else vol_ma_col
    vol_expanding = (
        not pd.isna(last[vol_ma_col_5])
        and last[vol_ma_col_5] > 0
        and last["vol"] > last[vol_ma_col_5] * 1.2
    )
    return had_dry_up and vol_expanding