fewfef

bitmart/api高频交易.py

@@ -82,7 +82,7 @@ class BitmartFuturesTransaction:
                     'close': float(k["close_price"])
                 })
             formatted.sort(key=lambda x: x['id'])
-            return formatted[-3:]  # 最近3根: kline_1 (最老), kline_2, kline_3 (最新)
+            return formatted  # 最近3根: kline_1 (最老), kline_2, kline_3 (最新)
         except Exception as e:
             logger.error(f"获取K线异常: {e}")
             self.ding(error=True, msg="获取K线异常")

交易/test.py

@@ -1 +1,227 @@
-print(407 / 1000)
+import time
+import uuid
+import datetime
+
+import requests
+from DrissionPage import ChromiumPage
+from DrissionPage import ChromiumOptions
+from tqdm import tqdm
+from loguru import logger
+from bitmart.api_contract import APIContract
+from bitmart.lib.cloud_exceptions import APIException
+from 交易.tools import send_dingtalk_message
+
+
+class BitmartFuturesTransaction:
+    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.api_key = "a0fb7b98464fd9bcce67e7c519d58ec10d0c38a8"
+        self.secret_key = "4eaeba78e77aeaab1c2027f846a276d164f264a44c2c1bb1c5f3be50c8de1ca5"
+        self.memo = "合约交易"
+
+        self.contract_symbol = "ETHUSDT"
+
+        self.contractAPI = APIContract(self.api_key, self.secret_key, self.memo, timeout=(5, 15))
+
+        self.start = 0  # 持仓状态: -1 空, 0 无, 1 多
+        self.direction = None
+
+        self.pbar = tqdm(total=30, desc="等待K线", ncols=80)
+
+        self.last_kline_time = None
+
+        self.leverage = "100"  # 高杠杆（全仓模式下可开更大仓位）
+        self.open_type = "cross"  # 全仓模式（你的“成本开仓”需求）
+        self.risk_percent = 0.01  # 每次开仓使用可用余额的 1%
+
+        self.open_avg_price = None  # 开仓价格
+        self.current_amount = None  # 持仓量
+
+    def get_klines(self):
+        """获取最近3根30分钟K线（step=30）"""
+        try:
+            end_time = int(time.time())
+            # 获取足够多的条目确保有最新3根
+            response = self.contractAPI.get_kline(
+                contract_symbol=self.contract_symbol,
+                step=30,  # 30分钟
+                start_time=end_time - 3600 * 10,  # 取最近10小时
+                end_time=end_time
+            )[0]["data"]
+
+            # 每根: [timestamp, open, high, low, close, volume]
+            formatted = []
+            for k in response:
+                formatted.append({
+                    'id': int(k["timestamp"]),
+                    'open': float(k["open_price"]),
+                    'high': float(k["high_price"]),
+                    'low': float(k["low_price"]),
+                    'close': float(k["close_price"])
+                })
+            formatted.sort(key=lambda x: x['id'])
+            return formatted  # 最近3根: kline_1 (最老), kline_2, kline_3 (最新)
+        except Exception as e:
+            logger.error(f"获取K线异常: {e}")
+            self.ding(error=True, msg="获取K线异常")
+            return None
+
+    def get_current_price(self):
+        """获取当前最新价格，用于计算张数"""
+        try:
+            end_time = int(time.time())
+            response = self.contractAPI.get_kline(
+                contract_symbol=self.contract_symbol,
+                step=1,  # 1分钟
+                start_time=end_time - 3600 * 3,  # 取最近10小时
+                end_time=end_time
+            )[0]
+            if response['code'] == 1000:
+                return float(response['data'][-1]["close_price"])
+            return None
+        except Exception as e:
+            logger.error(f"获取价格异常: {e}")
+            return None
+
+    def get_available_balance(self):
+        """获取合约账户可用USDT余额"""
+        try:
+            response = self.contractAPI.get_assets_detail()[0]
+            if response['code'] == 1000:
+                data = response['data']
+                if isinstance(data, dict):
+                    return float(data.get('available_balance', 0))
+                elif isinstance(data, list):
+                    for asset in data:
+                        if asset.get('currency') == 'USDT':
+                            return float(asset.get('available_balance', 0))
+            return None
+        except Exception as e:
+            logger.error(f"余额查询异常: {e}")
+            return None
+
+    # 获取当前持仓方向
+    def get_position_status(self):
+        """获取当前持仓方向"""
+        try:
+            response = self.contractAPI.get_position(contract_symbol=self.contract_symbol)[0]
+            if response['code'] == 1000:
+                positions = response['data']
+                if not positions:
+                    self.start = 0
+                    return True
+                self.start = 1 if positions[0]['position_type'] == 1 else -1
+                self.open_avg_price = positions[0]['open_avg_price']
+                self.current_amount = positions[0]['current_amount']
+                self.position_cross = positions[0]["position_cross"]
+                return True
+
+            else:
+                return False
+
+        except Exception as e:
+            logger.error(f"持仓查询异常: {e}")
+            return False
+
+    # 设置杠杆和全仓
+    def set_leverage(self):
+        """程序启动时设置全仓 + 高杠杆"""
+        try:
+            response = self.contractAPI.post_submit_leverage(
+                contract_symbol=self.contract_symbol,
+                leverage=self.leverage,
+                open_type=self.open_type
+            )[0]
+            if response['code'] == 1000:
+                logger.success(f"全仓模式 + {self.leverage}x 杠杆设置成功")
+                return True
+            else:
+                logger.error(f"杠杆设置失败: {response}")
+                return False
+        except Exception as e:
+            logger.error(f"设置杠杆异常: {e}")
+            return False
+
+    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
+
+    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 action(self):
+        # 启动时设置全仓高杠杆
+        if not self.set_leverage():
+            logger.error("杠杆设置失败，程序继续运行但可能下单失败")
+            return
+
+        # 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.click_safe('x://span[normalize-space(text()) ="市价"]')
+
+
+if __name__ == '__main__':
+    BitmartFuturesTransaction(tge_id=196495).action()