crypto.ai/cryptoai/api/binance_api.py

import os
import pandas as pd
from typing import List, Dict, Any, Optional
from binance.client import Client
from binance.exceptions import BinanceAPIException
import datetime
import time

class BinanceAPI:
    """Binance API交互类，用于获取市场数据和执行交易"""
    
    def __init__(self, api_key: str, api_secret: str, test_mode: bool = True):
        """
        初始化Binance API客户端
        
        Args:
            api_key: Binance API密钥
            api_secret: Binance API密钥
            test_mode: 是否使用测试模式（不执行实际交易）
        """
        self.api_key = api_key
        self.api_secret = api_secret
        self.test_mode = test_mode
        self.client = Client(api_key, api_secret)
        
        # 验证API连接
        try:
            self.client.get_account()
            print("Binance API连接成功")
        except BinanceAPIException as e:
            print(f"Binance API连接失败: {e}")

    def get_all_symbols(self) -> List[str]:
        """
        获取所有合约交易对
        
        Returns:
            List[str]: 所有合约交易对列表
        """
        symbols = self.client.get_exchange_info()['symbols']

        result = []
        # 选择USDT交易对以及 status 为 TRADING 的交易对
        for symbol in symbols:
            if symbol['quoteAsset'] == 'USDT' and symbol['status'] == 'TRADING':
                result.append(symbol['symbol'])

        return result


    def get_historical_klines(self, symbol: str, interval: str, start_str: str, end_str: Optional[str] = None) -> pd.DataFrame:
        """
        获取历史K线数据
        
        Args:
            symbol: 交易对符号，例如 'BTCUSDT'
            interval: K线间隔，例如 '1h', '1d'
            start_str: 开始时间，例如 '1 day ago UTC'
            end_str: 结束时间，例如 'now UTC'
            
        Returns:
            包含历史数据的DataFrame
        """
        try:
            klines = self.client.get_historical_klines(
                symbol=symbol,
                interval=interval,
                start_str=start_str,
                end_str=end_str
            )
            
            # 转换为DataFrame
            df = pd.DataFrame(klines, columns=[
                'timestamp', 'open', 'high', 'low', 'close', 'volume',
                'close_time', 'quote_asset_volume', 'number_of_trades',
                'taker_buy_base_asset_volume', 'taker_buy_quote_asset_volume', 'ignore'
            ])
            
            # 转换数据类型
            df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms', utc=True).map(lambda x: x.tz_convert('Asia/Shanghai'))
            df['close_time'] = pd.to_datetime(df['close_time'], unit='ms', utc=True).map(lambda x: x.tz_convert('Asia/Shanghai'))
            
            for col in ['open', 'high', 'low', 'close', 'volume', 'quote_asset_volume',
                        'taker_buy_base_asset_volume', 'taker_buy_quote_asset_volume']:
                df[col] = df[col].astype(float)
            
            return df
        
        except BinanceAPIException as e:
            print(f"获取历史K线数据时出错: {e}")
            return pd.DataFrame()
    
    def get_recent_trades(self, symbol: str, limit: int = 500) -> pd.DataFrame:
        """
        获取最近的交易
        
        Args:
            symbol: 交易对符号，例如 'BTCUSDT'
            limit: 返回的交易数量
            
        Returns:
            包含最近交易的DataFrame
        """
        try:
            trades = self.client.get_recent_trades(symbol=symbol, limit=limit)
            return pd.DataFrame(trades)
        except BinanceAPIException as e:
            print(f"获取最近交易时出错: {e}")
            return pd.DataFrame()
    
    def get_order_book(self, symbol: str, limit: int = 100) -> Dict[str, Any]:
        """
        获取订单簿
        
        Args:
            symbol: 交易对符号，例如 'BTCUSDT'
            limit: 返回的订单数量
            
        Returns:
            订单簿数据字典
        """
        try:
            return self.client.get_order_book(symbol=symbol, limit=limit)
        except BinanceAPIException as e:
            print(f"获取订单簿时出错: {e}")
            return {}
    
    def get_ticker(self, symbol: str = None) -> Dict[str, Any]:
        """
        获取当前价格
        
        Args:
            symbol: 交易对符号，例如 'BTCUSDT'，如果为None，则获取所有交易对
            
        Returns:
            当前价格数据
        """
        try:
            if symbol:
                return self.client.get_symbol_ticker(symbol=symbol)
            else:
                return self.client.get_all_tickers()
        except BinanceAPIException as e:
            print(f"获取当前价格时出错: {e}")
            return {}
    
    def place_order(self, symbol: str, side: str, type: str, quantity: float, price: float = None) -> Dict[str, Any]:
        """
        下单
        
        Args:
            symbol: 交易对符号，例如 'BTCUSDT'
            side: 买入或卖出，'BUY' 或 'SELL'
            type: 订单类型，例如 'LIMIT', 'MARKET'
            quantity: 数量
            price: 价格（对于限价单）
            
        Returns:
            订单响应
        """
        if self.test_mode:
            print(f"测试模式: 模拟下单 {side} {quantity} {symbol} @ {price if price else 'MARKET'}")
            return {
                "symbol": symbol,
                "orderId": "test_order_id",
                "status": "TEST",
                "side": side,
                "type": type,
                "quantity": quantity,
                "price": price
            }
        
        try:
            if type == 'LIMIT':
                return self.client.create_order(
                    symbol=symbol,
                    side=side,
                    type=type,
                    timeInForce='GTC',
                    quantity=quantity,
                    price=price
                )
            elif type == 'MARKET':
                return self.client.create_order(
                    symbol=symbol,
                    side=side,
                    type=type,
                    quantity=quantity
                )
        except BinanceAPIException as e:
            print(f"下单时出错: {e}")
            return {}