Update volume_analysis.py

volume_analysis.py

@@ -1,10 +1,14 @@
-import pandas as pd
-import numpy as np
 from typing import Dict, Any
+
+import numpy as np
+import pandas as pd
+
 from config import (
-    VOLUME_SPIKE_MULTIPLIER,
     VOLUME_MA_PERIOD,
-    CLIMAX_VOLUME_MULTIPLIER,
+    VOLUME_SPIKE_MULT,
+    VOLUME_CLIMAX_MULT,
+    VOLUME_WEAK_THRESHOLD,
+    BREAKOUT_LOOKBACK,
 )
 
 
@@ -12,93 +16,105 @@ def compute_volume_ma(df: pd.DataFrame, period: int = VOLUME_MA_PERIOD) -> pd.Se
     return df["volume"].rolling(period).mean()
 
 
-def detect_volume_spike(df: pd.DataFrame, period: int = VOLUME_MA_PERIOD) -> pd.Series:
+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_MULTIPLIER)
+    return df["volume"] > vol_ma * VOLUME_SPIKE_MULT
 
 
-def detect_climax_volume(df: pd.DataFrame, period: int = VOLUME_MA_PERIOD) -> pd.Series:
+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 * CLIMAX_VOLUME_MULTIPLIER)
+    return df["volume"] > vol_ma * VOLUME_CLIMAX_MULT
 
 
-def compute_breakout_confirmation(
-    df: pd.DataFrame, lookback: int = VOLUME_MA_PERIOD
-) -> pd.Series:
-    price_high = df["close"].rolling(lookback).max().shift(1)
-    price_low = df["close"].rolling(lookback).min().shift(1)
-    breakout_up = df["close"] > price_high
-    breakout_down = df["close"] < price_low
-    vol_spike = detect_volume_spike(df)
-    confirmed_up = breakout_up & vol_spike
-    confirmed_down = breakout_down & vol_spike
-    confirmation = pd.Series(0, index=df.index)
-    confirmation[confirmed_up] = 1
-    confirmation[confirmed_down] = -1
-    return confirmation
+def compute_obv(df: pd.DataFrame) -> pd.Series:
+    direction = np.sign(df["close"].diff()).fillna(0)
+    return (df["volume"] * direction).cumsum()
 
 
-def compute_obv(df: pd.DataFrame) -> pd.Series:
-    direction = np.sign(df["close"].diff())
-    direction.iloc[0] = 0
-    obv = (df["volume"] * direction).cumsum()
-    return obv
+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_volume_delta_approx(df: pd.DataFrame) -> pd.Series:
+def compute_delta_approx(df: pd.DataFrame) -> pd.Series:
     body = df["close"] - df["open"]
-    wick_range = (df["high"] - df["low"]).replace(0, np.nan)
-    buy_ratio = (body / wick_range).clip(0, 1).fillna(0.5)
-    buy_volume = df["volume"] * buy_ratio
-    sell_volume = df["volume"] * (1 - buy_ratio)
-    delta = buy_volume - sell_volume
-    return delta
+    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 = detect_volume_spike(df, VOLUME_MA_PERIOD)
-    climax = detect_climax_volume(df, VOLUME_MA_PERIOD)
-    breakout = compute_breakout_confirmation(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_volume_delta_approx(df)
+    delta = compute_delta_approx(df)
+    vwap_dev = compute_vwap_deviation(df, VOLUME_MA_PERIOD)
 
-    last_vol = df["volume"].iloc[-1]
-    last_vol_ma = vol_ma.iloc[-1]
-    last_spike = spike.iloc[-1]
-    last_climax = climax.iloc[-1]
-    last_breakout = breakout.iloc[-1]
+    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_slope = (obv.iloc[-1] - obv.iloc[-5]) / (obv.iloc[-5] + 1e-10)
-    delta_sum = delta.iloc[-5:].sum()
+    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:
-        volume_score = 0.4
+        base_score = 0.3
     elif last_spike and last_breakout != 0:
-        volume_score = 1.0
-    elif last_spike:
-        volume_score = 0.7
-    elif vol_ratio > 1.2:
-        volume_score = 0.5
-    elif vol_ratio > 0.8:
-        volume_score = 0.3
+        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:
-        volume_score = 0.1
+        base_score = 0.1
 
-    obv_bonus = 0.1 if obv_slope > 0 else -0.1
-    volume_score = float(np.clip(volume_score + obv_bonus, 0.0, 1.0))
+    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": vol_ratio,
-        "spike": bool(last_spike),
-        "climax": bool(last_climax),
-        "breakout": int(last_breakout),
-        "obv_slope": obv_slope,
-        "delta_sum": delta_sum,
-        "volume_score": volume_score,
-        "spike_series": spike,
-        "climax_series": climax,
-        "breakout_series": breakout,
+        "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,
     }