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 Broker Agnostic Architecture v2.1.0 and Python 3.13 optimization

Browse Source 
- Refactored TradingBot to use BrokerInterface for universal compatibility (MT5, CCXT, Mock).
- Integrated CCXT with support for Binance Demo Trading.
- Updated dependencies for Python 3.13 and fixed installation CPU overhead.
- Added comprehensive testing tools: test_agnostic_bot.py, visual_simulation.py.
- Updated documentation (README, Changelog, Roadmap) to reflect Agnostic Revolution.
This commit is contained in:
Reynov Christian
2025-12-31 14:58:36 +08:00
parent 72bcc1f284
commit f0b5a07545
16 changed files with 733 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+25
View File
@@ -8,6 +8,31 @@
---
## 📈 **v2.1.0 - "Agnostic Revolution"** ✨ (December 2025)
### 🌍 **Broker Agnostic Architecture** ⭐⭐⭐⭐⭐
- **Universal Broker Interface**: Complete abstraction of broker logic via `BrokerInterface`.
- **Multi-Platform Support**: Now supports Crypto Exchanges (Binance, Bybit) via CCXT integration.
- **Docker & Cloud Ready**: Crypto bots can now run on Linux/Docker without needing a local MT5 terminal.
- **Broker Factory**: Dynamic switching between MT5, CCXT, and Mock adapters.
- **Stateful Mock Testing**: Advanced simulation mode with `visual_simulation.py` for risk-free testing.
### ⚡ **System & Performance Optimization**
- **Python 3.13 Support**: Optimized for the latest stable Python versions (3.10 - 3.13).
- **Dependency Refresh**: Fixed "CPU 100%" issue during installation by optimizing `requirements.txt`.
- **Binary Wheels**: Configured dependencies to use pre-compiled binaries for faster setup.
- **Agnostic Symbol Mapping**: Automatic translation between MT5 (EURUSD) and CCXT (BTC/USDT) formats.
### 🧪 **New Testing Framework**
- **Agnostic Bot Validator**: `test_agnostic_bot.py` for cross-platform logic verification.
- **CCXT Connection Suite**: Dedicated tools for validating exchange connectivity and demo trading.
- **Visual Simulation Engine**: Real-time market mock-up for strategy debugging.
---
## 🎯 **Investment Highlights**
**Why QuantumBotX Stands Out:**
QUICK_START_GUIDE.md
+22 -16
View File
@@ -1,12 +1,16 @@
# QuantumBotX - Quick Start Guide
## First Time Setup
## Setup Your Broker
1. **Install MetaTrader 5** (Required)
- Download from: <https://www.metatrader5.com/>
- Install and create a demo account
- Keep MT5 running in the background
- ⚠️ **IMPORTANT:** MetaTrader 5 must be running for QuantumBotX to work
- **Option A: MetaTrader 5 (Forex/Gold)**
- Download MT5 from: <https://www.metatrader5.com/>
- Install and keep it running in the background.
- **Option B: Crypto Exchange (Binance/Bybit)**
- Create an account on your preferred exchange.
- For testing, use **Binance Futures Testnet**.
⚠️ **Note:** MT5 is only required if you choose to trade Forex/Gold via `MT5` broker type.
2. **Configure Your Settings**
- Copy `.env.example` to `.env`
@@ -18,18 +22,20 @@
MT5_SERVER=your_server_name
```
3. **Start the Application**
3. **Verify Connection**
- To test MT5: `python test_mt5_connection.py`
- To test Crypto: `python test_ccxt.py`
- To simulate without Internet: `python visual_simulation.py`
4. **Start the Application**
- Double-click `start.bat` (Windows)
- Open <http://127.0.0.1:5000> in your browser
- Open <http://127.0.0.1:5000>
## ✅ System Requirements
- **Windows 7 SP1 or later** (64-bit recommended)
- **MetaTrader 5** (must be installed separately)
- **4GB RAM minimum** (8GB recommended)
- **500MB free disk space**
- **Internet connection** for initial setup
- **❌ Python NOT required** (already bundled in the installer)
- **Windows 10/11** (Native MT5 support)
- **Linux/MacOS/Docker** (CCXT/Crypto support only)
- **Python 3.10 - 3.13**
- **4GB RAM minimum**
- **Stable internet connection**
## Daily Use
README.md
+36 -44
View File
@@ -1,28 +1,20 @@
# 🤖 QuantumBotX — AI-Powered Modular Trading Bot for MT5
# 🤖 QuantumBotX — AI-Powered Broker Agnostic Trading Bot
!MIT License
!Python Version
!Framework
!Made with Love
Welcome to **QuantumBotX**, your personal, modular, and smart trading assistant built with Python and MetaTrader5 (MT5).
Designed to be elegant, powerful, and flexible — whether you're a scalper, swing trader, or a strategy researcher.
Welcome to **QuantumBotX**, your personal, modular, and smart trading assistant. Now powered by a **Broker Agnostic Architecture**, allowing you to trade across MetaTrader5 (MT5) and major Crypto Exchanges (Binance, Bybit, etc.) using a single unified interface.
---
## ⚠️ Platform Support Notice
## 🌍 Platform & Broker Support
### **Primary Platform: Windows** 🪟
### **Flexible Deployment** 🚀
- **Windows**: Native support for MT5 and CCXT.
- **Linux/Docker/Cloud**: Full support for Crypto Trading via CCXT (no MT5 required!).
- **Local MT5**: Requires Windows (or Wine) for Forex/Gold trading.
This version of QuantumBotX is **optimized for Windows** and requires MetaTrader 5 terminal to be installed locally. It's designed for learning algorithmic trading on your personal computer.
### **Alternative Platforms** 🔄
- **Linux**: Can attempt using Wine (experimental - see Linux Setup guide below)
- **macOS**: Not officially supported (requires Wine or Windows VM)
- **Cloud/VPS**: Not compatible (requires local MT5 terminal)
> 💡 **Pro Tip**: For cloud deployment and multi-platform support, check out our upcoming **QuantumBotX API** version!
### **Supported Brokers** 🏦
- **MetaTrader 5**: XM, Exness, FBS, IC Markets, etc.
- **Crypto Exchanges (via CCXT)**: Binance (Spot/Futures), Bybit, OKX, and 100+ others.
- **Simulation**: Built-in **Mock Broker** for risk-free strategy testing without internet.
---
@@ -90,38 +82,28 @@ This version of QuantumBotX is **optimized for Windows** and requires MetaTrader
---
## 🚀 Development & Testing Framework
### 🧪 **Testing & Simulation Infrastructure**
### 🧪 **Testing Infrastructure**
- ✅ **30+ Test Scripts**: Comprehensive testing suite in dedicated `testing/` directory
- ✅ **Multi-Broker Testing**: XM Global, Exness, Alpari compatibility validation
- ✅ **Strategy Validation**: Individual strategy testing and parameter optimization
- ✅ **ATR Education Testing**: Interactive examples and beginner tutorials
- ✅ **Crypto Integration Tests**: Bitcoin/Ethereum weekend mode validation
- ✅ **Indonesian Market Tests**: XM Indonesia and IDR pairs testing
- ✅ **Risk Management Tests**: XAUUSD protection and ATR-based sizing validation
- ✅ **Agnostic Testing**: `test_agnostic_bot.py` validates logic across different broker types.
- ✅ **Visual Simulation**: `visual_simulation.py` provides a real-time "Mock Market" for strategy debugging.
- ✅ **CCXT Validation**: dedicated `test_ccxt.py` and `test_ccxt_order.py` for exchange connectivity.
- ✅ **30+ Test Scripts**: Comprehensive testing suite in dedicated `testing/` directory.
### 🔧 **Development Tools**
- ✅ **Symbol Migration Tools**: Automatic broker symbol discovery and mapping
- ✅ **Bot State Management**: Debug and fix tools for bot recovery
- ✅ **Performance Analysis**: Backtesting debugging and optimization tools
- ✅ **Market Diagnostics**: Real-time market condition analysis
- ✅ **Integration Demos**: Complete workflow demonstrations
- ✅ **Broker Factory**: Dynamic adapter switching between MT5, CCXT, and Mock providers.
- ✅ **Symbol Discovery**: Automatic mapping between Forex (EURUSD) and Crypto (BTC/USDT) formats.
> **Note**: All testing scripts are excluded from git repository for clean production deployment
---
## 📦 Tech Stack
- `Python 3.10+`
- `Flask` & `TailwindCSS`
- `Python 3.10 - 3.13` (Recommended: 3.13 for best library compatibility)
- `CCXT` (Crypto Exchange Hybrid Integration)
- `Flask` & `Vanilla CSS` (Modern Aesthetic)
- `MetaTrader5` Python Integration
- `pandas` & `pandas-ta` for data analysis
- `Chart.js` for data visualization
- `SQLite` for database
- `pandas` & `pandas-ta` (Financial Engineering)
- `Chart.js` (Simulasi & Result Visualization)
- `SQLite` (Local Database)
---
@@ -211,14 +193,24 @@ This version of QuantumBotX is **optimized for Windows** and requires MetaTrader
---
## 🔐 Environment Variables (`.env`)
Rename `.env.example` to `.env`, and fill in the following:
```env
# --- BROKER SELECTION ---
BROKER_TYPE="MT5" # Options: MT5, CCXT, MOCK
# --- MT5 CONFIG (If MT5 selected) ---
MT5_LOGIN="your_mt5_login"
MT5_PASSWORD="your_password"
MT5_SERVER="your_broker_server"
# --- CCXT CONFIG (If CCXT selected) ---
EXCHANGE_ID="binance"
CCXT_API_KEY="your_api_key"
CCXT_API_SECRET="your_api_secret"
CCXT_TESTNET=true
# --- APP CONFIG ---
SECRET_KEY="any_flask_secret_key"
DB_NAME=bots.db
```
README_NEW.md
+1
View File
@@ -193,6 +193,7 @@ python run.py
### Current Features ✅
- ✅ **Agnostic Architecture**: Support for MT5, CCXT (Binance), and Mock Brokers
- ✅ MT5 Integration with 50+ instruments
- ✅ 16 Trading strategies with risk management
- ✅ AI mentor in Indonesian
ROADMAP.md
+13 -5
View File
@@ -4,12 +4,20 @@
## 🎯 **What's Coming Next**
### **Q4 2025: Intelligence Enhancement**
### **Q4 2025: Agnostic Revolution (RELEASED)** 🚀
We've officially laid the foundation for the **QuantumBotX API** by implementing a Broker Agnostic Architecture:
- **Advanced AI Features**: Enhanced strategy analysis with machine learning
- **Real-time Notifications**: Telegram integration for trade alerts
- **Portfolio Analytics**: Advanced performance dashboards
- **Enterprise Features**: Multi-account management and compliance logging
- ✅ **Multi-Broker Support**: Dynamic switching between MT5, Binance, and Mock adapters via `BrokerFactory`.
- ✅ **Cloud Ready**: Run on Linux/Docker for Crypto trading without local MT5.
- ✅ **Stateful Simulation**: Real-time mock testing with `visual_simulation.py`.
- ✅ **Modern Foundation**: Optimized for Python 3.13 with seamingless dependency management.
### **Q1 2026: Intelligence Enhancement (NEXT)**
- 🔄 **Telegram Notifications**: Real-time trade and error alerts.
- 🔄 **Portfolio Analytics**: Advanced performance dashboards.
- 🔄 **AI Strategy Optimizer**: Automated parameter tuning based on market regime.
### **Exciting New Project** 🚀
core/adapters/ccxt_adapter.py
+9 -8
View File
@@ -29,20 +29,21 @@ class CCXTAdapter(BrokerInterface):
'options': {'defaultType': 'future'} # Default to futures for bots
}
# Enable testnet if configured
if credentials.get('PASSWORD'):
config['password'] = credentials.get('PASSWORD')
# Enable testnet/demo if configured
if credentials.get('TESTNET', False):
config['options']['demo'] = True
if self.exchange_id == 'binance':
# Manually point to Futures Testnet URL to be extra safe
config['urls'] = {
'api': {
'public': 'https://testnet.binance.vision/api',
'private': 'https://testnet.binance.vision/api',
'public': 'https://testnet.binancefuture.com/fapi/v1',
'private': 'https://testnet.binancefuture.com/fapi/v1',
}
}
logger.info("Using Binance TESTNET (https://testnet.binance.vision)")
# Add other exchange testnet URLs as needed
if credentials.get('PASSWORD'): # For exchanges like KuCoin
config['password'] = credentials.get('PASSWORD')
logger.info(f"Using {self.exchange_id} DEMO TRADING mode")
self.exchange = exchange_class(config)
core/adapters/mt5_adapter.py
+48 -29
View File
@@ -7,68 +7,79 @@ from typing import Dict, Any, List, Optional
try:
from core.utils.mt5 import (
initialize_mt5,
shutdown_mt5,
get_symbol_info,
get_rates as get_rates_mt5,
place_trade,
close_trade,
get_open_positions,
get_rates_mt5,
get_account_info_mt5,
get_open_positions_mt5,
find_mt5_symbol,
TIMEFRAME_MAP,
get_todays_profit_mt5
)
# Corrected import for trade functions
from core.mt5.trade import place_trade as mt5_place_trade, close_trade as mt5_close_trade
import MetaTrader5 as mt5
MT5_AVAILABLE = True
except ImportError:
MT5_AVAILABLE = False
logging.warning("MetaTrader5 module not found. MT5Adapter will not work.")
logger.error("MetaTrader5 module or dependencies not found. MT5Adapter will not work.")
logger = logging.getLogger(__name__)
class MT5Adapter(BrokerInterface):
"""
Adapter for MetaTrader 5 using the official python library.
Wraps the functions from core.utils.mt5.
Wraps the functions from core.utils.mt5 and core.mt5.trade.
"""
def initialize(self, credentials: Dict[str, Any]) -> bool:
# MT5 usually initialized via run.py, but we can support re-init here
return True
"""Login to MT5 if provided credentials, else assume already initialized."""
if not credentials:
return mt5.initialize() if MT5_AVAILABLE else False
login = credentials.get('MT5_LOGIN') or credentials.get('login')
password = credentials.get('MT5_PASSWORD') or credentials.get('password')
server = credentials.get('MT5_SERVER') or credentials.get('server', 'MetaQuotes-Demo')
if login and password:
return initialize_mt5(int(login), password, server)
return mt5.initialize() if MT5_AVAILABLE else False
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()
mt5_positions = get_open_positions_mt5()
standardized_positions = []
for pos in mt5_positions:
# Map MT5 type (0 for Buy, 1 for Sell) to string
standardized_type = 'BUY' if pos.get('type') == mt5.POSITION_TYPE_BUY else 'SELL'
pos['type'] = standardized_type
standardized_positions.append(pos)
return standardized_positions
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
action = mt5.TRADE_ACTION_DEAL
type_op = mt5.ORDER_TYPE_BUY if order_type == 'BUY' else mt5.ORDER_TYPE_SELL
mt5_order_type = mt5.ORDER_TYPE_BUY if order_type == 'BUY' else mt5.ORDER_TYPE_SELL
# We use the existing place_trade logic but wrap the arguments
# Wait, the existing place_trade uses ATR multipliers.
# For the universal adapter, we want raw SL/TP values.
request = {
"action": action,
"action": mt5.TRADE_ACTION_DEAL,
"symbol": valid_symbol,
"volume": volume,
"type": type_op,
"type": mt5_order_type,
"price": mt5.symbol_info_tick(valid_symbol).ask if order_type == 'BUY' else mt5.symbol_info_tick(valid_symbol).bid,
"sl": sl,
"tp": tp,
@@ -80,21 +91,29 @@ class MT5Adapter(BrokerInterface):
}
result = mt5.order_send(request)
if result.retcode != mt5.TRADE_RETCODE_DONE:
logger.error(f"Order failed: {result.comment}")
if result is None or result.retcode != mt5.TRADE_RETCODE_DONE:
logger.error(f"Order failed: {result.comment if result else 'No result'}")
return False
logger.info(f"Order placed: {result.order}")
return True
def close_position(self, position_id: str, volume: float = 0.0) -> bool:
def close_position(self, ticket_id: Any, volume: float = 0.0) -> bool:
"""Close an existing position in MT5."""
try:
# Logic to close position...
# For now, returning False as placeholder
pass
except:
pass
return False
# Find the position by ticket
positions = mt5.positions_get(ticket=int(ticket_id))
if not positions:
logger.warning(f"Position #{ticket_id} not found to close.")
return False
position = positions[0]
# Use the existing close_trade utility
result, msg = mt5_close_trade(position)
return result is not None
except Exception as e:
logger.error(f"Error closing position {ticket_id}: {e}")
return False
def get_symbol_info(self, symbol: str) -> Optional[Dict[str, Any]]:
valid_symbol = find_mt5_symbol(symbol)
core/bots/trading_bot.py
+3 -7
View File
@@ -6,8 +6,7 @@ import logging
from datetime import datetime
from core.strategies.strategy_map import STRATEGY_MAP
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
# from core.mt5.trade import place_trade, close_trade # DEPRECATED
# AI Mentor Integration
from core.db.models import log_trade_for_ai_analysis
# Holiday and market hours management
@@ -37,9 +36,6 @@ 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:
@@ -241,7 +237,7 @@ class TradingBot(threading.Thread):
# Logika untuk sinyal BUY
if signal == 'BUY':
# Jika ada posisi SELL, tutup dulu
if position and position.get('type') == 1: # 1 is SELL in MT5, Adapter should standardize this later
if position and position.get('type') == 'SELL':
self.log_activity('CLOSE SELL', "Menutup posisi JUAL untuk membuka posisi BELI.", is_notification=True)
# Log untuk AI mentor analysis
@@ -266,7 +262,7 @@ class TradingBot(threading.Thread):
# Logika untuk sinyal SELL
elif signal == 'SELL':
# Jika ada posisi BUY, tutup dulu
if position and position.get('type') == 0: # 0 is BUY in MT5
if position and position.get('type') == 'BUY':
self.log_activity('CLOSE BUY', "Menutup posisi BELI untuk membuka posisi JUAL.", is_notification=True)
# Log untuk AI mentor analysis
core/brokers/broker_factory.py
+4 -1
View File
@@ -9,6 +9,7 @@ from typing import Dict, Optional, List
from enum import Enum
from .base_broker import BaseBroker
from .mt5_broker import MT5Broker
from .binance_broker import BinanceBroker
from .ctrader_broker import CTraderBroker
from .interactive_brokers import InteractiveBrokersBroker
@@ -65,7 +66,9 @@ class BrokerFactory:
config = broker_config['config']
try:
if broker_type == BrokerType.BINANCE:
if broker_type == BrokerType.MT5:
broker = MT5Broker()
elif broker_type == BrokerType.BINANCE:
broker = BinanceBroker(testnet=config.get('testnet', True))
elif broker_type == BrokerType.BINANCE_FUTURES:
# Future implementation
core/brokers/mt5_broker.py
+198
View File
@@ -0,0 +1,198 @@
# core/brokers/mt5_broker.py
"""
MetaTrader 5 Broker Implementation
Connects QuantumBotX to MT5 terminals via the universal BaseBroker interface.
"""
import logging
import pandas as pd
import MetaTrader5 as mt5
from typing import Dict, List, Optional, Union
from datetime import datetime, timedelta
from .base_broker import BaseBroker, OrderType, OrderStatus, Timeframe, Position, Order, AccountInfo
from core.utils.mt5 import get_rates_mt5, TIMEFRAME_MAP as MT5_TIMEFRAME_MAP
logger = logging.getLogger(__name__)
class MT5Broker(BaseBroker):
"""
MT5 Implementation of the BaseBroker.
Wraps MetaTrader5 library calls into a unified API.
"""
def __init__(self, broker_name: str = "MetaTrader 5"):
super().__init__(broker_name)
self.timeframe_map = {
Timeframe.M1: mt5.TIMEFRAME_M1,
Timeframe.M5: mt5.TIMEFRAME_M5,
Timeframe.M15: mt5.TIMEFRAME_M15,
Timeframe.M30: mt5.TIMEFRAME_M30,
Timeframe.H1: mt5.TIMEFRAME_H1,
Timeframe.H4: mt5.TIMEFRAME_H4,
Timeframe.D1: mt5.TIMEFRAME_D1
}
def connect(self, credentials: Dict) -> bool:
"""Connect to MT5 terminal"""
try:
login = credentials.get('login')
password = credentials.get('password')
server = credentials.get('server', 'MetaQuotes-Demo')
if not mt5.initialize(login=int(login), password=password, server=server):
logger.error(f"MT5 initialization failed: {mt5.last_error()}")
self.is_connected = False
return False
self.is_connected = True
self.supported_symbols = [s.name for s in mt5.symbols_get()]
logger.info("MT5 connected successfully.")
return True
except Exception as e:
logger.error(f"Error connecting to MT5: {e}")
return False
def disconnect(self) -> bool:
"""Disconnect from MT5"""
mt5.shutdown()
self.is_connected = False
return True
def get_symbols(self) -> List[str]:
"""Get list of available trading symbols"""
if not self.is_connected:
return []
symbols = mt5.symbols_get()
return [s.name for s in symbols] if symbols else []
def get_market_data(self, symbol: str, timeframe: Timeframe,
count: int = 500) -> pd.DataFrame:
"""Get OHLCV market data from MT5"""
mt5_tf = self.timeframe_map.get(timeframe, mt5.TIMEFRAME_H1)
return get_rates_mt5(symbol, mt5_tf, count)
def get_current_price(self, symbol: str) -> Dict[str, float]:
"""Get current bid/ask prices"""
tick = mt5.symbol_info_tick(symbol)
if tick:
return {"bid": tick.bid, "ask": tick.ask}
return {"bid": 0.0, "ask": 0.0}
def place_order(self, symbol: str, order_type: OrderType, side: str,
size: float, price: Optional[float] = None,
stop_loss: Optional[float] = None,
take_profit: Optional[float] = None) -> Optional[Order]:
"""Place a trading order in MT5"""
if not self.is_connected:
return None
# Map OrderType to MT5 constant
mt5_type = None
if order_type == OrderType.MARKET_BUY:
mt5_type = mt5.ORDER_TYPE_BUY
elif order_type == OrderType.MARKET_SELL:
mt5_type = mt5.ORDER_TYPE_SELL
# ... support other types as needed
if mt5_type is None:
logger.error(f"Unsupported order type for MT5: {order_type}")
return None
curr_price = price or (self.get_current_price(symbol)['ask'] if mt5_type == mt5.ORDER_TYPE_BUY else self.get_current_price(symbol)['bid'])
request = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": symbol,
"volume": size,
"type": mt5_type,
"price": curr_price,
"sl": stop_loss or 0.0,
"tp": take_profit or 0.0,
"magic": 2024001, # Default magic number
"comment": "QuantumBotX Trade",
"type_time": mt5.ORDER_TIME_GTC,
"type_filling": mt5.ORDER_FILLING_FOK,
}
result = mt5.order_send(request)
if result and result.retcode == mt5.TRADE_RETCODE_DONE:
order = Order(str(result.order), symbol, order_type, side, size, curr_price)
order.status = OrderStatus.FILLED
return order
else:
logger.error(f"MT5 Order failed: {result.comment if result else 'Unknown error'}")
return None
def cancel_order(self, order_id: str) -> bool:
"""Cancel an existing order (MT5 usually handles this via close or delete pending)"""
# Simplistic implementation for now
return False
def get_positions(self) -> List[Position]:
"""Get all open positions from MT5"""
mt5_positions = mt5.positions_get()
positions = []
if mt5_positions:
for p in mt5_positions:
side = 'long' if p.type == mt5.POSITION_TYPE_BUY else 'short'
positions.append(Position(
symbol=p.symbol,
side=side,
size=p.volume,
entry_price=p.price_open,
current_price=p.price_current,
unrealized_pnl=p.profit
))
return positions
def get_orders(self) -> List[Order]:
"""Get all pending orders from MT5"""
mt5_orders = mt5.orders_get()
orders = []
if mt5_orders:
for o in mt5_orders:
# Map MT5 order types back to our OrderType
# This is a simplification
order_type = OrderType.LIMIT_BUY if o.type == mt5.ORDER_TYPE_BUY_LIMIT else OrderType.LIMIT_SELL
orders.append(Order(
order_id=str(o.ticket),
symbol=o.symbol,
order_type=order_type,
side='buy' if 'BUY' in order_type.name else 'sell',
size=o.volume_initial,
price=o.price_open
))
return orders
def get_account_info(self) -> AccountInfo:
"""Get account information from MT5"""
inf = mt5.account_info()
if inf:
return AccountInfo(
balance=inf.balance,
equity=inf.equity,
margin=inf.margin,
free_margin=inf.margin_free,
margin_level=inf.margin_level,
currency=inf.currency
)
return AccountInfo(0, 0, 0, 0, 0)
def get_trade_history(self, days: int = 30) -> List[Dict]:
"""Get trade history from MT5"""
from_date = datetime.now() - timedelta(days=days)
history = mt5.history_deals_get(from_date, datetime.now())
deals = []
if history:
for d in history:
deals.append({
"ticket": d.ticket,
"symbol": d.symbol,
"type": d.type,
"volume": d.volume,
"price": d.price,
"profit": d.profit,
"time": datetime.fromtimestamp(d.time)
})
return deals
requirements-mt5.txt
+1
View File
@@ -0,0 +1 @@
MetaTrader5==5.0.5120
requirements.txt
+3 -4
View File
@@ -8,10 +8,9 @@ idna==3.10
itsdangerous==2.2.0
Jinja2==3.1.6
MarkupSafe==3.0.2
MetaTrader5==5.0.5120
numpy==1.23.5
pandas==2.3.1
pandas_ta==0.3.14b0
numpy>=2.2.6
pandas>=2.2.3
pandas-ta
python-dateutil==2.9.0.post0
python-dotenv==1.1.1
pytz==2025.2
test_agnostic_bot.py
+142
View File
@@ -0,0 +1,142 @@
import logging
import pandas as pd
from core.bots.trading_bot import TradingBot
from core.interfaces.broker_interface import BrokerInterface
# Configure logging
logging.basicConfig(level=logging.INFO)
class MockBroker(BrokerInterface):
def __init__(self):
self.positions = []
self._ticket_counter = 1000
def initialize(self, credentials):
return True
def get_account_info(self):
return {'balance': 10000, 'equity': 10000}
def get_rates(self, symbol, timeframe, count=100):
# Mengembalikan DataFrame dengan format yang tepat
# Dibuat lebih banyak baris (count) untuk simulasi indikator strategi
data = {
'time': pd.date_range(end=pd.Timestamp.now(), periods=count, freq='H'),
'open': [1.0] * count,
'high': [1.1] * count,
'low': [0.9] * count,
'close': [1.05] * count,
'tick_volume': [100] * count
}
return pd.DataFrame(data)
def get_open_positions(self):
return self.positions
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 = ""):
self._ticket_counter += 1
# Ekstrak magic number dari comment (Bot-ID) agar bot bisa mengenali posisinya
magic = int(comment.split('-')[1]) if 'Bot-' in comment else 0
new_pos = {
'ticket': self._ticket_counter,
'symbol': symbol,
'type': order_type,
'volume': volume,
'price': price,
'sl': sl,
'tp': tp,
'magic': magic,
'profit': 10.5 # Dummy profit
}
self.positions.append(new_pos)
print(f"MOCK ORDER PLACED: {new_pos}")
return True
def close_position(self, position_id, volume: float = 0.0):
# Menghapus posisi dari list internal berdasarkan ticket
self.positions = [p for p in self.positions if str(p['ticket']) != str(position_id)]
print(f"MOCK POSITION CLOSED: ID {position_id}")
return True
def get_symbol_info(self, symbol):
return {'name': symbol, 'digits': 5}
def get_todays_profit(self):
return 50.0
def test_bot_initialization():
print("Testing Bot initialization with Mock Broker...")
mock_broker = MockBroker()
bot = TradingBot(
id=999,
name="MockBot",
market="EURUSD",
risk_percent=1.0,
sl_pips=50,
tp_pips=100,
timeframe="H1",
check_interval=1,
strategy="MA_CROSSOVER", # Real strategy from STRATEGY_MAP
broker=mock_broker
)
# We won't actually start the thread in this test to avoid loop
print(f"Bot '{bot.name}' initialized successfully with broker {bot.broker.__class__.__name__}")
assert bot.broker == mock_broker
print("Initialization Test passed!")
def test_full_trade_loop():
print("\n--- Testing Full Trade Loop (Simulation) ---")
mock_broker = MockBroker()
# Setup bot
bot = TradingBot(
id=777,
name="LoopTester",
market="EURUSD",
risk_percent=0.1,
sl_pips=50,
tp_pips=100,
timeframe="H1",
check_interval=0.1,
strategy="MA_CROSSOVER",
broker=mock_broker
)
# Pre-setup manually for testing internal logic
bot.market_for_mt5 = "EURUSD"
from core.strategies.strategy_map import STRATEGY_MAP
bot.strategy_instance = STRATEGY_MAP["MA_CROSSOVER"](bot_instance=bot)
# 1. Simulasikan Sinyal BUY
print("\n[Step 1] Simulating BUY Signal...")
# Bot handle signal (no position yet)
bot._handle_trade_signal('BUY', None)
positions = mock_broker.get_open_positions()
assert len(positions) == 1
assert positions[0]['type'] == 'BUY'
print(f"Verified: Position opened successfully. Ticket: {positions[0]['ticket']}")
# 2. Simulasikan Sinyal SELL (Bot harus tutup BUY dulu baru buka SELL)
print("\n[Step 2] Simulating SELL Signal while BUY is open...")
# Get current position like the bot loop does
current_pos = bot._get_open_position()
assert current_pos is not None
# Trigger signal handler
bot._handle_trade_signal('SELL', current_pos)
# Verify positions after switch
positions = mock_broker.get_open_positions()
assert len(positions) == 1
assert positions[0]['type'] == 'SELL'
print("Verification: Old BUY position was closed and new SELL position was opened!")
print("\nFull Trade Loop Test Passed!")
if __name__ == "__main__":
test_bot_initialization()
test_full_trade_loop()
test_ccxt.py
+29 -36
View File
@@ -1,49 +1,42 @@
import logging
import os
from dotenv import load_dotenv
from core.factory.broker_factory import BrokerFactory
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("TestCCXT")
load_dotenv()
def test_ccxt():
logger.info("1. Requesting CCXT adapter from Factory...")
def test_ccxt_minimal():
# 1. Pastikan env terisi
exchange = os.getenv('EXCHANGE_ID', 'binance')
logger.info(f"Targeting Exchange: {exchange}")
# 2. Ambil dari Factory (Factory akan otomatis inisialisasi pake .env)
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())
broker = BrokerFactory.get_broker('CCXT', exchange_id=exchange)
if broker and broker.exchange:
logger.info(f"✅ Instance {broker.__class__.__name__} ready.")
else:
logger.error(" Failed: DataFrame is empty.")
logger.error("❌ Failed to get initialized broker instance.")
return
except Exception as e:
logger.error(f" Failed: {e}")
logger.error(f"❌ Factory Error: {e}")
return
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.")
# 3. Test Ambil Data (Public)
symbol = 'BTC/USDT'
logger.info(f"Fetching Rates for {symbol}...")
try:
df = broker.get_rates(symbol, 'H1', 5)
if not df.empty:
logger.info("✅ Profit! Data received:")
print(df[['time', 'close', 'tick_volume']].tail())
else:
logger.warning("⚠️ Connected, but DataFrame is empty. Check if symbol exists.")
except Exception as e:
logger.error(f"❌ Request Error: {e}")
if __name__ == "__main__":
test_ccxt()
test_ccxt_minimal()
test_ccxt_order.py
+64
View File
@@ -0,0 +1,64 @@
import logging
import os
from dotenv import load_dotenv
from core.factory.broker_factory import BrokerFactory
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("FirstOrder")
load_dotenv()
def test_place_live_order():
exchange_id = os.getenv('EXCHANGE_ID', 'binance')
logger.info(f"1. Init Broker: {exchange_id} (Testnet)")
try:
# BrokerFactory akan inisialisasi pake .env (API Key, Secret, dll)
broker = BrokerFactory.get_broker('CCXT', exchange_id=exchange_id)
if not broker or not broker.exchange:
logger.error("❌ Gagal inisialisasi Broker. Cek .env Anda!")
return
symbol = 'BTC/USDT'
logger.info(f"2. Mengirim Order: BUY {symbol}")
# Kita coba beli jumlah kecil, misal 0.01 BTC (satuan kontrak di Futures)
success = broker.place_order(
symbol=symbol,
order_type='BUY',
volume=0.01,
comment="Order dari QuantumBotX"
)
if success:
logger.info("✅ BERHASIL! Order telah dieksekusi di Exchange.")
# Beri jeda sebentar agar data di bursa ter-update
import time
logger.info("Menunggu data posisi terupdate...")
time.sleep(2)
# Cek daftar posisi terbuka
positions = broker.get_open_positions()
print("\n" + "="*50)
print("POSISI TERBUKA SAAT INI:")
print("="*50)
if not positions:
print("Tidak ada posisi aktif (mungkin langsung tertutup atau error).")
for pos in positions:
side_str = "BUY (Long)" if pos['type'] == 0 else "SELL (Short)"
print(f"- Symbol: {pos['symbol']}")
print(f" Type : {side_str}")
print(f" Volume: {pos['volume']}")
print(f" Price : {pos['price']}")
print(f" Profit: {pos['profit']} USDT")
print("-" * 20)
else:
logger.error("❌ Gagal menempatkan order. Cek log error di atas.")
except Exception as e:
logger.error(f"❌ Terjadi kesalahan fatal: {e}", exc_info=True)
if __name__ == "__main__":
test_place_live_order()
visual_simulation.py
+135
View File
@@ -0,0 +1,135 @@
import logging
import time
import pandas as pd
import random
from core.bots.trading_bot import TradingBot
from core.interfaces.broker_interface import BrokerInterface
from core.strategies.strategy_map import STRATEGY_MAP
# 1. Setup Logging yang Cantik
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s | %(levelname)-7s | %(message)s',
datefmt='%H:%M:%S'
)
logger = logging.getLogger("QuantumSim")
# 2. MockBroker yang Lebih Hebat (Bisa update harga)
class AdvancedMockBroker(BrokerInterface):
def __init__(self):
self.positions = []
self._ticket_counter = 5000
self.current_price = 88000.0
self.history = []
# Generate some initial history
for i in range(100):
self.current_price += random.uniform(-100, 100)
self.history.append(self.current_price)
def initialize(self, credentials): return True
def get_account_info(self): return {'balance': 100000, 'equity': 100000}
def get_rates(self, symbol, timeframe, count=100):
# Update harga sedikit setiap kali dipanggil agar simulasi terasa hidup
self.current_price += random.uniform(-150, 150)
self.history.append(self.current_price)
last_prices = self.history[-count:]
df = pd.DataFrame({
'time': pd.date_range(end=pd.Timestamp.now(), periods=len(last_prices), freq='H'),
'open': [p * 0.999 for p in last_prices],
'high': [p * 1.002 for p in last_prices],
'low': [p * 0.998 for p in last_prices],
'close': last_prices,
'tick_volume': [random.randint(1000, 5000) for _ in last_prices]
})
return df
def get_open_positions(self): return self.positions
def place_order(self, symbol, order_type, volume, price=0.0, sl=0.0, tp=0.0, comment=""):
self._ticket_counter += 1
magic = int(comment.split('-')[1]) if 'Bot-' in comment else 0
new_pos = {
'ticket': self._ticket_counter,
'symbol': symbol,
'type': order_type,
'volume': volume,
'price': self.current_price,
'sl': sl, 'tp': tp, 'magic': magic,
'profit': 0.0
}
self.positions.append(new_pos)
logger.info(f"✨ [BROKER] ORDER BERHASIL: {order_type} {symbol} @ {self.current_price:.2f}")
return True
def close_position(self, position_id, volume=0.0):
self.positions = [p for p in self.positions if str(p['ticket']) != str(position_id)]
logger.info(f"🛑 [BROKER] POSISI DITUTUP: ID {position_id}")
return True
def get_symbol_info(self, symbol): return {'name': symbol, 'digits': 2}
def get_todays_profit(self): return 120.50
# 3. Script Simulasi Utama
def run_visual_simulation():
print("\n" + "="*60)
print(" QUANTUM BOT X - LIVE SIMULATION MODE (MOCK) ")
print("="*60)
broker = AdvancedMockBroker()
bot = TradingBot(
id=1337,
name="UltraBot-Sim",
market="BTC/USDT",
risk_percent=0.05,
sl_pips=1000,
tp_pips=2000,
timeframe="H1",
check_interval=2, # Cek setiap 2 detik
strategy="MA_CROSSOVER", # Pake strategi asli
broker=broker
)
# Kita jalankan loop bot secara manual agar bisa kita batasi jumlah iterasinya
# (Biasanya bot.start() akan jalan selamanya di thread terpisah)
# Setup Strategy Instance (biasanya dilakukan di bot.run())
bot.market_for_mt5 = "BTC/USDT"
bot.strategy_instance = STRATEGY_MAP["MA_CROSSOVER"](bot_instance=bot)
print(f"Bot '{bot.name}' Ready. Menggunakan Strategi: {bot.strategy_name}")
print("Memulai simulasi 10 iterasi...\n")
for i in range(1, 11):
print(f"\n--- Iterasi {i}/10 | Harga Saat Ini: {broker.current_price:.2f} ---")
# Ambil data kandel
df = broker.get_rates(bot.market_for_mt5, bot.timeframe, 50)
# Analisis Strategi
bot.last_analysis = bot.strategy_instance.analyze(df)
# PAKSA SINYAL untuk demo agar terlihat log-nya
if i == 2: signal = 'BUY'
elif i == 5: signal = 'SELL'
elif i == 8: signal = 'BUY'
else: signal = bot.last_analysis.get('signal', 'HOLD')
logger.info(f"Analisis: {signal} | Penjelasan: {bot.last_analysis.get('explanation', 'Manual Override for Demo' if i in [2,5,8] else '')}")
# Eksekusi (Logika di TradingBot._handle_trade_signal)
posisi_sekarang = bot._get_open_position()
bot._handle_trade_signal(signal, posisi_sekarang)
# Kasih jeda biar enak dilihat
time.sleep(1.5)
print("\n" + "="*60)
print(" SIMULASI SELESAI! ")
print("="*60)
print("Bot berhasil mensimulasikan logika trading tanpa menyentuh dana asli.")
if __name__ == "__main__":
run_visual_simulation()