diff --git a/mt5_h4_pattern_scanner.py b/mt5_h4_pattern_scanner.py deleted file mode 100644 index 182919d..0000000 --- a/mt5_h4_pattern_scanner.py +++ /dev/null @@ -1,2871 +0,0 @@ -#!/usr/bin/env python3 -""" -MT5 Multi-Timeframe Candlestick Pattern Scanner & Backtester v6 -=============================================================== -Expanded from v5: supports M5, M15, H1, H4, and D1 timeframes for both -live scanning and backtesting. All parameters are consolidated near the top. - -Must run on Windows with MT5 installed. -Credentials are loaded exclusively from the .env file in the same directory. - -Install: pip install MetaTrader5 pandas numpy colorama python-dotenv - -Usage: - # Live scanner — all 5 timeframes (default) - python mt5_multitf_pattern_scanner_v6.py - - # Live scanner — specific timeframes only - python mt5_multitf_pattern_scanner_v6.py --timeframes M5 H1 H4 - - # One-shot scan of latest closed candle on all timeframes - python mt5_multitf_pattern_scanner_v6.py --mode scan - - # Quick backtest (last 500 bars on H4) - python mt5_multitf_pattern_scanner_v6.py --mode backtest --bars 500 - - # Full backtest on one timeframe - python mt5_multitf_pattern_scanner_v6.py --mode fullbacktest --timeframes H4 - - # Full backtest on ALL timeframes, date-ranged - python mt5_multitf_pattern_scanner_v6.py --mode fullbacktest --from 2024-01-01 --to 2024-12-31 - - # Full backtest with filters - python mt5_multitf_pattern_scanner_v6.py --mode fullbacktest \\ - --d1-trend-filter --volume-filter --forward 15 --sl 1.5 --tp 1.5 - - # Live scanner with custom account sizing - python mt5_multitf_pattern_scanner_v6.py --mode live --account-balance 25000 --risk-percent 0.5 -""" - -import MetaTrader5 as mt5 -import pandas as pd -import numpy as np -from datetime import datetime, timedelta -import argparse -import time -import os -import sys -import json -import glob -import re -import warnings -warnings.filterwarnings('ignore') - -# ── Load credentials from .env file (REQUIRED) ───────────────────── -try: - from dotenv import load_dotenv - _dotenv_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), '.env') - if os.path.exists(_dotenv_path): - load_dotenv(_dotenv_path) - _ENV_LOADED = True - else: - _ENV_LOADED = False -except ImportError: - _ENV_LOADED = False - -# ── Color output for Windows terminal ────────────────────────────── -try: - from colorama import init, Fore, Style - init(autoreset=True) - _COLORAMA = True -except ImportError: - _COLORAMA = False - -def C(color, text): - """Return colour-wrapped text if colorama available, else plain text.""" - if not _COLORAMA: - return str(text) - _MAP = { - 'green': Fore.LIGHTGREEN_EX, - 'red': Fore.LIGHTRED_EX, - 'yellow': Fore.YELLOW, - 'cyan': Fore.CYAN, - 'blue': Fore.LIGHTBLUE_EX, - 'magenta': Fore.MAGENTA, - 'white': Fore.WHITE, - 'dim': Fore.BLACK, - 'bold': Style.BRIGHT, - 'reset': Style.RESET_ALL, - } - c = _MAP.get(color, '') - return f"{c}{text}{Style.RESET_ALL}" - - -# ============================================================ -# CONFIGURATION — ALL PARAMETERS IN ONE PLACE -# ============================================================ - -# ── MT5 Credentials (from .env — never hardcode) ─────────────────── -_MT5_PATH = os.getenv('MT5_PATH', r"C:\Program Files\Capital Point Trading MT5 Terminal\terminal64.exe") -_MT5_ACCOUNT = int(os.getenv('MT5_ACCOUNT', '52598748')) -_MT5_PASSWORD = os.getenv('MT5_PASSWORD', '') -_MT5_SERVER = os.getenv('MT5_SERVER', 'CapitalPointTrading-Demo') - -# ── Timeframe Map: label → MT5 constant + candle duration (minutes) ─ -TIMEFRAME_MAP = { - 'M5': {'mt5_tf': mt5.TIMEFRAME_M5, 'minutes': 5, 'label': 'M5'}, - 'M15': {'mt5_tf': mt5.TIMEFRAME_M15, 'minutes': 15, 'label': 'M15'}, - 'H1': {'mt5_tf': mt5.TIMEFRAME_H1, 'minutes': 60, 'label': 'H1'}, - 'H4': {'mt5_tf': mt5.TIMEFRAME_H4, 'minutes': 240, 'label': 'H4'}, - 'D1': {'mt5_tf': mt5.TIMEFRAME_D1, 'minutes': 1440, 'label': 'D1'}, -} - -CFG = { - # ── MT5 Connection ───────────────────────────────────────────── - 'mt5_path': _MT5_PATH, - 'account': _MT5_ACCOUNT, - 'password': _MT5_PASSWORD, - 'server': _MT5_SERVER, - - # ── Symbol & Timeframes ──────────────────────────────────────── - 'symbol': "EURUSD", - # Active timeframes for live scan & backtest. All 5 available: - # 'M5', 'M15', 'H1', 'H4', 'D1' - 'active_timeframes': ['M5', 'M15', 'H1', 'H4', 'D1'], - # Timeframe used as the D1 trend-filter source (should be >= 'H4') - 'trend_filter_tf': 'D1', - - # ── ATR / SL / TP ───────────────────────────────────────────── - 'atr_period': 14, - 'sl_multiplier': 1.5, - 'tp_multiplier': 1.5, # R:R = tp_multiplier / sl_multiplier - # Higher-timeframe ATR source per trading TF. - # M5/M15 ATR is tiny → use H1 ATR for SL/TP sizing on fast TFs. - # Set to None or same as trading TF to use native ATR. - 'atr_tf_by_tf': { - 'M5': 'H1', # Use H1 ATR for M5 signals (much wider, more realistic SL/TP) - 'M15': 'H1', # Use H1 ATR for M15 signals - 'H1': 'H1', # Native - 'H4': 'H4', # Native (H4 ATR is already meaningful) - 'D1': 'D1', # Native - }, - - # ── Pattern Detection Thresholds ────────────────────────────── - 'doji_body_ratio': 0.1, - 'spinning_top_body_ratio': 0.3, - 'marubozu_wick_ratio': 0.05, - 'hammer_lower_wick_ratio': 2.0, - 'hammer_upper_wick_ratio': 0.3, - 'long_candle_ratio': 0.7, - 'small_candle_ratio': 0.35, - 'tweezer_tolerance_pips': 3, - 'engulf_tolerance_pips': 2.0, - - # ── Trend Detection ──────────────────────────────────────────── - 'trend_lookback': 20, # SMA-based lookback bars - - # ── Forward Evaluation (per-timeframe multiples of candle duration) - # Default forward candles. For fast TFs this is auto-scaled in - # run_full_backtest() so the evaluation window is always ~60 hours. - 'default_forward_candles': 15, # used as-is for H4 (60 h) - # Per-timeframe forward candle overrides (set 0 to use auto-scaling) - 'forward_candles_by_tf': { - 'M5': 720, # 720 × 5 min = 60 h - 'M15': 240, # 240 × 15 min = 60 h - 'H1': 60, # 60 × 1 h = 60 h - 'H4': 15, # 15 × 4 h = 60 h - 'D1': 5, # 5 × 1 day = 5 days (~1 trading week) - }, - - # ── Full Backtest ────────────────────────────────────────────── - 'max_r_levels': 5, - 'pip_divisor': 0.0001, - 'warmup_bars': 30, - - # ── Live Scanner ─────────────────────────────────────────────── - 'bars_to_fetch': 50, # bars fetched per TF for pattern context - # Seconds between poll cycles per timeframe - # M5/M15 poll more frequently; D1 can poll once per minute - 'poll_interval_by_tf': { - 'M5': 15, - 'M15': 30, - 'H1': 30, - 'H4': 30, - 'D1': 60, - }, - - # ── Session Classifier ───────────────────────────────────────── - 'broker_utc_offset': 2, # UTC+2 broker server time - - # ── Signal Deduplication & Entry Verification ───────────────── - 'deduplicate_signals': True, - 'verify_entry': True, - - # ── Volume Confirmation ──────────────────────────────────────── - 'volume_filter': False, - 'volume_ma_period': 20, - 'volume_threshold': 1.0, # signal candle vol >= threshold × avg - - # ── D1 Trend Filter ──────────────────────────────────────────── - 'd1_trend_filter': True, - 'd1_sma_period': 20, - - # ── Auto-Reconnect ───────────────────────────────────────────── - 'max_reconnect_attempts': 5, - 'reconnect_backoff_base': 10, # seconds, doubles each retry - - # ── Live Stats Integration ───────────────────────────────────── - 'stats_cache_hours': 4, - 'min_signals_for_stats': 5, - 'min_historical_win_rate': 50.0, - - # ── Live Signal Filtering ────────────────────────────────────── - 'min_signal_score': 55.0, # 0 = disabled - 'alert_only_strong': True, - 'show_dashboard_on_start': True, - - # ── Position Sizing ──────────────────────────────────────────── - 'account_balance': 100000, - 'risk_percent': 1.0, # % of account per trade -} - -# ── Derived Paths ─────────────────────────────────────────────────── -_LOG_DIR = os.path.dirname(os.path.abspath(__file__)) -LOG_FILE = os.path.join(_LOG_DIR, "mt5_pattern_scan_log.txt") -DEFAULT_OUTPUT_DIR = os.path.join(_LOG_DIR, "backtest_results") - -# ── Pattern Priority for Deduplication ───────────────────────────── -PATTERN_PRIORITY = { - 'Doji': 1, 'Spinning Top': 1, - 'Hammer': 2, 'Inverted Hammer': 2, - 'Shooting Star': 2, 'Hanging Man': 2, - 'Marubozu (Bullish)': 3, 'Marubozu (Bearish)': 3, - 'Tweezer Tops': 4, 'Tweezer Bottoms': 4, - 'Near Bullish Engulfing': 4, 'Near Bearish Engulfing': 4, - 'Bullish Engulfing': 5, 'Bearish Engulfing': 5, - 'Bullish Harami': 6, 'Bearish Harami': 6, - 'Morning Star': 7, 'Evening Star': 7, - 'Three White Soldiers': 8, 'Three Black Crows': 8, - 'Rising Three Methods': 9, 'Falling Three Methods': 9, -} - - -# ============================================================ -# UTILITY FUNCTIONS -# ============================================================ - -def log_message(msg, cfg=None): - """Print and log a message. Strips ANSI colour codes for log file.""" - timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") - line = f"[{timestamp}] {msg}" - print(line) - clean_line = re.sub(r'\x1b\[[0-9;]*m', '', line) - try: - with open(LOG_FILE, "a", encoding='utf-8') as f: - f.write(clean_line + "\n") - except Exception: - pass - - -def classify_session(hour, cfg=None): - """Classify broker-time hour into a trading session.""" - if cfg is None: - cfg = CFG - offset = cfg.get('broker_utc_offset', 2) - utc_hour = (hour - offset) % 24 - if 0 <= utc_hour < 7: - return 'Asia' - elif 7 <= utc_hour < 9: - return 'London Open' - elif 9 <= utc_hour < 12: - return 'London Morning' - elif 12 <= utc_hour < 16: - return 'London/NY Overlap' - elif 16 <= utc_hour < 20: - return 'NY Afternoon' - elif 20 <= utc_hour < 24: - return 'Pacific' - else: - return 'Unknown' - - -def deduplicate_patterns(patterns, cfg=None): - """Keep only the highest-priority directional pattern per candle.""" - if cfg is None: - cfg = CFG - if not cfg.get('deduplicate_signals', True): - return patterns - directional = [p for p in patterns if p.get('direction') != 'Neutral'] - neutral = [p for p in patterns if p.get('direction') == 'Neutral'] - if directional: - directional.sort(key=lambda p: PATTERN_PRIORITY.get(p.get('name', ''), 0), reverse=True) - return [directional[0]] - else: - neutral.sort(key=lambda p: PATTERN_PRIORITY.get(p.get('name', ''), 0), reverse=True) - return [neutral[0]] if neutral else [] - - -def get_forward_candles(tf_label, cfg=None): - """Return the forward evaluation candle count for the given timeframe label.""" - if cfg is None: - cfg = CFG - overrides = cfg.get('forward_candles_by_tf', {}) - return overrides.get(tf_label, cfg.get('default_forward_candles', 15)) - - -def get_atr_tf(tf_label, cfg=None): - """Return the timeframe label used for ATR calculation for the given trading TF. - - If 'atr_tf_by_tf' maps a TF to a higher TF, that higher TF is returned. - If the mapping is None or the same as tf_label, returns tf_label (native ATR). - """ - if cfg is None: - cfg = CFG - atr_tf_map = cfg.get('atr_tf_by_tf', {}) - atr_tf = atr_tf_map.get(tf_label, tf_label) - if atr_tf is None: - return tf_label - return atr_tf - - -# ============================================================ -# PATTERN DETECTION — Structured-array version (scanner) -# ============================================================ - -def detect_trend(rates, cfg=None): - """Detect trend using SMA over configurable lookback.""" - if cfg is None: - cfg = CFG - lookback = cfg.get('trend_lookback', 20) - if isinstance(rates, pd.DataFrame): - if len(rates) < lookback + 1: - return 'ranging' - closes = rates['CLOSE'].values if 'CLOSE' in rates.columns else rates['close'].values - recent = closes[-(lookback+1):] - sma = np.mean(recent[:-1]) - current_close = recent[-1] - else: - if len(rates) < lookback + 1: - return 'ranging' - recent = rates[-(lookback+1):] - closes = np.array([r['close'] for r in recent]) - sma = np.mean(closes[:-1]) - current_close = closes[-1] - if current_close > sma: - return 'uptrend' - elif current_close < sma: - return 'downtrend' - else: - return 'ranging' - - -def compute_atr(rates, cfg=None): - """Compute ATR from structured-array rates.""" - if cfg is None: - cfg = CFG - period = cfg.get('atr_period', 14) - if len(rates) < period + 1: - ranges = [r['high'] - r['low'] for r in rates] - return np.mean(ranges) if ranges else 0.001 - tr_values = [] - for i in range(1, len(rates)): - hl = rates[i]['high'] - rates[i]['low'] - hc = abs(rates[i]['high'] - rates[i-1]['close']) - lc = abs(rates[i]['low'] - rates[i-1]['close']) - tr_values.append(max(hl, hc, lc)) - if len(tr_values) >= period: - return np.mean(tr_values[-period:]) - return np.mean(tr_values) - - -def get_candle_metrics(candle, cfg=None): - """Compute metrics for a single candle (structured array row).""" - body = abs(candle['close'] - candle['open']) - body_sign = 1 if candle['close'] >= candle['open'] else -1 - range_val = candle['high'] - candle['low'] - upper_wick = candle['high'] - max(candle['open'], candle['close']) - lower_wick = min(candle['open'], candle['close']) - candle['low'] - body_ratio = body / range_val if range_val > 0 else 0 - return { - 'body': body, 'body_sign': body_sign, 'range': range_val, - 'upper_wick': upper_wick, 'lower_wick': lower_wick, 'body_ratio': body_ratio - } - - -def detect_doji(m, cfg=None): - if cfg is None: cfg = CFG - return m['range'] > 0 and m['body_ratio'] <= cfg['doji_body_ratio'] - -def detect_spinning_top(m, cfg=None): - if cfg is None: cfg = CFG - if m['range'] == 0 or m['body'] == 0: return False - if m['body_ratio'] > cfg['spinning_top_body_ratio']: return False - return m['upper_wick'] >= m['body'] and m['lower_wick'] >= m['body'] - -def detect_marubozu(m, cfg=None): - if cfg is None: cfg = CFG - if m['body'] == 0: return False - if m['body_ratio'] < cfg['long_candle_ratio']: return False - return (m['upper_wick'] <= m['body'] * cfg['marubozu_wick_ratio'] and - m['lower_wick'] <= m['body'] * cfg['marubozu_wick_ratio']) - -def detect_hammer(m, trend, cfg=None): - if cfg is None: cfg = CFG - if trend not in ('downtrend', 'ranging'): return False - if m['body'] == 0: return False - return (m['lower_wick'] >= m['body'] * cfg['hammer_lower_wick_ratio'] and - m['upper_wick'] <= m['body'] * cfg['hammer_upper_wick_ratio']) - -def detect_inverted_hammer(m, trend, cfg=None): - if cfg is None: cfg = CFG - if trend not in ('downtrend', 'ranging'): return False - if m['body'] == 0: return False - return (m['upper_wick'] >= m['body'] * cfg['hammer_lower_wick_ratio'] and - m['lower_wick'] <= m['body'] * cfg['hammer_upper_wick_ratio']) - -def detect_shooting_star(m, trend, cfg=None): - if cfg is None: cfg = CFG - if trend not in ('uptrend', 'ranging'): return False - if m['body'] == 0: return False - return (m['upper_wick'] >= m['body'] * cfg['hammer_lower_wick_ratio'] and - m['lower_wick'] <= m['body'] * cfg['hammer_upper_wick_ratio']) - -def detect_hanging_man(m, trend, cfg=None): - if cfg is None: cfg = CFG - if trend not in ('uptrend', 'ranging'): return False - if m['body'] == 0: return False - return (m['lower_wick'] >= m['body'] * cfg['hammer_lower_wick_ratio'] and - m['upper_wick'] <= m['body'] * cfg['hammer_upper_wick_ratio']) - -def detect_near_engulfing_full(curr, prev, cm, pm, cfg=None): - if cfg is None: cfg = CFG - if cm['body'] == 0 or pm['body'] == 0: return None - tol = cfg.get('engulf_tolerance_pips', 2.0) * 0.0001 - if cm['body_sign'] == 1 and pm['body_sign'] == -1: - if not (curr['open'] <= prev['close'] and curr['close'] >= prev['open']): - if curr['open'] <= prev['close'] + tol and curr['close'] >= prev['open'] - tol: - return 'Near Bullish Engulfing' - if cm['body_sign'] == -1 and pm['body_sign'] == 1: - if not (curr['open'] >= prev['close'] and curr['close'] <= prev['open']): - if curr['open'] >= prev['close'] - tol and curr['close'] <= prev['open'] + tol: - return 'Near Bearish Engulfing' - return None - -def detect_engulfing_full(curr, prev, cm, pm, cfg=None): - if cm['body'] == 0 or pm['body'] == 0: return None - if cm['body_sign'] == 1 and pm['body_sign'] == -1: - if curr['open'] <= prev['close'] and curr['close'] >= prev['open']: - return 'Bullish Engulfing' - if cm['body_sign'] == -1 and pm['body_sign'] == 1: - if curr['open'] >= prev['close'] and curr['close'] <= prev['open']: - return 'Bearish Engulfing' - return None - -def detect_harami_full(curr, prev, cm, pm, cfg=None): - if cfg is None: cfg = CFG - if cm['body'] == 0 or pm['body'] == 0: return None - if pm['body_ratio'] < cfg['long_candle_ratio'] * 0.8: return None - ch = max(curr['open'], curr['close']); cl = min(curr['open'], curr['close']) - ph = max(prev['open'], prev['close']); pl = min(prev['open'], prev['close']) - if ch <= ph and cl >= pl: - if pm['body_sign'] == -1 and cm['body_sign'] == 1: return 'Bullish Harami' - if pm['body_sign'] == 1 and cm['body_sign'] == -1: return 'Bearish Harami' - return None - -def detect_morning_star(rates, idx, ml, cfg=None): - if cfg is None: cfg = CFG - if idx < 2: return False - f, s, t = ml[idx-2], ml[idx-1], ml[idx] - if f['body_sign'] != -1 or f['body_ratio'] < cfg['long_candle_ratio']: return False - if s['body_ratio'] > cfg['small_candle_ratio'] + 0.1: return False - if t['body_sign'] != 1 or t['body_ratio'] < cfg['long_candle_ratio'] * 0.7: return False - return rates[idx]['close'] > (rates[idx-2]['open'] + rates[idx-2]['close']) / 2 - -def detect_evening_star(rates, idx, ml, cfg=None): - if cfg is None: cfg = CFG - if idx < 2: return False - f, s, t = ml[idx-2], ml[idx-1], ml[idx] - if f['body_sign'] != 1 or f['body_ratio'] < cfg['long_candle_ratio']: return False - if s['body_ratio'] > cfg['small_candle_ratio'] + 0.1: return False - if t['body_sign'] != -1 or t['body_ratio'] < cfg['long_candle_ratio'] * 0.7: return False - return rates[idx]['close'] < (rates[idx-2]['open'] + rates[idx-2]['close']) / 2 - -def detect_three_white_soldiers(rates, idx, ml, cfg=None): - if idx < 2: return False - m1, m2, m3 = ml[idx-2], ml[idx-1], ml[idx] - if m1['body_sign'] != 1 or m2['body_sign'] != 1 or m3['body_sign'] != 1: return False - if m1['body_ratio'] < 0.5 or m2['body_ratio'] < 0.5 or m3['body_ratio'] < 0.5: return False - return rates[idx]['close'] > rates[idx-1]['close'] > rates[idx-2]['close'] - -def detect_three_black_crows(rates, idx, ml, cfg=None): - if idx < 2: return False - m1, m2, m3 = ml[idx-2], ml[idx-1], ml[idx] - if m1['body_sign'] != -1 or m2['body_sign'] != -1 or m3['body_sign'] != -1: return False - if m1['body_ratio'] < 0.5 or m2['body_ratio'] < 0.5 or m3['body_ratio'] < 0.5: return False - return rates[idx]['close'] < rates[idx-1]['close'] < rates[idx-2]['close'] - -def detect_tweezer(rates, idx, cfg=None): - if cfg is None: cfg = CFG - if idx < 1: return None - prev, curr = rates[idx-1], rates[idx] - tol = cfg['tweezer_tolerance_pips'] * 0.0001 - trend = detect_trend(rates[:idx+1], cfg) - if abs(prev['high'] - curr['high']) <= tol and trend in ('uptrend', 'ranging'): - return 'Tweezer Tops' - if abs(prev['low'] - curr['low']) <= tol and trend in ('downtrend', 'ranging'): - return 'Tweezer Bottoms' - return None - -def detect_rising_three_methods(rates, idx, ml, cfg=None): - if cfg is None: cfg = CFG - if idx < 4: return False - fm = ml[idx-4] - if fm['body_sign'] != 1 or fm['body_ratio'] < cfg['long_candle_ratio']: return False - first = rates[idx-4] - for i in range(1, 4): - c = rates[idx-4+i]; cm = ml[idx-4+i] - if cm['body_ratio'] > cfg['small_candle_ratio'] + 0.15: return False - if c['high'] > first['high'] or c['low'] < first['low']: return False - fm5 = ml[idx] - if fm5['body_sign'] != 1 or fm5['body_ratio'] < cfg['long_candle_ratio'] * 0.7: return False - return rates[idx]['close'] > first['close'] - -def detect_falling_three_methods(rates, idx, ml, cfg=None): - if cfg is None: cfg = CFG - if idx < 4: return False - fm = ml[idx-4] - if fm['body_sign'] != -1 or fm['body_ratio'] < cfg['long_candle_ratio']: return False - first = rates[idx-4] - for i in range(1, 4): - c = rates[idx-4+i]; cm = ml[idx-4+i] - if cm['body_ratio'] > cfg['small_candle_ratio'] + 0.15: return False - if c['high'] > first['high'] or c['low'] < first['low']: return False - fm5 = ml[idx] - if fm5['body_sign'] != -1 or fm5['body_ratio'] < cfg['long_candle_ratio'] * 0.7: return False - return rates[idx]['close'] < first['close'] - - -# ============================================================ -# VOLUME CONFIRMATION — Structured-array version -# ============================================================ - -def check_volume_confirmed(rates, idx, cfg=None): - """Check if the signal candle's tick_volume confirms the pattern.""" - if cfg is None: cfg = CFG - if not cfg.get('volume_filter', False): - return True - vol_period = cfg.get('volume_ma_period', 20) - vol_thresh = cfg.get('volume_threshold', 1.0) - start = max(0, idx - vol_period) - vols = [r['tick_volume'] for r in rates[start:idx+1]] - if len(vols) < 2: - return True - avg_vol = np.mean(vols[:-1]) - if avg_vol == 0: - return True - return vols[-1] >= vol_thresh * avg_vol - - -# ============================================================ -# ENTRY PRICE LOGIC -# ============================================================ - -def compute_entry_details(candle, pattern_name, direction, trend, idx, rates, cfg=None): - """Compute trade entry price based on pattern and candle structure.""" - if cfg is None: cfg = CFG - body_top = max(candle['open'], candle['close']) - body_bottom = min(candle['open'], candle['close']) - body_mid = (body_top + body_bottom) / 2.0 - is_bullish_candle = candle['close'] >= candle['open'] - - entry_type = entry_price = entry_reason = None - - if direction == 'Bullish': - if pattern_name in ('Hammer', 'Inverted Hammer', 'Morning Star', 'Three White Soldiers', - 'Tweezer Bottoms', 'Rising Three Methods') \ - or 'Bullish Engulfing' in pattern_name \ - or 'Bullish Harami' in pattern_name: - entry_type, entry_price = 'Buy Stop', round(body_top, 5) - entry_reason = f'Break above body top ({body_top:.5f}) confirms bullish signal' - elif 'Marubozu' in pattern_name and 'Bullish' in pattern_name: - entry_type, entry_price = 'Market Buy', round(candle['close'], 5) - entry_reason = f'Bullish Marubozu close ({candle["close"]:.5f}) — strong momentum' - else: - entry_type, entry_price = 'Buy Stop', round(body_top, 5) - entry_reason = f'Break above body top ({body_top:.5f})' - - elif direction == 'Bearish': - if pattern_name in ('Evening Star', 'Shooting Star', 'Hanging Man', 'Three Black Crows', - 'Falling Three Methods', 'Tweezer Tops') \ - or 'Bearish Engulfing' in pattern_name \ - or 'Bearish Harami' in pattern_name: - entry_type, entry_price = 'Sell Stop', round(body_bottom, 5) - entry_reason = f'Break below body bottom ({body_bottom:.5f}) confirms bearish signal' - elif 'Marubozu' in pattern_name and 'Bearish' in pattern_name: - entry_type, entry_price = 'Market Sell', round(candle['close'], 5) - entry_reason = f'Bearish Marubozu close ({candle["close"]:.5f}) — strong momentum' - else: - entry_type, entry_price = 'Sell Stop', round(body_bottom, 5) - entry_reason = f'Break below body bottom ({body_bottom:.5f})' - else: - entry_type = 'Breakout' - entry_price = None - entry_reason = 'Wait for breakout: Buy Stop above body top OR Sell Stop below body bottom' - - return { - 'body_top': round(body_top, 5), 'body_bottom': round(body_bottom, 5), - 'body_mid': round(body_mid, 5), 'is_bullish_candle': is_bullish_candle, - 'entry_type': entry_type, 'entry_price': entry_price, - 'aggressive_entry': round(candle['close'], 5), 'entry_reason': entry_reason, - } - - -def verify_entry_fill(entry_type, entry_price, next_candle, cfg=None): - """Verify whether a stop/market entry would be filled on the next candle.""" - if cfg is None: cfg = CFG - if not cfg.get('verify_entry', True): - return True, entry_price - if entry_type == 'Market Buy': - return True, next_candle['open'] - elif entry_type == 'Market Sell': - return True, next_candle['open'] - elif entry_type == 'Buy Stop': - if next_candle['high'] >= entry_price: - return True, max(next_candle['open'], entry_price) - return False, None - elif entry_type == 'Sell Stop': - if next_candle['low'] <= entry_price: - return True, min(next_candle['open'], entry_price) - return False, None - return True, entry_price - - -# ============================================================ -# BACKTEST STATS LOADER (for live scanner integration) -# ============================================================ - -def load_latest_backtest_stats(output_dir=None, symbol=None, cfg=None): - """Load pattern & session performance from latest backtest CSVs or JSON cache. - - Tries multiple sources in order: - 1. v5 JSON cache (latest_stats.json) — flat structure with patterns/sessions/cross - 2. v6 JSON cache (latest_stats_multitf.json) — multi-TF nested structure - 3. Fall back to parsing CSV files (supports both v5 and v6 naming) - - Returns dict with keys: patterns, sessions, overall, cross, generated_at - """ - if cfg is None: cfg = CFG - if output_dir is None: output_dir = DEFAULT_OUTPUT_DIR - if symbol is None: symbol = cfg.get('symbol', 'EURUSD') - cache_hours = cfg.get('stats_cache_hours', 4) - - # ── Try v5 JSON cache first (flat structure with pattern/session/cross stats) ── - v5_cache_path = os.path.join(output_dir, 'latest_stats.json') - if os.path.exists(v5_cache_path): - try: - with open(v5_cache_path, 'r', encoding='utf-8') as f: - stats = json.load(f) - # v5 cache has 'patterns', 'sessions', 'overall', 'cross' at top level - if stats.get('patterns') or stats.get('overall'): - generated = stats.get('generated_at', '') - if generated: - gen_dt = datetime.strptime(generated, '%Y-%m-%d %H:%M:%S') - age_hours = (datetime.now() - gen_dt).total_seconds() / 3600 - if age_hours < cache_hours * 24: # v5 cache: allow 24h staleness - return stats - except Exception: - pass - - # ── Try v6 multi-TF JSON cache ── - v6_cache_path = os.path.join(output_dir, 'latest_stats_multitf.json') - if os.path.exists(v6_cache_path): - try: - with open(v6_cache_path, 'r', encoding='utf-8') as f: - v6_stats = json.load(f) - generated = v6_stats.get('generated_at', '') - if generated: - gen_dt = datetime.strptime(generated, '%Y-%m-%d %H:%M:%S') - age_hours = (datetime.now() - gen_dt).total_seconds() / 3600 - if age_hours < cache_hours * 24: - # Convert v6 multi-TF structure to v5 flat structure - # Aggregate across all timeframes into unified stats - stats = _merge_multitf_stats(v6_stats, output_dir, symbol) - if stats.get('patterns') or stats.get('overall'): - return stats - except Exception: - pass - - # ── Fall back to parsing CSVs (both v5 and v6 naming) ── - stats = {'patterns': {}, 'sessions': {}, 'overall': {}, 'cross': {}} - - # Find latest CSVs — try v5 naming first, then v6 naming - pattern_csvs = sorted( - glob.glob(os.path.join(output_dir, f"{symbol}_*_pattern_summary.csv")) + - glob.glob(os.path.join(output_dir, f"{symbol}_*_*_to_*_pattern_summary.csv")), - reverse=True - ) - session_csvs = sorted( - glob.glob(os.path.join(output_dir, f"{symbol}_*_session_summary.csv")) + - glob.glob(os.path.join(output_dir, f"{symbol}_*_*_to_*_session_summary.csv")), - reverse=True - ) - det_csvs = sorted( - glob.glob(os.path.join(output_dir, f"{symbol}_*_detections.csv")) + - glob.glob(os.path.join(output_dir, f"{symbol}_*_*_to_*_detections.csv")), - reverse=True - ) - - if pattern_csvs: - try: - df_p = pd.read_csv(pattern_csvs[0]) - for _, row in df_p.iterrows(): - pat = row['Pattern'] - stats['patterns'][pat] = { - 'win_rate': round(float(row.get('Win_Rate_%', 0)), 1), - 'total': int(row.get('Total', 0)), - 'avg_max_r': round(float(row.get('Avg_Max_R', 0)), 2), - 'sl_hit_pct': round(float(row.get('SL_Hit_%', 0)), 1), - 'tp_hit_pct': round(float(row.get('TP_Hit_%', 0)), 1), - } - except Exception: - pass - - if session_csvs: - try: - df_s = pd.read_csv(session_csvs[0]) - for _, row in df_s.iterrows(): - sess = row['Session'] - stats['sessions'][sess] = { - 'win_rate': round(float(row.get('Win_Rate_%', 0)), 1), - 'signals': int(row.get('Signals', 0)), - 'avg_max_r': round(float(row.get('Avg_Max_R', 0)), 2), - 'sl_hit_pct': round(float(row.get('SL_Hit_%', 0)), 1), - 'tp_hit_pct': round(float(row.get('TP_Hit_%', 0)), 1), - } - except Exception: - pass - - # Overall + cross stats from detections CSV - if det_csvs: - try: - df_d = pd.read_csv(det_csvs[0]) - directional = df_d[df_d['Direction'] != 'Neutral'] - if len(directional) > 0: - s = int((directional['Prediction_Success'] == True).sum()) - f_ = int((directional['Prediction_Success'] == False).sum()) - stats['overall'] = { - 'win_rate': round(s / (s + f_) * 100, 1) if (s + f_) > 0 else 0, - 'total_signals': len(directional), - 'avg_max_r': round(float(directional['Max_R'].dropna().mean()), 2), - 'sl_hit_pct': round(float((directional['SL_Hit'] == True).sum() / len(directional) * 100), 1), - 'tp_hit_pct': round(float((directional['TP_Hit'] == True).sum() / len(directional) * 100), 1), - } - cross = {} - for pat_name in directional['Pattern'].unique(): - pat_dir = directional[directional['Pattern'] == pat_name] - for sess in pat_dir['Session'].unique(): - ps = pat_dir[pat_dir['Session'] == sess] - if len(ps) >= 3: - s_ps = int((ps['Prediction_Success'] == True).sum()) - f_ps = int((ps['Prediction_Success'] == False).sum()) - cross[f"{pat_name}|{sess}"] = { - 'win_rate': round(s_ps / (s_ps + f_ps) * 100, 1) if (s_ps + f_ps) > 0 else 0, - 'signals': len(ps), - 'avg_max_r': round(float(ps['Max_R'].dropna().mean()), 2), - } - stats['cross'] = cross - except Exception: - pass - - stats['generated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') - return stats - - -def _merge_multitf_stats(v6_stats, output_dir, symbol): - """Merge v6 multi-TF stats into v5 flat structure for display compatibility.""" - stats = {'patterns': {}, 'sessions': {}, 'overall': {}, 'cross': {}, 'generated_at': v6_stats.get('generated_at', '')} - # The v6 JSON only has per-TF overall stats, not pattern/session-level. - # We need to parse the CSVs for detailed stats. - pattern_csvs = sorted( - glob.glob(os.path.join(output_dir, f"{symbol}_*_*_to_*_pattern_summary.csv")), - reverse=True - ) - session_csvs = sorted( - glob.glob(os.path.join(output_dir, f"{symbol}_*_*_to_*_session_summary.csv")), - reverse=True - ) - det_csvs = sorted( - glob.glob(os.path.join(output_dir, f"{symbol}_*_*_to_*_detections.csv")), - reverse=True - ) - if pattern_csvs: - try: - df_p = pd.read_csv(pattern_csvs[0]) - for _, row in df_p.iterrows(): - pat = row['Pattern'] - stats['patterns'][pat] = { - 'win_rate': round(float(row.get('Win_Rate_%', 0)), 1), - 'total': int(row.get('Total', 0)), - 'avg_max_r': round(float(row.get('Avg_Max_R', 0)), 2), - 'sl_hit_pct': round(float(row.get('SL_Hit_%', 0)), 1), - 'tp_hit_pct': round(float(row.get('TP_Hit_%', 0)), 1), - } - except Exception: - pass - if session_csvs: - try: - df_s = pd.read_csv(session_csvs[0]) - for _, row in df_s.iterrows(): - sess = row['Session'] - stats['sessions'][sess] = { - 'win_rate': round(float(row.get('Win_Rate_%', 0)), 1), - 'signals': int(row.get('Signals', 0)), - 'avg_max_r': round(float(row.get('Avg_Max_R', 0)), 2), - 'sl_hit_pct': round(float(row.get('SL_Hit_%', 0)), 1), - 'tp_hit_pct': round(float(row.get('TP_Hit_%', 0)), 1), - } - except Exception: - pass - if det_csvs: - try: - df_d = pd.read_csv(det_csvs[0]) - directional = df_d[df_d['Direction'] != 'Neutral'] - if len(directional) > 0: - s = int((directional['Prediction_Success'] == True).sum()) - f_ = int((directional['Prediction_Success'] == False).sum()) - stats['overall'] = { - 'win_rate': round(s / (s + f_) * 100, 1) if (s + f_) > 0 else 0, - 'total_signals': len(directional), - 'avg_max_r': round(float(directional['Max_R'].dropna().mean()), 2), - 'sl_hit_pct': round(float((directional['SL_Hit'] == True).sum() / len(directional) * 100), 1), - 'tp_hit_pct': round(float((directional['TP_Hit'] == True).sum() / len(directional) * 100), 1), - } - cross = {} - for pat_name in directional['Pattern'].unique(): - pat_dir = directional[directional['Pattern'] == pat_name] - for sess in pat_dir['Session'].unique(): - ps = pat_dir[pat_dir['Session'] == sess] - if len(ps) >= 3: - s_ps = int((ps['Prediction_Success'] == True).sum()) - f_ps = int((ps['Prediction_Success'] == False).sum()) - cross[f"{pat_name}|{sess}"] = { - 'win_rate': round(s_ps / (s_ps + f_ps) * 100, 1) if (s_ps + f_ps) > 0 else 0, - 'signals': len(ps), - 'avg_max_r': round(float(ps['Max_R'].dropna().mean()), 2), - } - stats['cross'] = cross - except Exception: - pass - return stats - - -def compute_pattern_tier(pattern_name, stats, cfg=None): - """Classify a pattern into a tier (A/B/C/D) based on backtest statistics. - - Tier A (Elite): WR >= 58% AND sample >= 30 AND avg_max_r >= 0.35R - Tier B (Tradeable): WR >= 50% AND sample >= 10 AND avg_max_r >= 0.25R - Tier C (Marginal): WR >= 40% AND sample >= 5 - Tier D (Avoid): WR < 40% OR insufficient data - """ - if cfg is None: cfg = CFG - min_sig = cfg.get('min_signals_for_stats', 5) - pat_stats = stats.get('patterns', {}).get(pattern_name, {}) - n = pat_stats.get('total', 0) - wr = pat_stats.get('win_rate', 0) - amr = pat_stats.get('avg_max_r', 0) - if n < min_sig: - return ('D', 'AVOID', 'red') - if wr >= 58 and n >= 30 and amr >= 0.35: - return ('A', 'ELITE', 'green') - elif wr >= 50 and n >= 10 and amr >= 0.25: - return ('B', 'TRADEABLE', 'yellow') - elif wr >= 40: - return ('C', 'MARGINAL', 'red') - else: - return ('D', 'AVOID', 'red') - - -def compute_session_quality(session, stats, cfg=None): - """Classify session quality based on directional win rate and R-multiples. - - Returns: (quality_label, quality_color, sess_wr, sess_n, sess_amr) - """ - if cfg is None: cfg = CFG - min_sig = cfg.get('min_signals_for_stats', 5) - sess_stats = stats.get('sessions', {}).get(session, {}) - n = sess_stats.get('signals', 0) - wr = sess_stats.get('win_rate', 0) - amr = sess_stats.get('avg_max_r', 0) - if n < min_sig: - return ('UNKNOWN', 'dim', 0, n, 0) - if wr >= 55 and amr >= 0.35: - return ('PRIME', 'green', wr, n, amr) - elif wr >= 50: - return ('FAVORABLE', 'green', wr, n, amr) - elif wr >= 45: - return ('NEUTRAL', 'yellow', wr, n, amr) - else: - return ('UNFAVORABLE', 'red', wr, n, amr) - - -def compute_cross_quality(pattern_name, session, stats, cfg=None): - """Get the most specific win rate and quality for a pattern+session combo. - - Returns: (cross_wr, cross_n, cross_amr, cross_label, cross_color) - """ - if cfg is None: cfg = CFG - min_sig = cfg.get('min_signals_for_stats', 5) - cross_key = f"{pattern_name}|{session}" - cross_stats = stats.get('cross', {}).get(cross_key, {}) - pat_stats = stats.get('patterns', {}).get(pattern_name, {}) - if cross_stats and cross_stats.get('signals', 0) >= min_sig: - wr = cross_stats.get('win_rate', 0) - n = cross_stats.get('signals', 0) - amr = cross_stats.get('avg_max_r', 0) - elif pat_stats and pat_stats.get('total', 0) >= min_sig: - wr = pat_stats.get('win_rate', 0) - n = pat_stats.get('total', 0) - amr = pat_stats.get('avg_max_r', 0) - elif stats.get('overall', {}).get('total_signals', 0) >= min_sig: - wr = stats['overall'].get('win_rate', 0) - n = stats['overall'].get('total_signals', 0) - amr = stats['overall'].get('avg_max_r', 0) - else: - return (None, 0, 0, 'N/A', 'dim') - if wr >= 60: - return (wr, n, amr, 'HIGH EDGE', 'green') - elif wr >= 50: - return (wr, n, amr, 'EDGE', 'yellow') - elif wr >= 40: - return (wr, n, amr, 'WEAK', 'red') - else: - return (wr, n, amr, 'NO EDGE', 'red') - - -def compute_signal_score(pattern_name, session, direction, stats, cfg=None): - """Compute a 0-100 signal quality score based on historical backtest stats. - - Score formula: - base_score = pattern_win_rate (0-100) - session_bonus = +10 if session WR > 55%, -10 if < 45% - confidence_factor = min(1.0, signals / 30) - r_factor = avg_max_r / 1.0 - tier_bonus = +5 for Tier A, +3 for Tier B - """ - if cfg is None: cfg = CFG - min_signals = cfg.get('min_signals_for_stats', 5) - pat_stats = stats.get('patterns', {}).get(pattern_name, {}) - sess_stats = stats.get('sessions', {}).get(session, {}) - cross_key = f"{pattern_name}|{session}" - cross_stats = stats.get('cross', {}).get(cross_key, {}) - if cross_stats and cross_stats.get('signals', 0) >= min_signals: - wr = cross_stats.get('win_rate', 50) - n = cross_stats.get('signals', 0) - amr = cross_stats.get('avg_max_r', 0) - elif pat_stats and pat_stats.get('total', 0) >= min_signals: - wr = pat_stats.get('win_rate', 50) - n = pat_stats.get('total', 0) - amr = pat_stats.get('avg_max_r', 0) - elif stats.get('overall', {}).get('total_signals', 0) >= min_signals: - overall = stats['overall'] - wr = overall.get('win_rate', 50) - n = overall.get('total_signals', 0) - amr = overall.get('avg_max_r', 0) - else: - return None - base_score = wr - confidence = min(1.0, n / 30.0) - session_bonus = 0 - if sess_stats and sess_stats.get('signals', 0) >= min_signals: - sess_wr = sess_stats.get('win_rate', 50) - if sess_wr > 55: - session_bonus = 10 - elif sess_wr < 45: - session_bonus = -10 - r_factor = min(amr, 2.0) * 10 - tier_letter, _, _ = compute_pattern_tier(pattern_name, stats, cfg) - tier_bonus = 5 if tier_letter == 'A' else (3 if tier_letter == 'B' else 0) - score = base_score * confidence + session_bonus + r_factor + tier_bonus - return round(max(0, min(100, score)), 1) - - -def print_top_setups(stats, cfg=None): - """Print best historical setups at scanner start — top patterns, sessions, and cross-stats.""" - if cfg is None: cfg = CFG - min_sig = cfg.get('min_signals_for_stats', 5) - min_wr = cfg.get('min_historical_win_rate', 50.0) - lines = [] - lines.append("") - lines.append(C('cyan', "=" * 70)) - lines.append(C('bold', " TOP HISTORICAL SETUPS (from latest backtest)")) - lines.append(C('cyan', "=" * 70)) - overall = stats.get('overall', {}) - if overall: - owr = overall.get('win_rate', 0) - owr_color = 'green' if owr >= min_wr else 'red' - lines.append(f" Overall: {C(owr_color, f'WR {owr:.1f}%')} | {overall.get('total_signals',0)} signals | Avg Max R: {overall.get('avg_max_r',0):.2f}R") - pat_list = [] - for pat, data in stats.get('patterns', {}).items(): - n = data.get('total', 0) - wr = data.get('win_rate', 0) - amr = data.get('avg_max_r', 0) - if n >= min_sig: - confidence = min(1.0, n / 30.0) - weighted = wr * confidence + min(amr, 2.0) * 10 - tier_letter, tier_label, tier_clr = compute_pattern_tier(pat, stats, cfg) - pat_list.append((pat, wr, n, amr, weighted, tier_letter, tier_label, tier_clr)) - pat_list.sort(key=lambda x: x[4], reverse=True) - if pat_list: - lines.append("") - lines.append(f" {'Pattern':<30s} | {'Tier':>5s} | {'WR':>6s} | {'Sig':>5s} | {'MaxR':>5s} | {'Edge':>5s}") - lines.append(f" {'-'*30} | {'-'*5} | {'-'*6} | {'-'*5} | {'-'*5} | {'-'*5}") - for pat, wr, n, amr, weighted, tl, tlab, tc in pat_list[:7]: - edge_tag = "HIGH" if wr >= min_wr else "LOW" - edge_color = 'green' if wr >= min_wr else 'red' - lines.append(f" {pat:<30s} | {C(tc, f'{tl}:{tlab}'):>14s} | {C(edge_color, f'{wr:>5.1f}%')} | {n:>5d} | {amr:>4.2f}R | {C(edge_color, f'{edge_tag:>5s}')}") - all_sess = [(s, d) for s, d in stats.get('sessions', {}).items() - if d.get('signals', 0) >= min_sig] - if all_sess: - all_sess.sort(key=lambda x: x[1].get('win_rate', 0), reverse=True) - lines.append("") - lines.append(C('bold', " Session Quality:")) - for sess, data in all_sess: - swr = data.get('win_rate', 0) - sq, sqc, _, sn, samr = compute_session_quality(sess, stats, cfg) - lines.append(f" {sess:<22s} | {C(sqc, f'{swr:.1f}% WR [{sq}]')} ({sn} sig) | AvgMaxR: {samr:.2f}R") - cross_list = [] - for key, data in stats.get('cross', {}).items(): - n = data.get('signals', 0) - wr = data.get('win_rate', 0) - amr = data.get('avg_max_r', 0) - if n >= min_sig: - confidence = min(1.0, n / 20.0) - weighted = wr * confidence + min(amr, 2.0) * 10 - cross_list.append((key, wr, n, amr, weighted)) - cross_list.sort(key=lambda x: x[4], reverse=True) - if cross_list: - lines.append("") - lines.append(f" {'Pattern x Session':<40s} | {'WR':>6s} | {'Sig':>4s} | {'MaxR':>5s}") - lines.append(f" {'-'*40} | {'-'*6} | {'-'*4} | {'-'*5}") - for key, wr, n, amr, weighted in cross_list[:5]: - cwr_color = 'green' if wr >= min_wr else 'yellow' - lines.append(f" {key:<40s} | {C(cwr_color, f'{wr:>5.1f}%')} | {n:>4d} | {amr:>4.2f}R") - weak = [(p, d) for p, d in stats.get('patterns', {}).items() - if d.get('total', 0) >= min_sig and d.get('win_rate', 0) < 45] - if weak: - weak.sort(key=lambda x: x[1].get('win_rate', 0)) - lines.append("") - lines.append(C('red', " Tier D — AVOID (WR < 45%):")) - for pat, data in weak: - wwr = data.get('win_rate', 0) - _, _, wtc = compute_pattern_tier(pat, stats, cfg) - lines.append(f" {pat:<30s} | {C(wtc, f'WR: {wwr:.1f}%')} ({data.get('total',0)} sig)") - rec_setups = [] - for key, data in stats.get('cross', {}).items(): - n = data.get('signals', 0) - wr = data.get('win_rate', 0) - amr = data.get('avg_max_r', 0) - if n >= min_sig: - confidence = min(1.0, n / 20.0) - score = wr * confidence + min(amr, 2.0) * 10 - if score > 60: - rec_setups.append((key, wr, n, amr, score)) - for pat, data in stats.get('patterns', {}).items(): - n = data.get('total', 0) - wr = data.get('win_rate', 0) - amr = data.get('avg_max_r', 0) - if n >= min_sig: - confidence = min(1.0, n / 30.0) - score = wr * confidence + min(amr, 2.0) * 10 - if score > 60: - rec_key = f"{pat} (any session)" - if not any(pat in k for k, _, _, _, _ in rec_setups): - rec_setups.append((rec_key, wr, n, amr, score)) - if rec_setups: - rec_setups.sort(key=lambda x: x[4], reverse=True) - lines.append("") - lines.append(C('green', C('bold', " RECOMMENDED LIVE SETUPS (score > 60):"))) - for key, wr, n, amr, score in rec_setups[:8]: - lines.append(f" {key:<40s} | {C('green', f'WR: {wr:.1f}%')} | {n} sig | {C('bold', f'Score: {score:.1f}')}") - lines.append(C('cyan', "=" * 70)) - lines.append("") - for line in lines: - print(line) - - -def apply_signal_score_filter(patterns, stats, cfg=None): - """Filter patterns by min_signal_score. Attaches signal_score to each pattern dict.""" - if cfg is None: cfg = CFG - min_score = cfg.get('min_signal_score', 0) - if min_score <= 0 or not stats: - for p in patterns: - score = compute_signal_score(p['name'], p['session'], p['direction'], stats, cfg) - p['signal_score'] = score - return patterns - filtered = [] - for p in patterns: - score = compute_signal_score(p['name'], p['session'], p['direction'], stats, cfg) - p['signal_score'] = score - if score is None or score >= min_score: - filtered.append(p) - return filtered - - -# ============================================================ -# SCANNER: scan last closed candle on a given timeframe -# ============================================================ - -def scan_patterns(rates, cfg=None, d1_rates=None, tf_label='H4', htf_atr_rates=None): - """Scan the last closed candle for all patterns. Returns list of dicts. - - Args: - htf_atr_rates: Optional higher-timeframe rates (structured array) for ATR - calculation. If provided and atr_tf_by_tf maps this TF to a - higher TF, ATR is computed from these rates instead of native. - """ - if cfg is None: - cfg = CFG - if len(rates) < 6: - return [] - ml = [get_candle_metrics(r, cfg) for r in rates] - idx = len(rates) - 1 - curr = rates[idx]; cm = ml[idx] - patterns = [] - trend = detect_trend(rates[:idx+1], cfg) - - # Use higher-timeframe ATR if configured and rates provided - atr_src = get_atr_tf(tf_label, cfg) - if htf_atr_rates is not None and atr_src != tf_label: - atr = compute_atr(htf_atr_rates, cfg) - else: - atr = compute_atr(rates, cfg) - - _ct = curr['time'] - if isinstance(_ct, (int, float, np.integer, np.floating)): - hour = datetime.fromtimestamp(int(_ct)).hour - elif hasattr(_ct, 'hour'): - hour = _ct.hour - else: - hour = int(_ct) % 24 - session = classify_session(hour, cfg) - - vol_confirmed = check_volume_confirmed(rates, idx, cfg) - - # D1 trend filter - d1_trend = 'N/A' - if cfg.get('d1_trend_filter', False) and d1_rates is not None and len(d1_rates) >= cfg.get('d1_sma_period', 20): - d1_closes = [r['close'] for r in d1_rates[-cfg['d1_sma_period']:]] - d1_sma = np.mean(d1_closes) - d1_close = d1_rates[-1]['close'] - d1_trend = 'uptrend' if d1_close > d1_sma else ('downtrend' if d1_close < d1_sma else 'ranging') - - # Single-candle patterns - if detect_doji(cm, cfg): patterns.append({'name': 'Doji', 'category': 'Neutral', 'direction': 'Neutral'}) - if detect_spinning_top(cm, cfg): patterns.append({'name': 'Spinning Top', 'category': 'Neutral', 'direction': 'Neutral'}) - if detect_marubozu(cm, cfg): - d = 'Bullish' if cm['body_sign'] == 1 else 'Bearish' - patterns.append({'name': f'Marubozu ({d})', 'category': f'{d} Continuation', 'direction': d}) - if detect_hammer(cm, trend, cfg): patterns.append({'name': 'Hammer', 'category': 'Bullish Reversal', 'direction': 'Bullish'}) - if detect_inverted_hammer(cm, trend, cfg): patterns.append({'name': 'Inverted Hammer', 'category': 'Bullish Reversal', 'direction': 'Bullish'}) - if detect_shooting_star(cm, trend, cfg): patterns.append({'name': 'Shooting Star', 'category': 'Bearish Reversal', 'direction': 'Bearish'}) - if detect_hanging_man(cm, trend, cfg): patterns.append({'name': 'Hanging Man', 'category': 'Bearish Reversal', 'direction': 'Bearish'}) - - # Two-candle patterns - if idx >= 1: - prev = rates[idx-1]; pm = ml[idx-1] - e = detect_engulfing_full(curr, prev, cm, pm, cfg) - if e: - d = 'Bullish' if 'Bullish' in e else 'Bearish' - patterns.append({'name': e, 'category': f'{d} Reversal', 'direction': d}) - ne = detect_near_engulfing_full(curr, prev, cm, pm, cfg) - if ne: - d = 'Bullish' if 'Bullish' in ne else 'Bearish' - patterns.append({'name': ne, 'category': f'{d} Reversal', 'direction': d}) - h = detect_harami_full(curr, prev, cm, pm, cfg) - if h: - d = 'Bullish' if 'Bullish' in h else 'Bearish' - patterns.append({'name': h, 'category': f'{d} Reversal', 'direction': d}) - tw = detect_tweezer(rates, idx, cfg) - if tw: - d = 'Bearish' if 'Tops' in tw else 'Bullish' - patterns.append({'name': tw, 'category': f'{d} Reversal', 'direction': d}) - - # Three-candle patterns - if detect_morning_star(rates, idx, ml, cfg): patterns.append({'name': 'Morning Star', 'category': 'Bullish Reversal', 'direction': 'Bullish'}) - if detect_evening_star(rates, idx, ml, cfg): patterns.append({'name': 'Evening Star', 'category': 'Bearish Reversal', 'direction': 'Bearish'}) - if detect_three_white_soldiers(rates, idx, ml, cfg): patterns.append({'name': 'Three White Soldiers', 'category': 'Bullish Reversal', 'direction': 'Bullish'}) - if detect_three_black_crows(rates, idx, ml, cfg): patterns.append({'name': 'Three Black Crows', 'category': 'Bearish Reversal', 'direction': 'Bearish'}) - - # Five-candle patterns - if detect_rising_three_methods(rates, idx, ml, cfg): patterns.append({'name': 'Rising Three Methods', 'category': 'Bullish Continuation', 'direction': 'Bullish'}) - if detect_falling_three_methods(rates, idx, ml, cfg): patterns.append({'name': 'Falling Three Methods', 'category': 'Bearish Continuation', 'direction': 'Bearish'}) - - patterns = deduplicate_patterns(patterns, cfg) - - # D1 trend filter - if cfg.get('d1_trend_filter', False) and d1_trend != 'N/A': - filtered = [] - for pat in patterns: - if pat['direction'] == 'Bullish' and d1_trend == 'uptrend': filtered.append(pat) - elif pat['direction'] == 'Bearish' and d1_trend == 'downtrend': filtered.append(pat) - elif pat['direction'] == 'Neutral': filtered.append(pat) - patterns = filtered - - if cfg.get('volume_filter', False): - patterns = [p for p in patterns if p['direction'] == 'Neutral' or vol_confirmed] - - sl_mult = cfg['sl_multiplier'] - tp_mult = cfg['tp_multiplier'] - for pat in patterns: - d = pat['direction'] - if d == 'Bullish': - pat['sl'] = round(curr['low'] - sl_mult * atr, 5) - risk = curr['close'] - pat['sl'] - pat['tp'] = round(curr['close'] + risk * (tp_mult / sl_mult), 5) - elif d == 'Bearish': - pat['sl'] = round(curr['high'] + sl_mult * atr, 5) - risk = pat['sl'] - curr['close'] - pat['tp'] = round(curr['close'] - risk * (tp_mult / sl_mult), 5) - else: - pat['sl'] = round(curr['low'] - sl_mult * atr, 5) - risk_bull = curr['close'] - pat['sl'] - pat['tp_long'] = round(curr['close'] + risk_bull * (tp_mult / sl_mult), 5) - bearish_sl = round(curr['high'] + sl_mult * atr, 5) - risk_bear = bearish_sl - curr['close'] - pat['tp_short'] = round(curr['close'] - risk_bear * (tp_mult / sl_mult), 5) - - pat['atr'] = round(atr, 5) - pat['atr_tf'] = atr_src - pat['session'] = session - pat['trend'] = trend - pat['Volume_Confirmed'] = vol_confirmed - pat['D1_Trend'] = d1_trend - pat['timeframe'] = tf_label - - entry = compute_entry_details(curr, pat['name'], d, trend, idx, rates, cfg) - pat.update(entry) - - if d == 'Bullish' and entry['entry_price'] is not None: - pat['sl_dist_pips'] = round(abs(entry['entry_price'] - pat['sl']) * 10000, 1) - pat['tp_dist_pips'] = round(abs(pat['tp'] - entry['entry_price']) * 10000, 1) - pat['rr_ratio'] = round(pat['tp_dist_pips'] / pat['sl_dist_pips'], 2) if pat['sl_dist_pips'] > 0 else None - elif d == 'Bearish' and entry['entry_price'] is not None: - pat['sl_dist_pips'] = round(abs(pat['sl'] - entry['entry_price']) * 10000, 1) - pat['tp_dist_pips'] = round(abs(entry['entry_price'] - pat['tp']) * 10000, 1) - pat['rr_ratio'] = round(pat['tp_dist_pips'] / pat['sl_dist_pips'], 2) if pat['sl_dist_pips'] > 0 else None - else: - pat['sl_dist_pips'] = pat['tp_dist_pips'] = pat['rr_ratio'] = None - - return patterns - - -def format_pattern_output(candle, patterns, cfg=None, stats=None, tf_label='H4'): - """Format pattern detection results for display with colour-coded tier, - historical backtest edge, and signal quality score.""" - if cfg is None: cfg = CFG - if stats is None: stats = {} - ct = candle['time'] - if isinstance(ct, (int, float, np.integer, np.floating)): - ct = datetime.fromtimestamp(int(ct)) - time_str = ct.strftime("%Y-%m-%d %H:%M:%S") if hasattr(ct, 'strftime') else str(ct) - bt = max(candle['open'], candle['close']) - bb = min(candle['open'], candle['close']) - lines = [] - tf_color = {'M5': 'magenta', 'M15': 'blue', 'H1': 'cyan', 'H4': 'yellow', 'D1': 'white'}.get(tf_label, 'cyan') - lines.append(C(tf_color, "=" * 80)) - lines.append(C('bold', f" [{tf_label}] CANDLE CLOSE: {time_str}")) - lines.append(f" Symbol: {C('yellow', cfg['symbol'])} | Timeframe: {C(tf_color, tf_label)}") - close_str = f"{candle['close']:.5f}" - lines.append(f" O: {candle['open']:.5f} H: {candle['high']:.5f} L: {candle['low']:.5f} C: {C('bold', close_str)}") - lines.append(f" Body Top: {bt:.5f} | Body Bottom: {bb:.5f} | Size: {abs(candle['close']-candle['open'])*10000:.1f} pips") - lines.append(C(tf_color, "=" * 80)) - if not patterns: - lines.append(C('dim', f" No patterns detected on {tf_label}.")) - return "\n".join(lines) - lines.append(C('bold', f" PATTERNS DETECTED: {len(patterns)}")) - lines.append("-" * 80) - min_wr = cfg.get('min_historical_win_rate', 50.0) - min_sig = cfg.get('min_signals_for_stats', 5) - for i, pat in enumerate(patterns, 1): - direction = pat['direction'] - dir_color = 'green' if direction == 'Bullish' else ('red' if direction == 'Bearish' else 'yellow') - - # ── Pre-compute tier, session quality, cross quality ── - tier_letter, tier_label, tier_color = compute_pattern_tier(pat['name'], stats, cfg) - sess_quality, sess_q_color, sess_wr, sess_n, sess_amr = compute_session_quality(pat['session'], stats, cfg) - cross_wr, cross_n, cross_amr, cross_ql, cross_qc = compute_cross_quality(pat['name'], pat['session'], stats, cfg) - - # ── Pattern header with TIER badge ── - tier_badge = C(tier_color, C('bold', f'[Tier {tier_letter}]')) - lines.append(f"\n {C('bold', f'Pattern #{i}:')} {C(dir_color, pat['name'])} {tier_badge} {C(tier_color, tier_label)}") - lines.append(f" TF: {C(tf_color, tf_label)} | Category: {pat['category']} | Direction: {C(dir_color, direction)}") - atr_display_tf = pat.get('atr_tf', tf_label) - atr_label = f"ATR({cfg['atr_period']},{atr_display_tf})" if atr_display_tf != tf_label else f"ATR({cfg['atr_period']})" - lines.append(f" Session: {C('cyan', pat['session'])} | Trend: {pat['trend']} | {atr_label}: {pat['atr']:.5f}") - - vol_val = pat.get('Volume_Confirmed', 'N/A') - vol_color = 'green' if vol_val is True else ('red' if vol_val is False else 'dim') - lines.append(f" Volume Confirmed: {C(vol_color, str(vol_val))}") - - d1_val = pat.get('D1_Trend', 'N/A') - d1_color = 'green' if d1_val == 'uptrend' else ('red' if d1_val == 'downtrend' else 'yellow') - lines.append(f" D1 Trend: {C(d1_color, str(d1_val))}") - - # ── QUALITY SUMMARY LINE (tier + session + cross) ── - pat_stats = stats.get('patterns', {}).get(pat['name'], {}) - sess_stats = stats.get('sessions', {}).get(pat['session'], {}) - cross_key = f"{pat['name']}|{pat['session']}" - cross_stats = stats.get('cross', {}).get(cross_key, {}) - pat_wr = pat_stats.get('win_rate', 0) - pat_n = pat_stats.get('total', 0) - pat_amr = pat_stats.get('avg_max_r', 0) - - quality_parts = [] - quality_parts.append(f"Tier {tier_letter}:{C(tier_color, tier_label)}") - if pat_n >= min_sig: - wr_color = 'green' if pat_wr >= min_wr else ('yellow' if pat_wr >= 45 else 'red') - quality_parts.append(f"WR {C(wr_color, f'{pat_wr:.1f}%')} ({pat_n})") - quality_parts.append(f"Sess:{C(sess_q_color, f'{sess_quality}')}") - if sess_n >= min_sig: - quality_parts.append(f"SessWR {C(sess_q_color, f'{sess_wr:.1f}%')}") - if cross_wr is not None: - quality_parts.append(f"CrossWR {C(cross_qc, f'{cross_wr:.1f}%')} ({cross_n})") - quality_parts.append(f"AvgR {pat_amr:.2f}R") - lines.append(f" {C('bold', 'QUALITY:')} {' | '.join(quality_parts)}") - - lines.append(f" ENTRY: {C('bold', pat['entry_type'])} | Price: {C('bold', str(pat['entry_price']))} | Aggressive: {pat['aggressive_entry']:.5f}") - lines.append(f" Reason: {pat['entry_reason']}") - - # ── Gather historical probability data ── - best_tp_pct = None - best_wr = None - best_src = '' - if cross_stats and cross_stats.get('signals', 0) >= min_sig: - best_tp_pct = cross_stats.get('tp_hit_pct') - best_wr = cross_stats.get('win_rate') - best_src = 'cross' - if best_tp_pct is None and pat_stats and pat_stats.get('total', 0) >= min_sig: - best_tp_pct = pat_stats.get('tp_hit_pct') - best_wr = pat_stats.get('win_rate') - best_src = 'pattern' - if best_tp_pct is None: - overall = stats.get('overall', {}) - if overall.get('total_signals', 0) >= min_sig: - best_tp_pct = overall.get('tp_hit_pct') - best_wr = overall.get('win_rate') - best_src = 'overall' - - # ── BUY / SELL / BREAKOUT with probability ── - if direction == 'Bullish': - sl_str = C('red', f"SL: {pat['sl']:.5f}") - tp_str = C('green', f"TP: {pat['tp']:.5f}") - buy_line = f" >>> {C('green', C('bold', 'BUY'))} | Entry: {pat['entry_price']:.5f} {sl_str} {tp_str}" - if best_tp_pct is not None: - prob_color = 'green' if best_tp_pct >= 40 else ('yellow' if best_tp_pct >= 30 else 'red') - buy_line += f" {C(prob_color, C('bold', f'Prob(TP): {best_tp_pct:.1f}%'))}" - lines.append(buy_line) - if pat['sl_dist_pips'] is not None: - lines.append(f" >>> SL: {pat['sl_dist_pips']:.1f} pips | TP: {pat['tp_dist_pips']:.1f} pips | R:R 1:{pat['rr_ratio']:.2f}") - elif direction == 'Bearish': - sl_str = C('red', f"SL: {pat['sl']:.5f}") - tp_str = C('green', f"TP: {pat['tp']:.5f}") - sell_line = f" >>> {C('red', C('bold', 'SELL'))} | Entry: {pat['entry_price']:.5f} {sl_str} {tp_str}" - if best_tp_pct is not None: - prob_color = 'green' if best_tp_pct >= 40 else ('yellow' if best_tp_pct >= 30 else 'red') - sell_line += f" {C(prob_color, C('bold', f'Prob(TP): {best_tp_pct:.1f}%'))}" - lines.append(sell_line) - if pat['sl_dist_pips'] is not None: - lines.append(f" >>> SL: {pat['sl_dist_pips']:.1f} pips | TP: {pat['tp_dist_pips']:.1f} pips | R:R 1:{pat['rr_ratio']:.2f}") - else: - lines.append(f" >>> {C('yellow', 'BREAKOUT')} | Buy: {pat['body_top']:.5f} | Sell: {pat['body_bottom']:.5f}") - - # ── Historical edge ── - has_edge = False - if pat_stats and pat_stats.get('total', 0) >= min_sig: - has_edge = True - lines.append(f" {C('bold', 'HISTORICAL EDGE:')}") - wr = pat_stats.get('win_rate', 0) - n = pat_stats.get('total', 0) - amr = pat_stats.get('avg_max_r', 0) - sl_pct = pat_stats.get('sl_hit_pct', 0) - tp_pct = pat_stats.get('tp_hit_pct', 0) - wr_tag = "HIGH" if wr >= min_wr else "LOW" - wr_tag_color = 'green' if wr >= min_wr else 'red' - lines.append(f" {pat['name']}: {C(wr_tag_color, f'WR {wr:.1f}% [{wr_tag}]')} ({n} signals) | Avg Max R: {amr:.2f}R | {C('red', f'SL: {sl_pct:.1f}%')} {C('green', f'TP: {tp_pct:.1f}%')}") - if sess_stats and sess_stats.get('signals', 0) >= min_sig: - sess_wr_val = sess_stats.get('win_rate', 0) - sess_n_val = sess_stats.get('signals', 0) - lines.append(f" Session {pat['session']}: {C(sess_q_color, f'{sess_wr_val:.1f}% WR [{sess_quality}]')} ({sess_n_val} signals)") - if cross_stats and cross_stats.get('signals', 0) >= min_sig: - cross_wr_val = cross_stats.get('win_rate', 0) - cross_n_val = cross_stats.get('signals', 0) - cross_amr_val = cross_stats.get('avg_max_r', 0) - lines.append(f" {pat['name']} in {pat['session']}: {C(cross_qc, f'{cross_wr_val:.1f}% WR [{cross_ql}]')} ({cross_n_val} signals) | Avg Max R: {cross_amr_val:.2f}R") - - # ── Signal quality score ── - score = pat.get('signal_score') - if score is None: - score = compute_signal_score(pat['name'], pat['session'], pat['direction'], stats, cfg) - if score is not None: - if score >= 65: - score_label = "STRONG"; score_color = 'green' - elif score >= 52: - score_label = "MODERATE"; score_color = 'yellow' - else: - score_label = "WEAK"; score_color = 'red' - lines.append(f" Signal Score: {C(score_color, C('bold', f'{score:.1f}/100 [{score_label}]'))}") - elif not has_edge: - lines.append(C('dim', f" HISTORICAL EDGE: Insufficient data (<{min_sig} signals)")) - - # Position sizing (standard account — standard lots only) - risk_pct = cfg.get('risk_percent', 1.0) - balance = cfg.get('account_balance', 100000) - if pat.get('sl_dist_pips') and pat['sl_dist_pips'] > 0: - risk_amount = balance * risk_pct / 100.0 - pip_value = 10 # EURUSD std lot - lots = round(risk_amount / (pat['sl_dist_pips'] * pip_value), 2) - lines.append(f" Position Size ({risk_pct:.1f}% of ${balance:,.0f}): {C('cyan', f'{lots:.2f} lots')}") - - lines.append(f" {'─' * 40}") - lines.append(C(tf_color, "=" * 80)) - return "\n".join(lines) - - -# ============================================================ -# MT5 CONNECTION HELPERS -# ============================================================ - -def connect_mt5(cfg=None): - """Initialize and log in to MT5 using credentials from CFG (loaded from .env).""" - if cfg is None: cfg = CFG - if not _ENV_LOADED: - log_message(C('yellow', "WARNING: .env file not found — using fallback credentials"), cfg) - log_message("Initializing MT5 connection...", cfg) - if not mt5.initialize(path=cfg['mt5_path']): - log_message(f"MT5 initialization failed: {mt5.last_error()}", cfg) - return False - log_message(f"MT5 initialized. Version: {mt5.version()}", cfg) - if not mt5.login(login=cfg['account'], password=cfg['password'], server=cfg['server']): - log_message(f"MT5 login failed: {mt5.last_error()}", cfg) - mt5.shutdown() - return False - log_message(f"Connected to {cfg['server']} account {cfg['account']}", cfg) - return True - - -def mt5_reconnect(cfg=None): - """Attempt MT5 reconnection with exponential backoff.""" - if cfg is None: cfg = CFG - max_attempts = cfg.get('max_reconnect_attempts', 5) - base_backoff = cfg.get('reconnect_backoff_base', 10) - for attempt in range(1, max_attempts + 1): - log_message(f"Reconnection attempt {attempt}/{max_attempts}...", cfg) - try: - mt5.shutdown() - except Exception: - pass - time.sleep(base_backoff * (2 ** (attempt - 1))) - if connect_mt5(cfg): - log_message(f"Reconnection successful on attempt {attempt}.", cfg) - return True - log_message(f"Reconnection attempt {attempt} failed.", cfg) - log_message(f"All {max_attempts} reconnection attempts failed.", cfg) - return False - - -def fetch_rates(symbol, tf_label, num_bars, cfg=None): - """Fetch the most recent `num_bars` bars for the given symbol and TF label.""" - if cfg is None: cfg = CFG - tf_info = TIMEFRAME_MAP.get(tf_label) - if tf_info is None: - log_message(f"Unknown timeframe: {tf_label}", cfg) - return None - rates = mt5.copy_rates_from_pos(symbol, tf_info['mt5_tf'], 0, num_bars) - if rates is None: - log_message(f"Failed to fetch {tf_label} rates: {mt5.last_error()}", cfg) - return rates - - -def fetch_rates_range(symbol, tf_label, date_from, date_to, cfg=None): - """Fetch bars in a date range for full backtest.""" - if cfg is None: cfg = CFG - tf_info = TIMEFRAME_MAP.get(tf_label) - if tf_info is None: - return None - warmup_bars = cfg.get('warmup_bars', 30) - warmup_start = date_from - timedelta(minutes=warmup_bars * tf_info['minutes'] + 24 * 60) - rates = mt5.copy_rates_range(symbol, tf_info['mt5_tf'], warmup_start, date_to) - if rates is None or len(rates) == 0: - print(f"[MT5] No {tf_label} data for {symbol} from {warmup_start} to {date_to}") - return None - df = pd.DataFrame(rates) - df['DATETIME'] = pd.to_datetime(df['time'], unit='s') - df.sort_values('DATETIME', inplace=True) - df.reset_index(drop=True, inplace=True) - df.rename(columns={'open': 'OPEN', 'high': 'HIGH', 'low': 'LOW', 'close': 'CLOSE', - 'tick_volume': 'TICKVOL', 'real_volume': 'VOL', 'spread': 'SPREAD'}, inplace=True) - df['DATE'] = df['DATETIME'].dt.strftime('%Y.%m.%d') - df['TIME'] = df['DATETIME'].dt.strftime('%H:%M:%S') - df['IN_RANGE'] = (df['DATETIME'] >= date_from) & (df['DATETIME'] <= date_to) - df['BODY'] = abs(df['CLOSE'] - df['OPEN']) - df['BODY_SIGN'] = np.where(df['CLOSE'] >= df['OPEN'], 1, -1) - df['RANGE'] = df['HIGH'] - df['LOW'] - df['UPPER_WICK'] = df['HIGH'] - df[['OPEN', 'CLOSE']].max(axis=1) - df['LOWER_WICK'] = df[['OPEN', 'CLOSE']].min(axis=1) - df['LOW'] - df['BODY_RATIO'] = np.where(df['RANGE'] > 0, df['BODY'] / df['RANGE'], 0) - wc = (~df['IN_RANGE']).sum(); ic = df['IN_RANGE'].sum() - print(f" [{tf_label}] Fetched {len(df)} bars ({wc} warmup + {ic} in range)") - return df - - -# ============================================================ -# BACKTEST — DataFrame-based pattern detectors -# ============================================================ - -def fb_compute_atr(df, period): - tr = pd.DataFrame({ - 'hl': df['HIGH'] - df['LOW'], - 'hc': abs(df['HIGH'] - df['CLOSE'].shift(1)), - 'lc': abs(df['LOW'] - df['CLOSE'].shift(1)), - }).max(axis=1) - return tr.rolling(window=period, min_periods=1).mean() - - -def fb_compute_htf_atr(df, htf_df, atr_period): - """Compute ATR on a higher-timeframe DataFrame and map it to the lower-TF df. - - For each bar in `df`, finds the most recent HTF bar whose DATETIME <= df bar's - DATETIME and uses that HTF bar's ATR value. This allows M5/M15 bars to use H1 ATR - for SL/TP sizing, giving much more realistic stop distances. - - Args: - df: Lower-timeframe DataFrame (the one being scanned for patterns). - htf_df: Higher-timeframe DataFrame with at least HIGH, LOW, CLOSE, DATETIME columns. - atr_period: ATR period for the rolling mean. - - Returns: - pd.Series aligned with df's index, containing the HTF ATR values. - """ - if htf_df is None or len(htf_df) == 0: - # Fallback: compute native ATR - return fb_compute_atr(df, atr_period) - - # Compute ATR on the higher-timeframe - htf_atr = fb_compute_atr(htf_df, atr_period) - htf_with_atr = htf_df[['DATETIME']].copy() - htf_with_atr['ATR'] = htf_atr.values - - # Build an ATR lookup: for each lower-TF bar, find the most recent HTF ATR - # Use merge_asof for efficient time-based alignment - result = pd.merge_asof( - df[['DATETIME']].copy().sort_values('DATETIME'), - htf_with_atr.sort_values('DATETIME'), - on='DATETIME', - direction='backward' # use the HTF bar at or before the current bar - ) - # Re-align with original df index (merge_asof sorts by DATETIME) - result.index = df.index - return result['ATR'] - - -def fb_detect_trend(df, idx, cfg=None): - if cfg is None: cfg = CFG - lookback = cfg.get('trend_lookback', 20) - if idx < lookback: - return 'ranging' - subset = df.iloc[max(0, idx - lookback):idx + 1] - return detect_trend(subset, cfg) - - -def fb_detect_doji(df, idx, cfg=None): - if cfg is None: cfg = CFG - r = df.iloc[idx] - return r['RANGE'] > 0 and r['BODY_RATIO'] <= cfg['doji_body_ratio'] - -def fb_detect_spinning_top(df, idx, cfg=None): - if cfg is None: cfg = CFG - r = df.iloc[idx] - if r['RANGE'] == 0 or r['BODY'] == 0: return False - if r['BODY_RATIO'] > cfg['spinning_top_body_ratio']: return False - return r['UPPER_WICK'] >= r['BODY'] and r['LOWER_WICK'] >= r['BODY'] - -def fb_detect_marubozu(df, idx, cfg=None): - if cfg is None: cfg = CFG - r = df.iloc[idx] - if r['BODY'] == 0: return False - if r['BODY_RATIO'] < cfg['long_candle_ratio']: return False - return (r['UPPER_WICK'] <= r['BODY'] * cfg['marubozu_wick_ratio'] and - r['LOWER_WICK'] <= r['BODY'] * cfg['marubozu_wick_ratio']) - -def fb_detect_hammer(df, idx, cfg=None): - if cfg is None: cfg = CFG - r = df.iloc[idx] - if idx < 3: return False - trend = fb_detect_trend(df, idx, cfg) - if trend not in ('downtrend', 'ranging'): return False - if r['BODY'] == 0: return False - return r['LOWER_WICK'] >= r['BODY'] * cfg['hammer_lower_wick_ratio'] and r['UPPER_WICK'] <= r['BODY'] * cfg['hammer_upper_wick_ratio'] - -def fb_detect_inverted_hammer(df, idx, cfg=None): - if cfg is None: cfg = CFG - r = df.iloc[idx] - if idx < 3: return False - trend = fb_detect_trend(df, idx, cfg) - if trend not in ('downtrend', 'ranging'): return False - if r['BODY'] == 0: return False - return r['UPPER_WICK'] >= r['BODY'] * cfg['hammer_lower_wick_ratio'] and r['LOWER_WICK'] <= r['BODY'] * cfg['hammer_upper_wick_ratio'] - -def fb_detect_shooting_star(df, idx, cfg=None): - if cfg is None: cfg = CFG - r = df.iloc[idx] - if idx < 3: return False - trend = fb_detect_trend(df, idx, cfg) - if trend not in ('uptrend', 'ranging'): return False - if r['BODY'] == 0: return False - return r['UPPER_WICK'] >= r['BODY'] * cfg['hammer_lower_wick_ratio'] and r['LOWER_WICK'] <= r['BODY'] * cfg['hammer_upper_wick_ratio'] - -def fb_detect_hanging_man(df, idx, cfg=None): - if cfg is None: cfg = CFG - r = df.iloc[idx] - if idx < 3: return False - trend = fb_detect_trend(df, idx, cfg) - if trend not in ('uptrend', 'ranging'): return False - if r['BODY'] == 0: return False - return r['LOWER_WICK'] >= r['BODY'] * cfg['hammer_lower_wick_ratio'] and r['UPPER_WICK'] <= r['BODY'] * cfg['hammer_upper_wick_ratio'] - -def fb_detect_near_engulfing(df, idx, cfg=None): - if cfg is None: cfg = CFG - if idx < 1: return None - c = df.iloc[idx]; p = df.iloc[idx-1] - if c['BODY'] == 0 or p['BODY'] == 0: return None - tol = cfg.get('engulf_tolerance_pips', 2.0) * 0.0001 - if c['BODY_SIGN'] == 1 and p['BODY_SIGN'] == -1: - if not (c['OPEN'] <= p['CLOSE'] and c['CLOSE'] >= p['OPEN']): - if c['OPEN'] <= p['CLOSE'] + tol and c['CLOSE'] >= p['OPEN'] - tol: - return 'Near Bullish Engulfing' - if c['BODY_SIGN'] == -1 and p['BODY_SIGN'] == 1: - if not (c['OPEN'] >= p['CLOSE'] and c['CLOSE'] <= p['OPEN']): - if c['OPEN'] >= p['CLOSE'] - tol and c['CLOSE'] <= p['OPEN'] + tol: - return 'Near Bearish Engulfing' - return None - -def fb_detect_engulfing(df, idx, cfg=None): - if idx < 1: return None - c = df.iloc[idx]; p = df.iloc[idx-1] - if c['BODY'] == 0 or p['BODY'] == 0: return None - if c['BODY_SIGN'] == 1 and p['BODY_SIGN'] == -1 and c['OPEN'] <= p['CLOSE'] and c['CLOSE'] >= p['OPEN']: - return 'Bullish Engulfing' - if c['BODY_SIGN'] == -1 and p['BODY_SIGN'] == 1 and c['OPEN'] >= p['CLOSE'] and c['CLOSE'] <= p['OPEN']: - return 'Bearish Engulfing' - return None - -def fb_detect_harami(df, idx, cfg=None): - if cfg is None: cfg = CFG - if idx < 1: return None - c = df.iloc[idx]; p = df.iloc[idx-1] - if c['BODY'] == 0 or p['BODY'] == 0: return None - if p['BODY_RATIO'] < cfg['long_candle_ratio'] * 0.8: return None - ch = max(c['OPEN'], c['CLOSE']); cl = min(c['OPEN'], c['CLOSE']) - ph = max(p['OPEN'], p['CLOSE']); pl = min(p['OPEN'], p['CLOSE']) - if ch <= ph and cl >= pl: - if p['BODY_SIGN'] == -1 and c['BODY_SIGN'] == 1: return 'Bullish Harami' - if p['BODY_SIGN'] == 1 and c['BODY_SIGN'] == -1: return 'Bearish Harami' - return None - -def fb_detect_morning_star(df, idx, cfg=None): - if cfg is None: cfg = CFG - if idx < 2: return False - f, s, t = df.iloc[idx-2], df.iloc[idx-1], df.iloc[idx] - if f['BODY_SIGN'] != -1 or f['BODY_RATIO'] < cfg['long_candle_ratio']: return False - if s['BODY_RATIO'] > cfg['small_candle_ratio'] + 0.1: return False - if t['BODY_SIGN'] != 1 or t['BODY_RATIO'] < cfg['long_candle_ratio'] * 0.7: return False - return t['CLOSE'] > (f['OPEN'] + f['CLOSE']) / 2 - -def fb_detect_evening_star(df, idx, cfg=None): - if cfg is None: cfg = CFG - if idx < 2: return False - f, s, t = df.iloc[idx-2], df.iloc[idx-1], df.iloc[idx] - if f['BODY_SIGN'] != 1 or f['BODY_RATIO'] < cfg['long_candle_ratio']: return False - if s['BODY_RATIO'] > cfg['small_candle_ratio'] + 0.1: return False - if t['BODY_SIGN'] != -1 or t['BODY_RATIO'] < cfg['long_candle_ratio'] * 0.7: return False - return t['CLOSE'] < (f['OPEN'] + f['CLOSE']) / 2 - -def fb_detect_three_white_soldiers(df, idx, cfg=None): - if idx < 2: return False - c1, c2, c3 = df.iloc[idx-2], df.iloc[idx-1], df.iloc[idx] - if c1['BODY_SIGN'] != 1 or c2['BODY_SIGN'] != 1 or c3['BODY_SIGN'] != 1: return False - if c1['BODY_RATIO'] < 0.5 or c2['BODY_RATIO'] < 0.5 or c3['BODY_RATIO'] < 0.5: return False - return c3['CLOSE'] > c2['CLOSE'] > c1['CLOSE'] - -def fb_detect_three_black_crows(df, idx, cfg=None): - if idx < 2: return False - c1, c2, c3 = df.iloc[idx-2], df.iloc[idx-1], df.iloc[idx] - if c1['BODY_SIGN'] != -1 or c2['BODY_SIGN'] != -1 or c3['BODY_SIGN'] != -1: return False - if c1['BODY_RATIO'] < 0.5 or c2['BODY_RATIO'] < 0.5 or c3['BODY_RATIO'] < 0.5: return False - return c3['CLOSE'] < c2['CLOSE'] < c1['CLOSE'] - -def fb_detect_tweezer(df, idx, cfg=None): - if cfg is None: cfg = CFG - if idx < 1: return None - p = df.iloc[idx-1]; c = df.iloc[idx] - tol = cfg['tweezer_tolerance_pips'] * 0.0001 - trend = fb_detect_trend(df, idx, cfg) - if abs(p['HIGH'] - c['HIGH']) <= tol and trend in ('uptrend', 'ranging'): return 'Tweezer Tops' - if abs(p['LOW'] - c['LOW']) <= tol and trend in ('downtrend', 'ranging'): return 'Tweezer Bottoms' - return None - -def fb_detect_rising_three_methods(df, idx, cfg=None): - if cfg is None: cfg = CFG - if idx < 4: return False - first = df.iloc[idx-4]; fifth = df.iloc[idx] - if first['BODY_SIGN'] != 1 or first['BODY_RATIO'] < cfg['long_candle_ratio']: return False - for i in range(1, 4): - c = df.iloc[idx-4+i] - if c['BODY_RATIO'] > cfg['small_candle_ratio'] + 0.15: return False - if c['HIGH'] > first['HIGH'] or c['LOW'] < first['LOW']: return False - if fifth['BODY_SIGN'] != 1 or fifth['BODY_RATIO'] < cfg['long_candle_ratio'] * 0.7: return False - return fifth['CLOSE'] > first['CLOSE'] - -def fb_detect_falling_three_methods(df, idx, cfg=None): - if cfg is None: cfg = CFG - if idx < 4: return False - first = df.iloc[idx-4]; fifth = df.iloc[idx] - if first['BODY_SIGN'] != -1 or first['BODY_RATIO'] < cfg['long_candle_ratio']: return False - for i in range(1, 4): - c = df.iloc[idx-4+i] - if c['BODY_RATIO'] > cfg['small_candle_ratio'] + 0.15: return False - if c['HIGH'] > first['HIGH'] or c['LOW'] < first['LOW']: return False - if fifth['BODY_SIGN'] != -1 or fifth['BODY_RATIO'] < cfg['long_candle_ratio'] * 0.7: return False - return fifth['CLOSE'] < first['CLOSE'] - - -# ============================================================ -# FORWARD EVALUATION (intra-candle path simulation) -# ============================================================ - -def simulate_forward_evaluation(df, idx, direction, sl_price, tp_price, r_levels, - forward_candles, cfg=None, fill_price=None): - """Forward evaluation with intra-candle path simulation to avoid look-ahead bias.""" - if cfg is None: cfg = CFG - max_r_levels = cfg.get('max_r_levels', 5) - r_hits = {f'R{r}_Hit': None for r in range(1, max_r_levels+1)} - sl_hit = tp_hit = False - highest_r = 0 - outcome = 'Timeout' - - if direction not in ('Bullish', 'Bearish'): - return {'sl_hit': None, 'tp_hit': None, 'outcome': 'N/A', - 'max_r': None, 'r_hits': r_hits, 'fill_price': None, 'entry_filled': True} - - if idx + 1 >= len(df): - return {'sl_hit': None, 'tp_hit': None, 'outcome': 'Timeout', - 'max_r': 0, 'r_hits': r_hits, 'fill_price': None, 'entry_filled': False} - - end_idx = min(idx + 1 + forward_candles, len(df)) - future = df.iloc[idx+1:end_idx] - if len(future) == 0: - return {'sl_hit': None, 'tp_hit': None, 'outcome': 'Timeout', - 'max_r': 0, 'r_hits': r_hits, 'fill_price': None, 'entry_filled': False} - - stopped = False - for _, fc in future.iterrows(): - if stopped: - break - fc_high = fc['HIGH']; fc_low = fc['LOW'] - fc_open = fc['OPEN']; fc_close = fc['CLOSE'] - is_bullish_c = fc_close > fc_open - is_bearish_c = fc_close < fc_open - - sl_in_range = (fc_low <= sl_price if direction == 'Bullish' else fc_high >= sl_price) - tp_in_range = (fc_high >= tp_price if direction == 'Bullish' else fc_low <= tp_price) - - if sl_in_range and tp_in_range: - sl_hit = tp_hit = True - if direction == 'Bullish': - outcome = 'SL_Hit' if is_bullish_c else ('TP_Hit' if is_bearish_c else 'SL_Hit') - else: - outcome = 'TP_Hit' if is_bullish_c else ('SL_Hit' if is_bearish_c else 'SL_Hit') - stopped = True - elif sl_in_range: - sl_hit = True; outcome = 'SL_Hit'; stopped = True - elif tp_in_range: - tp_hit = True; outcome = 'TP_Hit' - for r in range(1, max_r_levels+1): - rv = r_levels.get(f'R{r}') - if rv is not None: - if (direction == 'Bullish' and fc_high >= rv) or (direction == 'Bearish' and fc_low <= rv): - r_hits[f'R{r}_Hit'] = True; highest_r = max(highest_r, r) - stopped = True - else: - for r in range(1, max_r_levels+1): - rv = r_levels.get(f'R{r}') - if rv is not None: - if (direction == 'Bullish' and fc_high >= rv) or (direction == 'Bearish' and fc_low <= rv): - r_hits[f'R{r}_Hit'] = True; highest_r = max(highest_r, r) - - for r in range(1, max_r_levels+1): - if r_hits[f'R{r}_Hit'] is None: - r_hits[f'R{r}_Hit'] = False - - if not stopped and len(future) > 0: - final_close = future.iloc[-1]['CLOSE'] - benchmark = fill_price if fill_price is not None else df.iloc[idx]['CLOSE'] - if direction == 'Bullish': - outcome = 'Marginal_Win' if final_close > benchmark else 'Marginal_Loss' - elif direction == 'Bearish': - outcome = 'Marginal_Win' if final_close < benchmark else 'Marginal_Loss' - - return {'sl_hit': sl_hit, 'tp_hit': tp_hit, 'outcome': outcome, - 'max_r': highest_r, 'r_hits': r_hits, 'fill_price': None, 'entry_filled': True} - - -# ============================================================ -# BACKTEST: detect all patterns on a DataFrame (one timeframe) -# ============================================================ - -def fb_detect_all_patterns(df, cfg=None, d1_df=None, tf_label='H4', htf_atr_df=None): - """Run all pattern detectors on in-range candles. Returns list of detection dicts. - - Args: - htf_atr_df: Optional higher-timeframe DataFrame for ATR calculation. - If provided and the atr_tf_by_tf mapping indicates a different - ATR source TF, ATR is computed from this DF instead of the native TF. - """ - if cfg is None: cfg = CFG - atr_period = cfg.get('atr_period', 14) - sl_mult = cfg.get('sl_multiplier', 1.5) - tp_mult = cfg.get('tp_multiplier', 1.5) - forward_candles = get_forward_candles(tf_label, cfg) - - # Determine ATR source: native or higher timeframe - atr_tf = get_atr_tf(tf_label, cfg) - if htf_atr_df is not None and atr_tf != tf_label: - atr = fb_compute_htf_atr(df, htf_atr_df, atr_period) - print(f" [{tf_label}] Using {atr_tf} ATR for SL/TP (native {tf_label} ATR too small)") - else: - atr = fb_compute_atr(df, atr_period) - df['ATR'] = atr - - vol_ma_period = cfg.get('volume_ma_period', 20) - df['VOL_MA'] = df['TICKVOL'].rolling(window=vol_ma_period, min_periods=1).mean() - - detections = [] - in_range = df.index[df['IN_RANGE']].tolist() - total = len(in_range) - - for count, idx in enumerate(in_range, 1): - if count % 200 == 0 or count == total: - print(f" [{tf_label}] Scanning candle {count}/{total} ...", end='\r') - - row = df.iloc[idx] - found = [] - - if fb_detect_doji(df, idx, cfg): - found.append({'Pattern': 'Doji', 'Category': 'Neutral', 'Direction': 'Neutral', 'Candles': 1}) - if fb_detect_spinning_top(df, idx, cfg): - found.append({'Pattern': 'Spinning Top', 'Category': 'Neutral', 'Direction': 'Neutral', 'Candles': 1}) - if fb_detect_marubozu(df, idx, cfg): - d = 'Bullish' if row['BODY_SIGN'] == 1 else 'Bearish' - found.append({'Pattern': f'Marubozu ({d})', 'Category': f'{d} Continuation', 'Direction': d, 'Candles': 1}) - if fb_detect_hammer(df, idx, cfg): - found.append({'Pattern': 'Hammer', 'Category': 'Bullish Reversal', 'Direction': 'Bullish', 'Candles': 1}) - if fb_detect_inverted_hammer(df, idx, cfg): - found.append({'Pattern': 'Inverted Hammer', 'Category': 'Bullish Reversal', 'Direction': 'Bullish', 'Candles': 1}) - if fb_detect_shooting_star(df, idx, cfg): - found.append({'Pattern': 'Shooting Star', 'Category': 'Bearish Reversal', 'Direction': 'Bearish', 'Candles': 1}) - if fb_detect_hanging_man(df, idx, cfg): - found.append({'Pattern': 'Hanging Man', 'Category': 'Bearish Reversal', 'Direction': 'Bearish', 'Candles': 1}) - - eng = fb_detect_engulfing(df, idx, cfg) - if eng: - d = 'Bullish' if 'Bullish' in eng else 'Bearish' - found.append({'Pattern': eng, 'Category': f'{d} Reversal', 'Direction': d, 'Candles': 2}) - - ne = fb_detect_near_engulfing(df, idx, cfg) - if ne: - d = 'Bullish' if 'Bullish' in ne else 'Bearish' - found.append({'Pattern': ne, 'Category': f'{d} Reversal', 'Direction': d, 'Candles': 2}) - - har = fb_detect_harami(df, idx, cfg) - if har: - d = 'Bullish' if 'Bullish' in har else 'Bearish' - found.append({'Pattern': har, 'Category': f'{d} Reversal', 'Direction': d, 'Candles': 2}) - - tw = fb_detect_tweezer(df, idx, cfg) - if tw: - d = 'Bearish' if 'Tops' in tw else 'Bullish' - found.append({'Pattern': tw, 'Category': f'{d} Reversal', 'Direction': d, 'Candles': 2}) - - if fb_detect_morning_star(df, idx, cfg): - found.append({'Pattern': 'Morning Star', 'Category': 'Bullish Reversal', 'Direction': 'Bullish', 'Candles': 3}) - if fb_detect_evening_star(df, idx, cfg): - found.append({'Pattern': 'Evening Star', 'Category': 'Bearish Reversal', 'Direction': 'Bearish', 'Candles': 3}) - if fb_detect_three_white_soldiers(df, idx, cfg): - found.append({'Pattern': 'Three White Soldiers', 'Category': 'Bullish Reversal', 'Direction': 'Bullish', 'Candles': 3}) - if fb_detect_three_black_crows(df, idx, cfg): - found.append({'Pattern': 'Three Black Crows', 'Category': 'Bearish Reversal', 'Direction': 'Bearish', 'Candles': 3}) - if fb_detect_rising_three_methods(df, idx, cfg): - found.append({'Pattern': 'Rising Three Methods', 'Category': 'Bullish Continuation', 'Direction': 'Bullish', 'Candles': 5}) - if fb_detect_falling_three_methods(df, idx, cfg): - found.append({'Pattern': 'Falling Three Methods', 'Category': 'Bearish Continuation', 'Direction': 'Bearish', 'Candles': 5}) - - # Deduplicate - if cfg.get('deduplicate_signals', True): - found_dicts = [{'name': f['Pattern'], 'category': f['Category'], 'direction': f['Direction']} for f in found] - deduped = deduplicate_patterns(found_dicts, cfg) - found = [f for f in found if any(f['Pattern'] == d['name'] for d in deduped)] - - # D1 trend filter - d1_trend = 'N/A' - if cfg.get('d1_trend_filter', False) and d1_df is not None: - d1_trend = fb_get_d1_trend_at_time(d1_df, row['DATETIME'], cfg) - filtered_found = [] - for pat in found: - if pat['Direction'] == 'Bullish' and d1_trend == 'uptrend': filtered_found.append(pat) - elif pat['Direction'] == 'Bearish' and d1_trend == 'downtrend': filtered_found.append(pat) - elif pat['Direction'] == 'Neutral': filtered_found.append(pat) - found = filtered_found - - # Volume filter - if cfg.get('volume_filter', False): - vol_confirmed = row['TICKVOL'] >= cfg.get('volume_threshold', 1.0) * row['VOL_MA'] if row['VOL_MA'] > 0 else True - found = [p for p in found if p['Direction'] == 'Neutral' or vol_confirmed] - else: - vol_confirmed = True - - for pat in found: - det = fb_compute_details(df, idx, pat, atr.iloc[idx], sl_mult, tp_mult, - forward_candles, cfg, d1_df, vol_confirmed, tf_label, d1_trend, atr_tf) - if det is not None: - detections.append(det) - - print(f" [{tf_label}] Scanning complete.{' '*30}") - return detections - - -def fb_get_d1_trend_at_time(d1_df, h4_datetime, cfg=None): - """Get D1 trend at a given datetime.""" - d1_bar = d1_df[d1_df['DATETIME'] <= h4_datetime] - if len(d1_bar) == 0: - return 'ranging' - return d1_bar.iloc[-1].get('D1_TREND', 'ranging') - - -def fb_compute_details(df, idx, pinfo, current_atr, sl_mult, tp_mult, - forward_candles, cfg=None, d1_df=None, vol_confirmed=True, - tf_label='H4', d1_trend='N/A', atr_tf=None): - """Compute SL/TP/R-levels + forward evaluation for one pattern occurrence.""" - if cfg is None: cfg = CFG - row = df.iloc[idx] - direction = pinfo['Direction'] - rr_ratio = tp_mult / sl_mult - pip_divisor = cfg.get('pip_divisor', 0.0001) - max_r_levels = cfg.get('max_r_levels', 5) - - if direction == 'Bullish': - sl = row['LOW'] - sl_mult * current_atr - risk = row['CLOSE'] - sl - tp = row['CLOSE'] + risk * rr_ratio - elif direction == 'Bearish': - sl = row['HIGH'] + sl_mult * current_atr - risk = sl - row['CLOSE'] - tp = row['CLOSE'] - risk * rr_ratio - else: - sl_val = row['LOW'] - sl_mult * current_atr - risk_bull = row['CLOSE'] - sl_val - risk_bear = (row['HIGH'] + sl_mult * current_atr) - row['CLOSE'] - sl = f"{sl_val:.5f}" - tp = f"Long:{row['CLOSE']+risk_bull*rr_ratio:.5f}|Short:{row['CLOSE']-risk_bear*rr_ratio:.5f}" - risk = None - - sl_pips = round(risk / pip_divisor, 1) if risk is not None else None - - r_levels = {} - if direction == 'Bullish' and risk is not None and risk > 0: - for r in range(1, max_r_levels+1): r_levels[f'R{r}'] = round(row['CLOSE'] + r * risk, 5) - elif direction == 'Bearish' and risk is not None and risk > 0: - for r in range(1, max_r_levels+1): r_levels[f'R{r}'] = round(row['CLOSE'] - r * risk, 5) - - body_top = max(row['OPEN'], row['CLOSE']) - body_bottom = min(row['OPEN'], row['CLOSE']) - - if direction == 'Bullish': - if pinfo['Pattern'] in ('Hammer', 'Inverted Hammer', 'Morning Star', 'Three White Soldiers', - 'Tweezer Bottoms', 'Rising Three Methods') \ - or 'Bullish Engulfing' in pinfo['Pattern'] or 'Bullish Harami' in pinfo['Pattern']: - entry_type = 'Buy Stop'; entry_price = round(body_top, 5) - elif 'Marubozu' in pinfo['Pattern'] and 'Bullish' in pinfo['Pattern']: - entry_type = 'Market Buy'; entry_price = round(row['CLOSE'], 5) - else: - entry_type = 'Buy Stop'; entry_price = round(body_top, 5) - elif direction == 'Bearish': - if pinfo['Pattern'] in ('Evening Star', 'Shooting Star', 'Hanging Man', 'Three Black Crows', - 'Falling Three Methods', 'Tweezer Tops') \ - or 'Bearish Engulfing' in pinfo['Pattern'] or 'Bearish Harami' in pinfo['Pattern']: - entry_type = 'Sell Stop'; entry_price = round(body_bottom, 5) - elif 'Marubozu' in pinfo['Pattern'] and 'Bearish' in pinfo['Pattern']: - entry_type = 'Market Sell'; entry_price = round(row['CLOSE'], 5) - else: - entry_type = 'Sell Stop'; entry_price = round(body_bottom, 5) - else: - entry_type = 'Breakout'; entry_price = None - - # Entry verification - entry_filled = True; fill_price = entry_price; no_fill = False; gap_fill = False - if cfg.get('verify_entry', True) and direction in ('Bullish', 'Bearish') and idx + 1 < len(df): - next_c = df.iloc[idx+1] - if entry_type == 'Buy Stop': - if next_c['HIGH'] >= entry_price: - fill_price = round(max(next_c['OPEN'], entry_price), 5) - gap_fill = next_c['OPEN'] > entry_price - else: - entry_filled = False; no_fill = True - elif entry_type == 'Sell Stop': - if next_c['LOW'] <= entry_price: - fill_price = round(min(next_c['OPEN'], entry_price), 5) - gap_fill = next_c['OPEN'] < entry_price - else: - entry_filled = False; no_fill = True - elif entry_type in ('Market Buy', 'Market Sell'): - fill_price = round(df.iloc[idx+1]['OPEN'], 5) - - # Recalculate risk from fill price if verified - if cfg.get('verify_entry', True) and fill_price is not None and entry_filled and direction in ('Bullish', 'Bearish'): - if direction == 'Bullish' and isinstance(sl, float): - risk = fill_price - sl - if risk > 0: - tp = fill_price + risk * rr_ratio - for r in range(1, max_r_levels+1): r_levels[f'R{r}'] = round(fill_price + r * risk, 5) - sl_pips = round(risk / pip_divisor, 1) - elif direction == 'Bearish' and isinstance(sl, float): - risk = sl - fill_price - if risk > 0: - tp = fill_price - risk * rr_ratio - for r in range(1, max_r_levels+1): r_levels[f'R{r}'] = round(fill_price - r * risk, 5) - sl_pips = round(risk / pip_divisor, 1) - - # Forward evaluation - prediction_success = sl_hit_result = tp_hit_result = None - outcome = 'Timeout'; max_r = 0 - r_hits = {f'R{r}_Hit': None for r in range(1, max_r_levels+1)} - - if no_fill: - outcome = 'No_Fill'; entry_filled = False - elif direction in ('Bullish', 'Bearish') and isinstance(sl, float) and idx + 1 < len(df): - fwd = simulate_forward_evaluation(df, idx, direction, sl, tp, r_levels, - forward_candles, cfg, fill_price=fill_price) - sl_hit_result = fwd['sl_hit']; tp_hit_result = fwd['tp_hit'] - outcome = fwd['outcome']; max_r = fwd['max_r']; r_hits = fwd['r_hits'] - prediction_success = (True if outcome in ('TP_Hit', 'Marginal_Win') else - False if outcome in ('SL_Hit', 'Marginal_Loss', 'No_Fill') else None) - - hour = row['DATETIME'].hour - session = classify_session(hour, cfg) - trend = fb_detect_trend(df, idx, cfg) - - result = { - 'Timeframe': tf_label, - 'DateTime': row['DATETIME'], 'Date': row['DATE'], 'Time': row['TIME'], - 'Pattern': pinfo['Pattern'], 'Category': pinfo['Category'], 'Direction': direction, - 'Session': session, 'Trend_Context': trend, 'D1_Trend': d1_trend, - 'Open': row['OPEN'], 'High': row['HIGH'], 'Low': row['LOW'], 'Close': row['CLOSE'], - 'ATR': round(current_atr, 5) if not pd.isna(current_atr) else None, - 'ATR_TF': atr_tf or tf_label, - 'SL': round(sl, 5) if isinstance(sl, float) else sl, - 'TP': round(tp, 5) if isinstance(tp, float) else tp, - 'SL_Pips': sl_pips, - 'Risk_1R': round(risk, 5) if risk is not None else None, - 'TP_R_Multiple': round(rr_ratio, 2), - 'Entry_Type': entry_type, 'Entry_Price': entry_price, - 'Fill_Price': fill_price, 'Entry_Filled': entry_filled, 'Gap_Fill': gap_fill, - 'Outcome': outcome, 'Max_R': max_r, - 'Prediction_Success': prediction_success, - 'SL_Hit': sl_hit_result, 'TP_Hit': tp_hit_result, - 'Volume_Confirmed': vol_confirmed, - 'Candles_in_Pattern': pinfo['Candles'], - 'Forward_Candles': forward_candles, - } - for r in range(1, max_r_levels+1): - rk = f'R{r}' - result[rk] = r_levels.get(rk) - result[f'R{r}_Hit'] = r_hits.get(f'R{r}_Hit') - result[f'R{r}_Pips'] = (round(r * risk / pip_divisor, 1) - if r_levels.get(rk) is not None and risk is not None else None) - return result - - -# ============================================================ -# BACKTEST REPORT GENERATOR -# ============================================================ - -def fb_generate_report(detections, df, symbol, tf_label, cfg=None): - """Generate full text report for one timeframe backtest.""" - if cfg is None: cfg = CFG - max_r_levels = cfg.get('max_r_levels', 5) - forward_candles = get_forward_candles(tf_label, cfg) - tf_minutes = TIMEFRAME_MAP.get(tf_label, {}).get('minutes', 240) - - lines = [] - L = lines.append - L("=" * 120) - L(f"{symbol} [{tf_label}] PRICE ACTION PATTERN BACKTEST REPORT") - L("=" * 120) - L(f"Timeframe : {tf_label} ({tf_minutes} min candles)") - L(f"Data Period : {df.loc[df['IN_RANGE'],'DATE'].iloc[0]} to {df.loc[df['IN_RANGE'],'DATE'].iloc[-1]}") - L(f"Total Candles : {df['IN_RANGE'].sum()}") - L(f"Total Detections : {len(detections)}") - L(f"ATR Period : {cfg.get('atr_period', 14)}") - atr_src = get_atr_tf(tf_label, cfg) - if atr_src != tf_label: - L(f"ATR Source TF : {atr_src} (higher-TF ATR for wider SL/TP)") - else: - L(f"ATR Source TF : {tf_label} (native)") - L(f"SL Multiplier : {cfg.get('sl_multiplier', 1.5)} x ATR") - L(f"TP R:R : 1:{cfg.get('tp_multiplier', 1.5)/cfg.get('sl_multiplier', 1.5):.1f}") - L(f"Forward Eval : {forward_candles} candles = {forward_candles * tf_minutes // 60:.0f} hours") - L(f"D1 Trend Filter : {cfg.get('d1_trend_filter', False)}") - L(f"Volume Filter : {cfg.get('volume_filter', False)}") - L(f"Verify Entry : {cfg.get('verify_entry', True)}") - L(f"Deduplicate : {cfg.get('deduplicate_signals', True)}") - L("") - - if not detections: - L("No patterns detected.") - return "\n".join(lines) - - det_df = pd.DataFrame(detections) - directional = det_df[det_df['Direction'] != 'Neutral'] - - def bc(s): - s_ = int((s == True).sum()); f_ = int((s == False).sum()) - return s_, f_, round(s_ / (s_ + f_) * 100, 1) if (s_ + f_) > 0 else 0 - - L("-" * 120); L("SECTION 1: PATTERN FREQUENCY"); L("-" * 120) - for pat, cnt in det_df['Pattern'].value_counts().items(): - L(f" {pat:30s} | {det_df[det_df['Pattern']==pat]['Direction'].iloc[0]:10s} | Count: {cnt}") - - L(""); L("-" * 120); L("SECTION 2: SESSION DISTRIBUTION"); L("-" * 120) - for sess, cnt in det_df['Session'].value_counts().items(): - L(f" {sess:25s} | {cnt:4d} | {cnt/len(det_df)*100:.1f}%") - - L(""); L("-" * 120); L("SECTION 3: WIN RATE BY PATTERN"); L("-" * 120) - hdr = f" {'Pattern':30s} | {'Total':>6s} | {'Win':>5s} | {'Loss':>5s} | {'WR%':>6s} | {'SL%':>6s} | {'TP%':>6s} | {'AvgSL pips':>11s}" - L(hdr); L(" " + "-" * (len(hdr)-2)) - for pat in det_df['Pattern'].unique(): - ds = det_df[(det_df['Pattern'] == pat) & (det_df['Direction'] != 'Neutral')] - total = len(det_df[det_df['Pattern'] == pat]) - if len(ds) > 0: - s, f_, wr = bc(ds['Prediction_Success']) - slp = round((ds['SL_Hit'] == True).sum() / len(ds) * 100, 1) - tpp = round((ds['TP_Hit'] == True).sum() / len(ds) * 100, 1) - asp = ds['SL_Pips'].dropna().mean() - else: - s = f_ = 0; wr = slp = tpp = 0; asp = float('nan') - L(f" {pat:30s} | {total:6d} | {s:5d} | {f_:5d} | {wr:5.1f}% | {slp:5.1f}% | {tpp:5.1f}% | {asp:.1f}" if not pd.isna(asp) else - f" {pat:30s} | {total:6d} | {s:5d} | {f_:5d} | {wr:5.1f}% | {slp:5.1f}% | {tpp:5.1f}% | N/A") - - L(""); L("-" * 120); L("SECTION 4: OUTCOME BREAKDOWN"); L("-" * 120) - for outcome_name in ['TP_Hit', 'SL_Hit', 'Marginal_Win', 'Marginal_Loss', 'Timeout', 'No_Fill']: - cnt = int((directional['Outcome'] == outcome_name).sum()) if len(directional) > 0 else 0 - pct = round(cnt / len(directional) * 100, 1) if len(directional) > 0 else 0 - L(f" {outcome_name:20s} | {cnt:6d} | {pct:5.1f}%") - - L(""); L("-" * 120); L("SECTION 5: R-LEVEL HIT RATES BY PATTERN"); L("-" * 120) - rh = f" {'Pattern':30s} | {'Sigs':>5s}" + ''.join([f" | {'R'+str(r)+'%':>6s}" for r in range(1, max_r_levels+1)]) + " | AvgMaxR" - L(rh); L(" " + "-" * (len(rh)-2)) - for pat in det_df['Pattern'].unique(): - ds = det_df[(det_df['Pattern'] == pat) & (det_df['Direction'] != 'Neutral')] - if len(ds) == 0: continue - rp = [f" {pat:30s} | {len(ds):5d}"] - for r in range(1, max_r_levels+1): - col = f'R{r}_Hit' - if col in ds.columns: - hc = int((ds[col] == True).sum()); ec = int((ds[col].notna()).sum()) - rp.append(f" | {round(hc/ec*100, 0) if ec > 0 else 0:5.0f}%") - else: - rp.append(f" | {'N/A':>6s}") - amr = ds['Max_R'].dropna().mean() - rp.append(f" | {amr:.2f}" if not pd.isna(amr) else " | N/A") - L(''.join(rp)) - - L(""); L("-" * 120); L("SECTION 6: WIN RATE BY SESSION"); L("-" * 120) - for sess in directional['Session'].unique() if len(directional) > 0 else []: - sd = directional[directional['Session'] == sess] - if len(sd) == 0: continue - s, f_, wr = bc(sd['Prediction_Success']) - amr = sd['Max_R'].dropna().mean() - L(f" {sess:25s} | Signals: {len(sd):4d} | WR: {wr:.1f}% | AvgMaxR: {amr:.2f}R") - - L(""); L("-" * 120); L("SECTION 7: KEY STATISTICS"); L("-" * 120) - td = len(directional) - if td > 0: - s, f_, owr = bc(directional['Prediction_Success']) - L(f" Total directional signals : {td}") - L(f" Overall win rate : {owr:.1f}%") - L(f" SL hit rate : {round((directional['SL_Hit']==True).sum()/td*100, 1):.1f}%") - L(f" TP hit rate : {round((directional['TP_Hit']==True).sum()/td*100, 1):.1f}%") - L(f" Avg SL pips : {directional['SL_Pips'].dropna().mean():.1f}") - L(f" Avg Max R : {directional['Max_R'].dropna().mean():.2f}R") - for r in range(1, max_r_levels+1): - col = f'R{r}_Hit' - if col in directional.columns: - hc = int((directional[col] == True).sum()); ec = int(directional[col].notna().sum()) - L(f" R{r} hit rate : {round(hc/ec*100,1) if ec>0 else 0:.1f}%") - - L(""); L("=" * 120) - return "\n".join(lines) - - -# ============================================================ -# MODE 1: LIVE MULTI-TIMEFRAME SCANNER -# ============================================================ - -def run_scanner(cfg=None): - """Live scanner loop — monitors all active timeframes for new candle closes.""" - if cfg is None: cfg = CFG - active_tfs = cfg.get('active_timeframes', ['M5', 'M15', 'H1', 'H4', 'D1']) - symbol = cfg['symbol'] - - log_message(C('cyan', '=' * 70), cfg) - log_message(C('bold', f" {symbol} MULTI-TIMEFRAME PATTERN SCANNER v6 — STARTING"), cfg) - log_message(C('cyan', '=' * 70), cfg) - log_message(f"Active timeframes: {C('yellow', ', '.join(active_tfs))}", cfg) - sl_str = f"{cfg['sl_multiplier']}x ATR" - log_message(f"SL: {C('red', sl_str)} | TP R:R = 1:{cfg['tp_multiplier']/cfg['sl_multiplier']:.1f}", cfg) - - # Show ATR source per TF - atr_map_display = [] - for tf in active_tfs: - atr_src = get_atr_tf(tf, cfg) - atr_map_display.append(f"{tf}→{atr_src}" if atr_src != tf else tf) - log_message(f"ATR Source: {', '.join(atr_map_display)}", cfg) - - if cfg.get('d1_trend_filter', False): - log_message(f"D1 Trend Filter: {C('green', 'ENABLED')} (SMA {cfg['d1_sma_period']})", cfg) - if cfg.get('volume_filter', False): - log_message(f"Volume Filter: {C('green', 'ENABLED')} ({cfg['volume_threshold']}x avg)", cfg) - - # Load backtest stats for historical edge display - stats = load_latest_backtest_stats(cfg=cfg) - stats_last_refresh = datetime.now() - if stats.get('overall', {}).get('total_signals', 0) > 0: - owr = stats['overall'].get('win_rate', 0) - on = stats['overall'].get('total_signals', 0) - log_message(f"Backtest stats loaded: Overall WR {owr:.1f}% ({on} signals)", cfg) - else: - log_message("No backtest stats found. Run fullbacktest first for historical edge data.", cfg) - - # Print dashboard on start - if cfg.get('show_dashboard_on_start', True) and stats.get('overall', {}).get('total_signals', 0) > 0: - print_top_setups(stats, cfg) - - if not connect_mt5(cfg): - return - - # Track last candle time per timeframe - last_candle_time = {tf: None for tf in active_tfs} - d1_rates_cache = None - - try: - while True: - try: - # Refresh D1 data periodically for trend filter - if cfg.get('d1_trend_filter', False): - d1_rates_cache = fetch_rates(symbol, 'D1', cfg.get('bars_to_fetch', 50), cfg) - - # Auto-refresh stats cache periodically - if (datetime.now() - stats_last_refresh).total_seconds() > cfg.get('stats_cache_hours', 4) * 3600: - stats = load_latest_backtest_stats(cfg=cfg) - stats_last_refresh = datetime.now() - if stats.get('overall', {}).get('total_signals', 0) > 0: - log_message(f"Stats refreshed: Overall WR {stats['overall'].get('win_rate',0):.1f}%", cfg) - - # Fetch higher-timeframe ATR rates once per loop iteration - htf_atr_rates_cache = {} # tf_label → rates - atr_tfs_needed = set() - for tf in active_tfs: - atr_src = get_atr_tf(tf, cfg) - if atr_src != tf: - atr_tfs_needed.add(atr_src) - for atr_src in atr_tfs_needed: - htf_rates = fetch_rates(symbol, atr_src, cfg.get('bars_to_fetch', 50), cfg) - if htf_rates is not None: - htf_atr_rates_cache[atr_src] = htf_rates - - for tf_label in active_tfs: - tf_info = TIMEFRAME_MAP[tf_label] - poll_interval = cfg.get('poll_interval_by_tf', {}).get(tf_label, 30) - rates = fetch_rates(symbol, tf_label, cfg.get('bars_to_fetch', 50), cfg) - if rates is None: - continue - - # Resolve ATR source for this TF - atr_src = get_atr_tf(tf_label, cfg) - htf_atr_rates = htf_atr_rates_cache.get(atr_src) if atr_src != tf_label else None - - bar_time = rates[-1]['time'] - if isinstance(bar_time, (int, float, np.integer, np.floating)): - bar_time = datetime.fromtimestamp(int(bar_time)) - - if last_candle_time[tf_label] is None: - last_candle_time[tf_label] = bar_time - # Scan the most recent closed candle on startup - if len(rates) >= 2: - pats = scan_patterns(list(rates[:-1]), cfg, d1_rates_cache, tf_label, htf_atr_rates) - pats = apply_signal_score_filter(pats, stats, cfg) - log_message(format_pattern_output(rates[-2], pats, cfg, stats, tf_label), cfg) - continue - - if bar_time != last_candle_time[tf_label]: - next_close = bar_time + timedelta(minutes=tf_info['minutes']) - log_message( - C('bold', C('yellow', - f"\nNEW {tf_label} CANDLE CLOSED! | Next: {next_close.strftime('%Y-%m-%d %H:%M')}" - )), cfg - ) - pats = scan_patterns(list(rates[:-1]), cfg, d1_rates_cache, tf_label, htf_atr_rates) - pats = apply_signal_score_filter(pats, stats, cfg) - output = format_pattern_output(rates[-2], pats, cfg, stats, tf_label) - log_message(output, cfg) - - # Write alert file for signals with patterns - if pats: - try: - _ts = rates[-2]['time'] - if isinstance(_ts, (int, float, np.integer, np.floating)): - _ts = datetime.fromtimestamp(int(_ts)) - fname = f"alert_{tf_label}_{_ts.strftime('%Y%m%d_%H%M%S')}.txt" - with open(os.path.join(_LOG_DIR, fname), "w", encoding='utf-8') as f: - f.write(f"SIGNAL ALERT [{tf_label}] — {_ts}\n\n" + output) - except Exception: - pass - - last_candle_time[tf_label] = bar_time - - time.sleep(min(cfg.get('poll_interval_by_tf', {}).get(tf, 30) for tf in active_tfs)) - - except Exception as e: - log_message(f"Scanner iteration error: {e}", cfg) - if not mt5_reconnect(cfg): - break - - except KeyboardInterrupt: - log_message("\nScanner stopped by user (Ctrl+C)", cfg) - finally: - try: mt5.shutdown() - except Exception: pass - log_message("MT5 connection closed.", cfg) - - -# ============================================================ -# MODE 2: ONE-SHOT SCAN (all active timeframes) -# ============================================================ - -def run_single_scan(cfg=None): - """Single scan of the latest closed candle on all active timeframes.""" - if cfg is None: cfg = CFG - active_tfs = cfg.get('active_timeframes', ['M5', 'M15', 'H1', 'H4', 'D1']) - log_message(f"Running single scan on: {', '.join(active_tfs)}", cfg) - - # Load backtest stats - stats = load_latest_backtest_stats(cfg=cfg) - if stats.get('overall', {}).get('total_signals', 0) > 0: - owr = stats['overall'].get('win_rate', 0) - on = stats['overall'].get('total_signals', 0) - log_message(f"Backtest stats loaded: Overall WR {owr:.1f}% ({on} signals)", cfg) - else: - log_message("No backtest stats found. Run fullbacktest first for historical edge data.", cfg) - - # Print dashboard - if cfg.get('show_dashboard_on_start', True) and stats.get('overall', {}).get('total_signals', 0) > 0: - print_top_setups(stats, cfg) - - if not connect_mt5(cfg): - return - - d1_rates = None - if cfg.get('d1_trend_filter', False): - d1_rates = fetch_rates(cfg['symbol'], 'D1', cfg.get('bars_to_fetch', 50), cfg) - - # Pre-fetch higher-timeframe ATR rates - htf_atr_rates_cache = {} - atr_tfs_needed = set() - for tf in active_tfs: - atr_src = get_atr_tf(tf, cfg) - if atr_src != tf: - atr_tfs_needed.add(atr_src) - for atr_src in atr_tfs_needed: - htf_rates = fetch_rates(cfg['symbol'], atr_src, cfg.get('bars_to_fetch', 50), cfg) - if htf_rates is not None: - htf_atr_rates_cache[atr_src] = htf_rates - - for tf_label in active_tfs: - rates = fetch_rates(cfg['symbol'], tf_label, cfg.get('bars_to_fetch', 50), cfg) - if rates is None: - log_message(f"No data for {tf_label}.", cfg) - continue - closed = rates[-2] if len(rates) >= 2 else rates[-1] - scan_src = list(rates[:-1]) if len(rates) >= 2 else list(rates) - # Resolve ATR source for this TF - atr_src = get_atr_tf(tf_label, cfg) - htf_atr_rates = htf_atr_rates_cache.get(atr_src) if atr_src != tf_label else None - pats = scan_patterns(scan_src, cfg, d1_rates, tf_label, htf_atr_rates) - pats = apply_signal_score_filter(pats, stats, cfg) - log_message(format_pattern_output(closed, pats, cfg, stats, tf_label), cfg) - - try: mt5.shutdown() - except Exception: pass - log_message("Single scan complete.", cfg) - - -# ============================================================ -# MODE 3: QUICK BACKTEST (N recent bars, single TF) -# ============================================================ - -def run_quick_backtest(num_bars=500, cfg=None): - """Quick backtest over N recent bars. Supports all active timeframes.""" - if cfg is None: cfg = CFG - active_tfs = cfg.get('active_timeframes', ['H4']) - symbol = cfg['symbol'] - log_message(f"Quick backtest: {num_bars} bars on {', '.join(active_tfs)}", cfg) - if not connect_mt5(cfg): - return - - # Pre-fetch higher-timeframe ATR rates - htf_atr_rates_cache = {} - atr_tfs_needed = set() - for tf in active_tfs: - atr_src = get_atr_tf(tf, cfg) - if atr_src != tf: - atr_tfs_needed.add(atr_src) - for atr_src in atr_tfs_needed: - htf_rates = fetch_rates(symbol, atr_src, num_bars, cfg) - if htf_rates is not None: - htf_atr_rates_cache[atr_src] = htf_rates - - for tf_label in active_tfs: - log_message(f"\n--- [{tf_label}] ---", cfg) - rates = fetch_rates(symbol, tf_label, num_bars, cfg) - if rates is None: - log_message(f"No data for {tf_label}.", cfg) - continue - - # Resolve ATR source - atr_src = get_atr_tf(tf_label, cfg) - htf_atr_rates = htf_atr_rates_cache.get(atr_src) if atr_src != tf_label else None - - rates_list = list(rates) - total_patterns = 0 - pattern_counts = {} - all_detections = [] - - for i in range(5, len(rates_list) - 1): - subset = rates_list[:i+1] - pats = scan_patterns(subset, cfg, tf_label=tf_label, htf_atr_rates=htf_atr_rates) - if pats: - candle = rates_list[i] - total_patterns += len(pats) - for p in pats: - pattern_counts[p['name']] = pattern_counts.get(p['name'], 0) + 1 - ct = candle['time'] - if isinstance(ct, (int, float, np.integer, np.floating)): - ct = datetime.fromtimestamp(ct) - all_detections.append({ - 'Timeframe': tf_label, - 'DateTime': ct, 'Pattern': p['name'], - 'Direction': p['direction'], 'Session': p['session'], - 'Open': candle['open'], 'High': candle['high'], - 'Low': candle['low'], 'Close': candle['close'], - 'Entry_Type': p.get('entry_type'), 'Entry_Price': p.get('entry_price'), - 'ATR': p.get('atr'), 'SL': p.get('sl'), 'TP': p.get('tp'), - 'SL_Dist_Pips': p.get('sl_dist_pips'), - }) - - log_message(f"[{tf_label}] Total patterns: {total_patterns}", cfg) - for name, count in sorted(pattern_counts.items(), key=lambda x: -x[1]): - log_message(f" {name:35s}: {count}", cfg) - - if all_detections: - csv_path = os.path.join(_LOG_DIR, f"quick_backtest_{symbol}_{tf_label}.csv") - pd.DataFrame(all_detections).to_csv(csv_path, index=False) - log_message(f" → {csv_path}", cfg) - - try: mt5.shutdown() - except Exception: pass - log_message("\nQuick backtest complete.", cfg) - - -# ============================================================ -# MODE 4: FULL DATE-RANGED BACKTEST (all active timeframes) -# ============================================================ - -def run_full_backtest(args, cfg=None): - """Full backtest with date range, R-levels, forward evaluation across all active TFs.""" - if cfg is None: cfg = CFG - active_tfs = cfg.get('active_timeframes', ['M5', 'M15', 'H1', 'H4', 'D1']) - symbol = args.symbol - date_from = datetime.strptime(args.date_from, "%Y-%m-%d") - date_to = datetime.strptime(args.date_to, "%Y-%m-%d") + timedelta(days=1) - timedelta(seconds=1) - out_dir = args.output - - print("=" * 70) - print(f" {symbol} MULTI-TIMEFRAME BACKTESTER v6") - print("=" * 70) - print(f" Timeframes : {', '.join(active_tfs)}") - print(f" From : {args.date_from}") - print(f" To : {args.date_to}") - print(f" ATR : {cfg.get('atr_period', 14)}") - # Show ATR source per TF - atr_map_display = [] - for tf in active_tfs: - atr_src = get_atr_tf(tf, cfg) - atr_map_display.append(f"{tf}→{atr_src}" if atr_src != tf else tf) - print(f" ATR Source : {', '.join(atr_map_display)}") - print(f" SL/TP : {cfg.get('sl_multiplier', 1.5)}x / {cfg.get('tp_multiplier', 1.5)}x ATR") - print(f" Output : {out_dir}") - print() - - print("[1] Connecting to MT5...") - if not connect_mt5(cfg): - print("FATAL: Could not connect to MT5."); sys.exit(1) - - # Fetch D1 data for trend filter (shared across TFs if needed) - d1_df_cache = None - if cfg.get('d1_trend_filter', False): - print("[2] Fetching D1 data for trend filter...") - d1_df_cache = fetch_rates_range(symbol, 'D1', date_from, date_to, cfg) - if d1_df_cache is not None: - d1_df_cache['D1_SMA'] = d1_df_cache['CLOSE'].rolling(window=cfg.get('d1_sma_period', 20), min_periods=1).mean() - d1_df_cache['D1_TREND'] = np.where(d1_df_cache['CLOSE'] > d1_df_cache['D1_SMA'], 'uptrend', - np.where(d1_df_cache['CLOSE'] < d1_df_cache['D1_SMA'], 'downtrend', 'ranging')) - - # Pre-fetch higher-timeframe data for ATR calculation - # For example, if M5 uses H1 ATR, fetch H1 data once and reuse it - htf_atr_cache = {} # tf_label → DataFrame - atr_tf_needed = set() - for tf in active_tfs: - atr_src = get_atr_tf(tf, cfg) - if atr_src != tf: - atr_tf_needed.add(atr_src) - for atr_src in atr_tf_needed: - print(f"[ATR] Fetching {atr_src} data for higher-timeframe ATR...") - htf_df = fetch_rates_range(symbol, atr_src, date_from, date_to, cfg) - if htf_df is not None: - htf_atr_cache[atr_src] = htf_df - print(f" [ATR] {atr_src} data ready ({len(htf_df)} bars)") - else: - print(f" [ATR] WARNING: Could not fetch {atr_src} data — will fall back to native ATR") - - all_results = {} # tf_label → list of detection dicts - - for tf_label in active_tfs: - print(f"\n[TF] Fetching {tf_label} data...") - df = fetch_rates_range(symbol, tf_label, date_from, date_to, cfg) - if df is None: - print(f" [{tf_label}] No data — skipping.") - continue - - # Use D1 data as trend filter for all TFs below D1 - d1_for_filter = None if tf_label == 'D1' else d1_df_cache - - # Resolve the ATR source TF and its DataFrame - atr_src = get_atr_tf(tf_label, cfg) - htf_atr_df = htf_atr_cache.get(atr_src) if atr_src != tf_label else None - - print(f" [{tf_label}] Running pattern detection...") - detections = fb_detect_all_patterns(df, cfg, d1_for_filter, tf_label, htf_atr_df) - all_results[tf_label] = detections - print(f" [{tf_label}] {len(detections)} pattern detections") - - if not detections: - print(f" [{tf_label}] No patterns found.") - continue - - os.makedirs(out_dir, exist_ok=True) - tag = f"{symbol}_{tf_label}_{args.date_from}_to_{args.date_to}" - det_csv = os.path.join(out_dir, f"{tag}_detections.csv") - sum_csv = os.path.join(out_dir, f"{tag}_pattern_summary.csv") - sess_csv = os.path.join(out_dir, f"{tag}_session_summary.csv") - rpt_file = os.path.join(out_dir, f"{tag}_report.txt") - - det_df_out = pd.DataFrame(detections) - det_df_out.to_csv(det_csv, index=False, encoding='utf-8') - print(f" → {det_csv}") - - # Pattern summary CSV - directional = det_df_out[det_df_out['Direction'] != 'Neutral'] - max_r_levels = cfg.get('max_r_levels', 5) - srows = [] - for pat in det_df_out['Pattern'].unique(): - ps = det_df_out[det_df_out['Pattern'] == pat] - ds = ps[ps['Direction'] != 'Neutral'] - row_s = {'Timeframe': tf_label, 'Pattern': pat, 'Category': ps['Category'].iloc[0], - 'Direction': ps['Direction'].iloc[0], 'Total': len(ps)} - if len(ds) > 0: - s_ = int((ds['Prediction_Success'] == True).sum()) - f_ = int((ds['Prediction_Success'] == False).sum()) - wr_ = round(s_ / (s_ + f_) * 100, 1) if (s_ + f_) > 0 else 0 - row_s.update({ - 'Wins': s_, 'Losses': f_, 'Win_Rate_%': wr_, - 'SL_Hit_%': round((ds['SL_Hit'] == True).sum() / len(ds) * 100, 1), - 'TP_Hit_%': round((ds['TP_Hit'] == True).sum() / len(ds) * 100, 1), - 'Avg_SL_Pips': round(ds['SL_Pips'].dropna().mean(), 1) if not ds['SL_Pips'].dropna().empty else 0, - 'Avg_Max_R': round(ds['Max_R'].dropna().mean(), 2) if not ds['Max_R'].dropna().empty else 0, - }) - for r in range(1, max_r_levels+1): - col = f'R{r}_Hit' - if col in ds.columns: - hc = int((ds[col] == True).sum()); ec = int(ds[col].notna().sum()) - row_s[f'R{r}_Hit_%'] = round(hc / ec * 100, 1) if ec > 0 else 0 - else: - row_s[f'R{r}_Hit_%'] = 0 - else: - row_s.update({'Wins': 0, 'Losses': 0, 'Win_Rate_%': 0, - 'SL_Hit_%': 0, 'TP_Hit_%': 0, 'Avg_SL_Pips': 0, 'Avg_Max_R': 0}) - for r in range(1, max_r_levels+1): row_s[f'R{r}_Hit_%'] = 0 - srows.append(row_s) - pd.DataFrame(srows).to_csv(sum_csv, index=False, encoding='utf-8') - print(f" → {sum_csv}") - - # Session summary CSV - srows2 = [] - for sess in directional['Session'].unique() if len(directional) > 0 else []: - sd = directional[directional['Session'] == sess] - if len(sd) == 0: continue - s_ = int((sd['Prediction_Success'] == True).sum()) - f_ = int((sd['Prediction_Success'] == False).sum()) - wr_ = round(s_ / (s_ + f_) * 100, 1) if (s_ + f_) > 0 else 0 - row_s2 = {'Timeframe': tf_label, 'Session': sess, 'Signals': len(sd), - 'Wins': s_, 'Losses': f_, 'Win_Rate_%': wr_, - 'SL_Hit_%': round((sd['SL_Hit'] == True).sum() / len(sd) * 100, 1), - 'TP_Hit_%': round((sd['TP_Hit'] == True).sum() / len(sd) * 100, 1), - 'Avg_SL_Pips': round(sd['SL_Pips'].dropna().mean(), 1) if not sd['SL_Pips'].dropna().empty else 0, - 'Avg_Max_R': round(sd['Max_R'].dropna().mean(), 2) if not sd['Max_R'].dropna().empty else 0} - for r in range(1, max_r_levels+1): - col = f'R{r}_Hit' - if col in sd.columns: - hc = int((sd[col] == True).sum()); ec = int(sd[col].notna().sum()) - row_s2[f'R{r}_Hit_%'] = round(hc / ec * 100, 1) if ec > 0 else 0 - else: - row_s2[f'R{r}_Hit_%'] = 0 - srows2.append(row_s2) - pd.DataFrame(srows2).to_csv(sess_csv, index=False, encoding='utf-8') - print(f" → {sess_csv}") - - # Text report - report = fb_generate_report(detections, df, symbol, tf_label, cfg) - with open(rpt_file, 'w', encoding='utf-8') as fh: - fh.write(report) - print(f" → {rpt_file}") - - # Quick TF summary - print(f"\n [{tf_label}] Quick Summary:") - if len(directional) > 0: - s_ = int((directional['Prediction_Success'] == True).sum()) - f_ = int((directional['Prediction_Success'] == False).sum()) - wr_ = round(s_ / (s_ + f_) * 100, 1) if (s_ + f_) > 0 else 0 - print(f" Directional signals : {len(directional)}") - print(f" Win rate : {wr_}%") - print(f" Avg SL pips : {directional['SL_Pips'].dropna().mean():.1f}") - print(f" Avg Max R : {directional['Max_R'].dropna().mean():.2f}R") - - # Save combined stats JSON - try: - combined_stats = {'symbol': symbol, 'generated_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), - 'backtest_range': f"{args.date_from} to {args.date_to}", - 'timeframes': {}} - for tf_label, dets in all_results.items(): - if not dets: continue - df_tf = pd.DataFrame(dets) - dirdf = df_tf[df_tf['Direction'] != 'Neutral'] - tf_stats = {} - if len(dirdf) > 0: - s_ = int((dirdf['Prediction_Success'] == True).sum()) - f_ = int((dirdf['Prediction_Success'] == False).sum()) - tf_stats['overall'] = { - 'win_rate': round(s_ / (s_ + f_) * 100, 1) if (s_ + f_) > 0 else 0, - 'total_signals': len(dirdf), - 'avg_max_r': round(float(dirdf['Max_R'].dropna().mean()), 2), - 'sl_hit_pct': round(float((dirdf['SL_Hit'] == True).sum() / len(dirdf) * 100), 1), - 'tp_hit_pct': round(float((dirdf['TP_Hit'] == True).sum() / len(dirdf) * 100), 1), - } - combined_stats['timeframes'][tf_label] = tf_stats - stats_path = os.path.join(out_dir, 'latest_stats_multitf.json') - os.makedirs(out_dir, exist_ok=True) - with open(stats_path, 'w', encoding='utf-8') as sf: - json.dump(combined_stats, sf, indent=2, ensure_ascii=False) - print(f"\n Stats → {stats_path}") - except Exception as e: - print(f" [WARNING] Could not save stats: {e}") - - try: mt5.shutdown() - except Exception: pass - print("\nFull backtest complete.") - - -# ============================================================ -# ARGUMENT PARSER & MAIN -# ============================================================ - -def parse_args(cfg=None): - if cfg is None: cfg = CFG - p = argparse.ArgumentParser( - description="MT5 Multi-Timeframe Candlestick Pattern Scanner & Backtester v6", - formatter_class=argparse.RawTextHelpFormatter, - epilog=""" -Examples: - # Live scanner — all timeframes - python mt5_multitf_pattern_scanner_v6.py - - # Live scanner — specific timeframes - python mt5_multitf_pattern_scanner_v6.py --timeframes M5 H1 H4 - - # One-shot scan - python mt5_multitf_pattern_scanner_v6.py --mode scan - - # Quick backtest (500 bars) on H4 only - python mt5_multitf_pattern_scanner_v6.py --mode backtest --bars 500 --timeframes H4 - - # Full backtest on all TFs, 2024 full year - python mt5_multitf_pattern_scanner_v6.py --mode fullbacktest --from 2024-01-01 --to 2024-12-31 - - # Full backtest with filters on H4 only - python mt5_multitf_pattern_scanner_v6.py --mode fullbacktest --timeframes H4 \\ - --d1-trend-filter --volume-filter --forward 15 - """ - ) - - p.add_argument("--mode", choices=['live', 'scan', 'backtest', 'fullbacktest'], default='live', - help="Operating mode (default: live)") - p.add_argument("--timeframes", nargs='+', choices=list(TIMEFRAME_MAP.keys()), - default=cfg['active_timeframes'], - help="Active timeframes (default: M5 M15 H1 H4 D1)") - p.add_argument("--bars", type=int, default=500, - help="Bars for quick backtest (default: 500)") - - # Full backtest - p.add_argument("--from", dest="date_from", default="2024-01-01", - help="Full backtest start date YYYY-MM-DD") - p.add_argument("--to", dest="date_to", default="2024-12-31", - help="Full backtest end date YYYY-MM-DD") - - # Core parameters - p.add_argument("--symbol", default=cfg['symbol']) - p.add_argument("--atr", type=int, default=cfg['atr_period']) - p.add_argument("--atr-tf", type=str, default=None, - help="Override ATR source timeframe for ALL timeframes (e.g. H1, H4). " - "By default, atr_tf_by_tf config is used (M5→H1, M15→H1, etc.)") - p.add_argument("--sl", type=float, default=cfg['sl_multiplier']) - p.add_argument("--tp", type=float, default=cfg['tp_multiplier']) - p.add_argument("--forward",type=int, default=cfg['default_forward_candles'], - help="Default forward candles for evaluation (H4 default)") - p.add_argument("--output", default=DEFAULT_OUTPUT_DIR) - - # MT5 connection overrides — credentials always come from .env - p.add_argument("--mt5-path", default=cfg['mt5_path']) - p.add_argument("--account", type=int, default=cfg['account']) - p.add_argument("--password", default=cfg['password']) - p.add_argument("--server", default=cfg['server']) - - # Pattern thresholds - p.add_argument("--doji-body-ratio", type=float, default=cfg['doji_body_ratio']) - p.add_argument("--spinning-top-body-ratio", type=float, default=cfg['spinning_top_body_ratio']) - p.add_argument("--marubozu-wick-ratio", type=float, default=cfg['marubozu_wick_ratio']) - p.add_argument("--hammer-lower-wick-ratio", type=float, default=cfg['hammer_lower_wick_ratio']) - p.add_argument("--hammer-upper-wick-ratio", type=float, default=cfg['hammer_upper_wick_ratio']) - p.add_argument("--long-candle-ratio", type=float, default=cfg['long_candle_ratio']) - p.add_argument("--small-candle-ratio", type=float, default=cfg['small_candle_ratio']) - p.add_argument("--tweezer-tolerance", type=float, default=cfg['tweezer_tolerance_pips']) - p.add_argument("--engulf-tolerance-pips", type=float, default=cfg['engulf_tolerance_pips']) - p.add_argument("--trend-lookback", type=int, default=cfg['trend_lookback']) - p.add_argument("--broker-utc-offset", type=int, default=cfg['broker_utc_offset']) - - # Filters - p.add_argument("--deduplicate", dest="deduplicate_signals", action="store_true") - p.add_argument("--no-deduplicate", dest="deduplicate_signals", action="store_false") - p.add_argument("--verify-entry", dest="verify_entry", action="store_true") - p.add_argument("--no-verify-entry", dest="verify_entry", action="store_false") - p.add_argument("--volume-filter", dest="volume_filter", action="store_true") - p.add_argument("--no-volume-filter", dest="volume_filter", action="store_false") - p.add_argument("--volume-ma-period", type=int, default=cfg['volume_ma_period']) - p.add_argument("--volume-threshold", type=float, default=cfg['volume_threshold']) - p.add_argument("--d1-trend-filter", dest="d1_trend_filter", action="store_true") - p.add_argument("--no-d1-trend-filter", dest="d1_trend_filter", action="store_false") - p.add_argument("--d1-sma-period", type=int, default=cfg['d1_sma_period']) - - # Reconnect - p.add_argument("--max-reconnect-attempts", type=int, default=cfg['max_reconnect_attempts']) - p.add_argument("--reconnect-backoff", type=int, default=cfg['reconnect_backoff_base']) - - # Live signal filtering - p.add_argument("--min-signal-score", type=float, default=cfg['min_signal_score']) - p.add_argument("--alert-only-strong", dest="alert_only_strong", action="store_true") - p.add_argument("--no-alert-only-strong", dest="alert_only_strong", action="store_false") - p.add_argument("--show-dashboard", dest="show_dashboard_on_start", action="store_true") - p.add_argument("--no-dashboard", dest="show_dashboard_on_start", action="store_false") - - # Position sizing - p.add_argument("--account-balance", type=float, default=cfg['account_balance']) - p.add_argument("--risk-percent", type=float, default=cfg['risk_percent']) - - # Misc - p.add_argument("--warmup-bars", type=int, default=cfg['warmup_bars']) - p.add_argument("--bars-to-fetch", type=int, default=cfg['bars_to_fetch']) - - p.set_defaults( - deduplicate_signals=cfg['deduplicate_signals'], - verify_entry=cfg['verify_entry'], - volume_filter=cfg['volume_filter'], - d1_trend_filter=cfg['d1_trend_filter'], - alert_only_strong=cfg['alert_only_strong'], - show_dashboard_on_start=cfg['show_dashboard_on_start'], - ) - return p.parse_args() - - -def main(): - args = parse_args() - - # Build runtime CFG from defaults + CLI overrides - runtime_cfg = dict(CFG) - runtime_cfg['active_timeframes'] = args.timeframes - cli_to_cfg = { - 'symbol': 'symbol', 'atr': 'atr_period', 'sl': 'sl_multiplier', 'tp': 'tp_multiplier', - 'forward': 'default_forward_candles', 'mt5_path': 'mt5_path', 'account': 'account', - 'password': 'password', 'server': 'server', - 'doji_body_ratio': 'doji_body_ratio', 'spinning_top_body_ratio': 'spinning_top_body_ratio', - 'marubozu_wick_ratio': 'marubozu_wick_ratio', - 'hammer_lower_wick_ratio': 'hammer_lower_wick_ratio', - 'hammer_upper_wick_ratio': 'hammer_upper_wick_ratio', - 'long_candle_ratio': 'long_candle_ratio', 'small_candle_ratio': 'small_candle_ratio', - 'tweezer_tolerance': 'tweezer_tolerance_pips', - 'engulf_tolerance_pips': 'engulf_tolerance_pips', - 'trend_lookback': 'trend_lookback', 'broker_utc_offset': 'broker_utc_offset', - 'deduplicate_signals': 'deduplicate_signals', 'verify_entry': 'verify_entry', - 'volume_filter': 'volume_filter', 'volume_ma_period': 'volume_ma_period', - 'volume_threshold': 'volume_threshold', - 'd1_trend_filter': 'd1_trend_filter', 'd1_sma_period': 'd1_sma_period', - 'max_reconnect_attempts': 'max_reconnect_attempts', - 'reconnect_backoff': 'reconnect_backoff_base', - 'warmup_bars': 'warmup_bars', 'bars_to_fetch': 'bars_to_fetch', - 'min_signal_score': 'min_signal_score', - 'alert_only_strong': 'alert_only_strong', - 'show_dashboard_on_start': 'show_dashboard_on_start', - 'account_balance': 'account_balance', - 'risk_percent': 'risk_percent', - } - for cli_key, cfg_key in cli_to_cfg.items(): - if hasattr(args, cli_key): - runtime_cfg[cfg_key] = getattr(args, cli_key) - - # Handle --atr-tf override: if specified, override all ATR source TFs - if hasattr(args, 'atr_tf') and args.atr_tf is not None: - override_tf = args.atr_tf.upper() - if override_tf not in TIMEFRAME_MAP: - print(f"ERROR: --atr-tf '{override_tf}' not in {list(TIMEFRAME_MAP.keys())}") - sys.exit(1) - # Override the per-TF mapping to use the specified TF for all - runtime_cfg['atr_tf_by_tf'] = {tf: override_tf for tf in TIMEFRAME_MAP.keys()} - - if args.mode == 'live': - run_scanner(runtime_cfg) - elif args.mode == 'scan': - run_single_scan(runtime_cfg) - elif args.mode == 'backtest': - run_quick_backtest(args.bars, runtime_cfg) - elif args.mode == 'fullbacktest': - run_full_backtest(args, runtime_cfg) - - -if __name__ == '__main__': - main() \ No newline at end of file