diff --git a/weex/stock_data.xlsx b/weex/stock_data.xlsx new file mode 100644 index 0000000..b08fa95 Binary files /dev/null and b/weex/stock_data.xlsx differ diff --git a/weex/test.py b/weex/test.py new file mode 100644 index 0000000..a93adab --- /dev/null +++ b/weex/test.py @@ -0,0 +1,253 @@ +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}') diff --git a/weex/时间戳查询数据.py b/weex/时间戳查询数据.py new file mode 100644 index 0000000..43a53ab --- /dev/null +++ b/weex/时间戳查询数据.py @@ -0,0 +1,28 @@ +import pandas as pd + + +def filter_data_by_single_date(file_path, date): + # 将传入的日期转换为 pandas 的 Timestamp 类型 + target_date = pd.Timestamp(date) + # 计算该天的开始时间（即 00:00:00），并转换为毫秒级时间戳 + start_timestamp_ms = int(target_date.normalize().timestamp() * 1000) + # 计算该天的结束时间（即 23:59:59.999），并转换为毫秒级时间戳 + end_timestamp_ms = int( + (target_date.normalize() + pd.Timedelta(days=1) - pd.Timedelta(milliseconds=1)).timestamp() * 1000) + + # 读取 Excel 文件 + df = pd.read_excel(file_path) + # 筛选出指定时间戳范围内的数据 + filtered_df = df[(df['id'] >= start_timestamp_ms) & (df['id'] <= end_timestamp_ms)] + # 将筛选后的 DataFrame 转换为列表嵌套字典的形式 + return filtered_df.to_dict(orient='records') + + +# 示例使用 +file_path = 'stock_data.xlsx' +# 替换为你要查询的日期 +date = '2025-10-1' + +result = filter_data_by_single_date(file_path, date) +for item in result: + print(item) diff --git a/weex/读取文件分析.py b/weex/读取文件分析.py new file mode 100644 index 0000000..6a8d6bf --- /dev/null +++ b/weex/读取文件分析.py @@ -0,0 +1,31 @@ +import datetime + +import pandas as pd +from itertools import groupby + + +def excel_to_list_of_dicts(file_path): + df = pd.read_excel(file_path) + return df.to_dict(orient='records') + +def fetch_kline(day: int, year: int = 2025, month: int = 9, period=1): + """获取某一天的分钟级 K线数据""" + # 构造该日的起止时间戳 + time_ser = datetime.datetime(year, month, day) # 修正为2024年9月 + 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) + + + + +if __name__ == '__main__': + + file_path = 'stock_data.xlsx' + data = excel_to_list_of_dicts(file_path) + + # 按字典内容排序 + data.sort(key=lambda x: tuple(sorted(x.items()))) + unique_data = [next(group) for _, group in groupby(data)] + datas = sorted(unique_data, key=lambda x: x["id"]) + for i in datas: + print(i) diff --git a/回测数据/推测策略，回测.py b/回测数据/推测策略，回测.py index 8e9fe05..238fc39 100644 --- a/回测数据/推测策略，回测.py +++ b/回测数据/推测策略，回测.py @@ -147,8 +147,8 @@ if __name__ == '__main__': datas.extend(sorted_data) - sorted(datas, key=lambda x: x["id"]) - + datas = sorted(datas, key=lambda x: x["id"]) + print(datas) daily_signals = 0 # 信号总数 daily_wins = 0 daily_profit = 0 # 价差总和 @@ -165,7 +165,8 @@ if __name__ == '__main__': 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) + 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 diff --git a/回测数据/通过止盈止损策略.py b/回测数据/通过止盈止损策略.py deleted file mode 100644 index 826e69d..0000000 --- a/回测数据/通过止盈止损策略.py +++ /dev/null @@ -1,169 +0,0 @@ -import datetime -import requests -from loguru import logger -from typing import List, Dict, Tuple, Optional - -# ================== 常量配置 ================== -BASE_URL = "https://capi.websea.com/webApi/market/getKline" -HEADERS = { - 'accept': 'application/json, text/plain, */*', - 'origin': 'https://www.websea.com', - 'referer': 'https://www.websea.com/', - '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', -} -SYMBOL = "ETH-USDT" -PERIOD = "15min" -TAKE_PROFIT_DIFF = 30 -STOP_LOSS_DIFF = -2 - - -# ================== 数据获取 ================== -def fetch_kline(day: int, year: int = 2025, month: int = 9) -> List[Dict]: - """获取指定日期的K线数据（按时间升序返回）""" - try: - date = datetime.datetime(year, month, day) - start = int(date.replace(hour=0, minute=0, second=0).timestamp()) - end = int(date.replace(hour=23, minute=59, second=59).timestamp()) - params = {"symbol": SYMBOL, "period": PERIOD, "start": start, "end": end} - resp = requests.get(BASE_URL, params=params, headers=HEADERS, timeout=10) - resp.raise_for_status() - data = resp.json().get("result", {}).get("data", []) - return sorted(data, key=lambda x: x["id"]) - except Exception as e: - logger.error(f"获取 {day} 号数据失败: {e}") - return [] - - -# ================== K线辅助函数 ================== -def is_bullish(candle: Dict) -> bool: - return float(candle["open"]) < float(candle["close"]) - - -def is_bearish(candle: Dict) -> bool: - return float(candle["open"]) > float(candle["close"]) - - -def check_signal(prev: Dict, curr: Dict) -> Tuple[Optional[str], Optional[str]]: - """检测K线形态信号""" - p_open, p_close = float(prev["open"]), float(prev["close"]) - c_open, c_close = float(curr["open"]), float(curr["close"]) - - if is_bullish(curr) and is_bearish(prev) and c_open < p_close and c_close > p_open: - return "long", "bear_bull_engulf" # 涨包跌 - if is_bearish(curr) and is_bullish(prev) and c_open > p_close and c_close < p_open: - return "short", "bull_bear_engulf" # 跌包涨 - return None, None - - -# ================== 回测函数 ================== -def simulate_trade(direction: str, entry_price: float, future: List[Dict], - take_profit_diff: float, stop_loss_diff: float) -> Tuple[float, float, int]: - """基于未来K线模拟止盈止损""" - for candle in future: - high, low, close = map(float, (candle["high"], candle["low"], candle["close"])) - cid = candle["id"] - - if direction == "long": - if high >= entry_price + take_profit_diff: - return entry_price + take_profit_diff, take_profit_diff, cid - if low <= entry_price + stop_loss_diff: - return entry_price + stop_loss_diff, stop_loss_diff, cid - - elif direction == "short": - if low <= entry_price - take_profit_diff: - return entry_price - take_profit_diff, take_profit_diff, cid - if high >= entry_price - stop_loss_diff: - return entry_price - stop_loss_diff, stop_loss_diff, cid - - final_close = float(future[-1]["close"]) - diff = final_close - entry_price if direction == "long" else entry_price - final_close - return final_close, diff, future[-1]["id"] - - -# ================== 主程序 ================== -def main(): - all_trades = [] - total_profit = 0.0 - - signal_stats = { - "bear_bull_engulf": {"count": 0, "wins": 0, "profit": 0.0, "name": "涨包跌"}, - "bull_bear_engulf": {"count": 0, "wins": 0, "profit": 0.0, "name": "跌包涨"}, - } - - for day in range(1, 31): - data = fetch_kline(year=2025, month=9, day=day) - if not data: - continue - - daily_signals = daily_wins = 0 - daily_profit = 0.0 - - for i in range(1, len(data) - 2): - prev, curr = data[i - 1], data[i] - entry_candle = data[i + 1] - future_candles = data[i + 2:] - - direction, signal_type = check_signal(prev, curr) - if not direction: - continue - - entry_price = float(entry_candle["open"]) - exit_price, diff, exit_id = simulate_trade( - direction, entry_price, future_candles, TAKE_PROFIT_DIFF, STOP_LOSS_DIFF - ) - - daily_signals += 1 - daily_profit += diff - signal_stats[signal_type]["count"] += 1 - signal_stats[signal_type]["profit"] += diff - if diff > 0: - daily_wins += 1 - signal_stats[signal_type]["wins"] += 1 - - all_trades.append({ - "date": f"{day}号", - "entry_id": entry_candle["id"], - "direction": "做多" if direction == "long" else "做空", - "signal": signal_stats[signal_type]["name"], - "entry": entry_price, - "exit": exit_price, - "diff": diff, - "exit_id": exit_id, - }) - - if daily_signals: - win_rate = daily_wins / daily_signals * 100 - logger.info(f"{day}号: 信号={daily_signals}, 胜率={win_rate:.2f}%, 盈亏={daily_profit:.2f}") - else: - logger.info(f"{day}号: 无信号") - - total_profit += daily_profit - - logger.success(f"✅ 综合盈亏：{total_profit:.2f}") - - # 信号统计 - logger.info("===== 信号类型统计 =====") - for k, v in signal_stats.items(): - if v["count"]: - wr = v["wins"] / v["count"] * 100 - logger.info( - f"{v['name']} 信号={v['count']}, 胜率={wr:.2f}%, 总盈亏={v['profit']:.2f}, 均盈亏={v['profit'] / v['count']:.2f}") - else: - logger.info(f"{v['name']} 无信号") - - # 交易明细 - total_fee = total_gain = 0.0 - logger.info("===== 每笔交易详情 =====") - for t in all_trades: - profit = t["diff"] / t["entry"] * 10000 - close_fee = 10000 / t["entry"] * t["exit"] * 0.0005 - total_gain += profit - total_fee += 5 + close_fee - logger.info( - f"{t['date']} {t['entry_id']} {t['direction']}({t['signal']}) 入={t['entry']:.2f} 出={t['exit']:.2f} 差={t['diff']:.2f} 盈={profit:.2f} 手续费={5 + close_fee:.2f}") - - logger.info(f"总笔数：{len(all_trades)} 总盈利：{total_gain:.2f} 总手续费：{total_fee:.2f}") - - -if __name__ == "__main__": - main()