import logging from typing import Any, Dict import numpy as np from freqtrade.freqai.prediction_models.ReinforcementLearner import ReinforcementLearner from freqtrade.freqai.RL.Base5ActionRLEnv import Actions, Base5ActionRLEnv, Positions logger = logging.getLogger(__name__) class MyCoolRLModel(ReinforcementLearner): """ 针对高波动资产 (如 PEPE) 优化的强化学习模型。 核心改进： 1. 移除入场奖励，引入开仓成本惩罚。 2. 引入非线性盈亏奖励，鼓励捕捉大趋势。 3. 严厉惩罚浮亏持仓（抗单行为）。 4. 增加对大户操纵的识别和防御机制。 """ class MyRLEnv(Base5ActionRLEnv): """ 自定义环境，重写 calculate_reward 以适应动量交易和大户操纵防御。 """ def calculate_reward(self, action: int) -> float: """ 奖励函数的示例。这是用户可能希望 注入自己创意的一个函数。 警告！ 此函数是功能展示，旨在展示尽可能多的 环境控制功能。它还设计用于在小型计算机上快速运行。 这是一个基准，*不* 用于生产环境。 :param action: int = 智能体为当前K线做出的动作。 :return: float = 给智能体当前步骤的奖励（用于优化 神经网络中的权重） """ # 首先，如果动作无效，则惩罚 if not self._is_valid(action): self.tensorboard_log("invalid", category="actions") return -2.0 # 获取核心状态数据 pnl = self.get_unrealized_profit() # 当前浮动盈亏 (百分比，如 0.01 代表 1%) # 获取持仓时间 (K线数量) trade_duration = 0 if self._last_trade_tick is not None: trade_duration = self._current_tick - self._last_trade_tick # 读取配置中的最大持仓时间，默认 100 根 K 线 max_trade_duration = self.rl_config.get("max_trade_duration_candles", 100) # 奖励累加器 reward = 0.0 # ========================================================= # 场景 A: 决定入场 (Long Enter / Short Enter) # ========================================================= if action in (Actions.Long_enter.value, Actions.Short_enter.value): if self._position == Positions.Neutral: # 入场给予轻微惩罚，避免过度交易，但不要太重 # 从 -0.05 改为 -0.01，让模型更愿意尝试入场 return -0.01 # ========================================================= # 场景 B: 观望 (Neutral) # ========================================================= if action == Actions.Neutral.value and self._position == Positions.Neutral: # 空仓观望给予轻微惩罚，鼓励模型寻找机会 # 从 0 改为 -0.001，避免模型一直观望 return -0.001 # ========================================================= # 场景 C: 持仓中 (Holding) - 每一根 K 线都会触发 # ========================================================= if self._position in (Positions.Short, Positions.Long): # 如果当前动作是继续持有 (Neutral) if action == Actions.Neutral.value: # 1. 时间管理：希望持仓，但不要超时 # - 前50%时间：轻微奖励，鼓励持仓捕捉趋势 # - 50%-80%时间：轻微惩罚 # - 超过80%时间：惩罚加重 time_ratio = trade_duration / max_trade_duration if time_ratio < 0.5: # 前半段时间：轻微奖励，鼓励持仓 time_reward = 0.005 * (1 - time_ratio * 2) reward += time_reward elif time_ratio < 0.8: # 50%-80%时间：轻微惩罚 time_penalty = -0.01 * ((time_ratio - 0.5) / 0.3) reward += time_penalty else: # 超过80%时间：惩罚加重，提醒模型该离场了 time_penalty = -0.02 * ((time_ratio - 0.8) / 0.2) - 0.01 reward += time_penalty # 2. 浮动盈亏反馈 (关键！) if pnl > 0: # 浮盈：给予微弱的正反馈，鼓励拿住趋势 # 使用 log 函数让奖励增长平缓，避免模型过于贪婪而不止盈 reward += np.log(1 + pnl) * 0.5 else: # 浮亏：给予惩罚，但不要太严厉 # 从 abs(pnl) * 2.0 改为 abs(pnl) * 1.0 reward -= (abs(pnl) * 1.0) # 3. 止损惩罚加速 # 如果浮亏超过 3%，给予额外的惩罚，但不要太严厉 if pnl < -0.03: reward -= 0.5 # 从 -1.0 改为 -0.5 # 4. 【新增】检测是否在大户操纵中被套 # 如果刚入场不久就出现大幅反向波动，可能是被大户收割 if trade_duration < 5 and pnl < -0.01: # 入场后5根K线内就亏损超过1%，很可能是被大户骗了 reward -= 1.0 # 从 -3.0 改为 -1.0，不要太严厉 # ========================================================= # 场景 D: 离场结算 (Exit) # ========================================================= if (action == Actions.Long_exit.value and self._position == Positions.Long) or \ (action == Actions.Short_exit.value and self._position == Positions.Short): # 基础因子 factor = 10.0 # 【核心修改 3】非线性结算奖励 if pnl > 0: # 盈利：奖励 = PnL * 因子 # 如果这笔交易是大赚 (比如 > 2%)，因子翻倍 if pnl > 0.02: factor *= 2.0 # 最终奖励 total_reward = pnl * factor # 额外奖励：如果这笔交易很快就赚了钱（高效率） if trade_duration < (max_trade_duration * 0.2): total_reward *= 1.5 # 【新增】奖励快速脱离陷阱 # 如果是因被大户收割而被迫止损，但及时离场，给予一定奖励 # 这鼓励模型快速识别并逃离陷阱 if trade_duration < 5 and pnl < 0 and pnl > -0.02: # 在5根K线内止损，亏损小于2%，说明及时逃离了陷阱 total_reward += 0.5 # 给予小奖励，奖励快速止损 return float(total_reward) else: # 亏损：惩罚 = PnL * 因子 # 亏损时的惩罚系数通常要比盈利系数大，模拟"损失厌恶" # 这会让模型非常忌惮亏损离场 loss_factor = 15.0 # 【新增】区分不同类型的亏损 # 如果是被大户收割导致的快速亏损，惩罚可以适当减轻 # 因为这是市场环境问题，而非模型判断错误 if trade_duration < 5 and pnl < -0.02: # 快速大幅亏损，可能是被大户收割 loss_factor = 10.0 # 减轻惩罚，因为这是市场陷阱 else: loss_factor = 15.0 # 正常惩罚 return float(pnl * loss_factor) return float(reward)