2 posts tagged with "bot"

import cProfile
import re

def my_strategy_execution():
    # Code thực thi chiến lược
    pass

cProfile.run('my_strategy_execution()')

# Ví dụ: Thay vì tính toán thủ công
def calculate_sma_manual(data, window):
    sma = []
    for i in range(len(data)):
        if i < window - 1:
            sma.append(None)
        else:
            sma.append(sum(data[i-window+1:i+1]) / window)
    return sma

# Sử dụng thư viện tối ưu
import pandas as pd

def calculate_sma_pandas(data, window):
    return data.rolling(window=window).mean()

# So sánh hiệu suất
import time

data = pd.Series(range(10000))

start_time = time.time()
sma_manual = calculate_sma_manual(data, 50)
end_time = time.time()
print(f"Manual SMA time: {end_time - start_time}")

start_time = time.time()
sma_pandas = calculate_sma_pandas(data, 50)
end_time = time.time()
print(f"Pandas SMA time: {end_time - start_time}")

class TradingBot:
    def __init__(self, api_key, api_secret):
        self.api_key = api_key
        self.api_secret = api_secret
        self.exchange = self.connect_exchange()
        self.strategy = None
        self.positions = {}
        
    def connect_exchange(self):
        # Kết nối với sàn giao dịch
        exchange = ccxt.binance({
            'apiKey': self.api_key,
            'secret': self.api_secret
        })
        return exchange

def handle_connection(self):
    try:
        # Kiểm tra kết nối
        self.exchange.load_markets()
        return True
    except Exception as e:
        logger.error(f"Lỗi kết nối: {str(e)}")
        return False

def get_market_data(self, symbol, timeframe='1m'):
    try:
        # Lấy dữ liệu OHLCV
        ohlcv = self.exchange.fetch_ohlcv(
            symbol,
            timeframe,
            limit=100
        )
        return pd.DataFrame(
            ohlcv,
            columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
        )
    except Exception as e:
        logger.error(f"Lỗi lấy dữ liệu: {str(e)}")
        return None

def process_signals(self, data):
    # Áp dụng chiến lược
    signals = self.strategy.generate_signals(data)
    
    # Lọc tín hiệu
    valid_signals = signals[signals['signal'] != 0]
    
    return valid_signals

def execute_trade(self, symbol, side, amount):
    try:
        # Tạo lệnh
        order = self.exchange.create_order(
            symbol=symbol,
            type='market',
            side=side,
            amount=amount
        )
        
        # Cập nhật vị thế
        self.update_positions(order)
        
        return order
    except Exception as e:
        logger.error(f"Lỗi thực hiện lệnh: {str(e)}")
        return None

def update_positions(self, order):
    symbol = order['symbol']
    side = order['side']
    amount = float(order['amount'])
    
    if symbol not in self.positions:
        self.positions[symbol] = 0
        
    if side == 'buy':
        self.positions[symbol] += amount
    else:
        self.positions[symbol] -= amount

def check_risk_limits(self, symbol, amount):
    # Kiểm tra số dư
    balance = self.exchange.fetch_balance()
    available = balance['free'].get(symbol.split('/')[0], 0)
    
    # Kiểm tra giới hạn
    if amount > available:
        logger.warning(f"Số dư không đủ: {amount} > {available}")
        return False
        
    return True

def manage_stop_loss(self, symbol, entry_price):
    current_price = self.exchange.fetch_ticker(symbol)['last']
    stop_loss = entry_price * 0.95  # 5% stop loss
    
    if current_price <= stop_loss:
        self.execute_trade(symbol, 'sell', self.positions[symbol])
        logger.info(f"Đã kích hoạt stop loss: {symbol}")

def monitor_performance(self):
    # Tính toán lợi nhuận
    total_pnl = 0
    for symbol, amount in self.positions.items():
        current_price = self.exchange.fetch_ticker(symbol)['last']
        pnl = amount * current_price
        total_pnl += pnl
        
    return total_pnl

def generate_report(self):
    report = {
        'timestamp': datetime.now(),
        'total_pnl': self.monitor_performance(),
        'positions': self.positions,
        'trades': self.trade_history
    }
    
    # Lưu báo cáo
    self.save_report(report)

def run(self):
    while True:
        try:
            # Cập nhật dữ liệu
            data = self.get_market_data('BTC/USDT')
            
            # Xử lý tín hiệu
            signals = self.process_signals(data)
            
            # Thực hiện giao dịch
            for signal in signals:
                if self.check_risk_limits(signal['symbol'], signal['amount']):
                    self.execute_trade(
                        signal['symbol'],
                        signal['side'],
                        signal['amount']
                    )
            
            # Quản lý rủi ro
            self.manage_risk()
            
            # Tạo báo cáo
            self.generate_report()
            
            # Đợi đến chu kỳ tiếp theo
            time.sleep(60)
            
        except Exception as e:
            logger.error(f"Lỗi trong vòng lặp chính: {str(e)}")
            time.sleep(60)

def update_strategy(self, new_strategy):
    # Cập nhật chiến lược
    self.strategy = new_strategy
    
    # Kiểm tra tính tương thích
    if not self.strategy.validate():
        raise ValueError("Chiến lược không hợp lệ")