gavindiaz/mt5-candlestick-scanner-backtester
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2eba3e574734e8394d450b4bb5273a8b9729eaeb
mt5-candlestick-scanner-bac…/mt5_multitf_pattern_scanner.py
T
Richard 2eba3e5747 Add files via upload 
removed - .txt file written every time a pattern is detected during live scanning. They're purely for record-keeping, the script doesn't read them back or need them to function. Safe to delete them all. They'll just keep accumulating as the scanner runs.
2026-05-07 14:19:25 +02:00

2951 lines
139 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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.
Reads ALL per-TF CSVs (pattern_summary, session_summary, detections) and:
- Builds per-TF pattern stats in stats['patterns_tf']
- Properly merges across all TFs for overall pattern/session/cross stats
- Carries per-TF overall stats from the v6 JSON
"""
tf_order = ['M5', 'M15', 'H1', 'H4', 'D1']
stats = {'patterns': {}, 'sessions': {}, 'overall': {}, 'cross': {},
'generated_at': v6_stats.get('generated_at', ''),
'patterns_tf': {tf: {} for tf in tf_order},
'timeframes': {}}
# Carry over per-TF overall stats from the v6 JSON
for tf_label, tf_data in v6_stats.get('timeframes', {}).items():
if 'overall' in tf_data:
stats['timeframes'][tf_label] = tf_data['overall']
# ── Parse ALL pattern_summary CSVs ──
pattern_csvs = sorted(
glob.glob(os.path.join(output_dir, f"{symbol}_*_*_to_*_pattern_summary.csv")),
reverse=True
)
if pattern_csvs:
try:
all_dfs = [pd.read_csv(p) for p in pattern_csvs]
df_all = pd.concat(all_dfs, ignore_index=True)
# Per-TF pattern stats
for tf in tf_order:
tf_rows = df_all[df_all.get('Timeframe', pd.Series(dtype=str)) == tf]
if len(tf_rows) > 0:
for _, row in tf_rows.iterrows():
pat = row['Pattern']
stats['patterns_tf'][tf][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),
}
# Merged across all TFs (weighted by signal count)
for pat in df_all['Pattern'].unique():
rows = df_all[df_all['Pattern'] == pat]
total_sig = int(rows['Total'].sum())
if total_sig > 0:
total_wins = 0
total_maxr_w = 0.0
total_sl_w = 0.0
total_tp_w = 0.0
for _, r in rows.iterrows():
n = int(r.get('Total', 0))
if n > 0:
total_wins += round(float(r.get('Win_Rate_%', 0)) * n / 100)
total_maxr_w += float(r.get('Avg_Max_R', 0)) * n
total_sl_w += float(r.get('SL_Hit_%', 0)) * n
total_tp_w += float(r.get('TP_Hit_%', 0)) * n
stats['patterns'][pat] = {
'win_rate': round(total_wins / total_sig * 100, 1),
'total': total_sig,
'avg_max_r': round(total_maxr_w / total_sig, 2),
'sl_hit_pct': round(total_sl_w / total_sig, 1),
'tp_hit_pct': round(total_tp_w / total_sig, 1),
}
except Exception:
pass
# ── Parse ALL session_summary CSVs ──
session_csvs = sorted(
glob.glob(os.path.join(output_dir, f"{symbol}_*_*_to_*_session_summary.csv")),
reverse=True
)
if session_csvs:
try:
all_dfs = [pd.read_csv(s) for s in session_csvs]
df_all = pd.concat(all_dfs, ignore_index=True)
for sess in df_all['Session'].unique():
rows = df_all[df_all['Session'] == sess]
total_sig = int(rows['Signals'].sum())
if total_sig > 0:
total_wins = 0
total_maxr_w = 0.0
total_sl_w = 0.0
total_tp_w = 0.0
for _, r in rows.iterrows():
n = int(r.get('Signals', 0))
if n > 0:
total_wins += round(float(r.get('Win_Rate_%', 0)) * n / 100)
total_maxr_w += float(r.get('Avg_Max_R', 0)) * n
total_sl_w += float(r.get('SL_Hit_%', 0)) * n
total_tp_w += float(r.get('TP_Hit_%', 0)) * n
stats['sessions'][sess] = {
'win_rate': round(total_wins / total_sig * 100, 1),
'signals': total_sig,
'avg_max_r': round(total_maxr_w / total_sig, 2),
'sl_hit_pct': round(total_sl_w / total_sig, 1),
'tp_hit_pct': round(total_tp_w / total_sig, 1),
}
except Exception:
pass
# ── Parse ALL detections CSVs for overall + cross stats ──
det_csvs = sorted(
glob.glob(os.path.join(output_dir, f"{symbol}_*_*_to_*_detections.csv")),
reverse=True
)
if det_csvs:
try:
all_dfs = [pd.read_csv(d) for d in det_csvs]
df_d = pd.concat(all_dfs, ignore_index=True)
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")
# Per-timeframe breakdown
tf_stats = stats.get('timeframes', {})
if tf_stats:
lines.append(f" {'Timeframe':<12s} | {'WR':>6s} | {'Signals':>8s} | {'Avg Max R':>10s}")
lines.append(f" {'-'*12} | {'-'*6} | {'-'*8} | {'-'*10}")
for tf_label, tf_overall in tf_stats.items():
twr = tf_overall.get('win_rate', 0)
tclr = 'green' if twr >= min_wr else ('yellow' if twr >= 45 else 'red')
lines.append(f" {tf_label:<12s} | {C(tclr, f'{twr:>5.1f}%')} | {tf_overall.get('total_signals',0):>8d} | {tf_overall.get('avg_max_r',0):>9.2f}R")
pat_list = []
patterns_tf = stats.get('patterns_tf', {})
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)
# Find best TF for this pattern
best_tf, best_tf_wr = '', 0
tf_wrs = {}
for tf in ['M5', 'M15', 'H1', 'H4', 'D1']:
if pat in patterns_tf.get(tf, {}):
tf_wr = patterns_tf[tf][pat].get('win_rate', 0)
tf_n = patterns_tf[tf][pat].get('total', 0)
tf_wrs[tf] = (tf_wr, tf_n) if tf_n >= min_sig else (None, tf_n)
if tf_n >= min_sig and tf_wr > best_tf_wr:
best_tf_wr = tf_wr
best_tf = tf
else:
tf_wrs[tf] = (None, 0)
pat_list.append((pat, wr, n, amr, weighted, tier_letter, tier_label, tier_clr, best_tf, tf_wrs))
pat_list.sort(key=lambda x: x[4], reverse=True)
tf_cols = ['M5', 'M15', 'H1', 'H4', 'D1']
if pat_list:
lines.append("")
lines.append(f" {'Pattern':<28s} | {'Tier':>14s} | {'M5':>5s} | {'M15':>5s} | {'H1':>5s} | {'H4':>5s} | {'D1':>5s} | {'Sig':>6s} | {'Edge':>5s}")
lines.append(f" {'-'*28} | {'-'*14} | {'-'*5} | {'-'*5} | {'-'*5} | {'-'*5} | {'-'*5} | {'-'*6} | {'-'*5}")
for pat, wr, n, amr, weighted, tl, tlab, tc, best_tf, tf_wrs in pat_list[:7]:
edge_tag = "HIGH" if wr >= min_wr else "LOW"
edge_color = 'green' if wr >= min_wr else 'red'
tf_cells = []
for tf in tf_cols:
tf_wr, tf_n = tf_wrs.get(tf, (None, 0))
if tf_wr is not None:
clr = 'green' if tf_wr >= min_wr else ('yellow' if tf_wr >= 45 else 'red')
tf_cells.append(C(clr, f'{tf_wr:>4.1f}%'))
else:
tf_cells.append(f" {'--' if tf_n < min_sig else '':>4s} ")
tf_str = ' | '.join(tf_cells)
lines.append(f" {pat:<28s} | {C(tc, f'{tl}:{tlab}'):>14s} | {tf_str} | {n:>6d} | {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)
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()
Reference in New Issue View Git Blame Copy Permalink