In soft
# ============================================
# MY QUANT BOT - Phase 18 (SMC Edition)
# Full professional entry system:
# 1. Multi-timeframe bias (4H + 1H + 15M)
# 2. Smart Money Concepts (FVG, BOS, CHOCH)
# 3. Liquidity sweep detection
# 4. Smart position scaling
# 5. Full entry confirmation flow
# 6. Everything from Phase 17 still works
# ============================================
import os
import time
import json
import numpy as np
import MetaTrader5 as mt5
import pandas as pd
import requests
from dotenv import load_dotenv
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler
from datetime import datetime, timezone
# --- Load credentials ---
load_dotenv()
LOGIN = int(os.getenv("MT5_LOGIN"))
PASSWORD = os.getenv("MT5_PASSWORD")
SERVER = os.getenv("MT5_SERVER")
# --- Pairs to trade ---
SYMBOLS = ["EURUSDm", "GBPUSDm", "XAUUSDm"]
# --- Timeframes ---
TIMEFRAME_M5 = mt5.TIMEFRAME_M5
TIMEFRAME_M15 = mt5.TIMEFRAME_M15
TIMEFRAME_H1 = mt5.TIMEFRAME_H1
TIMEFRAME_H4 = mt5.TIMEFRAME_H4
# --- Indicator Settings ---
RSI_PERIOD = 14
MACD_FAST = 12
MACD_SLOW = 26
MACD_SIGNAL = 9
ATR_PERIOD = 14
ADX_PERIOD = 14
# --- EMA Settings ---
EMA_FAST = 50
EMA_SLOW = 200
EMA_PULLBACK = 20
# --- Session Filter (UTC) ---
SESSION_START = 7
SESSION_END = 21
# --- Volatility Filter ---
ATR_MIN_THRESHOLD = {
"EURUSDm": 0.00030,
"GBPUSDm": 0.00040,
"XAUUSDm": 0.30,
}
# --- Spread Filter ---
MAX_SPREAD_MULTIPLIER = 1.5
# --- Chop Filter ---
CHOP_CANDLES = 8
CHOP_BODY_RATIO = 0.4
CHOP_MAX_COUNT = 5
# --- SMC Settings ---
FVG_LOOKBACK = 20 # Candles to look for FVGs
SWING_LOOKBACK = 50 # Candles for swing highs/lows
LIQ_ZONE_BUFFER = 0.3 # ATR multiplier for liquidity zone
BOS_LOOKBACK = 10 # Candles for BOS detection
# --- Position Scaling ---
SCALE_INITIAL = 0.4 # 40% of calculated lot for first entry
SCALE_ADD = 0.3 # 30% for each add
MAX_SCALE = 3 # Max scale-ins per symbol
# --- ADX Minimum ---
ADX_MIN_TREND = 20
# --- Market Regimes ---
REGIME_SETTINGS = {
"TRENDING": {
"atr_sl_mult": 1.2,
"atr_tp_mult": 3.0,
"min_conf": 55,
"use_pullback": True,
"use_breakout": True,
},
"RANGING": {
"atr_sl_mult": 1.0,
"atr_tp_mult": 1.5,
"min_conf": 65,
"use_pullback": True,
"use_breakout": False,
},
"HIGH_VOLATILITY": {
"atr_sl_mult": 2.0,
"atr_tp_mult": 2.5,
"min_conf": 75,
"use_pullback": False,
"use_breakout": True,
},
"CHOPPY": {
"atr_sl_mult": 1.5,
"atr_tp_mult": 2.0,
"min_conf": 999,
"use_pullback": False,
"use_breakout": False,
},
}
# --- News Filter Settings ---
NEWS_PAUSE_BEFORE = 30
NEWS_PAUSE_AFTER = 15
NEWS_CURRENCIES = ["USD", "EUR", "GBP", "XAU"]
# --- Risk Profiles ---
RISK_PROFILES = {
"LOW": {
"risk_pct": 0.005,
"min_rsi_gap": 5,
},
"BALANCED": {
"risk_pct": 0.01,
"min_rsi_gap": 0,
},
}
# --- Trade Management Styles ---
MGMT_STYLES = {
0: "TRAILING_SL",
1: "BREAK_EVEN",
2: "PARTIAL_CLOSE",
}
BE_TRIGGER = 0.5
PC_TRIGGER = 0.5
PC_VOLUME = 0.5
# --- Trading Costs ---
TRADING_COSTS = {
"EURUSDm": {
"spread": 0.00013,
"slippage": 0.00005,
"commission": 0.07,
"pip_value": 1.0,
"pip_size": 0.0001,
},
"GBPUSDm": {
"spread": 0.00018,
"slippage": 0.00007,
"commission": 0.07,
"pip_value": 1.0,
"pip_size": 0.0001,
},
"XAUUSDm": {
"spread": 0.30,
"slippage": 0.10,
"commission": 0.10,
"pip_value": 1.0,
"pip_size": 0.1,
},
}
# --- Risk Management ---
MAX_DRAWDOWN_PCT = 0.10
MAX_DAILY_LOSS = 0.05
RETRAIN_EVERY_DAYS = 7
# --- File paths ---
TRADES_FILE = "trades.csv"
SUMMARY_FILE = "summary.txt"
MODEL_LOG = "model_log.txt"
NEWS_LOG = "news_log.txt"
MEMORY_FILE = "trade_memory.json"
SLIPPAGE_FILE = "slippage_log.json"
# --- ML models per pair ---
models = {}
last_retrain = {}
fallback_models = {}
# --- Active trade management ---
managed_trades = {}
# --- Risk tracking ---
peak_balance = None
day_start_bal = None
# ============================================
# 1. MULTI-TIMEFRAME BIAS SYSTEM
# ============================================
def get_htf_bias_4h(symbol):
"""
4H timeframe bias using EMA50/200
This is the MACRO trend direction
"""
rates = mt5.copy_rates_from_pos(symbol, TIMEFRAME_H4, 0, 210)
if rates is None or len(rates) < 210:
return "NEUTRAL"
df = pd.DataFrame(rates)
ema50 = df["close"].ewm(span=50, adjust=False).mean()
ema200 = df["close"].ewm(span=200, adjust=False).mean()
price = df["close"].iloc[-1]
e50 = ema50.iloc[-1]
e200 = ema200.iloc[-1]
# Strong uptrend
if price > e50 > e200 and e50 > e200:
return "BUY"
# Strong downtrend
elif price < e50 < e200 and e50 < e200:
return "SELL"
return "NEUTRAL"
def get_htf_bias_1h(symbol):
"""
1H timeframe bias using EMA20/50
This is the INTERMEDIATE trend direction
"""
rates = mt5.copy_rates_from_pos(symbol, TIMEFRAME_H1, 0, 55)
if rates is None or len(rates) < 55:
return "NEUTRAL"
df = pd.DataFrame(rates)
ema20 = df["close"].ewm(span=20, adjust=False).mean()
ema50 = df["close"].ewm(span=50, adjust=False).mean()
price = df["close"].iloc[-1]
e20 = ema20.iloc[-1]
e50 = ema50.iloc[-1]
if price > e20 > e50:
return "BUY"
elif price < e20 < e50:
return "SELL"
return "NEUTRAL"
def get_ltf_bias_m5(symbol):
"""
5M timeframe for precise entry alignment
Uses MACD direction
"""
rates = mt5.copy_rates_from_pos(symbol, TIMEFRAME_M5, 0, 50)
if rates is None or len(rates) < 50:
return "NEUTRAL"
df = pd.DataFrame(rates)
ema_fast = df["close"].ewm(span=12, adjust=False).mean()
ema_slow = df["close"].ewm(span=26, adjust=False).mean()
macd = ema_fast - ema_slow
signal = macd.ewm(span=9, adjust=False).mean()
if macd.iloc[-1] > signal.iloc[-1]:
return "BUY"
elif macd.iloc[-1] < signal.iloc[-1]:
return "SELL"
return "NEUTRAL"
def get_mtf_bias(symbol):
"""
Combines 4H + 1H + 5M for full alignment
Returns direction only if at least 2/3 agree
"""
bias_4h = get_htf_bias_4h(symbol)
bias_1h = get_htf_bias_1h(symbol)
bias_m5 = get_ltf_bias_m5(symbol)
biases = [bias_4h, bias_1h, bias_m5]
buy_count = biases.count("BUY")
sell_count = biases.count("SELL")
if buy_count >= 2:
return "BUY", bias_4h, bias_1h, bias_m5
elif sell_count >= 2:
return "SELL", bias_4h, bias_1h, bias_m5
return "NEUTRAL", bias_4h, bias_1h, bias_m5
# ============================================
# 2. SMART MONEY CONCEPTS (SMC)
# ============================================
def detect_fvg(df, direction):
"""
Fair Value Gap (FVG) Detection:
A FVG is a 3-candle pattern where:
- For BUY FVG: candle[i-2].high < candle[i].low
- For SELL FVG: candle[i-2].low > candle[i].high
FVGs act as magnets — price often returns to fill them.
We avoid trading INTO unfilled FVGs.
"""
fvgs = []
for i in range(2, min(FVG_LOOKBACK, len(df))):
c0 = df.iloc[-i-1] # oldest
c1 = df.iloc[-i] # middle
c2 = df.iloc[-i+1] # newest
if direction == "BUY":
# Bullish FVG: gap between c0 high and c2 low
if c0["high"] < c2["low"]:
fvgs.append({
"type": "BULLISH",
"top": c2["low"],
"bottom": c0["high"],
"filled": False,
})
elif direction == "SELL":
# Bearish FVG: gap between c0 low and c2 high
if c0["low"] > c2["high"]:
fvgs.append({
"type": "BEARISH",
"top": c0["low"],
"bottom": c2["high"],
"filled": False,
})
return fvgs
def is_price_in_fvg(price, fvgs):
"""
Checks if current price is inside an unfilled FVG.
If yes, avoid entry (wait for gap to fill).
"""
for fvg in fvgs:
if fvg["bottom"] <= price <= fvg["top"]:
return True, fvg
return False, None
def detect_bos_choch(df, direction):
"""
Break of Structure (BOS) / Change of Character (CHOCH)
BOS: Trend continuation signal
- In uptrend: Price breaks above previous swing high
- In downtrend: Price breaks below previous swing low
CHOCH: Trend reversal signal
- Price breaks against the current trend structure
Returns: "BOS", "CHOCH", or None
"""
recent = df.iloc[-BOS_LOOKBACK:]
highs = recent["high"].values
lows = recent["low"].values
closes = recent["close"].values
current_price = closes[-1]
# Find previous swing high/low
prev_swing_high = max(highs[:-3])
prev_swing_low = min(lows[:-3])
if direction == "BUY":
# BOS: breaks above previous swing high
if current_price > prev_swing_high:
return "BOS"
# CHOCH: breaks below previous swing low (reversal)
if current_price < prev_swing_low:
return "CHOCH"
elif direction == "SELL":
# BOS: breaks below previous swing low
if current_price < prev_swing_low:
return "BOS"
# CHOCH: breaks above previous swing high (reversal)
if current_price > prev_swing_high:
return "CHOCH"
return None
def detect_liquidity_sweep(df, direction):
"""
Liquidity Sweep Detection:
Smart money drives price to sweep stop losses
clustered above highs / below lows,
then reverses.
BUY sweep: Price briefly dips below recent lows
(sweeping stop losses) then bounces up.
SELL sweep: Price briefly spikes above recent highs
then falls.
Returns True if sweep was just detected.
"""
recent = df.iloc[-10:]
latest = df.iloc[-1]
prev = df.iloc[-2]
atr = latest["atr"]
if direction == "BUY":
# Recent low that was swept
recent_low = recent["low"].min()
# Check if previous candle dipped below then recovered
swept = (prev["low"] < recent_low) and (latest["close"] > recent_low)
if swept:
return True, f"✅ Bullish liquidity sweep at {recent_low:.5f}"
elif direction == "SELL":
# Recent high that was swept
recent_high = recent["high"].max()
# Check if previous candle spiked above then fell
swept = (prev["high"] > recent_high) and (latest["close"] < recent_high)
if swept:
return True, f"✅ Bearish liquidity sweep at {recent_high:.5f}"
return False, "No sweep"
def detect_order_block(df, direction):
"""
Order Block Detection (MY OWN ADDITION):
An order block is the last bearish candle before a bullish move
or the last bullish candle before a bearish move.
These are high-probability entry zones.
BUY: Find last bearish candle before a significant up move
SELL: Find last bullish candle before a significant down move
"""
atr = df["atr"].iloc[-1]
for i in range(3, min(20, len(df))):
candle = df.iloc[-i]
future = df.iloc[-i+1: -i+4]
if len(future) < 3:
continue
candle_body = abs(candle["close"] - candle["open"])
future_move = abs(future["close"].iloc[-1] - candle["close"])
if direction == "BUY":
# Last bearish candle before strong up move
is_bearish = candle["close"] < candle["open"]
strong_move = future_move > atr * 1.5
if is_bearish and strong_move:
ob_zone = {"top": candle["high"], "bottom": candle["low"]}
current = df["close"].iloc[-1]
# Price returning to order block zone
if ob_zone["bottom"] <= current <= ob_zone["top"]:
return True, ob_zone
elif direction == "SELL":
# Last bullish candle before strong down move
is_bullish = candle["close"] > candle["open"]
strong_move = future_move > atr * 1.5
if is_bullish and strong_move:
ob_zone = {"top": candle["high"], "bottom": candle["low"]}
current = df["close"].iloc[-1]
if ob_zone["bottom"] <= current <= ob_zone["top"]:
return True, ob_zone
return False, None
# ============================================
# 3. LIQUIDITY ZONE DETECTION
# ============================================
def detect_liquidity_zones(df):
"""
Identifies areas where stop losses likely cluster:
- Just above recent swing highs (sell stop clusters)
- Just below recent swing lows (buy stop clusters)
Price is attracted to these zones before reversing.
"""
recent = df.iloc[-SWING_LOOKBACK:]
atr = df["atr"].iloc[-1]
price = df["close"].iloc[-1]
swing_high = recent["high"].max()
swing_low = recent["low"].min()
# Buy stops cluster above swing high
buy_stops_zone = (swing_high, swing_high + atr * LIQ_ZONE_BUFFER)
# Sell stops cluster below swing low
sell_stops_zone = (swing_low - atr * LIQ_ZONE_BUFFER, swing_low)
# Is price in a dangerous liquidity zone?
in_buy_stops = buy_stops_zone[0] <= price <= buy_stops_zone[1]
in_sell_stops = sell_stops_zone[0] <= price <= sell_stops_zone[1]
return {
"swing_high": swing_high,
"swing_low": swing_low,
"buy_stops": buy_stops_zone,
"sell_stops": sell_stops_zone,
"in_buy_stops": in_buy_stops,
"in_sell_stops": in_sell_stops,
}
def passes_liquidity_filter(direction, liq_zones):
"""
Avoids entering into low-liquidity danger zones.
BUY: Don't buy when price is right at sell stop zone
SELL: Don't sell when price is right at buy stop zone
"""
if direction == "BUY" and liq_zones["in_sell_stops"]:
return False, "⚠️ Price in sell-stop zone — avoid BUY"
if direction == "SELL" and liq_zones["in_buy_stops"]:
return False, "⚠️ Price in buy-stop zone — avoid SELL"
return True, "✅ Liquidity zone clear"
# ============================================
# 4. SMART POSITION SCALING
# ============================================
def get_scale_in_lot(symbol, sl_price, entry_price, risk_mode, scale_level):
"""
Smart scaling instead of fixed lot stacking.
Scale 1 (initial): 40% of calculated risk lot
Scale 2 (add): 30% of calculated risk lot
Scale 3 (final): 30% of calculated risk lot
Only adds if price moves favorably.
Total risk never exceeds 1% per trade series.
"""
balance, _ = get_account_info()
if balance is None:
return 0.01
profile = RISK_PROFILES[risk_mode]
risk_pct = profile["risk_pct"]
risk_amount = balance * risk_pct
costs = TRADING_COSTS.get(symbol, TRADING_COSTS["EURUSDm"])
pip_value = costs["pip_value"]
pip_size = costs["pip_size"]
sl_pips = abs(entry_price - sl_price) / pip_size
if sl_pips == 0:
return 0.01
base_lot = risk_amount / (sl_pips * pip_value)
scale_factors = {1: SCALE_INITIAL, 2: SCALE_ADD, 3: SCALE_ADD}
factor = scale_factors.get(scale_level, SCALE_ADD)
lot_size = base_lot * factor
return max(0.01, min(0.50, round(lot_size, 2)))
def can_scale_in(symbol, direction, atr):
"""
Check if we can add a scale-in position.
Only if:
1. Existing trades in profit
2. Price moved at least 0.5 ATR in our direction
3. Under max scale limit
"""
positions = get_open_positions(symbol)
if not positions:
return False, 0
count = len(positions)
if count >= MAX_SCALE:
return False, count
# All must be profitable
if not all(pos.profit > 0 for pos in positions):
return False, count
# Price must have moved favorably
first_pos = positions[0]
entry = first_pos.price_open
current = mt5.symbol_info_tick(symbol)
if current is None:
return False, count
price = current.bid if direction == "BUY" else current.ask
move = price - entry if direction == "BUY" else entry - price
if move >= atr * 0.5:
return True, count
return False, count
# ============================================
# INDICATORS
# ============================================
def calculate_rsi(df, period=14):
delta = df["close"].diff()
gain = delta.where(delta > 0, 0.0)
loss = -delta.where(delta < 0, 0.0)
avg_gain = gain.rolling(window=period).mean()
avg_loss = loss.rolling(window=period).mean()
rs = avg_gain / avg_loss
return 100 - (100 / (1 + rs))
def calculate_macd(df, fast=12, slow=26, signal=9):
ema_fast = df["close"].ewm(span=fast, adjust=False).mean()
ema_slow = df["close"].ewm(span=slow, adjust=False).mean()
macd_line = ema_fast - ema_slow
signal_line = macd_line.ewm(span=signal, adjust=False).mean()
return macd_line, signal_line, macd_line - signal_line
def calculate_atr(df, period=14):
df = df.copy()
df["h-l"] = df["high"] - df["low"]
df["h-pc"] = abs(df["high"] - df["close"].shift(1))
df["l-pc"] = abs(df["low"] - df["close"].shift(1))
df["tr"] = df[["h-l", "h-pc", "l-pc"]].max(axis=1)
return df["tr"].rolling(window=period).mean()
def calculate_adx(df, period=14):
df = df.copy()
df["h-ph"] = df["high"] - df["high"].shift(1)
df["pl-l"] = df["low"].shift(1) - df["low"]
df["+dm"] = np.where((df["h-ph"] > df["pl-l"]) & (df["h-ph"] > 0), df["h-ph"], 0)
df["-dm"] = np.where((df["pl-l"] > df["h-ph"]) & (df["pl-l"] > 0), df["pl-l"], 0)
df["h-l"] = df["high"] - df["low"]
df["h-pc"] = abs(df["high"] - df["close"].shift(1))
df["l-pc"] = abs(df["low"] - df["close"].shift(1))
df["tr"] = df[["h-l", "h-pc", "l-pc"]].max(axis=1)
atr_s = df["tr"].ewm(span=period, adjust=False).mean()
pdm_s = df["+dm"].ewm(span=period, adjust=False).mean()
ndm_s = df["-dm"].ewm(span=period, adjust=False).mean()
df["+di"] = 100 * pdm_s / atr_s
df["-di"] = 100 * ndm_s / atr_s
dx = 100 * abs(df["+di"] - df["-di"]) / (df["+di"] + df["-di"])
return dx.ewm(span=period, adjust=False).mean()
# ============================================
# TREND CONFIRMATION
# ============================================
def confirm_buy_trend(df, atr):
ema50 = df["close"].ewm(span=EMA_FAST, adjust=False).mean()
ema200 = df["close"].ewm(span=EMA_SLOW, adjust=False).mean()
adx = calculate_adx(df, ADX_PERIOD)
price = df["close"].iloc[-1]
rsi = df["rsi"].iloc[-1]
e50 = ema50.iloc[-1]
e200 = ema200.iloc[-1]
adx_v = adx.iloc[-1]
slope = ema50.iloc[-1] - ema50.iloc[-3]
recent_high = df["high"].rolling(20).max().iloc[-1]
near_resist = (recent_high - price) < atr * 0.5
results = {
"bullish_trend": e50 > e200,
"strong_trend": adx_v > ADX_MIN_TREND,
"price_above_ema": price > e50,
"not_near_resist": not near_resist,
"rsi_ok": rsi < 60,
"ema_rising": slope > 0,
}
return all(results.values()), results
def confirm_sell_trend(df, atr):
ema50 = df["close"].ewm(span=EMA_FAST, adjust=False).mean()
ema200 = df["close"].ewm(span=EMA_SLOW, adjust=False).mean()
adx = calculate_adx(df, ADX_PERIOD)
price = df["close"].iloc[-1]
rsi = df["rsi"].iloc[-1]
e50 = ema50.iloc[-1]
e200 = ema200.iloc[-1]
adx_v = adx.iloc[-1]
slope = ema50.iloc[-1] - ema50.iloc[-3]
recent_low = df["low"].rolling(20).min().iloc[-1]
near_support = (price - recent_low) < atr * 0.5
results = {
"bearish_trend": e50 < e200,
"strong_trend": adx_v > ADX_MIN_TREND,
"price_below_ema": price < e50,
"not_near_support": not near_support,
"rsi_ok": rsi > 40,
"ema_declining": slope < 0,
}
return all(results.values()), results
def confirm_trend(df, direction, atr):
if direction == "BUY":
return confirm_buy_trend(df, atr)
return confirm_sell_trend(df, atr)
# ============================================
# REGIME DETECTION
# ============================================
def is_choppy_market(df, lookback=8):
recent = df.iloc[-lookback:]
small_body_count = 0
for _, candle in recent.iterrows():
cr = candle["high"] - candle["low"]
bs = abs(candle["close"] - candle["open"])
if cr > 0 and bs / cr < CHOP_BODY_RATIO:
small_body_count += 1
if small_body_count >= CHOP_MAX_COUNT:
return True, f"⚠️ CHOPPY"
overlap_count = 0
for i in range(1, min(5, len(recent))):
curr = recent.iloc[i]
prev = recent.iloc[i-1]
overlap = min(curr["high"], prev["high"]) - max(curr["low"], prev["low"])
cr = curr["high"] - curr["low"]
if cr > 0 and overlap / cr > 0.7:
overlap_count += 1
if overlap_count >= 3:
return True, f"⚠️ CHOPPY (overlap)"
return False, "✅ Clean"
def detect_market_regime(df):
is_chop, _ = is_choppy_market(df)
if is_chop:
return "CHOPPY", 0, 1.0
adx = calculate_adx(df, ADX_PERIOD)
atr = df["atr"].iloc[-1]
atr_avg = df["atr"].iloc[-20:].mean()
adx_val = adx.iloc[-1]
atr_r = atr / atr_avg if atr_avg > 0 else 1.0
if atr_r > 1.8:
return "HIGH_VOLATILITY", adx_val, atr_r
if adx_val >= 25:
return "TRENDING", adx_val, atr_r
if adx_val < 20:
return "RANGING", adx_val, atr_r
return "RANGING", adx_val, atr_r
# ============================================
# MACD BIAS
# ============================================
def get_macd_bias(macd_now, signal_now):
if macd_now > signal_now:
return "BUY"
elif macd_now < signal_now:
return "SELL"
return None
# ============================================
# SESSION FILTERS
# ============================================
def is_trading_session():
hour = datetime.utcnow().hour
if SESSION_START <= hour < SESSION_END:
return True, f"✅ Active ({hour:02d}:00 UTC)"
hours_left = (SESSION_START - hour) % 24
return False, f"š“ Closed ~{hours_left}h"
def is_high_quality_session():
hour = datetime.utcnow().hour
if 7 <= hour <= 10:
return True, "š London open"
elif 13 <= hour <= 17:
return True, "š„ London/NY overlap"
elif 10 < hour < 12:
return True, "š Mid-London"
elif 17 < hour <= 19:
return True, "š NY afternoon"
return False, f"⚠️ Low quality ({hour:02d}:00 UTC)"
def passes_volatility_filter(symbol, atr):
threshold = ATR_MIN_THRESHOLD.get(symbol, 0.00030)
return (True, "✅ ATR OK") if atr >= threshold else (False, "⏸️ ATR low")
def passes_spread_filter(symbol):
tick = mt5.symbol_info_tick(symbol)
costs = TRADING_COSTS.get(symbol, TRADING_COSTS["EURUSDm"])
if tick is None:
return False, "No tick"
if (tick.ask - tick.bid) <= costs["spread"] * MAX_SPREAD_MULTIPLIER:
return True, "✅ Spread OK"
return False, "š« Spread high"
# ============================================
# ENTRY LOGIC
# ============================================
def get_pullback_signal(df, direction):
ema20 = df["close"].ewm(span=EMA_PULLBACK, adjust=False).mean()
latest = df.iloc[-1]
atr = latest["atr"]
ema_val = ema20.iloc[-1]
distance = abs(latest["close"] - ema_val)
near_ema = distance <= atr * 0.3
cr = latest["high"] - latest["low"]
bs = abs(latest["close"] - latest["open"])
strong = bs / cr >= 0.55 if cr > 0 else False
if direction == "BUY":
if near_ema and latest["close"] > latest["open"] and strong and latest["close"] > ema_val:
return "BUY", (bs / cr) * 100, "PULLBACK"
elif direction == "SELL":
if near_ema and latest["close"] < latest["open"] and strong and latest["close"] < ema_val:
return "SELL", (bs / cr) * 100, "PULLBACK"
return None, 0, None
def get_breakout_signal(df, direction):
latest = df.iloc[-1]
prev = df.iloc[-2]
cr = latest["high"] - latest["low"]
bs = abs(latest["close"] - latest["open"])
br = bs / cr if cr > 0 else 0
strong = br >= 0.6
if direction == "BUY":
if latest["close"] > prev["high"] and latest["close"] > latest["open"] and strong:
return "BUY", br * 100, "BREAKOUT"
elif direction == "SELL":
if latest["close"] < prev["low"] and latest["close"] < latest["open"] and strong:
return "SELL", br * 100, "BREAKOUT"
return None, 0, None
def get_best_entry(df, direction, regime):
settings = REGIME_SETTINGS[regime]
if settings["use_pullback"]:
d, s, t = get_pullback_signal(df, direction)
if d:
return d, s, t
if settings["use_breakout"]:
d, s, t = get_breakout_signal(df, direction)
if d:
return d, s, t
return None, 0, None
# ============================================
# CONSECUTIVE LOSS GUARD
# ============================================
def check_consecutive_losses(symbol, max_losses=3):
if not os.path.exists(TRADES_FILE):
return False, ""
df = pd.read_csv(TRADES_FILE)
if df.empty:
return False, ""
today = datetime.now().strftime("%Y-%m-%d")
sym_df = df[df["symbol"] == symbol]
today_df = sym_df[sym_df["time"].str.startswith(today)]
closed = today_df[today_df["result"] != "OPEN"]
if len(closed) < max_losses:
return False, ""
last_n = closed.tail(max_losses)["result"].tolist()
if all(r == "LOSS" for r in last_n):
return True, f"⛔ {symbol}: {max_losses} consecutive losses — paused!"
return False, ""
# ============================================
# RSI DIVERGENCE
# ============================================
def detect_rsi_divergence(df, lookback=10):
recent = df.iloc[-lookback:]
prices = recent["close"].values
rsis = recent["rsi"].values
price_lows = [(i, prices[i]) for i in range(1, len(prices)-1) if prices[i] < prices[i-1] and prices[i] < prices[i+1]]
rsi_lows = [(i, rsis[i]) for i in range(1, len(rsis)-1) if rsis[i] < rsis[i-1] and rsis[i] < rsis[i+1]]
price_highs = [(i, prices[i]) for i in range(1, len(prices)-1) if prices[i] > prices[i-1] and prices[i] > prices[i+1]]
rsi_highs = [(i, rsis[i]) for i in range(1, len(rsis)-1) if rsis[i] > rsis[i-1] and rsis[i] > rsis[i+1]]
if len(price_lows) >= 2 and len(rsi_lows) >= 2:
if price_lows[-1][1] < price_lows[-2][1] and rsi_lows[-1][1] > rsi_lows[-2][1]:
return "BULLISH"
if len(price_highs) >= 2 and len(rsi_highs) >= 2:
if price_highs[-1][1] > price_highs[-2][1] and rsi_highs[-1][1] < rsi_highs[-2][1]:
return "BEARISH"
return None
# ============================================
# TRADE QUALITY MEMORY
# ============================================
def load_trade_memory():
if os.path.exists(MEMORY_FILE):
with open(MEMORY_FILE, "r") as f:
return json.load(f)
return {"regime_stats": {}, "entry_stats": {}, "pair_stats": {}}
def save_trade_memory(memory):
with open(MEMORY_FILE, "w") as f:
json.dump(memory, f, indent=2)
def update_trade_memory(regime, entry_type, symbol, result):
memory = load_trade_memory()
for key, category in [
(regime, "regime_stats"),
(entry_type, "entry_stats"),
(symbol, "pair_stats"),
]:
if key not in memory[category]:
memory[category][key] = {"wins": 0, "total": 0, "win_rate": 0}
memory[category][key]["total"] += 1
if result == "WIN":
memory[category][key]["wins"] += 1
t = memory[category][key]["total"]
w = memory[category][key]["wins"]
memory[category][key]["win_rate"] = (w / t * 100) if t > 0 else 0
save_trade_memory(memory)
def should_skip_due_to_memory(regime, entry_type, symbol, min_wr=40):
memory = load_trade_memory()
for key, category in [
(regime, "regime_stats"),
(entry_type, "entry_stats"),
]:
stats = memory[category].get(key, {})
total = stats.get("total", 0)
wr = stats.get("win_rate", 50)
if total >= 10 and wr < min_wr:
return True, f"Memory: {key} WR={wr:.1f}%"
return False, ""
def show_trade_memory():
memory = load_trade_memory()
for category, label in [
("regime_stats", "Regime"),
("entry_stats", "Entry"),
("pair_stats", "Pair"),
]:
for key, stats in memory[category].items():
wr = stats.get("win_rate", 0)
t = stats.get("total", 0)
if t > 0:
print(f" {label}/{key}: {t} | {wr:.1f}% WR")
# ============================================
# DYNAMIC SLIPPAGE
# ============================================
def load_slippage_data():
if os.path.exists(SLIPPAGE_FILE):
with open(SLIPPAGE_FILE, "r") as f:
return json.load(f)
return {sym: [] for sym in SYMBOLS}
def save_slippage_data(data):
with open(SLIPPAGE_FILE, "w") as f:
json.dump(data, f, indent=2)
def record_slippage(symbol, requested_price, filled_price):
data = load_slippage_data()
if symbol not in data:
data[symbol] = []
data[symbol].append(abs(filled_price - requested_price))
data[symbol] = data[symbol][-20:]
save_slippage_data(data)
def get_dynamic_slippage(symbol):
data = load_slippage_data()
slips = data.get(symbol, [])
if len(slips) >= 5:
return (sum(slips) / len(slips)) * 1.2
return TRADING_COSTS.get(symbol, TRADING_COSTS["EURUSDm"])["slippage"]
def get_dynamic_sl_tp_buffer(symbol):
data = load_slippage_data()
slips = data.get(symbol, [])
if len(slips) >= 5:
return (sum(slips) / len(slips)) * 2
return 0
# ============================================
# ML FEATURES + TRAINING
# ============================================
def extract_ml_features(df, regime, structure):
latest = df.iloc[-1]
recent = df.iloc[-20:]
vol_now = df["close"].pct_change().rolling(5).std().iloc[-1]
vol_avg = df["close"].pct_change().rolling(20).std().iloc[-1]
vol_ratio = vol_now / vol_avg if vol_avg > 0 else 1.0
sma20 = df["close"].rolling(20).mean().iloc[-1]
ts = (latest["close"] - sma20) / sma20 * 100
atr_n = latest["atr"] / latest["close"] * 100
avg_vol = recent["tick_volume"].mean()
vol_p = latest["tick_volume"] / avg_vol if avg_vol > 0 else 1.0
cr = latest["high"] - latest["low"]
bs = abs(latest["close"] - latest["open"])
br = bs / cr if cr > 0 else 0.5
adx_v = calculate_adx(df, ADX_PERIOD).iloc[-1]
rm = {"TRENDING": 0, "RANGING": 1, "HIGH_VOLATILITY": 2, "CHOPPY": 3}.get(regime, 1)
pp = structure.get("price_position", 0.5)
return np.array([[vol_ratio, ts, atr_n, pp, vol_p, br, adx_v, rm]])
def label_trade_quality_directional(row, future, atr, direction):
sl_dist = atr * 1.0
tp_dist = atr * 1.5
entry = row["close"]
if direction == "BUY":
sl = entry - sl_dist
tp = entry + tp_dist
for _, c in future.iterrows():
if c["low"] <= sl:
return 0
if c["high"] >= tp:
return 1
return 0
else:
sl = entry + sl_dist
tp = entry - tp_dist
for _, c in future.iterrows():
if c["high"] >= sl:
return 0
if c["low"] <= tp:
return 1
return 0
def build_training_data(df, symbol="EURUSDm"):
costs = TRADING_COSTS.get(symbol, TRADING_COSTS["EURUSDm"])
total_cost = costs["spread"] + costs["slippage"]
records = []
for i in range(50, len(df) - 20):
row = df.iloc[i]
future = df.iloc[i+1: i+20]
recent = df.iloc[i-20: i]
if pd.isna(row["rsi"]) or pd.isna(row["atr"]):
continue
atr = row["atr"]
vol_now = df["close"].pct_change().rolling(5).std().iloc[i]
vol_avg = df["close"].pct_change().rolling(20).std().iloc[i]
vr = vol_now / vol_avg if vol_avg and vol_avg > 0 else 1.0
sma20 = recent["close"].mean()
ts = (row["close"] - sma20) / sma20 * 100
atr_n = atr / row["close"] * 100
hs = df["high"].iloc[i-50:i].max() if i >= 50 else recent["high"].max()
ls = df["low"].iloc[i-50:i].min() if i >= 50 else recent["low"].min()
rng = hs - ls
pp = (row["close"] - ls) / rng if rng > 0 else 0.5
avg_vol = recent["tick_volume"].mean()
vp = row["tick_volume"] / avg_vol if avg_vol > 0 else 1.0
cr = row["high"] - row["low"]
bs = abs(row["close"] - row["open"])
br = bs / cr if cr > 0 else 0.5
adx_s = calculate_adx(df.iloc[:i+1], ADX_PERIOD)
adx_v = adx_s.iloc[-1]
atr_r = atr / df["atr"].iloc[i-20:i].mean() if df["atr"].iloc[i-20:i].mean() > 0 else 1
hr = 3 if atr_r > 1.8 else (0 if adx_v >= 25 else (1 if adx_v < 20 else 2))
direction = "BUY" if row["macd"] > row["macd_signal"] else "SELL"
good_trade = label_trade_quality_directional(row, future, atr, direction)
bm = 0 if (vr < 0.8 and abs(ts) < 0.1) else 1
ms = 0 if abs(ts) > 0.15 else (1 if vr < 0.7 else 2)
records.append({
"vol_ratio": vr, "trend_strength": ts, "atr_norm": atr_n,
"price_pos": pp, "vol_pressure": vp, "body_ratio": br,
"adx_val": adx_v, "regime_num": hr,
"best_mode": bm, "mgmt_style": ms, "good_trade": good_trade,
})
return pd.DataFrame(records)
def walk_forward_validate(X, y, model_class, params):
n = len(X)
train_e = int(n * 0.70)
val_e = int(n * 0.85)
model = model_class(**params)
model.fit(X[:train_e], y[:train_e])
tr = model.score(X[:train_e], y[:train_e]) * 100
v = model.score(X[train_e:val_e], y[train_e:val_e]) * 100
te = model.score(X[val_e:], y[val_e:]) * 100
return model, tr, v, te
def train_all_models(df, symbol, force=False):
print(f" š§ Training ML for {symbol}...")
df = df.copy()
df["rsi"] = calculate_rsi(df, RSI_PERIOD)
df["macd"], df["macd_signal"], df["macd_hist"] = calculate_macd(df, MACD_FAST, MACD_SLOW, MACD_SIGNAL)
df["atr"] = calculate_atr(df, ATR_PERIOD)
training_data = build_training_data(df, symbol)
if len(training_data) < 50:
return None, None, None, None
feature_cols = ["vol_ratio", "trend_strength", "atr_norm", "price_pos", "vol_pressure", "body_ratio", "adx_val", "regime_num"]
X = training_data[feature_cols].values
y_mode = training_data["best_mode"].values
y_mgmt = training_data["mgmt_style"].values
y_filter = training_data["good_trade"].values
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)
params = {"n_estimators": 50, "max_depth": 4, "min_samples_leaf": 5, "random_state": 42}
model_mode, tr_m, v_m, te_m = walk_forward_validate(X_scaled, y_mode, RandomForestClassifier, params)
model_mgmt, tr_g, v_g, te_g = walk_forward_validate(X_scaled, y_mgmt, RandomForestClassifier, params)
model_filter, tr_f, v_f, te_f = walk_forward_validate(X_scaled, y_filter, RandomForestClassifier, params)
print(f" š Risk:{tr_m:.0f}/{v_m:.0f}/{te_m:.0f}% Mgmt:{tr_g:.0f}/{v_g:.0f}/{te_g:.0f}% Filter:{tr_f:.0f}/{v_f:.0f}/{te_f:.0f}%")
if te_f < 50 and not force:
existing = models.get(symbol)
if existing and existing.get("model_filter"):
print(f" ⚠️ Keeping existing model")
return (existing["model_mode"], existing["model_mgmt"], existing["model_filter"], existing["scaler"])
existing = models.get(symbol)
if existing and existing.get("model_filter"):
fallback_models[symbol] = existing.copy()
with open(MODEL_LOG, "a") as f:
f.write(f"{datetime.now()} - {symbol} | Filter:{te_f:.1f}%\n")
print(f" ✅ {symbol} ML ready!")
return model_mode, model_mgmt, model_filter, scaler
def predict_risk_mode(model_mode, scaler, features):
fs = scaler.transform(features)
pred = model_mode.predict(fs)[0]
prob = model_mode.predict_proba(fs)[0]
return ("BALANCED" if pred == 1 else "LOW"), max(prob) * 100
def predict_mgmt_style(model_mgmt, scaler, features):
fs = scaler.transform(features)
pred = model_mgmt.predict(fs)[0]
prob = model_mgmt.predict_proba(fs)[0]
return MGMT_STYLES[pred], max(prob) * 100
def ml_trade_filter(model_filter, scaler, features, regime):
settings = REGIME_SETTINGS[regime]
min_conf = settings["min_conf"]
fs = scaler.transform(features)
pred = model_filter.predict(fs)[0]
prob = model_filter.predict_proba(fs)[0]
conf = max(prob) * 100
return (True, conf) if pred == 1 and conf >= min_conf else (False, conf)
# ============================================
# ATR SL/TP
# ============================================
def calculate_atr_sl_tp(price, direction, atr, risk_mode, regime, structure, symbol):
settings = REGIME_SETTINGS[regime]
slip_buffer = get_dynamic_sl_tp_buffer(symbol)
sl_dist = atr * settings["atr_sl_mult"] + slip_buffer
tp_dist = atr * settings["atr_tp_mult"] + slip_buffer
if direction == "BUY":
sl = round(price - sl_dist, 5)
tp = round(price + tp_dist, 5)
sw = structure.get("swing_high", tp)
if price < sw < tp:
tp = round(sw, 5)
else:
sl = round(price + sl_dist, 5)
tp = round(price - tp_dist, 5)
sw = structure.get("swing_low", tp)
if tp < sw < price:
tp = round(sw, 5)
return sl, tp
# ============================================
# POSITION SIZING
# ============================================
def get_account_info():
info = mt5.account_info()
if info is None:
return None, None
return info.balance, info.equity
def calculate_lot_size(symbol, sl_price, entry_price, risk_mode):
balance, _ = get_account_info()
if balance is None:
return 0.01
profile = RISK_PROFILES[risk_mode]
risk_pct = profile["risk_pct"]
risk_amount = balance * risk_pct
costs = TRADING_COSTS.get(symbol, TRADING_COSTS["EURUSDm"])
sl_pips = abs(entry_price - sl_price) / costs["pip_size"]
if sl_pips == 0:
return 0.01
lot_size = risk_amount / (sl_pips * costs["pip_value"])
return max(0.01, min(1.00, round(lot_size, 2)))
# ============================================
# RISK CONTROL
# ============================================
def check_risk_limits():
global peak_balance, day_start_bal
balance, equity = get_account_info()
if balance is None:
return True, ""
if peak_balance is None:
peak_balance = balance
if day_start_bal is None:
day_start_bal = balance
if balance > peak_balance:
peak_balance = balance
dd = (peak_balance - equity) / peak_balance
dl = (day_start_bal - equity) / day_start_bal
if dd >= MAX_DRAWDOWN_PCT:
return False, f"šØ DRAWDOWN {dd*100:.1f}%"
if dl >= MAX_DAILY_LOSS:
return False, f"šØ DAILY LOSS {dl*100:.1f}%"
return True, ""
def show_risk_status():
global peak_balance, day_start_bal
balance, equity = get_account_info()
if balance is None:
return
dd = ((peak_balance - equity) / peak_balance * 100 if peak_balance else 0)
dl = ((day_start_bal - equity) / day_start_bal * 100 if day_start_bal else 0)
print(f" š¼ ${balance:.2f} | DD:{dd:.1f}% | DL:{dl:.1f}%")
# ============================================
# NEWS FILTER
# ============================================
def fetch_news_events():
try:
now = datetime.now()
date_str = now.strftime("%Y-%m-%d")
url = "https://nfs.faireconomy.media/ff_calendar_thisweek.json"
headers = {"User-Agent": "Mozilla/5.0"}
response = requests.get(url, headers=headers, timeout=10)
if response.status_code != 200:
return []
data = response.json()
events = []
for item in data:
impact = item.get("impact", "").lower()
currency = item.get("country", "").upper()
date = item.get("date", "")
if impact != "high":
continue
if date_str not in date:
continue
if currency not in NEWS_CURRENCIES:
continue
events.append({
"currency": currency,
"event": item.get("title", "Unknown"),
"time": item.get("date", ""),
"date": date_str,
})
print(f" š° Found {len(events)} high impact events today")
return events
except Exception as e:
print(f" ⚠️ News fetch error: {e}")
return []
def parse_news_time(time_str, date_str):
try:
if not time_str:
return None
return datetime.fromisoformat(time_str).replace(tzinfo=None)
except Exception:
return None
def is_news_time(events):
now = datetime.now()
for event in events:
event_dt = parse_news_time(event["time"], event["date"])
if not event_dt:
continue
mins_until = (event_dt - now).total_seconds() / 60
mins_since = (now - event_dt).total_seconds() / 60
if 0 <= mins_until <= NEWS_PAUSE_BEFORE:
return True, f"NEWS in {mins_until:.0f} mins: {event['currency']} {event['event']}"
if 0 <= mins_since <= NEWS_PAUSE_AFTER:
return True, f"Post-NEWS ({mins_since:.0f} mins ago)"
return False, ""
def log_news_pause(reason):
with open(NEWS_LOG, "a") as f:
f.write(f"{datetime.now()} - PAUSED: {reason}\n")
# ============================================
# STACKING + POSITIONS
# ============================================
def get_open_positions(symbol):
positions = mt5.positions_get(symbol=symbol)
return list(positions) if positions else []
def count_open_trades(symbol):
return len(get_open_positions(symbol))
def all_positions_profitable(symbol):
positions = get_open_positions(symbol)
if not positions:
return False
return all(pos.profit > 0 for pos in positions)
def get_stack_direction(symbol):
positions = get_open_positions(symbol)
if not positions:
return None
return "BUY" if positions[0].type == 0 else "SELL"
def detect_reversal(df, stack_direction):
latest = df.iloc[-1]
if stack_direction == "BUY" and latest["rsi"] > 65 and latest["macd"] < latest["macd_signal"]:
return True
if stack_direction == "SELL" and latest["rsi"] < 35 and latest["macd"] > latest["macd_signal"]:
return True
return False
def close_all_positions(symbol):
for pos in get_open_positions(symbol):
tick = mt5.symbol_info_tick(symbol)
if pos.type == 0:
price = tick.bid
order = mt5.ORDER_TYPE_SELL
else:
price = tick.ask
order = mt5.ORDER_TYPE_BUY
result = mt5.order_send({
"action": mt5.TRADE_ACTION_DEAL, "symbol": symbol,
"volume": pos.volume, "type": order, "position": pos.ticket,
"price": price, "deviation": 10, "magic": 123456,
"comment": "Close All Stack", "type_time": mt5.ORDER_TIME_GTC,
"type_filling": mt5.ORDER_FILLING_FOK,
})
if result.retcode == mt5.TRADE_RETCODE_DONE:
print(f" š Closed {pos.ticket} ({symbol})")
# ============================================
# SMART TRADE MANAGEMENT
# ============================================
def manage_open_trades():
for symbol in SYMBOLS:
for pos in get_open_positions(symbol):
ticket = pos.ticket
direction = "BUY" if pos.type == 0 else "SELL"
entry = pos.price_open
current = pos.price_current
sl = pos.sl
tp = pos.tp
volume = pos.volume
mgmt_info = managed_trades.get(ticket, {})
style = mgmt_info.get("style", "BREAK_EVEN")
partial_done = mgmt_info.get("partial_done", False)
be_done = mgmt_info.get("be_done", False)
tp_dist = abs(tp - entry)
if tp_dist == 0:
continue
progress = abs(current - entry) / tp_dist
if style == "TRAILING_SL":
if direction == "BUY" and current > entry:
new_sl = round(current - (tp_dist * 0.3), 5)
if new_sl > sl:
modify_sl(ticket, new_sl)
print(f" š TRAIL → {new_sl}")
elif direction == "SELL" and current < entry:
new_sl = round(current + (tp_dist * 0.3), 5)
if new_sl < sl:
modify_sl(ticket, new_sl)
print(f" š TRAIL → {new_sl}")
elif style == "BREAK_EVEN":
if not be_done and progress >= BE_TRIGGER:
new_sl = round(entry, 5)
if (direction == "BUY" and new_sl > sl) or (direction == "SELL" and new_sl < sl):
modify_sl(ticket, new_sl)
managed_trades[ticket]["be_done"] = True
print(f" šÆ BE → {new_sl}")
elif style == "PARTIAL_CLOSE":
if not partial_done and progress >= PC_TRIGGER:
close_vol = round(volume * PC_VOLUME, 2)
if close_vol >= 0.01:
partial_close(pos, close_vol, symbol)
managed_trades[ticket]["partial_done"] = True
print(f" š° PARTIAL {close_vol}")
def modify_sl(ticket, new_sl):
mt5.order_send({"action": mt5.TRADE_ACTION_SLTP, "position": ticket, "sl": new_sl})
def partial_close(pos, volume, symbol):
tick = mt5.symbol_info_tick(symbol)
price = tick.bid if pos.type == 0 else tick.ask
order = mt5.ORDER_TYPE_SELL if pos.type == 0 else mt5.ORDER_TYPE_BUY
mt5.order_send({
"action": mt5.TRADE_ACTION_DEAL, "symbol": symbol,
"volume": volume, "type": order, "position": pos.ticket,
"price": price, "deviation": 10, "magic": 123456,
"comment": "Partial Close", "type_time": mt5.ORDER_TIME_GTC,
"type_filling": mt5.ORDER_FILLING_FOK,
})
# ============================================
# TRADE LOGGING
# ============================================
def log_trade(symbol, direction, price, sl, tp, rsi, macd,
risk_mode, confidence, mgmt_style, lot_size,
regime, entry_type, smc_context, stack_num=1):
new_row = pd.DataFrame([{
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"symbol": symbol,
"direction": direction,
"price": price,
"sl": sl,
"tp": tp,
"lot_size": lot_size,
"rsi": round(rsi, 2),
"macd": round(macd, 6),
"risk_mode": risk_mode,
"confidence": round(confidence, 1),
"mgmt_style": mgmt_style,
"regime": regime,
"entry_type": entry_type,
"smc_context": smc_context,
"stack_num": stack_num,
"result": "OPEN",
"pnl": 0.0,
}])
file_exists = os.path.exists(TRADES_FILE)
new_row.to_csv(TRADES_FILE, mode="a", header=not file_exists, index=False)
print(f" š {symbol}|{regime}|{entry_type}|{smc_context}|Lot:{lot_size}")
def update_closed_trades():
if not os.path.exists(TRADES_FILE):
return
df = pd.read_csv(TRADES_FILE)
if df.empty:
return
deals = mt5.history_deals_get(
datetime(2020, 1, 1, tzinfo=timezone.utc),
datetime.now(timezone.utc)
)
if deals is None or len(deals) == 0:
return
deals_df = pd.DataFrame(list(deals), columns=deals[0]._asdict().keys())
deals_df = deals_df[deals_df["entry"] == 1]
for idx, row in df[df["result"] == "OPEN"].iterrows():
sym_deals = deals_df[deals_df["symbol"] == row["symbol"]]
matched = sym_deals[sym_deals["comment"].str.contains("ML RSI|Close All", na=False)]
if not matched.empty:
last = matched.iloc[-1]
pnl = last["profit"]
result = "WIN" if pnl > 0 else "LOSS"
df.at[idx, "result"] = result
df.at[idx, "pnl"] = pnl
update_trade_memory(row.get("regime", "RANGING"), row.get("entry_type", "BREAKOUT"), row["symbol"], result)
record_slippage(row["symbol"], row["price"], last["price"])
df.to_csv(TRADES_FILE, index=False)
def show_stats():
if not os.path.exists(TRADES_FILE):
return
df = pd.read_csv(TRADES_FILE)
if df.empty:
return
closed = df[df["result"] != "OPEN"]
total = len(closed)
wins = len(closed[closed["result"] == "WIN"])
losses = len(closed[closed["result"] == "LOSS"])
pnl = closed["pnl"].sum()
winrate = (wins / total * 100) if total > 0 else 0
print(f"\n{'='*50}")
print(f" š Total:{total} WR:{winrate:.1f}% P&L:${pnl:.2f}")
for sym in SYMBOLS:
s = closed[closed["symbol"] == sym]
sw = len(s[s["result"] == "WIN"])
st = len(s)
sr = (sw / st * 100) if st > 0 else 0
sp = s["pnl"].sum()
print(f" {sym}: {st} | {sr:.1f}% | ${sp:.2f}")
if "smc_context" in closed.columns:
for ctx in closed["smc_context"].unique():
c = closed[closed["smc_context"] == ctx]
cw = len(c[c["result"] == "WIN"])
ct = len(c)
cr = (cw / ct * 100) if ct > 0 else 0
print(f" SMC/{ctx}: {ct} | {cr:.1f}%")
show_risk_status()
show_trade_memory()
print(f"{'='*50}\n")
with open(SUMMARY_FILE, "w") as f:
f.write(f"SUMMARY {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
f.write(f"Total:{total} WR:{winrate:.1f}% P&L:${pnl:.2f}\n")
# ============================================
# PLACE TRADE
# ============================================
def get_market_structure(df, lookback=50):
recent = df.iloc[-lookback:]
swing_high = recent["high"].max()
swing_low = recent["low"].min()
today = df["time"].iloc[-1].date()
yesterday = df[df["time"].dt.date < today]
if len(yesterday) > 0:
prev_day_high = yesterday["high"].iloc[-24:].max() if len(yesterday) >= 24 else yesterday["high"].max()
prev_day_low = yesterday["low"].iloc[-24:].min() if len(yesterday) >= 24 else yesterday["low"].min()
else:
prev_day_high = swing_high
prev_day_low = swing_low
current_price = df["close"].iloc[-1]
range_size = swing_high - swing_low
price_position = (current_price - swing_low) / range_size if range_size > 0 else 0.5
return {
"swing_high": swing_high,
"swing_low": swing_low,
"prev_day_high": prev_day_high,
"prev_day_low": prev_day_low,
"price_position": price_position,
"range_size": range_size,
}
def passes_structure_filter(direction, price, structure):
pos = structure["price_position"]
if 0.4 <= pos <= 0.6:
return False, f"⏸️ Mid-range"
if direction == "BUY" and pos > 0.6:
return True, "✅ Upper range"
if direction == "SELL" and pos < 0.4:
return True, "✅ Lower range"
return False, "⏸️ Wrong structure"
def place_trade(symbol, order_type, sl, tp, rsi, macd,
risk_mode, confidence, mgmt_style, regime,
entry_type, smc_context, scale_level=1):
tick = mt5.symbol_info_tick(symbol)
slip = get_dynamic_slippage(symbol)
if order_type == "BUY":
raw_price = tick.ask
price = round(raw_price + slip, 5)
order = mt5.ORDER_TYPE_BUY
else:
raw_price = tick.bid
price = round(raw_price - slip, 5)
order = mt5.ORDER_TYPE_SELL
# Smart scaling lot size
lot_size = get_scale_in_lot(symbol, sl, price, risk_mode, scale_level)
request = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": symbol,
"volume": lot_size,
"type": order,
"price": price,
"sl": sl,
"tp": tp,
"deviation": 10,
"magic": 123456,
"comment": "ML RSI+MACD Bot",
"type_time": mt5.ORDER_TIME_GTC,
"type_filling": mt5.ORDER_FILLING_FOK,
}
result = mt5.order_send(request)
if result.retcode == mt5.TRADE_RETCODE_DONE:
print(f" ✅ {order_type} {symbol} Scale#{scale_level} [{entry_type}|{smc_context}]")
print(f" š {price:.5f} | SL:{sl} | TP:{tp} | Lot:{lot_size}")
managed_trades[result.order] = {
"style": mgmt_style, "partial_done": False, "be_done": False,
}
log_trade(symbol, order_type, price, sl, tp, rsi, macd,
risk_mode, confidence, mgmt_style, lot_size,
regime, entry_type, smc_context, scale_level)
else:
print(f" ❌ Failed! {result.retcode} - {result.comment}")
# ============================================
# FETCH & PREPARE
# ============================================
def fetch_and_prepare(symbol):
rates = mt5.copy_rates_from_pos(symbol, TIMEFRAME_M15, 0, 500)
if rates is None or len(rates) == 0:
return None
df = pd.DataFrame(rates)
df["time"] = pd.to_datetime(df["time"], unit="s")
df["rsi"] = calculate_rsi(df, RSI_PERIOD)
df["macd"], df["macd_signal"], df["macd_hist"] = calculate_macd(df, MACD_FAST, MACD_SLOW, MACD_SIGNAL)
df["atr"] = calculate_atr(df, ATR_PERIOD)
return df
# ============================================
# MAIN BOT LOOP
# ============================================
def run_bot():
global peak_balance, day_start_bal
if not mt5.initialize(login=LOGIN, password=PASSWORD, server=SERVER):
print(f"❌ Failed to connect: {mt5.last_error()}")
return
balance, _ = get_account_info()
peak_balance = balance
day_start_bal = balance
print("=" * 50)
print(" MY QUANT BOT - PHASE 18 (SMC Edition)")
print(f" Pairs : {', '.join(SYMBOLS)}")
print(f" MTF : 4H + 1H + 5M bias alignment")
print(f" SMC : FVG + BOS/CHOCH + Liq sweep")
print(f" Scaling : Smart position scaling")
print(f" Balance : ${balance:.2f}")
print("=" * 50)
print("\nš Training ML models for all pairs...")
for symbol in SYMBOLS:
df = fetch_and_prepare(symbol)
if df is not None:
result = train_all_models(df, symbol, force=True)
if result[0] is not None:
models[symbol] = {
"model_mode": result[0],
"model_mgmt": result[1],
"model_filter": result[2],
"scaler": result[3],
}
last_retrain[symbol] = datetime.now()
print("\nš° Fetching economic calendar...")
news_events = fetch_news_events()
last_news_fetch = datetime.now()
last_day = datetime.now().day
cycle_counter = 0
while True:
try:
if datetime.now().day != last_day:
day_start_bal = get_account_info()[0]
last_day = datetime.now().day
print(f" š New day — balance reset")
is_safe, risk_reason = check_risk_limits()
if not is_safe:
print(f"\n{risk_reason} — Sleeping 5 mins...")
time.sleep(300)
continue
in_session, session_msg = is_trading_session()
if not in_session:
print(f"\n{session_msg}")
time.sleep(1800)
continue
hq_session, hq_msg = is_high_quality_session()
if (datetime.now() - last_news_fetch).seconds / 60 >= 30:
news_events = fetch_news_events()
last_news_fetch = datetime.now()
dangerous, reason = is_news_time(news_events)
if dangerous:
print(f"\nš« {reason}")
log_news_pause(reason)
time.sleep(60)
continue
manage_open_trades()
print(f"\n{'='*50}")
print(f"⏰ {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} | {hq_msg}")
print(f"{'='*50}")
for symbol in SYMBOLS:
# Consecutive loss guard
consec_loss, consec_msg = check_consecutive_losses(symbol)
if consec_loss:
print(f"\n {consec_msg}")
continue
df = fetch_and_prepare(symbol)
if df is None:
continue
latest = df.iloc[-1]
rsi = latest["rsi"]
macd_now = latest["macd"]
signal_now = latest["macd_signal"]
price = latest["close"]
atr = latest["atr"]
# Basic filters
vol_ok, vol_msg = passes_volatility_filter(symbol, atr)
spread_ok, spr_msg = passes_spread_filter(symbol)
is_chop, chop_msg = is_choppy_market(df)
if is_chop or not vol_ok or not spread_ok:
continue
# MTF Bias (4H + 1H + 5M)
mtf_bias, b4h, b1h, bm5 = get_mtf_bias(symbol)
if mtf_bias == "NEUTRAL":
print(f"\n {symbol} | MTF NEUTRAL — skip")
continue
# MACD bias confirmation
macd_bias = get_macd_bias(macd_now, signal_now)
if macd_bias != mtf_bias:
print(f"\n {symbol} | MACD/{macd_bias} vs MTF/{mtf_bias} — skip")
continue
direction = mtf_bias
# Regime + Structure
regime, adx_val, _ = detect_market_regime(df)
structure = get_market_structure(df)
m = models.get(symbol)
if not m or m["model_mode"] is None:
continue
features = extract_ml_features(df, regime, structure)
risk_mode, conf = predict_risk_mode(m["model_mode"], m["scaler"], features)
mgmt_style, _ = predict_mgmt_style(m["model_mgmt"], m["scaler"], features)
open_count = count_open_trades(symbol)
stack_dir = get_stack_direction(symbol)
print(f"\n {symbol} | Open:{open_count}/{MAX_SCALE}")
print(f" {price:.5f} | RSI:{rsi:.2f} | ATR:{atr:.5f}")
print(f" MTF: 4H:{b4h} 1H:{b1h} 5M:{bm5} → {mtf_bias}")
print(f" Regime:{regime}(ADX:{adx_val:.1f}) | {chop_msg}")
# Reversal check
if stack_dir and detect_reversal(df, stack_dir):
print(f" ⚠️ REVERSAL! Closing {symbol}...")
close_all_positions(symbol)
continue
# Trend confirmation
trend_ok, trend_results = confirm_trend(df, direction, atr)
failed = [k for k, v in trend_results.items() if not v]
if failed:
print(f" ❌ Trend failed: {', '.join(failed)}")
continue
# === SMC LAYER ===
smc_context = "STANDARD"
# FVG check
fvgs = detect_fvg(df, direction)
in_fvg, fvg_data = is_price_in_fvg(price, fvgs)
if in_fvg:
print(f" ⚠️ Price inside FVG — waiting for fill")
continue
# Liquidity sweep (bonus signal)
swept, sweep_msg = detect_liquidity_sweep(df, direction)
if swept:
smc_context = "LIQ_SWEEP"
print(f" š§ {sweep_msg}")
# BOS/CHOCH
bos = detect_bos_choch(df, direction)
if bos == "BOS":
smc_context = "BOS"
print(f" š BOS confirmed!")
elif bos == "CHOCH":
print(f" ⚠️ CHOCH — potential reversal, skip")
continue
# Order block
ob_hit, ob_zone = detect_order_block(df, direction)
if ob_hit:
smc_context = "ORDER_BLOCK"
print(f" šÆ Price at Order Block!")
# Liquidity zones
liq_zones = detect_liquidity_zones(df)
liq_ok, liq_msg = passes_liquidity_filter(direction, liq_zones)
if not liq_ok:
print(f" {liq_msg}")
continue
# Structure filter
struct_ok, struct_msg = passes_structure_filter(direction, price, structure)
if not struct_ok:
print(f" {struct_msg}")
continue
# Memory check
skip_mem, mem_reason = should_skip_due_to_memory(regime, "BREAKOUT", symbol)
if skip_mem:
print(f" ⏸️ {mem_reason}")
continue
# Entry signal
entry_dir, pa_strength, entry_type = get_best_entry(df, direction, regime)
if not entry_dir:
print(f" ⏸️ No entry signal")
continue
# RSI divergence
divergence = detect_rsi_divergence(df)
if divergence and divergence != ("BULLISH" if direction == "BUY" else "BEARISH"):
print(f" ⚠️ RSI divergence conflicts — skip")
continue
# ML filter
trade_ok, filter_conf = ml_trade_filter(m["model_filter"], m["scaler"], features, regime)
if not trade_ok:
print(f" š« ML: SKIP ({filter_conf:.1f}%)")
continue
print(f" ✅ ALL CHECKS PASSED!")
print(f" SMC: {smc_context} | {entry_type} | ML:{filter_conf:.1f}%")
sl, tp = calculate_atr_sl_tp(
price, direction, atr, risk_mode, regime, structure, symbol)
# Smart scaling entry
if open_count == 0:
print(f" š¢ Initial entry Scale#1 [{smc_context}]")
place_trade(symbol, direction, sl, tp, rsi, macd_now,
risk_mode, conf, mgmt_style, regime,
entry_type, smc_context, scale_level=1)
else:
can_add, current_scale = can_scale_in(symbol, direction, atr)
if can_add and direction == stack_dir:
scale_level = current_scale + 1
print(f" š Scale-in #{scale_level} [{smc_context}]")
place_trade(symbol, direction, sl, tp, rsi, macd_now,
risk_mode, conf, mgmt_style, regime,
entry_type, smc_context,
scale_level=scale_level)
elif current_scale >= MAX_SCALE:
print(f" ⏸️ Max scale ({MAX_SCALE}) reached")
else:
print(f" ⏸️ Waiting for favorable move to scale")
cycle_counter += 1
if cycle_counter % 10 == 0:
update_closed_trades()
show_stats()
for symbol in SYMBOLS:
last_rt = last_retrain.get(symbol, datetime.min)
days_since = (datetime.now() - last_rt).days
if days_since >= RETRAIN_EVERY_DAYS:
print(f"\nš 7-day retrain for {symbol}...")
df = fetch_and_prepare(symbol)
if df is not None:
result = train_all_models(df, symbol)
if result[0] is not None:
models[symbol] = {
"model_mode": result[0],
"model_mgmt": result[1],
"model_filter": result[2],
"scaler": result[3],
}
last_retrain[symbol] = datetime.now()
print("\n š¤ Sleeping 60 seconds...")
time.sleep(60)
except KeyboardInterrupt:
print("\nš Bot stopped")
show_stats()
break
except Exception as e:
print(f"❌ Error: {e}")
time.sleep(10)
mt5.shutdown()
print("✅ Disconnected cleanly")
# --- Start ---
run_bot()
Comments
Post a Comment