import ccxt import time import os import pandas as pd import gradio as gr from datetime import datetime # --- CONFIGURATION FROM ENV --- # Users should set these in Hugging Face Space Secrets API_KEY = os.environ.get('BINANCE_API_KEY', '') API_SECRET = os.environ.get('BINANCE_API_SECRET', '') USE_TESTNET = os.environ.get('USE_TESTNET', 'True').lower() == 'true' BRIDGE_CURRENCY = 'USDT' START_AMOUNT = 100 FEE = 0.001 MIN_PROFIT_PERCENT = -0.5 MAX_CYCLE_LEN = 3 # Global exchange instance exchange = None all_cycles = [] execution_logs = [] trade_history = [] # List to store completed trades and profits def init_exchange(): global exchange, all_cycles if exchange is not None: return exchange exchange_class = getattr(ccxt, 'binance') exchange = exchange_class({ 'apiKey': API_KEY, 'secret': API_SECRET, 'enableRateLimit': True, }) if USE_TESTNET: exchange.set_sandbox_mode(True) markets = exchange.load_markets() all_cycles = find_cycles(markets, MAX_CYCLE_LEN) return exchange def find_cycles(markets, max_len=3): graph = {} for symbol, market in markets.items(): if not market['active'] or '/' not in symbol: continue base, quote = market['base'], market['quote'] if base not in graph: graph[base] = [] if quote not in graph: graph[quote] = [] graph[base].append((quote, symbol, 'sell')) graph[quote].append((base, symbol, 'buy')) found_cycles = [] def dfs(curr, path, symbols, sides, visited): if len(path) > max_len: return if curr not in graph: return for neighbor, symbol, side in graph[curr]: if neighbor == path[0] and len(path) >= 3: found_cycles.append({'path': path + [neighbor], 'symbols': symbols + [symbol], 'sides': sides + [side]}) elif neighbor not in visited and len(path) < max_len: dfs(neighbor, path + [neighbor], symbols + [symbol], sides + [side], visited | {neighbor}) dfs(BRIDGE_CURRENCY, [BRIDGE_CURRENCY], [], [], {BRIDGE_CURRENCY}) return found_cycles def execute_cycle(exchange, cycle_data, start_amount): """Executes trades for the given cycle""" global execution_logs, trade_history path_str = " -> ".join(cycle_data['path']) log_entry = f"[{datetime.now().strftime('%H:%M:%S')}] Starting execution: {path_str}" execution_logs.append(log_entry) current_amount = start_amount for i in range(len(cycle_data['symbols'])): symbol = cycle_data['symbols'][i] side = cycle_data['sides'][i] try: ticker = exchange.fetch_ticker(symbol) price = ticker['ask'] if side == 'buy' else ticker['bid'] if side == 'buy': amount = current_amount / price else: amount = current_amount amount = float(exchange.amount_to_precision(symbol, amount)) order = exchange.create_market_order(symbol, side, amount) if side == 'buy': current_amount = order['filled'] else: current_amount = order['cost'] execution_logs.append(f" - {side.upper()} {symbol} filled at {price}") except Exception as e: execution_logs.append(f" - ERROR: {str(e)}") return False profit_abs = current_amount - start_amount profit_pct = (profit_abs / start_amount) * 100 # Save to history trade_history.append({ 'timestamp': datetime.now().isoformat(), 'path': path_str, 'start_amount': start_amount, 'end_amount': round(current_amount, 6), 'profit_usdt': round(profit_abs, 6), 'profit_pct': round(profit_pct, 4) }) execution_logs.append(f" - FINISHED: Final amount {current_amount} {BRIDGE_CURRENCY} (Profit: {profit_pct:.4f}%)") return True def scan_once(auto_trade=False): global exchange, all_cycles, execution_logs, trade_history if exchange is None: init_exchange() try: tickers = exchange.fetch_tickers() usdt_eur_rate = 0.92 if 'USDT/EUR' in tickers: usdt_eur_rate = tickers['USDT/EUR']['last'] elif 'EUR/USDT' in tickers: usdt_eur_rate = 1 / tickers['EUR/USDT']['last'] opportunities = [] for c in all_cycles: try: kapital = START_AMOUNT steps = len(c['symbols']) valid = True for i in range(steps): symbol = c['symbols'][i] side = c['sides'][i] if symbol not in tickers or not tickers[symbol]['ask'] or not tickers[symbol]['bid']: valid = False; break price = tickers[symbol]['ask'] if side == 'buy' else tickers[symbol]['bid'] if side == 'buy': kapital = (kapital / price) * (1 - FEE) else: kapital = (kapital * price) * (1 - FEE) if not valid: continue profit_pct = ((kapital - START_AMOUNT) / START_AMOUNT) * 100 profit_eur = (kapital - START_AMOUNT) * usdt_eur_rate if profit_pct > MIN_PROFIT_PERCENT: path_str = " -> ".join(c['path']) opportunities.append({ 'Path': path_str, 'Profit %': round(profit_pct, 4), 'Profit EUR': round(profit_eur, 4), 'raw_data': c }) except: continue opportunities.sort(key=lambda x: x['Profit %'], reverse=True) # Auto-trade logic if auto_trade and opportunities and opportunities[0]['Profit %'] > 0: best = opportunities[0] execute_cycle(exchange, best['raw_data'], START_AMOUNT) # Prepare DF for display display_opps = [] for o in opportunities[:20]: display_opps.append({ 'Path': o['Path'], 'Profit %': o['Profit %'], 'Profit EUR': o['Profit EUR'] }) df = pd.DataFrame(display_opps) # Total profit calculation total_profit = sum(t['profit_usdt'] for t in trade_history) trade_count = len(trade_history) status = f"Last scan: {datetime.now().strftime('%H:%M:%S')} | Total Profit: {total_profit:.4f} USDT ({trade_count} trades)" logs_text = "\n".join(execution_logs[-15:]) # Prepare history DF history_df = pd.DataFrame(trade_history[-10:]) if trade_history else pd.DataFrame() return df, status, logs_text, history_df except Exception as e: return pd.DataFrame(), f"Error: {str(e)}", "\n".join(execution_logs[-15:]), pd.DataFrame() print(">>> BUILDING GRADIO UI...") with gr.Blocks(title="Crypto Arbitrage Scanner") as demo: gr.Markdown("# 🚀 Binance Triangular Arbitrage Scanner") gr.Markdown(f"Scanning cycles starting with **{BRIDGE_CURRENCY}**. Testnet: **{USE_TESTNET}**") with gr.Row(): status_text = gr.Textbox(label="Status & Stats", value="Initializing...", interactive=False) auto_trade_toggle = gr.Checkbox(label="Enable Auto-Trade (Only if Profit > 0%)", value=False) refresh_btn = gr.Button("Manual Scan", variant="primary") with gr.Row(): with gr.Column(scale=2): gr.Markdown("### 🔍 Live Opportunities") results_df = gr.Dataframe( headers=["Path", "Profit %", "Profit EUR"], datatype=["str", "number", "number"], label="Top 20 Opportunities" ) gr.Markdown("### 📜 Trade History (Last 10)") history_table = gr.Dataframe(label="Recent Profits") with gr.Column(scale=1): gr.Markdown("### 💻 Execution Logs") log_box = gr.TextArea(label="Bot Logs", interactive=False, lines=25) gr.Markdown("---") gr.Markdown("💡 **API Endpoint:** This space exposes a Gradio API. You can check it via the 'Use via API' link at the bottom.") # --- API ENDPOINT (Gradio Native) --- def get_profits_api(): total_profit = sum(t['profit_usdt'] for t in trade_history) return { "total_profit_usdt": round(total_profit, 6), "trade_count": len(trade_history), "history": trade_history[-50:] } api_btn = gr.Button("Get Profits JSON", visible=False) api_btn.click(fn=get_profits_api, api_name="profits") # Auto-refresh logic timer = gr.Timer(value=2) timer.tick(fn=scan_once, inputs=[auto_trade_toggle], outputs=[results_df, status_text, log_box, history_table]) refresh_btn.click(fn=scan_once, inputs=[auto_trade_toggle], outputs=[results_df, status_text, log_box, history_table]) print(">>> BOT STARTING ON HUGGING FACE...") demo.queue().launch()