OsirisPrimusV2

"""
OSIRIS PRIMUS V2 — Short-Biased Swing Strategy
================================================
Built from raw price data analysis of 10 crypto pairs over ~800 days.
Signals validated with real SL/TP simulation checking H/L of each daily candle.

SHORT signals are 3-4x stronger than long signals in the data.
This version is SHORT-BIASED with very selective long entries.

SHORT SIGNALS (proven on 9-10/10 pairs):
1. Wednesday (weight 3) — strongest day-of-week signal (+755% cross-pair)
2. ClosePos >= 0.85 (weight 3) — exhaustion at daily high (+343% cross-pair)
3. Up day + Volume spike (weight 2) — blow-off top
4. Bear trend (weight 1) — trend confirmation
5. 2 consecutive green days (weight 2) — mean reversion setup

LONG SIGNALS (very selective, only with strong confluence):
1. ClosePos <= 0.15 (weight 3) — deep accumulation only
2. Tuesday (weight 2) — in bull trend only
3. Down day + Volume spike (weight 3) — capitulation bounce
4. Bull trend (weight 2) — mandatory for longs
5. 2 consecutive red days (weight 1) — mean reversion

KEY: Short score_min=3 (frequent), Long score_min=5 (rare, needs confluence).

SL/TP: 2% SL / 4% TP for shorts, 2% SL / 4% TP for longs.
"""

import logging
import numpy as np
from pandas import DataFrame
from datetime import datetime

from freqtrade.strategy import IStrategy
from freqtrade.strategy import DecimalParameter, IntParameter

logger = logging.getLogger(__name__)


class OsirisPrimusV2(IStrategy):
    """OsirisPrimusV2 — short-biased swing from raw price patterns."""

    timeframe = "1d"
    can_short = True

    # ROI disabled — custom SL/TP
    minimal_roi = {"0": 100}

    stoploss = -0.03

    trailing_stop = False
    process_only_new_candles = True
    startup_candle_count: int = 55

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

    # ClosePos thresholds
    close_pos_long = DecimalParameter(0.10, 0.20, default=0.15, space="buy", optimize=True)
    close_pos_short = DecimalParameter(0.80, 0.95, default=0.85, space="sell", optimize=True)

    # Volume spike
    vol_spike_min = DecimalParameter(1.2, 2.0, default=1.5, space="buy", optimize=True)

    # SL/TP — optimized: 3% SL, 6% TP (1:2 asymmetry validated across 10 pairs)
    sl_pct = DecimalParameter(1.5, 4.0, default=3.0, space="sell", optimize=True)
    tp_pct = DecimalParameter(3.0, 8.0, default=6.0, space="sell", optimize=True)

    # Score thresholds — short needs 2+ signals, long needs 3+ signals
    long_score_min = IntParameter(4, 8, default=8, space="buy", optimize=True)
    short_score_min = IntParameter(3, 7, default=5, space="sell", optimize=True)

    # Max hold days
    max_hold = IntParameter(3, 7, default=5, space="sell", optimize=True)

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:

        # Close Position within daily range
        dataframe['range'] = dataframe['high'] - dataframe['low']
        dataframe['close_pos'] = np.where(
            dataframe['range'] > 0,
            (dataframe['close'] - dataframe['low']) / dataframe['range'],
            0.5
        )

        # Body
        dataframe['body_pct'] = (dataframe['close'] - dataframe['open']) / dataframe['open'] * 100
        dataframe['is_green'] = dataframe['close'] > dataframe['open']

        # Volume relative to 20-bar SMA
        dataframe['vol_sma20'] = dataframe['volume'].rolling(20).mean()
        dataframe['vol_ratio'] = dataframe['volume'] / dataframe['vol_sma20']

        # Trend
        dataframe['sma20'] = dataframe['close'].rolling(20).mean()
        dataframe['trend_bull'] = (dataframe['close'] > dataframe['sma20']).astype(int)

        # Day of week
        dataframe['day_of_week'] = dataframe['date'].dt.dayofweek

        # Prev candle colors
        dataframe['prev1_green'] = dataframe['is_green'].shift(1).fillna(False).astype(int)
        dataframe['prev2_green'] = dataframe['is_green'].shift(2).fillna(False).astype(int)

        # ─── SHORT SCORE (primary edge) ───────────────────
        dataframe['short_score'] = 0

        # Wednesday — strongest signal (weight 3)
        dataframe['short_score'] += np.where(
            dataframe['day_of_week'] == 2, 3, 0
        )

        # ClosePos high — exhaustion (weight 3)
        dataframe['short_score'] += np.where(
            dataframe['close_pos'] >= self.close_pos_short.value, 3, 0
        )

        # Up day + Volume spike — blow-off (weight 2)
        dataframe['short_score'] += np.where(
            (dataframe['body_pct'] > 0.5) &
            (dataframe['vol_ratio'] > self.vol_spike_min.value),
            2, 0
        )

        # Bear trend (weight 1)
        dataframe['short_score'] += np.where(
            dataframe['trend_bull'] == 0, 1, 0
        )

        # 2 consecutive green days — overbought (weight 2)
        dataframe['short_score'] += np.where(
            (dataframe['prev1_green'] == 1) & (dataframe['prev2_green'] == 1),
            2, 0
        )

        # ─── LONG SCORE (selective — needs strong confluence) ──
        dataframe['long_score'] = 0

        # ClosePos very low — deep accumulation (weight 3)
        dataframe['long_score'] += np.where(
            dataframe['close_pos'] <= self.close_pos_long.value, 3, 0
        )

        # Tuesday in bull trend (weight 2)
        dataframe['long_score'] += np.where(
            (dataframe['day_of_week'] == 1) & (dataframe['trend_bull'] == 1),
            2, 0
        )

        # Down day + Volume spike — capitulation (weight 3)
        dataframe['long_score'] += np.where(
            (dataframe['body_pct'] < -0.5) &
            (dataframe['vol_ratio'] > self.vol_spike_min.value),
            3, 0
        )

        # Bull trend — mandatory context (weight 2)
        dataframe['long_score'] += np.where(
            dataframe['trend_bull'] == 1, 2, 0
        )

        # 2 consecutive red days (weight 1)
        dataframe['long_score'] += np.where(
            (dataframe['prev1_green'] == 0) & (dataframe['prev2_green'] == 0),
            1, 0
        )

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:

        dataframe.loc[
            dataframe['long_score'] >= self.long_score_min.value,
            'enter_long'
        ] = 1

        dataframe.loc[
            dataframe['short_score'] >= self.short_score_min.value,
            'enter_short'
        ] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        return dataframe

    def custom_exit(self, pair: str, trade, current_time: datetime,
                    current_rate: float, current_profit: float,
                    **kwargs) -> str | bool:

        tp = self.tp_pct.value / 100
        sl = self.sl_pct.value / 100

        if current_profit >= tp:
            return f"tp_{self.tp_pct.value}pct"

        if current_profit <= -sl:
            return f"sl_{self.sl_pct.value}pct"

        trade_duration = (current_time - trade.open_date_utc).days
        if trade_duration >= self.max_hold.value:
            return "max_hold"

        return False

    def custom_stoploss(self, pair: str, trade, current_time: datetime,
                        current_rate: float, current_profit: float,
                        after_fill: bool, **kwargs) -> float:
        return -(self.sl_pct.value / 100)