import time import datetime from tqdm import * from loguru import * from DrissionPage import * 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: def __init__(self, tge_id): self.tge_port = None # tge浏览器使用端口 self.tge_id = tge_id # tge id self.tge_url = "http://127.0.0.1:50326" # tge本地服务url self.tge_headers = { "Authorization": f"Bearer asp_174003986c9b0799677c5b2c1adb76e402735d753bc91a91", "Content-Type": "application/json" } self.page = None # 浏览器对象 self.start = 0 # 持仓状态 -1:做空，0：维持仓，1：做多 self.kline_1 = None # 0：跌，1：涨 self.kline_2 = None # 0：跌，1：涨 self.kline_1 = self.kline_2 = self.kline_3 = None self.direction = None # 信号类型 self.pbar = None # 进度条对象 self.session = requests.Session() # 接口请求对象 self.headers = {} self.cookies = {} def openBrowser(self, ): # 直接指定ID打开窗口，也可以使用 createBrowser 方法返回的ID try: response = requests.post( f"{self.tge_url}/api/browser/start", json={"envId": self.tge_id}, headers=self.tge_headers ) self.tge_port = response.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 is_bullish(self, c): # 阳线 return float(c['close']) > float(c['open']) def is_bearish(self, c): # 阴线 return float(c['close']) < float(c['open']) def check_signal(self, prev, curr): """ 包住形态信号判定（仅15分钟K线）： - 前跌后涨包住 -> 做多 - 前涨后跌包住 -> 做空 """ p_open, p_close = float(prev['open']), float(prev['close']) c_open, c_close = float(curr['open']), float(curr['close']) # 前跌后涨包住 -> 做多 # if is_bullish(curr) and is_bearish(prev) and int(c_open) - 3 <= int(p_close) and int(c_close) >= int(p_open): # return "long", "bear_bull_engulf" if is_bullish(curr) and is_bearish(prev) and int(c_close) >= int(p_open): return "long", "bear_bull_engulf" # 前涨后跌包住 -> 做空 # if is_bearish(curr) and is_bullish(prev) and int(c_open) + 3 >= int(p_close) and int(c_close) <= int(p_open): # return "short", "bull_bear_engulf" if is_bearish(curr) and is_bullish(prev) and int(c_close) <= int(p_open): return "short", "bull_bear_engulf" return None, None def get_price(self): # 方法 1：最常用（推荐） now_ts = int(time.time()) # 当前毫秒时间戳 ten_days_ago_ts = int((time.time() - 86400 * 10)) # 10天前毫秒时间戳 params = { 'symbol': 'ETH-USDT', 'period': '30min', 'start': str(ten_days_ago_ts), 'end': str(now_ts), } datas = [] for i in range(3): logger.info(f"获取最新数据：{i + 1}次。。。") try: response = self.session.get('https://capi.websea.com/webApi/market/getKline', params=params, timeout=5) for data in response.json()["result"]['data']: insert_data = { 'id': int(data["id"]), 'open': float(data["open"]), 'high': float(data["high"]), 'low': float(data["low"]), 'close': float(data["close"]) } datas.append(insert_data) return datas except: time.sleep(1) return datas def to_do_page(self): # self.page.get("https://www.weeaxs.site/zh-CN/futures/demo-trading/ETH-SUSDT") self.page.ele('x://*[normalize-space(text())= "市价"]', timeout=15).click() time.sleep(1) num = self.get_num() if num: logger.info("获取可用余额成功！！！") else: logger.error("获取可用余额失败！！！") self.send_dingtalk_message("获取可用余额失败！！！", type=0) return self.page.ele('x://*[@id="amountInput"]').input(float(num) / 100) time.sleep(1) if self.direction == "long" and not self.start: logger.success(f"{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，开多") self.send_dingtalk_message(f"信号：{self.direction}，开多，开仓金额：{float(num) / 100}") self.page.ele('x://*[normalize-space(text())= "买入/做多"]').click() self.start = 1 elif self.direction == "short" and not self.start: logger.success(f"{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，开空") self.send_dingtalk_message(f"信号：{self.direction}，开空，开仓金额：{float(num) / 100}") self.page.ele('x://*[normalize-space(text())= "卖出/做空"]').click() self.start = -1 elif self.direction == "long" and self.start == -1: logger.success(f"{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，反手平空做多") self.send_dingtalk_message(f"信号：{self.direction}，反手平空做多，开仓金额：{float(num) / 100}") self.page.ele('x://*[normalize-space(text())= "市价全平"]').scroll.to_see(center=True) time.sleep(1) self.page.ele('x://*[normalize-space(text())= "市价全平"]').click() time.sleep(3) self.page.ele('x://*[normalize-space(text())= "买入/做多"]').click() self.start = 1 elif self.direction == "short" and self.start == 1: logger.success(f"{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，反手平多做空") self.send_dingtalk_message(f"信号：{self.direction}，反手平多做空，开仓金额：{float(num) / 100}") self.page.ele('x://*[normalize-space(text())= "市价全平"]').scroll.to_see(center=True) time.sleep(1) self.page.ele('x://*[normalize-space(text())= "市价全平"]').click() time.sleep(3) self.page.ele('x://*[normalize-space(text())= "卖出/做空"]').click() self.start = -1 def get_now_time(self): # 获取当前时间戳 current_timestamp = time.time() # 将当前时间戳转换为 datetime 对象 current_datetime = datetime.datetime.fromtimestamp(current_timestamp) # 计算距离当前时间最近的整点或 30 分时刻 if current_datetime.minute < 30: target_datetime = current_datetime.replace(minute=0, second=0, microsecond=0) else: target_datetime = current_datetime.replace(minute=30, second=0, microsecond=0) # 将目标 datetime 对象转换为时间戳 target_timestamp = target_datetime.timestamp() return int(target_timestamp) def close_extra_tabs_in_browser(self): try: for _, i in enumerate(self.page.get_tabs()): if _ == 0: continue i.close() return True except: pass return False def get_num(self): for i in range(3): try: response = self.session.get( 'https://capi.websea.com/webApi/asset/account', ) return response.json()["result"]["available"] except: time.sleep(1) return False def get_token(self): tab = self.page.new_tab() tab.listen.start("api.websea.com/webApi/zendesk/url") tab.get(url="https://www.websea.com/zh-CN/futures/ETH-USDT") res = tab.listen.wait() for i in res.request.cookies: if not i["name"].startswith(':'): self.cookies[i["name"]] = i["value"] for i in res.request.headers: if i == "token": self.headers[i] = res.request.headers[i] if self.cookies.get('shareToken'): self.session.cookies.update(self.cookies) self.session.headers.update(self.headers) tab.close() return True else: tab.close() return False def get_position_status(self): for i in range(3): try: response = self.session.get( 'https://capi.websea.com/webApi/entrust/holdPosition', ) resp_data = response.json() if not resp_data["result"]: self.start = 0 elif resp_data["result"][0]["type"] == 1: self.start = 1 elif resp_data["result"][0]["type"] == 2: self.start = -1 self.resp_data = resp_data["result"][0] return True except: time.sleep(1) return False def get_now_time1(self): timestamp = time.time() local_time = time.localtime(timestamp) formatted_time = time.strftime("%Y-%m-%d %H:%M:%S", local_time) return formatted_time def send_dingtalk_message(self, message_content, type=1): if type: send_dingtalk_message(f"🔔websea：" + message_content) else: for i in range(15): send_dingtalk_message(f"❌websea：" + message_content) def action(self): # 获取比特端口 if self.openBrowser(): logger.info("获取打开比特成功，成功获取端口！！！") else: logger.error("打开比特失败！！！") return # 接管浏览器 if self.take_over_browser(): logger.info("接管比特浏览器成功！！！") else: logger.error("接管浏览器失败！！！") return if self.close_extra_tabs_in_browser(): logger.info('关闭多余标签页成功!!!') else: logger.info('关闭多余标签页失败!!!') self.page.get(url="https://www.websea.com/zh-CN/futures/ETH-USDT") # 打开网页 self.pbar = tqdm(total=30, desc="等待时间中", ncols=80) # desc：进度条说明，ncols：长度 self.time_start = None # 时间状态 避免同一个时段，发生太多消息 while True: # 获取当前时间 current_time = time.localtime() current_minute = current_time.tm_min if current_minute < 30: self.pbar.n = current_minute self.pbar.refresh() else: self.pbar.n = current_minute - 30 self.pbar.refresh() # if current_minute not in [0, 1, 2, 3, 4, 5, 30, 31, 32, 33, 34, ]: # 判断是否是 新的30分钟了 # # if current_minute not in range(60): # 判断是否是 新的30分钟了 # time.sleep(10) # continue if self.kline_3 and self.get_now_time() == self.kline_3["id"]: continue new_price_datas = self.get_price() if not new_price_datas: logger.info("获取最新价格有问题！！！") self.send_dingtalk_message(message_content=f"获取价格有问题！！！", type=0) return new_price_datas1 = sorted(new_price_datas, key=lambda x: x["id"]) self.kline_1, self.kline_2, self.kline_3 = new_price_datas1[-3:] self.time_start = self.get_now_time() self.page.get(url="https://www.websea.com/zh-CN/futures/ETH-USDT") # 打开网页 if self.get_token(): # 获取token logger.info("获取token成功!!!") else: logger.info("获取token失败！！！") self.send_dingtalk_message(message_content=f"获取token失败！！！", type=0) return if self.get_position_status(): logger.info("获取仓位信息成功!!!") else: logger.info("获取仓位信息失败！！！") self.send_dingtalk_message(message_content=f"获取仓位信息失败！！！", type=0) return try: if self.start == 1: if is_bearish(self.kline_1) and is_bearish(self.kline_2): logger.success(f"{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，平多") self.send_dingtalk_message( message_content=f"第一根信号：{self.kline_1}，{self.kline_2}，平多") self.page.ele('x://*[normalize-space(text())= "市价全平"]').scroll.to_see(center=True) time.sleep(1) self.page.ele('x://*[normalize-space(text())= "市价全平"]').click() self.start = 0 elif self.start == -1: if is_bullish(self.kline_1) and is_bullish(self.kline_2): logger.success(f"{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，平空") self.send_dingtalk_message( message_content=f"第一根信号：{self.kline_1}，{self.kline_2}，平空") self.page.ele('x://*[normalize-space(text())= "市价全平"]').scroll.to_see(center=True) time.sleep(1) self.page.ele('x://*[normalize-space(text())= "市价全平"]').click() self.start = 0 except: self.send_dingtalk_message(message_content=f"止损平仓出错！！！", type=0) return self.direction, signal_key = self.check_signal(prev=self.kline_1, curr=self.kline_2) # 判断信号 if self.direction: try: self.to_do_page() except Exception as e: self.send_dingtalk_message(message_content=f"购买操作失败，{e}", type=0) return self.pbar.reset() # 重置进度条 # =============================================================================================== num = self.get_num() if self.start: # 提取并转换 direction = "多" if int(self.resp_data['type']) == 2 else "空" contracts = int(self.resp_data['number']) # 张数 eth_amount = float(self.resp_data['numberConvert']) # ETH数量 open_price = float(self.resp_data['openPriceAvg']) current_price = float(self.resp_data['markPrice']) # 标记价格（更准确用于盈亏） unrealized_pnl = float(self.resp_data['unProfitLoss']) pnl_rate = float(self.resp_data['profitRate']) available_balance = float(self.resp_data['accountAvailable']) # 钉钉消息示例 msg = ( f"【WebSea ETH-USDT 永续持仓】\n" f"持仓方向：{direction}\n" f"持仓张数：{contracts} 张\n" f"持仓数量(eth)：{eth_amount:.3f} ETH\n" f"持仓数量(usdt)：{float(self.resp_data["numberConvertU"]) / 100:.3f} usdt\n" f"开仓均价：{open_price:.2f} USDT\n" f"标记现价：{current_price:.2f} USDT\n" f"浮动盈亏：{unrealized_pnl:+.2f} USDT ({pnl_rate:+.2f}%)\n" f"账户可用：{available_balance:.2f} USDT" ) else: msg = ( f"【WebSea ETH-USDT 永续持仓】\n" f"持仓方向：无\n" f"账户可用：{float(num):.2f} USDT" ) self.send_dingtalk_message( message_content= msg ) if __name__ == '__main__': WeexTransaction( tge_id=191303, ).action()