#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ TradeHub - agente de recolha MT5 ================================ Corre em cada maquina (VPS ou local) que tenha terminais MetaTrader 5. Descobre sozinho todos os terminais em execucao, le a conta autenticada em cada um e envia historico + posicoes abertas para a API. CONFIGURACAO POR LOGIN ---------------------- Nao ha caminhos a escrever. A config so tem metadados por numero de conta (alias, grupo, saldo_inicial) e contas novas sao apanhadas automaticamente com o login como nome. Podes mover um terminal de VPS que nada muda. A ligacao em si tem de ser feita pelo caminho do executavel - a biblioteca MetaTrader5 nao sabe procurar "o terminal da conta X", e o login so e conhecido depois de ligar. Por isso o agente enumera os processos terminal64.exe da maquina e liga-se a cada um para descobrir que conta la esta. O agente NUNCA faz login com credenciais. Passar login/password ao mt5.initialize() trocaria a conta do terminal e derrubaria o EA a correr. Uso: python coletor_mt5.py # ciclo continuo python coletor_mt5.py --uma-vez # uma passagem e sai python coletor_mt5.py --listar # mostra as contas encontradas python coletor_mt5.py --reset # ignora estado e recolhe tudo """ import argparse import json import logging import os import socket import sys import time from collections import defaultdict from datetime import datetime, timedelta, timezone from urllib import error as urlerr from urllib import request as urlreq try: import MetaTrader5 as mt5 except ImportError: sys.exit("Falta a biblioteca: pip install MetaTrader5") try: import psutil except ImportError: psutil = None # sem psutil so funcionam os caminhos manuais da config BASE = os.path.dirname(os.path.abspath(__file__)) CONFIG = os.path.join(BASE, "config.json") ESTADO = os.path.join(BASE, "estado.json") LOG = os.path.join(BASE, "coletor.log") LOTE = 400 # trades por pedido HTTP # --------------------------------------------------------------------- # Configuracao e estado # --------------------------------------------------------------------- def carregar_config(): if not os.path.exists(CONFIG): sys.exit(f"Nao encontrei {CONFIG}. Copia config.exemplo.json.") with open(CONFIG, encoding="utf-8") as f: return json.load(f) def carregar_estado(): if os.path.exists(ESTADO): try: with open(ESTADO, encoding="utf-8") as f: return json.load(f) except (ValueError, OSError): logging.warning("estado.json ilegivel, a recomecar do zero") return {} def guardar_estado(estado): tmp = ESTADO + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(estado, f, indent=2) os.replace(tmp, ESTADO) # --------------------------------------------------------------------- # Descoberta de terminais # --------------------------------------------------------------------- NOMES_MT5 = ("terminal64.exe", "terminal.exe") def descobrir_terminais(cfg): """ Caminhos dos terminais MT5 a interrogar. Vem dos processos em execucao (ordem principal) mais quaisquer caminhos fixos indicados na config. Uma instancia por pasta de instalacao: dois terminais partilham o mesmo executavel apenas se forem a mesma instalacao, caso em que a MT5 so expoe um deles. """ caminhos = [] if psutil is None: logging.warning("psutil nao instalado (pip install psutil); " "so uso os caminhos_extra da config") else: for proc in psutil.process_iter(["name", "exe"]): try: nome = (proc.info.get("name") or "").lower() if nome in NOMES_MT5 and proc.info.get("exe"): caminhos.append(os.path.normpath(proc.info["exe"])) except (psutil.NoSuchProcess, psutil.AccessDenied): continue for extra in cfg.get("caminhos_extra", []): caminhos.append(os.path.normpath(extra)) # dedup preservando a ordem vistos = set() unicos = [] for c in caminhos: chave = c.lower() if chave not in vistos: vistos.add(chave) unicos.append(c) return unicos def metadados(cfg, login): """Alias/grupo/saldo inicial configurados para esta conta, se existirem.""" m = (cfg.get("contas") or {}).get(str(login), {}) return { "alias": m.get("alias") or str(login), "grupo": m.get("grupo"), "saldo_inicial": m.get("saldo_inicial"), "simbolo_referencia": m.get("simbolo_referencia"), } # --------------------------------------------------------------------- # Utilitarios MT5 # --------------------------------------------------------------------- def detetar_offset(simbolo_ref): """Horas entre a hora do servidor do broker e UTC (ex: 2 no inverno, 3 no verao).""" simbolos = [simbolo_ref] if simbolo_ref else [] simbolos += ["XAUUSD", "EURUSD", "GOLD"] for s in simbolos: if not s: continue tick = mt5.symbol_info_tick(s) if tick and tick.time: return int(round((tick.time - time.time()) / 3600.0)) return 0 def para_utc(ts_servidor, offset_h): """Epoch em hora de servidor -> string UTC 'YYYY-MM-DD HH:MM:SS'.""" dt = datetime.fromtimestamp(ts_servidor - offset_h * 3600, tz=timezone.utc) return dt.strftime("%Y-%m-%d %H:%M:%S") def prefixo_ea(comentario): """Extrai o prefixo do comentario (ex: 'FR-1234' -> 'FR').""" if not comentario: return None txt = comentario.strip() corte = len(txt) for i, ch in enumerate(txt): if not (ch.isalpha() or ch == "_"): corte = i break pref = txt[:corte].upper() return pref[:32] or None def reconstruir_trades(deals, sl_tp, offset_h): """ Agrupa deals por position_id e devolve uma linha por posicao fechada. Posicoes ainda abertas ou parcialmente fechadas sao ignoradas (aparecem na tabela de posicoes abertas ate fecharem por completo). """ por_pos = defaultdict(list) for d in deals: if d.type not in (mt5.DEAL_TYPE_BUY, mt5.DEAL_TYPE_SELL): continue # ignora deposito, levantamento, credito, correcao if d.position_id == 0: continue por_pos[d.position_id].append(d) trades = [] for pos_id, ds in por_pos.items(): ds.sort(key=lambda x: (x.time_msc, x.ticket)) entradas = [d for d in ds if d.entry == mt5.DEAL_ENTRY_IN] saidas = [d for d in ds if d.entry in (mt5.DEAL_ENTRY_OUT, mt5.DEAL_ENTRY_OUT_BY, mt5.DEAL_ENTRY_INOUT)] if not entradas or not saidas: continue vol_in = sum(d.volume for d in entradas) vol_out = sum(d.volume for d in saidas) if vol_out + 1e-8 < vol_in: continue # fecho parcial: ainda ha posicao viva p_abrir = sum(d.price * d.volume for d in entradas) / vol_in p_fechar = sum(d.price * d.volume for d in saidas) / vol_out lucro = sum(d.profit for d in ds) comissao = sum(getattr(d, "commission", 0.0) for d in ds) swap = sum(getattr(d, "swap", 0.0) for d in ds) taxa = sum(getattr(d, "fee", 0.0) for d in ds) comentario = (entradas[0].comment or saidas[-1].comment or "").strip() t_abrir = entradas[0].time t_fechar = saidas[-1].time sl, tp = sl_tp.get(pos_id, (None, None)) trades.append({ "position_id": int(pos_id), "simbolo": entradas[0].symbol, "tipo": "buy" if entradas[0].type == mt5.DEAL_TYPE_BUY else "sell", "volume": round(vol_in, 4), "abertura": para_utc(t_abrir, offset_h), "fecho": para_utc(t_fechar, offset_h), "preco_abrir": round(p_abrir, 8), "preco_fechar": round(p_fechar, 8), "sl": sl or None, "tp": tp or None, "lucro": round(lucro, 2), "comissao": round(comissao, 2), "swap": round(swap, 2), "taxa": round(taxa, 2), "liquido": round(lucro + comissao + swap + taxa, 2), "magic": int(entradas[0].magic or 0), "comentario": comentario[:64] or None, "ea": prefixo_ea(comentario), "duracao_s": max(0, int(t_fechar - t_abrir)), }) trades.sort(key=lambda t: t["fecho"]) return trades def mapa_sl_tp(desde, ate): """position_id -> (sl, tp) a partir das ordens historicas.""" ordens = mt5.history_orders_get(desde, ate) or [] mapa = {} for o in ordens: pid = getattr(o, "position_id", 0) if not pid: continue sl = float(o.sl) if o.sl else None tp = float(o.tp) if o.tp else None if sl or tp: mapa[pid] = (sl, tp) return mapa # --------------------------------------------------------------------- # Envio para a API # --------------------------------------------------------------------- def enviar(cfg, payload): corpo = json.dumps(payload, ensure_ascii=False).encode("utf-8") req = urlreq.Request( cfg["api_url"], data=corpo, headers={ "Content-Type": "application/json; charset=utf-8", "X-Api-Key": cfg["api_key"], }, method="POST", ) try: with urlreq.urlopen(req, timeout=cfg.get("timeout", 60)) as r: return json.loads(r.read().decode("utf-8")) except urlerr.HTTPError as e: detalhe = e.read().decode("utf-8", "ignore")[:300] raise RuntimeError(f"HTTP {e.code}: {detalhe}") from None except (urlerr.URLError, socket.timeout) as e: raise RuntimeError(f"rede: {e}") from None # --------------------------------------------------------------------- # Recolha de um terminal # --------------------------------------------------------------------- def processar_terminal(cfg, caminho, estado, forcar): ok = mt5.initialize(path=caminho) if caminho else mt5.initialize() if not ok: raise RuntimeError(f"initialize falhou: {mt5.last_error()}") try: conta = mt5.account_info() if conta is None: raise RuntimeError("terminal sem conta autenticada") login = int(conta.login) if login in set(cfg.get("ignorar", [])): logging.info("conta %s ignorada por configuracao", login) return login, 0, 0, {"novas": 0, "atualizadas": 0} meta = metadados(cfg, login) offset = detetar_offset(meta["simbolo_referencia"] or cfg.get("simbolo_referencia")) chave = str(login) anterior = estado.get(chave, {}) agora = datetime.now(timezone.utc) if forcar or not anterior.get("ultimo_fecho"): desde = agora - timedelta(days=cfg.get("dias_historico_inicial", 730)) else: ultimo = datetime.strptime(anterior["ultimo_fecho"], "%Y-%m-%d %H:%M:%S") ultimo = ultimo.replace(tzinfo=timezone.utc) desde = ultimo - timedelta(days=cfg.get("dias_rescan", 3)) # janela em hora de servidor, com folga de 1 dia de cada lado desde_srv = desde + timedelta(hours=offset) - timedelta(days=1) ate_srv = agora + timedelta(hours=offset) + timedelta(days=1) deals = mt5.history_deals_get(desde_srv, ate_srv) or [] sl_tp = mapa_sl_tp(desde_srv, ate_srv) trades = reconstruir_trades(deals, sl_tp, offset) abertas = [] for p in (mt5.positions_get() or []): com = (p.comment or "").strip() abertas.append({ "ticket": int(p.ticket), "simbolo": p.symbol, "tipo": "buy" if p.type == mt5.POSITION_TYPE_BUY else "sell", "volume": round(p.volume, 4), "abertura": para_utc(p.time, offset), "preco_abrir": round(p.price_open, 8), "preco_atual": round(p.price_current, 8), "sl": float(p.sl) or None, "tp": float(p.tp) or None, "lucro": round(p.profit, 2), "swap": round(p.swap, 2), "magic": int(p.magic or 0), "comentario": com[:64] or None, "ea": prefixo_ea(com), }) info_conta = { "login": login, "alias": meta["alias"], "grupo": meta["grupo"], "servidor": conta.server, "corretora": conta.company, "moeda": conta.currency, "alavancagem": int(conta.leverage or 0), "saldo": round(conta.balance, 2), "equity": round(conta.equity, 2), "margem_livre": round(conta.margin_free, 2), "saldo_inicial": meta["saldo_inicial"], "offset_servidor": offset, } finally: mt5.shutdown() # ---- envio em lotes; o ultimo leva o snapshot de posicoes ------- host = socket.gethostname() lotes = [trades[i:i + LOTE] for i in range(0, len(trades), LOTE)] or [[]] resumo = {"novas": 0, "atualizadas": 0} for i, lote in enumerate(lotes): ultimo_lote = (i == len(lotes) - 1) payload = { "host": host, "conta": info_conta, "trades": lote, "posicoes": abertas if ultimo_lote else None, "substituir_posicoes": ultimo_lote, "equity_ponto": { "ts": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:00"), "saldo": info_conta["saldo"], "equity": info_conta["equity"], } if ultimo_lote else None, } r = enviar(cfg, payload) resumo["novas"] += r.get("novas", 0) resumo["atualizadas"] += r.get("atualizadas", 0) if trades: estado[str(login)] = {"ultimo_fecho": trades[-1]["fecho"]} return login, len(trades), len(abertas), resumo # --------------------------------------------------------------------- # Ciclo principal # --------------------------------------------------------------------- def passagem(cfg, estado, forcar): caminhos = descobrir_terminais(cfg) if not caminhos: logging.warning("Nenhum terminal MT5 em execucao nesta maquina") return for caminho in caminhos: try: login, n_tr, n_pos, r = processar_terminal(cfg, caminho, estado, forcar) logging.info("conta %s | %d trades (%d novas, %d atualizadas) | %d abertas | %s", login, n_tr, r["novas"], r["atualizadas"], n_pos, os.path.basename(os.path.dirname(caminho))) except Exception as e: # noqa: BLE001 - queremos continuar nos outros logging.error("%s | falhou: %s", caminho, e) finally: mt5.shutdown() guardar_estado(estado) def listar(cfg): """Mostra que contas estao acessiveis nesta maquina.""" caminhos = descobrir_terminais(cfg) if not caminhos: print("Nenhum terminal MT5 em execucao.") return print(f"{len(caminhos)} terminal(is) encontrado(s):\n") for caminho in caminhos: try: if not mt5.initialize(path=caminho): print(f" [erro] {caminho}: {mt5.last_error()}") continue c = mt5.account_info() if c is None: print(f" [sem conta] {caminho}") continue m = metadados(cfg, int(c.login)) print(f" {c.login:>12} {m['alias']:<22} {c.company} / {c.server}" f" saldo {c.balance:.2f} {c.currency}") finally: mt5.shutdown() print("\nCopia os numeros de conta para a seccao \"contas\" do config.json.") def main(): ap = argparse.ArgumentParser(description="Agente TradeHub para MT5") ap.add_argument("--uma-vez", action="store_true", help="uma passagem e sai") ap.add_argument("--listar", action="store_true", help="lista as contas encontradas") ap.add_argument("--reset", action="store_true", help="recolhe todo o historico") args = ap.parse_args() logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)-7s %(message)s", handlers=[logging.FileHandler(LOG, encoding="utf-8"), logging.StreamHandler()], ) cfg = carregar_config() if args.listar: listar(cfg) return estado = {} if args.reset else carregar_estado() if args.uma_vez: passagem(cfg, estado, args.reset) return intervalo = cfg.get("intervalo_segundos", 60) forcar = args.reset logging.info("Agente iniciado. Intervalo: %ds. Terminais detetados: %d", intervalo, len(descobrir_terminais(cfg))) while True: try: passagem(cfg, estado, forcar) forcar = False except KeyboardInterrupt: logging.info("Terminado pelo utilizador") return except Exception as e: # noqa: BLE001 logging.exception("Erro no ciclo: %s", e) time.sleep(intervalo) if __name__ == "__main__": main()