advanced_candle_pattern
Identify cryptocurrency trading opportunities by detecting progressive candle size increase patterns across multiple timeframes to reveal momentum shifts.
Instructions
Advanced candle pattern analysis using multi-timeframe data.
Args:
exchange: Exchange name (BINANCE, KUCOIN, etc.)
base_timeframe: Base timeframe for analysis (5m, 15m, 1h, 4h)
pattern_length: Number of consecutive periods to analyze (2-4)
min_size_increase: Minimum percentage increase in candle size
limit: Maximum number of results to return
Returns:
Coins with progressive candle size increase patternsInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| exchange | No | KUCOIN | |
| base_timeframe | No | 15m | |
| pattern_length | No | ||
| min_size_increase | No | ||
| limit | No |
Implementation Reference
- src/tradingview_mcp/server.py:678-802 (handler)Main handler function implementing the advanced_candle_pattern tool logic. Registered via @mcp.tool() decorator. Performs advanced candle pattern scanning using multi-timeframe data from TradingView, with fallback single timeframe analysis and scoring.
@mcp.tool() def advanced_candle_pattern( exchange: str = "KUCOIN", base_timeframe: str = "15m", pattern_length: int = 3, min_size_increase: float = 10.0, limit: int = 15 ) -> dict: """Advanced candle pattern analysis using multi-timeframe data. Args: exchange: Exchange name (BINANCE, KUCOIN, etc.) base_timeframe: Base timeframe for analysis (5m, 15m, 1h, 4h) pattern_length: Number of consecutive periods to analyze (2-4) min_size_increase: Minimum percentage increase in candle size limit: Maximum number of results to return Returns: Coins with progressive candle size increase patterns """ try: exchange = sanitize_exchange(exchange, "KUCOIN") base_timeframe = sanitize_timeframe(base_timeframe, "15m") pattern_length = max(2, min(4, pattern_length)) min_size_increase = max(5.0, min(50.0, min_size_increase)) limit = max(1, min(30, limit)) # Get symbols symbols = load_symbols(exchange) if not symbols: return { "error": f"No symbols found for exchange: {exchange}", "exchange": exchange } # Limit for performance symbols = symbols[:min(limit * 2, 100)] # Use tradingview-screener for multi-timeframe data if available if TRADINGVIEW_SCREENER_AVAILABLE: try: # Get multiple timeframe data using screener results = _fetch_multi_timeframe_patterns( exchange, symbols, base_timeframe, pattern_length, min_size_increase ) return { "exchange": exchange, "base_timeframe": base_timeframe, "pattern_length": pattern_length, "min_size_increase": min_size_increase, "method": "multi-timeframe", "total_found": len(results), "data": results[:limit] } except Exception as e: # Fallback to single timeframe analysis pass # Fallback: Use single timeframe with enhanced pattern detection screener = EXCHANGE_SCREENER.get(exchange, "crypto") analysis = get_multiple_analysis( screener=screener, interval=base_timeframe, symbols=symbols ) pattern_results = [] for symbol, data in analysis.items(): if data is None: continue try: indicators = data.indicators # Enhanced pattern detection using available indicators pattern_score = _calculate_candle_pattern_score( indicators, pattern_length, min_size_increase ) if pattern_score['detected']: metrics = compute_metrics(indicators) result = { "symbol": symbol, "pattern_score": pattern_score['score'], "pattern_details": pattern_score['details'], "current_price": pattern_score['price'], "total_change": pattern_score['total_change'], "volume": indicators.get("volume", 0), "bollinger_rating": metrics.get('rating', 0) if metrics else 0, "technical_strength": { "rsi": round(indicators.get("RSI", 50), 2), "momentum": "Strong" if abs(pattern_score['total_change']) > min_size_increase else "Moderate", "volume_trend": "High" if indicators.get("volume", 0) > 10000 else "Low" } } pattern_results.append(result) except Exception as e: continue # Sort by pattern score and total change pattern_results.sort(key=lambda x: (x['pattern_score'], abs(x['total_change'])), reverse=True) return { "exchange": exchange, "base_timeframe": base_timeframe, "pattern_length": pattern_length, "min_size_increase": min_size_increase, "method": "enhanced-single-timeframe", "total_found": len(pattern_results), "data": pattern_results[:limit] } except Exception as e: return { "error": f"Advanced pattern analysis failed: {str(e)}", "exchange": exchange, "base_timeframe": base_timeframe } - Helper function to calculate the candle pattern score used in the advanced_candle_pattern tool. Analyzes candle characteristics and technical indicators to determine pattern strength.
def _calculate_candle_pattern_score(indicators: dict, pattern_length: int, min_increase: float) -> dict: """Calculate candle pattern score based on available indicators.""" try: open_price = indicators.get("open", 0) close_price = indicators.get("close", 0) high_price = indicators.get("high", 0) low_price = indicators.get("low", 0) volume = indicators.get("volume", 0) rsi = indicators.get("RSI", 50) if not all([open_price, close_price, high_price, low_price]): return {"detected": False, "score": 0} # Current candle analysis candle_body = abs(close_price - open_price) candle_range = high_price - low_price body_ratio = candle_body / candle_range if candle_range > 0 else 0 # Price change price_change = ((close_price - open_price) / open_price) * 100 # Pattern scoring score = 0 details = [] # Strong candle body if body_ratio > 0.7: score += 2 details.append("Strong candle body") elif body_ratio > 0.5: score += 1 details.append("Moderate candle body") # Significant price movement if abs(price_change) >= min_increase: score += 2 details.append(f"Strong momentum ({price_change:.1f}%)") elif abs(price_change) >= min_increase / 2: score += 1 details.append(f"Moderate momentum ({price_change:.1f}%)") # Volume confirmation if volume > 5000: score += 1 details.append("Good volume") # RSI momentum if (price_change > 0 and 50 < rsi < 80) or (price_change < 0 and 20 < rsi < 50): score += 1 details.append("RSI momentum aligned") # Trend consistency (using EMA vs price) ema50 = indicators.get("EMA50", close_price) if (price_change > 0 and close_price > ema50) or (price_change < 0 and close_price < ema50): score += 1 details.append("Trend alignment") detected = score >= 3 # Minimum threshold return { "detected": detected, "score": score, "details": details, "price": round(close_price, 6), "total_change": round(price_change, 3), "body_ratio": round(body_ratio, 3), "volume": volume } except Exception as e: return {"detected": False, "score": 0, "error": str(e)} - Helper function for fetching multi-timeframe OHLC data using tradingview_screener API and applying pattern scoring for the advanced_candle_pattern tool.
def _fetch_multi_timeframe_patterns(exchange: str, symbols: List[str], base_tf: str, length: int, min_increase: float) -> List[dict]: """Fetch multi-timeframe pattern data using tradingview-screener.""" try: from tradingview_screener import Query from tradingview_screener.column import Column # Map timeframe to TradingView format tf_map = {"5m": "5", "15m": "15", "1h": "60", "4h": "240", "1D": "1D"} tv_interval = tf_map.get(base_tf, "15") # Create query for OHLC data cols = [ f"open|{tv_interval}", f"close|{tv_interval}", f"high|{tv_interval}", f"low|{tv_interval}", f"volume|{tv_interval}", "RSI" ] q = Query().set_markets("crypto").select(*cols) q = q.where(Column("exchange") == exchange.upper()) q = q.limit(len(symbols)) total, df = q.get_scanner_data() if df is None or df.empty: return [] results = [] for _, row in df.iterrows(): symbol = row.get("ticker", "") try: open_val = row.get(f"open|{tv_interval}") close_val = row.get(f"close|{tv_interval}") high_val = row.get(f"high|{tv_interval}") low_val = row.get(f"low|{tv_interval}") volume_val = row.get(f"volume|{tv_interval}", 0) rsi_val = row.get("RSI", 50) if not all([open_val, close_val, high_val, low_val]): continue # Calculate pattern metrics pattern_score = _calculate_candle_pattern_score({ "open": open_val, "close": close_val, "high": high_val, "low": low_val, "volume": volume_val, "RSI": rsi_val }, length, min_increase) if pattern_score['detected']: results.append({ "symbol": symbol, "pattern_score": pattern_score['score'], "price": pattern_score['price'], "change": pattern_score['total_change'], "body_ratio": pattern_score['body_ratio'], "volume": volume_val, "rsi": round(rsi_val, 2), "details": pattern_score['details'] }) except Exception as e: continue return sorted(results, key=lambda x: x['pattern_score'], reverse=True) except Exception as e: return []