732 lines
30 KiB
Python
732 lines
30 KiB
Python
# analyze_market_smc.py
|
|
# Enhanced Market Analysis using Pure Smart Money Concepts (SMC)
|
|
# No Traditional Indicators - Focus on Market Structure, Liquidity, and Institutional Behavior
|
|
|
|
import pandas as pd
|
|
import numpy as np
|
|
import logging
|
|
from typing import Dict, Any, List, Optional, Tuple
|
|
from enum import Enum
|
|
from datetime import datetime, timedelta
|
|
import yfinance as yf
|
|
|
|
# Import our Pure SMC Technical Indicators
|
|
from technical_indicators_smc import (
|
|
analyze_smc_full,
|
|
calculate_optimal_entry_exit,
|
|
MarketStructure,
|
|
LiquidityType,
|
|
OrderBlockType,
|
|
FVGType
|
|
)
|
|
|
|
# =========================
|
|
# Trading Profile Enums
|
|
# =========================
|
|
|
|
class TradingProfile(Enum):
|
|
SCALPER = "scalper"
|
|
INTRADAY = "intraday"
|
|
SWING = "swing"
|
|
POSITION = "position"
|
|
|
|
class MarketPhase(Enum):
|
|
ACCUMULATION = "accumulation"
|
|
MARKUP = "markup"
|
|
DISTRIBUTION = "distribution"
|
|
MARKDOWN = "markdown"
|
|
TRANSITION = "transition"
|
|
|
|
# =========================
|
|
# Pair-Specific Configurations
|
|
# =========================
|
|
|
|
SPREAD_RULES = {
|
|
"XAUUSD": {"max_abs": 0.5, "max_rel_atr": 0.25, "smc_sensitivity": 1.2},
|
|
"BTCUSD": {"max_abs": 30.0, "max_rel_atr": 0.12, "smc_sensitivity": 1.0},
|
|
"EURUSD": {"max_abs": 0.00015, "max_rel_atr": 0.20, "smc_sensitivity": 0.8},
|
|
"GBPUSD": {"max_abs": 0.00020, "max_rel_atr": 0.25, "smc_sensitivity": 0.9},
|
|
"USDJPY": {"max_abs": 0.015, "max_rel_atr": 0.18, "smc_sensitivity": 0.85},
|
|
"DEFAULT": {"max_rel_atr": 0.20, "smc_sensitivity": 1.0}
|
|
}
|
|
|
|
# SMC-specific configurations for different pairs
|
|
SMC_CONFIGS = {
|
|
"XAUUSD": {
|
|
"swing_length": 7, # Gold needs longer swing detection
|
|
"liquidity_lookback": 30,
|
|
"order_block_lookback": 100,
|
|
"min_gap_atr": 0.15,
|
|
"eq_tolerance_pips": 10.0
|
|
},
|
|
"BTCUSD": {
|
|
"swing_length": 5,
|
|
"liquidity_lookback": 20,
|
|
"order_block_lookback": 50,
|
|
"min_gap_atr": 0.08,
|
|
"eq_tolerance_pips": 50.0 # Crypto needs higher tolerance
|
|
},
|
|
"EURUSD": {
|
|
"swing_length": 4,
|
|
"liquidity_lookback": 15,
|
|
"order_block_lookback": 40,
|
|
"min_gap_atr": 0.12,
|
|
"eq_tolerance_pips": 3.0
|
|
},
|
|
"DEFAULT": {
|
|
"swing_length": 5,
|
|
"liquidity_lookback": 20,
|
|
"order_block_lookback": 50,
|
|
"min_gap_atr": 0.1,
|
|
"eq_tolerance_pips": 5.0
|
|
}
|
|
}
|
|
|
|
# =========================
|
|
# Utility Functions
|
|
# =========================
|
|
|
|
def _match_symbol_key(symbol: str) -> str:
|
|
"""Match symbol to configuration key"""
|
|
if not symbol:
|
|
return "DEFAULT"
|
|
s = str(symbol).upper()
|
|
for k in SPREAD_RULES.keys():
|
|
if k != "DEFAULT" and s.startswith(k):
|
|
return k
|
|
return "DEFAULT"
|
|
|
|
def is_spread_ok_for_symbol(symbol: str, spread: float, atr: float) -> bool:
|
|
"""SMC-aware spread validation"""
|
|
try:
|
|
if spread is None or (isinstance(spread, float) and spread != spread):
|
|
return False
|
|
|
|
key = _match_symbol_key(symbol)
|
|
rules = SPREAD_RULES.get(key, SPREAD_RULES["DEFAULT"])
|
|
|
|
# SMC sensitivity adjustment
|
|
smc_sensitivity = rules.get("smc_sensitivity", 1.0)
|
|
|
|
if atr is not None and atr > 0:
|
|
rel = float(spread) / float(atr)
|
|
max_rel = rules.get("max_rel_atr", 0.20) * smc_sensitivity
|
|
if rel > max_rel:
|
|
return False
|
|
|
|
max_abs = rules.get("max_abs")
|
|
if max_abs is not None:
|
|
adjusted_max = max_abs * smc_sensitivity
|
|
if float(spread) > adjusted_max:
|
|
return False
|
|
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
def fetch_data(ticker: str, period: str = '1y', interval: str = '1d') -> pd.DataFrame:
|
|
"""Enhanced data fetching with SMC requirements"""
|
|
try:
|
|
df = yf.download(ticker, period=period, interval=interval)
|
|
if df.empty:
|
|
raise ValueError(f"No data available for ticker '{ticker}'")
|
|
|
|
df = df.rename(columns={'Open': 'open', 'High': 'high', 'Low': 'low', 'Close': 'close', 'Volume': 'volume'})
|
|
df.index.name = 'time'
|
|
df = df.reset_index()
|
|
|
|
if 'Adj Close' in df.columns:
|
|
df = df.drop(columns=['Adj Close'])
|
|
|
|
# Ensure time column exists for SMC analysis
|
|
if 'time' not in df.columns:
|
|
df['time'] = df.index
|
|
|
|
# Convert time to datetime if it's not already
|
|
if 'time' in df.columns:
|
|
try:
|
|
df['time'] = pd.to_datetime(df['time'])
|
|
except:
|
|
df['time'] = pd.date_range(start='2023-01-01', periods=len(df), freq='1H')
|
|
|
|
return df
|
|
except Exception as e:
|
|
logging.error(f"Error fetching data for {ticker}: {e}")
|
|
return pd.DataFrame()
|
|
|
|
# =========================
|
|
# Enhanced Profile Analyzer
|
|
# =========================
|
|
|
|
class SMCProfileAnalyzer:
|
|
"""SMC-based market analysis for different trading profiles"""
|
|
|
|
def __init__(self, profile: TradingProfile, symbol: str = "DEFAULT"):
|
|
self.profile = profile
|
|
self.symbol = symbol
|
|
self.smc_config = SMC_CONFIGS.get(_match_symbol_key(symbol), SMC_CONFIGS["DEFAULT"])
|
|
|
|
# Profile-specific SMC adjustments
|
|
if profile == TradingProfile.SCALPER:
|
|
self.smc_config = self.smc_config.copy()
|
|
self.smc_config['swing_length'] = max(3, self.smc_config['swing_length'] - 2)
|
|
self.smc_config['liquidity_lookback'] = min(10, self.smc_config['liquidity_lookback'])
|
|
self.smc_config['min_gap_atr'] = self.smc_config['min_gap_atr'] * 0.8
|
|
|
|
elif profile == TradingProfile.POSITION:
|
|
self.smc_config = self.smc_config.copy()
|
|
self.smc_config['swing_length'] = self.smc_config['swing_length'] + 3
|
|
self.smc_config['liquidity_lookback'] = self.smc_config['liquidity_lookback'] * 2
|
|
self.smc_config['order_block_lookback'] = self.smc_config['order_block_lookback'] * 2
|
|
|
|
def analyze(self, df: pd.DataFrame) -> Dict[str, Any]:
|
|
"""Complete SMC analysis based on trading profile"""
|
|
if len(df) < 50: # Minimum data requirement
|
|
return self._empty_analysis()
|
|
|
|
# Core SMC Analysis
|
|
smc_analysis = analyze_smc_full(df, self.smc_config)
|
|
|
|
# Profile-specific interpretation
|
|
if self.profile == TradingProfile.SCALPER:
|
|
return self._analyze_scalper(df, smc_analysis)
|
|
elif self.profile == TradingProfile.INTRADAY:
|
|
return self._analyze_intraday(df, smc_analysis)
|
|
elif self.profile == TradingProfile.SWING:
|
|
return self._analyze_swing(df, smc_analysis)
|
|
else: # POSITION
|
|
return self._analyze_position(df, smc_analysis)
|
|
|
|
def _empty_analysis(self) -> Dict[str, Any]:
|
|
"""Return empty analysis when insufficient data"""
|
|
return {
|
|
'smc_bias': 'NEUTRAL',
|
|
'confidence': 0.0,
|
|
'entry_signals': [],
|
|
'risk_level': 'HIGH',
|
|
'market_phase': MarketPhase.TRANSITION.value,
|
|
'key_levels': {},
|
|
'warnings': ['Insufficient data for SMC analysis']
|
|
}
|
|
|
|
def _analyze_scalper(self, df: pd.DataFrame, smc_analysis: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Scalper-focused SMC analysis"""
|
|
result = {
|
|
'smc_bias': smc_analysis['overall_bias'],
|
|
'confidence': min(smc_analysis['bias_strength'] / 5.0, 1.0),
|
|
'entry_signals': [],
|
|
'risk_level': 'MEDIUM',
|
|
'market_phase': self._determine_scalper_phase(smc_analysis),
|
|
'warnings': []
|
|
}
|
|
|
|
# Focus on recent FVGs and Order Blocks for scalping
|
|
recent_fvgs = [fvg for fvg in smc_analysis['fair_value_gaps'] if fvg['strength'] > 1.0]
|
|
recent_obs = [ob for ob in smc_analysis['order_blocks'] if not ob['tested']]
|
|
|
|
if recent_fvgs:
|
|
result['entry_signals'].append({
|
|
'type': 'FVG_RETEST',
|
|
'zones': recent_fvgs[:2], # Top 2 FVGs
|
|
'timeframe': 'M1-M5'
|
|
})
|
|
|
|
if recent_obs:
|
|
result['entry_signals'].append({
|
|
'type': 'ORDER_BLOCK_RETEST',
|
|
'zones': recent_obs[:1], # Top 1 OB
|
|
'timeframe': 'M1-M15'
|
|
})
|
|
|
|
# Scalper warnings
|
|
if smc_analysis['premium_discount_zones']['current_zone'] == 'EQUILIBRIUM':
|
|
result['warnings'].append('Price in equilibrium - consider waiting for clear bias')
|
|
|
|
return result
|
|
|
|
def _analyze_intraday(self, df: pd.DataFrame, smc_analysis: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Intraday trader SMC analysis"""
|
|
result = {
|
|
'smc_bias': smc_analysis['overall_bias'],
|
|
'confidence': min(smc_analysis['bias_strength'] / 7.0, 1.0),
|
|
'entry_signals': [],
|
|
'risk_level': 'MEDIUM',
|
|
'market_phase': self._determine_intraday_phase(smc_analysis),
|
|
'session_analysis': self._analyze_trading_session(df),
|
|
'warnings': []
|
|
}
|
|
|
|
# Liquidity sweep opportunities
|
|
if smc_analysis['liquidity_sweeps']:
|
|
strong_sweeps = [s for s in smc_analysis['liquidity_sweeps'] if s['strength'] > 1.5]
|
|
if strong_sweeps:
|
|
result['entry_signals'].append({
|
|
'type': 'LIQUIDITY_SWEEP_REVERSAL',
|
|
'sweeps': strong_sweeps,
|
|
'timeframe': 'M15-H1'
|
|
})
|
|
|
|
# Premium/Discount zone analysis
|
|
zones = smc_analysis['premium_discount_zones']
|
|
if zones['current_zone'] in ['PREMIUM', 'DISCOUNT']:
|
|
result['entry_signals'].append({
|
|
'type': 'PREMIUM_DISCOUNT_REVERSAL',
|
|
'zone': zones['current_zone'],
|
|
'levels': zones[zones['current_zone'].lower()],
|
|
'timeframe': 'H1-H4'
|
|
})
|
|
|
|
return result
|
|
|
|
def _analyze_swing(self, df: pd.DataFrame, smc_analysis: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Swing trader SMC analysis"""
|
|
result = {
|
|
'smc_bias': smc_analysis['overall_bias'],
|
|
'confidence': min(smc_analysis['bias_strength'] / 8.0, 1.0),
|
|
'entry_signals': [],
|
|
'risk_level': 'LOW',
|
|
'market_phase': self._determine_swing_phase(smc_analysis),
|
|
'structure_analysis': smc_analysis['structure'],
|
|
'warnings': []
|
|
}
|
|
|
|
# Structure-based signals
|
|
structure = smc_analysis['structure']
|
|
if structure['bos_detected']:
|
|
result['entry_signals'].append({
|
|
'type': 'BREAK_OF_STRUCTURE',
|
|
'direction': structure['trend_direction'],
|
|
'strength': structure['structure_strength'],
|
|
'timeframe': 'H4-D1'
|
|
})
|
|
|
|
# CHoCH signals
|
|
if smc_analysis['change_of_character']:
|
|
choch = smc_analysis['change_of_character']
|
|
result['entry_signals'].append({
|
|
'type': 'CHANGE_OF_CHARACTER',
|
|
'direction': choch['new_trend'],
|
|
'strength': choch['strength'],
|
|
'timeframe': 'H4-D1'
|
|
})
|
|
|
|
# High-quality Order Blocks
|
|
strong_obs = [ob for ob in smc_analysis['order_blocks']
|
|
if ob['strength'] > 2.0 and not ob['tested']]
|
|
if strong_obs:
|
|
result['entry_signals'].append({
|
|
'type': 'HIGH_QUALITY_ORDER_BLOCK',
|
|
'blocks': strong_obs[:2],
|
|
'timeframe': 'H1-H4'
|
|
})
|
|
|
|
return result
|
|
|
|
def _analyze_position(self, df: pd.DataFrame, smc_analysis: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Position trader SMC analysis"""
|
|
result = {
|
|
'smc_bias': smc_analysis['overall_bias'],
|
|
'confidence': min(smc_analysis['bias_strength'] / 10.0, 1.0),
|
|
'entry_signals': [],
|
|
'risk_level': 'LOW',
|
|
'market_phase': self._determine_position_phase(smc_analysis),
|
|
'long_term_structure': self._analyze_long_term_structure(df, smc_analysis),
|
|
'warnings': []
|
|
}
|
|
|
|
# Major structure changes
|
|
if smc_analysis['structure']['bos_detected'] and smc_analysis['structure']['structure_strength'] > 0.7:
|
|
result['entry_signals'].append({
|
|
'type': 'MAJOR_STRUCTURE_SHIFT',
|
|
'direction': smc_analysis['structure']['trend_direction'],
|
|
'strength': smc_analysis['structure']['structure_strength'],
|
|
'timeframe': 'D1-W1'
|
|
})
|
|
|
|
# Institutional Order Blocks
|
|
institutional_obs = [ob for ob in smc_analysis['order_blocks']
|
|
if ob['strength'] > 3.0]
|
|
if institutional_obs:
|
|
result['entry_signals'].append({
|
|
'type': 'INSTITUTIONAL_ORDER_BLOCK',
|
|
'blocks': institutional_obs,
|
|
'timeframe': 'H4-D1'
|
|
})
|
|
|
|
return result
|
|
|
|
def _analyze_trading_session(self, df: pd.DataFrame) -> Dict[str, Any]:
|
|
"""Analyze current trading session impact"""
|
|
if 'time' not in df.columns:
|
|
return {'session': 'UNKNOWN', 'activity': 'NORMAL'}
|
|
|
|
current_time = pd.to_datetime(df['time'].iloc[-1])
|
|
utc_hour = current_time.hour
|
|
|
|
if 0 <= utc_hour < 8:
|
|
session = "ASIAN"
|
|
activity = "LOW" if utc_hour < 2 else "MEDIUM"
|
|
elif 8 <= utc_hour < 16:
|
|
session = "LONDON"
|
|
activity = "HIGH"
|
|
else:
|
|
session = "NEW_YORK"
|
|
activity = "HIGH" if utc_hour < 20 else "MEDIUM"
|
|
|
|
return {'session': session, 'activity': activity}
|
|
|
|
def _determine_scalper_phase(self, smc_analysis: Dict[str, Any]) -> str:
|
|
"""Determine market phase for scalpers"""
|
|
if smc_analysis['liquidity_sweeps']:
|
|
return MarketPhase.TRANSITION.value
|
|
elif smc_analysis['premium_discount_zones']['current_zone'] == 'EQUILIBRIUM':
|
|
return MarketPhase.ACCUMULATION.value
|
|
else:
|
|
return MarketPhase.MARKUP.value if 'BULLISH' in smc_analysis['overall_bias'] else MarketPhase.MARKDOWN.value
|
|
|
|
def _determine_intraday_phase(self, smc_analysis: Dict[str, Any]) -> str:
|
|
"""Determine market phase for intraday traders"""
|
|
structure_strength = smc_analysis['structure']['structure_strength']
|
|
|
|
if structure_strength > 0.7:
|
|
return MarketPhase.MARKUP.value if 'BULLISH' in smc_analysis['overall_bias'] else MarketPhase.MARKDOWN.value
|
|
elif structure_strength > 0.4:
|
|
return MarketPhase.DISTRIBUTION.value if smc_analysis['premium_discount_zones']['current_zone'] == 'PREMIUM' else MarketPhase.ACCUMULATION.value
|
|
else:
|
|
return MarketPhase.TRANSITION.value
|
|
|
|
def _determine_swing_phase(self, smc_analysis: Dict[str, Any]) -> str:
|
|
"""Determine market phase for swing traders"""
|
|
if smc_analysis['change_of_character']:
|
|
return MarketPhase.TRANSITION.value
|
|
elif smc_analysis['structure']['bos_detected']:
|
|
return MarketPhase.MARKUP.value if 'BULLISH' in smc_analysis['overall_bias'] else MarketPhase.MARKDOWN.value
|
|
else:
|
|
zone = smc_analysis['premium_discount_zones']['current_zone']
|
|
if zone == 'PREMIUM':
|
|
return MarketPhase.DISTRIBUTION.value
|
|
elif zone == 'DISCOUNT':
|
|
return MarketPhase.ACCUMULATION.value
|
|
else:
|
|
return MarketPhase.TRANSITION.value
|
|
|
|
def _determine_position_phase(self, smc_analysis: Dict[str, Any]) -> str:
|
|
"""Determine market phase for position traders"""
|
|
bias_strength = smc_analysis['bias_strength']
|
|
|
|
if bias_strength >= 8:
|
|
return MarketPhase.MARKUP.value if 'BULLISH' in smc_analysis['overall_bias'] else MarketPhase.MARKDOWN.value
|
|
elif bias_strength >= 5:
|
|
zone = smc_analysis['premium_discount_zones']['current_zone']
|
|
if zone == 'PREMIUM' and 'BEARISH' in smc_analysis['overall_bias']:
|
|
return MarketPhase.DISTRIBUTION.value
|
|
elif zone == 'DISCOUNT' and 'BULLISH' in smc_analysis['overall_bias']:
|
|
return MarketPhase.ACCUMULATION.value
|
|
else:
|
|
return MarketPhase.TRANSITION.value
|
|
else:
|
|
return MarketPhase.ACCUMULATION.value
|
|
|
|
def _analyze_long_term_structure(self, df: pd.DataFrame, smc_analysis: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Analyze long-term market structure for position trading"""
|
|
# Use longer lookback for position trading
|
|
long_term_high = df['high'].tail(200).max() if len(df) >= 200 else df['high'].max()
|
|
long_term_low = df['low'].tail(200).min() if len(df) >= 200 else df['low'].min()
|
|
current_price = df['close'].iloc[-1]
|
|
|
|
# Position in long-term range
|
|
range_size = long_term_high - long_term_low
|
|
position_pct = (current_price - long_term_low) / range_size if range_size > 0 else 0.5
|
|
|
|
return {
|
|
'long_term_high': long_term_high,
|
|
'long_term_low': long_term_low,
|
|
'range_position_pct': position_pct,
|
|
'in_upper_third': position_pct > 0.67,
|
|
'in_lower_third': position_pct < 0.33,
|
|
'trend_bias': 'BULLISH' if position_pct > 0.6 else 'BEARISH' if position_pct < 0.4 else 'NEUTRAL'
|
|
}
|
|
|
|
# =========================
|
|
# Enhanced Market Context
|
|
# =========================
|
|
|
|
def get_market_context_smc(df: pd.DataFrame, profile: Optional[str] = None, symbol: str = "DEFAULT") -> Dict[str, Any]:
|
|
"""Enhanced market context analysis using pure SMC"""
|
|
if df.empty or len(df) < 50:
|
|
return {
|
|
'error': 'Insufficient data for SMC analysis',
|
|
'min_required': 50,
|
|
'current_length': len(df)
|
|
}
|
|
|
|
# Get SMC configuration for symbol
|
|
smc_config = SMC_CONFIGS.get(_match_symbol_key(symbol), SMC_CONFIGS["DEFAULT"])
|
|
|
|
# Core SMC Analysis
|
|
smc_analysis = analyze_smc_full(df, smc_config)
|
|
|
|
# Calculate optimal entry/exit points
|
|
entry_exit = calculate_optimal_entry_exit(smc_analysis, risk_reward_ratio=2.0)
|
|
|
|
# Profile-specific analysis
|
|
profile_analysis = None
|
|
if profile:
|
|
try:
|
|
profile_enum = TradingProfile(profile.lower())
|
|
analyzer = SMCProfileAnalyzer(profile_enum, symbol)
|
|
profile_analysis = analyzer.analyze(df)
|
|
except ValueError:
|
|
logging.warning(f"Unknown trading profile: {profile}")
|
|
|
|
# Enhanced context
|
|
context = {
|
|
'timestamp': datetime.now(),
|
|
'symbol': symbol,
|
|
'current_price': smc_analysis['current_price'],
|
|
'atr': smc_analysis['atr'],
|
|
|
|
# Core SMC Analysis
|
|
'smc_analysis': smc_analysis,
|
|
'entry_exit_analysis': entry_exit,
|
|
|
|
# Market bias with SMC reasoning
|
|
'market_bias': smc_analysis['overall_bias'],
|
|
'bias_strength': smc_analysis['bias_strength'],
|
|
'bias_components': {
|
|
'bullish_signals': smc_analysis['bullish_signals'],
|
|
'bearish_signals': smc_analysis['bearish_signals']
|
|
},
|
|
|
|
# Key SMC levels
|
|
'key_levels': {
|
|
'premium_discount': smc_analysis['premium_discount_zones'],
|
|
'active_order_blocks': len([ob for ob in smc_analysis['order_blocks'] if ob['status'] == 'ACTIVE']),
|
|
'active_fvgs': len([fvg for fvg in smc_analysis['fair_value_gaps'] if fvg['status'] == 'ACTIVE']),
|
|
'liquidity_zones': len(smc_analysis['liquidity_zones'])
|
|
},
|
|
|
|
# Profile-specific insights
|
|
'profile_analysis': profile_analysis,
|
|
|
|
# Risk assessment
|
|
'risk_assessment': {
|
|
'overall_risk': _calculate_overall_risk(smc_analysis),
|
|
'volatility_risk': 'HIGH' if smc_analysis['atr'] > df['close'].iloc[-1] * 0.02 else 'NORMAL',
|
|
'structure_risk': 'LOW' if smc_analysis['structure']['structure_strength'] > 0.6 else 'MEDIUM'
|
|
}
|
|
}
|
|
|
|
return context
|
|
|
|
def _calculate_overall_risk(smc_analysis: Dict[str, Any]) -> str:
|
|
"""Calculate overall risk based on SMC factors"""
|
|
risk_factors = 0
|
|
|
|
# High bias strength = lower risk
|
|
if smc_analysis['bias_strength'] < 3:
|
|
risk_factors += 1
|
|
|
|
# Conflicting signals = higher risk
|
|
total_signals = smc_analysis['bullish_signals'] + smc_analysis['bearish_signals']
|
|
if total_signals > 0:
|
|
balance = abs(smc_analysis['bullish_signals'] - smc_analysis['bearish_signals']) / total_signals
|
|
if balance < 0.4: # Signals are too balanced
|
|
risk_factors += 1
|
|
|
|
# Equilibrium zone = higher risk
|
|
if smc_analysis['premium_discount_zones']['current_zone'] == 'EQUILIBRIUM':
|
|
risk_factors += 1
|
|
|
|
# No clear structure = higher risk
|
|
if smc_analysis['structure']['structure_strength'] < 0.3:
|
|
risk_factors += 1
|
|
|
|
if risk_factors >= 3:
|
|
return 'HIGH'
|
|
elif risk_factors >= 2:
|
|
return 'MEDIUM'
|
|
else:
|
|
return 'LOW'
|
|
|
|
# =========================
|
|
# Backtesting with SMC
|
|
# =========================
|
|
|
|
def backtest_smc_strategy(df: pd.DataFrame, context: Dict[str, Any],
|
|
initial_capital: float = 10000,
|
|
risk_per_trade: float = 0.01,
|
|
symbol: str = "DEFAULT") -> Dict[str, Any]:
|
|
"""Enhanced backtesting using SMC analysis"""
|
|
trades = []
|
|
equity = [initial_capital]
|
|
positions = []
|
|
|
|
# Get symbol-specific spread rules
|
|
spread_rules = SPREAD_RULES.get(_match_symbol_key(symbol), SPREAD_RULES["DEFAULT"])
|
|
|
|
for i in range(100, len(df)): # Start after sufficient data
|
|
current_data = df.iloc[:i+1]
|
|
|
|
# Get SMC analysis for current point
|
|
sub_context = get_market_context_smc(current_data, symbol=symbol)
|
|
|
|
if 'error' in sub_context:
|
|
continue
|
|
|
|
smc_analysis = sub_context['smc_analysis']
|
|
entry_exit = sub_context['entry_exit_analysis']
|
|
|
|
current_price = df['close'].iloc[i]
|
|
atr = smc_analysis['atr']
|
|
|
|
# Check spread conditions
|
|
spread_val = df.get('spread', pd.Series([0.001] * len(df))).iloc[i]
|
|
if not is_spread_ok_for_symbol(symbol, spread_val, atr):
|
|
continue
|
|
|
|
# Entry logic based on SMC
|
|
if not positions and entry_exit['direction'] != 'WAIT':
|
|
if len(entry_exit['entry_zones']) > 0:
|
|
best_zone = entry_exit['entry_zones'][0]
|
|
|
|
# Check if current price is near entry zone
|
|
entry_price = best_zone['entry_price']
|
|
if abs(current_price - entry_price) / current_price < 0.01: # Within 1%
|
|
|
|
# Calculate position size based on risk
|
|
stop_loss = best_zone.get('stop_loss', entry_price - atr * 2)
|
|
risk_amount = initial_capital * risk_per_trade
|
|
|
|
if entry_exit['direction'] == 'BUY':
|
|
risk_per_share = entry_price - stop_loss
|
|
else:
|
|
risk_per_share = stop_loss - entry_price
|
|
|
|
if risk_per_share > 0:
|
|
position_size = risk_amount / risk_per_share
|
|
|
|
positions.append({
|
|
'direction': entry_exit['direction'],
|
|
'entry_price': current_price,
|
|
'stop_loss': stop_loss,
|
|
'take_profit': best_zone.get('take_profit'),
|
|
'size': position_size,
|
|
'entry_time': df['time'].iloc[i] if 'time' in df.columns else i,
|
|
'smc_reason': best_zone['type']
|
|
})
|
|
|
|
# Exit logic
|
|
elif positions:
|
|
pos = positions[0]
|
|
pnl = 0
|
|
exit_reason = None
|
|
|
|
if pos['direction'] == 'BUY':
|
|
if current_price <= pos['stop_loss']:
|
|
pnl = pos['size'] * (current_price - pos['entry_price'])
|
|
exit_reason = 'STOP_LOSS'
|
|
elif pos['take_profit'] and current_price >= pos['take_profit']:
|
|
pnl = pos['size'] * (current_price - pos['entry_price'])
|
|
exit_reason = 'TAKE_PROFIT'
|
|
else: # SELL
|
|
if current_price >= pos['stop_loss']:
|
|
pnl = pos['size'] * (pos['entry_price'] - current_price)
|
|
exit_reason = 'STOP_LOSS'
|
|
elif pos['take_profit'] and current_price <= pos['take_profit']:
|
|
pnl = pos['size'] * (pos['entry_price'] - current_price)
|
|
exit_reason = 'TAKE_PROFIT'
|
|
|
|
if exit_reason:
|
|
trades.append({
|
|
'entry_price': pos['entry_price'],
|
|
'exit_price': current_price,
|
|
'pnl': pnl,
|
|
'direction': pos['direction'],
|
|
'exit_reason': exit_reason,
|
|
'smc_reason': pos['smc_reason'],
|
|
'hold_time': i - df[df['time'] == pos['entry_time']].index[0] if 'time' in df.columns else 1
|
|
})
|
|
positions = []
|
|
|
|
# Update equity
|
|
current_equity = equity[-1] + (trades[-1]['pnl'] if trades else 0)
|
|
equity.append(current_equity)
|
|
|
|
# Calculate metrics
|
|
if not trades:
|
|
return {'error': 'No trades generated', 'trades': 0}
|
|
|
|
total_pnl = sum(t['pnl'] for t in trades)
|
|
winning_trades = [t for t in trades if t['pnl'] > 0]
|
|
win_rate = len(winning_trades) / len(trades)
|
|
|
|
avg_win = np.mean([t['pnl'] for t in winning_trades]) if winning_trades else 0
|
|
losing_trades = [t for t in trades if t['pnl'] <= 0]
|
|
avg_loss = np.mean([t['pnl'] for t in losing_trades]) if losing_trades else 0
|
|
|
|
return {
|
|
'total_pnl': total_pnl,
|
|
'total_trades': len(trades),
|
|
'win_rate': win_rate,
|
|
'avg_win': avg_win,
|
|
'avg_loss': avg_loss,
|
|
'profit_factor': abs(avg_win / avg_loss) if avg_loss != 0 else float('inf'),
|
|
'final_equity': equity[-1],
|
|
'max_drawdown': min(equity) - max(equity[:equity.index(min(equity))+1]) if len(equity) > 1 else 0,
|
|
'smc_strategy_breakdown': {
|
|
'order_block_trades': len([t for t in trades if 'ORDER_BLOCK' in t['smc_reason']]),
|
|
'fvg_trades': len([t for t in trades if 'FVG' in t['smc_reason']]),
|
|
'zone_trades': len([t for t in trades if 'ZONE' in t['smc_reason']])
|
|
}
|
|
}
|
|
|
|
# =========================
|
|
# Example Usage
|
|
# =========================
|
|
|
|
if __name__ == "__main__":
|
|
# Test with Bitcoin data
|
|
print("Testing SMC Analysis with BTC-USD...")
|
|
df_btc = fetch_data('BTC-USD', period='3mo', interval='1h')
|
|
|
|
if not df_btc.empty:
|
|
context_btc = get_market_context_smc(df_btc, profile='swing', symbol='BTCUSD')
|
|
print(f"\nBTC-USD SMC Analysis:")
|
|
print(f"Market Bias: {context_btc['market_bias']}")
|
|
print(f"Bias Strength: {context_btc['bias_strength']}")
|
|
print(f"Current Zone: {context_btc['smc_analysis']['premium_discount_zones']['current_zone']}")
|
|
print(f"Active Order Blocks: {context_btc['key_levels']['active_order_blocks']}")
|
|
print(f"Active FVGs: {context_btc['key_levels']['active_fvgs']}")
|
|
|
|
if context_btc['entry_exit_analysis']['direction'] != 'WAIT':
|
|
direction = context_btc['entry_exit_analysis']['direction']
|
|
zones = context_btc['entry_exit_analysis']['entry_zones']
|
|
print(f"\nEntry Recommendation: {direction}")
|
|
if zones:
|
|
best_zone = zones[0]
|
|
print(f"Best Entry Zone: {best_zone['type']} at {best_zone['entry_price']:.2f}")
|
|
|
|
# Run backtest
|
|
backtest_results = backtest_smc_strategy(df_btc, context_btc, symbol='BTCUSD')
|
|
if 'error' not in backtest_results:
|
|
print(f"\nBacktest Results:")
|
|
print(f"Total PnL: ${backtest_results['total_pnl']:.2f}")
|
|
print(f"Win Rate: {backtest_results['win_rate']:.1%}")
|
|
print(f"Total Trades: {backtest_results['total_trades']}")
|
|
|
|
# Test with Gold data
|
|
print("\n" + "="*50)
|
|
print("Testing SMC Analysis with Gold (GC=F)...")
|
|
df_gold = fetch_data('GC=F', period='3mo', interval='1h')
|
|
|
|
if not df_gold.empty:
|
|
context_gold = get_market_context_smc(df_gold, profile='intraday', symbol='XAUUSD')
|
|
print(f"\nXAU-USD SMC Analysis:")
|
|
print(f"Market Bias: {context_gold['market_bias']}")
|
|
print(f"Risk Level: {context_gold['risk_assessment']['overall_risk']}")
|
|
|
|
if context_gold.get('profile_analysis'):
|
|
profile_data = context_gold['profile_analysis']
|
|
print(f"Market Phase: {profile_data['market_phase']}")
|
|
print(f"Entry Signals: {len(profile_data['entry_signals'])}")
|