"""TaxBrainV29 ����·��������Э���㡣 """ from __future__ import annotations from typing import Optional, Any, Dict import talib.abstract as ta import copy import json import sys import time import uuid from dataclasses import dataclass, field, replace from datetime import datetime from pathlib import Path import types from typing import Any, Dict, Optional, Set, List, Tuple import pandas as pd from pandas import Timestamp import math ROOT_PATH = Path(__file__).resolve().parents[2] if str(ROOT_PATH) not in sys.path: sys.path.insert(0, str(ROOT_PATH)) MODULE_ALIAS = 'user_data.strategies.TaxBrainV29' if MODULE_ALIAS not in sys.modules: sys.modules[MODULE_ALIAS] = sys.modules.get(__name__) module_obj = sys.modules.get(__name__) if module_obj is None: module_obj = types.ModuleType(__name__) sys.modules[__name__] = module_obj module_obj.__dict__.update(globals()) try: from user_data.strategies.agents.exits.facade import ExitFacade from user_data.strategies.agents.exits.router import EXIT_ROUTER, SLContext, TPContext from user_data.strategies.agents.exits.exit import ExitTags except Exception: # pragma: no cover ExitFacade = None # type: ignore EXIT_ROUTER = None # type: ignore SLContext = None # type: ignore TPContext = None # type: ignore ExitTags = None # type: ignore try: from freqtrade.enums import RunMode except Exception: RunMode = None try: from freqtrade.strategy import ( IStrategy, Trade, Order, PairLocks, informative, # @informative decorator # Hyperopt Parameters BooleanParameter, CategoricalParameter, DecimalParameter, IntParameter, RealParameter, # timeframe helpers timeframe_to_minutes, timeframe_to_next_date, timeframe_to_prev_date, # Strategy helper functions merge_informative_pair, stoploss_from_absolute, stoploss_from_open, ) except Exception: # pragma: no cover class IStrategy: # type: ignore def __init__(self, *args, **kwargs): if not hasattr(self, "dp"): self.dp = types.SimpleNamespace() class Trade: # type: ignore pass class Order: # type: ignore pass class PairLocks: # type: ignore pass class BooleanParameter: # type: ignore pass class CategoricalParameter: # type: ignore pass class DecimalParameter: # type: ignore pass class IntParameter: # type: ignore pass class RealParameter: # type: ignore pass def informative(timeframe: str): # type: ignore def decorator(func): return func return decorator def timeframe_to_minutes(timeframe: str): # type: ignore return 0 def timeframe_to_next_date(timeframe: str, date): # type: ignore return date def timeframe_to_prev_date(timeframe: str, date): # type: ignore return date def merge_informative_pair(*args, **kwargs): # type: ignore return None def stoploss_from_absolute(*args, **kwargs): # type: ignore return 0.0 def stoploss_from_open(*args, **kwargs): # type: ignore return 0.0 # TaxBrainV29.py 顶部（和其它 import 放一起） from user_data.strategies.agents.exits import rules_threshold # 注册 SL/TP + 即时退出规则 from user_data.strategies.config.v29_config import V29Config, apply_overrides from user_data.strategies.agents.portfolio.treasury import TreasuryAgent from user_data.strategies.agents.portfolio.tier import TierAgent, TierManager from user_data.strategies.agents.portfolio.sizer import SizerAgent from user_data.strategies.agents.signals.builder import ( collect_factor_requirements, collect_indicator_requirements, ) from user_data.strategies.agents.signals import builder, indicators, schemas from user_data.strategies.agents.portfolio.risk import RiskAgent from user_data.strategies.agents.portfolio.reservation import ReservationAgent from user_data.strategies.agents.portfolio.persist import StateStore from user_data.strategies.agents.exits.exit import ExitPolicyV29 from user_data.strategies.agents.portfolio.execution import ExecutionAgent from user_data.strategies.agents.portfolio.cycle import CycleAgent from user_data.strategies.agents.portfolio.analytics import AnalyticsAgent # safe, optional import – file lives next to your strategy agents try: from . import evaluate_sl_override except Exception: # pragma: no cover # type: ignore def evaluate_sl_override(ctx: Dict[str, Any], base_sl_pct: float | None) -> Optional[float]: return None # 让所有内置入场信号完成“注册” from user_data.strategies.agents.signals import builtin_signals as _signals # noqa: F401 class _NoopStateStore: """No-op persistence adapter used during backtests/hyperopt to keep state ephemeral.""" def save(self) -> None: # pragma: no cover - trivial return def load_if_exists(self) -> None: # pragma: no cover - trivial return @dataclass class ActiveTradeMeta: """ActiveTradeMeta 用于记录在市订单在 TaxBrainV29 流程中的关键元数据，方便执行、退出与风控代理在不同 hook 间共享统一视图。 Attributes: sl_pct: 建仓时设定的止损百分比（通常为负值，代表最大亏损距离）。 tp_pct: 建仓时设定的止盈百分比。 direction: 信号方向，取值为 'long' 或 'short'。 entry_bar_tick: 建仓发生时的全局 bar_tick 序号，用于与 CycleAgent 对齐。 entry_price: 建仓价格，供执行/退出时计算盈亏。 bucket: 拨款桶信息，fast 或 slow。 icu_bars_left: ICU 强制回收剩余 bar 数，None 表示当前不受 ICU 约束。 """ sl_pct: float tp_pct: float direction: str entry_bar_tick: int entry_price: float bucket: str icu_bars_left: Optional[int] exit_profile: Optional[str] = None recipe: Optional[str] = None plan_timeframe: Optional[str] = None plan_atr_pct: Optional[float] = None tier_name: Optional[str] = None @dataclass class PairState: """PairState 汇聚单个交易对的即时状态，用于信号评估、财政拨款与风险检查之间的信息同步。 Attributes: closs: 连续亏损计数，驱动 TierPolicy 的升降级与冷却时长。 local_loss: 累计局部亏损金额，用于疼痛加权财政拨款。 cooldown_bars_left: 当前剩余冷却 bar 数，>0 时禁止开新仓。 last_dir: 最近一次信号方向，None 表示暂无有效信号。 last_score: 最近一次信号期望收益（edge）数值。 last_squad: 最近一次信号所属 squad（MRL/PBL/TRS 等）。 last_sl_pct: 最近一次信号建议的止损百分比。 last_tp_pct: 最近一次信号建议的止盈百分比。 last_atr_pct: 最近一次记录的 ATR 百分比，供早锁盈估算使用。 active_trades: 当前在市的交易字典，键为 trade_id，值为 ActiveTradeMeta。 """ closs: int = 0 local_loss: float = 0.0 cooldown_bars_left: int = 0 last_dir: Optional[str] = None last_score: float = 0.0 last_kind: Optional[str] = None last_squad: Optional[str] = None last_sl_pct: float = 0.0 last_tp_pct: float = 0.0 last_atr_pct: float = 0.0 last_exit_profile: Optional[str] = None last_recipe: Optional[str] = None active_trades: Dict[str, ActiveTradeMeta] = field(default_factory=dict) @dataclass class TreasuryState: """TreasuryState 持久化 fast/slow 拨款结果以及财政周期起点，保证跨周期及崩溃恢复后的拨款连续性。 Attributes: fast_alloc_risk: fast 拨款桶中每个交易对的名义风险额度。 slow_alloc_risk: slow 拨款桶中每个交易对的名义风险额度。 cycle_start_tick: 当前财政周期开始时记录的 bar_tick。 cycle_start_equity: 当前财政周期开始时的权益快照。 """ fast_alloc_risk: Dict[str, float] = field(default_factory=dict) slow_alloc_risk: Dict[str, float] = field(default_factory=dict) cycle_start_tick: int = 0 cycle_start_equity: float = 0.0 def to_snapshot(self) -> Dict[str, Any]: """导出财政子模块的可序列化快照数据，供 StateStore 写入磁盘。 Returns: Dict[str, Any]: fast/slow 拨款与周期起点组成的字典。 """ return { "fast_alloc_risk": self.fast_alloc_risk, "slow_alloc_risk": self.slow_alloc_risk, "cycle_start_tick": self.cycle_start_tick, "cycle_start_equity": self.cycle_start_equity, } def restore_snapshot(self, payload: Dict[str, Any]) -> None: """从 StateStore 读取的字典中恢复财政拨款状态。 Args: payload: 包含 fast/slow 拨款及周期起点信息的字典。 """ self.fast_alloc_risk = {k: float(v) for k, v in payload.get( "fast_alloc_risk", {}).items()} self.slow_alloc_risk = {k: float(v) for k, v in payload.get( "slow_alloc_risk", {}).items()} self.cycle_start_tick = int(payload.get("cycle_start_tick", 0)) self.cycle_start_equity = float(payload.get("cycle_start_equity", 0.0)) class GlobalState: """GlobalState 负责维护 TaxBrainV29 的组合级状态，包括债务池、各交易对风险、财政拨款以及 bar_tick 进度。 该容器为多个代理提供读写接口，并承担快照持久化与恢复的职责。 """ def __init__(self, cfg: V29Config) -> None: """初始化全局状态容器。 Args: cfg: 当前加载的 V29Config，用于读取 CAP、税率等参数。 """ self.cfg = cfg self.per_pair: Dict[str, PairState] = {} self.debt_pool: float = 0.0 self.trade_risk_ledger: Dict[str, float] = {} self.pair_risk_open: Dict[str, float] = {} self.treasury = TreasuryState() self.bar_tick: int = 0 self.current_cycle_ts: Optional[float] = None self.last_finalized_bar_ts: Optional[float] = None self.reported_pairs_for_current_cycle: set[str] = set() self.last_finalize_walltime: float = time.time() # === Nominal position ledgers (amount * price, leverage-agnostic) === self.trade_stake_ledger: Dict[str, float] = {} self.pair_stake_open: Dict[str, float] = {} def _canonical_pair(self, pair: str | None) -> str: if not pair: return "" return str(pair).split(":")[0] def get_pair_state(self, pair: str) -> PairState: """获取指定交易对的 PairState，如不存在则初始化默认实例。 Args: pair: 交易对名称，例如 'BTC/USDT'。 Returns: PairState: 可供调用方修改的状态对象。 """ key = self._canonical_pair(pair) if key not in self.per_pair: self.per_pair[key] = PairState() return self.per_pair[key] def get_total_open_risk(self) -> float: """汇总所有交易对当前在市风险，用于组合 CAP 控制。 Returns: float: 当前组合层面的风险敞口总和。 """ return sum(self.pair_risk_open.values()) def get_dynamic_portfolio_cap_pct(self, equity: float) -> float: """根据当前权益与债务压力计算组合 VaR 上限占比。 Args: equity: 当前账户权益。 Returns: float: 调整后的组合 CAP 百分比；债务率过高或 equity<=0 时会自动折减。 """ base = self.cfg.portfolio_cap_pct_base if equity <= 0: return base * 0.5 if (self.debt_pool / equity) > self.cfg.drawdown_threshold_pct: return base * 0.5 return base def per_pair_cap_room(self, pair: str, equity: float, tier_pol, reserved: float) -> float: """计算指定交易对在单票 CAP 约束下剩余可分配的风险空间。 Args: pair: 交易对名称。 equity: 当前账户权益。 tier_pol: 对应 CLOSS 等级的 TierPolicy。 reserved: 当前交易对已预约的风险额度。 Returns: float: 在 CAP 范围内仍可追加的风险额度。 """ key = self._canonical_pair(pair) cap = tier_pol.per_pair_risk_cap_pct * equity used = self.pair_risk_open.get(key, 0.0) + reserved return max(0.0, cap - used) def record_trade_open( self, pair: str, trade_id: str, real_risk: float, sl_pct: float, tp_pct: float, direction: str, bucket: str, entry_price: float, tier_pol, exit_profile: Optional[str] = None, recipe: Optional[str] = None, plan_timeframe: Optional[str] = None, plan_atr_pct: Optional[float] = None, tier_name: Optional[str] = None, stake_nominal: Optional[float] = None, # nominal position size (amount * price) ) -> None: """记录新建仓的风险占用，并创建 ActiveTradeMeta 以便执行与退出逻辑后续引用。 Args: pair: 交易对名称。 trade_id: Freqtrade 分配的交易标识。 real_risk: 根据止损距离计算出的真实风险敞口。 sl_pct: 建仓时的止损百分比。 tp_pct: 建仓时的止盈百分比。 direction: 方向标识 'long' 或 'short'。 bucket: 当前订单使用的拨款桶（fast/slow）。 entry_price: 建仓价格。 tier_pol: 所属 TierPolicy，用于初始化 ICU 倒计时与冷却时长。 """ pst = self.get_pair_state(pair) pair_key = self._canonical_pair(pair) tier_name = tier_name or (getattr(tier_pol, "name", None) if tier_pol else None) # 记录的风险应为含杠杆的实际损失额 self.trade_risk_ledger[trade_id] = float(real_risk) self.pair_risk_open[pair_key] = self.pair_risk_open.get( pair_key, 0.0) + float(real_risk) # === Derive nominal stake from risk/sl_pct and accumulate per-pair exposure === if stake_nominal is None: if sl_pct and sl_pct > 0: stake_nominal = real_risk / sl_pct else: stake_nominal = 0.0 stake_nominal = float(max(stake_nominal or 0.0, 0.0)) self.trade_stake_ledger[trade_id] = stake_nominal self.pair_stake_open[pair_key] = self.pair_stake_open.get(pair_key, 0.0) + stake_nominal icu_left = tier_pol.icu_force_exit_bars if tier_pol.icu_force_exit_bars > 0 else None pst.active_trades[trade_id] = ActiveTradeMeta( sl_pct=float(sl_pct), tp_pct=float(tp_pct), direction=str(direction), entry_bar_tick=self.bar_tick, entry_price=float(entry_price), bucket=str(bucket), icu_bars_left=icu_left, exit_profile=exit_profile, recipe=recipe, plan_timeframe=plan_timeframe, plan_atr_pct=float(plan_atr_pct) if plan_atr_pct is not None else None, tier_name=tier_name, ) pst.cooldown_bars_left = max( pst.cooldown_bars_left, tier_pol.cooldown_bars) def record_trade_close(self, pair: str, trade_id: str, profit_abs: float, tier_mgr) -> None: """在平仓时回收风险、更新债务池，并依据盈亏调整冷却与连续亏损计数。 Args: pair: 交易对名称。 trade_id: 交易标识。 profit_abs: 本次交易的绝对盈亏（正数表示盈利或打平）。 tier_mgr: TierManager，用于在状态更新后获取新的策略参数。 """ pst = self.get_pair_state(pair) pair_key = self._canonical_pair(pair) # 1) 回收风险账本 was_risk = self.trade_risk_ledger.pop(trade_id, 0.0) self.pair_risk_open[pair_key] = max( 0.0, self.pair_risk_open.get(pair_key, 0.0) - was_risk, ) if self.pair_risk_open.get(pair_key, 0.0) <= 1e-12: self.pair_risk_open[pair_key] = 0.0 # 2) 回收保证金账本 was_stake = self.trade_stake_ledger.pop(trade_id, 0.0) self.pair_stake_open[pair_key] = max( 0.0, self.pair_stake_open.get(pair_key, 0.0) - was_stake, ) if self.pair_stake_open.get(pair_key, 0.0) <= 1e-12: self.pair_stake_open[pair_key] = 0.0 # 3) 从活跃订单中移除 pst.active_trades.pop(trade_id, None) prev_closs = pst.closs # 4) 根据 TierRouting / DEFAULT_TIER_ROUTING_MAP 计算最大 closs 等级 # 这里使用 TierManager 内部已经构造好的 _routing_map， # 实际上就是你在 v29_config.DEFAULT_TIER_ROUTING_MAP / tier_routing.loss_tier_map 配的那一份。 routing_map = getattr(tier_mgr, "_routing_map", None) or {} max_closs = max(routing_map.keys()) if routing_map else 3 # 没配置时退回旧行为 0–3 # 5) 盈利或打平：一律回到 T0，并使用 after_win 冷却 if profit_abs >= 0: tax = profit_abs * self.cfg.tax_rate_on_wins # 盈利用于偿还债务池 self.debt_pool = max(0.0, self.debt_pool - tax) # local_loss 也可以适度回收（避免永远挂着历史亏损） pst.local_loss = max(0.0, pst.local_loss - profit_abs) # if pst.local_loss<=0: # 债务为0的时候,回到原点 # pst.closs = 0 pol = tier_mgr.get(pst.closs) pst.cooldown_bars_left = max( pst.cooldown_bars_left, pol.cooldown_bars_after_win, ) return # 6) 亏损：先统一记账，再按 closs 规则路由 loss = abs(profit_abs) pst.local_loss += loss self.debt_pool += loss if prev_closs >= max_closs: # 处于“最后一级”（例如 ICU 最高级）时再亏一次： # 按你的要求，这一笔视为“最后一次”，打完必须回到 T0。 pst.closs = 0 pol = tier_mgr.get(pst.closs) pst.cooldown_bars_left = max( pst.cooldown_bars_left, pol.cooldown_bars, ) else: # 尚未到达最高 closs：按连续亏损 +1 计数。 # 例如：0→1，1→2，2→3，3→4 ... 直到 max_closs。 pst.closs = prev_closs + 1 pol = tier_mgr.get(pst.closs) pst.cooldown_bars_left = max( pst.cooldown_bars_left, pol.cooldown_bars, ) def to_snapshot(self) -> Dict[str, Any]: """序列化全局状态为字典，便于持久化或调试。 Returns: Dict[str, Any]: 包含债务池、各交易对状态与财政信息的快照。 """ per_pair_snap: Dict[str, Any] = {} for pair, pst in self.per_pair.items(): per_pair_snap[pair] = { "closs": pst.closs, "local_loss": pst.local_loss, "cooldown_bars_left": pst.cooldown_bars_left, "last_dir": pst.last_dir, "last_score": pst.last_score, "last_kind": pst.last_kind, "last_squad": pst.last_squad, "last_sl_pct": pst.last_sl_pct, "last_tp_pct": pst.last_tp_pct, "last_atr_pct": pst.last_atr_pct, "last_exit_profile": pst.last_exit_profile, "last_recipe": pst.last_recipe, "active_trades": { tid: { "sl_pct": meta.sl_pct, "tp_pct": meta.tp_pct, "direction": meta.direction, "entry_bar_tick": meta.entry_bar_tick, "entry_price": meta.entry_price, "bucket": meta.bucket, "icu_bars_left": meta.icu_bars_left, "exit_profile": meta.exit_profile, "recipe": meta.recipe, } for tid, meta in pst.active_trades.items() }, } return { "debt_pool": self.debt_pool, "per_pair": per_pair_snap, "trade_risk_ledger": self.trade_risk_ledger, "pair_risk_open": self.pair_risk_open, "treasury": self.treasury.to_snapshot(), "bar_tick": self.bar_tick, "current_cycle_ts": self.current_cycle_ts, "last_finalized_bar_ts": self.last_finalized_bar_ts, "last_finalize_walltime": self.last_finalize_walltime, } def restore_snapshot(self, payload: Dict[str, Any]) -> None: """从 `to_snapshot` 生成的字典恢复全局状态，用于断点续跑或崩溃恢复。 Args: payload: 先前保存的状态字典。 """ self.debt_pool = float(payload.get("debt_pool", 0.0)) self.trade_risk_ledger = {k: float(v) for k, v in payload.get( "trade_risk_ledger", {}).items()} self.pair_risk_open = {k: float(v) for k, v in payload.get( "pair_risk_open", {}).items()} self.bar_tick = int(payload.get("bar_tick", 0)) self.current_cycle_ts = payload.get("current_cycle_ts") self.last_finalized_bar_ts = payload.get("last_finalized_bar_ts") self.last_finalize_walltime = float( payload.get("last_finalize_walltime", time.time())) self.per_pair = {} for pair, pst_payload in payload.get("per_pair", {}).items(): pst = PairState( closs=int(pst_payload.get("closs", 0)), local_loss=float(pst_payload.get("local_loss", 0.0)), cooldown_bars_left=int( pst_payload.get("cooldown_bars_left", 0)), last_dir=pst_payload.get("last_dir"), last_score=float(pst_payload.get("last_score", 0.0)), last_kind=pst_payload.get("last_kind"), last_squad=pst_payload.get("last_squad"), last_sl_pct=float(pst_payload.get("last_sl_pct", 0.0)), last_tp_pct=float(pst_payload.get("last_tp_pct", 0.0)), last_atr_pct=float(pst_payload.get("last_atr_pct", 0.0)), last_exit_profile=pst_payload.get("last_exit_profile"), last_recipe=pst_payload.get("last_recipe"), ) for tid, meta_payload in pst_payload.get("active_trades", {}).items(): pst.active_trades[tid] = ActiveTradeMeta( sl_pct=float(meta_payload.get("sl_pct", 0.0)), tp_pct=float(meta_payload.get("tp_pct", 0.0)), direction=str(meta_payload.get("direction", "")), entry_bar_tick=int(meta_payload.get("entry_bar_tick", 0)), entry_price=float(meta_payload.get("entry_price", 0.0)), bucket=str(meta_payload.get("bucket", "slow")), icu_bars_left=( int(meta_payload["icu_bars_left"]) if meta_payload.get("icu_bars_left") is not None else None ), exit_profile=meta_payload.get("exit_profile"), recipe=meta_payload.get("recipe"), ) self.per_pair[pair] = pst self.treasury.restore_snapshot(payload.get("treasury", {})) def reset_cycle_after_restore(self) -> None: """在完成 StateStore 恢复后重置周期追踪计数，使后续 finalize 流程重新对齐当前时间。""" self.current_cycle_ts = None self.reported_pairs_for_current_cycle = set() self.last_finalize_walltime = time.time() class EquityProvider: """EquityProvider 封装权益读写接口，供 CycleAgent、SizerAgent 等组件查询或更新账户权益。""" def __init__(self, initial_equity: float) -> None: """初始化 EquityProvider。 Args: initial_equity: Dry-run 或实盘环境下的初始权益。 """ self.equity_current = float(initial_equity) def to_snapshot(self) -> Dict[str, float]: """导出权益数值快照，便于持久化。 Returns: Dict[str, float]: 仅包含当前权益的字典。 """ return {"equity_current": self.equity_current} def restore_snapshot(self, payload: Dict[str, Any]) -> None: """从快照字典恢复权益数值。 Args: payload: 持久化层读取的权益字典。 """ self.equity_current = float(payload.get( "equity_current", self.equity_current)) def get_equity(self) -> float: """返回当前记录的权益数值。 Returns: float: 最新权益。 """ return self.equity_current def on_trade_closed_update(self, profit_abs: float) -> None: """在交易结束后，根据绝对盈亏增减权益。 Args: profit_abs: 交易的绝对盈利（亏损为负值）。 """ self.equity_current += float(profit_abs) class TaxBrainV29(IStrategy): """TaxBrainV29 是遵循 AGENTS.md 规范实现的多代理路由策略类。 该类负责在 Freqtrade 各个 hook 中调度信号、财政、风险、预约、执行等子代理， 并保持 V29.1 的五项修订（动态 ADX、盈利周期清债、timeframe/startup 覆盖、早锁盈兜底、仅释放预约）持续生效。 """ timeframe = V29Config().timeframe can_short = True startup_candle_count = V29Config().startup_candle_count # minimal_roi = {"0": 0.03} # stoploss = -0.2 use_custom_roi = False use_custom_stoploss = False trailing_stop = False use_exit_signal = True exit_profit_only = False ignore_buy_signals = False ignore_sell_signals = True _indicator_requirements_map: Dict[Optional[str], Set[str]] = {} def __init__(self, config: Dict[str, Any]) -> None: """Initialise strategy state, analytics, and persistence layer.""" base_cfg = V29Config() self.cfg = apply_overrides(base_cfg, config.get("strategy_params", {})) self.timeframe = self.cfg.timeframe self.startup_candle_count = self.cfg.startup_candle_count self.stoploss = self.cfg.sizing.enforce_leverage*-0.2 self.minimal_roi = {"0": 0.50*self.cfg.sizing.enforce_leverage} try: self.__class__.timeframe = self.cfg.timeframe self.__class__.startup_candle_count = self.cfg.startup_candle_count except Exception: pass extra_signal_factors = getattr(self.cfg, "extra_signal_factors", None) self._factor_requirements = collect_factor_requirements( extra_signal_factors) self._indicator_requirements = collect_indicator_requirements( extra_signal_factors) self.__class__._indicator_requirements_map = self._indicator_requirements config_timeframes = tuple( getattr(self.cfg, "informative_timeframes", ())) inferred_timeframes = tuple( tf for tf in self._factor_requirements.keys() if tf) self._informative_timeframes = self._derive_informative_timeframes( config_timeframes, inferred_timeframes ) self._informative_last: Dict[str, Dict[str, pd.Series]] = {} self._informative_cache: Dict[str, Dict[str, pd.DataFrame]] = {} self._register_informative_methods() super().__init__(config) self.tier_mgr = TierManager(self.cfg) self.tier_agent = TierAgent() user_data_dir = Path(config.get("user_data_dir", ".")) initial_equity = float(config.get( "dry_run_wallet", self.cfg.dry_run_wallet_fallback)) self.eq_provider = EquityProvider(initial_equity) self.state = GlobalState(self.cfg) self.analytics = AnalyticsAgent(user_data_dir / "logs") self.reservation = ReservationAgent(self.cfg, analytics=self.analytics) self.treasury_agent = TreasuryAgent(self.cfg, self.tier_mgr) self.risk_agent = RiskAgent(self.cfg, self.reservation, self.tier_mgr) self.exit_facade = ExitFacade(self.cfg, self.tier_mgr) if ExitFacade else None if self.exit_facade: self.exit_facade.attach_strategy(self) self.exit_facade.set_dataprovider(getattr(self, "dp", None)) self.exit_policy = ExitPolicyV29(self.state, self.eq_provider, self.cfg, dp=getattr(self, "dp", None)) self.exit_policy.set_strategy(self) state_file = (user_data_dir / "taxbrain_v29_state.json").resolve() # ���봰�岻�� bot_start() �� backtest/hyperopt ҲҪ����������־û� self._runmode_token: str = self._compute_runmode_token() self._persist_enabled: bool = not self._is_backtest_like_runmode() self.persist = StateStore( filepath=str(state_file), state=self.state, eq_provider=self.eq_provider, reservation_agent=self.reservation, ) if not self._persist_enabled: # backtest/hyperopt ��չʾ���ڴ�״̬��ֹ���� closs �����ݣ�����ְ�� JSON �ļ� self.persist = _NoopStateStore() try: self._tier_debug( f"init runmode_token={self._runmode_token or 'unknown'} backtest_like={self._is_backtest_like_runmode()} persist_enabled={self._persist_enabled}" ) except Exception: pass self.cycle_agent = CycleAgent( cfg=self.cfg, state=self.state, reservation=self.reservation, treasury=self.treasury_agent, risk=self.risk_agent, analytics=self.analytics, persist=self.persist, tier_mgr=self.tier_mgr, ) self.sizer = SizerAgent( self.state, self.reservation, self.eq_provider, self.cfg, self.tier_mgr) self.execution = ExecutionAgent( self.state, self.reservation, self.eq_provider, self.cfg) self._last_signal: Dict[str, Optional[schemas.Candidate]] = {} self._pending_entry_meta: Dict[str, Dict[str, Any]] = {} self._tf_sec = self._tf_to_sec(self.cfg.timeframe) self._candidate_pool_limit = int( max(1, getattr(self.cfg, "candidate_pool_max_per_side", 4)) ) def set_dataprovider(self, dp): """ Freqtrade 在策略实例创建后，会回调此方法注入 DataProvider。 这里顺便把 dp 传给 exit_policy，供 Router 的 Immediate/Vector/Threshold 规则使用。 """ # 调用父类实现（如果父类没有该方法也不会出错） try: super().set_dataprovider(dp) except AttributeError: pass # 桥接给策略本身与退出策略 self.dp = dp if hasattr(self, "exit_policy") and self.exit_policy is not None: self.exit_policy.dp = dp if self.exit_facade: self.exit_facade.set_dataprovider(dp) @staticmethod def _tf_to_sec(tf: str) -> int: """将 Freqtrade 的 timeframe 字符串转换为秒数，用于 CycleAgent 的节奏控制。 Args: tf: 形如 '5m'、'1h' 的时间粒度。 Returns: int: 对应的秒数，未识别时回退到 300 秒。 """ if tf.endswith("m"): return int(tf[:-1]) * 60 if tf.endswith("h"): return int(tf[:-1]) * 3600 if tf.endswith("d"): return int(tf[:-1]) * 86400 return 300 def _tier_debug(self, message: str) -> None: """Append lightweight tier debug info into user_data/logs/tier_debug.log.""" try: user_dir = Path(self.config.get("user_data_dir", "user_data")) log_path = user_dir / "logs" / "tier_debug.log" log_path.parent.mkdir(parents=True, exist_ok=True) ts = datetime.utcnow().isoformat() with open(log_path, "a", encoding="utf-8") as handle: handle.write(f"{ts} {message}\n") except Exception: pass def _compute_runmode_token(self) -> str: cfg_mode = self.config.get("runmode") if hasattr( self, "config") else None if cfg_mode is not None: return str(getattr(cfg_mode, "value", cfg_mode)).lower() strategy_mode = getattr(self, "runmode", None) if strategy_mode is not None: return str(getattr(strategy_mode, "value", strategy_mode)).lower() dp_mode = getattr(getattr(self, "dp", None), "runmode", None) if dp_mode is not None: return str(getattr(dp_mode, "value", dp_mode)).lower() return "" def _ensure_runmode_token(self) -> str: if not self._runmode_token: self._runmode_token = self._compute_runmode_token() return self._runmode_token def _is_backtest_like_runmode(self) -> bool: token = self._ensure_runmode_token() or "" return any(key in token for key in ("backtest", "hyperopt")) def _derive_informative_timeframes( self, manual: tuple[str, ...], inferred: tuple[str, ...] ) -> tuple[str, ...]: """ """ ordered: list[str] = [] for tf in (*manual, *inferred): if not tf or tf == self.timeframe: continue normalized = str(tf) if normalized not in ordered: ordered.append(normalized) return tuple(ordered) @staticmethod def _timeframe_suffix_token(timeframe: Optional[str]) -> str: token = (timeframe or "").strip() return token.replace("/", "_") if token else "" def _prepare_informative_frame(self, frame: Optional[pd.DataFrame], timeframe: str) -> Optional[pd.DataFrame]: if frame is None or frame.empty: return frame suffix = self._timeframe_suffix_token(timeframe) if not suffix: return frame rename_map: dict[str, str] = {} for col in list(frame.columns): if col == "date": continue if col.endswith(f"_{suffix}"): continue rename_map[col] = f"{col}_{suffix}" if rename_map: frame = frame.rename(columns=rename_map) return frame def _register_informative_methods(self) -> None: if not self._informative_timeframes: return for tf in self._informative_timeframes: self.__class__._ensure_informative_method(tf) @classmethod def _ensure_informative_method(cls, timeframe: str) -> None: func_name = f"populate_indicators_{timeframe.replace('/', '_')}" if hasattr(cls, func_name): return @informative(timeframe) def _informative_populator(self, dataframe, metadata, tf=timeframe): needs = getattr( self.__class__, "_indicator_requirements_map", {}).get(tf) return indicators.compute_indicators( dataframe, self.cfg, suffix=None, required=needs or set(), duplicate_ohlc=True, # 保留原始 OHLC（列名保持 open/close/volume） ) _informative_populator.__name__ = func_name setattr(cls, func_name, _informative_populator) def _get_informative_dataframe(self, pair: str, timeframe: str) -> Optional[pd.DataFrame]: cached = self._informative_cache.get(pair, {}).get(timeframe) if cached is not None and not cached.empty: return cached getter = getattr(self.dp, 'get_informative_dataframe', None) if callable(getter): result = getter(pair, timeframe) if isinstance(result, tuple): result = result[0] if isinstance(result, pd.DataFrame) and not result.empty: prepared = self._prepare_informative_frame(result.copy(), timeframe) self._informative_cache.setdefault(pair, {})[timeframe] = prepared return prepared return self._prepare_informative_frame(result, timeframe) result = self.dp.get_analyzed_dataframe(pair, timeframe) if isinstance(result, tuple): result = result[0] if isinstance(result, pd.DataFrame) and not result.empty: prepared = self._prepare_informative_frame(result.copy(), timeframe) self._informative_cache.setdefault(pair, {})[timeframe] = prepared return prepared return self._prepare_informative_frame(result, timeframe) def informative_pairs(self): if not self._informative_timeframes: return [] try: whitelist = self.dp.current_whitelist() except Exception: whitelist = self.config.get( "exchange", {}).get("pair_whitelist", []) pairs: list[tuple[str, str]] = [] if not whitelist: return pairs seen: set[tuple[str, str]] = set() for pair in whitelist: for tf in self._informative_timeframes: key = (pair, tf) if key not in seen: seen.add(key) pairs.append(key) return pairs def bot_start(self, **kwargs) -> None: """ """ self._runmode_token = self._compute_runmode_token() backtest_like = self._is_backtest_like_runmode() self._persist_enabled = not backtest_like try: self.logger.info( f"[tier] runmode={self._runmode_token or 'unknown'} backtest_like={backtest_like} persist={'on' if self._persist_enabled else 'off'}" ) except Exception: pass if self._persist_enabled: self.persist.load_if_exists() else: self.persist = _NoopStateStore() self.cycle_agent.persist = self.persist if self.state.treasury.cycle_start_tick == 0: self.state.treasury.cycle_start_tick = self.state.bar_tick self.state.treasury.cycle_start_equity = self.eq_provider.get_equity() def populate_indicators(self, df: pd.DataFrame, metadata: dict) -> pd.DataFrame: """计算指标、生成候选信号并触发周期 finalize。 Args: df: 原始 K 线数据。 metadata: Freqtrade 提供的上下文字典，至少包含 pair。 Returns: pd.DataFrame: 附加指标列后的数据帧。 Notes: 当 cfg.adx_len != 14 时会记录动态列名日志，对应 V29.1 修订 #1。 """ pair = metadata["pair"] print(">>> base_needs:", self._indicator_requirements.get(None)) base_needs = self._indicator_requirements.get(None) df = indicators.compute_indicators(df, self.cfg, required=base_needs) informative_rows: Dict[str, pd.Series] = {} if self._informative_timeframes: for tf in self._informative_timeframes: try: info_df = self._get_informative_dataframe(pair, tf) except Exception: continue if info_df is None or info_df.empty: continue informative_rows[tf] = info_df.iloc[-1].copy() cache = self._informative_cache.setdefault(pair, {}) cache[tf] = info_df.copy() if informative_rows: self._informative_last[pair] = informative_rows elif pair in self._informative_last: self._informative_last.pop(pair, None) self._informative_cache.pop(pair, None) if len(df) == 0: return df last = df.iloc[-1].copy() last_ts = float(pd.to_datetime( last.get("date", pd.Timestamp.utcnow())).timestamp()) try: whitelist = self.dp.current_whitelist() except Exception: whitelist = [pair] self.cycle_agent.maybe_finalize( pair=pair, bar_ts=last_ts, whitelist=whitelist, timeframe_sec=self._tf_sec, eq_provider=self.eq_provider, ) return df def get_informative_row(self, pair: str, timeframe: str) -> Optional[pd.Series]: """""" return self._informative_last.get(pair, {}).get(timeframe) def get_informative_value( self, pair: str, timeframe: str, column: str, default: Optional[float] = None ) -> Optional[float]: """""" row = self.get_informative_row(pair, timeframe) if row is None: return default try: if column in row: return row[column] except Exception: pass suffix = self._timeframe_suffix_token(timeframe) if suffix: alt_col = f"{column}_{suffix}" try: if alt_col in row: return row[alt_col] except Exception: pass try: return row.get(column, default) # pandas Series get fallback except Exception: return default def _eval_entry_on_row(self, row: pd.Series, inf_rows: Optional[Dict[str, pd.Series]], pst_state) -> Optional[schemas.Candidate]: """ ??????????K??????????? Candidate """ candidates = builder.build_candidates( row, self.cfg, informative=inf_rows) policy = self.tier_mgr.get(pst_state.closs) return self.tier_agent.filter_best(policy, candidates) def _candidate_with_plan( self, pair: str, candidate: Optional[schemas.Candidate], row: pd.Series, inf_rows: Optional[Dict[str, pd.Series]], ) -> Optional[schemas.Candidate]: if not candidate or not getattr(self, "exit_facade", None): return candidate try: _, _, plan = self.exit_facade.resolve_entry_plan(pair, candidate, row, inf_rows or {}) except Exception: plan = None if not plan: return candidate try: return replace( candidate, sl_pct=float(plan.sl_pct) if getattr(plan, "sl_pct", None) else candidate.sl_pct, tp_pct=float(plan.tp_pct) if getattr(plan, "tp_pct", None) else candidate.tp_pct, plan_timeframe=getattr(plan, "timeframe", getattr(candidate, "plan_timeframe", None)), plan_atr_pct=getattr(plan, "atr_pct", getattr(candidate, "plan_atr_pct", None)), ) except Exception: return candidate def _aligned_informative_for_df(self, pair: str, df: pd.DataFrame) -> Dict[str, pd.DataFrame]: """一次性把所有 informative timeframe merge_asof 到主周期索引，避免每行重复合并。""" out = {} for tf in getattr(self, "_informative_timeframes", []): info_df = self._get_informative_dataframe(pair, tf) if info_df is None or info_df.empty: continue left = pd.DataFrame({"_t": pd.to_datetime( df["date"]) if "date" in df.columns else pd.to_datetime(df.index)}) right = info_df.copy() right["_tinfo"] = pd.to_datetime( right["date"]) if "date" in right.columns else pd.to_datetime(right.index) merged = pd.merge_asof(left.sort_values("_t"), right.sort_values("_tinfo"), left_on="_t", right_on="_tinfo", direction="backward") merged.index = df.index out[tf] = merged return out def _informative_rows_for_index(self, aligned_info: Dict[str, pd.DataFrame], idx) -> Dict[str, pd.Series]: """""" rows: Dict[str, pd.Series] = {} for tf, frame in aligned_info.items(): try: rows[tf] = frame.loc[idx] except Exception: continue return rows def _candidate_to_payload(self, candidate: schemas.Candidate) -> dict[str, Any]: return { "direction": candidate.direction, "kind": candidate.kind, "raw_score": candidate.raw_score, "rr_ratio": candidate.rr_ratio, "expected_edge": candidate.expected_edge, "win_prob": candidate.win_prob, "squad": candidate.squad, "sl_pct": candidate.sl_pct, "tp_pct": candidate.tp_pct, "exit_profile": candidate.exit_profile, "recipe": candidate.recipe, "plan_timeframe": getattr(candidate, "plan_timeframe", None), "plan_atr_pct": getattr(candidate, "plan_atr_pct", None), } def _candidate_allowed_by_policy(self, policy, candidate: Optional[schemas.Candidate]) -> bool: if not policy or not candidate: return False if not policy.permits(kind=candidate.kind, squad=candidate.squad, recipe=candidate.recipe): return False if candidate.raw_score < policy.min_raw_score: return False if candidate.rr_ratio < policy.min_rr_ratio: return False if candidate.expected_edge < policy.min_edge: return False return True def _candidate_allowed_any_tier(self, candidate: Optional[schemas.Candidate]) -> bool: if not candidate: return False iter_policies = getattr(self.tier_mgr, "policies", None) policies = iter_policies() if callable(iter_policies) else () if not policies: policies = (self.tier_mgr.get(0),) for policy in policies: if self._candidate_allowed_by_policy(policy, candidate): return True return False def _group_candidates_by_direction( self, candidates: list[schemas.Candidate] ) -> dict[str, list[schemas.Candidate]]: grouped: dict[str, list[schemas.Candidate]] = {"long": [], "short": []} for cand in candidates: grouped.setdefault(cand.direction, []).append(cand) return grouped def _trim_candidate_pool( self, grouped: dict[str, list[schemas.Candidate]] ) -> dict[str, list[schemas.Candidate]]: limited: dict[str, list[schemas.Candidate]] = {"long": [], "short": []} for direction, items in grouped.items(): ordered = sorted(items, key=lambda c: (c.expected_edge, c.raw_score), reverse=True) limited[direction] = ordered[: self._candidate_pool_limit] return limited def _apply_entry_signal( self, df: pd.DataFrame, idx, candidates_by_dir: dict[str, list[schemas.Candidate]] ) -> None: has_long = bool(candidates_by_dir.get("long")) has_short = bool(candidates_by_dir.get("short")) df.at[idx, "enter_long"] = 1 if has_long else 0 df.at[idx, "enter_short"] = 1 if has_short else 0 if not has_long and not has_short: df.at[idx, "enter_tag"] = None return payload = { "version": 2, "candidates": { "long": [ self._candidate_to_payload(cand) for cand in candidates_by_dir.get("long", []) ], "short": [ self._candidate_to_payload(cand) for cand in candidates_by_dir.get("short", []) ], }, } df.at[idx, "enter_tag"] = json.dumps(payload) def _update_last_signal(self, pair: str, candidate: Optional[schemas.Candidate], row: pd.Series) -> None: pst = self.state.get_pair_state(pair) try: pst.last_atr_pct = float(row.get("atr_pct", 0.0) or 0.0) except Exception: pst.last_atr_pct = 0.0 if candidate: pst.last_dir = candidate.direction pst.last_kind = candidate.kind # ← 信号名（mean_rev_long / trend_short / …） pst.last_squad = getattr(candidate, "squad", None) # ← 所属 squad（MRL / PBL / TRS） pst.last_score = candidate.expected_edge pst.last_sl_pct = candidate.sl_pct pst.last_tp_pct = candidate.tp_pct pst.last_exit_profile = candidate.exit_profile pst.last_recipe = candidate.recipe else: pst.last_dir = None pst.last_squad = None pst.last_score = 0.0 pst.last_sl_pct = 0.0 pst.last_tp_pct = 0.0 pst.last_exit_profile = None pst.last_recipe = None self._last_signal[pair] = candidate def populate_entry_trend(self, df: pd.DataFrame, metadata: dict) -> pd.DataFrame: pair = metadata["pair"] if "BNB" in pair: print("BNB") df["enter_long"] = 0 df["enter_short"] = 0 df["enter_tag"] = None aligned_info = self._aligned_informative_for_df(pair, df) runmode = getattr(self.dp, "runmode", None) is_vector_pass = (RunMode and runmode in { RunMode.BACKTEST, RunMode.HYPEROPT, RunMode.PLOT}) pst_snapshot = copy.deepcopy(self.state.get_pair_state(pair)) if is_vector_pass: for idx in df.index: row = df.loc[idx].copy() row["LOSS_TIER_STATE"] = pst_snapshot.closs inf_rows = self._informative_rows_for_index(aligned_info, idx) raw_candidates = builder.build_candidates(row, self.cfg, informative=inf_rows) if not raw_candidates: continue planned: list[schemas.Candidate] = [] for candidate in raw_candidates: cand_with_plan = self._candidate_with_plan(pair, candidate, row, inf_rows) if cand_with_plan: planned.append(cand_with_plan) if not planned: continue if is_vector_pass: planned = [ cand for cand in planned if self._candidate_allowed_any_tier(cand) ] if not planned: continue grouped = self._trim_candidate_pool(self._group_candidates_by_direction(planned)) if not grouped.get("long") and not grouped.get("short"): continue self._apply_entry_signal(df, idx, grouped) # pst_snapshot = advance_state(pst_snapshot, sig, row) actual_state = self.state.get_pair_state(pair) last_idx = df.index[-1] last_row = df.loc[last_idx].copy() last_row["LOSS_TIER_STATE"] = actual_state.closs last_inf_rows = self._informative_rows_for_index(aligned_info, last_idx) last_candidate = self._eval_entry_on_row( last_row, last_inf_rows, actual_state) if last_candidate: last_candidate = self._candidate_with_plan(pair, last_candidate, last_row, last_inf_rows) if not is_vector_pass and last_candidate: grouped = self._trim_candidate_pool( self._group_candidates_by_direction([last_candidate]) ) self._apply_entry_signal(df, last_idx, grouped) self._update_last_signal(pair, last_candidate, last_row) return df def populate_exit_trend(self, df: pd.DataFrame, metadata: dict) -> pd.DataFrame: """ 通过 Exit Router 的 VectorExit 渠道应用“可插拔退出规则”。 - 规则可以直接改 df 并返回 df，或返回一个包含列的 dict（exit_long/exit_short/exit_tag） - Router 负责 OR 聚合、exit_tag 覆盖策略（按优先级） """ # 兜底：Freqtrade 要求存在列 if "exit_long" not in df.columns: df["exit_long"] = 0 if "exit_short" not in df.columns: df["exit_short"] = 0 if "exit_tag" not in df.columns: df["exit_tag"] = None meta = { "pair": metadata.get("pair"), "timeframe": getattr(self, "timeframe", None) or self.timeframe, } # 交给 Router 聚合所有 vector exit 规则 try: if self.exit_facade: df = self.exit_facade.apply_vector( df=df, metadata=meta, state=self.state, timeframe_col=None, ) elif EXIT_ROUTER is not None: df = EXIT_ROUTER.apply_vector_exits( df=df, metadata=meta, dp=self.dp, cfg=self.cfg, state=self.state, timeframe_col=None, ) except Exception: # 安全兜底：如果外部规则出错，不阻断主流程 pass return df def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: str | None, side: str, **kwargs) -> bool: runmode_cfg = self.config.get("runmode", None) runmode_token = str( getattr(runmode_cfg, "value", runmode_cfg) or "").lower() dp_runmode = getattr(self.dp, "runmode", None) dp_mode_token = str( getattr(dp_runmode, "value", dp_runmode) or "").lower() backtest_modes = {"runmode.backtest", "backtest", "hyperopt", "runmode.hyperopt"} is_backtest_mode = any( token in backtest_modes for token in (runmode_token, dp_mode_token) if token) pst = self.state.get_pair_state(pair) tier_pol = self.tier_mgr.get(pst.closs) try: self._tier_debug( f"cte called pair={pair} runmode_token={runmode_token} dp_token={dp_mode_token} backtest={is_backtest_mode} closs={pst.closs}" ) except Exception: pass if is_backtest_mode and pst.cooldown_bars_left > 0: try: now_ts = float(current_time.timestamp()) except Exception: now_ts = None if now_ts is not None: last_ts = getattr(pst, "_cooldown_last_ts", None) if last_ts is None: pst._cooldown_last_ts = now_ts else: bars_elapsed = int((now_ts - last_ts) // self._tf_sec) if bars_elapsed > 0: pst.cooldown_bars_left = max( 0, pst.cooldown_bars_left - bars_elapsed) pst._cooldown_last_ts = now_ts # 冷却期一律禁止开新仓 # print(f"pst.closs_{pst.closs}") if pst.cooldown_bars_left > 0: return False # 若当前 tier 配置了 single_position_only，则有仓位时禁止再开 if tier_pol and getattr(tier_pol, "single_position_only", False) and pst.active_trades: return False # 实盘 / 干跑：仍用“最新一拍”的 _last_signal 语义 requested_dir = "long" if side.lower() in ("buy", "long") else "short" payload = None candidate_from_tag: Optional[schemas.Candidate] = None if entry_tag: try: payload = json.loads(entry_tag) except Exception: payload = None if isinstance(payload, dict): candidate_from_tag = self._resolve_candidate_from_tag(pair, payload, side) elif payload is not None: normalized = self._normalize_entry_meta(payload, side) if normalized and pair not in self._pending_entry_meta: self._pending_entry_meta[pair] = normalized if is_backtest_mode: try: self._tier_debug( f"cte precheck pair={pair} candidate_present={bool(candidate_from_tag)} entry_tag_present={bool(entry_tag)}" ) if not candidate_from_tag: tier_name = getattr(tier_pol, "name", None) if tier_pol else None try: self.logger.info( f"[tier] bt-entry-skip pair={pair} closs={pst.closs} tier={tier_name} reason=no_candidate" ) except Exception: pass self._tier_debug( f"bt-entry-skip pair={pair} closs={pst.closs} tier={tier_name} reason=no_candidate" ) return False if pair not in self._pending_entry_meta: self._pending_entry_meta[pair] = self._candidate_meta_from_candidate(candidate_from_tag) tier_name = getattr(tier_pol, "name", None) if tier_pol else None try: self.logger.info( f"[tier] bt-entry pair={pair} closs={pst.closs} tier={tier_name} recipe={getattr(candidate_from_tag, 'recipe', None)} kind={getattr(candidate_from_tag, 'kind', None)} sl={getattr(candidate_from_tag, 'sl_pct', None)} tp={getattr(candidate_from_tag, 'tp_pct', None)}" ) except Exception: pass self._tier_debug( f"bt-entry pair={pair} closs={pst.closs} tier={tier_name} recipe={getattr(candidate_from_tag, 'recipe', None)} kind={getattr(candidate_from_tag, 'kind', None)} sl={getattr(candidate_from_tag, 'sl_pct', None)} tp={getattr(candidate_from_tag, 'tp_pct', None)}" ) return True except Exception as exc: self._tier_debug(f"bt-entry-error pair={pair} err={exc}") return False sig = self._last_signal.get(pair) if not sig or sig.direction != requested_dir: return False self._pending_entry_meta[pair] = { "sl_pct": sig.sl_pct, "tp_pct": sig.tp_pct, "dir": sig.direction, "exit_profile": sig.exit_profile, "recipe": sig.recipe, "plan_timeframe": getattr(sig, "plan_timeframe", None), "atr_pct": getattr(sig, "plan_atr_pct", None), } return True @staticmethod def _dir_from_side(side: str) -> str: return "long" if str(side).lower() in ("buy", "long") else "short" def _candidate_meta_from_candidate(self, candidate: schemas.Candidate) -> dict[str, Any]: return { "dir": candidate.direction, "sl_pct": candidate.sl_pct, "tp_pct": candidate.tp_pct, "exit_profile": candidate.exit_profile, "recipe": candidate.recipe, "plan_timeframe": getattr(candidate, "plan_timeframe", None), "atr_pct": getattr(candidate, "plan_atr_pct", None), } def _normalize_entry_meta(self, meta: Optional[Dict[str, Any]], side: str) -> Optional[Dict[str, Any]]: """ """ if meta is None: return None normalized = dict(meta) sl_val = normalized.get('sl_pct') if sl_val is None and 'sl' in normalized: sl_val = normalized.get('sl') try: normalized['sl_pct'] = float(sl_val) if sl_val is not None else 0.0 except Exception: normalized['sl_pct'] = 0.0 tp_val = normalized.get('tp_pct') if tp_val is None and 'tp' in normalized: tp_val = normalized.get('tp') try: normalized['tp_pct'] = float(tp_val) if tp_val is not None else 0.0 except Exception: normalized['tp_pct'] = 0.0 direction = normalized.get('dir') if not direction: direction = self._dir_from_side(side) normalized['dir'] = direction if 'plan_timeframe' in normalized: ptf = normalized.get('plan_timeframe') if ptf is None: normalized['plan_timeframe'] = None else: token = str(ptf).strip() normalized['plan_timeframe'] = token or None atr_val = normalized.get('atr_pct') if atr_val is not None: try: normalized['atr_pct'] = float(atr_val) except Exception: normalized['atr_pct'] = None else: normalized['atr_pct'] = None return normalized def _resolve_candidate_from_tag( self, pair: str, payload: Dict[str, Any], side: str ) -> Optional[schemas.Candidate]: candidates_pool = payload.get("candidates") if not isinstance(candidates_pool, dict): return None direction = self._dir_from_side(side) raw_list = candidates_pool.get(direction) or [] if not raw_list: return None hydrated: list[schemas.Candidate] = [] for raw in raw_list: try: plan_atr_val = raw.get("plan_atr_pct") plan_atr = None if plan_atr_val not in (None, ""): plan_atr = float(plan_atr_val) hydrated.append( schemas.Candidate( direction=direction, kind=str(raw["kind"]), raw_score=float(raw["raw_score"]), rr_ratio=float(raw["rr_ratio"]), win_prob=float(raw.get("win_prob", 0.0)), expected_edge=float(raw["expected_edge"]), squad=str(raw["squad"]), sl_pct=float(raw["sl_pct"]), tp_pct=float(raw["tp_pct"]), exit_profile=raw.get("exit_profile"), recipe=raw.get("recipe"), plan_timeframe=raw.get("plan_timeframe"), plan_atr_pct=plan_atr, ) ) except (KeyError, TypeError, ValueError): continue if not hydrated: return None pst = self.state.get_pair_state(pair) tier_pol = self.tier_mgr.get(pst.closs) best = self.tier_agent.filter_best(tier_pol, hydrated) if best is None: try: tier_name = getattr(tier_pol, "name", None) if tier_pol else None recipes = [getattr(h, "recipe", None) for h in hydrated] kinds = [getattr(h, "kind", None) for h in hydrated] self.logger.info( f"[tier] reject_by_tier pair={pair} closs={pst.closs} tier={tier_name} recipes={recipes} kinds={kinds}" ) self._tier_debug( f"reject_by_tier pair={pair} closs={pst.closs} tier={tier_name} recipes={recipes} kinds={kinds}" ) except Exception: pass return best def _compute_profit_abs(self, trade, rate: float) -> float: """Compute absolute PnL for a trade, falling back to a simple delta calc.""" try: for attr in ("close_profit_abs", "profit_abs"): val = getattr(trade, attr, None) if val is not None: return float(val) except Exception: pass try: open_rate = float(getattr(trade, "open_rate", rate) or rate) amount = float(getattr(trade, "amount", 0.0) or 0.0) direction = -1.0 if getattr(trade, "is_short", False) else 1.0 return direction * (float(rate) - open_rate) * amount except Exception: return 0.0 def confirm_trade_exit( self, pair: str, trade, order_type: str, amount: float, rate: float, time_in_force: str, exit_reason: str, current_time: datetime | None = None, **kwargs, ) -> bool: """Ensure backtests/hyperopts update tier state on exits.""" backtest_like = self._is_backtest_like_runmode() if backtest_like and self.tier_mgr: trade_id = str(getattr(trade, "trade_id", getattr(trade, "id", "NA"))) processed = getattr(self, "_bt_closed_trades", None) if processed is None: processed = set() self._bt_closed_trades = processed if trade_id not in processed: profit_abs = self._compute_profit_abs(trade, rate) try: self._tier_debug( f"ctx called pair={pair} id={trade_id} pnl={profit_abs}" ) before = self.state.get_pair_state(pair).closs self.state.record_trade_close(pair, trade_id, profit_abs, self.tier_mgr) after = self.state.get_pair_state(pair).closs try: self.logger.info( f"[tier] bt-exit {pair} id={trade_id} pnl={profit_abs:.4f} closs {before}->{after}" ) self._tier_debug( f"bt-exit pair={pair} id={trade_id} pnl={profit_abs:.4f} closs {before}->{after}" ) except Exception: pass except Exception as exc: try: self.logger.warning( f"[tier] failed to record backtest exit {trade_id} {pair}: {exc}" ) except Exception: pass processed.add(trade_id) return True def custom_stake_amount( self, pair: str, current_time, current_rate: float, proposed_stake: float, min_stake: float | None, max_stake: float, leverage: float, entry_tag: str | None, side: str, **kwargs, ) -> float: """Compute margin stake for Freqtrade while keeping nominal sizing internal. self.sizer.compute(ctx) returns: - stake_margin: amount passed back to Freqtrade as stake_amount (margin) - real_risk: max loss (USDT) at current stop based on the nominal position When recording trades via state.record_trade_open(), always use nominal position sizes; do not confuse stake_margin with any *_nominal field. """ meta: Optional[Dict[str, Any]] = None if entry_tag: try: parsed = json.loads(entry_tag) except Exception: parsed = None if isinstance(parsed, dict) and "candidates" in parsed: selected = self._resolve_candidate_from_tag(pair, parsed, side) if selected: meta = self._normalize_entry_meta( self._candidate_meta_from_candidate(selected), side ) elif parsed is not None: meta = self._normalize_entry_meta(parsed, side) if meta is None: cached = self._pending_entry_meta.get(pair) if cached is not None: meta = self._normalize_entry_meta(cached, side) else: sig = self._last_signal.get(pair) if sig: meta = self._normalize_entry_meta({ 'sl_pct': float(sig.sl_pct), 'tp_pct': float(sig.tp_pct), 'dir': sig.direction, 'exit_profile': sig.exit_profile, 'recipe': sig.recipe, 'plan_timeframe': getattr(sig, "plan_timeframe", None), 'atr_pct': getattr(sig, "plan_atr_pct", None), }, side) if meta is None: return 0.0 sl = float(meta.get('sl_pct', 0.0)) tp = float(meta.get('tp_pct', 0.0)) direction = str(meta.get('dir')) stake, risk, bucket = self.sizer.compute( pair=pair, sl_pct=sl, tp_pct=tp, direction=direction, proposed_stake=proposed_stake, min_stake=min_stake, max_stake=max_stake, leverage=leverage, plan_atr_pct=meta.get("atr_pct"), exit_profile=meta.get("exit_profile"), bucket=meta.get("bucket"), ) if stake <= 0 or risk <= 0: return 0.0 # In backtests/hyperopt there is no asynchronous order lifecycle, so skip # reservation bookkeeping to avoid leaking reserved_risk and starving sizing. if self._is_backtest_like_runmode(): meta = meta or {} meta.update( { "sl_pct": sl, "tp_pct": tp, "stake_final": stake, "risk_final": risk, "bucket": bucket, "entry_price": current_rate, "dir": direction, } ) self._pending_entry_meta[pair] = meta return float(stake) rid = f"{pair}:{bucket}:{uuid.uuid4().hex}" self.reservation.reserve(pair, rid, risk, bucket) meta.update({ 'sl_pct': sl, 'tp_pct': tp, 'stake_final': stake, 'risk_final': risk, 'reservation_id': rid, 'bucket': bucket, 'entry_price': current_rate, 'dir': direction, }) self._pending_entry_meta[pair] = meta return float(stake) def leverage(self, *args, **kwargs) -> float: """返回配置中的强制杠杆倍数，供 Freqtrade 下单使用。""" return self.cfg.sizing.enforce_leverage def order_filled(self, pair: str, trade, order, current_time: datetime, **kwargs) -> None: """处理开仓或平仓成交事件，更新风险账本、预约状态并触发持久化。""" trade_id = str(getattr(trade, "trade_id", getattr(trade, "id", "NA"))) side = getattr(order, "ft_order_side", None) # Freqtrade 约定：entry_side / exit_side 与 ft_order_side 匹配 is_entry = side == getattr(trade, "entry_side", None) is_exit = side == getattr(trade, "exit_side", None) if is_entry: # 对 entry：尽量带上 meta，但没有也要能工作 meta = self._pending_entry_meta.pop(pair, None) opened = self.execution.on_open_filled( pair=pair, trade=trade, order=order, pending_meta=meta, tier_mgr=self.tier_mgr, ) if opened: try: self.state.get_pair_state(pair)._cooldown_last_ts = float( current_time.timestamp()) except Exception: pass if opened and self._persist_enabled: self.persist.save() elif is_exit: # 对 exit：完全不依赖 _pending_entry_meta closed = False if self._is_backtest_like_runmode() and trade_id in getattr(self, "_bt_closed_trades", set()): # 已在 confirm_trade_exit 中计数，避免重复累加 closs closed = True else: closed = self.execution.on_close_filled( pair=pair, trade=trade, order=order, tier_mgr=self.tier_mgr, ) if closed: try: self.state.get_pair_state(pair)._cooldown_last_ts = float( current_time.timestamp()) except Exception: pass if closed and self._persist_enabled: self.persist.save() else: # 其它类型（比如 reduce_only / 手动改单等），可以先忽略或简单打个日志 return def order_cancelled(self, pair: str, trade, order, current_time: datetime, **kwargs) -> None: """当订单被撤销时，释放对应预约并记录分析日志，符合 V29.1 修订 #5（仅释放预约，不回灌财政字段）。""" released, meta = self._handle_cancel_or_reject(pair) if released: trade_id = str(getattr(trade, "trade_id", getattr(trade, "id", meta.get( "reservation_id") if meta else "NA"))) if trade else meta.get("reservation_id", "NA") if meta else "NA" tag = ExitTags.ORDER_CANCELLED if ExitTags else "order_cancelled" self.analytics.log_exit(pair, trade_id, tag) def order_rejected(self, pair: str, trade, order, current_time: datetime, **kwargs) -> None: """当订单被拒绝时，与撤单路径一致，仅释放预约并写入分析日志。""" released, meta = self._handle_cancel_or_reject(pair) if released: trade_id = str(getattr(trade, "trade_id", getattr(trade, "id", meta.get( "reservation_id") if meta else "NA"))) if trade else meta.get("reservation_id", "NA") if meta else "NA" tag = ExitTags.ORDER_REJECTED if ExitTags else "order_rejected" self.analytics.log_exit(pair, trade_id, tag) def _handle_cancel_or_reject(self, pair: str) -> tuple[bool, Optional[dict[str, Any]]]: """内部工具：统一处理撤单/拒单路径，释放预约并返回是否释放及原始元数据快照。""" meta = self._pending_entry_meta.get(pair) meta_snapshot = dict(meta) if meta else None rid = meta.get("reservation_id") if meta else None released = self.execution.on_cancel_or_reject(pair, rid) if released: self._pending_entry_meta.pop(pair, None) if self._persist_enabled: self.persist.save() return released, meta_snapshot @staticmethod def _apply_leverage_pct(pct: Optional[float], trade) -> Optional[float]: """ 将基于“价格百分比”的 pct（例如 0.02 表示价格±2%） 转换为 freqtrade 期望的“含杠杆 ROI 百分比”。 - 对于期货：roi_pct = pct * trade.leverage - 对于现货：leverage=1，相当于不变 """ if pct is None: return None try: pct = float(pct) except (TypeError, ValueError): return None if pct <= 0: return None lev = getattr(trade, "leverage", 1.0) or 1.0 try: lev = float(lev) except (TypeError, ValueError): lev = 1.0 if lev <= 0: lev = 1.0 return pct * lev def custom_stoploss( self, pair: str, trade, current_time: datetime, current_rate: float, current_profit: float, after_fill: bool, **kwargs, ) -> Optional[float]: """Delegate stoploss calculation to ExitRouter rules.""" if EXIT_ROUTER is None or SLContext is None: return self.stoploss try: ctx = SLContext( pair=pair, trade=trade, now=current_time, profit=float(current_profit or 0.0), dp=self.dp, cfg=self.cfg, state=self.state, strategy=self, ) sl_pct = EXIT_ROUTER.sl_best(ctx, base_sl_pct=None) except Exception: sl_pct = None # 转成含杠杆 ROI 百分比 sl_pct = self._apply_leverage_pct(sl_pct, trade) if sl_pct is None or sl_pct <= 0: return self.stoploss return -abs(float(sl_pct)) def custom_exit( self, pair: str, trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs, ) -> Optional[str]: """ 自定义离场逻辑： 1. 先走 ExitPolicyV29 的风控 / flip 等“策略级规则”。 2. 如果没有触发，再用“持仓均价 ± N×ATR(当前K线)”价格触发离场。 """ # 1) 策略级风控 / flip / ICU / 债务风险 等 tid = str(getattr(trade, "trade_id", getattr(trade, "id", "NA"))) # reason = self.exit_policy.decide( # pair, # tid, # float(current_profit) if current_profit is not None else None, # trade=trade, # ) reason = None # 2) 如果策略级没有给出退出，再用 ATR 价格逻辑 if not reason: reason = self._atr_entry_exit_reason( pair, trade, current_time, current_rate, current_profit, ) if not reason: return None # 3) 记录退出原因，方便回测 / 复盘分析 try: if hasattr(trade, "exit_reason"): setattr(trade, "exit_reason", reason) except Exception: pass try: if hasattr(trade, "set_custom_data"): trade.set_custom_data("exit_tag", reason) except Exception: pass try: self.analytics.log_exit(pair, tid, reason) except Exception: pass return reason def custom_roi( self, pair: str, trade, current_time: datetime, trade_duration: int, entry_tag: Optional[str], side: str, **kwargs, ) -> Optional[float]: """ 自定义 ROI 回调： - 由 freqtrade 在每次循环、每个未平仓交易上调用 - 返回当前应使用的最小 ROI 阈值（比例，如 0.05 表示 +5%） - 实际阈值由 ExitRouter/ExitFacade 中的 TP 规则聚合得到 """ # 如果 Router / TPContext 尚未初始化，则回退到 minimal_roi 逻辑 if EXIT_ROUTER is None or TPContext is None: return None try: # 当前我们的 TP 规则并未使用 ctx.profit，因此这里给一个占位值即可 ctx = TPContext( pair=pair, trade=trade, now=current_time, profit=0.0, dp=self.dp, cfg=self.cfg, state=self.state, strategy=self, ) # 从所有 TP 规则中取“最保守的”（最小的正值）tp_pct tp_pct = EXIT_ROUTER.tp_best(ctx, base_tp_pct=None) except Exception: tp_pct = None # 把 tp_pct 换算成“账户收益百分比”的格式（考虑杠杆） tp_pct = self._apply_leverage_pct(tp_pct, trade) # 返回 None 表示“保持原有 minimal_roi” if tp_pct is None or tp_pct <= 0: return None # freqtrade 期望的是一个 ROI 阈值（比例），由 min_roi_reached() 与当前利润对比后决定是否离场 return float(tp_pct) def _router_sl_tp_pct( self, pair: str, trade, current_time: datetime, current_profit: float, ) -> tuple[Optional[float], Optional[float]]: """ 使用 EXIT_ROUTER + ExitFacade 计算当前 trade 对应的 sl_pct / tp_pct。 返回值为“相对开仓价的百分比（ΔP / entry_price）”，不含杠杆。 """ if EXIT_ROUTER is None or SLContext is None or TPContext is None: return None, None try: sl_ctx = SLContext( pair=pair, trade=trade, now=current_time, profit=float(current_profit or 0.0), dp=self.dp, cfg=self.cfg, state=self.state, strategy=self, ) sl_pct = EXIT_ROUTER.sl_best(sl_ctx, base_sl_pct=None) except Exception: sl_pct = None try: tp_ctx = TPContext( pair=pair, trade=trade, now=current_time, profit=float(current_profit or 0.0), dp=self.dp, cfg=self.cfg, state=self.state, strategy=self, ) tp_pct = EXIT_ROUTER.tp_best(tp_ctx, base_tp_pct=None) except Exception: tp_pct = None # 这里刻意不调用 self._apply_leverage_pct，保留为“价格百分比” return (float(sl_pct) if sl_pct and sl_pct > 0 else None, float(tp_pct) if tp_pct and tp_pct > 0 else None) def _get_current_atr_abs( self, pair: str, current_time: datetime, profile: Optional[Any] = None, ) -> Optional[float]: """ 从 DataProvider 的指标表里，按 current_time 精确地取这一根K线的 ATR 绝对值。 优先使用 profile.atr_timeframe；否则退回到策略 / 配置的主 timeframe。 """ dp = getattr(self, "dp", None) if dp is None: return None # 1) 决定 ATR 使用的 timeframe atr_tf = None if profile is not None: atr_tf = getattr(profile, "atr_timeframe", None) if not atr_tf: # cfg.timeframe 一般等于策略主周期 atr_tf = getattr(self.cfg, "timeframe", None) or self.timeframe try: analyzed = dp.get_analyzed_dataframe(pair, atr_tf) except Exception: return None df = analyzed[0] if isinstance(analyzed, (list, tuple)) else analyzed if df is None or df.empty: return None # 2) 和 atr_pct_from_dp 一样的时间对齐逻辑：df.loc[:current_time] try: upto = df.loc[:current_time] if current_time is not None else df except Exception: try: ct = ( current_time.replace(tzinfo=None) if getattr(current_time, "tzinfo", None) else current_time ) upto = df.loc[:ct] except Exception: upto = df if upto is None or upto.empty: return None row = upto.iloc[-1] # 3) 优先直接用 ATR 列；没有的话用 atr_pct * close 反推 try: if "atr" in row.index: atr_val = float(row["atr"]) if math.isfinite(atr_val) and atr_val > 0: return atr_val except Exception: pass try: atr_pct = float(row.get("atr_pct", 0.0)) close = float(row.get("close", 0.0)) if atr_pct > 0 and close > 0: atr_val = atr_pct * close if math.isfinite(atr_val) and atr_val > 0: return atr_val except Exception: return None return None def _atr_entry_exit_reason( self, pair: str, trade, current_time: datetime, current_rate: float, current_profit: float, ) -> Optional[str]: """ 仅根据“持仓均价 ± N×ATR(当前K线)”判断是否离场： - 多单: SL = entry - atr_mul_sl * ATR(t), TP = entry + atr_mul_tp * ATR(t) - 空单: SL = entry + atr_mul_sl * ATR(t), TP = entry - atr_mul_tp * ATR(t) N (atr_mul_sl / atr_mul_tp) 完全由 exit_profiles + ExitFacade 决定。 """ if self.exit_facade is None: return None # 1) 解析当前 trade 使用的 exit_profile try: profile_name, profile_def, _ = self.exit_facade.resolve_trade_plan( pair, trade, current_time ) except Exception: profile_def = None if not profile_def: return None # 2) 读取配置中的 ATR 倍数 k_sl = float(getattr(profile_def, "atr_mul_sl", 0.0) or 0.0) k_tp = getattr(profile_def, "atr_mul_tp", None) if k_tp is None or k_tp <= 0: # 和 compute_plan_from_atr 保持一致：没配置 atr_mul_tp 就 >= 2×SL k_tp = max(k_sl * 2.0, 0.0) k_tp = float(k_tp) # 3) 按 current_time 从原始指标数据里取这一根K线的 ATR 绝对值 atr_abs = self._get_current_atr_abs(pair, current_time, profile_def) if atr_abs is None or atr_abs <= 0: return None entry = float(trade.open_rate) price = float(current_rate) is_short = bool(getattr(trade, "is_short", False)) sl_abs = k_sl * atr_abs if k_sl > 0 else None tp_abs = k_tp * atr_abs if k_tp > 0 else None # 4) 按“持仓均价 ± N×ATR”比较当前价格 if not is_short: # 多单 if sl_abs is not None: sl_price = entry - sl_abs if price <= sl_price: # print(f"price_{price},atr_abs{atr_abs} sl_abs{sl_abs} current_time{current_time}\n") return "atr_entry_sl" if tp_abs is not None: tp_price = entry + tp_abs if price >= tp_price: # print(f"price_{price},atr_abs{atr_abs} tp_price{tp_price} current_time{current_time}\n") return "atr_entry_tp" else: # 空单 if sl_abs is not None: sl_price = entry + sl_abs if price >= sl_price: # print(f"price_{price},atr_abs{atr_abs} sl_abs{sl_abs} current_time{current_time}\n") return "atr_entry_sl" if tp_abs is not None: tp_price = entry - tp_abs if price <= tp_price: # print(f"price_{price},atr_abs{atr_abs} tp_price{tp_price} current_time{current_time}\n") return "atr_entry_tp" return None