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

fsefesf

Browse Source
This commit is contained in:
27942
2025-10-03 18:24:43 +08:00
parent dea758323f
commit 53e9936cc0
2 changed files with 136 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
回测数据/haha
View File

@@ -120,8 +120,15 @@ if __name__ == '__main__':
这样就是当前价格从k线图中看，就是前一笔的开盘价和结盘价处于当前一笔的开盘价和结盘价中间包裹住的情况，然后需要查询出来
# 第一种情况：当前一笔是涨的，前一笔是跌的，当前一笔的开盘价格低于前一笔的结盘价格，当前一笔的结盘价格大于前一笔的开盘价格，
# 第二种情况：当前一笔是跌的，前一笔是涨的，当前一笔的开盘价格大于前一笔的结盘价格，当前价格的结盘价格小于前一笔的开盘价格，
ui:8000
前端：15000
一次性，

221
回测数据/test2.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,120 +22,156 @@ headers = {
def fetch_kline(day: int):
"""获取某一天的分钟K线数据"""
"""获取某一天的分钟级 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()),
'symbol': 'ETH-USDT', # 交易对
'period': '1min', # 分钟级K线
'start': int(start_of_day.timestamp()), # 开始时间
'end': int(end_of_day.timestamp()), # 结束时间
}
# 请求 API 获取数据
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 is_bullish(candle):
"""判断是否是阳线（收盘价 > 开盘价）"""
return float(candle['close']) > float(candle['open'])
def is_bearish(candle):
"""判断是否是阴线（收盘价 < 开盘价）"""
return float(candle['close']) < float(candle['open'])
def check_signal(prev, curr):
"""
判断是否出现反包形态：
1. 前一根阴线，后一根阳线，并且阳线包住前一根阴线 -> 做多信号
2. 前一根阳线，后一根阴线，并且阴线包住前一根阳线 -> 做空信号
"""
p_open, p_close = float(prev['open']), float(prev['close'])
c_open, c_close = float(curr['open']), float(curr['close'])
# 情况1：前阴后阳，且阳线包住前阴
if is_bullish(curr) and is_bearish(prev):
if c_open < p_close and c_close > p_open:
return "long"
# 情况2：前阳后阴，且阴线包住前阳
if is_bearish(curr) and is_bullish(prev):
if c_open > p_close and c_close < p_open:
return "short"
return None
def simulate_trade(direction, entry_price, future_candles, capital=10000, take_profit_ratio=5,
stop_loss_ratio=-2):
"""
模拟交易（逐根K线回测）
使用资金金额来控制止盈止损：
- 盈利达到 capital * take_profit_ratio 就止盈
- 亏损达到 capital * stop_loss_ratio 就止损
"""
for candle in future_candles:
high, low, close = float(candle['high']), float(candle['low']), float(candle['close'])
if direction == "long":
if high - entry_price >= take_profit_ratio:
return high, take_profit_ratio, candle['id'] # 止盈
if low - entry_price <= stop_loss_ratio:
return low, stop_loss_ratio, candle['id'] # 止损
elif direction == "short":
if entry_price - low >= take_profit_ratio:
return high, take_profit_ratio, candle['id'] # 止盈
if entry_price - low <= stop_loss_ratio:
return low, stop_loss_ratio, candle['id'] # 止损
# 如果未来都没触发，最后一根收盘平仓
final_price = float(future_candles[-1]['close'])
if direction == "long":
diff_money = (final_price - entry_price) / entry_price * capital
else:
diff_money = (entry_price - final_price) / entry_price * capital
return final_price, diff_money, future_candles[-1]['id']
# ================= 主程序 =================
if __name__ == '__main__':
if __name__ == '__main__':
zh_project = 0 # 累计盈亏
all_trades = [] # 保存所有交易明细
zh_project = 0
all_trades = [] # 存放所有交易记录
# 遍历 9月1日 ~ 9月27日
for i in range(1, 30):
sorted_data = fetch_kline(i) # 获取数据
for i in range(27, 28):
signals = wins = 0
lig_price = low_price = 0 # 分别记录多/空的净收益
sorted_data = fetch_kline(i) # 获取交易数据
# 遍历每根K线，寻找信号
for idx in range(1, len(sorted_data) - 2): # 留出未来K线
prev, curr = sorted_data[idx - 1], sorted_data[idx] # 前一笔，当前一笔
entry_candle = sorted_data[idx + 1] # 下一根开盘价作为入场价
future_candles = sorted_data[idx + 2:] # 未来行情
# 信号数 盈利的笔数
signals = wins = 0
# 涨方向盈利多少 跌方向盈利多少
lig_price = low_price = 0
entry_open = float(entry_candle['open']) # 开仓价格
direction = check_signal(prev, curr) # 判断开仓方向
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线
if direction:
signals += 1 # 总共信号数
exit_price, diff, exit_time = simulate_trade(direction, entry_open, future_candles)
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 direction == "long":
lig_price += diff
if diff > 0: wins += 1
else:
low_price += diff
if diff > 0: wins += 1
# 当前为涨
if curr_close > curr_open:
# 前一笔涨 + 包裹
if prev_close > prev_open and curr_open < prev_open and curr_close > prev_close:
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"], "做多" if direction == "long" else "做空",
entry_open, exit_price, diff, exit_time))
if future_close > entry_open:
wins += 1
# 输出每日统计结果
if signals > 0:
logger.info(
f"日期：{i}号，信号数={signals}, 胜率={wins / signals * 100:.2f}%，"
f"上涨方向={lig_price:.2f}，下跌方向={low_price:.2f}，综合={lig_price + low_price:.2f}"
)
else:
logger.info(f"日期：{i}号，没有信号")
# 前一笔跌 + 反包
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"]))
# 累计盈亏
zh_project += (lig_price + low_price)
if future_close > entry_open:
wins += 1
logger.success(f"综合价格：{zh_project:.2f}")
# 当前为跌
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("===== 每笔交易详情 =====")
n = n1 = 0 # n = 总盈利，n1 = 总手续费
for date, time_str, direction, entry, exit, diff, end_time in all_trades:
logger.info(
f"日期：{i}号，信号数={signals}, 胜率={wins / signals * 100:.2f}%，上涨方向：{lig_price:.2f}，下跌方向：{low_price:.2f}，综合价格：{(lig_price + low_price):.2f}"
f"{date} {time_str} {direction} 入场={entry:.2f} 出场={exit:.2f} 出场时间={end_time} "
f"差价={diff:.2f} 盈利={diff / entry * 10000:.2f} "
f"开仓手续费=5u 平仓手续费={10000 / entry * exit * 0.0005:.2f}"
)
else:
logger.info(f"日期：{i}号，没有信号")
n1 += 5 + (10000 / entry * exit * 0.0005)
n += (diff / entry) * 10000
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}')
print(f'一共笔数：{len(all_trades)}')
print(f"一共盈利：{n:.2f}")
print(f'一共手续费：{n1:.2f}')