Bitget Scalp Trading Bot - Script Documentation

#!/usr/bin/env python3
"""
Bitget Scalp Trading Bot
========================
A comprehensive automated scalp trading bot for Bitget exchange using Spot REST API.
Features candlestick pattern detection, risk management, and modular architecture.
Author: AI Assistant
License: MIT
"""
import os
import time
import hmac
import hashlib
import base64
import requests
import pandas as pd
import numpy as np
import csv
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
from dotenv import load_dotenv
import json
import logging
from dataclasses import dataclass
from collections import deque
import threading
from concurrent.futures import ThreadPoolExecutor
# Load environment variables
load_dotenv()
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('trading_bot.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
@dataclass
class TechnicalIndicators:
"""Data class for technical indicators"""
rsi: float = 0.0
macd: float = 0.0
macd_signal: float = 0.0
bb_upper: float = 0.0
bb_lower: float = 0.0
bb_middle: float = 0.0
ema_fast: float = 0.0
ema_slow: float = 0.0
volume_sma: float = 0.0
atr: float = 0.0
@dataclass
class TradeSignal:
"""Data class for trade signals"""
symbol: str
signal_type: str # 'BUY', 'SELL', 'HOLD'
confidence: float # 0.0 to 1.0
patterns_detected: List[str]
indicators: TechnicalIndicators
timestamp: datetime
price: float
reason: str
class TechnicalAnalysis:
"""Technical analysis and indicator calculation class"""

@staticmethod
def calculate_rsi(prices: pd.Series, period: int = 14) -> pd.Series:
"""Calculate Relative Strength Index"""
delta = prices.diff()
gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
rs = gain / loss
 rsi = 100 - (100 / (1 + rs))
return rsi

@staticmethod
def calculate_macd(prices: pd.Series, fast: int = 12, slow: int = 26, signal: in
t = 9) -> Tuple[pd.Series, pd.Series, pd.Series]:
"""Calculate MACD, MACD Signal, and MACD Histogram"""
ema_fast = prices.ewm(span=fast).mean()
ema_slow = prices.ewm(span=slow).mean()
macd = ema_fast - ema_slow
macd_signal = macd.ewm(span=signal).mean()
macd_histogram = macd - macd_signal
return macd, macd_signal, macd_histogram

@staticmethod
def calculate_bollinger_bands(prices: pd.Series, period: int = 20, std_dev: int
= 2) -> Tuple[pd.Series, pd.Series, pd.Series]:
"""Calculate Bollinger Bands"""
sma = prices.rolling(window=period).mean()
std = prices.rolling(window=period).std()
upper_band = sma + (std * std_dev)
lower_band = sma - (std * std_dev)
return upper_band, sma, lower_band

@staticmethod
def calculate_ema(prices: pd.Series, period: int) -> pd.Series:
"""Calculate Exponential Moving Average"""
return prices.ewm(span=period).mean()

@staticmethod
def calculate_atr(high: pd.Series, low: pd.Series, close: pd.Series, period: int
= 14) -> pd.Series:
"""Calculate Average True Range"""
high_low = high - low
high_close = np.abs(high - close.shift())
low_close = np.abs(low - close.shift())
true_range = np.maximum(high_low, np.maximum(high_close, low_close))
atr = true_range.rolling(window=period).mean()
return atr
class PerformanceMonitor:
"""Monitor and track bot performance metrics"""

def __init__(self):
self.trades_history = deque(maxlen=1000) # Store last 1000 trades
self.daily_pnl = {}
self.win_streak = 0
self.loss_streak = 0
self.max_win_streak = 0
self.max_loss_streak = 0
self.total_trades = 0
self.winning_trades = 0
self.losing_trades = 0
self.total_pnl = 0.0
self.max_drawdown = 0.0
self.peak_balance = 0.0

def record_trade(self, pnl: float, entry_price: float, exit_price: float, quanti

ty: float):
"""Record a completed trade"""
trade_data = {
'timestamp': datetime.now(),
'pnl': pnl,
'entry_price': entry_price,
'exit_price': exit_price,
'quantity': quantity,
'return_pct': pnl / (entry_price * quantity) if entry_price > 0 else 0
}

self.trades_history.append(trade_data)
self.total_trades += 1
self.total_pnl += pnl

# Track win/loss streaks

if pnl > 0:
self.winning_trades += 1
self.win_streak += 1
self.loss_streak = 0
 self.max_win_streak = max(self.max_win_streak, self.win_streak)
else:
self.losing_trades += 1
self.loss_streak += 1
self.win_streak = 0
self.max_loss_streak = max(self.max_loss_streak, self.loss_streak)

# Daily P&L; tracking

today = datetime.now().date()
if today not in self.daily_pnl:
self.daily_pnl[today] = 0
self.daily_pnl[today] += pnl

def update_balance(self, current_balance: float):

"""Update balance and calculate drawdown"""
if current_balance > self.peak_balance:
self.peak_balance = current_balance

if self.peak_balance > 0:
drawdown = (self.peak_balance - current_balance) / self.peak_balance
self.max_drawdown = max(self.max_drawdown, drawdown)

def get_statistics(self) -> Dict:

"""Get comprehensive performance statistics"""
win_rate = self.winning_trades / self.total_trades if self.total_trades > 0
else 0
avg_win = sum(t['pnl'] for t in self.trades_history if t['pnl'] > 0) / max(s
elf.winning_trades, 1)
avg_loss = sum(t['pnl'] for t in self.trades_history if t['pnl'] < 0) / max(
self.losing_trades, 1)
profit_factor = abs(avg_win * self.winning_trades / (avg_loss * self.losing_
trades)) if self.losing_trades > 0 else float('inf')

return {
'total_trades': self.total_trades,
'winning_trades': self.winning_trades,
'losing_trades': self.losing_trades,
'win_rate': win_rate,
'total_pnl': self.total_pnl,
'avg_win': avg_win,
'avg_loss': avg_loss,
'profit_factor': profit_factor,
'max_win_streak': self.max_win_streak,
'max_loss_streak': self.max_loss_streak,
'max_drawdown': self.max_drawdown,
'current_streak': self.win_streak if self.win_streak > 0 else -self.loss
_streak
}
class BitgetTradingBot:
"""
Bitget Scalp Trading Bot with pattern detection and risk management
"""

def __init__(self):
"""Initialize the trading bot with API credentials and configuration"""
self.api_key = os.getenv('BITGET_API_KEY')
self.api_secret = os.getenv('BITGET_API_SECRET')
self.api_passphrase = os.getenv('BITGET_PASSPHRASE') # Bitget requires pass
phrase
self.base_url = 'https://api.bitget.com'

if not self.api_key or not self.api_secret or not self.api_passphrase:

raise ValueError("API credentials not found. Please set BITGET_API_KEY,
BITGET_API_SECRET, and BITGET_PASSPHRASE in .env file")

# Trading configuration
self.symbol = ""
self.profit_target = 0.002 # 0.2%
self.stop_loss = 0.003 # -0.3%
self.max_position_percent = 0.05 # 5% of balance
self.order_timeout = 60 # seconds
self.price_adjustment = 0.0001 # Small adjustment for better execution
self.max_spread_percent = 0.001 # 0.1% max spread
self.min_depth_threshold = 1000 # Minimum depth in USDT

# Enhanced trading controls

self.max_daily_trades = 50
 self.cooling_period = 300 # 5 minutes between trades
self.min_confidence_threshold = 0.6
self.daily_trade_count = 0
self.last_trade_time = 0
self.last_reset_date = datetime.now().date()

# Active positions tracking

self.active_orders = {}
self.positions = {}

# Initialize components
self.technical_analyzer = TechnicalAnalysis()
self.performance_monitor = PerformanceMonitor()

# Initialize CSV logging

self.init_csv_logging()

def reset_daily_counters(self):
"""Reset daily counters if new day"""
today = datetime.now().date()
if today != self.last_reset_date:
self.daily_trade_count = 0
self.last_reset_date = today
logger.info("Daily counters reset for new trading day")

def init_csv_logging(self):
"""Initialize CSV file for trade logging"""
self.csv_filename = f"trades_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
with open(self.csv_filename, 'w', newline='') as f:
writer = csv.writer(f)
writer.writerow([
'timestamp', 'action', 'symbol', 'price', 'quantity',
'order_id', 'pnl', 'balance_usdt', 'pattern_detected'
])

def log_trade(self, action: str, symbol: str, price: float, quantity: float,
order_id: str = "", pnl: float = 0, pattern: str = ""):
"""Log trade action to CSV file"""
try:
balance = self.get_balance('USDT')
with open(self.csv_filename, 'a', newline='') as f:
writer = csv.writer(f)
writer.writerow([
datetime.now().isoformat(),
action,
symbol,
price,
quantity,
order_id,
pnl,
balance,
pattern
])
except Exception as e:
logger.error(f"Failed to log trade: {e}")

def _generate_signature(self, method: str, request_path: str, body: str = '') ->
str:
"""Generate signature for Bitget API authentication"""
timestamp = str(int(time.time() * 1000))
message = timestamp + method.upper() + request_path + body

signature = base64.b64encode(
hmac.new(
self.api_secret.encode('utf-8'),
message.encode('utf-8'),
hashlib.sha256
).digest()
).decode('utf-8')

return signature, timestamp

def _make_request(self, method: str, endpoint: str, params: dict = None,

signed: bool = False, retries: int = 3) -> dict:
"""Make HTTP request to Bitget API with error handling and retries"""
url = f"{self.base_url}{endpoint}"
 for attempt in range(retries):
try:
headers = {
'Content-Type': 'application/json',
}

if signed:
body = json.dumps(params) if params and method != 'GET' else ''
signature, timestamp = self._generate_signature(method, endpoint
, body)

headers.update({
'ACCESS-KEY': self.api_key,
'ACCESS-SIGN': signature,
'ACCESS-TIMESTAMP': timestamp,
'ACCESS-PASSPHRASE': self.api_passphrase,
})

if method == 'GET':
response = requests.get(url, params=params, headers=headers, tim
eout=10)
elif method == 'POST':
response = requests.post(url, json=params, headers=headers, time
out=10)
elif method == 'DELETE':
response = requests.delete(url, json=params, headers=headers, ti
meout=10)

response.raise_for_status()
result = response.json()

# Bitget API response format

if result.get('code') == '00000':
return result.get('data', {})
else:
logger.error(f"API error: {result.get('msg', 'Unknown error')}")
return {}

except requests.exceptions.RequestException as e:
logger.warning(f"Request attempt {attempt + 1} failed: {e}")
if attempt == retries - 1:
raise
time.sleep(2 ** attempt) # Exponential backoff

def get_symbol_info(self, symbol: str) -> dict:

"""Get trading symbol information"""
try:
response = self._make_request('GET', '/api/spot/v1/public/symbols')

if isinstance(response, list):
for s in response:
if s['symbol'] == symbol:
return s

raise ValueError(f"Symbol {symbol} not found")

except Exception as e:
logger.error(f"Failed to get symbol info: {e}")
return {}

def get_candles(self, symbol: str, granularity: str = '1min', limit: int = 100)
-> pd.DataFrame:
"""
Fetch candlestick data from Bitget API

Args:
symbol: Trading pair (e.g., 'BTCUSDT')
granularity: Kline interval (1min, 5min, 15min, etc.)
limit: Number of candles to fetch

Returns:
DataFrame with OHLCV data
"""
try:
params = {
'symbol': symbol,
'granularity': granularity,
'limit': str(limit)
 }

response = self._make_request('GET', '/api/spot/v1/market/candles', para

ms)

if not response:
return pd.DataFrame()

# Convert to DataFrame - Bitget returns [timestamp, open, high, low, clo

se, volume]
df = pd.DataFrame(response, columns=[
'timestamp', 'open', 'high', 'low', 'close', 'volume'
])

# Convert to appropriate data types

numeric_columns = ['open', 'high', 'low', 'close', 'volume']
for col in numeric_columns:
df[col] = pd.to_numeric(df[col])

df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')

df = df.sort_values('timestamp').reset_index(drop=True)

logger.info(f"Fetched {len(df)} candles for {symbol}")

return df

except Exception as e:
logger.error(f"Failed to fetch candles: {e}")
return pd.DataFrame()

def detect_bullish_engulfing(self, df: pd.DataFrame) -> bool:

"""Detect bullish engulfing pattern"""
if len(df) < 2:
return False

prev_candle = df.iloc[-2]
curr_candle = df.iloc[-1]

# Previous candle is bearish (red)

prev_bearish = prev_candle['close'] < prev_candle['open']
# Current candle is bullish (green)
curr_bullish = curr_candle['close'] > curr_candle['open']
# Current candle engulfs previous
engulfs = (curr_candle['open'] < prev_candle['close'] and
curr_candle['close'] > prev_candle['open'])

return prev_bearish and curr_bullish and engulfs

def detect_hammer(self, df: pd.DataFrame) -> bool:

"""Detect hammer pattern"""
if len(df) < 1:
return False

candle = df.iloc[-1]
body_size = abs(candle['close'] - candle['open'])
upper_shadow = candle['high'] - max(candle['open'], candle['close'])
lower_shadow = min(candle['open'], candle['close']) - candle['low']

# Hammer characteristics
small_body = body_size < (candle['high'] - candle['low']) * 0.3
long_lower_shadow = lower_shadow > body_size * 2
short_upper_shadow = upper_shadow < body_size * 0.5

return small_body and long_lower_shadow and short_upper_shadow

def detect_doji(self, df: pd.DataFrame) -> bool:

"""Detect doji pattern"""
if len(df) < 1:
return False

candle = df.iloc[-1]
body_size = abs(candle['close'] - candle['open'])
total_range = candle['high'] - candle['low']

# Doji has very small body relative to range

return body_size < total_range * 0.05 if total_range > 0 else False

def detect_bearish_engulfing(self, df: pd.DataFrame) -> bool:

 """Detect bearish engulfing pattern"""
if len(df) < 2:
return False

prev_candle = df.iloc[-2]
curr_candle = df.iloc[-1]

# Previous candle is bullish

prev_bullish = prev_candle['close'] > prev_candle['open']
# Current candle is bearish
curr_bearish = curr_candle['close'] < curr_candle['open']
# Current candle engulfs previous
engulfs = (curr_candle['open'] > prev_candle['close'] and
curr_candle['close'] < prev_candle['open'])

return prev_bullish and curr_bearish and engulfs

def detect_patterns(self, df: pd.DataFrame) -> dict:

"""
Detect various candlestick patterns

Returns:
Dictionary with pattern names and detection results
"""
patterns = {
'bullish_engulfing': self.detect_bullish_engulfing(df),
'hammer': self.detect_hammer(df),
'doji': self.detect_doji(df),
'bearish_engulfing': self.detect_bearish_engulfing(df)
}

detected = [name for name, detected in patterns.items() if detected]

if detected:
logger.info(f"Patterns detected: {', '.join(detected)}")

return patterns

def calculate_technical_indicators(self, df: pd.DataFrame) -> TechnicalIndicator

s:
"""Calculate comprehensive technical indicators"""
if len(df) < 50: # Need sufficient data
return TechnicalIndicators()

try:
close_prices = df['close']
high_prices = df['high']
low_prices = df['low']
volume = df['volume']

# Calculate indicators
rsi = self.technical_analyzer.calculate_rsi(close_prices).iloc[-1]
macd, macd_signal, _ = self.technical_analyzer.calculate_macd(close_pric
es)
bb_upper, bb_middle, bb_lower = self.technical_analyzer.calculate_bollin
ger_bands(close_prices)
ema_fast = self.technical_analyzer.calculate_ema(close_prices, 12).iloc[
-1]
ema_slow = self.technical_analyzer.calculate_ema(close_prices, 26).iloc[
-1]
volume_sma = volume.rolling(window=20).mean().iloc[-1]
atr = self.technical_analyzer.calculate_atr(high_prices, low_prices, clo
se_prices).iloc[-1]

return TechnicalIndicators(
rsi=rsi if not pd.isna(rsi) else 50.0,
macd=macd.iloc[-1] if not pd.isna(macd.iloc[-1]) else 0.0,
macd_signal=macd_signal.iloc[-1] if not pd.isna(macd_signal.iloc[-1]
) else 0.0,
bb_upper=bb_upper.iloc[-1] if not pd.isna(bb_upper.iloc[-1]) else 0.
0,
bb_lower=bb_lower.iloc[-1] if not pd.isna(bb_lower.iloc[-1]) else 0.
0,
bb_middle=bb_middle.iloc[-1] if not pd.isna(bb_middle.iloc[-1]) else
0.0,
ema_fast=ema_fast if not pd.isna(ema_fast) else 0.0,
ema_slow=ema_slow if not pd.isna(ema_slow) else 0.0,
volume_sma=volume_sma if not pd.isna(volume_sma) else 0.0,
 atr=atr if not pd.isna(atr) else 0.0
)

except Exception as e:
logger.error(f"Error calculating technical indicators: {e}")
return TechnicalIndicators()

def generate_trade_signal(self, df: pd.DataFrame, current_price: float) -> Trade

Signal:
"""Generate comprehensive trade signal with confidence scoring"""
patterns = self.detect_patterns(df)
indicators = self.calculate_technical_indicators(df)

# Initialize signal
signal_type = 'HOLD'
confidence = 0.0
reasons = []
patterns_detected = []

# Pattern-based signals
pattern_score = 0.0
if patterns['bullish_engulfing']:
pattern_score += 0.3
patterns_detected.append('bullish_engulfing')
reasons.append('Bullish engulfing pattern')

if patterns['hammer']:
pattern_score += 0.25
patterns_detected.append('hammer')
reasons.append('Hammer pattern')

if patterns['doji']:
pattern_score += 0.1
patterns_detected.append('doji')
reasons.append('Doji indecision')

# Technical indicator signals

indicator_score = 0.0

# RSI conditions
if indicators.rsi < 30: # Oversold
indicator_score += 0.2
reasons.append(f'RSI oversold ({indicators.rsi:.1f})')
elif indicators.rsi > 70: # Overbought
indicator_score -= 0.2
reasons.append(f'RSI overbought ({indicators.rsi:.1f})')

# MACD conditions
if indicators.macd > indicators.macd_signal:
indicator_score += 0.15
reasons.append('MACD bullish crossover')
else:
indicator_score -= 0.1

# Bollinger Bands
if current_price <= indicators.bb_lower:
indicator_score += 0.15
reasons.append('Price at BB lower band')
elif current_price >= indicators.bb_upper:
indicator_score -= 0.15
reasons.append('Price at BB upper band')

# EMA trend
if indicators.ema_fast > indicators.ema_slow:
indicator_score += 0.1
reasons.append('EMA uptrend')
else:
indicator_score -= 0.05

# Volume confirmation
current_volume = df['volume'].iloc[-1]
if current_volume > indicators.volume_sma * 1.2:
indicator_score += 0.1
reasons.append('High volume confirmation')

# Calculate total confidence

total_score = pattern_score + indicator_score
 confidence = max(0.0, min(1.0, total_score))

# Determine signal type

if confidence >= self.min_confidence_threshold and total_score > 0:
signal_type = 'BUY'
elif total_score < -0.3:
signal_type = 'SELL'

return TradeSignal(
symbol=self.symbol,
signal_type=signal_type,
confidence=confidence,
patterns_detected=patterns_detected,
indicators=indicators,
timestamp=datetime.now(),
price=current_price,
reason='; '.join(reasons)
)

def get_order_book(self, symbol: str) -> dict:

"""Get order book depth"""
try:
params = {'symbol': symbol}
response = self._make_request('GET', '/api/spot/v1/market/depth', params
)

# Calculate spread and depth metrics

if response and 'bids' in response and 'asks' in response:
best_bid = float(response['bids'][0][0]) if response['bids'] else 0
best_ask = float(response['asks'][0][0]) if response['asks'] else 0

spread = (best_ask - best_bid) / best_bid if best_bid > 0 else float

('inf')

# Calculate depth in USDT

bid_depth = sum(float(bid[0]) * float(bid[1]) for bid in response['b
ids'][:10])
ask_depth = sum(float(ask[0]) * float(ask[1]) for ask in response['a
sks'][:10])

response['spread_percent'] = spread
response['bid_depth_usdt'] = bid_depth
response['ask_depth_usdt'] = ask_depth
response['best_bid'] = best_bid
response['best_ask'] = best_ask

return response
except Exception as e:
logger.error(f"Failed to get order book: {e}")
return {}

def check_market_conditions(self, symbol: str) -> bool:

"""Check if market conditions are suitable for trading"""
try:
order_book = self.get_order_book(symbol)

if not order_book:
return False

# Check spread
spread = order_book.get('spread_percent', float('inf'))
if spread > self.max_spread_percent:
logger.warning(f"Spread too wide: {spread:.4f}%")
return False

# Check depth
bid_depth = order_book.get('bid_depth_usdt', 0)
ask_depth = order_book.get('ask_depth_usdt', 0)

if bid_depth < self.min_depth_threshold or ask_depth < self.min_depth_th

reshold:
logger.warning(f"Insufficient liquidity - Bid: {bid_depth}, Ask: {as
k_depth}")
return False

logger.info(f"Market conditions OK - Spread: {spread:.4f}%, Depth: {min(

bid_depth, ask_depth):.0f} USDT")
 return True

except Exception as e:
logger.error(f"Failed to check market conditions: {e}")
return False

def get_balance(self, asset: str) -> float:

"""Get account balance for specific asset"""
try:
response = self._make_request('GET', '/api/spot/v1/account/assets', sign
ed=True)

if isinstance(response, list):
for balance in response:
if balance.get('coinName') == asset:
return float(balance.get('available', 0))

return 0.0
except Exception as e:
logger.error(f"Failed to get balance: {e}")
return 0.0

def place_order(self, symbol: str, side: str, quantity: float,

price: float = None, order_type: str = 'limit') -> dict:
"""Place an order on Bitget"""
try:
params = {
'symbol': symbol,
'side': side.lower(), # Bitget uses lowercase
'orderType': order_type.lower(),
'quantity': str(quantity),
}

if order_type.lower() == 'limit':
params['price'] = str(price)

response = self._make_request('POST', '/api/spot/v1/trade/orders', param

s, signed=True)

if response:
order_id = response.get('orderId')
logger.info(f"Order placed: {side} {quantity} {symbol} at {price} (I
D: {order_id})")

# Track active order

self.active_orders[order_id] = {
'symbol': symbol,
'side': side,
'quantity': quantity,
'price': price,
'timestamp': time.time()
}

return response
except Exception as e:
logger.error(f"Failed to place order: {e}")
return {}

def cancel_order(self, symbol: str, order_id: str) -> bool:

"""Cancel an existing order"""
try:
params = {
'symbol': symbol,
'orderId': order_id
}

response = self._make_request('POST', '/api/spot/v1/trade/cancel-order',

params, signed=True)

if response:
logger.info(f"Order cancelled: {order_id}")
self.active_orders.pop(order_id, None)
return True

return False
except Exception as e:
logger.error(f"Failed to cancel order: {e}")
 return False

def get_order_status(self, symbol: str, order_id: str) -> dict:

"""Get order status"""
try:
params = {
'symbol': symbol,
'orderId': order_id
}

response = self._make_request('GET', '/api/spot/v1/trade/orderInfo', par

ams, signed=True)
return response
except Exception as e:
logger.error(f"Failed to get order status: {e}")
return {}

def calculate_position_size(self, symbol: str, price: float) -> float:

"""Calculate position size based on risk management rules"""
try:
usdt_balance = self.get_balance('USDT')
max_position_value = usdt_balance * self.max_position_percent
quantity = max_position_value / price

# Get symbol info for precision

symbol_info = self.get_symbol_info(symbol)
if symbol_info:
# Round to appropriate precision based on Bitget symbol info
quantity_precision = int(symbol_info.get('quantityPrecision', 6))
quantity = round(quantity, quantity_precision)

logger.info(f"Calculated position size: {quantity} (${max_position_value

:.2f})")
return quantity

except Exception as e:
logger.error(f"Failed to calculate position size: {e}")
return 0.0

def check_trading_conditions(self) -> Tuple[bool, str]:

"""Check if trading conditions allow new trades"""
self.reset_daily_counters()

# Check daily trade limit

if self.daily_trade_count >= self.max_daily_trades:
return False, f"Daily trade limit reached ({self.daily_trade_count}/{sel
f.max_daily_trades})"

# Check cooling period

time_since_last_trade = time.time() - self.last_trade_time
if time_since_last_trade < self.cooling_period:
remaining = self.cooling_period - time_since_last_trade
return False, f"Cooling period active ({remaining:.0f}s remaining)"

# Check existing positions

if self.symbol in self.positions:
return False, "Already have active position"

return True, "Trading conditions OK"

def adaptive_position_sizing(self, symbol: str, price: float, confidence: float)

-> float:
"""Calculate adaptive position size based on confidence and performance"""
base_quantity = self.calculate_position_size(symbol, price)

# Adjust based on confidence

confidence_multiplier = 0.5 + (confidence * 0.5) # 0.5x to 1.0x based on co
nfidence

# Adjust based on recent performance

performance_stats = self.performance_monitor.get_statistics()
if performance_stats['total_trades'] > 10:
win_rate = performance_stats['win_rate']
if win_rate > 0.6:
confidence_multiplier *= 1.2 # Increase size if performing well
elif win_rate < 0.4:
confidence_multiplier *= 0.8 # Decrease size if performing poorly
 # Adjust based on current streak
current_streak = performance_stats.get('current_streak', 0)
if current_streak < -3: # Losing streak
confidence_multiplier *= 0.7
elif current_streak > 3: # Winning streak
confidence_multiplier = min(confidence_multiplier * 1.1, 1.0)

adjusted_quantity = base_quantity * confidence_multiplier

logger.info(f"Position sizing: base={base_quantity:.6f}, confidence={confide
nce:.2f}, "
f"multiplier={confidence_multiplier:.2f}, final={adjusted_quantit
y:.6f}")

return adjusted_quantity

def dynamic_exit_strategy(self, symbol: str, entry_price: float, entry_side: str

, indicators: TechnicalIndicators) -> Tuple[bool, str]:
"""Enhanced exit strategy with multiple conditions"""
try:
order_book = self.get_order_book(symbol)
if not order_book:
return False, ""

current_price = order_book['best_bid'] if entry_side == 'BUY' else order

_book['best_ask']

if entry_side == 'BUY':
# Long position
profit_pct = (current_price - entry_price) / entry_price

# Basic profit/loss targets

if profit_pct >= self.profit_target:
return True, f"PROFIT_TARGET ({profit_pct:.4f})"
elif profit_pct <= -self.stop_loss:
return True, f"STOP_LOSS ({profit_pct:.4f})"

# Technical exit conditions

if indicators.rsi > 80: # Extremely overbought
return True, f"RSI_OVERBOUGHT ({indicators.rsi:.1f})"

if current_price >= indicators.bb_upper * 1.01: # Price well above

upper BB
return True, "BOLLINGER_UPPER_BREACH"

# Bearish pattern detection

current_df = self.get_candles(symbol, limit=20)
if not current_df.empty:
patterns = self.detect_patterns(current_df)
if patterns['bearish_engulfing']:
return True, "BEARISH_ENGULFING_EXIT"

return False, ""

except Exception as e:
logger.error(f"Failed to check exit conditions: {e}")
return False, ""

def execute_trade_cycle(self, symbol: str):

"""Execute one complete trade cycle with enhanced analysis"""
try:
logger.info(f"Starting enhanced trade cycle for {symbol}")

# 1. Check trading conditions

can_trade, reason = self.check_trading_conditions()
if not can_trade:
logger.info(f"Trading conditions not met: {reason}")
return

# 2. Check market conditions

if not self.check_market_conditions(symbol):
logger.warning("Market conditions not suitable for trading")
return

# 3. Get candlestick data

df = self.get_candles(symbol, limit=100) # More data for better analysi
s
 if df.empty or len(df) < 50:
logger.warning("Insufficient candlestick data available")
return

# 4. Generate comprehensive trade signal

current_price = df['close'].iloc[-1]
signal = self.generate_trade_signal(df, current_price)

logger.info(f"Trade signal: {signal.signal_type} with confidence {signal

.confidence:.2f}")
logger.info(f"Signal reason: {signal.reason}")

if signal.signal_type != 'BUY':
logger.info("No strong buy signal detected, skipping trade")
return

# 5. Get order book and calculate entry

order_book = self.get_order_book(symbol)
if not order_book:
return

entry_price = order_book['best_bid'] + self.price_adjustment

quantity = self.adaptive_position_sizing(symbol, entry_price, signal.con
fidence)

if quantity <= 0:
logger.warning("Calculated quantity is 0 or negative")
return

# 6. Place buy order

buy_order = self.place_order(symbol, 'BUY', quantity, entry_price)
if not buy_order:
return

buy_order_id = buy_order['orderId']
patterns_str = ','.join(signal.patterns_detected)
self.log_trade('BUY_ORDER', symbol, entry_price, quantity, buy_order_id,
0, patterns_str)

# 7. Monitor order execution

start_time = time.time()
order_filled = False

while time.time() - start_time < self.order_timeout:

order_status = self.get_order_status(symbol, buy_order_id)

# Bitget order status mapping

if order_status.get('state') == 'filled':
filled_price = float(order_status['priceAvg'])
filled_qty = float(order_status['size'])

logger.info(f"Buy order filled: {filled_qty} at {filled_price}")

self.log_trade('BUY_FILLED', symbol, filled_price, filled_qty, b
uy_order_id)

# Track position with signal data

self.positions[symbol] = {
'side': 'BUY',
'quantity': filled_qty,
'entry_price': filled_price,
'entry_time': time.time(),
'signal': signal,
'indicators': signal.indicators
}

self.daily_trade_count += 1
self.last_trade_time = time.time()
order_filled = True
break

time.sleep(5)

# 8. Handle unfilled order

if not order_filled:
logger.warning("Buy order not filled within timeout, cancelling")
if self.cancel_order(symbol, buy_order_id):
 self.log_trade('BUY_CANCELLED', symbol, entry_price, quantity, b
uy_order_id)
return

# 9. Enhanced exit monitoring

position = self.positions[symbol]

while symbol in self.positions:

# Get fresh data for exit analysis
exit_df = self.get_candles(symbol, limit=50)
if not exit_df.empty:
exit_indicators = self.calculate_technical_indicators(exit_df)
else:
exit_indicators = position['indicators']

should_exit, reason = self.dynamic_exit_strategy(

symbol, position['entry_price'], position['side'], exit_indicato
rs
)

if should_exit:
logger.info(f"Exit condition met: {reason}")

# Place sell order

current_order_book = self.get_order_book(symbol)
exit_price = current_order_book['best_ask'] - self.price_adjustm
ent

sell_order = self.place_order(symbol, 'SELL', position['quantity

'], exit_price)

if sell_order:
sell_order_id = sell_order['orderId']

# Wait for sell order to fill

sell_start = time.time()
while time.time() - sell_start < 30:
sell_status = self.get_order_status(symbol, sell_order_i
d)

if sell_status.get('state') == 'filled':
exit_price_actual = float(sell_status['priceAvg'])
exit_qty = float(sell_status['size'])

# Calculate P&L; and update performance

pnl = (exit_price_actual - position['entry_price'])
* exit_qty
pnl_pct = pnl / (position['entry_price'] * exit_qty)

logger.info(f"Sell order filled: {exit_qty} at {exit

_price_actual}")
logger.info(f"Trade P&L;: ${pnl:.4f} ({pnl_pct:.4f}%
)")

self.log_trade('SELL_FILLED', symbol, exit_price_act

ual,
exit_qty, sell_order_id, pnl)

# Update performance monitor

self.performance_monitor.record_trade(
pnl, position['entry_price'], exit_price_actual,
exit_qty
)

# Remove position
del self.positions[symbol]
break

time.sleep(2)
else:
# Sell order timeout - use market order
logger.warning("Sell limit order timeout, placing market
order")
self.cancel_order(symbol, sell_order_id)

market_sell = self.place_order(symbol, 'SELL',

 position['quantity'], order
_type='market')
if market_sell:
logger.info("Market sell order placed")
# Estimate P&L; for market order
estimated_exit_price = current_order_book['best_ask'
]
estimated_pnl = (estimated_exit_price - position['en
try_price']) * position['quantity']
self.performance_monitor.record_trade(
estimated_pnl, position['entry_price'], estimate
d_exit_price, position['quantity']
)
del self.positions[symbol]
break

time.sleep(10) # Check exit conditions every 10 seconds

except Exception as e:
logger.error(f"Error in trade cycle: {e}")

def get_performance_summary(self) -> Dict:

"""Get current performance summary"""
try:
stats = self.performance_monitor.get_statistics()
current_balance = self.get_balance('USDT')
self.performance_monitor.update_balance(current_balance)

return {
'current_balance': current_balance,
'total_trades': stats['total_trades'],
'win_rate': stats['win_rate'],
'total_pnl': stats['total_pnl'],
'max_drawdown': stats['max_drawdown'],
'profit_factor': stats['profit_factor'],
'active_positions': len(self.positions),
'daily_trades': self.daily_trade_count
}
except Exception as e:
logger.error(f"Error getting performance summary: {e}")
return {}

def emergency_stop(self):
"""Emergency stop - cancel all orders and close positions"""
logger.warning("Emergency stop initiated!")

try:
# Cancel all active orders
for order_id in list(self.active_orders.keys()):
order = self.active_orders[order_id]
self.cancel_order(order['symbol'], order_id)

# Close all positions with market orders

for symbol, position in list(self.positions.items()):
logger.warning(f"Emergency closing position in {symbol}")
market_sell = self.place_order(
symbol, 'SELL', position['quantity'], order_type='market'
)
if market_sell:
self.log_trade('EMERGENCY_SELL', symbol, 0, position['quantity']
,
market_sell.get('orderId', ''), 0, 'EMERGENCY_STOP'
)
del self.positions[symbol]

logger.info("Emergency stop completed")

except Exception as e:
logger.error(f"Error during emergency stop: {e}")

def run_bot(self, symbol: str, profit_target: float = 0.002,

stop_loss: float = 0.003, max_position_percent: float = 0.05):
"""
Main bot execution loop

Args:
symbol: Trading pair (e.g., 'BTCUSDT')
 profit_target: Profit target as decimal (0.002 = 0.2%)
stop_loss: Stop loss as decimal (0.003 = 0.3%)
max_position_percent: Max position size as % of balance
"""
self.symbol = symbol
self.profit_target = profit_target
self.stop_loss = stop_loss
self.max_position_percent = max_position_percent

logger.info(f"Starting Bitget Scalp Bot for {symbol}")

logger.info(f"Profit Target: {profit_target:.2%}")
logger.info(f"Stop Loss: {stop_loss:.2%}")
logger.info(f"Max Position: {max_position_percent:.1%} of balance")

try:
while True:
self.execute_trade_cycle(symbol)

# Print performance summary every 10 cycles

if hasattr(self, '_cycle_count'):
self._cycle_count += 1
else:
self._cycle_count = 1

if self._cycle_count % 10 == 0:
summary = self.get_performance_summary()
if summary:
logger.info(f"Performance Summary: "
f"Trades: {summary.get('total_trades', 0)}, "
f"Win Rate: {summary.get('win_rate', 0):.1%}, "
f"P&L;: ${summary.get('total_pnl', 0):.2f}, "
f"Balance: ${summary.get('current_balance', 0):.2f
}")

# Wait before next cycle (avoid rate limits)

time.sleep(60) # 1 minute between cycles

except KeyboardInterrupt:
logger.info("Bot stopped by user")
self.emergency_stop()
except Exception as e:
logger.error(f"Bot error: {e}")
self.emergency_stop()
finally:
# Clean up any remaining open orders
for order_id in list(self.active_orders.keys()):
order = self.active_orders[order_id]
self.cancel_order(order['symbol'], order_id)
def main():
"""Main function to run the trading bot"""
print("Bitget Scalp Trading Bot")
print("=" * 50)

# Configuration
symbol = input("Enter trading pair (e.g., BTCUSDT): ").upper()

try:
profit_target = float(input("Enter profit target % (default 0.2): ") or "0.2
") / 100
stop_loss = float(input("Enter stop loss % (default 0.3): ") or "0.3") / 100
max_position = float(input("Enter max position % of balance (default 5): ")
or "5") / 100
except ValueError:
print("Invalid input, using default values")
profit_target = 0.002
stop_loss = 0.003
max_position = 0.05

# Create and run bot

try:
bot = BitgetTradingBot()
bot.run_bot(symbol, profit_target, stop_loss, max_position)
except Exception as e:
print(f"Failed to start bot: {e}")
print("\nMake sure you have:")
print("1. Created a .env file with BITGET_API_KEY, BITGET_API_SECRET, and BI
TGET_PASSPHRASE")
 print("2. Installed required packages: pip install pandas requests python-do
tenv")
if __name__ == "__main__":
main()