From bb391d502cc0f12572a680de5fd8841df9f96a35 Mon Sep 17 00:00:00 2001 From: agentson Date: Fri, 27 Feb 2026 00:45:18 +0900 Subject: [PATCH 1/2] feat: add triple barrier labeler with first-touch logic (TASK-CODE-004) --- src/analysis/triple_barrier.py | 111 +++++++++++++++++++++++++++++++++ tests/test_triple_barrier.py | 91 +++++++++++++++++++++++++++ 2 files changed, 202 insertions(+) create mode 100644 src/analysis/triple_barrier.py create mode 100644 tests/test_triple_barrier.py diff --git a/src/analysis/triple_barrier.py b/src/analysis/triple_barrier.py new file mode 100644 index 0000000..cf7a961 --- /dev/null +++ b/src/analysis/triple_barrier.py @@ -0,0 +1,111 @@ +"""Triple barrier labeler utilities. + +Implements first-touch labeling with upper/lower/time barriers. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Literal, Sequence + + +TieBreakMode = Literal["stop_first", "take_first"] + + +@dataclass(frozen=True) +class TripleBarrierSpec: + take_profit_pct: float + stop_loss_pct: float + max_holding_bars: int + tie_break: TieBreakMode = "stop_first" + + +@dataclass(frozen=True) +class TripleBarrierLabel: + label: int # +1 take-profit first, -1 stop-loss first, 0 timeout + touched: Literal["take_profit", "stop_loss", "time"] + touch_bar: int + entry_price: float + upper_barrier: float + lower_barrier: float + + +def label_with_triple_barrier( + *, + highs: Sequence[float], + lows: Sequence[float], + closes: Sequence[float], + entry_index: int, + side: int, + spec: TripleBarrierSpec, +) -> TripleBarrierLabel: + """Label one entry using triple-barrier first-touch rule. + + Args: + highs/lows/closes: OHLC components with identical length. + entry_index: Entry bar index in the sequences. + side: +1 for long, -1 for short. + spec: Barrier specification. + """ + if side not in {1, -1}: + raise ValueError("side must be +1 or -1") + if len(highs) != len(lows) or len(highs) != len(closes): + raise ValueError("highs, lows, closes lengths must match") + if entry_index < 0 or entry_index >= len(closes): + raise IndexError("entry_index out of range") + if spec.max_holding_bars <= 0: + raise ValueError("max_holding_bars must be positive") + + entry_price = float(closes[entry_index]) + if entry_price <= 0: + raise ValueError("entry price must be positive") + + if side == 1: + upper = entry_price * (1.0 + spec.take_profit_pct) + lower = entry_price * (1.0 - spec.stop_loss_pct) + else: + # For short side, favorable move is down. + upper = entry_price * (1.0 + spec.stop_loss_pct) + lower = entry_price * (1.0 - spec.take_profit_pct) + + last_index = min(len(closes) - 1, entry_index + spec.max_holding_bars) + for idx in range(entry_index + 1, last_index + 1): + h = float(highs[idx]) + l = float(lows[idx]) + + up_touch = h >= upper + down_touch = l <= lower + if not up_touch and not down_touch: + continue + + if up_touch and down_touch: + if spec.tie_break == "stop_first": + touched = "stop_loss" if side == 1 else "take_profit" + label = -1 if side == 1 else 1 + else: + touched = "take_profit" if side == 1 else "stop_loss" + label = 1 if side == 1 else -1 + elif up_touch: + touched = "take_profit" if side == 1 else "stop_loss" + label = 1 if side == 1 else -1 + else: + touched = "stop_loss" if side == 1 else "take_profit" + label = -1 if side == 1 else 1 + + return TripleBarrierLabel( + label=label, + touched=touched, + touch_bar=idx, + entry_price=entry_price, + upper_barrier=upper, + lower_barrier=lower, + ) + + return TripleBarrierLabel( + label=0, + touched="time", + touch_bar=last_index, + entry_price=entry_price, + upper_barrier=upper, + lower_barrier=lower, + ) diff --git a/tests/test_triple_barrier.py b/tests/test_triple_barrier.py new file mode 100644 index 0000000..73efc47 --- /dev/null +++ b/tests/test_triple_barrier.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +from src.analysis.triple_barrier import TripleBarrierSpec, label_with_triple_barrier + + +def test_long_take_profit_first() -> None: + highs = [100, 101, 103] + lows = [100, 99.6, 100] + closes = [100, 100, 102] + spec = TripleBarrierSpec(take_profit_pct=0.02, stop_loss_pct=0.01, max_holding_bars=3) + out = label_with_triple_barrier( + highs=highs, + lows=lows, + closes=closes, + entry_index=0, + side=1, + spec=spec, + ) + assert out.label == 1 + assert out.touched == "take_profit" + assert out.touch_bar == 2 + + +def test_long_stop_loss_first() -> None: + highs = [100, 100.5, 101] + lows = [100, 98.8, 99] + closes = [100, 99.5, 100] + spec = TripleBarrierSpec(take_profit_pct=0.02, stop_loss_pct=0.01, max_holding_bars=3) + out = label_with_triple_barrier( + highs=highs, + lows=lows, + closes=closes, + entry_index=0, + side=1, + spec=spec, + ) + assert out.label == -1 + assert out.touched == "stop_loss" + assert out.touch_bar == 1 + + +def test_time_barrier_timeout() -> None: + highs = [100, 100.8, 100.7] + lows = [100, 99.3, 99.4] + closes = [100, 100, 100] + spec = TripleBarrierSpec(take_profit_pct=0.02, stop_loss_pct=0.02, max_holding_bars=2) + out = label_with_triple_barrier( + highs=highs, + lows=lows, + closes=closes, + entry_index=0, + side=1, + spec=spec, + ) + assert out.label == 0 + assert out.touched == "time" + assert out.touch_bar == 2 + + +def test_tie_break_stop_first_default() -> None: + highs = [100, 102.1] + lows = [100, 98.9] + closes = [100, 100] + spec = TripleBarrierSpec(take_profit_pct=0.02, stop_loss_pct=0.01, max_holding_bars=1) + out = label_with_triple_barrier( + highs=highs, + lows=lows, + closes=closes, + entry_index=0, + side=1, + spec=spec, + ) + assert out.label == -1 + assert out.touched == "stop_loss" + + +def test_short_side_inverts_barrier_semantics() -> None: + highs = [100, 100.5, 101.2] + lows = [100, 97.8, 98.0] + closes = [100, 99, 99] + spec = TripleBarrierSpec(take_profit_pct=0.02, stop_loss_pct=0.01, max_holding_bars=3) + out = label_with_triple_barrier( + highs=highs, + lows=lows, + closes=closes, + entry_index=0, + side=-1, + spec=spec, + ) + assert out.label == 1 + assert out.touched == "take_profit" From 9f64c9944aee684ff1df3a598eb2af2b16916af5 Mon Sep 17 00:00:00 2001 From: agentson Date: Fri, 27 Feb 2026 00:47:09 +0900 Subject: [PATCH 2/2] fix: correct short-side tie-break semantics in triple barrier --- src/analysis/triple_barrier.py | 8 +++---- tests/test_triple_barrier.py | 40 ++++++++++++++++++++++++++++++++++ 2 files changed, 44 insertions(+), 4 deletions(-) diff --git a/src/analysis/triple_barrier.py b/src/analysis/triple_barrier.py index cf7a961..f609496 100644 --- a/src/analysis/triple_barrier.py +++ b/src/analysis/triple_barrier.py @@ -80,11 +80,11 @@ def label_with_triple_barrier( if up_touch and down_touch: if spec.tie_break == "stop_first": - touched = "stop_loss" if side == 1 else "take_profit" - label = -1 if side == 1 else 1 + touched = "stop_loss" + label = -1 else: - touched = "take_profit" if side == 1 else "stop_loss" - label = 1 if side == 1 else -1 + touched = "take_profit" + label = 1 elif up_touch: touched = "take_profit" if side == 1 else "stop_loss" label = 1 if side == 1 else -1 diff --git a/tests/test_triple_barrier.py b/tests/test_triple_barrier.py index 73efc47..1fff8e3 100644 --- a/tests/test_triple_barrier.py +++ b/tests/test_triple_barrier.py @@ -89,3 +89,43 @@ def test_short_side_inverts_barrier_semantics() -> None: ) assert out.label == 1 assert out.touched == "take_profit" + + +def test_short_tie_break_modes() -> None: + highs = [100, 101.1] + lows = [100, 97.9] + closes = [100, 100] + + stop_first = TripleBarrierSpec( + take_profit_pct=0.02, + stop_loss_pct=0.01, + max_holding_bars=1, + tie_break="stop_first", + ) + out_stop = label_with_triple_barrier( + highs=highs, + lows=lows, + closes=closes, + entry_index=0, + side=-1, + spec=stop_first, + ) + assert out_stop.label == -1 + assert out_stop.touched == "stop_loss" + + take_first = TripleBarrierSpec( + take_profit_pct=0.02, + stop_loss_pct=0.01, + max_holding_bars=1, + tie_break="take_first", + ) + out_take = label_with_triple_barrier( + highs=highs, + lows=lows, + closes=closes, + entry_index=0, + side=-1, + spec=take_first, + ) + assert out_take.label == 1 + assert out_take.touched == "take_profit"