import time from tqdm import tqdm from loguru import logger from bit_tools import openBrowser from DrissionPage import ChromiumPage from DrissionPage import ChromiumOptions from bitmart.api_contract import APIContract class BitmartFuturesTransaction: def __init__(self, bit_id): self.page: ChromiumPage | None = None self.api_key = "a0fb7b98464fd9bcce67e7c519d58ec10d0c38a8" self.secret_key = "4eaeba78e77aeaab1c2027f846a276d164f264a44c2c1bb1c5f3be50c8de1ca5" self.memo = "合约交易" self.contract_symbol = "ETHUSDT" self.contractAPI = APIContract(self.api_key, self.secret_key, self.memo, timeout=(5, 15)) self.start = 0 # 持仓状态: -1 空, 0 无, 1 多 self.direction = None self.pbar = tqdm(total=30, desc="等待K线", ncols=80) self.last_kline_time = None self.leverage = "100" # 高杠杆（全仓模式下可开更大仓位） self.open_type = "cross" # 全仓模式（你的“成本开仓”需求） self.risk_percent = 0.01 # 每次开仓使用可用余额的 1% self.open_avg_price = None # 开仓价格 self.current_amount = None # 持仓量 self.bit_id = bit_id def get_klines(self): """获取最近3根30分钟K线（step=30）""" try: end_time = int(time.time()) # 获取足够多的条目确保有最新3根 response = self.contractAPI.get_kline( contract_symbol=self.contract_symbol, step=30, # 30分钟 start_time=end_time - 3600 * 10, # 取最近10小时 end_time=end_time )[0]["data"] # 每根: [timestamp, open, high, low, close, volume] formatted = [] for k in response: formatted.append({ 'id': int(k["timestamp"]), 'open': float(k["open_price"]), 'high': float(k["high_price"]), 'low': float(k["low_price"]), 'close': float(k["close_price"]) }) formatted.sort(key=lambda x: x['id']) return formatted # 最近3根: kline_1 (最老), kline_2, kline_3 (最新) except Exception as e: logger.error(f"获取K线异常: {e}") self.ding(error=True, msg="获取K线异常") return None def get_current_price(self): """获取当前最新价格，用于计算张数""" try: end_time = int(time.time()) response = self.contractAPI.get_kline( contract_symbol=self.contract_symbol, step=1, # 1分钟 start_time=end_time - 3600 * 3, # 取最近10小时 end_time=end_time )[0] if response['code'] == 1000: return float(response['data'][-1]["close_price"]) return None except Exception as e: logger.error(f"获取价格异常: {e}") return None def get_available_balance(self): """获取合约账户可用USDT余额""" try: response = self.contractAPI.get_assets_detail()[0] if response['code'] == 1000: data = response['data'] if isinstance(data, dict): return float(data.get('available_balance', 0)) elif isinstance(data, list): for asset in data: if asset.get('currency') == 'USDT': return float(asset.get('available_balance', 0)) return None except Exception as e: logger.error(f"余额查询异常: {e}") return None # 获取当前持仓方向 def get_position_status(self): """获取当前持仓方向""" try: response = self.contractAPI.get_position(contract_symbol=self.contract_symbol)[0] if response['code'] == 1000: positions = response['data'] if not positions: self.start = 0 return True self.start = 1 if positions[0]['position_type'] == 1 else -1 self.open_avg_price = positions[0]['open_avg_price'] self.current_amount = positions[0]['current_amount'] self.position_cross = positions[0]["position_cross"] return True else: return False except Exception as e: logger.error(f"持仓查询异常: {e}") return False # 设置杠杆和全仓 def set_leverage(self): """程序启动时设置全仓 + 高杠杆""" try: response = self.contractAPI.post_submit_leverage( contract_symbol=self.contract_symbol, leverage=self.leverage, open_type=self.open_type )[0] if response['code'] == 1000: logger.success(f"全仓模式 + {self.leverage}x 杠杆设置成功") return True else: logger.error(f"杠杆设置失败: {response}") return False except Exception as e: logger.error(f"设置杠杆异常: {e}") return False def openBrowser(self): """打开 TGE 对应浏览器实例""" try: bit_port = openBrowser(id=self.bit_id) co = ChromiumOptions() co.set_local_port(port=bit_port) self.page = ChromiumPage(addr_or_opts=co) return True except: return False def take_over_browser(self): """接管浏览器""" try: co = ChromiumOptions() co.set_local_port(self.tge_port) self.page = ChromiumPage(addr_or_opts=co) self.page.set.window.max() return True except: return False def close_extra_tabs(self): """关闭多余 tab""" try: for idx, tab in enumerate(self.page.get_tabs()): if idx > 0: tab.close() return True except: return False def click_safe(self, xpath, sleep=0.5): """安全点击""" try: ele = self.page.ele(xpath) if not ele: return False ele.scroll.to_see(center=True) time.sleep(sleep) ele.click() return True except: return False def 全平仓(self): self.click_safe('x://span[normalize-space(text()) ="市价"]') def 平一半多仓(self): self.click_safe('x://button[normalize-space(text()) ="平仓"]') self.click_safe('x://*[@id="futureTradeForm"]/div[5]/div[3]/div[3]/span[3]') self.click_safe('x://span[normalize-space(text()) ="卖出/平多"]') def 平一半空仓(self): self.click_safe('x://button[normalize-space(text()) ="平仓"]') self.click_safe('x://*[@id="futureTradeForm"]/div[5]/div[3]/div[3]/span[3]') self.click_safe('x://span[normalize-space(text()) ="买入/平空"]') def 开单(self, marketPriceLongOrder=0, limitPriceShortOrder=0, size=None, price=None): """ marketPriceLongOrder 市价最多或者做空，1是做多，-1是做空 limitPriceShortOrder 限价最多或者做空 """ if marketPriceLongOrder == -1: self.click_safe('x://button[normalize-space(text()) ="市价"]') self.page.ele('x://*[@id="size_0"]').input(size) self.click_safe('x://span[normalize-space(text()) ="卖出/做空"]') elif marketPriceLongOrder == 1: self.click_safe('x://button[normalize-space(text()) ="市价"]') self.page.ele('x://*[@id="size_0"]').input(size) self.click_safe('x://span[normalize-space(text()) ="买入/做多"]') if limitPriceShortOrder == -1: self.click_safe('x://button[normalize-space(text()) ="限价"]') self.page.ele('x://*[@id="price_0"]').input(vals=price, clear=True) time.sleep(1) self.page.ele('x://*[@id="size_0"]').input(1) self.click_safe('x://span[normalize-space(text()) ="卖出/做空"]') elif limitPriceShortOrder == 1: self.click_safe('x://button[normalize-space(text()) ="限价"]') self.page.ele('x://*[@id="price_0"]').input(vals=price, clear=True) time.sleep(1) self.page.ele('x://*[@id="size_0"]').input(1) self.click_safe('x://span[normalize-space(text()) ="买入/做多"]') def ding(self, text, error=False): logger.info(text) def action(self): # 启动时设置全仓高杠杆 if not self.set_leverage(): logger.error("杠杆设置失败，程序继续运行但可能下单失败") return # 1. 打开浏览器 if not self.openBrowser(): self.ding("打开 TGE 失败！", error=True) return logger.info("TGE 端口获取成功") self.get_klines() # self.close_extra_tabs() # self.page.get("https://derivatives.bitmart.com/zh-CN/futures/ETHUSDT") # self.click_safe('x://button[normalize-space(text()) ="市价"]') # self.click_safe('x://button[normalize-space(text()) ="限价"]') # # self.page.ele('x://*[@id="price_0"]').input(vals=3000, clear=True) # self.page.ele('x://*[@id="size_0"]').input(1) # self.click_safe('x://span[normalize-space(text()) ="买入/做多"]') # self.click_safe('x://span[normalize-space(text()) ="卖出/做空"]') self.click_safe('x://button[normalize-space(text()) ="开仓"]') self.click_safe('x://button[normalize-space(text()) ="平仓"]') self.click_safe('x://span[normalize-space(text()) ="买入/平空"]') self.click_safe('x://span[normalize-space(text()) ="卖出/平多"]') self.click_safe('x://*[@id="futureTradeForm"]/div[5]/div[3]/div[3]/span[3]') if __name__ == '__main__': BitmartFuturesTransaction(bit_id="f2320f57e24c45529a009e1541e25961").action()