134 lines
3.8 KiB
Python
134 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
📊 Market Regime Checker - Quick Status
|
|
Checks if market is Trending or Ranging
|
|
"""
|
|
|
|
import MetaTrader5 as mt
|
|
import pandas as pd
|
|
import numpy as np
|
|
from datetime import datetime
|
|
|
|
SYMBOL = "XAUUSD"
|
|
TIMEFRAME = mt.TIMEFRAME_M15
|
|
|
|
def calculate_adx(df, period=14):
|
|
"""Calculate ADX indicator"""
|
|
|
|
# True Range
|
|
df['high_low'] = df['high'] - df['low']
|
|
df['high_close'] = np.abs(df['high'] - df['close'].shift())
|
|
df['low_close'] = np.abs(df['low'] - df['close'].shift())
|
|
df['true_range'] = df[['high_low', 'high_close', 'low_close']].max(axis=1)
|
|
|
|
# Directional Movement
|
|
df['up_move'] = df['high'] - df['high'].shift()
|
|
df['down_move'] = df['low'].shift() - df['low']
|
|
|
|
df['plus_dm'] = np.where((df['up_move'] > df['down_move']) & (df['up_move'] > 0), df['up_move'], 0)
|
|
df['minus_dm'] = np.where((df['down_move'] > df['up_move']) & (df['down_move'] > 0), df['down_move'], 0)
|
|
|
|
# Smoothed values
|
|
df['atr'] = df['true_range'].rolling(window=period).mean()
|
|
df['plus_di'] = 100 * (df['plus_dm'].rolling(window=period).mean() / df['atr'])
|
|
df['minus_di'] = 100 * (df['minus_dm'].rolling(window=period).mean() / df['atr'])
|
|
|
|
# ADX
|
|
df['dx'] = 100 * np.abs(df['plus_di'] - df['minus_di']) / (df['plus_di'] + df['minus_di'])
|
|
df['adx'] = df['dx'].rolling(window=period).mean()
|
|
|
|
return df['adx'].iloc[-1]
|
|
|
|
def check_market_regime():
|
|
"""Check current market regime"""
|
|
|
|
print("=" * 70)
|
|
print(f"📊 MARKET REGIME CHECK: {SYMBOL}")
|
|
print("=" * 70)
|
|
|
|
# Initialize MT5
|
|
if not mt.initialize():
|
|
print("❌ MT5 initialization failed")
|
|
return None
|
|
|
|
# Get current price
|
|
tick = mt.symbol_info_tick(SYMBOL)
|
|
if not tick:
|
|
print("❌ Could not get price data")
|
|
mt.shutdown()
|
|
return None
|
|
|
|
current_price = tick.bid
|
|
timestamp = datetime.fromtimestamp(tick.time)
|
|
|
|
print(f"\n💹 Current Price: ${current_price:.2f}")
|
|
print(f"⏰ Time: {timestamp.strftime('%Y-%m-%d %H:%M:%S')}")
|
|
|
|
# Get historical data for ADX calculation
|
|
rates = mt.copy_rates_from_pos(SYMBOL, TIMEFRAME, 0, 100)
|
|
if rates is None or len(rates) == 0:
|
|
print("❌ Could not get historical data")
|
|
mt.shutdown()
|
|
return None
|
|
|
|
df = pd.DataFrame(rates)
|
|
df['time'] = pd.to_datetime(df['time'], unit='s')
|
|
|
|
# Calculate ADX
|
|
adx = calculate_adx(df, period=14)
|
|
|
|
# Determine regime
|
|
if adx < 25:
|
|
regime = "ranging"
|
|
can_trade = False
|
|
symbol = "🛑"
|
|
status = "RANGING MARKET"
|
|
decision = "Trading BLOCKED"
|
|
reason = "ADX < 25 = No clear trend"
|
|
advice = "Wait for trending market (ADX ≥ 25)"
|
|
else:
|
|
regime = "trending"
|
|
can_trade = True
|
|
symbol = "✅"
|
|
status = "TRENDING MARKET"
|
|
decision = "Trading ALLOWED"
|
|
reason = "ADX ≥ 25 = Strong trend"
|
|
advice = "Good conditions for trading!"
|
|
|
|
print(f"\n📈 REGIME ANALYSIS:")
|
|
print(f" Regime: {status}")
|
|
print(f" ADX: {adx:.1f}")
|
|
print(f" Status: {symbol} {regime.upper()}")
|
|
|
|
print(f"\n🎯 TRADING DECISION:")
|
|
print(f" {symbol} {decision}")
|
|
print(f" 📊 {reason}")
|
|
print(f" 💡 {advice}")
|
|
|
|
# Visual indicator
|
|
print(f"\n📊 ADX SCALE:")
|
|
print(" 0-20: Very Weak/Ranging ❌")
|
|
print(" 20-25: Weak/Ranging ⚠️")
|
|
print(" 25-40: Trending ✅")
|
|
print(" 40+: Strong Trending ✅✅")
|
|
print(f" YOUR ADX: {adx:.1f} {'━' * int(adx/2)}")
|
|
|
|
print("\n" + "=" * 70)
|
|
|
|
mt.shutdown()
|
|
|
|
return {
|
|
'regime': regime,
|
|
'adx': adx,
|
|
'can_trade': can_trade,
|
|
'price': current_price,
|
|
'timestamp': timestamp
|
|
}
|
|
|
|
if __name__ == "__main__":
|
|
result = check_market_regime()
|
|
|
|
if result:
|
|
import sys
|
|
sys.exit(0 if result['can_trade'] else 1)
|