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

Keep main focused on MT5 platform

Browse Source
This commit is contained in:
chrisnov-it
2026-05-19 12:38:30 +08:00
parent 2715bfc8ef
commit 5bf46f5fe1
12 changed files with 21 additions and 3433 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ROADMAP.md
+21 -44
View File
@@ -1,52 +1,29 @@
# 📋 QuantumBotX Development Roadmap
# QuantumBotX Roadmap
**👀 Stay Tuned for Exciting Updates!**
QuantumBotX `main` is currently maintained as a Windows-first MetaTrader 5
trading platform. Cross-platform broker work is intentionally developed outside
`main` until it is mature enough to merge cleanly.
## 🎯 **What's Coming Next**
## Current Focus
### **Q4 2025: Intelligence Enhancement**
- **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
- Keep MT5 demo/live workflow stable on Windows.
- Keep strategy registration, backtesting, and dashboard modules installable on
modern Python.
- Improve public-safe tests and documentation.
- Keep broker/account-specific diagnostics out of the public repository.
### **Exciting New Project** 🚀
We're developing **QuantumBotX API** - a revolutionary cloud-based trading platform that will give users unprecedented freedom:
## Near-Term Work
> **"Trade anywhere, anytime, with any broker - no local installations required!"**
- Harden setup for Python 3.12.
- Improve MT5 connection diagnostics and clearer user-facing error messages.
- Expand regression checks for strategy imports and backtesting.
- Review packaging scripts for Windows installer reliability.
**QuantumBotX API will feature:**
- 🌐 **Cloud-native architecture** - Run on any device, anywhere
- 🔄 **Direct broker integration** - No intermediaries, pure API trading
- 🌏 **Cross-broker support** - IC Markets, Pepperstone, and beyond
- ⚡ **Real-time execution** - Ultra-low latency trade processing
- 🎓 **Advanced education** - Built-in learning with community support
## Deferred Work
### **Timeline**
- **Q4 2025**: Closed beta testing with select users
- **Q1 2026**: Public beta launch with premium support
- **Q2 2026**: Full global launch with subscription tiers
- Cross-platform broker backends on dedicated development branches.
- Broker-neutral order and market-data interfaces outside `main`.
- Cloud/API trading platform concepts outside `main`.
---
## 🤝 **Community & Support**
**We're building more than software - we're building a trading community!**
- **Discord Community**: Join our growing trader community
- **Educational Content**: Free trading courses and tutorials
- **Open Source**: Contribute to the project and shape its future
- **Mentorship Program**: One-on-one guidance for serious traders
---
## 🗺️ **Our Mission**
**Empowering traders worldwide with safe, educational, and profitable algorithmic trading solutions.**
*From local learning platform → Global trading ecosystem!* 🚀💫
---
**Roadmap Updated: September 2025**
</content>
Those items should stay on dedicated development branches until the MT5 platform
on `main` remains clean and stable.
core/brokers/base_broker.py
-172
View File
@@ -1,172 +0,0 @@
# core/brokers/base_broker.py
"""
Universal Broker Interface for Multi-Platform Trading
Supports MT5, Binance, and other brokers through unified API
"""
from abc import ABC, abstractmethod
from typing import Dict, List, Optional, Union
from enum import Enum
import pandas as pd
from datetime import datetime
class OrderType(Enum):
MARKET_BUY = "market_buy"
MARKET_SELL = "market_sell"
LIMIT_BUY = "limit_buy"
LIMIT_SELL = "limit_sell"
STOP_LOSS = "stop_loss"
TAKE_PROFIT = "take_profit"
class OrderStatus(Enum):
PENDING = "pending"
FILLED = "filled"
CANCELLED = "cancelled"
REJECTED = "rejected"
class Timeframe(Enum):
M1 = "1m"
M5 = "5m"
M15 = "15m"
M30 = "30m"
H1 = "1h"
H4 = "4h"
D1 = "1d"
class Position:
def __init__(self, symbol: str, side: str, size: float, entry_price: float,
current_price: float, unrealized_pnl: float, realized_pnl: float = 0):
self.symbol = symbol
self.side = side # 'long' or 'short'
self.size = size
self.entry_price = entry_price
self.current_price = current_price
self.unrealized_pnl = unrealized_pnl
self.realized_pnl = realized_pnl
self.timestamp = datetime.now()
class Order:
def __init__(self, order_id: str, symbol: str, order_type: OrderType,
side: str, size: float, price: Optional[float] = None):
self.order_id = order_id
self.symbol = symbol
self.order_type = order_type
self.side = side
self.size = size
self.price = price
self.status = OrderStatus.PENDING
self.filled_size = 0.0
self.avg_fill_price = 0.0
self.timestamp = datetime.now()
class AccountInfo:
def __init__(self, balance: float, equity: float, margin: float,
free_margin: float, margin_level: float, currency: str = "USD"):
self.balance = balance
self.equity = equity
self.margin = margin
self.free_margin = free_margin
self.margin_level = margin_level
self.currency = currency
self.timestamp = datetime.now()
class BaseBroker(ABC):
"""
Abstract base class for all broker implementations.
Provides unified interface for MT5, Binance, and other brokers.
"""
def __init__(self, broker_name: str):
self.broker_name = broker_name
self.is_connected = False
self.supported_symbols = []
@abstractmethod
def connect(self, credentials: Dict) -> bool:
"""Connect to broker with credentials"""
pass
@abstractmethod
def disconnect(self) -> bool:
"""Disconnect from broker"""
pass
@abstractmethod
def get_symbols(self) -> List[str]:
"""Get list of available trading symbols"""
pass
@abstractmethod
def get_market_data(self, symbol: str, timeframe: Timeframe,
count: int = 500) -> pd.DataFrame:
"""
Get OHLCV market data
Returns: DataFrame with columns [time, open, high, low, close, volume]
"""
pass
@abstractmethod
def get_current_price(self, symbol: str) -> Dict[str, float]:
"""
Get current bid/ask prices
Returns: {"bid": price, "ask": price}
"""
pass
@abstractmethod
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) -> Order:
"""Place a trading order"""
pass
@abstractmethod
def cancel_order(self, order_id: str) -> bool:
"""Cancel an existing order"""
pass
@abstractmethod
def get_positions(self) -> List[Position]:
"""Get all open positions"""
pass
@abstractmethod
def get_orders(self) -> List[Order]:
"""Get all pending orders"""
pass
@abstractmethod
def get_account_info(self) -> AccountInfo:
"""Get account information"""
pass
@abstractmethod
def get_trade_history(self, days: int = 30) -> List[Dict]:
"""Get trade history"""
pass
# Utility methods (implemented in base class)
def normalize_symbol(self, symbol: str) -> str:
"""Normalize symbol format for the broker"""
return symbol.upper().replace("/", "").replace("-", "")
def calculate_position_size(self, account_balance: float, risk_percent: float,
entry_price: float, stop_loss: float) -> float:
"""Calculate position size based on risk management"""
risk_amount = account_balance * (risk_percent / 100)
price_difference = abs(entry_price - stop_loss)
if price_difference == 0:
return 0
position_size = risk_amount / price_difference
return position_size
def validate_symbol(self, symbol: str) -> bool:
"""Check if symbol is supported by broker"""
return symbol in self.supported_symbols
def is_market_open(self) -> bool:
"""Check if market is currently open (override for specific markets)"""
return True # Crypto markets are always open
core/brokers/binance_broker.py
-359
View File
@@ -1,359 +0,0 @@
# core/brokers/binance_broker.py
"""
Binance Exchange Integration for QuantumBotX
Implements crypto trading through Binance API
"""
import pandas as pd
import time
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import logging
from .base_broker import (
BaseBroker, OrderType, OrderStatus, Timeframe,
Position, Order, AccountInfo
)
logger = logging.getLogger(__name__)
class BinanceBroker(BaseBroker):
"""
Binance exchange implementation of the universal broker interface.
Supports spot and futures trading.
"""
def __init__(self, testnet: bool = True):
super().__init__("Binance")
self.testnet = testnet
self.client = None
self.base_url = "https://testnet.binance.vision" if testnet else "https://api.binance.com"
# Timeframe mapping
self.timeframe_map = {
Timeframe.M1: "1m",
Timeframe.M5: "5m",
Timeframe.M15: "15m",
Timeframe.M30: "30m",
Timeframe.H1: "1h",
Timeframe.H4: "4h",
Timeframe.D1: "1d"
}
def connect(self, credentials: Dict) -> bool:
"""
Connect to Binance with API credentials
credentials: {"api_key": "...", "secret_key": "..."}
"""
try:
# Import here to avoid dependency issues if not installed
from binance.client import Client
from binance.exceptions import BinanceAPIException
api_key = credentials.get("api_key")
secret_key = credentials.get("secret_key")
if not api_key or not secret_key:
logger.error("Binance API key and secret key are required")
return False
# Initialize Binance client
self.client = Client(
api_key=api_key,
api_secret=secret_key,
testnet=self.testnet
)
# Test connection
account_info = self.client.get_account()
self.is_connected = True
# Get supported symbols
exchange_info = self.client.get_exchange_info()
self.supported_symbols = [s['symbol'] for s in exchange_info['symbols']
if s['status'] == 'TRADING']
logger.info(f"Connected to Binance {'Testnet' if self.testnet else 'Mainnet'}")
logger.info(f"Account status: {account_info.get('accountType', 'Unknown')}")
return True
except Exception as e:
logger.error(f"Failed to connect to Binance: {e}")
self.is_connected = False
return False
def disconnect(self) -> bool:
"""Disconnect from Binance"""
self.client = None
self.is_connected = False
logger.info("Disconnected from Binance")
return True
def get_symbols(self) -> List[str]:
"""Get list of available trading symbols"""
if not self.is_connected:
return []
return self.supported_symbols
def get_market_data(self, symbol: str, timeframe: Timeframe, count: int = 500) -> pd.DataFrame:
"""
Get OHLCV market data from Binance
"""
if not self.is_connected:
raise Exception("Not connected to Binance")
try:
# Convert timeframe
interval = self.timeframe_map[timeframe]
# Get klines (candlestick data)
klines = self.client.get_klines(
symbol=symbol,
interval=interval,
limit=count
)
# Convert to DataFrame
df = pd.DataFrame(klines, columns=[
'timestamp', 'open', 'high', 'low', 'close', 'volume',
'close_time', 'quote_asset_volume', 'number_of_trades',
'taker_buy_base_asset_volume', 'taker_buy_quote_asset_volume', 'ignore'
])
# Clean and format data
df['time'] = pd.to_datetime(df['timestamp'], unit='ms')
df['open'] = pd.to_numeric(df['open'])
df['high'] = pd.to_numeric(df['high'])
df['low'] = pd.to_numeric(df['low'])
df['close'] = pd.to_numeric(df['close'])
df['volume'] = pd.to_numeric(df['volume'])
# Return standardized format
return df[['time', 'open', 'high', 'low', 'close', 'volume']].copy()
except Exception as e:
logger.error(f"Failed to get market data for {symbol}: {e}")
return pd.DataFrame()
def get_current_price(self, symbol: str) -> Dict[str, float]:
"""Get current bid/ask prices"""
if not self.is_connected:
raise Exception("Not connected to Binance")
try:
ticker = self.client.get_orderbook_ticker(symbol=symbol)
return {
"bid": float(ticker['bidPrice']),
"ask": float(ticker['askPrice'])
}
except Exception as e:
logger.error(f"Failed to get current price for {symbol}: {e}")
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) -> Order:
"""Place a trading order on Binance"""
if not self.is_connected:
raise Exception("Not connected to Binance")
try:
# Convert order parameters
binance_side = side.upper() # 'BUY' or 'SELL'
# Determine order type
if order_type == OrderType.MARKET_BUY or order_type == OrderType.MARKET_SELL:
binance_type = "MARKET"
elif order_type == OrderType.LIMIT_BUY or order_type == OrderType.LIMIT_SELL:
binance_type = "LIMIT"
else:
raise ValueError(f"Unsupported order type: {order_type}")
# Prepare order parameters
order_params = {
'symbol': symbol,
'side': binance_side,
'type': binance_type,
'quantity': size,
}
if binance_type == "LIMIT":
order_params['price'] = price
order_params['timeInForce'] = 'GTC' # Good Till Cancelled
# Place order
result = self.client.create_order(**order_params)
# Create Order object
order = Order(
order_id=str(result['orderId']),
symbol=symbol,
order_type=order_type,
side=side.lower(),
size=size,
price=price
)
# Update status based on result
if result['status'] == 'FILLED':
order.status = OrderStatus.FILLED
order.filled_size = float(result.get('executedQty', 0))
order.avg_fill_price = float(result.get('price', price or 0))
elif result['status'] == 'NEW':
order.status = OrderStatus.PENDING
logger.info(f"Order placed: {order.order_id} for {symbol}")
return order
except Exception as e:
logger.error(f"Failed to place order: {e}")
# Return failed order
order = Order(
order_id="failed",
symbol=symbol,
order_type=order_type,
side=side.lower(),
size=size,
price=price
)
order.status = OrderStatus.REJECTED
return order
def cancel_order(self, order_id: str) -> bool:
"""Cancel an existing order"""
if not self.is_connected:
return False
try:
# Note: Need symbol to cancel order in Binance
# This is a limitation - may need to store order info
logger.warning("Cancel order requires symbol - implement order tracking")
return False
except Exception as e:
logger.error(f"Failed to cancel order {order_id}: {e}")
return False
def get_positions(self) -> List[Position]:
"""Get all open positions (for futures)"""
if not self.is_connected:
return []
try:
# For spot trading, positions are just balances
account = self.client.get_account()
positions = []
for balance in account['balances']:
free = float(balance['free'])
locked = float(balance['locked'])
total = free + locked
if total > 0:
# Create position for non-zero balances
position = Position(
symbol=balance['asset'],
side='long', # Spot is always long
size=total,
entry_price=0.0, # Not available for spot
current_price=0.0, # Would need to fetch
unrealized_pnl=0.0 # Not calculated for spot
)
positions.append(position)
return positions
except Exception as e:
logger.error(f"Failed to get positions: {e}")
return []
def get_orders(self) -> List[Order]:
"""Get all pending orders"""
if not self.is_connected:
return []
try:
# Get open orders for all symbols (limitation: need symbol)
# For now, return empty - would need to track symbols
logger.warning("Get orders requires symbol tracking - implement order cache")
return []
except Exception as e:
logger.error(f"Failed to get orders: {e}")
return []
def get_account_info(self) -> AccountInfo:
"""Get account information"""
if not self.is_connected:
return AccountInfo(0, 0, 0, 0, 0, "USDT")
try:
account = self.client.get_account()
# Calculate total balance in USDT
total_balance = 0.0
for balance in account['balances']:
free = float(balance['free'])
locked = float(balance['locked'])
total = free + locked
if total > 0:
asset = balance['asset']
if asset == 'USDT':
total_balance += total
else:
# Convert to USDT (simplified - would need price conversion)
# For demo purposes, assume small balances
if asset in ['BTC', 'ETH']:
total_balance += total * 30000 # Rough estimate
else:
total_balance += total # Assume stablecoin or ignore
return AccountInfo(
balance=total_balance,
equity=total_balance, # Same for spot
margin=0.0, # Not applicable for spot
free_margin=total_balance,
margin_level=100.0, # Not applicable for spot
currency="USDT"
)
except Exception as e:
logger.error(f"Failed to get account info: {e}")
return AccountInfo(0, 0, 0, 0, 0, "USDT")
def get_trade_history(self, days: int = 30) -> List[Dict]:
"""Get trade history"""
if not self.is_connected:
return []
try:
# Get trades for major symbols (limitation: need symbol)
logger.warning("Trade history requires symbol tracking - implement symbol cache")
return []
except Exception as e:
logger.error(f"Failed to get trade history: {e}")
return []
def normalize_symbol(self, symbol: str) -> str:
"""Normalize symbol format for Binance"""
# Binance uses format like 'BTCUSDT', 'ETHUSDT'
symbol = symbol.upper().replace("/", "").replace("-", "")
# Common conversions
if symbol.endswith("USD") and not symbol.endswith("USDT"):
symbol = symbol.replace("USD", "USDT")
return symbol
def is_market_open(self) -> bool:
"""Crypto markets are always open"""
return True
# Convenience function to create Binance broker
def create_binance_broker(testnet: bool = True) -> BinanceBroker:
"""Create a Binance broker instance"""
return BinanceBroker(testnet=testnet)
core/brokers/broker_factory.py
-234
View File
@@ -1,234 +0,0 @@
# core/brokers/broker_factory.py
"""
Broker Factory for QuantumBotX
Manages multiple brokers and provides unified interface
"""
import logging
from typing import Dict, Optional, List
from enum import Enum
from .base_broker import BaseBroker
from .binance_broker import BinanceBroker
from .ctrader_broker import CTraderBroker
from .interactive_brokers import InteractiveBrokersBroker
from .tradingview_broker import TradingViewBroker
from .indonesian_brokers import (
IndopremierBroker, XMIndonesiaBroker,
OctaFXIndonesiaBroker, HSBCIndonesiaBroker
)
logger = logging.getLogger(__name__)
class BrokerType(Enum):
MT5 = "mt5"
BINANCE = "binance"
BINANCE_FUTURES = "binance_futures"
CTRADER = "ctrader"
INTERACTIVE_BROKERS = "interactive_brokers"
TRADINGVIEW = "tradingview"
# Indonesian brokers
INDOPREMIER = "indopremier"
XM_INDONESIA = "xm_indonesia"
OCTAFX_INDONESIA = "octafx_indonesia"
HSBC_INDONESIA = "hsbc_indonesia"
class BrokerFactory:
"""
Factory class to create and manage different broker instances
"""
_brokers: Dict[str, BaseBroker] = {}
_configs: Dict[str, Dict] = {}
@classmethod
def register_broker_config(cls, broker_id: str, broker_type: BrokerType, config: Dict):
"""Register broker configuration"""
cls._configs[broker_id] = {
'type': broker_type,
'config': config
}
@classmethod
def create_broker(cls, broker_id: str) -> Optional[BaseBroker]:
"""Create broker instance from registered configuration"""
if broker_id in cls._brokers:
return cls._brokers[broker_id]
if broker_id not in cls._configs:
logger.error(f"No configuration found for broker: {broker_id}")
return None
broker_config = cls._configs[broker_id]
broker_type = broker_config['type']
config = broker_config['config']
try:
if broker_type == BrokerType.BINANCE:
broker = BinanceBroker(testnet=config.get('testnet', True))
elif broker_type == BrokerType.BINANCE_FUTURES:
# Future implementation
broker = BinanceBroker(testnet=config.get('testnet', True))
elif broker_type == BrokerType.CTRADER:
broker = CTraderBroker(demo=config.get('demo', True))
elif broker_type == BrokerType.INTERACTIVE_BROKERS:
broker = InteractiveBrokersBroker(paper_trading=config.get('paper_trading', True))
elif broker_type == BrokerType.TRADINGVIEW:
broker = TradingViewBroker(paper_trading=config.get('paper_trading', True))
elif broker_type == BrokerType.INDOPREMIER:
broker = IndopremierBroker(demo=config.get('demo', True))
elif broker_type == BrokerType.XM_INDONESIA:
broker = XMIndonesiaBroker(demo=config.get('demo', True))
elif broker_type == BrokerType.OCTAFX_INDONESIA:
broker = OctaFXIndonesiaBroker(demo=config.get('demo', True))
elif broker_type == BrokerType.HSBC_INDONESIA:
broker = HSBCIndonesiaBroker(demo=config.get('demo', True))
elif broker_type == BrokerType.MT5:
# Import MT5 broker when implemented
from .mt5_broker import MT5Broker
broker = MT5Broker()
else:
logger.error(f"Unsupported broker type: {broker_type}")
return None
# Connect broker
if broker.connect(config.get('credentials', {})):
cls._brokers[broker_id] = broker
logger.info(f"Successfully created and connected broker: {broker_id}")
return broker
else:
logger.error(f"Failed to connect broker: {broker_id}")
return None
except Exception as e:
logger.error(f"Error creating broker {broker_id}: {e}")
return None
@classmethod
def get_broker(cls, broker_id: str) -> Optional[BaseBroker]:
"""Get existing broker instance"""
return cls._brokers.get(broker_id)
@classmethod
def disconnect_all(cls):
"""Disconnect all brokers"""
for broker_id, broker in cls._brokers.items():
try:
broker.disconnect()
logger.info(f"Disconnected broker: {broker_id}")
except Exception as e:
logger.error(f"Error disconnecting broker {broker_id}: {e}")
cls._brokers.clear()
@classmethod
def get_all_brokers(cls) -> Dict[str, BaseBroker]:
"""Get all connected brokers"""
return cls._brokers.copy()
@classmethod
def get_supported_symbols(cls, broker_id: str) -> List[str]:
"""Get supported symbols for a broker"""
broker = cls.get_broker(broker_id)
if broker:
return broker.get_symbols()
return []
@classmethod
def is_broker_connected(cls, broker_id: str) -> bool:
"""Check if broker is connected"""
broker = cls.get_broker(broker_id)
return broker.is_connected if broker else False
# Configuration helper functions
def setup_demo_brokers():
"""Setup demo brokers for testing"""
# Binance Testnet configuration
BrokerFactory.register_broker_config(
broker_id="binance_testnet",
broker_type=BrokerType.BINANCE,
config={
'testnet': True,
'credentials': {
'api_key': '', # Add your testnet API key
'secret_key': '' # Add your testnet secret key
}
}
)
# MT5 Demo configuration
BrokerFactory.register_broker_config(
broker_id="mt5_demo",
broker_type=BrokerType.MT5,
config={
'credentials': {
'login': '', # Add your MT5 demo login
'password': '', # Add your MT5 demo password
'server': 'MetaQuotes-Demo'
}
}
)
def load_brokers_from_env():
"""Load broker configurations from environment variables"""
import os
# Binance configuration
binance_api_key = os.getenv('BINANCE_API_KEY')
binance_secret = os.getenv('BINANCE_SECRET_KEY')
binance_testnet = os.getenv('BINANCE_TESTNET', 'true').lower() == 'true'
if binance_api_key and binance_secret:
BrokerFactory.register_broker_config(
broker_id="binance",
broker_type=BrokerType.BINANCE,
config={
'testnet': binance_testnet,
'credentials': {
'api_key': binance_api_key,
'secret_key': binance_secret
}
}
)
# MT5 configuration
mt5_login = os.getenv('MT5_LOGIN')
mt5_password = os.getenv('MT5_PASSWORD')
mt5_server = os.getenv('MT5_SERVER', 'MetaQuotes-Demo')
if mt5_login and mt5_password:
BrokerFactory.register_broker_config(
broker_id="mt5",
broker_type=BrokerType.MT5,
config={
'credentials': {
'login': mt5_login,
'password': mt5_password,
'server': mt5_server
}
}
)
# Example usage
if __name__ == "__main__":
# Load configurations
load_brokers_from_env()
# Create brokers
binance_broker = BrokerFactory.create_broker("binance")
mt5_broker = BrokerFactory.create_broker("mt5")
if binance_broker:
print(f"Binance connected: {binance_broker.is_connected}")
symbols = binance_broker.get_symbols()[:10] # First 10 symbols
print(f"Binance symbols: {symbols}")
if mt5_broker:
print(f"MT5 connected: {mt5_broker.is_connected}")
account_info = mt5_broker.get_account_info()
print(f"MT5 balance: {account_info.balance}")
# Cleanup
BrokerFactory.disconnect_all()
core/brokers/ctrader_broker.py
-415
View File
@@ -1,415 +0,0 @@
# core/brokers/ctrader_broker.py
"""
cTrader Broker Integration for QuantumBotX
Modern forex/CFD platform with excellent API
"""
import pandas as pd
import time
import requests
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import logging
from .base_broker import (
BaseBroker, OrderType, OrderStatus, Timeframe,
Position, Order, AccountInfo
)
logger = logging.getLogger(__name__)
class CTraderBroker(BaseBroker):
"""
cTrader (cTID) implementation of the universal broker interface.
Uses cTrader REST API for modern forex trading.
"""
def __init__(self, demo: bool = True):
super().__init__("cTrader")
self.demo = demo
self.client_id = None
self.client_secret = None
self.access_token = None
self.account_id = None
self.base_url = "https://demo-api.ctraderapi.com" if demo else "https://api.ctraderapi.com"
# Timeframe mapping
self.timeframe_map = {
Timeframe.M1: "M1",
Timeframe.M5: "M5",
Timeframe.M15: "M15",
Timeframe.M30: "M30",
Timeframe.H1: "H1",
Timeframe.H4: "H4",
Timeframe.D1: "D1"
}
def connect(self, credentials: Dict) -> bool:
"""
Connect to cTrader with OAuth credentials
credentials: {"client_id": "...", "client_secret": "...", "account_id": "..."}
"""
try:
self.client_id = credentials.get("client_id")
self.client_secret = credentials.get("client_secret")
self.account_id = credentials.get("account_id")
if not all([self.client_id, self.client_secret, self.account_id]):
logger.error("cTrader client_id, client_secret, and account_id are required")
return False
# OAuth token request
token_url = f"{self.base_url}/oauth/v2/token"
token_data = {
'grant_type': 'client_credentials',
'client_id': self.client_id,
'client_secret': self.client_secret,
'scope': 'trading'
}
response = requests.post(token_url, data=token_data)
if response.status_code == 200:
token_info = response.json()
self.access_token = token_info['access_token']
self.is_connected = True
# Get supported symbols
self._load_symbols()
logger.info(f"Connected to cTrader {'Demo' if self.demo else 'Live'}")
return True
else:
logger.error(f"cTrader authentication failed: {response.text}")
return False
except Exception as e:
logger.error(f"Failed to connect to cTrader: {e}")
self.is_connected = False
return False
def disconnect(self) -> bool:
"""Disconnect from cTrader"""
self.access_token = None
self.is_connected = False
logger.info("Disconnected from cTrader")
return True
def _make_request(self, endpoint: str, method: str = "GET", data: Dict = None) -> Dict:
"""Make authenticated request to cTrader API"""
if not self.access_token:
raise Exception("Not authenticated with cTrader")
headers = {
'Authorization': f'Bearer {self.access_token}',
'Content-Type': 'application/json'
}
url = f"{self.base_url}{endpoint}"
if method == "GET":
response = requests.get(url, headers=headers, params=data)
elif method == "POST":
response = requests.post(url, headers=headers, json=data)
elif method == "PUT":
response = requests.put(url, headers=headers, json=data)
elif method == "DELETE":
response = requests.delete(url, headers=headers)
if response.status_code in [200, 201]:
return response.json()
else:
raise Exception(f"cTrader API error: {response.status_code} - {response.text}")
def _load_symbols(self):
"""Load available symbols from cTrader"""
try:
symbols_data = self._make_request("/v2/symbols")
self.supported_symbols = [s['symbolName'] for s in symbols_data.get('symbols', [])]
except Exception as e:
logger.warning(f"Failed to load cTrader symbols: {e}")
# Common forex symbols as fallback
self.supported_symbols = [
'EURUSD', 'GBPUSD', 'USDJPY', 'USDCHF', 'AUDUSD', 'USDCAD',
'NZDUSD', 'EURGBP', 'EURJPY', 'GBPJPY', 'XAUUSD', 'XAGUSD'
]
def get_symbols(self) -> List[str]:
"""Get list of available trading symbols"""
return self.supported_symbols
def get_market_data(self, symbol: str, timeframe: Timeframe, count: int = 500) -> pd.DataFrame:
"""Get OHLCV market data from cTrader"""
if not self.is_connected:
raise Exception("Not connected to cTrader")
try:
# Convert timeframe
ct_timeframe = self.timeframe_map[timeframe]
# Calculate from time (count bars back)
now = datetime.utcnow()
# Estimate time per bar
minutes_per_bar = {
'M1': 1, 'M5': 5, 'M15': 15, 'M30': 30,
'H1': 60, 'H4': 240, 'D1': 1440
}
minutes_back = count * minutes_per_bar.get(ct_timeframe, 60)
from_time = now - timedelta(minutes=minutes_back)
# Request historical data
params = {
'symbolName': symbol,
'periodName': ct_timeframe,
'fromTimestamp': int(from_time.timestamp() * 1000),
'toTimestamp': int(now.timestamp() * 1000),
'count': count
}
data = self._make_request("/v2/bars", params=params)
bars = data.get('bars', [])
if not bars:
return pd.DataFrame()
# Convert to DataFrame
df_data = []
for bar in bars:
df_data.append({
'time': datetime.fromtimestamp(bar['timestamp'] / 1000),
'open': bar['open'],
'high': bar['high'],
'low': bar['low'],
'close': bar['close'],
'volume': bar.get('volume', 0)
})
return pd.DataFrame(df_data)
except Exception as e:
logger.error(f"Failed to get market data for {symbol}: {e}")
return pd.DataFrame()
def get_current_price(self, symbol: str) -> Dict[str, float]:
"""Get current bid/ask prices"""
if not self.is_connected:
raise Exception("Not connected to cTrader")
try:
data = self._make_request(f"/v2/symbols/{symbol}/tick")
return {
"bid": data['bid'],
"ask": data['ask']
}
except Exception as e:
logger.error(f"Failed to get current price for {symbol}: {e}")
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) -> Order:
"""Place a trading order on cTrader"""
if not self.is_connected:
raise Exception("Not connected to cTrader")
try:
# Convert order parameters
ct_side = "BUY" if side.lower() == "buy" else "SELL"
# Convert volume to lots (cTrader uses volume in units)
volume = int(size * 100000) # Convert lots to units
# Determine order type
if order_type in [OrderType.MARKET_BUY, OrderType.MARKET_SELL]:
ct_type = "MARKET"
elif order_type in [OrderType.LIMIT_BUY, OrderType.LIMIT_SELL]:
ct_type = "LIMIT"
else:
raise ValueError(f"Unsupported order type: {order_type}")
# Prepare order data
order_data = {
'accountId': self.account_id,
'symbolName': symbol,
'orderType': ct_type,
'tradeSide': ct_side,
'volume': volume,
}
if ct_type == "LIMIT" and price:
order_data['limitPrice'] = price
if stop_loss:
order_data['stopLoss'] = stop_loss
if take_profit:
order_data['takeProfit'] = take_profit
# Place order
result = self._make_request("/v2/orders", method="POST", data=order_data)
# Create Order object
order = Order(
order_id=str(result.get('orderId', 'unknown')),
symbol=symbol,
order_type=order_type,
side=side.lower(),
size=size,
price=price
)
order.status = OrderStatus.PENDING
if result.get('executionType') == 'TRADE':
order.status = OrderStatus.FILLED
logger.info(f"cTrader order placed: {order.order_id} for {symbol}")
return order
except Exception as e:
logger.error(f"Failed to place cTrader order: {e}")
order = Order(
order_id="failed",
symbol=symbol,
order_type=order_type,
side=side.lower(),
size=size,
price=price
)
order.status = OrderStatus.REJECTED
return order
def cancel_order(self, order_id: str) -> bool:
"""Cancel an existing order"""
if not self.is_connected:
return False
try:
self._make_request(f"/v2/orders/{order_id}", method="DELETE")
return True
except Exception as e:
logger.error(f"Failed to cancel cTrader order {order_id}: {e}")
return False
def get_positions(self) -> List[Position]:
"""Get all open positions"""
if not self.is_connected:
return []
try:
data = self._make_request(f"/v2/accounts/{self.account_id}/positions")
positions = []
for pos_data in data.get('positions', []):
position = Position(
symbol=pos_data['symbolName'],
side='long' if pos_data['tradeSide'] == 'BUY' else 'short',
size=pos_data['volume'] / 100000, # Convert units to lots
entry_price=pos_data['entryPrice'],
current_price=pos_data['currentPrice'],
unrealized_pnl=pos_data['unrealizedGrossProfit']
)
positions.append(position)
return positions
except Exception as e:
logger.error(f"Failed to get cTrader positions: {e}")
return []
def get_orders(self) -> List[Order]:
"""Get all pending orders"""
if not self.is_connected:
return []
try:
data = self._make_request(f"/v2/accounts/{self.account_id}/orders")
orders = []
for order_data in data.get('orders', []):
order = Order(
order_id=str(order_data['orderId']),
symbol=order_data['symbolName'],
order_type=OrderType.LIMIT_BUY, # Simplified
side=order_data['tradeSide'].lower(),
size=order_data['volume'] / 100000,
price=order_data.get('limitPrice')
)
order.status = OrderStatus.PENDING
orders.append(order)
return orders
except Exception as e:
logger.error(f"Failed to get cTrader orders: {e}")
return []
def get_account_info(self) -> AccountInfo:
"""Get account information"""
if not self.is_connected:
return AccountInfo(0, 0, 0, 0, 0, "USD")
try:
data = self._make_request(f"/v2/accounts/{self.account_id}")
balance = data.get('balance', 0)
equity = data.get('equity', balance)
margin = data.get('margin', 0)
free_margin = data.get('freeMargin', balance)
margin_level = data.get('marginLevel', 100)
currency = data.get('currency', 'USD')
return AccountInfo(
balance=balance,
equity=equity,
margin=margin,
free_margin=free_margin,
margin_level=margin_level,
currency=currency
)
except Exception as e:
logger.error(f"Failed to get cTrader account info: {e}")
return AccountInfo(0, 0, 0, 0, 0, "USD")
def get_trade_history(self, days: int = 30) -> List[Dict]:
"""Get trade history"""
if not self.is_connected:
return []
try:
from_time = datetime.now() - timedelta(days=days)
params = {
'fromTimestamp': int(from_time.timestamp() * 1000),
'toTimestamp': int(datetime.now().timestamp() * 1000)
}
data = self._make_request(f"/v2/accounts/{self.account_id}/deals", params=params)
return data.get('deals', [])
except Exception as e:
logger.error(f"Failed to get cTrader trade history: {e}")
return []
def normalize_symbol(self, symbol: str) -> str:
"""Normalize symbol format for cTrader"""
# cTrader typically uses format like 'EURUSD', 'GBPUSD'
return symbol.upper().replace("/", "").replace("-", "")
def is_market_open(self) -> bool:
"""Check if forex market is open"""
now = datetime.utcnow()
# Forex market is open from Sunday 22:00 UTC to Friday 22:00 UTC
if now.weekday() == 5: # Saturday
return False
if now.weekday() == 6 and now.hour < 22: # Sunday before 22:00 UTC
return False
if now.weekday() == 4 and now.hour >= 22: # Friday after 22:00 UTC
return False
return True
# Convenience function
def create_ctrader_broker(demo: bool = True) -> CTraderBroker:
"""Create a cTrader broker instance"""
return CTraderBroker(demo=demo)
core/brokers/indonesian_brokers.py
-546
View File
@@ -1,546 +0,0 @@
# core/brokers/indonesian_brokers.py
"""
Indonesian Market Brokers Integration for QuantumBotX
Supporting local Indonesian brokers and international brokers popular in Indonesia
"""
import pandas as pd
import time
import requests
import json
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import logging
from .base_broker import (
BaseBroker, OrderType, OrderStatus, Timeframe,
Position, Order, AccountInfo
)
logger = logging.getLogger(__name__)
class IndopremierBroker(BaseBroker):
"""
Indopremier Securities (IPOT) - Popular Indonesian broker
Known for good demo accounts and local market access
"""
def __init__(self, demo: bool = True):
super().__init__("Indopremier")
self.demo = demo
self.base_url = "https://demo-api.indopremier.com" if demo else "https://api.indopremier.com"
self.session = requests.Session()
# Indonesian market symbols
self.supported_symbols = [
# IDX (Indonesian Stock Exchange) - Blue chips
'BBCA.JK', # Bank Central Asia
'BBRI.JK', # Bank Rakyat Indonesia
'BMRI.JK', # Bank Mandiri
'TLKM.JK', # Telkom Indonesia
'ASII.JK', # Astra International
'UNVR.JK', # Unilever Indonesia
'ICBP.JK', # Indofood CBP
'INDF.JK', # Indofood Sukses Makmur
'GGRM.JK', # Gudang Garam
'HMSP.JK', # HM Sampoerna
# IDX ETFs and Indices
'LQ45.JK', # LQ45 Index
'IHSG.JK', # Jakarta Composite Index
# International through Indopremier
'USDID', # USD/IDR
'USDIDR', # USD/IDR alternative
'XAUIDR', # Gold in IDR
]
def connect(self, credentials: Dict) -> bool:
"""Connect to Indopremier"""
try:
username = credentials.get("username")
password = credentials.get("password")
if not all([username, password]):
logger.error("Indopremier username and password required")
return False
# Simulate authentication for demo
if self.demo:
self.is_connected = True
logger.info("Connected to Indopremier Demo")
return True
# Real implementation would use actual API
auth_data = {
'username': username,
'password': password
}
# This would be actual API call
self.is_connected = True
logger.info("Connected to Indopremier Live")
return True
except Exception as e:
logger.error(f"Failed to connect to Indopremier: {e}")
return False
def get_market_data(self, symbol: str, timeframe: Timeframe, count: int = 500) -> pd.DataFrame:
"""Get Indonesian market data"""
try:
# For demo, generate realistic Indonesian stock data
dates = pd.date_range(end=datetime.now(), periods=count, freq='1h')
# Realistic prices for Indonesian stocks
base_prices = {
'BBCA.JK': 9000, # BCA around 9,000 IDR
'BBRI.JK': 4500, # BRI around 4,500 IDR
'BMRI.JK': 8500, # Mandiri around 8,500 IDR
'TLKM.JK': 3200, # Telkom around 3,200 IDR
'ASII.JK': 6800, # Astra around 6,800 IDR
'UNVR.JK': 7200, # Unilever around 7,200 IDR
'USDID': 15400, # USD/IDR around 15,400
'XAUIDR': 1000000, # Gold around 1M IDR per oz
}
base_price = base_prices.get(symbol, 5000)
# Indonesian market volatility (generally lower than crypto)
volatility = 0.015 if '.JK' in symbol else 0.008 # 1.5% for stocks, 0.8% for forex
# Generate price movements
returns = np.random.randn(count) * volatility
prices = base_price * (1 + returns).cumprod()
df = pd.DataFrame({
'time': dates,
'open': prices,
'high': prices * (1 + np.random.uniform(0, 0.01, count)),
'low': prices * (1 - np.random.uniform(0, 0.01, count)),
'close': prices,
'volume': np.random.randint(100000, 1000000, count) # Indonesian market volumes
})
# Ensure OHLC integrity
df['high'] = df[['high', 'close', 'open']].max(axis=1)
df['low'] = df[['low', 'close', 'open']].min(axis=1)
# Adjust for Indonesian market hours (09:00-16:00 WIB, Mon-Fri)
# Filter out weekend data for stock symbols
if '.JK' in symbol:
df = df[df['time'].dt.weekday < 5] # Monday=0, Sunday=6
return df
except Exception as e:
logger.error(f"Failed to get Indopremier market data for {symbol}: {e}")
return pd.DataFrame()
def disconnect(self) -> bool:
"""Disconnect from Indopremier"""
self.is_connected = False
logger.info("Disconnected from Indopremier")
return True
def get_symbols(self) -> List[str]:
"""Get list of available trading symbols"""
return self.supported_symbols
def get_current_price(self, symbol: str) -> Dict[str, float]:
"""Get current bid/ask prices"""
try:
# For demo, use last price from market data
df = self.get_market_data(symbol, Timeframe.M1, 1)
if not df.empty:
last_price = df.iloc[-1]['close']
spread = last_price * 0.001 # 0.1% spread for Indonesian stocks
return {
"bid": last_price - spread/2,
"ask": last_price + spread/2
}
return {"bid": 0.0, "ask": 0.0}
except Exception as e:
logger.error(f"Failed to get Indopremier current price for {symbol}: {e}")
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) -> Order:
"""Place order (simulated for demo)"""
try:
order_id = str(int(time.time()))
# For Indonesian stocks, size is in lots (100 shares)
if '.JK' in symbol:
size = max(1, int(size)) # Minimum 1 lot
order = Order(
order_id=order_id,
symbol=symbol,
order_type=order_type,
side=side.lower(),
size=size,
price=price
)
# Simulate immediate execution for demo
order.status = OrderStatus.FILLED
order.filled_size = size
current_price = self.get_current_price(symbol)
order.avg_fill_price = current_price['ask'] if side.lower() == 'buy' else current_price['bid']
logger.info(f"Indopremier demo order: {side} {size} {symbol} at {order.avg_fill_price}")
return order
except Exception as e:
logger.error(f"Failed to place Indopremier order: {e}")
order = Order(
order_id="failed",
symbol=symbol,
order_type=order_type,
side=side.lower(),
size=size,
price=price
)
order.status = OrderStatus.REJECTED
return order
def cancel_order(self, order_id: str) -> bool:
"""Cancel an existing order"""
logger.info(f"Indopremier demo: Order {order_id} cancelled")
return True
def get_positions(self) -> List[Position]:
"""Get all open positions"""
# For demo, return empty list
return []
def get_orders(self) -> List[Order]:
"""Get all pending orders"""
# For demo, return empty list
return []
def get_account_info(self) -> AccountInfo:
"""Get account information"""
try:
return AccountInfo(
balance=1000000000, # 1 billion IDR demo balance
equity=1000000000,
margin=0.0,
free_margin=1000000000,
margin_level=100.0,
currency="IDR"
)
except Exception as e:
logger.error(f"Failed to get Indopremier account info: {e}")
return AccountInfo(0, 0, 0, 0, 0, "IDR")
def get_trade_history(self, days: int = 30) -> List[Dict]:
"""Get trade history"""
# For demo, return empty list
return []
class XMIndonesiaBroker(BaseBroker):
"""
XM Indonesia - Popular international broker in Indonesia
Offers forex, commodities, and indices with good demo accounts
"""
def __init__(self, demo: bool = True):
super().__init__("XM Indonesia")
self.demo = demo
# XM Indonesia popular symbols
self.supported_symbols = [
# Major Forex pairs
'EURUSD', 'GBPUSD', 'USDJPY', 'USDCHF', 'AUDUSD', 'USDCAD',
'NZDUSD', 'EURGBP', 'EURJPY', 'GBPJPY',
# IDR pairs (if available)
'USDIDR', 'EURIDR', 'GBPIDR', 'JPYIDR',
# Commodities popular in Indonesia
'XAUUSD', 'XAGUSD', 'USOIL', 'UKOIL', 'NGAS',
# Indices
'US30', 'SPX500', 'NAS100', 'UK100', 'GER30', 'FRA40',
'AUS200', 'JPN225', 'HK50',
# Cryptocurrency CFDs
'BTCUSD', 'ETHUSD', 'LTCUSD', 'XRPUSD'
]
def connect(self, credentials: Dict) -> bool:
"""Connect to XM Indonesia"""
try:
login = credentials.get("login")
password = credentials.get("password")
server = credentials.get("server", "XM-Demo" if self.demo else "XM-Real")
if not all([login, password]):
logger.error("XM Indonesia login and password required")
return False
# XM uses MT4/MT5 platform, so similar to existing MT5 integration
self.is_connected = True
logger.info(f"Connected to XM Indonesia {'Demo' if self.demo else 'Live'}")
return True
except Exception as e:
logger.error(f"Failed to connect to XM Indonesia: {e}")
return False
def disconnect(self) -> bool:
self.is_connected = False
return True
def get_symbols(self) -> List[str]:
return self.supported_symbols
def get_market_data(self, symbol: str, timeframe: Timeframe, count: int = 500) -> pd.DataFrame:
# Generate simulated forex data
dates = pd.date_range(end=datetime.now(), periods=count, freq='1h')
base_prices = {'EURUSD': 1.0850, 'USDIDR': 15400, 'XAUUSD': 2020}
base_price = base_prices.get(symbol, 1.0)
returns = np.random.randn(count) * 0.01
prices = base_price * (1 + returns).cumprod()
return pd.DataFrame({
'time': dates, 'open': prices, 'high': prices * 1.002,
'low': prices * 0.998, 'close': prices, 'volume': np.random.randint(1000, 10000, count)
})
def get_current_price(self, symbol: str) -> Dict[str, float]:
df = self.get_market_data(symbol, Timeframe.M1, 1)
if not df.empty:
price = df.iloc[-1]['close']
return {"bid": price - 0.0001, "ask": price + 0.0001}
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) -> Order:
order = Order(str(int(time.time())), symbol, order_type, side.lower(), size, price)
order.status = OrderStatus.FILLED
return order
def cancel_order(self, order_id: str) -> bool:
return True
def get_positions(self) -> List[Position]:
return []
def get_orders(self) -> List[Order]:
return []
def get_account_info(self) -> AccountInfo:
return AccountInfo(10000, 10000, 0, 10000, 100, "USD")
def get_trade_history(self, days: int = 30) -> List[Dict]:
return []
class OctaFXIndonesiaBroker(BaseBroker):
"""
OctaFX Indonesia - Another popular international broker
Known for good spreads and demo accounts
"""
def __init__(self, demo: bool = True):
super().__init__("OctaFX Indonesia")
self.demo = demo
self.supported_symbols = [
# Forex majors and minors
'EURUSD', 'GBPUSD', 'USDJPY', 'USDCHF', 'AUDUSD', 'USDCAD',
'NZDUSD', 'EURGBP', 'EURJPY', 'GBPJPY', 'AUDJPY', 'NZDJPY',
'EURCHF', 'GBPCHF', 'AUDCHF', 'NZDCHF', 'CADCHF', 'CHFJPY',
# Exotic pairs including IDR
'USDIDR', 'USDSGD', 'USDTHB', 'USDMYR',
# Metals
'XAUUSD', 'XAGUSD', 'XPDUSD', 'XPTUSD',
# Energies
'USOIL', 'UKOIL', 'NGAS',
# Indices
'SPX500', 'NAS100', 'US30', 'UK100', 'GER30', 'FRA40', 'ESP35',
'ITA40', 'AUS200', 'JPN225', 'HK50'
]
def connect(self, credentials: Dict) -> bool:
self.is_connected = True
return True
def disconnect(self) -> bool:
self.is_connected = False
return True
def get_symbols(self) -> List[str]:
return self.supported_symbols
def get_market_data(self, symbol: str, timeframe: Timeframe, count: int = 500) -> pd.DataFrame:
dates = pd.date_range(end=datetime.now(), periods=count, freq='1h')
base_price = 1.0850 if 'EUR' in symbol else 15400 if 'IDR' in symbol else 100
returns = np.random.randn(count) * 0.01
prices = base_price * (1 + returns).cumprod()
return pd.DataFrame({
'time': dates, 'open': prices, 'high': prices * 1.001,
'low': prices * 0.999, 'close': prices, 'volume': np.random.randint(1000, 5000, count)
})
def get_current_price(self, symbol: str) -> Dict[str, float]:
df = self.get_market_data(symbol, Timeframe.M1, 1)
if not df.empty:
price = df.iloc[-1]['close']
return {"bid": price - 0.0001, "ask": price + 0.0001}
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) -> Order:
order = Order(str(int(time.time())), symbol, order_type, side.lower(), size, price)
order.status = OrderStatus.FILLED
return order
def cancel_order(self, order_id: str) -> bool:
return True
def get_positions(self) -> List[Position]:
return []
def get_orders(self) -> List[Order]:
return []
def get_account_info(self) -> AccountInfo:
return AccountInfo(10000, 10000, 0, 10000, 100, "USD")
def get_trade_history(self, days: int = 30) -> List[Dict]:
return []
class HSBCIndonesiaBroker(BaseBroker):
"""
HSBC Indonesia - International bank with trading platform
Good for forex and international markets
"""
def __init__(self, demo: bool = True):
super().__init__("HSBC Indonesia")
self.demo = demo
self.supported_symbols = [
# Major currencies
'EURUSD', 'GBPUSD', 'USDJPY', 'USDCHF', 'AUDUSD', 'USDCAD',
# Asian currencies (HSBC specialty)
'USDIDR', 'USDSGD', 'USDHKD', 'USDKRW', 'USDCNY', 'USDTHB',
'USDMYR', 'USDPHP', 'USDVND',
# Cross currencies
'EURIDR', 'GBPIDR', 'AUDIDR', 'JPYIDR', 'SGDIDR',
# Precious metals
'XAUUSD', 'XAGUSD'
]
def connect(self, credentials: Dict) -> bool:
self.is_connected = True
return True
def disconnect(self) -> bool:
self.is_connected = False
return True
def get_symbols(self) -> List[str]:
return self.supported_symbols
def get_market_data(self, symbol: str, timeframe: Timeframe, count: int = 500) -> pd.DataFrame:
dates = pd.date_range(end=datetime.now(), periods=count, freq='1h')
base_price = 15400 if 'IDR' in symbol else 1.0850 if 'EUR' in symbol else 100
returns = np.random.randn(count) * 0.008
prices = base_price * (1 + returns).cumprod()
return pd.DataFrame({
'time': dates, 'open': prices, 'high': prices * 1.001,
'low': prices * 0.999, 'close': prices, 'volume': np.random.randint(500, 2000, count)
})
def get_current_price(self, symbol: str) -> Dict[str, float]:
df = self.get_market_data(symbol, Timeframe.M1, 1)
if not df.empty:
price = df.iloc[-1]['close']
return {"bid": price - 0.0002, "ask": price + 0.0002}
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) -> Order:
order = Order(str(int(time.time())), symbol, order_type, side.lower(), size, price)
order.status = OrderStatus.FILLED
return order
def cancel_order(self, order_id: str) -> bool:
return True
def get_positions(self) -> List[Position]:
return []
def get_orders(self) -> List[Order]:
return []
def get_account_info(self) -> AccountInfo:
return AccountInfo(10000, 10000, 0, 10000, 100, "USD")
def get_trade_history(self, days: int = 30) -> List[Dict]:
return []
# Factory function for Indonesian brokers
def create_indonesian_broker(broker_name: str, demo: bool = True) -> BaseBroker:
"""Create Indonesian broker instance"""
brokers = {
'indopremier': IndopremierBroker,
'xm_indonesia': XMIndonesiaBroker,
'octafx_indonesia': OctaFXIndonesiaBroker,
'hsbc_indonesia': HSBCIndonesiaBroker
}
broker_class = brokers.get(broker_name.lower())
if broker_class:
return broker_class(demo=demo)
else:
raise ValueError(f"Unknown Indonesian broker: {broker_name}")
# Indonesian market information
INDONESIAN_MARKET_INFO = {
'market_hours': {
'idx_stocks': 'Monday-Friday 09:00-16:00 WIB (GMT+7)',
'forex_local': '24/5 (follows global forex)',
'commodities': '24/5 (follows global commodities)'
},
'popular_stocks': {
'BBCA.JK': 'Bank Central Asia - Largest private bank',
'BBRI.JK': 'Bank Rakyat Indonesia - State-owned bank',
'BMRI.JK': 'Bank Mandiri - Largest bank by assets',
'TLKM.JK': 'Telkom Indonesia - Telecom giant',
'ASII.JK': 'Astra International - Automotive conglomerate',
'UNVR.JK': 'Unilever Indonesia - Consumer goods',
'ICBP.JK': 'Indofood CBP - Food and beverages',
'GGRM.JK': 'Gudang Garam - Cigarette manufacturer',
'HMSP.JK': 'HM Sampoerna - Tobacco company'
},
'currency_info': {
'base_currency': 'IDR (Indonesian Rupiah)',
'typical_usd_idr': '15,000-16,000 IDR per USD',
'volatility': 'Moderate, influenced by commodity prices'
},
'regulatory_info': {
'regulator': 'OJK (Otoritas Jasa Keuangan)',
'stock_exchange': 'IDX (Indonesia Stock Exchange)',
'trading_lot': '100 shares minimum for most stocks'
}
}
core/brokers/interactive_brokers.py
-491
View File
@@ -1,491 +0,0 @@
# core/brokers/interactive_brokers.py
"""
Interactive Brokers Integration for QuantumBotX
Professional-grade multi-asset trading platform
"""
import pandas as pd
import time
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import logging
import threading
from .base_broker import (
BaseBroker, OrderType, OrderStatus, Timeframe,
Position, Order, AccountInfo
)
logger = logging.getLogger(__name__)
class InteractiveBrokersBroker(BaseBroker):
"""
Interactive Brokers (IBKR) implementation using TWS API.
Supports stocks, forex, futures, options, and more.
"""
def __init__(self, paper_trading: bool = True):
super().__init__("Interactive Brokers")
self.paper_trading = paper_trading
self.ib_app = None
self.client_id = 1 # Unique client ID
self.port = 7497 if paper_trading else 7496 # Paper vs Live port
self.host = "127.0.0.1"
self.is_connected_flag = False
# Data storage
self.positions_data = {}
self.orders_data = {}
self.account_data = {}
self.market_data_cache = {}
# Timeframe mapping (IB uses specific duration/bar size combinations)
self.timeframe_map = {
Timeframe.M1: ("1 D", "1 min"), # 1 day of 1-minute bars
Timeframe.M5: ("5 D", "5 mins"), # 5 days of 5-minute bars
Timeframe.M15: ("10 D", "15 mins"), # 10 days of 15-minute bars
Timeframe.M30: ("1 M", "30 mins"), # 1 month of 30-minute bars
Timeframe.H1: ("1 M", "1 hour"), # 1 month of 1-hour bars
Timeframe.H4: ("3 M", "4 hours"), # 3 months of 4-hour bars
Timeframe.D1: ("1 Y", "1 day"), # 1 year of daily bars
}
def connect(self, credentials: Dict) -> bool:
"""
Connect to Interactive Brokers TWS/Gateway
credentials: {"host": "127.0.0.1", "port": 7497, "client_id": 1}
"""
try:
# Import here to avoid dependency issues if not installed
from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.contract import Contract
# Override connection parameters if provided
self.host = credentials.get("host", self.host)
self.port = credentials.get("port", self.port)
self.client_id = credentials.get("client_id", self.client_id)
# Create IB App class that combines EClient and EWrapper
class IBApp(EWrapper, EClient):
def __init__(self, broker_instance):
EClient.__init__(self, self)
self.broker = broker_instance
self.next_order_id = None
def nextValidId(self, orderId: int):
"""Callback when connection is established"""
self.next_order_id = orderId
self.broker.is_connected_flag = True
logger.info(f"IB connection established. Next order ID: {orderId}")
def accountSummary(self, reqId: int, account: str, tag: str, value: str, currency: str):
"""Account summary callback"""
if account not in self.broker.account_data:
self.broker.account_data[account] = {}
self.broker.account_data[account][tag] = {
'value': value,
'currency': currency
}
def position(self, account: str, contract, position: float, avgCost: float):
"""Position callback"""
symbol = contract.symbol
self.broker.positions_data[symbol] = {
'account': account,
'symbol': symbol,
'position': position,
'avg_cost': avgCost,
'contract': contract
}
def openOrder(self, orderId, contract, order, orderState):
"""Open order callback"""
self.broker.orders_data[orderId] = {
'order_id': orderId,
'contract': contract,
'order': order,
'state': orderState
}
def historicalData(self, reqId, bar):
"""Historical data callback"""
if reqId not in self.broker.market_data_cache:
self.broker.market_data_cache[reqId] = []
self.broker.market_data_cache[reqId].append({
'date': bar.date,
'open': bar.open,
'high': bar.high,
'low': bar.low,
'close': bar.close,
'volume': bar.volume
})
def error(self, reqId, errorCode, errorString, advancedOrderRejectJson=""):
"""Error callback"""
logger.error(f"IB Error {errorCode}: {errorString}")
# Create and connect IB app
self.ib_app = IBApp(self)
self.ib_app.connect(self.host, self.port, self.client_id)
# Start message processing in separate thread
def run_loop():
self.ib_app.run()
api_thread = threading.Thread(target=run_loop, daemon=True)
api_thread.start()
# Wait for connection
timeout = 10 # 10 seconds timeout
for _ in range(timeout * 10): # Check every 0.1 seconds
if self.is_connected_flag:
break
time.sleep(0.1)
if self.is_connected_flag:
self.is_connected = True
# Request account summary
self.ib_app.reqAccountSummary(1, "All", "$LEDGER")
time.sleep(2) # Wait for data
# Load supported symbols (simplified list)
self.supported_symbols = [
# Forex
'EUR.USD', 'GBP.USD', 'USD.JPY', 'USD.CHF', 'AUD.USD', 'USD.CAD',
# Stocks
'AAPL', 'GOOGL', 'MSFT', 'TSLA', 'AMZN', 'META',
# Futures
'ES', 'NQ', 'YM', 'RTY', # Stock index futures
'GC', 'SI', 'CL', # Commodity futures
]
logger.info(f"Connected to Interactive Brokers {'Paper' if self.paper_trading else 'Live'}")
return True
else:
logger.error("Failed to establish IB connection within timeout")
return False
except ImportError:
logger.error("ibapi package not installed. Install with: pip install ibapi")
return False
except Exception as e:
logger.error(f"Failed to connect to Interactive Brokers: {e}")
self.is_connected = False
return False
def disconnect(self) -> bool:
"""Disconnect from Interactive Brokers"""
if self.ib_app:
self.ib_app.disconnect()
self.is_connected = False
self.is_connected_flag = False
logger.info("Disconnected from Interactive Brokers")
return True
def get_symbols(self) -> List[str]:
"""Get list of available trading symbols"""
return self.supported_symbols
def _create_contract(self, symbol: str) -> 'Contract':
"""Create IB Contract object for symbol"""
from ibapi.contract import Contract
contract = Contract()
# Determine contract type based on symbol format
if '.' in symbol: # Forex (EUR.USD format)
base, quote = symbol.split('.')
contract.symbol = base
contract.secType = "CASH"
contract.currency = quote
contract.exchange = "IDEALPRO"
elif symbol in ['ES', 'NQ', 'YM', 'RTY', 'GC', 'SI', 'CL']: # Futures
contract.symbol = symbol
contract.secType = "FUT"
contract.exchange = "CME" # Simplified
contract.lastTradeDateOrContractMonth = "202412" # Would need dynamic
else: # Stocks
contract.symbol = symbol
contract.secType = "STK"
contract.currency = "USD"
contract.exchange = "SMART"
return contract
def get_market_data(self, symbol: str, timeframe: Timeframe, count: int = 500) -> pd.DataFrame:
"""Get OHLCV market data from Interactive Brokers"""
if not self.is_connected:
raise Exception("Not connected to Interactive Brokers")
try:
contract = self._create_contract(symbol)
duration, bar_size = self.timeframe_map[timeframe]
# Request historical data
req_id = int(time.time()) # Unique request ID
self.market_data_cache[req_id] = []
self.ib_app.reqHistoricalData(
req_id, contract, "", duration, bar_size, "TRADES", 1, 1, False, []
)
# Wait for data
timeout = 10
for _ in range(timeout * 10):
if req_id in self.market_data_cache and len(self.market_data_cache[req_id]) > 0:
break
time.sleep(0.1)
# Convert to DataFrame
data = self.market_data_cache.get(req_id, [])
if not data:
return pd.DataFrame()
df_data = []
for bar in data:
# Parse IB date format
try:
if len(bar['date']) == 8: # Daily format: 20231201
date_obj = datetime.strptime(bar['date'], '%Y%m%d')
else: # Intraday format: 20231201 10:30:00
date_obj = datetime.strptime(bar['date'], '%Y%m%d %H:%M:%S')
except:
date_obj = datetime.now()
df_data.append({
'time': date_obj,
'open': bar['open'],
'high': bar['high'],
'low': bar['low'],
'close': bar['close'],
'volume': bar['volume']
})
# Clean up cache
del self.market_data_cache[req_id]
return pd.DataFrame(df_data)
except Exception as e:
logger.error(f"Failed to get IB market data for {symbol}: {e}")
return pd.DataFrame()
def get_current_price(self, symbol: str) -> Dict[str, float]:
"""Get current bid/ask prices"""
if not self.is_connected:
raise Exception("Not connected to Interactive Brokers")
try:
# IB requires market data subscription for real-time prices
# For demo purposes, return last close price as both bid/ask
# In real implementation, would use reqMktData
df = self.get_market_data(symbol, Timeframe.M1, 1)
if not df.empty:
last_price = df.iloc[-1]['close']
return {"bid": last_price - 0.0001, "ask": last_price + 0.0001}
else:
return {"bid": 0.0, "ask": 0.0}
except Exception as e:
logger.error(f"Failed to get IB current price for {symbol}: {e}")
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) -> Order:
"""Place a trading order on Interactive Brokers"""
if not self.is_connected:
raise Exception("Not connected to Interactive Brokers")
try:
from ibapi.order import Order as IBOrder
contract = self._create_contract(symbol)
# Create IB order
ib_order = IBOrder()
ib_order.action = "BUY" if side.lower() == "buy" else "SELL"
ib_order.totalQuantity = size
# Set order type
if order_type in [OrderType.MARKET_BUY, OrderType.MARKET_SELL]:
ib_order.orderType = "MKT"
elif order_type in [OrderType.LIMIT_BUY, OrderType.LIMIT_SELL]:
ib_order.orderType = "LMT"
ib_order.lmtPrice = price
# Get next order ID
if not self.ib_app.next_order_id:
logger.error("No valid order ID available")
raise Exception("No valid order ID")
order_id = self.ib_app.next_order_id
self.ib_app.next_order_id += 1
# Place order
self.ib_app.placeOrder(order_id, contract, ib_order)
# Create Order object
order = Order(
order_id=str(order_id),
symbol=symbol,
order_type=order_type,
side=side.lower(),
size=size,
price=price
)
order.status = OrderStatus.PENDING
logger.info(f"IB order placed: {order_id} for {symbol}")
return order
except Exception as e:
logger.error(f"Failed to place IB order: {e}")
order = Order(
order_id="failed",
symbol=symbol,
order_type=order_type,
side=side.lower(),
size=size,
price=price
)
order.status = OrderStatus.REJECTED
return order
def cancel_order(self, order_id: str) -> bool:
"""Cancel an existing order"""
if not self.is_connected:
return False
try:
self.ib_app.cancelOrder(int(order_id))
return True
except Exception as e:
logger.error(f"Failed to cancel IB order {order_id}: {e}")
return False
def get_positions(self) -> List[Position]:
"""Get all open positions"""
if not self.is_connected:
return []
try:
# Request positions
self.ib_app.reqPositions()
time.sleep(2) # Wait for data
positions = []
for symbol, pos_data in self.positions_data.items():
if pos_data['position'] != 0: # Only non-zero positions
position = Position(
symbol=symbol,
side='long' if pos_data['position'] > 0 else 'short',
size=abs(pos_data['position']),
entry_price=pos_data['avg_cost'],
current_price=pos_data['avg_cost'], # Would need market price
unrealized_pnl=0.0 # Would need calculation
)
positions.append(position)
return positions
except Exception as e:
logger.error(f"Failed to get IB positions: {e}")
return []
def get_orders(self) -> List[Order]:
"""Get all pending orders"""
if not self.is_connected:
return []
try:
# Request open orders
self.ib_app.reqOpenOrders()
time.sleep(2) # Wait for data
orders = []
for order_id, order_data in self.orders_data.items():
order = Order(
order_id=str(order_id),
symbol=order_data['contract'].symbol,
order_type=OrderType.LIMIT_BUY, # Simplified
side=order_data['order'].action.lower(),
size=order_data['order'].totalQuantity,
price=getattr(order_data['order'], 'lmtPrice', None)
)
order.status = OrderStatus.PENDING
orders.append(order)
return orders
except Exception as e:
logger.error(f"Failed to get IB orders: {e}")
return []
def get_account_info(self) -> AccountInfo:
"""Get account information"""
if not self.is_connected:
return AccountInfo(0, 0, 0, 0, 0, "USD")
try:
# Use cached account data
account_data = list(self.account_data.values())[0] if self.account_data else {}
net_liquidation = float(account_data.get('NetLiquidation', {}).get('value', 0))
total_cash = float(account_data.get('TotalCashValue', {}).get('value', 0))
buying_power = float(account_data.get('BuyingPower', {}).get('value', 0))
return AccountInfo(
balance=total_cash,
equity=net_liquidation,
margin=0.0, # Would need calculation
free_margin=buying_power,
margin_level=100.0, # Would need calculation
currency="USD"
)
except Exception as e:
logger.error(f"Failed to get IB account info: {e}")
return AccountInfo(0, 0, 0, 0, 0, "USD")
def get_trade_history(self, days: int = 30) -> List[Dict]:
"""Get trade history"""
if not self.is_connected:
return []
try:
# IB trade history would require execution reports
# For now, return empty list
logger.warning("IB trade history not implemented - requires execution report handling")
return []
except Exception as e:
logger.error(f"Failed to get IB trade history: {e}")
return []
def normalize_symbol(self, symbol: str) -> str:
"""Normalize symbol format for Interactive Brokers"""
# Convert common formats to IB format
symbol = symbol.upper()
# Forex: EURUSD -> EUR.USD
forex_pairs = ['EURUSD', 'GBPUSD', 'USDJPY', 'USDCHF', 'AUDUSD', 'USDCAD']
for pair in forex_pairs:
if symbol == pair:
return f"{pair[:3]}.{pair[3:]}"
return symbol
def is_market_open(self) -> bool:
"""Check if markets are open (simplified)"""
now = datetime.now()
# US market hours: weekdays, roughly 9:30 AM - 4:00 PM ET
return now.weekday() < 5 # Simplified
# Convenience function
def create_ib_broker(paper_trading: bool = True) -> InteractiveBrokersBroker:
"""Create an Interactive Brokers broker instance"""
return InteractiveBrokersBroker(paper_trading=paper_trading)
core/brokers/tradingview_broker.py
-449
View File
@@ -1,449 +0,0 @@
# core/brokers/tradingview_broker.py
"""
TradingView Integration for QuantumBotX
Social trading platform with Pine Script integration
"""
import pandas as pd
import time
import requests
import json
import websocket
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import logging
import threading
from .base_broker import (
BaseBroker, OrderType, OrderStatus, Timeframe,
Position, Order, AccountInfo
)
logger = logging.getLogger(__name__)
class TradingViewBroker(BaseBroker):
"""
TradingView integration for QuantumBotX.
Note: This is a conceptual implementation as TradingView doesn't have
a traditional trading API. In practice, this would work through:
1. Webhook signals from TradingView alerts
2. Screen scraping (not recommended)
3. Third-party integrations
This implementation shows how it would work architecturally.
"""
def __init__(self, paper_trading: bool = True):
super().__init__("TradingView")
self.paper_trading = paper_trading
self.session = requests.Session()
self.websocket = None
self.webhook_server = None
# TradingView doesn't provide direct API access
# This would work through webhook alerts
self.base_url = "https://www.tradingview.com"
# Simulated data for demo purposes
self.portfolio = {}
self.pending_orders = {}
self.trade_history = []
self.current_capital = 10000.0
# Timeframe mapping
self.timeframe_map = {
Timeframe.M1: "1",
Timeframe.M5: "5",
Timeframe.M15: "15",
Timeframe.M30: "30",
Timeframe.H1: "60",
Timeframe.H4: "240",
Timeframe.D1: "1D"
}
def connect(self, credentials: Dict) -> bool:
"""
Connect to TradingView (conceptual)
credentials: {"username": "...", "password": "...", "webhook_secret": "..."}
"""
try:
username = credentials.get("username")
password = credentials.get("password")
webhook_secret = credentials.get("webhook_secret")
if not all([username, webhook_secret]):
logger.error("TradingView username and webhook_secret are required")
return False
# In real implementation, would set up webhook server
self._setup_webhook_server(webhook_secret)
self.is_connected = True
# Popular tradingview symbols
self.supported_symbols = [
# Forex
'EURUSD', 'GBPUSD', 'USDJPY', 'USDCHF', 'AUDUSD', 'USDCAD',
'NZDUSD', 'EURGBP', 'EURJPY', 'GBPJPY',
# Crypto
'BTCUSD', 'ETHUSD', 'ADAUSD', 'SOLUSD', 'DOGEUSD',
# Stocks
'AAPL', 'GOOGL', 'MSFT', 'TSLA', 'AMZN', 'META', 'NVDA',
# Commodities
'XAUUSD', 'XAGUSD', 'USOIL', 'UKOIL',
# Indices
'SPX', 'DJI', 'NDX', 'RUT'
]
logger.info(f"Connected to TradingView {'Paper' if self.paper_trading else 'Live'}")
return True
except Exception as e:
logger.error(f"Failed to connect to TradingView: {e}")
self.is_connected = False
return False
def _setup_webhook_server(self, webhook_secret: str):
"""Setup webhook server to receive TradingView alerts"""
try:
from flask import Flask, request, jsonify
webhook_app = Flask(__name__)
@webhook_app.route('/tradingview-webhook', methods=['POST'])
def handle_webhook():
try:
# Verify webhook secret
received_secret = request.headers.get('X-Webhook-Secret')
if received_secret != webhook_secret:
return jsonify({'error': 'Invalid webhook secret'}), 401
# Parse alert data
alert_data = request.get_json()
self._process_tradingview_alert(alert_data)
return jsonify({'status': 'success'}), 200
except Exception as e:
logger.error(f"Webhook error: {e}")
return jsonify({'error': str(e)}), 500
# Run webhook server in background thread
def run_webhook():
webhook_app.run(host='0.0.0.0', port=5001, debug=False)
webhook_thread = threading.Thread(target=run_webhook, daemon=True)
webhook_thread.start()
logger.info("TradingView webhook server started on port 5001")
except ImportError:
logger.warning("Flask not available for webhook server")
except Exception as e:
logger.error(f"Failed to setup webhook server: {e}")
def _process_tradingview_alert(self, alert_data: Dict):
"""Process incoming TradingView alert"""
try:
# Expected alert format:
# {
# "symbol": "EURUSD",
# "action": "buy" or "sell",
# "price": 1.0850,
# "stop_loss": 1.0800,
# "take_profit": 1.0900,
# "quantity": 1.0,
# "strategy": "My Strategy"
# }
symbol = alert_data.get('symbol')
action = alert_data.get('action', '').lower()
price = float(alert_data.get('price', 0))
quantity = float(alert_data.get('quantity', 1.0))
if action in ['buy', 'sell'] and symbol and price > 0:
# Execute the trade
order_type = OrderType.MARKET_BUY if action == 'buy' else OrderType.MARKET_SELL
order = self.place_order(
symbol=symbol,
order_type=order_type,
side=action,
size=quantity,
price=price,
stop_loss=alert_data.get('stop_loss'),
take_profit=alert_data.get('take_profit')
)
logger.info(f"TradingView alert processed: {action} {quantity} {symbol} at {price}")
except Exception as e:
logger.error(f"Failed to process TradingView alert: {e}")
def disconnect(self) -> bool:
"""Disconnect from TradingView"""
self.is_connected = False
logger.info("Disconnected from TradingView")
return True
def get_symbols(self) -> List[str]:
"""Get list of available trading symbols"""
return self.supported_symbols
def get_market_data(self, symbol: str, timeframe: Timeframe, count: int = 500) -> pd.DataFrame:
"""
Get market data from TradingView
Note: This would require web scraping or third-party API
"""
try:
# For demo purposes, generate simulated data
# In real implementation, would scrape TradingView charts or use third-party API
logger.warning("TradingView market data: Using simulated data (real implementation would require scraping)")
# Generate simulated price data
dates = pd.date_range(end=datetime.now(), periods=count, freq='1h')
# Base prices for different symbols
base_prices = {
'EURUSD': 1.0850, 'GBPUSD': 1.2650, 'USDJPY': 148.50,
'BTCUSD': 42000, 'ETHUSD': 2500, 'AAPL': 190.0,
'XAUUSD': 2020.0, 'SPX': 4500.0
}
base_price = base_prices.get(symbol, 100.0)
# Generate price movements
returns = np.random.randn(count) * 0.01 # 1% volatility
prices = base_price * (1 + returns).cumprod()
df = pd.DataFrame({
'time': dates,
'open': prices,
'high': prices * (1 + np.random.uniform(0, 0.005, count)),
'low': prices * (1 - np.random.uniform(0, 0.005, count)),
'close': prices,
'volume': np.random.randint(1000, 10000, count)
})
# Ensure OHLC integrity
df['high'] = df[['high', 'close', 'open']].max(axis=1)
df['low'] = df[['low', 'close', 'open']].min(axis=1)
return df
except Exception as e:
logger.error(f"Failed to get TradingView market data for {symbol}: {e}")
return pd.DataFrame()
def get_current_price(self, symbol: str) -> Dict[str, float]:
"""Get current bid/ask prices"""
try:
# In real implementation, would scrape TradingView or use websocket
df = self.get_market_data(symbol, Timeframe.M1, 1)
if not df.empty:
last_price = df.iloc[-1]['close']
spread = last_price * 0.0001 # Typical spread
return {
"bid": last_price - spread/2,
"ask": last_price + spread/2
}
return {"bid": 0.0, "ask": 0.0}
except Exception as e:
logger.error(f"Failed to get TradingView current price for {symbol}: {e}")
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) -> Order:
"""
Place order (simulated for TradingView)
In practice, this would trigger through connected broker
"""
try:
order_id = str(int(time.time()))
# Simulate order execution
if order_type in [OrderType.MARKET_BUY, OrderType.MARKET_SELL]:
current_price = self.get_current_price(symbol)
execution_price = current_price['ask'] if side.lower() == 'buy' else current_price['bid']
else:
execution_price = price
# Create order
order = Order(
order_id=order_id,
symbol=symbol,
order_type=order_type,
side=side.lower(),
size=size,
price=execution_price
)
# Simulate immediate execution for market orders
if order_type in [OrderType.MARKET_BUY, OrderType.MARKET_SELL]:
order.status = OrderStatus.FILLED
order.filled_size = size
order.avg_fill_price = execution_price
# Update portfolio
if symbol not in self.portfolio:
self.portfolio[symbol] = {'long': 0, 'short': 0, 'avg_price': 0}
if side.lower() == 'buy':
self.portfolio[symbol]['long'] += size
else:
self.portfolio[symbol]['short'] += size
# Add to trade history
self.trade_history.append({
'time': datetime.now(),
'symbol': symbol,
'side': side.lower(),
'size': size,
'price': execution_price,
'order_id': order_id
})
logger.info(f"TradingView simulated order executed: {side} {size} {symbol} at {execution_price}")
else:
order.status = OrderStatus.PENDING
self.pending_orders[order_id] = order
return order
except Exception as e:
logger.error(f"Failed to place TradingView order: {e}")
order = Order(
order_id="failed",
symbol=symbol,
order_type=order_type,
side=side.lower(),
size=size,
price=price
)
order.status = OrderStatus.REJECTED
return order
def cancel_order(self, order_id: str) -> bool:
"""Cancel an existing order"""
try:
if order_id in self.pending_orders:
del self.pending_orders[order_id]
return True
return False
except Exception as e:
logger.error(f"Failed to cancel TradingView order {order_id}: {e}")
return False
def get_positions(self) -> List[Position]:
"""Get all open positions"""
try:
positions = []
for symbol, pos_data in self.portfolio.items():
long_size = pos_data['long']
short_size = pos_data['short']
net_size = long_size - short_size
if net_size != 0:
current_price_data = self.get_current_price(symbol)
current_price = current_price_data['bid'] if net_size > 0 else current_price_data['ask']
position = Position(
symbol=symbol,
side='long' if net_size > 0 else 'short',
size=abs(net_size),
entry_price=pos_data.get('avg_price', current_price),
current_price=current_price,
unrealized_pnl=0.0 # Would calculate based on entry vs current
)
positions.append(position)
return positions
except Exception as e:
logger.error(f"Failed to get TradingView positions: {e}")
return []
def get_orders(self) -> List[Order]:
"""Get all pending orders"""
return list(self.pending_orders.values())
def get_account_info(self) -> AccountInfo:
"""Get account information"""
try:
# Simulate account info
return AccountInfo(
balance=self.current_capital,
equity=self.current_capital, # Simplified
margin=0.0,
free_margin=self.current_capital,
margin_level=100.0,
currency="USD"
)
except Exception as e:
logger.error(f"Failed to get TradingView account info: {e}")
return AccountInfo(0, 0, 0, 0, 0, "USD")
def get_trade_history(self, days: int = 30) -> List[Dict]:
"""Get trade history"""
try:
cutoff_date = datetime.now() - timedelta(days=days)
recent_trades = [
trade for trade in self.trade_history
if trade['time'] >= cutoff_date
]
return recent_trades
except Exception as e:
logger.error(f"Failed to get TradingView trade history: {e}")
return []
def normalize_symbol(self, symbol: str) -> str:
"""Normalize symbol format for TradingView"""
# TradingView uses various symbol formats
symbol = symbol.upper()
# Convert some common formats
if symbol == 'XAUUSD':
return 'GOLD'
elif symbol == 'XAGUSD':
return 'SILVER'
elif symbol.endswith('USDT'):
return symbol.replace('USDT', 'USD')
return symbol
def is_market_open(self) -> bool:
"""TradingView shows global markets - always something open"""
return True
def create_pine_script_strategy(self, strategy_code: str) -> str:
"""
Create a Pine Script strategy (conceptual)
Returns strategy ID for webhook alerts
"""
try:
# In real implementation, would create TradingView strategy
# and set up webhook alerts
strategy_id = f"strategy_{int(time.time())}"
logger.info(f"Pine Script strategy created (simulated): {strategy_id}")
logger.info("Set up TradingView alerts with webhook URL: http://your-server.com:5001/tradingview-webhook")
return strategy_id
except Exception as e:
logger.error(f"Failed to create Pine Script strategy: {e}")
return ""
# Convenience function
def create_tradingview_broker(paper_trading: bool = True) -> TradingViewBroker:
"""Create a TradingView broker instance"""
return TradingViewBroker(paper_trading=paper_trading)
core/interfaces/__init__.py
-1
View File
@@ -1 +0,0 @@
# Init file for core/interfaces
testing/indonesian_market_demo.py
-353
View File
@@ -1,353 +0,0 @@
#!/usr/bin/env python3
"""
Indonesian Market Trading Demo for QuantumBotX
Showcasing opportunities in Indonesian financial markets
"""
import sys
import os
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
# Add the project root to the path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
def demo_indonesian_market_overview():
"""Overview of Indonesian trading opportunities"""
print("🇮🇩 Indonesian Market Trading Opportunities")
print("=" * 60)
print("Welcome to the Indonesian Financial Markets!")
print("=" * 60)
market_segments = {
'IDX Stocks (Jakarta Stock Exchange)': {
'description': 'Local Indonesian companies',
'examples': ['BBCA.JK (BCA)', 'BBRI.JK (BRI)', 'TLKM.JK (Telkom)'],
'trading_hours': '09:00-16:00 WIB (GMT+7)',
'currency': 'IDR (Indonesian Rupiah)',
'min_lot': '100 shares',
'opportunities': ['Banking sector growth', 'Infrastructure development', 'Consumer goods expansion']
},
'USD/IDR Forex': {
'description': 'Indonesian Rupiah currency trading',
'examples': ['USDIDR', 'EURIDR', 'JPYIDR'],
'trading_hours': '24/5 (Global forex hours)',
'currency': 'IDR pairs',
'min_lot': 'Varies by broker',
'opportunities': ['Commodity-driven moves', 'Central bank policy', 'Tourism recovery']
},
'International Markets via Indonesian Brokers': {
'description': 'Global markets through local brokers',
'examples': ['XAUUSD', 'US stocks', 'Major forex pairs'],
'trading_hours': 'Varies by market',
'currency': 'USD typically',
'min_lot': 'Standard international',
'opportunities': ['Global diversification', 'USD income', 'Hedge against IDR']
}
}
print("\\n📊 Indonesian Market Segments:")
for i, (segment, details) in enumerate(market_segments.items(), 1):
print(f"\\n{i}. {segment}")
print(f" 📝 Description: {details['description']}")
print(f" 📈 Examples: {', '.join(details['examples'])}")
print(f" ⏰ Hours: {details['trading_hours']}")
print(f" 💰 Currency: {details['currency']}")
print(f" 🎯 Opportunities: {', '.join(details['opportunities'][:2])}")
def demo_indonesian_brokers():
"""Showcase Indonesian brokers with demo accounts"""
print("\\n🏢 Indonesian Brokers with Demo Accounts")
print("=" * 60)
brokers = [
{
'name': 'Indopremier Securities (IPOT)',
'type': 'Local Indonesian Broker',
'specialties': ['IDX Stocks', 'Local bonds', 'Indonesian mutual funds'],
'demo_account': 'Yes - Full IDX access',
'advantages': ['Local market expertise', 'IDR-based trading', 'Indonesian customer service'],
'website': 'https://www.indopremier.com/',
'best_for': 'Indonesian stock market and local investments'
},
{
'name': 'XM Indonesia',
'type': 'International Broker (Indonesia Office)',
'specialties': ['Forex', 'CFDs', 'Commodities', 'Crypto CFDs'],
'demo_account': 'Yes - $10,000 virtual',
'advantages': ['Global markets', 'MT4/MT5 platform', 'Indonesian support'],
'website': 'https://www.xm.com/id/',
'best_for': 'Forex and international markets'
},
{
'name': 'OctaFX Indonesia',
'type': 'International Broker (Popular in Indonesia)',
'specialties': ['Forex', 'Metals', 'Indices', 'Energies'],
'demo_account': 'Yes - Unlimited time',
'advantages': ['Tight spreads', 'Fast execution', 'Indonesian community'],
'website': 'https://www.octafx.com/id/',
'best_for': 'Professional forex trading'
},
{
'name': 'HSBC Indonesia',
'type': 'International Bank',
'specialties': ['Forex', 'Asian currencies', 'Trade finance'],
'demo_account': 'Available for qualified clients',
'advantages': ['Banking integration', 'Asian market focus', 'Multi-currency'],
'website': 'Contact local HSBC branch',
'best_for': 'Currency hedging and international business'
}
]
print("\\n🎯 Recommended Brokers for Indonesian Traders:")
for i, broker in enumerate(brokers, 1):
print(f"\\n{i}. {broker['name']}")
print(f" 🏢 Type: {broker['type']}")
print(f" 📈 Specialties: {', '.join(broker['specialties'][:3])}")
print(f" 🧪 Demo Account: {broker['demo_account']}")
print(f" ⭐ Best For: {broker['best_for']}")
print(f" 🌐 Website: {broker['website']}")
def demo_idx_stocks_trading():
"""Demo trading Indonesian stocks"""
print("\\n📈 IDX Stock Trading Simulation")
print("=" * 60)
# Simulate some popular Indonesian stocks
idx_stocks = [
{'symbol': 'BBCA.JK', 'name': 'Bank Central Asia', 'price': 9150, 'sector': 'Banking'},
{'symbol': 'BBRI.JK', 'name': 'Bank Rakyat Indonesia', 'price': 4520, 'sector': 'Banking'},
{'symbol': 'TLKM.JK', 'name': 'Telkom Indonesia', 'price': 3280, 'sector': 'Telecommunications'},
{'symbol': 'ASII.JK', 'name': 'Astra International', 'price': 6750, 'sector': 'Automotive'},
{'symbol': 'UNVR.JK', 'name': 'Unilever Indonesia', 'price': 7100, 'sector': 'Consumer Goods'},
]
print("\\n🏦 Popular IDX Stocks (Simulated Prices):")
print("Symbol | Company | Price (IDR) | Sector")
print("-" * 70)
total_portfolio_value = 0
for stock in idx_stocks:
# Simulate small price movements
current_price = stock['price'] * (1 + np.random.uniform(-0.02, 0.02))
change_pct = ((current_price - stock['price']) / stock['price']) * 100
# Simulate trading with 1000 IDR capital per stock
shares_affordable = int(100000 / current_price) # 100k IDR investment
position_value = shares_affordable * current_price
total_portfolio_value += position_value
color = "📈" if change_pct > 0 else "📉" if change_pct < 0 else "➡️"
print(f"{stock['symbol']:10} | {stock['name']:25} | {current_price:8.0f} {color} | {stock['sector']}")
print(f"\\n💼 Simulated Portfolio Value: {total_portfolio_value:,.0f} IDR")
print(f"💰 Equivalent in USD: ${total_portfolio_value/15400:.2f} (assuming 1 USD = 15,400 IDR)")
def demo_usd_idr_trading():
"""Demo USD/IDR forex trading"""
print("\\n💱 USD/IDR Forex Trading Simulation")
print("=" * 60)
# Current USD/IDR around 15,400
base_rate = 15400
# Simulate daily USD/IDR movements
days = 30
dates = pd.date_range(end=datetime.now(), periods=days, freq='D')
# IDR volatility (typically 0.5-1% daily)
daily_changes = np.random.randn(days) * 0.008 # 0.8% daily volatility
rates = base_rate * (1 + daily_changes).cumprod()
print(f"\\n📊 USD/IDR Rate Simulation (Last {days} days):")
print(f"Starting Rate: {base_rate:,.0f} IDR per USD")
print(f"Ending Rate: {rates[-1]:,.0f} IDR per USD")
print(f"Total Change: {((rates[-1] - base_rate) / base_rate) * 100:+.2f}%")
# Trading simulation
position_size = 10000 # $10,000 USD position
entry_rate = rates[0]
exit_rate = rates[-1]
if rates[-1] > rates[0]: # USD strengthened
pnl_usd = position_size * ((exit_rate - entry_rate) / entry_rate)
direction = "USD strengthened"
else: # USD weakened
pnl_usd = position_size * ((exit_rate - entry_rate) / entry_rate)
direction = "USD weakened"
pnl_idr = pnl_usd * exit_rate
print(f"\\n💹 Trading Simulation:")
print(f"Position: Long ${position_size:,} USD vs IDR")
print(f"Entry Rate: {entry_rate:,.0f} IDR/USD")
print(f"Exit Rate: {exit_rate:,.0f} IDR/USD")
print(f"Market Move: {direction}")
print(f"P&L: ${pnl_usd:+,.2f} USD (or {pnl_idr:+,.0f} IDR)")
def demo_strategy_performance_indonesia():
"""Test strategies on Indonesian markets"""
print("\\n🤖 Strategy Performance on Indonesian Markets")
print("=" * 60)
from core.brokers.indonesian_brokers import IndopremierBroker
# Create Indonesian broker instance
broker = IndopremierBroker(demo=True)
# Test symbols
test_symbols = [
('BBCA.JK', 'Bank Central Asia'),
('USDIDR', 'USD/IDR Forex'),
('XAUIDR', 'Gold in IDR')
]
print("\\n📈 Testing QuantumBotX Strategies on Indonesian Markets:")
for symbol, name in test_symbols:
try:
# Get simulated market data
df = broker.get_market_data(symbol, broker.timeframe_map[broker.Timeframe.H1] if hasattr(broker, 'timeframe_map') else 'H1', 500)
if not df.empty:
# Calculate basic metrics
volatility = (df['close'].std() / df['close'].mean()) * 100
price_range = f"{df['close'].min():.0f} - {df['close'].max():.0f}"
# Assess suitability for different strategies
if volatility < 2:
strategy_rec = "Bollinger Reversion (Low volatility)"
elif volatility > 5:
strategy_rec = "Conservative MA Crossover (High volatility)"
else:
strategy_rec = "QuantumBotX Hybrid (Moderate volatility)"
print(f"\\n📊 {symbol} ({name}):")
print(f" Price Range: {price_range}")
print(f" Volatility: {volatility:.1f}%")
print(f" Recommended Strategy: {strategy_rec}")
print(f" Data Points: {len(df)} bars")
else:
print(f"\\n❌ {symbol}: No data available")
except Exception as e:
print(f"\\n❌ {symbol}: Error - {e}")
def demo_regulatory_compliance():
"""Indonesian regulatory information"""
print("\\n⚖️ Indonesian Regulatory Compliance")
print("=" * 60)
regulatory_info = {
'Primary Regulator': {
'name': 'OJK (Otoritas Jasa Keuangan)',
'role': 'Financial Services Authority',
'website': 'https://www.ojk.go.id/',
'oversight': 'Banks, capital markets, insurance, pension funds'
},
'Stock Exchange': {
'name': 'IDX (Indonesia Stock Exchange)',
'location': 'Jakarta',
'website': 'https://www.idx.co.id/',
'trading_currency': 'Indonesian Rupiah (IDR)'
},
'Key Regulations': [
'Foreign investment limits in certain sectors',
'Tax obligations for trading profits',
'Anti-money laundering (AML) requirements',
'Know Your Customer (KYC) procedures'
],
'Tax Considerations': [
'Capital gains tax on stock trading',
'Forex trading taxation rules',
'Withholding tax on foreign investments',
'Professional trader vs investor classification'
]
}
print("\\n🏛️ Regulatory Framework:")
print(f"Primary Regulator: {regulatory_info['Primary Regulator']['name']}")
print(f"Stock Exchange: {regulatory_info['Stock Exchange']['name']}")
print("\\n⚠️ Important Considerations:")
for consideration in regulatory_info['Key Regulations'][:3]:
print(f" • {consideration}")
print("\\n💰 Tax Implications:")
for tax_item in regulatory_info['Tax Considerations'][:3]:
print(f" • {tax_item}")
print("\\n📝 Recommendation:")
print(" • Consult with Indonesian tax advisor")
print(" • Understand local broker regulations")
print(" • Keep detailed trading records")
print(" • Consider professional trader registration if applicable")
def main():
"""Main Indonesian market demo"""
print("🇮🇩 SELAMAT DATANG! Welcome to Indonesian Market Trading!")
print("Your QuantumBotX system now supports Indonesian markets!")
print()
# Run all demos
demo_indonesian_market_overview()
demo_indonesian_brokers()
demo_idx_stocks_trading()
demo_usd_idr_trading()
demo_strategy_performance_indonesia()
demo_regulatory_compliance()
print("\\n" + "=" * 60)
print("🎯 NEXT STEPS FOR INDONESIAN TRADING")
print("=" * 60)
next_steps = [
{
'step': '1. Choose Your Indonesian Broker',
'recommendation': 'Start with XM Indonesia demo (easiest setup)',
'action': 'Sign up for demo account at xm.com/id/'
},
{
'step': '2. Add Indonesian Configuration',
'recommendation': 'Update .env file with Indonesian broker credentials',
'action': 'Add XM_INDONESIA_LOGIN and XM_INDONESIA_PASSWORD'
},
{
'step': '3. Test IDX Stocks Strategy',
'recommendation': 'Start with banking stocks (BBCA, BBRI, BMRI)',
'action': 'Run backtests on Indonesian blue-chip stocks'
},
{
'step': '4. Explore USD/IDR Trading',
'recommendation': 'Great for Indonesian traders to earn USD',
'action': 'Test forex strategies on USD/IDR pair'
},
{
'step': '5. Regulatory Compliance',
'recommendation': 'Understand Indonesian tax obligations',
'action': 'Consult with local financial advisor'
}
]
for step_info in next_steps:
print(f"\\n{step_info['step']}")
print(f" 💡 Recommendation: {step_info['recommendation']}")
print(f" 🎯 Action: {step_info['action']}")
print("\\n🎉 AMAZING OPPORTUNITY!")
print("=" * 60)
print("You're now building a trading system that covers:")
print("✅ Global Forex (MT5, cTrader, XM)")
print("✅ Cryptocurrency (Binance)")
print("✅ US Stocks (Interactive Brokers)")
print("✅ Social Trading (TradingView)")
print("✅ Indonesian Markets (Local brokers)")
print()
print("🌏 FROM INDONESIA TO THE WORLD!")
print("Your trading system now spans the entire globe! 🚀")
if __name__ == "__main__":
main()
testing/multi_broker_universe_demo.py
-310
View File
@@ -1,310 +0,0 @@
#!/usr/bin/env python3
"""
Multi-Broker Universe Demo for QuantumBotX
Shows how to trade across all major platforms simultaneously
"""
import sys
import os
import pandas as pd
import numpy as np
from datetime import datetime
# Add the project root to the path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
def demo_all_brokers():
"""Demonstrate all broker integrations"""
print("🌍 QuantumBotX Multi-Broker Universe Demo")
print("=" * 60)
print("Your trading system now supports ALL major platforms!")
print("=" * 60)
brokers_info = [
{
'name': 'MetaTrader 5',
'type': 'Forex/CFD Platform',
'assets': ['EURUSD', 'GBPUSD', 'XAUUSD', 'US30', 'AAPL'],
'advantages': ['Most forex brokers', 'Expert Advisors', 'Built-in indicators'],
'best_for': 'Forex and traditional CFD trading'
},
{
'name': 'Binance',
'type': 'Crypto Exchange',
'assets': ['BTCUSDT', 'ETHUSDT', 'ADAUSDT', 'SOLUSDT', 'DOGEUSDT'],
'advantages': ['24/7 trading', 'High liquidity', 'Low fees'],
'best_for': 'Cryptocurrency trading and DeFi'
},
{
'name': 'cTrader',
'type': 'Modern Forex Platform',
'assets': ['EURUSD', 'GBPUSD', 'USDJPY', 'XAUUSD', 'USOIL'],
'advantages': ['Advanced charting', 'Level II pricing', 'Fast execution'],
'best_for': 'Professional forex trading'
},
{
'name': 'Interactive Brokers',
'type': 'Multi-Asset Broker',
'assets': ['AAPL', 'ES', 'EURUSD', 'GC', 'Options'],
'advantages': ['Global markets', 'Low commissions', 'Advanced tools'],
'best_for': 'Stocks, futures, and options'
},
{
'name': 'TradingView',
'type': 'Social Trading Platform',
'assets': ['All markets', 'Pine Script', 'Social signals'],
'advantages': ['Community strategies', 'Advanced charts', 'Alerts'],
'best_for': 'Strategy development and social trading'
}
]
print("\\n🏢 Broker Overview:")
print("=" * 60)
for i, broker in enumerate(brokers_info, 1):
print(f"\\n{i}. {broker['name']} ({broker['type']})")
print(f" 📈 Assets: {', '.join(broker['assets'][:3])}{'...' if len(broker['assets']) > 3 else ''}")
print(f" ⭐ Best For: {broker['best_for']}")
print(f" 🎯 Key Advantages: {', '.join(broker['advantages'][:2])}")
return brokers_info
def demo_unified_portfolio():
"""Show how to create a unified portfolio across all brokers"""
print("\\n💼 Unified Portfolio Management")
print("=" * 60)
portfolio_allocation = {
'MT5 (Forex)': {
'allocation': '30%',
'symbols': ['EURUSD', 'GBPUSD', 'USDJPY'],
'strategy': 'QuantumBotX Hybrid',
'capital': '$3,000'
},
'Binance (Crypto)': {
'allocation': '25%',
'symbols': ['BTCUSDT', 'ETHUSDT', 'ADAUSDT'],
'strategy': 'MA Crossover (Crypto-tuned)',
'capital': '$2,500'
},
'cTrader (Forex Pro)': {
'allocation': '20%',
'symbols': ['XAUUSD', 'USOIL'],
'strategy': 'Bollinger Reversion',
'capital': '$2,000'
},
'Interactive Brokers (Stocks)': {
'allocation': '20%',
'symbols': ['AAPL', 'MSFT', 'TSLA'],
'strategy': 'Quantum Velocity',
'capital': '$2,000'
},
'TradingView (Signals)': {
'allocation': '5%',
'symbols': ['Community strategies'],
'strategy': 'Pine Script alerts',
'capital': '$500'
}
}
print("\\n📊 Portfolio Distribution ($10,000 total):")
print("-" * 60)
total_expected_return = 0
for broker, details in portfolio_allocation.items():
print(f"\\n{broker}")
print(f" 💰 Capital: {details['capital']} ({details['allocation']})")
print(f" 📈 Assets: {', '.join(details['symbols'][:3])}")
print(f" 🤖 Strategy: {details['strategy']}")
# Simulate expected returns
expected_monthly = np.random.uniform(2, 8) # 2-8% monthly return
total_expected_return += expected_monthly * float(details['allocation'].strip('%')) / 100
print(f" 📊 Expected Monthly Return: {expected_monthly:.1f}%")
print(f"\\n🎯 Portfolio Expected Monthly Return: {total_expected_return:.1f}%")
print(f"🎯 Portfolio Expected Annual Return: {total_expected_return * 12:.1f}%")
def demo_risk_management():
"""Show unified risk management across all brokers"""
print("\\n🛡️ Unified Risk Management System")
print("=" * 60)
risk_rules = [
{
'rule': 'Maximum Portfolio Risk',
'value': '15% of total capital',
'implementation': 'Sum of all open positions across all brokers'
},
{
'rule': 'Per-Broker Risk Limit',
'value': '5% per broker maximum',
'implementation': 'Individual broker position sizing limits'
},
{
'rule': 'Correlation Protection',
'value': 'Max 3 correlated positions',
'implementation': 'Cross-broker correlation monitoring'
},
{
'rule': 'Volatility Scaling',
'value': 'Dynamic position sizing',
'implementation': 'ATR-based sizing per asset class'
},
{
'rule': 'Emergency Brake',
'value': 'Auto-stop at 10% daily loss',
'implementation': 'Real-time P&L monitoring across all accounts'
}
]
print("\\n🔒 Global Risk Rules:")
for i, rule in enumerate(risk_rules, 1):
print(f"\\n{i}. {rule['rule']}: {rule['value']}")
print(f" Implementation: {rule['implementation']}")
def demo_24_7_opportunities():
"""Show 24/7 trading opportunities"""
print("\\n⏰ 24/7 Global Trading Opportunities")
print("=" * 60)
trading_schedule = [
{'time': '00:00-08:00 UTC', 'active': ['Crypto (Binance)', 'Forex (Asian session)'], 'opportunity': 'Crypto volatility + Asian forex'},
{'time': '08:00-16:00 UTC', 'active': ['All Forex', 'European Stocks', 'Crypto'], 'opportunity': 'European session overlap'},
{'time': '13:00-17:00 UTC', 'active': ['US Stocks (IB)', 'US/EU Forex overlap', 'Crypto'], 'opportunity': 'Maximum liquidity window'},
{'time': '17:00-00:00 UTC', 'active': ['Crypto (Binance)', 'Asian prep', 'After-hours'], 'opportunity': 'Crypto focus + overnight gaps'}
]
print("\\n🌍 Global Trading Sessions:")
for session in trading_schedule:
print(f"\\n⏰ {session['time']}")
print(f" 🎯 Active: {', '.join(session['active'])}")
print(f" 💡 Opportunity: {session['opportunity']}")
print("\\n🔥 Never Miss a Move:")
print(" • Forex: 24/5 traditional markets")
print(" • Crypto: 24/7/365 never stops")
print(" • Stocks: Pre/post market + global exchanges")
print(" • Commodities: Global futures markets")
def demo_integration_benefits():
"""Show the benefits of integrated multi-broker system"""
print("\\n🚀 Integration Benefits")
print("=" * 60)
benefits = [
{
'category': 'Market Coverage',
'benefits': [
'Trade forex, crypto, stocks, and commodities',
'Access to global markets 24/7',
'Never limited by single broker restrictions'
]
},
{
'category': 'Risk Diversification',
'benefits': [
'Spread risk across multiple platforms',
'Reduce broker-specific risks',
'Currency and asset class diversification'
]
},
{
'category': 'Strategy Optimization',
'benefits': [
'Different strategies for different markets',
'Platform-specific advantages utilization',
'Cross-market arbitrage opportunities'
]
},
{
'category': 'Operational Excellence',
'benefits': [
'Single dashboard for all trading',
'Unified risk management',
'Consolidated reporting and analytics'
]
}
]
for benefit_group in benefits:
print(f"\\n📈 {benefit_group['category']}:")
for benefit in benefit_group['benefits']:
print(f" ✅ {benefit}")
def main():
"""Main demo function"""
print("🎉 Welcome to the Financial Universe!")
print("Your QuantumBotX system now connects to EVERYTHING!")
print()
# Demo all components
brokers_info = demo_all_brokers()
demo_unified_portfolio()
demo_risk_management()
demo_24_7_opportunities()
demo_integration_benefits()
print("\\n" + "=" * 60)
print("🎯 IMPLEMENTATION ROADMAP")
print("=" * 60)
roadmap = [
{
'phase': 'Week 1: Crypto Integration',
'tasks': ['Set up Binance testnet', 'Test crypto strategies', 'Validate risk management'],
'impact': 'Add 24/7 trading capability'
},
{
'phase': 'Week 2: cTrader Setup',
'tasks': ['Create cTrader demo account', 'Test modern forex features', 'Compare with MT5'],
'impact': 'Enhanced forex trading experience'
},
{
'phase': 'Week 3: Interactive Brokers',
'tasks': ['Set up TWS paper trading', 'Test stock strategies', 'Explore futures'],
'impact': 'Access to US stocks and global markets'
},
{
'phase': 'Week 4: TradingView Integration',
'tasks': ['Set up webhook alerts', 'Create Pine Script strategies', 'Social trading'],
'impact': 'Community-driven strategy development'
},
{
'phase': 'Month 2: Unified Platform',
'tasks': ['Portfolio manager', 'Cross-broker risk management', 'Performance analytics'],
'impact': 'Complete multi-broker trading ecosystem'
}
]
for i, phase in enumerate(roadmap, 1):
print(f"\\n{i}. {phase['phase']}")
print(f" 📋 Tasks: {', '.join(phase['tasks'][:2])}...")
print(f" 🎯 Impact: {phase['impact']}")
print("\\n" + "=" * 60)
print("🏆 THE BIG PICTURE")
print("=" * 60)
print("\\n🌟 What You're Building:")
print(" • Universal Trading Platform - One system, all markets")
print(" • Risk-Managed Portfolio - Diversified across asset classes")
print(" • 24/7 Profit Machine - Never miss opportunities")
print(" • Future-Proof Architecture - Ready for any new broker")
print("\\n💰 Potential Impact:")
current_profit = 4649.94
projected_increase = 2.5 # Conservative 2.5x increase
projected_profit = current_profit * projected_increase
print(f" Current Demo Profit: ${current_profit:,.2f}")
print(f" With Multi-Broker: ${projected_profit:,.2f} (estimated)")
print(f" Improvement Factor: {projected_increase}x")
print("\\n🎉 Congratulations!")
print("You've just designed a trading system that rivals")
print("what hedge funds and prop trading firms use!")
print("\\nFrom learning to trade → Building a financial empire! 🚀")
if __name__ == "__main__":
main()
testing/test_ctrader_broker.py
-59
View File
@@ -1,59 +0,0 @@
# testing/test_ctrader_broker.py
import unittest
from unittest.mock import patch
from datetime import datetime
from core.brokers.ctrader_broker import CTraderBroker
class TestCTraderBroker(unittest.TestCase):
"""
Test cases for the cTrader broker implementation, focusing on market hours.
"""
def setUp(self):
"""Set up a CTraderBroker instance for testing."""
self.broker = CTraderBroker(demo=True)
@patch('core.brokers.ctrader_broker.datetime')
def test_is_market_open_weekday(self, mock_datetime):
"""Test that the market is open on a standard weekday."""
# Wednesday, 12:00 UTC
mock_datetime.utcnow.return_value = datetime(2023, 1, 4, 12, 0, 0)
self.assertTrue(self.broker.is_market_open())
@patch('core.brokers.ctrader_broker.datetime')
def test_is_market_closed_saturday(self, mock_datetime):
"""Test that the market is closed on Saturday."""
# Saturday, 12:00 UTC
mock_datetime.utcnow.return_value = datetime(2023, 1, 7, 12, 0, 0)
self.assertFalse(self.broker.is_market_open())
@patch('core.brokers.ctrader_broker.datetime')
def test_is_market_opens_sunday_evening(self, mock_datetime):
"""Test that the market opens on Sunday evening."""
# Sunday, 22:01 UTC (market is open)
mock_datetime.utcnow.return_value = datetime(2023, 1, 8, 22, 1, 0)
self.assertTrue(self.broker.is_market_open())
@patch('core.brokers.ctrader_broker.datetime')
def test_is_market_closed_sunday_morning(self, mock_datetime):
"""Test that the market is closed on Sunday morning."""
# Sunday, 10:00 UTC (market is closed)
mock_datetime.utcnow.return_value = datetime(2023, 1, 8, 10, 0, 0)
self.assertFalse(self.broker.is_market_open())
@patch('core.brokers.ctrader_broker.datetime')
def test_is_market_closes_friday_evening(self, mock_datetime):
"""Test that the market closes on Friday evening."""
# Friday, 22:01 UTC (market is closed)
mock_datetime.utcnow.return_value = datetime(2023, 1, 6, 22, 1, 0)
self.assertFalse(self.broker.is_market_open())
@patch('core.brokers.ctrader_broker.datetime')
def test_is_market_open_friday_morning(self, mock_datetime):
"""Test that the market is open on Friday morning."""
# Friday, 10:00 UTC (market is open)
mock_datetime.utcnow.return_value = datetime(2023, 1, 6, 10, 0, 0)
self.assertTrue(self.broker.is_market_open())
if __name__ == '__main__':
unittest.main()