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

gfrdegdergr

Browse Source
This commit is contained in:
27942
2025-09-30 13:58:23 +08:00
parent 53806a6c2a
commit 669490a857
4 changed files with 320 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
回测数据/test.py
View File

@@ -1,79 +1 @@
import datetime
import requests
headers = {
'accept': 'application/json, text/plain, */*',
'accept-language': 'zh,zh-CN;q=0.9,zh-HK;q=0.8,en;q=0.7',
'cache-control': 'no-cache',
'origin': 'https://www.websea.com',
'pragma': 'no-cache',
'priority': 'u=1, i',
'referer': 'https://www.websea.com/',
'sec-ch-ua': '"Chromium";v="140", "Not=A?Brand";v="24", "Google Chrome";v="140"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
'sec-fetch-dest': 'empty',
'sec-fetch-mode': 'cors',
'sec-fetch-site': 'same-site',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36',
}
if __name__ == '__main__':
datas = []
# 定义开始日期和结束日期
time_ser = datetime.datetime(2025, 9, 25)
# 获取当天开始时刻（00:00:00）
start_of_day = time_ser.replace(hour=0, minute=0, second=0, microsecond=0)
# 获取当天结束时刻（23:59:59）
end_of_day = time_ser.replace(hour=23, minute=59, second=59, microsecond=0)
# 将开始时刻和结束时刻转换为时间戳
start_timestamp = start_of_day.timestamp()
end_timestamp = end_of_day.timestamp()
params = {
'symbol': 'ETH-USDT',
'period': '1min',
'start': int(start_timestamp),
'end': int(end_timestamp),
}
response = requests.get('https://capi.websea.com/webApi/market/getKline', params=params, headers=headers)
# 提取数据
data = response.json()['result']['data']
# 根据 id 进行排序
sorted_data = sorted(data, key=lambda x: x['id'])
prev_data = None
n = 0
for i in range(1, len(sorted_data)):
current_data = sorted_data[i]
prev_data = sorted_data[i - 1]
# 当前一笔是涨的
if float(current_data['open']) < float(current_data['close']):
# 前一笔是涨的
if float(prev_data['open']) < float(prev_data['close']):
if float(current_data['open']) < float(prev_data['open']) and float(current_data['close']) > float(
prev_data['close']):
print(f"前一笔数据: {prev_data}")
print(f"当前一笔数据: {current_data}")
print("做多" + "-" * 50)
n += 1
# 前一笔是跌的
else:
if float(current_data['open']) < float(prev_data['close']) and float(current_data['close']) > float(
prev_data['open']):
print(f"前一笔数据: {prev_data}")
print(f"当前一笔数据: {current_data}")
print("做多" + "-" * 50)
n += 1
print(n)
print(10000 * 0.001)

88
回测数据/test2.py
View File

@@ -41,18 +41,20 @@ if __name__ == '__main__':
zh_project = 0
all_trades = [] # 存放所有交易记录
for i in range(30, 31):
for i in range(27, 28):
sorted_data = fetch_kline(i)
sorted_data = fetch_kline(i) # 获取交易数据
# 信号数 盈利的笔数
signals = wins = 0
# 涨方向盈利多少 跌方向盈利多少
lig_price = low_price = 0
for idx in range(1, len(sorted_data) - 6):
prev = sorted_data[idx - 1] # 前一根K
curr = sorted_data[idx] # 当前K
for idx in range(1, len(sorted_data) - 4):
prev = sorted_data[idx - 1] # 前一根K
curr = sorted_data[idx] # 当前K
entry_candle = sorted_data[idx + 1] # ✅ 下一根K线作为开仓点
future = sorted_data[idx + 5] # 5根后的K线
future = sorted_data[idx + 4] # 5根后的K线
prev_open, prev_close = float(prev['open']), float(prev['close'])
curr_open, curr_close = float(curr['open']), float(curr['close'])
@@ -66,42 +68,46 @@ if __name__ == '__main__':
signals += 1
diff = future_close - entry_open
lig_price += diff
all_trades.append((f"{i}号", entry_candle["id"], "做多", entry_open, future_close, diff, future["id"]))
all_trades.append(
(f"{i}号", entry_candle["id"], "做多", entry_open, future_close, diff, future["id"]))
if future_close > entry_open:
wins += 1
# # 前一笔跌 + 反包
# elif prev_close < prev_open and curr_open < prev_close and curr_close > prev_open:
# signals += 1
# diff = future_close - entry_open
# lig_price += diff
# all_trades.append((f"{i}号", entry_candle["id"], "做多", entry_open, future_close, diff, future["id"]))
#
# if future_close > entry_open:
# wins += 1
#
# # 当前为跌
# elif curr_close < curr_open:
# # 前一笔跌 + 包裹
# if prev_close < prev_open and curr_open > prev_open and curr_close < prev_close:
# signals += 1
# diff = entry_open - future_close
# low_price += diff
# all_trades.append((f"{i}号", entry_candle["id"], "做空", entry_open, future_close, diff, future["id"]))
#
# if future_close < entry_open:
# wins += 1
#
# # 前一笔涨 + 反包
# elif prev_close > prev_open and curr_open > prev_close and curr_close < prev_open:
# signals += 1
# diff = entry_open - future_close
# low_price += diff
# all_trades.append((f"{i}号", entry_candle["id"], "做空", entry_open, future_close, diff, future["id"]))
#
# if future_close < entry_open:
# wins += 1
# 前一笔跌 + 反包
elif prev_close < prev_open and curr_open < prev_close and curr_close > prev_open:
signals += 1
diff = future_close - entry_open
lig_price += diff
all_trades.append(
(f"{i}号", entry_candle["id"], "做多", entry_open, future_close, diff, future["id"]))
if future_close > entry_open:
wins += 1
# 当前为跌
elif curr_close < curr_open:
# 前一笔跌 + 包裹
if prev_close < prev_open and curr_open > prev_open and curr_close < prev_close:
signals += 1
diff = entry_open - future_close
low_price += diff
all_trades.append(
(f"{i}号", entry_candle["id"], "做空", entry_open, future_close, diff, future["id"]))
if future_close < entry_open:
wins += 1
# 前一笔涨 + 反包
elif prev_close > prev_open and curr_open > prev_close and curr_close < prev_open:
signals += 1
diff = entry_open - future_close
low_price += diff
all_trades.append(
(f"{i}号", entry_candle["id"], "做空", entry_open, future_close, diff, future["id"]))
if future_close < entry_open:
wins += 1
if signals > 0:
logger.info(
@@ -129,6 +135,6 @@ if __name__ == '__main__':
n1 += 5 + (10000 / entry * exit * 0.0005)
n += (diff / entry) * 10000
print(len(all_trades))
print(n)
print(n1)
print(f'一共笔数：{len(all_trades)}')
print(f"一共盈利：{n}")
print(f'一共手续费：{n1}')

145
回测数据/推测策略，回测.py Normal file
View File

@@ -0,0 +1,145 @@
import datetime
import requests
from loguru import logger
headers = {
'accept': 'application/json, text/plain, */*',
'accept-language': 'zh,zh-CN;q=0.9,zh-HK;q=0.8,en;q=0.7',
'cache-control': 'no-cache',
'origin': 'https://www.websea.com',
'pragma': 'no-cache',
'priority': 'u=1, i',
'referer': 'https://www.websea.com/',
'sec-ch-ua': '"Chromium";v="140", "Not=A?Brand";v="24", "Google Chrome";v="140"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
'sec-fetch-dest': 'empty',
'sec-fetch-mode': 'cors',
'sec-fetch-site': 'same-site',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36',
}
def fetch_kline(day: int):
"""获取某一天的分钟K线数据"""
time_ser = datetime.datetime(2025, 9, day)
start_of_day = time_ser.replace(hour=0, minute=0, second=0, microsecond=0)
end_of_day = time_ser.replace(hour=23, minute=59, second=59, microsecond=0)
params = {
'symbol': 'ETH-USDT',
'period': '1min',
'start': int(start_of_day.timestamp()),
'end': int(end_of_day.timestamp()),
}
response = requests.get('https://capi.websea.com/webApi/market/getKline', params=params, headers=headers)
data = response.json()['result']['data']
return sorted(data, key=lambda x: x['id'])
if __name__ == '__main__':
zh_project = 0
all_trades = [] # 存放所有交易记录
for i in range(27, 28):
sorted_data = fetch_kline(i) # 获取交易数据
# 信号数 盈利的笔数
signals = wins = 0
# 涨方向盈利多少 跌方向盈利多少
lig_price = low_price = 0
for idx in range(1, len(sorted_data) - 4):
prev = sorted_data[idx - 1] # 前一根K
curr = sorted_data[idx] # 当前K
entry_candle = sorted_data[idx + 1] # ✅ 下一根K线作为开仓点
future = sorted_data[idx + 4] # 5根后的K线
prev_open, prev_close = float(prev['open']), float(prev['close']) # 前一笔开盘，前一笔结盘
curr_open, curr_close = float(curr['open']), float(curr['close']) # 当前一笔开盘，当前一笔结盘
entry_open = float(entry_candle['open']) # 下一笔开仓
future_close = float(future['close']) # 四根后结盘
# 当前为涨
if curr_close > curr_open:
# 前一笔涨 + 包裹
if prev_close > prev_open and curr_open < prev_open and curr_close > prev_close:
future = sorted_data[idx + 7]
future_close = float(future['close']) # 四根后结盘
signals += 1
diff = future_close - entry_open
lig_price += diff
if future_close > entry_open:
wins += 1
all_trades.append(
(f"{i}号", entry_candle["id"], "做多", entry_open, future_close, diff, future["id"]))
# # 前一笔跌 + 反包
# elif prev_close < prev_open and curr_open < prev_close and curr_close > prev_open:
# signals += 1
# diff = future_close - entry_open
# lig_price += diff
# all_trades.append((f"{i}号", entry_candle["id"], "做多", entry_open, future_close, diff, future["id"]))
#
# if future_close > entry_open:
# wins += 1
#
# # 当前为跌
# elif curr_close < curr_open:
# # 前一笔跌 + 包裹
# if prev_close < prev_open and curr_open > prev_open and curr_close < prev_close:
# signals += 1
# diff = entry_open - future_close
# low_price += diff
# all_trades.append((f"{i}号", entry_candle["id"], "做空", entry_open, future_close, diff, future["id"]))
#
# if future_close < entry_open:
# wins += 1
#
# # 前一笔涨 + 反包
# elif prev_close > prev_open and curr_open > prev_close and curr_close < prev_open:
# signals += 1
# diff = entry_open - future_close
# low_price += diff
# all_trades.append((f"{i}号", entry_candle["id"], "做空", entry_open, future_close, diff, future["id"]))
#
# if future_close < entry_open:
# wins += 1
if signals > 0:
logger.info(
f"日期：{i}号，信号数={signals}, 胜率={wins / signals * 100:.2f}%，上涨方向：{lig_price:.2f}，下跌方向：{low_price:.2f}，综合价格：{(lig_price + low_price):.2f}"
)
else:
logger.info(f"日期：{i}号，没有信号")
zh_project += (lig_price + low_price)
logger.success(f"综合价格：{zh_project:.2f}")
# 输出每笔交易
logger.info("===== 每笔交易详情 =====")
n = n1 = 0
for trade in all_trades:
date, time_str, direction, entry, exit, diff, end_time = trade
logger.info(
f"{date} {time_str} {direction} 入场={entry:.2f} 出场={exit:.2f} 出场时间：{end_time} 差价={diff:.2f} 盈利：{diff / entry * 10000} "
f"开仓手续费：5u 平仓手续费：{10000 / entry * exit * 0.0005}"
)
n1 += 5 + (10000 / entry * exit * 0.0005)
n += (diff / entry) * 10000
print(f'一共笔数：{len(all_trades)}')
print(f"一共盈利：{n}")
print(f'一共手续费：{n1}')

127
回测数据/通过止盈止损策略.py Normal file
View File

@@ -0,0 +1,127 @@
import datetime
import requests
from loguru import logger
headers = {
'accept': 'application/json, text/plain, */*',
'accept-language': 'zh,zh-CN;q=0.9,zh-HK;q=0.8,en;q=0.7',
'cache-control': 'no-cache',
'origin': 'https://www.websea.com',
'pragma': 'no-cache',
'priority': 'u=1, i',
'referer': 'https://www.websea.com/',
'sec-ch-ua': '"Chromium";v="140", "Not=A?Brand";v="24", "Google Chrome";v="140"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
'sec-fetch-dest': 'empty',
'sec-fetch-mode': 'cors',
'sec-fetch-site': 'same-site',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36',
}
def fetch_kline(day: int):
"""获取某一天的分钟K线数据"""
time_ser = datetime.datetime(2025, 9, day)
start_of_day = time_ser.replace(hour=0, minute=0, second=0, microsecond=0)
end_of_day = time_ser.replace(hour=23, minute=59, second=59, microsecond=0)
params = {
'symbol': 'ETH-USDT',
'period': '1min',
'start': int(start_of_day.timestamp()),
'end': int(end_of_day.timestamp()),
}
response = requests.get('https://capi.websea.com/webApi/market/getKline', params=params, headers=headers)
data = response.json()['result']['data']
return sorted(data, key=lambda x: x['id'])
def backtest_kline(sorted_data, day):
"""执行回测逻辑"""
signals = wins = 0
lig_price = low_price = 0
all_trades = []
for idx in range(1, len(sorted_data) - 2):
prev = sorted_data[idx - 1]
curr = sorted_data[idx]
entry_candle = sorted_data[idx + 1] # ✅ 下一根开仓
prev_open, prev_close = float(prev['open']), float(prev['close'])
curr_open, curr_close = float(curr['open']), float(curr['close'])
entry_open = float(entry_candle['open'])
# 当前为涨，前一笔涨 + 包裹 => 做多信号
if curr_close > curr_open and prev_close > prev_open and curr_open < prev_open and curr_close > prev_close:
signals += 1
take_profit = entry_open * 1.10 # 止盈 10%
stop_loss = entry_open * 0.95 # 止损 5%
exit_price = None
exit_time = None
# ✅ 往后遍历直到触发止盈 / 止损
for future in sorted_data[idx + 2:]:
high, low, close = float(future['high']), float(future['low']), float(future['close'])
# 先判断是否触发止盈
if high >= take_profit:
exit_price = take_profit
exit_time = future["id"]
break
# 判断是否触发止损
if low <= stop_loss:
exit_price = stop_loss
exit_time = future["id"]
break
# 如果没触发止盈止损，用最后一根收盘价离场
if exit_price is None:
exit_price = float(sorted_data[-1]['close'])
exit_time = sorted_data[-1]["id"]
diff = exit_price - entry_open
lig_price += diff
all_trades.append((f"{day}号", entry_candle["id"], "做多", entry_open, exit_price, diff, exit_time))
if diff > 0:
wins += 1
return signals, wins, lig_price, low_price, all_trades
if __name__ == '__main__':
zh_project = 0
all_trades = []
for i in range(27, 28):
sorted_data = fetch_kline(i)
signals, wins, lig_price, low_price, trades = backtest_kline(sorted_data, i)
if signals > 0:
logger.info(
f"日期：{i}号，信号数={signals}, 胜率={wins / signals * 100:.2f}%，上涨方向：{lig_price:.2f}，下跌方向：{low_price:.2f}，综合价格：{(lig_price + low_price):.2f}"
)
else:
logger.info(f"日期：{i}号，没有信号")
zh_project += (lig_price + low_price)
all_trades.extend(trades)
logger.success(f"综合价格：{zh_project:.2f}")
# 输出每笔交易
logger.info("===== 每笔交易详情 =====")
n = n1 = 0
for trade in all_trades:
date, time_str, direction, entry, exit, diff, end_time = trade
fee_open = 5
fee_close = 10000 / entry * exit * 0.0005
logger.info(
f"{date} {time_str} {direction} 入场={entry:.2f} 出场={exit:.2f} 出场时间：{end_time} 差价={diff:.2f} 盈利：{diff / entry * 10000:.2f} "
f"开仓手续费：{fee_open:.2f} 平仓手续费：{fee_close:.2f}"
)
n1 += fee_open + fee_close
n += (diff / entry) * 10000
print(f'一共笔数：{len(all_trades)}')
print(f"一共盈利：{n:.2f}")
print(f'一共手续费：{n1:.2f}')