EA/EA

backend/
 ├─ main.py
 ├─ engine.py
 ├─ mt5_adapter.py
 ├─ strategies.py
 ├─ backtest.py
 ├─ requirements.txt
 └─ .env.example
 
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from dotenv import load_dotenv
import os

load_dotenv()

from engine import Engine

app = FastAPI(title="FX Bot - Demo (MT5 Adapter)")
engine = Engine()

class StartRequest(BaseModel):
    adapter: str = "mt5"
    demo: bool = True

@app.post("/start")
async def start(req: StartRequest, bg: BackgroundTasks):
    if engine.running:
        return {"ok": False, "msg": "Engine already running"}
    engine.select_adapter(req.adapter, demo=req.demo)
    bg.add_task(engine.run)
    return {"ok": True, "adapter": req.adapter}

@app.post("/stop")
def stop():
    engine.stop()
    return {"ok": True}

@app.get("/status")
def status():
    return engine.get_status()

@app.post("/backtest")
def backtest(payload: dict):
    return engine.backtest(payload)
import asyncio
from datetime import datetime
from mt5_adapter import MT5Adapter
from strategies import ema_atr_strategy

class Engine:
    def _init_(self):
        self.adapter = None
        self.running = False
        self.equity = 10000.0
        self.open_positions = []
        self.logs = []
        self.adapter_name = None
        self._stop_requested = False
        self.max_concurrent = 2
        self.risk_per_trade = 0.005

    def select_adapter(self, name: str, demo: bool=True):
        name = name.lower()
        self.adapter_name = name
        if name == "mt5":
            self.adapter = MT5Adapter(demo=demo, engine=self)
        else:
            raise ValueError("Unknown adapter")

    async def _run_loop(self):
        self.running = True
        self._stop_requested = False
        self.logs.append(f"{datetime.utcnow().isoformat()}Z starting engine with {self.adapter_name}")
        await self.adapter.connect()
        try:
            while not self._stop_requested:
                df = await self.adapter.get_latest_candles("EURUSD", "M15")
                signal = ema_atr_strategy(df)
                if signal != 0:
                    if len(self.open_positions) < self.max_concurrent:
                        volume = 0.01
                        order = await self.adapter.create_market_order("EURUSD", signal, volume=volume)
                        self.open_positions.append(order)
                        self.logs.append(f"{datetime.utcnow().isoformat()}Z order placed: {order}")
                await asyncio.sleep(5)
        finally:
            await self.adapter.disconnect()
            self.running = False
            self.logs.append(f"{datetime.utcnow().isoformat()}Z engine stopped")

    def run(self):
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        loop.run_until_complete(self._run_loop())
        loop.close()

    def stop(self):
        self._stop_requested = True

    def get_status(self):
        return {
            "running": self.running,
            "mode": self.adapter_name,
            "equity": self.equity,
            "open_positions": len(self.open_positions),
            "logs": self.logs[-200:]
        }

    def backtest(self, payload):
        from backtest import run_backtest
        return run_backtest(payload)
import asyncio
from datetime import datetime
from mt5_adapter import MT5Adapter
from strategies import ema_atr_strategy

class Engine:
    def _init_(self):
        self.adapter = None
        self.running = False
        self.equity = 10000.0
        self.open_positions = []
        self.logs = []
        self.adapter_name = None
        self._stop_requested = False
        self.max_concurrent = 2
        self.risk_per_trade = 0.005

    def select_adapter(self, name: str, demo: bool=True):
        name = name.lower()
        self.adapter_name = name
        if name == "mt5":
            self.adapter = MT5Adapter(demo=demo, engine=self)
        else:
            raise ValueError("Unknown adapter")

    async def _run_loop(self):
        self.running = True
…
import os, time, pandas as pd
import MetaTrader5 as mt5
from datetime import datetime

class MT5Adapter:
    def _init_(self, demo=True, engine=None):
        self.demo = demo
        self.engine = engine
        self.login = int(os.getenv("MT5_LOGIN") or 0)
        self.password = os.getenv("MT5_PASSWORD")
        self.server = os.getenv("MT5_SERVER")

    async def connect(self):
        if not mt5.initialize():
            raise RuntimeError("MT5 initialize failed")
        if self.login:
            logged = mt5.login(self.login, password=self.password, server=self.server)
            if not logged:
                last = mt5.last_error()
                raise RuntimeError(f"MT5 login failed: {last}")
        time.sleep(1)
        return True

    async …
import numpy as np

def ema(series, n):
    return series.ewm(span=n, adjust=False).mean()

def atr(df, n=14):
    h, l, c = df["high"], df["low"], df["close"]
    prev = c.shift(1)
    tr = np.maximum(h - l, np.maximum((h - prev).abs(), (l - prev).abs()))
    return tr.rolling(n).mean()

def ema_atr_strategy(df):
    df = df.copy().dropna()
    if len(df) < 60:
        return 0
    df["ema_fast"] = ema(df["close"], 20)
    df["ema_slow"] = ema(df["close"], 50)
    df["atr"] = atr(df, 14)
    last = df.iloc[-1]
    prev = df.iloc[-2]
    if last["ema_fast"] > last["ema_slow"] and prev["ema_fast"] <= prev["ema_slow"]:
        return 1
    if last["ema_fast"] < last["ema_slow"] and prev["ema_fast"] >= prev["ema_slow"]:
        return -1
    return 0
import pandas as pd
import numpy as np

def run_backtest(payload):
    start = payload.get("start", "2024-01-01")
    end = payload.get("end", "2024-02-01")
    freq = "15T"
    rng = pd.date_range(start=start, end=end, freq=freq)
    px = 1.0 + np.cumsum(np.random.normal(0, 0.0005, len(rng)))
    df = pd.DataFrame({"time": rng, "open": px, "high": px + 0.0005, "low": px - 0.0005, "close": px}).set_index("time")
    trades = []
    for i in range(5):
        trades.append({"time": str(rng[i]), "pnl": float(np.random.normal(5,20))})
    stats = {"trades": len(trades), "net_pnl": sum(t["pnl"] for t in trades), "win_rate": 0.5}
    return {"stats": stats, "trades": trades}
fastapi
uvicorn[standard]
pydantic
numpy
pandas
python-dotenv
MetaTrader5
requests
websockets