""" 加密货币K线形态策略模块 基于原有K线形态策略，适配加密货币市场 """ import pandas as pd from typing import Dict, List, Any from datetime import datetime, timedelta from loguru import logger from ..data.binance_fetcher import BinanceFetcher from ..utils.notification import NotificationManager from ..database.mysql_database_manager import MySQLDatabaseManager from .kline_pattern_strategy import KLinePatternStrategy from .base_strategy import StrategyResult class CryptoKLinePatternStrategy(KLinePatternStrategy): """加密货币K线形态策略类""" def __init__(self, data_fetcher: BinanceFetcher, notification_manager: NotificationManager, config: Dict[str, Any], db_manager: MySQLDatabaseManager = None): """ 初始化加密货币K线形态策略 Args: data_fetcher: Binance数据获取器 notification_manager: 通知管理器 config: 策略配置 db_manager: 数据库管理器 """ # 先调用父类构造函数初始化基本属性 super().__init__(data_fetcher, notification_manager, config, db_manager) # 然后替换和定制加密货币特有属性 self.data_fetcher = data_fetcher # 替换为Binance数据获取器 self.strategy_name = "加密货币K线形态策略" self.timeframes = config.get('timeframes', ['4hour', 'daily', 'weekly']) self.max_turnover_ratio = config.get('max_turnover_ratio', 100.0) # 加密货币换手率通常更高 # 加密货币特有参数 self.quote_asset = config.get('quote_asset', 'USDT') # 计价货币 self.min_volume_usdt = config.get('min_volume_usdt', 1000000) # 最小24h交易量（USDT） # 热门币种缓存机制 self._hot_symbols_cache = None self._cache_timestamp = None self._cache_duration = 300 # 缓存5分钟 # 更新策略ID为加密货币版本 self.strategy_id = self.db_manager.create_or_update_strategy( strategy_name=self.strategy_name, strategy_type="crypto_kline_pattern", description="加密货币强势上涨+多空博弈+突破确认形态识别策略", config=config ) logger.info(f"加密货币K线形态策略初始化完成 (策略ID: {self.strategy_id})") def _get_cached_hot_symbols(self, max_symbols: int) -> List[str]: """ 获取缓存的热门交易对列表 Args: max_symbols: 最大交易对数量 Returns: 交易对列表 """ import time current_time = time.time() # 检查缓存是否有效 if (self._hot_symbols_cache is not None and self._cache_timestamp is not None and current_time - self._cache_timestamp < self._cache_duration): logger.info(f"🔄 使用缓存的热门交易对数据 ({len(self._hot_symbols_cache)} 个)") return self._hot_symbols_cache[:max_symbols] # 缓存失效或不存在，重新获取 logger.info(f"🔥 获取热门交易对 (TOP{max_symbols})...") hot_symbols = self.data_fetcher.get_top_volume_symbols( quote_asset=self.quote_asset, limit=max_symbols * 2 # 多获取一些以备过滤 ) if hot_symbols: self._hot_symbols_cache = hot_symbols self._cache_timestamp = current_time logger.info(f"✅ 热门交易对获取成功，已缓存 {len(self._hot_symbols_cache)} 个") return self._hot_symbols_cache[:max_symbols] else: logger.error("❌ 热门交易对数据为空") return [] def clear_hot_symbols_cache(self): """清除热门交易对缓存，强制下次重新获取""" self._hot_symbols_cache = None self._cache_timestamp = None logger.info("🔄 热门交易对缓存已清除") def _check_current_price_validity(self, symbol: str, entry_price: float, tolerance: float = 0.05) -> bool: """ 检查当前价格是否仍然适合入场（不低于入场价的5%） Args: symbol: 交易对符号 entry_price: 建议入场价格 tolerance: 价格下跌容忍度，默认5% Returns: bool: 当前价格是否仍然有效 """ try: # 获取最新价格数据（使用24小时数据） current_data = self.data_fetcher.get_historical_klines( symbol, '1d', limit=2 # 获取最近2天数据 ) if current_data.empty: logger.warning(f"无法获取 {symbol} 的最新价格数据") return True # 无法获取价格时保守处理，保留信号 current_price = current_data.iloc[-1]['close'] min_valid_price = entry_price * (1 - tolerance) is_valid = current_price >= min_valid_price if not is_valid: price_drop_pct = (entry_price - current_price) / entry_price * 100 logger.info(f"🔻 {symbol} 价格过滤: 当前价 {current_price:.4f} < 入场价 {entry_price:.4f} (下跌{price_drop_pct:.1f}%)") return is_valid except Exception as e: logger.error(f"检查 {symbol} 当前价格失败: {e}") return True # 出错时保守处理，保留信号 def _filter_recent_signals(self, signals: List[Dict[str, Any]], days: int = 7) -> List[Dict[str, Any]]: """ 过滤最近N天内产生的信号，并检查当前价格有效性（加密货币版本） Args: signals: 信号列表 days: 最近天数，默认7天 Returns: 过滤后的信号列表 """ if not signals: return signals from datetime import datetime, date import pandas as pd current_date = datetime.now().date() recent_signals = [] price_filtered_count = 0 for signal in signals: signal_date = signal.get('confirmation_date') or signal.get('date') # 处理不同的日期格式 if isinstance(signal_date, str): try: signal_date = pd.to_datetime(signal_date).date() except: continue elif hasattr(signal_date, 'date'): signal_date = signal_date.date() elif not isinstance(signal_date, date): continue # 计算信号距今天数 days_ago = (current_date - signal_date).days # 只保留最近N天内的信号 if days_ago <= days: # 检查当前价格是否仍然有效 symbol = signal.get('stock_code', '') # 在加密货币中，stock_code存储的是交易对符号 entry_price = signal.get('yin_high', 0) if symbol and entry_price > 0: if self._check_current_price_validity(symbol, entry_price): recent_signals.append(signal) logger.debug(f"✅ 保留有效信号: {symbol} {signal_date} (距今{days_ago}天)") else: price_filtered_count += 1 logger.debug(f"🔻 价格过滤信号: {symbol} {signal_date} (价格已跌破入场价)") else: # 缺少必要信息时保守处理 recent_signals.append(signal) logger.debug(f"✅ 保留信号(缺少价格信息): {signal_date} (距今{days_ago}天)") else: logger.debug(f"🗓️ 过滤历史信号: {signal_date} (距今{days_ago}天)") # 统计过滤结果 time_filtered_count = len(signals) - len(recent_signals) - price_filtered_count if len(recent_signals) != len(signals): logger.info(f"📅 加密货币信号过滤统计: 总共{len(signals)}个 → 保留{len(recent_signals)}个") if time_filtered_count > 0: logger.info(f" 🗓️ 时间过滤: {time_filtered_count}个") if price_filtered_count > 0: logger.info(f" 🔻 价格过滤: {price_filtered_count}个") return recent_signals def analyze_symbol(self, symbol: str, timeframes: List[str] = None, session_id: int = None) -> Dict[str, StrategyResult]: """ 分析单个交易对的K线形态 Args: symbol: 交易对符号，如'BTCUSDT' timeframes: 时间周期列表，如果为None则使用策略默认周期 session_id: 扫描会话ID Returns: 时间周期到策略结果的映射 """ if timeframes is None: timeframes = self.timeframes results = {} symbol_name = self.data_fetcher.get_symbol_name(symbol) for timeframe in timeframes: start_time = datetime.now() try: # 转换时间周期格式 binance_interval = self.data_fetcher.convert_timeframe(timeframe) # 计算需要获取的K线数量(使用limit更稳定) if timeframe == 'daily': limit = 60 # 获取60根日线 elif timeframe == '4hour': limit = 180 # 获取180根4小时线(约30天) elif timeframe == 'weekly': limit = 26 # 获取26根周线(约半年) else: limit = 100 # 默认100根 logger.info(f"🔍 分析交易对: {symbol} | 周期: {timeframe}") # 获取历史数据(使用limit参数,更可靠) df = self.data_fetcher.get_historical_klines( symbol, binance_interval, limit=limit ) if df.empty: logger.warning(f"{symbol} {timeframe} 数据为空") results[timeframe] = StrategyResult( strategy_name=self.strategy_name, stock_code=symbol, timeframe=timeframe, signals=[], success=False, error="数据为空", execution_time=(datetime.now() - start_time).total_seconds() ) continue # 计算K线特征 df_with_features = self.calculate_kline_features(df) # 检测形态（返回已确认信号和形态形成） confirmed_signals, formed_patterns = self.detect_pattern(df_with_features) # 过滤3天内的信号，7天内的形态 recent_signals = self._filter_recent_signals(confirmed_signals, days=3) recent_patterns = self._filter_recent_signals(formed_patterns, days=7) # 处理确认信号格式 formatted_signals = [] for signal in recent_signals: formatted_signal = { 'date': signal['date'], 'signal_type': signal['pattern_type'], 'price': signal['breakout_price'], 'confidence': signal['final_yang_entity_ratio'], 'stock_code': symbol, # 添加交易对代码 'stock_name': symbol_name, 'status': 'confirmed', 'details': { 'yin_high': signal['yin_high'], 'breakout_amount': signal['breakout_amount'], 'breakout_pct': signal['breakout_pct'], 'ema20_price': signal['ema20_price'], 'turnover_ratio': signal['turnover_ratio'], 'breakout_position': signal.get('breakout_position', 4), 'pattern_subtype': signal.get('pattern_subtype', '') # 添加形态子类型 } } # 添加回踩确认信息 if not signal.get('confirmation_pending', True): formatted_signal['details'].update({ 'new_high_confirmed': signal.get('new_high_confirmed', True), 'new_high_price': signal.get('new_high_price'), 'new_high_date': signal.get('new_high_date'), 'confirmation_date': signal.get('confirmation_date'), 'confirmation_days': signal.get('confirmation_days'), 'pullback_distance': signal.get('pullback_distance') }) formatted_signals.append(formatted_signal) # 将信号添加到监控列表 signal['stock_code'] = symbol signal['stock_name'] = symbol_name signal['timeframe'] = timeframe self.add_triggered_signal(signal) # 保存信号到数据库(标记为加密货币) if session_id is not None: try: signal_id = self.db_manager.save_stock_signal( session_id=session_id, strategy_id=self.strategy_id, signal=signal, asset_type='crypto' # 标记为加密货币 ) logger.debug(f"加密货币信号已保存到数据库: signal_id={signal_id}") except Exception as e: logger.error(f"保存加密货币信号到数据库失败: {e}") # 处理形态形成格式 formatted_patterns = [] for pattern in recent_patterns: formatted_pattern = { 'date': pattern['date'], 'pattern_type': pattern['pattern_type'], 'price': pattern['breakout_price'], 'confidence': pattern['final_yang_entity_ratio'], 'stock_code': symbol, # 添加交易对代码 'stock_name': symbol_name, 'status': 'formed', 'details': { 'yin_high': pattern['yin_high'], 'breakout_amount': pattern['breakout_amount'], 'breakout_pct': pattern['breakout_pct'], 'ema20_price': pattern['ema20_price'], 'turnover_ratio': pattern['turnover_ratio'], 'breakout_position': pattern.get('breakout_position', 4), 'pattern_subtype': pattern.get('pattern_subtype', '') # 添加形态子类型 } } formatted_patterns.append(formatted_pattern) execution_time = (datetime.now() - start_time).total_seconds() results[timeframe] = StrategyResult( strategy_name=self.strategy_name, stock_code=symbol, timeframe=timeframe, signals=formatted_signals, patterns=formatted_patterns, success=True, execution_time=execution_time ) # 美化统计日志 if formatted_signals or formatted_patterns: logger.info(f"✅ {symbol} {timeframe}周期: 信号={len(formatted_signals)}个, 形态={len(formatted_patterns)}个") if formatted_signals: logger.info(f" 🎯 已确认信号:") for i, signal in enumerate(formatted_signals, 1): logger.info(f" {i}. {signal['date']} | 价格: {signal['price']:.4f} | {signal['signal_type']}") if formatted_patterns: logger.info(f" 📊 形态形成:") for i, pattern in enumerate(formatted_patterns, 1): logger.info(f" {i}. {pattern['date']} | 价格: {pattern['price']:.4f} | {pattern['pattern_type']}") else: logger.debug(f"📭 {symbol} {timeframe}周期: 无信号和形态") except Exception as e: logger.error(f"分析交易对 {symbol} {timeframe}周期失败: {e}") execution_time = (datetime.now() - start_time).total_seconds() results[timeframe] = StrategyResult( strategy_name=self.strategy_name, stock_code=symbol, timeframe=timeframe, signals=[], patterns=[], success=False, error=str(e), execution_time=execution_time ) return results def scan_market(self, symbol_list: List[str] = None, max_symbols: int = 100) -> Dict[str, Dict[str, List[Dict[str, Any]]]]: """ 扫描加密货币市场 Args: symbol_list: 交易对列表，如果为None则使用热门交易对 max_symbols: 最大扫描交易对数量 Returns: 所有交易对的分析结果 """ logger.info("🚀" + "="*70) logger.info("🌍 开始加密货币市场K线形态扫描") logger.info("🚀" + "="*70) # 创建扫描会话 scan_config = { 'max_symbols': max_symbols, 'data_source': 'Binance', 'quote_asset': self.quote_asset, 'timeframes': self.timeframes } session_id = self.db_manager.create_scan_session( strategy_id=self.strategy_id, scan_config=scan_config ) if symbol_list is None: # 使用缓存的热门交易对数据 symbol_list = self._get_cached_hot_symbols(max_symbols) if symbol_list: logger.info(f"📊 数据源: Binance热门交易对 | 扫描交易对: {len(symbol_list)} 个") else: logger.error("❌ 热门交易对数据为空，无法进行扫描") return {} results = {} total_signals = 0 total_patterns = 0 for i, symbol in enumerate(symbol_list): logger.info(f"⏳ 扫描进度: [{i+1:3d}/{len(symbol_list):3d}] 🔍 {symbol}") try: symbol_results = self.analyze_symbol(symbol, session_id=session_id) # 统计信号和形态数量 symbol_signal_count = sum(result.get_signal_count() for result in symbol_results.values()) symbol_pattern_count = sum(result.get_pattern_count() for result in symbol_results.values()) # 只保留有确认信号的交易对结果，不包括仅有形态的交易对 if symbol_signal_count > 0: results[symbol] = symbol_results total_signals += symbol_signal_count total_patterns += symbol_pattern_count except Exception as e: logger.error(f"扫描交易对 {symbol} 失败: {e}") continue # 更新扫描会话统计 try: self.db_manager.update_scan_session_stats( session_id=session_id, total_scanned=len(symbol_list), total_signals=total_signals ) logger.debug(f"扫描会话统计已更新: {session_id}") except Exception as e: logger.error(f"更新扫描会话统计失败: {e}") # 美化最终扫描结果 logger.info("🎉" + "="*70) logger.info(f"🌍 加密货币市场K线形态扫描完成!") logger.info(f"📊 扫描统计:") logger.info(f" 🔍 总扫描交易对: {len(symbol_list)} 个") logger.info(f" 🎯 确认信号: {total_signals} 个") logger.info(f" 📊 形态形成: {total_patterns} 个") logger.info(f" 📈 涉及交易对: {len(results)} 个") logger.info(f" 💾 扫描会话ID: {session_id}") if results: logger.info(f"📋 详细结果:") signal_count = 0 pattern_count = 0 for symbol, symbol_results in results.items(): for timeframe, result in symbol_results.items(): # 显示确认信号 for signal in result.signals: signal_count += 1 logger.info(f" 🎯 信号#{signal_count}: {symbol} | {timeframe} | {signal['date']} | {signal['price']:.4f}") # 显示形态形成 for pattern in result.patterns: pattern_count += 1 logger.info(f" 📊 形态#{pattern_count}: {symbol} | {timeframe} | {pattern['date']} | {pattern['price']:.4f}") logger.info("🎉" + "="*70) # 发送汇总通知 if results: scan_stats = { 'total_scanned': len(symbol_list), 'data_source': f'Binance-{self.quote_asset}' } try: success = self.notification_manager.send_strategy_summary(results, scan_stats) if success: logger.info("📱 策略信号汇总通知发送完成") else: logger.warning("📱 策略信号汇总通知发送失败") except Exception as e: logger.error(f"发送汇总通知失败: {e}") return results def get_strategy_description(self) -> str: """获取策略描述""" trend_desc = "" if self.strong_trend_enabled: trend_desc = f""" 【多头排列先决条件】✅ - EMA5 > EMA10 > EMA20（三重多头排列） - 在形态识别、突破确认、回踩信号三个阶段都进行检查 - 确保整个过程中始终保持强势趋势 """ else: trend_desc = "\n【多头排列先决条件】❌ 未启用\n" return f"""加密货币K线形态策略 - 强势上涨+多空博弈+突破确认 {trend_desc} 新形态设计：强势上涨 → 多空博弈 → 突破确认 【形态A：双阳+博弈】 1. 强势阶段：2根连续阳线 2. 博弈阶段：2-3根K线，平均实体≤40%（阴线/十字星/小阳线） 3. 突破确认：博弈后1-3根K线内，大阳线（实体≥55%）突破博弈阶段最高价 【形态B：高实体+博弈】 1. 强势阶段：1根高实体阳线（实体≥60%） 2. 博弈阶段：2-3根K线，平均实体≤40%（阴线/十字星/小阳线） 3. 突破确认：博弈后1-3根K线内，大阳线（实体≥55%）突破博弈阶段最高价 【严格约束条件】 - EMA5 > EMA10 > EMA20（三重多头排列）全程检查 - 价格不能跌破EMA10支撑（博弈、突破、回踩全程） - 价格必须创新高后回踩到博弈阶段最高点附近才产生正式信号 - 回踩容忍度：{self.pullback_tolerance:.0%} - 确认窗口：{self.pullback_confirmation_days}天 【加密货币市场特点】 - 计价货币: {self.quote_asset} - 最小24h交易量: ${self.min_volume_usdt:,.0f} - 支持时间周期：{', '.join(self.timeframes)} 【策略优势】 - 真实市场心理：强势→博弈→突破的完整过程 - 多重验证：EMA多头排列 + EMA10支撑 + 创新高回踩确认 - 降低假突破：严格的形态和趋势验证 - 优质入场时机：回踩确认提供更好的风险收益比 """ if __name__ == "__main__": # 测试代码 from ..data.binance_fetcher import BinanceFetcher from ..utils.notification import NotificationManager # 模拟配置 strategy_config = { 'min_entity_ratio': 0.55, 'timeframes': ['daily', '4hour'], 'quote_asset': 'USDT' } notification_config = { 'dingtalk': { 'enabled': False, 'webhook_url': '' } } # 初始化组件 data_fetcher = BinanceFetcher() notification_manager = NotificationManager(notification_config) strategy = CryptoKLinePatternStrategy(data_fetcher, notification_manager, strategy_config) print("加密货币K线形态策略初始化完成") print(strategy.get_strategy_description())