diff --git a/交易/weex-三分之一策略-5分钟交易.py b/交易/weex-三分之一策略-5分钟交易.py index af5598f..e3d5886 100644 --- a/交易/weex-三分之一策略-5分钟交易.py +++ b/交易/weex-三分之一策略-5分钟交易.py @@ -9,6 +9,7 @@ WEEX ETH-USDT 永续合约 — 三分之一策略(5分钟K线) 4. 反手二:持多且上根上影线>0.01%且当前跌到上根开盘→反手空;持空且上根下影线>0.01%且当前涨到上根开盘→反手多 5. 同一根5分钟K线内只交易一次 """ +import random import time import datetime from typing import Optional, Dict, List, Tuple @@ -27,7 +28,7 @@ class Config: CONTRACT_ID = "10000002" PRODUCT_CODE = "cmt_ethusdt" - KLINE_TYPE = "MINUTE_5" # 5分钟K线(三分之一策略) + KLINE_TYPE = "MINUTE_5" # 5分钟K线(三分之一策略) KLINE_LIMIT = 100 TRADING_URL = "https://www.weex.com/zh-CN/futures/ETH-USDT" @@ -50,7 +51,8 @@ class OneThirdStrategyAnalyzer: return abs(float(candle['open']) - float(candle['close'])) @staticmethod - def find_valid_prev_bar(all_data: List[Dict], current_idx: int, min_body_size: float = 0.1) -> Tuple[Optional[int], Optional[Dict]]: + def find_valid_prev_bar(all_data: List[Dict], current_idx: int, min_body_size: float = 0.1) -> Tuple[ + Optional[int], Optional[Dict]]: if current_idx <= 0: return None, None for i in range(current_idx - 1, -1, -1): @@ -96,9 +98,11 @@ class OneThirdStrategyAnalyzer: curr_low = float(curr['low']) prev_high = float(prev['high']) prev_low = float(prev['low']) - if start == -1 and curr_high >= prev_high and OneThirdStrategyAnalyzer.upper_shadow_pct(prev) > Config.MIN_SHADOW_PCT: + if start == -1 and curr_high >= prev_high and OneThirdStrategyAnalyzer.upper_shadow_pct( + prev) > Config.MIN_SHADOW_PCT: return 'long', prev - if start == 1 and curr_low <= prev_low and OneThirdStrategyAnalyzer.lower_shadow_pct(prev) > Config.MIN_SHADOW_PCT: + if start == 1 and curr_low <= prev_low and OneThirdStrategyAnalyzer.lower_shadow_pct( + prev) > Config.MIN_SHADOW_PCT: return 'short', prev return None, None @@ -110,17 +114,19 @@ class OneThirdStrategyAnalyzer: curr_high = float(curr['high']) curr_low = float(curr['low']) prev_open = float(prev['open']) - if start == 1 and OneThirdStrategyAnalyzer.upper_shadow_pct(prev) > Config.MIN_SHADOW_PCT and curr_low <= prev_open: + if start == 1 and OneThirdStrategyAnalyzer.upper_shadow_pct( + prev) > Config.MIN_SHADOW_PCT and curr_low <= prev_open: return 'short', prev - if start == -1 and OneThirdStrategyAnalyzer.lower_shadow_pct(prev) > Config.MIN_SHADOW_PCT and curr_high >= prev_open: + if start == -1 and OneThirdStrategyAnalyzer.lower_shadow_pct( + prev) > Config.MIN_SHADOW_PCT and curr_high >= prev_open: return 'long', prev return None, None @staticmethod def check_realtime_trigger( - kline_data: List[Dict], - last_trigger_kline_id: Optional[int], - last_trigger_direction: Optional[str], + kline_data: List[Dict], + last_trigger_kline_id: Optional[int], + last_trigger_direction: Optional[str], ) -> Tuple[Optional[str], Optional[float], Optional[Dict], Optional[Dict]]: if len(kline_data) < 2: return None, None, None, None @@ -223,11 +229,11 @@ class WEEXApiClient: self.headers = headers def get_kline_data( - self, - contract_id: str = Config.CONTRACT_ID, - product_code: str = Config.PRODUCT_CODE, - kline_type: str = Config.KLINE_TYPE, - limit: int = Config.KLINE_LIMIT, + self, + contract_id: str = Config.CONTRACT_ID, + product_code: str = Config.PRODUCT_CODE, + kline_type: str = Config.KLINE_TYPE, + limit: int = Config.KLINE_LIMIT, ) -> List[Dict]: params = { 'contractId': contract_id, @@ -487,9 +493,12 @@ class WeexOneThirdTransaction: self.last_trade_kline_id: Optional[int] = None def initialize(self) -> bool: - if not self.browser_manager.openBrowser(): + for i in range(3): + if self.browser_manager.openBrowser(): + break + else: MessageSender.send("打开浏览器失败", is_error=True) - return False + if not self.browser_manager.take_over_browser(): MessageSender.send("接管浏览器失败", is_error=True) return False @@ -619,6 +628,10 @@ class WeexOneThirdTransaction: MessageSender.send(f"运行出错: {e}", is_error=True) time.sleep(10) + if random.randint(1, 10) > 7: + self.browser_manager.page.close() + time.sleep(5) + def action(self) -> None: self.run()