# source: https://raw.githubusercontent.com/shatianming5/Agent_market/389cd928020b2247909d47cecf41a78974d62755/workspace/strategies/type_F_ml/freqtrade_rl_strategy.py
"""Freqtrade RL Strategy — uses a trained RL model for trading decisions.

Loads RL model (Q-table, DQN, etc) from workspace/models/ or artifacts/models/
and uses it to make buy/sell decisions based on market state.

Since most RL models output discrete actions (buy/sell/hold), this maps:
  action=1 → enter_long
  action=0 → hold
  action=-1 or 2 → exit_long
"""
from __future__ import annotations

import json
import pickle
import sys
from pathlib import Path
from typing import Any, Dict, List, Optional

import numpy as np
from pandas import DataFrame

_HERE = Path(__file__).resolve()
_ROOT = _HERE.parents[3]
for p in [str(_ROOT / "src"), str(_ROOT)]:
    if p not in sys.path:
        sys.path.insert(0, p)

from freqtrade.strategy import IStrategy


def _resolve_summary_artifact(
    raw_path: Optional[str],
    *,
    model_dir: Path,
    fallback_names: Optional[List[str]] = None,
) -> Optional[Path]:
    """Resolve copied training artifacts when summaries contain stale absolute paths."""
    candidates: List[Path] = []
    if raw_path:
        raw = Path(str(raw_path))
        if raw.is_absolute():
            candidates.append(model_dir / raw.name)
            candidates.append(raw)
        else:
            candidates.append(_ROOT / raw)
            candidates.append(model_dir / raw.name)
    for fallback in fallback_names or []:
        fallback_path = Path(str(fallback))
        candidates.append(fallback_path if fallback_path.is_absolute() else (model_dir / fallback_path))

    seen = set()
    for candidate in candidates:
        key = str(candidate)
        if key in seen:
            continue
        seen.add(key)
        if candidate.exists():
            return candidate
    return candidates[0] if candidates else None


def _iter_model_candidates(directory: Path, *suffixes: str) -> List[Path]:
    items: List[Path] = []
    for suffix in suffixes:
        items.extend(
            path for path in directory.glob(f"*{suffix}")
            if path.is_file() and not path.name.startswith("._")
        )
    return sorted(items)


def _looks_like_rl_summary(directory: Path, summary: Dict[str, Any], model_path: Optional[Path]) -> bool:
    model_name = str(summary.get("model") or "").lower()
    algo_name = str(summary.get("algo", {}).get("name") or "").lower() if isinstance(summary.get("algo"), dict) else ""
    suffix = model_path.suffix.lower() if model_path is not None else ""
    if "rl" in directory.name.lower():
        return True
    if model_name in {"ppo", "a2c", "dqn", "sac", "td3", "ddpg", "reinforce", "rl"}:
        return True
    if algo_name in {"ppo", "a2c", "dqn", "sac", "td3", "ddpg", "reinforce"}:
        return True
    return suffix == ".zip"


class Github_shatianming5_Agent_market__freqtrade_rl_strategy__20260514_202634(IStrategy):
    """RL-based trading strategy for freqtrade."""

    timeframe = "1h"
    can_short = False
    minimal_roi = {"0": 0.10, "240": 0.03}
    stoploss = -0.05
    trailing_stop = True
    trailing_stop_positive = 0.012
    trailing_stop_positive_offset = 0.025
    use_exit_signal = True
    startup_candle_count = 60
    process_only_new_candles = True

    _model = None
    _features: Optional[List[str]] = None
    _feature_cfg = None
    _expression_specs = None
    _loaded = False

    def _load(self):
        if self._loaded:
            return

        models_root = _ROOT / "artifacts" / "models"
        model_path = None
        summary = None
        model_dir = None

        # Find RL model (prefer RL-tagged summaries / dirs).
        for d in sorted(models_root.iterdir(), key=lambda x: x.stat().st_mtime, reverse=True):
            if not d.is_dir():
                continue
            summary_f = d / "training_summary.json"
            if not summary_f.exists():
                continue
            s = json.loads(summary_f.read_text())
            mp = _resolve_summary_artifact(
                s.get("model_path"),
                model_dir=d,
                fallback_names=["ppo_trading_env.zip", "model.zip", "model.pkl", "model.pt", "model.pth"],
            )
            if mp is None or not _looks_like_rl_summary(d, s, mp):
                continue
            if mp.exists():
                model_path = mp
                summary = s
                model_dir = d
                break

        if model_path is None or summary is None or model_dir is None:
            raise FileNotFoundError("No RL model found")

        self._features = [str(c) for c in (summary.get("features") or []) if str(c).strip()]

        for candidate in [
            _resolve_summary_artifact(
                summary.get("feature_snapshot") or summary.get("feature_file"),
                model_dir=model_dir,
                fallback_names=["feature_snapshot.json"],
            ),
            _ROOT / "user_data" / "freqai_features_real.json",
        ]:
            if candidate is None:
                continue
            if candidate.exists():
                self._feature_cfg = json.loads(candidate.read_text(encoding="utf-8-sig"))
                break

        expr_path = _resolve_summary_artifact(
            summary.get("expressions_snapshot") or summary.get("expressions_file"),
            model_dir=model_dir,
            fallback_names=["expressions_snapshot.json"],
        )
        if expr_path and expr_path.exists():
            from agent_market.freqai.expression_engine import load_expression_file
            self._expression_specs = load_expression_file(expr_path)

        # Load model
        suffix = model_path.suffix.lower()
        if suffix == ".pkl":
            with open(model_path, "rb") as f:
                self._model = pickle.load(f)
        elif suffix in (".pt", ".pth"):
            import torch
            self._model = torch.load(model_path, map_location="cpu", weights_only=False)
        elif suffix == ".zip":
            # Stable-Baselines3 model
            try:
                from stable_baselines3 import PPO
                self._model = PPO.load(str(model_path))
            except ImportError:
                raise ImportError("stable_baselines3 required for .zip RL models")
        else:
            with open(model_path, "rb") as f:
                self._model = pickle.load(f)

        self._loaded = True

    def _predict_sb3_actions(self, X: np.ndarray, prices: Optional[np.ndarray]) -> np.ndarray:
        """Run PPO-style policies one candle at a time with env-compatible state."""
        if prices is None:
            prices = np.zeros(len(X), dtype=np.float32)

        preds = np.zeros(len(X), dtype=np.float32)
        position = 0
        entry_price = 0.0

        for idx, row in enumerate(X):
            price = float(prices[idx]) if idx < len(prices) else 0.0
            unrealized = 0.0
            if position == 1 and entry_price > 0 and price > 0:
                unrealized = (price / entry_price) - 1.0
            obs = np.concatenate(
                [
                    row.astype(np.float32, copy=False),
                    np.asarray([float(position), float(unrealized)], dtype=np.float32),
                ]
            )
            action, _ = self._model.predict(obs, deterministic=True)
            action_id = int(np.asarray(action).reshape(-1)[0])
            if action_id == 1:
                preds[idx] = 1.0
                if position == 0:
                    position = 1
                    entry_price = price
            elif action_id == 2:
                preds[idx] = -1.0
                if position == 1:
                    position = 0
                    entry_price = 0.0
            else:
                preds[idx] = 0.0

        return preds

    def _predict_action(self, X: np.ndarray, prices: Optional[np.ndarray] = None) -> np.ndarray:
        """Predict RL action for each row. Returns array of floats."""
        if hasattr(self._model, "policy") and hasattr(self._model, "predict"):
            return self._predict_sb3_actions(X, prices)
        if hasattr(self._model, "predict"):
            # Sklearn-like or custom model with predict()
            preds = self._model.predict(X)
            return np.array(preds, dtype=float).reshape(-1)
        elif isinstance(self._model, dict):
            # Q-table or weight-based model
            if "weights" in self._model:
                w = self._model["weights"]
                b = self._model.get("bias", 0)
                mean = self._model.get("mean", np.zeros(X.shape[1]))
                std = self._model.get("std", np.ones(X.shape[1]))
                X_norm = (X - mean) / (std + 1e-8)
                return (X_norm @ w + b).reshape(-1)
        elif hasattr(self._model, "forward"):
            import torch
            with torch.no_grad():
                t = torch.from_numpy(X.astype(np.float32))
                return self._model(t).cpu().numpy().reshape(-1)

        # Fallback: treat as regression model output
        return np.zeros(len(X))

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        self._load()
        if self._feature_cfg:
            from agent_market.freqai.features import apply_configured_features
            dataframe = apply_configured_features(dataframe, self._feature_cfg)
        if self._expression_specs:
            from agent_market.freqai.expression_engine import apply_expressions
            dataframe, _ = apply_expressions(dataframe, self._expression_specs, on_error="raise")
        for col in self._features:
            if col not in dataframe.columns:
                dataframe[col] = 0.0
        matrix = dataframe[self._features].astype(float).replace([np.inf, -np.inf], np.nan).ffill().fillna(0.0)
        dataframe["rl_pred"] = self._predict_action(
            matrix.to_numpy(dtype=np.float32),
            dataframe["close"].to_numpy(dtype=np.float32) if "close" in dataframe.columns else None,
        )
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # RL pred > 0 means "buy signal" (regression output) or action=1
        dataframe.loc[(dataframe["volume"] > 0) & (dataframe["rl_pred"] > 0.003),
                      ["enter_long", "enter_tag"]] = (1, "rl_long")
        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe.loc[(dataframe["volume"] > 0) & (dataframe["rl_pred"] < 0.0),
                      "exit_long"] = 1
        return dataframe