import datetime import pandas as pd import requests from loguru import logger # ================= 辅助函数 ================= def is_bullish(candle): """判断是否是阳线（开盘价 < 收盘价，即涨）""" return float(candle['open']) < float(candle['close']) def is_bearish(candle): """判断是否是阴线（开盘价 > 收盘价，即跌）""" return float(candle['open']) > float(candle['close']) def check_signal(prev, curr): """ 判断是否出现包住形态，返回信号类型和方向： 1. 前跌后涨包住 -> 做多信号 (bear_bull_engulf) 2. 前涨后跌包住 -> 做空信号 (bull_bear_engulf) 3. 前涨后涨包住 -> 做多信号 (bull_bull_engulf) 4. 前跌后跌包住 -> 做空信号 (bear_bear_engulf) """ 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", "bear_bull_engulf" # 情况2：前涨后跌，且跌线包住前涨线 -> 做空信号 if is_bearish(curr) and is_bullish(prev): if c_open >= p_close and c_close <= p_open: return "short", "bull_bear_engulf" # # 情况3：前涨后涨，且后涨线包住前涨线 -> 做多信号 # if is_bullish(curr) and is_bullish(prev): # if c_open < p_open and c_close > p_close: # return "long", "bull_bull_engulf" # # 情况4：前跌后跌，且后跌线包住前跌线 -> 做空信号 # if is_bearish(curr) and is_bearish(prev): # if c_open > p_open and c_close < p_close: # return "short", "bear_bear_engulf" return None, None def simulate_trade(direction, entry_price, future_candles, take_profit_diff=30, stop_loss_diff=-10): """ 模拟交易（逐根K线回测） 使用价差来控制止盈止损： - 盈利达到 take_profit_diff 就止盈 - 亏损达到 stop_loss_diff 就止损 direction：信号类型 entry_price：开盘价格 future_candles：未来的行情数据 """ 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_diff: return entry_price + take_profit_diff, take_profit_diff, candle['id'] # 止盈 if low <= entry_price + stop_loss_diff: return entry_price + stop_loss_diff, stop_loss_diff, candle['id'] # 止损 elif direction == "short": # 做空：检查最低价是否达到止盈，最高价是否触及止损 if low <= entry_price - take_profit_diff: return entry_price - take_profit_diff, take_profit_diff, candle['id'] # 止盈 if high >= entry_price - stop_loss_diff: return entry_price - stop_loss_diff, stop_loss_diff, candle['id'] # 止损 # 如果未来都没触发，最后一根收盘平仓 final_price = float(future_candles[-1]['close']) if direction == "long": diff_money = final_price - entry_price else: diff_money = entry_price - final_price return final_price, diff_money, future_candles[-1]['id'] if __name__ == '__main__': zh_project = 0 # 累计盈亏 all_trades = [] # 保存所有交易明细 # 四种信号类型的统计 signal_stats = { "bear_bull_engulf": {"count": 0, "wins": 0, "total_profit": 0, "name": "涨包跌"}, "bull_bear_engulf": {"count": 0, "wins": 0, "total_profit": 0, "name": "跌包涨"}, # "bull_bull_engulf": {"count": 0, "wins": 0, "total_profit": 0, "name": "涨包涨"}, # "bear_bear_engulf": {"count": 0, "wins": 0, "total_profit": 0, "name": "跌包跌"} } headers = { 'accept': 'application/json, text/plain, */*', 'accept-language': 'zh-CN,zh;q=0.9', 'appversion': '2.0.0', 'bundleid': '', 'cache-control': 'no-cache', 'language': 'zh_CN', 'origin': 'https://www.weeaxs.site', 'pragma': 'no-cache', 'priority': 'u=1, i', 'referer': 'https://www.weeaxs.site/', 'sec-ch-ua': '"Google Chrome";v="141", "Not?A_Brand";v="8", "Chromium";v="141"', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-platform': '"Windows"', 'sec-fetch-dest': 'empty', 'sec-fetch-mode': 'cors', 'sec-fetch-site': 'cross-site', 'sidecar': '0148e5be00be63a67540447fd47c4d0977aeb6aa56c3fde73f5841b534db3bf815', 'terminalcode': '4a2cc45598d8543222359a255cdbef17', 'terminaltype': '1', 'traceid': 'mgkm2gayjxzpnfjpuv', 'u-token': 'eyJhbGciOiJSUzI1NiJ9.eyJqdGkiOiI1NjBlNDg4Yy1jYzYxLTQzNTUtOTRjOC0yYWYwNTdkMGIyMzkxNDIyODc0MDY0IiwidWlkIjoidEQzQ1FIaFJUblVYcm5MNFNwckw3UT09Iiwic3ViIjoieXgyMDI1KioqKkBnbWFpbC5jb20iLCJpcCI6ImcwRzMydVRYUDF1ZGt3MjVFanZQenc9PSIsImRpZCI6Ii9DckplOTBGbURHREFCTnVzY0N5V0x0Nkk3R04yemRJS3RxZ0VPeU9HRk9nMk12cVptaUhFNmJ0YSt0OUgrcUEiLCJzdHMiOjAsImlhdCI6MTc2MDA4MDk5OSwiZXhwIjoxNzY3ODU2OTk5LCJwdXNoaWQiOiJvTmpMNm1ab2h4T203V3ZyZlIvcWdBPT0iLCJhdGwiOiIwIiwiaXNzIjoidXBleCJ9.64PHFr48cwtphejC-bFw8aLu9lx5jP81GBvrb0IHwBYM8EaWrcMU9VGT7zLL1mFYYUpedmTlS7EHzNvjuHNb8cUEEZGpAXKIGgQkyE48LrzhlQVASn3h0P7Wd9hWlLwcu1bOswo4Xgocecl0tpXaZVwAZccq4n13bqkEAouZLFM', 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/141.0.0.0 Safari/537.36', 'vs': 'v5b7p4i8zCMwEj9kSmzH7KMajbx3b6cS', 'x-sig': 'e178e9659b38b516ed81ad87a8c5e741', 'x-timestamp': '1760086517003', } datas = [] klineId = None klineTime = None for i in range(500): print(i) params = { 'languageType': '1', 'sign': 'SIGN', 'timeZone': 'string', 'contractId': '10000002', 'productCode': 'ethusdt', 'priceType': 'LAST_PRICE', 'klineType': 'MINUTE_15', 'limit': '200', 'nextKey.contractId': '10000002', 'nextKey.klineId': klineId if klineId else '6593862509827714', 'nextKey.klineTime': klineTime if klineTime else '1759851900000', } response = requests.get( 'https://http-gateway2.janapw.com/api/v1/public/quote/v1/getKlineV2', params=params, headers=headers ) klineId = response.json()["data"]["nextKey"]["klineId"] klineTime = response.json()["data"]["nextKey"]["klineTime"] for data in response.json()["data"]["dataList"]: # print(data) datas.append( { "open": data[3], "high": data[1], "low": data[2], "close": data[0], "id": data[4], } ) # {'open': '4514.40', 'high': '4533.91', 'low': '4513.51', 'close': '4532.93', 'id': '1758003300000'} datas = sorted(datas, key=lambda x: x["id"]) for data in datas: print(data) # 将列表转换为 DataFrame df = pd.DataFrame(datas) # 将 DataFrame 保存为 Excel 文件 df.to_excel('stock_data.xlsx', index=False) print("数据已成功保存到 stock_data.xlsx 文件中。") daily_signals = 0 # 信号总数 daily_wins = 0 daily_profit = 0 # 价差总和 # 遍历每根K线，寻找信号 for idx in range(1, len(datas) - 2): # 留出未来K线 prev, curr = datas[idx - 1], datas[idx] # 前一笔，当前一笔 entry_candle = datas[idx + 1] # 下一根开盘价作为入场价 future_candles = datas[idx + 2:] # 未来行情 entry_open = float(entry_candle['open']) # 开仓价格 direction, signal_type = check_signal(prev, curr) # 判断开仓方向和信号类型 if direction and signal_type: daily_signals += 1 exit_price, diff, exit_time = simulate_trade(direction, entry_open, future_candles, take_profit_diff=30, stop_loss_diff=-2) # 统计该信号类型的表现 signal_stats[signal_type]["count"] += 1 signal_stats[signal_type]["total_profit"] += diff if diff > 0: signal_stats[signal_type]["wins"] += 1 daily_wins += 1 daily_profit += diff # 将时间戳转换为本地时间 local_time = datetime.datetime.fromtimestamp(int(entry_candle['id']) / 1000) formatted_time = local_time.strftime("%Y-%m-%d %H:%M:%S") # 保存交易详情 all_trades.append( ( f"{formatted_time}号", "做多" if direction == "long" else "做空", signal_stats[signal_type]["name"], entry_open, exit_price, diff, exit_time ) ) # ===== 输出每笔交易详情 ===== logger.info("===== 每笔交易详情 =====") n = n1 = 0 # n = 总盈利，n1 = 总手续费 for date, direction, signal_name, entry, exit, diff, end_time in all_trades: profit_amount = diff / entry * 10000 # 计算盈利金额 close_fee = 10000 / entry * exit * 0.0005 # 平仓手续费 logger.info( f"{date} {direction}({signal_name}) 入场={entry:.2f} 出场={exit:.2f} 出场时间={end_time} " f"差价={diff:.2f} 盈利={profit_amount:.2f} " f"开仓手续费=5u 平仓手续费={close_fee:.2f}" ) n1 += 5 + close_fee n += profit_amount print(f'一共笔数：{len(all_trades)}') print(f"一共盈利：{n:.2f}") print(f'一共手续费：{n1:.2f}')