diff --git a/回测数据/haha b/回测数据/haha index e13ef32..cc1e065 100644 --- a/回测数据/haha +++ b/回测数据/haha @@ -120,8 +120,15 @@ if __name__ == '__main__': 这样就是当前价格从k线图中看，就是前一笔的开盘价和结盘价处于当前一笔的开盘价和结盘价中间包裹住的情况，然后需要查询出来 +# 第一种情况：当前一笔是涨的，前一笔是跌的，当前一笔的开盘价格低于前一笔的结盘价格，当前一笔的结盘价格大于前一笔的开盘价格， +# 第二种情况：当前一笔是跌的，前一笔是涨的，当前一笔的开盘价格大于前一笔的结盘价格，当前价格的结盘价格小于前一笔的开盘价格， + + ui:8000 前端：15000 一次性， + + + diff --git a/回测数据/test2.py b/回测数据/test2.py index 61ebe56..cbb746e 100644 --- a/回测数据/test2.py +++ b/回测数据/test2.py @@ -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}')