import datetime import json import time 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', 'cache-control': 'no-cache', 'origin': 'https://www.okx.com', 'pragma': 'no-cache', 'priority': 'u=1, i', 'referer': 'https://www.okx.com/', 'sec-ch-ua': '"Chromium";v="140", "Not=A?Brand";v="24", "Google Chrome";v="140"', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-platform': '"Windows"', 'sec-fetch-dest': 'empty', 'sec-fetch-mode': 'cors', 'sec-fetch-site': 'same-site', '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', } # ================== 获取 OKX K线数据 ================== def fetch_kline_okx(day: int, year: int = 2025, month: int = 9, period='15m', symbol="ETH-USDT"): """ 从 OKX 获取指定日期的K线数据 :param day: 日期 :param year: 年份 :param month: 月份 :param period: K线周期 (如 '1m', '5m', '15m', '1H') :param symbol: 交易对 """ time_ser = datetime.datetime(year, month, day) start_of_day = int(time_ser.replace(hour=0, minute=0, second=0, microsecond=0).timestamp() * 1000) end_of_day = int(time_ser.replace(hour=23, minute=59, second=59, microsecond=0).timestamp() * 1000) url = "https://www.okx.com/api/v5/market/candles" params = { "instId": symbol, "bar": period, "limit": "1000" } all_data = [] next_start = end_of_day while True: params["after"] = next_start resp = requests.get(url, params=params, headers=headers) data = resp.json().get("data", []) if not data: break for d in data: candle = { "id": int(int(d[0]) / 1000), # 秒级时间戳 "open": d[1], "high": d[2], "low": d[3], "close": d[4], "volume": d[5] } all_data.append(candle) next_start = int(data[-1][0]) if len(data) < 1000: break # OKX返回数据按时间倒序排列 return sorted(all_data, key=lambda x: x['id']) # ================== 辅助函数 ================== 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) """ 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): if c_open < p_close and c_close > p_open: return "long", "bear_bull_engulf" # 前涨后跌包住 -> 做空 if is_bearish(curr) and is_bullish(prev): if c_open > p_close and c_close < p_open: return "short", "bull_bear_engulf" return None, None def simulate_trade(direction, entry_price, future_candles, take_profit_diff=30, stop_loss_diff=-10): """ 模拟逐根K线的止盈止损 """ 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']) diff_money = (final_price - entry_price) if direction == "long" else (entry_price - final_price) return final_price, diff_money, future_candles[-1]['id'] # ================== 主程序 ================== if __name__ == '__main__': 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": "跌包涨"} } datas = [] for i in range(1, 31): logger.info(f"Fetching 2025-09-{i}") sorted_data = fetch_kline_okx(year=2025, month=1, day=i, period='15m', symbol="ETH-USDT") datas.extend(sorted_data) time.sleep(0.3) # # 可选：保存数据方便离线分析 # with open("okx_eth_15m_sep2025.json", "w", encoding="utf-8") as f: # json.dump(datas, f, ensure_ascii=False, indent=2) datas = sorted(datas, key=lambda x: x["id"]) daily_signals = daily_wins = daily_profit = 0 for idx in range(1, len(datas) - 2): 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(entry_candle['id']) formatted_time = local_time.strftime("%Y-%m-%d %H:%M") all_trades.append(( formatted_time, "做多" if direction == "long" else "做空", signal_stats[signal_type]["name"], entry_open, exit_price, diff, exit_time )) # ================= 输出交易详情 ================= logger.info("===== 每笔交易详情 =====") n = n1 = 0 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("=" * 60) print(f"交易总数：{len(all_trades)}") print(f"总盈利：{n:.2f} u") print(f"总手续费：{n1:.2f} u") print(f"净利润：{n - n1:.2f} u") print("=" * 60)