drl-trading-bot-dev2 / src /env /htf_env.py
DRL Trading Bot
Feature: HTF Agent integration — live trading, API endpoints, UI tab
fc115d5
Raw
History Blame Contribute Delete
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,
)