# source: https://raw.githubusercontent.com/manwallet/ft-userdata-llm-share/9eaa080b568605e96e65401c06de125dfedaac27/user_data/strategies/LLMFunctionStrategy.py """ LLM Function Calling Strategy 基于LLM函数调用的智能交易策略 作者: Claude Code 版本: 1.0.0 """ import logging from typing import Dict, Any, Optional, List import pandas as pd from datetime import datetime, timedelta, timezone from freqtrade.strategy import IStrategy, informative, merge_informative_pair import talib.abstract as ta # 导入自定义模块 from llm_modules.utils.config_loader import ConfigLoader from llm_modules.utils.context_builder import ContextBuilder from llm_modules.tools.trading_tools import TradingTools from llm_modules.llm.llm_client import LLMClient from llm_modules.llm.function_executor import FunctionExecutor from llm_modules.experience.trade_logger import TradeLogger from llm_modules.experience.experience_manager import ExperienceManager # 导入新的指标计算器 from llm_modules.indicators.indicator_calculator import IndicatorCalculator # 初始化 logger（必须在使用前定义） logger = logging.getLogger(__name__) # 历史上下文系统 from llm_modules.experience.trade_reviewer import TradeReviewer # 增强模块导入 from llm_modules.utils.position_tracker import PositionTracker from llm_modules.utils.market_comparator import MarketStateComparator from llm_modules.utils.decision_checker import DecisionQualityChecker # 自我学习系统导入 from llm_modules.learning.historical_query import HistoricalQueryEngine from llm_modules.learning.pattern_analyzer import PatternAnalyzer from llm_modules.learning.self_reflection import SelfReflectionEngine from llm_modules.learning.trade_evaluator import TradeEvaluator from llm_modules.learning.reward_learning import RewardLearningSystem class Github_manwallet_ft_userdata_llm_share__LLMFunctionStrategy__20251217_072326(IStrategy): """ LLM函数调用策略 特性: - OpenAI Function Calling 完整交易控制 - 支持期货、多空双向、动态杠杆 - 经验学习和持续优化 """ # 策略基本配置 INTERFACE_VERSION = 3 can_short = True timeframe = '4h' # 4小时K线：专注中长期趋势，大幅过滤日内噪音 startup_candle_count = 300 # === 风控硬开关（按用户要求：全部关闭止损/止盈/锁利润） === # 说明：Freqtrade 仍需要一个 stoploss 数值，因此这里给一个极大容忍值（接近“禁用”）。 stoploss = -0.99 use_custom_stoploss = False # 仓位调整 position_adjustment_enable = True max_entry_position_adjustment = 10 # 订单类型 - 支持限价（LLM可选提供 limit_price；未提供则使用系统默认定价） order_types = { 'entry': 'limit', 'exit': 'limit', } @property def protections(self): """ Freqtrade 2024.10+ 已移除 config.json 中的 protections 配置。 保护机制必须在策略里通过 property 定义。 目标：当模型连续吃亏/频繁进出时，强制暂停交易，避免“老韭菜式反复开单”。 """ # 按需求：关闭所有保护机制（含冷静期等） return [] def informative_pairs(self): """ 定义需要的额外时间框架数据 告诉freqtrade需要下载哪些时间框架的数据 """ pairs = self.dp.current_whitelist() informative_pairs = [] # 4h为主框架，1d确认大趋势，1h用于精确入场 for pair in pairs: informative_pairs.extend([ (pair, '1h'), # 1小时（精确入场时机） (pair, '1d'), # 1天（确认大趋势） ]) return informative_pairs def __init__(self, config: dict) -> None: """初始化策略""" super().__init__(config) logger.info("=" * 60) logger.info("LLM Function Calling Strategy - 正在初始化...") logger.info("=" * 60) try: # 1. 加载配置 self.config_loader = ConfigLoader() self.llm_config = self.config_loader.get_llm_config() self.risk_config = self.config_loader.get_risk_config() self.experience_config = self.config_loader.get_experience_config() self.context_config = self.config_loader.get_context_config() # 自动锁利润配置（独立于LLM） all_cfg = self.config_loader.get_all_config() # 自动止盈/锁利润配置（按需开启） self.auto_profit_config = all_cfg.get("auto_profit_protection", {}) if isinstance(all_cfg, dict) else {} self.stoploss = -0.99 self.use_custom_stoploss = False # 2. 初始化自我学习系统 trade_log_path = self.experience_config.get('trade_log_path', './user_data/logs/trade_experience.jsonl') self.historical_query = HistoricalQueryEngine(trade_log_path) self.pattern_analyzer = PatternAnalyzer(min_sample_size=5) self.self_reflection = SelfReflectionEngine() self.trade_evaluator = TradeEvaluator() # 初始化奖励学习系统 reward_config = { 'storage_path': './user_data/logs/reward_learning.json', 'learning_rate': 0.1, 'discount_factor': 0.95 } self.reward_learning = RewardLearningSystem(reward_config) logger.info("✓ 自我学习系统已初始化 (HistoricalQuery, PatternAnalyzer, SelfReflection, TradeEvaluator, RewardLearning)") # 3. 初始化上下文构建器（注入学习组件） self.context_builder = ContextBuilder( context_config=self.context_config, historical_query_engine=self.historical_query, pattern_analyzer=self.pattern_analyzer, tradable_balance_ratio=config.get('tradable_balance_ratio', 1.0), max_open_trades=config.get('max_open_trades', 1), main_timeframe=self.timeframe ) # 4. 初始化函数执行器 self.function_executor = FunctionExecutor() # 5. 初始化交易工具（简化版 - 只保留交易控制工具） self.trading_tools = TradingTools(self) # 5.1 是否允许LLM控制投入金额（stake_amount） # 默认允许；若关闭，则只让LLM管杠杆/方向，仓位由freqtrade按config stake_amount计算 self.allow_llm_stake_amount = bool(self.risk_config.get("allow_llm_stake_amount", True)) # 6. 初始化LLM客户端 self.llm_client = LLMClient(self.llm_config, self.function_executor) # 8. 注册所有工具函数 self._register_all_tools() # 9. 初始化经验系统（注入反思引擎） self.trade_logger = TradeLogger(self.experience_config) self.experience_manager = ExperienceManager( trade_logger=self.trade_logger, self_reflection_engine=self.self_reflection, trade_evaluator=self.trade_evaluator, reward_learning=self.reward_learning ) # 10. 缓存 self._leverage_cache = {} self._position_adjustment_cache = {} self._stake_request_cache = {} self._entry_price_cache = {} # LLM限价入场缓存 {pair: price} self._exit_price_cache = {} # LLM限价出场缓存 {pair: price} self._model_score_cache = {} # 存储模型对交易的自我评分 self._cancelled_orders = {} # 追踪被取消的订单 {pair: {time, reason, decision}} self._last_entry_decision = {} # 记录上一次的开仓决策 {pair: {action, time}} # 自动止盈/锁利润：记录每笔交易的历史峰值收益（ratio，比如 0.05=+5%） self._profit_peak_by_trade_id: Dict[int, float] = {} # === 模型止盈目标（仅止盈，不启用止损）=== self._tp_by_trade_id: Dict[int, float] = {} self._tp_update_by_pair: Dict[str, float] = {} self._pending_tp_by_pair: Dict[str, float] = {} self._pending_tp_by_temp_key: Dict[str, float] = {} # 10.1 交易频率硬限制（不依赖LLM自觉） limits = config.get("llm_trade_limits", {}) if isinstance(config, dict) else {} # 按需求：关闭冷静期/币对锁定 self._cooldown_minutes_after_exit = 0 self._cooldown_until_by_pair: Dict[str, datetime] = {} logger.info( f"✓ 交易频率限制: 平仓后冷静期 {self._cooldown_minutes_after_exit} 分钟" ) # 11. 初始化增强模块 self.position_tracker = PositionTracker() self.market_comparator = MarketStateComparator() self.decision_checker = DecisionQualityChecker() self.trade_reviewer = TradeReviewer() logger.info("✓ 增强模块已初始化 (PositionTracker, MarketStateComparator, DecisionChecker, TradeReviewer)") # 12. 系统提示词（两套：开仓和持仓） self.entry_system_prompt = self.context_builder.build_entry_system_prompt() self.position_system_prompt = self.context_builder.build_position_system_prompt() logger.info("✓ 已加载两套系统提示词（开仓/持仓管理）") logger.info("✓ 策略初始化完成") logger.info(f" - LLM模型: {self.llm_config.get('model')}") logger.info(f" - 交易工具已注册: {len(self.function_executor.list_functions())} 个") logger.info(f" - 自我学习系统: 已启用（历史查询+模式分析+自我反思）") logger.info("=" * 60) except Exception as e: logger.error(f"策略初始化失败: {e}", exc_info=True) raise def _get_system_prompt(self, has_position: bool) -> str: """ 根据是否有仓位选择系统提示词 Args: has_position: 是否有仓位 Returns: 对应的系统提示词 """ if has_position: return self.position_system_prompt else: return self.entry_system_prompt def _register_all_tools(self): """注册所有工具函数（简化版 - 只注册交易控制工具）""" # 只注册交易工具（市场数据、账户信息已在context中提供） if self.trading_tools: self.function_executor.register_tools_from_instance( self.trading_tools, self.trading_tools.get_tools_schema() ) logger.debug(f"已注册 {len(self.trading_tools.get_tools_schema())} 个交易控制函数") def _collect_multi_timeframe_history(self, pair: str) -> Dict[str, pd.DataFrame]: """根据ContextBuilder配置获取多时间框架K线数据""" if not getattr(self.context_builder, 'include_multi_timeframe_data', True): return {} if not hasattr(self, 'dp') or not self.dp: return {} if not hasattr(self.context_builder, 'get_multi_timeframe_history_config'): return {} tf_config = self.context_builder.get_multi_timeframe_history_config() if not tf_config: return {} history: Dict[str, pd.DataFrame] = {} for timeframe, cfg in tf_config.items(): candles = cfg.get('candles', 0) fields = cfg.get('fields', []) tf_df = self._fetch_timeframe_dataframe(pair, timeframe, candles, fields) if tf_df is not None and not tf_df.empty: history[timeframe] = tf_df return history def _fetch_timeframe_dataframe( self, pair: str, timeframe: str, candles: int, fields: List[str] ) -> Optional[pd.DataFrame]: if candles <= 0: return None try: raw_df = self.dp.get_pair_dataframe(pair=pair, timeframe=timeframe) except Exception as e: logger.warning(f"获取{timeframe}数据失败: {e}") return None if raw_df is None or raw_df.empty: return None padding = max(candles + 100, 200) df = raw_df.tail(padding).copy() self._append_indicator_columns(df, fields) return df.tail(candles) def _append_indicator_columns(self, dataframe: pd.DataFrame, fields: List[str]): """ 在给定dataframe上补齐所需指标列 使用统一的 IndicatorCalculator 简化逻辑 """ if not fields: return # 简单粗暴：直接添加所有指标（IndicatorCalculator会跳过已存在的列） # 这比之前的逐个判断更简洁，且计算成本可忽略 IndicatorCalculator.add_all_indicators(dataframe) def bot_start(self, **kwargs) -> None: """ 策略启动时调用（此时dp和wallets已初始化） """ logger.info("✓ Bot已启动，策略运行中...") logger.info(f"✓ 交易工具: {len(self.function_executor.list_functions())} 个函数可用") def confirm_trade_entry( self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: Optional[str], side: str, **kwargs ) -> bool: """ 开仓确认回调 - 保存市场状态到 MarketComparator 注意：此时 trade 对象还未创建，无法获取 trade_id 暂时先获取技术指标，等 trade 创建后再关联 """ try: # 获取最新的dataframe dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return True latest = dataframe.iloc[-1] # 提取技术指标 indicators = { 'atr': latest.get('atr', 0), 'rsi': latest.get('rsi', 50), 'ema_20': latest.get('ema_20', 0), 'ema_50': latest.get('ema_50', 0), 'macd': latest.get('macd', 0), 'macd_signal': latest.get('macd_signal', 0), 'adx': latest.get('adx', 0) } # 暂存开仓信息（将在下一次 populate 中关联 trade_id） # 使用 pair+rate 作为临时key temp_key = f"{pair}_{rate}" self._pending_entry_states = getattr(self, '_pending_entry_states', {}) self._pending_entry_states[temp_key] = { 'pair': pair, 'rate': rate, 'indicators': indicators, 'entry_tag': entry_tag or '', 'side': side, 'time': current_time, } # 关联本次开仓的止盈目标（如果模型提供） try: pending_tp = self._pending_tp_by_pair.pop(pair, None) if hasattr(self, "_pending_tp_by_pair") else None if pending_tp is not None: self._pending_tp_by_temp_key[temp_key] = float(pending_tp) except Exception: pass logger.debug(f"开仓确认: {pair} @ {rate}, 等待trade_id关联") # 清除上一次的开仓决策记录（订单已成功） if pair in self._last_entry_decision: del self._last_entry_decision[pair] # 清除可能存在的取消订单记录 if pair in self._cancelled_orders: del self._cancelled_orders[pair] except Exception as e: logger.error(f"confirm_trade_entry 失败: {e}") return True # ====== Auto take-profit / profit protection (strategy-controlled) ====== def _get_auto_profit_cfg(self) -> Dict[str, Any]: cfg = getattr(self, "auto_profit_config", {}) or {} if not isinstance(cfg, dict): return {} return cfg def custom_exit( self, pair: str, trade: Any, current_time: datetime, current_rate: float, current_profit: float, **kwargs ) -> Optional[str]: """ 自动止盈/锁利润（不依赖LLM退出理由）： - 目标止盈：profit >= take_profit_pct 时平仓 - 峰值回撤止盈：达到 min_peak_profit 后，从峰值回撤超过 max_drawdown_from_peak 时平仓 """ try: cfg = self._get_auto_profit_cfg() if not bool(cfg.get("enabled", False)): return None tid = getattr(trade, "id", None) if tid is None: return None tid = int(tid) # 0) 模型止盈目标优先（按 trade_id） try: tp_llm = self._tp_by_trade_id.get(tid) if hasattr(self, "_tp_by_trade_id") else None if tp_llm is not None: tp_llm = float(tp_llm) current_profit_val = float(current_profit) if tp_llm > 0: logger.debug(f"🔍 {pair} trade_id={tid} 检查模型止盈: tp_llm={tp_llm:.4f}, current_profit={current_profit_val:.4f}") if current_profit_val >= tp_llm: logger.info(f"✅ {pair} trade_id={tid} 触发模型止盈目标: tp_llm={tp_llm:.4f}, current_profit={current_profit_val:.4f}") return "llm_take_profit_target" except Exception as e: logger.error(f"❌ {pair} trade_id={tid} 检查模型止盈失败: {e}") # 最小持仓时间（可选，避免刚成交就被规则扫掉） min_hold_minutes = float(cfg.get("min_hold_minutes", 0) or 0) if min_hold_minutes > 0: open_dt = getattr(trade, "open_date_utc", None) or getattr(trade, "open_date", None) try: if open_dt and open_dt.tzinfo is None: open_dt = open_dt.replace(tzinfo=timezone.utc) held_seconds = (current_time - open_dt).total_seconds() if open_dt else 0 except Exception: held_seconds = 0 if held_seconds < (min_hold_minutes * 60): return None # 更新峰值收益（ratio） peak = self._profit_peak_by_trade_id.get(tid, None) if peak is None: peak = float(current_profit) else: peak = max(float(peak), float(current_profit)) self._profit_peak_by_trade_id[tid] = peak # 可选：按杠杆缩放阈值（把阈值解释为“标的价格波动比例”） scale_by_leverage = bool(cfg.get("scale_by_leverage", False)) lev = getattr(trade, "leverage", None) or 1.0 try: lev = float(lev) if lev <= 0: lev = 1.0 except Exception: lev = 1.0 take_profit = cfg.get("take_profit_pct", None) min_peak = float(cfg.get("min_peak_profit", 0.03) or 0.0) max_dd = float(cfg.get("max_drawdown_from_peak", 0.015) or 0.0) min_peak = max(0.0, min_peak) max_dd = max(0.0, max_dd) if scale_by_leverage and lev > 1: min_peak = min_peak * lev max_dd = max_dd * lev if take_profit is not None: try: take_profit = float(take_profit) * lev except Exception: take_profit = None # 1) 目标止盈（可选） if take_profit is not None: try: tp = float(take_profit) except Exception: tp = None if tp is not None and tp > 0 and float(current_profit) >= tp: return "auto_take_profit_target" # 2) 峰值回撤止盈 if peak >= min_peak: drawdown = peak - float(current_profit) if drawdown >= max_dd: return "auto_trailing_drawdown_exit" return None except Exception: return None def confirm_trade_exit( self, pair: str, trade: Any, order_type: str, amount: float, rate: float, time_in_force: str, exit_reason: str, current_time: datetime, **kwargs ) -> bool: """ 平仓确认回调 - 生成交易复盘 """ try: # 按需求：冷静期/币对锁定已关闭 # 获取持仓追踪数据 position_metrics = self.position_tracker.get_position_metrics(trade.id) # 获取市场状态变化 dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if not dataframe.empty: latest = dataframe.iloc[-1] current_indicators = { 'atr': latest.get('atr', 0), 'rsi': latest.get('rsi', 50), 'ema_20': latest.get('ema_20', 0), 'ema_50': latest.get('ema_50', 0), 'macd': latest.get('macd', 0), 'adx': latest.get('adx', 0) } market_changes = self.market_comparator.compare_with_entry( trade_id=trade.id, current_price=rate, current_indicators=current_indicators ) else: market_changes = {} # 手动计算盈亏百分比（因为此时 trade.close_profit 可能为 None） if trade.is_short: profit_pct = (trade.open_rate - rate) / trade.open_rate * trade.leverage * 100 else: profit_pct = (rate - trade.open_rate) / trade.open_rate * trade.leverage * 100 # 计算持仓时长（处理时区兼容性） # freqtrade使用naive UTC时间，根据是否有tzinfo选择对应的now if trade.open_date.tzinfo is None: # trade.open_date 是 naive，current_time 也应该是 naive exit_time = current_time.replace(tzinfo=None) if current_time.tzinfo else current_time else: # trade.open_date 是 aware，current_time 也应该是 aware exit_time = current_time if current_time.tzinfo else current_time.replace(tzinfo=timezone.utc) duration_minutes = int((exit_time - trade.open_date).total_seconds() / 60) # 生成交易复盘（如果 TradeReviewer 可用） if self.trade_reviewer: review = self.trade_reviewer.generate_trade_review( pair=pair, side='short' if trade.is_short else 'long', entry_price=trade.open_rate, exit_price=rate, entry_reason=getattr(trade, 'enter_tag', '') or '', exit_reason=exit_reason, profit_pct=profit_pct, duration_minutes=duration_minutes, leverage=trade.leverage, position_metrics=position_metrics, market_changes=market_changes ) # 输出复盘报告 report = self.trade_reviewer.format_review_report(review) logger.info(f"\n{report}") # 记录交易到历史日志（供未来决策参考） if self.experience_manager: # 格式化持仓时间 if duration_minutes < 60: duration_str = f"{duration_minutes}分钟" elif duration_minutes < 1440: duration_str = f"{duration_minutes / 60:.1f}小时" else: duration_str = f"{duration_minutes / 1440:.1f}天" # 记录交易 max_loss_pct = position_metrics.get('max_loss_pct', 0) if position_metrics else 0 max_profit_pct = position_metrics.get('max_profit_pct', 0) if position_metrics else 0 # 获取模型评分 model_score = self._model_score_cache.pop(pair, None) model_score_str = f"模型评分 {model_score:.0f}/100" if model_score else "" market_condition = f"MFE {max_profit_pct:+.2f}% / MAE {max_loss_pct:+.2f}% / 持仓 {duration_str} / {model_score_str}" self.experience_manager.log_trade_completion( trade_id=trade.id, pair=pair, side='short' if trade.is_short else 'long', entry_time=trade.open_date, entry_price=trade.open_rate, entry_reason=getattr(trade, 'enter_tag', '') or '未记录', exit_time=exit_time, exit_price=rate, exit_reason=exit_reason, profit_pct=profit_pct, profit_abs=trade.stake_amount * profit_pct / 100, leverage=trade.leverage, stake_amount=trade.stake_amount, max_drawdown=max_loss_pct, market_condition=market_condition, position_metrics=position_metrics, # 【新增】传递持仓指标 market_changes=market_changes # 【新增】传递市场变化 ) logger.info(f"✓ 交易 {trade.id} 已记录到历史日志") # 清理追踪数据 if trade.id in self.position_tracker.positions: del self.position_tracker.positions[trade.id] if trade.id in self.market_comparator.entry_states: del self.market_comparator.entry_states[trade.id] # 清理自动止盈峰值记录 try: if hasattr(self, "_profit_peak_by_trade_id") and trade.id in self._profit_peak_by_trade_id: del self._profit_peak_by_trade_id[trade.id] except Exception: pass # 清理模型止盈目标 try: if hasattr(self, "_tp_by_trade_id") and trade.id in self._tp_by_trade_id: del self._tp_by_trade_id[trade.id] except Exception: pass except Exception as e: logger.error(f"生成交易复盘失败: {e}", exc_info=True) return True # 多时间框架数据支持 # 多时间框架已禁用 - 只使用30m单时间框架降低成本 # @informative('1h') # def populate_indicators_1h(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: # """1小时数据指标 - 使用统一的 IndicatorCalculator""" # return IndicatorCalculator.add_all_indicators(dataframe) # @informative('4h') # def populate_indicators_4h(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: # """4小时数据指标 - 使用统一的 IndicatorCalculator""" # return IndicatorCalculator.add_all_indicators(dataframe) # @informative('1d') # def populate_indicators_1d(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: # """日线数据指标（注意：8天数据只有8根日线K线，EMA50勉强可用，已删除EMA200）""" # dataframe['ema_20'] = ta.EMA(dataframe, timeperiod=20) # dataframe['ema_50'] = ta.EMA(dataframe, timeperiod=50) # # dataframe['ema_200'] = ta.EMA(dataframe, timeperiod=200) # 需要200天数据，删除 # dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) # macd = ta.MACD(dataframe) # dataframe['macd'] = macd['macd'] # dataframe['macd_signal'] = macd['macdsignal'] # bollinger = ta.BBANDS(dataframe, timeperiod=20) # dataframe['bb_upper'] = bollinger['upperband'] # dataframe['bb_lower'] = bollinger['lowerband'] # dataframe['atr'] = ta.ATR(dataframe, timeperiod=14) # dataframe['adx'] = ta.ADX(dataframe, timeperiod=14) # return dataframe def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: """ 计算技术指标（30分钟基础数据）- 使用统一的 IndicatorCalculator """ return IndicatorCalculator.add_all_indicators(dataframe) def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: """ 开仓信号 - 由LLM决策 """ pair = metadata['pair'] # 默认不开仓 dataframe.loc[:, 'enter_long'] = 0 dataframe.loc[:, 'enter_short'] = 0 dataframe.loc[:, 'enter_tag'] = '' # 只在最新的K线上做决策 if len(dataframe) < self.startup_candle_count: return dataframe # 【调试】记录每次调用的时间点 if len(dataframe) >= 2: latest_time = dataframe.iloc[-1]['date'] if 'date' in dataframe.iloc[-1] else "unknown" logger.debug(f"[populate_entry_trend] {pair} | Latest time: {latest_time} | DataFrame length: {len(dataframe)}") try: # 按需求：冷静期/币对锁定已关闭，不再跳过开仓决策 # 获取当前所有持仓（用于传给context_builder） from freqtrade.persistence import Trade try: current_trades = Trade.get_open_trades() except (AttributeError, RuntimeError): # 测试模式下可能没有数据库session current_trades = [] # 检查当前是否有持仓 pair_trades = [t for t in current_trades if t.pair == pair] # 冷静期由Freqtrade的lock_pair原生功能处理，无需手动检查 # 检测上一次决策是否执行成功（订单是否被取消） last_decision = self._cancelled_orders.get(pair, None) # 检查上一次开仓决策是否成功执行 last_entry = self._last_entry_decision.get(pair) if last_entry and not last_decision: # 有上一次决策记录，但没有取消订单记录 last_action = last_entry.get('action') last_time = last_entry.get('time') # 如果上一次决策是开仓，但现在没有持仓 if last_action in ['enter_long', 'enter_short'] and len(pair_trades) == 0: # 从配置读取超时时间，默认5分钟 timeout_minutes = self.config.get('unfilledtimeout', {}).get('entry', 5) # 检查是否已经过了足够的时间（超时时间 + 1分钟缓冲） now = datetime.now(timezone.utc) check_threshold = (timeout_minutes + 1) * 60 # 转换成秒，多等1分钟 if last_time and (now - last_time).total_seconds() > check_threshold: # 判断订单可能被取消了 self._cancelled_orders[pair] = { 'time': now, 'type': 'entry', 'reason': '开仓订单可能超时未成交', 'side': 'long' if last_action == 'enter_long' else 'short', 'price': 0, # 没有订单价格信息 } last_decision = self._cancelled_orders[pair] logger.warning(f"⚠️ 检测到订单可能被取消: {pair} {last_decision['side']}") # 如果有记录的取消订单，检查是否应该清除 if last_decision: cancel_time = last_decision.get('time') if cancel_time: now = datetime.now(timezone.utc) # 如果取消时间超过10分钟，清除记录 if (now - cancel_time).total_seconds() > 600: del self._cancelled_orders[pair] # 同时清除上一次的开仓决策记录 if pair in self._last_entry_decision: del self._last_entry_decision[pair] last_decision = None # 构建完整的市场上下文（包含技术指标、账户信息、持仓情况） # 获取exchange对象用于市场情绪数据 exchange = None if hasattr(self, 'dp') and self.dp: if hasattr(self.dp, '_exchange'): exchange = self.dp._exchange elif hasattr(self.dp, 'exchange'): exchange = self.dp.exchange multi_tf_history = ( self._collect_multi_timeframe_history(pair) if getattr(self.context_builder, 'include_multi_timeframe_data', True) else {} ) market_context = self.context_builder.build_market_context( dataframe=dataframe, metadata=metadata, wallets=self.wallets, current_trades=current_trades, exchange=exchange, position_tracker=self.position_tracker, market_comparator=self.market_comparator, multi_timeframe_data=multi_tf_history, cancelled_order=last_decision # 传递取消订单信息 ) # 检查当前交易对是否有持仓（前面已定义，这里注释掉避免重复） # pair_trades = [t for t in current_trades if t.pair == pair] has_position = len(pair_trades) > 0 # 根据是否有持仓选择action_type和系统提示词 action_type = "hold" if has_position else "entry" # 构建决策请求 decision_request = self.context_builder.build_decision_request( action_type=action_type, market_context=market_context, position_context="" # 已包含在market_context中 ) # 调用LLM决策（根据是否有持仓选择提示词） messages = [ {"role": "system", "content": self._get_system_prompt(has_position=has_position)}, {"role": "user", "content": decision_request} ] response = self.llm_client.call_with_functions( messages=messages, max_iterations=10 # 限制迭代次数，防止无限循环 ) # 处理响应 if response.get("success"): function_calls = response.get("function_calls", []) llm_message = response.get("message", "") # 检查是否有交易信号 signal = self.trading_tools.get_signal(pair) # 提取置信度用于记录决策 confidence = signal.get("confidence_score", 50) / 100 if signal else 0.5 # 记录决策 self.experience_manager.log_decision_with_context( pair=pair, action="entry", decision=llm_message, reasoning=str(function_calls), confidence=confidence, market_context={"indicators": market_context}, function_calls=function_calls ) if signal: action = signal.get("action") reason = signal.get("reason", llm_message) # pair_trades已在前面定义，这里不需要重复定义 # 提取新增参数 confidence_score = signal.get("confidence_score", 0) key_support = signal.get("key_support", 0) key_resistance = signal.get("key_resistance", 0) rsi_value = signal.get("rsi_value", 0) trend_strength = signal.get("trend_strength", "未知") stake_amount = signal.get("stake_amount") take_profit_pct = signal.get("take_profit_pct") if self.allow_llm_stake_amount and stake_amount and stake_amount > 0: self._stake_request_cache[pair] = stake_amount # LLM限价委托（入场/开仓） limit_price = signal.get("limit_price") if limit_price and float(limit_price) > 0: self._entry_price_cache[pair] = float(limit_price) # 缓存模型的止盈目标（等订单成交后关联trade_id） try: if take_profit_pct is not None: tp = abs(float(take_profit_pct)) if 0 < tp < 5: self._pending_tp_by_pair[pair] = tp except Exception: pass if action == "enter_long": dataframe.loc[dataframe.index[-1], 'enter_long'] = 1 dataframe.loc[dataframe.index[-1], 'enter_tag'] = reason logger.info(f"📈 {pair} | 做多 | 置信度: {confidence_score}") logger.info(f" 支撑: {key_support} | 阻力: {key_resistance}") logger.info(f" RSI: {rsi_value} | 趋势强度: {trend_strength}") logger.info(f" 理由: {reason}") # 记录开仓决策 self._last_entry_decision[pair] = { 'action': action, 'time': datetime.now(timezone.utc) } elif action == "enter_short": dataframe.loc[dataframe.index[-1], 'enter_short'] = 1 dataframe.loc[dataframe.index[-1], 'enter_tag'] = reason logger.info(f"📉 {pair} | 做空 | 置信度: {confidence_score}") logger.info(f" 支撑: {key_support} | 阻力: {key_resistance}") logger.info(f" RSI: {rsi_value} | 趋势强度: {trend_strength}") logger.info(f" 理由: {reason}") # 记录开仓决策 self._last_entry_decision[pair] = { 'action': action, 'time': datetime.now(timezone.utc) } elif action == "hold": logger.info(f"🔒 {pair} | 保持持仓 | 置信度: {confidence_score} | RSI: {rsi_value}") logger.info(f" 理由: {reason}") # 记录hold决策，累加hold次数 current_price = dataframe.iloc[-1]['close'] for trade in pair_trades: try: self.position_tracker.update_position( trade_id=trade.id, pair=pair, current_price=current_price, open_price=trade.open_rate, is_short=trade.is_short, leverage=trade.leverage, decision_type='hold', decision_reason=reason ) except Exception as e: logger.debug(f"更新hold计数失败: {e}") elif action == "wait": logger.info(f"⏸️ {pair} | 空仓等待 | 置信度: {confidence_score} | RSI: {rsi_value}") logger.info(f" 理由: {reason}") else: # 没有交易信号 = 观望，显示LLM的完整分析 logger.info(f"⏸️ {pair} | 未提供明确信号\n{llm_message}") # 清空信号缓存 self.trading_tools.clear_signals() # 若模型在“持仓管理/开仓流程”里调用了 set_risk_targets（止盈更新），映射到当前 trade_id 生效 try: if has_position and pair_trades and getattr(self, "_tp_update_by_pair", None): tp_upd = self._tp_update_by_pair.pop(pair, None) if tp_upd is not None: for t in pair_trades: tid = getattr(t, "id", None) if tid is None: continue self._tp_by_trade_id[int(tid)] = float(tp_upd) logger.info(f"✅ {pair} 已应用模型止盈更新: take_profit_pct={tp_upd}") except Exception: pass except Exception as e: logger.error(f"开仓决策失败 {pair}: {e}") return dataframe def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: """ 平仓信号 - 由LLM决策 """ pair = metadata['pair'] # 默认不平仓 dataframe.loc[:, 'exit_long'] = 0 dataframe.loc[:, 'exit_short'] = 0 dataframe.loc[:, 'exit_tag'] = '' # 只在最新的K线上做决策 if len(dataframe) < self.startup_candle_count: return dataframe try: # 获取当前所有持仓 from freqtrade.persistence import Trade current_trades = Trade.get_open_trades() # 当前交易对持仓（必须先定义，后续会多处使用） pair_trades = [t for t in current_trades if t.pair == pair] # 检查当前交易对是否有持仓 pair_has_position = len(pair_trades) > 0 if not pair_has_position: return dataframe # 无持仓，不需要决策 # 构建完整的市场上下文（包含技术指标、账户信息、持仓情况） # 获取exchange对象用于市场情绪数据 exchange = None if hasattr(self, 'dp') and self.dp: if hasattr(self.dp, '_exchange'): exchange = self.dp._exchange elif hasattr(self.dp, 'exchange'): exchange = self.dp.exchange multi_tf_history = ( self._collect_multi_timeframe_history(pair) if getattr(self.context_builder, 'include_multi_timeframe_data', True) else {} ) market_context = self.context_builder.build_market_context( dataframe=dataframe, metadata=metadata, wallets=self.wallets, current_trades=current_trades, exchange=exchange, position_tracker=self.position_tracker, market_comparator=self.market_comparator, multi_timeframe_data=multi_tf_history, cancelled_order=None # exit不需要显示取消订单信息 ) # 按需求：止损/止盈/锁利润功能已关闭，不再向上下文注入风控目标 # 更新 PositionTracker 和关联 MarketComparator # pair_trades 已在上方计算 # 检查dataframe是否为空 if dataframe.empty: logger.warning(f"{pair} dataframe为空，跳过持仓追踪更新") return dataframe current_price = dataframe.iloc[-1]['close'] for trade in pair_trades: try: # 更新持仓追踪数据 self.position_tracker.update_position( trade_id=trade.id, pair=pair, current_price=current_price, open_price=trade.open_rate, is_short=trade.is_short, leverage=trade.leverage, decision_type='check', # 正在检查是否平仓 decision_reason='' # 稍后在决策后更新 ) # 关联待定的开仓状态（如果存在） temp_key = f"{pair}_{trade.open_rate}" if hasattr(self, '_pending_entry_states') and temp_key in self._pending_entry_states: pending = self._pending_entry_states[temp_key] # 保存到 MarketComparator self.market_comparator.save_entry_state( trade_id=trade.id, pair=pair, price=trade.open_rate, indicators=pending['indicators'], entry_reason=pending['entry_tag'], trend_alignment='', market_sentiment='' ) # 清除待定状态 del self._pending_entry_states[temp_key] logger.debug(f"已关联开仓状态到 trade_id={trade.id}") # 同时把“模型止盈目标”关联到 trade_id try: if hasattr(self, "_pending_tp_by_temp_key") and temp_key in self._pending_tp_by_temp_key: tp_value = float(self._pending_tp_by_temp_key.pop(temp_key)) self._tp_by_trade_id[int(trade.id)] = tp_value logger.info(f"✅ {pair} trade_id={trade.id} 已关联模型止盈目标: take_profit_pct={tp_value}") else: logger.debug(f"⚠️ {pair} trade_id={trade.id} temp_key={temp_key} 未找到待关联的止盈目标") except Exception as e: logger.error(f"❌ {pair} trade_id={trade.id} 关联模型止盈目标失败: {e}") except Exception as e: logger.debug(f"更新持仓追踪失败: {e}") # 构建决策请求 decision_request = self.context_builder.build_decision_request( action_type="exit", market_context=market_context, position_context="" # 已包含在market_context中 ) # 调用LLM决策（使用持仓管理提示词） messages = [ {"role": "system", "content": self._get_system_prompt(has_position=True)}, {"role": "user", "content": decision_request} ] response = self.llm_client.call_with_functions( messages=messages, max_iterations=10 # 限制迭代次数，防止无限循环 ) if response.get("success"): llm_message = response.get("message", "") signal = self.trading_tools.get_signal(pair) if signal and signal.get("action") == "exit": reason = signal.get("reason", llm_message) # 提取新增参数 confidence_score = signal.get("confidence_score", 0) rsi_value = signal.get("rsi_value", 0) trade_score = signal.get("trade_score", None) # 模型自我评分 # 缓存模型评分（在 confirm_trade_exit 中使用） if trade_score is not None: self._model_score_cache[pair] = trade_score # LLM限价委托（出场/平仓） limit_price = signal.get("limit_price") if limit_price and float(limit_price) > 0: self._exit_price_cache[pair] = float(limit_price) dataframe.loc[dataframe.index[-1], 'exit_long'] = 1 dataframe.loc[dataframe.index[-1], 'exit_short'] = 1 dataframe.loc[dataframe.index[-1], 'exit_tag'] = reason logger.info(f"🔚 {pair} | 平仓 | 置信度: {confidence_score} | 自我评分: {trade_score}/100") logger.info(f" RSI: {rsi_value}") logger.info(f" 理由: {reason}") # 【立即生成交易复盘】- 在平仓信号发出时 if pair_trades and self.trade_reviewer: try: trade = pair_trades[0] # 获取持仓追踪数据 position_metrics = self.position_tracker.get_position_metrics(trade.id) # 获取市场状态变化 latest = dataframe.iloc[-1] current_indicators = { 'atr': latest.get('atr', 0), 'rsi': latest.get('rsi', 50), 'ema_20': latest.get('ema_20', 0), 'ema_50': latest.get('ema_50', 0), 'macd': latest.get('macd', 0), 'adx': latest.get('adx', 0) } market_changes = self.market_comparator.compare_with_entry( trade_id=trade.id, current_price=current_price, current_indicators=current_indicators ) # 计算持仓时长（分钟） now = datetime.utcnow() if trade.open_date.tzinfo is None else datetime.now(timezone.utc) duration_minutes = int((now - trade.open_date).total_seconds() / 60) # 计算预期平仓盈亏（使用当前市价） exit_price = current_price if trade.is_short: profit_pct = (trade.open_rate - exit_price) / trade.open_rate * trade.leverage * 100 else: profit_pct = (exit_price - trade.open_rate) / trade.open_rate * trade.leverage * 100 # 生成交易复盘 review = self.trade_reviewer.generate_trade_review( pair=pair, side='short' if trade.is_short else 'long', entry_price=trade.open_rate, exit_price=exit_price, entry_reason=getattr(trade, 'enter_tag', '') or '', exit_reason=reason, profit_pct=profit_pct, duration_minutes=duration_minutes, leverage=trade.leverage, position_metrics=position_metrics, market_changes=market_changes ) # 输出复盘报告 report = self.trade_reviewer.format_review_report(review) logger.info(f"\n{report}") except Exception as e: logger.error(f"生成交易复盘失败: {e}", exc_info=True) else: logger.info(f"💎 {pair} | 继续持有\n{llm_message}") # 记录决策到 DecisionChecker（用于检测重复模式和盈利回撤） if signal: action = signal.get("action") reason = signal.get("reason", llm_message) # 计算当前盈亏（用于决策质量分析） if pair_trades: trade = pair_trades[0] if trade.is_short: profit_pct = (trade.open_rate - current_price) / trade.open_rate * trade.leverage * 100 else: profit_pct = (current_price - trade.open_rate) / trade.open_rate * trade.leverage * 100 # 记录决策 decision_type = 'exit' if action == 'exit' else 'hold' try: quality_check = self.decision_checker.record_decision( pair=pair, decision_type=decision_type, reason=reason, profit_pct=profit_pct ) # 如果有警告，记录到日志（不阻止交易） if quality_check.get('warnings'): for warning in quality_check['warnings']: if warning.get('level') == 'high': logger.warning(f"[决策质量警告] {warning.get('message')}") if warning.get('suggestion'): logger.warning(f" 建议: {warning.get('suggestion')}") except Exception as e: logger.debug(f"决策质量检查失败: {e}") self.trading_tools.clear_signals() # 应用 set_risk_targets 的止盈更新（如果模型在exit阶段调用） try: tp_upd = self._tp_update_by_pair.pop(pair, None) if hasattr(self, "_tp_update_by_pair") else None if tp_upd is not None and pair_trades: for t in pair_trades: tid = getattr(t, "id", None) if tid is None: continue self._tp_by_trade_id[int(tid)] = float(tp_upd) logger.info(f"✅ {pair} 已应用模型止盈更新(exit阶段): take_profit_pct={tp_upd}") except Exception: pass except Exception as e: logger.error(f"平仓决策失败 {pair}: {e}") return dataframe def leverage( self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: Optional[str], side: str, **kwargs ) -> float: """ 动态杠杆 - 由LLM决定或使用缓存值 """ # 检查缓存 if pair in self._leverage_cache: leverage_value = self._leverage_cache[pair] del self._leverage_cache[pair] # 使用后清除 return min(leverage_value, max_leverage) # 默认杠杆 default_leverage = self.risk_config.get("default_leverage", 10) return min(default_leverage, max_leverage) def custom_stake_amount( self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: Optional[float], max_stake: float, leverage: float, entry_tag: Optional[str], side: str, **kwargs ) -> float: """ 动态仓位大小 - 可由LLM调整 """ # 关闭LLM仓位控制：永远使用freqtrade计算出的 proposed_stake if not getattr(self, "allow_llm_stake_amount", True): return proposed_stake stake_request = None if hasattr(self, '_stake_request_cache'): stake_request = self._stake_request_cache.pop(pair, None) if stake_request is None: return proposed_stake desired = stake_request # 只检查最小值，不限制最大值（由tradable_balance_ratio自然限制） if min_stake and desired < min_stake: logger.warning(f"{pair} 指定投入 {stake_request:.2f} USDT 低于最小要求 {min_stake:.2f}，已调整为最小值") desired = min_stake logger.info(f"{pair} 使用LLM指定仓位: {desired:.2f} USDT (请求 {stake_request:.2f})") return desired def adjust_trade_position( self, trade: Any, current_time: datetime, current_rate: float, current_profit: float, min_stake: Optional[float], max_stake: float, current_entry_rate: float, current_exit_rate: float, current_entry_profit: float, current_exit_profit: float, **kwargs ) -> Optional[float]: """ 仓位调整 - 允许LLM加仓或减仓 重要：此方法在每根K线都会被调用，确保回测时逐根K线处理（而非批量处理） 这对于LLM策略至关重要，因为每次决策都依赖最新的上下文。 Args: trade: 当前交易对象 current_rate: 当前价格 其他参数... Returns: Optional[float]: 要增加的stake金额（正数=加仓，负数=减仓），None=不调整 """ pair = trade.pair # === 强制逐根K线处理逻辑（DCA风格） === # 即使不调整仓位，也要执行检查，确保Freqtrade逐根K线调用 # 这避免了回测时的批量处理，保证了LLM决策的真实性 # 1. 记录调用次数（用于调试和验证逐根K线处理） if not hasattr(self, '_adjust_call_count'): self._adjust_call_count = {} self._adjust_call_count[pair] = self._adjust_call_count.get(pair, 0) + 1 # 2. 计算持仓时间（即使不用，也确保方法执行了有意义的计算） if trade.open_date.tzinfo is None: now = current_time.replace(tzinfo=None) if current_time.tzinfo else current_time else: now = current_time if current_time.tzinfo else current_time.replace(tzinfo=timezone.utc) holding_minutes = (now - trade.open_date).total_seconds() / 60 # 3. 每10次调用输出一次日志（证明确实在逐根K线处理） if self._adjust_call_count[pair] % 10 == 0: logger.debug( f"[DCA检查] {pair} | 调用#{self._adjust_call_count[pair]} | " f"持仓{holding_minutes/60:.1f}h | 盈亏{current_profit:.2%}" ) # 检查LLM是否有仓位调整决策 if pair in self._position_adjustment_cache: adjustment_info = self._position_adjustment_cache[pair] del self._position_adjustment_cache[pair] adjustment_pct = adjustment_info.get("adjustment_pct", 0) reason = adjustment_info.get("reason", "") limit_price = adjustment_info.get("limit_price") # 计算调整金额 current_stake = trade.stake_amount adjustment_stake = current_stake * (adjustment_pct / 100) if adjustment_pct > 0: # 加仓 adjustment_stake = min(adjustment_stake, max_stake) if min_stake and adjustment_stake < min_stake: logger.warning(f"{pair} 加仓金额 {adjustment_stake} 低于最小stake {min_stake}") return None # 加仓属于额外入场订单，复用 custom_entry_price if limit_price and float(limit_price) > 0: self._entry_price_cache[pair] = float(limit_price) logger.info(f"{pair} 加仓 {adjustment_pct:.1f}% = {adjustment_stake:.2f} USDT | {reason}") return adjustment_stake elif adjustment_pct < 0: # 减仓 max_reduce = -current_stake * 0.99 # 最多减99%（保留一点避免完全平仓） adjustment_stake = max(adjustment_stake, max_reduce) logger.info(f"{pair} 减仓 {abs(adjustment_pct):.1f}% = {adjustment_stake:.2f} USDT | {reason}") return adjustment_stake # 无调整 - 但已执行完整的检查逻辑 return None # ====== Limit price hooks ====== # config.json 已把 entry/exit 改为 limit。 # 如果 LLM 在函数里提供 limit_price，则这里覆盖 proposed_rate； # 否则使用系统默认 proposed_rate（受 entry_pricing/exit_pricing 影响）。 def custom_entry_price( self, pair: str, current_time: datetime, proposed_rate: float, entry_tag: Optional[str] = None, side: Optional[str] = None, **kwargs ) -> float: price = None if hasattr(self, "_entry_price_cache"): price = self._entry_price_cache.pop(pair, None) return float(price) if price else proposed_rate def custom_exit_price( self, pair: str, trade: Any, current_time: datetime, proposed_rate: float, exit_tag: Optional[str] = None, **kwargs ) -> float: price = None if hasattr(self, "_exit_price_cache"): price = self._exit_price_cache.pop(pair, None) return float(price) if price else proposed_rate