Spaces:

chen470
/

drl-trading-bot-dev2

Paused

drl-trading-bot-dev2 / src /env /htf_env.py
DRL Trading Bot
Feature: HTF Agent integration — live trading, API endpoints, UI tab
fc115d5 8 months ago
26.1 kB
"""
HTF Trading Environment
=======================
Hierarchical Multi-Timeframe (HTF) Gymnasium trading environment for DRL.
Executes at 15-minute resolution.  At each step it reads pre-computed
hierarchical features across 1D / 4H / 1H / 15M timeframes plus cross-TF
alignment signals, building a 117-dimensional observation:
    [20 1D feats | 25 4H feats | 30 1H feats | 35 15M feats |
     4 align feats | 3 pos state]
Reward shaping rewards entries that align with the HTF cascade and penalises
counter-trend positions, idle inaction during strong setups, and drawdowns.
"""

import logging
from typing import Dict, Optional, Tuple

import gymnasium as gym
import numpy as np
import pandas as pd
from gymnasium import spaces

try:
    from src.features.htf_features import HTFDataAligner, HTFFeatureEngine
    _HTF_AVAILABLE = True
except ImportError:
    _HTF_AVAILABLE = False

logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Observation dimension constants (must match htf_features.py)
# ---------------------------------------------------------------------------
N_1D = 20
N_4H = 25
N_1H = 30
N_15M = 35
N_ALIGN = 4
N_POS = 3
N_OBS = N_1D + N_4H + N_1H + N_15M + N_ALIGN + N_POS  # 117

# Indices of the per-TF "summary score" features inside each TF block.
# These are used to compute overall HTF alignment.
IDX_1D_SCORE = N_1D - 1          # feature index 19 within 1D block
IDX_4H_SCORE = N_4H - 1          # feature index 24 within 4H block
IDX_1H_SCORE = N_1H - 1          # feature index 29 within 1H block
IDX_15M_SCORE = N_15M - 1         # feature index 34 within 15M block

# Within the concatenated 114-feature vector (before pos state):
#   1D:  [0 .. 19]
#   4H:  [20 .. 44]
#   1H:  [45 .. 74]
#   15M: [75 .. 109]
#   align: [110 .. 113]
OFFSET_1D = 0
OFFSET_4H = N_1D
OFFSET_1H = N_1D + N_4H
OFFSET_15M = N_1D + N_4H + N_1H
OFFSET_ALIGN = N_1D + N_4H + N_1H + N_15M

# Position of the overall_alignment scalar inside the concatenated feature vec
# (last of the 4 alignment features)
IDX_OVERALL_ALIGN = OFFSET_ALIGN + 3  # index 113 in the 114-feature prefix


class HTFTradingEnv(gym.Env):
    """
    Hierarchical Multi-Timeframe Trading Environment.
    Parameters
    ----------
    df_15m : pd.DataFrame
        Required.  OHLCV DataFrame with DatetimeIndex at 15-minute resolution.
    df_1h : pd.DataFrame, optional
        1-hour OHLCV data.  Auto-resampled from df_15m when not provided.
    df_4h : pd.DataFrame, optional
        4-hour OHLCV data.  Auto-resampled from df_15m when not provided.
    df_1d : pd.DataFrame, optional
        Daily OHLCV data.  Auto-resampled from df_15m when not provided.
    initial_balance : float
        Starting portfolio balance (default 10 000).
    position_size : float
        Fraction of balance committed per trade (default 0.25).
    stop_loss_pct : float
        Stop-loss distance as a fraction of entry price (default 0.015 = 1.5%).
    take_profit_pct : float
        Take-profit distance as a fraction of entry price (default 0.03 = 3%).
    trading_fee : float
        Taker fee applied on open and close (default 0.0004 = 0.04%).
    lookback_window : int
        Minimum bars consumed before the first tradeable step (default 96).
    training_mode : bool
        When True randomises the episode start position for diversity.
    """

    metadata = {"render_modes": ["human"]}

    def __init__(
        self,
        df_15m: pd.DataFrame,
        df_1h: Optional[pd.DataFrame] = None,
        df_4h: Optional[pd.DataFrame] = None,
        df_1d: Optional[pd.DataFrame] = None,
        initial_balance: float = 10_000.0,
        position_size: float = 0.25,
        stop_loss_pct: float = 0.015,
        take_profit_pct: float = 0.03,
        trading_fee: float = 0.0004,
        lookback_window: int = 96,
        training_mode: bool = True,
    ) -> None:
        super().__init__()

        if not _HTF_AVAILABLE:
            raise ImportError(
                "HTFFeatureEngine / HTFDataAligner could not be imported from "
                "src.features.htf_features.  Ensure the module is on the Python path."
            )

        # ---- hyperparameters ------------------------------------------------
        self.initial_balance = float(initial_balance)
        self.position_size = float(position_size)
        self.stop_loss_pct = float(stop_loss_pct)
        self.take_profit_pct = float(take_profit_pct)
        self.trading_fee = float(trading_fee)
        self.lookback_window = int(lookback_window)
        self.training_mode = bool(training_mode)

        # ---- build / align timeframe DataFrames -----------------------------
        def _to_datetime_index(df: pd.DataFrame, label: str) -> pd.DataFrame:
            """Ensure a DataFrame has a DatetimeIndex."""
            if isinstance(df.index, pd.DatetimeIndex):
                return df.copy()
            if "open_time" in df.columns:
                df = df.copy()
                df["open_time"] = pd.to_datetime(df["open_time"])
                return df.set_index("open_time")
            raise ValueError(f"{label} must have a DatetimeIndex or 'open_time' column.")

        self.df_15m = _to_datetime_index(df_15m, "df_15m")

        aligner = HTFDataAligner()
        aligned = aligner.align_timestamps(self.df_15m)

        self.df_15m = aligned["15m"]
        self.df_1h = _to_datetime_index(df_1h, "df_1h") if df_1h is not None else aligned["1h"]
        self.df_4h = _to_datetime_index(df_4h, "df_4h") if df_4h is not None else aligned["4h"]
        self.df_1d = _to_datetime_index(df_1d, "df_1d") if df_1d is not None else aligned["1d"]

        self._n_15m = len(self.df_15m)

        # ---- pre-cache parent-bar index arrays ------------------------------
        logger.info("HTFTradingEnv: caching parent-bar index arrays (%d 15M bars).", self._n_15m)
        self._idx_1h = np.empty(self._n_15m, dtype=np.int32)
        self._idx_4h = np.empty(self._n_15m, dtype=np.int32)
        self._idx_1d = np.empty(self._n_15m, dtype=np.int32)

        for i in range(self._n_15m):
            self._idx_1h[i] = aligner.get_parent_idx(self.df_15m, self.df_1h, i)
            self._idx_4h[i] = aligner.get_parent_idx(self.df_15m, self.df_4h, i)
            self._idx_1d[i] = aligner.get_parent_idx(self.df_15m, self.df_1d, i)

        # ---- pre-compute all features (critical for training speed) ---------
        logger.info("HTFTradingEnv: pre-computing features for all %d bars.", self._n_15m)
        self._engine = HTFFeatureEngine()
        self._precompute_all_features()

        # ---- action / observation spaces ------------------------------------
        self.action_space = spaces.Discrete(3)  # 0=Hold, 1=Long, 2=Short
        self.observation_space = spaces.Box(
            low=-np.inf,
            high=np.inf,
            shape=(N_OBS,),
            dtype=np.float32,
        )

        # ---- initialise episode state (calls reset logic) -------------------
        self.balance: float = self.initial_balance
        self.position: int = 0
        self.position_price: float = 0.0
        self.position_size_units: float = 0.0
        self.position_entry_step: int = 0
        self.current_step: int = self.lookback_window
        self.trades = []
        self.equity_curve = [self.initial_balance]
        self.max_balance: float = self.initial_balance

    # =========================================================================
    # Pre-computation
    # =========================================================================

    def _precompute_all_features(self) -> None:
        """
        Pre-compute and cache feature arrays for every 15M bar.
        Stores:
            _feat_1d   : (N, 20) float32
            _feat_4h   : (N, 25) float32
            _feat_1h   : (N, 30) float32
            _feat_15m  : (N, 35) float32
            _feat_align: (N,  4) float32
            _htf_align_at: (N,) float32  — overall_alignment scalar per bar
        """
        n = self._n_15m
        feat_1d = np.zeros((n, N_1D), dtype=np.float32)
        feat_4h = np.zeros((n, N_4H), dtype=np.float32)
        feat_1h = np.zeros((n, N_1H), dtype=np.float32)
        feat_15m = np.zeros((n, N_15M), dtype=np.float32)
        feat_align = np.zeros((n, N_ALIGN), dtype=np.float32)

        eng = self._engine

        for i in range(n):
            idx_1d = int(self._idx_1d[i])
            idx_4h = int(self._idx_4h[i])
            idx_1h = int(self._idx_1h[i])

            f1d = eng.compute_1d_features(self.df_1d, idx_1d)
            f4h = eng.compute_4h_features(self.df_4h, idx_4h)
            f1h = eng.compute_1h_features(self.df_1h, idx_1h)
            f15m = eng.compute_15m_features(self.df_15m, i)
            align = eng.compute_alignment_full(
                float(f1d[IDX_1D_SCORE]),
                float(f4h[IDX_4H_SCORE]),
                float(f1h[IDX_1H_SCORE]),
                float(f15m[IDX_15M_SCORE]),
            )

            feat_1d[i] = f1d
            feat_4h[i] = f4h
            feat_1h[i] = f1h
            feat_15m[i] = f15m
            feat_align[i] = align

        self._feat_1d = feat_1d
        self._feat_4h = feat_4h
        self._feat_1h = feat_1h
        self._feat_15m = feat_15m
        self._feat_align = feat_align
        # overall_alignment is the 4th alignment feature (index 3)
        self._htf_align_at = feat_align[:, 3].astype(np.float32)

        logger.info("HTFTradingEnv: feature pre-computation complete.")

    # =========================================================================
    # Gymnasium interface
    # =========================================================================

    def reset(
        self,
        seed: Optional[int] = None,
        options: Optional[Dict] = None,
    ) -> Tuple[np.ndarray, Dict]:
        super().reset(seed=seed)

        if self.training_mode:
            max_start = self._n_15m - self.lookback_window - 100
            low_start = self.lookback_window
            if max_start <= low_start:
                self.current_step = low_start
            else:
                self.current_step = int(
                    self.np_random.integers(low_start, max_start)
                )
        else:
            self.current_step = self.lookback_window

        self.balance = self.initial_balance
        self.position = 0
        self.position_price = 0.0
        self.position_size_units = 0.0
        self.position_entry_step = 0

        self.trades = []
        self.equity_curve = [self.initial_balance]
        self.max_balance = self.initial_balance

        return self._get_observation(), {}

    def step(
        self, action: int
    ) -> Tuple[np.ndarray, float, bool, bool, Dict]:
        """
        Execute one environment step (one 15M bar).
        Parameters
        ----------
        action : int
            0 = Hold, 1 = Long, 2 = Short.
        Returns
        -------
        obs, reward, terminated, truncated, info
        """
        current_price: float = float(self.df_15m.iloc[self.current_step]["close"])
        reward: float = 0.0
        trade_made: bool = False

        # ---- HTF alignment at this bar (pre-computed) -----------------------
        htf_align: float = float(self._htf_align_at[self.current_step])

        # Derive higher-TF directional consensus from 1D + 4H trend scores
        sig_1d: float = float(self._feat_1d[self.current_step, IDX_1D_SCORE])
        sig_4h: float = float(self._feat_4h[self.current_step, IDX_4H_SCORE])
        htf_bearish: bool = sig_1d < -0.1 and sig_4h < -0.1
        htf_bullish: bool = sig_1d > 0.1 and sig_4h > 0.1

        steps_in_position: int = self.current_step - self.position_entry_step
        min_hold_steps: int = 4  # ~1 hour minimum hold for 15M bars

        # ---- execute action -------------------------------------------------
        if action == 1:  # Long
            if self.position == -1:  # close short first
                if steps_in_position < min_hold_steps:
                    reward -= 0.015
                pnl = self._close_position(current_price)
                reward += pnl
                trade_made = True

            if self.position == 0:
                # Counter-trend penalty: going long when 1D+4H are bearish
                if htf_bearish:
                    reward -= 0.015
                # HTF alignment multiplier on trade cost
                htf_mult = 1.0 + 0.5 * max(0.0, htf_align)
                self._open_position(current_price, 1)
                self.position_entry_step = self.current_step
                reward -= 0.003 * htf_mult  # weighted trade cost signal
                trade_made = True

        elif action == 2:  # Short
            if self.position == 1:  # close long first
                if steps_in_position < min_hold_steps:
                    reward -= 0.015
                pnl = self._close_position(current_price)
                reward += pnl
                trade_made = True

            if self.position == 0:
                # Counter-trend penalty: going short when 1D+4H are bullish
                if htf_bullish:
                    reward -= 0.015
                htf_mult = 1.0 + 0.5 * max(0.0, -htf_align)
                self._open_position(current_price, -1)
                self.position_entry_step = self.current_step
                reward -= 0.003 * htf_mult
                trade_made = True

        # ---- advance step ---------------------------------------------------
        self.current_step += 1
        done: bool = self.current_step >= self._n_15m - 1

        # ---- SL / TP check at the new bar -----------------------------------
        if self.position != 0 and not done:
            new_price: float = float(self.df_15m.iloc[self.current_step]["close"])

            if self.position == 1:
                pnl_pct: float = (new_price - self.position_price) / self.position_price
            else:
                pnl_pct = (self.position_price - new_price) / self.position_price

            if pnl_pct <= -self.stop_loss_pct:
                pnl = self._close_position(new_price)
                reward += pnl
                reward -= 0.05  # strong SL penalty
                trade_made = True

            elif pnl_pct >= self.take_profit_pct:
                pnl = self._close_position(new_price)
                reward += pnl
                # Bonus scaled by HTF alignment: aligned TP hit earns more
                align_bonus = 0.10 * (1.0 + 0.5 * abs(htf_align))
                reward += align_bonus
                trade_made = True

            else:
                # Unrealized PnL credit (0.2x) — encourages holding winners
                # Position-size-invariant: pure percentage, no scaling
                reward += pnl_pct * 0.2

        # ---- idle penalty when strong setup is being ignored ----------------
        if self.position == 0 and abs(htf_align) > 0.6:
            reward -= 0.0005

        # ---- equity tracking and drawdown penalty ---------------------------
        equity: float = self._calculate_equity()
        self.equity_curve.append(equity)
        if equity > self.max_balance:
            self.max_balance = equity

        drawdown: float = (self.max_balance - equity) / (self.max_balance + 1e-10)
        if drawdown > 0.05:
            reward -= drawdown * 0.1

        # ---- build observation ----------------------------------------------
        if done:
            obs = np.zeros(self.observation_space.shape, dtype=np.float32)
        else:
            obs = self._get_observation()

        info: Dict = {
            "balance": self.balance,
            "equity": equity,
            "position": self.position,
            "htf_alignment": htf_align,
            "trade_made": trade_made,
        }

        return obs, float(reward), done, False, info

    # =========================================================================
    # Observation
    # =========================================================================

    def _get_observation(self) -> np.ndarray:
        """
        Build and return the 117-dim observation for the current step.
        Layout: [20 1D | 25 4H | 30 1H | 35 15M | 4 align | 3 pos]
        Reads directly from the pre-computed cache — no recomputation.
        """
        i = self.current_step
        feats_114 = np.concatenate([
            self._feat_1d[i],
            self._feat_4h[i],
            self._feat_1h[i],
            self._feat_15m[i],
            self._feat_align[i],
        ])  # shape (114,)

        # Position state
        current_price: float = float(self.df_15m.iloc[i]["close"])
        if self.position != 0:
            if self.position == 1:
                unrealized_pnl = (current_price - self.position_price) / (self.position_price + 1e-10)
            else:
                unrealized_pnl = (self.position_price - current_price) / (self.position_price + 1e-10)
        else:
            unrealized_pnl = 0.0

        balance_ratio = (self.balance - self.initial_balance) / (self.initial_balance + 1e-10)

        pos_state = np.array([
            float(self.position),
            float(np.clip(unrealized_pnl, -0.5, 0.5)),
            float(np.clip(balance_ratio, -0.5, 0.5)),
        ], dtype=np.float32)

        obs = np.concatenate([feats_114, pos_state])  # (117,)
        return obs.astype(np.float32)

    # =========================================================================
    # HTF alignment helper
    # =========================================================================

    def _compute_htf_alignment(self) -> float:
        """
        Return the overall HTF alignment score [-1, 1] at the current step.
        Reads from the pre-computed cache.
        """
        return float(self._htf_align_at[self.current_step])

    # =========================================================================
    # Position management
    # =========================================================================

    def _open_position(self, price: float, direction: int) -> None:
        """
        Open a long (direction=+1) or short (direction=-1) position.
        Deducts the opening taker fee from balance.
        """
        trade_amount: float = self.balance * self.position_size
        fee: float = trade_amount * self.trading_fee
        self.balance -= fee
        self.position = direction
        self.position_price = price
        self.position_size_units = (trade_amount - fee) / (price + 1e-10)

    def _close_position(self, price: float) -> float:
        """
        Close the current position and credit balance.
        Returns
        -------
        float
            Normalised PnL: raw_pnl / initial_balance.
        """
        if self.position == 0:
            return 0.0

        if self.position == 1:
            raw_pnl: float = (price - self.position_price) * self.position_size_units
        else:
            raw_pnl = (self.position_price - price) * self.position_size_units

        # Taker fee on the closing leg (applied only to proceeds when profitable)
        fee: float = abs(price * self.position_size_units) * self.trading_fee
        raw_pnl -= fee

        self.balance += raw_pnl

        # Record trade for metrics
        entry_val: float = self.position_price * self.position_size_units + 1e-10
        self.trades.append({
            "direction": "long" if self.position == 1 else "short",
            "entry": self.position_price,
            "exit": price,
            "pnl": raw_pnl,
            "pnl_pct": raw_pnl / entry_val,
            "entry_step": self.position_entry_step,
            "exit_step": self.current_step,
        })

        self.position = 0
        self.position_price = 0.0
        self.position_size_units = 0.0

        # Position-size-invariant reward: percentage return on position value
        return raw_pnl / (entry_val + 1e-10)

    # =========================================================================
    # Equity
    # =========================================================================

    def _calculate_equity(self) -> float:
        """Return balance plus unrealised PnL."""
        if self.position == 0:
            return self.balance

        current_price: float = float(self.df_15m.iloc[self.current_step]["close"])
        if self.position == 1:
            unrealised: float = (current_price - self.position_price) * self.position_size_units
        else:
            unrealised = (self.position_price - current_price) * self.position_size_units

        return self.balance + unrealised

    # =========================================================================
    # Episode metrics
    # =========================================================================

    def get_episode_metrics(self) -> Dict:
        """
        Return a dictionary of episode performance statistics.
        Metrics include total return, Sharpe, Sortino, max drawdown, win rate,
        profit factor, trade count, and final balance.  Annualisation uses
        4 × 24 × 365 = 35 040 fifteen-minute bars per year.
        """
        equity_curve = np.array(self.equity_curve, dtype=np.float64)
        _BARS_PER_YEAR = 4 * 24 * 365  # 15-minute bars

        total_return: float = (
            (equity_curve[-1] - self.initial_balance) / (self.initial_balance + 1e-10)
        )

        returns: np.ndarray = (
            np.diff(equity_curve) / (equity_curve[:-1] + 1e-10)
            if len(equity_curve) > 1
            else np.zeros(1)
        )

        # Sharpe ratio
        ret_std = float(np.std(returns))
        if ret_std > 1e-12:
            sharpe: float = float(np.mean(returns)) / ret_std * np.sqrt(_BARS_PER_YEAR)
        else:
            sharpe = 0.0

        # Sortino ratio
        downside = returns[returns < 0]
        down_std = float(np.std(downside)) if len(downside) > 0 else 0.0
        if down_std > 1e-12:
            sortino: float = float(np.mean(returns)) / down_std * np.sqrt(_BARS_PER_YEAR)
        else:
            sortino = 0.0

        # Max drawdown
        peak = np.maximum.accumulate(equity_curve)
        drawdown_series = (peak - equity_curve) / (peak + 1e-10)
        max_drawdown: float = float(np.max(drawdown_series))

        # Trade statistics
        n_trades: int = len(self.trades)
        if n_trades > 0:
            pnl_list = [t["pnl"] for t in self.trades]
            wins = sum(1 for p in pnl_list if p > 0)
            win_rate: float = wins / n_trades
            avg_trade_pnl: float = float(np.mean(pnl_list))
            gross_profit: float = sum(p for p in pnl_list if p > 0)
            gross_loss: float = abs(sum(p for p in pnl_list if p < 0))
            profit_factor: float = gross_profit / (gross_loss + 1e-10)
            avg_win: float = (
                float(np.mean([p for p in pnl_list if p > 0])) if wins > 0 else 0.0
            )
            avg_loss: float = (
                float(np.mean([p for p in pnl_list if p < 0]))
                if (n_trades - wins) > 0
                else 0.0
            )
        else:
            win_rate = 0.0
            avg_trade_pnl = 0.0
            profit_factor = 0.0
            avg_win = 0.0
            avg_loss = 0.0

        return {
            "total_return": total_return,
            "total_return_pct": round(total_return * 100.0, 2),
            "sharpe_ratio": round(sharpe, 4),
            "sortino_ratio": round(sortino, 4),
            "max_drawdown": max_drawdown,
            "max_drawdown_pct": round(max_drawdown * 100.0, 2),
            "total_trades": n_trades,
            "win_rate": win_rate,
            "profit_factor": round(profit_factor, 4),
            "avg_trade_pnl": avg_trade_pnl,
            "avg_win": avg_win,
            "avg_loss": avg_loss,
            "final_balance": float(equity_curve[-1]),
            "initial_balance": self.initial_balance,
        }

    # =========================================================================
    # Render
    # =========================================================================

    def render(self, mode: str = "human") -> None:
        equity = self._calculate_equity()
        step = self.current_step
        pos_str = {1: "LONG", -1: "SHORT", 0: "FLAT"}.get(self.position, "?")
        align = self._compute_htf_alignment()
        logger.info(
            "Step %d | %s @ %.4f | equity=%.2f | htf_align=%.3f",
            step,
            pos_str,
            self.position_price,
            equity,
            align,
        )


# =============================================================================
# Factory function
# =============================================================================

def create_htf_env(
    df_15m: pd.DataFrame,
    df_1h: Optional[pd.DataFrame] = None,
    df_4h: Optional[pd.DataFrame] = None,
    df_1d: Optional[pd.DataFrame] = None,
    config: Optional[Dict] = None,
) -> HTFTradingEnv:
    """
    Factory function for HTFTradingEnv.
    Parameters
    ----------
    df_15m : pd.DataFrame
        Required 15-minute OHLCV data (DatetimeIndex).
    df_1h, df_4h, df_1d : pd.DataFrame, optional
        Higher timeframe data; auto-resampled from df_15m when omitted.
    config : dict, optional
        Override any HTFTradingEnv constructor keyword argument.
        Recognised keys: initial_balance, position_size, stop_loss_pct,
        take_profit_pct, trading_fee, lookback_window, training_mode.
    Returns
    -------
    HTFTradingEnv
    """
    kwargs: Dict = {}
    if config is not None:
        for key in (
            "initial_balance",
            "position_size",
            "stop_loss_pct",
            "take_profit_pct",
            "trading_fee",
            "lookback_window",
            "training_mode",
        ):
            if key in config:
                kwargs[key] = config[key]

    return HTFTradingEnv(
        df_15m=df_15m,
        df_1h=df_1h,
        df_4h=df_4h,
        df_1d=df_1d,
        **kwargs,
    )