""" ========================================= CV6 AI Trading OS — Volume Analyzer ========================================= Real volume analysis — no fake data. What we detect: • Volume spike — current vs 20-day avg • Price-Volume — breakout confirmed or not • Delivery % — conviction (NSE bhavcopy) • Volume trend — rising / falling / flat • Climax volume — exhaustion signal All data from yfinance (real NSE historical). ========================================= """ from __future__ import annotations import time from dataclasses import dataclass, field from typing import Any, Dict, List, Optional from loguru import logger # ── Cache ───────────────────────────────────────────────────────────────────── _cache: Dict[str, Any] = {} _cache_ts: Dict[str, float] = {} _CACHE_TTL = 300 # 5 minutes def _cached(key: str, fn): now = time.time() if key in _cache and (now - _cache_ts.get(key, 0)) < _CACHE_TTL: return _cache[key] result = fn() if result is not None: _cache[key] = result _cache_ts[key] = now return _cache.get(key) # ── Result dataclass ─────────────────────────────────────────────────────────── @dataclass class VolumeAnalysis: symbol: str current_volume: int = 0 avg_volume_20d: int = 0 volume_ratio: float = 0.0 # current / avg (>2.0 = spike) volume_signal: str = "NORMAL" # SPIKE | HIGH | NORMAL | LOW | CLIMAX price_volume: str = "UNCONFIRMED" # CONFIRMED_UP | CONFIRMED_DOWN | UNCONFIRMED | DIVERGENCE volume_trend: str = "FLAT" # RISING | FALLING | FLAT delivery_pct: Optional[float] = None # % of volume that is delivery (conviction) spike_factor: float = 0.0 # how many σ above mean interpretation: str = "" last_5_volumes: List[int] = field(default_factory=list) last_close: float = 0.0 last_change_pct: float = 0.0 # ── Core analysis ────────────────────────────────────────────────────────────── def analyse_volume(symbol: str, period: str = "30d") -> Optional[VolumeAnalysis]: """ Fetch real NSE volume data via yfinance and analyse. symbol: NSE ticker e.g. "RELIANCE" (no .NS needed — auto-added) """ cache_key = f"vol_{symbol}_{period}" return _cached(cache_key, lambda: _fetch_and_analyse(symbol, period)) def _fetch_and_analyse(symbol: str, period: str) -> Optional[VolumeAnalysis]: try: import yfinance as yf import statistics ticker = symbol.upper() if not ticker.endswith(".NS") and not ticker.endswith(".BO"): ticker_ns = ticker + ".NS" else: ticker_ns = ticker df = yf.download(ticker_ns, period=period, interval="1d", auto_adjust=True, progress=False) if df is None or len(df) < 5: logger.warning(f"[Volume] {symbol}: insufficient data") return None volumes = df["Volume"].dropna().astype(int).tolist() closes = df["Close"].dropna().tolist() opens = df["Open"].dropna().tolist() if len(volumes) < 5: return None current_vol = volumes[-1] recent_20 = volumes[-21:-1] if len(volumes) >= 21 else volumes[:-1] avg_vol = int(sum(recent_20) / len(recent_20)) if recent_20 else current_vol vol_ratio = round(current_vol / avg_vol, 2) if avg_vol > 0 else 1.0 # Standard deviation for spike detection if len(recent_20) >= 5: mean_v = sum(recent_20) / len(recent_20) std_v = statistics.stdev(recent_20) if len(recent_20) > 1 else 1 spike_factor = round((current_vol - mean_v) / std_v, 2) if std_v > 0 else 0.0 else: spike_factor = 0.0 # Volume signal if vol_ratio >= 4.0: vol_signal = "CLIMAX" # exhaustion / major event elif vol_ratio >= 2.5: vol_signal = "SPIKE" # strong unusual activity elif vol_ratio >= 1.5: vol_signal = "HIGH" # above average elif vol_ratio < 0.5: vol_signal = "LOW" # below average else: vol_signal = "NORMAL" # Price-Volume confirmation last_close = float(closes[-1]) prev_close = float(closes[-2]) if len(closes) >= 2 else last_close change_pct = round((last_close - prev_close) / prev_close * 100, 2) if prev_close else 0.0 if change_pct > 0.3 and vol_ratio >= 1.5: pv = "CONFIRMED_UP" # price up + volume up = genuine move elif change_pct < -0.3 and vol_ratio >= 1.5: pv = "CONFIRMED_DOWN" # price down + volume up = genuine selling elif change_pct > 0.3 and vol_ratio < 0.8: pv = "DIVERGENCE" # price up but low volume = weak/fake breakout elif change_pct < -0.3 and vol_ratio < 0.8: pv = "DIVERGENCE" # price down + low volume = probably not sustained else: pv = "UNCONFIRMED" # Volume trend (last 5 days) last5 = volumes[-5:] if len(last5) >= 3: rising = sum(1 for i in range(1, len(last5)) if last5[i] > last5[i-1]) falling = sum(1 for i in range(1, len(last5)) if last5[i] < last5[i-1]) if rising >= 3: vol_trend = "RISING" elif falling >= 3: vol_trend = "FALLING" else: vol_trend = "FLAT" else: vol_trend = "FLAT" # Human-readable interpretation parts = [] if vol_signal == "CLIMAX": parts.append(f"CLIMAX volume ({vol_ratio:.1f}x avg) — possible trend exhaustion") elif vol_signal == "SPIKE": parts.append(f"Volume SPIKE {vol_ratio:.1f}x avg — institutional activity likely") elif vol_signal == "HIGH": parts.append(f"High volume ({vol_ratio:.1f}x avg) — above-average interest") elif vol_signal == "LOW": parts.append(f"Low volume ({vol_ratio:.1f}x avg) — weak conviction") if pv == "CONFIRMED_UP": parts.append("Price ↑ + Volume ↑ → breakout CONFIRMED") elif pv == "CONFIRMED_DOWN": parts.append("Price ↓ + Volume ↑ → breakdown CONFIRMED") elif pv == "DIVERGENCE": parts.append("Price moved but volume weak → signal UNRELIABLE") if vol_trend == "RISING": parts.append("Volume trend RISING — accumulation likely") elif vol_trend == "FALLING": parts.append("Volume trend FALLING — interest waning") interpretation = ". ".join(parts) if parts else f"Normal volume ({vol_ratio:.1f}x avg)" return VolumeAnalysis( symbol = symbol.upper(), current_volume = current_vol, avg_volume_20d = avg_vol, volume_ratio = vol_ratio, volume_signal = vol_signal, price_volume = pv, volume_trend = vol_trend, delivery_pct = None, # NSE bhavcopy requires separate fetch spike_factor = spike_factor, interpretation = interpretation, last_5_volumes = last5, last_close = last_close, last_change_pct = change_pct, ) except Exception as e: logger.error(f"[Volume] {symbol}: {e}") return None def analyse_volume_multi(symbols: List[str]) -> Dict[str, Optional[VolumeAnalysis]]: """Analyse volume for multiple symbols. Returns dict {symbol: VolumeAnalysis}.""" results = {} for sym in symbols: results[sym] = analyse_volume(sym) return results def top_volume_spikes(symbols: List[str], min_ratio: float = 2.0) -> List[VolumeAnalysis]: """ From a list of symbols, return those with volume spike >= min_ratio. Sorted by volume_ratio descending. """ results = [] for sym in symbols: va = analyse_volume(sym) if va and va.volume_ratio >= min_ratio: results.append(va) return sorted(results, key=lambda x: x.volume_ratio, reverse=True) def fmt_volume(va: VolumeAnalysis) -> str: """One-line summary for AI prompt.""" dvol = f" | Delivery N/A" return ( f"{va.symbol}: Vol {va.current_volume:,} " f"({va.volume_ratio:.1f}x avg) | {va.volume_signal} | " f"P-V: {va.price_volume} | Trend: {va.volume_trend}" f"{dvol} | {va.interpretation}" )