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

dededdew

Browse Source
This commit is contained in:
27942
2025-10-14 10:47:40 +08:00
parent a269989e4b
commit c01f0465a4
4 changed files with 271 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

171
bit_tools.py Normal file
View File

@@ -0,0 +1,171 @@
import requests
import json
import time
# 官方文档地址
# https://doc2.bitbrowser.cn/jiekou/ben-di-fu-wu-zhi-nan.html
# 此demo仅作为参考使用，以下使用的指纹参数仅是部分参数，完整参数请参考文档
url = "http://127.0.0.1:54345"
headers = {'Content-Type': 'application/json'}
tg_url = "https://web.telegram.org/a/"
def createBrowser(
groupId=None
): # 创建或者更新窗口，指纹参数 browserFingerPrint 如没有特定需求，只需要指定下内核即可，如果需要更详细的参数，请参考文档
json_data = {
"groupId": groupId, # 分组id
'name': 'google', # 窗口名称
'remark': '', # 备注
'proxyMethod': 2, # 代理方式 2自定义 3 提取IP
# 代理类型 ['noproxy', 'http', 'https', 'socks5', 'ssh']
'proxyType': 'noproxy',
'host': '', # 代理主机
'port': '', # 代理端口
'proxyUserName': '', # 代理账号
"browserFingerPrint": { # 指纹对象
'coreVersion': '124' # 内核版本，注意，win7/win8/winserver 2012 已经不支持112及以上内核了，无法打开
}
}
res = requests.post(f"{url}/browser/update",
data=json.dumps(json_data), headers=headers).json()
browserId = res['data']['id']
return browserId
def updateBrowser(): # 更新窗口，支持批量更新和按需更新，ids 传入数组，单独更新只传一个id即可，只传入需要修改的字段即可，比如修改备注，具体字段请参考文档，browserFingerPrint指纹对象不修改，则无需传入
json_data = {'ids': ['93672cf112a044f08b653cab691216f0'],
'remark': '我是一个备注', 'browserFingerPrint': {}}
res = requests.post(f"{url}/browser/update/partial",
data=json.dumps(json_data), headers=headers).json()
print(res)
def openBrowser(id): # 直接指定ID打开窗口，也可以使用 createBrowser 方法返回的ID
json_data = {"id": f'{id}', "args": [
# "--disable-application-cache",
# "--disable-cache",
# "--disable-gpu-shader-disk-cache",
# "--media-cache-size=1",
# "--disk-cache-size=1",
# "--incognito"
]}
res = requests.post(f"{url}/browser/open",
data=json.dumps(json_data), headers=headers).json()
return res["data"]["http"].split(":")[1]
def closeBrowser(id): # 关闭窗口
json_data = {'id': f'{id}'}
res = requests.post(f"{url}/browser/close",
data=json.dumps(json_data), headers=headers)
return res.json()
def deleteBrowser(id): # 删除窗口
json_data = {'id': f'{id}'}
print(requests.post(f"{url}/browser/delete",
data=json.dumps(json_data), headers=headers).json())
def query_bit_browser(page, page_size):
data = {"page": page, "pageSize": page_size, 'sort': 'asc'}
res = requests.post(f'{url}/browser/list', data=json.dumps(data), headers=headers)
return res.json()["data"]["list"]
def update_proxy_Browser(
id,
host, port, proxyUserName, proxyPassword
):
json_data = {
"ids": [id],
# "ipCheckService": "ip123in",
"proxyMethod": 2,
"proxyType": "socks5",
"host": host,
"port": port,
"proxyUserName": proxyUserName,
"proxyPassword": proxyPassword
}
res = requests.post(f'{url}/browser/proxy/update', data=json.dumps(json_data), headers=headers)
return res.json()
def get_group_lists_Browser():
json_data = {
"page": 0,
"pageSize": 100,
"all": True
}
res = requests.post(f'{url}/group/list', data=json.dumps(json_data), headers=headers)
return res.json()["data"]["list"]
def get_browser_lists_Browser(id):
json_data = {
"groupId": id,
"page": 0,
"pageSize": 100
}
res = requests.post(f'{url}/browser/list', data=json.dumps(json_data), headers=headers)
return res.json()["data"]["list"]
def get_group_lists(): # 获取全部分组的信息
# url = "/group/list"
json_data = {
"page": 0,
"pageSize": 10,
"all": True
}
res = requests.post(f'{url}/group/list', data=json.dumps(json_data), headers=headers)
data = {}
for i in res.json()["data"]["list"]:
data[i["groupName"]] = i["id"]
return data
def group_add(groupName):
json_data = {
"groupName": groupName,
"sortNum": 0
}
res = requests.post(f'{url}/group/add', data=json.dumps(json_data), headers=headers)
return res.json()
# if __name__ == '__main__':
# for i in get_group_lists():
# print(i)
# pass
# browser_id = createBrowser()
# openBrowser(browser_id)
#
# time.sleep(10) # 等待10秒自动关闭窗口
#
# closeBrowser(browser_id)
#
# time.sleep(10) # 等待10秒自动删掉窗口
#
# deleteBrowser(browser_id)
# deleteBrowser(id="43f0d0978f6a4bb7bbcb0b7786f436e9")

BIN
models/database.db
View File

Binary file not shown.

11
weex/自动化抓取数据.py
View File

@@ -1,16 +1,23 @@
from DrissionPage import ChromiumPage, ChromiumOptions
from bit_tools import openBrowser
from models.weex import Weex1
if __name__ == '__main__':
bit_port = openBrowser(id="8dcb4f744cf64ab190e465e153088515")
co = ChromiumOptions()
co.set_local_port(1001)
co.set_local_port(port=bit_port)
co = ChromiumOptions()
co.set_local_port(bit_port)
page = ChromiumPage(addr_or_opts=co)
page.set.window.max()
page.listen.start("gateway1.janapw.com/api/v1/public/quote/v1/getKlineV2")
page.listen.start("https://http-gateway2.janapw.com/api/v1/public/quote/v1/getKlineV2")
page.get(url="https://www.weeaxs.site/zh-CN/futures/ETH-USDT")

109
weex/读取数据库分析数据2.0.py
View File

@@ -1,9 +1,82 @@
"""
量化交易回测系统 - 优化版
功能：基于包住形态的交易信号识别和回测分析
作者：量化交易团队
版本：2.0
"""
import datetime
from typing import List, Dict, Tuple, Optional, Any
from dataclasses import dataclass
from loguru import logger
from peewee import fn
from models.weex import Weex15, Weex1
# ===============================================================
# 📊 配置管理类
# ===============================================================
@dataclass
class BacktestConfig:
"""回测配置类"""
# 交易参数
take_profit: float = 8.0 # 止盈点数
stop_loss: float = -1.0 # 止损点数
contract_size: float = 10000 # 合约规模
open_fee: float = 5.0 # 开仓手续费
close_fee_rate: float = 0.0005 # 平仓手续费率
# 回测日期范围
start_date: str = "2025-7-1"
end_date: str = "2025-7-31"
# 信号参数
enable_bear_bull_engulf: bool = True # 涨包跌信号
enable_bull_bear_engulf: bool = True # 跌包涨信号
def __post_init__(self):
"""验证配置参数"""
if self.take_profit <= 0:
raise ValueError("止盈点数必须大于0")
if self.stop_loss >= 0:
raise ValueError("止损点数必须小于0")
@dataclass
class TradeRecord:
"""交易记录类"""
entry_time: datetime.datetime
exit_time: datetime.datetime
signal_type: str
direction: str
entry_price: float
exit_price: float
profit_loss: float
profit_amount: float
total_fee: float
net_profit: float
@dataclass
class SignalStats:
"""信号统计类"""
signal_name: str
count: int = 0
wins: int = 0
total_profit: float = 0.0
@property
def win_rate(self) -> float:
"""胜率计算"""
return (self.wins / self.count * 100) if self.count > 0 else 0.0
@property
def avg_profit(self) -> float:
"""平均盈利"""
return self.total_profit / self.count if self.count > 0 else 0.0
# ===============================================================
# 📊 数据获取模块
# ===============================================================
@@ -58,9 +131,9 @@ def check_signal(prev, curr):
if is_bullish(curr) and is_bearish(prev) and c_open <= p_close and c_close >= p_open:
return "long", "bear_bull_engulf"
# 前涨后跌包住 -> 做空
if is_bearish(curr) and is_bullish(prev) and c_open >= p_close and c_close <= p_open:
return "short", "bull_bear_engulf"
# # 前涨后跌包住 -> 做空
# if is_bearish(curr) and is_bullish(prev) and c_open >= p_close and c_close <= p_open:
# return "short", "bull_bear_engulf"
return None, None
@@ -180,17 +253,17 @@ if __name__ == '__main__':
total_profit = sum(t['diff'] / t['entry'] * 10000 for t in trades)
total_fee = sum(5 + 10000 / t['entry'] * t['exit'] * 0.0005 for t in trades)
logger.info("===== 每笔交易详情 =====")
for t in trades:
logger.info(f"{t['entry_time']} {t['direction']}({t['signal']}) "
f"入场={t['entry']:.2f} 出场={t['exit']:.2f} 出场时间={t['exit_time']} "
f"差价={t['diff']:.2f}")
print(i1, i)
print(f"\n一共交易笔数：{len(trades)}")
print(f"一共盈利：{total_profit:.2f}")
print(f"一共手续费：{total_fee:.2f}")
print(f"净利润：{total_profit - total_fee:.2f}")
# logger.info("===== 每笔交易详情 =====")
# for t in trades:
# logger.info(f"{t['entry_time']} {t['direction']}({t['signal']}) "
# f"入场={t['entry']:.2f} 出场={t['exit']:.2f} 出场时间={t['exit_time']} "
# f"差价={t['diff']:.2f}")
#
# print(i1, i)
# print(f"\n一共交易笔数：{len(trades)}")
# print(f"一共盈利：{total_profit:.2f}")
# print(f"一共手续费：{total_fee:.2f}")
# print(f"净利润：{total_profit - total_fee:.2f}")
if total_profit > total_fee * 0.1:
print(i1, i)
@@ -201,7 +274,7 @@ if __name__ == '__main__':
print("\n===== 信号统计 =====")
for k, v in stats.items():
win_rate = (v['wins'] / v['count'] * 100) if v['count'] > 0 else 0
print(
f"{v['name']} ({k}) - 信号数: {v['count']} | 胜率: {win_rate:.2f}% | 总盈利: {v['total_profit']:.2f}")
# for k, v in stats.items():
# win_rate = (v['wins'] / v['count'] * 100) if v['count'] > 0 else 0
# print(
# f"{v['name']} ({k}) - 信号数: {v['count']} | 胜率: {win_rate:.2f}% | 总盈利: {v['total_profit']:.2f}")