mt5-chart-browser

> **Skill:** Mt5 Chart Browser | **Domain:** trading | **Category:** infrastructure | **Level:** intermediate > **Tags:** `trading`, `infrastructure`, `mt5`, `chart`, `browser`, `analysis-tools` # 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 ```python 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 ```python 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 ```python 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 ```python 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) ```python 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)