diff --git a/weex/读取数据库分析数据2.0-优化信号版.py b/weex/读取数据库分析数据2.0-优化信号版.py new file mode 100644 index 0000000..ee54563 --- /dev/null +++ b/weex/读取数据库分析数据2.0-优化信号版.py @@ -0,0 +1,258 @@ +""" +量化交易回测系统 - 优化版（v2.1） +功能：基于包住形态的交易信号识别和回测分析 +作者：量化交易团队 +""" + +import datetime +from dataclasses import dataclass +from loguru import logger +from models.weex import Weex15, Weex1 + + +# =============================================================== +# 📊 配置管理类 +# =============================================================== + +@dataclass +class BacktestConfig: + """回测配置类""" + take_profit: float = 8.0 # 止盈点数 + stop_loss: float = -1.0 # 止损点数 + contract_size: float = 10000 # 合约规模 + open_fee: float = 5.0 # 开仓手续费 + close_fee_rate: float = 0.0005 # 平仓手续费率 + + start_date: str = "2025-7-1" + end_date: str = "2025-7-31" + + enable_bear_bull_engulf: bool = True # 涨包跌信号 + enable_bull_bear_engulf: bool = True # 跌包涨信号 + + def __post_init__(self): + if self.take_profit <= 0: + raise ValueError("止盈点数必须大于0") + if self.stop_loss >= 0: + raise ValueError("止损点数必须小于0") + + +# =============================================================== +# 📊 数据模块 +# =============================================================== + +def get_data_by_date(model, date_str): + """按天获取指定表的数据""" + try: + target_date = datetime.datetime.strptime(date_str, '%Y-%m-%d') + except ValueError: + logger.error("日期格式不正确，请使用 YYYY-MM-DD 格式。") + return [] + + start_ts = int(target_date.timestamp() * 1000) + end_ts = int((target_date + datetime.timedelta(days=1)).timestamp() * 1000) - 1 + + query = (model + .select() + .where(model.id.between(start_ts, end_ts)) + .order_by(model.id.asc())) + + return [ + {'id': i.id, 'open': i.open, 'high': i.high, 'low': i.low, 'close': i.close} + for i in query + ] + + +def get_future_data_1min(start_ts, end_ts): + """获取指定时间范围内的 1 分钟数据""" + query = (Weex1 + .select() + .where(Weex1.id.between(start_ts, end_ts)) + .order_by(Weex1.id.asc())) + return [{'id': i.id, 'open': i.open, 'high': i.high, 'low': i.low, 'close': i.close} for i in query] + + +# =============================================================== +# 📈 信号模块 +# =============================================================== + +def is_bullish(c): return float(c['open']) < float(c['close']) +def is_bearish(c): return float(c['open']) > float(c['close']) + + +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']) + + # 前跌后涨包住 -> 做多 + 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 + + +# =============================================================== +# 💹 模拟模块（1分钟级止盈止损） +# =============================================================== + +def simulate_trade(direction, entry_price, entry_time, next_15min_time, tp=8, sl=-1): + """用 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 + + 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: # 跳空止盈 + 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, tp, candle['id'] + if low <= sl_price: + return sl_price, sl, candle['id'] + + 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'] + if low <= tp_price: + return tp_price, tp, candle['id'] + if high >= sl_price: + return sl_price, sl, candle['id'] + + # 未触发止盈止损，用最后一根收盘价平仓 + final = future_candles[-1] + final_price = float(final['close']) + diff = (final_price - entry_price) if direction == "long" else (entry_price - final_price) + return final_price, diff, final['id'] + + +# =============================================================== +# 📊 回测主流程 +# =============================================================== + +def backtest(dates, tp, sl): + all_data = [] + for date_str in dates: + all_data.extend(get_data_by_date(Weex15, date_str)) + + all_data.sort(key=lambda x: x['id']) + + stats = { + "bear_bull_engulf": {"count": 0, "wins": 0, "total_profit": 0, "name": "涨包跌"}, + "bull_bear_engulf": {"count": 0, "wins": 0, "total_profit": 0, "name": "跌包涨"}, + } + + trades = [] + idx = 1 + open_position = None + + while idx < len(all_data) - 1: + prev, curr = all_data[idx - 1], all_data[idx] + entry_candle = all_data[idx + 1] + direction, signal = check_signal(prev, curr) + + # === 检查信号 === + if not direction: + idx += 1 + continue + + # === 当前有持仓 === + if open_position: + if direction == open_position['direction']: + # 同方向信号：忽略 + idx += 1 + continue + else: + # 反方向信号：立即平仓 + exit_price = float(entry_candle['open']) + diff = (exit_price - open_position['entry_price']) if open_position['direction'] == 'long' else ( + open_position['entry_price'] - exit_price) + trades.append({ + "entry_time": datetime.datetime.fromtimestamp(open_position['entry_time'] / 1000), + "exit_time": datetime.datetime.fromtimestamp(entry_candle['id'] / 1000), + "signal": "反向平仓", + "direction": "平仓", + "entry": open_position['entry_price'], + "exit": exit_price, + "diff": diff + }) + open_position = None # 平仓后可立即反手 + + # === 开新仓 === + next_15min_time = all_data[idx + 50]['id'] if idx + 50 < 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 + ) + if exit_price is None: + idx += 1 + continue + + # 记录统计 + stats[signal]['count'] += 1 + stats[signal]['total_profit'] += diff + if diff > 0: + stats[signal]['wins'] += 1 + + trades.append({ + "entry_time": datetime.datetime.fromtimestamp(entry_candle['id'] / 1000), + "exit_time": datetime.datetime.fromtimestamp(exit_time / 1000), + "signal": stats[signal]['name'], + "direction": "做多" if direction == "long" else "做空", + "entry": entry_price, + "exit": exit_price, + "diff": diff + }) + + # === 跳过到平仓时间点 === + # 找到 exit_time 对应的 candle 索引，防止未平仓时重复触发信号 + while idx < len(all_data) - 1 and all_data[idx]['id'] < exit_time: + idx += 1 + + open_position = None # 已平仓 + idx += 1 + + return trades, stats + + +# =============================================================== +# 🚀 主入口 +# =============================================================== + +if __name__ == '__main__': + dates = [f"2025-9-{i}" for i in range(1, 31)] + + trades, stats = backtest(dates, tp=50, sl=-10) + + 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}" + ) + + 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(f"\n一共交易笔数：{len(trades)}") + print(f"总盈利：{total_profit:.2f}") + print(f"总手续费：{total_fee:.2f}") + print(f"净利润：{total_profit - total_fee:.2f}") + 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}")