BlackboxAI
0206ef7cbb
Real-time orderflow trading system with absorption, initiative, sweep, exhaustion, and divergence detection. Features volume profile framing, state machine trade lifecycle, MT5 + Bybit feeds, FastAPI dashboard, and Telegram alerts for 30+ instruments. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
344 lines
14 KiB
Python
344 lines
14 KiB
Python
"""
|
|
Profile Framing — Daily Bias Engine
|
|
Implements Fabio's profile framing methodology for determining directional bias.
|
|
|
|
Core logic:
|
|
1. Build daily cash-session volume profiles
|
|
2. Classify profile shape → P-shape (long), b-shape (short), D (neutral), double (transition)
|
|
3. Track value acceptance/rejection across days
|
|
4. Merge overlapping profiles (2-3 days) for refined VAL/VAH
|
|
5. Detect market shifts: failed auctions, hooks, distribution warnings
|
|
6. Output: daily bias direction + qualified levels for orderflow execution
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from typing import Optional
|
|
|
|
from orderflow_system.data.models import VolumeProfileResult, Side
|
|
from orderflow_system.config.settings import BiasDirection, ProfileShape
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class LevelType(Enum):
|
|
VAH = "vah"
|
|
VAL = "val"
|
|
POC = "poc"
|
|
LVN = "lvn"
|
|
MERGED_VAH = "merged_vah"
|
|
MERGED_VAL = "merged_val"
|
|
|
|
|
|
@dataclass
|
|
class QualifiedLevel:
|
|
"""A price level qualified by profile framing for orderflow execution."""
|
|
price: float
|
|
level_type: LevelType
|
|
direction: Side # Expected trade direction at this level
|
|
strength: float = 0.0 # How many confirmations (rejection days, merges)
|
|
source_dates: list[str] = field(default_factory=list)
|
|
notes: str = ""
|
|
|
|
|
|
@dataclass
|
|
class DailyBias:
|
|
"""Output of the profile framing analysis for the current session."""
|
|
date: str
|
|
direction: BiasDirection = BiasDirection.NEUTRAL
|
|
confidence: float = 0.0 # 0-100
|
|
profile_shape: str = "unknown"
|
|
qualified_levels: list[QualifiedLevel] = field(default_factory=list)
|
|
poc: float = 0.0
|
|
vah: float = 0.0
|
|
val: float = 0.0
|
|
lvn_levels: list[float] = field(default_factory=list)
|
|
merged_vah: Optional[float] = None
|
|
merged_val: Optional[float] = None
|
|
notes: str = ""
|
|
|
|
|
|
class ProfileFramingEngine:
|
|
"""
|
|
Analyzes multi-day volume profiles to determine directional bias
|
|
and qualify key levels for orderflow execution.
|
|
|
|
Fabio's methodology:
|
|
- P-shape profile (POC > 65% position) → buyers in control → bias LONG
|
|
- b-shape profile (POC < 35%) → sellers in control → bias SHORT
|
|
- D-shape → balanced/neutral → fade extremes
|
|
- Profile merging: when days overlap at same level, merge for precision
|
|
- Rejection tracking: 2-3 days rejecting same level → strong wall
|
|
- Market shift detection: accepted value moving direction
|
|
- Failed auction / hook: price tries to break VA boundary, gets rejected
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._profile_history: list[VolumeProfileResult] = []
|
|
self._bias_history: list[DailyBias] = []
|
|
self._max_history = 30
|
|
self._rejection_tracker: dict[str, list[str]] = {}
|
|
# key = 'vah_zone' or 'val_zone', value = list of dates that rejected
|
|
|
|
def add_profile(self, profile: VolumeProfileResult):
|
|
"""Add a daily profile to history."""
|
|
self._profile_history.append(profile)
|
|
if len(self._profile_history) > self._max_history:
|
|
self._profile_history = self._profile_history[-self._max_history:]
|
|
|
|
def analyze(self, current_price: float = 0.0) -> DailyBias:
|
|
"""
|
|
Analyze the most recent profiles to produce a directional bias
|
|
and qualified levels for today's trading.
|
|
"""
|
|
if not self._profile_history:
|
|
return DailyBias(date="unknown")
|
|
|
|
latest = self._profile_history[-1]
|
|
bias = DailyBias(
|
|
date=latest.session_date,
|
|
poc=latest.poc,
|
|
vah=latest.vah,
|
|
val=latest.val,
|
|
lvn_levels=latest.lvn_levels,
|
|
profile_shape=latest.shape,
|
|
)
|
|
|
|
# ── Step 1: Determine direction from profile shape ──
|
|
self._classify_direction(bias, latest)
|
|
|
|
# ── Step 2: Check multi-day context ──
|
|
if len(self._profile_history) >= 2:
|
|
self._check_multi_day_context(bias, current_price)
|
|
|
|
# ── Step 3: Build qualified levels ──
|
|
self._build_qualified_levels(bias, current_price)
|
|
|
|
# ── Step 4: Try multi-day merge for refined levels ──
|
|
if len(self._profile_history) >= 2:
|
|
self._try_merge_profiles(bias)
|
|
|
|
self._bias_history.append(bias)
|
|
if len(self._bias_history) > self._max_history:
|
|
self._bias_history = self._bias_history[-self._max_history:]
|
|
|
|
return bias
|
|
|
|
def _classify_direction(self, bias: DailyBias, profile: VolumeProfileResult):
|
|
"""
|
|
Classify bias from profile shape.
|
|
P-shape = buyers in control = LONG bias
|
|
b-shape = sellers in control = SHORT bias
|
|
"""
|
|
shape = profile.shape
|
|
poc_pct = profile.poc_position_pct
|
|
|
|
if shape == "p_shape":
|
|
bias.direction = BiasDirection.LONG
|
|
bias.confidence = 40 + poc_pct * 30 # Higher POC = stronger
|
|
bias.notes = f"P-shape profile, POC at {poc_pct:.0%} — buyers in control"
|
|
elif shape == "b_shape":
|
|
bias.direction = BiasDirection.SHORT
|
|
bias.confidence = 40 + (1 - poc_pct) * 30
|
|
bias.notes = f"b-shape profile, POC at {poc_pct:.0%} — sellers in control"
|
|
elif shape == "double_dist":
|
|
bias.direction = BiasDirection.NEUTRAL
|
|
bias.confidence = 30
|
|
bias.notes = "Double distribution — transition day, watch for direction"
|
|
else:
|
|
bias.direction = BiasDirection.NEUTRAL
|
|
bias.confidence = 20
|
|
bias.notes = f"D-shape balanced profile, POC at {poc_pct:.0%}"
|
|
|
|
def _check_multi_day_context(self, bias: DailyBias, current_price: float):
|
|
"""
|
|
Check value acceptance/rejection across recent days.
|
|
- Value moving UP across days → strengthen LONG bias
|
|
- Value moving DOWN → strengthen SHORT bias
|
|
- Repeated rejection at same VAH → warning of distribution
|
|
- Failed auction (hook at VA boundary) → continuation setup
|
|
"""
|
|
recent = self._profile_history[-3:] # Last 3 days
|
|
if len(recent) < 2:
|
|
return
|
|
|
|
prev = recent[-2]
|
|
latest = recent[-1]
|
|
|
|
# Value acceptance direction
|
|
poc_shift = latest.poc - prev.poc
|
|
vah_shift = latest.vah - prev.vah
|
|
val_shift = latest.val - prev.val
|
|
|
|
if poc_shift > 0 and vah_shift > 0:
|
|
# Value accepted higher
|
|
if bias.direction == BiasDirection.LONG:
|
|
bias.confidence = min(100, bias.confidence + 15)
|
|
bias.notes += " | Value accepted higher — momentum confirmed"
|
|
elif bias.direction == BiasDirection.NEUTRAL:
|
|
bias.direction = BiasDirection.LONG
|
|
bias.confidence = min(100, bias.confidence + 10)
|
|
elif poc_shift < 0 and val_shift < 0:
|
|
# Value accepted lower
|
|
if bias.direction == BiasDirection.SHORT:
|
|
bias.confidence = min(100, bias.confidence + 15)
|
|
bias.notes += " | Value accepted lower — downtrend confirmed"
|
|
elif bias.direction == BiasDirection.NEUTRAL:
|
|
bias.direction = BiasDirection.SHORT
|
|
bias.confidence = min(100, bias.confidence + 10)
|
|
|
|
# Check for VAH rejection across days (distribution warning)
|
|
if len(recent) >= 2:
|
|
vah_tolerance = (latest.vah - latest.val) * 0.1
|
|
vahs_similar = all(
|
|
abs(p.vah - latest.vah) < vah_tolerance for p in recent[-2:]
|
|
)
|
|
if vahs_similar and latest.shape != "p_shape":
|
|
bias.direction = BiasDirection.WARNING
|
|
bias.confidence = min(100, bias.confidence + 10)
|
|
bias.notes += " | WARNING: VAH rejected for multiple days — possible distribution"
|
|
|
|
# Failed auction detection (hook)
|
|
# Use VA boundaries as proxies since VolumeProfileResult doesn't have high/low
|
|
if current_price > 0:
|
|
# Price is above VAL after a session that traded below it → bullish hook
|
|
if current_price > latest.val and prev.val < latest.val:
|
|
bias.notes += " | Failed auction below VAL — hook setup (bullish)"
|
|
bias.confidence = min(100, bias.confidence + 10)
|
|
# Price is below VAH after a session that traded above it → bearish hook
|
|
elif current_price < latest.vah and prev.vah > latest.vah:
|
|
bias.notes += " | Failed auction above VAH — hook setup (bearish)"
|
|
bias.confidence = min(100, bias.confidence + 10)
|
|
|
|
def _build_qualified_levels(self, bias: DailyBias, current_price: float):
|
|
"""Build the list of qualified levels for orderflow execution."""
|
|
latest = self._profile_history[-1]
|
|
levels = []
|
|
|
|
# VAL — primary support / long entry zone in uptrend
|
|
val_dir = Side.BUY if bias.direction in (BiasDirection.LONG, BiasDirection.NEUTRAL) else Side.SELL
|
|
levels.append(QualifiedLevel(
|
|
price=latest.val,
|
|
level_type=LevelType.VAL,
|
|
direction=val_dir,
|
|
strength=50,
|
|
source_dates=[latest.session_date],
|
|
notes="Value Area Low — fade for longs in uptrend, break confirms short",
|
|
))
|
|
|
|
# VAH — primary resistance / short entry zone in downtrend
|
|
vah_dir = Side.SELL if bias.direction in (BiasDirection.SHORT, BiasDirection.NEUTRAL) else Side.BUY
|
|
levels.append(QualifiedLevel(
|
|
price=latest.vah,
|
|
level_type=LevelType.VAH,
|
|
direction=vah_dir,
|
|
strength=50,
|
|
source_dates=[latest.session_date],
|
|
notes="Value Area High — fade for shorts in downtrend, break confirms long",
|
|
))
|
|
|
|
# POC — fair value / mean reversion target
|
|
levels.append(QualifiedLevel(
|
|
price=latest.poc,
|
|
level_type=LevelType.POC,
|
|
direction=val_dir, # Same as general direction
|
|
strength=30,
|
|
source_dates=[latest.session_date],
|
|
notes="Point of Control — fair value, mean reversion target",
|
|
))
|
|
|
|
# LVN levels — rebalancing magnets / rejection points
|
|
for lvn in latest.lvn_levels:
|
|
# Direction at LVN: price above → expect rejection → SELL; price below → bounce → BUY
|
|
if current_price > 0:
|
|
lvn_dir = Side.SELL if current_price > lvn else Side.BUY
|
|
else:
|
|
lvn_dir = val_dir
|
|
levels.append(QualifiedLevel(
|
|
price=lvn,
|
|
level_type=LevelType.LVN,
|
|
direction=lvn_dir,
|
|
strength=40,
|
|
source_dates=[latest.session_date],
|
|
notes="Low Volume Node — rebalancing pivot, expect rejection",
|
|
))
|
|
|
|
# Strengthen levels that appear across multiple days
|
|
if len(self._profile_history) >= 2:
|
|
prev = self._profile_history[-2]
|
|
tolerance = (latest.vah - latest.val) * 0.05
|
|
for level in levels:
|
|
# Check if level aligns with previous day's levels
|
|
for prev_level in [prev.val, prev.vah, prev.poc]:
|
|
if abs(level.price - prev_level) < tolerance:
|
|
level.strength = min(100, level.strength + 20)
|
|
level.source_dates.append(prev.session_date)
|
|
level.notes += " | Confluent with previous day"
|
|
|
|
bias.qualified_levels = levels
|
|
|
|
def _try_merge_profiles(self, bias: DailyBias):
|
|
"""
|
|
Merge recent profiles if they overlap at similar levels.
|
|
Fabio merges 2-3 day profiles when value areas overlap to get
|
|
more precise VAL/VAH.
|
|
"""
|
|
from orderflow_system.analytics.volume_profile import (
|
|
VolumeProfileEngine,
|
|
VolumeProfileConfig,
|
|
)
|
|
|
|
recent = self._profile_history[-3:]
|
|
if len(recent) < 2:
|
|
return
|
|
|
|
# Check if profiles overlap (value areas intersect)
|
|
latest = recent[-1]
|
|
to_merge = [latest]
|
|
|
|
for prev in recent[:-1]:
|
|
overlap = min(latest.vah, prev.vah) - max(latest.val, prev.val)
|
|
range_avg = ((latest.vah - latest.val) + (prev.vah - prev.val)) / 2
|
|
if range_avg > 0 and overlap / range_avg > 0.3:
|
|
to_merge.append(prev)
|
|
|
|
if len(to_merge) < 2:
|
|
return
|
|
|
|
# Merge the overlapping profiles
|
|
engine = VolumeProfileEngine(VolumeProfileConfig())
|
|
merged = engine.merge_profiles(to_merge)
|
|
|
|
bias.merged_vah = merged.vah
|
|
bias.merged_val = merged.val
|
|
bias.notes += f" | Merged {len(to_merge)}-day profile: VAH={merged.vah:.2f}, VAL={merged.val:.2f}"
|
|
|
|
# Add merged levels as qualified
|
|
bias.qualified_levels.append(QualifiedLevel(
|
|
price=merged.val,
|
|
level_type=LevelType.MERGED_VAL,
|
|
direction=Side.BUY,
|
|
strength=70,
|
|
source_dates=[p.session_date for p in to_merge],
|
|
notes=f"Merged {len(to_merge)}-day VAL — high precision support",
|
|
))
|
|
bias.qualified_levels.append(QualifiedLevel(
|
|
price=merged.vah,
|
|
level_type=LevelType.MERGED_VAH,
|
|
direction=Side.SELL,
|
|
strength=70,
|
|
source_dates=[p.session_date for p in to_merge],
|
|
notes=f"Merged {len(to_merge)}-day VAH — high precision resistance",
|
|
))
|
|
|
|
@property
|
|
def current_bias(self) -> Optional[DailyBias]:
|
|
return self._bias_history[-1] if self._bias_history else None
|
|
|
|
@property
|
|
def profile_history(self) -> list[VolumeProfileResult]:
|
|
return self._profile_history
|