Technical Volume Profile

---




name: technical-volume-profile
compatibility: opencode
completeness: 95
content-types:
- code
- guidance
- config
- do-dont
description: '"Provides Volume analysis techniques for understanding market structure"'
license: MIT
maturity: stable
metadata:
  domain: trading
  output-format: code
  related-skills: fundamentals-trading-plan, technical-cycle-analysis
  role: implementation
  scope: implementation
  triggers: analysis, technical volume profile, technical-volume-profile, techniques,
    understanding
  archetypes:
  - tactical
  anti_triggers:
  - brainstorming
  - vague ideation
  - no risk management
  response_profile:
    verbosity: low
    directive_strength: high
    abstraction_level: operational
version: "1.0.0"




---




**Role:** Interpret volume distribution to identify accumulation, distribution, and liquidity zones

**Philosophy:** Volume confirms price moves; volume profile reveals where smart money is active

## Key Principles

1. **Volume-Price Relationship**: High volume at price levels indicates strong interest
2. **Point of Control**: Highest volume level acts as magnet or barrier
3. **Value Area**: 70% of session volume defines fair price range
4. **Volume Imbalance**: Large one-sided volume suggests trend strength
5. **Volume Divergence**: Price moving on low volume signals weakness

## Implementation Guidelines

### Structure
- Core logic: technical_analysis/volume_profile.py
- Helper functions: technical_analysis/volume_helpers.py
- Tests: tests/test_volume_profile.py

### Patterns to Follow
- Use pandas DataFrame for efficient volume calculations
- Implement multiple volume metrics (VPVR, OBV, Volume MA)
- Support session-based and custom time windows

## Adherence Checklist
Before completing your task, verify:
- [ ] Volume profile recalculates only on new session
- [ ] Volume-based indicators update in real-time
- [ ] Volume spikes are flagged (2x+ average volume)
- [ ] Volume divergence detection runs on all timeframes
- [ ] Session types (regular, extended) are properly handled


Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory.

## Python Implementation

```python
import pandas as pd
import numpy as np
from typing import Dict, List, Tuple
from dataclasses import dataclass
from datetime import datetime

@dataclass
class VolumeProfile:
    """Volume profile for a trading session."""
    session_start: datetime
    session_end: datetime
    point_of_control: float  # Highest volume price level
    value_area: Tuple[float, float]  # 70% volume range
    volume_distribution: Dict[float, float]  # Price -> Volume mapping
    total_volume: float

class VolumeProfileAnalyzer:
    """Analyzes volume distribution and volume-based indicators."""
    
    def __init__(self, tick_size: float = 0.01, session_start: str = "09:30"):
        self.tick_size = tick_size
        self.session_start = session_start
        self.session_end = "16:00"
    
    def calculate_volume_profile(
        self, candles: pd.DataFrame, session_only: bool = True
    ) -> VolumeProfile:
        """Calculate Volume Profile Visible Range (VPVR)."""
        # Filter to session hours if required
        if session_only:
            candles = self._filter_session_hours(candles)
        
        # Bin prices into discrete levels
        price_levels = np.arange(
            candles['low'].min(),
            candles['high'].max() + self.tick_size,
            self.tick_size
        )
        
        # Calculate volume at each price level
        volume_dist = {}
        for _, candle in candles.iterrows():
            # Volume is attributed to all prices the candle traversed
            for price in np.arange(
                candle['low'], candle['high'] + self.tick_size, self.tick_size
            ):
                price_rounded = round(price / self.tick_size) * self.tick_size
                volume_dist[price_rounded] = volume_dist.get(price_rounded, 0) + candle['volume']
        
        # Find Point of Control (POC)
        poc_price = max(volume_dist, key=volume_dist.get)
        
        # Calculate Value Area (70% of total volume)
        sorted_levels = sorted(volume_dist.items(), key=lambda x: x[1], reverse=True)
        cumulative_volume = 0
        total_volume = sum(volume_dist.values())
        target_volume = total_volume * 0.7
        
        va_low = sorted_levels[0][0]
        va_high = sorted_levels[0][0]
        
        for price, vol in sorted_levels:
            if cumulative_volume >= target_volume:
                break
            cumulative_volume += vol
            va_low = min(va_low, price)
            va_high = max(va_high, price)
        
        return VolumeProfile(
            session_start=candles.index[0],
            session_end=candles.index[-1],
            point_of_control=poc_price,
            value_area=(va_low, va_high),
            volume_distribution=volume_dist,
            total_volume=total_volume
        )
    
    def calculate_obv(self, candles: pd.DataFrame) -> pd.Series:
        """Calculate On-Balance Volume (OBV)."""
        obv = [0]
        
        for i in range(1, len(candles)):
            if candles['close'].iloc[i] > candles['close'].iloc[i-1]:
                obv.append(obv[-1] + candles['volume'].iloc[i])
            elif candles['close'].iloc[i] < candles['close'].iloc[i-1]:
                obv.append(obv[-1] - candles['volume'].iloc[i])
            else:
                obv.append(obv[-1])
        
        return pd.Series(obv, index=candles.index)
    
    def calculate_volume_ma(self, candles: pd.DataFrame, period: int = 20) -> pd.Series:
        """Calculate Volume Moving Average."""
        return candles['volume'].rolling(window=period).mean()
    
    def detect_volume_spikes(self, candles: pd.DataFrame, threshold: float = 2.0) -> List[datetime]:
        """Detect unusual volume spikes."""
        vol_ma = self.calculate_volume_ma(candles)
        spikes = []
        
        for i in range(len(candles)):
            if candles['volume'].iloc[i] > threshold * vol_ma.iloc[i]:
                spikes.append(candles.index[i])
        
        return spikes
    
    def identify_volume_divergence(
        self, candles: pd.DataFrame, period: int = 20
    ) -> List[Dict]:
        """Identify volume-price divergences."""
        divergences = []
        prices = candles['close'].values
        volumes = candles['volume'].values
        
        for i in range(period, len(candles)):
            # Price higher, volume lower = bearish divergence
            if prices[i] > prices[i-period] and volumes[i] < volumes[i-period]:
                divergences.append({
                    'type': 'bearish',
                    'date': candles.index[i],
                    'price_change': (prices[i] - prices[i-period]) / prices[i-period] * 100,
                    'volume_change': (volumes[i] - volumes[i-period]) / volumes[i-period] * 100
                })
            
            # Price lower, volume higher = bullish divergence
            elif prices[i] < prices[i-period] and volumes[i] > volumes[i-period]:
                divergences.append({
                    'type': 'bullish',
                    'date': candles.index[i],
                    'price_change': (prices[i] - prices[i-period]) / prices[i-period] * 100,
                    'volume_change': (volumes[i] - volumes[i-period]) / volumes[i-period] * 100
                })
        
        return divergences
    
    def _filter_session_hours(self, candles: pd.DataFrame) -> pd.DataFrame:
        """Filter candles to regular trading hours."""
        candles_copy = candles.copy()
        candles_copy['hour'] = candles_copy.index.hour
        candles_copy['minute'] = candles_copy.index.minute
        
        # Regular session: 9:30 AM to 4:00 PM
        start_hour, start_min = 9, 30
        end_hour, end_min = 16, 0
        
        mask = (
            ((candles_copy['hour'] == start_hour) & (candles_copy['minute'] >= start_min)) |
            (candles_copy['hour'] > start_hour)
        ) & (
            ((candles_copy['hour'] == end_hour) & (candles_copy['minute'] <= end_min)) |
            (candles_copy['hour'] < end_hour)
        )
        
        return candles_copy[mask]
```

---

---



### Pattern 2: Risk-Managed Trading Logic with Validation

```python
from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import Optional


logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class TradeSignal:
    """Immutable trade signal with all required validation constraints."""
    symbol: str
    side: str  # "buy" or "sell"
    price: float
    quantity: float
    confidence: float  # 0.0 to 1.0
    reason: str

    def validate(self) -> bool:
        """Validate that the trade signal meets all business constraints."""
        if self.quantity <= 0:
            raise ValueError(f"Quantity must be positive, got {self.quantity}")
        if self.price <= 0:
            raise ValueError(f"Price must be positive, got {self.price}")
        if not 0.0 <= self.confidence <= 1.0:
            raise ValueError(f"Confidence must be between 0 and 1, got {self.confidence}")
        return True


def generate_trade_signal(
    symbol: str,
    side: str,
    price: float,
    quantity: float,
    confidence: float,
    reason: str,
) -> TradeSignal:
    """Generate a validated trade signal with guard clause checks."""
    if side not in ("buy", "sell"):
        raise ValueError(f"Invalid side '{side}', must be 'buy' or 'sell'")

    signal = TradeSignal(
        symbol=symbol,
        side=side,
        price=price,
        quantity=quantity,
        confidence=confidence,
        reason=reason,
    )
    signal.validate()
    logger.info("Trade signal generated: %s %s %.4f @ %.2f (confidence=%.2f)",
                 symbol, side, quantity, price, confidence)
    return signal


def execute_with_risk_check(signal: TradeSignal, max_position_pct: float = 0.05) -> dict:
    """Execute a trade signal after applying risk management checks."""
    adjusted_quantity = signal.quantity
    if signal.side == "buy" and signal.quantity > max_position_pct:
        logger.warning("Position %s exceeds max %.1f%% — capping to %.4f",
                        signal.symbol, max_position_pct * 100, max_position_pct)
        adjusted_quantity = max_position_pct

    return {
        "symbol": signal.symbol,
        "side": signal.side,
        "price": signal.price,
        "quantity": adjusted_quantity,
        "capped": adjusted_quantity < signal.quantity,
        "confidence": signal.confidence,
        "status": "submitted",
    }
```

## Constraints

### MUST DO
- Implement indicator calculations using rolling windows with explicit lookback periods; never use full-history data for online indicators
- Validate signal generation by confirming alignment across multiple independent indicators before acting on a single signal
- Calculate all price-based indicators (SMA, EMA, RSI) on closing prices unless specifically designed for tick data
- Include proper handling of missing/NaN candles in indicator pipelines — forward-fill only within session boundaries
- Log signal generation with the full context window of indicator values that led to each signal

### MUST NOT DO
- Do not use look-ahead bias: never reference future bars or prices when calculating indicators during backtesting
- Avoid recalculating all indicators from scratch on every tick — maintain running state for efficiency
- Never combine indicators with different timeframes without explicit resampling and clear documentation of the alignment logic
- Do not generate signals based on a single indicator crossover; require confirmation from price action or volume
- Avoid hardcoding parameter values (e.g., RSI period = 14) without testing regime-specific optima


## Live References

> Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.

- [Volume Profile Guide](https://www.investopedia.com/articles/trading/11/volume-profile.asp)
- [Point of Control and Value Area](https://www.tradingview.com/wiki/Volume_Profile)
- [Volume Profile Trading Strategies](https://en.wikipedia.org/wiki/Volume_profile_(trading))
- [Footprint Charts and Order Flow](https://www.tradingview.com/wiki/Order_Flow)
- [Volume Analysis in Technical Trading](https://www.investopedia.com/terms/v/volume.asp)