# source: https://raw.githubusercontent.com/dvcrn/openclaw-skills-marketplace/67b1c2373b5ad4a345051bc48b47ca2a5d8f933c/plugins/procaross--aicoin/skills/aicoin/strategies/WhaleFollowStrategy.py # Github_dvcrn_openclaw_skills_marketplace__WhaleFollowStrategy__20260612_213313 - Follow whale/institutional order flow # Powered by AiCoin's exclusive big_orders data (200+ exchanges aggregated) # # How it works: # - Standard indicators (RSI + EMA) provide base signals (works in backtest) # - In live/dry-run mode, AiCoin whale data adds an edge: # * big_orders: detect large institutional buy/sell pressure # * ls_ratio: cross-exchange long/short ratio as contrarian signal # - When whales are buying AND retail is short -> strong long signal # - When whales are selling AND retail is long -> strong short signal # # AiCoin tier required: Normal ($99/mo) for big_orders, Basic ($29/mo) for ls_ratio # Backtest: works with standard indicators only (conservative estimate) # Live: AiCoin data adds alpha on top of base signals # from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter from pandas import DataFrame import logging logger = logging.getLogger(__name__) class Github_dvcrn_openclaw_skills_marketplace__WhaleFollowStrategy__20260612_213313(IStrategy): INTERFACE_VERSION = 3 timeframe = '15m' can_short = True # ROI: take profit at these thresholds (optimized via hyperopt) minimal_roi = {"0": 0.316, "107": 0.106, "178": 0.047, "217": 0} stoploss = -0.236 use_exit_signal = False # ROI + trailing stop exits outperform signal-based exits trailing_stop = True trailing_stop_positive = 0.042 trailing_stop_positive_offset = 0.061 # Hyperopt-optimizable parameters (defaults from hyperopt optimization) rsi_buy = IntParameter(20, 40, default=34, space='buy') rsi_sell = IntParameter(60, 80, default=65, space='sell') ema_fast_len = IntParameter(5, 15, default=9, space='buy') ema_slow_len = IntParameter(15, 30, default=23, space='buy') whale_weight = DecimalParameter(0.0, 1.0, default=0.324, space='buy') # AiCoin data (updated periodically in live mode) _ac_whale_signal = 0.0 # -1 (selling) to +1 (buying) _ac_ls_ratio = 0.5 # 0-1, >0.5 = more longs _ac_last_update = 0.0 def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # ── Standard indicators (always available) ── # RSI delta = dataframe['close'].diff() gain = delta.clip(lower=0).rolling(window=14).mean() loss = (-delta.clip(upper=0)).rolling(window=14).mean() rs = gain / loss dataframe['rsi'] = 100 - (100 / (1 + rs)) # EMA dataframe['ema_fast'] = dataframe['close'].ewm( span=self.ema_fast_len.value, adjust=False).mean() dataframe['ema_slow'] = dataframe['close'].ewm( span=self.ema_slow_len.value, adjust=False).mean() # Volume SMA (for volume confirmation) dataframe['vol_sma'] = dataframe['volume'].rolling(window=20).mean() # ── AiCoin whale data (live/dry-run only) ── dataframe['whale_signal'] = 0.0 dataframe['ls_ratio'] = 0.5 if self.dp and self.dp.runmode.value in ('live', 'dry_run'): import time now = time.time() # Update AiCoin data every 5 minutes if now - self._ac_last_update > 300: self._update_aicoin_data(metadata) self._ac_last_update = now # Apply to last row (current candle) dataframe.iloc[-1, dataframe.columns.get_loc('whale_signal')] = self._ac_whale_signal dataframe.iloc[-1, dataframe.columns.get_loc('ls_ratio')] = self._ac_ls_ratio return dataframe def _update_aicoin_data(self, metadata: dict): """Fetch latest AiCoin whale data.""" try: import sys, os _sd = os.path.dirname(os.path.abspath(__file__)) if _sd not in sys.path: sys.path.insert(0, _sd) from aicoin_data import AiCoinData, ccxt_to_aicoin ac = AiCoinData(cache_ttl=300) pair = metadata.get('pair', 'BTC/USDT:USDT') exchange = self.config.get('exchange', {}).get('name', 'binance') symbol = ccxt_to_aicoin(pair, exchange) # Whale big orders try: orders = ac.big_orders(symbol) self._ac_whale_signal = self._parse_whale_signal(orders) logger.info(f"AiCoin whale signal for {pair}: {self._ac_whale_signal:.2f}") except Exception as e: logger.debug(f"AiCoin big_orders unavailable: {e}") # Long/short ratio try: ls = ac.ls_ratio() self._ac_ls_ratio = self._parse_ls_ratio(ls) logger.info(f"AiCoin L/S ratio: {self._ac_ls_ratio:.2f}") except Exception as e: logger.debug(f"AiCoin ls_ratio unavailable: {e}") except ImportError: logger.warning("aicoin_data module not found. Run ft-deploy.mjs to install.") except Exception as e: logger.warning(f"AiCoin data error: {e}") @staticmethod def _parse_whale_signal(data: dict) -> float: """Parse big_orders response into a -1 to +1 signal. Positive = whale buying pressure, Negative = whale selling pressure. """ try: # big_orders response may contain buy/sell volume or order lists # Try common response structures if 'data' in data: items = data['data'] if isinstance(items, list): buy_vol = sum(float(o.get('amount', 0)) for o in items if o.get('side', '').lower() in ('buy', 'bid', 'long')) sell_vol = sum(float(o.get('amount', 0)) for o in items if o.get('side', '').lower() in ('sell', 'ask', 'short')) total = buy_vol + sell_vol if total > 0: return (buy_vol - sell_vol) / total # -1 to +1 elif isinstance(items, dict): buy_vol = float(items.get('buyAmount', items.get('buy_amount', 0))) sell_vol = float(items.get('sellAmount', items.get('sell_amount', 0))) total = buy_vol + sell_vol if total > 0: return (buy_vol - sell_vol) / total except Exception: pass return 0.0 @staticmethod def _parse_ls_ratio(data: dict) -> float: """Parse ls_ratio response into a 0-1 value. >0.5 = more longs. AiCoin ls_ratio API returns: { data: { detail: { last: "0.87", last_day: 0.81, last_week: 2.12 } } } 'last' is the long/short ratio (longs / shorts). Convert to 0-1: ratio / (1 + ratio). e.g., 0.87 -> 0.465, 1.0 -> 0.5, 2.0 -> 0.667 """ try: detail = data.get('data', {}).get('detail', {}) if detail: ratio = float(detail.get('last', 1.0)) return max(0.0, min(1.0, ratio / (1.0 + ratio))) except Exception: pass return 0.5 def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: w = self.whale_weight.value # Long entry: uptrend + RSI low + whale buying + retail short dataframe.loc[ (dataframe['rsi'] < self.rsi_buy.value) & (dataframe['ema_fast'] > dataframe['ema_slow']) & (dataframe['volume'] > dataframe['vol_sma'] * 0.5) & # AiCoin boost: whale buying (signal > 0) or no data (signal == 0) (dataframe['whale_signal'] >= -0.3 * w) & # AiCoin boost: contrarian - retail is short (ls_ratio < 0.5) (dataframe['ls_ratio'] <= 0.5 + 0.2 * (1 - w)), 'enter_long'] = 1 # Short entry: downtrend + RSI high + whale selling + retail long dataframe.loc[ (dataframe['rsi'] > self.rsi_sell.value) & (dataframe['ema_fast'] < dataframe['ema_slow']) & (dataframe['volume'] > dataframe['vol_sma'] * 0.5) & (dataframe['whale_signal'] <= 0.3 * w) & (dataframe['ls_ratio'] >= 0.5 - 0.2 * (1 - w)), 'enter_short'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ (dataframe['rsi'] > 75), 'exit_long'] = 1 dataframe.loc[ (dataframe['rsi'] < 25), 'exit_short'] = 1 return dataframe