Combining AI With EMA Strategies: Dynamic Lookback Tuning & Z-Score Filtering

import numpy as np
import pandas as pd
from xgboost import XGBClassifier

class AIEMACrossoverFilter:
    """
    Machine Learning Filter for Exponential Moving Average (EMA) Crossover Strategies.
    Evaluates raw EMA crossover triggers against order book liquidity, CVD volume delta,
    and spatial Z-score distance to eliminate false breakout signals during sideways chop.
    """
    def __init__(self, probability_threshold: float = 0.70):
        self.probability_threshold = probability_threshold
        self.model = XGBClassifier(
            n_estimators=200,
            max_depth=4,
            learning_rate=0.03,
            subsample=0.8,
            random_state=42
        )

    def extract_ema_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Calculates stationary features relative to EMA indicator vectors.
        """
        features = pd.DataFrame(index=df.index)
        
        # 1. Standard Exponential Moving Averages
        df['ema_9'] = df['close'].ewm(span=9, adjust=False).mean()
        df['ema_21'] = df['close'].ewm(span=21, adjust=False).mean()
        df['ema_200'] = df['close'].ewm(span=200, adjust=False).mean()

        # 2. Normalized EMA Distance Z-Score (Spatial Dispersion)
        ema_spread = df['ema_9'] - df['ema_21']
        spread_mean = ema_spread.rolling(window=30).mean()
        spread_std = ema_spread.rolling(window=30).std()
        features['ema_z_score'] = (ema_spread - spread_mean) / (spread_std + 1e-8)

        # 3. Macro Trend Alignment (Distance from 200 EMA)
        features['dist_200_ema'] = (df['close'] - df['ema_200']) / df['ema_200']

        # 4. Cumulative Volume Delta (CVD) Acceleration Rate
        buy_sell_delta = df['buy_volume'] - df['sell_volume']
        features['cvd_slope_10'] = buy_sell_delta.rolling(10).sum() / (df['volume'].rolling(10).sum() + 1e-8)

        # 5. Volatility Expansion Metric (ATR ratio)
        high_low = df['high'] - df['low']
        atr_14 = high_low.rolling(14).mean()
        features['atr_ratio'] = atr_14 / df['close']

        return features.dropna()

    def evaluate_crossover_signal(self, live_features: pd.DataFrame, is_bullish_cross: bool) -> dict:
        """
        Processes a live EMA crossover setup and generates AI validation status.
        """
        if not is_bullish_cross:
            return {"status": "NO_CROSSOVER", "probability": 0.0, "execute": False}

        # Predict probability of sustainable macro trend extension
        win_probability = float(self.model.predict_proba(live_features)[0][1])
        should_execute = win_probability >= self.probability_threshold

        return {
            "status": "APPROVED" if should_execute else "FILTERED_OUT",
            "confidence_score": round(win_probability * 100, 2),
            "execute": should_execute,
            "filter_reason": "High probability macro alignment" if should_execute else "Low volume delta or structural chop trap"
        }

import numpy as np
import pandas as pd

def calculate_adaptive_ai_ema(df: pd.DataFrame, base_span: int = 20, min_span: int = 8, max_span: int = 50) -> pd.Series:
    """
    Computes an Adaptive Dynamic EMA lookback window based on real-time market volatility.
    When volatility compresses (chop), lookback expands to prevent whipsaws.
    When momentum accelerates (breakout), lookback shortens for instant responsiveness.
    """
    # Calculate Efficiency Ratio (Kaufman ER concept adapted for ML pipelines)
    change = (df['close'] - df['close'].shift(10)).abs()
    volatility = (df['close'] - df['close'].shift(1)).abs().rolling(10).sum()
    efficiency_ratio = change / (volatility + 1e-8)

    # Dynamic smoothing multiplier calculation
    fast_alpha = 2.0 / (min_span + 1.0)
    slow_alpha = 2.0 / (max_span + 1.0)
    
    # Scale alpha based on market efficiency ratio
    adaptive_alpha = (efficiency_ratio * (fast_alpha - slow_alpha) + slow_alpha) ** 2

    # Compute dynamic adaptive EMA series
    adaptive_ema = np.zeros(len(df))
    adaptive_ema[0] = df['close'].iloc[0]

    for i in range(1, len(df)):
        alpha = adaptive_alpha.iloc[i]
        adaptive_ema[i] = alpha * df['close'].iloc[i] + (1 - alpha) * adaptive_ema[i - 1]

    return pd.Series(adaptive_ema, index=df.index)

Role: Quantitative Algorithmic Trend Validator
Context: A primary crossover signal has occurred on the 15-minute chart for the BTC/USDT pair (9 EMA has crossed above the 21 EMA). You must analyze multi-timeframe structural data to verify if this cross represents a highly sustainable macro expansion.

Input Parameters for Validation:
- Target Asset Class: BTC/USDT
- Immediate Signal Profile: 15-Min Bullish Crossover (9 EMA / 21 EMA)
- 4-Hour Timeframe Structural State: Price is trading safely above the 200 EMA line; macro structure is upward trending.
- Volume Profile State: Current candle volume registers 140% above the 24-hour rolling average baseline.
- Derivative Funding Rate State: Funding rate is highly neutral, indicating no excessive retail leverage saturation.

Analysis Directives:
1. Confirm the trend as "SUSTAINABLE" if the 4-Hour macro trend aligns with the 15-minute breakout, and funding variables reveal low leverage saturation.
2. If the higher timeframe data shows the asset is trading directly below major historical 200 EMA resistance, flag the pattern as a potential false breakout trap and return "ABORT".

Output Constraints:
Return strictly a valid, minified JSON object payload. Do not include conversational text, markdown code block backticks, or introduction statements.

Target JSON Layout:
{
  "trend_validated": boolean,
  "confidence_coefficient": float, // Value scaled from 0.0 to 100.0
  "risk_grade": "LOW" | "MEDIUM" | "HIGH",
  "recommended_stop_loss_coordinate": "EMA_21" | "SWING_LOW" | "INVALIDATE",
  "structural_summary": "STRING"
}