gavindiaz/XauBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
7af9183af3ef7ce67ed3cb28a335862d1d514301
XauBot/src/smc_polars.py
T
GifariKemalandClaude Opus 4.5 7af9183af3 feat: Smart AI Trading Bot for XAUUSD with ML and SMC 
- XGBoost ML model with 37 features for market direction prediction
- Smart Money Concepts (SMC): Order Blocks, FVG, BOS, CHoCH
- HMM market regime detection (trending/ranging/volatile)
- ATR-based stop loss with 1.5 ATR minimum distance
- Broker-level SL protection with fallback
- Time-based exit (max 6 hours per trade)
- Session-aware trading optimized for London/NY overlap
- Auto-retraining based on market conditions
- Telegram notifications and web dashboard
- Backtest results: 63.9% win rate, 2.64 profit factor, 4.83 Sharpe

Backtest period: Jan 2025 - Feb 2026, 654 trades, $4,189 net P/L

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-02-06 09:01:35 +07:00

828 lines
30 KiB
Python
Raw Blame History

"""
Smart Money Concepts (SMC) Implementation - Pure Polars
========================================================
Native implementation of SMC concepts using Polars expressions.
NO PANDAS. NO smartmoneyconcepts library.
Implements:
- Fair Value Gaps (FVG)
- Swing Points (Fractal High/Low)
- Order Blocks
- Break of Structure (BOS)
- Change of Character (CHoCH)
- Liquidity Zones
"""
import polars as pl
import numpy as np
from typing import Tuple, Optional, Dict
from dataclasses import dataclass
from loguru import logger
@dataclass
class SMCSignal:
"""SMC trading signal."""
signal_type: str # "BUY" or "SELL"
entry_price: float
stop_loss: float
take_profit: float
confidence: float
reason: str
@property
def risk_reward(self) -> float:
"""Calculate risk/reward ratio."""
risk = abs(self.entry_price - self.stop_loss)
reward = abs(self.take_profit - self.entry_price)
return reward / risk if risk > 0 else 0
class SMCAnalyzer:
"""
Smart Money Concepts Analyzer using Pure Polars.
All calculations are vectorized using Polars expressions.
No loops, no Pandas, maximum performance.
"""
def __init__(
self,
swing_length: int = 5,
fvg_min_gap_pips: float = 2.0,
ob_lookback: int = 10,
):
"""
Initialize SMC Analyzer.
Args:
swing_length: Number of bars for swing detection
fvg_min_gap_pips: Minimum FVG gap size in pips
ob_lookback: Order block lookback period
"""
self.swing_length = swing_length
self.fvg_min_gap_pips = fvg_min_gap_pips
self.ob_lookback = ob_lookback
def calculate_all(self, df: pl.DataFrame) -> pl.DataFrame:
"""
Calculate all SMC indicators.
Args:
df: Polars DataFrame with OHLCV data
Returns:
DataFrame with all SMC columns added
"""
df = self.calculate_swing_points(df)
df = self.calculate_fvg(df)
df = self.calculate_order_blocks(df)
df = self.calculate_bos_choch(df)
return df
def calculate_fvg(self, df: pl.DataFrame) -> pl.DataFrame:
"""
Calculate Fair Value Gaps (FVG) using Polars expressions.
Bullish FVG: Current Low > Previous-2 High (gap up)
Bearish FVG: Current High < Previous-2 Low (gap down)
This is a vectorized implementation - no loops.
Args:
df: DataFrame with OHLCV data
Returns:
DataFrame with FVG columns:
- is_fvg_bull: Boolean for bullish FVG
- is_fvg_bear: Boolean for bearish FVG
- fvg_top: Top of FVG zone
- fvg_bottom: Bottom of FVG zone
- fvg_mid: Midpoint of FVG (50% retracement target)
"""
# Get shifted values using Polars expressions
df = df.with_columns([
# Previous candle values (t-1)
pl.col("high").shift(1).alias("_prev_high"),
pl.col("low").shift(1).alias("_prev_low"),
# Candle before previous (t-2)
pl.col("high").shift(2).alias("_prev2_high"),
pl.col("low").shift(2).alias("_prev2_low"),
# Next candle values (t+1) - for detecting FVG on middle candle
pl.col("high").shift(-1).alias("_next_high"),
pl.col("low").shift(-1).alias("_next_low"),
])
# Calculate FVG conditions
# For the MIDDLE candle of a 3-candle pattern:
# Bullish FVG: prev2_high < next_low (gap between candle 1's high and candle 3's low)
# Bearish FVG: prev2_low > next_high (gap between candle 1's low and candle 3's high)
df = df.with_columns([
# Bullish FVG detection
(pl.col("_prev2_high") < pl.col("_next_low")).alias("is_fvg_bull"),
# Bearish FVG detection
(pl.col("_prev2_low") > pl.col("_next_high")).alias("is_fvg_bear"),
])
# Calculate FVG zones
df = df.with_columns([
# Bullish FVG zone: from prev2_high to next_low
pl.when(pl.col("is_fvg_bull"))
.then(pl.col("_next_low"))
.otherwise(None)
.alias("fvg_top"),
pl.when(pl.col("is_fvg_bull"))
.then(pl.col("_prev2_high"))
.otherwise(
pl.when(pl.col("is_fvg_bear"))
.then(pl.col("_prev2_low"))
.otherwise(None)
)
.alias("fvg_bottom"),
])
# Update fvg_top for bearish FVG
df = df.with_columns([
pl.when(pl.col("is_fvg_bear"))
.then(pl.col("_prev2_low"))
.otherwise(pl.col("fvg_top"))
.alias("fvg_top"),
pl.when(pl.col("is_fvg_bear"))
.then(pl.col("_next_high"))
.otherwise(pl.col("fvg_bottom"))
.alias("fvg_bottom"),
])
# Calculate FVG midpoint (50% retracement)
df = df.with_columns([
((pl.col("fvg_top") + pl.col("fvg_bottom")) / 2).alias("fvg_mid"),
])
# Combined FVG signal: 1 for bullish, -1 for bearish, 0 for none
df = df.with_columns([
pl.when(pl.col("is_fvg_bull"))
.then(1)
.when(pl.col("is_fvg_bear"))
.then(-1)
.otherwise(0)
.alias("fvg_signal"),
])
# Drop temporary columns
df = df.drop([
"_prev_high", "_prev_low", "_prev2_high", "_prev2_low",
"_next_high", "_next_low"
])
logger.debug(f"FVG calculation complete. Bullish: {df['is_fvg_bull'].sum()}, Bearish: {df['is_fvg_bear'].sum()}")
return df
def calculate_swing_points(self, df: pl.DataFrame) -> pl.DataFrame:
"""
Calculate Swing Points (Fractal Highs/Lows) using rolling windows.
A Swing High is when the current high is the highest in the window.
A Swing Low is when the current low is the lowest in the window.
Uses centered rolling window for look-ahead detection.
Args:
df: DataFrame with OHLCV data
Returns:
DataFrame with swing point columns:
- swing_high: 1 if swing high, 0 otherwise
- swing_low: -1 if swing low, 0 otherwise
- swing_high_level: Price level of swing high
- swing_low_level: Price level of swing low
"""
window_size = 2 * self.swing_length + 1
# Calculate rolling max/min with centered window
df = df.with_columns([
pl.col("high")
.rolling_max(window_size=window_size, center=True)
.alias("_roll_max"),
pl.col("low")
.rolling_min(window_size=window_size, center=True)
.alias("_roll_min"),
])
# Detect swing points where current price equals rolling extreme
df = df.with_columns([
# Swing High: current high is the rolling max
pl.when(pl.col("high") == pl.col("_roll_max"))
.then(1)
.otherwise(0)
.alias("swing_high"),
# Swing Low: current low is the rolling min
pl.when(pl.col("low") == pl.col("_roll_min"))
.then(-1)
.otherwise(0)
.alias("swing_low"),
])
# Store swing levels
df = df.with_columns([
pl.when(pl.col("swing_high") == 1)
.then(pl.col("high"))
.otherwise(None)
.alias("swing_high_level"),
pl.when(pl.col("swing_low") == -1)
.then(pl.col("low"))
.otherwise(None)
.alias("swing_low_level"),
])
# Forward fill last swing levels for reference
df = df.with_columns([
pl.col("swing_high_level")
.forward_fill()
.alias("last_swing_high"),
pl.col("swing_low_level")
.forward_fill()
.alias("last_swing_low"),
])
# Drop temporary columns
df = df.drop(["_roll_max", "_roll_min"])
swing_highs = (df["swing_high"] == 1).sum()
swing_lows = (df["swing_low"] == -1).sum()
logger.debug(f"Swing points: {swing_highs} highs, {swing_lows} lows")
return df
def calculate_order_blocks(self, df: pl.DataFrame) -> pl.DataFrame:
"""
Calculate Order Blocks using vectorized Polars operations.
Bullish Order Block: Last bearish candle before a bullish impulse
that creates a swing low and breaks structure.
Bearish Order Block: Last bullish candle before a bearish impulse
that creates a swing high and breaks structure.
This implementation uses numpy for the complex lookback logic,
then converts back to Polars for performance.
Args:
df: DataFrame with OHLCV and swing point data
Returns:
DataFrame with Order Block columns:
- ob: 1 for bullish OB, -1 for bearish OB, 0 for none
- ob_top: Top of order block zone
- ob_bottom: Bottom of order block zone
- ob_mitigated: True if OB has been mitigated
"""
# Ensure swing points are calculated
if "swing_high" not in df.columns:
df = self.calculate_swing_points(df)
# Extract numpy arrays for complex logic
opens = df["open"].to_numpy()
highs = df["high"].to_numpy()
lows = df["low"].to_numpy()
closes = df["close"].to_numpy()
swing_highs = df["swing_high"].to_numpy()
swing_lows = df["swing_low"].to_numpy()
n = len(df)
ob = np.zeros(n, dtype=np.int8)
ob_top = np.full(n, np.nan)
ob_bottom = np.full(n, np.nan)
for i in range(self.ob_lookback, n):
# Check for swing low -> Bullish Order Block
if swing_lows[i] == -1:
# Look for last bearish candle before swing low
for j in range(i - 1, max(0, i - self.ob_lookback), -1):
if closes[j] < opens[j]: # Bearish candle
# Check if this is a valid OB (price moved up significantly after)
if i + 1 < n and closes[i + 1] > highs[j]:
ob[j] = 1 # Bullish OB
ob_top[j] = highs[j]
ob_bottom[j] = lows[j]
break
# Check for swing high -> Bearish Order Block
if swing_highs[i] == 1:
# Look for last bullish candle before swing high
for j in range(i - 1, max(0, i - self.ob_lookback), -1):
if closes[j] > opens[j]: # Bullish candle
# Check if this is a valid OB (price moved down significantly after)
if i + 1 < n and closes[i + 1] < lows[j]:
ob[j] = -1 # Bearish OB
ob_top[j] = highs[j]
ob_bottom[j] = lows[j]
break
# Add to DataFrame
df = df.with_columns([
pl.Series("ob", ob),
pl.Series("ob_top", ob_top),
pl.Series("ob_bottom", ob_bottom),
])
# Calculate OB mitigation (price has revisited the OB zone)
df = df.with_columns([
# Forward fill OB zones for mitigation checking
pl.col("ob_top").forward_fill().alias("_ob_top_ff"),
pl.col("ob_bottom").forward_fill().alias("_ob_bottom_ff"),
pl.col("ob").forward_fill().alias("_ob_ff"),
])
# Check if current price has entered OB zone (mitigation)
df = df.with_columns([
pl.when(
(pl.col("_ob_ff") == 1) &
(pl.col("low") <= pl.col("_ob_top_ff")) &
(pl.col("high") >= pl.col("_ob_bottom_ff"))
)
.then(True)
.when(
(pl.col("_ob_ff") == -1) &
(pl.col("high") >= pl.col("_ob_bottom_ff")) &
(pl.col("low") <= pl.col("_ob_top_ff"))
)
.then(True)
.otherwise(False)
.alias("ob_mitigated"),
])
# Drop temporary columns
df = df.drop(["_ob_top_ff", "_ob_bottom_ff", "_ob_ff"])
bullish_obs = (df["ob"] == 1).sum()
bearish_obs = (df["ob"] == -1).sum()
logger.debug(f"Order Blocks: {bullish_obs} bullish, {bearish_obs} bearish")
return df
def calculate_bos_choch(self, df: pl.DataFrame) -> pl.DataFrame:
"""
Calculate Break of Structure (BOS) and Change of Character (CHoCH).
BOS: Structure break in the direction of the trend (continuation)
CHoCH: Structure break against the trend (reversal signal)
Uses numpy for stateful trend tracking, then converts to Polars.
Args:
df: DataFrame with OHLCV and swing point data
Returns:
DataFrame with BOS/CHoCH columns:
- bos: 1 for bullish BOS, -1 for bearish BOS
- choch: 1 for bullish CHoCH, -1 for bearish CHoCH
- market_structure: Current market structure (1=bullish, -1=bearish)
"""
# Ensure swing points are calculated
if "swing_high" not in df.columns:
df = self.calculate_swing_points(df)
# Extract arrays
highs = df["high"].to_numpy()
lows = df["low"].to_numpy()
closes = df["close"].to_numpy()
swing_highs = df["swing_high"].to_numpy()
swing_lows = df["swing_low"].to_numpy()
swing_high_levels = df["swing_high_level"].to_numpy() if "swing_high_level" in df.columns else np.full(len(df), np.nan)
swing_low_levels = df["swing_low_level"].to_numpy() if "swing_low_level" in df.columns else np.full(len(df), np.nan)
n = len(df)
bos = np.zeros(n, dtype=np.int8)
choch = np.zeros(n, dtype=np.int8)
market_structure = np.zeros(n, dtype=np.int8)
# Track last significant swing levels
last_swing_high = np.nan
last_swing_low = np.nan
trend = 0 # 0=neutral, 1=bullish, -1=bearish
for i in range(self.swing_length, n):
# Update last swing levels
if swing_highs[i] == 1 and not np.isnan(swing_high_levels[i]):
last_swing_high = swing_high_levels[i]
if swing_lows[i] == -1 and not np.isnan(swing_low_levels[i]):
last_swing_low = swing_low_levels[i]
market_structure[i] = trend
# Check for break of swing high (bullish break)
if not np.isnan(last_swing_high):
if closes[i] > last_swing_high:
if trend == 1: # Continuing bullish trend
bos[i] = 1 # Bullish BOS
elif trend == -1: # Was bearish, now breaking up
choch[i] = 1 # Bullish CHoCH (reversal)
trend = 1
last_swing_high = np.nan # Reset after break
# Check for break of swing low (bearish break)
if not np.isnan(last_swing_low):
if closes[i] < last_swing_low:
if trend == -1: # Continuing bearish trend
bos[i] = -1 # Bearish BOS
elif trend == 1: # Was bullish, now breaking down
choch[i] = -1 # Bearish CHoCH (reversal)
trend = -1
last_swing_low = np.nan # Reset after break
market_structure[i] = trend
# Add to DataFrame
df = df.with_columns([
pl.Series("bos", bos),
pl.Series("choch", choch),
pl.Series("market_structure", market_structure),
])
bullish_bos = (df["bos"] == 1).sum()
bearish_bos = (df["bos"] == -1).sum()
bullish_choch = (df["choch"] == 1).sum()
bearish_choch = (df["choch"] == -1).sum()
logger.debug(f"BOS: {bullish_bos} bullish, {bearish_bos} bearish")
logger.debug(f"CHoCH: {bullish_choch} bullish, {bearish_choch} bearish")
return df
def calculate_liquidity_zones(self, df: pl.DataFrame) -> pl.DataFrame:
"""
Calculate Liquidity Zones (Equal Highs/Lows and BSL/SSL).
OPTIMIZED: Uses native Polars expressions instead of rolling_map for better performance.
Buy Side Liquidity (BSL): Clusters of equal highs (stop losses of shorts)
Sell Side Liquidity (SSL): Clusters of equal lows (stop losses of longs)
Args:
df: DataFrame with OHLCV data
Returns:
DataFrame with liquidity columns:
- bsl_level: Buy side liquidity level
- ssl_level: Sell side liquidity level
- liquidity_sweep: True when liquidity is swept
"""
window_size = 20
# OPTIMIZED: Use rolling_std to detect price clusters
# Low standard deviation = prices are similar (potential liquidity zone)
# This is much faster than rolling_map with lambda
df = df.with_columns([
# Rolling std of highs - low std means similar prices (cluster)
pl.col("high")
.rolling_std(window_size=window_size)
.alias("_high_std"),
# Rolling std of lows
pl.col("low")
.rolling_std(window_size=window_size)
.alias("_low_std"),
# Rolling mean for reference
pl.col("high")
.rolling_mean(window_size=window_size)
.alias("_high_mean"),
pl.col("low")
.rolling_mean(window_size=window_size)
.alias("_low_mean"),
])
# Calculate coefficient of variation (std/mean) - lower = more clustered
# Threshold: if CV < 0.001 (0.1%), prices are very similar
cv_threshold = 0.001
df = df.with_columns([
# High cluster detection
pl.when(
(pl.col("_high_std") / pl.col("_high_mean")) < cv_threshold
)
.then(pl.col("high"))
.otherwise(None)
.alias("bsl_level"),
# Low cluster detection
pl.when(
(pl.col("_low_std") / pl.col("_low_mean")) < cv_threshold
)
.then(pl.col("low"))
.otherwise(None)
.alias("ssl_level"),
])
# Forward fill liquidity levels
df = df.with_columns([
pl.col("bsl_level").forward_fill().alias("_bsl_ff"),
pl.col("ssl_level").forward_fill().alias("_ssl_ff"),
])
# Detect liquidity sweeps
df = df.with_columns([
# BSL sweep: high goes above BSL then closes below
pl.when(
(pl.col("high") > pl.col("_bsl_ff").shift(1)) &
(pl.col("close") < pl.col("_bsl_ff").shift(1))
)
.then(pl.lit("BSL"))
.when(
(pl.col("low") < pl.col("_ssl_ff").shift(1)) &
(pl.col("close") > pl.col("_ssl_ff").shift(1))
)
.then(pl.lit("SSL"))
.otherwise(None)
.alias("liquidity_sweep"),
])
# Drop temporary columns
df = df.drop(["_high_std", "_low_std", "_high_mean", "_low_mean", "_bsl_ff", "_ssl_ff"])
return df
def generate_signal(self, df: pl.DataFrame) -> Optional[SMCSignal]:
"""
Generate trading signal based on SMC analysis.
Signal Logic (RELAXED for active trading):
1. Check market structure (BOS/CHoCH) - extended lookback
2. Find valid FVG OR Order Block in recent candles
3. Generate signal based on best available setup
Args:
df: DataFrame with all SMC indicators
Returns:
SMCSignal if valid setup found, None otherwise
"""
# Get latest row
if len(df) < 10:
return None
latest = df.tail(1)
current_close = latest["close"].item()
current_high = latest["high"].item()
current_low = latest["low"].item()
market_structure = latest["market_structure"].item() if "market_structure" in df.columns else 0
# Check for recent BOS/CHoCH (extended to 10 candles)
recent_df = df.tail(10)
recent_bos = recent_df["bos"].to_list() if "bos" in df.columns else []
recent_choch = recent_df["choch"].to_list() if "choch" in df.columns else []
# Check for FVG in recent candles (not just current)
recent_fvg_bull = recent_df["is_fvg_bull"].to_list() if "is_fvg_bull" in df.columns else []
recent_fvg_bear = recent_df["is_fvg_bear"].to_list() if "is_fvg_bear" in df.columns else []
# Get FVG zones from recent candles
fvg_bottoms = recent_df["fvg_bottom"].to_list() if "fvg_bottom" in df.columns else []
fvg_tops = recent_df["fvg_top"].to_list() if "fvg_top" in df.columns else []
# Check for Order Block in recent candles
recent_obs = recent_df["ob"].to_list() if "ob" in df.columns else []
ob_tops = recent_df["ob_top"].to_list() if "ob_top" in df.columns else []
ob_bottoms = recent_df["ob_bottom"].to_list() if "ob_bottom" in df.columns else []
# Get swing levels for SL
last_swing_high = latest["last_swing_high"].item() if "last_swing_high" in df.columns else None
last_swing_low = latest["last_swing_low"].item() if "last_swing_low" in df.columns else None
signal = None
# Determine if there's a recent bullish/bearish setup
has_bullish_break = 1 in recent_bos or 1 in recent_choch
has_bearish_break = -1 in recent_bos or -1 in recent_choch
has_bullish_fvg = any(recent_fvg_bull)
has_bearish_fvg = any(recent_fvg_bear)
has_bullish_ob = 1 in recent_obs
has_bearish_ob = -1 in recent_obs
# Get valid FVG/OB zone for entry
def get_valid_bullish_zone():
# Find most recent bullish FVG or OB
for i in range(len(recent_fvg_bull) - 1, -1, -1):
if recent_fvg_bull[i] and fvg_bottoms[i] is not None:
return fvg_bottoms[i], "FVG"
for i in range(len(recent_obs) - 1, -1, -1):
if recent_obs[i] == 1 and ob_bottoms[i] is not None:
return ob_bottoms[i], "OB"
return None, None
def get_valid_bearish_zone():
# Find most recent bearish FVG or OB
for i in range(len(recent_fvg_bear) - 1, -1, -1):
if recent_fvg_bear[i] and fvg_tops[i] is not None:
return fvg_tops[i], "FVG"
for i in range(len(recent_obs) - 1, -1, -1):
if recent_obs[i] == -1 and ob_tops[i] is not None:
return ob_tops[i], "OB"
return None, None
# Get ATR for dynamic SL/TP calculation
atr = latest["atr"].item() if "atr" in df.columns else current_close * 0.01 # Fallback 1%
min_sl_distance = 1.5 * atr # Minimum 1.5 ATR untuk SL
max_tp_distance = 4.0 * atr # Maximum 4 ATR untuk TP
# BULLISH SIGNAL CONDITIONS (RELAXED)
# Need: bullish structure OR recent bullish break, AND (FVG OR OB)
if ((market_structure == 1 or has_bullish_break) and
(has_bullish_fvg or has_bullish_ob)):
entry_zone, zone_type = get_valid_bullish_zone()
entry = entry_zone if entry_zone else current_close
# SL below swing low or ATR-based (use the FURTHER one to prevent whipsaw)
swing_sl = last_swing_low if last_swing_low and last_swing_low < entry else None
atr_sl = entry - min_sl_distance
if swing_sl:
# Use the further SL (more protection)
sl = min(swing_sl, atr_sl)
else:
sl = atr_sl
# TP at 2:1 RR minimum, capped at max distance
risk = entry - sl
tp = entry + (risk * 2)
# Cap TP at reasonable distance
if tp > entry + max_tp_distance:
tp = entry + max_tp_distance
# Confidence based on confirmations
conf = 0.55 # Base
if has_bullish_break:
conf += 0.1
if has_bullish_fvg:
conf += 0.1
if has_bullish_ob:
conf += 0.1
reason_parts = []
if has_bullish_break:
reason_parts.append("BOS/CHoCH")
if zone_type == "FVG":
reason_parts.append("FVG")
if zone_type == "OB":
reason_parts.append("OB")
signal = SMCSignal(
signal_type="BUY",
entry_price=entry,
stop_loss=sl,
take_profit=tp,
confidence=min(conf, 0.85),
reason="Bullish " + " + ".join(reason_parts),
)
# BEARISH SIGNAL CONDITIONS (RELAXED)
elif ((market_structure == -1 or has_bearish_break) and
(has_bearish_fvg or has_bearish_ob)):
entry_zone, zone_type = get_valid_bearish_zone()
entry = entry_zone if entry_zone else current_close
# SL above swing high or ATR-based (use the FURTHER one to prevent whipsaw)
swing_sl = last_swing_high if last_swing_high and last_swing_high > entry else None
atr_sl = entry + min_sl_distance
if swing_sl:
# Use the further SL (more protection)
sl = max(swing_sl, atr_sl)
else:
sl = atr_sl
# TP at 2:1 RR minimum, capped at max distance
risk = sl - entry
tp = entry - (risk * 2)
# Cap TP at reasonable distance
if tp < entry - max_tp_distance:
tp = entry - max_tp_distance
# Confidence based on confirmations
conf = 0.55 # Base
if has_bearish_break:
conf += 0.1
if has_bearish_fvg:
conf += 0.1
if has_bearish_ob:
conf += 0.1
reason_parts = []
if has_bearish_break:
reason_parts.append("BOS/CHoCH")
if zone_type == "FVG":
reason_parts.append("FVG")
if zone_type == "OB":
reason_parts.append("OB")
signal = SMCSignal(
signal_type="SELL",
entry_price=entry,
stop_loss=sl,
take_profit=tp,
confidence=min(conf, 0.85),
reason="Bearish " + " + ".join(reason_parts),
)
if signal:
logger.info(f"SMC Signal: {signal.signal_type} @ {signal.entry_price:.5f}, "
f"SL: {signal.stop_loss:.5f}, TP: {signal.take_profit:.5f}, "
f"RR: {signal.risk_reward:.2f}, Confidence: {signal.confidence:.2f}")
return signal
def calculate_smc_summary(df: pl.DataFrame) -> Dict:
"""
Calculate summary statistics for SMC analysis.
Args:
df: DataFrame with SMC indicators
Returns:
Dictionary with summary statistics
"""
summary = {
"total_bars": len(df),
"swing_highs": (df["swing_high"] == 1).sum() if "swing_high" in df.columns else 0,
"swing_lows": (df["swing_low"] == -1).sum() if "swing_low" in df.columns else 0,
"bullish_fvg": df["is_fvg_bull"].sum() if "is_fvg_bull" in df.columns else 0,
"bearish_fvg": df["is_fvg_bear"].sum() if "is_fvg_bear" in df.columns else 0,
"bullish_ob": (df["ob"] == 1).sum() if "ob" in df.columns else 0,
"bearish_ob": (df["ob"] == -1).sum() if "ob" in df.columns else 0,
"bullish_bos": (df["bos"] == 1).sum() if "bos" in df.columns else 0,
"bearish_bos": (df["bos"] == -1).sum() if "bos" in df.columns else 0,
"bullish_choch": (df["choch"] == 1).sum() if "choch" in df.columns else 0,
"bearish_choch": (df["choch"] == -1).sum() if "choch" in df.columns else 0,
}
# Current market structure
if "market_structure" in df.columns:
current_structure = df["market_structure"].tail(1).item()
summary["current_structure"] = "BULLISH" if current_structure == 1 else "BEARISH" if current_structure == -1 else "NEUTRAL"
return summary
if __name__ == "__main__":
# Test SMC analyzer with synthetic data
import numpy as np
from datetime import datetime, timedelta
# Create synthetic OHLCV data
np.random.seed(42)
n = 500
base_price = 2000.0
returns = np.random.randn(n) * 0.002
prices = base_price * np.exp(np.cumsum(returns))
df = pl.DataFrame({
"time": [datetime.now() - timedelta(minutes=15*i) for i in range(n-1, -1, -1)],
"open": prices,
"high": prices * (1 + np.abs(np.random.randn(n)) * 0.001),
"low": prices * (1 - np.abs(np.random.randn(n)) * 0.001),
"close": prices * (1 + np.random.randn(n) * 0.0005),
"volume": np.random.randint(1000, 10000, n),
})
# Initialize analyzer
analyzer = SMCAnalyzer(swing_length=5)
# Calculate all SMC indicators
df = analyzer.calculate_all(df)
# Print summary
summary = calculate_smc_summary(df)
print("\n=== SMC Analysis Summary ===")
for key, value in summary.items():
print(f"{key}: {value}")
# Generate signal
signal = analyzer.generate_signal(df)
if signal:
print(f"\n=== Trading Signal ===")
print(f"Type: {signal.signal_type}")
print(f"Entry: {signal.entry_price:.2f}")
print(f"SL: {signal.stop_loss:.2f}")
print(f"TP: {signal.take_profit:.2f}")
print(f"R:R: {signal.risk_reward:.2f}")
print(f"Confidence: {signal.confidence:.2%}")
print(f"Reason: {signal.reason}")
else:
print("\nNo valid signal")
# Show columns
print(f"\n=== DataFrame Columns ===")
print(df.columns)
Reference in New Issue View Git Blame Copy Permalink