MidasBot/freqtrade/user_data/strategies/AiBiasStrategy.py

# pragma pylint: disable=missing-docstring, invalid-name, too-few-public-methods
"""
AiBiasStrategy — stratégie directionnelle pilotée par le biais IA (Phase 3).

Base technique : croisement EMA + RSI (comme SampleStrategy).
Surcouche IA : un biais de marché produit par Claude (service ai_analyzer) est lu
depuis Redis et filtre/renforce les entrées.

- Live / dry-run : lit le biais COURANT de la paire à chaque nouvelle bougie.
- Backtest : Redis ne contient pas d'historique de biais -> repli `neutral` (la
  surcouche IA est neutralisée, seule la base technique joue). Backtester finement
  l'IA nécessiterait d'enregistrer les biais historiques (amélioration future).
"""
from __future__ import annotations

import json
import os
from functools import lru_cache
from pathlib import Path

import pandas as pd
import talib.abstract as ta
from pandas import DataFrame

from freqtrade.enums import RunMode
from freqtrade.strategy import IStrategy

# Répertoire de l'historique des biais (monté dans le conteneur freqtrade).
_HISTORY_DIR = os.environ.get(
    "AI_HISTORY_DIR", "/freqtrade/user_data/ai_bias_history"
)


@lru_cache(maxsize=1)
def _redis_client():
    """Client Redis paresseux et tolérant aux pannes (None si indisponible)."""
    try:
        import redis  # importé ici pour ne pas casser si le paquet manque

        url = os.environ.get("REDIS_URL", "redis://redis:6379/0")
        client = redis.Redis.from_url(
            url, decode_responses=True, socket_connect_timeout=1, socket_timeout=1
        )
        client.ping()
        return client
    except Exception:  # noqa: BLE001
        return None


def _read_bias(pair: str) -> dict:
    """Renvoie {'direction', 'confidence'} COURANT pour la paire (live/dry-run)."""
    client = _redis_client()
    if client is None:
        return {"direction": "neutral", "confidence": 0.0}
    try:
        raw = client.get(f"bias:{pair}")
        if not raw:
            return {"direction": "neutral", "confidence": 0.0}
        data = json.loads(raw)
        return {
            "direction": data.get("direction", "neutral"),
            "confidence": float(data.get("confidence", 0.0)),
        }
    except Exception:  # noqa: BLE001
        return {"direction": "neutral", "confidence": 0.0}


def _load_history(pair: str) -> pd.DataFrame:
    """Charge l'historique horodaté des biais d'une paire (vide si absent)."""
    path = Path(_HISTORY_DIR) / f"{pair.replace('/', '_')}.csv"
    if not path.exists():
        return pd.DataFrame(columns=["timestamp", "direction", "confidence"])
    try:
        hist = pd.read_csv(path)
        hist["timestamp"] = pd.to_datetime(hist["timestamp"], utc=True)
        hist["confidence"] = pd.to_numeric(hist["confidence"], errors="coerce").fillna(0.0)
        return hist.sort_values("timestamp").reset_index(drop=True)
    except Exception:  # noqa: BLE001 — historique corrompu : on l'ignore
        return pd.DataFrame(columns=["timestamp", "direction", "confidence"])


def _merge_history(dataframe: DataFrame, pair: str) -> DataFrame:
    """Associe à chaque bougie le dernier biais connu À CETTE DATE (merge_asof)."""
    hist = _load_history(pair)
    if hist.empty:
        dataframe["ai_direction"] = "neutral"
        dataframe["ai_confidence"] = 0.0
        return dataframe
    # Aligner la résolution temporelle (Freqtrade=ms, pandas parse=us) sinon merge_asof échoue.
    hist = hist.copy()
    hist["timestamp"] = hist["timestamp"].astype(dataframe["date"].dtype)
    # Le dataframe Freqtrade est déjà trié par date croissante (prérequis merge_asof).
    merged = pd.merge_asof(
        dataframe[["date"]],
        hist[["timestamp", "direction", "confidence"]],
        left_on="date",
        right_on="timestamp",
        direction="backward",
    )
    dataframe["ai_direction"] = merged["direction"].fillna("neutral").values
    dataframe["ai_confidence"] = merged["confidence"].fillna(0.0).values
    return dataframe


class AiBiasStrategy(IStrategy):
    INTERFACE_VERSION = 3

    timeframe = "1h"

    minimal_roi = {"0": 0.05, "120": 0.03, "360": 0.01, "720": 0}
    stoploss = -0.10
    trailing_stop = True
    trailing_stop_positive = 0.02
    trailing_stop_positive_offset = 0.03
    trailing_only_offset_is_reached = True

    startup_candle_count: int = 50
    process_only_new_candles = True
    use_exit_signal = True

    # Confiance minimale du biais IA pour qu'il influence les décisions.
    ai_min_confidence: float = float(os.environ.get("AI_MIN_CONFIDENCE", "0.6"))

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe["ema_fast"] = ta.EMA(dataframe, timeperiod=9)
        dataframe["ema_slow"] = ta.EMA(dataframe, timeperiod=21)
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)

        # Source du biais IA selon le mode :
        # - backtest/hyperopt/plot : historique horodaté (biais valide à chaque bougie)
        # - dry-run/live           : biais COURANT depuis Redis (dernière bougie)
        runmode = self.dp.runmode if self.dp is not None else RunMode.OTHER
        if runmode in (RunMode.BACKTEST, RunMode.HYPEROPT, RunMode.PLOT):
            dataframe = _merge_history(dataframe, metadata["pair"])
        else:
            bias = _read_bias(metadata["pair"])
            dataframe["ai_direction"] = bias["direction"]
            dataframe["ai_confidence"] = bias["confidence"]
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        base_long = (
            (dataframe["ema_fast"] > dataframe["ema_slow"])
            & (dataframe["ema_fast"].shift(1) <= dataframe["ema_slow"].shift(1))
            & (dataframe["rsi"] < 70)
            & (dataframe["volume"] > 0)
        )

        # Filtre IA : si le biais est suffisamment confiant, on bloque les longs
        # quand il est baissier ; on les laisse passer sinon (neutre/haussier).
        ai_blocks_long = (dataframe["ai_direction"] == "bearish") & (
            dataframe["ai_confidence"] >= self.ai_min_confidence
        )

        dataframe.loc[base_long & (~ai_blocks_long), "enter_long"] = 1
        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        base_exit = (
            (dataframe["ema_fast"] < dataframe["ema_slow"])
            & (dataframe["ema_fast"].shift(1) >= dataframe["ema_slow"].shift(1))
            & (dataframe["volume"] > 0)
        )

        # Sortie anticipée si l'IA devient nettement baissière.
        ai_exit = (dataframe["ai_direction"] == "bearish") & (
            dataframe["ai_confidence"] >= self.ai_min_confidence
        )

        dataframe.loc[base_exit | ai_exit, "exit_long"] = 1
        return dataframe