""" WebSea ETH-USDT 永续合约自动交易系统 优化版本：改进代码结构、模块化和可读性 """ import time import datetime from typing import Optional, Dict, List, Tuple from tqdm import tqdm from loguru import logger from DrissionPage import ChromiumOptions, ChromiumPage from curl_cffi import requests from 交易.tools import send_dingtalk_message class KlineAnalyzer: """K线分析器：负责K线数据分析和信号判断""" @staticmethod def is_bullish(candle: Dict) -> bool: """判断是否为阳线""" return float(candle['close']) > float(candle['open']) @staticmethod def is_bearish(candle: Dict) -> bool: """判断是否为阴线""" return float(candle['close']) < float(candle['open']) @staticmethod def check_engulfing_signal(prev: Dict, curr: Dict) -> Tuple[Optional[str], Optional[str]]: """ 包住形态信号判定（仅30分钟K线）： - 前跌后涨包住 -> 做多 - 前涨后跌包住 -> 做空 Returns: (direction, signal_type): direction为"long"或"short"，signal_type为信号类型 """ c_open, c_close = float(curr['open']), float(curr['close']) p_open = float(prev['open']) # 前跌后涨包住 -> 做多 if (KlineAnalyzer.is_bullish(curr) and KlineAnalyzer.is_bearish(prev) and int(c_close) >= int(p_open)): return "long", "bear_bull_engulf" # 前涨后跌包住 -> 做空 if (KlineAnalyzer.is_bearish(curr) and KlineAnalyzer.is_bullish(prev) and int(c_close) <= int(p_open)): return "short", "bull_bear_engulf" return None, None class BrowserManager: """浏览器管理器：负责浏览器的启动、接管和标签页管理""" def __init__(self, tge_id: int, tge_url: str, tge_headers: Dict): self.tge_id = tge_id self.tge_url = tge_url self.tge_headers = tge_headers self.tge_port: Optional[int] = None self.page: Optional[ChromiumPage] = None def open_browser(self) -> bool: """打开浏览器并获取端口""" try: response = requests.post( f"{self.tge_url}/api/browser/start", json={"envId": self.tge_id}, headers=self.tge_headers, timeout=10 ) self.tge_port = response.json()["data"]["port"] return True except Exception as e: logger.error(f"打开浏览器失败: {e}") return False def take_over_browser(self) -> bool: """接管浏览器""" if not self.tge_port: logger.error("浏览器端口未设置") return False 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 Exception as e: logger.error(f"接管浏览器失败: {e}") return False def close_extra_tabs(self) -> bool: """关闭多余的标签页，只保留第一个""" if not self.page: return False try: tabs = self.page.get_tabs() for idx, tab in enumerate(tabs): if idx == 0: continue tab.close() return True except Exception as e: logger.warning(f"关闭多余标签页失败: {e}") return False class ApiClient: """API客户端：负责与WebSea API的交互""" def __init__(self): self.session = requests.Session() self.headers: Dict = {} self.cookies: Dict = {} def update_session(self, cookies: Dict, headers: Dict) -> None: """更新session的cookies和headers""" self.session.cookies.update(cookies) self.session.headers.update(headers) def get_kline_data(self, symbol: str = 'ETH-USDT', period: str = '30min', days: int = 10) -> List[Dict]: """ 获取K线数据 Args: symbol: 交易对符号 period: K线周期 days: 获取最近N天的数据 Returns: K线数据列表 """ now_ts = int(time.time()) start_ts = int(time.time() - 86400 * days) params = { 'symbol': symbol, 'period': period, 'start': str(start_ts), 'end': str(now_ts), } for attempt in range(3): logger.info(f"获取最新K线数据：第{attempt + 1}次尝试...") try: response = self.session.get( 'https://capi.websea.com/webApi/market/getKline', params=params, timeout=5 ) result = response.json()["result"]['data'] kline_data = [] for data in result: kline_data.append({ 'id': int(data["id"]), 'open': float(data["open"]), 'high': float(data["high"]), 'low': float(data["low"]), 'close': float(data["close"]) }) return kline_data except Exception as e: logger.warning(f"获取K线数据失败: {e}") if attempt < 2: time.sleep(1) return [] def get_available_balance(self) -> Optional[float]: """获取可用余额""" for attempt in range(3): try: response = self.session.get( 'https://capi.websea.com/webApi/asset/account', timeout=5 ) return float(response.json()["result"]["available"]) except Exception as e: logger.warning(f"获取余额失败: {e}") if attempt < 2: time.sleep(1) return None def get_position_status(self) -> Tuple[bool, Optional[Dict]]: """ 获取持仓状态 Returns: (success, position_data): success为是否成功，position_data为持仓信息 """ for attempt in range(3): try: response = self.session.get( 'https://capi.websea.com/webApi/entrust/holdPosition', timeout=5 ) resp_data = response.json() if not resp_data["result"]: return True, {"type": 0, "position": None} position = resp_data["result"][0] position_info = { "type": position["type"], # 1: 多, 2: 空 "position": position } return True, position_info except Exception as e: logger.warning(f"获取持仓状态失败: {e}") if attempt < 2: time.sleep(1) return False, None class TokenManager: """Token管理器：负责获取和更新认证token""" def __init__(self, api_client: ApiClient, page: ChromiumPage): self.api_client = api_client self.page = page def get_token(self) -> bool: """从浏览器中获取token和cookies""" try: 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() # 提取cookies cookies = {} for cookie in res.request.cookies: if not cookie["name"].startswith(':'): cookies[cookie["name"]] = cookie["value"] # 提取headers headers = {} if "token" in res.request.headers: headers["token"] = res.request.headers["token"] tab.close() # 检查是否有shareToken if cookies.get('shareToken'): self.api_client.update_session(cookies, headers) return True else: logger.warning("未获取到shareToken") return False except Exception as e: logger.error(f"获取token失败: {e}") return False class TradingExecutor: """交易执行器：负责执行交易操作""" def __init__(self, page: ChromiumPage, api_client: ApiClient): self.page = page self.api_client = api_client def navigate_to_trading_page(self) -> bool: """导航到交易页面""" try: self.page.ele('x://*[normalize-space(text())= "市价"]', timeout=15).click() time.sleep(1) return True except Exception as e: logger.error(f"导航到交易页面失败: {e}") return False def close_all_positions(self) -> bool: """平仓所有持仓""" try: self.page.ele('x://*[normalize-space(text())= "市价全平"]').scroll.to_see(center=True) time.sleep(1) self.page.actions.click('x://*[normalize-space(text())= "市价全平"]') time.sleep(3) return True except Exception as e: logger.error(f"平仓失败: {e}") return False def open_long(self, amount: float) -> bool: """开多""" try: self.page.ele('x://*[@id="amountInput"]').input(amount) time.sleep(1) self.page.actions.click('x://*[normalize-space(text())= "买入/做多"]') return True except Exception as e: logger.error(f"开多失败: {e}") return False def open_short(self, amount: float) -> bool: """开空""" try: self.page.ele('x://*[@id="amountInput"]').input(amount) time.sleep(1) self.page.actions.click('x://*[normalize-space(text())= "卖出/做空"]') return True except Exception as e: logger.error(f"开空失败: {e}") return False def execute_trade(self, direction: str, current_position: int, amount: float) -> bool: """ 执行交易操作 Args: direction: 交易方向 "long" 或 "short" current_position: 当前持仓状态 -1:空, 0:无, 1:多 amount: 交易金额 Returns: 是否成功 """ if not self.navigate_to_trading_page(): return False try: # 开多 if direction == "long": if current_position == 0: logger.success(f"{datetime.datetime.now()}，开多，金额：{amount}") self.open_long(amount) elif current_position == -1: logger.success(f"{datetime.datetime.now()}，反手平空做多，金额：{amount}") self.close_all_positions() self.open_long(amount) # 开空 elif direction == "short": if current_position == 0: logger.success(f"{datetime.datetime.now()}，开空，金额：{amount}") self.open_short(amount) elif current_position == 1: logger.success(f"{datetime.datetime.now()}，反手平多做空，金额：{amount}") self.close_all_positions() self.open_short(amount) return True except Exception as e: logger.error(f"执行交易失败: {e}") return False class PositionManager: """持仓管理器：负责持仓状态管理和止损""" def __init__(self, trading_executor: TradingExecutor): self.trading_executor = trading_executor self.current_position: int = 0 # -1:空, 0:无, 1:多 def update_position(self, position_type: int) -> None: """更新持仓状态""" self.current_position = position_type def check_stop_loss(self, kline_1: Dict, kline_2: Dict) -> bool: """ 检查止损条件 Args: kline_1: 倒数第二根K线 kline_2: 倒数第一根K线 Returns: 是否执行了止损 """ try: # 持多仓，连续两根阴线，平多 if self.current_position == 1: if (KlineAnalyzer.is_bearish(kline_1) and KlineAnalyzer.is_bearish(kline_2)): logger.success( f"{datetime.datetime.now()}，止损信号：连续两根阴线，平多" ) self.trading_executor.close_all_positions() self.current_position = 0 return True # 持空仓，连续两根阳线，平空 elif self.current_position == -1: if (KlineAnalyzer.is_bullish(kline_1) and KlineAnalyzer.is_bullish(kline_2)): logger.success( f"{datetime.datetime.now()}，止损信号：连续两根阳线，平空" ) self.trading_executor.close_all_positions() self.current_position = 0 return True except Exception as e: logger.error(f"止损检查失败: {e}") return False class MessageSender: """消息发送器：负责发送钉钉消息""" @staticmethod def send_dingtalk_message(message_content: str, is_error: bool = False) -> None: """ 发送钉钉消息 Args: message_content: 消息内容 is_error: 是否为错误消息（错误消息会发送多次） """ if is_error: prefix = "❌websea：" count = 15 else: prefix = "🔔websea：" count = 1 for _ in range(count): send_dingtalk_message(f"{prefix}{message_content}") @staticmethod def format_position_message(position_data: Optional[Dict], available_balance: float) -> str: """格式化持仓信息消息""" if position_data and position_data.get("position"): pos = position_data["position"] direction = "多" if int(pos['type']) == 1 else "空" contracts = int(pos['number']) eth_amount = float(pos['numberConvert']) open_price = float(pos['openPriceAvg']) current_price = float(pos['markPrice']) unrealized_pnl = float(pos['unProfitLoss']) pnl_rate = float(pos['profitRate']) return ( f"【WebSea ETH-USDT 永续持仓】\n" f"持仓方向：{direction}\n" f"持仓张数：{contracts} 张\n" f"持仓数量(eth)：{eth_amount:.3f} ETH\n" f"持仓数量(usdt)：{float(pos['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: return ( f"【WebSea ETH-USDT 永续持仓】\n" f"持仓方向：无\n" f"账户可用：{available_balance:.2f} USDT" ) class TimeUtils: """时间工具类""" @staticmethod def get_current_kline_timestamp() -> int: """ 获取当前K线的时间戳（30分钟K线的整点或30分时刻） Returns: 时间戳 """ current_timestamp = time.time() 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 ) return int(target_datetime.timestamp()) @staticmethod def get_formatted_time() -> str: """获取格式化的当前时间字符串""" timestamp = time.time() local_time = time.localtime(timestamp) return time.strftime("%Y-%m-%d %H:%M:%S", local_time) @staticmethod def get_progress_bar_value() -> int: """获取进度条的当前值（0-29）""" current_time = time.localtime() current_minute = current_time.tm_min return current_minute if current_minute < 30 else current_minute - 30 class WeexTransaction: """WebSea自动交易主类""" def __init__(self, tge_id: int): # 配置 self.tge_id = tge_id self.tge_url = "http://127.0.0.1:50326" self.tge_headers = { "Authorization": f"Bearer asp_174003986c9b0799677c5b2c1adb76e402735d753bc91a91", "Content-Type": "application/json" } self.trading_url = "https://www.websea.com/zh-CN/futures/ETH-USDT" # 组件初始化 self.browser_manager = BrowserManager(self.tge_id, self.tge_url, self.tge_headers) self.api_client = ApiClient() self.position_manager = None # 需要在浏览器初始化后创建 self.trading_executor = None # 需要在浏览器初始化后创建 self.token_manager = None # 需要在浏览器初始化后创建 # 状态 self.pbar: Optional[tqdm] = None self.last_kline_id: Optional[int] = None self.kline_1: Optional[Dict] = None self.kline_2: Optional[Dict] = None self.kline_3: Optional[Dict] = None def initialize(self) -> bool: """初始化所有组件""" # 打开浏览器 if not self.browser_manager.open_browser(): logger.error("打开浏览器失败") MessageSender.send_dingtalk_message("打开浏览器失败", is_error=True) return False # 接管浏览器 if not self.browser_manager.take_over_browser(): logger.error("接管浏览器失败") MessageSender.send_dingtalk_message("接管浏览器失败", is_error=True) return False # 关闭多余标签页 self.browser_manager.close_extra_tabs() # 初始化需要page的组件 page = self.browser_manager.page self.trading_executor = TradingExecutor(page, self.api_client) self.position_manager = PositionManager(self.trading_executor) self.token_manager = TokenManager(self.api_client, page) # 打开交易页面 page.get(url=self.trading_url) # 初始化进度条 self.pbar = tqdm(total=30, desc="等待时间中", ncols=80) return True def update_progress_bar(self) -> None: """更新进度条""" if self.pbar: self.pbar.n = TimeUtils.get_progress_bar_value() self.pbar.refresh() def fetch_and_update_kline(self) -> bool: """获取并更新K线数据""" kline_data = self.api_client.get_kline_data() if not kline_data: logger.warning("获取K线数据失败") MessageSender.send_dingtalk_message("获取K线数据失败", is_error=True) return False # 排序并取最后三根 sorted_data = sorted(kline_data, key=lambda x: x["id"]) self.kline_1, self.kline_2, self.kline_3 = sorted_data[-3:] # 检查是否是新K线 current_kline_id = self.kline_3["id"] if self.last_kline_id == current_kline_id: return False # 没有新K线 self.last_kline_id = current_kline_id return True def sync_position_status(self) -> bool: """同步持仓状态""" success, position_info = self.api_client.get_position_status() if not success: logger.error("获取持仓状态失败") MessageSender.send_dingtalk_message("获取持仓状态失败", is_error=True) return False if position_info: pos_type = position_info.get("type", 0) if pos_type == 1: self.position_manager.current_position = 1 elif pos_type == 2: self.position_manager.current_position = -1 else: self.position_manager.current_position = 0 return True def process_trading_logic(self) -> None: """处理交易逻辑""" page = self.browser_manager.page # 刷新页面 page.get(url=self.trading_url) # 获取token if not self.token_manager.get_token(): logger.error("获取token失败") MessageSender.send_dingtalk_message("获取token失败", is_error=True) return # 同步持仓状态 if not self.sync_position_status(): return # 检查止损 try: self.position_manager.check_stop_loss(self.kline_1, self.kline_2) except Exception as e: logger.error(f"止损检查出错: {e}") MessageSender.send_dingtalk_message(f"止损检查出错: {e}", is_error=True) return # 检查交易信号 direction, signal_type = KlineAnalyzer.check_engulfing_signal( prev=self.kline_1, curr=self.kline_2 ) if direction: # 获取可用余额 balance = self.api_client.get_available_balance() if balance is None: logger.error("获取可用余额失败") MessageSender.send_dingtalk_message("获取可用余额失败", is_error=True) return amount = balance / 100 # 执行交易 if not self.trading_executor.execute_trade( direction=direction, current_position=self.position_manager.current_position, amount=amount ): MessageSender.send_dingtalk_message( f"交易执行失败，方向：{direction}", is_error=True ) return # 更新持仓状态 if direction == "long": self.position_manager.current_position = 1 elif direction == "short": self.position_manager.current_position = -1 # 发送交易消息 MessageSender.send_dingtalk_message( f"信号：{direction}，开仓金额：{amount:.2f}", is_error=False ) # 再次同步持仓状态（交易后） if not self.sync_position_status(): return # 发送持仓信息 balance = self.api_client.get_available_balance() if balance is not None: success, position_info = self.api_client.get_position_status() if success: msg = MessageSender.format_position_message(position_info, balance) MessageSender.send_dingtalk_message(msg, is_error=False) def run(self) -> None: """主运行循环""" # 初始化 if not self.initialize(): return logger.info("系统初始化完成，开始运行...") # 主循环 while True: try: # 更新进度条 self.update_progress_bar() # 获取当前K线时间戳 current_timestamp = TimeUtils.get_current_kline_timestamp() # 检查是否是新K线（避免重复处理） if self.kline_3 and self.kline_3["id"] == current_timestamp: time.sleep(10) continue # 获取并更新K线数据 if not self.fetch_and_update_kline(): time.sleep(10) continue # 处理交易逻辑 self.process_trading_logic() # 重置进度条 if self.pbar: self.pbar.reset() # 等待一段时间 time.sleep(5) except KeyboardInterrupt: logger.info("收到中断信号，正在退出...") break except Exception as e: logger.error(f"运行出错: {e}") MessageSender.send_dingtalk_message(f"运行出错: {e}", is_error=True) time.sleep(10) def action(self) -> None: """兼容旧接口的方法名""" self.run() if __name__ == '__main__': transaction = WeexTransaction(tge_id=191303) transaction.action()