gavindiazandClaude Code
1bcfe6e374
research/ now runs end-to-end (fetch -> features/labels -> walk-forward -> train -> OOS evaluate -> save), and backtest_strategy() replays the same signal(ctx) strategy over a historical frame with the same manage_position hook the live loop calls, simulating SL/TP, trailing, partial closes and cost per bar. --backtest wires it into the CLI. Co-Authored-By: Claude Code <noreply@anthropic.com>
176 lines
6.7 KiB
Python
176 lines
6.7 KiB
Python
"""Feature engineering and forward (triple-barrier-style) label construction.
|
|
|
|
Indicator computation prefers ``pandas_ta`` (pandas-ta-classic) when installed;
|
|
every indicator has a small stdlib/numpy fallback so the pipeline runs on the
|
|
core dependency set. Labels are built forward-looking over ``label_horizon``
|
|
bars: +1 if take-profit hits first, -1 if stop-loss hits first, 0 otherwise
|
|
(same-bar policy: sl_first).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from .config import ResearchConfig
|
|
|
|
try: # pandas-ta-classic is not a core dependency; degrade gracefully.
|
|
import pandas_ta as ta # type: ignore
|
|
|
|
_PANDAS_TA = True
|
|
except ImportError:
|
|
_PANDAS_TA = False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# indicator fallbacks (pure numpy, close-based)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _sma(values: np.ndarray, period: int) -> np.ndarray:
|
|
out = np.full(len(values), np.nan)
|
|
if len(values) < period:
|
|
return out
|
|
cs = np.cumsum(values)
|
|
out[period - 1 :] = (cs[period - 1 :] - np.concatenate([[0], cs[:-period]])) / period
|
|
return out
|
|
|
|
|
|
def _rsi(values: np.ndarray, period: int = 14) -> np.ndarray:
|
|
out = np.full(len(values), np.nan)
|
|
if len(values) <= period:
|
|
return out
|
|
delta = np.diff(values)
|
|
gain = np.where(delta > 0, delta, 0.0)
|
|
loss = np.where(delta < 0, -delta, 0.0)
|
|
avg_gain = np.full(len(delta), np.nan)
|
|
avg_loss = np.full(len(delta), np.nan)
|
|
avg_gain[period - 1] = gain[:period].mean()
|
|
avg_loss[period - 1] = loss[:period].mean()
|
|
for i in range(period, len(delta)):
|
|
avg_gain[i] = (avg_gain[i - 1] * (period - 1) + gain[i]) / period
|
|
avg_loss[i] = (avg_loss[i - 1] * (period - 1) + loss[i]) / period
|
|
with np.errstate(divide="ignore", invalid="ignore"):
|
|
rsi_series = 100 - 100 / (1 + avg_gain / avg_loss)
|
|
rsi_series[np.isnan(rsi_series)] = 100.0 # no losing bars -> 100, not NaN
|
|
out[period:] = rsi_series[period - 1 :]
|
|
return out
|
|
|
|
|
|
def _ema(values: np.ndarray, period: int) -> np.ndarray:
|
|
out = np.full(len(values), np.nan)
|
|
if len(values) < period:
|
|
return out
|
|
alpha = 2.0 / (period + 1)
|
|
out[period - 1] = values[:period].mean()
|
|
for i in range(period, len(values)):
|
|
out[i] = alpha * values[i] + (1 - alpha) * out[i - 1]
|
|
return out
|
|
|
|
|
|
def _atr(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 14) -> np.ndarray:
|
|
out = np.full(len(close), np.nan)
|
|
if len(close) <= period:
|
|
return out
|
|
prev = np.roll(close, 1)
|
|
prev[0] = close[0]
|
|
tr = np.maximum(high - low, np.maximum(np.abs(high - prev), np.abs(low - prev)))
|
|
out[period - 1] = tr[:period].mean()
|
|
for i in range(period, len(close)):
|
|
out[i] = (out[i - 1] * (period - 1) + tr[i]) / period
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# public builders
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def build_features(frame: pd.DataFrame) -> pd.DataFrame:
|
|
"""Return a DataFrame with indicator columns aligned to the input rows."""
|
|
closes = frame["close"].to_numpy(dtype=float)
|
|
highs = frame["high"].to_numpy(dtype=float)
|
|
lows = frame["low"].to_numpy(dtype=float)
|
|
result = pd.DataFrame(index=frame.index)
|
|
|
|
if _PANDAS_TA:
|
|
result["rsi"] = ta.rsi(frame["close"], length=14).to_numpy()
|
|
result["sma20"] = ta.sma(frame["close"], length=20).to_numpy()
|
|
result["sma60"] = ta.sma(frame["close"], length=60).to_numpy()
|
|
result["ema12"] = ta.ema(frame["close"], length=12).to_numpy()
|
|
result["atr14"] = ta.atr(frame["high"], frame["low"], frame["close"], length=14).to_numpy()
|
|
bb = ta.bbands(frame["close"], length=20, std=2)
|
|
if bb is not None and not bb.empty:
|
|
result["bb_upper"] = bb["BBU_20_2.0"].to_numpy()
|
|
result["bb_lower"] = bb["BBL_20_2.0"].to_numpy()
|
|
else:
|
|
result["rsi"] = _rsi(closes, 14)
|
|
result["sma20"] = _sma(closes, 20)
|
|
result["sma60"] = _sma(closes, 60)
|
|
result["ema12"] = _ema(closes, 12)
|
|
result["atr14"] = _atr(highs, lows, closes, 14)
|
|
|
|
# simple derived features that cost nothing and are always available
|
|
result["ret_1"] = np.concatenate([[np.nan], np.diff(closes)]) / np.where(closes == 0, np.nan, closes)
|
|
result["ret_5"] = closes / np.roll(closes, 5) - 1
|
|
result["ret_20"] = closes / np.roll(closes, 20) - 1
|
|
result["range_pct"] = (highs - lows) / np.where(closes == 0, np.nan, closes)
|
|
return result
|
|
|
|
|
|
def build_labels(
|
|
frame: pd.DataFrame, config: ResearchConfig, direction: str = "long"
|
|
) -> pd.Series:
|
|
"""Forward triple-barrier-style labels aligned to each bar.
|
|
|
|
Entry is the bar's close. Looking ahead ``label_horizon`` bars: for long,
|
|
+1 when high >= entry * (1 + tp), -1 when low <= entry * (1 - sl); ties
|
|
resolve sl_first. Return NaN where not enough future bars exist.
|
|
"""
|
|
closes = frame["close"].to_numpy(dtype=float)
|
|
highs = frame["high"].to_numpy(dtype=float)
|
|
lows = frame["low"].to_numpy(dtype=float)
|
|
n = len(closes)
|
|
horizon = config.label_horizon
|
|
tp = config.label_tp_pct / 100.0
|
|
sl = config.label_sl_pct / 100.0
|
|
labels = np.full(n, np.nan)
|
|
|
|
up_level = 1.0 + tp if direction == "long" else 1.0 - tp
|
|
dn_level = 1.0 - sl if direction == "long" else 1.0 + sl
|
|
|
|
for i in range(n - horizon):
|
|
entry = closes[i]
|
|
if entry <= 0:
|
|
continue
|
|
window_high = highs[i + 1 : i + horizon + 1]
|
|
window_low = lows[i + 1 : i + horizon + 1]
|
|
hit_up = (window_high >= entry * up_level).any()
|
|
hit_dn = (window_low <= entry * dn_level).any()
|
|
if hit_dn and hit_up: # same-bar policy: sl_first
|
|
labels[i] = -1.0
|
|
elif hit_up:
|
|
labels[i] = 1.0
|
|
elif hit_dn:
|
|
labels[i] = -1.0
|
|
else:
|
|
labels[i] = 0.0
|
|
return pd.Series(labels, index=frame.index)
|
|
|
|
|
|
def make_dataset(frame: pd.DataFrame, config: ResearchConfig) -> tuple[pd.DataFrame, pd.Series]:
|
|
"""Combine features + labels into a supervised (X, y) pair, dropping warmup.
|
|
|
|
The target is collapsed to binary: ``1`` when take-profit hits first
|
|
(label +1) and ``0`` for stop-loss-first or timeout. The downstream
|
|
contract is "P(+1)" — a single probability drives the long-entry signal —
|
|
so ``predict_proba_positive`` is only meaningful on a binary target.
|
|
"""
|
|
feats = build_features(frame)
|
|
labels = build_labels(frame, config)
|
|
nans = feats.isna().any(axis=1) | labels.isna()
|
|
keep = ~nans
|
|
X = feats.loc[keep].reset_index(drop=True)
|
|
y = (labels.loc[keep] > 0).astype(int).reset_index(drop=True)
|
|
return X, y
|