OsirisFlow

"""
OSIRIS FLOW STRATEGY v1.0 — Real Order Flow Day Trading
=========================================================
Usa tick data real (aggTrades da Binance) processado em features de order flow.
NÃO usa proxies OHLCV — usa delta, CVD, big trades, absorption, sweeps REAIS.

FEATURES DE ORDER FLOW (derivados de tick data):
1. Delta (buy_vol - sell_vol) — pressão real compradora/vendedora
2. CVD (Cumulative Volume Delta) — tendência de fluxo acumulado
3. Buy Ratio — fração do volume que é compra agressiva
4. Big Trades Imbalance — desbalanceamento de trades grandes (institucional)
5. Delta Z-Score — delta normalizado para comparação entre períodos
6. Volume Spike — anomalias de volume (entrada institucional)
7. Absorption — alto volume sem movimento de preço (parede institucional)
8. Sweep Up/Down — liquidity grabs (stop hunting)
9. Delta Divergence — preço sobe mas delta negativo (ou vice versa)
10. Aggression Ratio — buy_qty/sell_qty (urgência direcional)
11. Trade Intensity — trades/segundo (atividade do mercado)

FILOSOFIA:
- Day trade: entries e exits em minutos/horas, não dias
- 10 trades/dia target via high-frequency order flow signals
- SL apertado (0.5-1%), TP 0.3-1%, trailing para runs
- Edge vem de dados que 99% dos traders não tem acesso

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__)

# Path to order flow feather data
ORDERFLOW_PATH = Path(__file__).parent.parent / "data" / "orderflow" / "BTCUSDT-orderflow-5m.feather"


class OsirisFlow(IStrategy):
    """
    OSIRIS Flow — Real order flow day trading.
    """

    timeframe = "5m"
    can_short = True

    # Day trade exits — wider range, let hyperopt decide
    minimal_roi = {
        "0": 0.02,
        "15": 0.01,
        "45": 0.005,
        "90": 0.002,
        "180": 0,
    }

    stoploss = -0.03   # -3% max loss per trade (wider for BTC)

    trailing_stop = True
    trailing_stop_positive = 0.005
    trailing_stop_positive_offset = 0.01
    trailing_only_offset_is_reached = True

    use_custom_stoploss = False
    process_only_new_candles = True

    startup_candle_count: int = 30

    # ─── Hyperopt Parameters ─────────────────────────────────────────────

    # Long entry thresholds
    buy_delta_zscore = DecimalParameter(0.5, 3.0, default=1.2, space="buy", optimize=True)
    buy_buy_ratio = DecimalParameter(0.52, 0.75, default=0.58, space="buy", optimize=True)
    buy_big_imbalance = DecimalParameter(0.05, 0.5, default=0.15, space="buy", optimize=True)
    buy_vol_spike = DecimalParameter(1.0, 3.0, default=1.5, space="buy", optimize=True)
    buy_aggression = DecimalParameter(1.1, 3.0, default=1.3, space="buy", optimize=True)
    buy_score_min = IntParameter(5, 10, default=7, space="buy", optimize=True)

    # Short entry thresholds
    sell_delta_zscore = DecimalParameter(0.5, 3.0, default=1.2, space="buy", optimize=True)
    sell_buy_ratio = DecimalParameter(0.25, 0.48, default=0.42, space="buy", optimize=True)
    sell_big_imbalance = DecimalParameter(-0.5, -0.05, default=-0.15, space="buy", optimize=True)
    sell_aggression = DecimalParameter(0.3, 0.9, default=0.7, space="buy", optimize=True)

    # Exit parameters
    exit_delta_reversal = DecimalParameter(0.5, 2.5, default=1.0, space="sell", optimize=True)
    exit_absorption_thr = DecimalParameter(1.5, 4.0, default=2.0, space="sell", optimize=True)

    # Order flow data cache
    _orderflow_df: Optional[pd.DataFrame] = None

    def _load_orderflow(self) -> pd.DataFrame:
        """Load and cache order flow feather data."""
        if self._orderflow_df is None:
            if ORDERFLOW_PATH.exists():
                self._orderflow_df = pd.read_feather(ORDERFLOW_PATH)
                # Ensure datetime index
                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 data: {len(self._orderflow_df)} candles "
                           f"from {self._orderflow_df.index.min()} to {self._orderflow_df.index.max()}")
            elif ORDERFLOW_PATH.is_symlink():
                target = ORDERFLOW_PATH.resolve()
                if target.exists():
                    self._orderflow_df = pd.read_feather(target)
                    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 data (symlink): {len(self._orderflow_df)} candles")
                else:
                    logger.error(f"Order flow data not found: {ORDERFLOW_PATH} -> {target}")
                    self._orderflow_df = pd.DataFrame()
            else:
                logger.error(f"Order flow data not found: {ORDERFLOW_PATH}")
                self._orderflow_df = pd.DataFrame()
        return self._orderflow_df

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """Merge order flow features into the Freqtrade dataframe."""

        of = self._load_orderflow()

        if of.empty:
            logger.warning("No order flow data — all features will be NaN")
            for col in ['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']:
                dataframe[col] = 0
        else:
            # Merge by timestamp
            dataframe['date_key'] = pd.to_datetime(dataframe['date'])
            of_reset = of.reset_index()
            of_reset.rename(columns={'index': 'date_key'} if 'index' in of_reset.columns else {}, inplace=True)
            if 'date_key' not in of_reset.columns and of.index.name:
                of_reset = of_reset.rename(columns={of.index.name: 'date_key'})

            # Select only order flow columns
            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']
            available_cols = [c for c in flow_cols if c in of_reset.columns]
            of_subset = of_reset[['date_key'] + available_cols].copy()
            of_subset['date_key'] = pd.to_datetime(of_subset['date_key']).dt.tz_localize(None)
            dataframe['date_key'] = pd.to_datetime(dataframe['date_key']).dt.tz_localize(None)

            dataframe = dataframe.merge(of_subset, on='date_key', how='left', suffixes=('', '_of'))
            dataframe.drop(columns=['date_key'], inplace=True)

            # Fill NaN with neutral values
            for col in available_cols:
                if col in dataframe.columns:
                    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)} candles matched "
                       f"({matched/len(dataframe)*100:.1f}%)")

        # ─── Derived indicators from order flow ──────────────────────────

        # CVD momentum (rate of change of CVD)
        dataframe['cvd_roc'] = dataframe['cvd'].pct_change(5) * 100

        # Delta momentum (sum of last 3 deltas)
        dataframe['delta_sum3'] = dataframe['delta'].rolling(3).sum()
        dataframe['delta_sum5'] = dataframe['delta'].rolling(5).sum()

        # Smoothed buy ratio (EMA of buy_ratio)
        dataframe['buy_ratio_ema'] = dataframe['buy_ratio'].ewm(span=10).mean()

        # Big trades cumulative (directional big trade flow)
        dataframe['big_flow'] = (dataframe['big_buys'] - dataframe['big_sells']).rolling(5).sum()

        # Absorption strength (rolling max)
        dataframe['absorption_peak'] = dataframe['absorption'].rolling(10).max()

        # Sweep preceded by large volume (confirmation)
        dataframe['sweep_confirmed_up'] = (
            (dataframe['sweep_up'] > 1.5) & (dataframe['vol_spike'] > 1.5)
        ).astype(int)
        dataframe['sweep_confirmed_down'] = (
            (dataframe['sweep_down'] > 1.5) & (dataframe['vol_spike'] > 1.5)
        ).astype(int)

        # Trade intensity z-score
        ti_mean = dataframe['trade_intensity'].rolling(20).mean()
        ti_std = dataframe['trade_intensity'].rolling(20).std().replace(0, np.nan)
        dataframe['ti_zscore'] = (dataframe['trade_intensity'] - ti_mean) / ti_std

        # VWAP deviation (price relative to VWAP)
        dataframe['vwap_dev'] = np.where(
            dataframe['vwap'] > 0,
            (dataframe['close'] - dataframe['vwap']) / dataframe['vwap'] * 100,
            0
        )

        # Simple trend filter (EMA 20)
        dataframe['ema20'] = ta.EMA(dataframe['close'], timeperiod=20)
        dataframe['ema50'] = ta.EMA(dataframe['close'], timeperiod=50)
        dataframe['trend'] = np.where(dataframe['ema20'] > dataframe['ema50'], 1, -1)

        # RSI for overbought/oversold filter
        dataframe['rsi'] = ta.RSI(dataframe['close'], timeperiod=14)

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """Score-based entry using real order flow signals."""

        # ─── LONG SCORING ────────────────────────────────────────────────
        score_long = pd.Series(0, index=dataframe.index, dtype=int)

        # 1. Delta Z-Score positive (buying pressure above normal)
        score_long += np.where(dataframe['delta_zscore'] > self.buy_delta_zscore.value, 1, 0)

        # 2. Buy ratio high (more aggressive buyers)
        score_long += np.where(dataframe['buy_ratio'] > self.buy_buy_ratio.value, 1, 0)

        # 3. Buy ratio EMA trending up
        score_long += np.where(dataframe['buy_ratio_ema'] > self.buy_buy_ratio.value, 1, 0)

        # 4. Big trades imbalance positive (institutions buying)
        score_long += np.where(dataframe['big_imbalance'] > self.buy_big_imbalance.value, 1, 0)

        # 5. Volume spike with buying (smart money entry)
        score_long += np.where(
            (dataframe['vol_spike'] > self.buy_vol_spike.value) & (dataframe['buy_ratio'] > 0.52), 1, 0)

        # 6. Aggression ratio high (urgent buying)
        score_long += np.where(dataframe['aggression'] > self.buy_aggression.value, 1, 0)

        # 7. Delta sum positive (sustained buying over 3 candles)
        score_long += np.where(dataframe['delta_sum3'] > 0, 1, 0)

        # 8. CVD trending up (cumulative buying trend)
        score_long += np.where(dataframe['cvd_roc'] > 0.1, 1, 0)

        # 9. Big institutional flow positive
        score_long += np.where(dataframe['big_flow'] > 10, 1, 0)

        # 10. Trade intensity elevated (active market, not dead)
        score_long += np.where(dataframe['ti_zscore'] > 0.5, 1, 0)

        # 11. No bearish sweep (not a bull trap)
        score_long += np.where(dataframe['sweep_confirmed_down'].shift(1) == 0, 1, 0)

        # Entry condition
        dataframe['enter_long'] = (
            (score_long >= self.buy_score_min.value) &
            (dataframe['volume'] > 0)
        ).astype(int)

        # ─── SHORT SCORING ───────────────────────────────────────────────
        score_short = pd.Series(0, index=dataframe.index, dtype=int)

        # 1. Delta Z-Score negative (selling pressure above normal)
        score_short += np.where(dataframe['delta_zscore'] < -self.sell_delta_zscore.value, 1, 0)

        # 2. Buy ratio low (more aggressive sellers)
        score_short += np.where(dataframe['buy_ratio'] < self.sell_buy_ratio.value, 1, 0)

        # 3. Buy ratio EMA trending down
        score_short += np.where(dataframe['buy_ratio_ema'] < self.sell_buy_ratio.value, 1, 0)

        # 4. Big trades imbalance negative (institutions selling)
        score_short += np.where(dataframe['big_imbalance'] < self.sell_big_imbalance.value, 1, 0)

        # 5. Volume spike with selling
        score_short += np.where(
            (dataframe['vol_spike'] > self.buy_vol_spike.value) & (dataframe['buy_ratio'] < 0.48), 1, 0)

        # 6. Aggression ratio low (urgent selling)
        score_short += np.where(dataframe['aggression'] < self.sell_aggression.value, 1, 0)

        # 7. Delta sum negative (sustained selling over 3 candles)
        score_short += np.where(dataframe['delta_sum3'] < 0, 1, 0)

        # 8. CVD trending down
        score_short += np.where(dataframe['cvd_roc'] < -0.1, 1, 0)

        # 9. Big institutional flow negative
        score_short += np.where(dataframe['big_flow'] < -10, 1, 0)

        # 10. Trade intensity elevated
        score_short += np.where(dataframe['ti_zscore'] > 0.5, 1, 0)

        # 11. No bullish sweep
        score_short += np.where(dataframe['sweep_confirmed_up'].shift(1) == 0, 1, 0)

        dataframe['enter_short'] = (
            (score_short >= self.buy_score_min.value) &
            (dataframe['volume'] > 0)
        ).astype(int)

        # Store scores for analysis
        dataframe['score_long'] = score_long
        dataframe['score_short'] = score_short

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """Exit on order flow reversal signals."""

        # Exit long when selling pressure appears
        dataframe['exit_long'] = (
            (dataframe['delta_zscore'] < -self.exit_delta_reversal.value) |
            (dataframe['absorption'] > self.exit_absorption_thr.value)
        ).astype(int)

        # Exit short when buying pressure appears
        dataframe['exit_short'] = (
            (dataframe['delta_zscore'] > self.exit_delta_reversal.value) |
            (dataframe['absorption'] > self.exit_absorption_thr.value)
        ).astype(int)

        return dataframe