3 posts tagged with "automation"

import sys
sys.path.append('../Common')  # Thêm thư mục Common vào đường dẫn
from CommonYFinance import CommonYFinance

symbol = 'VCB.VN'
from_date = '2025-01-01'
to_date = '2025-03-28'
interval = '1d'

data = CommonYFinance.loaddataYFinance(symbol, from_date, to_date, interval)
print(data.head())

import pandas as pd
import yfinance as yf

class CommonYFinance:
    @staticmethod
    def loaddataYFinance(symbol, from_date, to_date, interval):
        """
        Tải dữ liệu lịch sử của một mã chứng khoán từ Yahoo Finance.
        
        Parameters:
        - symbol: Mã chứng khoán (VD: 'VCB.VN')
        - from_date: Ngày bắt đầu (YYYY-MM-DD)
        - to_date: Ngày kết thúc (YYYY-MM-DD)
        - interval: Khung thời gian ('1d', '1h', v.v.)
        
        Returns:
        - DataFrame chứa dữ liệu giá
        """
        data = yf.download(symbol, start=from_date, end=to_date, interval=interval, auto_adjust=False)
        data.reset_index(inplace=True)
        data = data.rename(columns={'Date': 'Datetime'})
        data.columns = data.columns.droplevel(1)  # Xóa cấp độ dư thừa nếu có
        data.columns.name = None
        return data

    @staticmethod
    def Hello():
        """Hàm test đơn giản."""
        print('Hello')
        return '1'

import sys
sys.path.append("../Common")  # Thêm thư mục Common vào đường dẫn
from CommonYFinance import CommonYFinance

# Cấu hình thông số
symbol = 'VCB.VN'
from_date = '2023-11-01'
to_date = '2023-11-30'
interval = '1d'

# Tải dữ liệu
data = CommonYFinance.loaddataYFinance(symbol, from_date, to_date, interval)
print(data.head())

pip install python-binance pandas numpy matplotlib

from binance.client import Client
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime
import time

class BinanceConnection:
    def __init__(self, api_key, api_secret):
        self.client = Client(api_key, api_secret)
        
    def test_connection(self):
        try:
            # Kiểm tra kết nối
            self.client.ping()
            print("Kết nối thành công!")
            return True
        except Exception as e:
            print(f"Lỗi kết nối: {str(e)}")
            return False
    
    def get_account_info(self):
        try:
            # Lấy thông tin tài khoản
            account = self.client.get_account()
            balances = {asset['asset']: float(asset['free'])
                       for asset in account['balances']
                       if float(asset['free']) > 0}
            return balances
        except Exception as e:
            print(f"Lỗi lấy thông tin tài khoản: {str(e)}")
            return None

# Tạo kết nối
api_key = 'your_api_key'
api_secret = 'your_api_secret'
binance = BinanceConnection(api_key, api_secret)

# Kiểm tra kết nối
if binance.test_connection():
    # Lấy thông tin tài khoản
    balances = binance.get_account_info()
    print("\nSố dư tài khoản:")
    for asset, amount in balances.items():
        print(f"{asset}: {amount}")

class MarketData:
    def __init__(self, client, symbol):
        self.client = client
        self.symbol = symbol
    
    def get_historical_data(self, interval='1h', limit=100):
        try:
            # Lấy dữ liệu lịch sử
            klines = self.client.get_klines(
                symbol=self.symbol,
                interval=interval,
                limit=limit
            )
            
            # Chuyển đổi dữ liệu
            data = pd.DataFrame(klines, columns=[
                'timestamp', 'open', 'high', 'low', 'close', 'volume',
                'close_time', 'quote_volume', 'trades', 'taker_buy_base',
                'taker_buy_quote', 'ignore'
            ])
            
            # Xử lý dữ liệu
            data['timestamp'] = pd.to_datetime(data['timestamp'], unit='ms')
            for col in ['open', 'high', 'low', 'close', 'volume']:
                data[col] = data[col].astype(float)
            
            return data
        except Exception as e:
            print(f"Lỗi lấy dữ liệu: {str(e)}")
            return None

# Lấy dữ liệu thị trường
market_data = MarketData(binance.client, 'BTCUSDT')
data = market_data.get_historical_data()

# Vẽ biểu đồ giá
plt.figure(figsize=(12, 6))
plt.plot(data['timestamp'], data['close'])
plt.title('Giá BTC/USDT')
plt.xlabel('Thời gian')
plt.ylabel('Giá (USDT)')
plt.grid(True)
plt.savefig('price_chart.png')

class TechnicalIndicators:
    def __init__(self, data):
        self.data = data
    
    def calculate_ma(self, window):
        """Tính toán đường trung bình động"""
        return self.data['close'].rolling(window=window).mean()
    
    def calculate_rsi(self, window=14):
        """Tính toán chỉ báo RSI"""
        delta = self.data['close'].diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=window).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=window).mean()
        rs = gain / loss
        return 100 - (100 / (1 + rs))
    
    def calculate_macd(self, fast=12, slow=26, signal=9):
        """Tính toán chỉ báo MACD"""
        exp1 = self.data['close'].ewm(span=fast, adjust=False).mean()
        exp2 = self.data['close'].ewm(span=slow, adjust=False).mean()
        macd = exp1 - exp2
        signal_line = macd.ewm(span=signal, adjust=False).mean()
        return macd, signal_line

# Tính toán các chỉ báo
indicators = TechnicalIndicators(data)
data['MA20'] = indicators.calculate_ma(20)
data['MA50'] = indicators.calculate_ma(50)
data['RSI'] = indicators.calculate_rsi()
data['MACD'], data['Signal'] = indicators.calculate_macd()

# Vẽ biểu đồ chỉ báo
fig, (ax1, ax2, ax3) = plt.subplots(3, 1, figsize=(12, 12), sharex=True)

# Biểu đồ giá và MA
ax1.plot(data['timestamp'], data['close'], label='Giá')
ax1.plot(data['timestamp'], data['MA20'], label='MA20')
ax1.plot(data['timestamp'], data['MA50'], label='MA50')
ax1.set_title('Giá và Đường MA')
ax1.legend()
ax1.grid(True)

# Biểu đồ RSI
ax2.plot(data['timestamp'], data['RSI'], label='RSI')
ax2.axhline(y=70, color='r', linestyle='--')
ax2.axhline(y=30, color='g', linestyle='--')
ax2.set_title('RSI')
ax2.legend()
ax2.grid(True)

# Biểu đồ MACD
ax3.plot(data['timestamp'], data['MACD'], label='MACD')
ax3.plot(data['timestamp'], data['Signal'], label='Signal')
ax3.set_title('MACD')
ax3.legend()
ax3.grid(True)

plt.tight_layout()
plt.savefig('technical_indicators.png')

class TradingSignals:
    def __init__(self, data):
        self.data = data
    
    def generate_signals(self):
        """Tạo tín hiệu giao dịch"""
        signals = pd.DataFrame(index=self.data.index)
        signals['Signal'] = 0
        
        # Tín hiệu từ MA
        signals.loc[self.data['MA20'] > self.data['MA50'], 'Signal'] = 1
        signals.loc[self.data['MA20'] < self.data['MA50'], 'Signal'] = -1
        
        # Tín hiệu từ RSI
        signals.loc[self.data['RSI'] < 30, 'Signal'] = 1
        signals.loc[self.data['RSI'] > 70, 'Signal'] = -1
        
        # Tín hiệu từ MACD
        signals.loc[self.data['MACD'] > self.data['Signal'], 'Signal'] = 1
        signals.loc[self.data['MACD'] < self.data['Signal'], 'Signal'] = -1
        
        return signals

# Tạo tín hiệu giao dịch
signals = TradingSignals(data).generate_signals()

# Vẽ biểu đồ tín hiệu
plt.figure(figsize=(12, 6))
plt.plot(data['timestamp'], data['close'], label='Giá')
plt.scatter(data[signals['Signal'] == 1]['timestamp'],
            data[signals['Signal'] == 1]['close'],
            marker='^', color='g', label='Mua')
plt.scatter(data[signals['Signal'] == -1]['timestamp'],
            data[signals['Signal'] == -1]['close'],
            marker='v', color='r', label='Bán')
plt.title('Tín Hiệu Giao Dịch')
plt.legend()
plt.grid(True)
plt.savefig('trading_signals.png')

class SimpleTradingBot:
    def __init__(self, client, symbol, initial_capital=1000):
        self.client = client
        self.symbol = symbol
        self.initial_capital = initial_capital
        self.capital = initial_capital
        self.position = None
        self.trades = []
    
    def run(self, interval='1h'):
        """Chạy bot giao dịch"""
        print(f"Bot đang chạy cho {self.symbol}")
        
        while True:
            try:
                # Lấy dữ liệu thị trường
                market_data = MarketData(self.client, self.symbol)
                data = market_data.get_historical_data(interval=interval)
                
                if data is None:
                    continue
                
                # Tính toán chỉ báo
                indicators = TechnicalIndicators(data)
                data['MA20'] = indicators.calculate_ma(20)
                data['MA50'] = indicators.calculate_ma(50)
                data['RSI'] = indicators.calculate_rsi()
                
                # Tạo tín hiệu
                signals = TradingSignals(data).generate_signals()
                current_signal = signals['Signal'].iloc[-1]
                
                # Thực hiện giao dịch
                if current_signal == 1 and self.position is None:
                    # Mở vị thế mua
                    price = data['close'].iloc[-1]
                    quantity = self.capital / price
                    
                    order = self.client.create_order(
                        symbol=self.symbol,
                        side='BUY',
                        type='MARKET',
                        quantity=quantity
                    )
                    
                    self.position = {
                        'type': 'long',
                        'entry_price': price,
                        'quantity': quantity
                    }
                    print(f"Mở vị thế mua: {price}")
                
                elif current_signal == -1 and self.position is not None:
                    # Đóng vị thế
                    price = data['close'].iloc[-1]
                    
                    order = self.client.create_order(
                        symbol=self.symbol,
                        side='SELL',
                        type='MARKET',
                        quantity=self.position['quantity']
                    )
                    
                    profit = (price - self.position['entry_price']) * self.position['quantity']
                    self.capital += profit
                    
                    self.trades.append({
                        'entry_price': self.position['entry_price'],
                        'exit_price': price,
                        'profit': profit
                    })
                    
                    print(f"Đóng vị thế: {price}, Lợi nhuận: {profit}")
                    self.position = None
                
                # Đợi một khoảng thời gian
                time.sleep(60)
                
            except Exception as e:
                print(f"Lỗi: {str(e)}")
                time.sleep(60)
    
    def get_performance(self):
        """Lấy thông tin hiệu suất"""
        if not self.trades:
            return {
                'total_trades': 0,
                'win_rate': 0,
                'total_profit': 0,
                'return': 0
            }
        
        profits = [trade['profit'] for trade in self.trades]
        winning_trades = [p for p in profits if p > 0]
        
        return {
            'total_trades': len(self.trades),
            'win_rate': len(winning_trades) / len(self.trades),
            'total_profit': sum(profits),
            'return': (self.capital - self.initial_capital) / self.initial_capital
        }

# Tạo và chạy bot
bot = SimpleTradingBot(binance.client, 'BTCUSDT')
bot.run()

# In thông tin hiệu suất
performance = bot.get_performance()
print("\nHiệu suất bot:")
for key, value in performance.items():
    print(f"{key}: {value:.2f}")

class OptimizedTradingBot(SimpleTradingBot):
    def __init__(self, client, symbol, initial_capital=1000):
        super().__init__(client, symbol, initial_capital)
        self.optimization_params = {
            'ma_short': range(10, 31, 5),
            'ma_long': range(40, 101, 10),
            'rsi_period': range(10, 21, 2),
            'rsi_overbought': range(65, 81, 5),
            'rsi_oversold': range(20, 36, 5),
            'stop_loss': [0.01, 0.02, 0.03],
            'take_profit': [0.03, 0.05, 0.07]
        }
    
    def optimize_parameters(self, data):
        """Tối ưu hóa tham số"""
        best_performance = {
            'total_return': -float('inf'),
            'params': None
        }
        
        # Tìm kiếm tham số tối ưu
        for ma_short in self.optimization_params['ma_short']:
            for ma_long in self.optimization_params['ma_long']:
                if ma_short >= ma_long:
                    continue
                    
                for rsi_period in self.optimization_params['rsi_period']:
                    for rsi_overbought in self.optimization_params['rsi_overbought']:
                        for rsi_oversold in self.optimization_params['rsi_oversold']:
                            if rsi_oversold >= rsi_overbought:
                                continue
                                
                            for stop_loss in self.optimization_params['stop_loss']:
                                for take_profit in self.optimization_params['take_profit']:
                                    params = {
                                        'ma_short': ma_short,
                                        'ma_long': ma_long,
                                        'rsi_period': rsi_period,
                                        'rsi_overbought': rsi_overbought,
                                        'rsi_oversold': rsi_oversold,
                                        'stop_loss': stop_loss,
                                        'take_profit': take_profit
                                    }
                                    
                                    # Kiểm tra hiệu suất
                                    performance = self._test_parameters(data, params)
                                    
                                    if performance['total_return'] > best_performance['total_return']:
                                        best_performance['total_return'] = performance['total_return']
                                        best_performance['params'] = params
        
        return best_performance['params']
    
    def _test_parameters(self, data, params):
        """Kiểm tra hiệu suất với bộ tham số"""
        # Tính toán chỉ báo với tham số mới
        indicators = TechnicalIndicators(data)
        data['MA_short'] = indicators.calculate_ma(params['ma_short'])
        data['MA_long'] = indicators.calculate_ma(params['ma_long'])
        data['RSI'] = indicators.calculate_rsi(params['rsi_period'])
        
        # Tạo tín hiệu
        signals = TradingSignals(data).generate_signals()
        
        # Mô phỏng giao dịch
        capital = self.initial_capital
        position = None
        trades = []
        
        for i in range(1, len(data)):
            if signals['Signal'].iloc[i] == 1 and position is None:
                # Mở vị thế mua
                entry_price = data['close'].iloc[i]
                quantity = capital / entry_price
                position = {
                    'type': 'long',
                    'entry_price': entry_price,
                    'quantity': quantity,
                    'stop_loss': entry_price * (1 - params['stop_loss']),
                    'take_profit': entry_price * (1 + params['take_profit'])
                }
            
            elif signals['Signal'].iloc[i] == -1 and position is not None:
                # Đóng vị thế
                exit_price = data['close'].iloc[i]
                profit = (exit_price - position['entry_price']) * position['quantity']
                capital += profit
                
                trades.append({
                    'entry_price': position['entry_price'],
                    'exit_price': exit_price,
                    'profit': profit
                })
                position = None
            
            # Kiểm tra stop loss và take profit
            if position is not None:
                current_price = data['close'].iloc[i]
                if current_price <= position['stop_loss'] or current_price >= position['take_profit']:
                    profit = (current_price - position['entry_price']) * position['quantity']
                    capital += profit
                    
                    trades.append({
                        'entry_price': position['entry_price'],
                        'exit_price': current_price,
                        'profit': profit
                    })
                    position = None
        
        return {
            'total_return': (capital - self.initial_capital) / self.initial_capital,
            'total_trades': len(trades),
            'win_rate': len([t for t in trades if t['profit'] > 0]) / len(trades) if trades else 0
        }

class RiskManager:
    def __init__(self, max_position_size=0.1, max_drawdown=0.2, max_trades_per_day=5):
        self.max_position_size = max_position_size
        self.max_drawdown = max_drawdown
        self.max_trades_per_day = max_trades_per_day
        self.daily_trades = 0
        self.initial_capital = None
        self.current_capital = None
        self.peak_capital = None
    
    def calculate_position_size(self, capital, price):
        """Tính toán khối lượng giao dịch"""
        return (capital * self.max_position_size) / price
    
    def check_drawdown(self, current_capital):
        """Kiểm tra drawdown"""
        if self.initial_capital is None:
            self.initial_capital = current_capital
            self.current_capital = current_capital
            self.peak_capital = current_capital
            return True
        
        self.current_capital = current_capital
        self.peak_capital = max(self.peak_capital, current_capital)
        drawdown = (self.peak_capital - self.current_capital) / self.peak_capital
        
        return drawdown <= self.max_drawdown
    
    def check_daily_trades(self):
        """Kiểm tra số lượng giao dịch trong ngày"""
        if self.daily_trades >= self.max_trades_per_day:
            return False
        self.daily_trades += 1
        return True
    
    def reset_daily_trades(self):
        """Reset số lượng giao dịch hàng ngày"""
        self.daily_trades = 0

class OptimizedTradingBotWithRisk(OptimizedTradingBot):
    def __init__(self, client, symbol, initial_capital=1000):
        super().__init__(client, symbol, initial_capital)
        self.risk_manager = RiskManager()
    
    def run(self, interval='1h'):
        """Chạy bot với quản lý rủi ro"""
        print(f"Bot đang chạy cho {self.symbol}")
        
        while True:
            try:
                # Lấy dữ liệu thị trường
                market_data = MarketData(self.client, self.symbol)
                data = market_data.get_historical_data(interval=interval)
                
                if data is None:
                    continue
                
                # Tối ưu tham số
                if len(self.trades) % 100 == 0:  # Tối ưu mỗi 100 giao dịch
                    best_params = self.optimize_parameters(data)
                    self.strategy_params = best_params
                
                # Tính toán chỉ báo
                indicators = TechnicalIndicators(data)
                data['MA20'] = indicators.calculate_ma(self.strategy_params['ma_short'])
                data['MA50'] = indicators.calculate_ma(self.strategy_params['ma_long'])
                data['RSI'] = indicators.calculate_rsi(self.strategy_params['rsi_period'])
                
                # Tạo tín hiệu
                signals = TradingSignals(data).generate_signals()
                current_signal = signals['Signal'].iloc[-1]
                
                # Kiểm tra rủi ro
                if not self.risk_manager.check_drawdown(self.capital):
                    print("Đã vượt quá mức drawdown cho phép")
                    break
                
                if not self.risk_manager.check_daily_trades():
                    print("Đã đạt giới hạn giao dịch trong ngày")
                    time.sleep(3600)  # Đợi 1 giờ
                    self.risk_manager.reset_daily_trades()
                    continue
                
                # Thực hiện giao dịch
                if current_signal == 1 and self.position is None:
                    # Mở vị thế mua
                    price = data['close'].iloc[-1]
                    quantity = self.risk_manager.calculate_position_size(self.capital, price)
                    
                    order = self.client.create_order(
                        symbol=self.symbol,
                        side='BUY',
                        type='MARKET',
                        quantity=quantity
                    )
                    
                    self.position = {
                        'type': 'long',
                        'entry_price': price,
                        'quantity': quantity,
                        'stop_loss': price * (1 - self.strategy_params['stop_loss']),
                        'take_profit': price * (1 + self.strategy_params['take_profit'])
                    }
                    print(f"Mở vị thế mua: {price}")
                
                elif current_signal == -1 and self.position is not None:
                    # Đóng vị thế
                    price = data['close'].iloc[-1]
                    
                    order = self.client.create_order(
                        symbol=self.symbol,
                        side='SELL',
                        type='MARKET',
                        quantity=self.position['quantity']
                    )
                    
                    profit = (price - self.position['entry_price']) * self.position['quantity']
                    self.capital += profit
                    
                    self.trades.append({
                        'entry_price': self.position['entry_price'],
                        'exit_price': price,
                        'profit': profit
                    })
                    
                    print(f"Đóng vị thế: {price}, Lợi nhuận: {profit}")
                    self.position = None
                
                # Đợi một khoảng thời gian
                time.sleep(60)
                
            except Exception as e:
                print(f"Lỗi: {str(e)}")
                time.sleep(60)

class TradingBotMonitor:
    def __init__(self, bot):
        self.bot = bot
        self.performance_history = []
        self.alerts = []
    
    def monitor_performance(self):
        """Giám sát hiệu suất bot"""
        performance = self.bot.get_performance()
        self.performance_history.append(performance)
        
        # Kiểm tra các điều kiện cảnh báo
        if performance['win_rate'] < 0.4:
            self.alerts.append({
                'type': 'warning',
                'message': f"Tỷ lệ thắng thấp: {performance['win_rate']:.2f}"
            })
        
        if performance['return'] < -0.1:
            self.alerts.append({
                'type': 'error',
                'message': f"Lỗi lớn: {performance['return']:.2f}"
            })
        
        return performance
    
    def generate_report(self):
        """Tạo báo cáo hiệu suất"""
        if not self.performance_history:
            return "Chưa có dữ liệu hiệu suất"
        
        latest = self.performance_history[-1]
        report = f"""
        Báo cáo hiệu suất bot:
        - Tổng số giao dịch: {latest['total_trades']}
        - Tỷ lệ thắng: {latest['win_rate']:.2f}
        - Tổng lợi nhuận: {latest['total_profit']:.2f}
        - Lợi nhuận: {latest['return']:.2f}
        
        Cảnh báo:
        """
        
        for alert in self.alerts:
            report += f"- {alert['type']}: {alert['message']}\n"
        
        return report

def deploy_bot(api_key, api_secret, symbol, initial_capital=1000):
    """Triển khai bot giao dịch"""
    try:
        # Tạo bot
        bot = OptimizedTradingBotWithRisk(
            client=Client(api_key, api_secret),
            symbol=symbol,
            initial_capital=initial_capital
        )
        
        # Tạo monitor
        monitor = TradingBotMonitor(bot)
        
        # Chạy bot trong thread riêng
        import threading
        bot_thread = threading.Thread(target=bot.run)
        bot_thread.start()
        
        # Giám sát hiệu suất
        while bot_thread.is_alive():
            performance = monitor.monitor_performance()
            print(monitor.generate_report())
            time.sleep(300)  # Cập nhật mỗi 5 phút
        
        return True
    except Exception as e:
        print(f"Lỗi triển khai: {str(e)}")
        return False

# Triển khai bot
if deploy_bot('your_api_key', 'your_api_secret', 'BTCUSDT'):
    print("Bot đã được triển khai thành công")
else:
    print("Lỗi triển khai bot")