Abdoul Ahad Binizi
239 lines
8.6 KiB
Python
239 lines
8.6 KiB
Python
"""
|
||
AHAD QUANT Forex V5 — Sequence Dataset Preparation
|
||
Converts per-pair tabular features → sliding windows (B, SEQ_LEN, 62) for DL models.
|
||
|
||
Window i = X[i : i+SEQ_LEN] → sequence input for TFT / TransformerGRU
|
||
Label i = y[i+SEQ_LEN-1] → same label as the last step in the window
|
||
Tab row i = X[i+SEQ_LEN-1] → tabular features at the end of the window
|
||
(used to align tabular models in train.py)
|
||
|
||
Returned arrays are ALL aligned on the same M samples, making it trivial to
|
||
train tabular and DL models together with consistent train/val/test splits.
|
||
|
||
Usage (called from train.py when HAS_DL=True):
|
||
X_tab, X_seq, y = prepare_seq_dataset()
|
||
# X_tab : (M, 62) — same features as prepare_dataset(), but aligned
|
||
# X_seq : (M, 168, 62) — sequence tensors
|
||
# y : (M,) — labels
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
|
||
import numpy as np
|
||
|
||
import config
|
||
from features import build_features, NUM_FEATURES
|
||
|
||
# ─── Constants ───────────────────────────────────────────────────────────────
|
||
|
||
SEQ_LEN = 168 # 7 days × 24h — window length fed into TFT / TGRU
|
||
|
||
# Memory cap: 25 pairs × MAX_SEQ_PER_PAIR × 168 × 62 × 4B ~ 1.5 GB at 6000
|
||
# Increase if you have >16 GB RAM; decrease if training crashes with OOM.
|
||
MAX_SEQ_PER_PAIR: int = int(os.environ.get("MAX_SEQ_PER_PAIR", "6000"))
|
||
LOOKAHEAD = 3 # must match LOOKAHEAD in train.py
|
||
WARMUP = 30 # feature warm-up candles (same as train.py)
|
||
MIN_CANDLES = SEQ_LEN + LOOKAHEAD + WARMUP + 10 # minimum pair length
|
||
|
||
|
||
# ─── Core builders ───────────────────────────────────────────────────────────
|
||
|
||
def _load_candles(pair: str) -> list | None:
|
||
path = os.path.join(config.DATA_DIR, f"{pair}_1h.json")
|
||
if not os.path.exists(path):
|
||
return None
|
||
with open(path) as f:
|
||
return json.load(f)
|
||
|
||
|
||
def _make_labels(close: np.ndarray, lookahead: int = LOOKAHEAD) -> np.ndarray:
|
||
labels = np.zeros(len(close))
|
||
for i in range(len(close) - lookahead):
|
||
labels[i] = 1.0 if close[i + lookahead] > close[i] else 0.0
|
||
return labels
|
||
|
||
|
||
def build_sequences(
|
||
X: np.ndarray,
|
||
y: np.ndarray,
|
||
seq_len: int = SEQ_LEN,
|
||
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
|
||
"""
|
||
Convert a single pair's aligned (X, y) arrays into sliding windows.
|
||
|
||
Parameters
|
||
----------
|
||
X : (N, 62) feature matrix — already NaN-free and warmed up
|
||
y : (N,) label vector
|
||
seq_len : window length
|
||
|
||
Returns
|
||
-------
|
||
X_seq : (M, seq_len, 62) — sequences
|
||
X_tab : (M, 62) — tabular row at end of each window (aligned)
|
||
y_seq : (M,) — labels aligned with end of each window
|
||
|
||
where M = N - seq_len - LOOKAHEAD
|
||
(last LOOKAHEAD samples excluded to avoid future-leakage in labels)
|
||
"""
|
||
N = len(X)
|
||
# Last valid window ends at index N-LOOKAHEAD-1
|
||
n_windows = N - seq_len - LOOKAHEAD
|
||
if n_windows <= 0:
|
||
raise ValueError(
|
||
f"Not enough samples ({N}) for seq_len={seq_len} + lookahead={LOOKAHEAD}. "
|
||
f"Need at least {seq_len + LOOKAHEAD + 1}."
|
||
)
|
||
|
||
X_seq = np.empty((n_windows, seq_len, X.shape[1]), dtype=np.float32)
|
||
for i in range(n_windows):
|
||
X_seq[i] = X[i : i + seq_len]
|
||
|
||
# Tabular: last row of each window
|
||
X_tab = X[seq_len - 1 : seq_len - 1 + n_windows].astype(np.float32)
|
||
|
||
# Label: at the end of the window (same index as X_tab row)
|
||
y_seq = y[seq_len - 1 : seq_len - 1 + n_windows].astype(np.float32)
|
||
|
||
return X_seq, X_tab, y_seq
|
||
|
||
|
||
# ─── Full dataset builder ────────────────────────────────────────────────────
|
||
|
||
def prepare_seq_dataset(
|
||
seq_len: int = SEQ_LEN,
|
||
verbose: bool = True,
|
||
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
|
||
"""
|
||
Build the full aligned (X_tab, X_seq, y) dataset across all pairs.
|
||
|
||
Mirrors prepare_dataset() in train.py but:
|
||
1. Uses a sliding window of `seq_len` candles
|
||
2. Returns BOTH tabular (aligned) AND sequence data
|
||
3. All three output arrays share the same index space → consistent splits
|
||
|
||
Returns
|
||
-------
|
||
X_tab : (M_total, 62) — tabular features at end of each window
|
||
X_seq : (M_total, seq_len, 62) — sequence windows
|
||
y : (M_total,) — binary labels
|
||
"""
|
||
all_X_tab: list[np.ndarray] = []
|
||
all_X_seq: list[np.ndarray] = []
|
||
all_y: list[np.ndarray] = []
|
||
|
||
# Reference pair for correlation feature (same logic as train.py)
|
||
_ref = "EURUSD"
|
||
ref_data = _load_candles(_ref)
|
||
ref_close = np.array([c["c"] for c in ref_data]) if ref_data else None
|
||
|
||
for pair in config.COINS:
|
||
data = _load_candles(pair)
|
||
if data is None or len(data) < MIN_CANDLES:
|
||
if verbose:
|
||
print(f" [{pair:8s}] skipped — insufficient data ({len(data) if data else 0} candles)")
|
||
continue
|
||
|
||
close = np.array([c["c"] for c in data])
|
||
pair_ref = (
|
||
ref_close[-len(close) :]
|
||
if ref_close is not None and len(ref_close) >= len(close)
|
||
else None
|
||
)
|
||
|
||
X_all = build_features(data, btc_closes=pair_ref)
|
||
y_all = _make_labels(close)
|
||
|
||
# Apply same warm-up crop as train.py
|
||
X_all = X_all[WARMUP : -LOOKAHEAD]
|
||
y_all = y_all[WARMUP : -LOOKAHEAD]
|
||
|
||
# Drop NaN rows (should be none after warm-up, but safety check)
|
||
valid = ~np.isnan(X_all).any(axis=1)
|
||
X_all, y_all = X_all[valid], y_all[valid]
|
||
|
||
if len(X_all) < MIN_CANDLES:
|
||
if verbose:
|
||
print(f" [{pair:8s}] skipped after cleaning ({len(X_all)} samples left)")
|
||
continue
|
||
|
||
try:
|
||
X_seq_pair, X_tab_pair, y_seq_pair = build_sequences(X_all, y_all, seq_len)
|
||
except ValueError as e:
|
||
if verbose:
|
||
print(f" [{pair:8s}] skipped — {e}")
|
||
continue
|
||
|
||
# ── Memory cap: stratified sample to MAX_SEQ_PER_PAIR ──────────
|
||
n_pair = len(y_seq_pair)
|
||
if n_pair > MAX_SEQ_PER_PAIR:
|
||
rng = np.random.default_rng(seed=42)
|
||
# Preserve class balance (stratified)
|
||
idx_pos = np.where(y_seq_pair == 1)[0]
|
||
idx_neg = np.where(y_seq_pair == 0)[0]
|
||
half = MAX_SEQ_PER_PAIR // 2
|
||
sel_pos = rng.choice(idx_pos, size=min(half, len(idx_pos)), replace=False)
|
||
sel_neg = rng.choice(idx_neg, size=min(half, len(idx_neg)), replace=False)
|
||
sel = np.sort(np.concatenate([sel_pos, sel_neg]))
|
||
X_seq_pair = X_seq_pair[sel]
|
||
X_tab_pair = X_tab_pair[sel]
|
||
y_seq_pair = y_seq_pair[sel]
|
||
if verbose:
|
||
print(f" → sampled {len(y_seq_pair):,} / {n_pair:,} sequences (RAM cap)")
|
||
|
||
all_X_tab.append(X_tab_pair)
|
||
all_X_seq.append(X_seq_pair)
|
||
all_y.append(y_seq_pair)
|
||
|
||
if verbose:
|
||
print(f" [{pair:8s}] {len(y_seq_pair):>7,} sequences")
|
||
|
||
if not all_X_tab:
|
||
raise RuntimeError(
|
||
"No pairs produced sequences. Check DATA_DIR and MIN_CANDLES."
|
||
)
|
||
|
||
X_tab = np.concatenate(all_X_tab, axis=0)
|
||
X_seq = np.concatenate(all_X_seq, axis=0)
|
||
y = np.concatenate(all_y, axis=0)
|
||
|
||
if verbose:
|
||
print(f"\n Total : {len(y):,} sequences | shape X_seq={X_seq.shape} | "
|
||
f"{y.mean():.2%} long labels")
|
||
|
||
return X_tab, X_seq, y
|
||
|
||
|
||
# ─── Sequence normalisation helpers ─────────────────────────────────────────
|
||
|
||
def fit_seq_scaler(X_tab: np.ndarray):
|
||
"""
|
||
Fit a StandardScaler on tabular training data.
|
||
The same scaler is applied to sequences by reshaping (B, T, F) → (B*T, F).
|
||
Returns a fitted sklearn StandardScaler.
|
||
"""
|
||
from sklearn.preprocessing import StandardScaler
|
||
scaler = StandardScaler()
|
||
scaler.fit(X_tab)
|
||
return scaler
|
||
|
||
|
||
def transform_sequences(scaler, X_seq: np.ndarray) -> np.ndarray:
|
||
"""
|
||
Apply a fitted StandardScaler to a 3-D sequence array.
|
||
|
||
Parameters
|
||
----------
|
||
scaler : fitted sklearn StandardScaler
|
||
X_seq : (B, T, F) float32
|
||
|
||
Returns
|
||
-------
|
||
X_norm : (B, T, F) float32
|
||
"""
|
||
B, T, F = X_seq.shape
|
||
flat = X_seq.reshape(-1, F)
|
||
normed = scaler.transform(flat).astype(np.float32)
|
||
return normed.reshape(B, T, F)
|