Bot & Automation
Treding Forex Ai
/root/hermes-projects/Treding Forex Ai
mt5-bridge/app/main.py
text
from __future__ import annotations
from enum import Enum
from typing import Annotated
from uuid import uuid4
from fastapi import Depends, FastAPI, Header, HTTPException, status
from pydantic import BaseModel, Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
try:
import MetaTrader5 as mt5
except ImportError: # pragma: no cover - only happens off the MT5 bridge host.
mt5 = None
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
bridge_token: str = Field(min_length=24)
bridge_dry_run: bool = True
allow_demo_trading: bool = False
allow_real_trading: bool = False
real_trading_confirm: str | None = None
mt5_login: int | None = None
mt5_password: str | None = None
mt5_server: str | None = None
mt5_path: str | None = None
required_account_login: int | None = None
required_account_server: str | None = None
allowed_symbols: str = "XAUUSD,EURUSD,GBPUSD,USDJPY,GBPJPY,AUDUSD"
max_lot_per_order: float = 0.01
max_slippage_points: int = 30
@field_validator("mt5_login", "required_account_login", mode="before")
@classmethod
def empty_login_to_none(cls, value: object) -> object:
if value == "":
return None
return value
@field_validator(
"mt5_password",
"mt5_server",
"mt5_path",
"required_account_server",
"real_trading_confirm",
mode="before",
)
@classmethod
def empty_string_to_none(cls, value: object) -> object:
if value == "":
return None
return value
@property
def allowed_symbol_set(self) -> set[str]:
return {symbol.strip().upper() for symbol in self.allowed_symbols.split(",") if symbol.strip()}
settings = Settings()
app = FastAPI(title="Treding Forex Ai MT5 Bridge", version="0.2.0")
REAL_TRADING_CONFIRMATION = "I_UNDERSTAND_REAL_MONEY_RISK"
class OrderSide(str, Enum):
buy = "buy"
sell = "sell"
class OrderRequest(BaseModel):
clientOrderId: str
symbol: str
side: OrderSide
lotSize: float = Field(gt=0)
entry: float
stopLoss: float
takeProfit: float
metadata: dict[str, str | int | float] = Field(default_factory=dict)
class OrderResponse(BaseModel):
orderId: str
dryRun: bool
message: str
accountTradeMode: str | None = None
class PositionInfo(BaseModel):
ticket: str
symbol: str
side: str
volume: float
openPrice: float
sl: float
tp: float
profit: float
openTime: str
clientOrderId: str
class PositionsResponse(BaseModel):
positions: list[PositionInfo]
class CandleInfo(BaseModel):
time: str
open: float
high: float
low: float
close: float
volume: float
class CandlesResponse(BaseModel):
candles: list[CandleInfo]
class ClosePositionRequest(BaseModel):
symbol: str
class ClosePositionResponse(BaseModel):
ticket: str
pnl: float
dryRun: bool
message: str
def require_auth(authorization: Annotated[str | None, Header()] = None) -> None:
expected = f"Bearer {settings.bridge_token}"
if authorization != expected:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized")
@app.get("/health", dependencies=[Depends(require_auth)])
def health() -> dict[str, object]:
connected = ensure_mt5_initialized(raise_on_error=False)
account = mt5.account_info() if connected and mt5 is not None else None
terminal = mt5.terminal_info() if connected and mt5 is not None else None
return {
"ok": True,
"dryRun": settings.bridge_dry_run,
"allowDemoTrading": settings.allow_demo_trading,
"allowRealTrading": settings.allow_real_trading,
"realTradingConfirmationReady": settings.real_trading_confirm == REAL_TRADING_CONFIRMATION,
"mt5Available": mt5 is not None,
"mt5Connected": connected,
"loginConfigured": settings.mt5_login is not None,
"serverConfigured": bool(settings.mt5_server),
"requiredAccountLoginConfigured": effective_required_login() is not None,
"requiredAccountServerConfigured": bool(effective_required_server()),
"accountLoginPresent": account is not None,
"accountLoginLast4": str(account.login)[-4:] if account else None,
"accountServer": account.server if account else None,
"accountTradeMode": "demo" if account and account.trade_mode == 0 else "real" if account and account.trade_mode == 2 else "unknown" if account else None,
"terminalConnected": bool(getattr(terminal, "connected", False)) if terminal else False,
"terminalTradeAllowed": bool(getattr(terminal, "trade_allowed", False)) if terminal else False,
"tradeAllowed": bool(getattr(account, "trade_allowed", False)) if account else False,
"allowedSymbols": sorted(settings.allowed_symbol_set),
"maxLotPerOrder": settings.max_lot_per_order,
}
@app.post("/orders", response_model=OrderResponse, dependencies=[Depends(require_auth)])
def create_order(order: OrderRequest) -> OrderResponse:
symbol = order.symbol.upper()
if symbol not in settings.allowed_symbol_set:
raise HTTPException(status_code=400, detail="Symbol is not allowed")
if order.lotSize > settings.max_lot_per_order:
raise HTTPException(status_code=400, detail="Lot size exceeds MAX_LOT_PER_ORDER")
if settings.bridge_dry_run:
return OrderResponse(
orderId=f"dry-run-{uuid4()}",
dryRun=True,
message="Dry-run accepted. No real order was sent to MT5.",
)
ensure_mt5_initialized(raise_on_error=True)
account_trade_mode = validate_trading_gate()
ticket = send_market_order(order)
return OrderResponse(
orderId=str(ticket),
dryRun=False,
message="Order sent to MT5",
accountTradeMode=account_trade_mode,
)
@app.get("/positions", response_model=PositionsResponse, dependencies=[Depends(require_auth)])
def get_positions() -> PositionsResponse:
"""
Ambil semua posisi terbuka dari MT5.
Dalam mode dry-run, kembalikan list kosong.
"""
if settings.bridge_dry_run:
return PositionsResponse(positions=[])
connected = ensure_mt5_initialized(raise_on_error=True)
if not connected or mt5 is None:
raise HTTPException(status_code=500, detail="MT5 tidak terhubung")
positions_raw = mt5.positions_get()
if positions_raw is None:
return PositionsResponse(positions=[])
result: list[PositionInfo] = []
for pos in positions_raw:
side = "buy" if pos.type == mt5.POSITION_TYPE_BUY else "sell"
# clientOrderId disimpan di comment field (12 karakter pertama)
comment = getattr(pos, "comment", "") or ""
client_order_id = comment.replace("TredingForexAi ", "")
result.append(
PositionInfo(
ticket=str(pos.ticket),
symbol=pos.symbol,
side=side,
volume=pos.volume,
openPrice=pos.price_open,
sl=pos.sl,
tp=pos.tp,
profit=pos.profit,
openTime=str(pos.time),
clientOrderId=client_order_id,
)
)
return PositionsResponse(positions=result)
@app.get("/market-data/candles", response_model=CandlesResponse, dependencies=[Depends(require_auth)])
def get_candles(symbol: str, timeframe: str, limit: int = 200) -> CandlesResponse:
normalized_symbol = symbol.upper()
if normalized_symbol not in settings.allowed_symbol_set:
raise HTTPException(status_code=400, detail="Symbol is not allowed")
if limit < 2 or limit > 1000:
raise HTTPException(status_code=400, detail="Limit must be between 2 and 1000")
ensure_mt5_initialized(raise_on_error=True)
if mt5 is None:
raise HTTPException(status_code=500, detail="MetaTrader5 package is not available")
mt5_timeframe = mt5_timeframe_constant(timeframe)
if not mt5.symbol_select(normalized_symbol, True):
raise HTTPException(status_code=400, detail="Symbol cannot be selected in MT5")
rates = mt5.copy_rates_from_pos(normalized_symbol, mt5_timeframe, 0, limit)
if rates is None or len(rates) == 0:
raise HTTPException(status_code=404, detail="No candle data from MT5")
candles = [
CandleInfo(
time=str(int(rate["time"])),
open=float(rate["open"]),
high=float(rate["high"]),
low=float(rate["low"]),
close=float(rate["close"]),
volume=float(rate["tick_volume"]),
)
for rate in rates
]
return CandlesResponse(candles=candles)
@app.post(
"/positions/{ticket}/close",
response_model=ClosePositionResponse,
dependencies=[Depends(require_auth)],
)
def close_position(ticket: str, body: ClosePositionRequest) -> ClosePositionResponse:
"""
Tutup posisi berdasarkan ticket number.
Dalam mode dry-run, simulasi close tanpa benar-benar mengirim ke MT5.
"""
if settings.bridge_dry_run:
return ClosePositionResponse(
ticket=ticket,
pnl=0.0,
dryRun=True,
message=f"Dry-run: position {ticket} close simulated.",
)
ensure_mt5_initialized(raise_on_error=True)
validate_trading_gate()
if mt5 is None:
raise HTTPException(status_code=500, detail="MetaTrader5 package is not available")
symbol = body.symbol.upper()
# Cari posisi berdasarkan ticket
positions = mt5.positions_get(ticket=int(ticket))
if not positions:
raise HTTPException(status_code=404, detail=f"Position {ticket} tidak ditemukan")
pos = positions[0]
tick = mt5.symbol_info_tick(symbol)
if tick is None:
raise HTTPException(status_code=400, detail="Tidak bisa ambil tick untuk close")
# Close posisi: BUY ditutup dengan SELL, SELL ditutup dengan BUY
close_type = mt5.ORDER_TYPE_SELL if pos.type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY
close_price = tick.bid if pos.type == mt5.POSITION_TYPE_BUY else tick.ask
request = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": symbol,
"volume": pos.volume,
"type": close_type,
"position": int(ticket),
"price": close_price,
"deviation": settings.max_slippage_points,
"magic": 265590,
"comment": f"AutoClose {ticket[:8]}",
"type_time": mt5.ORDER_TIME_GTC,
"type_filling": mt5.ORDER_FILLING_IOC,
}
result = mt5.order_send(request)
if result is None:
raise HTTPException(status_code=500, detail="MT5 order_send returned no result")
if result.retcode != mt5.TRADE_RETCODE_DONE:
raise HTTPException(status_code=400, detail=f"MT5 rejected close retcode={result.retcode}")
pnl = round(pos.profit, 2)
return ClosePositionResponse(
ticket=ticket,
pnl=pnl,
dryRun=False,
message=f"Position {ticket} closed. P&L: {pnl}",
)
def ensure_mt5_initialized(raise_on_error: bool) -> bool:
if mt5 is None:
if raise_on_error:
raise HTTPException(status_code=500, detail="MetaTrader5 package is not installed")
return False
# Optimasi: Jika terminal sudah terhubung dan akun sudah terlogin, lewati inisialisasi ulang
try:
terminal_info = mt5.terminal_info()
account_info = mt5.account_info()
if terminal_info is not None and account_info is not None:
if settings.mt5_login is None or account_info.login == settings.mt5_login:
return True
except Exception:
pass
initialize_kwargs = {}
if settings.mt5_path:
initialize_kwargs["path"] = settings.mt5_path
if settings.mt5_login and settings.mt5_password and settings.mt5_server:
initialize_kwargs.update(
{
"login": settings.mt5_login,
"password": settings.mt5_password,
"server": settings.mt5_server,
}
)
initialized = mt5.initialize(**initialize_kwargs)
if not initialized:
if raise_on_error:
raise HTTPException(status_code=503, detail=safe_mt5_error("MT5 initialize failed"))
return False
if initialized and settings.mt5_login and settings.mt5_password and settings.mt5_server:
logged_in = mt5.login(settings.mt5_login, password=settings.mt5_password, server=settings.mt5_server)
if not logged_in:
if raise_on_error:
raise HTTPException(status_code=401, detail=safe_mt5_error("MT5 login failed"))
return False
return True
def safe_mt5_error(prefix: str) -> str:
if mt5 is None:
return prefix
code, message = mt5.last_error()
return f"{prefix}: code={code}, message={message}"
def effective_required_login() -> int | None:
return settings.required_account_login
def effective_required_server() -> str | None:
return settings.required_account_server
def mask_account_login(login: int) -> str:
login_text = str(login)
if len(login_text) <= 4:
return "****"
return f"***{login_text[-4:]}"
def account_trade_mode_label(account: object | None) -> str:
if account is None:
return "unknown"
trade_mode = getattr(account, "trade_mode", None)
if mt5 is not None:
demo_value = getattr(mt5, "ACCOUNT_TRADE_MODE_DEMO", None)
contest_value = getattr(mt5, "ACCOUNT_TRADE_MODE_CONTEST", None)
real_value = getattr(mt5, "ACCOUNT_TRADE_MODE_REAL", None)
if demo_value is not None and trade_mode == demo_value:
return "demo"
if contest_value is not None and trade_mode == contest_value:
return "contest"
if real_value is not None and trade_mode == real_value:
return "real"
fallback_labels = {0: "demo", 1: "contest", 2: "real"}
return fallback_labels.get(trade_mode, "unknown")
def mt5_timeframe_constant(timeframe: str) -> int:
if mt5 is None:
raise HTTPException(status_code=500, detail="MetaTrader5 package is not available")
normalized = timeframe.upper()
mapping = {
"M1": mt5.TIMEFRAME_M1,
"M5": mt5.TIMEFRAME_M5,
"M15": mt5.TIMEFRAME_M15,
"H1": mt5.TIMEFRAME_H1,
"H4": mt5.TIMEFRAME_H4,
"D1": mt5.TIMEFRAME_D1,
"W1": mt5.TIMEFRAME_W1,
}
mt5_value = mapping.get(normalized)
if mt5_value is None:
raise HTTPException(status_code=400, detail="Unsupported timeframe")
return mt5_value
def validate_trading_gate() -> str:
if mt5 is None:
raise HTTPException(status_code=500, detail="MetaTrader5 package is not available")
terminal = mt5.terminal_info()
account = mt5.account_info()
if terminal is None or account is None:
raise HTTPException(status_code=503, detail="MT5 terminal/account is not available")
if not getattr(terminal, "connected", False):
raise HTTPException(status_code=503, detail="MT5 terminal is not connected")
expected_login = effective_required_login()
if expected_login is not None and getattr(account, "login", None) != expected_login:
raise HTTPException(status_code=403, detail="Connected MT5 account does not match REQUIRED_ACCOUNT_LOGIN")
expected_server = effective_required_server()
if expected_server and str(getattr(account, "server", "")).lower() != expected_server.lower():
raise HTTPException(status_code=403, detail="Connected MT5 server does not match REQUIRED_ACCOUNT_SERVER")
if not getattr(terminal, "trade_allowed", False):
raise HTTPException(status_code=403, detail="MT5 terminal trading/algo trading is disabled")
if not getattr(account, "trade_allowed", False):
raise HTTPException(status_code=403, detail="MT5 account trading is disabled")
trade_mode = account_trade_mode_label(account)
if trade_mode == "demo":
if not settings.allow_demo_trading:
raise HTTPException(
status_code=403,
detail="Demo trading is blocked. Set ALLOW_DEMO_TRADING=true after verifying the demo account.",
)
return trade_mode
if trade_mode == "real":
if not settings.allow_real_trading:
raise HTTPException(
status_code=403,
detail="Real trading is blocked. Set ALLOW_REAL_TRADING=true only after explicit approval.",
)
if settings.real_trading_confirm != REAL_TRADING_CONFIRMATION:
raise HTTPException(
status_code=403,
detail=f"Real trading confirmation missing. Set REAL_TRADING_CONFIRM={REAL_TRADING_CONFIRMATION}.",
)
return trade_mode
raise HTTPException(status_code=403, detail=f"Unsupported MT5 account trade mode: {trade_mode}")
def send_market_order(order: OrderRequest) -> int:
if mt5 is None:
raise HTTPException(status_code=500, detail="MetaTrader5 package is not available")
symbol = order.symbol.upper()
if not mt5.symbol_select(symbol, True):
raise HTTPException(status_code=400, detail="Symbol cannot be selected in MT5")
symbol_info = mt5.symbol_info(symbol)
if symbol_info is None:
raise HTTPException(status_code=400, detail="Symbol info is not available in MT5")
validate_symbol_volume(symbol, order.lotSize)
tick = mt5.symbol_info_tick(symbol)
if tick is None:
raise HTTPException(status_code=400, detail="No tick data from MT5")
order_type = mt5.ORDER_TYPE_BUY if order.side == OrderSide.buy else mt5.ORDER_TYPE_SELL
price = tick.ask if order.side == OrderSide.buy else tick.bid
# Pembulatan desimal otomatis sesuai spesifikasi digit simbol broker
digits = getattr(symbol_info, "digits", 2)
point = getattr(symbol_info, "point", 0.01)
stops_level = getattr(symbol_info, "trade_stops_level", 0)
# Batas minimal jarak stop (minimal 20 points untuk pengaman jika broker return 0)
min_stop_points = max(stops_level, 20) + 5
min_distance = min_stop_points * point
raw_sl = order.stopLoss if order.stopLoss else 0.0
raw_tp = order.takeProfit if order.takeProfit else 0.0
# Penyesuaian dinamis SL/TP agar memenuhi syarat minimal stops_level broker
if order.side == OrderSide.buy:
# BUY: SL harus di bawah price, TP di atas price
sl = raw_sl
if sl > 0.0 and (price - sl) < min_distance:
sl = price - min_distance
tp = raw_tp
if tp > 0.0 and (tp - price) < min_distance:
tp = price + min_distance
else:
# SELL: SL harus di atas price, TP di bawah price
sl = raw_sl
if sl > 0.0 and (sl - price) < min_distance:
sl = price + min_distance
tp = raw_tp
if tp > 0.0 and (price - tp) < min_distance:
tp = price - min_distance
# Bulatkan kembali ke spesifikasi digit simbol
sl = round(sl, digits) if sl > 0.0 else 0.0
tp = round(tp, digits) if tp > 0.0 else 0.0
request = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": symbol,
"volume": order.lotSize,
"type": order_type,
"price": price,
"sl": sl,
"tp": tp,
"deviation": settings.max_slippage_points,
"magic": 265590,
"comment": f"TredingForexAi {order.clientOrderId[:12]}",
"type_time": mt5.ORDER_TIME_GTC,
"type_filling": mt5.ORDER_FILLING_IOC,
}
result = mt5.order_send(request)
if result is None:
raise HTTPException(status_code=500, detail="MT5 order_send returned no result")
if result.retcode != mt5.TRADE_RETCODE_DONE:
raise HTTPException(status_code=400, detail=f"MT5 rejected order retcode={result.retcode}")
return int(result.order)
def validate_symbol_volume(symbol: str, lot_size: float) -> None:
if mt5 is None:
raise HTTPException(status_code=500, detail="MetaTrader5 package is not available")
symbol_info = mt5.symbol_info(symbol)
if symbol_info is None:
raise HTTPException(status_code=400, detail="Symbol info is not available in MT5")
volume_min = float(getattr(symbol_info, "volume_min", 0.0) or 0.0)
volume_max = float(getattr(symbol_info, "volume_max", 0.0) or 0.0)
volume_step = float(getattr(symbol_info, "volume_step", 0.0) or 0.0)
if volume_min > 0 and lot_size < volume_min:
raise HTTPException(status_code=400, detail=f"Lot size is below symbol minimum volume {volume_min}")
if volume_max > 0 and lot_size > volume_max:
raise HTTPException(status_code=400, detail=f"Lot size is above symbol maximum volume {volume_max}")
if volume_step > 0:
steps = round((lot_size - volume_min) / volume_step)
normalized = volume_min + (steps * volume_step)
if abs(lot_size - normalized) > 1e-8:
raise HTTPException(status_code=400, detail=f"Lot size must follow symbol volume step {volume_step}")