"""基于研究结论的真实交易策略。 策略核心： 1. 入场：综合评分 >= MIN_ENTRY_SCORE (0.60)，结合 RSI/MACD/成交量趋势确认 2. 止损追踪：盈利超过 TRAILING_STOP_TRIGGER (2%) 后启用移动止损 3. 退出：盈利达到 PROFIT_EXIT_RATIO (5%) 或持仓超过 MAX_HOLDING_HOURS (48h) 参数配置： - QUANT_STRATEGY_MIN_ENTRY_SCORE = 0.60 - QUANT_STRATEGY_TRAILING_STOP_TRIGGER = 0.02 - QUANT_STRATEGY_TRAILING_STOP_DISTANCE = 0.01 - QUANT_STRATEGY_PROFIT_EXIT_RATIO = 0.05 - QUANT_STRATEGY_MAX_HOLDING_HOURS = 48 """ from __future__ import annotations from datetime import datetime, timezone, timedelta from decimal import Decimal from logging import getLogger from typing import Optional from pandas import DataFrame import pandas as pd import numpy as np try: import requests except ImportError: requests = None # pragma: no cover - fallback when requests not installed from freqtrade.strategy import IStrategy, merge_informative_pair # Fix import path for Docker container - user_data is mounted at /freqtrade/user_data import sys sys.path.insert(0, '/freqtrade/user_data') from config_helper import get_config # 交易告警集成 try: TRADE_ALERT_ENABLED = get_config("QUANT_TRADE_ALERT_ENABLED", default="true", as_type="str").lower() == "true" except Exception: TRADE_ALERT_ENABLED = False logger = getLogger(__name__) # 策略配置参数（使用统一配置接口） # 回测优化：降低入场阈值，更容易触发交易信号 MIN_ENTRY_SCORE = float(get_config("QUANT_STRATEGY_MIN_ENTRY_SCORE", default=Decimal("0.35"), as_type="decimal")) TRAILING_STOP_TRIGGER = float(get_config("QUANT_STRATEGY_TRAILING_STOP_TRIGGER", default=Decimal("0.02"), as_type="decimal")) TRAILING_STOP_DISTANCE = float(get_config("QUANT_STRATEGY_TRAILING_STOP_DISTANCE", default=Decimal("0.01"), as_type="decimal")) PROFIT_EXIT_RATIO = float(get_config("QUANT_STRATEGY_PROFIT_EXIT_RATIO", default=Decimal("0.05"), as_type="decimal")) MAX_HOLDING_HOURS = get_config("QUANT_STRATEGY_MAX_HOLDING_HOURS", default=48, as_type="int") # 技术指标参数 RSI_PERIOD = get_config("QUANT_RSI_PERIOD", default=14, as_type="int") RSI_OVERBUY_THRESHOLD = float(get_config("QUANT_RSI_OVERBUY_THRESHOLD", default=Decimal("70"), as_type="decimal")) RSI_OVERSELL_THRESHOLD = float(get_config("QUANT_RSI_OVERSELL_THRESHOLD", default=Decimal("30"), as_type="decimal")) MACD_FAST_PERIOD = get_config("QUANT_MACD_FAST_PERIOD", default=12, as_type="int") MACD_SLOW_PERIOD = get_config("QUANT_MACD_SLOW_PERIOD", default=26, as_type="int") MACD_SIGNAL_PERIOD = get_config("QUANT_MACD_SIGNAL_PERIOD", default=9, as_type="int") EMA_FAST_PERIOD = get_config("QUANT_EMA_FAST_PERIOD", default=20, as_type="int") EMA_SLOW_PERIOD = get_config("QUANT_EMA_SLOW_PERIOD", default=55, as_type="int") VOLUME_TREND_PERIOD = get_config("QUANT_VOLUME_TREND_PERIOD", default=20, as_type="int") # 权重配置（用于本地计算的综合评分） TREND_CONFIRM_WEIGHT = 0.20 # 趋势确认权重 INDICATOR_WEIGHT = 0.20 # 技术指标权重 # 研究评分 API 配置 QUANT_API_BASE_URL = get_config("QUANT_API_BASE_URL", default="http://127.0.0.1:9011", as_type="str") RESEARCH_API_TIMEOUT = get_config("QUANT_RESEARCH_API_TIMEOUT", default=5, as_type="int") # 综合评分权重：研究评分权重 0.4，本地评分权重 0.6（当 API 可用时） RESEARCH_SCORE_FALLBACK_WEIGHT = 0.40 class SampleStrategy(IStrategy): """基于研究结论的真实交易策略。""" INTERFACE_VERSION = 3 can_short = False minimal_roi = {"0": PROFIT_EXIT_RATIO} # 使用配置的盈利目标 stoploss = -0.10 # 基础止损 10% timeframe = "1h" startup_candle_count = 100 # 需要足够数据计算 MACD process_only_new_candles = True use_exit_signal = True use_custom_stoploss = True # 持仓追踪（用于移动止损和时间限制） _position_tracker: dict[str, dict] = {} # 研究评分缓存（避免重复调用 API） _research_score_cache: dict[str, dict] = {} _cache_timestamp: Optional[datetime] = None _cache_ttl_seconds: int = 300 # 5分钟缓存 # 交易告警追踪 _trade_alert_tracker: dict[str, dict] = {} _daily_pnl_tracker: dict[str, float] = {} _consecutive_losses: dict[str, int] = {} _last_alert_reset_date: Optional[str] = None def _check_trade_alerts(self, pair: str, profit_ratio: float, entry_price: float, exit_price: float) -> None: """检查交易告警条件并发送告警。 Args: pair: 币种对 profit_ratio: 盈亏比例 entry_price: 入场价格 exit_price: 出场价格 """ if not TRADE_ALERT_ENABLED: return try: import requests # 告警配置阈值 large_loss_threshold = 0.05 # 5% consecutive_loss_threshold = 3 daily_loss_threshold = 0.10 # 10% symbol = pair.replace("/", "").replace(":", "") is_loss = profit_ratio < 0 # 检查单笔大额亏损 if is_loss and abs(profit_ratio) > large_loss_threshold: self._send_trade_alert( alert_type="large_loss", symbol=symbol, message=f"单笔亏损 {abs(profit_ratio):.2%} 超过阈值 {large_loss_threshold:.2%}", details={ "loss_percent": f"{abs(profit_ratio):.2%}", "entry_price": str(entry_price), "exit_price": str(exit_price), } ) # 检查连续亏损 if is_loss: self._consecutive_losses[symbol] = self._consecutive_losses.get(symbol, 0) + 1 if self._consecutive_losses[symbol] >= consecutive_loss_threshold: self._send_trade_alert( alert_type="consecutive_loss", symbol=symbol, message=f"连续亏损 {self._consecutive_losses[symbol]} 笔", details={ "consecutive_count": str(self._consecutive_losses[symbol]), } ) # 重置计数 self._consecutive_losses[symbol] = 0 else: self._consecutive_losses[symbol] = 0 # 更新日盈亏追踪 today = datetime.now(timezone.utc).strftime("%Y-%m-%d") if self._last_alert_reset_date != today: self._daily_pnl_tracker.clear() self._last_alert_reset_date = today self._daily_pnl_tracker[symbol] = self._daily_pnl_tracker.get(symbol, 0.0) + profit_ratio # 检查日亏损限制 daily_pnl = self._daily_pnl_tracker.get(symbol, 0.0) if daily_pnl < -daily_loss_threshold: self._send_trade_alert( alert_type="daily_loss_limit", symbol=symbol, message=f"日亏损 {abs(daily_pnl):.2%} 超过阈值 {daily_loss_threshold:.2%}", details={ "daily_loss": f"{abs(daily_pnl):.2%}", "date": today, } ) except Exception as e: logger.warning("交易告警检查失败: %s", e) def _send_trade_alert(self, alert_type: str, symbol: str, message: str, details: dict) -> None: """发送交易告警到飞书。 Args: alert_type: 告警类型 symbol: 币种符号 message: 告警消息 details: 告警详情 """ try: import requests # 飞书告警 API alert_url = f"{QUANT_API_BASE_URL}/api/v1/alert/trade" payload = { "alert_type": alert_type, "symbol": symbol, "message": message, "details": details, "timestamp": datetime.now(timezone.utc).isoformat(), } response = requests.post(alert_url, json=payload, timeout=5) if response.status_code == 200: logger.info("交易告警发送成功: %s - %s", alert_type, symbol) else: logger.warning("交易告警发送失败: HTTP %d", response.status_code) except requests.exceptions.Timeout: logger.warning("交易告警发送超时") except Exception as e: logger.warning("交易告警发送异常: %s", e) def confirm_trade_exit( self, pair: str, order: dict, trade: object, current_time: datetime, proposed_rate: float, current_profit: float, exit_reason: str, **kwargs, ) -> bool: """交易退出确认回调 - 检查告警条件。 Args: pair: 币种对 order: 订单信息 trade: 交易对象 current_time: 当前时间 proposed_rate: 提议的退出价格 current_profit: 当前盈利比例 exit_reason: 退出原因 Returns: 是否确认退出（始终返回 True） """ # 检查交易告警 entry_price = float(trade.open_rate) if hasattr(trade, 'open_rate') else proposed_rate self._check_trade_alerts( pair=pair, profit_ratio=current_profit, entry_price=entry_price, exit_price=proposed_rate, ) logger.info( "交易退出确认: %s, 盈亏=%.2f%%, 原因=%s", pair, current_profit * 100, exit_reason, ) return True def bot_loop_start(self, current_time: datetime, **kwargs) -> None: """Bot循环开始回调 - 每日开始时重置统计数据。 Args: current_time: 当前时间 """ today = current_time.strftime("%Y-%m-%d") # 检查是否需要重置日统计 if self._last_alert_reset_date != today: self._daily_pnl_tracker.clear() self._last_alert_reset_date = today logger.info("交易日统计数据已重置: %s", today) def _fetch_research_score(self, symbol: str) -> Optional[float]: """从研究 API 获取币种评分。 Args: symbol: 币种符号（如 ETHUSDT） Returns: 评分（0.0-1.0），API 不可用时返回 None """ if requests is None: logger.debug("requests 库未安装，跳过研究评分获取") return None # 检查缓存是否有效 now = datetime.now(timezone.utc) if ( self._cache_timestamp and (now - self._cache_timestamp).total_seconds() < self._cache_ttl_seconds and symbol in self._research_score_cache ): return self._research_score_cache[symbol].get("score") try: api_url = f"{QUANT_API_BASE_URL}/api/v1/evaluation/workspace" logger.debug("调用研究评分 API: %s", api_url) response = requests.get(api_url, timeout=RESEARCH_API_TIMEOUT) response.raise_for_status() data = response.json() if not data or data.get("error"): logger.warning("研究 API 返回错误: %s", data.get("error")) return None workspace = data.get("data", {}).get("item", {}) if workspace.get("status") != "ready": logger.debug("研究评估状态非 ready: %s", workspace.get("status")) return None # 更新缓存 self._cache_timestamp = now leaderboard = workspace.get("leaderboard", []) for entry in leaderboard: entry_symbol = entry.get("symbol", "") # 处理可能的符号格式差异（ETHUSDT vs ETH/USDT） normalized_symbol = entry_symbol.replace("/", "") self._research_score_cache[normalized_symbol] = { "score": float(entry.get("score", 0)), "next_action": entry.get("next_action", ""), "recommendation_reason": entry.get("recommendation_reason", ""), } # 返回请求的币种评分 normalized_request = symbol.replace("/", "") if normalized_request in self._research_score_cache: cached = self._research_score_cache[normalized_request] logger.info( "研究评分获取成功: %s, score=%.4f, action=%s", symbol, cached.get("score", 0), cached.get("next_action", ""), ) return cached.get("score") logger.debug("币种 %s 未在研究排行榜中", symbol) return None except requests.exceptions.Timeout: logger.warning("研究 API 请求超时 (%d秒)", RESEARCH_API_TIMEOUT) return None except requests.exceptions.RequestException as e: logger.warning("研究 API 请求失败: %s", str(e)) return None except (KeyError, ValueError, TypeError) as e: logger.warning("研究 API 响应解析失败: %s", str(e)) return None def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """计算技术指标：RSI、MACD、EMA、成交量趋势。""" if dataframe is None or len(dataframe) < 30: logger.warning("数据不足，无法计算指标") return dataframe # 计算 EMA 快线（20周期） dataframe["ema_fast"] = self._calculate_ema( dataframe["close"], EMA_FAST_PERIOD ) # 计算 EMA 慢线（55周期） dataframe["ema_slow"] = self._calculate_ema( dataframe["close"], EMA_SLOW_PERIOD ) # 计算 RSI dataframe["rsi"] = self._calculate_rsi(dataframe["close"], RSI_PERIOD) # 计算 MACD macd_result = self._calculate_macd( dataframe["close"], MACD_FAST_PERIOD, MACD_SLOW_PERIOD, MACD_SIGNAL_PERIOD, ) dataframe["macd_line"] = macd_result["macd_line"] dataframe["macd_signal"] = macd_result["signal_line"] dataframe["macd_histogram"] = macd_result["histogram"] # 计算成交量趋势 dataframe["volume_sma"] = dataframe["volume"].rolling( window=VOLUME_TREND_PERIOD, min_periods=1 ).mean() dataframe["volume_ratio"] = ( dataframe["volume"] / dataframe["volume_sma"] ).replace([np.inf, -np.inf], 1.0).fillna(1.0) # 量价配合分析 dataframe["price_change"] = dataframe["close"].pct_change() dataframe["volume_signal"] = self._calculate_volume_signal( dataframe["volume_ratio"], dataframe["price_change"], ) # 计算综合评分（传入 metadata 以获取币种信息） dataframe["entry_score"] = self._calculate_entry_score(dataframe, metadata) # RSI 信号 dataframe["rsi_signal"] = self._calculate_rsi_signal(dataframe["rsi"]) # MACD 趋势信号 dataframe["macd_trend"] = self._calculate_macd_trend(dataframe["macd_histogram"]) # 趋势确认 dataframe["trend_confirmed"] = self._check_trend_confirmation(dataframe) logger.debug( "指标计算完成: %s, 最新评分=%.4f, RSI=%.2f, MACD趋势=%s", metadata.get("pair", "unknown"), dataframe["entry_score"].iloc[-1] if len(dataframe) > 0 else 0, dataframe["rsi"].iloc[-1] if len(dataframe) > 0 else 50, dataframe["macd_trend"].iloc[-1] if len(dataframe) > 0 else "neutral", ) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """入场信号：综合评分 >= MIN_ENTRY_SCORE 且趋势确认。""" dataframe["enter_long"] = 0 dataframe["enter_tag"] = "" if len(dataframe) < 2: return dataframe # 入场条件： # 1. 综合评分 >= MIN_ENTRY_SCORE # 2. 趋势确认（EMA、RSI、MACD、成交量） # 3. RSI 不在超买区域（避免追高） # 4. MACD 趋势向上或中性 latest = dataframe.iloc[-1] entry_score = latest.get("entry_score", 0) trend_confirmed = latest.get("trend_confirmed", False) rsi_signal = latest.get("rsi_signal", "neutral") macd_trend = latest.get("macd_trend", "neutral") # 检查入场条件 score_met = entry_score >= MIN_ENTRY_SCORE trend_ok = trend_confirmed rsi_ok = rsi_signal != "overbought_risk" # 不在超买风险区 macd_ok = macd_trend in ("bullish", "neutral") # MACD向上或中性 if score_met and trend_ok and rsi_ok and macd_ok: dataframe.loc[dataframe.index[-1], "enter_long"] = 1 dataframe.loc[dataframe.index[-1], "enter_tag"] = ( f"score={entry_score:.2f}|rsi={rsi_signal}|macd={macd_trend}" ) logger.info( "入场信号触发: %s, 评分=%.4f, 趋势确认=%s, RSI信号=%s, MACD趋势=%s", metadata.get("pair", "unknown"), entry_score, trend_confirmed, rsi_signal, macd_trend, ) return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """退出信号：反向信号检测。""" dataframe["exit_long"] = 0 if len(dataframe) < 2: return dataframe latest = dataframe.iloc[-1] # 退出条件： # 1. RSI 进入超买区域且价格开始下跌 # 2. MACD 趋势转为向下 # 3. 成交量萎缩且价格下跌 rsi_signal = latest.get("rsi_signal", "neutral") macd_trend = latest.get("macd_trend", "neutral") volume_signal = latest.get("volume_signal", "neutral") price_change = latest.get("price_change", 0) should_exit = False exit_reason = "" # RSI 超买警告 if rsi_signal == "overbought_risk" and price_change < 0: should_exit = True exit_reason = "rsi_overbought" # MACD 趋势反转 if macd_trend == "bearish" and latest.get("macd_histogram", 0) < 0: should_exit = True exit_reason = "macd_bearish" # 成交量萎缩下跌 if volume_signal == "low_volume_fall": should_exit = True exit_reason = "volume_fall" if should_exit: dataframe.loc[dataframe.index[-1], "exit_long"] = 1 logger.info( "退出信号触发: %s, 原因=%s, RSI=%s, MACD=%s, VOL=%s", metadata.get("pair", "unknown"), exit_reason, rsi_signal, macd_trend, volume_signal, ) return dataframe def custom_stoploss( self, pair: str, trade: object, current_time: datetime, current_rate: float, current_profit: float, after_fill: bool, **kwargs, ) -> Optional[float]: """实现追踪止损。 当盈利超过 TRAILING_STOP_TRIGGER (2%) 后，启用移动止损。 止损距离为 TRAILING_STOP_DISTANCE (1%)。 """ if not after_fill: return None # 初始化持仓追踪 if pair not in self._position_tracker: self._position_tracker[pair] = { "entry_time": current_time, "peak_profit": current_profit, "trailing_active": False, } tracker = self._position_tracker[pair] # 更峰值盈利 if current_profit > tracker["peak_profit"]: tracker["peak_profit"] = current_profit # 检查是否触发追踪止损 if not tracker["trailing_active"]: if current_profit >= TRAILING_STOP_TRIGGER: tracker["trailing_active"] = True logger.info( "追踪止损激活: %s, 当前盈利=%.2f%%, 触发阈值=%.2f%%", pair, current_profit * 100, TRAILING_STOP_TRIGGER * 100, ) # 计算止损水平 if tracker["trailing_active"]: # 追踪止损：峰值盈利 - 止损距离 stop_profit = tracker["peak_profit"] - TRAILING_STOP_DISTANCE stoploss = stop_profit / current_profit if current_profit > 0 else -self.stoploss # 止损不能低于基础止损 if stoploss < -self.stoploss: stoploss = -self.stoploss logger.debug( "追踪止损: %s, 峰值=%.2f%%, 止损位=%.2f%%, 当前=%.2f%%", pair, tracker["peak_profit"] * 100, stop_profit * 100, current_profit * 100, ) return stoploss # 未激活时返回基础止损 return None def custom_exit( self, pair: str, trade: object, current_time: datetime, current_rate: float, current_profit: float, **kwargs, ) -> Optional[str]: """检查退出条件：盈利目标和持仓时间限制。""" # 初始化追踪 if pair not in self._position_tracker: self._position_tracker[pair] = { "entry_time": trade.open_date_utc or current_time, "peak_profit": current_profit, "trailing_active": False, } tracker = self._position_tracker[pair] entry_time = tracker["entry_time"] # 检查盈利目标 if current_profit >= PROFIT_EXIT_RATIO: logger.info( "盈利目标达成退出: %s, 盈利=%.2f%% >= 目标=%.2f%%", pair, current_profit * 100, PROFIT_EXIT_RATIO * 100, ) self._position_tracker.pop(pair, None) return f"profit_target_{PROFIT_EXIT_RATIO:.0%}" # 检查持仓时间限制 if entry_time: holding_duration = current_time - entry_time holding_hours = holding_duration.total_seconds() / 3600 if holding_hours >= MAX_HOLDING_HOURS: logger.info( "持仓时间超限退出: %s, 持仓=%.1f小时 >= 上限=%d小时", pair, holding_hours, MAX_HOLDING_HOURS, ) self._position_tracker.pop(pair, None) return f"time_limit_{MAX_HOLDING_HOURS}h" return None def _calculate_ema(self, series: pd.Series, period: int) -> pd.Series: """计算指数移动平均。""" return series.ewm(span=period, adjust=False).mean() def _calculate_rsi(self, series: pd.Series, period: int) -> pd.Series: """计算相对强弱指标。""" delta = series.diff() gain = delta.where(delta > 0, 0) loss = -delta.where(delta < 0, 0) avg_gain = gain.rolling(window=period, min_periods=1).mean() avg_loss = loss.rolling(window=period, min_periods=1).mean() rs = avg_gain / avg_loss.replace(0, np.inf) rsi = 100 - (100 / (1 + rs)) return rsi.fillna(50) def _calculate_macd( self, series: pd.Series, fast_period: int, slow_period: int, signal_period: int, ) -> dict: """计算 MACD 指标。""" ema_fast = self._calculate_ema(series, fast_period) ema_slow = self._calculate_ema(series, slow_period) macd_line = ema_fast - ema_slow signal_line = self._calculate_ema(macd_line, signal_period) histogram = macd_line - signal_line return { "macd_line": macd_line, "signal_line": signal_line, "histogram": histogram, } def _calculate_volume_signal( self, volume_ratio: pd.Series, price_change: pd.Series, ) -> pd.Series: """计算成交量信号。""" result = pd.Series("neutral", index=volume_ratio.index) # 成交量放大 high_volume = volume_ratio > 1.2 # 成交量萎缩 low_volume = volume_ratio < 0.8 # 量价齐升 result.loc[high_volume & (price_change > 0)] = "bullish_volume" # 量价齐跌 result.loc[high_volume & (price_change < 0)] = "bearish_volume" # 缩量上涨 result.loc[low_volume & (price_change > 0)] = "low_volume_rise" # 缩量下跌 result.loc[low_volume & (price_change < 0)] = "low_volume_fall" # 放量横盘 result.loc[high_volume & (price_change == 0)] = "high_volume_neutral" return result def _calculate_rsi_signal(self, rsi: pd.Series) -> pd.Series: """计算 RSI 信号。""" result = pd.Series("neutral", index=rsi.index) # 超买区域（卖出机会，但做多时是风险） result.loc[rsi > RSI_OVERBUY_THRESHOLD] = "overbought_risk" # 超卖区域（买入机会） result.loc[rsi < RSI_OVERSELL_THRESHOLD] = "oversold_buy" return result def _calculate_macd_trend(self, histogram: pd.Series) -> pd.Series: """计算 MACD 趋势。""" result = pd.Series("neutral", index=histogram.index) result.loc[histogram > 0] = "bullish" result.loc[histogram < 0] = "bearish" return result def _check_trend_confirmation(self, dataframe: DataFrame) -> pd.Series: """检查趋势确认（综合 EMA、RSI、MACD、成交量）。""" if len(dataframe) < 2: return pd.Series(False, index=dataframe.index) # EMA 趋势：价格在 EMA 快线之上 ema_trend = dataframe["close"] > dataframe["ema_fast"] # RSI 趋势：不在超买区域 rsi_trend = dataframe["rsi"] < RSI_OVERBUY_THRESHOLD # MACD 趋势：向上或中性 macd_trend = dataframe["macd_histogram"] >= 0 # 成交量趋势：放大或正常 volume_trend = dataframe["volume_ratio"] >= 0.8 # 计算确认得分（需要至少3项满足） confirmation_score = ( ema_trend.astype(int) + rsi_trend.astype(int) + macd_trend.astype(int) + volume_trend.astype(int) ) # 需要至少3项确认 return confirmation_score >= 3 def _calculate_entry_score(self, dataframe: DataFrame, metadata: dict = None) -> pd.Series: """计算综合入场评分。 评分公式： - 有研究评分时：基础评分 * 0.6 + 研究评分 * 0.4 - 无研究评分时：使用纯本地计算（fallback） """ if len(dataframe) < 2: return pd.Series(0.0, index=dataframe.index) # 获取研究评分（从 metadata 中获取币种信息） research_score = None if metadata: pair = metadata.get("pair", "") # 标准化币种符号（去掉斜杠） symbol = pair.replace("/", "").replace(":", "") research_score = self._fetch_research_score(symbol) # 趋势评分：EMA 金叉 trend_score = 0.0 ema_cross = dataframe["ema_fast"] > dataframe["ema_slow"] trend_score = ema_cross.astype(float) * TREND_CONFIRM_WEIGHT # 指标评分 indicator_score = 0.0 # RSI 评分：超卖加分，超买减分 rsi_score = pd.Series(0.0, index=dataframe.index) rsi_score.loc[dataframe["rsi"] < RSI_OVERSELL_THRESHOLD] = 0.08 rsi_score.loc[dataframe["rsi"] > RSI_OVERBUY_THRESHOLD] = -0.05 rsi_score.loc[(dataframe["rsi"] >= RSI_OVERSELL_THRESHOLD) & (dataframe["rsi"] <= RSI_OVERBUY_THRESHOLD)] = 0.04 # MACD 评分：向上加分 macd_score = pd.Series(0.0, index=dataframe.index) macd_score.loc[dataframe["macd_histogram"] > 0] = 0.08 macd_score.loc[dataframe["macd_histogram"] == 0] = 0.02 # 成交量评分：放量加分 volume_score = pd.Series(0.0, index=dataframe.index) volume_score.loc[dataframe["volume_ratio"] > 1.2] = 0.04 indicator_score = (rsi_score + macd_score + volume_score) / 3 * INDICATOR_WEIGHT # 本地计算的基础评分 base_score = 0.50 local_score = base_score + trend_score + indicator_score # 综合评分：有研究评分时加权融合 if research_score is not None: # 研究评分权重 0.4，本地权重 0.6 total_score = local_score * 0.6 + research_score * RESEARCH_SCORE_FALLBACK_WEIGHT logger.info( "综合评分计算: %s, 本地=%.4f, 研究=%.4f, 综合=%.4f", metadata.get("pair", "unknown") if metadata else "unknown", local_score.iloc[-1] if len(local_score) > 0 else 0, research_score, total_score.iloc[-1] if len(total_score) > 0 else 0, ) else: # API 不可用时使用纯本地计算 total_score = local_score logger.debug( "研究评分不可用，使用本地评分: %s, score=%.4f", metadata.get("pair", "unknown") if metadata else "unknown", total_score.iloc[-1] if len(total_score) > 0 else 0, ) # 限制在 0-1 之间 return total_score.clip(0.0, 1.0) # 保持向后兼容：类名使用 SampleStrategy # 真实策略逻辑在 QuantStrategy 实现