Market Structure Shifts: BOS and ChoCh Detection

import yfinance as yf
import pandas as pd
import numpy as np

df = yf.download('BTC-USD', period='6mo', progress=False)

def find_swing_points(df, window=5):
    """Find swing highs and lows."""
    df['Swing_High'] = False
    df['Swing_Low'] = False

    for i in range(window, len(df) - window):
        # Swing High
        if df['High'].iloc[i] == df['High'].iloc[i-window:i+window+1].max():
            df.loc[df.index[i], 'Swing_High'] = True

        # Swing Low
        if df['Low'].iloc[i] == df['Low'].iloc[i-window:i+window+1].min():
            df.loc[df.index[i], 'Swing_Low'] = True

    return df

def detect_structure_breaks(df):
    """
    Detect Break of Structure (BOS) and Change of Character (ChoCh).
    """
    df['BOS'] = None
    df['ChoCh'] = None

    # Get swing points
    swing_highs = df[df['Swing_High']]['High'].to_dict()
    swing_lows = df[df['Swing_Low']]['Low'].to_dict()

    swing_high_dates = list(swing_highs.keys())
    swing_low_dates = list(swing_lows.keys())

    # Track last significant swings
    for i in range(len(df)):
        current_date = df.index[i]

        # Get recent swings before this date
        recent_highs = [swing_highs[d] for d in swing_high_dates if d < current_date]
        recent_lows = [swing_lows[d] for d in swing_low_dates if d < current_date]

        if len(recent_highs) < 2 or len(recent_lows) < 2:
            continue

        last_high = recent_highs[-1]
        prev_high = recent_highs[-2]
        last_low = recent_lows[-1]
        prev_low = recent_lows[-2]

        current_high = df['High'].iloc[i]
        current_low = df['Low'].iloc[i]

        # Bullish BOS: Price breaks above last swing high in uptrend
        if last_high > prev_high and last_low > prev_low:  # Uptrend
            if current_high > last_high:
                df.loc[current_date, 'BOS'] = 'Bullish'

        # Bearish BOS: Price breaks below last swing low in downtrend
        if last_high < prev_high and last_low < prev_low:  # Downtrend
            if current_low < last_low:
                df.loc[current_date, 'BOS'] = 'Bearish'

        # Bullish ChoCh: Price breaks above last lower high (trend reversal)
        if last_high < prev_high and last_low < prev_low:  # Downtrend
            if current_high > last_high:
                df.loc[current_date, 'ChoCh'] = 'Bullish'

        # Bearish ChoCh: Price breaks below last higher low (trend reversal)
        if last_high > prev_high and last_low > prev_low:  # Uptrend
            if current_low < last_low:
                df.loc[current_date, 'ChoCh'] = 'Bearish'

    return df

# Apply detection
df = find_swing_points(df, window=5)
df = detect_structure_breaks(df)

print("Structure Break Detection:")
print(f"Bullish BOS: {(df['BOS'] == 'Bullish').sum()}")
print(f"Bearish BOS: {(df['BOS'] == 'Bearish').sum()}")
print(f"Bullish ChoCh: {(df['ChoCh'] == 'Bullish').sum()}")
print(f"Bearish ChoCh: {(df['ChoCh'] == 'Bearish').sum()}")

# Show recent breaks
recent_bos = df[df['BOS'].notna()].tail(5)
print("\nRecent BOS Events:")
print(recent_bos[['Close', 'BOS']])

recent_choch = df[df['ChoCh'].notna()].tail(5)
print("\nRecent ChoCh Events:")
print(recent_choch[['Close', 'ChoCh']])

# Calculate returns after structure breaks
df['Return_5d'] = df['Close'].pct_change(5).shift(-5) * 100
df['Return_10d'] = df['Close'].pct_change(10).shift(-10) * 100

def test_signal_performance(df, signal_column, signal_value):
    """Test returns following a signal."""
    signals = df[df[signal_column] == signal_value].copy()

    if len(signals) == 0:
        return None

    return {
        'signal': f"{signal_column}={signal_value}",
        'count': len(signals),
        'avg_return_5d': signals['Return_5d'].mean(),
        'avg_return_10d': signals['Return_10d'].mean(),
        'win_rate_5d': (signals['Return_5d'] > 0).sum() / len(signals) * 100,
        'win_rate_10d': (signals['Return_10d'] > 0).sum() / len(signals) * 100
    }

# Test all signal types
signals = [
    ('BOS', 'Bullish'),
    ('BOS', 'Bearish'),
    ('ChoCh', 'Bullish'),
    ('ChoCh', 'Bearish')
]

print("Structure Break Performance Analysis:")
print(f"{'Signal':<20} {'N':>4} {'5d Ret':>8} {'10d Ret':>8} {'Win% 5d':>9} {'Win% 10d':>10}")
print("-" * 75)

for col, val in signals:
    stats = test_signal_performance(df, col, val)
    if stats:
        print(f"{stats['signal']:<20} {stats['count']:>4} "
              f"{stats['avg_return_5d']:>7.2f}% {stats['avg_return_10d']:>7.2f}% "
              f"{stats['win_rate_5d']:>8.1f}% {stats['win_rate_10d']:>9.1f}%")

def enhanced_structure_signals(df):
    """
    Enhanced structure break signals with confirmation filters.

    Filters:
    1. Volume > 1.5x average
    2. Strong momentum (RSI)
    3. Near liquidity zone
    """
    # Calculate filters
    df['Volume_MA'] = df['Volume'].rolling(20).mean()
    df['Volume_Ratio'] = df['Volume'] / df['Volume_MA']

    # RSI
    delta = df['Close'].diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
    rs = gain / loss
    df['RSI'] = 100 - (100 / (1 + rs))

    # Enhanced signals
    df['Enhanced_Bullish'] = (
        ((df['BOS'] == 'Bullish') | (df['ChoCh'] == 'Bullish')) &
        (df['Volume_Ratio'] > 1.5) &
        (df['RSI'] > 50)
    )

    df['Enhanced_Bearish'] = (
        ((df['BOS'] == 'Bearish') | (df['ChoCh'] == 'Bearish')) &
        (df['Volume_Ratio'] > 1.5) &
        (df['RSI'] < 50)
    )

    return df

# Apply enhanced signals
df = enhanced_structure_signals(df)

print("Enhanced Signal Performance:")
print(f"Enhanced Bullish signals: {df['Enhanced_Bullish'].sum()}")
print(f"Enhanced Bearish signals: {df['Enhanced_Bearish'].sum()}")

# Test enhanced signals
enhanced_bull_stats = test_signal_performance(df, 'Enhanced_Bullish', True)
enhanced_bear_stats = test_signal_performance(df, 'Enhanced_Bearish', True)

print("\nEnhanced Bullish Performance:")
if enhanced_bull_stats:
    print(f"  Count: {enhanced_bull_stats['count']}")
    print(f"  5d Return: {enhanced_bull_stats['avg_return_5d']:.2f}%")
    print(f"  10d Return: {enhanced_bull_stats['avg_return_10d']:.2f}%")
    print(f"  Win Rate 5d: {enhanced_bull_stats['win_rate_5d']:.1f}%")
    print(f"  Win Rate 10d: {enhanced_bull_stats['win_rate_10d']:.1f}%")

print("\nEnhanced Bearish Performance:")
if enhanced_bear_stats:
    print(f"  Count: {enhanced_bear_stats['count']}")
    print(f"  5d Return: {enhanced_bear_stats['avg_return_5d']:.2f}%")
    print(f"  10d Return: {enhanced_bear_stats['avg_return_10d']:.2f}%")
    print(f"  Win Rate 5d: {enhanced_bear_stats['win_rate_5d']:.1f}%")
    print(f"  Win Rate 10d: {enhanced_bear_stats['win_rate_10d']:.1f}%")