From aab552d7db91af8577f85057c4c831875a7114c5 Mon Sep 17 00:00:00 2001 From: 27942 <2794236280@qq.com> Date: Tue, 14 Oct 2025 15:59:27 +0800 Subject: [PATCH] dededdew --- weex/读取数据库分析数据.py | 152 ++++++++++++++++++---------------- weex/读取数据库分析数据2.0.py | 100 ++++++++++++---------- 2 files changed, 137 insertions(+), 115 deletions(-) diff --git a/weex/读取数据库分析数据.py b/weex/读取数据库分析数据.py index 73bc6ee..67c7aa5 100644 --- a/weex/读取数据库分析数据.py +++ b/weex/读取数据库分析数据.py @@ -1,7 +1,7 @@ import datetime from loguru import logger -from models.weex import Weex15 +from models.weex import Weex15, Weex1 def is_bullish(candle): @@ -30,16 +30,16 @@ def check_signal(prev, curr): 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" + # # 情况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: @@ -119,7 +119,7 @@ def get_data_by_date(date_str): end_timestamp = int((target_date + datetime.timedelta(days=1)).timestamp() * 1000) - 1 # 查询该天的数据，并按照 id 字段从小到大排序 - query = Weex15.select().where(Weex15.id.between(start_timestamp, end_timestamp)).order_by(Weex15.id.asc()) + query = Weex1.select().where(Weex1.id.between(start_timestamp, end_timestamp)).order_by(Weex1.id.asc()) results = list(query) # 将结果转换为列表嵌套字典的形式 @@ -142,86 +142,96 @@ if __name__ == '__main__': # 示例调用 datas = [] for i in range(1, 31): - date_str = f'2025-6-{i}' + date_str = f'2025-9-{i}' data = get_data_by_date(date_str) datas.extend(data) datas = sorted(datas, key=lambda x: x["id"]) - zh_project = 0 # 累计盈亏 - all_trades = [] # 保存所有交易明细 - daily_signals = 0 # 信号总数 - daily_wins = 0 - daily_profit = 0 # 价差总和 + for i in range(1, 11): + for i1 in range(1, 51): - # 四种信号类型的统计 - 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": "跌包跌"} - } + zh_project = 0 # 累计盈亏 + all_trades = [] # 保存所有交易明细 + daily_signals = 0 # 信号总数 + daily_wins = 0 + daily_profit = 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:] # 未来行情 + # 四种信号类型的统计 + 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": "跌包跌"} + } - entry_open = float(entry_candle['open']) # 开仓价格 - direction, signal_type = check_signal(prev, curr) # 判断开仓方向和信号类型 + # 遍历每根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:] # 未来行情 - if direction and signal_type: - daily_signals += 1 + entry_open = float(entry_candle['open']) # 开仓价格 + direction, signal_type = check_signal(prev, curr) # 判断开仓方向和信号类型 - exit_price, diff, exit_time = simulate_trade(direction, entry_open, future_candles, take_profit_diff=6, - stop_loss_diff=-2) + if direction and signal_type: + daily_signals += 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 + exit_price, diff, exit_time = simulate_trade( + direction, + entry_open, + future_candles, + take_profit_diff=i1, + stop_loss_diff=-i + ) - daily_profit += diff + # 统计该信号类型的表现 + 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) - formatted_time = local_time.strftime("%Y-%m-%d %H:%M:%S") + daily_profit += diff - exit_time = datetime.datetime.fromtimestamp(exit_time / 1000) - exit_time1 = exit_time.strftime("%Y-%m-%d %H:%M:%S") + # 将时间戳转换为本地时间 + local_time = datetime.datetime.fromtimestamp(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_time1 - ) - ) + exit_time = datetime.datetime.fromtimestamp(exit_time / 1000) + exit_time1 = exit_time.strftime("%Y-%m-%d %H:%M:%S") - # ===== 输出每笔交易详情 ===== - 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 # 平仓手续费 + # 保存交易详情 + all_trades.append( + ( + f"{formatted_time}号", + "做多" if direction == "long" else "做空", + signal_stats[signal_type]["name"], + entry_open, + exit_price, + diff, + exit_time1 + ) + ) - 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 + # ===== 输出每笔交易详情 ===== + logger.info("===== 每笔交易详情 =====") + logger.info(f"{i},{i1}") + 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 # 平仓手续费 - print(f'一共笔数：{len(all_trades)}') - print(f"一共盈利：{n:.2f}") - print(f'一共手续费：{n1:.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 + + if n > n1 * 0.1: + print(f'一共笔数：{len(all_trades)}') + print(f"一共盈利：{n:.2f}") + print(f'一共手续费：{n1:.2f}') diff --git a/weex/读取数据库分析数据2.0.py b/weex/读取数据库分析数据2.0.py index 4104848..80e1998 100644 --- a/weex/读取数据库分析数据2.0.py +++ b/weex/读取数据库分析数据2.0.py @@ -131,9 +131,9 @@ def check_signal(prev, curr): 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" + # 前涨后跌包住 -> 做空 + 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 @@ -147,20 +147,26 @@ def simulate_trade(direction, entry_price, entry_time, next_15min_time, tp=8, sl 用 1 分钟数据进行精细化止盈止损模拟 entry_time: 当前信号的 entry candle id（毫秒时间戳） next_15min_time: 下一个15min时间戳，用于界定止盈止损分析范围 + + direction：信号类型 + entry_price：开仓价格 + entry_time：开仓时间 + next_15min_time：15分钟未来行情 + """ # 查 15 分钟之间的 1 分钟数据 future_candles = get_future_data_1min(entry_time, next_15min_time) if not future_candles: return None, 0, None - tp_price = entry_price + tp if direction == "long" else entry_price - tp - sl_price = entry_price + sl if direction == "long" else entry_price - sl + tp_price = entry_price + tp if direction == "long" else entry_price - tp # 止盈价位 + sl_price = entry_price + sl if direction == "long" else entry_price - sl # 止损价位 for candle in future_candles: open_p, high, low = map(float, (candle['open'], candle['high'], candle['low'])) - if direction == "long": - if open_p >= tp_price: # 开盘跳空止盈 + if direction == "long": # long + 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'] @@ -169,8 +175,8 @@ def simulate_trade(direction, entry_price, entry_time, next_15min_time, tp=8, sl if low <= sl_price: return sl_price, sl, candle['id'] - else: # short - if open_p <= tp_price: + else: # short 跌信号 + 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'] @@ -190,7 +196,7 @@ def simulate_trade(direction, entry_price, entry_time, next_15min_time, tp=8, sl # 📊 主回测流程 # =============================================================== -def backtest(dates, tp=8, sl=-1): +def backtest(dates, tp, sl): """ datas：日期的列表 @@ -202,7 +208,7 @@ def backtest(dates, tp=8, sl=-1): all_data = [] for date_str in dates: - all_data.extend(get_data_by_date(Weex15, date_str)) # 获取天的数据，15分钟k线数据 + all_data.extend(get_data_by_date(Weex15, date_str)) # 获取每天的数据，15分钟k线数据 all_data.sort(key=lambda x: x['id']) @@ -215,18 +221,24 @@ def backtest(dates, tp=8, sl=-1): for idx in range(1, len(all_data) - 1): prev, curr = all_data[idx - 1], all_data[idx] # 前一笔，当前一笔 - entry_candle = all_data[idx + 1] # 开仓 + entry_candle = all_data[idx + 1] # 下一笔开仓k线 direction, signal = check_signal(prev, curr) if not direction: continue # 下一个 15 分钟K线的时间范围 - next_15min_time = all_data[idx + 1]['id'] if idx + 2 < len(all_data) else all_data[-1]['id'] + next_15min_time = all_data[idx + 5]['id'] if idx + 5 < len(all_data) else all_data[-1]['id'] - entry_price = float(entry_candle['open']) - exit_price, diff, exit_time = simulate_trade(direction, entry_price, entry_candle['id'], next_15min_time, tp=tp, - sl=sl) + entry_price = float(entry_candle['open']) # 开仓价格 + exit_price, diff, exit_time = simulate_trade( + direction, + entry_price, + entry_candle['id'], + next_15min_time, + tp=tp, + sl=sl + ) if exit_price is None: continue @@ -254,36 +266,36 @@ def backtest(dates, tp=8, sl=-1): # =============================================================== if __name__ == '__main__': - dates = [f"2025-7-{i}" for i in range(1, 31)] - for i in range(1, 11): - for i1 in range(1, 51): - trades, stats = backtest(dates, tp=i1, sl=0 - i) + dates = [f"2025-9-{i}" for i in range(1, 31)] - total_profit = sum(t['diff'] / t['entry'] * 10000 for t in trades) - total_fee = sum(5 + 10000 / t['entry'] * t['exit'] * 0.0005 for t in trades) + trades, stats = backtest(dates, tp=10, sl=-2) - # logger.info("===== 每笔交易详情 =====") - # for t in trades: - # logger.info(f"{t['entry_time']} {t['direction']}({t['signal']}) " - # f"入场={t['entry']:.2f} 出场={t['exit']:.2f} 出场时间={t['exit_time']} " - # f"差价={t['diff']:.2f}") - # - # print(i1, i) - # print(f"\n一共交易笔数：{len(trades)}") - # print(f"一共盈利：{total_profit:.2f}") - # print(f"一共手续费：{total_fee:.2f}") - # print(f"净利润：{total_profit - total_fee:.2f}") + logger.info("===== 每笔交易详情 =====") + for t in trades: + logger.info( + f"{t['entry_time']} {t['direction']}({t['signal']}) " + f"入场={t['entry']:.2f} 出场={t['exit']:.2f} 出场时间={t['exit_time']} " + f"差价={t['diff']:.2f}" + ) - if total_profit > total_fee * 0.1: - print(i1, i) - print(f"\n一共交易笔数：{len(trades)}") - print(f"一共盈利：{total_profit:.2f}") - print(f"一共手续费：{total_fee:.2f}") - print(f"净利润：{total_profit - total_fee * 0.1}") + total_profit = sum(t['diff'] / t['entry'] * 10000 for t in trades) + total_fee = sum(5 + 10000 / t['entry'] * t['exit'] * 0.0005 for t in trades) - print("\n===== 信号统计 =====") + print(f"\n一共交易笔数：{len(trades)}") + print(f"一共盈利：{total_profit:.2f}") + print(f"一共手续费：{total_fee:.2f}") + print(f"净利润：{total_profit - total_fee:.2f}") - # for k, v in stats.items(): - # win_rate = (v['wins'] / v['count'] * 100) if v['count'] > 0 else 0 - # print( - # f"{v['name']} ({k}) - 信号数: {v['count']} | 胜率: {win_rate:.2f}% | 总盈利: {v['total_profit']:.2f}") + # if total_profit > total_fee * 0.1: + # print(i1, i) + # print(f"\n一共交易笔数：{len(trades)}") + # print(f"一共盈利：{total_profit:.2f}") + # print(f"一共手续费：{total_fee:.2f}") + # print(f"净利润：{total_profit - total_fee * 0.1}") + # + # print("\n===== 信号统计 =====") + # + # for k, v in stats.items(): + # win_rate = (v['wins'] / v['count'] * 100) if v['count'] > 0 else 0 + # print( + # f"{v['name']} ({k}) - 信号数: {v['count']} | 胜率: {win_rate:.2f}% | 总盈利: {v['total_profit']:.2f}")