import gradio as gr from transformers import pipeline import hmac import hashlib import requests import time import os from datetime import datetime import plotly.express as px # Initialize hotdog classification pipeline with error handling hotdog_pipeline = None try: hotdog_pipeline = pipeline(task="image-classification", model="julien-c/hotdog-not-hotdog") except RuntimeError as e: print(f"Warning: Hotdog classifier failed to initialize: {e}. Image classification will be unavailable.") # Global state for API credentials (updated by user input) api_key = gr.State(os.getenv('BINGX_API_KEY', "")) api_secret = gr.State(os.getenv('BINGX_API_SECRET', "")) # Configuration API_BASE_URL = "https://open-api.bingx.com" # Generate Signature def generate_signature(api_secret, params_str): return hmac.new(api_secret.encode("utf-8"), params_str.encode("utf-8"), hashlib.sha256).hexdigest() # Parse Parameters def parse_params(params): sorted_keys = sorted(params.keys()) param_pairs = [f"{key}={params[key]}" for key in sorted_keys] params_str = "&".join(param_pairs) return params_str + "×tamp=" + str(int(time.time() * 1000)) # Fetch from BingX API def fetch_from_api(endpoint, params=None, api_key=api_key, api_secret=api_secret): if not api_key.value or not api_secret.value: raise ValueError("BingX API Key and Secret are not set. Please enter them in the Manage API Credentials section.") params = params or {} params['recvWindow'] = params.get('recvWindow', '5000') params_str = parse_params(params) signature = generate_signature(api_secret.value, params_str) url = f"{API_BASE_URL}{endpoint}?{params_str}&signature={signature}" try: response = requests.get(url, headers={'X-BX-APIKEY': api_key.value, 'Content-Type': 'application/json'}) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: raise Exception(f"API Error: {str(e)}") # Fetch Specific Data def fetch_balance(api_key=api_key, api_secret=api_secret): data = fetch_from_api('/openApi/swap/v3/user/balance', api_key=api_key, api_secret=api_secret) return data['data'][0]['walletBalance'] if data['data'] and data['data'][0]['asset'] == 'USDT' else 0 def fetch_open_positions(api_key=api_key, api_secret=api_secret): data = fetch_from_api('/openApi/swap/v2/user/positions', {'symbol': 'BTC-USDT'}, api_key=api_key, api_secret=api_secret) return data['data'] or [] def fetch_trade_history(api_key=api_key, api_secret=api_secret): data = fetch_from_api('/openApi/swap/v2/user/income', {'limit': 100}, api_key=api_key, api_secret=api_secret) return data['data'] or [] # Helper Functions def calculate_today_profit(trades): today = datetime.now().date() return sum(trade['income'] for trade in trades if datetime.fromtimestamp(trade['time'] / 1000).date() == today) def calculate_advanced_stats(trades): total_profit, total_loss, wins = 0, 0, 0 for trade in trades: pl = trade['income'] if pl > 0: total_profit += pl wins += 1 else: total_loss += abs(pl) profit_factor = total_loss and (total_profit / total_loss) or 0 win_rate = trades and (wins / len(trades) * 100) or 0 return {'profit_factor': profit_factor, 'win_rate': win_rate} def calculate_portfolio_allocation(positions): total_value = sum(pos['positionValue'] for pos in positions if 'positionValue' in pos) by_symbol = {} for pos in positions: by_symbol[pos['symbol']] = by_symbol.get(pos['symbol'], 0) + (pos.get('positionValue', 0)) labels = list(by_symbol.keys()) data = [val / total_value * 100 if total_value else 0 for val in by_symbol.values()] return {'labels': labels, 'data': data} # Update UI Functions def update_trading_table(positions, trades): table_rows = [] for pos in positions: table_rows.append(f"""
SahabatPipHijau
Premium Account
Total Balance
Open Trades
Today's Profit
Risk Exposure
Last synced: Just now