# source: https://raw.githubusercontent.com/honginp/vibefoundry-dev/2c7696ea251de507030e8b4b833358342e3565a4/legacy/freqtrade_dump/AlphaEngineStrategy.py import numpy as np from scipy.stats import norm from freqtrade.strategy import IStrategy from pandas import DataFrame from typing import Optional, Tuple class Runner: def __init__(self, delta_up: float, delta_down: float, d_star_up: float, d_star_down: float): self.delta_up = delta_up self.delta_down = delta_down self.d_star_up = d_star_up self.d_star_down = d_star_down self.mode = 1 # 1: up, -1: down self.extreme = None self.reference = None self.expected_dc_level = None self.expected_os_level = None def run(self, price: float) -> int: if self.extreme is None: self.extreme = self.reference = price self._update_expected_levels() return 0 if self.mode == -1: if price >= self.expected_dc_level: self.mode = 1 self.extreme = self.reference = price self._update_expected_levels() return 1 # DC up if price < self.extreme: self.extreme = price self._update_expected_levels() if price < self.expected_os_level: self.reference = self.extreme self._update_expected_levels() return -2 # OS down elif self.mode == 1: if price <= self.expected_dc_level: self.mode = -1 self.extreme = self.reference = price self._update_expected_levels() return -1 # DC down if price > self.extreme: self.extreme = price self._update_expected_levels() if price > self.expected_os_level: self.reference = self.extreme self._update_expected_levels() return 2 # OS up return 0 def _update_expected_levels(self): if self.mode == -1: self.expected_dc_level = self.extreme * np.exp(self.delta_up) self.expected_os_level = self.reference * np.exp(-self.d_star_down) else: self.expected_dc_level = self.extreme * np.exp(-self.delta_down) self.expected_os_level = self.reference * np.exp(self.d_star_up) def get_expected_dc_level(self) -> float: return self.expected_dc_level def get_expected_os_level(self) -> float: return self.expected_os_level class LocalLiquidity: def __init__(self, delta: float, delta_star: float, alpha: float): self.delta = delta self.delta_star = delta_star self.alpha = alpha self.alpha_weight = np.exp(-2.0 / (alpha + 1.0)) self.surp = 0.0 self.liq = 0.5 self.H1 = -np.exp(-delta_star/delta) * np.log(np.exp(-delta_star/delta)) - \ (1.0 - np.exp(-delta_star/delta)) * np.log(1.0 - np.exp(-delta_star/delta)) self.H2 = np.exp(-delta_star/delta) * np.log(np.exp(-delta_star/delta))**2 + \ (1.0 - np.exp(-delta_star/delta)) * np.log(1.0 - np.exp(-delta_star/delta))**2 - self.H1**2 def compute(self, event: int) -> float: if event != 0: surprise = 0.08338161 if abs(event) == 1 else 2.525729 self.surp = self.alpha_weight * surprise + (1.0 - self.alpha_weight) * self.surp delta = np.sqrt(self.alpha) * (self.surp - self.H1) / np.sqrt(self.H2) self.liq = 1.0 - norm.cdf(delta) return self.liq class Github_honginp_vibefoundry_dev__AlphaEngineStrategy__20260712_173606(IStrategy): INTERFACE_VERSION = 3 minimal_roi = {"0": 0.05} # 5% profit target stoploss = -0.10 # 10% stop loss timeframe = "5m" def __init__(self, config: dict) -> None: super().__init__(config) self.base_unit_size = 1.0 # Base position size per agent self.profit_target = 0.05 # 5% target PnL self.agents = [] deltas = [0.0025, 0.005, 0.01, 0.015] # Multi-scale thresholds for delta in deltas: # Normal runner r0 = Runner(delta, delta, delta, delta) # Adjusted for inventory 15-30 r1 = Runner(0.75 * delta, 1.5 * delta, 0.75 * delta, 0.75 * delta) # Adjusted for inventory >30 r2 = Runner(0.5 * delta, 2.0 * delta, 0.5 * delta, 0.5 * delta) liquidity = LocalLiquidity(delta, delta * 2.525729, 50.0) self.agents.append({ 'delta': delta, 'runners': [r0, r1, r2], 'liquidity': liquidity, 'virtual_position': 0.0, 'entry_price': None }) def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe['enter_long'] = 0 dataframe['exit_long'] = 0 dataframe['increase_position'] = 0.0 dataframe['decrease_position'] = 0.0 for i in range(1, len(dataframe)): price = dataframe['close'].iloc[i] low = dataframe['low'].iloc[i] high = dataframe['high'].iloc[i] inventory = self._get_inventory(metadata) for agent in self.agents: runner = self._select_runner(agent, inventory) event = runner.run(price) liq = agent['liquidity'].compute(event) # Cascading: Enter or increase position if event < 0 and liq < 0.5 and agent['virtual_position'] == 0: size = self._compute_size(liq) agent['virtual_position'] = size agent['entry_price'] = price dataframe['increase_position'].iloc[i] += size if inventory == 0: dataframe['enter_long'].iloc[i] = 1 # De-cascading: Reduce position elif event > 0 and agent['virtual_position'] > 0: if price >= agent['entry_price'] * (1 + agent['delta']): agent['virtual_position'] = 0 agent['entry_price'] = None dataframe['decrease_position'].iloc[i] += size # Exit if profit target reached open_trades = self.dp.trades_open if open_trades: trade = open_trades[0] total_pnl = (trade.realized_profit + trade.calc_profit_ratio(dataframe['close'].iloc[-1])) / trade.stake_amount if total_pnl >= self.profit_target: dataframe['exit_long'].iloc[-1] = 1 return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[dataframe['enter_long'] == 1, 'enter_long'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[dataframe['exit_long'] == 1, 'exit_long'] = 1 return dataframe def adjust_trade_position(self, trade: 'Trade', current_time: 'datetime', current_rate: float, current_profit: float, min_stake: float, max_stake: float, current_entry_rate: float, current_exit_rate: float, current_profit_ratio: float) -> Optional[float]: df, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) if df.empty: return None last_candle = df.iloc[-1] desired_amount = last_candle['increase_position'] - last_candle['decrease_position'] current_amount = trade.amount diff = desired_amount - current_amount if diff > 0: stake_amount = min(diff * current_rate, max_stake - trade.stake_amount) if stake_amount > min_stake: return stake_amount elif diff < 0: stake_amount = max(diff * current_rate, -trade.stake_amount) if abs(stake_amount) > min_stake: return stake_amount return None def _select_runner(self, agent: dict, inventory: float) -> Runner: if inventory < 15: return agent['runners'][0] elif 15 <= inventory < 30: return agent['runners'][1] else: return agent['runners'][2] def _compute_size(self, liq: float) -> float: if liq < 0.1: return self.base_unit_size * 0.1 elif liq < 0.5: return self.base_unit_size * 0.5 return self.base_unit_size def _get_inventory(self, metadata: dict) -> float: open_trades = self.dp.trades_open if not open_trades: return 0.0 return open_trades[0].amount # Assuming one trade per pair