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 14:55:59 +08:00
parent 669490a857
commit 5f0c14c9ef
2 changed files with 120 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
回测数据/test.py
View File

@@ -1 +1,5 @@
print(10000 * 0.001)
print(4000 * 1.001)
print(3954.11 * 1.002)
print(4000 * 0.998)
print(4000 * 0.998)
print(4000 * 0.998)

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

@@ -2,6 +2,7 @@ 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',
@@ -21,67 +22,133 @@ headers = {
def fetch_kline(day: int):
"""获取某一天的分钟K线数据"""
"""
获取某一天的分钟K线数据
:param day: 日期中的“日”，比如27表示2025-09-27
:return: 按时间顺序排序后的K线数据
"""
time_ser = datetime.datetime(2025, 9, day)
# 当天 0点
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)
# API 参数
params = {
'symbol': 'ETH-USDT',
'period': '1min',
'start': int(start_of_day.timestamp()),
'end': int(end_of_day.timestamp()),
'symbol': 'ETH-USDT', # 交易对
'period': '1min', # 1分钟K线
'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']
# 返回按 id(时间) 排序的列表
return sorted(data, key=lambda x: x['id'])
def backtest_kline(sorted_data, day):
"""执行回测逻辑"""
signals = wins = 0
lig_price = low_price = 0
all_trades = []
"""
回测策略逻辑
规则：
- 做多：连续两根阳线，第二根包住第一根
- 做空：连续两根阴线，第二根包住第一根
止盈止损：
- 多单：止盈+0.15%，止损-0.2%
- 空单：止盈-0.15%，止损+0.2%
"""
signals = wins = 0 # 统计信号总数、胜率
lig_price = low_price = 0 # 多单收益、空单收益
all_trades = [] # 存储所有交易详情
# 遍历 K 线，至少需要 3 根（前一根+当前+下一根开仓）
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 = sorted_data[idx - 1] # 前一根K线
curr = sorted_data[idx] # 当前K线
entry_candle = sorted_data[idx + 1] # 下一根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'])
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:
trade_signal = None # 信号类型：long / short
# ======================== 做多条件 ========================
if (
curr_close > curr_open and prev_close > prev_open and # 连续两根阳线
curr_open < prev_open and curr_close > prev_close # 当前包住前一根
):
trade_signal = "long"
# ======================== 做空条件 ========================
elif (
curr_close < curr_open and prev_close < prev_open and # 连续两根阴线
curr_open > prev_open and curr_close < prev_close # 当前包住前一根
):
trade_signal = "short"
# ======================== 执行交易 ========================
if trade_signal:
signals += 1
take_profit = entry_open * 1.10 # 止盈 10%
stop_loss = entry_open * 0.95 # 止损 5%
exit_price = None
exit_time = None
# ✅ 往后遍历直到触发止盈 / 止损
# 设定止盈止损
if trade_signal == "long":
take_profit = entry_open * 1.002
stop_loss = entry_open * 0.998
else:
take_profit = entry_open * 0.9985
stop_loss = entry_open * 1.001
exit_price = None # 出场价格
exit_time = None # 出场时间
# 遍历后续K线，寻找止盈/止损
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 trade_signal == "long":
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
else: # short
if low <= take_profit: # 触发止盈
exit_price = take_profit
exit_time = future["id"]
break
if high >= 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 trade_signal == "long":
diff = exit_price - entry_open # 多单盈亏
lig_price += diff
else:
diff = entry_open - exit_price # 空单盈亏
low_price += diff
# 保存交易详情
all_trades.append(
(f"{day}号", entry_candle["id"], "做多" if trade_signal == "long" else "做空",
entry_open, exit_price, diff, exit_time)
)
# 统计胜率
if diff > 0:
wins += 1
@@ -92,13 +159,17 @@ if __name__ == '__main__':
zh_project = 0
all_trades = []
# 回测 9月27日的数据
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}"
f"日期：{i}号，信号数={signals}, 胜率={wins / signals * 100:.2f}%，"
f"上涨方向：{lig_price:.2f}，下跌方向：{low_price:.2f}，"
f"综合价格：{(lig_price + low_price):.2f}"
)
else:
logger.info(f"日期：{i}号，没有信号")
@@ -106,22 +177,27 @@ if __name__ == '__main__':
zh_project += (lig_price + low_price)
all_trades.extend(trades)
# ======================== 汇总输出 ========================
logger.success(f"综合价格：{zh_project:.2f}")
# 输出每笔交易
logger.info("===== 每笔交易详情 =====")
n = n1 = 0
n = n1 = 0 # n: 总盈利, n1: 总手续费
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
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"{date} {time_str} {direction} 入场={entry:.2f} 出场={exit:.2f} 出场时间：{end_time} "
f"差价={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}')