# --- Do not remove these libs ---
"""
Wyckoff BTC V1.0 BASELINE — FROZEN

Status: BASELINE FROZEN
Evidence: PASS (+ Limited Evidence, N=20)
Cost Adjusted: PASS (net PF 1.45 @ fee+slip 5bps)
Risk: small sample — 目标积累 N>=50 再谈规模

Branch A: Spring Reversal
  8h bias + 4h structure + 1h Spring/UTAD
  Range disabled（regime_mode=trend）
  ATR + 结构止损
  setup_type: SPRING / UTAD

证据: user_data/Chan/scripts/wyckoff_v1_baseline_phase2.json
LPS 是独立 Setup 研究，禁止并入本文件调参。
"""
from freqtrade.strategy import (
	IStrategy, IntParameter, DecimalParameter, CategoricalParameter,
	merge_informative_pair, stoploss_from_open, stoploss_from_absolute,
)
from freqtrade.persistence import Trade
import talib.abstract as ta
from pandas import DataFrame
import pandas as pd
import numpy as np
from datetime import datetime
from typing import Optional
import logging

logger = logging.getLogger(__name__)

# freqtrade backtesting -c ./user_data/Chan/config/Wyckoff_BTC_V1_BASELINE.json \
#   --strategy Wyckoff_BTC_V1_BASELINE --strategy-path ./user_data/Chan/strategies --timerange=20230101-


class Wyckoff_BTC_V1_BASELINE(IStrategy):
	"""冻结基线：Spring 反转。禁止继续调参；对比实验请用独立分支。"""
	INTERFACE_VERSION = 3
	STRATEGY_VERSION = "V1.0_BASELINE"
	SETUP_FAMILY = "SPRING"

	timeframe = "1h"
	structure_timeframe = "4h"
	bias_timeframe: Optional[str] = "8h"
	use_bias_filter = True
	# trend = bull|bear only（Range disabled — 理论一致性约束，非调参）
	regime_mode: str = "trend"

	can_short = True
	process_only_new_candles = True
	startup_candle_count = 220

	minimal_roi = {
		"0": 0.10,
		"1440": 0.05,
		"4320": 0.025,
		"10080": 0,
	}
	stoploss = -0.10
	use_custom_stoploss = True
	trailing_stop = True
	trailing_stop_positive = 0.02
	trailing_stop_positive_offset = 0.04
	trailing_only_offset_is_reached = True
	use_exit_signal = True
	exit_profit_only = False

	# ---- 冻结默认值（optimize=False）----
	range_lookback = IntParameter(12, 48, default=24, space="buy", optimize=False)
	spring_pierce_pct = DecimalParameter(0.001, 0.012, default=0.004, decimals=3, space="buy", optimize=False)
	vol_spike_mult = DecimalParameter(1.1, 2.5, default=1.8, decimals=1, space="buy", optimize=False)
	adx_min = IntParameter(10, 28, default=14, space="buy", optimize=False)
	tr_pos_long_max = DecimalParameter(0.35, 0.55, default=0.45, decimals=2, space="buy", optimize=False)
	tr_pos_short_min = DecimalParameter(0.45, 0.65, default=0.55, decimals=2, space="buy", optimize=False)
	atr_sl_mult = DecimalParameter(1.2, 3.5, default=1.5, decimals=1, space="sell", optimize=False)
	atr_sl_min = DecimalParameter(0.012, 0.04, default=0.018, decimals=3, space="sell", optimize=False)
	atr_sl_max = DecimalParameter(0.05, 0.12, default=0.08, decimals=2, space="sell", optimize=False)
	time_stop_hours = IntParameter(48, 240, default=120, space="sell", optimize=False)

	# Branch A：仅 Spring / UTAD
	use_spring_sig = CategoricalParameter([True, False], default=True, space="buy", optimize=False)
	use_utad_sig = CategoricalParameter([True, False], default=True, space="buy", optimize=False)
	use_sos_sig = CategoricalParameter([True, False], default=False, space="buy", optimize=False)
	use_sow_sig = CategoricalParameter([True, False], default=False, space="buy", optimize=False)

	lev = 1.0

	def informative_pairs(self):
		pairs = self.dp.current_whitelist() if self.dp else []
		tfs = {self.structure_timeframe}
		if self.bias_timeframe and self.use_bias_filter:
			tfs.add(self.bias_timeframe)
		return [(pair, tf) for pair in pairs for tf in tfs]

	def _add_wyckoff_structure(self, df: DataFrame) -> DataFrame:
		lb = int(self.range_lookback.value)

		df["atr"] = ta.ATR(df, timeperiod=14)
		df["ema50"] = ta.EMA(df, timeperiod=50)
		df["ema200"] = ta.EMA(df, timeperiod=200)
		df["adx"] = ta.ADX(df, timeperiod=14)
		df["rsi"] = ta.RSI(df, timeperiod=14)
		df["volume_ma"] = ta.SMA(df, timeperiod=20, price="volume")

		df["tr_high"] = df["high"].rolling(lb).max()
		df["tr_low"] = df["low"].rolling(lb).min()
		df["tr_mid"] = (df["tr_high"] + df["tr_low"]) / 2.0
		df["tr_width"] = (df["tr_high"] - df["tr_low"]) / df["tr_mid"].replace(0, np.nan)
		df["tr_width_ma"] = df["tr_width"].rolling(lb).mean()

		rng = (df["tr_high"] - df["tr_low"]).replace(0, np.nan)
		df["tr_pos"] = (df["close"] - df["tr_low"]) / rng

		df["in_range"] = (df["tr_width"] < df["tr_width_ma"] * 1.35) & (df["adx"] < 28)
		df["ema50_slope"] = df["ema50"] - df["ema50"].shift(8)
		df["prior_down"] = df["ema50_slope"].shift(lb) < 0
		df["prior_up"] = df["ema50_slope"].shift(lb) > 0

		down_bar = df["close"] < df["open"]
		up_bar = df["close"] > df["open"]
		vol_down = np.where(down_bar, df["volume"], np.nan)
		vol_up = np.where(up_bar, df["volume"], np.nan)
		df["vol_down_ma"] = pd.Series(vol_down, index=df.index).rolling(10, min_periods=3).mean()
		df["vol_up_ma"] = pd.Series(vol_up, index=df.index).rolling(10, min_periods=3).mean()
		df["effort_absorb"] = (
			df["vol_down_ma"].notna()
			& df["vol_up_ma"].notna()
			& (df["vol_up_ma"] > df["vol_down_ma"] * 1.05)
		)

		df["accum_ctx"] = (
			df["in_range"]
			& (df["prior_down"] | (df["close"] < df["ema50"]))
			& (df["tr_pos"] < float(self.tr_pos_long_max.value))
		)
		df["distrib_ctx"] = (
			df["in_range"]
			& (df["prior_up"] | (df["close"] > df["ema50"]))
			& (df["tr_pos"] > float(self.tr_pos_short_min.value))
		)
		df["bull_bias"] = (df["close"] > df["ema200"]) & (df["ema50"] > df["ema200"])
		df["bear_bias"] = (df["close"] < df["ema200"]) & (df["ema50"] < df["ema200"])
		df["vol_spike"] = df["volume"] > df["volume_ma"] * float(self.vol_spike_mult.value)
		return df

	def _merge_tf(self, dataframe: DataFrame, pair: str, tf: str) -> DataFrame:
		inf = self.dp.get_pair_dataframe(pair=pair, timeframe=tf)
		inf = self._add_wyckoff_structure(inf)
		keep = [
			"date", "atr", "ema50", "ema200", "adx", "rsi",
			"tr_high", "tr_low", "tr_mid", "tr_width", "tr_pos",
			"in_range", "accum_ctx", "distrib_ctx",
			"vol_spike", "effort_absorb", "prior_down", "prior_up",
			"bull_bias", "bear_bias",
		]
		inf = inf[[c for c in keep if c in inf.columns]].copy()
		return merge_informative_pair(dataframe, inf, self.timeframe, tf, ffill=True)

	def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
		pair = metadata["pair"]
		stf = self.structure_timeframe
		dataframe = self._merge_tf(dataframe, pair, stf)

		btf = self.bias_timeframe
		if btf and self.use_bias_filter and btf != stf:
			dataframe = self._merge_tf(dataframe, pair, btf)

		ss = f"_{stf}"
		dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)
		dataframe["ema21"] = ta.EMA(dataframe, timeperiod=21)
		dataframe["ema50"] = ta.EMA(dataframe, timeperiod=50)
		dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
		dataframe["volume_ma"] = ta.SMA(dataframe, timeperiod=20, price="volume")
		dataframe["vol_ok"] = dataframe["volume"] > dataframe["volume_ma"] * float(self.vol_spike_mult.value)

		tr_high = dataframe[f"tr_high{ss}"]
		tr_low = dataframe[f"tr_low{ss}"]
		pierce = float(self.spring_pierce_pct.value)

		accum_soft = (
			dataframe[f"accum_ctx{ss}"].fillna(False).astype(bool)
			| (
				dataframe[f"in_range{ss}"].fillna(False).astype(bool)
				& dataframe[f"prior_down{ss}"].fillna(False).astype(bool)
				& (dataframe[f"tr_pos{ss}"] < float(self.tr_pos_long_max.value))
			)
		)
		distrib_soft = (
			dataframe[f"distrib_ctx{ss}"].fillna(False).astype(bool)
			| (
				dataframe[f"in_range{ss}"].fillna(False).astype(bool)
				& dataframe[f"prior_up{ss}"].fillna(False).astype(bool)
				& (dataframe[f"tr_pos{ss}"] > float(self.tr_pos_short_min.value))
			)
		)

		if btf and self.use_bias_filter:
			bs = f"_{btf}" if btf != stf else ss
			if f"bear_bias{bs}" in dataframe.columns:
				dataframe["bias_long_ok"] = ~dataframe[f"bear_bias{bs}"].fillna(False).astype(bool)
				dataframe["bias_short_ok"] = ~dataframe[f"bull_bias{bs}"].fillna(False).astype(bool)
			else:
				dataframe["bias_long_ok"] = True
				dataframe["bias_short_ok"] = True
		else:
			dataframe["bias_long_ok"] = True
			dataframe["bias_short_ok"] = True

		vol_mild = dataframe["volume"] > dataframe["volume_ma"] * max(1.1, float(self.vol_spike_mult.value) * 0.85)

		dataframe["spring"] = (
			tr_low.notna()
			& (dataframe["low"] < tr_low * (1.0 - pierce))
			& (dataframe["close"] > tr_low)
			& (dataframe["close"] > dataframe["open"])
			& accum_soft
			& vol_mild
			& (dataframe["rsi"] < 58)
			& dataframe["bias_long_ok"]
		)
		dataframe["utad"] = (
			tr_high.notna()
			& (dataframe["high"] > tr_high * (1.0 + pierce))
			& (dataframe["close"] < tr_high)
			& (dataframe["close"] < dataframe["open"])
			& distrib_soft
			& vol_mild
			& (dataframe["rsi"] > 42)
			& dataframe["bias_short_ok"]
		)
		# 基线不进 SOS/SOW；保留列供 exit 参考
		dataframe["sos"] = False
		dataframe["sow"] = False

		for col in ["spring", "utad", "sos", "sow", "vol_ok", "bias_long_ok", "bias_short_ok"]:
			dataframe[col] = dataframe[col].fillna(False).astype(bool)
		dataframe["setup_type"] = ""
		dataframe.loc[dataframe["spring"], "setup_type"] = "SPRING_LONG"
		dataframe.loc[dataframe["utad"], "setup_type"] = "UTAD_SHORT"
		return dataframe

	def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
		dataframe["enter_long"] = 0
		dataframe["enter_short"] = 0
		dataframe["enter_tag"] = ""

		vol_ok = dataframe["volume"] > 0

		# 分开标签：禁止把 SPRING / UTAD 混成同一统计桶
		if bool(self.use_spring_sig.value):
			cond = vol_ok & dataframe["spring"]
			dataframe.loc[cond, ["enter_long", "enter_tag"]] = (1, "SPRING_LONG")

		if bool(self.use_utad_sig.value):
			cond = vol_ok & dataframe["utad"]
			dataframe.loc[cond, ["enter_short", "enter_tag"]] = (1, "UTAD_SHORT")

		self._apply_regime_filter(dataframe)
		return dataframe

	def _apply_regime_filter(self, dataframe: DataFrame) -> None:
		rm = getattr(self, "regime_mode", "all")
		if rm == "all" or not self.bias_timeframe:
			return
		bs = f"_{self.bias_timeframe}"
		bc, ec = f"bull_bias{bs}", f"bear_bias{bs}"
		if bc not in dataframe.columns or ec not in dataframe.columns:
			return
		bull = dataframe[bc].fillna(False).astype(bool)
		bear = dataframe[ec].fillna(False).astype(bool)
		both = bull & bear
		bull, bear = bull & ~both, bear & ~both
		range_m = (~bull) & (~bear)
		if rm == "bull":
			mask = ~bull
		elif rm == "bear":
			mask = ~bear
		elif rm == "range":
			mask = ~range_m
		elif rm == "trend":
			mask = range_m  # Range disabled
		else:
			return
		dataframe.loc[mask, ["enter_long", "enter_short"]] = (0, 0)
		dataframe.loc[mask, "enter_tag"] = ""

	def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
		dataframe["exit_long"] = 0
		dataframe["exit_short"] = 0
		dataframe["exit_tag"] = ""
		ss = f"_{self.structure_timeframe}"

		exit_long = dataframe["utad"] | (
			dataframe[f"distrib_ctx{ss}"].fillna(False).astype(bool)
			& (dataframe["close"] < dataframe["ema21"])
			& (dataframe["rsi"] < 45)
		)
		exit_short = dataframe["spring"] | (
			dataframe[f"accum_ctx{ss}"].fillna(False).astype(bool)
			& (dataframe["close"] > dataframe["ema21"])
			& (dataframe["rsi"] > 55)
		)
		dataframe.loc[exit_long, ["exit_long", "exit_tag"]] = (1, "wyckoff_phase_flip")
		dataframe.loc[exit_short, ["exit_short", "exit_tag"]] = (1, "wyckoff_phase_flip")
		return dataframe

	def custom_stoploss(
		self, pair: str, trade: Trade, current_time: datetime,
		current_rate: float, current_profit: float, after_fill: bool, **kwargs,
	) -> Optional[float]:
		dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
		if dataframe.empty:
			return None
		last = dataframe.iloc[-1]
		atr = float(last["atr"]) if pd.notna(last["atr"]) else 0.0
		if atr <= 0 or trade.open_rate <= 0:
			return None

		atr_dist = float(self.atr_sl_mult.value) * atr
		tag = trade.enter_tag or ""
		buffer = atr * 0.15

		if after_fill and trade.get_custom_data("struct_stop") is None:
			if trade.is_short:
				trade.set_custom_data("struct_stop", float(last["high"]) + buffer)
			else:
				trade.set_custom_data("struct_stop", float(last["low"]) - buffer)

		struct = trade.get_custom_data("struct_stop")
		if trade.is_short:
			atr_stop = trade.open_rate + atr_dist
			stop_price = min(atr_stop, float(struct)) if struct is not None else atr_stop
		else:
			atr_stop = trade.open_rate - atr_dist
			stop_price = max(atr_stop, float(struct)) if struct is not None else atr_stop

		raw = abs(trade.open_rate - stop_price) / trade.open_rate
		raw = min(max(raw, float(self.atr_sl_min.value)), float(self.atr_sl_max.value))
		if struct is not None and tag in (
			"SPRING_LONG", "UTAD_SHORT", "SPRING", "UTAD", "wyckoff_spring", "wyckoff_utad",
		):
			sl = stoploss_from_absolute(
				stop_price, current_rate, is_short=trade.is_short, leverage=trade.leverage
			)
			return sl if sl and sl > 0 else None
		return stoploss_from_open(
			-raw, current_profit, is_short=trade.is_short, leverage=trade.leverage
		) or None

	def custom_exit(
		self, pair: str, trade: Trade, current_time: datetime,
		current_rate: float, current_profit: float, **kwargs,
	) -> Optional[str]:
		hours = (current_time - trade.open_date_utc).total_seconds() / 3600
		if hours > float(self.time_stop_hours.value) and current_profit < 0:
			return "wyckoff_time_stop"
		if hours > float(self.time_stop_hours.value) * 2:
			return "wyckoff_time_stop_max"
		return None

	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:
		return min(self.lev, max_leverage)