# source: https://raw.githubusercontent.com/secondlinger/WinnieBari/73bedc96c87439cd3ca5ae33aca9cb01ed125468/user_data/strategies/NASOSv7_RiskPlus.py from datetime import datetime, timedelta from functools import reduce from typing import Optional, Dict, List import numpy as np import pandas as pd import talib.abstract as ta import freqtrade.vendor.qtpylib.indicators as qtpylib from pandas import DataFrame from freqtrade.persistence import Trade from freqtrade.strategy.interface import IStrategy from freqtrade.strategy import ( stoploss_from_open, merge_informative_pair, DecimalParameter, IntParameter, ) # ---- params (初期値は攻め寄り) ---- buy_params = { "base_nb_candles_buy": 8, "ewo_high": 2.403, "ewo_high_2": -5.585, "ewo_low": -14.378, "lookback_candles": 3, "low_offset": 0.984, "low_offset_2": 0.942, "profit_threshold": 1.008, "rsi_buy": 72, } sell_params = { "base_nb_candles_sell": 16, "high_offset": 1.084, "high_offset_2": 1.401, # カスタムSLの閾値は維持（伸ばす設計なのでやや攻め） "pHSL": -0.20, "pPF_1": 0.020, "pPF_2": 0.040, "pSL_1": 0.010, "pSL_2": 0.025, } def EWO(dataframe: DataFrame, ema_length=5, ema2_length=35) -> pd.Series: df = dataframe ema1 = ta.EMA(df, timeperiod=ema_length) ema2 = ta.EMA(df, timeperiod=ema2_length) emadif = (ema1 - ema2) / df['low'] * 100 return emadif class Github_secondlinger_WinnieBari__NASOSv7_RiskPlus__20250919_011108(IStrategy): INTERFACE_VERSION = 3 # ROIは低く設定（カスタムSLで伸ばす） minimal_roi = {"0": 0.01} # ベースSL（ハードストップ） stoploss = -0.35 # trailing_stopはカスタムSLと競合するので無効化（ドキュメント推奨） trailing_stop = False use_custom_stoploss = True use_exit_signal = True exit_profit_only = False exit_profit_offset = 0.01 ignore_roi_if_entry_signal = False timeframe = '5m' inf_1h = '1h' process_only_new_candles = True startup_candle_count = 300 # DCA（ポジション調整）を有効化 position_adjustment_enable = True # ---------------- Hyper-params ---------------- base_nb_candles_buy = IntParameter(2, 20, default=buy_params["base_nb_candles_buy"], space="buy", optimize=True) base_nb_candles_sell = IntParameter(2, 25, default=sell_params["base_nb_candles_sell"], space="sell", optimize=True) low_offset = DecimalParameter(0.9, 0.99, default=buy_params["low_offset"], space="buy", optimize=False) low_offset_2 = DecimalParameter(0.9, 0.99, default=buy_params["low_offset_2"], space="buy", optimize=False) high_offset = DecimalParameter(0.95, 1.1, default=sell_params["high_offset"], space="sell", optimize=True) high_offset_2 = DecimalParameter(0.99, 1.5, default=sell_params["high_offset_2"], space="sell", optimize=True) fast_ewo = 50 slow_ewo = 200 lookback_candles = IntParameter(1, 24, default=buy_params["lookback_candles"], space="buy", optimize=True) profit_threshold = DecimalParameter(1.0, 1.03, default=buy_params["profit_threshold"], space="buy", optimize=True) ewo_low = DecimalParameter(-20.0, -8.0, default=buy_params["ewo_low"], space="buy", optimize=False) ewo_high = DecimalParameter(2.0, 12.0, default=buy_params["ewo_high"], space="buy", optimize=False) ewo_high_2 = DecimalParameter(-6.0, 12.0, default=buy_params["ewo_high_2"], space="buy", optimize=False) rsi_buy = IntParameter(50, 100, default=buy_params["rsi_buy"], space="buy", optimize=False) # カスタムSLのしきい値 pHSL = DecimalParameter(-0.300, -0.080, default=sell_params["pHSL"], decimals=3, space="sell", optimize=False, load=True) pPF_1 = DecimalParameter(0.010, 0.040, default=sell_params["pPF_1"], decimals=3, space="sell", optimize=False, load=True) pSL_1 = DecimalParameter(0.008, 0.025, default=sell_params["pSL_1"], decimals=3, space="sell", optimize=False, load=True) pPF_2 = DecimalParameter(0.030, 0.100, default=sell_params["pPF_2"], decimals=3, space="sell", optimize=False, load=True) pSL_2 = DecimalParameter(0.015, 0.060, default=sell_params["pSL_2"], decimals=3, space="sell", optimize=False, load=True) # 追加パラメータ（ボラ＆ブレイクアウト） donchian_len = IntParameter(18, 55, default=20, space="buy", optimize=True) adx_th = IntParameter(18, 35, default=22, space="buy", optimize=True) atr_len = IntParameter(10, 21, default=14, space="sell", optimize=True) plot_config = { 'main_plot': { 'ma_buy': {'color': 'orange'}, 'ma_sell': {'color': 'orange'}, }, } # 簡易スリッページ保護（エグジット時） slippage_protection = { 'retries': 3, 'max_slippage': -0.02 } # ---------- レバレッジ（先物ペアに有効） ---------- def leverage(self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, side: str, **kwargs) -> float: return 2.0 # ---------- 動的ステーク ---------- def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: float | None, max_stake: float, leverage: float, entry_tag: str | None, side: str, **kwargs) -> float: # 強いブレイクアウト時は積極配分、通常は提案値 if entry_tag in ("breakout", "ewo1"): return max_stake return proposed_stake # ---------- DCA（ポジション調整） ---------- def adjust_trade_position(self, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs): # 強いモメンタム持続時に1ユニット追撃（控えめ設計） pair = trade.pair dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe is None or len(dataframe) < 2: return None last = dataframe.iloc[-1] prev = dataframe.iloc[-2] strong = (last["close"] > last["donchian_upper"]) and (last["adx"] > self.adx_th.value) and (last["ema_50_1h"] > last["ema_200_1h"]) new_signal = (last.get("enter_long", 0) == 1) and (prev.get("enter_long", 0) == 0) if strong and new_signal and trade.nr_of_successful_entries < 3: return (None, "buy", trade.stake_amount) # 1口分追加 return None # ---------- 情報足 ---------- def informative_pairs(self): pairs = self.dp.current_whitelist() informative_pairs = [(pair, self.inf_1h) for pair in pairs] return informative_pairs def informative_1h_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: assert self.dp, "DataProvider is required for multiple timeframes." informative_1h = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.inf_1h) informative_1h['ema_50'] = ta.EMA(informative_1h, timeperiod=50) informative_1h['ema_200'] = ta.EMA(informative_1h, timeperiod=200) informative_1h['adx'] = ta.ADX(informative_1h, timeperiod=14) return informative_1h # 1) 1時間足にEMA追加（既存のinformative_1h_indicatorsを拡張） def informative_1h_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: assert self.dp, "DataProvider is required for multiple timeframes." informative_1h = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.inf_1h) # 1hトレンド判定用 informative_1h['ema_50'] = ta.EMA(informative_1h, timeperiod=50) informative_1h['ema_200'] = ta.EMA(informative_1h, timeperiod=200) return informative_1h def normal_tf_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: for val in self.base_nb_candles_buy.range: dataframe[f'ma_buy_{val}'] = ta.EMA(dataframe, timeperiod=val) for val in self.base_nb_candles_sell.range: dataframe[f'ma_sell_{val}'] = ta.EMA(dataframe, timeperiod=val) dataframe['hma_50'] = qtpylib.hull_moving_average(dataframe['close'], window=50) dataframe['ema_100'] = ta.EMA(dataframe, timeperiod=100) dataframe['sma_9'] = ta.SMA(dataframe, timeperiod=9) dataframe['EWO'] = EWO(dataframe, self.fast_ewo, self.slow_ewo) dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) dataframe['rsi_fast'] = ta.RSI(dataframe, timeperiod=4) dataframe['rsi_slow'] = ta.RSI(dataframe, timeperiod=20) dataframe['atr_14'] = ta.ATR(dataframe, timeperiod=14) # 追加：ATR/ADX/Donchian dataframe['atr'] = ta.ATR(dataframe, timeperiod=self.atr_len.value) dlen = int(self.donchian_len.value) dataframe['donchian_upper'] = dataframe['high'].rolling(dlen).max() dataframe['donchian_lower'] = dataframe['low'].rolling(dlen).min() dataframe['adx'] = ta.ADX(dataframe, timeperiod=14) return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: informative_1h = self.informative_1h_indicators(dataframe, metadata) dataframe = merge_informative_pair(dataframe, informative_1h, self.timeframe, self.inf_1h, ffill=True) dataframe.rename(columns={ 'ema_50_1h': 'ema_50_1h', 'ema_200_1h': 'ema_200_1h', 'adx_1h': 'adx_1h' }, inplace=True) dataframe = self.normal_tf_indicators(dataframe, metadata) # 1h EMAを正しく参照（mergeの命名に合わせる） dataframe['ema_50_1h'] = dataframe['ema_50_1h'] dataframe['ema_200_1h'] = dataframe['ema_200_1h'] return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe['enter_long'] = 0 dataframe['buy_tag'] = None # 元の「利益しきい値」フィルタ（攻めるが、異常な直近高値追いは避ける） dont_buy = (dataframe['close_1h'].rolling(self.lookback_candles.value).max() < (dataframe['close'] * self.profit_threshold.value)) # 既存EWOパターン cond_ewo1 = ( (dataframe['rsi_fast'] < 35) & (dataframe['close'] < (dataframe[f'ma_buy_{self.base_nb_candles_buy.value}'] * self.low_offset.value)) & (dataframe['EWO'] > self.ewo_high.value) & (dataframe['rsi'] < self.rsi_buy.value) & (dataframe['volume'] > 0) & (dataframe['close'] < (dataframe[f'ma_sell_{self.base_nb_candles_sell.value}'] * self.high_offset.value)) ) cond_ewo2 = ( (dataframe['rsi_fast'] < 35) & (dataframe['close'] < (dataframe[f'ma_buy_{self.base_nb_candles_buy.value}'] * self.low_offset_2.value)) & (dataframe['EWO'] > self.ewo_high_2.value) & (dataframe['rsi'] < self.rsi_buy.value) & (dataframe['volume'] > 0) & (dataframe['close'] < (dataframe[f'ma_sell_{self.base_nb_candles_sell.value}'] * self.high_offset.value)) & (dataframe['rsi'] < 25) ) cond_ewolow = ( (dataframe['rsi_fast'] < 35) & (dataframe['close'] < (dataframe[f'ma_buy_{self.base_nb_candles_buy.value}'] * self.low_offset.value)) & (dataframe['EWO'] < self.ewo_low.value) & (dataframe['volume'] > 0) & (dataframe['close'] < (dataframe[f'ma_sell_{self.base_nb_candles_sell.value}'] * self.high_offset.value)) ) # 追加：ブレイクアウト（Donchian×ADX×1hトレンド） cond_breakout = ( (dataframe['close'] > dataframe['donchian_upper']) & (dataframe['adx'] > self.adx_th.value) & (dataframe['ema_50_1h'] > dataframe['ema_200_1h']) & (dataframe['volume'] > 0) ) dataframe.loc[cond_ewo1, ['enter_long', 'buy_tag']] = (1, 'ewo1') dataframe.loc[cond_ewo2, ['enter_long', 'buy_tag']] = (1, 'ewo2') dataframe.loc[cond_ewolow, ['enter_long', 'buy_tag']] = (1, 'ewolow') dataframe.loc[cond_breakout, ['enter_long', 'buy_tag']] = (1, 'breakout') dataframe.loc[dont_buy, 'enter_long'] = 0 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: conditions = [] # 利益を伸ばす方針なので、明確な弱含みのみでシグナル売り cond1 = ( (dataframe['close'] < dataframe['hma_50']) & (dataframe['close'] > (dataframe[f'ma_sell_{self.base_nb_candles_sell.value}'] * self.high_offset.value)) & (dataframe['volume'] > 0) & (dataframe['rsi_fast'] < dataframe['rsi_slow']) ) # ADX低下＋Donchianミドル割れ傾向（弱気） mid = (dataframe['donchian_upper'] + dataframe['donchian_lower']) / 2.0 cond2 = ( (dataframe['close'] < mid) & (dataframe['adx'] < 18) & (dataframe['volume'] > 0) ) conditions += [cond1 | cond2] dataframe['exit_long'] = 0 if conditions: dataframe.loc[reduce(lambda x, y: x | y, conditions), 'exit_long'] = 1 return dataframe # ---------- カスタムSL：段階ロック＋ATR距離のハイブリッド ---------- def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: HSL, PF_1, SL_1, PF_2, SL_2 = ( self.pHSL.value, self.pPF_1.value, self.pSL_1.value, self.pPF_2.value, self.pSL_2.value ) # ATR距離（現在値基準の追従用） df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) atr = df['atr'].iloc[-1] if df is not None and 'atr' in df.columns else 0.0 atr_dist = -(atr / max(current_rate, 1e-9)) * 2.0 # 2*ATRを目安（攻め） if current_profit > PF_2: sl_profit = SL_2 + (current_profit - PF_2) elif current_profit > PF_1: ratio = (current_profit - PF_1) / max((PF_2 - PF_1), 1e-6) sl_profit = SL_1 + (ratio * (SL_2 - SL_1)) else: sl_profit = HSL # 段階ロック（open基準）とATR追従（現在値基準）のうち厳しい方を採用 sl_open_based = stoploss_from_open(sl_profit, current_profit) return max(sl_open_based, atr_dist) # ---------- エグジット確定前の最終確認（スリッページ保護） ---------- def confirm_trade_exit(self, pair: str, trade: Trade, order_type: str, amount: float, rate: float, time_in_force: str, sell_reason: str, current_time: datetime, **kwargs) -> bool: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1] if dataframe is not None and len(dataframe) > 0 else None if last_candle is not None: if sell_reason in ['sell_signal']: if (last_candle['hma_50'] * 1.149 > last_candle['ema_100']) and ( last_candle['close'] < last_candle['ema_100'] * 0.951): return False # 簡易スリッページ保護 state = self.slippage_protection.setdefault('__pair_retries', {}) slippage = (rate / last_candle['close']) - 1 if slippage < self.slippage_protection['max_slippage']: pair_retries = state.get(pair, 0) if pair_retries < self.slippage_protection['retries']: state[pair] = pair_retries + 1 return False state[pair] = 0 return True