OsirisFlow v2 — Pattern detection from real tick data.
Timeframe
5m
Direction
Long & Short
Stoploss
-2.0%
Trailing Stop
Yes
ROI
0m: 5.0%, 30m: 2.5%, 60m: 1.0%, 120m: 0.5%
Interface Version
N/A
Startup Candles
N/A
Indicators
3
freqtrade/freqtrade-strategies
"""
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 OsirisFlowV2(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