Ai Order Flow Analysis

---




name: ai-order-flow-analysis
compatibility: opencode
completeness: 95
content-types:
- code
- guidance
- config
- do-dont
description: '"Provides Analyze order flow to detect market pressure and anticipate
  price moves"'
license: MIT
maturity: stable
metadata:
  domain: trading
  output-format: code
  related-skills: data-order-book, data-validation, exchange-health, exchange-order-execution-api
  role: implementation
  scope: implementation
  triggers: ai order flow analysis, ai-order-flow-analysis, analyze, detect, market
  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:** Extract actionable insights from order book dynamics and trade flow data

**Philosophy:** Order flow reveals hidden liquidity and order imbalances. Prioritize microsecond timing, bid-ask spread analysis, and volume profiling for predictive signals.

## Key Principles

1. **Hidden Liquidity Detection**: Identify hidden orders through trade size and timing patterns
2. **Order Flow Imbalance**: Track buy vs sell pressure at each price level
3. **Liquidity Profiling**: Map liquidity across price levels and time
4. **Trade Size Analysis**: Detect large institutional orders via size anomalies
5. **Time and Sales Processing**: Analyze tick-by-tick data for microstructure signals

## Implementation Guidelines

### Structure
- Core logic: `orderflow/parser.py` - Order flow data parsing
- Analyzer: `orderflow/analyzers.py` - Order flow analysis
- Detector: `orderflow/detectors.py` - Order flow pattern detection
- Config: `config/orderflow_config.yaml` - Order flow parameters

### Patterns to Follow
- Process order book snapshots and trade streams separately
- Calculate cumulative delta over multiple time scales
- Track order flow imbalance with volume-weighted pricing
- Detect order flow divergence from price action

## Adherence Checklist
Before completing your task, verify:
- [ ] Order book imbalance (bid-ask pressure) calculated
- [ ] Cumulative delta tracked over time
- [ ] Hidden liquidity detected via trade size patterns
- [ ] Volume profile constructed at key levels
- [ ] Order flow divergence from price identified



## Code Examples

### Order Book Imbalance Calculator

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

@dataclass
class OrderBook:
    """Order book state at a point in time."""
    timestamp: int
    bids: List[Tuple[float, float]]  # (price, size)
    asks: List[Tuple[float, float]]  # (price, size)

class OrderBookImbalance:
    """Calculate order book imbalance metrics."""
    
    def __init__(self, depth: int = 5):
        self.depth = depth
    
    def calculate_imbalance(self, order_book: OrderBook) -> Dict:
        """Calculate order book imbalance metrics."""
        bids = order_book.bids[:self.depth]
        asks = order_book.asks[:self.depth]
        
        if not bids or not asks:
            return {
                'imbalance': 0.5,
                'bid_volume': 0,
                'ask_volume': 0,
                'imbalance_signed': 0,
                'price_weighted_imbalance': 0
            }
        
        # Calculate volumes
        bid_volume = sum(size for _, size in bids)
        ask_volume = sum(size for _, size in asks)
        total_volume = bid_volume + ask_volume
        
        # Basic imbalance (0 = all bids, 1 = all asks)
        imbalance = ask_volume / total_volume if total_volume > 0 else 0.5
        
        # Signed imbalance (-1 = all bids, +1 = all asks)
        imbalance_signed = (ask_volume - bid_volume) / total_volume if total_volume > 0 else 0
        
        # Price-weighted imbalance (better pressure indicator)
        bid_weighted = sum(price * size for price, size in bids)
        ask_weighted = sum(price * size for price, size in asks)
        price_weighted_imbalance = (ask_weighted - bid_weighted) / (bid_weighted + ask_weighted) if (bid_weighted + ask_weighted) > 0 else 0
        
        return {
            'imbalance': imbalance,
            'imbalance_signed': imbalance_signed,
            'bid_volume': bid_volume,
            'ask_volume': ask_volume,
            'total_volume': total_volume,
            'price_weighted_imbalance': price_weighted_imbalance,
            'bid_ask_ratio': bid_volume / ask_volume if ask_volume > 0 else float('inf')
        }
    
    def calculate_imbalance_change(self, imbalance_series: np.ndarray,
                                  window: int = 10) -> np.ndarray:
        """Calculate change in order book imbalance."""
        if len(imbalance_series) < window:
            return np.zeros_like(imbalance_series)
        
        current = np.array(imbalance_series)
        previous = np.roll(current, window)
        previous[:window] = current[:window]
        
        return current - previous
    
    def calculate_cumulative_delta(self, prices: np.ndarray,
                                  volumes: np.ndarray,
                                  buy_signals: np.ndarray) -> np.ndarray:
        """Calculate cumulative delta (net buying volume)."""
        delta = np.where(buy_signals, volumes, -volumes)
        return np.cumsum(delta)
```

### Trade Flow Analyzer

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

@dataclass
class Trade:
    """Single trade record."""
    timestamp: int
    price: float
    size: float
    side: str  # 'buy' or 'sell'

class TradeFlowAnalyzer:
    """Analyze trade flow patterns for market signals."""
    
    def __init__(self, high_frequency_threshold: float = 1000):
        """High frequency = trades faster than this threshold (ms)."""
        self.high_freq_threshold = high_frequency_threshold
    
    def detect_large_liquidity_orders(self, trades: List[Trade],
                                     volume_threshold: float = 3.0) -> List[Dict]:
        """Detect large hidden liquidity orders."""
        if not trades:
            return []
        
        volumes = [t.size for t in trades]
        avg_volume = np.mean(volumes)
        std_volume = np.std(volumes)
        
        large_orders = []
        for trade in trades:
            z_score = (trade.size - avg_volume) / (std_volume + 1e-6)
            
            if z_score > volume_threshold:
                large_orders.append({
                    'timestamp': trade.timestamp,
                    'price': trade.price,
                    'size': trade.size,
                    'z_score': z_score,
                    'side': trade.side,
                    'type': 'large_liquidity'
                })
        
        return large_orders
    
    def detect_aggressive_trades(self, trades: List[Trade],
                                spread: float = 0.01) -> List[Dict]:
        """Detect aggressive trades that hit posted orders."""
        if not trades or len(trades) < 2:
            return []
        
        aggressive = []
        
        for i in range(1, len(trades)):
            prev_trade = trades[i-1]
            curr_trade = trades[i]
            
            # Check if trade crossed spread aggressively
            if curr_trade.side == 'buy' and curr_trade.price >= prev_trade.price + spread:
                aggressive.append({
                    'timestamp': curr_trade.timestamp,
                    'price': curr_trade.price,
                    'size': curr_trade.size,
                    'side': 'buy',
                    'type': 'aggressive_buy'
                })
            elif curr_trade.side == 'sell' and curr_trade.price <= prev_trade.price - spread:
                aggressive.append({
                    'timestamp': curr_trade.timestamp,
                    'price': curr_trade.price,
                    'size': curr_trade.size,
                    'side': 'sell',
                    'type': 'aggressive_sell'
                })
        
        return aggressive
    
    def calculate_volume_profile(self, trades: List[Trade],
                                price_bins: int = 100) -> Dict:
        """Calculate volume profile across price levels."""
        if not trades:
            return {}
        
        prices = [t.price for t in trades]
        volumes = [t.size for t in trades]
        
        min_price = min(prices)
        max_price = max(prices)
        bin_size = (max_price - min_price) / price_bins
        
        volume_profile = np.zeros(price_bins)
        
        for price, volume in zip(prices, volumes):
            bin_idx = int((price - min_price) / bin_size)
            bin_idx = min(bin_idx, price_bins - 1)
            volume_profile[bin_idx] += volume
        
        return {
            'min_price': min_price,
            'max_price': max_price,
            'bin_size': bin_size,
            'volume_profile': volume_profile,
            'price_levels': np.linspace(min_price, max_price, price_bins),
            'total_volume': sum(volumes)
        }
    
    def detect_volume_spike(self, volumes: np.ndarray,
                           window: int = 60,
                           threshold: float = 2.0) -> List[Dict]:
        """Detect volume spikes relative to recent average."""
        if len(volumes) < window:
            return []
        
        # Rolling statistics
        rolling_mean = pd.Series(volumes).rolling(window).mean().fillna(np.mean(volumes))
        rolling_std = pd.Series(volumes).rolling(window).std().fillna(np.std(volumes))
        
        spikes = []
        for i in range(len(volumes)):
            if rolling_std[i] > 0:
                z_score = (volumes[i] - rolling_mean[i]) / rolling_std[i]
                
                if z_score > threshold:
                    spikes.append({
                        'timestamp': i,
                        'volume': volumes[i],
                        'z_score': z_score,
                        'rolling_mean': rolling_mean[i],
                        'rolling_std': rolling_std[i]
                    })
        
        return spikes
```

### Hidden Liquidity Detector

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

@dataclass
class Trade:
    timestamp: int
    price: float
    size: float
    side: str

class HiddenLiquidityDetector:
    """Detect hidden liquidity through trade flow patterns."""
    
    def __init__(self, min_order_size: float = 1000,
                 timing_window: int = 10):
        self.min_order_size = min_order_size
        self.timing_window = timing_window
    
    def detect_iceberg_orders(self, trades: List[Trade]) -> List[Dict]:
        """Detect iceberg orders (large orders broken into smaller pieces)."""
        if len(trades) < 3:
            return []
        
        # Filter for large individual trades
        large_trades = [t for t in trades if t.size >= self.min_order_size]
        
        if len(large_trades) < 2:
            return []
        
        # Find clusters of trades at same price
        iceberg_orders = []
        i = 0
        
        while i < len(large_trades):
            current_price = large_trades[i].price
            current_side = large_trades[i].side
            
            # Count consecutive trades at same price
            cluster_size = 1
            cluster_start = i
            cluster_time = large_trades[i].timestamp
            
            while (i + cluster_size < len(large_trades) and
                   large_trades[i + cluster_size].price == current_price and
                   large_trades[i + cluster_size].side == current_side and
                   large_trades[i + cluster_size].timestamp - cluster_start < self.timing_window):
                cluster_size += 1
            
            # Iceberg pattern: multiple trades of similar size
            if cluster_size >= 3:
                sizes = [large_trades[cluster_start + j].size for j in range(cluster_size)]
                avg_size = np.mean(sizes)
                
                # Check if sizes are consistent (iceberg signature)
                size_variance = np.var(sizes)
                
                if size_variance < avg_size * 0.5:  # Low variance relative to size
                    iceberg_orders.append({
                        'start_time': large_trades[cluster_start].timestamp,
                        'end_time': large_trades[cluster_start + cluster_size - 1].timestamp,
                        'price': current_price,
                        'side': current_side,
                        'total_volume': sum(sizes),
                        'avg_chunk_size': avg_size,
                        'chunk_count': cluster_size,
                        'estimate_total': avg_size * 5  # Assume 5 chunks typically
                    })
            
            i += max(1, cluster_size)
        
        return iceberg_orders
    
    def detect_sweep_orders(self, trades: List[Trade],
                           book_levels: int = 5) -> List[Dict]:
        """Detect sweep orders that take out multiple liquidity levels."""
        if len(trades) < 2:
            return []
        
        sweeps = []
        
        for i in range(1, len(trades)):
            prev = trades[i-1]
            curr = trades[i]
            
            # Check if price moved against expected direction
            if curr.side == 'buy' and curr.price > prev.price * 1.001:
                # Price surged up - likely took out ask levels
                sweeps.append({
                    'timestamp': curr.timestamp,
                    'type': 'sweep_up',
                    'price': curr.price,
                    'size': curr.size,
                    'direction': 'up',
                    'confidence': 0.8
                })
            elif curr.side == 'sell' and curr.price < prev.price * 0.999:
                # Price dropped - likely took out bid levels
                sweeps.append({
                    'timestamp': curr.timestamp,
                    'type': 'sweep_down',
                    'price': curr.price,
                    'size': curr.size,
                    'direction': 'down',
                    'confidence': 0.8
                })
        
        return sweeps
    
    def detect_absorption(self, trades: List[Trade],
                         price_support: float = None,
                         price_resistance: float = None) -> List[Dict]:
        """Detect absorption (large orders absorbing liquidity at support/resistance)."""
        if not trades:
            return []
        
        absptions = []
        
        for i in range(1, len(trades)):
            curr = trades[i]
            prev = trades[i-1]
            
            # Check for absorption at support
            if price_support and abs(curr.price - price_support) < 0.01:
                if curr.side == 'buy' and curr.size > np.mean([t.size for t in trades]):
                    absptions.append({
                        'timestamp': curr.timestamp,
                        'type': 'absorption_at_support',
                        'price': curr.price,
                        'size': curr.size,
                        'support_level': price_support
                    })
            
            # Check for absorption at resistance
            if price_resistance and abs(curr.price - price_resistance) < 0.01:
                if curr.side == 'sell' and curr.size > np.mean([t.size for t in trades]):
                    absptions.append({
                        'timestamp': curr.timestamp,
                        'type': 'absorption_at_resistance',
                        'price': curr.price,
                        'size': curr.size,
                        'resistance_level': price_resistance
                    })
        
        return absptions
```

### Order Flow Divergence Detector

```python
import numpy as np
import pandas as pd
from typing import List, Dict

class OrderFlowDivergenceDetector:
    """Detect divergence between order flow and price action."""
    
    def __init__(self, window: int = 20):
        self.window = window
    
    def calculate_cumulative_volume_delta(self, prices: np.ndarray,
                                         volumes: np.ndarray,
                                         buy_signals: np.ndarray) -> np.ndarray:
        """Calculate cumulative volume delta."""
        delta = np.where(buy_signals, volumes, -volumes)
        return np.cumsum(delta)
    
    def detect_divergence(self, prices: np.ndarray,
                         cumulative_delta: np.ndarray) -> List[Dict]:
        """Detect bullish and bearish divergences."""
        if len(prices) < 2 * self.window:
            return []
        
        divergences = []
        
        # Calculate price and delta trends
        price_returns = np.diff(prices)
        delta_changes = np.diff(cumulative_delta)
        
        for i in range(self.window, len(prices) - self.window):
            # Price high/low
            price_window = prices[i-self.window:i+self.window]
            current_price = prices[i]
            
            local_max = np.max(price_window)
            local_min = np.min(price_window)
            
            # Delta high/low
            delta_window = cumulative_delta[i-self.window:i+self.window]
            current_delta = cumulative_delta[i]
            
            local_delta_max = np.max(delta_window)
            local_delta_min = np.min(delta_window)
            
            # Bullish divergence: price lower, delta higher
            if current_price < local_min and current_delta > local_delta_min * 0.9:
                divergences.append({
                    'timestamp': i,
                    'type': 'bullish',
                    'price_level': current_price,
                    'delta_level': current_delta,
                    'price_min': local_min,
                    'delta_min': local_delta_min,
                    'signal': 'potential_bottom'
                })
            
            # Bearish divergence: price higher, delta lower
            if current_price > local_max * 0.99 and current_delta < local_delta_max * 0.9:
                divergences.append({
                    'timestamp': i,
                    'type': 'bearish',
                    'price_level': current_price,
                    'delta_level': current_delta,
                    'price_max': local_max,
                    'delta_max': local_delta_max,
                    'signal': 'potential_top'
                })
        
        return divergences
    
    def calculate_order_flow_strength(self, prices: np.ndarray,
                                     volumes: np.ndarray,
                                     buy_signals: np.ndarray) -> np.ndarray:
        """Calculate order flow strength indicator."""
        delta = np.where(buy_signals, volumes, -volumes)
        
        # Rolling statistics
        delta_sma = pd.Series(delta).rolling(self.window).mean().fillna(0)
        delta_std = pd.Series(delta).rolling(self.window).std().fillna(1)
        
        # Strength = delta / (std * volume)
        strength = delta_sma / (delta_std * np.sqrt(volumes) + 1e-6)
        
        return strength
```

---

---

## Constraints

### MUST DO
- Validate input feature distributions against training data baselines; flag drift exceeding 2 standard deviations
- Implement model versioning with reproducibility tags — every prediction must be traceable to the exact model artifact and config
- Include confidence intervals or probability estimates alongside all point predictions, never return raw scores without context
- Log all model inputs, outputs, and metadata to enable post-hoc analysis of prediction failures
- Implement feature computation consistently between training and inference — use the same transformation pipeline for both

### MUST NOT DO
- Do not train models on look-ahead biased features (e.g., using future prices or events in training data)
- Avoid deploying a new model version without shadow-testing against the current production model first
- Never retrain a model on a data window that includes regime changes without explicit regime-aware validation
- Do not use accuracy as the primary metric for imbalanced datasets — use precision/recall, F1, or AUC-ROC
- Avoid hardcoding feature names; load them from a schema or config file to prevent mismatches between training and inference


## 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.

- [Order Flow Analysis Tutorial](https://docs.quantconnect.com/tutorials/order-flow-analysis)
- [Volume Profile and Order Flow Concepts](https://www.tradingview.com/wiki/Order_Flow)
- [Microstructure Theory Overview](https://en.wikipedia.org/wiki/Market_microstructure)
- [Large Trade Detection Methods](https://arxiv.org/abs/1805.01469)
- [Order Flow Imbalance Signals](https://www.investopedia.com/terms/o/order-flow.asp)