volume_analysis.py

from typing import Dict, Any

import numpy as np
import pandas as pd

from config import (
    VOLUME_MA_PERIOD,
    VOLUME_SPIKE_MULT,
    VOLUME_CLIMAX_MULT,
    VOLUME_WEAK_THRESHOLD,
    BREAKOUT_LOOKBACK,
)


def compute_volume_ma(df: pd.DataFrame, period: int = VOLUME_MA_PERIOD) -> pd.Series:
    return df["volume"].rolling(period).mean()


def detect_spikes(df: pd.DataFrame, period: int = VOLUME_MA_PERIOD) -> pd.Series:
    vol_ma = compute_volume_ma(df, period)
    return df["volume"] > vol_ma * VOLUME_SPIKE_MULT


def detect_climax(df: pd.DataFrame, period: int = VOLUME_MA_PERIOD) -> pd.Series:
    vol_ma = compute_volume_ma(df, period)
    return df["volume"] > vol_ma * VOLUME_CLIMAX_MULT


def compute_obv(df: pd.DataFrame) -> pd.Series:
    direction = np.sign(df["close"].diff()).fillna(0)
    return (df["volume"] * direction).cumsum()


def compute_vwap_deviation(df: pd.DataFrame, period: int = VOLUME_MA_PERIOD) -> pd.Series:
    typical = (df["high"] + df["low"] + df["close"]) / 3
    vol = df["volume"]
    cum_vp = (typical * vol).rolling(period).sum()
    cum_vol = vol.rolling(period).sum().replace(0, np.nan)
    vwap = cum_vp / cum_vol
    return (df["close"] - vwap) / vwap


def compute_delta_approx(df: pd.DataFrame) -> pd.Series:
    body = df["close"] - df["open"]
    wick = (df["high"] - df["low"]).replace(0, np.nan)
    buy_ratio = ((body / wick) * 0.5 + 0.5).clip(0.0, 1.0).fillna(0.5)
    buy_vol = df["volume"] * buy_ratio
    sell_vol = df["volume"] * (1 - buy_ratio)
    return buy_vol - sell_vol


def compute_breakout_signal(df: pd.DataFrame, lookback: int = BREAKOUT_LOOKBACK) -> pd.Series:
    prior_high = df["close"].rolling(lookback).max().shift(1)
    prior_low = df["close"].rolling(lookback).min().shift(1)
    spikes = detect_spikes(df)
    signal = pd.Series(0, index=df.index)
    signal[(df["close"] > prior_high) & spikes] = 1
    signal[(df["close"] < prior_low) & spikes] = -1
    return signal


def analyze_volume(df: pd.DataFrame) -> Dict[str, Any]:
    vol_ma = compute_volume_ma(df, VOLUME_MA_PERIOD)
    spike_series = detect_spikes(df, VOLUME_MA_PERIOD)
    climax_series = detect_climax(df, VOLUME_MA_PERIOD)
    breakout_series = compute_breakout_signal(df, BREAKOUT_LOOKBACK)
    obv = compute_obv(df)
    delta = compute_delta_approx(df)
    vwap_dev = compute_vwap_deviation(df, VOLUME_MA_PERIOD)

    last_vol = float(df["volume"].iloc[-1])
    last_vol_ma = float(vol_ma.iloc[-1]) if not np.isnan(vol_ma.iloc[-1]) else 1.0
    last_spike = bool(spike_series.iloc[-1])
    last_climax = bool(climax_series.iloc[-1])
    last_breakout = int(breakout_series.iloc[-1])
    last_vwap_dev = float(vwap_dev.iloc[-1]) if not np.isnan(vwap_dev.iloc[-1]) else 0.0

    vol_ratio = last_vol / last_vol_ma if last_vol_ma > 0 else 1.0

    obv_recent = obv.iloc[-10:]
    obv_slope = float(np.polyfit(range(len(obv_recent)), obv_recent.values, 1)[0])
    obv_normalized = obv_slope / (abs(obv_recent.mean()) + 1e-10)

    delta_sum_5 = float(delta.iloc[-5:].sum())
    delta_sign = 1 if delta_sum_5 > 0 else -1

    weak_vol = vol_ratio < VOLUME_WEAK_THRESHOLD

    if last_climax:
        base_score = 0.3
    elif last_spike and last_breakout != 0:
        base_score = 1.0
    elif last_spike and last_breakout == 0:
        base_score = 0.65
    elif vol_ratio >= 1.2:
        base_score = 0.5
    elif vol_ratio >= 0.8:
        base_score = 0.35
    else:
        base_score = 0.1

    obv_bonus = float(np.clip(obv_normalized * 0.1, -0.1, 0.1))
    vwap_bonus = 0.05 if last_vwap_dev > 0 and last_breakout == 1 else 0.0
    volume_score = float(np.clip(base_score + obv_bonus + vwap_bonus, 0.0, 1.0))

    return {
        "vol_ratio": round(vol_ratio, 3),
        "spike": last_spike,
        "climax": last_climax,
        "weak": weak_vol,
        "breakout": last_breakout,
        "obv_slope_norm": round(obv_normalized, 4),
        "delta_sum_5": round(delta_sum_5, 2),
        "delta_sign": delta_sign,
        "vwap_deviation": round(last_vwap_dev, 4),
        "volume_score": round(volume_score, 4),
        "spike_series": spike_series,
        "climax_series": climax_series,
        "breakout_series": breakout_series,
    }