| """ |
| 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__) |
|
|
| |
| |
| |
| 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 |
|
|
| |
| |
| IDX_1D_SCORE = N_1D - 1 |
| IDX_4H_SCORE = N_4H - 1 |
| IDX_1H_SCORE = N_1H - 1 |
| IDX_15M_SCORE = N_15M - 1 |
|
|
| |
| |
| |
| |
| |
| |
| 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 |
|
|
| |
| |
| IDX_OVERALL_ALIGN = OFFSET_ALIGN + 3 |
|
|
|
|
| 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." |
| ) |
|
|
| |
| 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) |
|
|
| |
| 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) |
|
|
| |
| 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) |
|
|
| |
| logger.info("HTFTradingEnv: pre-computing features for all %d bars.", self._n_15m) |
| self._engine = HTFFeatureEngine() |
| self._precompute_all_features() |
|
|
| |
| self.action_space = spaces.Discrete(3) |
| self.observation_space = spaces.Box( |
| low=-np.inf, |
| high=np.inf, |
| shape=(N_OBS,), |
| dtype=np.float32, |
| ) |
|
|
| |
| 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 |
|
|
| |
| |
| |
|
|
| 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 |
| |
| self._htf_align_at = feat_align[:, 3].astype(np.float32) |
|
|
| logger.info("HTFTradingEnv: feature pre-computation complete.") |
|
|
| |
| |
| |
|
|
| 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_align: float = float(self._htf_align_at[self.current_step]) |
|
|
| |
| 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 |
|
|
| |
| if action == 1: |
| if self.position == -1: |
| 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: |
| |
| if htf_bearish: |
| 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 |
|
|
| elif action == 2: |
| if self.position == 1: |
| 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: |
| |
| 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 |
|
|
| |
| self.current_step += 1 |
| done: bool = self.current_step >= self._n_15m - 1 |
|
|
| |
| 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 |
| trade_made = True |
|
|
| elif pnl_pct >= self.take_profit_pct: |
| pnl = self._close_position(new_price) |
| reward += pnl |
| |
| align_bonus = 0.10 * (1.0 + 0.5 * abs(htf_align)) |
| reward += align_bonus |
| trade_made = True |
|
|
| else: |
| |
| |
| reward += pnl_pct * 0.2 |
|
|
| |
| if self.position == 0 and abs(htf_align) > 0.6: |
| reward -= 0.0005 |
|
|
| |
| 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 |
|
|
| |
| 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 |
|
|
| |
| |
| |
|
|
| 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], |
| ]) |
|
|
| |
| 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]) |
| return obs.astype(np.float32) |
|
|
| |
| |
| |
|
|
| 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]) |
|
|
| |
| |
| |
|
|
| 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 |
|
|
| |
| fee: float = abs(price * self.position_size_units) * self.trading_fee |
| raw_pnl -= fee |
|
|
| self.balance += raw_pnl |
|
|
| |
| 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 |
|
|
| |
| return raw_pnl / (entry_val + 1e-10) |
|
|
| |
| |
| |
|
|
| 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 |
|
|
| |
| |
| |
|
|
| 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 |
|
|
| 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) |
| ) |
|
|
| |
| 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 |
|
|
| |
| 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 |
|
|
| |
| peak = np.maximum.accumulate(equity_curve) |
| drawdown_series = (peak - equity_curve) / (peak + 1e-10) |
| max_drawdown: float = float(np.max(drawdown_series)) |
|
|
| |
| 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, |
| } |
|
|
| |
| |
| |
|
|
| 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, |
| ) |
|
|
|
|
| |
| |
| |
|
|
| 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, |
| ) |
|
|