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:

Usage Conventions

1. Always call `connect_mt5()` first — every session starts with connection
2. Enable symbols before pulling data — enable_symbol() is auto-called in get_ohlcv()
3. Use vectorized indicators — never loop over bars manually
4. GPU analysis is optional — falls back to CPU if no CUDA device
5. Chart images are base64 — can be piped directly to Claude's vision API for LLM-level analysis
6. 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(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:

Usage Conventions

1. Always call `connect_mt5()` first — every session starts with connection
2. Enable symbols before pulling data — enable_symbol() is auto-called in get_ohlcv()
3. Use vectorized indicators — never loop over bars manually
4. GPU analysis is optional — falls back to CPU if no CUDA device
5. Chart images are base64 — can be piped directly to Claude's vision API for LLM-level analysis
6. Multi-timeframe is standard — always check at least 3 timeframes before conclusions

Multi Account Manager

Multi-Account Manager

import pandas as pd
import numpy as np
from dataclasses import dataclass, field
from datetime import datetime
import json, os

@dataclass
class BrokerAccount:
    id: str
    broker: str
    account_number: str
    currency: str = "USD"
    balance: float = 0
    equity: float = 0
    margin_used: float = 0
    open_positions: int = 0
    daily_pnl: float = 0
    total_pnl: float = 0
    leverage: int = 100
    last_updated: str = ""

class MultiAccountManager:

    def __init__(self, path: str = "/home/claude/trading_data_store/accounts"):
        self.path = path
        os.makedirs(path, exist_ok=True)
        self.file = f"{path}/accounts.json"
        self.accounts = self._load()

    def _load(self) -> list[dict]:
        if os.path.exists(self.file):
            with open(self.file) as f: return json.load(f)
        return []

    def _save(self):
        with open(self.file, "w") as f: json.dump(self.accounts, f, indent=2, default=str)

    def add_account(self, account: BrokerAccount):
        self.accounts.append(account.__dict__)
        self._save()

    def update_account(self, account_id: str, **kwargs):
        for acc in self.accounts:
            if acc["id"] == account_id:
                acc.update(kwargs)
                acc["last_updated"] = datetime.utcnow().isoformat()
        self._save()

    def aggregate_summary(self) -> dict:
        if not self.accounts:
            return {"error": "No accounts registered"}
        total_balance = sum(a["balance"] for a in self.accounts)
        total_equity = sum(a["equity"] for a in self.accounts)
        total_daily_pnl = sum(a["daily_pnl"] for a in self.accounts)
        total_margin = sum(a["margin_used"] for a in self.accounts)
        return {
            "n_accounts": len(self.accounts),
            "total_balance": round(total_balance, 2),
            "total_equity": round(total_equity, 2),
            "total_daily_pnl": round(total_daily_pnl, 2),
            "total_margin_used": round(total_margin, 2),
            "margin_utilization_pct": round(total_margin / max(total_equity, 1) * 100, 2),
            "accounts": self.accounts,
            "best_performer": max(self.accounts, key=lambda a: a["daily_pnl"])["id"] if self.accounts else None,
            "worst_performer": min(self.accounts, key=lambda a: a["daily_pnl"])["id"] if self.accounts else None,
        }

    def compare_accounts(self) -> pd.DataFrame:
        return pd.DataFrame(self.accounts)[[
            "id", "broker", "balance", "equity", "daily_pnl", "total_pnl", "leverage"
        ]].sort_values("total_pnl", ascending=False)

Pair Scanner Screener

Pair Scanner & Screener

import pandas as pd, numpy as np
from typing import Callable

class PairScanner:

    @staticmethod
    def scan(pairs_data: dict, conditions: list[dict]) -> list[dict]:
        """
        Scan all pairs against a list of conditions.
        pairs_data: {"EURUSD": df, "GBPUSD": df, ...}
        conditions: [{"name": "RSI_oversold", "fn": lambda df: rsi(df) < 30}, ...]
        """
        results = []
        for symbol, df in pairs_data.items():
            if df.empty or len(df) < 50: continue
            matches = []
            for cond in conditions:
                try:
                    if cond["fn"](df):
                        matches.append(cond["name"])
                except: continue
            if matches:
                results.append({
                    "symbol": symbol,
                    "conditions_met": matches,
                    "n_conditions": len(matches),
                    "score": len(matches) / len(conditions),
                })
        return sorted(results, key=lambda r: r["n_conditions"], reverse=True)

    @staticmethod
    def preset_scans() -> dict:
        """Pre-built scan conditions for common setups."""
        def _rsi(df, period=14):
            d = df["close"].diff()
            g = d.where(d > 0, 0).rolling(period).mean()
            l = (-d.where(d < 0, 0)).rolling(period).mean()
            return (100 - 100 / (1 + g / l.replace(0, np.nan))).iloc[-1]

        return {
            "oversold_bounce": [
                {"name": "RSI<30", "fn": lambda df: _rsi(df) < 30},
                {"name": "above_SMA200", "fn": lambda df: df["close"].iloc[-1] > df["close"].rolling(200).mean().iloc[-1]},
                {"name": "bullish_candle", "fn": lambda df: df["close"].iloc[-1] > df["open"].iloc[-1]}],
            "overbought_fade": [
                {"name": "RSI>70", "fn": lambda df: _rsi(df) > 70},
                {"name": "below_SMA200", "fn": lambda df: df["close"].iloc[-1] < df["close"].rolling(200).mean().iloc[-1]},
                {"name": "bearish_candle", "fn": lambda df: df["close"].iloc[-1] < df["open"].iloc[-1]}],
            "breakout_candidate": [
                {"name": "BB_squeeze", "fn": lambda df: (df["close"].rolling(20).std().iloc[-1] / df["close"].rolling(20).mean().iloc[-1]) < 0.005},
                {"name": "volume_rising", "fn": lambda df: df["volume"].iloc[-1] > df["volume"].rolling(20).mean().iloc[-1] * 1.3},
                {"name": "near_20bar_high", "fn": lambda df: df["close"].iloc[-1] > df["high"].rolling(20).max().iloc[-2] * 0.998}],
            "trend_pullback": [
                {"name": "above_EMA50", "fn": lambda df: df["close"].iloc[-1] > df["close"].ewm(span=50).mean().iloc[-1]},
                {"name": "touching_EMA20", "fn": lambda df: abs(df["close"].iloc[-1] - df["close"].ewm(span=20).mean().iloc[-1]) < (df["high"] - df["low"]).rolling(14).mean().iloc[-1] * 0.5},
                {"name": "RSI_40_60", "fn": lambda df: 40 < _rsi(df) < 60}],
        }

    @staticmethod
    def quick_scan(pairs_data: dict, scan_name: str = "oversold_bounce") -> list[dict]:
        presets = PairScanner.preset_scans()
        conditions = presets.get(scan_name, presets["oversold_bounce"])
        return PairScanner.scan(pairs_data, conditions)

AsyncMT5Wrapper & PositionStateMachine (Enhancement)

Ctrl click to launch VS Code Native REPL

Related Skills

• Mt5 Development
• Mt5 Chart Browser
• Trading Automation
• Market Data Ingestion