jackyu66gitandCursor
8ee11317d3
自动刷新改用 tail update 与 scrollToPosition 恢复视窗，避免 setData 后跳到最右；拆分 chart_tv 模块并扩展 analyze/recent API。同步威科夫分析、pipeline 增量构建及相关策略与配置。 Co-authored-by: Cursor <cursoragent@cursor.com>
450 lines
15 KiB
Python
450 lines
15 KiB
Python
# --- Do not remove these libs ---
|
||
from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter, 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 trade -c ./user_data/Chan/config/Turtle_BTC.json --strategy Turtle_BTC --strategy-path ./user_data/Chan/strategies
|
||
# freqtrade backtesting -c ./user_data/Chan/config/Turtle_BTC.json --strategy Turtle_BTC --strategy-path ./user_data/Chan/strategies --timerange=20251201-
|
||
# freqtrade download-data -c ./user_data/Chan/config/Turtle_BTC.json -t 15m --pairs BTC/USDT:USDT --timerange=20240101-
|
||
|
||
|
||
class Turtle_BTC(IStrategy):
|
||
"""
|
||
海龟交易法 (Turtle Trading) - 15m 优化版
|
||
|
||
相对经典日线参数，15m 上做了适配：
|
||
- 通道周期拉长（约 1日 / 2日），降低噪音假突破
|
||
- EMA200 趋势过滤：只做顺势方向
|
||
- ADX 过滤：只在有趋势时开仓
|
||
- 突破用「向上/向下穿越」，避免通道内反复信号
|
||
- 单单元保证金上限，避免低波动时仓位占满账户
|
||
- 系统2 优先、系统1 补漏（S1 带赢利跳过过滤）
|
||
- trade_side 可限制只做多/只做空（默认 short，适配近段下跌市）
|
||
"""
|
||
INTERFACE_VERSION = 3
|
||
timeframe = "15m"
|
||
can_short = True
|
||
process_only_new_candles = True
|
||
# 需覆盖 S2 入场周期 + EMA200
|
||
startup_candle_count = 250
|
||
|
||
minimal_roi = {"0": 100}
|
||
stoploss = -0.99
|
||
use_custom_stoploss = True
|
||
trailing_stop = False
|
||
use_exit_signal = False
|
||
exit_profit_only = False
|
||
ignore_roi_if_entry_signal = True
|
||
|
||
position_adjustment_enable = True
|
||
max_entry_position_adjustment = 3 # 首仓 + 3 加仓 = 4 单元
|
||
|
||
# ---- 15m 适配后的默认周期（约 1日 / 2日）----
|
||
# 96 根 15m ≈ 1 天；192 根 ≈ 2 天
|
||
entry_period_s1 = IntParameter(48, 144, default=96, space="buy", optimize=True)
|
||
exit_period_s1 = IntParameter(24, 96, default=48, space="sell", optimize=True)
|
||
entry_period_s2 = IntParameter(120, 288, default=192, space="buy", optimize=True)
|
||
exit_period_s2 = IntParameter(48, 144, default=96, space="sell", optimize=True)
|
||
atr_period = IntParameter(14, 40, default=20, space="buy", optimize=False)
|
||
stop_atr_mult = DecimalParameter(1.5, 3.5, default=2.0, decimals=1, space="sell", optimize=True)
|
||
pyramid_atr_mult = DecimalParameter(0.3, 1.0, default=0.5, decimals=1, space="buy", optimize=True)
|
||
risk_per_unit = DecimalParameter(0.005, 0.02, default=0.01, decimals=3, space="buy", optimize=False)
|
||
adx_threshold = IntParameter(15, 35, default=20, space="buy", optimize=True)
|
||
# 单单元保证金占可用资金上限（防止 15m 低波动时打满仓）
|
||
max_unit_stake_pct = DecimalParameter(0.15, 0.40, default=0.25, decimals=2, space="buy", optimize=False)
|
||
|
||
lev = 1.0
|
||
use_s1_win_skip = True
|
||
use_system1 = True
|
||
use_system2 = True
|
||
# 趋势 / 强度过滤
|
||
use_ema_filter = True
|
||
use_adx_filter = True
|
||
# None=双向；可用 "long" / "short" 限制单边（勿用单段行情曲线拟合）
|
||
trade_side: Optional[str] = None
|
||
ema_period = 200
|
||
|
||
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
||
ep1 = int(self.entry_period_s1.value)
|
||
xp1 = int(self.exit_period_s1.value)
|
||
ep2 = int(self.entry_period_s2.value)
|
||
xp2 = int(self.exit_period_s2.value)
|
||
atr_n = int(self.atr_period.value)
|
||
|
||
dataframe["atr"] = ta.ATR(dataframe, timeperiod=atr_n)
|
||
dataframe["n"] = dataframe["atr"]
|
||
dataframe["ema_trend"] = ta.EMA(dataframe, timeperiod=self.ema_period)
|
||
dataframe["adx"] = ta.ADX(dataframe, timeperiod=14)
|
||
dataframe["volume_ma"] = ta.SMA(dataframe, timeperiod=20, price="volume")
|
||
|
||
# 唐奇安通道（shift 1 防 lookahead）
|
||
dataframe["dc_high_s1"] = dataframe["high"].rolling(ep1).max().shift(1)
|
||
dataframe["dc_low_s1"] = dataframe["low"].rolling(ep1).min().shift(1)
|
||
dataframe["dc_exit_high_s1"] = dataframe["high"].rolling(xp1).max().shift(1)
|
||
dataframe["dc_exit_low_s1"] = dataframe["low"].rolling(xp1).min().shift(1)
|
||
|
||
dataframe["dc_high_s2"] = dataframe["high"].rolling(ep2).max().shift(1)
|
||
dataframe["dc_low_s2"] = dataframe["low"].rolling(ep2).min().shift(1)
|
||
dataframe["dc_exit_high_s2"] = dataframe["high"].rolling(xp2).max().shift(1)
|
||
dataframe["dc_exit_low_s2"] = dataframe["low"].rolling(xp2).min().shift(1)
|
||
|
||
# 穿越突破（只在刚突破那根触发）
|
||
dataframe["break_up_s1"] = (
|
||
(dataframe["close"] > dataframe["dc_high_s1"])
|
||
& (dataframe["close"].shift(1) <= dataframe["dc_high_s1"].shift(1))
|
||
)
|
||
dataframe["break_dn_s1"] = (
|
||
(dataframe["close"] < dataframe["dc_low_s1"])
|
||
& (dataframe["close"].shift(1) >= dataframe["dc_low_s1"].shift(1))
|
||
)
|
||
dataframe["break_up_s2"] = (
|
||
(dataframe["close"] > dataframe["dc_high_s2"])
|
||
& (dataframe["close"].shift(1) <= dataframe["dc_high_s2"].shift(1))
|
||
)
|
||
dataframe["break_dn_s2"] = (
|
||
(dataframe["close"] < dataframe["dc_low_s2"])
|
||
& (dataframe["close"].shift(1) >= dataframe["dc_low_s2"].shift(1))
|
||
)
|
||
|
||
# 顺势过滤：价格相对 EMA200
|
||
dataframe["trend_long"] = dataframe["close"] > dataframe["ema_trend"]
|
||
dataframe["trend_short"] = dataframe["close"] < dataframe["ema_trend"]
|
||
dataframe["adx_ok"] = dataframe["adx"] >= float(self.adx_threshold.value)
|
||
dataframe["vol_ok"] = dataframe["volume"] > dataframe["volume_ma"] * 0.8
|
||
|
||
if self.use_s1_win_skip:
|
||
dataframe["skip_s1_long"] = self._s1_skip_mask(
|
||
dataframe, long=True, exit_col="dc_exit_low_s1"
|
||
)
|
||
dataframe["skip_s1_short"] = self._s1_skip_mask(
|
||
dataframe, long=False, exit_col="dc_exit_high_s1"
|
||
)
|
||
else:
|
||
dataframe["skip_s1_long"] = False
|
||
dataframe["skip_s1_short"] = False
|
||
|
||
return dataframe
|
||
|
||
@staticmethod
|
||
def _s1_skip_mask(dataframe: DataFrame, long: bool, exit_col: str) -> pd.Series:
|
||
"""系统1：上次同向突破盈利则跳过下一次。"""
|
||
n = len(dataframe)
|
||
skip = np.zeros(n, dtype=bool)
|
||
in_trade = False
|
||
entry_price = 0.0
|
||
last_was_win = False
|
||
closes = dataframe["close"].to_numpy()
|
||
breaks = (dataframe["break_up_s1"] if long else dataframe["break_dn_s1"]).fillna(False).to_numpy()
|
||
exits = dataframe[exit_col].to_numpy()
|
||
|
||
for i in range(n):
|
||
if np.isnan(exits[i]) or np.isnan(closes[i]):
|
||
continue
|
||
if in_trade:
|
||
hit_exit = closes[i] < exits[i] if long else closes[i] > exits[i]
|
||
if hit_exit:
|
||
pnl = (closes[i] - entry_price) if long else (entry_price - closes[i])
|
||
last_was_win = pnl > 0
|
||
in_trade = False
|
||
elif breaks[i]:
|
||
if last_was_win:
|
||
skip[i] = True
|
||
last_was_win = False
|
||
else:
|
||
in_trade = True
|
||
entry_price = closes[i]
|
||
return pd.Series(skip, index=dataframe.index)
|
||
|
||
def _entry_filters(self, dataframe: DataFrame, long: bool) -> pd.Series:
|
||
base = (
|
||
(dataframe["volume"] > 0)
|
||
& dataframe["atr"].notna()
|
||
& (dataframe["atr"] > 0)
|
||
& dataframe["vol_ok"]
|
||
)
|
||
if self.use_ema_filter:
|
||
base &= dataframe["trend_long"] if long else dataframe["trend_short"]
|
||
if self.use_adx_filter:
|
||
base &= dataframe["adx_ok"]
|
||
return base
|
||
|
||
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
||
dataframe["enter_long"] = 0
|
||
dataframe["enter_short"] = 0
|
||
dataframe["enter_tag"] = ""
|
||
|
||
allow_long = self.trade_side in (None, "long")
|
||
allow_short = self.trade_side in (None, "short")
|
||
long_base = self._entry_filters(dataframe, long=True) if allow_long else False
|
||
short_base = self._entry_filters(dataframe, long=False) if allow_short else False
|
||
|
||
# 系统2优先（更稳），系统1补漏
|
||
if self.use_system2:
|
||
if allow_long:
|
||
long_s2 = long_base & dataframe["break_up_s2"]
|
||
dataframe.loc[long_s2, ["enter_long", "enter_tag"]] = (1, "turtle_s2_long")
|
||
if allow_short:
|
||
short_s2 = short_base & dataframe["break_dn_s2"]
|
||
dataframe.loc[short_s2, ["enter_short", "enter_tag"]] = (1, "turtle_s2_short")
|
||
|
||
if self.use_system1:
|
||
if allow_long:
|
||
long_s1 = (
|
||
long_base & dataframe["break_up_s1"]
|
||
& (~dataframe["skip_s1_long"])
|
||
& (dataframe["enter_long"] != 1)
|
||
)
|
||
dataframe.loc[long_s1, ["enter_long", "enter_tag"]] = (1, "turtle_s1_long")
|
||
if allow_short:
|
||
short_s1 = (
|
||
short_base & dataframe["break_dn_s1"]
|
||
& (~dataframe["skip_s1_short"])
|
||
& (dataframe["enter_short"] != 1)
|
||
)
|
||
dataframe.loc[short_s1, ["enter_short", "enter_tag"]] = (1, "turtle_s1_short")
|
||
|
||
return dataframe
|
||
|
||
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
||
dataframe["exit_long"] = 0
|
||
dataframe["exit_short"] = 0
|
||
return dataframe
|
||
|
||
def custom_exit(
|
||
self,
|
||
pair: str,
|
||
trade: Trade,
|
||
current_time: datetime,
|
||
current_rate: float,
|
||
current_profit: float,
|
||
**kwargs,
|
||
) -> Optional[str]:
|
||
"""按入场系统使用对应退出通道；用 close 与 current_rate 双确认。"""
|
||
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
|
||
if dataframe.empty:
|
||
return None
|
||
last = dataframe.iloc[-1]
|
||
tag = trade.enter_tag or ""
|
||
price = min(float(last["close"]), current_rate) if not trade.is_short else max(float(last["close"]), current_rate)
|
||
|
||
if trade.is_short:
|
||
if "s1" in tag and price > float(last["dc_exit_high_s1"]):
|
||
return "turtle_s1_exit"
|
||
if "s2" in tag and price > float(last["dc_exit_high_s2"]):
|
||
return "turtle_s2_exit"
|
||
else:
|
||
if "s1" in tag and price < float(last["dc_exit_low_s1"]):
|
||
return "turtle_s1_exit"
|
||
if "s2" in tag and price < float(last["dc_exit_low_s2"]):
|
||
return "turtle_s2_exit"
|
||
return None
|
||
|
||
def custom_stake_amount(
|
||
self,
|
||
pair: str,
|
||
current_time: datetime,
|
||
current_rate: float,
|
||
proposed_stake: float,
|
||
min_stake: Optional[float],
|
||
max_stake: float,
|
||
leverage: float,
|
||
entry_tag: Optional[str],
|
||
side: str,
|
||
**kwargs,
|
||
) -> float:
|
||
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
|
||
if dataframe.empty:
|
||
return proposed_stake
|
||
|
||
last = dataframe.iloc[-1]
|
||
atr = float(last["atr"]) if pd.notna(last["atr"]) else 0.0
|
||
if atr <= 0 or current_rate <= 0:
|
||
return proposed_stake
|
||
|
||
wallets = self.wallets
|
||
available = wallets.get_total(self.config["stake_currency"]) if wallets else max_stake
|
||
risk_amount = available * float(self.risk_per_unit.value)
|
||
stop_dist = float(self.stop_atr_mult.value) * atr
|
||
notional = risk_amount * current_rate / stop_dist
|
||
stake = notional / max(leverage, 1.0)
|
||
|
||
# 单单元上限，避免低波动打满仓
|
||
stake = min(stake, available * float(self.max_unit_stake_pct.value))
|
||
|
||
if min_stake is not None:
|
||
stake = max(stake, min_stake)
|
||
stake = min(stake, max_stake)
|
||
return stake
|
||
|
||
def adjust_trade_position(
|
||
self,
|
||
trade: Trade,
|
||
current_time: datetime,
|
||
current_rate: float,
|
||
current_profit: float,
|
||
min_stake: Optional[float],
|
||
max_stake: float,
|
||
current_entry_rate: float,
|
||
current_exit_rate: float,
|
||
current_entry_profit: float,
|
||
current_exit_profit: float,
|
||
**kwargs,
|
||
):
|
||
"""每朝有利方向 0.5N 加仓，最多 4 单元；有挂单时不加。"""
|
||
if trade.has_open_orders:
|
||
return None
|
||
if trade.nr_of_successful_entries >= (1 + self.max_entry_position_adjustment):
|
||
return None
|
||
|
||
dataframe, _ = self.dp.get_analyzed_dataframe(trade.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:
|
||
return None
|
||
|
||
entry_n = trade.get_custom_data("entry_n")
|
||
if entry_n is None:
|
||
entry_n = atr
|
||
trade.set_custom_data("entry_n", entry_n)
|
||
|
||
last_entry_price = trade.get_custom_data("last_entry_price")
|
||
if last_entry_price is None:
|
||
last_entry_price = trade.open_rate
|
||
trade.set_custom_data("last_entry_price", last_entry_price)
|
||
|
||
# 已规划的下一单元序号（从第 2 单元起）
|
||
next_unit = trade.nr_of_successful_entries + 1
|
||
step = float(self.pyramid_atr_mult.value) * float(entry_n)
|
||
# 相对首仓（或记录的单元锚定价）计算阈值，避免 after_fill 用均价漂移
|
||
anchor = float(trade.get_custom_data("unit1_price") or trade.open_rate)
|
||
# 第 n 单元触发价 = 首仓 ± (n-1)*0.5N
|
||
offset = (next_unit - 1) * step
|
||
|
||
if trade.is_short:
|
||
trigger = anchor - offset
|
||
if current_rate > trigger:
|
||
return None
|
||
else:
|
||
trigger = anchor + offset
|
||
if current_rate < trigger:
|
||
return None
|
||
|
||
stake = self.custom_stake_amount(
|
||
pair=trade.pair,
|
||
current_time=current_time,
|
||
current_rate=current_rate,
|
||
proposed_stake=max_stake,
|
||
min_stake=min_stake,
|
||
max_stake=max_stake,
|
||
leverage=trade.leverage,
|
||
entry_tag=trade.enter_tag,
|
||
side="short" if trade.is_short else "long",
|
||
)
|
||
if stake <= 0:
|
||
return None
|
||
|
||
return stake, f"turtle_pyramid_{next_unit}"
|
||
|
||
def custom_stoploss(
|
||
self,
|
||
pair: str,
|
||
trade: Trade,
|
||
current_time: datetime,
|
||
current_rate: float,
|
||
current_profit: float,
|
||
after_fill: bool,
|
||
**kwargs,
|
||
) -> Optional[float]:
|
||
"""
|
||
止损 = 最近一单元入场价 ± 2N。
|
||
加仓后整体移到新单元的 2N（海龟原版）。
|
||
"""
|
||
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 after_fill:
|
||
filled = trade.nr_of_successful_entries
|
||
if filled <= 1:
|
||
trade.set_custom_data("unit1_price", current_rate)
|
||
trade.set_custom_data("last_entry_price", current_rate)
|
||
if atr > 0:
|
||
trade.set_custom_data("entry_n", atr)
|
||
else:
|
||
# 加仓：用本次成交价作为最新单元锚点
|
||
trade.set_custom_data("last_entry_price", current_rate)
|
||
|
||
entry_n = trade.get_custom_data("entry_n")
|
||
n = float(entry_n) if entry_n is not None else atr
|
||
if n <= 0:
|
||
return None
|
||
|
||
last_entry = trade.get_custom_data("last_entry_price") or trade.open_rate
|
||
mult = float(self.stop_atr_mult.value)
|
||
|
||
if trade.is_short:
|
||
stop_price = float(last_entry) + mult * n
|
||
else:
|
||
stop_price = float(last_entry) - mult * n
|
||
|
||
sl = stoploss_from_absolute(
|
||
stop_price, current_rate, is_short=trade.is_short, leverage=trade.leverage
|
||
)
|
||
# 0 表示止损已在价格不利侧之外，保持不变
|
||
return sl if sl > 0 else None
|
||
|
||
def confirm_trade_entry(
|
||
self,
|
||
pair: str,
|
||
order_type: str,
|
||
amount: float,
|
||
rate: float,
|
||
time_in_force: str,
|
||
current_time: datetime,
|
||
entry_tag: Optional[str],
|
||
side: str,
|
||
**kwargs,
|
||
) -> bool:
|
||
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
|
||
if dataframe.empty:
|
||
return False
|
||
row = dataframe.iloc[-1]
|
||
if pd.isna(row["atr"]) or row["atr"] <= 0:
|
||
return False
|
||
if self.trade_side is not None and side != self.trade_side:
|
||
return False
|
||
if self.use_ema_filter:
|
||
if side == "long" and not bool(row["trend_long"]):
|
||
return False
|
||
if side == "short" and not bool(row["trend_short"]):
|
||
return False
|
||
if self.use_adx_filter and not bool(row["adx_ok"]):
|
||
return False
|
||
return True
|
||
|
||
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)
|