gavindiaz/quantumbotx
1
0
Fork 0
mirror of https://github.com/chrisnov-it/quantumbotx.git synced 2026-07-27 18:57:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Implement trading bot with a pluggable broker adapter pattern, market hour management, and Docker deployment infrastructure.

Browse Source
This commit is contained in:
Reynov Christian
2025-11-21 22:00:48 +08:00
parent 0724f07f3f
commit 5c7f41c817
14 changed files with 615 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Dockerfile
+23
View File
@@ -0,0 +1,23 @@
# Use an official Python runtime as a parent image
FROM python:3.9-slim
# Set the working directory in the container
WORKDIR /app
# Copy the current directory contents into the container at /app
COPY . /app
# Install any needed packages specified in requirements-docker.txt
RUN pip install --no-cache-dir -r requirements-docker.txt
# Make port 5000 available to the world outside this container
EXPOSE 5000
# Define environment variable
ENV FLASK_APP=run.py
ENV FLASK_RUN_HOST=0.0.0.0
ENV BROKER_TYPE=CCXT
# Default to CCXT in Docker since MT5 doesn't run on Linux easily
# Run run.py when the container launches
CMD ["python", "run.py"]
core/adapters/ccxt_adapter.py
+147
View File
@@ -0,0 +1,147 @@
import ccxt
import pandas as pd
from typing import Dict, Any, List, Optional
from datetime import datetime
import logging
from core.interfaces.broker_interface import BrokerInterface
logger = logging.getLogger(__name__)
class CCXTAdapter(BrokerInterface):
def __init__(self):
self.exchange = None
self.exchange_id = None
# Map standard timeframes to CCXT format
self.timeframe_map = {
'M1': '1m', 'M5': '5m', 'M15': '15m', 'M30': '30m',
'H1': '1h', 'H4': '4h', 'D1': '1d', 'W1': '1w', 'MN1': '1M'
}
def initialize(self, credentials: Dict[str, Any]) -> bool:
try:
self.exchange_id = credentials.get('EXCHANGE_ID', 'binance')
exchange_class = getattr(ccxt, self.exchange_id)
config = {
'apiKey': credentials.get('API_KEY'),
'secret': credentials.get('API_SECRET'),
'enableRateLimit': True,
'options': {'defaultType': 'future'} # Default to futures for bots
}
if credentials.get('PASSWORD'): # For exchanges like KuCoin
config['password'] = credentials.get('PASSWORD')
self.exchange = exchange_class(config)
# Test connection
self.exchange.load_markets()
logger.info(f"Connected to {self.exchange_id} successfully.")
return True
except Exception as e:
logger.error(f"Failed to initialize CCXT exchange {self.exchange_id}: {e}")
return False
def get_account_info(self) -> Optional[Dict[str, Any]]:
try:
balance = self.exchange.fetch_balance()
# Normalize to standard format
return {
'balance': balance.get('total', {}).get('USDT', 0.0),
'equity': balance.get('total', {}).get('USDT', 0.0), # Approx for spot
'margin': 0.0, # Complex to calculate across exchanges
'free_margin': balance.get('free', {}).get('USDT', 0.0)
}
except Exception as e:
logger.error(f"Error fetching account info: {e}")
return None
def get_rates(self, symbol: str, timeframe: str, count: int = 100) -> pd.DataFrame:
try:
tf = self.timeframe_map.get(timeframe, '1h')
ohlcv = self.exchange.fetch_ohlcv(symbol, tf, limit=count)
df = pd.DataFrame(ohlcv, columns=['time', 'open', 'high', 'low', 'close', 'tick_volume'])
df['time'] = pd.to_datetime(df['time'], unit='ms')
return df
except Exception as e:
logger.error(f"Error fetching rates for {symbol}: {e}")
return pd.DataFrame()
def get_open_positions(self) -> List[Dict[str, Any]]:
try:
# This works best for Futures. Spot exchanges might not return "positions" in the same way.
positions = self.exchange.fetch_positions()
normalized_positions = []
for pos in positions:
if float(pos['contracts']) > 0: # Only active positions
normalized_positions.append({
'ticket': pos.get('id', f"{pos['symbol']}_{pos['side']}"),
'symbol': pos['symbol'],
'type': 0 if pos['side'] == 'long' else 1, # 0=BUY, 1=SELL (MT5 convention)
'volume': float(pos['contracts']),
'price': float(pos['entryPrice']),
'profit': float(pos.get('unrealizedPnl', 0.0)),
'sl': float(pos.get('stopLossPrice', 0.0) or 0.0),
'tp': float(pos.get('takeProfitPrice', 0.0) or 0.0),
'magic': 0 # CCXT doesn't support magic numbers natively usually
})
return normalized_positions
except Exception as e:
# Fallback for Spot: check balance? No, simpler to just return empty for now or log warning
# logger.warning(f"Could not fetch positions (might be Spot market): {e}")
return []
def place_order(self, symbol: str, order_type: str, volume: float, price: float = 0.0, sl: float = 0.0, tp: float = 0.0, comment: str = "") -> bool:
try:
side = 'buy' if order_type == 'BUY' or order_type == 0 else 'sell'
type = 'limit' if price > 0 else 'market'
params = {}
# CCXT unified stopLoss/takeProfit is tricky, often exchange specific params
# For simplicity in this POC, we might skip SL/TP attachment or use params
if sl > 0:
params['stopLoss'] = sl
if tp > 0:
params['takeProfit'] = tp
if type == 'limit':
self.exchange.create_order(symbol, type, side, volume, price, params)
else:
self.exchange.create_order(symbol, type, side, volume, None, params)
logger.info(f"Order placed: {side} {symbol} {volume}")
return True
except Exception as e:
logger.error(f"Error placing order: {e}")
return False
def close_position(self, position_id: str, volume: float = 0.0) -> bool:
# Closing positions in CCXT usually means placing an opposite order
# Or using close_position method if supported
try:
# Try to find position info to know symbol and amount
# This is tricky without state.
# For now, we assume the bot logic handles "Close" by sending an opposite order signal
# But the interface demands close_position.
# We might need to implement this by fetching position first.
logger.warning("close_position not fully implemented for CCXT yet. Use place_order with opposite side.")
return False
except Exception as e:
logger.error(f"Error closing position: {e}")
return False
def get_symbol_info(self, symbol: str) -> Optional[Dict[str, Any]]:
try:
market = self.exchange.market(symbol)
return {
'name': market['symbol'],
'digits': market['precision']['price'],
'min_volume': market['limits']['amount']['min'],
'max_volume': market['limits']['amount']['max'],
'volume_step': market['precision']['amount'], # Approx
'point': 1.0 / (10 ** market['precision']['price'])
}
except Exception as e:
logger.error(f"Error getting symbol info for {symbol}: {e}")
return None
core/adapters/mt5_adapter.py
+104
View File
@@ -0,0 +1,104 @@
import logging
import pandas as pd
from typing import Dict, Any, List, Optional
from core.interfaces.broker_interface import BrokerInterface
from core.utils.mt5 import (
initialize_mt5,
get_account_info_mt5,
get_rates_mt5,
get_open_positions_mt5,
find_mt5_symbol,
TIMEFRAME_MAP
)
import MetaTrader5 as mt5
logger = logging.getLogger(__name__)
class MT5Adapter(BrokerInterface):
"""
Adapter for MetaTrader 5 using the official python library.
Wraps the functions from core.utils.mt5.
"""
def initialize(self, credentials: Dict[str, Any]) -> bool:
account = int(credentials.get('MT5_LOGIN', 0))
password = credentials.get('MT5_PASSWORD', '')
server = credentials.get('MT5_SERVER', '')
return initialize_mt5(account, password, server)
def get_account_info(self) -> Optional[Dict[str, Any]]:
return get_account_info_mt5()
def get_rates(self, symbol: str, timeframe: str, count: int = 100) -> pd.DataFrame:
# Convert string timeframe (e.g. "H1") to MT5 constant
mt5_timeframe = TIMEFRAME_MAP.get(timeframe, mt5.TIMEFRAME_H1)
# Ensure symbol is valid for this broker
valid_symbol = find_mt5_symbol(symbol)
if not valid_symbol:
logger.error(f"Symbol {symbol} not found in MT5")
return pd.DataFrame()
return get_rates_mt5(valid_symbol, mt5_timeframe, count)
def get_open_positions(self) -> List[Dict[str, Any]]:
return get_open_positions_mt5()
def place_order(self, symbol: str, order_type: str, volume: float, price: float = 0.0, sl: float = 0.0, tp: float = 0.0, comment: str = "") -> bool:
valid_symbol = find_mt5_symbol(symbol)
if not valid_symbol:
return False
# Basic order logic - simplified for adapter POC
# In a full implementation, we would move the order construction logic here
# For now, we will use a direct MT5 call to keep it simple, or we could import a helper if it existed.
# Since core.utils.mt5 doesn't have a 'place_order' function (it seems to be in the bot logic),
# we will implement a basic version here.
action = mt5.TRADE_ACTION_DEAL
type_op = mt5.ORDER_TYPE_BUY if order_type == 'BUY' else mt5.ORDER_TYPE_SELL
request = {
"action": action,
"symbol": valid_symbol,
"volume": volume,
"type": type_op,
"price": mt5.symbol_info_tick(valid_symbol).ask if order_type == 'BUY' else mt5.symbol_info_tick(valid_symbol).bid,
"sl": sl,
"tp": tp,
"deviation": 20,
"magic": 123456,
"comment": comment,
"type_time": mt5.ORDER_TIME_GTC,
"type_filling": mt5.ORDER_FILLING_IOC,
}
result = mt5.order_send(request)
if result.retcode != mt5.TRADE_RETCODE_DONE:
logger.error(f"Order failed: {result.comment}")
return False
logger.info(f"Order placed: {result.order}")
return True
def close_position(self, position_id: str, volume: float = 0.0) -> bool:
# Implementation for closing position
# Simplified for POC
try:
position_id_int = int(position_id)
# Logic to close position...
# For now, returning False as placeholder or we can implement if needed immediately.
# But the user asked for the Adapter structure first.
pass
except:
pass
return False
def get_symbol_info(self, symbol: str) -> Optional[Dict[str, Any]]:
valid_symbol = find_mt5_symbol(symbol)
if not valid_symbol:
return None
info = mt5.symbol_info(valid_symbol)
if info:
return info._asdict()
return None
core/bots/controller.py
+18 -1
View File
@@ -5,6 +5,9 @@ import logging
from core.db import queries
from .trading_bot import TradingBot
from core.strategies.strategy_map import STRATEGY_MAP
from core.factory.broker_factory import BrokerFactory
import os
from dotenv import load_dotenv
logger = logging.getLogger(__name__)
@@ -126,13 +129,27 @@ def mulai_bot(bot_id: int):
params_dict = json.loads(bot_data.get('strategy_params', '{}'))
try:
# Initialize Broker Adapter
load_dotenv()
broker_type = os.getenv('BROKER_TYPE', 'MT5') # Default to MT5
broker = BrokerFactory.get_broker(broker_type)
if broker:
creds = {
'MT5_LOGIN': os.getenv('MT5_LOGIN'),
'MT5_PASSWORD': os.getenv('MT5_PASSWORD'),
'MT5_SERVER': os.getenv('MT5_SERVER')
}
broker.initialize(creds)
bot_thread = TradingBot(
id=bot_data['id'], name=bot_data['name'], market=bot_data['market'],
risk_percent=bot_data['lot_size'], sl_pips=bot_data['sl_pips'],
tp_pips=bot_data['tp_pips'], timeframe=bot_data['timeframe'],
check_interval=bot_data['check_interval_seconds'], strategy=bot_data['strategy'],
strategy_params=params_dict,
enable_strategy_switching=bool(bot_data.get('enable_strategy_switching', 0))
enable_strategy_switching=bool(bot_data.get('enable_strategy_switching', 0)),
broker=broker
)
bot_thread.start()
active_bots[bot_id] = bot_thread
core/bots/trading_bot.py
+70 -30
View File
@@ -4,10 +4,10 @@ import threading
import time
import logging
from datetime import datetime
import MetaTrader5 as mt5
from core.strategies.strategy_map import STRATEGY_MAP
from core.mt5.trade import place_trade, close_trade
from core.utils.mt5 import TIMEFRAME_MAP # <-- Impor dari lokasi terpusat
from core.factory.broker_factory import BrokerFactory
# from core.mt5.trade import place_trade, close_trade <-- DEPRECATED
from core.utils.mt5 import TIMEFRAME_MAP # Keep for now or move to adapter
# AI Mentor Integration
from core.db.models import log_trade_for_ai_analysis
# Holiday and market hours management
@@ -18,7 +18,7 @@ logger = logging.getLogger(__name__)
class TradingBot(threading.Thread):
def __init__(self, id, name, market, risk_percent, sl_pips, tp_pips, timeframe, check_interval, strategy, strategy_params={}, status='Dijeda', enable_strategy_switching=False):
def __init__(self, id, name, market, risk_percent, sl_pips, tp_pips, timeframe, check_interval, strategy, strategy_params={}, status='Dijeda', enable_strategy_switching=False, broker=None):
super().__init__()
self.id = id
self.name = name
@@ -37,24 +37,48 @@ class TradingBot(threading.Thread):
self.last_analysis = {"signal": "MEMUAT", "explanation": "Bot sedang memulai, menunggu analisis pertama..."}
self._stop_event = threading.Event()
self.strategy_instance = None
self.strategy_instance = None
# Gunakan map yang diimpor untuk menjaga konsistensi
self.tf_map = TIMEFRAME_MAP
# Initialize Broker Adapter
if broker:
self.broker = broker
else:
# Default to MT5 if not provided (for backward compatibility or default behavior)
self.broker = BrokerFactory.get_broker('MT5')
# We assume credentials are in env or handled by adapter internally for now
# In a real scenario, we might pass credentials here
from dotenv import load_dotenv
import os
load_dotenv()
creds = {
'MT5_LOGIN': os.getenv('MT5_LOGIN'),
'MT5_PASSWORD': os.getenv('MT5_PASSWORD'),
'MT5_SERVER': os.getenv('MT5_SERVER')
}
self.broker.initialize(creds)
def run(self):
"""Metode utama yang dijalankan oleh thread, kini dengan eksekusi trade."""
self.status = 'Aktif'
self.log_activity('START', f"Bot '{self.name}' dimulai.", is_notification=True)
# --- PERBAIKAN: Verifikasi Simbol Cerdas ---
from core.utils.mt5 import find_mt5_symbol
self.market_for_mt5 = find_mt5_symbol(self.market)
if not self.market_for_mt5:
msg = f"Simbol '{self.market}' atau variasinya tidak dapat ditemukan/diaktifkan di Market Watch MT5."
# --- PERBAIKAN: Verifikasi Simbol Cerdas via Adapter ---
# from core.utils.mt5 import find_mt5_symbol <-- DEPRECATED
# self.market_for_mt5 = find_mt5_symbol(self.market)
# We use get_symbol_info to verify if symbol exists and get the correct name
symbol_info = self.broker.get_symbol_info(self.market)
if not symbol_info:
msg = f"Simbol '{self.market}' atau variasinya tidak dapat ditemukan/diaktifkan di Broker."
self.log_activity('ERROR', msg, is_notification=True)
self.status = 'Error'
self.last_analysis = {"signal": "ERROR", "explanation": msg}
return # Hentikan eksekusi jika simbol tidak valid
self.market_for_mt5 = symbol_info['name'] # Use the resolved name from broker
try:
strategy_class = STRATEGY_MAP.get(self.strategy_name)
@@ -71,10 +95,8 @@ class TradingBot(threading.Thread):
while not self._stop_event.is_set():
try:
# Simbol sudah diverifikasi, jadi pemeriksaan ini menjadi redundan
# if not mt5.symbol_select(self.market_for_mt5, True): ...
symbol_info = mt5.symbol_info(self.market_for_mt5) # type: ignore
# Simbol sudah diverifikasi
symbol_info = self.broker.get_symbol_info(self.market_for_mt5)
if not symbol_info:
msg = f"Tidak dapat mengambil info untuk simbol {self.market_for_mt5}."
self.log_activity('WARNING', msg)
@@ -83,9 +105,12 @@ class TradingBot(threading.Thread):
continue
# Bot sekarang yang mengambil data
from core.utils.mt5 import get_rates_mt5
tf_const = self.tf_map.get(self.timeframe, mt5.TIMEFRAME_H1)
df = get_rates_mt5(self.market_for_mt5, tf_const, 250)
# Bot sekarang yang mengambil data via Adapter
# from core.utils.mt5 import get_rates_mt5 <-- DEPRECATED
# tf_const = self.tf_map.get(self.timeframe, mt5.TIMEFRAME_H1)
# df = get_rates_mt5(self.market_for_mt5, tf_const, 250)
df = self.broker.get_rates(self.market_for_mt5, self.timeframe, 250)
if df.empty:
msg = f"Gagal mengambil data harga untuk {self.market_for_mt5}. Periksa koneksi atau ketersediaan data historis."
@@ -142,10 +167,11 @@ class TradingBot(threading.Thread):
def _get_open_position(self):
"""Mendapatkan posisi terbuka untuk bot ini berdasarkan magic number (ID bot)."""
try:
positions = mt5.positions_get(symbol=self.market_for_mt5) # type: ignore
positions = self.broker.get_open_positions()
if positions:
for pos in positions:
if pos.magic == self.id:
# Adapter should return dicts, check magic
if pos.get('magic') == self.id and pos.get('symbol') == self.market_for_mt5:
return pos
return None
except Exception as e:
@@ -215,52 +241,66 @@ class TradingBot(threading.Thread):
# Logika untuk sinyal BUY
if signal == 'BUY':
# Jika ada posisi SELL, tutup dulu
if position and position.type == mt5.ORDER_TYPE_SELL:
if position and position.get('type') == 1: # 1 is SELL in MT5, Adapter should standardize this later
self.log_activity('CLOSE SELL', "Menutup posisi JUAL untuk membuka posisi BELI.", is_notification=True)
# Log untuk AI mentor analysis
profit_loss = position.profit if hasattr(position, 'profit') else 0
profit_loss = position.get('profit', 0)
self._log_trade_for_ai_mentor(position, profit_loss, 'CLOSE_SELL')
close_trade(position)
self.broker.close_position(position['ticket'])
position = None # Reset posisi setelah ditutup
# Jika tidak ada posisi, buka posisi BUY baru
if not position:
self.log_activity('OPEN BUY', "Membuka posisi BELI berdasarkan sinyal.", is_notification=True)
place_trade(self.market_for_mt5, mt5.ORDER_TYPE_BUY, self.risk_percent, self.sl_pips, self.tp_pips, self.id, self.timeframe)
self.broker.place_order(
symbol=self.market_for_mt5,
order_type='BUY',
volume=self.risk_percent,
sl=self.sl_pips,
tp=self.tp_pips,
comment=f"Bot-{self.id}"
)
# Logika untuk sinyal SELL
elif signal == 'SELL':
# Jika ada posisi BUY, tutup dulu
if position and position.type == mt5.ORDER_TYPE_BUY:
if position and position.get('type') == 0: # 0 is BUY in MT5
self.log_activity('CLOSE BUY', "Menutup posisi BELI untuk membuka posisi JUAL.", is_notification=True)
# Log untuk AI mentor analysis
profit_loss = position.profit if hasattr(position, 'profit') else 0
profit_loss = position.get('profit', 0)
self._log_trade_for_ai_mentor(position, profit_loss, 'CLOSE_BUY')
close_trade(position)
self.broker.close_position(position['ticket'])
position = None # Reset posisi setelah ditutup
# Jika tidak ada posisi, buka posisi SELL baru
if not position:
self.log_activity('OPEN SELL', "Membuka posisi JUAL berdasarkan sinyal.", is_notification=True)
place_trade(self.market_for_mt5, mt5.ORDER_TYPE_SELL, self.risk_percent, self.sl_pips, self.tp_pips, self.id, self.timeframe)
self.broker.place_order(
symbol=self.market_for_mt5,
order_type='SELL',
volume=self.risk_percent,
sl=self.sl_pips,
tp=self.tp_pips,
comment=f"Bot-{self.id}"
)
def _log_trade_for_ai_mentor(self, position, profit_loss, action_type):
"""Log trade data untuk analisis AI mentor"""
try:
# Hitung apakah stop loss dan take profit digunakan
stop_loss_used = hasattr(position, 'sl') and position.sl > 0
take_profit_used = hasattr(position, 'tp') and position.tp > 0
stop_loss_used = position.get('sl', 0) > 0 if position else False
take_profit_used = position.get('tp', 0) > 0 if position else False
# Log ke database untuk AI analysis
log_trade_for_ai_analysis(
bot_id=self.id,
symbol=self.market_for_mt5 or self.market,
profit_loss=profit_loss,
lot_size=position.volume if hasattr(position, 'volume') else self.risk_percent,
lot_size=position.get('volume') if position else self.risk_percent,
stop_loss_used=stop_loss_used,
take_profit_used=take_profit_used,
risk_percent=self.risk_percent,
core/factory/broker_factory.py
+30
View File
@@ -0,0 +1,30 @@
from typing import Dict, Any, Optional
from core.interfaces.broker_interface import BrokerInterface
from core.adapters.mt5_adapter import MT5Adapter
from core.adapters.ccxt_adapter import CCXTAdapter
class BrokerFactory:
"""
Factory class to create and return the appropriate broker adapter
based on configuration.
"""
@staticmethod
def get_broker(broker_type: str) -> Optional[BrokerInterface]:
"""
Returns an instance of a broker adapter.
Args:
broker_type (str): The type of broker (e.g., 'MT5', 'BINANCE', 'OANDA').
Returns:
Optional[BrokerInterface]: An instance of the requested broker adapter, or None if not supported.
"""
broker_type = broker_type.upper()
if broker_type == 'MT5':
return MT5Adapter()
elif broker_type == 'CCXT' or broker_type in ['BINANCE', 'BYBIT', 'KUCOIN']:
return CCXTAdapter()
else:
raise ValueError(f"Unknown broker type: {broker_type}")
core/interfaces/broker_interface.py
+61
View File
@@ -0,0 +1,61 @@
from abc import ABC, abstractmethod
from typing import Dict, Any, List, Optional
import pandas as pd
from datetime import datetime
class BrokerInterface(ABC):
"""
Abstract Base Class for all broker adapters.
This defines the universal contract that the bot uses to interact with any broker.
"""
@abstractmethod
def initialize(self, credentials: Dict[str, Any]) -> bool:
"""
Initialize connection to the broker.
"""
pass
@abstractmethod
def get_account_info(self) -> Optional[Dict[str, Any]]:
"""
Get account balance, equity, and other info.
"""
pass
@abstractmethod
def get_rates(self, symbol: str, timeframe: str, count: int = 100) -> pd.DataFrame:
"""
Get historical price data as a DataFrame.
DataFrame must have index 'time' and columns: open, high, low, close, tick_volume.
"""
pass
@abstractmethod
def get_open_positions(self) -> List[Dict[str, Any]]:
"""
Get list of currently open positions.
"""
pass
@abstractmethod
def place_order(self, symbol: str, order_type: str, volume: float, price: float = 0.0, sl: float = 0.0, tp: float = 0.0, comment: str = "") -> bool:
"""
Place a trade order.
order_type: 'BUY' or 'SELL' (or 'BUY_LIMIT', etc.)
"""
pass
@abstractmethod
def close_position(self, position_id: str, volume: float = 0.0) -> bool:
"""
Close an existing position.
"""
pass
@abstractmethod
def get_symbol_info(self, symbol: str) -> Optional[Dict[str, Any]]:
"""
Get specifications for a symbol (min lot, max lot, tick size, etc.)
"""
pass
docker-compose.yml
+18
View File
@@ -0,0 +1,18 @@
version: '3.8'
services:
quantumbotx:
build: .
ports:
- "5000:5000"
volumes:
- .:/app
- ./instance:/app/instance # Persist database
- ./logs:/app/logs # Persist logs
environment:
- FLASK_ENV=development
- BROKER_TYPE=CCXT
- EXCHANGE_ID=binance
# - API_KEY=your_api_key
# - API_SECRET=your_api_secret
restart: unless-stopped
last_broker.json
+3 -3
View File
@@ -1,5 +1,5 @@
{
"broker": "FBS-Demo",
"company": "FBS Markets Inc.",
"last_check": "2025-10-22T14:43:57.762205"
"broker": "XMGlobal-MT5 7",
"company": "XM Global Limited",
"last_check": "2025-11-21T18:33:20.944015"
}
requirements-docker.txt
+22
View File
@@ -0,0 +1,22 @@
blinker
certifi
charset-normalizer
click
colorama
Flask
idna
itsdangerous
Jinja2
MarkupSafe
numpy
pandas
pandas_ta
python-dateutil
python-dotenv
pytz
requests
six
tzdata
urllib3
Werkzeug
ccxt
requirements.txt
+1
View File
@@ -20,3 +20,4 @@ six==1.17.0
tzdata==2025.2
urllib3==2.5.0
Werkzeug==3.1.3
ccxt
run.py
+1 -1
View File
@@ -29,7 +29,7 @@ app = create_app()
@app.route('/api/health')
def health_check():
"""Endpoint untuk memastikan server berjalan."""
mt5_status = "MT5 connected" if mt5.isinitialize() else "MT5 not connected" # pyright: ignore[reportAttributeAccessIssue]
mt5_status = "MT5 connected" if mt5.terminal_info() is not None else "MT5 not connected" # pyright: ignore[reportAttributeAccessIssue]
return jsonify({"status": "ok", "message": "Server is running", "mt5": mt5_status})
if __name__ == '__main__':
test_adapter.py
+68
View File
@@ -0,0 +1,68 @@
import os
import logging
from dotenv import load_dotenv
from core.factory.broker_factory import BrokerFactory
# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("TestAdapter")
def main():
# Load environment variables
load_dotenv()
# Get credentials
mt5_login = os.getenv('MT5_LOGIN')
mt5_password = os.getenv('MT5_PASSWORD')
mt5_server = os.getenv('MT5_SERVER')
if not all([mt5_login, mt5_password, mt5_server]):
logger.error("Missing MT5 credentials in .env file")
return
credentials = {
'MT5_LOGIN': mt5_login,
'MT5_PASSWORD': mt5_password,
'MT5_SERVER': mt5_server
}
# 1. Use Factory to get Adapter
logger.info("1. Requesting MT5 adapter from Factory...")
broker = BrokerFactory.get_broker('MT5')
if not broker:
logger.error("Failed to get broker adapter!")
return
logger.info(" Success: Got MT5Adapter instance.")
# 2. Initialize Connection
logger.info("2. Initializing connection...")
if broker.initialize(credentials):
logger.info(" Success: Connected to MT5.")
else:
logger.error(" Failed: Could not connect to MT5.")
return
# 3. Get Account Info
logger.info("3. Fetching account info...")
info = broker.get_account_info()
if info:
logger.info(f" Success: Balance = {info.get('balance')}, Equity = {info.get('equity')}")
else:
logger.error(" Failed: Could not fetch account info.")
# 4. Get Rates (Test Data Fetching)
symbol = "XAUUSD" # Or any symbol you know exists
logger.info(f"4. Fetching rates for {symbol}...")
rates = broker.get_rates(symbol, "H1", count=5)
if not rates.empty:
logger.info(f" Success: Fetched {len(rates)} rows.")
print(rates.head())
else:
logger.warning(f" Warning: Could not fetch rates for {symbol} (Market might be closed or symbol wrong).")
logger.info("Test Complete.")
if __name__ == "__main__":
main()
test_ccxt.py
+49
View File
@@ -0,0 +1,49 @@
import logging
from core.factory.broker_factory import BrokerFactory
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("TestCCXT")
def test_ccxt():
logger.info("1. Requesting CCXT adapter from Factory...")
try:
broker = BrokerFactory.get_broker('CCXT')
logger.info(" Success: Got CCXTAdapter instance.")
except Exception as e:
logger.error(f" Failed: {e}")
return
logger.info("2. Initializing connection (Binance Public)...")
# No keys needed for public data
creds = {
'EXCHANGE_ID': 'binance',
'API_KEY': '',
'API_SECRET': ''
}
if broker.initialize(creds):
logger.info(" Success: Connected to Binance.")
else:
logger.error(" Failed: Could not connect.")
return
logger.info("3. Fetching Rates for BTC/USDT...")
try:
df = broker.get_rates('BTC/USDT', 'H1', 10)
if not df.empty:
logger.info(f" Success: Fetched {len(df)} rows.")
print(df.head())
else:
logger.error(" Failed: DataFrame is empty.")
except Exception as e:
logger.error(f" Failed: {e}")
logger.info("4. Getting Symbol Info...")
info = broker.get_symbol_info('BTC/USDT')
if info:
logger.info(f" Success: {info}")
else:
logger.error(" Failed: Could not get symbol info.")
if __name__ == "__main__":
test_ccxt()