Xây Dựng Bot Giao Dịch Tự Động với Python cho Forex, Chứng Khoán và Tiền Điện Tử
2024-01-10 — Admin
Giới thiệu
Giao dịch tự động là một trong những ứng dụng nổi bật nhất của lập trình trong lĩnh vực tài chính. Các bot giao dịch có thể giúp loại bỏ yếu tố cảm xúc, thực hiện giao dịch 24/7 và thực thi các chiến lược phức tạp một cách chính xác. Với Python và các thư viện phong phú của nó, việc xây dựng một bot giao dịch tự động đã trở nên dễ dàng hơn rất nhiều.
Trong bài viết này, chúng ta sẽ khám phá cách tạo bot giao dịch cho ba thị trường phổ biến: Forex (ngoại hối), chứng khoán và tiền điện tử.
Phần 1: Cấu trúc cơ bản của một Bot Giao dịch
Một bot giao dịch hiệu quả thường bao gồm bốn thành phần chính:
- Thu thập dữ liệu: Kết nối với các API hoặc nguồn dữ liệu để lấy thông tin thị trường.
- Phân tích và tạo tín hiệu: Xử lý dữ liệu để tạo ra các tín hiệu giao dịch.
- Ra quyết định: Áp dụng chiến lược giao dịch và quản lý rủi ro.
- Thực thi giao dịch: Kết nối với các sàn giao dịch để đặt lệnh.
Hãy bắt đầu bằng việc thiết lập môi trường Python và các thư viện cần thiết:
# Thư viện cần thiết để xây dựng bot giao dịch
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import datetime
import time
import requests
import json
# Thư viện cho giao dịch và phân tích
import backtrader as bt # Thư viện backtesting và giao dịch
import ccxt # Thư viện kết nối với sàn giao dịch tiền điện tử
import MetaTrader5 as mt5 # Kết nối với MT5 cho Forex
import alpaca_trade_api as tradeapi # API cho giao dịch chứng khoán
Phần 2: Bot Giao dịch Forex
Forex (Foreign Exchange) là thị trường ngoại hối, nơi các loại tiền tệ được giao dịch. Đây là thị trường lớn nhất và thanh khoản nhất trên thế giới.
Kết nối với MetaTrader 5 (MT5)
MetaTrader 5 là một trong những nền tảng giao dịch Forex phổ biến nhất. Python có thể kết nối với MT5 thông qua gói MetaTrader5:
import MetaTrader5 as mt5
import pandas as pd
import numpy as np
from datetime import datetime
import time
# Khởi tạo kết nối với MT5
def initialize_mt5():
if not mt5.initialize():
print("Khởi tạo MT5 thất bại")
mt5.shutdown()
return False
return True
# Đăng nhập vào tài khoản
def login_mt5(account, password, server):
if not mt5.login(account, password, server):
print("Đăng nhập thất bại")
mt5.shutdown()
return False
return True
# Lấy dữ liệu lịch sử
def get_historical_data(symbol, timeframe, start_date, end_date=None):
if end_date is None:
end_date = datetime.now()
# Chuyển đổi khung thời gian
timeframe_dict = {
"M1": mt5.TIMEFRAME_M1,
"M5": mt5.TIMEFRAME_M5,
"M15": mt5.TIMEFRAME_M15,
"M30": mt5.TIMEFRAME_M30,
"H1": mt5.TIMEFRAME_H1,
"H4": mt5.TIMEFRAME_H4,
"D1": mt5.TIMEFRAME_D1,
"W1": mt5.TIMEFRAME_W1,
"MN1": mt5.TIMEFRAME_MN1
}
# Lấy dữ liệu
rates = mt5.copy_rates_range(symbol, timeframe_dict[timeframe], start_date, end_date)
# Chuyển đổi sang DataFrame
df = pd.DataFrame(rates)
df['time'] = pd.to_datetime(df['time'], unit='s')
df.set_index('time', inplace=True)
return df
# Đặt lệnh
def place_order(symbol, order_type, volume, price=0, stop_loss=0, take_profit=0, comment="Python Order"):
if order_type == "BUY":
order_type = mt5.ORDER_TYPE_BUY
price = mt5.symbol_info_tick(symbol).ask
elif order_type == "SELL":
order_type = mt5.ORDER_TYPE_SELL
price = mt5.symbol_info_tick(symbol).bid
request = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": symbol,
"volume": float(volume),
"type": order_type,
"price": price,
"sl": stop_loss,
"tp": take_profit,
"comment": comment,
"type_time": mt5.ORDER_TIME_GTC,
"type_filling": mt5.ORDER_FILLING_IOC,
}
result = mt5.order_send(request)
return result
# Đóng tất cả các vị thế
def close_all_positions():
positions = mt5.positions_get()
for position in positions:
order_type = mt5.ORDER_TYPE_SELL if position.type == 0 else mt5.ORDER_TYPE_BUY
price = mt5.symbol_info_tick(position.symbol).bid if position.type == 0 else mt5.symbol_info_tick(position.symbol).ask
request = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": position.symbol,
"volume": position.volume,
"type": order_type,
"position": position.ticket,
"price": price,
"comment": "Close position",
"type_time": mt5.ORDER_TIME_GTC,
"type_filling": mt5.ORDER_FILLING_IOC,
}
mt5.order_send(request)
# Ví dụ về chiến lược Moving Average đơn giản
def ma_crossover_strategy(symbol, fast_period=20, slow_period=50, timeframe="H1"):
# Lấy dữ liệu gần đây
end_date = datetime.now()
start_date = end_date - datetime.timedelta(days=100)
df = get_historical_data(symbol, timeframe, start_date, end_date)
# Tính toán các đường trung bình động
df['MA_fast'] = df['close'].rolling(window=fast_period).mean()
df['MA_slow'] = df['close'].rolling(window=slow_period).mean()
# Kiểm tra tín hiệu giao cắt
df['signal'] = 0
df.loc[df['MA_fast'] > df['MA_slow'], 'signal'] = 1
df.loc[df['MA_fast'] < df['MA_slow'], 'signal'] = -1
# Tạo tín hiệu giao dịch
df['position'] = df['signal'].diff()
# Lấy tín hiệu gần nhất
last_position = df['position'].iloc[-1]
current_signal = df['signal'].iloc[-1]
# Thực hiện giao dịch dựa trên tín hiệu
if last_position == 2: # Chuyển từ SELL sang BUY
close_all_positions()
place_order(symbol, "BUY", 0.01)
print(f"BUY {symbol} at {datetime.now()}")
elif last_position == -2: # Chuyển từ BUY sang SELL
close_all_positions()
place_order(symbol, "SELL", 0.01)
print(f"SELL {symbol} at {datetime.now()}")
return df
# Chạy bot
def run_forex_bot(symbols, interval=3600): # 1 giờ
initialize_mt5()
login_mt5(12345, "password", "Broker-Server")
while True:
for symbol in symbols:
try:
df = ma_crossover_strategy(symbol)
print(f"Kiểm tra tín hiệu cho {symbol} lúc {datetime.now()}")
except Exception as e:
print(f"Lỗi: {str(e)}")
time.sleep(interval)
# Ví dụ sử dụng
if __name__ == "__main__":
symbols = ["EURUSD", "USDJPY", "GBPUSD"]
run_forex_bot(symbols)
Phần 3: Bot Giao dịch Chứng Khoán
Thị trường chứng khoán có những đặc thù riêng, như thời gian giao dịch giới hạn và quy định khác nhau. Để giao dịch chứng khoán, chúng ta có thể sử dụng API của các nhà môi giới như Alpaca.
import alpaca_trade_api as tradeapi
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import time
# Thiết lập kết nối với Alpaca
API_KEY = "YOUR_API_KEY"
API_SECRET = "YOUR_API_SECRET"
BASE_URL = "https://paper-api.alpaca.markets" # URL cho tài khoản demo
api = tradeapi.REST(API_KEY, API_SECRET, BASE_URL, api_version='v2')
# Lấy dữ liệu lịch sử
def get_stock_data(symbol, timeframe, limit=100):
# Lấy dữ liệu theo khung thời gian
if timeframe == "1Min":
barset = api.get_barset(symbol, timeframe, limit=limit)
elif timeframe == "1D":
barset = api.get_barset(symbol, timeframe, limit=limit)
# Chuyển đổi sang DataFrame
df = pd.DataFrame()
for bar in barset[symbol]:
df = df.append({
'time': bar.t,
'open': bar.o,
'high': bar.h,
'low': bar.l,
'close': bar.c,
'volume': bar.v
}, ignore_index=True)
df['time'] = pd.to_datetime(df['time'])
df.set_index('time', inplace=True)
return df
# Chiến lược MACD đơn giản
def macd_strategy(symbol, fast_period=12, slow_period=26, signal_period=9):
# Lấy dữ liệu
df = get_stock_data(symbol, "1D", 200)
# Tính MACD
df['EMA_fast'] = df['close'].ewm(span=fast_period).mean()
df['EMA_slow'] = df['close'].ewm(span=slow_period).mean()
df['MACD'] = df['EMA_fast'] - df['EMA_slow']
df['MACD_signal'] = df['MACD'].ewm(span=signal_period).mean()
df['MACD_histogram'] = df['MACD'] - df['MACD_signal']
# Tạo tín hiệu giao dịch
df['signal'] = 0
df.loc[df['MACD'] > df['MACD_signal'], 'signal'] = 1
df.loc[df['MACD'] < df['MACD_signal'], 'signal'] = -1
df['position'] = df['signal'].diff()
# Lấy tín hiệu gần nhất
last_position = df['position'].iloc[-1]
return df, last_position
# Đặt lệnh
def place_stock_order(symbol, side, qty, order_type="market", time_in_force="day"):
try:
order = api.submit_order(
symbol=symbol,
qty=qty,
side=side,
type=order_type,
time_in_force=time_in_force
)
return order
except Exception as e:
print(f"Lỗi khi đặt lệnh: {str(e)}")
return None
# Kiểm tra và đóng tất cả vị thế
def close_all_stock_positions():
try:
api.close_all_positions()
print("Đã đóng tất cả vị thế")
except Exception as e:
print(f"Lỗi khi đóng vị thế: {str(e)}")
# Chạy bot
def run_stock_bot(symbols, qty=1, interval=3600): # 1 giờ
while True:
try:
# Kiểm tra xem thị trường có mở cửa không
clock = api.get_clock()
if clock.is_open:
for symbol in symbols:
print(f"Đang kiểm tra tín hiệu cho {symbol}...")
df, last_position = macd_strategy(symbol)
# Thực hiện giao dịch dựa trên tín hiệu
if last_position == 2: # Tín hiệu mua
print(f"Tín hiệu MUA cho {symbol}")
place_stock_order(symbol, "buy", qty)
elif last_position == -2: # Tín hiệu bán
print(f"Tín hiệu BÁN cho {symbol}")
place_stock_order(symbol, "sell", qty)
else:
next_open = clock.next_open.strftime("%Y-%m-%d %H:%M")
print(f"Thị trường đóng cửa. Mở cửa tiếp theo vào: {next_open}")
except Exception as e:
print(f"Lỗi: {str(e)}")
time.sleep(interval)
# Ví dụ sử dụng
if __name__ == "__main__":
symbols = ["AAPL", "MSFT", "GOOGL"]
run_stock_bot(symbols)
Phần 4: Bot Giao dịch Tiền Điện Tử
Thị trường tiền điện tử hoạt động 24/7 và cung cấp nhiều API khác nhau. Chúng ta sẽ sử dụng thư viện CCXT, cho phép kết nối với hơn 100 sàn giao dịch tiền điện tử.
import ccxt
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime
import time
# Khởi tạo kết nối với sàn giao dịch
def initialize_exchange(exchange_id, api_key=None, secret=None):
exchange_class = getattr(ccxt, exchange_id)
exchange = exchange_class({
'apiKey': api_key,
'secret': secret,
'enableRateLimit': True,
})
return exchange
# Lấy dữ liệu lịch sử
def get_crypto_data(exchange, symbol, timeframe, limit=100):
try:
# Lấy OHLCV (Open, High, Low, Close, Volume)
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
# Chuyển đổi sang DataFrame
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
return df
except Exception as e:
print(f"Lỗi khi lấy dữ liệu: {str(e)}")
return None
# Chiến lược RSI (Relative Strength Index)
def rsi_strategy(df, rsi_period=14, overbought=70, oversold=30):
# Tính toán RSI
delta = df['close'].diff()
gain = delta.where(delta > 0, 0)
loss = -delta.where(delta < 0, 0)
avg_gain = gain.rolling(window=rsi_period).mean()
avg_loss = loss.rolling(window=rsi_period).mean()
rs = avg_gain / avg_loss
df['RSI'] = 100 - (100 / (1 + rs))
# Tạo tín hiệu
df['signal'] = 0
df.loc[df['RSI'] < oversold, 'signal'] = 1 # Tín hiệu mua khi RSI < 30
df.loc[df['RSI'] > overbought, 'signal'] = -1 # Tín hiệu bán khi RSI > 70
# Chỉ giao dịch khi có sự thay đổi tín hiệu
df['position'] = df['signal'].diff()
return df
# Đặt lệnh
def place_crypto_order(exchange, symbol, order_type, amount):
try:
if order_type == "buy":
order = exchange.create_market_buy_order(symbol, amount)
elif order_type == "sell":
order = exchange.create_market_sell_order(symbol, amount)
print(f"Đặt lệnh {order_type} thành công: {order}")
return order
except Exception as e:
print(f"Lỗi khi đặt lệnh: {str(e)}")
return None
# Chạy bot
def run_crypto_bot(exchange_id, symbols, amount, timeframe='1h', interval=300): # 5 phút
# Khởi tạo sàn giao dịch
exchange = initialize_exchange(exchange_id, "YOUR_API_KEY", "YOUR_SECRET_KEY")
while True:
for symbol in symbols:
try:
# Lấy dữ liệu và phân tích
df = get_crypto_data(exchange, symbol, timeframe, 100)
if df is not None:
df = rsi_strategy(df)
# Lấy vị thế hiện tại
current_position = 0
try:
positions = exchange.fetch_balance()
base_currency = symbol.split('/')[0]
if base_currency in positions:
current_position = positions[base_currency]['free']
except:
pass
# Lấy tín hiệu gần nhất
last_position = df['position'].iloc[-1]
# Thực hiện giao dịch
if last_position == 1: # Tín hiệu mua mới
print(f"Tín hiệu MUA cho {symbol}")
place_crypto_order(exchange, symbol, "buy", amount)
elif last_position == -1: # Tín hiệu bán mới
print(f"Tín hiệu BÁN cho {symbol}")
if current_position > 0:
place_crypto_order(exchange, symbol, "sell", current_position)
except Exception as e:
print(f"Lỗi khi xử lý {symbol}: {str(e)}")
time.sleep(interval)
# Ví dụ sử dụng
if __name__ == "__main__":
exchange_id = 'binance' # Có thể thay đổi thành 'coinbase', 'kraken', v.v.
symbols = ["BTC/USDT", "ETH/USDT", "SOL/USDT"]
amount = 0.001 # Số lượng giao dịch
run_crypto_bot(exchange_id, symbols, amount)
Phần 5: Cải thiện Bot Giao dịch với Machine Learning
Các chiến lược truyền thống như Moving Average, MACD, RSI có thể được cải thiện bằng cách áp dụng Machine Learning để dự đoán xu hướng giá.
import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score
import ccxt
import time
# Lấy dữ liệu và thêm các chỉ báo kỹ thuật
def prepare_data(exchange, symbol, timeframe, limit=1000):
# Lấy dữ liệu
df = get_crypto_data(exchange, symbol, timeframe, limit)
if df is None:
return None
# Thêm các chỉ báo
# 1. SMA - Simple Moving Average
df['SMA20'] = df['close'].rolling(window=20).mean()
df['SMA50'] = df['close'].rolling(window=50).mean()
# 2. EMA - Exponential Moving Average
df['EMA12'] = df['close'].ewm(span=12, adjust=False).mean()
df['EMA26'] = df['close'].ewm(span=26, adjust=False).mean()
# 3. MACD
df['MACD'] = df['EMA12'] - df['EMA26']
df['MACD_signal'] = df['MACD'].ewm(span=9, adjust=False).mean()
df['MACD_hist'] = df['MACD'] - df['MACD_signal']
# 4. RSI
delta = df['close'].diff()
gain = delta.where(delta > 0, 0)
loss = -delta.where(delta < 0, 0)
avg_gain = gain.rolling(window=14).mean()
avg_loss = loss.rolling(window=14).mean()
rs = avg_gain / avg_loss
df['RSI'] = 100 - (100 / (1 + rs))
# 5. Bollinger Bands
df['BB_middle'] = df['close'].rolling(window=20).mean()
std = df['close'].rolling(window=20).std()
df['BB_upper'] = df['BB_middle'] + (std * 2)
df['BB_lower'] = df['BB_middle'] - (std * 2)
# Tạo cột target - Tăng (1) hoặc Giảm (0)
df['target'] = np.where(df['close'].shift(-1) > df['close'], 1, 0)
# Loại bỏ các dòng có giá trị NaN
df.dropna(inplace=True)
return df
# Huấn luyện mô hình
def train_model(df):
# Các tính năng
features = ['open', 'high', 'low', 'close', 'volume',
'SMA20', 'SMA50', 'EMA12', 'EMA26',
'MACD', 'MACD_signal', 'MACD_hist', 'RSI',
'BB_middle', 'BB_upper', 'BB_lower']
# Chuẩn bị dữ liệu
X = df[features]
y = df['target']
# Chia tập dữ liệu
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
# Huấn luyện mô hình
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)
# Đánh giá mô hình
predictions = model.predict(X_test)
accuracy = accuracy_score(y_test, predictions)
print(f"Độ chính xác: {accuracy:.2f}")
return model, features
# Bot giao dịch với ML
def run_ml_crypto_bot(exchange_id, symbols, amount, timeframe='1h', interval=300):
# Khởi tạo sàn giao dịch
exchange = initialize_exchange(exchange_id, "YOUR_API_KEY", "YOUR_SECRET_KEY")
# Từ điển để lưu trữ mô hình cho mỗi cặp giao dịch
models = {}
feature_lists = {}
# Huấn luyện mô hình cho mỗi cặp giao dịch
for symbol in symbols:
df = prepare_data(exchange, symbol, timeframe)
if df is not None:
model, features = train_model(df)
models[symbol] = model
feature_lists[symbol] = features
# Vòng lặp chính
while True:
for symbol in symbols:
try:
if symbol in models:
# Lấy dữ liệu mới nhất
df = prepare_data(exchange, symbol, timeframe, 100)
if df is not None:
# Lấy dữ liệu gần nhất để dự đoán
latest_data = df.iloc[-1][feature_lists[symbol]].values.reshape(1, -1)
# Dự đoán
prediction = models[symbol].predict(latest_data)[0]
prediction_proba = models[symbol].predict_proba(latest_data)[0]
# Hành động dựa trên dự đoán
if prediction == 1 and prediction_proba[1] > 0.7: # Dự đoán tăng với độ tin cậy > 70%
print(f"Tín hiệu MUA cho {symbol} (tin cậy: {prediction_proba[1]:.2f})")
place_crypto_order(exchange, symbol, "buy", amount)
elif prediction == 0 and prediction_proba[0] > 0.7: # Dự đoán giảm với độ tin cậy > 70%
print(f"Tín hiệu BÁN cho {symbol} (tin cậy: {prediction_proba[0]:.2f})")
# Kiểm tra số dư
try:
balance = exchange.fetch_balance()
base_currency = symbol.split('/')[0]
if base_currency in balance and balance[base_currency]['free'] > 0:
place_crypto_order(exchange, symbol, "sell", balance[base_currency]['free'])
except Exception as e:
print(f"Lỗi khi kiểm tra số dư: {str(e)}")
except Exception as e:
print(f"Lỗi khi xử lý {symbol}: {str(e)}")
time.sleep(interval)
# Ví dụ sử dụng
if __name__ == "__main__":
exchange_id = 'binance'
symbols = ["BTC/USDT", "ETH/USDT"]
amount = 0.001
run_ml_crypto_bot(exchange_id, symbols, amount)
Phần 6: Quản lý rủi ro và tối ưu hóa
Quản lý rủi ro là một phần quan trọng trong giao dịch tự động. Dưới đây là một số kỹ thuật để bảo vệ vốn:
# Hàm quản lý rủi ro
def risk_management(exchange, symbol, risk_percent=2, stop_loss_percent=2, take_profit_percent=4):
try:
# Lấy số dư tài khoản
balance = exchange.fetch_balance()
quote_currency = symbol.split('/')[1] # USDT trong BTC/USDT
available_balance = balance[quote_currency]['free']
# Tính kích thước vị thế dựa trên % rủi ro
risk_amount = available_balance * (risk_percent / 100)
# Lấy giá hiện tại
ticker = exchange.fetch_ticker(symbol)
current_price = ticker['last']
# Tính kích thước lệnh
position_size = risk_amount / current_price
# Tính stop loss và take profit
stop_loss = current_price * (1 - stop_loss_percent / 100)
take_profit = current_price * (1 + take_profit_percent / 100)
return {
'position_size': position_size,
'stop_loss': stop_loss,
'take_profit': take_profit
}
except Exception as e:
print(f"Lỗi khi tính toán quản lý rủi ro: {str(e)}")
return None
# Theo dõi và điều chỉnh stop loss theo giá di chuyển (trailing stop)
def trailing_stop(exchange, symbol, order_id, trail_percent=1):
try:
# Lấy thông tin lệnh
order = exchange.fetch_order(order_id)
if order['status'] == 'closed':
entry_price = order['price']
while True:
# Lấy giá hiện tại
ticker = exchange.fetch_ticker(symbol)
current_price = ticker['last']
# Tính toán trailing stop
if order['side'] == 'buy': # Lệnh mua
stop_price = current_price * (1 - trail_percent / 100)
# Đặt lệnh stop loss
if stop_price > entry_price:
exchange.create_order(
symbol=symbol,
type='stop_loss',
side='sell',
amount=order['amount'],
price=stop_price
)
print(f"Đặt trailing stop tại {stop_price}")
time.sleep(60) # Kiểm tra mỗi phút
except Exception as e:
print(f"Lỗi khi thiết lập trailing stop: {str(e)}")
Phần 7: Backtesting với Backtrader
Trước khi triển khai một chiến lược giao dịch, bạn nên kiểm tra lại hiệu suất của nó trên dữ liệu lịch sử:
import backtrader as bt
import datetime
import pandas as pd
# Định nghĩa chiến lược
class MACrossStrategy(bt.Strategy):
params = (
('fast_period', 10),
('slow_period', 30),
)
def __init__(self):
# Tính toán các đường trung bình động
self.fast_ma = bt.indicators.SMA(self.data.close, period=self.params.fast_period)
self.slow_ma = bt.indicators.SMA(self.data.close, period=self.params.slow_period)
# Tín hiệu giao cắt
self.crossover = bt.indicators.CrossOver(self.fast_ma, self.slow_ma)
# Theo dõi lệnh, vị thế và giá mua/bán
self.order = None
self.buyprice = None
self.buycomm = None
def next(self):
# Kiểm tra nếu đang có lệnh đang chờ
if self.order:
return
# Kiểm tra nếu đang có vị thế
if not self.position:
# Mua nếu MA nhanh cắt lên trên MA chậm
if self.crossover > 0:
self.log(f'BUY CREATE, {self.data.close[0]}')
self.order = self.buy()
else:
# Bán nếu MA nhanh cắt xuống dưới MA chậm
if self.crossover < 0:
self.log(f'SELL CREATE, {self.data.close[0]}')
self.order = self.sell()
def log(self, txt, dt=None):
dt = dt or self.datas[0].datetime.date(0)
print(f'{dt.isoformat()}, {txt}')
def notify_order(self, order):
if order.status in [order.Submitted, order.Accepted]:
return
if order.status in [order.Completed]:
if order.isbuy():
self.log(f'BUY EXECUTED, Price: {order.executed.price}, Cost: {order.executed.value}, Comm: {order.executed.comm}')
self.buyprice = order.executed.price
self.buycomm = order.executed.comm
else:
self.log(f'SELL EXECUTED, Price: {order.executed.price}, Cost: {order.executed.value}, Comm: {order.executed.comm}')
self.bar_executed = len(self)
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log('Order Canceled/Margin/Rejected')
self.order = None
def notify_trade(self, trade):
if not trade.isclosed:
return
self.log(f'OPERATION PROFIT, GROSS: {trade.pnl}, NET: {trade.pnlcomm}')
# Hàm chạy backtest
def run_backtest(data_feed, strategy_class, **strategy_params):
# Tạo cerebro
cerebro = bt.Cerebro()
# Thêm dữ liệu
cerebro.adddata(data_feed)
# Thêm chiến lược
cerebro.addstrategy(strategy_class, **strategy_params)
# Thiết lập vốn ban đầu
cerebro.broker.setcash(100000.0)
# Thiết lập phí giao dịch (0.1%)
cerebro.broker.setcommission(commission=0.001)
# Thêm phân tích
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
# In giá trị ban đầu
print(f'Vốn ban đầu: {cerebro.broker.getvalue():.2f}')
# Chạy backtest
results = cerebro.run()
strat = results[0]
# In kết quả
print(f'Vốn sau cùng: {cerebro.broker.getvalue():.2f}')
print(f'Lợi nhuận: {cerebro.broker.getvalue() - 100000:.2f}')
print(f'Sharpe Ratio: {strat.analyzers.sharpe.get_analysis()["sharperatio"]:.3f}')
print(f'Max Drawdown: {strat.analyzers.drawdown.get_analysis()["max"]["drawdown"]:.2f}%')
print(f'Lợi nhuận hàng năm: {strat.analyzers.returns.get_analysis()["rtot"]:.2f}%')
# Vẽ đồ thị
cerebro.plot()
# Ví dụ sử dụng
if __name__ == "__main__":
# Lấy dữ liệu từ file CSV hoặc API
exchange = initialize_exchange('binance')
df = get_crypto_data(exchange, 'BTC/USDT', '1d', 500)
# Chuyển đổi DataFrame sang định dạng Backtrader
data = bt.feeds.PandasData(
dataname=df,
datetime=None, # Đã được đặt làm index
open=0,
high=1,
low=2,
close=3,
volume=4,
openinterest=-1
)
# Chạy backtest
run_backtest(data, MACrossStrategy, fast_period=10, slow_period=30)
Kết luận
Xây dựng một bot giao dịch tự động với Python là một quá trình đầy thử thách nhưng cũng rất thú vị. Bot có thể giúp bạn thực hiện giao dịch không cảm xúc và tuân theo chiến lược đã định sẵn.
Tuy nhiên, cần lưu ý rằng:
- Kiểm tra kỹ trước khi triển khai: Luôn backtesting chiến lược trên dữ liệu lịch sử và thử nghiệm trên tài khoản demo trước.
- Quản lý rủi ro: Không bao giờ đầu tư số tiền bạn không sẵn sàng mất và luôn sử dụng stop loss.
- Tuân thủ quy định: Đảm bảo bot của bạn tuân thủ các quy định pháp lý về giao dịch tự động.
- Giám sát liên tục: Ngay cả bot tự động cũng cần được giám sát và điều chỉnh để thích ứng với thay đổi của thị trường.
Với sự phát triển của AI và Machine Learning, tương lai của bot giao dịch tự động còn nhiều tiềm năng để khám phá và cải tiến.