# source: https://raw.githubusercontent.com/DOUGLASGUEDESATRIA/OSIRIS_TRADE/fe3c09567f3798d1cad6d7a306fa4f6131d91e17/user_data/strategies/OsirisFlowV2.py """ OSIRIS FLOW v2 — Pattern-Based Order Flow Day Trading ====================================================== Em vez de score numérico, detecta PADRÕES ESPECÍFICOS de order flow que traders institucionais reconhecem. Cada pattern é uma condição AND completa — não precisa de confluência genérica. PATTERNS IMPLEMENTADOS: LONG PATTERNS: 1. ABSORPTION_BUY: Grande venda mas preço não cai → parede de bid → delta negativo + corpo verde + high volume + low price_impact 2. SWEEP_REVERSAL_LONG: Stop hunt pra baixo seguido de compra agressiva → sweep_down[prev] + delta positivo + buy_ratio alto + delta_accel >0 3. WHALE_BUY: Trade gigante comprador + burst de atividade → max_trade_size alto + is_buy + delta positivo forte 4. DELTA_FLIP_LONG: Delta flipa de negativo pra positivo com aceleração → delta[prev] < 0, delta[curr] > 0, delta_accel > 0, volume crescente 5. INSTITUTIONAL_ACCUMULATION: Preço para mas volume de compra cresce → price_impact baixo + buy_volume crescente + corpo pequeno SHORT PATTERNS: 6-10: Espelhos dos LONG com sinais invertidos EXITS: - Trailing stop dinâmico baseado em ATR - Exit em absorção reversa (sinal de que o padrão falhou) - Exit em delta reversal forte 100% proprietário. Desenvolvido exclusivamente para OSIRIS. """ import logging import numpy as np import pandas as pd from pathlib import Path from pandas import DataFrame from typing import Optional from freqtrade.strategy import IStrategy from freqtrade.strategy import DecimalParameter, IntParameter from freqtrade.persistence import Trade import talib.abstract as ta logger = logging.getLogger(__name__) ORDERFLOW_PATH = Path(__file__).parent.parent / "data" / "orderflow" / "BTCUSDT-orderflow-5m.feather" class Github_DOUGLASGUEDESATRIA_OSIRIS_TRADE__OsirisFlowV2__20260329_060843(IStrategy): """OsirisFlow v2 — Pattern detection from real tick data.""" timeframe = "5m" can_short = True minimal_roi = { "0": 0.05, "30": 0.025, "60": 0.01, "120": 0.005, "240": 0, } stoploss = -0.02 trailing_stop = True trailing_stop_positive = 0.006 trailing_stop_positive_offset = 0.012 trailing_only_offset_is_reached = True process_only_new_candles = True startup_candle_count: int = 30 # ─── Hyperopt Parameters (thresholds for pattern detection) ─────── # Absorption pattern abs_vol_spike = DecimalParameter(1.2, 3.0, default=1.5, space="buy", optimize=True) abs_price_impact = DecimalParameter(1.0, 5.0, default=2.5, space="buy", optimize=True) # Sweep reversal sweep_threshold = DecimalParameter(0.8, 2.5, default=1.2, space="buy", optimize=True) sweep_buy_ratio = DecimalParameter(0.55, 0.75, default=0.6, space="buy", optimize=True) # Whale detection whale_size = DecimalParameter(5.0, 50.0, default=15.0, space="buy", optimize=True) # Delta flip flip_delta_zscore = DecimalParameter(0.5, 2.5, default=1.0, space="buy", optimize=True) # Minimum patterns required (1 = any single pattern, 2 = need 2 patterns simultaneously) min_patterns = IntParameter(1, 3, default=1, space="buy", optimize=True) # Exit exit_delta_reversal = DecimalParameter(0.8, 3.0, default=1.5, space="sell", optimize=True) _orderflow_df: Optional[pd.DataFrame] = None def _load_orderflow(self) -> pd.DataFrame: if self._orderflow_df is None: path = ORDERFLOW_PATH if path.is_symlink(): path = path.resolve() if path.exists(): self._orderflow_df = pd.read_feather(path) if 'date' in self._orderflow_df.columns: self._orderflow_df['date'] = pd.to_datetime(self._orderflow_df['date']) self._orderflow_df = self._orderflow_df.set_index('date') logger.info(f"Loaded order flow: {len(self._orderflow_df)} candles") else: logger.error(f"Order flow not found: {path}") self._orderflow_df = pd.DataFrame() return self._orderflow_df def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: of = self._load_orderflow() flow_cols = ['delta', 'cvd', 'buy_ratio', 'buy_volume', 'sell_volume', 'trade_intensity', 'big_imbalance', 'delta_zscore', 'vol_spike', 'absorption', 'sweep_up', 'sweep_down', 'delta_divergence', 'aggression', 'big_buys', 'big_sells', 'vwap', 'max_trade_size', 'trade_burst', 'delta_flips', 'delta_accel', 'price_impact', 'max_run'] if of.empty: for col in flow_cols: dataframe[col] = 0 else: dataframe['date_key'] = pd.to_datetime(dataframe['date']).dt.tz_localize(None) of_reset = of.reset_index() if of.index.name: of_reset.rename(columns={of.index.name: 'date_key'}, inplace=True) available = [c for c in flow_cols if c in of_reset.columns] of_sub = of_reset[['date_key'] + available].copy() of_sub['date_key'] = pd.to_datetime(of_sub['date_key']).dt.tz_localize(None) dataframe = dataframe.merge(of_sub, on='date_key', how='left', suffixes=('', '_of')) dataframe.drop(columns=['date_key'], inplace=True) for col in available: if col == 'buy_ratio': dataframe[col] = dataframe[col].fillna(0.5) elif col == 'aggression': dataframe[col] = dataframe[col].fillna(1.0) else: dataframe[col] = dataframe[col].fillna(0) matched = dataframe['delta'].ne(0).sum() logger.info(f"Order flow merge: {matched}/{len(dataframe)} ({matched/len(dataframe)*100:.1f}%)") # ── Derived rolling features ── dataframe['delta_sum3'] = dataframe['delta'].rolling(3).sum() dataframe['buy_ratio_ema'] = dataframe['buy_ratio'].ewm(span=10).mean() dataframe['vol_spike_ema'] = dataframe['vol_spike'].ewm(span=5).mean() # Z-scores for pattern features for col in ['max_trade_size', 'trade_burst', 'max_run']: mean = dataframe[col].rolling(50).mean() std = dataframe[col].rolling(50).std().replace(0, np.nan) dataframe[f'{col}_z'] = (dataframe[col] - mean) / std # Price context dataframe['ema20'] = ta.EMA(dataframe['close'], timeperiod=20) dataframe['atr'] = ta.ATR(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14) # Candle body analysis dataframe['body'] = dataframe['close'] - dataframe['open'] dataframe['body_pct'] = abs(dataframe['body']) / dataframe['close'] * 100 dataframe['is_green'] = (dataframe['body'] > 0).astype(int) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # ─── LONG PATTERNS ─── patterns_long = pd.Series(0, index=dataframe.index, dtype=int) # P1: ABSORPTION BUY — big selling absorbed, price holds # Institutional bid wall: volume spike but price doesn't drop p1_long = ( (dataframe['vol_spike'] > self.abs_vol_spike.value) & (dataframe['delta'] < 0) & # sellers hitting (dataframe['is_green'] == 1) & # but candle still green (dataframe['price_impact'] < self.abs_price_impact.value) # low price impact = absorption ) patterns_long += p1_long.astype(int) # P2: SWEEP REVERSAL — stop hunt followed by reversal # Previous candle swept lows, current candle buying aggressively p2_long = ( (dataframe['sweep_down'].shift(1) > self.sweep_threshold.value) & (dataframe['buy_ratio'] > self.sweep_buy_ratio.value) & (dataframe['delta'] > 0) & (dataframe['delta_accel'] > 0) # buying accelerating ) patterns_long += p2_long.astype(int) # P3: WHALE BUY — anomalously large buy trade p3_long = ( (dataframe['max_trade_size_z'] > 2.0) & (dataframe['buy_ratio'] > 0.55) & (dataframe['delta_zscore'] > self.flip_delta_zscore.value) ) patterns_long += p3_long.astype(int) # P4: DELTA FLIP — sellers give up, buyers take over p4_long = ( (dataframe['delta'].shift(1) < 0) & (dataframe['delta'].shift(2) < 0) & (dataframe['delta'] > 0) & (dataframe['delta_zscore'] > self.flip_delta_zscore.value) & (dataframe['delta_accel'] > 0) & (dataframe['vol_spike'] > 1.0) ) patterns_long += p4_long.astype(int) # P5: INSTITUTIONAL ACCUMULATION — price flat but buying growing p5_long = ( (dataframe['body_pct'] < 0.1) & # tiny body (dataframe['buy_ratio'] > 0.55) & (dataframe['buy_ratio_ema'] > 0.52) & (dataframe['max_run_z'] > 1.5) & # anomalous run of buys (dataframe['vol_spike'] > self.abs_vol_spike.value) ) patterns_long += p5_long.astype(int) dataframe['enter_long'] = ( (patterns_long >= self.min_patterns.value) & (dataframe['volume'] > 0) ).astype(int) # ─── SHORT PATTERNS ─── patterns_short = pd.Series(0, index=dataframe.index, dtype=int) # P1: ABSORPTION SELL p1_short = ( (dataframe['vol_spike'] > self.abs_vol_spike.value) & (dataframe['delta'] > 0) & # buyers hitting (dataframe['is_green'] == 0) & # but candle still red (dataframe['price_impact'] < self.abs_price_impact.value) ) patterns_short += p1_short.astype(int) # P2: SWEEP REVERSAL SHORT p2_short = ( (dataframe['sweep_up'].shift(1) > self.sweep_threshold.value) & (dataframe['buy_ratio'] < (1 - self.sweep_buy_ratio.value)) & (dataframe['delta'] < 0) & (dataframe['delta_accel'] < 0) ) patterns_short += p2_short.astype(int) # P3: WHALE SELL p3_short = ( (dataframe['max_trade_size_z'] > 2.0) & (dataframe['buy_ratio'] < 0.45) & (dataframe['delta_zscore'] < -self.flip_delta_zscore.value) ) patterns_short += p3_short.astype(int) # P4: DELTA FLIP SHORT p4_short = ( (dataframe['delta'].shift(1) > 0) & (dataframe['delta'].shift(2) > 0) & (dataframe['delta'] < 0) & (dataframe['delta_zscore'] < -self.flip_delta_zscore.value) & (dataframe['delta_accel'] < 0) & (dataframe['vol_spike'] > 1.0) ) patterns_short += p4_short.astype(int) # P5: INSTITUTIONAL DISTRIBUTION p5_short = ( (dataframe['body_pct'] < 0.1) & (dataframe['buy_ratio'] < 0.45) & (dataframe['buy_ratio_ema'] < 0.48) & (dataframe['max_run_z'] > 1.5) & (dataframe['vol_spike'] > self.abs_vol_spike.value) ) patterns_short += p5_short.astype(int) dataframe['enter_short'] = ( (patterns_short >= self.min_patterns.value) & (dataframe['volume'] > 0) ).astype(int) dataframe['patterns_long'] = patterns_long dataframe['patterns_short'] = patterns_short return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # Exit on strong reversal signal dataframe['exit_long'] = ( (dataframe['delta_zscore'] < -self.exit_delta_reversal.value) & (dataframe['delta_accel'] < 0) ).astype(int) dataframe['exit_short'] = ( (dataframe['delta_zscore'] > self.exit_delta_reversal.value) & (dataframe['delta_accel'] > 0) ).astype(int) return dataframe