ddrwode/lm_code
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

加入 weex

Browse Source
This commit is contained in:
ddrwode
2026-02-03 17:28:53 +08:00
parent fdcec74585
commit a7c6b4e1fc
1 changed files with 29 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
交易/weex-三分之一策略-5分钟交易.py
View File

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