From 9e17a26d48ea7a93987eeba93e2dd7fae0aec258 Mon Sep 17 00:00:00 2001 From: 27942 <2794236280@qq.com> Date: Mon, 13 Oct 2025 14:45:53 +0800 Subject: [PATCH] dededdew --- weex/读取数据库分析数据.py | 2 +- weex/读取数据库分析数据2.0.py | 310 ++++++++++++++++++++-------------- 2 files changed, 188 insertions(+), 124 deletions(-) diff --git a/weex/读取数据库分析数据.py b/weex/读取数据库分析数据.py index 2a1c67a..cb86543 100644 --- a/weex/读取数据库分析数据.py +++ b/weex/读取数据库分析数据.py @@ -142,7 +142,7 @@ if __name__ == '__main__': # 示例调用 datas = [] for i in range(1, 31): - date_str = f'2025-9-{i}' + date_str = f'2025-8-{i}' data = get_data_by_date(date_str) datas.extend(data) diff --git a/weex/读取数据库分析数据2.0.py b/weex/读取数据库分析数据2.0.py index f8fed4e..cb86543 100644 --- a/weex/读取数据库分析数据2.0.py +++ b/weex/读取数据库分析数据2.0.py @@ -1,163 +1,227 @@ import datetime + from loguru import logger from models.weex import Weex +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线回测） + 改进版：考虑开盘跳空触发止盈/止损的情况 + """ + for candle in future_candles: + open_p = float(candle['open']) + high = float(candle['high']) + low = float(candle['low']) + close = float(candle['close']) + + if direction == "long": + tp_price = entry_price + take_profit_diff + sl_price = entry_price + stop_loss_diff + + # 🧩 开盘就跳空止盈 + if open_p >= tp_price: + return open_p, open_p - entry_price, candle['id'] + + # 🧩 开盘就跳空止损 + if open_p <= sl_price: + return open_p, open_p - entry_price, candle['id'] + + # 正常区间内触发止盈/止损 + if high >= tp_price: + return tp_price, take_profit_diff, candle['id'] + if low <= sl_price: + return sl_price, stop_loss_diff, candle['id'] + + elif direction == "short": + tp_price = entry_price - take_profit_diff + sl_price = entry_price - stop_loss_diff + + # 🧩 开盘就跳空止盈 + if open_p <= tp_price: + return open_p, entry_price - open_p, candle['id'] + + # 🧩 开盘就跳空止损 + if open_p >= sl_price: + return open_p, entry_price - open_p, candle['id'] + + # 正常区间内触发止盈/止损 + if low <= tp_price: + return tp_price, take_profit_diff, candle['id'] + if high >= sl_price: + return sl_price, 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'] + + def get_data_by_date(date_str): - """按日期获取K线数据""" + # 将日期字符串转换为 datetime 对象 try: target_date = datetime.datetime.strptime(date_str, '%Y-%m-%d') except ValueError: print("日期格式不正确，请使用 YYYY-MM-DD 格式。") return [] + # 计算该天的开始时间戳（毫秒级） start_timestamp = int(target_date.timestamp() * 1000) + # 计算该天的结束时间戳（毫秒级），即下一天 00:00:00 的时间戳减去 1 毫秒 end_timestamp = int((target_date + datetime.timedelta(days=1)).timestamp() * 1000) - 1 + # 查询该天的数据，并按照 id 字段从小到大排序 query = Weex.select().where(Weex.id.between(start_timestamp, end_timestamp)).order_by(Weex.id.asc()) results = list(query) - return [ - {'id': item.id, 'open': item.open, 'high': item.high, 'low': item.low, 'close': item.close} - for item in results - ] + # 将结果转换为列表嵌套字典的形式 + data_list = [] + for item in results: + item_dict = { + 'id': item.id, + 'open': item.open, + 'high': item.high, + 'low': item.low, + 'close': item.close + } + data_list.append(item_dict) + + return data_list -# ========== 信号逻辑 ========== -def is_bullish(candle): - return float(candle['open']) < float(candle['close']) +if __name__ == '__main__': + # 示例调用 + datas = [] + for i in range(1, 31): + date_str = f'2025-8-{i}' + data = get_data_by_date(date_str) -def is_bearish(candle): - return float(candle['open']) > float(candle['close']) + datas.extend(data) + datas = sorted(datas, key=lambda x: x["id"]) -def check_signal(prev, curr): - """ - 检查是否形成包住信号 - """ - p_open, p_close = float(prev['open']), float(prev['close']) - c_open, c_close = float(curr['open']), float(curr['close']) + zh_project = 0 # 累计盈亏 + all_trades = [] # 保存所有交易明细 + daily_signals = 0 # 信号总数 + daily_wins = 0 + daily_profit = 0 # 价差总和 - # 前跌后涨包住 -> 做多 - 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, entry_candle, future_candles, take_profit_diff=30, stop_loss_diff=-10): - """ - 模拟一笔交易 - direction: long / short - entry_price: 入场价格（下一根开盘） - entry_candle: 开仓K线（判断是否立刻止损） - future_candles: 开仓后未来K线 - """ - entry_high = float(entry_candle['high']) - entry_low = float(entry_candle['low']) - - if direction == "long": - take_profit_price = entry_price + take_profit_diff - stop_loss_price = entry_price + stop_loss_diff - else: - take_profit_price = entry_price - take_profit_diff - stop_loss_price = entry_price - stop_loss_diff - - # ✅【优化点1】：如果开仓当根已经触及止损价，则仍按开仓价计算，不立即止损 - if direction == "long" and entry_low <= stop_loss_price: - logger.debug(f"多单开仓价{entry_price}，但当根最低价{entry_low}已触及止损{stop_loss_price}，按开仓价计算。") - stop_loss_price = entry_price - - if direction == "short" and entry_high >= stop_loss_price: - logger.debug(f"空单开仓价{entry_price}，但当根最高价{entry_high}已触及止损{stop_loss_price}，按开仓价计算。") - stop_loss_price = entry_price - - # ✅【优化点2】：从 entry_candle 下一根开始计算止盈止损 - for candle in future_candles: - high, low, close = float(candle['high']), float(candle['low']), float(candle['close']) - - if direction == "long": - if high >= take_profit_price: - return take_profit_price, take_profit_diff, candle['id'] - if low <= stop_loss_price: - return stop_loss_price, stop_loss_price - entry_price, candle['id'] - - elif direction == "short": - if low <= take_profit_price: - return take_profit_price, entry_price - take_profit_price, candle['id'] - if high >= stop_loss_price: - return stop_loss_price, entry_price - stop_loss_price, candle['id'] - - # ✅【未触发止盈止损，最后一根收盘平仓】 - final_candle = future_candles[-1] - final_price = float(final_candle['close']) - diff = final_price - entry_price if direction == "long" else entry_price - final_price - - return final_price, diff, final_candle['id'] - - -# ========== 主逻辑 ========== -def run_backtest(date_str='2025-10-08'): - data = get_data_by_date(date_str) - if not data: - print("无数据") - return - - 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": "跌包跌"} } - total_profit = 0 - total_fee = 0 + # 遍历每根K线，寻找信号 + for idx in range(1, len(datas) - 1): # 留出未来K线 + prev, curr = datas[idx - 1], datas[idx] # 前一笔，当前一笔 + entry_candle = datas[idx + 1] # 下一根开盘价作为入场价 + future_candles = datas[idx + 1:] # 未来行情 - for idx in range(1, len(data) - 2): - prev, curr = data[idx - 1], data[idx] - entry_candle = data[idx + 1] - future_candles = data[idx + 2:] - entry_open = float(entry_candle['open']) + entry_open = float(entry_candle['open']) # 开仓价格 + direction, signal_type = check_signal(prev, curr) # 判断开仓方向和信号类型 - direction, signal_type = check_signal(prev, curr) - if not direction: - continue + if direction and signal_type: + daily_signals += 1 - exit_price, diff, exit_time = simulate_trade(direction, entry_open, entry_candle, 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=6, + 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 + # 统计该信号类型的表现 + 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 - local_time = datetime.datetime.fromtimestamp(entry_candle['id'] / 1000).strftime("%Y-%m-%d %H:%M:%S") + daily_profit += diff - all_trades.append( - (local_time, direction, signal_stats[signal_type]["name"], entry_open, exit_price, diff, exit_time)) + # 将时间戳转换为本地时间 + local_time = datetime.datetime.fromtimestamp(entry_candle['id'] / 1000) + formatted_time = local_time.strftime("%Y-%m-%d %H:%M:%S") - # ===== 输出结果 ===== + exit_time = datetime.datetime.fromtimestamp(exit_time / 1000) + exit_time1 = exit_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_time1 + ) + ) + + # ===== 输出每笔交易详情 ===== logger.info("===== 每笔交易详情 =====") - for date, direction, name, entry, exit, diff, end_time in all_trades: - profit_amount = diff / entry * 10000 - close_fee = 10000 / entry * exit * 0.0005 - total_profit += profit_amount - total_fee += 5 + close_fee + 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} {('做多' if direction == 'long' else '做空')}({name}) 入场={entry:.2f} 出场={exit:.2f} " - f"出场时间={end_time} 差价={diff:.2f} 盈利={profit_amount:.2f} 手续费={close_fee:.2f}") + 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"总盈利：{total_profit:.2f}") - print(f"总手续费：{total_fee:.2f}") - - -if __name__ == "__main__": - run_backtest("2025-10-08") + print(f'一共笔数：{len(all_trades)}') + print(f"一共盈利：{n:.2f}") + print(f'一共手续费：{n1:.2f}')