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

rgfewfger

Browse Source
This commit is contained in:
27942
2025-12-11 16:52:46 +08:00
parent e02231f5ed
commit 9f99a116c0
2 changed files with 35 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
交易/bitmart_交易.py
View File

@@ -382,7 +382,7 @@ class WeexTransaction:
if not new_price_datas:
logger.info("获取最新价格有问题！！！")
self.send_dingtalk_message(message_content=f"weex：{self.get_now_time()}，获取最新价格有问题！！！")
self.send_dingtalk_message(message_content=f"bitmart：{self.get_now_time()}，获取最新价格有问题！！！")
continue
# 解析价格
@@ -400,7 +400,7 @@ class WeexTransaction:
logger.info("获取token成功!!!")
else:
logger.info("获取token失败！！！")
self.send_dingtalk_message(message_content=f"weex：{self.get_now_time()}，获取token失败！！！")
self.send_dingtalk_message(message_content=f"bitmart：{self.get_now_time()}，获取token失败！！！")
continue
if self.get_position_status():
@@ -408,7 +408,7 @@ class WeexTransaction:
else:
logger.info("获取仓位信息失败！！！")
self.send_dingtalk_message(message_content=f"weex：{self.get_now_time()}，获取仓位信息失败！！！")
self.send_dingtalk_message(message_content=f"bitmart：{self.get_now_time()}，获取仓位信息失败！！！")
continue
if self.start == 1:
@@ -438,7 +438,7 @@ class WeexTransaction:
self.pbar.reset() # 重置进度条
self.send_dingtalk_message(
message_content=
f"{datetime.datetime.now()}，"
f"bitmart：{datetime.datetime.now()}，"
f"目前有持仓：{"无" if self.start == 0 else ("多" if self.start == 1 else "空")}，"
f"当前信号：{"无" if not self.direction else ("多" if self.direction == "long" else "空")}"
)

78
交易/websea_交易.py
View File

@@ -1,5 +1,6 @@
import re
import time
import asyncio
import datetime
from tqdm import *
@@ -28,11 +29,6 @@ class WeexTransaction:
"Content-Type": "application/json"
}
# 替换为你自己的钉钉机器人 Webhook 地址
self.webhook_url = "https://oapi.dingtalk.com/robot/send?access_token=e2fafb3f46866d50fe52cbb29650ba9ef1cbc97915dde238192f04c906fe4125"
# 替换为你自己的钉钉机器人秘钥
self.secret = "SEC5f320e72d7a4eaca540c66c3d09edff2f74936517390dee99ece6dd1b3611998"
self.page = None # 浏览器对象
self.start = 0 # 持仓状态 -1:做空，0：维持仓，1：做多
@@ -46,24 +42,24 @@ class WeexTransaction:
self.session = requests.Session() # 接口请求对象
def send_dingtalk_message(self, message_content):
result = asyncio.run(self.to_do_tg(message_content))
return result
pass
def openBrowser(self, ): # 直接指定ID打开窗口，也可以使用 createBrowser 方法返回的ID
# url = "http://8.137.99.82:9005/api/send_click?token=fegergauiernguie&phone=8613661496481"
#
# res = requests.post(
# url=url,
# json={
# "phone": "8613661496481",
# "bot_name": "ergggreef",
# "datas": [
# {"send_message": [message_content], "click_button": [""], },
# ]
#
# }
# )
#
# print(res.json())
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 openBrowser(self, ): # 直接指定ID打开窗口，也可以使用 createBrowser 方法返回的ID
@@ -183,19 +179,22 @@ class WeexTransaction:
if self.direction == "long" and not self.start:
logger.success(f"{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，开多")
self.send_dingtalk_message(
message_content=f"{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，开多")
message_content=f"websea：{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，开多")
self.mn_tab.ele('x://*[contains(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(
message_content=f"{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，开空")
message_content=f"websea：{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，开空")
self.mn_tab.ele('x://*[contains(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(
message_content=f"{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，反手平空做多")
message_content=f"websea：{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，反手平空做多")
self.mn_tab.ele('x://*[contains(text(), "市价全平")]').scroll.to_see(center=True)
self.mn_tab.ele('x://*[contains(text(), "市价全平")]').click()
time.sleep(3)
@@ -204,30 +203,14 @@ class WeexTransaction:
elif self.direction == "short" and self.start == 1:
logger.success(f"{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，反手平多做空")
self.send_dingtalk_message(
message_content=f"{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，反手平多做空")
message_content=f"websea：{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，反手平多做空")
self.mn_tab.ele('x://*[contains(text(), "市价全平")]').scroll.to_see(center=True)
self.mn_tab.ele('x://*[contains(text(), "市价全平")]').click()
time.sleep(3)
self.mn_tab.ele('x://*[contains(text(), "卖出/做空")]').click()
self.start = -1
def get_text(self, target_text):
# 去除目标文本中的空白字符
cleaned_target_text = re.sub(r'\s', '', target_text)
# 创建 BeautifulSoup 对象
soup = BeautifulSoup(self.mn_tab.html, 'html.parser')
# 遍历所有标签的文本内容
for tag in soup.find_all():
tag_text = tag.get_text()
# 去除标签文本中的空白字符
cleaned_tag_text = re.sub(r'\s', '', tag_text)
if cleaned_target_text in cleaned_tag_text:
return True
else:
return False
def get_now_time(self):
# 获取当前时间戳
current_timestamp = time.time()
@@ -360,7 +343,7 @@ class WeexTransaction:
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分钟了
# if current_minute not in range(60): # 判断是否是 新的30分钟了
time.sleep(10)
continue
@@ -389,7 +372,8 @@ class WeexTransaction:
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"{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，平多")
message_content=f"websea：{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，平多")
self.mn_tab.ele('x://*[contains(text(), "市价全平")]').scroll.to_see(center=True)
self.mn_tab.ele('x://*[contains(text(), "市价全平")]').click()
self.start = 0
@@ -397,7 +381,7 @@ class WeexTransaction:
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"{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，平空")
message_content=f"websea：{datetime.datetime.now()}，第一根信号：{self.kline_1}，{self.kline_2}，平空")
self.mn_tab.ele('x://*[contains(text(), "市价全平")]').scroll.to_see(center=True)
self.mn_tab.ele('x://*[contains(text(), "市价全平")]').click()
@@ -410,12 +394,12 @@ class WeexTransaction:
self.to_do_page()
except Exception as e:
self.send_dingtalk_message(
message_content=f"{datetime.datetime.now()}，{e}")
message_content=f"websea：{datetime.datetime.now()}，{e}")
self.pbar.reset() # 重置进度条
self.send_dingtalk_message(
message_content=
f"{datetime.datetime.now()}，"
f"websea：{datetime.datetime.now()}，"
f"目前有持仓：{"无" if self.start == 0 else ("多" if self.start == 1 else "空")}，"
f"当前信号：{"无" if not self.direction else ("多" if self.direction == "long" else "空")}"
)