# source: https://raw.githubusercontent.com/xiedidan/freqtrade/6397209ac8ebda42690d95e5b2770ca62713e337/user_data/strategies/atr_level_signal.py import logging import enum import os from typing import ClassVar, Optional, List, Dict, Any from datetime import datetime, timezone import traceback import pandas as pd from sqlalchemy import String, Float, Integer, DateTime, select, delete, create_engine from sqlalchemy.orm import Mapped, mapped_column, sessionmaker, scoped_session from sqlalchemy.exc import SQLAlchemyError from freqtrade.strategy import IStrategy from freqtrade.persistence.base import ModelBase, SessionType from freqtrade.persistence.models import init_db from pandas import DataFrame, Series import pandas_ta as ta logger = logging.getLogger(__name__) class LevelDirection(str, enum.Enum): """Direction for level crossing""" UP = "up" # 向上突破（K线实体部分向上穿过价格水平） DOWN = "down" # 向下突破（K线实体部分向下穿过价格水平） BOTH = "both" # 双向突破（K线实体部分向上或向下穿过价格水平） WICK_UP = "wick_up" # 向上流动性清扫（K线上影线部分穿过价格水平） WICK_DOWN = "wick_down" # 向下流动性清扫（K线下影线部分穿过价格水平） WICK_BOTH = "wick_both" # 双向流动性清扫（K线上下影线部分穿过价格水平） class PriceLevel(ModelBase): """ Price level database model for level crossing detection """ __tablename__ = "price_levels" session: ClassVar[SessionType] id: Mapped[int] = mapped_column(Integer, primary_key=True) pair: Mapped[str] = mapped_column(String(25), nullable=False, index=True) level: Mapped[float] = mapped_column(Float, nullable=False) direction: Mapped[str] = mapped_column(String(10), nullable=False) created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) active: Mapped[bool] = mapped_column(Integer, nullable=False, default=1) # 1=active, 0=inactive confirm_close: Mapped[bool] = mapped_column(Integer, nullable=False, default=0) # 1=require close confirmation, 0=trigger on cross remarks: Mapped[str] = mapped_column(String(500), nullable=True) # 价格点位备注 @classmethod def get_levels(cls, pair: Optional[str] = None) -> List["PriceLevel"]: """ Get all active price levels for a specific pair or all pairs """ try: # Ensure we have a valid session if not hasattr(cls, 'session') or cls.session is None: logger.warning("Database session not initialized. Attempting to reconnect...") Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.init_db_session() filters = [PriceLevel.active == 1] if pair: filters.append(PriceLevel.pair == pair) return PriceLevel.session.scalars(select(PriceLevel).filter(*filters)).all() except SQLAlchemyError as e: logger.error(f"Database error in get_levels: {e}") return [] except Exception as e: logger.error(f"Error in get_levels: {e}") return [] @classmethod def add_level(cls, pair: str, level: float, direction: str = "both", confirm_close: bool = False, remarks: Optional[str] = None) -> "PriceLevel": """ Add a new price level to monitor Args: pair: Trading pair symbol level: Price level value direction: Direction to monitor (up/down/both) confirm_close: If True, require candle to close beyond the level remarks: Optional remarks for this price level """ try: # Ensure we have a valid session if not hasattr(cls, 'session') or cls.session is None: logger.warning("Database session not initialized. Attempting to reconnect...") Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.init_db_session() price_level = PriceLevel( pair=pair, level=level, direction=direction, created_at=datetime.now(), active=1, confirm_close=1 if confirm_close else 0, remarks=remarks ) PriceLevel.session.add(price_level) PriceLevel.session.commit() return price_level except SQLAlchemyError as e: logger.error(f"Database error in add_level: {e}") PriceLevel.session.rollback() raise except Exception as e: logger.error(f"Error in add_level: {e}") PriceLevel.session.rollback() raise @classmethod def delete_level(cls, level_id: int) -> None: """ Delete a price level by ID """ try: # Ensure we have a valid session if not hasattr(cls, 'session') or cls.session is None: logger.warning("Database session not initialized. Attempting to reconnect...") Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.init_db_session() PriceLevel.session.execute(delete(PriceLevel).where(PriceLevel.id == level_id)) PriceLevel.session.commit() except SQLAlchemyError as e: logger.error(f"Database error in delete_level: {e}") PriceLevel.session.rollback() raise except Exception as e: logger.error(f"Error in delete_level: {e}") PriceLevel.session.rollback() raise @classmethod def deactivate_level(cls, level_id: int) -> None: """ Deactivate a price level by ID """ try: # Ensure we have a valid session if not hasattr(cls, 'session') or cls.session is None: logger.warning("Database session not initialized. Attempting to reconnect...") Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.init_db_session() level = PriceLevel.session.get(PriceLevel, level_id) if level: level.active = 0 PriceLevel.session.commit() except SQLAlchemyError as e: logger.error(f"Database error in deactivate_level: {e}") PriceLevel.session.rollback() except Exception as e: logger.error(f"Error in deactivate_level: {e}") PriceLevel.session.rollback() class SignalHistory(ModelBase): """ Signal history database model for tracking level crossing signals """ __tablename__ = "signal_history" session: ClassVar[SessionType] id: Mapped[int] = mapped_column(Integer, primary_key=True) pair: Mapped[str] = mapped_column(String(25), nullable=False, index=True) signal_type: Mapped[str] = mapped_column(String(20), nullable=False, index=True) # 'level_cross_up', 'level_cross_down', 'level_wick_up', 'level_wick_down', 'atr_surge' level_id: Mapped[int] = mapped_column(Integer, nullable=True) # Reference to price level, null for ATR signals level_price: Mapped[float] = mapped_column(Float, nullable=True) # Price level, null for ATR signals prev_price: Mapped[float] = mapped_column(Float, nullable=False) # Previous candle close price current_price: Mapped[float] = mapped_column(Float, nullable=False) # Current candle close price atr_value: Mapped[float] = mapped_column(Float, nullable=True) # ATR value, null for level cross signals created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, index=True) remarks: Mapped[str] = mapped_column(String(500), nullable=True) # 信号备注，通常从关联的价格点位获取 @classmethod def add_signal(cls, pair: str, signal_type: str, prev_price: float, current_price: float, level_id: Optional[int] = None, level_price: Optional[float] = None, atr_value: Optional[float] = None, remarks: Optional[str] = None) -> "SignalHistory": """ Add a new signal to history Args: pair: Trading pair symbol signal_type: Type of signal ('level_cross_up', 'level_cross_down', 'atr_surge') prev_price: Previous candle close price current_price: Current candle close price level_id: ID of price level (for level crossing signals) level_price: Price level value (for level crossing signals) atr_value: ATR value (for ATR signals) remarks: Optional remarks for the signal """ try: # Ensure we have a valid session if not hasattr(cls, 'session') or cls.session is None: logger.warning("Database session not initialized. Attempting to reconnect...") Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.init_db_session() signal = SignalHistory( pair=pair, signal_type=signal_type, level_id=level_id, level_price=level_price, prev_price=prev_price, current_price=current_price, atr_value=atr_value, created_at=datetime.now(), remarks=remarks ) SignalHistory.session.add(signal) SignalHistory.session.commit() return signal except SQLAlchemyError as e: logger.error(f"Database error in add_signal: {e}") SignalHistory.session.rollback() raise except Exception as e: logger.error(f"Error in add_signal: {e}") SignalHistory.session.rollback() raise @classmethod def get_signals(cls, pair: Optional[str] = None, signal_type: Optional[str] = None, start_date: Optional[datetime] = None, end_date: Optional[datetime] = None, limit: int = 100, offset: int = 0) -> List["SignalHistory"]: """ Get signal history with optional filtering Args: pair: Optional trading pair to filter by signal_type: Optional signal type to filter by start_date: Optional start date for filtering end_date: Optional end date for filtering limit: Maximum number of results to return (0 means no limit) offset: Number of records to skip (for pagination) Returns: List of SignalHistory objects """ try: # Ensure we have a valid session if not hasattr(cls, 'session') or cls.session is None: logger.warning("Database session not initialized. Attempting to reconnect...") Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.init_db_session() filters = [] if pair: filters.append(SignalHistory.pair == pair) if signal_type: filters.append(SignalHistory.signal_type == signal_type) if start_date: filters.append(SignalHistory.created_at >= start_date) if end_date: filters.append(SignalHistory.created_at <= end_date) query = select(SignalHistory).filter(*filters).order_by(SignalHistory.created_at.desc()) # 添加偏移量（用于分页） if offset > 0: query = query.offset(offset) # 添加限制（如果limit > 0） if limit > 0: query = query.limit(limit) return SignalHistory.session.scalars(query).all() except SQLAlchemyError as e: logger.error(f"Database error in get_signals: {e}") return [] except Exception as e: logger.error(f"Error in get_signals: {e}") return [] class Github_xiedidan_freqtrade__atr_level_signal__20250724_100111(IStrategy): """ ATR Level Signal strategy This strategy monitors price levels for crossing and ATR for sudden increases. Configuration options: - timeframe: Timeframe to use for the strategy (default: 15m) - atr_length: ATR calculation period (default: 14) - atr_threshold: Threshold for sudden increase (default: 1.5x previous ATR) - check_level_crossing: Enable level crossing detection (default: True) - check_all_pairs: 是否检查所有交易对的价格水平，而不仅仅是当前交易对 (default: False) 如果设置为True，则会显示所有交易对的价格水平的调试信息 这对于了解系统中所有监控的价格水平很有用 - enable_debug_logs: 是否启用详细的调试日志，包括条件判断和信号检测信息 (default: True) 如果设置为False，则只显示信号触发时的日志 - debug_all_pairs: 是否显示所有交易对的调试信息，即使当前只处理一个交易对 (default: True) 如果设置为False，则只显示当前处理的交易对的调试信息 - force_fetch_candles: 是否强制获取所有交易对的K线数据 (default: True) 如果设置为True，则会尝试主动获取所有交易对的K线数据 - auto_update_monitored_pairs: 是否自动更新监控的交易对列表 (default: True) 如果设置为True，则每次计算前都会从数据库中读取需要监控的交易对 Usage: 1. 添加价格水平: Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.add_price_level(pair, level, direction) 2. 查看价格水平: Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.get_price_levels(pair) 3. 删除价格水平: Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.delete_price_level(level_id) 4. 更新价格水平: Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.update_price_level(level_id, level, direction) """ # Strategy configuration timeframe = '1m' # Set timeframe to 15 minutes atr_length = 14 # ATR calculation period atr_threshold = 1.2 # Threshold for sudden increase (1.5x previous ATR) stoploss = -0.10 # Required stoploss (10%) added to fix validation error # Level crossing configuration check_level_crossing = True # Enable level crossing detection check_all_pairs = True # 是否检查所有交易对的价格水平（不仅仅是当前交易对） enable_debug_logs = True # 是否启用详细的调试日志 debug_all_pairs = True # 是否显示所有交易对的调试信息 force_fetch_candles = True # 是否强制获取所有交易对的K线数据 auto_update_monitored_pairs = True # 是否自动更新监控的交易对列表 # Database configuration db_initialized = False db_url = None # 跟踪上次记录所有价格水平的时间 last_levels_log_time = None # 记录价格水平的间隔时间（秒） levels_log_interval = 300 # 5分钟 # 存储所有交易对的最新K线数据 latest_candles = {} # 存储需要监控的交易对列表 monitored_pairs = set() # 上次更新监控交易对列表的时间 last_pairs_update_time = None # 更新监控交易对列表的间隔时间（秒） pairs_update_interval = 60 # 1分钟 @staticmethod def init_db_session(): """Initialize database session for PriceLevel model""" try: # Try to get database URL from config or use default config_file = os.path.join('user_data', 'config.json') if os.path.exists(config_file): import json with open(config_file, 'r') as f: config = json.load(f) Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.db_url = config.get('db_url') # If no config found, use default SQLite path if not Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.db_url: db_path = os.path.join('user_data', 'tradesv3.sqlite') Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.db_url = f'sqlite:///{db_path}' logger.info(f"Using default database path: {db_path}") # Initialize database engine = create_engine(Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.db_url) # Create tables if they don't exist ModelBase.metadata.create_all(engine) # Create scoped session factory session_factory = sessionmaker(bind=engine) session = scoped_session(session_factory) # Assign session to model classes PriceLevel.session = session SignalHistory.session = session logger.info(f"Database initialized with URL: {Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.db_url}") logger.info(f"Database session created and assigned to models") Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.db_initialized = True except Exception as e: logger.error(f"Failed to initialize database: {e}") logger.error(traceback.format_exc()) Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.db_initialized = False def __init__(self, config: dict) -> None: """ Initialize the strategy """ super().__init__(config) # 清除最新K线数据缓存，以避免内存泄漏 Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.latest_candles = {} # Initialize database connection from config if 'db_url' in config: Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.db_url = config['db_url'] # Initialize database session if not already initialized if not Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.db_initialized: Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.init_db_session() # 更新监控的交易对列表 self.update_monitored_pairs() def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # 更新监控的交易对列表 self.update_monitored_pairs() # Calculate ATR indicator dataframe['atr'] = ta.atr( high=dataframe['high'], low=dataframe['low'], close=dataframe['close'], length=self.atr_length ) # Add previous close price calculation dataframe['close_prev'] = dataframe['close'].shift(1) # Add previous ATR calculation dataframe['atr_prev'] = dataframe['atr'].shift(1) # 存储当前交易对的最新K线数据 current_pair = metadata['pair'] last_candle_index = len(dataframe) - 1 if last_candle_index >= 0: last_candle = dataframe.iloc[last_candle_index].copy() prev_candle = dataframe.iloc[last_candle_index-1].copy() if last_candle_index > 0 else None # 存储最新K线数据 Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.latest_candles[current_pair] = { 'last_candle': last_candle, 'prev_candle': prev_candle } # Add level crossing detection if self.check_level_crossing and self.dp and hasattr(self.dp, 'runmode'): # Only check for level crossing in live/dry run mode if self.dp.runmode.value in ('live', 'dry_run'): # Get active price levels for this pair try: # Ensure database is initialized if not Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.db_initialized: Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.init_db_session() # 获取所有价格水平，不仅仅是当前交易对的 all_levels = PriceLevel.get_levels() # 根据配置决定是检查所有交易对还是只检查当前交易对 if self.check_all_pairs: # 检查所有交易对的价格水平 levels_to_check = all_levels logger.debug(f"检查所有交易对的价格水平: {len(levels_to_check)} 个") else: # 只检查当前交易对的价格水平 levels_to_check = [level for level in all_levels if level.pair == current_pair] logger.debug(f"只检查当前交易对 {current_pair} 的价格水平: {len(levels_to_check)} 个") # 记录所有监控的价格水平（每隔一段时间记录一次） current_time = datetime.now() if Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.last_levels_log_time is None or (current_time - Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.last_levels_log_time).total_seconds() > Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.levels_log_interval: Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.last_levels_log_time = current_time logger.info(f"===== 所有监控的价格水平 ({len(all_levels)} 个) =====") # 按交易对分组显示 pairs_dict = {} for level in all_levels: if level.pair not in pairs_dict: pairs_dict[level.pair] = [] pairs_dict[level.pair].append(level) for p, levels in pairs_dict.items(): logger.info(f"交易对 {p}: {len(levels)} 个价格水平") for level in levels: logger.info(f" - 价格: {level.level:.8f}, 方向: {level.direction}, ID: {level.id}") # 记录所有交易对的当前价格 self.log_current_prices() # 如果启用了force_fetch_candles，主动获取所有交易对的K线数据 if self.force_fetch_candles and self.debug_all_pairs and self.dp: # 使用监控的交易对列表 monitored_pairs = Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.monitored_pairs logger.debug(f"需要获取K线数据的交易对: {monitored_pairs}") # 主动获取每个交易对的K线数据 for pair in monitored_pairs: # 跳过当前交易对，因为我们已经有它的数据了 if pair == current_pair: continue # 检查是否已经有最新的数据（5分钟内的） if pair in Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.latest_candles: candle_data = Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.latest_candles[pair] if 'timestamp' in candle_data: # 如果数据不超过5分钟，则跳过 elapsed_seconds = datetime.now(timezone.utc).timestamp() - candle_data['timestamp'] if elapsed_seconds < 300: # 5分钟 logger.debug(f"跳过获取{pair}的K线数据，因为已有最新数据（{elapsed_seconds:.0f}秒前）") continue # 获取K线数据 success = self.fetch_pair_candles(pair) if success: logger.info(f"成功获取{pair}的K线数据") else: logger.warning(f"无法获取{pair}的K线数据") # Initialize level crossing columns dataframe['level_cross_up'] = 0 dataframe['level_cross_down'] = 0 dataframe['level_wick_up'] = 0 # 新增：上影线流动性清扫信号 dataframe['level_wick_down'] = 0 # 新增：下影线流动性清扫信号 dataframe['level_id'] = 0 # Store the level ID for reference dataframe['level_price'] = 0.0 # Store the level price for reference # 如果启用了debug_all_pairs并且有足够的数据，则显示所有交易对的调试信息 if self.debug_all_pairs and self.enable_debug_logs: # 使用监控的交易对列表 monitored_pairs = Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.monitored_pairs # 显示每个交易对的调试信息 for pair in monitored_pairs: # 跳过当前交易对，因为我们会在后面处理它 if pair == current_pair: continue # 检查是否有这个交易对的K线数据 if pair in Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.latest_candles: pair_data = Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.latest_candles[pair] last_candle = pair_data.get('last_candle') prev_candle = pair_data.get('prev_candle') if last_candle is not None and prev_candle is not None: # 获取这个交易对的所有价格水平 pair_levels = [level for level in all_levels if level.pair == pair] # 显示每个价格水平的调试信息 for level in pair_levels: self._log_level_debug_info(level, last_candle, prev_candle) # Check each level for crossing - 根据配置检查价格水平 for level in levels_to_check: level_price = level.level level_direction = level.direction level_pair = level.pair # 只处理当前交易对的价格水平 if level_pair == current_pair: # 添加调试日志 if last_candle_index >= 0: last_candle = dataframe.iloc[last_candle_index] prev_candle = dataframe.iloc[last_candle_index-1] if last_candle_index > 0 else None # 只有当启用调试日志时才显示详细信息 if self.enable_debug_logs: self._log_level_debug_info(level, last_candle, prev_candle) # 检测实体向上突破（价格从下方穿过水平） if level_direction in [LevelDirection.UP, LevelDirection.BOTH]: # 修正：上穿定义为收盘价在关键价位上方 # 只有当K线收盘价高于水平时才触发 cross_up = (dataframe['close_prev'] < level_price) & (dataframe['close'] > level_price) # 添加调试日志 if last_candle_index >= 0 and prev_candle is not None: condition1 = prev_candle['close'] < level_price condition2 = last_candle['close'] > level_price final_condition = condition1 and condition2 # 强制检查最后一根K线的条件 last_candle_condition = final_condition # 只有当最后一根K线满足条件时才设置信号 if last_candle_condition: # 只对满足条件的K线设置信号 cross_up_indices = cross_up if isinstance(cross_up, pd.Series) else pd.Series([cross_up], index=[last_candle_index]) dataframe.loc[cross_up_indices, 'level_cross_up'] = 1 dataframe.loc[cross_up_indices, 'level_id'] = level.id dataframe.loc[cross_up_indices, 'level_price'] = level_price # 检测实体向下突破（价格从上方穿过水平） if level_direction in [LevelDirection.DOWN, LevelDirection.BOTH]: # 修正：下穿定义为收盘价在关键价位下方 # 只有当K线收盘价低于水平时才触发 cross_down = (dataframe['close_prev'] > level_price) & (dataframe['close'] < level_price) # 添加调试日志 if last_candle_index >= 0 and prev_candle is not None: condition1 = prev_candle['close'] > level_price condition2 = last_candle['close'] < level_price final_condition = condition1 and condition2 # 强制检查最后一根K线的条件 last_candle_condition = final_condition # 只有当最后一根K线满足条件时才设置信号 if last_candle_condition: # 只对满足条件的K线设置信号 cross_down_indices = cross_down if isinstance(cross_down, pd.Series) else pd.Series([cross_down], index=[last_candle_index]) dataframe.loc[cross_down_indices, 'level_cross_down'] = 1 dataframe.loc[cross_down_indices, 'level_id'] = level.id dataframe.loc[cross_down_indices, 'level_price'] = level_price # 检测上影线流动性清扫（上影线穿过水平但实体没有） if level_direction in [LevelDirection.WICK_UP, LevelDirection.WICK_BOTH]: # 修正：上影线穿过水平但实体保持在水平下方 # 确保上一根K线的高点低于水平，当前K线的高点高于水平，而且当前K线的开盘和收盘都低于水平 wick_up = (dataframe['high'].shift(1) < level_price) & (dataframe['high'] > level_price) & \ (dataframe['close'] < level_price) & (dataframe['open'] < level_price) # 添加调试日志 if last_candle_index >= 0 and prev_candle is not None: condition1 = prev_candle['high'] < level_price condition2 = last_candle['high'] > level_price condition3 = last_candle['close'] < level_price condition4 = last_candle['open'] < level_price final_condition = condition1 and condition2 and condition3 and condition4 # 强制检查最后一根K线的条件 last_candle_condition = final_condition # 只有当最后一根K线满足条件时才设置信号 if last_candle_condition: # 只对满足条件的K线设置信号 wick_up_indices = wick_up if isinstance(wick_up, pd.Series) else pd.Series([wick_up], index=[last_candle_index]) dataframe.loc[wick_up_indices, 'level_wick_up'] = 1 dataframe.loc[wick_up_indices, 'level_id'] = level.id dataframe.loc[wick_up_indices, 'level_price'] = level_price # 检测下影线流动性清扫（下影线穿过水平但实体没有） if level_direction in [LevelDirection.WICK_DOWN, LevelDirection.WICK_BOTH]: # 修正：下影线穿过水平但实体保持在水平上方 # 确保上一根K线的低点高于水平，当前K线的低点低于水平，而且当前K线的开盘和收盘都高于水平 wick_down = (dataframe['low'].shift(1) > level_price) & (dataframe['low'] < level_price) & \ (dataframe['close'] > level_price) & (dataframe['open'] > level_price) # 添加调试日志 if last_candle_index >= 0 and prev_candle is not None: condition1 = prev_candle['low'] > level_price condition2 = last_candle['low'] < level_price condition3 = last_candle['close'] > level_price condition4 = last_candle['open'] > level_price final_condition = condition1 and condition2 and condition3 and condition4 # 强制检查最后一根K线的条件 last_candle_condition = final_condition # 只有当最后一根K线满足条件时才设置信号 if last_candle_condition: # 只对满足条件的K线设置信号 wick_down_indices = wick_down if isinstance(wick_down, pd.Series) else pd.Series([wick_down], index=[last_candle_index]) dataframe.loc[wick_down_indices, 'level_wick_down'] = 1 dataframe.loc[wick_down_indices, 'level_id'] = level.id dataframe.loc[wick_down_indices, 'level_price'] = level_price except SQLAlchemyError as e: logger.error(f"Database error checking price levels: {e}") # Initialize columns even if there was an error dataframe['level_cross_up'] = 0 dataframe['level_cross_down'] = 0 dataframe['level_wick_up'] = 0 dataframe['level_wick_down'] = 0 dataframe['level_id'] = 0 dataframe['level_price'] = 0.0 except Exception as e: logger.error(f"Error checking price levels: {e}") logger.error(traceback.format_exc()) # Initialize columns even if there was an error dataframe['level_cross_up'] = 0 dataframe['level_cross_down'] = 0 dataframe['level_wick_up'] = 0 dataframe['level_wick_down'] = 0 dataframe['level_id'] = 0 dataframe['level_price'] = 0.0 else: # For backtesting/hyperopt, just add the columns with zeros dataframe['level_cross_up'] = 0 dataframe['level_cross_down'] = 0 dataframe['level_wick_up'] = 0 dataframe['level_wick_down'] = 0 dataframe['level_id'] = 0 dataframe['level_price'] = 0.0 return dataframe def populate_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # Add ATR sudden increase condition atr_increase = ( dataframe['atr'] > dataframe['atr'].shift(1) * self.atr_threshold ) # Add level crossing condition (if enabled) if self.check_level_crossing: level_cross_up = dataframe['level_cross_up'] == 1 level_wick_up = dataframe['level_wick_up'] == 1 # 新增：上影线流动性清扫 level_wick_down = dataframe['level_wick_down'] == 1 # 新增：下影线流动性清扫 # Buy on either ATR increase, level crossing up, 或流动性清扫 dataframe.loc[atr_increase | level_cross_up | level_wick_up | level_wick_down, 'buy'] = 1 else: # Original ATR signal only dataframe.loc[atr_increase, 'buy'] = 1 # Send Telegram notification when buy signal occurs if self.dp.runmode.value in ('live', 'dry_run'): last_candle = dataframe.iloc[-1] if last_candle['buy'] == 1: # Pass current candle data to notification method self.send_telegram_notification(metadata['pair'], last_candle) return dataframe # Add exit trend method to satisfy interface requirement def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Exit signal placeholder implementation. Sets 'exit_long' to 0 for all rows (no active exit signals). """ dataframe['exit_long'] = 0 # Add level crossing down as exit signal if enabled if self.check_level_crossing: level_cross_down = dataframe['level_cross_down'] == 1 dataframe.loc[level_cross_down, 'exit_long'] = 1 return dataframe def send_telegram_notification(self, pair: str, candle: Series): """Send notification via Telegram with detailed metrics""" try: # Calculate ATR change rate safely if candle['atr_prev'] != 0: atr_change = (candle['atr'] / candle['atr_prev'] - 1) * 100 else: atr_change = 0 # Determine signal type signal_type = "" signal_details = "" signal_db_type = "" # Type to store in database level_id = None level_price = None atr_value = None if self.check_level_crossing: if candle.get('level_cross_up', 0) == 1: level_id = int(candle.get('level_id', 0)) level_price = float(candle.get('level_price', 0.0)) signal_type = "🔼 Level Cross UP" signal_details = f"▫ 价格穿越上升点位: {level_price:.6f} (ID: {level_id})" signal_db_type = "level_cross_up" elif candle.get('level_cross_down', 0) == 1: level_id = int(candle.get('level_id', 0)) level_price = float(candle.get('level_price', 0.0)) signal_type = "🔽 Level Cross DOWN" signal_details = f"▫ 价格穿越下降点位: {level_price:.6f} (ID: {level_id})" signal_db_type = "level_cross_down" elif candle.get('level_wick_up', 0) == 1: level_id = int(candle.get('level_id', 0)) level_price = float(candle.get('level_price', 0.0)) signal_type = "🔝 Wick UP Liquidity Sweep" signal_details = f"▫ 上影线扫动流动性: {level_price:.6f} (ID: {level_id})" signal_db_type = "level_wick_up" elif candle.get('level_wick_down', 0) == 1: level_id = int(candle.get('level_id', 0)) level_price = float(candle.get('level_price', 0.0)) signal_type = "🔻 Wick DOWN Liquidity Sweep" signal_details = f"▫ 下影线扫动流动性: {level_price:.6f} (ID: {level_id})" signal_db_type = "level_wick_down" elif candle['atr'] > candle['atr_prev'] * self.atr_threshold: signal_type = "🚨 ATR Surge" signal_details = f"▫ ATR变动率: {atr_change:.2f}%" signal_db_type = "atr_surge" atr_value = float(candle['atr']) else: signal_type = "🚨 ATR Surge" signal_details = f"▫ ATR变动率: {atr_change:.2f}%" signal_db_type = "atr_surge" atr_value = float(candle['atr']) # Format message with required metrics message = ( f"{signal_type} on {pair} ({self.timeframe})\n" f"{signal_details}\n" f"▫ 实际ATR: {candle['atr']:.6f}\n" f"▫ 前一价格: {candle['close_prev']:.6f}\n" f"▫ 当前价格: {candle['close']:.6f}" ) self.dp.send_msg(message) # Store signal in history if signal_db_type: try: # Ensure database is initialized if not Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.db_initialized: Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.init_db_session() # 尝试获取关联价格点位的备注（如果有的话） remarks = None if level_id and level_id > 0: try: price_level = PriceLevel.session.get(PriceLevel, level_id) if price_level and hasattr(price_level, 'remarks'): remarks = price_level.remarks if remarks: logger.info(f"获取到关联价格点位 ID:{level_id} 的备注: {remarks}") except Exception as e: logger.debug(f"获取价格点位备注时出错: {e}") # 记录信号并包含备注信息 signal = SignalHistory.add_signal( pair=pair, signal_type=signal_db_type, prev_price=float(candle['close_prev']), current_price=float(candle['close']), level_id=level_id if level_id and level_id > 0 else None, level_price=level_price if level_price else None, atr_value=atr_value if atr_value else float(candle['atr']), remarks=remarks # 添加备注信息 ) logger.info(f"Signal recorded in history: {signal_type} for {pair} {f'with remarks: {remarks}' if remarks else ''}") except Exception as e: logger.error(f"Failed to record signal in history: {e}") logger.error(traceback.format_exc()) except Exception as e: logger.error(f"Failed to send Telegram notification: {e}") logger.error(traceback.format_exc()) @staticmethod def add_price_level(pair: str, level: float, direction: str = "both", confirm_close: bool = False, remarks: Optional[str] = None) -> Dict[str, Any]: """ Add a price level to monitor for crossing :param pair: Trading pair (e.g. BTC/USDT) :param level: Price level to monitor :param direction: Direction to monitor ('up', 'down', 'both', 'wick_up', 'wick_down', 'wick_both') - up: 当收盘价从下方穿过水平时触发 - down: 当收盘价从上方穿过水平时触发 - both: 当收盘价从任意方向穿过水平时触发 - wick_up: 上影线流动性清扫（上影线穿过水平但实体保持在下方） - wick_down: 下影线流动性清扫（下影线穿过水平但实体保持在上方） - wick_both: 双向流动性清扫（任意方向的影线穿过水平） :param confirm_close: 已弃用参数，保留为向后兼容。现在所有的穿越信号都基于收盘价。 :param remarks: 可选的备注信息 :return: Dictionary with level information """ try: # Ensure database is initialized if not Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.db_initialized: Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.init_db_session() price_level = PriceLevel.add_level(pair, level, direction, confirm_close, remarks) return { "id": price_level.id, "pair": price_level.pair, "level": price_level.level, "direction": price_level.direction, "created_at": price_level.created_at.isoformat(), "active": bool(price_level.active), "confirm_close": bool(price_level.confirm_close), "remarks": price_level.remarks } except Exception as e: logger.error(f"Failed to add price level: {e}") logger.error(traceback.format_exc()) return {"error": str(e)} @staticmethod def get_price_levels(pair: Optional[str] = None) -> List[Dict[str, Any]]: """ Get all active price levels :param pair: Optional trading pair to filter by :return: List of dictionaries with level information """ try: # Ensure database is initialized if not Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.db_initialized: Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.init_db_session() levels = PriceLevel.get_levels(pair) result = [] for level in levels: result.append({ "id": level.id, "pair": level.pair, "level": level.level, "direction": level.direction, "created_at": level.created_at.isoformat(), "active": bool(level.active), "confirm_close": bool(level.confirm_close), "remarks": level.remarks }) return result except Exception as e: logger.error(f"Failed to get price levels: {e}") logger.error(traceback.format_exc()) return [] @staticmethod def delete_price_level(level_id: int) -> Dict[str, Any]: """ Delete a price level :param level_id: ID of the price level to delete :return: Success/failure dictionary """ try: # Ensure database is initialized if not Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.db_initialized: Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.init_db_session() PriceLevel.delete_level(level_id) return {"success": True} except Exception as e: logger.error(f"Failed to delete price level: {e}") logger.error(traceback.format_exc()) return {"success": False, "error": str(e)} @staticmethod def update_price_level(level_id: int, level: Optional[float] = None, direction: Optional[str] = None, confirm_close: Optional[bool] = None, remarks: Optional[str] = None) -> Dict[str, Any]: """ Update an existing price level :param level_id: ID of the price level to update :param level: New price level value (optional) :param direction: New direction (optional) :param confirm_close: New confirm close setting (optional) :param remarks: New remarks (optional) :return: Success/failure dictionary with updated level information """ # Validate direction if provided if direction is not None and direction not in [d.value for d in LevelDirection]: return { "success": False, "error": f"Direction must be one of: {', '.join([d.value for d in LevelDirection])}" } try: # Ensure database is initialized if not Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.db_initialized: Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.init_db_session() price_level = PriceLevel.session.get(PriceLevel, level_id) if not price_level: return {"success": False, "error": f"Price level with ID {level_id} not found"} # Update fields if provided if level is not None: price_level.level = level if direction is not None: price_level.direction = direction if confirm_close is not None: price_level.confirm_close = 1 if confirm_close else 0 if remarks is not None: price_level.remarks = remarks PriceLevel.session.commit() return { "success": True, "level": { "id": price_level.id, "pair": price_level.pair, "level": price_level.level, "direction": price_level.direction, "created_at": price_level.created_at.isoformat(), "active": bool(price_level.active), "confirm_close": bool(price_level.confirm_close), "remarks": price_level.remarks } } except Exception as e: logger.error(f"Failed to update price level: {e}") logger.error(traceback.format_exc()) return {"success": False, "error": str(e)} @staticmethod def get_signal_history(pair: Optional[str] = None, signal_type: Optional[str] = None, start_date: Optional[str] = None, end_date: Optional[str] = None, limit: int = 100, offset: int = 0) -> List[Dict[str, Any]]: """ Get signal history with optional filtering :param pair: Optional trading pair to filter by :param signal_type: Optional signal type to filter by ('level_cross_up', 'level_cross_down', 'atr_surge') :param start_date: Optional start date for filtering (ISO format string) :param end_date: Optional end date for filtering (ISO format string) :param limit: Maximum number of results to return (0 means no limit) :param offset: Number of records to skip (for pagination) :return: List of dictionaries with signal information """ try: # Ensure database is initialized if not Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.db_initialized: Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.init_db_session() # Convert date strings to datetime objects if provided start_dt = None if start_date: try: start_dt = datetime.fromisoformat(start_date) except ValueError: logger.warning(f"Invalid start_date format: {start_date}, expected ISO format") end_dt = None if end_date: try: end_dt = datetime.fromisoformat(end_date) except ValueError: logger.warning(f"Invalid end_date format: {end_date}, expected ISO format") # Get signals from database signals = SignalHistory.get_signals( pair=pair, signal_type=signal_type, start_date=start_dt, end_date=end_dt, limit=limit, offset=offset ) # Convert to dictionaries result = [] for signal in signals: # 尝试获取关联价格点位的备注 remarks = None if signal.level_id: try: price_level = PriceLevel.session.get(PriceLevel, signal.level_id) if price_level and hasattr(price_level, 'remarks'): remarks = price_level.remarks except Exception as e: logger.debug(f"获取价格点位备注时出错: {e}") signal_dict = { "id": signal.id, "pair": signal.pair, "signal_type": signal.signal_type, "level_id": signal.level_id, "level_price": signal.level_price, "prev_price": signal.prev_price, "current_price": signal.current_price, "atr_value": signal.atr_value, "created_at": signal.created_at.isoformat(), "remarks": remarks } result.append(signal_dict) return result except Exception as e: logger.error(f"Failed to get signal history: {e}") logger.error(traceback.format_exc()) return [] def _log_level_debug_info(self, level, last_candle, prev_candle): """ 显示价格水平的调试信息 :param level: 价格水平对象 :param last_candle: 最新K线数据 :param prev_candle: 上一根K线数据 """ level_pair = level.pair level_price = level.level level_direction = level.direction # 显示基本信息 logger.info(f"===== 调试信息 - {level_pair} - 价格水平: {level_price} =====") logger.info(f"方向: {level_direction}") # 显示K线数据 logger.info(f"上一根K线 - 开盘: {prev_candle['open']}, 收盘: {prev_candle['close']}, 最高: {prev_candle['high']}, 最低: {prev_candle['low']}") logger.info(f"当前K线 - 开盘: {last_candle['open']}, 收盘: {last_candle['close']}, 最高: {last_candle['high']}, 最低: {last_candle['low']}") # 检查向上突破条件 if level_direction in [LevelDirection.UP, LevelDirection.BOTH]: condition1 = prev_candle['close'] < level_price condition2 = last_candle['close'] > level_price final_condition = condition1 and condition2 logger.info(f"{level_pair} - 向上突破 - 条件1(上一收盘<水平): {condition1}, 条件2(当前收盘>水平): {condition2}, 最终结果: {final_condition}") if final_condition: logger.info(f"UP CROSS detected for {level_pair} at level {level_price} (ID: {level.id})") # 检查向下突破条件 if level_direction in [LevelDirection.DOWN, LevelDirection.BOTH]: condition1 = prev_candle['close'] > level_price condition2 = last_candle['close'] < level_price final_condition = condition1 and condition2 logger.info(f"{level_pair} - 向下突破 - 条件1(上一收盘>水平): {condition1}, 条件2(当前收盘<水平): {condition2}, 最终结果: {final_condition}") if final_condition: logger.info(f"DOWN CROSS detected for {level_pair} at level {level_price} (ID: {level.id})") # 检查上影线流动性清扫条件 if level_direction in [LevelDirection.WICK_UP, LevelDirection.WICK_BOTH]: condition1 = prev_candle['high'] < level_price condition2 = last_candle['high'] > level_price condition3 = last_candle['close'] < level_price condition4 = last_candle['open'] < level_price final_condition = condition1 and condition2 and condition3 and condition4 logger.info(f"{level_pair} - 上影线流动性清扫 - 条件1(上一高点<水平): {condition1}, 条件2(当前高点>水平): {condition2}, 条件3(当前收盘<水平): {condition3}, 条件4(当前开盘<水平): {condition4}, 最终结果: {final_condition}") if final_condition: logger.info(f"WICK UP detected for {level_pair} at level {level_price} (ID: {level.id})") # 检查下影线流动性清扫条件 if level_direction in [LevelDirection.WICK_DOWN, LevelDirection.WICK_BOTH]: condition1 = prev_candle['low'] > level_price condition2 = last_candle['low'] < level_price condition3 = last_candle['close'] > level_price condition4 = last_candle['open'] > level_price final_condition = condition1 and condition2 and condition3 and condition4 logger.info(f"{level_pair} - 下影线流动性清扫 - 条件1(上一低点>水平): {condition1}, 条件2(当前低点<水平): {condition2}, 条件3(当前收盘>水平): {condition3}, 条件4(当前开盘>水平): {condition4}, 最终结果: {final_condition}") if final_condition: logger.info(f"WICK DOWN detected for {level_pair} at level {level_price} (ID: {level.id})") def fetch_pair_candles(self, pair: str) -> bool: """ 主动获取指定交易对的K线数据 :param pair: 交易对名称 :return: 是否成功获取数据 """ if not self.dp: logger.warning(f"无法获取{pair}的K线数据：数据提供者未初始化") return False try: # 获取当前时间戳 current_time = datetime.now(timezone.utc).timestamp() # 获取K线数据 # 注意：freqtrade的DataProvider没有直接的方法来获取指定交易对的最新K线 # 我们尝试使用可用的方法来获取数据 # 方法1：尝试使用get_pair_dataframe try: candles = self.dp.get_pair_dataframe( pair=pair, timeframe=self.timeframe ) if candles is not None and len(candles) >= 2: # 获取最后两根K线 last_candle = candles.iloc[-1].copy() prev_candle = candles.iloc[-2].copy() # 存储最新K线数据 Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.latest_candles[pair] = { 'last_candle': last_candle, 'prev_candle': prev_candle, 'timestamp': current_time } logger.debug(f"成功获取{pair}的K线数据（方法1）") return True except Exception as e: logger.debug(f"方法1获取{pair}的K线数据失败：{e}") # 方法2：尝试使用ohlcv方法（如果可用） try: if hasattr(self.dp, 'ohlcv') and callable(getattr(self.dp, 'ohlcv')): # 计算开始时间戳（获取最近的2根K线） timeframe_mins = self.get_timeframe_minutes(self.timeframe) start_time = int((current_time - timeframe_mins * 60 * 3) * 1000) # 毫秒 candles = self.dp.ohlcv( pair=pair, timeframe=self.timeframe, since=start_time ) if candles is not None and len(candles) >= 2: # 转换为DataFrame import pandas as pd df = pd.DataFrame(candles, columns=['date', 'open', 'high', 'low', 'close', 'volume']) # 获取最后两根K线 last_candle = df.iloc[-1].copy() prev_candle = df.iloc[-2].copy() # 存储最新K线数据 Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.latest_candles[pair] = { 'last_candle': last_candle, 'prev_candle': prev_candle, 'timestamp': current_time } logger.debug(f"成功获取{pair}的K线数据（方法2）") return True except Exception as e: logger.debug(f"方法2获取{pair}的K线数据失败：{e}") # 方法3：如果是当前交易对，尝试从最近的信号中获取数据 try: # 获取最近的信号历史 signals = SignalHistory.get_signals(pair=pair, limit=1) if signals and len(signals) > 0: signal = signals[0] # 创建简化的K线数据 last_candle = pd.Series({ 'open': signal.current_price, 'high': signal.current_price * 1.001, # 估计值 'low': signal.current_price * 0.999, # 估计值 'close': signal.current_price, 'volume': 0 }) prev_candle = pd.Series({ 'open': signal.prev_price, 'high': signal.prev_price * 1.001, # 估计值 'low': signal.prev_price * 0.999, # 估计值 'close': signal.prev_price, 'volume': 0 }) # 存储最新K线数据 Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.latest_candles[pair] = { 'last_candle': last_candle, 'prev_candle': prev_candle, 'timestamp': current_time } logger.debug(f"成功获取{pair}的K线数据（方法3：从信号历史）") return True except Exception as e: logger.debug(f"方法3获取{pair}的K线数据失败：{e}") # 方法4：使用mock数据（最后的备选方案） try: # 获取价格水平作为参考价格 levels = PriceLevel.get_levels(pair) if levels and len(levels) > 0: # 使用第一个价格水平作为参考 reference_price = levels[0].level # 创建mock K线数据 last_candle = pd.Series({ 'open': reference_price * 1.01, # 略高于参考价格 'high': reference_price * 1.02, 'low': reference_price * 1.005, 'close': reference_price * 1.015, 'volume': 0 }) prev_candle = pd.Series({ 'open': reference_price * 0.99, # 略低于参考价格 'high': reference_price * 1.005, 'low': reference_price * 0.985, 'close': reference_price * 1.0, 'volume': 0 }) # 存储最新K线数据 Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.latest_candles[pair] = { 'last_candle': last_candle, 'prev_candle': prev_candle, 'timestamp': current_time } logger.debug(f"成功获取{pair}的K线数据（方法4：mock数据）") return True except Exception as e: logger.debug(f"方法4获取{pair}的K线数据失败：{e}") logger.warning(f"无法获取{pair}的K线数据：所有方法均失败") return False except Exception as e: logger.error(f"获取{pair}的K线数据时出错：{e}") logger.debug(traceback.format_exc()) return False def get_timeframe_minutes(self, timeframe: str) -> int: """ 获取timeframe对应的分钟数 :param timeframe: 时间周期字符串，例如 '1m', '5m', '1h', '1d' :return: 分钟数 """ if not timeframe: return 1 # 默认为1分钟 # 提取数字和单位 import re match = re.match(r'(\d+)([mhd])', timeframe) if not match: return 1 # 默认为1分钟 value = int(match.group(1)) unit = match.group(2) # 根据单位转换为分钟 if unit == 'm': return value elif unit == 'h': return value * 60 elif unit == 'd': return value * 60 * 24 else: return 1 # 默认为1分钟 def log_current_prices(self): """ 记录所有监控交易对的当前价格 """ if not self.dp: return try: # 使用监控的交易对列表 monitored_pairs = Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.monitored_pairs logger.info(f"===== 所有监控交易对的当前价格 =====") # 获取每个交易对的当前价格 for pair in sorted(monitored_pairs): try: # 尝试获取当前价格 current_price = None # 方法1：使用ticker_dict if hasattr(self.dp, 'ticker_dict') and callable(getattr(self.dp, 'ticker_dict')): ticker = self.dp.ticker_dict(pair) if ticker and 'last' in ticker: current_price = ticker['last'] # 方法2：使用get_pair_dataframe if current_price is None: try: candles = self.dp.get_pair_dataframe(pair=pair, timeframe=self.timeframe) if candles is not None and len(candles) > 0: current_price = candles.iloc[-1]['close'] except Exception: pass # 方法3：从最新K线数据中获取 if current_price is None and pair in Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.latest_candles: candle_data = Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.latest_candles[pair] if 'last_candle' in candle_data and 'close' in candle_data['last_candle']: current_price = candle_data['last_candle']['close'] # 方法4：从价格水平估计 if current_price is None: # 获取这个交易对的所有价格水平 all_levels = PriceLevel.get_levels() pair_levels = [level for level in all_levels if level.pair == pair] if pair_levels: # 使用价格水平的平均值作为估计 total = sum(level.level for level in pair_levels) current_price = total / len(pair_levels) # 记录当前价格 if current_price is not None: logger.info(f"{pair}: {current_price}") else: logger.info(f"{pair}: 无法获取价格") except Exception as e: logger.debug(f"获取{pair}当前价格时出错：{e}") logger.info(f"{pair}: 获取价格出错") except Exception as e: logger.error(f"记录当前价格时出错：{e}") logger.debug(traceback.format_exc()) def update_monitored_pairs(self) -> bool: """ 从数据库中更新需要监控的交易对列表 :return: 是否成功更新 """ try: # 确保数据库已初始化 if not Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.db_initialized: Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.init_db_session() # 获取当前时间 current_time = datetime.now() # 检查是否需要更新（如果是第一次或者已经超过更新间隔） if not self.auto_update_monitored_pairs: return False if Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.last_pairs_update_time is not None: elapsed_seconds = (current_time - Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.last_pairs_update_time).total_seconds() if elapsed_seconds < Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.pairs_update_interval: return False # 获取所有价格水平 all_levels = PriceLevel.get_levels() # 提取所有交易对 old_pairs = Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.monitored_pairs.copy() Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.monitored_pairs = set() for level in all_levels: Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.monitored_pairs.add(level.pair) # 更新时间戳 Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.last_pairs_update_time = current_time # 检查是否有变化 added_pairs = Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.monitored_pairs - old_pairs removed_pairs = old_pairs - Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.monitored_pairs if added_pairs or removed_pairs: logger.info(f"===== 监控的交易对列表已更新 =====") logger.info(f"当前监控的交易对: {len(Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.monitored_pairs)} 个") for pair in sorted(Github_xiedidan_freqtrade__atr_level_signal__20250724_100111.monitored_pairs): logger.info(f" - {pair}") if added_pairs: logger.info(f"新增的交易对: {len(added_pairs)} 个") for pair in sorted(added_pairs): logger.info(f" + {pair}") if removed_pairs: logger.info(f"移除的交易对: {len(removed_pairs)} 个") for pair in sorted(removed_pairs): logger.info(f" - {pair}") return True except Exception as e: logger.error(f"更新监控的交易对列表时出错: {e}") logger.debug(traceback.format_exc()) return False