import time import datetime from tqdm import tqdm from loguru import logger from DrissionPage import ChromiumOptions, ChromiumPage from curl_cffi import requests from 交易.tools import send_dingtalk_message def is_bullish(c): return float(c['close']) > float(c['open']) def is_bearish(c): return float(c['close']) < float(c['open']) class WeexTransaction: """BitMart 自动交易脚本（优化版）""" def __init__(self, tge_id): self.tge_url = "http://127.0.0.1:50326" self.tge_id = tge_id self.tge_headers = { "Authorization": "Bearer asp_174003986c9b0799677c5b2c1adb76e402735d753bc91a91", "Content-Type": "application/json" } self.page: ChromiumPage | None = None self.session = requests.Session() self.cookies = {} self.tge_port = None # 当前仓位（-1 空 / 0 无 / 1 多） self.start = 0 # 最新 3 根 kline self.kline_1 = None self.kline_2 = None self.kline_3 = None # 当前信号 self.direction = None # 防止同一时段重复执行 self.time_start = None self.pbar = None # ------------------------------------------------------------- # ------------------------- 通用工具 ---------------------------- # ------------------------------------------------------------- def now_text(self): """格式化当前时间""" return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") def ding(self, msg, error=False): """统一钉钉消息格式""" prefix = "❌bitmart：" if error else "🔔bitmart：" send_dingtalk_message(f"{prefix}{self.now_text()}，{msg}") def get_half_hour_timestamp(self): """返回当前最近的整点或半点时间戳""" ts = time.time() dt = datetime.datetime.fromtimestamp(ts) target = dt.replace(minute=0, second=0, microsecond=0) if dt.minute < 30 \ else dt.replace(minute=30, second=0, microsecond=0) return int(target.timestamp()) # ------------------------------------------------------------- # ---------------------- TGE 浏览器相关 ------------------------ # ------------------------------------------------------------- def openBrowser(self): """打开 TGE 对应浏览器实例""" try: res = requests.post( f"{self.tge_url}/api/browser/start", json={"envId": self.tge_id}, headers=self.tge_headers ) self.tge_port = res.json()["data"]["port"] 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 # ------------------------------------------------------------- # -------------------- BitMart Token & 接口 -------------------- # ------------------------------------------------------------- def get_token(self): """从浏览器监听接口中提取 token 和 cookies""" tab = self.page.new_tab() tab.listen.start("ifcontract/shareguide/status") for _ in range(3): try: tab.get("https://derivatives.bitmart.com/zh-CN/futures/ETHUSDT") res = tab.listen.wait(timeout=15) # 请求头 + Cookies for i in res.request.headers: if ":" not in i: self.session.headers[i] = res.request.headers[i] for c in res.request.cookies: self.cookies[c["name"]] = c["value"] if "accessKey" in self.cookies: self.session.cookies.update(self.cookies) tab.close() return True except: time.sleep(1) tab.close() return False def get_account_balance(self): """获取可用余额""" params = {'account_type': '3', 'isServerCal': '1'} for _ in range(3): try: res = self.session.get( 'https://derivatives.bitmart.com/gw-api/contract-tiger/forward/v1/ifcontract/accounts', params=params ) return float(res.json()["data"]["accounts"][0]["available_balance"]) except: time.sleep(1) return None def get_position_status(self): """获取当前仓位：1 多、-1 空、0 无""" params = {'status': '1'} for _ in range(3): try: res = self.session.get( 'https://derivatives.bitmart.com/gw-api/contract-tiger/forward/v1/ifcontract/userPositions', params=params ).json() pos = res.get("data", {}).get("positions", []) if not pos: self.start = 0 else: ptype = pos[0].get("position_type") self.start = 1 if ptype == 1 else -1 return True except: time.sleep(1) return False # ------------------------------------------------------------- # --------------------------- K线相关 --------------------------- # ------------------------------------------------------------- def get_price(self): """获取最新 15 分钟 K 线""" params = { 'unit': '30', 'resolution': 'M', 'contractID': '2', 'offset': '1', 'endTime': str(int(time.time())), } for _ in range(3): try: res = self.session.get( 'https://contract-v2.bitmart.com/v1/ifcontract/quote/kline', params=params, timeout=15 ) datas = [] for k in res.json()["data"]: datas.append({ 'id': int(k["timestamp"]) - 1, 'open': float(k["open"]), 'high': float(k["high"]), 'low': float(k["low"]), 'close': float(k["close"]) }) return sorted(datas, key=lambda x: x["id"]) except: time.sleep(1) return None def check_signal(self, prev, curr): """包住形态判定""" p_open, p_close = prev["open"], prev["close"] c_open, c_close = curr["open"], curr["close"] # 前跌后涨包住 -> 多 if is_bullish(curr) and is_bearish(prev) and c_close >= p_open: return "long" # 前涨后跌包住 -> 空 if is_bearish(curr) and is_bullish(prev) and c_close <= p_open: return "short" return None # ------------------------------------------------------------- # ---------------------- 下单逻辑封装 --------------------------- # ------------------------------------------------------------- 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 do_order(self): """执行交易动作（开仓/反手）""" # 选择市价 self.click_safe('x:(//button[normalize-space(text())="市价"])') # 余额 balance = None for i in range(3): balance = self.get_account_balance() if not balance: self.ding("获取可用余额失败！", error=True) return amount = balance / 100 self.page.ele('x://*[@id="size_0"]').input(amount) # ---- 开仓逻辑 ---- if self.direction == "long": if self.start == 0: self.ding(f"信号：多，开多：{amount}") self.click_safe('x://span[normalize-space(text()) ="买入/做多"]') self.start = 1 elif self.start == -1: self.ding(f"信号：多，反手空转多：{amount}") self.click_safe('x://span[normalize-space(text()) ="市价"]') time.sleep(2) self.click_safe('x://span[normalize-space(text()) ="买入/做多"]') self.start = 1 elif self.direction == "short": if self.start == 0: self.ding(f"信号：空，开空：{amount}") self.click_safe('x://span[normalize-space(text()) ="卖出/做空"]') self.start = -1 elif self.start == 1: self.ding(f"信号：空，反手多转空：{amount}") self.click_safe('x://span[normalize-space(text()) ="市价"]') time.sleep(2) self.click_safe('x://span[normalize-space(text()) ="卖出/做空"]') self.start = -1 # ------------------------------------------------------------- # ----------------------------- 主流程 --------------------------- # ------------------------------------------------------------- def action(self): # 1. 打开浏览器 if not self.openBrowser(): self.ding("打开 TGE 失败！", error=True) return logger.info("TGE 端口获取成功") # 2. 接管浏览器 if not self.take_over_browser(): self.ding("接管浏览器失败！", error=True) return logger.info("浏览器接管成功") self.close_extra_tabs() self.page.get("https://derivatives.bitmart.com/zh-CN/futures/ETHUSDT") self.pbar = tqdm(total=30, desc="等待半小时周期", ncols=80) while True: now = time.localtime() minute = now.tm_min # 更新进度条 self.pbar.n = minute if minute < 30 else minute - 30 self.pbar.refresh() # 必须是整点或半点及前 5 分钟 # if minute not in [0, 1, 2, 3, 4, 5, 30, 31, 32, 33, 34, 35]: # time.sleep(8) # return # 时间重复跳过 if self.time_start == self.get_half_hour_timestamp(): time.sleep(5) continue # ---- 获取 Token ---- if not self.get_token(): self.ding("获取 token 失败！", error=True) return logger.info("Token 获取成功") # ---- 获取价格 ---- kdatas = self.get_price() if not kdatas: self.ding("获取价格失败！", error=True) return self.kline_1, self.kline_2, self.kline_3 = kdatas[-3:] if int(self.kline_3["id"]) != self.get_half_hour_timestamp(): continue logger.success("K线获取成功") self.time_start = self.get_half_hour_timestamp() # ---- 刷新页面 ---- self.page.get("https://derivatives.bitmart.com/zh-CN/futures/ETHUSDT") for i in range(3): # ---- 获取仓位 ---- if not self.get_position_status(): self.ding("获取仓位失败！", error=True) continue if self.start: break # ---- 止损平仓 ---- try: if self.start == 1 and is_bearish(self.kline_1) and is_bearish(self.kline_2): self.ding("两根大阴线，平多") self.click_safe('x://span[normalize-space(text()) ="市价"]') self.start = 0 elif self.start == -1 and is_bullish(self.kline_1) and is_bullish(self.kline_2): self.ding("两根大阳线，平空") self.click_safe('x://span[normalize-space(text()) ="市价"]') self.start = 0 except: self.ding("止损平仓错误！", error=True) # continue return # ---- 生成新信号 ---- self.direction = self.check_signal(prev=self.kline_1, curr=self.kline_2) # ---- 执行交易 ---- if self.direction: try: self.do_order() except: self.ding("下单失败！", error=True) # ---- 周期结束消息 ---- self.pbar.reset() self.ding( f"持仓：{'无' if self.start == 0 else ('多' if self.start == 1 else '空')}，" f"信号：{'无' if not self.direction else ('多' if self.direction == 'long' else '空')}" ) if __name__ == '__main__': while True: try: WeexTransaction(tge_id=196495).action() except: time.sleep(5)