forex-scalping.com

Code: Select all

import uvicorn
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel
import MetaTrader5 as mt5

app = FastAPI()

# A secret token you will include in your TradingView alert message
# e.g., {"token": "YOUR_SECURE_TOKEN_HERE", "action": "buy", "volume": 0.1, "symbol": "EURUSD"}
SECRET_TOKEN = "YOUR_SECURE_TOKEN_HERE"

class TradePayload(BaseModel):
    token: str
    action: str
    volume: float
    symbol: str

@app.on_event("startup")
def startup_event():
    # Initialize connection to the local MT5 terminal
    if not mt5.initialize():
        print("MT5 initialization failed")
        mt5.shutdown()

@app.post("/tv-webhook")
async def execute_trade(payload: TradePayload):
    # 1. Validate the source
    if payload.token != SECRET_TOKEN:
        raise HTTPException(status_code=401, detail="Unauthorized")

    # 2. Prepare the MT5 order request
    symbol = payload.symbol
    action = payload.action.lower()
    
    # Ensure symbol is visible in market watch
    if not mt5.symbol_select(symbol, True):
        raise HTTPException(status_code=400, detail=f"Symbol {symbol} not found")

    symbol_info = mt5.symbol_info(symbol)
    
    if action == "buy":
        order_type = mt5.ORDER_TYPE_BUY
        price = symbol_info.ask
    elif action == "sell":
        order_type = mt5.ORDER_TYPE_SELL
        price = symbol_info.bid
    else:
        raise HTTPException(status_code=400, detail="Invalid action")

    request = {
        "action": mt5.TRADE_ACTION_DEAL,
        "symbol": symbol,
        "volume": payload.volume,
        "type": order_type,
        "price": price,
        "deviation": 5,
        "magic": 123456,
        "comment": "TV Webhook",
        "type_time": mt5.ORDER_TIME_GTC,
        "type_filling": mt5.ORDER_FILLING_FOK,
    }

    # 3. Fire the order natively into the MT5 terminal
    result = mt5.order_send(request)
    
    if result.retcode != mt5.TRADE_RETCODE_DONE:
        return {"status": "error", "message": result.comment, "retcode": result.retcode}
        
    return {"status": "success", "ticket": result.deal}

if __name__ == "__main__":
    # Run locally, exposed only to Nginx
    uvicorn.run(app, host="127.0.0.1", port=8000)