Files
mt5-candlestick-scanner-bac…/mt5_multitf_pattern_scanner.py
2026-05-30 08:22:47 +01:00

5056 lines
241 KiB
Python
Raw Permalink 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 v9
===============================================================
50+ candlestick patterns across M5, M15, H1, H4, D1 with TheStrat composites,
multi-timeframe backtesting, live scanner with signal scoring, tier-aware sound
alerts, auto-detected broker timezone, and local time display.
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
Sound Alerts (Windows only):
- Tier-aware: Tier A (Elite) always alerts, Tier B (Tradeable) alerts if score >= 60, Tier C/D never alerts
- High-Hz triple beep for BUY signals, Low-Hz triple beep for SELL signals
- Configurable via sound_alert_tier ('A', 'B', 'C') and sound_alert_tier_b_min_score
- Type 'm' + Enter at any time to mute/unmute sound alerts
Usage:
# Live scanner — all 5 timeframes (default)
python mt5_multitf_pattern_scanner.py
# Live scanner — specific timeframes only
python mt5_multitf_pattern_scanner.py --timeframes M5 H1 H4
# One-shot scan of latest closed candle on all timeframes
python mt5_multitf_pattern_scanner.py --mode scan
# Quick backtest (last 500 bars on H4)
python mt5_multitf_pattern_scanner.py --mode backtest --bars 500
# Full backtest on one timeframe
python mt5_multitf_pattern_scanner.py --mode fullbacktest --timeframes H4
# Full backtest on ALL timeframes, date-ranged
python mt5_multitf_pattern_scanner.py --mode fullbacktest --from 2024-01-01 --to 2024-12-31
# Full backtest with filters
python mt5_multitf_pattern_scanner.py --mode fullbacktest \\
--d1-trend-filter --volume-filter --forward 15 --sl 1.5 --tp 1.5
# Full backtest with structure-based SL (pattern invalidation levels)
python mt5_multitf_pattern_scanner.py --mode fullbacktest --sl-mode structure
# Full backtest with breakeven trade management
python mt5_multitf_pattern_scanner.py --mode fullbacktest --trade-management breakeven
# Full backtest with trailing stop management
python mt5_multitf_pattern_scanner.py --mode fullbacktest --trade-management trail
# Full backtest with partial close + trailing management
python mt5_multitf_pattern_scanner.py --mode fullbacktest --trade-management partial
# Full backtest with expired timeout (0R flat) instead of marginal win/loss
python mt5_multitf_pattern_scanner.py --mode fullbacktest --timeout-mode expired
# Multi-symbol watchlist scan
python mt5_multitf_pattern_scanner.py --mode scan --symbols EURUSD GBPUSD USDJPY
# Multi-symbol full backtest
python mt5_multitf_pattern_scanner.py --mode fullbacktest --symbols EURUSD GBPUSD
# Live scanner with custom account sizing
python mt5_multitf_pattern_scanner.py --mode live --account-balance 25000 --risk-percent 0.5
# Test sound alerts (plays both BUY and SELL test beeps)
python mt5_multitf_pattern_scanner.py --test-sound
"""
import MetaTrader5 as mt5
import pandas as pd
import numpy as np
from datetime import datetime, timedelta, timezone
import argparse
import time
import os
import sys
import json
import glob
import re
import warnings
warnings.filterwarnings('ignore', category=FutureWarning)
warnings.filterwarnings('ignore', category=DeprecationWarning)
warnings.filterwarnings('ignore', category=UserWarning)
# ── Windows sound & keyboard support ────────────────────────────────
try:
import winsound
_WINSOUND = True
except ImportError:
_WINSOUND = False
import threading
# Global mute state for sound alerts (toggled by 'm' key)
# Starts MUTED — type 'm' + Enter to unmute and hear alerts
_sound_muted = True
_sound_listener_thread = None
# ── 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', '0')) # Must be set in .env — 0 will fail login
_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
# Variable R:R overrides per pattern (tp_multiplier override).
# If a pattern is listed here, its TP multiplier is overridden.
# Example: Engulfing at 1.5:1, Morning Star at 2.5:1
'rr_by_pattern': {},
# Higher-timeframe ATR source per trading TF.
# H1 ATR used for all TFs — backtest proven to give better R-relative results.
# Native H4/D1 ATR produces stops too wide for price to reach 1R (0.26-0.36R avg).
# H1 ATR gives tighter stops: H4 avg SL 38.8p (was 60p), D1 avg SL 72.3p (was 158.8p).
# Set to None or same as trading TF to use native ATR.
'atr_tf_by_tf': {
'M5': 'H1', # H1 ATR for M5 signals (avoids tiny native M5 ATR)
'M15': 'H1', # H1 ATR for M15 signals
'H1': 'H1', # Native
'H4': 'H1', # H1 ATR — tighter stops, +1.5% win rate, +39% avg max R vs native H4
'D1': 'H1', # H1 ATR — -54% SL pips, +77% avg max R vs native D1
},
# ── 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,
# ── New Pattern Thresholds ──────────────────────────────────────
'belt_hold_wick_ratio': 0.1, # Max wick/body ratio for belt holds
'kicker_min_body_ratio': 0.6, # Min body ratio for kicker candles
'separating_lines_tolerance_pips': 2, # Tolerance for matching open prices
'meeting_lines_tolerance_pips': 2, # Tolerance for matching close prices
'gap_tolerance_pips': 2, # Minimum gap size in pips for gap patterns
# ── TheStrat Pattern Detection ─────────────────────────────────
'thestrat_enabled': True, # Enable TheStrat composite patterns
# ── 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': 20, # 20 × 1 day = 20 days (~4 trading weeks)
},
# ── 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, # Standard (winter) UTC offset for NY-close brokers (GMT+2)
# DST is handled automatically via broker_dst_rule — do NOT set this
# to 3 for summer; the code adds +1 during US daylight saving.
'broker_dst_rule': 'us', # DST rule: 'us' (2nd Sun Mar → 1st Sun Nov), 'eu', or 'none'
# ── 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
# ── Sound Alerts (Windows only) ─────────────────────────────────
'sound_enabled': True, # master switch for sound alerts
'sound_buy_hz': 1200, # Hz for strong buy triple beep
'sound_sell_hz': 400, # Hz for strong sell triple beep
'sound_beep_duration': 150, # ms per individual beep
'sound_beep_pause': 100, # ms pause between beeps
'sound_alert_tier': 'B', # Minimum tier to trigger sound: 'A' (elite only), 'B' (tradeable+), 'C' (all directional)
'sound_alert_tier_b_min_score': 60.0, # Tier B patterns must also have score >= this to alert
# ── Trade Management (Backtest) ────────────────────────────────
'trade_management_mode': 'fixed', # 'fixed', 'breakeven', 'trail', 'partial'
'breakeven_at_r': 1.0, # Move SL to breakeven when price hits this R level
'trail_at_r': 1.5, # Start trailing stop when price hits this R level
'trail_atr_mult': 1.0, # Trail SL by this multiple of ATR behind price
'partial_close_r': 1.0, # Close partial position at this R level
'partial_close_pct': 0.5, # Fraction of position to close at partial_close_r (0.5 = 50%)
'time_stop_pct': 0.7, # If this fraction of forward_candles elapsed without TP, tighten SL
# 0 = disabled. E.g. 0.7 with 15 forward = tighten after 10 bars
# ── SL Placement Mode ──────────────────────────────────────────
'sl_mode': 'atr', # 'atr' (current) or 'structure' (pattern-based)
'sl_structure_buffer_pips': 2, # Buffer in pips below pattern extreme for structure SL
# ── Timeout Classification ──────────────────────────────────────
'timeout_mode': 'marginal', # 'marginal' (current: Marginal_Win/Loss) or 'expired' (flat 0R)
# ── Multi-Symbol Watchlist ──────────────────────────────────────
'watchlist': ['EURUSD'], # Symbols to scan/backtest (default: EURUSD only)
# ── Equity Curve ────────────────────────────────────────────────
'equity_curve_enabled': True, # Generate equity curve in full backtest
}
# ── 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 = {
# ── Neutral / Low Priority (1) ──
'Doji': 1, 'Spinning Top': 1,
'Dragonfly Doji': 1, 'Gravestone Doji': 1,
# ── Single-Candle Directional (2) ──
'Hammer': 2, 'Inverted Hammer': 2,
'Shooting Star': 2, 'Hanging Man': 2,
'Bullish Belt Hold': 2, 'Bearish Belt Hold': 2,
# ── Strong Single-Candle (3) ──
'Marubozu (Bullish)': 3, 'Marubozu (Bearish)': 3,
# ── Two-Candle Reversal (4) ──
'Tweezer Tops': 4, 'Tweezer Bottoms': 4,
'Near Bullish Engulfing': 4, 'Near Bearish Engulfing': 4,
'Bullish Harami': 4, 'Bearish Harami': 4,
'Piercing Line': 4, 'Dark Cloud Cover': 4,
'Meeting Lines (Bullish)': 4, 'Meeting Lines (Bearish)': 4,
'Bullish Separating Lines': 4, 'Bearish Separating Lines': 4,
'Bearish Doji Star': 4,
# ── Strong Two-Candle (5) ──
'Bullish Engulfing': 5, 'Bearish Engulfing': 5,
'Bullish Kicker': 5, 'Bearish Kicker': 5,
# ── Three-Candle (6-7) ──
'Three Inside Up': 6, 'Three Inside Down': 6,
'Three Outside Up': 6, 'Three Outside Down': 6,
'Morning Star': 7, 'Evening Star': 7,
'Bullish Abandoned Baby': 7, 'Bearish Abandoned Baby': 7,
'Upside Gap Two Crows': 7,
# ── Strong Three-Candle (8) ──
'Three White Soldiers': 8, 'Three Black Crows': 8,
# ── Four-Candle (9) ──
'Bullish Three-Line Strike': 9, 'Bearish Three-Line Strike': 9,
'Concealing Baby Swallow': 9,
# ── Five-Candle (10) ──
'Rising Three Methods': 10, 'Falling Three Methods': 10,
'Mat Hold (Bullish)': 10, 'Mat Hold (Bearish)': 10,
'Ladder Bottom': 10,
# ── TheStrat Composite (11-12) ──
'TheStrat 2-2': 11, 'TheStrat 2-1-2': 11, 'TheStrat 3-1-2': 11,
'TheStrat 1-2-2 Rev': 12, 'TheStrat 1-3 Rev': 12,
}
# ============================================================
# UTILITY FUNCTIONS
# ============================================================
# ============================================================
# SOUND ALERT FUNCTIONS
# ============================================================
def play_signal_beep(direction, score, tier='D', cfg=None):
"""Play a triple beep sound alert for tier-qualified signals.
Tier-aware alert logic (replaces flat score threshold):
- Tier A (ELITE): Always alerts — highest confidence signals
- Tier B (TRADEABLE): Alerts if score >= sound_alert_tier_b_min_score (default 60)
- Tier C (MARGINAL): Never alerts — use only with strong confluence
- Tier D (AVOID): Never alerts — negative edge
High-Hz triple beep for BUY signals, Low-Hz triple beep for SELL signals.
Args:
direction: 'Bullish' or 'Bearish'
score: signal quality score (0-100)
tier: pattern tier letter 'A', 'B', 'C', or 'D'
cfg: configuration dict
"""
global _sound_muted
if cfg is None:
cfg = CFG
# Check master switch and mute state
if not cfg.get('sound_enabled', True) or _sound_muted:
return
if not _WINSOUND:
return
# Tier-based alert gate
min_tier = cfg.get('sound_alert_tier', 'B')
tier_b_min = cfg.get('sound_alert_tier_b_min_score', 60.0)
# Tier rank: A=1, B=2, C=3, D=4 — only alert if pattern's tier is at least as good as min_tier
tier_rank = {'A': 1, 'B': 2, 'C': 3, 'D': 4}
if tier_rank.get(tier, 4) > tier_rank.get(min_tier, 2):
return # Pattern tier is worse than minimum allowed
# Tier B requires minimum score (good confluence/session needed)
if tier == 'B' and (score is None or score < tier_b_min):
return
# Tier A always passes; Tier C/D already blocked above
if score is None:
return
# Determine frequency based on direction
if direction == 'Bullish':
freq = cfg.get('sound_buy_hz', 1200)
elif direction == 'Bearish':
freq = cfg.get('sound_sell_hz', 400)
else:
return # Neutral signals don't beep
duration = cfg.get('sound_beep_duration', 150)
pause = cfg.get('sound_beep_pause', 100)
# Play triple beep
try:
for i in range(3):
winsound.Beep(int(freq), int(duration))
if i < 2: # No pause after last beep
time.sleep(pause / 1000.0)
except Exception:
pass # Silently ignore sound errors
def _sound_key_listener():
"""Background daemon thread that listens for 'm' + Enter to toggle mute.
Uses input() which works in ALL terminals including PyCharm.
Runs as a daemon thread so it dies automatically when the main program exits.
"""
global _sound_muted
while True:
try:
cmd = input().strip().lower()
if cmd == 'm':
_sound_muted = not _sound_muted
status = C('red', 'MUTED') if _sound_muted else C('green', 'UNMUTED')
print(f"\n Sound {status} | Type 'm' + Enter to toggle")
except (EOFError, KeyboardInterrupt):
break
except Exception:
pass
def start_sound_key_listener():
"""Start the background keyboard listener thread (if not already running)."""
global _sound_listener_thread
if _sound_listener_thread is not None and _sound_listener_thread.is_alive():
return
_sound_listener_thread = threading.Thread(target=_sound_key_listener, daemon=True)
_sound_listener_thread.start()
# check_mute_key() removed — keyboard handled by background thread (start_sound_key_listener)
def test_sound(cfg=None):
"""Play test beeps for Tier A BUY and SELL signals so the user can verify audio.
Temporarily forces sound enabled and unmuted for the test, then restores
the original state. Exits after playing both test beeps.
"""
global _sound_muted
if cfg is None:
cfg = CFG
if not _WINSOUND:
print(C('red', "ERROR: winsound not available — sound requires Windows"))
return
# Save original mute state, force unmute for the test
was_muted = _sound_muted
_sound_muted = False
buy_hz = cfg.get('sound_buy_hz', 1200)
sell_hz = cfg.get('sound_sell_hz', 400)
duration = cfg.get('sound_beep_duration', 150)
pause = cfg.get('sound_beep_pause', 100)
alert_tier = cfg.get('sound_alert_tier', 'B')
print("")
print(C('cyan', "=" * 50))
print(C('bold', " SOUND TEST"))
print(C('cyan', "=" * 50))
print(f" Alert Tier: {alert_tier} (Tier A always, Tier B if score >= {cfg.get('sound_alert_tier_b_min_score', 60):.0f})")
print(f" BUY beep: {buy_hz} Hz x 3")
print(f" SELL beep: {sell_hz} Hz x 3")
print(f" Beep duration: {duration} ms")
print(f" Pause between: {pause} ms")
print("")
# Test Tier A BUY
print(C('green', " >>> Playing Tier A BUY test beep..."))
try:
for i in range(3):
winsound.Beep(int(buy_hz), int(duration))
if i < 2:
time.sleep(pause / 1000.0)
except Exception as e:
print(C('red', f" ERROR: {e}"))
time.sleep(0.5)
# Test Tier A SELL
print(C('red', " >>> Playing Tier A SELL test beep..."))
try:
for i in range(3):
winsound.Beep(int(sell_hz), int(duration))
if i < 2:
time.sleep(pause / 1000.0)
except Exception as e:
print(C('red', f" ERROR: {e}"))
# Restore original mute state
_sound_muted = was_muted
print("")
print(C('cyan', "=" * 50))
if was_muted:
print(f" Sound is {C('red', 'MUTED')} (restored to original state)")
print(f" Type {C('yellow', '\"m\" + Enter')} to unmute during live scanner")
else:
print(f" Sound is {C('green', 'UNMUTED')} (restored to original state)")
print(C('cyan', "=" * 50))
print("")
def local_now():
"""Return current local machine time for log timestamps.
Uses datetime.now() so log timestamps match the user's wall clock.
Candle display times are converted to local time separately via
to_local_time().
"""
return datetime.now()
# Backward-compatible alias (old name was misleading — returns local time, not broker time)
broker_now = local_now
def broker_time(ts):
"""Convert MT5 Unix timestamp to broker server clock time.
MT5 encodes broker server time directly in Unix timestamps as if it were
UTC. For example, if the broker is at UTC+2 and it's 12:00 broker time,
the Unix timestamp represents 12:00 UTC (not the true 10:00 UTC).
Decoding as UTC therefore returns the broker's clock time directly.
"""
return datetime.fromtimestamp(int(ts), tz=timezone.utc).replace(tzinfo=None)
def to_local_time(broker_dt, cfg=None):
"""Convert a broker-time datetime to local machine time for display.
Formula: local_time = broker_time - broker_utc_offset + local_utc_offset
The broker UTC offset is date-aware (accounts for US DST transitions).
The local UTC offset is computed from the system clock (handles local
DST automatically).
"""
if cfg is None:
cfg = CFG
# Date-aware broker offset (handles GMT+2/GMT+3 DST)
broker_offset = get_broker_offset_for_date(broker_dt, cfg)
# Modern replacement for deprecated datetime.utcnow()
local_offset = datetime.now().astimezone().utcoffset()
if local_offset is None:
local_offset = timedelta(0)
return broker_dt - timedelta(hours=broker_offset) + local_offset
def auto_detect_broker_offset(cfg=None):
"""Auto-detect broker UTC offset by comparing the latest MT5 candle time
with the current UTC time.
MT5 timestamps encode broker time as UTC, so the difference between
the decoded broker time and true UTC gives the broker's offset.
Returns the detected offset (integer hours), or the configured default
if detection fails.
"""
if cfg is None:
cfg = CFG
configured = cfg.get('broker_utc_offset', 2)
try:
symbol = cfg.get('symbol', 'EURUSD')
rates = mt5.copy_rates_from_pos(symbol, mt5.TIMEFRAME_M1, 0, 1)
if rates is not None and len(rates) > 0:
# Use timezone-aware UTC comparison (no deprecated datetime.utcnow())
broker_clock = datetime.fromtimestamp(int(rates[-1]['time']), tz=timezone.utc)
utc_now = datetime.now(timezone.utc)
diff_hours = (broker_clock - utc_now).total_seconds() / 3600
detected = round(diff_hours)
if abs(detected - diff_hours) < 0.5:
return detected
except Exception:
pass
return configured
_log_lock = threading.Lock()
def log_message(msg, cfg=None):
"""Print and log a message. Strips ANSI colour codes for log file.
Thread-safe: uses a lock to prevent interleaved writes from the
sound listener thread and the main scanner thread.
"""
timestamp = broker_now().strftime("%Y-%m-%d %H:%M:%S")
line = f"[{timestamp}] {msg}"
print(line)
clean_line = re.sub(r'\x1b\[[0-9;]*m', '', line)
with _log_lock:
try:
with open(LOG_FILE, "a", encoding='utf-8') as f:
f.write(clean_line + "\n")
except Exception:
pass
def get_broker_offset_for_date(dt, cfg=None):
"""Return the broker's UTC offset for a specific date, accounting for DST.
NY-close brokers follow US DST: GMT+2 (standard) / GMT+3 (daylight).
US DST: 2nd Sunday of March → 1st Sunday of November.
For brokers that don't follow US DST (e.g. Asian brokers at UTC+8),
set broker_dst_rule='none' in CFG.
Args:
dt: datetime (naive broker-time, or any date-aware datetime)
cfg: configuration dict
Returns:
Integer UTC offset (e.g. 2 or 3)
"""
if cfg is None:
cfg = CFG
base_offset = cfg.get('broker_utc_offset', 2)
dst_rule = cfg.get('broker_dst_rule', 'us')
if dst_rule != 'us' or base_offset != 2:
# No DST adjustment for non-NY-close brokers or non-standard offsets
return base_offset
# ── US DST calculation ──
date = dt.date() if isinstance(dt, datetime) else dt
year = date.year
# 2nd Sunday of March
mar1 = datetime(year, 3, 1)
dow_mar1 = mar1.weekday() # 0=Mon .. 6=Sun
days_to_first_sun = (6 - dow_mar1) % 7
second_sunday_mar = mar1 + timedelta(days=days_to_first_sun + 7)
spring_date = second_sunday_mar.date()
# 1st Sunday of November
nov1 = datetime(year, 11, 1)
dow_nov1 = nov1.weekday()
days_to_first_sun_nov = (6 - dow_nov1) % 7
first_sunday_nov = nov1 + timedelta(days=days_to_first_sun_nov)
fall_date = first_sunday_nov.date()
if spring_date <= date < fall_date:
return base_offset + 1 # Daylight saving: GMT+3
return base_offset # Standard: GMT+2
def classify_session(broker_hour, broker_dt=None, cfg=None):
"""Classify broker-time hour into a trading session.
Standard forex session boundaries (UTC):
Pacific: 21:00 00:00 Sydney open
Asia: 00:00 07:00 Tokyo active
London Open: 07:00 09:00 London open + Tokyo/London overlap
London Morning: 09:00 13:00 London active
London/NY Overlap: 13:00 16:00 Highest volume window
NY Afternoon: 16:00 21:00 NY active, London closed
If broker_dt is provided, uses date-aware DST offset for accurate
session classification across DST transitions (GMT+2/GMT+3).
Otherwise falls back to the configured broker_utc_offset.
"""
if cfg is None:
cfg = CFG
# Date-aware offset: handles GMT+2 winter / GMT+3 summer
if broker_dt is not None:
offset = get_broker_offset_for_date(broker_dt, cfg)
else:
offset = cfg.get('broker_utc_offset', 2)
utc_hour = (broker_hour - offset) % 24
# Classify by UTC hour — matches standard forex session times
if utc_hour >= 21: # 21:00 23:59 Sydney open
return 'Pacific'
elif utc_hour < 7: # 00:00 07:00 Tokyo active
return 'Asia'
elif utc_hour < 9: # 07:00 09:00 Tokyo/London overlap
return 'London Open'
elif utc_hour < 13: # 09:00 13:00 London active
return 'London Morning'
elif utc_hour < 16: # 13:00 16:00 London/NY overlap
return 'London/NY Overlap'
else: # 16:00 21:00 NY active
return 'NY Afternoon'
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
def get_pip_value(symbol=None, cfg=None):
"""Compute pip value for position sizing based on MT5 symbol info.
Falls back to a static lookup table if MT5 is not connected.
Returns the value of 1 pip movement per standard lot in account currency.
For most forex pairs: pip_value = contract_size * pip_size / (current_rate for cross pairs)
For EURUSD standard: 100000 * 0.0001 = 10 USD per pip per lot
"""
if cfg is None: cfg = CFG
if symbol is None: symbol = cfg.get('symbol', 'EURUSD')
# Try MT5 symbol_info (only works when connected)
try:
info = mt5.symbol_info(symbol)
if info is not None:
contract_size = info.trade_contract_size or 100000
tick_size = info.trade_tick_size or 0.00001
tick_value = info.trade_tick_value or 0
if tick_size > 0 and tick_value > 0:
# pip_value = value of 1 pip (0.0001 for 5-digit, 0.01 for 3-digit)
pip_size = 0.0001 if 'JPY' not in symbol else 0.01
return round(tick_value * (pip_size / tick_size), 4)
except Exception:
pass
# Fallback static lookup for common symbols
_STATIC_PIP_VALUES = {
'EURUSD': 10, 'GBPUSD': 10, 'AUDUSD': 10, 'NZDUSD': 10, 'USDCAD': 7.5,
'USDCHF': 11, 'USDJPY': 6.5, 'EURJPY': 6.5, 'GBPJPY': 6.5,
'XAUUSD': 1, 'XAGUSD': 5, 'US30': 1, 'NAS100': 1, 'SPX500': 1,
}
return _STATIC_PIP_VALUES.get(symbol, 10)
def compute_structure_sl(pattern_name, direction, rates_or_df, idx, cfg=None):
"""Compute structure-based SL using the pattern's natural invalidation level.
Structure SL places the stop at the pattern's extreme (e.g. below the Hammer's
low, below the engulfing candle's low) plus a small buffer, rather than using
a fixed ATR multiple. This gives tighter, more logical stops.
Returns (sl_price, sl_reason) or None if not applicable.
Args:
pattern_name: e.g. 'Hammer', 'Bullish Engulfing', 'Morning Star'
direction: 'Bullish' or 'Bearish'
rates_or_df: structured array (scanner) or DataFrame (backtest)
idx: index of the signal candle
cfg: configuration dict
"""
if cfg is None: cfg = CFG
buffer_pips = cfg.get('sl_structure_buffer_pips', 2)
pip_divisor = cfg.get('pip_divisor', 0.0001)
buffer = buffer_pips * pip_divisor
# For DataFrame, use uppercase column names; for structured arrays, use lowercase
is_df = isinstance(rates_or_df, pd.DataFrame)
low_key = 'LOW' if is_df else 'low'
high_key = 'HIGH' if is_df else 'high'
close_key = 'CLOSE' if is_df else 'close'
open_key = 'OPEN' if is_df else 'open'
try:
if is_df:
curr = rates_or_df.iloc[idx]
else:
curr = rates_or_df[idx]
except (IndexError, KeyError):
return None
curr_low = curr[low_key]
curr_high = curr[high_key]
if direction == 'Bullish':
# For bullish patterns, SL goes below the pattern's lowest point
if pattern_name in ('Hammer', 'Inverted Hammer'):
# Below the signal candle's low (the wick IS the pattern)
sl = curr_low - buffer
reason = f'Below Hammer low ({curr_low:.5f}) - buffer {buffer_pips}p'
elif pattern_name in ('Morning Star',):
# Below the lowest point of the 3-candle pattern
if idx >= 2:
if is_df:
prev2 = rates_or_df.iloc[idx-2]
prev1 = rates_or_df.iloc[idx-1]
else:
prev2 = rates_or_df[idx-2]
prev1 = rates_or_df[idx-1]
pattern_low = min(prev2[low_key], prev1[low_key], curr_low)
sl = pattern_low - buffer
reason = f'Below Morning Star low ({pattern_low:.5f}) - buffer {buffer_pips}p'
else:
sl = curr_low - buffer
reason = f'Below candle low ({curr_low:.5f}) - buffer {buffer_pips}p'
elif pattern_name in ('Three White Soldiers', 'Rising Three Methods'):
# Below the first candle's low of the multi-candle pattern
lookback = 4 if 'Three Methods' in pattern_name else 2
if idx >= lookback:
if is_df:
first = rates_or_df.iloc[idx-lookback]
else:
first = rates_or_df[idx-lookback]
sl = first[low_key] - buffer
reason = f'Below pattern first candle low ({first[low_key]:.5f}) - buffer {buffer_pips}p'
else:
sl = curr_low - buffer
reason = f'Below candle low ({curr_low:.5f}) - buffer {buffer_pips}p'
elif 'Bullish Engulfing' in pattern_name:
# Below the engulfing candle's low (current candle = engulfing)
sl = curr_low - buffer
reason = f'Below Engulfing candle low ({curr_low:.5f}) - buffer {buffer_pips}p'
elif 'Bullish Harami' in pattern_name:
# Below the mother candle's low (previous candle)
if idx >= 1:
if is_df:
prev = rates_or_df.iloc[idx-1]
else:
prev = rates_or_df[idx-1]
sl = min(prev[low_key], curr_low) - buffer
reason = f'Below Harami pattern low ({min(prev[low_key], curr_low):.5f}) - buffer {buffer_pips}p'
else:
sl = curr_low - buffer
reason = f'Below candle low - buffer {buffer_pips}p'
elif pattern_name == 'Tweezer Bottoms':
# Below the tweezer lows
sl = curr_low - buffer
reason = f'Below Tweezer Bottoms low ({curr_low:.5f}) - buffer {buffer_pips}p'
else:
# Default: below current candle low
sl = curr_low - buffer
reason = f'Below candle low ({curr_low:.5f}) - buffer {buffer_pips}p'
return (round(sl, 5), reason)
elif direction == 'Bearish':
# For bearish patterns, SL goes above the pattern's highest point
if pattern_name in ('Shooting Star', 'Hanging Man'):
sl = curr_high + buffer
reason = f'Above Shooting Star high ({curr_high:.5f}) + buffer {buffer_pips}p'
elif pattern_name in ('Evening Star',):
if idx >= 2:
if is_df:
prev2 = rates_or_df.iloc[idx-2]
prev1 = rates_or_df.iloc[idx-1]
else:
prev2 = rates_or_df[idx-2]
prev1 = rates_or_df[idx-1]
pattern_high = max(prev2[high_key], prev1[high_key], curr_high)
sl = pattern_high + buffer
reason = f'Above Evening Star high ({pattern_high:.5f}) + buffer {buffer_pips}p'
else:
sl = curr_high + buffer
reason = f'Above candle high + buffer {buffer_pips}p'
elif pattern_name in ('Three Black Crows', 'Falling Three Methods'):
lookback = 4 if 'Three Methods' in pattern_name else 2
if idx >= lookback:
if is_df:
first = rates_or_df.iloc[idx-lookback]
else:
first = rates_or_df[idx-lookback]
sl = first[high_key] + buffer
reason = f'Above pattern first candle high ({first[high_key]:.5f}) + buffer {buffer_pips}p'
else:
sl = curr_high + buffer
reason = f'Above candle high + buffer {buffer_pips}p'
elif 'Bearish Engulfing' in pattern_name:
sl = curr_high + buffer
reason = f'Above Engulfing candle high ({curr_high:.5f}) + buffer {buffer_pips}p'
elif 'Bearish Harami' in pattern_name:
if idx >= 1:
if is_df:
prev = rates_or_df.iloc[idx-1]
else:
prev = rates_or_df[idx-1]
sl = max(prev[high_key], curr_high) + buffer
reason = f'Above Harami pattern high ({max(prev[high_key], curr_high):.5f}) + buffer {buffer_pips}p'
else:
sl = curr_high + buffer
reason = f'Above candle high + buffer {buffer_pips}p'
elif pattern_name == 'Tweezer Tops':
sl = curr_high + buffer
reason = f'Above Tweezer Tops high ({curr_high:.5f}) + buffer {buffer_pips}p'
else:
sl = curr_high + buffer
reason = f'Above candle high ({curr_high:.5f}) + buffer {buffer_pips}p'
return (round(sl, 5), reason)
return None
# ============================================================
# PATTERN DETECTION — Unified (scanner uses fb_detect_* via DataFrame adapter)
# ============================================================
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 using Wilder's exponential smoothing (standard ATR method).
Uses alpha = 1/period for EMA smoothing, matching MT5's built-in ATR.
This is the industry-standard method and differs from simple moving average
by 5-15% on typical data.
"""
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)
# Wilder's smoothing: seed with SMA of first 'period' values, then EMA
atr_val = np.mean(tr_values[:period])
alpha = 1.0 / period
for i in range(period, len(tr_values)):
atr_val = alpha * tr_values[i] + (1.0 - alpha) * atr_val
return atr_val
def mt5_rates_to_df(rates, cfg=None):
"""Convert MT5 structured array to a minimal DataFrame for unified pattern detection.
This adapter allows the scanner to use the same fb_detect_* functions as the
backtest, eliminating ~400 lines of code duplication. The DataFrame has the
same column names (UPPERCASE) as the backtest DataFrame.
Also pre-computes BODY, BODY_SIGN, RANGE, UPPER_WICK, LOWER_WICK, BODY_RATIO
columns so the fb_detect_* functions work without modification.
Args:
rates: MT5 structured array (from mt5.copy_rates_from_pos)
cfg: configuration dict
Returns:
pd.DataFrame with UPPER-CASE column names matching backtest format
"""
if rates is None or len(rates) == 0:
return pd.DataFrame()
# Build DataFrame from structured array (or list of numpy.void)
# Detect available field names (numpy.void supports [] but not .get())
first = rates[0]
field_names = first.dtype.names if hasattr(first, 'dtype') and hasattr(first.dtype, 'names') else None
rows = []
for r in rates:
row = {
'time': r['time'],
'OPEN': r['open'],
'HIGH': r['high'],
'LOW': r['low'],
'CLOSE': r['close'],
'TICKVOL': r['tick_volume'],
}
if field_names and 'real_volume' in field_names:
row['VOL'] = r['real_volume']
else:
row['VOL'] = 0
if field_names and 'spread' in field_names:
row['SPREAD'] = r['spread']
else:
row['SPREAD'] = 0
rows.append(row)
df = pd.DataFrame(rows)
df['DATETIME'] = pd.to_datetime(df['time'], unit='s')
df['DATE'] = df['DATETIME'].dt.strftime('%Y.%m.%d')
df['TIME'] = df['DATETIME'].dt.strftime('%H:%M:%S')
df['IN_RANGE'] = True # All scanner bars are "in range"
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)
return df
# ── Scanner-specific detect_* functions removed (v8, unified in v9) ────────────
# Pattern detection is now unified: scan_patterns() converts MT5
# structured arrays to DataFrames and uses the fb_detect_* functions
# that are also used by the backtest. This eliminates ~400 lines of
# code duplication and ensures scanner/backtest always use identical
# detection logic.
#
# Removed functions:
# get_candle_metrics, detect_doji, detect_spinning_top, detect_marubozu,
# detect_hammer, detect_inverted_hammer, detect_shooting_star,
# detect_hanging_man, detect_near_engulfing_full, detect_engulfing_full,
# detect_harami_full, detect_morning_star, detect_evening_star,
# detect_three_white_soldiers, detect_three_black_crows, detect_tweezer,
# detect_rising_three_methods, detect_falling_three_methods,
# check_volume_confirmed
# ============================================================
# 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']
# Carry over per-TF pattern stats from enriched JSON (v7+)
if 'patterns' in tf_data:
for pat, pat_data in tf_data['patterns'].items():
stats['patterns_tf'][tf_label][pat] = pat_data
# ── Build merged stats from enriched JSON if available (v7+), else fall back to CSVs ──
# Check if the JSON has per-TF pattern/session/cross data (v7 enrichment)
has_enriched_data = any('patterns' in tf_data for tf_data in v6_stats.get('timeframes', {}).values())
if has_enriched_data:
# Build merged stats directly from JSON — no CSV parsing needed
# Merged patterns (weighted average across TFs)
all_pat_data = {} # pat -> list of (n, wr, amr, sl_pct, tp_pct)
for tf_label, tf_data in v6_stats.get('timeframes', {}).items():
for pat, pat_data in tf_data.get('patterns', {}).items():
if pat not in all_pat_data:
all_pat_data[pat] = []
all_pat_data[pat].append(pat_data)
for pat, entries in all_pat_data.items():
total_sig = sum(e.get('total', 0) for e in entries)
if total_sig > 0:
total_wins = sum(round(e.get('win_rate', 0) * e.get('total', 0) / 100) for e in entries)
total_maxr_w = sum(e.get('avg_max_r', 0) * e.get('total', 0) for e in entries)
total_sl_w = sum(e.get('sl_hit_pct', 0) * e.get('total', 0) for e in entries)
total_tp_w = sum(e.get('tp_hit_pct', 0) * e.get('total', 0) for e in entries)
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),
}
# Merged sessions (weighted average across TFs)
all_sess_data = {}
for tf_label, tf_data in v6_stats.get('timeframes', {}).items():
for sess, sess_data in tf_data.get('sessions', {}).items():
if sess not in all_sess_data:
all_sess_data[sess] = []
all_sess_data[sess].append(sess_data)
for sess, entries in all_sess_data.items():
total_sig = sum(e.get('signals', 0) for e in entries)
if total_sig > 0:
total_wins = sum(round(e.get('win_rate', 0) * e.get('signals', 0) / 100) for e in entries)
total_maxr_w = sum(e.get('avg_max_r', 0) * e.get('signals', 0) for e in entries)
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),
}
# Merged cross stats (weighted average across TFs)
all_cross_data = {}
for tf_label, tf_data in v6_stats.get('timeframes', {}).items():
for cross_key, cross_data in tf_data.get('cross', {}).items():
if cross_key not in all_cross_data:
all_cross_data[cross_key] = []
all_cross_data[cross_key].append(cross_data)
for cross_key, entries in all_cross_data.items():
total_sig = sum(e.get('signals', 0) for e in entries)
if total_sig >= 3:
total_wins = sum(round(e.get('win_rate', 0) * e.get('signals', 0) / 100) for e in entries)
total_maxr_w = sum(e.get('avg_max_r', 0) * e.get('signals', 0) for e in entries)
stats['cross'][cross_key] = {
'win_rate': round(total_wins / total_sig * 100, 1),
'signals': total_sig,
'avg_max_r': round(total_maxr_w / total_sig, 2),
}
# Overall (aggregate across all TFs)
total_all = sum(stats['timeframes'].get(tf, {}).get('total_signals', 0) for tf in tf_order)
if total_all > 0:
total_wins = sum(round(stats['timeframes'].get(tf, {}).get('win_rate', 0) *
stats['timeframes'].get(tf, {}).get('total_signals', 0) / 100)
for tf in tf_order)
total_maxr = sum(stats['timeframes'].get(tf, {}).get('avg_max_r', 0) *
stats['timeframes'].get(tf, {}).get('total_signals', 0)
for tf in tf_order)
stats['overall'] = {
'win_rate': round(total_wins / total_all * 100, 1),
'total_signals': total_all,
'avg_max_r': round(total_maxr / total_all, 2),
}
else:
# Fall back to CSV parsing for older JSON formats (v6 without enrichment)
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, tf_label=None):
"""Compute a 0-100 signal quality score based on historical backtest stats.
Enhanced formula (v7):
base = raw win rate (0-100)
sample_penalty = penalty if sample size is small (avoids over-scoring rare patterns)
session_gradient = proportional session bonus/penalty (not binary)
confluence_bonus = +5 if confluence >= 3 (from backtest confluence stats)
tier_bonus = +5 for Tier A, +3 for Tier B
Prefers TF-specific stats when available (tf_label provided and stats have
per-TF breakdown), falling back to merged aggregate stats.
"""
if cfg is None: cfg = CFG
min_signals = cfg.get('min_signals_for_stats', 5)
# Try TF-specific stats first if tf_label is provided
pat_stats = {}
sess_stats = {}
cross_key = f"{pattern_name}|{session}"
cross_stats = {}
if tf_label and tf_label in stats.get('timeframes', {}):
tf_data = stats['timeframes'][tf_label]
pat_stats = tf_data.get('patterns', {}).get(pattern_name, {})
sess_stats = tf_data.get('sessions', {}).get(session, {})
cross_stats = tf_data.get('cross', {}).get(cross_key, {})
# Fallback to merged stats if TF-specific not available
if not pat_stats:
pat_stats = stats.get('patterns', {}).get(pattern_name, {})
if not sess_stats:
sess_stats = stats.get('sessions', {}).get(session, {})
if not cross_stats:
cross_stats = stats.get('cross', {}).get(cross_key, {})
# Select most specific data source with sufficient sample
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 = raw win rate
base_score = wr
# Sample penalty: penalize low sample sizes instead of multiplying
# At 30+ signals, no penalty. Below 30, linear penalty down to -15 at min_signals.
if n >= 30:
sample_penalty = 0
elif n >= min_signals:
sample_penalty = -15.0 * (1.0 - (n - min_signals) / (30.0 - min_signals))
else:
sample_penalty = -20.0
# Session gradient: proportional bonus/penalty based on session WR
session_gradient = 0
if sess_stats and sess_stats.get('signals', 0) >= min_signals:
sess_wr = sess_stats.get('win_rate', 50)
# Gradient: +8 at 60% WR, +4 at 55%, 0 at 50%, -4 at 45%, -8 at 40%
session_gradient = round((sess_wr - 50) * 0.8, 1)
# Confluence bonus: +5 if high-confluence signals (upper half of range) perform well
# With d1_trend_filter active, effective range is 0-6, so high = scores 3,4,5
# Without the filter, effective range is 0-7, so high = scores 4,5,6
confluence_bonus = 0
if tf_label and tf_label in stats.get('timeframes', {}):
conf_data = stats['timeframes'][tf_label].get('confluence', {})
# Determine threshold based on d1_trend_filter
if cfg.get('d1_trend_filter', False):
high_keys = ['3', '4', '5'] # upper half of 0-6 range
else:
high_keys = ['4', '5', '6'] # upper half of 0-7 range
high_conf = None
for k in high_keys:
entry = conf_data.get(k, {})
if entry.get('signals', 0) >= 5:
high_conf = entry
break
if high_conf and high_conf.get('win_rate', 0) >= 55:
confluence_bonus = 5
# Tier bonus
tier_letter, _, _ = compute_pattern_tier(pattern_name, stats, cfg)
tier_bonus = 5 if tier_letter == 'A' else (3 if tier_letter == 'B' else 0)
# MFE bonus: patterns with higher MFE have more profit potential
mfe_bonus = 0
if amr >= 0.8:
mfe_bonus = 3
elif amr >= 0.5:
mfe_bonus = 1
score = base_score + sample_penalty + session_gradient + confluence_bonus + tier_bonus + mfe_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, tf_label=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, tf_label=tf_label)
p['signal_score'] = score
return patterns
filtered = []
for p in patterns:
score = compute_signal_score(p['name'], p['session'], p['direction'], stats, cfg, tf_label=tf_label)
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
# ── Use DataFrame-based detection (unified with backtest) ────────
scan_df = mt5_rates_to_df(rates, cfg)
if len(scan_df) < 6:
return []
idx = len(scan_df) - 1
row = scan_df.iloc[idx]
cm = {
'body': row['BODY'], 'body_sign': row['BODY_SIGN'], 'range': row['RANGE'],
'upper_wick': row['UPPER_WICK'], 'lower_wick': row['LOWER_WICK'],
'body_ratio': row['BODY_RATIO']
}
curr = {'open': row['OPEN'], 'high': row['HIGH'], 'low': row['LOW'], 'close': row['CLOSE'], 'time': row['time']}
patterns = []
trend = detect_trend(scan_df[: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)):
bt = broker_time(int(_ct))
hour = bt.hour
elif hasattr(_ct, 'hour'):
bt = _ct
hour = _ct.hour
else:
bt = None
hour = int(_ct) % 24
session = classify_session(hour, bt, cfg)
# Volume confirmation (DataFrame-based, matching backtest logic)
vol_confirmed = True
if cfg.get('volume_filter', False) and len(scan_df) > 0:
vol_ma_period = cfg.get('volume_ma_period', 20)
vol_thresh = cfg.get('volume_threshold', 1.0)
start = max(0, idx - vol_ma_period)
vols = scan_df.iloc[start:idx+1]['TICKVOL'].values
if len(vols) >= 2:
avg_vol = np.mean(vols[:-1])
if avg_vol > 0:
vol_confirmed = vols[-1] >= vol_thresh * avg_vol
# 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 (using unified fb_detect_* functions)
if fb_detect_doji(scan_df, idx, cfg): patterns.append({'name': 'Doji', 'category': 'Neutral', 'direction': 'Neutral'})
if fb_detect_spinning_top(scan_df, idx, cfg): patterns.append({'name': 'Spinning Top', 'category': 'Neutral', 'direction': 'Neutral'})
if fb_detect_marubozu(scan_df, idx, cfg):
d = 'Bullish' if cm['body_sign'] == 1 else 'Bearish'
patterns.append({'name': f'Marubozu ({d})', 'category': f'{d} Continuation', 'direction': d})
if fb_detect_hammer(scan_df, idx, cfg): patterns.append({'name': 'Hammer', 'category': 'Bullish Reversal', 'direction': 'Bullish'})
if fb_detect_inverted_hammer(scan_df, idx, cfg): patterns.append({'name': 'Inverted Hammer', 'category': 'Bullish Reversal', 'direction': 'Bullish'})
if fb_detect_shooting_star(scan_df, idx, cfg): patterns.append({'name': 'Shooting Star', 'category': 'Bearish Reversal', 'direction': 'Bearish'})
if fb_detect_hanging_man(scan_df, idx, cfg): patterns.append({'name': 'Hanging Man', 'category': 'Bearish Reversal', 'direction': 'Bearish'})
# Two-candle patterns
eng = fb_detect_engulfing(scan_df, idx, cfg)
if eng:
d = 'Bullish' if 'Bullish' in eng else 'Bearish'
patterns.append({'name': eng, 'category': f'{d} Reversal', 'direction': d})
ne = fb_detect_near_engulfing(scan_df, idx, cfg)
if ne:
d = 'Bullish' if 'Bullish' in ne else 'Bearish'
patterns.append({'name': ne, 'category': f'{d} Reversal', 'direction': d})
har = fb_detect_harami(scan_df, idx, cfg)
if har:
d = 'Bullish' if 'Bullish' in har else 'Bearish'
patterns.append({'name': har, 'category': f'{d} Reversal', 'direction': d})
tw = fb_detect_tweezer(scan_df, 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 fb_detect_morning_star(scan_df, idx, cfg): patterns.append({'name': 'Morning Star', 'category': 'Bullish Reversal', 'direction': 'Bullish'})
if fb_detect_evening_star(scan_df, idx, cfg): patterns.append({'name': 'Evening Star', 'category': 'Bearish Reversal', 'direction': 'Bearish'})
if fb_detect_three_white_soldiers(scan_df, idx, cfg): patterns.append({'name': 'Three White Soldiers', 'category': 'Bullish Reversal', 'direction': 'Bullish'})
if fb_detect_three_black_crows(scan_df, idx, cfg): patterns.append({'name': 'Three Black Crows', 'category': 'Bearish Reversal', 'direction': 'Bearish'})
# Five-candle patterns
if fb_detect_rising_three_methods(scan_df, idx, cfg): patterns.append({'name': 'Rising Three Methods', 'category': 'Bullish Continuation', 'direction': 'Bullish'})
if fb_detect_falling_three_methods(scan_df, idx, 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']
sl_mode = cfg.get('sl_mode', 'atr')
for pat in patterns:
d = pat['direction']
sl_reason = ''
# Structure-based SL: use pattern's natural invalidation level
struct_sl = None
if sl_mode == 'structure' and d in ('Bullish', 'Bearish'):
struct_result = compute_structure_sl(pat['name'], d, rates, idx, cfg)
if struct_result is not None:
struct_sl, sl_reason = struct_result
if d == 'Bullish':
if struct_sl is not None:
pat['sl'] = struct_sl
else:
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)
if sl_reason:
pat['sl_reason'] = sl_reason
elif d == 'Bearish':
if struct_sl is not None:
pat['sl'] = struct_sl
else:
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)
if sl_reason:
pat['sl_reason'] = sl_reason
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 = broker_time(int(ct))
# MT5's 'time' field is the candle's OPEN time in broker time.
# 1. Add candle duration to get the CLOSE time
# 2. Convert broker time to local time for display
tf_minutes = TIMEFRAME_MAP.get(tf_label, {}).get('minutes', 0)
if tf_minutes and hasattr(ct, '__add__'):
close_ct_broker = ct + timedelta(minutes=tf_minutes)
else:
close_ct_broker = ct
close_ct = to_local_time(close_ct_broker, cfg)
time_str = close_ct.strftime("%Y-%m-%d %H:%M:%S") if hasattr(close_ct, 'strftime') else str(close_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, tf_label=tf_label)
if score is not None:
if tier_letter == 'A':
score_label = "ELITE"; score_color = 'green'
elif tier_letter == 'B':
score_label = "TRADEABLE"; score_color = 'yellow'
elif score >= 65:
score_label = "STRONG"; score_color = 'green'
elif score >= 52:
score_label = "MODERATE"; score_color = 'yellow'
else:
score_label = "WEAK"; score_color = 'red'
alert_tag = " [ALERT]" if (tier_letter in ('A', 'B')) else ""
lines.append(f" Signal Score: {C(score_color, C('bold', f'{score:.1f}/100 [{score_label}{alert_tag}]'))}")
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 = get_pip_value(cfg.get('symbol', 'EURUSD'), cfg)
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')} (pip val: {pip_value})")
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):
"""Compute ATR using Wilder's exponential smoothing (standard ATR method).
Returns a pd.Series aligned with df, using alpha = 1/period for EMA smoothing.
This matches MT5's built-in ATR indicator and the industry standard.
"""
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)
# Wilder's smoothing: EMA with alpha = 1/period
alpha = 1.0 / period
return tr.ewm(alpha=alpha, adjust=False).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
# ── NEW SINGLE-CANDLE PATTERNS ──────────────────────────────────────
def fb_detect_dragonfly_doji(df, idx, cfg=None):
"""Dragonfly Doji: very small body at the top with long lower wick, nearly no upper wick.
Appears at bottom of downtrend → Bullish Reversal."""
if cfg is None: cfg = CFG
r = df.iloc[idx]
if r['RANGE'] == 0: return False
if r['BODY_RATIO'] > cfg['doji_body_ratio'] * 1.5: return False # Slightly looser than pure doji
if r['UPPER_WICK'] > r['BODY'] * 2: return False # Must have tiny/no upper wick
if r['LOWER_WICK'] < r['RANGE'] * 0.6: return False # Long lower wick is key
return True
def fb_detect_gravestone_doji(df, idx, cfg=None):
"""Gravestone Doji: very small body at the bottom with long upper wick, nearly no lower wick.
Appears at top of uptrend → Bearish Reversal."""
if cfg is None: cfg = CFG
r = df.iloc[idx]
if r['RANGE'] == 0: return False
if r['BODY_RATIO'] > cfg['doji_body_ratio'] * 1.5: return False
if r['LOWER_WICK'] > r['BODY'] * 2: return False # Must have tiny/no lower wick
if r['UPPER_WICK'] < r['RANGE'] * 0.6: return False # Long upper wick is key
return True
def fb_detect_bullish_belt_hold(df, idx, cfg=None):
"""Bullish Belt Hold: opens at/near the low, closes near the high, very small lower wick.
Appears at bottom of downtrend → Bullish Reversal."""
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_SIGN'] != 1: return False # Must be bullish candle
if r['BODY_RATIO'] < cfg['long_candle_ratio'] * 0.8: return False # Must be substantial body
if r['LOWER_WICK'] > r['BODY'] * cfg['belt_hold_wick_ratio']: return False # Tiny/no lower wick
return True
def fb_detect_bearish_belt_hold(df, idx, cfg=None):
"""Bearish Belt Hold: opens at/near the high, closes near the low, very small upper wick.
Appears at top of uptrend → Bearish Reversal."""
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_SIGN'] != -1: return False # Must be bearish candle
if r['BODY_RATIO'] < cfg['long_candle_ratio'] * 0.8: return False
if r['UPPER_WICK'] > r['BODY'] * cfg['belt_hold_wick_ratio']: return False
return True
# ── NEW TWO-CANDLE PATTERNS ─────────────────────────────────────────
def fb_detect_piercing_line(df, idx, cfg=None):
"""Piercing Line: bearish candle followed by bullish candle that opens below
prior close but closes above the midpoint of the bearish candle's body.
Bullish Reversal."""
if idx < 1: return False
c = df.iloc[idx]; p = df.iloc[idx-1]
if p['BODY_SIGN'] != -1: return False # Prior must be bearish
if c['BODY_SIGN'] != 1: return False # Current must be bullish
if p['BODY_RATIO'] < cfg.get('long_candle_ratio', 0.7) * 0.6: return False # Prior must be meaningful
p_mid = (p['OPEN'] + p['CLOSE']) / 2
if c['OPEN'] >= p['CLOSE']: return False # Must open below prior close
if c['CLOSE'] <= p_mid: return False # Must close above midpoint
return True
def fb_detect_dark_cloud_cover(df, idx, cfg=None):
"""Dark Cloud Cover: bullish candle followed by bearish candle that opens above
prior close but closes below the midpoint of the bullish candle's body.
Bearish Reversal."""
if idx < 1: return False
c = df.iloc[idx]; p = df.iloc[idx-1]
if p['BODY_SIGN'] != 1: return False # Prior must be bullish
if c['BODY_SIGN'] != -1: return False # Current must be bearish
if p['BODY_RATIO'] < cfg.get('long_candle_ratio', 0.7) * 0.6: return False
p_mid = (p['OPEN'] + p['CLOSE']) / 2
if c['OPEN'] <= p['CLOSE']: return False # Must open above prior close
if c['CLOSE'] >= p_mid: return False # Must close below midpoint
return True
def fb_detect_bullish_kicker(df, idx, cfg=None):
"""Bullish Kicker: long bearish candle followed by an even longer bullish candle
that opens above the prior close (gap up) and closes higher.
Strong Bullish Reversal."""
if idx < 1: return False
c = df.iloc[idx]; p = df.iloc[idx-1]
if p['BODY_SIGN'] != -1: return False
if c['BODY_SIGN'] != 1: return False
min_body = cfg.get('kicker_min_body_ratio', 0.6)
if p['BODY_RATIO'] < min_body: return False
if c['BODY_RATIO'] < min_body: return False
gap_tol = cfg.get('gap_tolerance_pips', 2) * cfg.get('pip_divisor', 0.0001)
if c['OPEN'] < p['CLOSE'] + gap_tol: return False # Must gap up above prior close
if c['CLOSE'] <= p['OPEN']: return False # Must close above prior open
return True
def fb_detect_bearish_kicker(df, idx, cfg=None):
"""Bearish Kicker: long bullish candle followed by an even longer bearish candle
that opens below the prior open (gap down) and closes lower.
Strong Bearish Reversal."""
if idx < 1: return False
c = df.iloc[idx]; p = df.iloc[idx-1]
if p['BODY_SIGN'] != 1: return False
if c['BODY_SIGN'] != -1: return False
min_body = cfg.get('kicker_min_body_ratio', 0.6)
if p['BODY_RATIO'] < min_body: return False
if c['BODY_RATIO'] < min_body: return False
gap_tol = cfg.get('gap_tolerance_pips', 2) * cfg.get('pip_divisor', 0.0001)
if c['OPEN'] > p['CLOSE'] - gap_tol: return False # Must gap down below prior close
if c['CLOSE'] >= p['OPEN']: return False # Must close below prior open
return True
def fb_detect_meeting_lines_bullish(df, idx, cfg=None):
"""Bullish Meeting Lines: long bearish candle followed by long bullish candle
that opens lower but closes at approximately the same level as the prior close.
Bullish Reversal."""
if idx < 1: return False
c = df.iloc[idx]; p = df.iloc[idx-1]
if p['BODY_SIGN'] != -1: return False
if c['BODY_SIGN'] != 1: return False
if p['BODY_RATIO'] < 0.5: return False
if c['BODY_RATIO'] < 0.5: return False
tol = cfg.get('meeting_lines_tolerance_pips', 2) * cfg.get('pip_divisor', 0.0001)
if abs(c['CLOSE'] - p['CLOSE']) > tol: return False # Closes must be approximately equal
return True
def fb_detect_meeting_lines_bearish(df, idx, cfg=None):
"""Bearish Meeting Lines: long bullish candle followed by long bearish candle
that opens higher but closes at approximately the same level as the prior close.
Bearish Reversal."""
if idx < 1: return False
c = df.iloc[idx]; p = df.iloc[idx-1]
if p['BODY_SIGN'] != 1: return False
if c['BODY_SIGN'] != -1: return False
if p['BODY_RATIO'] < 0.5: return False
if c['BODY_RATIO'] < 0.5: return False
tol = cfg.get('meeting_lines_tolerance_pips', 2) * cfg.get('pip_divisor', 0.0001)
if abs(c['CLOSE'] - p['CLOSE']) > tol: return False
return True
def fb_detect_bullish_separating_lines(df, idx, cfg=None):
"""Bullish Separating Lines: bearish candle followed by bullish candle that
opens at approximately the same price as the prior open. Bullish Continuation."""
if idx < 1: return False
c = df.iloc[idx]; p = df.iloc[idx-1]
if p['BODY_SIGN'] != -1: return False
if c['BODY_SIGN'] != 1: return False
if p['BODY_RATIO'] < 0.4: return False
if c['BODY_RATIO'] < 0.4: return False
tol = cfg.get('separating_lines_tolerance_pips', 2) * cfg.get('pip_divisor', 0.0001)
if abs(c['OPEN'] - p['OPEN']) > tol: return False # Opens must be approximately equal
return True
def fb_detect_bearish_separating_lines(df, idx, cfg=None):
"""Bearish Separating Lines: bullish candle followed by bearish candle that
opens at approximately the same price as the prior open. Bearish Continuation."""
if idx < 1: return False
c = df.iloc[idx]; p = df.iloc[idx-1]
if p['BODY_SIGN'] != 1: return False
if c['BODY_SIGN'] != -1: return False
if p['BODY_RATIO'] < 0.4: return False
if c['BODY_RATIO'] < 0.4: return False
tol = cfg.get('separating_lines_tolerance_pips', 2) * cfg.get('pip_divisor', 0.0001)
if abs(c['OPEN'] - p['OPEN']) > tol: return False
return True
def fb_detect_bearish_doji_star(df, idx, cfg=None):
"""Bearish Doji Star: long bullish candle followed by a Doji (small body).
Shows indecision after a strong move up → Bearish Reversal signal."""
if idx < 1: return False
c = df.iloc[idx]; p = df.iloc[idx-1]
if p['BODY_SIGN'] != 1: return False
if p['BODY_RATIO'] < cfg.get('long_candle_ratio', 0.7): return False
if c['BODY_RATIO'] > cfg.get('doji_body_ratio', 0.1) * 2: return False # Must be doji-like
return True
# ── NEW THREE-CANDLE PATTERNS ────────────────────────────────────────
def fb_detect_three_inside_up(df, idx, cfg=None):
"""Three Inside Up: large bearish → small bullish inside it (harami) →
third bullish closing above first candle's open. Confirms Bullish Harami."""
if idx < 2: return False
f, s, t = df.iloc[idx-2], df.iloc[idx-1], df.iloc[idx]
# First: large bearish
if f['BODY_SIGN'] != -1 or f['BODY_RATIO'] < cfg.get('long_candle_ratio', 0.7) * 0.8: return False
# Second: small bullish inside first's body
if s['BODY_SIGN'] != 1: return False
if s['BODY_RATIO'] > cfg.get('small_candle_ratio', 0.35) + 0.15: return False
if max(s['OPEN'], s['CLOSE']) > max(f['OPEN'], f['CLOSE']): return False
if min(s['OPEN'], s['CLOSE']) < min(f['OPEN'], f['CLOSE']): return False
# Third: bullish closing above first's open
if t['BODY_SIGN'] != 1: return False
if t['CLOSE'] <= f['OPEN']: return False
return True
def fb_detect_three_inside_down(df, idx, cfg=None):
"""Three Inside Down: large bullish → small bearish inside it (harami) →
third bearish closing below first candle's open. Confirms Bearish Harami."""
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.get('long_candle_ratio', 0.7) * 0.8: return False
if s['BODY_SIGN'] != -1: return False
if s['BODY_RATIO'] > cfg.get('small_candle_ratio', 0.35) + 0.15: return False
if max(s['OPEN'], s['CLOSE']) > max(f['OPEN'], f['CLOSE']): return False
if min(s['OPEN'], s['CLOSE']) < min(f['OPEN'], f['CLOSE']): return False
if t['BODY_SIGN'] != -1: return False
if t['CLOSE'] >= f['OPEN']: return False
return True
def fb_detect_three_outside_up(df, idx, cfg=None):
"""Three Outside Up: bearish candle → bullish engulfing → another bullish
closing higher. Confirms Bullish Engulfing."""
if idx < 2: return False
f, s, t = df.iloc[idx-2], df.iloc[idx-1], df.iloc[idx]
# First: bearish
if f['BODY_SIGN'] != -1: return False
# Second: bullish engulfing
if s['BODY_SIGN'] != 1: return False
if s['OPEN'] > f['CLOSE'] or s['CLOSE'] < f['OPEN']: return False
# Third: bullish closing higher than second
if t['BODY_SIGN'] != 1: return False
if t['CLOSE'] <= s['CLOSE']: return False
return True
def fb_detect_three_outside_down(df, idx, cfg=None):
"""Three Outside Down: bullish candle → bearish engulfing → another bearish
closing lower. Confirms Bearish Engulfing."""
if idx < 2: return False
f, s, t = df.iloc[idx-2], df.iloc[idx-1], df.iloc[idx]
if f['BODY_SIGN'] != 1: return False
if s['BODY_SIGN'] != -1: return False
if s['OPEN'] < f['CLOSE'] or s['CLOSE'] > f['OPEN']: return False
if t['BODY_SIGN'] != -1: return False
if t['CLOSE'] >= s['CLOSE']: return False
return True
def fb_detect_bullish_abandoned_baby(df, idx, cfg=None):
"""Bullish Abandoned Baby: long bearish → Doji (gap down) → long bullish (gap up).
Rare, strong Bullish Reversal."""
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.get('long_candle_ratio', 0.7): return False
if s['BODY_RATIO'] > cfg.get('doji_body_ratio', 0.1) * 2: return False # Doji
if t['BODY_SIGN'] != 1 or t['BODY_RATIO'] < cfg.get('long_candle_ratio', 0.7) * 0.6: return False
gap_tol = cfg.get('gap_tolerance_pips', 2) * cfg.get('pip_divisor', 0.0001)
# Doji gaps down from first (doji high < first low)
if s['HIGH'] > f['LOW'] + gap_tol: return False
# Third gaps up from doji (third low > doji high)
if t['LOW'] < s['HIGH'] + gap_tol: return False
return True
def fb_detect_bearish_abandoned_baby(df, idx, cfg=None):
"""Bearish Abandoned Baby: long bullish → Doji (gap up) → long bearish (gap down).
Rare, strong Bearish Reversal."""
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.get('long_candle_ratio', 0.7): return False
if s['BODY_RATIO'] > cfg.get('doji_body_ratio', 0.1) * 2: return False
if t['BODY_SIGN'] != -1 or t['BODY_RATIO'] < cfg.get('long_candle_ratio', 0.7) * 0.6: return False
gap_tol = cfg.get('gap_tolerance_pips', 2) * cfg.get('pip_divisor', 0.0001)
# Doji gaps up from first (doji low > first high)
if s['LOW'] < f['HIGH'] - gap_tol: return False
# Third gaps down from doji (third high < doji low)
if t['HIGH'] > s['LOW'] - gap_tol: return False
return True
def fb_detect_upside_gap_two_crows(df, idx, cfg=None):
"""Upside Gap Two Crows: long bullish → small bearish gap up →
larger bearish that engulfs the second but closes below first's close.
Bearish Reversal."""
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.get('long_candle_ratio', 0.7): return False
if s['BODY_SIGN'] != -1: return False # Second: small bearish
gap_tol = cfg.get('gap_tolerance_pips', 2) * cfg.get('pip_divisor', 0.0001)
if s['LOW'] < f['HIGH'] - gap_tol: return False # Must gap up
if t['BODY_SIGN'] != -1: return False # Third: bearish
if t['CLOSE'] >= s['CLOSE']: return False # Must close below second's close
if t['OPEN'] >= s['OPEN']: return False # Must open above second's open (engulf body)
if t['CLOSE'] >= f['CLOSE']: return False # Must close below first's close
return True
# ── NEW FOUR-CANDLE PATTERNS ─────────────────────────────────────────
def fb_detect_bullish_three_line_strike(df, idx, cfg=None):
"""Bullish Three-Line Strike: three consecutive bullish candles followed by
a long bearish candle that opens above third's close and closes below
first candle's open. Bullish Continuation (pullback before resumption)."""
if idx < 3: return False
c1, c2, c3, c4 = df.iloc[idx-3], df.iloc[idx-2], df.iloc[idx-1], df.iloc[idx]
# First three: consecutive bullish with progressive closes
if c1['BODY_SIGN'] != 1 or c2['BODY_SIGN'] != 1 or c3['BODY_SIGN'] != 1: return False
if c1['BODY_RATIO'] < 0.4 or c2['BODY_RATIO'] < 0.4 or c3['BODY_RATIO'] < 0.4: return False
if not (c3['CLOSE'] > c2['CLOSE'] > c1['CLOSE']): return False
# Fourth: long bearish opening above third, closing below first
if c4['BODY_SIGN'] != -1: return False
if c4['OPEN'] < c3['CLOSE']: return False
if c4['CLOSE'] > c1['OPEN']: return False
return True
def fb_detect_bearish_three_line_strike(df, idx, cfg=None):
"""Bearish Three-Line Strike: three consecutive bearish candles followed by
a long bullish candle that opens below third's close and closes above
first candle's open. Bearish Continuation."""
if idx < 3: return False
c1, c2, c3, c4 = df.iloc[idx-3], 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.4 or c2['BODY_RATIO'] < 0.4 or c3['BODY_RATIO'] < 0.4: return False
if not (c3['CLOSE'] < c2['CLOSE'] < c1['CLOSE']): return False
if c4['BODY_SIGN'] != 1: return False
if c4['OPEN'] > c3['CLOSE']: return False
if c4['CLOSE'] < c1['OPEN']: return False
return True
def fb_detect_concealing_baby_swallow(df, idx, cfg=None):
"""Concealing Baby Swallow: two long bearish → gap down with a small-bodied candle →
another long bearish that engulfs the small candle. Bullish Reversal.
Very rare pattern."""
if idx < 3: return False
c1, c2, c3, c4 = df.iloc[idx-3], df.iloc[idx-2], df.iloc[idx-1], df.iloc[idx]
# First two: long bearish
if c1['BODY_SIGN'] != -1 or c2['BODY_SIGN'] != -1: return False
if c1['BODY_RATIO'] < 0.5 or c2['BODY_RATIO'] < 0.5: return False
# Third: small body that gaps down
if c3['BODY_RATIO'] > cfg.get('small_candle_ratio', 0.35) + 0.1: return False
gap_tol = cfg.get('gap_tolerance_pips', 2) * cfg.get('pip_divisor', 0.0001)
if c3['HIGH'] > min(c1['LOW'], c2['LOW']) + gap_tol: return False # Must gap down
# Fourth: long bearish engulfing the third
if c4['BODY_SIGN'] != -1: return False
if c4['BODY_RATIO'] < 0.5: return False
if c4['HIGH'] < c3['HIGH'] or c4['LOW'] > c3['LOW']: return False # Engulfs third
return True
# ── NEW FIVE-CANDLE PATTERNS ─────────────────────────────────────────
def fb_detect_mat_hold_bullish(df, idx, cfg=None):
"""Mat Hold (Bullish): long bullish → small bearish candles that gap up
and stay within first candle's range → another long bullish closing
above first candle's close. Bullish Continuation (stronger than Rising Three)."""
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.get('long_candle_ratio', 0.7): return False
gap_tol = cfg.get('gap_tolerance_pips', 2) * cfg.get('pip_divisor', 0.0001)
for i in range(1, 4):
c = df.iloc[idx-4+i]
if c['BODY_RATIO'] > cfg.get('small_candle_ratio', 0.35) + 0.15: return False
if c['HIGH'] > first['HIGH'] or c['LOW'] < first['LOW']: return False
# Mat Hold: second candle must gap up from first
if i == 1 and c['LOW'] < first['CLOSE'] - gap_tol: return False
if fifth['BODY_SIGN'] != 1 or fifth['BODY_RATIO'] < cfg.get('long_candle_ratio', 0.7) * 0.7: return False
if fifth['CLOSE'] <= first['CLOSE']: return False
return True
def fb_detect_mat_hold_bearish(df, idx, cfg=None):
"""Mat Hold (Bearish): long bearish → small bullish candles that gap down
and stay within first candle's range → another long bearish closing
below first candle's close. Bearish Continuation."""
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.get('long_candle_ratio', 0.7): return False
gap_tol = cfg.get('gap_tolerance_pips', 2) * cfg.get('pip_divisor', 0.0001)
for i in range(1, 4):
c = df.iloc[idx-4+i]
if c['BODY_RATIO'] > cfg.get('small_candle_ratio', 0.35) + 0.15: return False
if c['HIGH'] > first['HIGH'] or c['LOW'] < first['LOW']: return False
if i == 1 and c['HIGH'] > first['CLOSE'] + gap_tol: return False
if fifth['BODY_SIGN'] != -1 or fifth['BODY_RATIO'] < cfg.get('long_candle_ratio', 0.7) * 0.7: return False
if fifth['CLOSE'] >= first['CLOSE']: return False
return True
def fb_detect_ladder_bottom(df, idx, cfg=None):
"""Ladder Bottom: three consecutive long bearish candles → small bearish/bullish
candle → long bullish candle. Bullish Reversal (buying pressure taking over)."""
if cfg is None: cfg = CFG
if idx < 4: return False
c1, c2, c3, c4, c5 = df.iloc[idx-4], df.iloc[idx-3], df.iloc[idx-2], df.iloc[idx-1], df.iloc[idx]
# First three: long bearish with progressive lower closes
if c1['BODY_SIGN'] != -1 or c2['BODY_SIGN'] != -1 or c3['BODY_SIGN'] != -1: return False
if c1['BODY_RATIO'] < 0.4 or c2['BODY_RATIO'] < 0.4 or c3['BODY_RATIO'] < 0.4: return False
if not (c3['CLOSE'] < c2['CLOSE'] < c1['CLOSE']): return False
# Fourth: small-bodied candle
if c4['BODY_RATIO'] > cfg.get('small_candle_ratio', 0.35) + 0.15: return False
# Fifth: long bullish
if c5['BODY_SIGN'] != 1: return False
if c5['BODY_RATIO'] < cfg.get('long_candle_ratio', 0.7) * 0.6: return False
return True
# ── THESTRAT PATTERN DETECTION ────────────────────────────────────────
def thestrat_classify_candle(df, idx, cfg=None):
"""Classify a candle using TheStrat 1-2-3 system.
Returns:
'1' — Inside bar: current high ≤ previous high AND current low ≥ previous low
'2↑' — Directional up: current high > previous high AND current low ≥ previous low
'2↓' — Directional down: current low < previous low AND current high ≤ previous high
'3' — Outside bar: current high > previous high AND current low < previous low
None — If idx < 1 or data invalid
"""
if idx < 1: return None
c = df.iloc[idx]; p = df.iloc[idx-1]
broke_high = c['HIGH'] > p['HIGH']
broke_low = c['LOW'] < p['LOW']
if broke_high and broke_low:
return '3' # Outside bar
elif broke_high and not broke_low:
return '2↑' # Directional up
elif broke_low and not broke_high:
return '2↓' # Directional down
else:
return '1' # Inside bar
def fb_detect_thestrat_22(df, idx, cfg=None):
"""TheStrat 2-2 Pattern: Two consecutive Type 2 candles.
Can be continuation (same direction) or reversal (opposite direction).
Returns dict with Pattern/Category/Direction or None.
"""
if idx < 2: return None
t1 = thestrat_classify_candle(df, idx-1, cfg)
t2 = thestrat_classify_candle(df, idx, cfg)
if t1 not in ('2↑', '2↓') or t2 not in ('2↑', '2↓'): return None
if t1 == t2:
# Same direction → continuation
d = 'Bullish' if t2 == '2↑' else 'Bearish'
return {'Pattern': 'TheStrat 2-2', 'Category': f'{d} Continuation', 'Direction': d, 'Candles': 2}
else:
# Opposite direction → reversal
d = 'Bullish' if t2 == '2↑' else 'Bearish'
return {'Pattern': 'TheStrat 2-2', 'Category': f'{d} Reversal', 'Direction': d, 'Candles': 2}
def fb_detect_thestrat_312(df, idx, cfg=None):
"""TheStrat 3-1-2 Reversal: Type 3 (outside) → Type 1 (inside) → Type 2 (directional breakout).
Direction is set by the final Type 2 candle's direction."""
if idx < 3: return None
t3 = thestrat_classify_candle(df, idx-2, cfg) # First candle: Type 3
t1 = thestrat_classify_candle(df, idx-1, cfg) # Second: Type 1
t2 = thestrat_classify_candle(df, idx, cfg) # Third: Type 2
if t3 != '3' or t1 != '1' or t2 not in ('2↑', '2↓'): return None
d = 'Bullish' if t2 == '2↑' else 'Bearish'
return {'Pattern': 'TheStrat 3-1-2', 'Category': f'{d} Reversal', 'Direction': d, 'Candles': 3}
def fb_detect_thestrat_212(df, idx, cfg=None):
"""TheStrat 2-1-2 Reversal: Type 2 (directional) → Type 1 (inside) → Type 2 (directional breakout).
Direction is set by the final Type 2 candle's direction."""
if idx < 3: return None
t2a = thestrat_classify_candle(df, idx-2, cfg)
t1 = thestrat_classify_candle(df, idx-1, cfg)
t2b = thestrat_classify_candle(df, idx, cfg)
if t2a not in ('2↑', '2↓') or t1 != '1' or t2b not in ('2↑', '2↓'): return None
d = 'Bullish' if t2b == '2↑' else 'Bearish'
return {'Pattern': 'TheStrat 2-1-2', 'Category': f'{d} Reversal', 'Direction': d, 'Candles': 3}
def fb_detect_thestrat_122_rev(df, idx, cfg=None):
"""TheStrat 1-2-2 Rev: Type 1 (inside) → Type 2 (breakout) → Type 2 (reversal).
The third candle must be in the opposite direction from the second.
Direction is set by the third candle."""
if idx < 3: return None
t1 = thestrat_classify_candle(df, idx-2, cfg)
t2a = thestrat_classify_candle(df, idx-1, cfg)
t2b = thestrat_classify_candle(df, idx, cfg)
if t1 != '1' or t2a not in ('2↑', '2↓') or t2b not in ('2↑', '2↓'): return None
# Must be opposite directions for reversal
if t2a == t2b: return None
d = 'Bullish' if t2b == '2↑' else 'Bearish'
return {'Pattern': 'TheStrat 1-2-2 Rev', 'Category': f'{d} Reversal', 'Direction': d, 'Candles': 3}
def fb_detect_thestrat_13_rev(df, idx, cfg=None):
"""TheStrat 1-3 Rev: Type 1 (inside) → Type 3 (outside/broadening).
A broadening formation suggests reversal potential.
Direction is inferred from where price goes after the outside bar.
We use the outside bar's body direction as a hint."""
if idx < 2: return None
t1 = thestrat_classify_candle(df, idx-1, cfg)
t3 = thestrat_classify_candle(df, idx, cfg)
if t1 != '1' or t3 != '3': return None
c = df.iloc[idx]
# Use body direction as a hint for the likely direction
d = 'Bullish' if c['CLOSE'] >= c['OPEN'] else 'Bearish'
return {'Pattern': 'TheStrat 1-3 Rev', 'Category': f'{d} Reversal', 'Direction': d, 'Candles': 2}
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 trade management and intra-candle path simulation.
Supports four trade management modes via cfg['trade_management_mode']:
'fixed' — Static SL/TP (original behavior, default)
'breakeven' — Move SL to entry (breakeven) when price hits breakeven_at_r R
'trail' — After trail_at_r R hit, trail SL by trail_atr_mult × ATR behind price
'partial' — Close partial_close_pct at partial_close_r R, trail the rest
Also supports time-based stop tightening: if time_stop_pct fraction of forward_candles
elapsed without TP, SL is tightened to breakeven.
Returns dict with: sl_hit, tp_hit, outcome, max_r, r_hits, fill_price,
entry_filled, bars_to_sl, bars_to_tp, mae_r, mfe_r,
exit_r (R-multiple at actual exit), sl_moved_to_be,
partial_closed, remaining_pct
"""
if cfg is None: cfg = CFG
max_r_levels = cfg.get('max_r_levels', 5)
pip_divisor = cfg.get('pip_divisor', 0.0001)
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'
bars_to_sl = None
bars_to_tp = None
mae_r = 0.0
mfe_r = 0.0
exit_r = 0.0 # R-multiple at actual exit
sl_moved_to_be = False
partial_closed = False
remaining_pct = 1.0 # Fraction of position still open
# Trade management config
tm_mode = cfg.get('trade_management_mode', 'fixed')
be_at_r = cfg.get('breakeven_at_r', 1.0)
trail_at_r = cfg.get('trail_at_r', 1.5)
trail_atr_mult = cfg.get('trail_atr_mult', 1.0)
partial_r = cfg.get('partial_close_r', 1.0)
partial_pct = cfg.get('partial_close_pct', 0.5)
time_stop_pct = cfg.get('time_stop_pct', 0.7)
time_stop_bar = int(forward_candles * time_stop_pct) if time_stop_pct > 0 else 0
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, 'bars_to_sl': None, 'bars_to_tp': None,
'mae_r': None, 'mfe_r': None, 'exit_r': None,
'sl_moved_to_be': False, 'partial_closed': False, 'remaining_pct': 1.0}
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, 'bars_to_sl': None, 'bars_to_tp': None,
'mae_r': 0.0, 'mfe_r': 0.0, 'exit_r': 0.0,
'sl_moved_to_be': False, 'partial_closed': False, 'remaining_pct': 1.0}
entry = fill_price if fill_price is not None else df.iloc[idx]['CLOSE']
risk = abs(entry - sl_price) if sl_price is not None else 0.001
if risk == 0:
risk = 0.001
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, 'bars_to_sl': None, 'bars_to_tp': None,
'mae_r': 0.0, 'mfe_r': 0.0, 'exit_r': 0.0,
'sl_moved_to_be': False, 'partial_closed': False, 'remaining_pct': 1.0}
# Active SL (may move during trade)
active_sl = sl_price
# Breakeven price (entry price, used when SL moves to BE)
be_price = entry
# Track whether trailing has started
trailing_active = False
# Track whether breakeven has been triggered (for partial mode)
be_triggered = False
bar_count = 0
stopped = False
for _, fc in future.iterrows():
if stopped:
break
bar_count += 1
fc_high = fc['HIGH']; fc_low = fc['LOW']
fc_open = fc['OPEN']; fc_close = fc['CLOSE']
# Current favorable R (for trade management decisions)
if direction == 'Bullish':
current_favorable_r = (fc_high - entry) / risk
else:
current_favorable_r = (entry - fc_low) / risk
# ── Trade management: adjust SL before checking hits ──────────
if tm_mode == 'breakeven' and not sl_moved_to_be:
if current_favorable_r >= be_at_r:
active_sl = be_price
sl_moved_to_be = True
elif tm_mode == 'trail':
# First move to breakeven at trail_at_r
if not sl_moved_to_be and current_favorable_r >= trail_at_r:
active_sl = be_price
sl_moved_to_be = True
trailing_active = True
# Then trail by ATR
if trailing_active and 'ATR' in fc.index and not pd.isna(fc.get('ATR', None)):
trail_dist = trail_atr_mult * fc['ATR']
if direction == 'Bullish':
new_sl = fc_high - trail_dist
if new_sl > active_sl:
active_sl = new_sl
else:
new_sl = fc_low + trail_dist
if new_sl < active_sl:
active_sl = new_sl
elif tm_mode == 'partial':
# Close partial position at partial_r
if not partial_closed and current_favorable_r >= partial_r:
partial_closed = True
remaining_pct = 1.0 - partial_pct
# Move SL to breakeven for the remainder
if not sl_moved_to_be:
active_sl = be_price
sl_moved_to_be = True
# After partial close, start trailing
if partial_closed and 'ATR' in fc.index and not pd.isna(fc.get('ATR', None)):
trail_dist = trail_atr_mult * fc['ATR']
if direction == 'Bullish':
new_sl = fc_high - trail_dist
if new_sl > active_sl:
active_sl = new_sl
else:
new_sl = fc_low + trail_dist
if new_sl < active_sl:
active_sl = new_sl
# Time-based stop tightening: if X% of forward window elapsed, tighten SL
# Only applies when trade management mode is not 'fixed'
if tm_mode != 'fixed' and time_stop_pct > 0 and bar_count >= time_stop_bar and not sl_moved_to_be:
active_sl = be_price
sl_moved_to_be = True
# ── Track MAE/MFE before checking stops ──────────────────────
if direction == 'Bullish':
adverse = entry - fc_low
favorable = fc_high - entry
else:
adverse = fc_high - entry
favorable = entry - fc_low
mae_r = max(mae_r, adverse / risk)
mfe_r = max(mfe_r, favorable / risk)
# ── Check SL/TP using the (possibly adjusted) active_sl ──────
sl_in_range = (fc_low <= active_sl if direction == 'Bullish' else fc_high >= active_sl)
tp_in_range = (fc_high >= tp_price if direction == 'Bullish' else fc_low <= tp_price)
if sl_in_range and tp_in_range:
sl_dist = abs(fc_open - active_sl)
tp_dist = abs(fc_open - tp_price)
if tp_dist <= sl_dist:
tp_hit = True; outcome = 'TP_Hit'
bars_to_tp = bar_count
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)
sl_hit = True
bars_to_sl = bar_count
# Exit R for TP
if direction == 'Bullish':
exit_r = (tp_price - entry) / risk
else:
exit_r = (entry - tp_price) / risk
exit_r *= remaining_pct # Scale by remaining position
else:
sl_hit = True; outcome = 'SL_Hit'
bars_to_sl = bar_count
tp_hit = True
bars_to_tp = bar_count
# Exit R for SL (with adjusted SL)
if direction == 'Bullish':
exit_r = (active_sl - entry) / risk
else:
exit_r = (entry - active_sl) / risk
exit_r *= remaining_pct
stopped = True
elif sl_in_range:
sl_hit = True; outcome = 'SL_Hit'; stopped = True
bars_to_sl = bar_count
# Calculate exit R with adjusted SL
if direction == 'Bullish':
exit_r = (active_sl - entry) / risk
else:
exit_r = (entry - active_sl) / risk
exit_r *= remaining_pct
elif tp_in_range:
tp_hit = True; outcome = 'TP_Hit'
bars_to_tp = bar_count
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
if direction == 'Bullish':
exit_r = (tp_price - entry) / risk
else:
exit_r = (entry - tp_price) / risk
exit_r *= remaining_pct
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:
if cfg.get('timeout_mode', 'marginal') == 'expired':
outcome = 'Expired'
else:
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'
# Calculate exit R for timeout
if len(future) > 0:
final_close = future.iloc[-1]['CLOSE']
if direction == 'Bullish':
exit_r = (final_close - entry) / risk
else:
exit_r = (entry - final_close) / risk
exit_r *= remaining_pct
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, 'bars_to_sl': bars_to_sl, 'bars_to_tp': bars_to_tp,
'mae_r': round(mae_r, 3), 'mfe_r': round(mfe_r, 3),
'exit_r': round(exit_r, 3),
'sl_moved_to_be': sl_moved_to_be, 'partial_closed': partial_closed,
'remaining_pct': remaining_pct}
# ============================================================
# 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 = []
# ── Single-candle: Neutral ──
if fb_detect_doji(df, idx, cfg):
found.append({'Pattern': 'Doji', 'Category': 'Neutral', 'Direction': 'Neutral', 'Candles': 1})
if fb_detect_dragonfly_doji(df, idx, cfg):
found.append({'Pattern': 'Dragonfly Doji', 'Category': 'Bullish Reversal', 'Direction': 'Bullish', 'Candles': 1})
if fb_detect_gravestone_doji(df, idx, cfg):
found.append({'Pattern': 'Gravestone Doji', 'Category': 'Bearish Reversal', 'Direction': 'Bearish', '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})
# ── Single-candle: Directional (trend-context required) ──
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})
if fb_detect_bullish_belt_hold(df, idx, cfg):
found.append({'Pattern': 'Bullish Belt Hold', 'Category': 'Bullish Reversal', 'Direction': 'Bullish', 'Candles': 1})
if fb_detect_bearish_belt_hold(df, idx, cfg):
found.append({'Pattern': 'Bearish Belt Hold', 'Category': 'Bearish Reversal', 'Direction': 'Bearish', 'Candles': 1})
# ── Two-candle patterns ──
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_piercing_line(df, idx, cfg):
found.append({'Pattern': 'Piercing Line', 'Category': 'Bullish Reversal', 'Direction': 'Bullish', 'Candles': 2})
if fb_detect_dark_cloud_cover(df, idx, cfg):
found.append({'Pattern': 'Dark Cloud Cover', 'Category': 'Bearish Reversal', 'Direction': 'Bearish', 'Candles': 2})
if fb_detect_bullish_kicker(df, idx, cfg):
found.append({'Pattern': 'Bullish Kicker', 'Category': 'Bullish Reversal', 'Direction': 'Bullish', 'Candles': 2})
if fb_detect_bearish_kicker(df, idx, cfg):
found.append({'Pattern': 'Bearish Kicker', 'Category': 'Bearish Reversal', 'Direction': 'Bearish', 'Candles': 2})
if fb_detect_meeting_lines_bullish(df, idx, cfg):
found.append({'Pattern': 'Meeting Lines (Bullish)', 'Category': 'Bullish Reversal', 'Direction': 'Bullish', 'Candles': 2})
if fb_detect_meeting_lines_bearish(df, idx, cfg):
found.append({'Pattern': 'Meeting Lines (Bearish)', 'Category': 'Bearish Reversal', 'Direction': 'Bearish', 'Candles': 2})
if fb_detect_bullish_separating_lines(df, idx, cfg):
found.append({'Pattern': 'Bullish Separating Lines', 'Category': 'Bullish Continuation', 'Direction': 'Bullish', 'Candles': 2})
if fb_detect_bearish_separating_lines(df, idx, cfg):
found.append({'Pattern': 'Bearish Separating Lines', 'Category': 'Bearish Continuation', 'Direction': 'Bearish', 'Candles': 2})
if fb_detect_bearish_doji_star(df, idx, cfg):
found.append({'Pattern': 'Bearish Doji Star', 'Category': 'Bearish Reversal', 'Direction': 'Bearish', 'Candles': 2})
# ── Three-candle patterns ──
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_three_inside_up(df, idx, cfg):
found.append({'Pattern': 'Three Inside Up', 'Category': 'Bullish Reversal', 'Direction': 'Bullish', 'Candles': 3})
if fb_detect_three_inside_down(df, idx, cfg):
found.append({'Pattern': 'Three Inside Down', 'Category': 'Bearish Reversal', 'Direction': 'Bearish', 'Candles': 3})
if fb_detect_three_outside_up(df, idx, cfg):
found.append({'Pattern': 'Three Outside Up', 'Category': 'Bullish Reversal', 'Direction': 'Bullish', 'Candles': 3})
if fb_detect_three_outside_down(df, idx, cfg):
found.append({'Pattern': 'Three Outside Down', 'Category': 'Bearish Reversal', 'Direction': 'Bearish', 'Candles': 3})
if fb_detect_bullish_abandoned_baby(df, idx, cfg):
found.append({'Pattern': 'Bullish Abandoned Baby', 'Category': 'Bullish Reversal', 'Direction': 'Bullish', 'Candles': 3})
if fb_detect_bearish_abandoned_baby(df, idx, cfg):
found.append({'Pattern': 'Bearish Abandoned Baby', 'Category': 'Bearish Reversal', 'Direction': 'Bearish', 'Candles': 3})
if fb_detect_upside_gap_two_crows(df, idx, cfg):
found.append({'Pattern': 'Upside Gap Two Crows', 'Category': 'Bearish Reversal', 'Direction': 'Bearish', 'Candles': 3})
# ── Four-candle patterns ──
if fb_detect_bullish_three_line_strike(df, idx, cfg):
found.append({'Pattern': 'Bullish Three-Line Strike', 'Category': 'Bullish Continuation', 'Direction': 'Bullish', 'Candles': 4})
if fb_detect_bearish_three_line_strike(df, idx, cfg):
found.append({'Pattern': 'Bearish Three-Line Strike', 'Category': 'Bearish Continuation', 'Direction': 'Bearish', 'Candles': 4})
if fb_detect_concealing_baby_swallow(df, idx, cfg):
found.append({'Pattern': 'Concealing Baby Swallow', 'Category': 'Bullish Reversal', 'Direction': 'Bullish', 'Candles': 4})
# ── Five-candle patterns ──
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})
if fb_detect_mat_hold_bullish(df, idx, cfg):
found.append({'Pattern': 'Mat Hold (Bullish)', 'Category': 'Bullish Continuation', 'Direction': 'Bullish', 'Candles': 5})
if fb_detect_mat_hold_bearish(df, idx, cfg):
found.append({'Pattern': 'Mat Hold (Bearish)', 'Category': 'Bearish Continuation', 'Direction': 'Bearish', 'Candles': 5})
if fb_detect_ladder_bottom(df, idx, cfg):
found.append({'Pattern': 'Ladder Bottom', 'Category': 'Bullish Reversal', 'Direction': 'Bullish', 'Candles': 5})
# ── TheStrat composite patterns ──
if cfg.get('thestrat_enabled', True):
ts22 = fb_detect_thestrat_22(df, idx, cfg)
if ts22:
found.append(ts22)
ts312 = fb_detect_thestrat_312(df, idx, cfg)
if ts312:
found.append(ts312)
ts212 = fb_detect_thestrat_212(df, idx, cfg)
if ts212:
found.append(ts212)
ts122 = fb_detect_thestrat_122_rev(df, idx, cfg)
if ts122:
found.append(ts122)
ts13 = fb_detect_thestrat_13_rev(df, idx, cfg)
if ts13:
found.append(ts13)
# 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')
# ============================================================
# SUPPORT/RESISTANCE & RSI HELPERS
# ============================================================
def fb_detect_swing_levels(df, idx, lookback=50, cfg=None):
"""Detect nearby swing highs and lows for support/resistance context.
Returns dict with:
- near_support: bool — price is within 1 ATR of a recent swing low
- near_resistance: bool — price is within 1 ATR of a recent swing high
- at_swing_low: bool — candle low is the lowest in the lookback window
- at_swing_high: bool — candle high is the highest in the lookback window
"""
if cfg is None: cfg = CFG
lookback = min(lookback, idx)
if lookback < 5:
return {'near_support': False, 'near_resistance': False,
'at_swing_low': False, 'at_swing_high': False}
subset = df.iloc[idx - lookback:idx + 1]
row = df.iloc[idx]
current_atr = row.get('ATR', 0)
if pd.isna(current_atr) or current_atr <= 0:
current_atr = subset['HIGH'].sub(subset['LOW']).mean()
if current_atr <= 0:
current_atr = 0.001
# Find swing highs and lows (local extremes using a 5-bar window)
swing_highs = []
swing_lows = []
for i in range(2, len(subset) - 2):
s = subset.iloc[i]
if s['HIGH'] >= subset.iloc[i-1]['HIGH'] and s['HIGH'] >= subset.iloc[i-2]['HIGH'] \
and s['HIGH'] >= subset.iloc[i+1]['HIGH'] and s['HIGH'] >= subset.iloc[i+2]['HIGH']:
swing_highs.append(s['HIGH'])
if s['LOW'] <= subset.iloc[i-1]['LOW'] and s['LOW'] <= subset.iloc[i-2]['LOW'] \
and s['LOW'] <= subset.iloc[i+1]['LOW'] and s['LOW'] <= subset.iloc[i+2]['LOW']:
swing_lows.append(s['LOW'])
near_support = any(abs(row['LOW'] - sl) <= current_atr for sl in swing_lows) if swing_lows else False
near_resistance = any(abs(row['HIGH'] - sh) <= current_atr for sh in swing_highs) if swing_highs else False
at_swing_low = row['LOW'] <= subset['LOW'].min() + current_atr * 0.1
at_swing_high = row['HIGH'] >= subset['HIGH'].max() - current_atr * 0.1
return {
'near_support': near_support,
'near_resistance': near_resistance,
'at_swing_low': at_swing_low,
'at_swing_high': at_swing_high,
}
def fb_compute_rsi(df, idx, period=14):
"""Compute RSI at a given index using Wilder's smoothing method.
Returns the RSI value (0-100) or None if insufficient data.
"""
if idx < period + 1:
return None
subset = df.iloc[max(0, idx - period * 3):idx + 1]
if len(subset) < period + 1:
return None
deltas = subset['CLOSE'].diff().dropna()
if len(deltas) < period:
return None
gains = deltas.where(deltas > 0, 0.0)
losses = (-deltas).where(deltas < 0, 0.0)
# Seed with SMA
avg_gain = gains.iloc[:period].mean()
avg_loss = losses.iloc[:period].mean()
if avg_loss == 0:
return 100.0
# Wilder's smoothing
for i in range(period, len(gains)):
avg_gain = (avg_gain * (period - 1) + gains.iloc[i]) / period
avg_loss = (avg_loss * (period - 1) + losses.iloc[i]) / period
if avg_loss == 0:
return 100.0
rs = avg_gain / avg_loss
return round(100.0 - (100.0 / (1.0 + rs)), 1)
def fb_compute_confluence(direction, trend, d1_trend, vol_confirmed, sr_context, rsi_value, session_quality, cfg=None):
"""Compute a confluence score (0-6 with D1 filter, 0-7 without) based on multiple confirming factors.
Each factor that aligns with the trade direction adds 1 point:
1. Trend alignment — trade with the local trend
2. D1 trend alignment — trade with the daily trend (SKIPPED when d1_trend_filter is active, since it's guaranteed)
3. Volume confirmation — above-average volume on signal candle
4. Support/Resistance context — near key level
5. RSI extreme — oversold for bullish, overbought for bearish
6. At swing extreme — at a swing high/low
7. Session quality — PRIME or FAVORABLE session
Returns (confluence_score, confluence_factors_list)
"""
if cfg is None: cfg = CFG
score = 0
factors = []
# 1. Trend alignment
if direction == 'Bullish' and trend == 'uptrend':
score += 1; factors.append('trend')
elif direction == 'Bearish' and trend == 'downtrend':
score += 1; factors.append('trend')
# 2. D1 trend alignment — skip when d1_trend_filter is active
# (the filter already guarantees alignment, so counting it would
# inflate every score by +1 and destroy score differentiation)
if not cfg.get('d1_trend_filter', False):
if direction == 'Bullish' and d1_trend == 'uptrend':
score += 1; factors.append('d1_trend')
elif direction == 'Bearish' and d1_trend == 'downtrend':
score += 1; factors.append('d1_trend')
# 3. Volume confirmation
if vol_confirmed:
score += 1; factors.append('volume')
# 4. S/R context — near support for bullish, near resistance for bearish
if sr_context.get('near_support') and direction == 'Bullish':
score += 1; factors.append('support')
elif sr_context.get('near_resistance') and direction == 'Bearish':
score += 1; factors.append('resistance')
# 5. RSI extreme
if rsi_value is not None:
if direction == 'Bullish' and rsi_value < 35:
score += 1; factors.append(f'rsi_{rsi_value}')
elif direction == 'Bearish' and rsi_value > 65:
score += 1; factors.append(f'rsi_{rsi_value}')
# 6. At swing extreme
if sr_context.get('at_swing_low') and direction == 'Bullish':
score += 1; factors.append('swing_low')
elif sr_context.get('at_swing_high') and direction == 'Bearish':
score += 1; factors.append('swing_high')
# 7. Session quality
if session_quality in ('PRIME', 'FAVORABLE'):
score += 1; factors.append(f'session_{session_quality}')
return score, factors
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 + confluence for one pattern occurrence.
Enhanced with:
- Variable R:R by pattern (rr_by_pattern config override)
- Support/Resistance context (swing level detection)
- RSI value at signal candle
- Confluence score (0-7) and factor list
- Time-to-SL/TP (bars_to_sl, bars_to_tp)
- MAE/MFE (Max Adverse/Favorable Excursion in R)
"""
if cfg is None: cfg = CFG
row = df.iloc[idx]
direction = pinfo['Direction']
pip_divisor = cfg.get('pip_divisor', 0.0001)
max_r_levels = cfg.get('max_r_levels', 5)
# Variable R:R by pattern — override tp_mult if pattern is listed
rr_overrides = cfg.get('rr_by_pattern', {})
pattern_name = pinfo['Pattern']
if pattern_name in rr_overrides:
effective_tp_mult = rr_overrides[pattern_name]
rr_ratio = effective_tp_mult / sl_mult
else:
effective_tp_mult = tp_mult
rr_ratio = tp_mult / sl_mult
# SL placement: ATR-based (default) or structure-based
sl_mode = cfg.get('sl_mode', 'atr')
struct_sl = None
sl_reason = ''
if sl_mode == 'structure' and direction in ('Bullish', 'Bearish'):
struct_result = compute_structure_sl(pattern_name, direction, df, idx, cfg)
if struct_result is not None:
struct_sl, sl_reason = struct_result
if direction == 'Bullish':
if struct_sl is not None:
sl = struct_sl
else:
sl = row['LOW'] - sl_mult * current_atr
risk = row['CLOSE'] - sl
tp = row['CLOSE'] + risk * rr_ratio
elif direction == 'Bearish':
if struct_sl is not None:
sl = struct_sl
else:
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)
# ── Context analysis: S/R, RSI, confluence ──────────────────────
hour = row['DATETIME'].hour
session = classify_session(hour, row['DATETIME'], cfg)
trend = fb_detect_trend(df, idx, cfg)
# Support/Resistance context
sr_context = fb_detect_swing_levels(df, idx, lookback=50, cfg=cfg)
# RSI
rsi_value = fb_compute_rsi(df, idx, period=14)
# Session quality (for confluence scoring)
# We need stats to compute session quality, but during backtest we don't have stats yet.
# Use a simple heuristic based on session name instead.
prime_sessions = {'London/NY Overlap', 'London Open'}
favorable_sessions = {'London Morning'}
if session in prime_sessions:
session_quality = 'PRIME'
elif session in favorable_sessions:
session_quality = 'FAVORABLE'
elif session in ('NY Afternoon', 'Asia'):
session_quality = 'NEUTRAL'
else:
session_quality = 'UNFAVORABLE'
# Confluence score
confluence_score, confluence_factors = fb_compute_confluence(
direction, trend, d1_trend, vol_confirmed, sr_context, rsi_value, session_quality, cfg)
# 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)}
bars_to_sl = None; bars_to_tp = None
mae_r = 0.0; mfe_r = 0.0
exit_r = 0.0; sl_moved_to_be = False; partial_closed = False; remaining_pct = 1.0
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']
bars_to_sl = fwd.get('bars_to_sl'); bars_to_tp = fwd.get('bars_to_tp')
mae_r = fwd.get('mae_r', 0.0); mfe_r = fwd.get('mfe_r', 0.0)
exit_r = fwd.get('exit_r', 0.0)
sl_moved_to_be = fwd.get('sl_moved_to_be', False)
partial_closed = fwd.get('partial_closed', False)
remaining_pct = fwd.get('remaining_pct', 1.0)
prediction_success = (True if outcome in ('TP_Hit', 'Marginal_Win') else
False if outcome in ('SL_Hit', 'Marginal_Loss', 'No_Fill', 'Expired') else None)
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,
# New fields
'Bars_to_SL': bars_to_sl, 'Bars_to_TP': bars_to_tp,
'MAE_R': mae_r, 'MFE_R': mfe_r,
'RSI': rsi_value,
'Near_Support': sr_context['near_support'],
'Near_Resistance': sr_context['near_resistance'],
'At_Swing_Low': sr_context['at_swing_low'],
'At_Swing_High': sr_context['at_swing_high'],
'Confluence_Score': confluence_score,
'Confluence_Factors': '|'.join(confluence_factors) if confluence_factors else '',
'RR_Override': pattern_name if pattern_name in rr_overrides else '',
'Exit_R': exit_r,
'SL_Moved_to_BE': sl_moved_to_be,
'Partial_Closed': partial_closed,
'Remaining_Pct': remaining_pct,
}
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
def compute_equity_curve(detections, cfg=None):
"""Compute equity curve and drawdown statistics from backtest detections.
Simulates sequential trading with fixed position sizing (1R risk per trade),
tracking cumulative P&L in R-multiples, then derives key metrics:
- Cumulative P&L curve
- Max drawdown (R and %)
- Sharpe ratio (annualised, assuming 252 trading days)
- Calmar ratio (annualised return / max drawdown)
- Max consecutive wins/losses
- Profit factor (gross profit / gross loss)
- Expectancy (average R per trade)
Returns dict with equity curve data and statistics, or None if insufficient data.
"""
if cfg is None: cfg = CFG
if not detections or len(detections) < 5:
return None
det_df = pd.DataFrame(detections)
directional = det_df[det_df['Direction'] != 'Neutral'].copy()
if len(directional) < 5:
return None
# Sort by DateTime to ensure sequential order
if 'DateTime' in directional.columns:
directional = directional.sort_values('DateTime').reset_index(drop=True)
# Determine R-multiple for each trade
# Use exit_r if available (v8 trade management), else derive from outcome
r_multiples = []
for _, row in directional.iterrows():
if 'Exit_R' in row and not pd.isna(row.get('Exit_R')):
r_mult = float(row['Exit_R'])
elif row.get('Outcome') == 'TP_Hit':
r_mult = float(row.get('TP_R_Multiple', 1.0))
elif row.get('Outcome') == 'SL_Hit':
# Check if SL was moved to breakeven
if row.get('SL_Moved_to_BE', False):
r_mult = 0.0
else:
r_mult = -1.0
elif row.get('Outcome') == 'Marginal_Win':
r_mult = 0.1 # Small positive
elif row.get('Outcome') == 'Marginal_Loss':
r_mult = -0.1 # Small negative
elif row.get('Outcome') == 'Expired':
r_mult = 0.0
elif row.get('Outcome') == 'No_Fill':
r_mult = 0.0
else:
r_mult = 0.0 # Timeout
r_multiples.append(r_mult)
directional = directional.copy()
directional['R_Multiple'] = r_multiples
# Cumulative equity curve (starting at 0)
directional['Cumulative_R'] = directional['R_Multiple'].cumsum()
# Drawdown calculation
directional['Peak_R'] = directional['Cumulative_R'].cummax()
directional['Drawdown_R'] = directional['Cumulative_R'] - directional['Peak_R']
max_dd_r = directional['Drawdown_R'].min()
# Max drawdown percentage (relative to peak equity)
peak_at_dd = directional.loc[directional['Drawdown_R'].idxmin(), 'Peak_R'] if max_dd_r < 0 else 0
max_dd_pct = abs(max_dd_r / peak_at_dd * 100) if peak_at_dd > 0 else 0
# Consecutive streaks
wins = (directional['R_Multiple'] > 0).values
losses = (directional['R_Multiple'] < 0).values
max_consec_wins = 0
max_consec_losses = 0
current_streak = 0
current_type = None
for w, l in zip(wins, losses):
if w:
if current_type == 'win':
current_streak += 1
else:
current_type = 'win'
current_streak = 1
max_consec_wins = max(max_consec_wins, current_streak)
elif l:
if current_type == 'loss':
current_streak += 1
else:
current_type = 'loss'
current_streak = 1
max_consec_losses = max(max_consec_losses, current_streak)
else:
current_streak = 0
current_type = None
# Profit factor
gross_profit = directional.loc[directional['R_Multiple'] > 0, 'R_Multiple'].sum()
gross_loss = abs(directional.loc[directional['R_Multiple'] < 0, 'R_Multiple'].sum())
profit_factor = gross_profit / gross_loss if gross_loss > 0 else float('inf')
# Expectancy
expectancy = directional['R_Multiple'].mean()
# Sharpe ratio (annualised)
if directional['R_Multiple'].std() > 0:
# Assume ~4 trades per day average across all TFs
trades_per_year = 252 * 4
sharpe = (directional['R_Multiple'].mean() / directional['R_Multiple'].std()) * np.sqrt(trades_per_year)
else:
sharpe = 0.0
# Calmar ratio (annualised return / max drawdown)
total_trades = len(directional)
annual_return = directional['Cumulative_R'].iloc[-1] * (252 * 4 / max(total_trades, 1))
calmar = annual_return / abs(max_dd_r) if max_dd_r != 0 else 0.0
# Win/loss statistics
n_wins = int((directional['R_Multiple'] > 0).sum())
n_losses = int((directional['R_Multiple'] < 0).sum())
avg_win = directional.loc[directional['R_Multiple'] > 0, 'R_Multiple'].mean() if n_wins > 0 else 0
avg_loss = directional.loc[directional['R_Multiple'] < 0, 'R_Multiple'].mean() if n_losses > 0 else 0
return {
'total_trades': total_trades,
'n_wins': n_wins,
'n_losses': n_losses,
'final_equity_r': round(directional['Cumulative_R'].iloc[-1], 2),
'max_dd_r': round(max_dd_r, 2),
'max_dd_pct': round(max_dd_pct, 1),
'max_consec_wins': max_consec_wins,
'max_consec_losses': max_consec_losses,
'profit_factor': round(profit_factor, 2),
'expectancy': round(expectancy, 3),
'sharpe': round(sharpe, 2),
'calmar': round(calmar, 2),
'avg_win_r': round(avg_win, 3) if avg_win else 0,
'avg_loss_r': round(avg_loss, 3) if avg_loss else 0,
'gross_profit_r': round(gross_profit, 2),
'gross_loss_r': round(gross_loss, 2),
'equity_curve': directional['Cumulative_R'].tolist(),
'drawdown_curve': directional['Drawdown_R'].tolist(),
}
# ============================================================
# 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"SL Mode : {cfg.get('sl_mode', '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"Trade Management : {cfg.get('trade_management_mode', 'fixed')}")
if cfg.get('trade_management_mode', 'fixed') != 'fixed':
L(f" Breakeven at : {cfg.get('breakeven_at_r', 1.0)}R")
L(f" Trail at : {cfg.get('trail_at_r', 1.5)}R x {cfg.get('trail_atr_mult', 1.0)} ATR")
L(f" Partial close : {cfg.get('partial_close_pct', 0.5)*100:.0f}% at {cfg.get('partial_close_r', 1.0)}R")
L(f" Time stop : {cfg.get('time_stop_pct', 0.7)*100:.0f}% of forward window")
L(f"Deduplicate : {cfg.get('deduplicate_signals', True)}")
L(f"Timeout Mode : {cfg.get('timeout_mode', 'marginal')}")
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', 'Expired', '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}%")
# ── Equity Curve & Drawdown ──
if cfg.get('equity_curve_enabled', True) and len(directional) >= 5:
eq = compute_equity_curve(detections, cfg)
if eq:
L(""); L("-" * 120); L("SECTION 8: EQUITY CURVE & DRAWDOWN"); L("-" * 120)
L(f" Total Trades : {eq['total_trades']}")
L(f" Final Equity : {eq['final_equity_r']:.2f}R")
L(f" Max Drawdown : {eq['max_dd_r']:.2f}R ({eq['max_dd_pct']:.1f}%)")
L(f" Max Consec Wins : {eq['max_consec_wins']}")
L(f" Max Consec Losses : {eq['max_consec_losses']}")
L(f" Profit Factor : {eq['profit_factor']:.2f}")
L(f" Expectancy : {eq['expectancy']:.3f}R per trade")
L(f" Avg Win : {eq['avg_win_r']:.3f}R")
L(f" Avg Loss : {eq['avg_loss_r']:.3f}R")
L(f" Gross Profit : {eq['gross_profit_r']:.2f}R")
L(f" Gross Loss : {eq['gross_loss_r']:.2f}R")
L(f" Sharpe Ratio : {eq['sharpe']:.2f}")
L(f" Calmar Ratio : {eq['calmar']:.2f}")
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']
watchlist = cfg.get('watchlist', [symbol])
if len(watchlist) > 1:
log_message(f"Multi-symbol watchlist: {', '.join(watchlist)}", cfg)
log_message(C('cyan', '=' * 70), cfg)
log_message(C('bold', f" {symbol} MULTI-TIMEFRAME PATTERN SCANNER v9 — STARTING"), cfg)
log_message(C('cyan', '=' * 70), cfg)
watchlist_display = ', '.join(cfg.get('watchlist', [symbol]))
if len(cfg.get('watchlist', [])) > 1:
log_message(f" Watchlist: {watchlist_display} (live scanner: {symbol})", cfg)
log_message(f"Active timeframes: {C('yellow', ', '.join(active_tfs))}", cfg)
current_offset = get_broker_offset_for_date(datetime.now(), cfg)
log_message(f"Timestamps: Local time ({datetime.now().strftime('%Z')}), broker GMT+{current_offset} (base {cfg.get('broker_utc_offset', 2)}, DST rule: {cfg.get('broker_dst_rule', 'us')})", 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)
# Sound alert status
if cfg.get('sound_enabled', True) and _WINSOUND:
buy_hz = cfg.get('sound_buy_hz', 1200)
sell_hz = cfg.get('sound_sell_hz', 400)
alert_tier = cfg.get('sound_alert_tier', 'B')
tier_b_min = cfg.get('sound_alert_tier_b_min_score', 60.0)
tier_labels = {'A': 'Elite only', 'B': 'Tradeable+', 'C': 'All directional'}
tier_desc = tier_labels.get(alert_tier, alert_tier)
log_message(f"Sound Alerts: {C('green', 'ENABLED')} | Buy: {buy_hz}Hz | Sell: {sell_hz}Hz | Alert Tier: {alert_tier} ({tier_desc})", cfg)
if alert_tier == 'B':
log_message(f" Tier B requires score >= {tier_b_min:.0f}", cfg)
log_message(f"Sound is {C('red', 'MUTED')} — Type {C('yellow', '\"m\" + Enter')} to unmute", cfg)
elif not _WINSOUND:
log_message(C('yellow', "Sound Alerts: DISABLED (winsound not available — Windows only)"), cfg)
if len(cfg.get('watchlist', [])) > 1:
log_message(f"Watchlist: {C('yellow', ', '.join(cfg['watchlist']))} (live scanner uses first symbol; use --mode scan for multi-symbol)", cfg)
# Start background keyboard listener for mute toggle
if cfg.get('sound_enabled', True) and _WINSOUND:
start_sound_key_listener()
# 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
# Auto-detect broker UTC offset (validates against DST calendar)
detected_offset = auto_detect_broker_offset(cfg)
expected_offset = get_broker_offset_for_date(datetime.now(), cfg)
if detected_offset == expected_offset:
log_message(f"Broker UTC offset: {detected_offset} (confirmed, matches DST calendar)", cfg)
else:
log_message(
C('yellow', f"Broker UTC offset MISMATCH: auto-detected {detected_offset}, "
f"expected {expected_offset} for today's date. "
f"Check broker_utc_offset ({cfg.get('broker_utc_offset', 2)}) and broker_dst_rule ({cfg.get('broker_dst_rule', 'us')})"), cfg)
# 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 = broker_time(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, tf_label=tf_label)
log_message(format_pattern_output(rates[-2], pats, cfg, stats, tf_label), cfg)
# Play sound alert for strong signals on startup scan
for pat in pats:
if pat.get('direction') in ('Bullish', 'Bearish'):
score = pat.get('signal_score')
if score is None:
score = compute_signal_score(pat['name'], pat['session'], pat['direction'], stats, cfg, tf_label=tf_label)
tier_letter, _, _ = compute_pattern_tier(pat['name'], stats, cfg)
play_signal_beep(pat['direction'], score, tier_letter, cfg)
continue
if bar_time != last_candle_time[tf_label]:
# bar_time is the NEW (forming) candle's open time (broker time).
# Its close time = bar_time + tf_minutes, converted to local time.
next_close_local = to_local_time(
bar_time + timedelta(minutes=tf_info['minutes']), cfg)
log_message(
C('bold', C('yellow',
f"\nNEW {tf_label} CANDLE CLOSED! | Next {tf_label} close: {next_close_local.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, tf_label=tf_label)
output = format_pattern_output(rates[-2], pats, cfg, stats, tf_label)
log_message(output, cfg)
# Play sound alert for strong directional signals
for pat in pats:
if pat.get('direction') in ('Bullish', 'Bearish'):
score = pat.get('signal_score')
if score is None:
score = compute_signal_score(pat['name'], pat['session'], pat['direction'], stats, cfg, tf_label=tf_label)
tier_letter, _, _ = compute_pattern_tier(pat['name'], stats, cfg)
play_signal_beep(pat['direction'], score, tier_letter, cfg)
last_candle_time[tf_label] = bar_time
# Keyboard listener runs in background thread — no manual check needed
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, tf_label=tf_label)
log_message(format_pattern_output(closed, pats, cfg, stats, tf_label), cfg)
# Play sound alert for strong directional signals
for pat in pats:
if pat.get('direction') in ('Bullish', 'Bearish'):
score = pat.get('signal_score')
if score is None:
score = compute_signal_score(pat['name'], pat['session'], pat['direction'], stats, cfg, tf_label=tf_label)
tier_letter, _, _ = compute_pattern_tier(pat['name'], stats, cfg)
play_signal_beep(pat['direction'], score, tier_letter, 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 = broker_time(int(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 v9")
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)}")
sl_m = cfg.get('sl_multiplier', 1.5)
tp_m = cfg.get('tp_multiplier', 1.5)
print(f" SL/TP : {sl_m}x / {tp_m}x ATR | R:R 1:{tp_m/sl_m:.1f}")
print(f" SL Mode : {cfg.get('sl_mode', 'atr')}")
print(f" Trade Mgmt : {cfg.get('trade_management_mode', 'fixed')}")
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,
# New fields
'Avg_MAE_R': round(ds['MAE_R'].dropna().mean(), 3) if 'MAE_R' in ds.columns and not ds['MAE_R'].dropna().empty else None,
'Avg_MFE_R': round(ds['MFE_R'].dropna().mean(), 3) if 'MFE_R' in ds.columns and not ds['MFE_R'].dropna().empty else None,
'Avg_Bars_to_TP': round(ds['Bars_to_TP'].dropna().mean(), 1) if 'Bars_to_TP' in ds.columns and not ds['Bars_to_TP'].dropna().empty else None,
'Avg_RSI': round(ds['RSI'].dropna().mean(), 1) if 'RSI' in ds.columns and not ds['RSI'].dropna().empty else None,
'At_Support_Pct': round((ds['Near_Support'] == True).sum() / max(len(ds),1) * 100, 1) if 'Near_Support' in ds.columns else None,
'At_Resistance_Pct': round((ds['Near_Resistance'] == True).sum() / max(len(ds),1) * 100, 1) if 'Near_Resistance' in ds.columns else None,
'Avg_Confluence': round(ds['Confluence_Score'].dropna().mean(), 1) if 'Confluence_Score' in ds.columns and not ds['Confluence_Score'].dropna().empty else None,
'Avg_Exit_R': round(ds['Exit_R'].dropna().mean(), 3) if 'Exit_R' in ds.columns and not ds['Exit_R'].dropna().empty else None,
'BE_Move_Pct': round((ds['SL_Moved_to_BE'] == True).sum() / max(len(ds),1) * 100, 1) if 'SL_Moved_to_BE' in ds.columns else None,
})
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 — ENRICHED with per-TF patterns/sessions/cross/confluence
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}",
'broker_utc_offset': cfg.get('broker_utc_offset', 2),
'broker_dst_rule': cfg.get('broker_dst_rule', 'us'),
'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),
'avg_mae_r': round(float(dirdf['MAE_R'].dropna().mean()), 3) if 'MAE_R' in dirdf.columns else None,
'avg_mfe_r': round(float(dirdf['MFE_R'].dropna().mean()), 3) if 'MFE_R' in dirdf.columns else None,
'avg_bars_to_tp': round(float(dirdf['Bars_to_TP'].dropna().mean()), 1) if 'Bars_to_TP' in dirdf.columns else None,
'avg_exit_r': round(float(dirdf['Exit_R'].dropna().mean()), 3) if 'Exit_R' in dirdf.columns else None,
'be_move_pct': round(float((dirdf['SL_Moved_to_BE'] == True).sum() / max(len(dirdf),1) * 100), 1) if 'SL_Moved_to_BE' in dirdf.columns else None,
}
# Per-pattern stats
tf_stats['patterns'] = {}
for pat in dirdf['Pattern'].unique():
ps = dirdf[dirdf['Pattern'] == pat]
ps_ = int((ps['Prediction_Success'] == True).sum())
pf_ = int((ps['Prediction_Success'] == False).sum())
tf_stats['patterns'][pat] = {
'win_rate': round(ps_ / (ps_ + pf_) * 100, 1) if (ps_ + pf_) > 0 else 0,
'total': len(ps),
'avg_max_r': round(float(ps['Max_R'].dropna().mean()), 2) if not ps['Max_R'].dropna().empty else 0,
'sl_hit_pct': round(float((ps['SL_Hit'] == True).sum() / len(ps) * 100), 1),
'tp_hit_pct': round(float((ps['TP_Hit'] == True).sum() / len(ps) * 100), 1),
}
# Per-session stats
tf_stats['sessions'] = {}
for sess in dirdf['Session'].unique():
sd = dirdf[dirdf['Session'] == sess]
ss_ = int((sd['Prediction_Success'] == True).sum())
sf_ = int((sd['Prediction_Success'] == False).sum())
tf_stats['sessions'][sess] = {
'win_rate': round(ss_ / (ss_ + sf_) * 100, 1) if (ss_ + sf_) > 0 else 0,
'signals': len(sd),
'avg_max_r': round(float(sd['Max_R'].dropna().mean()), 2) if not sd['Max_R'].dropna().empty else 0,
}
# Cross stats (pattern x session)
tf_stats['cross'] = {}
for pat in dirdf['Pattern'].unique():
for sess in dirdf['Session'].unique():
cs = dirdf[(dirdf['Pattern'] == pat) & (dirdf['Session'] == sess)]
if len(cs) >= 3:
cs_ = int((cs['Prediction_Success'] == True).sum())
cf_ = int((cs['Prediction_Success'] == False).sum())
tf_stats['cross'][f"{pat}|{sess}"] = {
'win_rate': round(cs_ / (cs_ + cf_) * 100, 1) if (cs_ + cf_) > 0 else 0,
'signals': len(cs),
'avg_max_r': round(float(cs['Max_R'].dropna().mean()), 2) if not cs['Max_R'].dropna().empty else 0,
}
# Confluence breakdown
if 'Confluence_Score' in dirdf.columns:
tf_stats['confluence'] = {}
for cs_val in sorted(dirdf['Confluence_Score'].dropna().unique()):
csd = dirdf[dirdf['Confluence_Score'] == cs_val]
cs_ = int((csd['Prediction_Success'] == True).sum())
cf_ = int((csd['Prediction_Success'] == False).sum())
tf_stats['confluence'][str(int(cs_val))] = {
'win_rate': round(cs_ / (cs_ + cf_) * 100, 1) if (cs_ + cf_) > 0 else 0,
'signals': len(csd),
'avg_max_r': round(float(csd['Max_R'].dropna().mean()), 2) if not csd['Max_R'].dropna().empty else 0,
}
# Equity curve
if cfg.get('equity_curve_enabled', True):
eq = compute_equity_curve(dets, cfg)
if eq:
tf_stats['equity'] = {
'final_equity_r': eq['final_equity_r'],
'max_dd_r': eq['max_dd_r'],
'max_dd_pct': eq['max_dd_pct'],
'max_consec_wins': eq['max_consec_wins'],
'max_consec_losses': eq['max_consec_losses'],
'profit_factor': eq['profit_factor'],
'expectancy': eq['expectancy'],
'sharpe': eq['sharpe'],
'calmar': eq['calmar'],
'avg_win_r': eq['avg_win_r'],
'avg_loss_r': eq['avg_loss_r'],
}
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 v9",
formatter_class=argparse.RawTextHelpFormatter,
epilog="""
Examples:
# Live scanner — all timeframes
python mt5_multitf_pattern_scanner.py
# Live scanner — specific timeframes
python mt5_multitf_pattern_scanner.py --timeframes M5 H1 H4
# One-shot scan
python mt5_multitf_pattern_scanner.py --mode scan
# Quick backtest (500 bars) on H4 only
python mt5_multitf_pattern_scanner.py --mode backtest --bars 500 --timeframes H4
# Full backtest on all TFs, 2024 full year
python mt5_multitf_pattern_scanner.py --mode fullbacktest --from 2024-01-01 --to 2024-12-31
# Full backtest with filters on H4 only
python mt5_multitf_pattern_scanner.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('--symbols', nargs='+', default=None,
help='Watchlist of symbols to scan/backtest (default: EURUSD from CFG). '
'Example: --symbols EURUSD GBPUSD USDJPY')
p.add_argument('--sl-mode', choices=['atr', 'structure'], default=None,
help='SL placement mode: atr (default, ATR-based) or structure (pattern invalidation level)')
p.add_argument('--trade-management', choices=['fixed', 'breakeven', 'trail', 'partial'], default=None,
help='Trade management mode in backtest: fixed (default), breakeven, trail, or partial')
p.add_argument('--timeout-mode', choices=['marginal', 'expired'], default=None,
help='Timeout classification: marginal (Marginal_Win/Loss) or expired (flat 0R)')
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'],
help="TP distance as ATR multiplier (e.g. 3.0). "
"Alternatively use --tp-rr for R:R-based setting")
p.add_argument("--tp-rr", type=float, default=None,
help="TP R:R ratio relative to SL (default: derived from --tp/--sl). "
"E.g. --tp-rr 2.0 sets TP at 2x the SL distance (1:2 R:R). "
"Overrides --tp if both are given")
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'],
help="Broker standard (winter) UTC offset (default: 2 for NY-close brokers)")
p.add_argument("--broker-dst-rule", type=str, default=cfg.get('broker_dst_rule', 'us'),
choices=['us', 'eu', 'none'],
help="DST rule: 'us' (2nd Sun Mar→1st Sun Nov), 'eu', or 'none' (default: us)")
# 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.add_argument("--test-sound", action="store_true",
help="Play test beeps (Tier A BUY then SELL) and exit")
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',
'broker_dst_rule': 'broker_dst_rule',
'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()}
# Handle --sl-mode, --trade-management, --timeout-mode overrides
if args.sl_mode:
runtime_cfg['sl_mode'] = args.sl_mode
if args.trade_management:
runtime_cfg['trade_management_mode'] = args.trade_management
if args.timeout_mode:
runtime_cfg['timeout_mode'] = args.timeout_mode
# Handle --tp-rr: override tp_multiplier from R:R ratio
# tp_multiplier = sl_multiplier × tp_rr (e.g. sl=1.5, tp-rr=2.0 → tp=3.0 → R:R 1:2)
if args.tp_rr is not None:
runtime_cfg['tp_multiplier'] = runtime_cfg['sl_multiplier'] * args.tp_rr
# Handle --test-sound: play both test beeps and exit
if args.test_sound:
test_sound(runtime_cfg)
return
# Determine symbol watchlist
if args.symbols:
watchlist = args.symbols
else:
watchlist = runtime_cfg.get('watchlist', [runtime_cfg.get('symbol', 'EURUSD')])
if args.mode == 'live':
run_scanner(runtime_cfg) # Scanner handles its own symbol via cfg
elif args.mode == 'scan':
for sym in watchlist:
if len(watchlist) > 1:
print(f"\n{'='*60}")
print(f" SCANNING: {sym}")
print(f"{'='*60}")
runtime_cfg['symbol'] = sym
run_single_scan(runtime_cfg)
elif args.mode == 'backtest':
for sym in watchlist:
if len(watchlist) > 1:
print(f"\n{'='*60}")
print(f" QUICK BACKTEST: {sym}")
print(f"{'='*60}")
runtime_cfg['symbol'] = sym
run_quick_backtest(args.bars, runtime_cfg)
elif args.mode == 'fullbacktest':
for sym in watchlist:
if len(watchlist) > 1:
print(f"\n{'='*60}")
print(f" FULL BACKTEST: {sym}")
print(f"{'='*60}")
args.symbol = sym
runtime_cfg['symbol'] = sym
run_full_backtest(args, runtime_cfg)
if __name__ == '__main__':
main()