strategy_3_trend_range_volume.py

# Strategy 3: Trend/Range/Volume Engine
# Primary Timeframe: 15m (for LR, ATR, STDDEV)
# Requires: technicalindicators (TA-Lib or similar)

from typing import List, Dict, Any
import numpy as np
# We will use a simple implementation for TA functions, but in the final system, 
# we should use a robust library like TA-Lib or pandas_ta.
# For now, we replicate the logic based on the Node.js script's intent.

# --- Core Indicator Functions (Simplified for Logic Extraction) ---

def calculate_atr(highs: List[float], lows: List[float], closes: List[float], period: int = 14) -> float:
    """Simplified ATR calculation (returns the last value)."""
    # In a real implementation, this would use a library.
    # For logic extraction, we assume we get the final ATR value.
    # Placeholder: returns a dummy value or average range for now.
    if len(closes) < period: return 0.0
    return np.mean([h - l for h, l in zip(highs[-period:], lows[-period:])])

def calculate_linear_regression_slope(closes: List[float], period: int = 50) -> float:
    """Simplified Linear Regression Slope calculation (returns the slope)."""
    if len(closes) < period: return 0.0
    
    y = np.array(closes[-period:])
    x = np.arange(period)
    
    # Linear regression: y = mx + c
    # m = slope
    m, c = np.polyfit(x, y, 1)
    return m

def calculate_stddev(closes: List[float], period: int = 50) -> float:
    """Standard Deviation (used for range detection)."""
    if len(closes) < period: return 0.0
    return np.std(closes[-period:])

# --- Strategy Core Functions ---

def check_trend(slope: float, price: float, atr: float) -> str:
    """Determines trend based on LR slope relative to price/ATR."""
    # The original script's logic for trend strength is complex.
    # We simplify it to: positive slope = LONG, negative slope = SHORT, near zero = RANGE.
    
    # Threshold based on ATR (a common way to normalize slope)
    # If slope is > 0.1 * ATR, it's a strong trend.
    trend_threshold = 0.1 * atr
    
    if slope > trend_threshold:
        return 'LONG'
    elif slope < -trend_threshold:
        return 'SHORT'
    else:
        return 'RANGE'

def check_range(atr: float, stddev_val: float, close: float, range_atr_mult: float = 0.5) -> bool:
    """Checks if the market is in a tight range."""
    # Range is detected if ATR is small relative to price, or if STDDEV is small.
    # Original logic: ATR small relative to price -> range
    
    # Normalize ATR by price
    normalized_atr = atr / close
    
    # If normalized ATR is below a threshold, it's a tight range.
    # We use the original RANGE_ATR_MULT (0.5%) as a proxy for the threshold.
    return normalized_atr < range_atr_mult / 100.0

def check_volume_spike(volumes: List[float], current_volume: float, multiplier: float = 2.0) -> bool:
    """Checks for a volume spike (current volume > multiplier * average volume)."""
    if len(volumes) < 10: return False
    
    avg_volume = np.mean(volumes[-10:-1]) # Average of last 9 candles (excluding current)
    return current_volume > multiplier * avg_volume

# --- Main Analysis Function ---

def analyze_strategy_3(klines: List[Dict[str, float]], 
                       lr_period: int = 50, 
                       atr_period: int = 14, 
                       stddev_period: int = 50,
                       volume_spike_mult: float = 2.0,
                       range_atr_mult: float = 0.5) -> Dict[str, Any]:
    """
    Analyzes a symbol using the Trend/Range/Volume Engine strategy.
    
    Args:
        klines: List of candle data (must contain 'close', 'high', 'low', 'volume').
    
    Returns:
        A dictionary with 'signal' ('BUY', 'SELL', 'NO_SIGNAL') and debug data.
    """
    
    if len(klines) < max(lr_period, atr_period, stddev_period) + 1:
        return {'signal': 'NO_DATA'}

    closes = [k['close'] for k in klines]
    highs = [k['high'] for k in klines]
    lows = [k['low'] for k in klines]
    volumes = [k['volume'] for k in klines]
    
    current_close = closes[-1]
    current_volume = volumes[-1]

    # 1. Calculate Indicators
    atr = calculate_atr(highs, lows, closes, atr_period)
    lr_slope = calculate_linear_regression_slope(closes, lr_period)
    stddev_val = calculate_stddev(closes, stddev_period)

    # 2. Check Trend and Range
    trend = check_trend(lr_slope, current_close, atr)
    is_in_range = check_range(atr, stddev_val, current_close, range_atr_mult)
    volume_spiked = check_volume_spike(volumes, current_volume, volume_spike_mult)

    # 3. Generate Signal (Replicating the core logic intent)
    signal = 'NO_SIGNAL'
    
    # Simplified Logic:
    # If in a strong LONG trend AND volume spike, BUY.
    # If in a strong SHORT trend AND volume spike, SELL.
    # If in a tight RANGE, look for mean reversion (not explicitly defined, so we skip for now).
    
    if trend == 'LONG' and volume_spiked:
        # This is a simplified interpretation. The original script had more complex confirmations 
        # (Footprint, Depth, Maker-Taker) which are impossible to replicate without the full data 
        # and the specific logic. We focus on the core: Trend + Volume.
        signal = 'BUY'
    elif trend == 'SHORT' and volume_spiked:
        signal = 'SELL'
    elif is_in_range:
        # In a range, we might look for reversal signals. 
        # Since the original logic is missing, we'll keep it NO_SIGNAL for safety.
        pass

    return {
        'signal': signal,
        'trend': trend,
        'is_in_range': is_in_range,
        'volume_spiked': volume_spiked,
        'lr_slope': lr_slope,
        'atr': atr,
        'stddev': stddev_val
    }

if __name__ == '__main__':
    print("Strategy 3 (Trend/Range/Volume) Analysis Logic Extracted.")
    print("This script contains the pure logic and will be integrated into the main Python system.")