import pandas as pd
import ccxt
import time
import logging
from datetime import datetime, timezone, timedelta
from dotenv import load_dotenv
import os
# Setup logging
[Link](
level=[Link],
format='%(asctime)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
logger = [Link](__name__)
# Load environment variables
load_dotenv()
# Initialize exchange
exchange = [Link]({
'apiKey': [Link]('MEXC_API_KEY'),
'secret': [Link]('MEXC_SECRET_KEY'),
'enableRateLimit': True,
'options': {'defaultType': 'spot'}
})
# Configuration - 1 hour TIMEFRAME
SYMBOL = 'ORDI_USDT' # Trading pair
TIMEFRAME = '1h' # 1-Hour timeframe
MIN_BARS = 72 # 3 days of data (72 hours)
INVESTMENT_AMOUNT = 10 # USD per trade
def fetch_ohlcv(symbol, timeframe, limit=100):
"""Fetch OHLCV data with retry logic"""
attempts = 0
while attempts < 3:
try:
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
df = [Link](ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
return df.set_index('timestamp')
except Exception as e:
[Link](f"Attempt {attempts+1}/3 failed: {str(e)[:100]}")
[Link](5)
attempts += 1
raise ConnectionError("Failed to fetch OHLCV data")
def compute_indicators(df):
"""Calculate technical indicators for 1h timeframe"""
# 20-period SMA
df['SMA20'] = df['close'].rolling(window=20).mean()
# 50-period EMA
df['EMA50'] = df['close'].ewm(span=50, adjust=False).mean()
# RSI (14-period)
delta = df['close'].diff()
gain = [Link](delta > 0, 0)
loss = -[Link](delta < 0, 0)
avg_gain = [Link](14).mean()
avg_loss = [Link](14).mean()
rs = avg_gain / avg_loss
df['RSI'] = 100 - (100 / (1 + rs))
return [Link]()
def generate_signal(df):
"""Generate trading signal for 1h candles"""
if len(df) < 20:
return "HOLD"
current = [Link][-1]
prev = [Link][-2]
# Trend + Momentum strategy
trend_up = current['close'] > current['EMA50']
momentum_up = current['RSI'] > 50 and current['RSI'] > prev['RSI']
if trend_up and momentum_up:
return "CALL"
elif not trend_up and not momentum_up:
return "PUT"
return "HOLD"
def execute_trade(signal):
"""Execute spot trade (simulated)"""
if signal == "HOLD":
return
[Link](f"Executing {signal} trade on {SYMBOL}")
try:
# Replace with actual MEXC API call:
# order = exchange.create_order(SYMBOL, 'market', [Link](),
INVESTMENT_AMOUNT)
[Link](f"✅ Simulated {signal} order for ${INVESTMENT_AMOUNT}")
except Exception as e:
[Link](f"Trade failed: {e}")
def main():
[Link]("="*50)
[Link](f"🚀 MEXC Trading Bot - 1 Hour Timeframe")
[Link](f"📌 Pair: {SYMBOL} | Minimum Bars: {MIN_BARS}")
[Link]("="*50)
# Test connection
try:
ticker = exchange.fetch_ticker(SYMBOL)
[Link](f"✅ Connected | Current Price: ${ticker['last']}")
except Exception as e:
[Link](f"Connection failed: {e}")
return
# Data collection
[Link]("Loading initial data...")
df = fetch_ohlcv(SYMBOL, TIMEFRAME, limit=MIN_BARS)
while True:
try:
# Update data
new_data = fetch_ohlcv(SYMBOL, TIMEFRAME, limit=5)
df = [Link]([df, new_data])
df = df[~[Link](keep='last')]
df = df.sort_index().tail(MIN_BARS)
# Calculate indicators
df = compute_indicators(df)
# Generate and log signal
signal = generate_signal(df)
[Link](f"\n{'='*30}")
[Link](f"🕒 {[Link]([Link]).strftime('%Y-%m-%d %H:%M:%S')} UTC")
[Link](f"💰 Price: ${df['close'].iloc[-1]:.2f}")
[Link](f"📊 Indicators: SMA20=${df['SMA20'].iloc[-1]:.2f} | EMA50=$
{df['EMA50'].iloc[-1]:.2f} | RSI={df['RSI'].iloc[-1]:.2f}")
[Link](f"📢 Signal: {signal}")
[Link](f"{'='*30}\n")
# Execute trade
execute_trade(signal)
# Wait for next candle (1 hour)
now = [Link]([Link])
next_candle = (now + timedelta(hours=1)).replace(minute=0, second=0, microsecond=0)
sleep_seconds = (next_candle - now).total_seconds()
if sleep_seconds > 0:
[Link](f"⏳ Next analysis at: {next_candle.strftime('%H:%M:%S')} UTC (in
{sleep_seconds/60:.1f} minutes)")
[Link](sleep_seconds)
except KeyboardInterrupt:
[Link]("Bot stopped by user")
break
except Exception as e:
[Link](f"Error: {e} | Retrying in 30s")
[Link](30)
if __name__ == "__main__":
main()