poc-bounce-strategy

# POC Bounce Strategy Complete institutional-grade volume profile trading system. Trades bounces off recent session POC levels where institutions accumulated positions. Applicable to any instrument/timeframe, optimized for XAUUSD H1/H4 and intraday futures M5–M30. ## When This Skill Triggers - "POC bounce" / "trade the POC" / "volume profile strategy" - "Where are the volume levels?" / "Key levels from volume" - "Naked POC" / "Virgin POC" / "untested POC" - "Value area strategy" / "VAH VAL bounce" / "80% rule" - "Where did institutions trade?" / "Fair value zone" - "Volume profile on gold/forex/ES/NQ" - "What levels should I watch today?" - Any request for volume-based support/resistance identification - Any request to analyze, detect, or trade POC levels ## Quick Reference — For detailed rules, read `references/full-strategy-rules.md` ## Architecture ``` User Query │ ├─[1] DETECT LEVELS → POCDetector │ ├── Session POCs (daily/weekly/monthly) │ ├── Naked POC filter │ ├── POC Cluster detection │ └── Value Area boundaries (VAH/VAL) │ ├─[2] CLASSIFY REGIME → RegimeClassifier │ ├── Balanced (D-shape) → Optimal for bounces │ ├── Trending (P/b-shape) → Trend-direction only │ └── Breakout/Discovery → Skip bounces │ ├─[3] SCORE SETUP → ConfluenceScorer │ └── 10-point scoring system (min 5 to trade) │ ├─[4] CONFIRM ENTRY → BounceConfirmation │ ├── Rejection candle (wick/body ≥ 1.5) │ ├── RSI confirmation │ ├── Volume spike check │ └── Trend alignment │ ├─[5] EXECUTE TRADE → TradeManager │ ├── ATR-based SL/TP/sizing │ ├── Trailing stop behind developing POC │ └── Partial exit at primary target │ └─[6] ROUTE TO SKILLS → SkillRouter └── Recommend linked skills for deeper analysis ``` ## Core Engine ```python import pandas as pd, numpy as np from dataclasses import dataclass, field from typing import List, Optional, Dict, Tuple from enum import Enum # ============================================================ # DATA STRUCTURES # ============================================================ class Regime(Enum): BALANCED = "balanced" # D-shape, optimal for bounces TRENDING_UP = "trending_up" # P-shape, long bounces only TRENDING_DN = "trending_dn" # b-shape, short bounces only BREAKOUT = "breakout" # Skip all bounce setups class BounceDirection(Enum): LONG = "long" SHORT = "short" @dataclass class POCLevel: price: float zone_high: float # POC + 0.3×ATR zone_low: float # POC - 0.3×ATR volume: float # Total volume at POC bin timeframe: str # "D1", "W1", "MN1", "session" session_date: str # When this POC was created is_naked: bool = True # Never retested = True is_cluster: bool = False # Part of multi-POC cluster cluster_strength: int = 1 # Number of POCs in cluster tested_count: int = 0 # 0 = virgin, 1 = tested once @dataclass class ValueArea: vah: float # Value Area High (70% boundary) val: float # Value Area Low (70% boundary) poc: float # Point of Control total_volume: float timeframe: str session_date: str @dataclass class TradeSetup: direction: BounceDirection poc_level: POCLevel entry_price: float stop_loss: float take_profit_1: float # Primary: next HVN/POC take_profit_2: float # Secondary: opposite VA boundary lot_size: float confluence_score: int grade: str # "A" (≥7), "B" (5-6), "SKIP" (<5) regime: Regime confirmations: List[str] # Which factors confirmed # ============================================================ # 1. POC DETECTOR — Level Detection Engine # ============================================================ class POCDetector: """Builds volume profile from OHLCV data and extracts POC, VA, HVN, LVN.""" @staticmethod def build_volume_profile(df: pd.DataFrame, n_bins: int = 100) -> Dict: """ Build volume profile from OHLCV dataframe. df must have columns: open, high, low, close, volume (tick or real). Returns dict with bins, poc, vah, val, hvn_list, lvn_list. """ price_high = df["high"].max() price_low = df["low"].min() bin_size = (price_high - price_low) / n_bins if bin_size <= 0: return {"error": "Invalid price range"} bins = np.zeros(n_bins) bin_prices = [price_low + (i + 0.5) * bin_size for i in range(n_bins)] for _, row in df.iterrows(): vol = row.get("volume", row.get("tick_volume", 1)) low_bin = max(0, min(int((row["low"] - price_low) / bin_size), n_bins - 1)) high_bin = max(0, min(int((row["high"] - price_low) / bin_size), n_bins - 1)) close_bin = max(0, min(int((row["close"] - price_low) / bin_size), n_bins - 1)) span = high_bin - low_bin + 1 for b in range(low_bin, high_bin + 1): weight = 1.0 / (abs(b - close_bin) + 1) bins[b] += vol * weight / span # POC = bin with max volume poc_idx = int(np.argmax(bins)) poc_price = round(bin_prices[poc_idx], 5) # Value Area = 70% of total volume centered on POC total_vol = bins.sum() target_vol = total_vol * 0.70 va_vol = bins[poc_idx] lo, hi = poc_idx, poc_idx while va_vol < target_vol and (lo > 0 or hi < n_bins - 1): up_vol = bins[hi + 1] if hi + 1 < n_bins else 0 dn_vol = bins[lo - 1] if lo - 1 >= 0 else 0 if up_vol >= dn_vol and hi + 1 < n_bins: hi += 1; va_vol += bins[hi] elif lo - 1 >= 0: lo -= 1; va_vol += bins[lo] else: hi = min(hi + 1, n_bins - 1); va_vol += bins[hi] vah = round(bin_prices[hi], 5) val = round(bin_prices[lo], 5) # HVN/LVN detection mean_vol = bins.mean() std_vol = bins.std() hvn = [round(bin_prices[i], 5) for i in range(n_bins) if bins[i] > mean_vol + 0.5 * std_vol] lvn = [round(bin_prices[i], 5) for i in range(n_bins) if bins[i] < mean_vol - 0.5 * std_vol and bins[i] > 0] return { "poc": poc_price, "vah": vah, "val": val, "hvn_list": hvn, "lvn_list": lvn, "total_volume": round(total_vol, 2), "bin_prices": bin_prices, "bin_volumes": bins.tolist(), } @staticmethod def detect_session_pocs(data_by_session: Dict[str, pd.DataFrame], atr: float, n_bins: int = 100) -> List[POCLevel]: """Detect POC for each session and mark naked/cluster status.""" pocs = [] for session_id, df in data_by_session.items(): if df.empty or len(df) < 10: continue profile = POCDetector.build_volume_profile(df, n_bins) if "error" in profile: continue zone_width = atr * 0.3 pocs.append(POCLevel( price=profile["poc"], zone_high=round(profile["poc"] + zone_width, 5), zone_low=round(profile["poc"] - zone_width, 5), volume=profile["total_volume"], timeframe="session", session_date=session_id, )) # Mark naked POCs (not yet retested by current price) # Caller should update is_naked based on current price history # Detect clusters (POCs within 0.5×ATR of each other) merge_threshold = atr * 0.5 for i in range(len(pocs)): for j in range(i + 1, len(pocs)): if abs(pocs[i].price - pocs[j].price) < merge_threshold: pocs[i].is_cluster = True pocs[j].is_cluster = True pocs[i].cluster_strength += 1 pocs[j].cluster_strength += 1 return pocs @staticmethod def update_naked_status(pocs: List[POCLevel], price_history: pd.Series) -> List[POCLevel]: """Mark POCs as tested if price has visited their zone.""" for poc in pocs: touches = ((price_history >= poc.zone_low) & (price_history <= poc.zone_high)).sum() if touches > 1: # >1 because the original session counts as 1 poc.is_naked = False poc.tested_count = int(touches) - 1 return pocs # ============================================================ # 2. REGIME CLASSIFIER # ============================================================ class RegimeClassifier: """Classifies market regime from profile shape and price action.""" @staticmethod def classify(profile: Dict, df: pd.DataFrame, atr: float) -> Regime: """ Classify regime based on: - Profile shape (distribution of volume) - Price position relative to VA - Recent price action (HH/HL vs LH/LL) """ poc = profile["poc"] vah = profile["vah"] val = profile["val"] va_range = vah - val current = df["close"].iloc[-1] # Check if price is outside all known value areas (breakout) if current > vah + atr or current < val - atr: return Regime.BREAKOUT # Check trend from recent swing structure highs = df["high"].rolling(20).max() lows = df["low"].rolling(20).min() recent_highs = df["high"].tail(40) recent_lows = df["low"].tail(40) # Simple HH/HL check mid = len(recent_highs) // 2 first_half_high = recent_highs.iloc[:mid].max() second_half_high = recent_highs.iloc[mid:].max() first_half_low = recent_lows.iloc[:mid].min() second_half_low = recent_lows.iloc[mid:].min() if second_half_high > first_half_high and second_half_low > first_half_low: return Regime.TRENDING_UP elif second_half_high < first_half_high and second_half_low < first_half_low: return Regime.TRENDING_DN return Regime.BALANCED @staticmethod def is_bounce_allowed(regime: Regime, direction: BounceDirection) -> bool: """Check if bounce direction is compatible with regime.""" if regime == Regime.BREAKOUT: return False if regime == Regime.TRENDING_UP and direction == BounceDirection.SHORT: return False if regime == Regime.TRENDING_DN and direction == BounceDirection.LONG: return False return True # ============================================================ # 3. CONFLUENCE SCORER — 10-Point System # ============================================================ CONFLUENCE_FACTORS = { "naked_poc": {"points": 3, "desc": "Naked (Virgin) POC — first test ever"},