402 lines
13 KiB
Python
402 lines
13 KiB
Python
import time
|
||
import datetime
|
||
from tqdm import tqdm
|
||
from loguru import logger
|
||
from DrissionPage import ChromiumOptions, ChromiumPage
|
||
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:
|
||
"""BitMart 自动交易脚本（优化版）"""
|
||
|
||
def __init__(self, tge_id):
|
||
self.tge_url = "http://127.0.0.1:50326"
|
||
self.tge_id = tge_id
|
||
|
||
self.tge_headers = {
|
||
"Authorization": "Bearer asp_174003986c9b0799677c5b2c1adb76e402735d753bc91a91",
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
self.page: ChromiumPage | None = None
|
||
self.session = requests.Session()
|
||
|
||
self.cookies = {}
|
||
self.tge_port = None
|
||
|
||
# 当前仓位（-1 空 / 0 无 / 1 多）
|
||
self.start = 0
|
||
|
||
# 最新 3 根 kline
|
||
self.kline_1 = None
|
||
self.kline_2 = None
|
||
self.kline_3 = None
|
||
|
||
# 当前信号
|
||
self.direction = None
|
||
|
||
# 防止同一时段重复执行
|
||
self.time_start = None
|
||
self.pbar = None
|
||
|
||
# -------------------------------------------------------------
|
||
# ------------------------- 通用工具 ----------------------------
|
||
# -------------------------------------------------------------
|
||
|
||
def now_text(self):
|
||
"""格式化当前时间"""
|
||
return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
def ding(self, msg, error=False):
|
||
"""统一钉钉消息格式"""
|
||
prefix = "❌bitmart：" if error else "🔔bitmart："
|
||
send_dingtalk_message(f"{prefix}{self.now_text()}，{msg}")
|
||
|
||
def get_half_hour_timestamp(self):
|
||
"""返回当前最近的整点或半点时间戳"""
|
||
ts = time.time()
|
||
dt = datetime.datetime.fromtimestamp(ts)
|
||
target = dt.replace(minute=0, second=0, microsecond=0) if dt.minute < 30 \
|
||
else dt.replace(minute=30, second=0, microsecond=0)
|
||
return int(target.timestamp())
|
||
|
||
# -------------------------------------------------------------
|
||
# ---------------------- TGE 浏览器相关 ------------------------
|
||
# -------------------------------------------------------------
|
||
|
||
def openBrowser(self):
|
||
"""打开 TGE 对应浏览器实例"""
|
||
try:
|
||
res = requests.post(
|
||
f"{self.tge_url}/api/browser/start",
|
||
json={"envId": self.tge_id},
|
||
headers=self.tge_headers
|
||
)
|
||
self.tge_port = res.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 close_extra_tabs(self):
|
||
"""关闭多余 tab"""
|
||
try:
|
||
for idx, tab in enumerate(self.page.get_tabs()):
|
||
if idx > 0:
|
||
tab.close()
|
||
return True
|
||
except:
|
||
return False
|
||
|
||
# -------------------------------------------------------------
|
||
# -------------------- BitMart Token & 接口 --------------------
|
||
# -------------------------------------------------------------
|
||
|
||
def get_token(self):
|
||
"""从浏览器监听接口中提取 token 和 cookies"""
|
||
tab = self.page.new_tab()
|
||
tab.listen.start("ifcontract/shareguide/status")
|
||
|
||
for _ in range(3):
|
||
try:
|
||
tab.get("https://derivatives.bitmart.com/zh-CN/futures/ETHUSDT")
|
||
res = tab.listen.wait(timeout=15)
|
||
|
||
# 请求头 + Cookies
|
||
for i in res.request.headers:
|
||
if ":" not in i:
|
||
self.session.headers[i] = res.request.headers[i]
|
||
|
||
for c in res.request.cookies:
|
||
self.cookies[c["name"]] = c["value"]
|
||
|
||
if "accessKey" in self.cookies:
|
||
self.session.cookies.update(self.cookies)
|
||
tab.close()
|
||
return True
|
||
|
||
except:
|
||
time.sleep(1)
|
||
|
||
tab.close()
|
||
return False
|
||
|
||
def get_account_balance(self):
|
||
"""获取可用余额"""
|
||
params = {'account_type': '3', 'isServerCal': '1'}
|
||
|
||
for _ in range(3):
|
||
try:
|
||
res = self.session.get(
|
||
'https://derivatives.bitmart.com/gw-api/contract-tiger/forward/v1/ifcontract/accounts',
|
||
params=params
|
||
)
|
||
return float(res.json()["data"]["accounts"][0]["available_balance"])
|
||
except:
|
||
time.sleep(1)
|
||
return None
|
||
|
||
def get_position_status(self):
|
||
"""获取当前仓位：1 多、-1 空、0 无"""
|
||
params = {'status': '1'}
|
||
|
||
for _ in range(3):
|
||
try:
|
||
res = self.session.get(
|
||
'https://derivatives.bitmart.com/gw-api/contract-tiger/forward/v1/ifcontract/userPositions',
|
||
params=params
|
||
).json()
|
||
|
||
pos = res.get("data", {}).get("positions", [])
|
||
if not pos:
|
||
self.start = 0
|
||
else:
|
||
ptype = pos[0].get("position_type")
|
||
self.start = 1 if ptype == 1 else -1
|
||
return True
|
||
|
||
except:
|
||
time.sleep(1)
|
||
|
||
return False
|
||
|
||
# -------------------------------------------------------------
|
||
# --------------------------- K线相关 ---------------------------
|
||
# -------------------------------------------------------------
|
||
|
||
def get_price(self):
|
||
"""获取最新 15 分钟 K 线"""
|
||
params = {
|
||
'unit': '30',
|
||
'resolution': 'M',
|
||
'contractID': '2',
|
||
'offset': '340',
|
||
'endTime': str(int(time.time())),
|
||
}
|
||
|
||
for _ in range(3):
|
||
try:
|
||
res = self.session.get(
|
||
'https://contract-v2.bitmart.com/v1/ifcontract/quote/kline',
|
||
params=params, timeout=5
|
||
)
|
||
|
||
datas = []
|
||
for k in res.json()["data"]:
|
||
datas.append({
|
||
'id': int(k["timestamp"]) - 1,
|
||
'open': float(k["open"]),
|
||
'high': float(k["high"]),
|
||
'low': float(k["low"]),
|
||
'close': float(k["close"])
|
||
})
|
||
return sorted(datas, key=lambda x: x["id"])
|
||
|
||
except:
|
||
time.sleep(1)
|
||
|
||
return None
|
||
|
||
def check_signal(self, prev, curr):
|
||
"""包住形态判定"""
|
||
p_open, p_close = prev["open"], prev["close"]
|
||
c_open, c_close = curr["open"], curr["close"]
|
||
|
||
# 前跌后涨包住 -> 多
|
||
if is_bullish(curr) and is_bearish(prev) and c_close >= p_open:
|
||
return "long"
|
||
|
||
# 前涨后跌包住 -> 空
|
||
if is_bearish(curr) and is_bullish(prev) and c_close <= p_open:
|
||
return "short"
|
||
|
||
return None
|
||
|
||
# -------------------------------------------------------------
|
||
# ---------------------- 下单逻辑封装 ---------------------------
|
||
# -------------------------------------------------------------
|
||
|
||
def click_safe(self, xpath, sleep=0.5):
|
||
"""安全点击"""
|
||
try:
|
||
ele = self.page.ele(xpath)
|
||
if not ele:
|
||
return False
|
||
ele.scroll.to_see(center=True)
|
||
time.sleep(sleep)
|
||
ele.click()
|
||
return True
|
||
except:
|
||
return False
|
||
|
||
def do_order(self):
|
||
"""执行交易动作（开仓/反手）"""
|
||
|
||
# 选择市价
|
||
self.click_safe('x:(//button[normalize-space(text())="市价"])')
|
||
|
||
# 余额
|
||
balance = None
|
||
for i in range(3):
|
||
balance = self.get_account_balance()
|
||
if not balance:
|
||
self.ding("获取可用余额失败！", error=True)
|
||
return
|
||
|
||
amount = balance / 100
|
||
self.page.ele('x://*[@id="size_0"]').input(amount)
|
||
|
||
# ---- 开仓逻辑 ----
|
||
if self.direction == "long":
|
||
if self.start == 0:
|
||
self.ding(f"信号：多，开多：{amount}")
|
||
self.click_safe('x://span[normalize-space(text()) ="买入/做多"]')
|
||
self.start = 1
|
||
elif self.start == -1:
|
||
self.ding(f"信号：多，反手空转多：{amount}")
|
||
self.click_safe('x://span[normalize-space(text()) ="市价"]')
|
||
time.sleep(2)
|
||
self.click_safe('x://span[normalize-space(text()) ="买入/做多"]')
|
||
self.start = 1
|
||
|
||
elif self.direction == "short":
|
||
if self.start == 0:
|
||
self.ding(f"信号：空，开空：{amount}")
|
||
self.click_safe('x://span[normalize-space(text()) ="卖出/做空"]')
|
||
self.start = -1
|
||
elif self.start == 1:
|
||
self.ding(f"信号：空，反手多转空：{amount}")
|
||
self.click_safe('x://span[normalize-space(text()) ="市价"]')
|
||
time.sleep(2)
|
||
self.click_safe('x://span[normalize-space(text()) ="卖出/做空"]')
|
||
self.start = -1
|
||
|
||
# -------------------------------------------------------------
|
||
# ----------------------------- 主流程 ---------------------------
|
||
# -------------------------------------------------------------
|
||
|
||
def action(self):
|
||
# 1. 打开浏览器
|
||
if not self.openBrowser():
|
||
self.ding("打开 TGE 失败！", error=True)
|
||
return
|
||
logger.info("TGE 端口获取成功")
|
||
|
||
# 2. 接管浏览器
|
||
if not self.take_over_browser():
|
||
self.ding("接管浏览器失败！", error=True)
|
||
return
|
||
logger.info("浏览器接管成功")
|
||
|
||
self.close_extra_tabs()
|
||
self.page.get("https://derivatives.bitmart.com/zh-CN/futures/ETHUSDT")
|
||
|
||
self.pbar = tqdm(total=30, desc="等待半小时周期", ncols=80)
|
||
|
||
while True:
|
||
now = time.localtime()
|
||
minute = now.tm_min
|
||
|
||
# 更新进度条
|
||
self.pbar.n = minute if minute < 30 else minute - 30
|
||
self.pbar.refresh()
|
||
|
||
# 必须是整点或半点及前 5 分钟
|
||
if minute not in [0, 1, 2, 3, 4, 5, 30, 31, 32, 33, 34, 35]:
|
||
time.sleep(8)
|
||
return
|
||
|
||
# 时间重复跳过
|
||
if self.time_start == self.get_half_hour_timestamp():
|
||
time.sleep(5)
|
||
continue
|
||
|
||
# ---- 获取 Token ----
|
||
if not self.get_token():
|
||
self.ding("获取 token 失败！", error=True)
|
||
return
|
||
logger.info("Token 获取成功")
|
||
|
||
# ---- 获取价格 ----
|
||
kdatas = self.get_price()
|
||
if not kdatas:
|
||
self.ding("获取价格失败！", error=True)
|
||
return
|
||
|
||
self.kline_1, self.kline_2, self.kline_3 = kdatas[-3:]
|
||
if int(self.kline_3["id"]) != self.get_half_hour_timestamp():
|
||
continue
|
||
|
||
logger.success("K线获取成功")
|
||
|
||
self.time_start = self.get_half_hour_timestamp()
|
||
|
||
# ---- 刷新页面 ----
|
||
self.page.get("https://derivatives.bitmart.com/zh-CN/futures/ETHUSDT")
|
||
|
||
for i in range(3):
|
||
# ---- 获取仓位 ----
|
||
if not self.get_position_status():
|
||
self.ding("获取仓位失败！", error=True)
|
||
continue
|
||
|
||
if self.start:
|
||
break
|
||
|
||
# ---- 止损平仓 ----
|
||
try:
|
||
if self.start == 1 and is_bearish(self.kline_1) and is_bearish(self.kline_2):
|
||
self.ding("两根大阴线，平多")
|
||
self.click_safe('x://span[normalize-space(text()) ="市价"]')
|
||
self.start = 0
|
||
|
||
elif self.start == -1 and is_bullish(self.kline_1) and is_bullish(self.kline_2):
|
||
self.ding("两根大阳线，平空")
|
||
self.click_safe('x://span[normalize-space(text()) ="市价"]')
|
||
self.start = 0
|
||
except:
|
||
self.ding("止损平仓错误！", error=True)
|
||
# continue
|
||
return
|
||
|
||
# ---- 生成新信号 ----
|
||
self.direction = self.check_signal(prev=self.kline_1, curr=self.kline_2)
|
||
|
||
# ---- 执行交易 ----
|
||
if self.direction:
|
||
try:
|
||
self.do_order()
|
||
except:
|
||
self.ding("下单失败！", error=True)
|
||
|
||
# ---- 周期结束消息 ----
|
||
self.pbar.reset()
|
||
self.ding(
|
||
f"持仓：{'无' if self.start == 0 else ('多' if self.start == 1 else '空')}，"
|
||
f"信号：{'无' if not self.direction else ('多' if self.direction == 'long' else '空')}"
|
||
)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
WeexTransaction(tge_id=196495).action()
|