Building a Self-Optimizing Python Trading Bot with Reinforcement Learning and Binance API

pip install python-binance gym numpy pandas torch stable-baselines3

   from binance.client import Client

   API_KEY = "your_api_key"
   API_SECRET = "your_api_secret"
   client = Client(API_KEY, API_SECRET)

import gym
import numpy as np
from gym import spaces
from binance.client import Client

class TradingEnv(gym.Env):
    def __init__(self, client, symbol="BTCUSDT", window_size=10):
        super(TradingEnv, self).__init__()
        self.client = client
        self.symbol = symbol
        self.window_size = window_size

        # Action space: 0=hold, 1=buy, 2=sell
        self.action_space = spaces.Discrete(3)

        # Observation space: normalized price history
        self.observation_space = spaces.Box(
            low=0, high=1, shape=(window_size,), dtype=np.float32
        )

        self.reset()

    def _get_observation(self):
        # Fetch historical klines (1m candles)
        klines = self.client.get_historical_klines(
            self.symbol, Client.KLINE_INTERVAL_1MINUTE, f"{self.window_size} minutes ago"
        )
        closes = [float(k[4]) for k in klines]
        closes = np.array(closes)

        # Normalize prices
        if self.max_price is None:
            self.max_price = closes.max()
        closes = closes / self.max_price

        return closes

    def reset(self):
        self.balance = 1000  # Starting balance (USD)
        self.position = 0    # Current BTC position
        self.max_price = None
        return self._get_observation()

    def step(self, action):
        current_price = self._get_observation()[-1] * self.max_price
        reward = 0

        if action == 1:  # Buy
            if self.balance > 0:
                self.position = self.balance / current_price
                self.balance = 0
        elif action == 2:  # Sell
            if self.position > 0:
                self.balance = self.position * current_price
                self.position = 0
                reward = self.balance - 1000  # Profit/loss

        # Update observation
        obs = self._get_observation()
        done = False  # Episode ends when balance hits 0 or time limit
        info = {"balance": self.balance, "position": self.position}

        return obs, reward, done, info

from stable_baselines3 import PPO
from stable_baselines3.common.env_checker import check_env
from trading_env import TradingEnv

# Initialize environment
client = Client(API_KEY, API_SECRET)
env = TradingEnv(client)
check_env(env)  # Validate the environment

# Train PPO agent
model = PPO("MlpPolicy", env, verbose=1)
model.learn(total_timesteps=10000)
model.save("trading_bot_ppo")

  model.learn(total_timesteps=10000, tb_log_name="ppo_trading")

def _get_observation(self):
    # Load pre-downloaded historical data (e.g., from Binance)
    closes = np.load("btc_historical_closes.npy")[-self.window_size:]
    closes = closes / closes.max()
    return closes

def evaluate(model, env, episodes=10):
    returns = []
    for _ in range(episodes):
        obs = env.reset()
        done = False
        episode_return = 0
        while not done:
            action, _ = model.predict(obs)
            obs, reward, done, info = env.step(action)
            episode_return += reward
        returns.append(episode_return)
    return np.mean(returns), np.std(returns)

def _get_observation(self):
    klines = self.client.get_klines(
        symbol=self.symbol, interval=Client.KLINE_INTERVAL_1MINUTE, limit=self.window_size
    )
    closes = [float(k[4]) for k in klines]
    return np.array(closes) / np.max(closes)

def step(self, action):
    current_price = self._get_observation()[-1] * self.max_price
    if action == 1 and self.balance > 0:  # Buy
        self.entry_price = current_price
        self.position = self.balance / current_price
        self.balance = 0
    elif action == 2 and self.position > 0:  # Sell
        self.balance = self.position * current_price
        self.position = 0
    elif self.position > 0 and current_price < self.entry_price * 0.95:  # 5% stop-loss
        self.balance = self.position * current_price
        self.position = 0
    ...

def _get_observation(self):
    klines = self.client.get_historical_klines(...)
    closes = np.array([float(k[4]) for k in klines])
    rsi = talib.RSI(closes, timeperiod=14)
    macd = talib.MACD(closes)[0]
    return np.column_stack([closes, rsi, macd])

python
import optuna
from stable_baselines3.common.evaluation import evaluate_policy