mahmoud20138 Skill: Mt5 Integration | Domain: trading | Category: infrastructure | Level: intermediate Tags:
trading,infrastructure,mt5,metatrader,scanner,ea
MT5 Chart Browser & GPU Image Analysis Skill
Overview
This skill provides a complete interface for connecting to MetaTrader 5, browsing all available symbols/pairs, pulling OHLCV data across all timeframes, computing any indicator, capturing chart images, and performing GPU-accelerated image analysis on chart screenshots for pattern recognition.
Architecture
┌─────────────────────────────────────────────────┐
│ MT5 Chart Browser │
├──────────┬──────────┬──────────┬────────────────┤
│ MT5 Conn │ Symbol │ Chart │ GPU Image │
│ Manager │ Browser │ Engine │ Analyzer │
└──────────┴──────────┴──────────┴────────────────┘
1. MT5 Connection & Symbol Browser
Connect to MT5
import MetaTrader5 as mt5
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Optional
def connect_mt5(
path: Optional[str] = None,
login: Optional[int] = None,
password: Optional[str] = None,
server: Optional[str] = None,
timeout: int = 10000
) -> bool:
"""Initialize MT5 connection. Call once per session."""
kwargs = {"timeout": timeout}
if path: kwargs["path"] = path
if login: kwargs["login"] = login
if password: kwargs["password"] = password
if server: kwargs["server"] = server
if not mt5.initialize(**kwargs):
print(f"MT5 init failed: {mt5.last_error()}")
return False
info = mt5.terminal_info()
print(f"Connected: {info.name} | Build {info.build} | {info.company}")
return True
def shutdown_mt5():
mt5.shutdown()
Browse All Available Symbols
def get_all_symbols(group: Optional[str] = None, visible_only: bool = False) -> pd.DataFrame:
"""
Get all symbols available in the broker.
group: filter like "Forex*", "Crypto*", "Index*", "*USD*"
visible_only: only symbols shown in Market Watch
"""
if group:
symbols = mt5.symbols_get(group=group)
else:
symbols = mt5.symbols_get()
if not symbols:
print(f"No symbols found. Error: {mt5.last_error()}")
return pd.DataFrame()
data = []
for s in symbols:
if visible_only and not s.visible:
continue
data.append({
"symbol": s.name,
"description": s.description,
"path": s.path,
"spread": s.spread,
"digits": s.digits,
"point": s.point,
"trade_mode": s.trade_mode,
"volume_min": s.volume_min,
"volume_max": s.volume_max,
"volume_step": s.volume_step,
"currency_base": s.currency_base,
"currency_profit": s.currency_profit,
"category": _categorize_symbol(s.path),
})
return pd.DataFrame(data)
def _categorize_symbol(path: str) -> str:
"""Auto-categorize symbol from broker path."""
p = path.lower()
if "forex" in p or "fx" in p: return "Forex"
if "crypto" in p: return "Crypto"
if "index" in p or "indices" in p: return "Index"
if "commodity" in p or "metal" in p: return "Commodity"
if "stock" in p or "share" in p: return "Stock"
if "energy" in p: return "Energy"
return "Other"
def enable_symbol(symbol: str) -> bool:
"""Make a symbol visible in Market Watch (required before data access)."""
selected = mt5.symbol_select(symbol, True)
if not selected:
print(f"Cannot enable {symbol}: {mt5.last_error()}")
return selected
def get_symbol_info(symbol: str) -> dict:
"""Full symbol specification — spread, margin, swap, session times, etc."""
info = mt5.symbol_info(symbol)
if info is None:
return {}
return info._asdict()
def get_current_price(symbol: str) -> dict:
"""Real-time bid/ask/last for a symbol."""
tick = mt5.symbol_info_tick(symbol)
if tick is None:
return {}
return {"symbol": symbol, "bid": tick.bid, "ask": tick.ask,
"last": tick.last, "volume": tick.volume, "time": tick.time}
Browse by Category
def browse_forex_pairs() -> pd.DataFrame:
return get_all_symbols(group="*Forex*")
def browse_crypto() -> pd.DataFrame:
return get_all_symbols(group="*Crypto*")
def browse_indices() -> pd.DataFrame:
return get_all_symbols(group="*Index*")
def browse_commodities() -> pd.DataFrame:
return get_all_symbols(group="*Commodity*")
def browse_by_currency(currency: str = "USD") -> pd.DataFrame:
"""All pairs containing a specific currency."""
return get_all_symbols(group=f"*{currency}*")
2. Chart Data Engine — All Timeframes & All Data
Timeframe Map
TIMEFRAMES = {
"M1": mt5.TIMEFRAME_M1, "M2": mt5.TIMEFRAME_M2,
"M3": mt5.TIMEFRAME_M3, "M4": mt5.TIMEFRAME_M4,
"M5": mt5.TIMEFRAME_M5, "M6": mt5.TIMEFRAME_M6,
"M10": mt5.TIMEFRAME_M10, "M12": mt5.TIMEFRAME_M12,
"M15": mt5.TIMEFRAME_M15, "M20": mt5.TIMEFRAME_M20,
"M30": mt5.TIMEFRAME_M30, "H1": mt5.TIMEFRAME_H1,
"H2": mt5.TIMEFRAME_H2, "H3": mt5.TIMEFRAME_H3,
"H4": mt5.TIMEFRAME_H4, "H6": mt5.TIMEFRAME_H6,
"H8": mt5.TIMEFRAME_H8, "H12": mt5.TIMEFRAME_H12,
"D1": mt5.TIMEFRAME_D1, "W1": mt5.TIMEFRAME_W1,
"MN1": mt5.TIMEFRAME_MN1,
}
def get_ohlcv(
symbol: str,
timeframe: str = "H1",
bars: int = 1000,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None
) -> pd.DataFrame:
"""
Pull OHLCV data. Supports bar count OR date range.
Returns: DataFrame with columns [time, open, high, low, close, tick_volume, spread]
"""
enable_symbol(symbol)
tf = TIMEFRAMES.get(timeframe.upper())
if tf is None:
raise ValueError(f"Unknown timeframe: {timeframe}. Use one of: {list(TIMEFRAMES.keys())}")
if start_date and end_date:
rates = mt5.copy_rates_range(symbol, tf, start_date, end_date)
elif start_date:
rates = mt5.copy_rates_from(symbol, tf, start_date, bars)
else:
rates = mt5.copy_rates_from_pos(symbol, tf, 0, bars)
if rates is None or len(rates) == 0:
print(f"No data for {symbol} {timeframe}: {mt5.last_error()}")
return pd.DataFrame()
df = pd.DataFrame(rates)
df["time"] = pd.to_datetime(df["time"], unit="s")
df.set_index("time", inplace=True)
df.rename(columns={"tick_volume": "volume"}, inplace=True)
return df
def get_ticks(
symbol: str,
start: datetime,
end: Optional[datetime] = None,
count: int = 10000,
flags: int = mt5.COPY_TICKS_ALL
) -> pd.DataFrame:
"""Raw tick data — bid/ask/last at millisecond granularity."""
enable_symbol(symbol)
if end:
ticks = mt5.copy_ticks_range(symbol, start, end, flags)
else:
ticks = mt5.copy_ticks_from(symbol, start, count, flags)
if ticks is None or len(ticks) == 0:
return pd.DataFrame()
df = pd.DataFrame(ticks)
df["time"] = pd.to_datetime(df["time"], unit="s")
return df
def multi_timeframe_snapshot(symbol: str, bars: int = 200) -> dict[str, pd.DataFrame]:
"""Pull data across all major timeframes for a single symbol."""
key_tfs = ["M5", "M15", "H1", "H4", "D1", "W1"]
return {tf: get_ohlcv(symbol, tf, bars) for tf in key_tfs}
3. Indicator Engine — Compute Any Indicator
Built-in Indicator Wrappers (vectorized, no look-ahead)
def sma(series: pd.Series, period: int) -> pd.Series:
return series.rolling(period).mean()
def ema(series: pd.Series, period: int) -> pd.Series:
return series.ewm(span=period, adjust=False).mean()
def rsi(series: pd.Series, period: int = 14) -> pd.Series:
delta = series.diff()
gain = delta.where(delta > 0, 0.0).rolling(period).mean()
loss = (-delta.where(delta < 0, 0.0)).rolling(period).mean()
rs = gain / loss.replace(0, np.nan)
return 100 - (100 / (1 + rs))
def macd(series: pd.Series, fast: int = 12, slow: int = 26, signal: int = 9) -> pd.DataFrame:
fast_ema = ema(series, fast)
slow_ema = ema(series, slow)
macd_line = fast_ema - slow_ema
signal_line = ema(macd_line, signal)
histogram = macd_line - signal_line
return pd.DataFrame({"macd": macd_line, "signal": signal_line, "histogram": histogram})
def bollinger_bands(series: pd.Series, period: int = 20, std_dev: float = 2.0) -> pd.DataFrame:
mid = sma(series, period)
std = series.rolling(period).std()
return pd.DataFrame({"upper": mid + std_dev * std, "middle": mid, "lower": mid - std_dev * std})
def atr(df: pd.DataFrame, period: int = 14) -> pd.Series:
high_low = df["high"] - df["low"]
high_close = (df["high"] - df["close"].shift(1)).abs()
low_close = (df["low"] - df["close"].shift(1)).abs()
tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
return tr.rolling(period).mean()
def stochastic(df: pd.DataFrame, k_period: int = 14, d_period: int = 3) -> pd.DataFrame:
low_min = df["low"].rolling(k_period).min()
high_max = df["high"].rolling(k_period).max()
k = 100 * (df["close"] - low_min) / (high_max - low_min).replace(0, np.nan)
d = k.rolling(d_period).mean()
return pd.DataFrame({"k": k, "d": d})
def ichimoku(df: pd.DataFrame, tenkan: int = 9, kijun: int = 26, senkou_b: int = 52) -> pd.DataFrame:
high_tenkan = df["high"].rolling(tenkan).max()
low_tenkan = df["low"].rolling(tenkan).min()
tenkan_sen = (high_tenkan + low_tenkan) / 2
high_kijun = df["high"].rolling(kijun).max()
low_kijun = df["low"].rolling(kijun).min()
kijun_sen = (high_kijun + low_kijun) / 2
senkou_a = ((tenkan_sen + kijun_sen) / 2).shift(kijun)
high_senkou = df["high"].rolling(senkou_b).max()
low_senkou = df["low"].rolling(senkou_b).min()
senkou_b_line = ((high_senkou + low_senkou) / 2).shift(kijun)
chikou = df["close"].shift(-kijun)
return pd.DataFrame({"tenkan": tenkan_sen, "kijun": kijun_sen,
"senkou_a": senkou_a, "senkou_b": senkou_b_line, "chikou": chikou})
def adx(df: pd.DataFrame, period: int = 14) -> pd.DataFrame:
plus_dm = df["high"].diff().clip(lower=0)
minus_dm = (-df["low"].diff()).clip(lower=0)
mask = plus_dm > minus_dm
plus_dm = plus_dm.where(mask, 0)
minus_dm = minus_dm.where(~mask, 0)
atr_val = atr(df, period)
plus_di = 100 * ema(plus_dm, period) / atr_val.replace(0, np.nan)
minus_di = 100 * ema(minus_dm, period) / atr_val.replace(0, np.nan)
dx = 100 * (plus_di - minus_di).abs() / (plus_di + minus_di).replace(0, np.nan)
adx_val = ema(dx, period)
return pd.DataFrame({"adx": adx_val, "plus_di": plus_di, "minus_di": minus_di})
def vwap(df: pd.DataFrame) -> pd.Series:
"""Session VWAP — requires intraday data with volume."""
tp = (df["high"] + df["low"] + df["close"]) / 3
return (tp * df["volume"]).cumsum() / df["volume"].cumsum()
def pivot_points(df: pd.DataFrame) -> pd.DataFrame:
"""Classic pivot points from previous bar's HLC."""
p = (df["high"].shift(1) + df["low"].shift(1) + df["close"].shift(1)) / 3
return pd.DataFrame({
"pivot": p, "r1": 2 * p - df["low"].shift(1), "s1": 2 * p - df["high"].shift(1),
"r2": p + (df["high"].shift(1) - df["low"].shift(1)),
"s2": p - (df["high"].shift(1) - df["low"].shift(1)),
})
# Master indicator dispatcher
INDICATORS = {
"sma": lambda df, **kw: sma(df["close"], kw.get("period", 20)),
"ema": lambda df, **kw: ema(df["close"], kw.get("period", 20)),
"rsi": lambda df, **kw: rsi(df["close"], kw.get("period", 14)),
"macd": lambda df, **kw: macd(df["close"], kw.get("fast", 12), kw.get("slow", 26), kw.get("signal", 9)),
"bollinger": lambda df, **kw: bollinger_bands(df["close"], kw.get("period", 20), kw.get("std", 2.0)),
"atr": lambda df, **kw: atr(df, kw.get("period", 14)),
"stochastic": lambda df, **kw: stochastic(df, kw.get("k", 14), kw.get("d", 3)),
"ichimoku": lambda df, **kw: ichimoku(df),
"adx": lambda df, **kw: adx(df, kw.get("period", 14)),
"vwap": lambda df, **kw: vwap(df),
"pivot": lambda df, **kw: pivot_points(df),
}
def apply_indicator(df: pd.DataFrame, name: str, **params):
"""Apply any indicator by name. Returns Series or DataFrame."""
fn = INDICATORS.get(name.lower())
if fn is None:
raise ValueError(f"Unknown indicator: {name}. Available: {list(INDICATORS.keys())}")
return fn(df, **params)
def apply_all_indicators(df: pd.DataFrame) -> pd.DataFrame:
"""Apply all indicators to a single DataFrame. For full analysis snapshots."""
result = df.copy()
result["sma_20"] = sma(df["close"], 20)
result["sma_50"] = sma(df["close"], 50)
result["ema_20"] = ema(df["close"], 20)
result["rsi_14"] = rsi(df["close"], 14)
macd_df = macd(df["close"])
result = pd.concat([result, macd_df], axis=1)
bb = bollinger_bands(df["close"])
result = pd.concat([result, bb.add_prefix("bb_")], axis=1)
result["atr_14"] = atr(df, 14)
stoch = stochastic(df)
result = pd.concat([result, stoch.add_prefix("stoch_")], axis=1)
return result
4. Chart Rendering & Screenshot Capture
Render Chart to Image (matplotlib-based, GPU-ready)
import matplotlib
matplotlib.use("Agg") # headless rendering
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from mplfinance import plot as mpf_plot
import io, base64
def render_candlestick_chart(
df: pd.DataFrame,
symbol: str,
timeframe: str,
indicators: list[str] = None,
width: int = 1920,
height: int = 1080,
save_path: Optional[str] = None
) -> str:
"""
Render a publication-quality candlestick chart with optional indicators.
Returns base64-encoded PNG for GPU image analysis or saves to file.
"""
import mplfinance as mpf
# Prepare indicator overlays
addplots = []
if indicators:
for ind_name in indicators:
ind_data = apply_indicator(df, ind_name)
if isinstance(ind_data, pd.Series):
addplots.append(mpf.make_addplot(ind_data, panel=0 if ind_name in ["sma", "ema", "bollinger"] else 2))
elif isinstance(ind_data, pd.DataFrame):
for col in ind_data.columns:
panel = 0 if "senkou" in col or "tenkan" in col or "kijun" in col else 2
addplots.append(mpf.make_addplot(ind_data[col], panel=panel, label=col))
style = mpf.make_mpf_style(base_mpf_style="nightclouds", gridstyle="", y_on_right=True)
fig, axes = mpf.plot(
df, type="candle", style=style, volume=True,
addplot=addplots if addplots else None,
title=f"{symbol} {timeframe}",
figsize=(width / 100, height / 100),
returnfig=True
)
buf = io.BytesIO()
fig.savefig(buf, format="png", dpi=100, bbox_inches="tight", facecolor="#1a1a2e")
plt.close(fig)
buf.seek(0)
if save_path:
with open(save_path, "wb") as f:
f.write(buf.read())
buf.seek(0)
return base64.b64encode(buf.read()).decode("utf-8")
def capture_mt5_chart_screenshot(symbol: str, timeframe: str, bars: int = 200) -> str:
"""Full pipeline: pull data → render → return base64 PNG."""
df = get_ohlcv(symbol, timeframe, bars)
if df.empty:
raise RuntimeError(f"No data for {symbol} {timeframe}")
return render_candlestick_chart(df, symbol, timeframe, indicators=["sma", "rsi", "macd"])
5. GPU-Accelerated Chart Image Analysis
Pattern Recognition via Image Analysis
import cv2
import torch
import torchvision.transforms as T
from PIL import Image
class ChartImageAnalyzer:
"""
GPU-accelerated chart analysis using computer vision + deep learning.
Analyzes candlestick chart screenshots for patterns, structure, and signals.
"""
def __init__(self, device: str = "auto"):
self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") if device == "auto" else torch.device(device)
self.transform = T.Compose([T.Resize((448, 448)), T.ToTensor(), T.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])])
print(f"ChartImageAnalyzer initialized on {self.device}")
def load_image(self, source: str) -> np.ndarray:
"""Load from file path or base64 string."""
if source.startswith("/") or source.startswith("."):
return cv2.imread(source)
else:
img_bytes = base64.b64decode(source)
nparr = np.frombuffer(img_bytes, np.uint8)
return cv2.imdecode(nparr, cv2.IMREAD_COLOR)
def detect_candle_colors(self, img: np.ndarray) -> dict:
"""Count bullish (green) vs bearish (red) candles via color segmentation."""
hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)
green_mask = cv2.inRange(hsv, (35, 50, 50), (85, 255, 255))
red_mask = cv2.inRange(hsv, (0, 50, 50), (15, 255, 255)) | cv2.inRange(hsv, (165, 50, 50), (180, 255, 255))
green_px = cv2.countNonZero(green_mask)
red_px = cv2.countNonZero(red_mask)
total = green_px + red_px or 1
return {
"bullish_ratio": round(green_px / total, 3),
"bearish_ratio": round(red_px / total, 3),
"dominant_bias": "bullish" if green_px > red_px else "bearish",
}
def detect_trend_lines(self, img: np.ndarray) -> list[dict]:
"""Detect straight lines using Hough Transform — potential support/resistance."""
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
edges = cv2.Canny(gray, 50, 150)
lines = cv2.HoughLinesP(edges, 1, np.pi / 180, threshold=100, minLineLength=100, maxLineGap=10)
if lines is None:
return []
results = []
for line in lines:
x1, y1, x2, y2 = line[0]
angle = np.degrees(np.arctan2(y2 - y1, x2 - x1))
length = np.sqrt((x2 - x1) ** 2 + (y2 - y1) ** 2)
results.append({"start": (x1, y1), "end": (x2, y2), "angle": round(angle, 1), "length": round(length, 1)})
return sorted(results, key=lambda x: x["length"], reverse=True)[:20]
def detect_support_resistance_zones(self, img: np.ndarray) -> list[dict]:
"""Find horizontal zones with high pixel density — likely S/R levels."""
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
h, w = gray.shape
# Horizontal projection: sum pixel intensity per row
projection = np.sum(gray, axis=1)
# Find peaks = rows with high intensity (price levels touched often)
from scipy.signal import find_peaks
peaks, properties = find_peaks(projection, height=np.percentile(projection, 90), distance=h // 20)
zones = []
for peak in peaks:
y_pct = round(1 - peak / h, 3) # Invert: top of chart = high price
zones.append({"y_pixel": int(peak), "price_level_pct": y_pct, "strength": round(projection[peak] / projection.max(), 3)})
return zones
def analyze_chart_structure(self, img: np.ndarray) -> dict:
"""Full structural analysis of a chart image."""
return {
"candle_bias": self.detect_candle_colors(img),
"trend_lines": self.detect_trend_lines(img),
"sr_zones": self.detect_support_resistance_zones(img),
"image_shape": img.shape,
}
def full_analysis(self, source: str) -> dict:
"""Complete GPU image analysis pipeline from file/base64."""
img = self.load_image(source)
if img is None:
return {"error": "Failed to load image"}
return self.analyze_chart_structure(img)
Multi-Pair Visual Scan
def visual_scan_all_pairs(
symbols: list[str],
timeframe: str = "H4",
bars: int = 200
) -> list[dict]:
"""
Screenshot and GPU-analyze every pair. Returns ranked analysis.
Use for quick visual scanning of market conditions across pairs.
"""
analyzer = ChartImageAnalyzer()
results = []
for sym in symbols:
try:
df = get_ohlcv(sym, timeframe, bars)
if df.empty:
continue
b64 = render_candlestick_chart(df, sym, timeframe)
analysis = analyzer.full_analysis(b64)
analysis["symbol"] = sym
analysis["timeframe"] = timeframe
results.append(analysis)
except Exception as e:
print(f"SKIP {sym}: {e}")
return results
6. Integration Points
This skill feeds data to other skills in the trading ecosystem:
| Downstream Skill | Data Provided |
|---|---|
pair-correlation-engine |
OHLCV data, multi-pair snapshots |
trading-data-science |
Raw data for analysis pipelines |
event-timeline-linker |
Timestamped price data for event alignment |
institutional-behavior-monitor |
Price reaction data around news events |
trading-brain |
Structured analysis results for decision-making |
Usage Conventions
- Always call
connect_mt5()first — every session starts with connection - Enable symbols before pulling data —
enable_symbol()is auto-called inget_ohlcv() - Use vectorized indicators — never loop over bars manually
- GPU analysis is optional — falls back to CPU if no CUDA device
- Chart images are base64 — can be piped directly to Claude's vision API for LLM-level analysis
- Multi-timeframe is standard — always check at least 3 timeframes before conclusions
MT5 EA Code Generator & Development
Mt5 Ea Code Generator
MT5 EA Code Generator
Generates complete, compilable MQL5 EAs from strategy specifications.
Always follows mq5-mq4-how-to-work-with conventions.
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class EASpec:
name: str
strategy_type: str # "breakout", "mean_reversion", "trend_follow", "scalp"
entry_conditions: list[str] # ["rsi < 30", "price > sma_50", "macd_cross_up"]
exit_conditions: list[str] # ["rsi > 70", "trailing_stop_hit"]
timeframe: str = "H1"
default_lot: float = 0.1
default_sl_pips: int = 50
default_tp_pips: int = 100
use_trailing_stop: bool = True
trailing_start_pips: int = 30
trailing_step_pips: int = 10
max_trades: int = 1
magic_number: int = 123456
trade_comment: str = "EA_Generated"
class EACodeGenerator:
@staticmethod
def generate(spec: EASpec) -> str:
"""Generate complete MQL5 EA source code from spec."""
indicator_handles = EACodeGenerator._generate_indicator_handles(spec.entry_conditions + spec.exit_conditions)
entry_logic = EACodeGenerator._generate_entry_logic(spec)
exit_logic = EACodeGenerator._generate_exit_logic(spec)
trailing = EACodeGenerator._generate_trailing(spec) if spec.use_trailing_stop else ""
return f'''//+------------------------------------------------------------------+
//| {spec.name}.mq5 — Auto-Generated EA
//| Strategy: {spec.strategy_type}
//+------------------------------------------------------------------+
#property copyright "Generated by Claude"
#property version "1.00"
#property strict
#include <Trade\\Trade.mqh>
// ═══ INPUTS ═══
input double LotSize = {spec.default_lot};
input int StopLoss = {spec.default_sl_pips};
input int TakeProfit = {spec.default_tp_pips};
input int MagicNumber = {spec.magic_number};
input int MaxTrades = {spec.max_trades};
{"input int TrailingStart = " + str(spec.trailing_start_pips) + ";" if spec.use_trailing_stop else ""}
{"input int TrailingStep = " + str(spec.trailing_step_pips) + ";" if spec.use_trailing_stop else ""}
// ═══ GLOBALS ═══
CTrade trade;
datetime lastBarTime = 0;
{indicator_handles}
// ═══ INIT ═══
int OnInit() {{
ChartSetInteger(0, CHART_SHOW_GRID, false);
ChartSetInteger(0, CHART_SHOW_TRADE_HISTORY, false);
trade.SetMagicNumber(MagicNumber);
// Initialize indicator handles here
return INIT_SUCCEEDED;
}}
void OnDeinit(const int reason) {{
// Release indicator handles
}}
// ═══ NEW BAR DETECTION ═══
bool IsNewBar() {{
datetime currentBarTime = iTime(_Symbol, _Period, 0);
if (currentBarTime != lastBarTime) {{
lastBarTime = currentBarTime;
return true;
}}
return false;
}}
// ═══ COUNT POSITIONS ═══
int CountPositions() {{
int count = 0;
for (int i = PositionsTotal() - 1; i >= 0; i--) {{
ulong ticket = PositionGetTicket(i);
if (PositionGetString(POSITION_SYMBOL) == _Symbol &&
PositionGetInteger(POSITION_MAGIC) == MagicNumber)
count++;
}}
return count;
}}
// ═══ MAIN TICK ═══
void OnTick() {{
if (!IsNewBar()) return;
if (CountPositions() >= MaxTrades) {{
{trailing}
return;
}}
{entry_logic}
{trailing}
}}
'''
@staticmethod
def _generate_indicator_handles(conditions: list[str]) -> str:
handles = []
if any("rsi" in c.lower() for c in conditions):
handles.append("int rsiHandle;")
if any("macd" in c.lower() for c in conditions):
handles.append("int macdHandle;")
if any("sma" in c.lower() or "ma" in c.lower() for c in conditions):
handles.append("int maHandle;")
return "\n".join(handles)
@staticmethod
def _generate_entry_logic(spec: EASpec) -> str:
return f"""
// Entry Logic ({spec.strategy_type})
double ask = SymbolInfoDouble(_Symbol, SYMBOL_ASK);
double bid = SymbolInfoDouble(_Symbol, SYMBOL_BID);
double sl_buy = ask - StopLoss * _Point;
double tp_buy = ask + TakeProfit * _Point;
double sl_sell = bid + StopLoss * _Point;
double tp_sell = bid - TakeProfit * _Point;
// TODO: Implement specific conditions from spec:
// Entry conditions: {', '.join(spec.entry_conditions)}
// Buy example:
// if (buyCondition) trade.Buy(LotSize, _Symbol, ask, sl_buy, tp_buy, "{spec.trade_comment}");
// Sell example:
// if (sellCondition) trade.Sell(LotSize, _Symbol, bid, sl_sell, tp_sell, "{spec.trade_comment}");
"""
@staticmethod
def _generate_exit_logic(spec: EASpec) -> str:
return f"// Exit conditions: {', '.join(spec.exit_conditions)}"
@staticmethod
def _generate_trailing(spec: EASpec) -> str:
return f"""
// Trailing Stop
for (int i = PositionsTotal() - 1; i >= 0; i--) {{
ulong ticket = PositionGetTicket(i);
if (PositionGetString(POSITION_SYMBOL) != _Symbol) continue;
if (PositionGetInteger(POSITION_MAGIC) != MagicNumber) continue;
double posPrice = PositionGetDouble(POSITION_PRICE_OPEN);
double posSL = PositionGetDouble(POSITION_SL);
double posTP = PositionGetDouble(POSITION_TP);
if (PositionGetInteger(POSITION_TYPE) == POSITION_TYPE_BUY) {{
double newSL = SymbolInfoDouble(_Symbol, SYMBOL_BID) - TrailingStep * _Point;
if (SymbolInfoDouble(_Symbol, SYMBOL_BID) - posPrice > TrailingStart * _Point && newSL > posSL)
trade.PositionModify(ticket, newSL, posTP);
}} else {{
double newSL = SymbolInfoDouble(_Symbol, SYMBOL_ASK) + TrailingStep * _Point;
if (posPrice - SymbolInfoDouble(_Symbol, SYMBOL_ASK) > TrailingStart * _Point && (newSL < posSL || posSL == 0))
trade.PositionModify(ticket, newSL, posTP);
}}
}}"""
Mt5 Chart Browser
MT5 Chart Browser & GPU Image Analysis Skill
Overview
This skill provides a complete interface for connecting to MetaTrader 5, browsing all available symbols/pairs, pulling OHLCV data across all timeframes, computing any indicator, capturing chart images, and performing GPU-accelerated image analysis on chart screenshots for pattern recognition.
Architecture
┌─────────────────────────────────────────────────┐
│ MT5 Chart Browser │
├──────────┬──────────┬──────────┬────────────────┤
│ MT5 Conn │ Symbol │ Chart │ GPU Image │
│ Manager │ Browser │ Engine │ Analyzer │
└──────────┴──────────┴──────────┴────────────────┘
1. MT5 Connection & Symbol Browser
Connect to MT5
import MetaTrader5 as mt5
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Optional
def connect_mt5(
path: Optional[str] = None,
login: Optional[int] = None,
password: Optional[str] = None,
server: Optional[str] = None,
timeout: int = 10000
) -> bool:
"""Initialize MT5 connection. Call once per session."""
kwargs = {"timeout": timeout}
if path: kwargs["path"] = path
if login: kwargs["login"] = login
if password: kwargs["password"] = password
if server: kwargs["server"] = server
if not mt5.initialize(**kwargs):
print(f"MT5 init failed: {mt5.last_error()}")
return False
info = mt5.terminal_info()
print(f"Connected: {info.name} | Build {info.build} | {info.company}")
return True
def shutdown_mt5():
mt5.shutdown()
Browse All Available Symbols
def get_all_symbols(group: Optional[str] = None, visible_only: bool = False) -> pd.DataFrame:
"""
Get all symbols available in the broker.
group: filter like "Forex*", "Crypto*", "Index*", "*USD*"
visible_only: only symbols shown in Market Watch
"""
if group:
symbols = mt5.symbols_get(group=group)
else:
symbols = mt5.symbols_get()
if not symbols:
print(f"No symbols found. Error: {mt5.last_error()}")
return pd.DataFrame()
data = []
for s in symbols:
if visible_only and not s.visible:
continue
data.append({
"symbol": s.name,
"description": s.description,
"path": s.path,
"spread": s.spread,
"digits": s.digits,
"point": s.point,
"trade_mode": s.trade_mode,
"volume_min": s.volume_min,
"volume_max": s.volume_max,
"volume_step": s.volume_step,
"currency_base": s.currency_base,
"currency_profit": s.currency_profit,
"category": _categorize_symbol(s.path),
})
return pd.DataFrame(data)
def _categorize_symbol(path: str) -> str:
"""Auto-categorize symbol from broker path."""
p = path.lower()
if "forex" in p or "fx" in p: return "Forex"
if "crypto" in p: return "Crypto"
if "index" in p or "indices" in p: return "Index"
if "commodity" in p or "metal" in p: return "Commodity"
if "stock" in p or "share" in p: return "Stock"
if "energy" in p: return "Energy"
return "Other"
def enable_symbol(symbol: str) -> bool:
"""Make a symbol visible in Market Watch (required before data access)."""
selected = mt5.symbol_select(symbol, True)
if not selected:
print(f"Cannot enable {symbol}: {mt5.last_error()}")
return selected
def get_symbol_info(symbol: str) -> dict:
"""Full symbol specification — spread, margin, swap, session times, etc."""
info = mt5.symbol_info(symbol)
if info is None:
return {}
return info._asdict()
def get_current_price(symbol: str) -> dict:
"""Real-time bid/ask/last for a symbol."""
tick = mt5.symbol_info_tick(symbol)
if tick is None:
return {}
return {"symbol": symbol, "bid": tick.bid, "ask": tick.ask,
"last": tick.last, "volume": tick.volume, "time": tick.time}
Browse by Category
def browse_forex_pairs() -> pd.DataFrame:
return get_all_symbols(group="*Forex*")
def browse_crypto() -> pd.DataFrame:
return get_all_symbols(group="*Crypto*")
def browse_indices() -> pd.DataFrame:
return get_all_symbols(group="*Index*")
def browse_commodities() -> pd.DataFrame:
return get_all_symbols(group="*Commodity*")
def browse_by_currency(currency: str = "USD") -> pd.DataFrame:
"""All pairs containing a specific currency."""
return get_all_symbols(group=f"*{currency}*")
2. Chart Data Engine — All Timeframes & All Data
Timeframe Map
TIMEFRAMES = {
"M1": mt5.TIMEFRAME_M1, "M2": mt5.TIMEFRAME_M2,
"M3": mt5.TIMEFRAME_M3, "M4": mt5.TIMEFRAME_M4,
"M5": mt5.TIMEFRAME_M5, "M6": mt5.TIMEFRAME_M6,
"M10": mt5.TIMEFRAME_M10, "M12": mt5.TIMEFRAME_M12,
"M15": mt5.TIMEFRAME_M15, "M20": mt5.TIMEFRAME_M20,
"M30": mt5.TIMEFRAME_M30, "H1": mt5.TIMEFRAME_H1,
"H2": mt5.TIMEFRAME_H2, "H3": mt5.TIMEFRAME_H3,
"H4": mt5.TIMEFRAME_H4, "H6": mt5.TIMEFRAME_H6,
"H8": mt5.TIMEFRAME_H8, "H12": mt5.TIMEFRAME_H12,
"D1": mt5.TIMEFRAME_D1, "W1": mt5.TIMEFRAME_W1,
"MN1": mt5.TIMEFRAME_MN1,
}
def get_ohlcv(
symbol: str,
timeframe: str = "H1",
bars: int = 1000,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None
) -> pd.DataFrame:
"""
Pull OHLCV data. Supports bar count OR date range.
Returns: DataFrame with columns [time, open, high, low, close, tick_volume, spread]
"""
enable_symbol(symbol)
tf = TIMEFRAMES.get(timeframe.upper())
if tf is None:
raise ValueError(f"Unknown timeframe: {timeframe}. Use one of: {list(TIMEFRAMES.keys())}")
if start_date and end_date:
rates = mt5.copy_rates_range(symbol, tf, start_date, end_date)
elif start_date:
rates = mt5.copy_rates_from(symbol, tf, start_date, bars)
else:
rates = mt5.copy_rates_from_pos(symbol, tf, 0, bars)
if rates is None or len(rates) == 0:
print(f"No data for {symbol} {timeframe}: {mt5.last_error()}")
return pd.DataFrame()
df = pd.DataFrame(rates)
df["time"] = pd.to_datetime(df["time"], unit="s")
df.set_index("time", inplace=True)
df.rename(columns={"tick_volume": "volume"}, inplace=True)
return df
def get_ticks(
symbol: str,
start: datetime,
end: Optional[datetime] = None,
count: int = 10000,
flags: int = mt5.COPY_TICKS_ALL
) -> pd.DataFrame:
"""Raw tick data — bid/ask/last at millisecond granularity."""
enable_symbol(symbol)
if end:
ticks = mt5.copy_ticks_range(symbol, start, end, flags)
else:
ticks = mt5.copy_ticks_from(symbol, start, count, flags)
if ticks is None or len(ticks) == 0:
return pd.DataFrame()
df = pd.DataFrame(ticks)
df["time"] = pd.to_datetime(df["time"], unit="s")
return df
def multi_timeframe_snapshot(symbol: str, bars: int = 200) -> dict[str, pd.DataFrame]:
"""Pull data across all major timeframes for a single symbol."""
key_tfs = ["M5", "M15", "H1", "H4", "D1", "W1"]
return {tf: get_ohlcv(symbol, tf, bars) for tf in key_tfs}
3. Indicator Engine — Compute Any Indicator
Built-in Indicator Wrappers (vectorized, no look-ahead)
def sma(series: pd.Series, period: int) -> pd.Series:
return series.rolling(period).mean()
def ema(series: pd.Series, period: int) -> pd.Series:
return series.ewm(span=period, adjust=False).mean()
def rsi(series: pd.Series, period: int = 14) -> pd.Series:
delta = series.diff()
gain = delta.where(delta > 0, 0.0).rolling(period).mean()
loss = (-delta.where(delta < 0, 0.0)).rolling(period).mean()
rs = gain / loss.replace(0, np.nan)
return 100 - (100 / (1 + rs))
def macd(series: pd.Series, fast: int = 12, slow: int = 26, signal: int = 9) -> pd.DataFrame:
fast_ema = ema(series, fast)
slow_ema = ema(series, slow)
macd_line = fast_ema - slow_ema
signal_line = ema(macd_line, signal)
histogram = macd_line - signal_line
return pd.DataFrame({"macd": macd_line, "signal": signal_line, "histogram": histogram})
def bollinger_bands(series: pd.Series, period: int = 20, std_dev: float = 2.0) -> pd.DataFrame:
mid = sma(series, period)
std = series.rolling(period).std()
return pd.DataFrame({"upper": mid + std_dev * std, "middle": mid, "lower": mid - std_dev * std})
def atr(df: pd.DataFrame, period: int = 14) -> pd.Series:
high_low = df["high"] - df["low"]
high_close = (df["high"] - df["close"].shift(1)).abs()
low_close = (df["low"] - df["close"].shift(1)).abs()
tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
return tr.rolling(period).mean()
def stochastic(df: pd.DataFrame, k_period: int = 14, d_period: int = 3) -> pd.DataFrame:
low_min = df["low"].rolling(k_period).min()
high_max = df["high"].rolling(k_period).max()
k = 100 * (df["close"] - low_min) / (high_max - low_min).replace(0, np.nan)
d = k.rolling(d_period).mean()
return pd.DataFrame({"k": k, "d": d})
def ichimoku(df: pd.DataFrame, tenkan: int = 9, kijun: int = 26, senkou_b: int = 52) -> pd.DataFrame:
high_tenkan = df["high"].rolling(tenkan).max()
low_tenkan = df["low"].rolling(tenkan).min()
tenkan_sen = (high_tenkan + low_tenkan) / 2
high_kijun = df["high"].rolling(kijun).max()
low_kijun = df["low"].rolling(kijun).min()
kijun_sen = (high_kijun + low_kijun) / 2
senkou_a = ((tenkan_sen + kijun_sen) / 2).shift(kijun)
high_senkou = df["high"].rolling(senkou_b).max()
low_senkou = df["low"].rolling(senkou_b).min()
senkou_b_line = ((high_senkou + low_senkou) / 2).shift(kijun)
chikou = df["close"].shift(-kijun)
return pd.DataFrame({"tenkan": tenkan_sen, "kijun": kijun_sen,
"senkou_a": senkou_a, "senkou_b": senkou_b_line, "chikou": chikou})
def adx(df: pd.DataFrame, period: int = 14) -> pd.DataFrame:
plus_dm = df["high"].diff().clip(lower=0)
minus_dm = (-df["low"].diff()).clip(lower=0)
mask = plus_dm > minus_dm
plus_dm = plus_dm.where(mask, 0)
minus_dm = minus_dm.where(~mask, 0)
atr_val = atr(df, period)
plus_di = 100 * ema(plus_dm, period) / atr_val.replace(0, np.nan)
minus_di = 100 * ema(minus_dm, period) / atr_val.replace(0, np.nan)
dx = 100 * (plus_di - minus_di).abs() / (plus_di + minus_di).replace(0, np.nan)
adx_val = ema(dx, period)
return pd.DataFrame({"adx": adx_val, "plus_di": plus_di, "minus_di": minus_di})
def vwap(df: pd.DataFrame) -> pd.Series:
"""Session VWAP — requires intraday data with volume."""
tp = (df["high"] + df["low"] + df["close"]) / 3
return (tp * df["volume"]).cumsum() / df["volume"].cumsum()
def pivot_points(df: pd.DataFrame) -> pd.DataFrame:
"""Classic pivot points from previous bar's HLC."""
p = (df["high"].shift(1) + df["low"].shift(1) + df["close"].shift(1)) / 3
return pd.DataFrame({
"pivot": p, "r1": 2 * p - df["low"].shift(1), "s1": 2 * p - df["high"].shift(1),
"r2": p + (df["high"].shift(1) - df["low"].shift(1)),
"s2": p - (df["high"].shift(1) - df["low"].shift(1)),
})
# Master indicator dispatcher
INDICATORS = {
"sma": lambda df, **kw: sma(df["close"], kw.get("period", 20)),
"ema": lambda df, **kw: ema(df["close"], kw.get("period", 20)),
"rsi": lambda df, **kw: rsi(df["close"], kw.get("period", 14)),
"macd": lambda df, **kw: macd(df["close"], kw.get("fast", 12), kw.get("slow", 26), kw.get("signal", 9)),
"bollinger": lambda df, **kw: bollinger_bands(df["close"], kw.get("period", 20), kw.get("std", 2.0)),
"atr": lambda df, **kw: atr(df, kw.get("period", 14)),
"stochastic": lambda df, **kw: stochastic(df, kw.get("k", 14), kw.get("d", 3)),
"ichimoku": lambda df, **kw: ichimoku(df),
"adx": lambda df, **kw: adx(df, kw.get("period", 14)),
"vwap": lambda df, **kw: vwap(df),
"pivot": lambda df, **kw: pivot_points(df),
}
def apply_indicator(df: pd.DataFrame, name: str, **params):
"""Apply any indicator by name. Returns Series or DataFrame."""
fn = INDICATORS.get(name.lower())
if fn is None:
raise ValueError(f"Unknown indicator: {name}. Available: {list(INDICATORS.keys())}")
return fn(
…(truncated)