36 行
1.4 KiB
Python
36 行
1.4 KiB
Python
|
import numpy as np
|
||
|
|
def barrier_labels(high, low, close, atr, sl_mult, tp_mult, horizon, spread):
|
||
|
|
"""Faithful port of CExpertSignalAIBase::TripleBarrierLabel.
|
||
|
|
Arrays are CHRONOLOGICAL (index 0 = oldest). Stop is tested before target on each
|
||
|
|
side, so a bar spanning both barriers scores as the loss. Returns 0=Buy 1=Sell 2=Neutral,
|
||
|
|
plus a timed-out mask (neither side resolved at all)."""
|
||
|
|
n = len(close)
|
||
|
|
lab = np.full(n, 2, dtype=np.int8)
|
||
|
|
timed = np.zeros(n, dtype=bool)
|
||
|
|
valid = np.zeros(n, dtype=bool)
|
||
|
|
for j in range(n):
|
||
|
|
a = atr[j]
|
||
|
|
if not np.isfinite(a) or a <= 0: continue
|
||
|
|
last = min(j + horizon, n - 1)
|
||
|
|
if last <= j: continue
|
||
|
|
e = close[j]; risk = sl_mult*a; rew = tp_mult*a
|
||
|
|
lTp = e + spread + rew; lSl = e + spread - risk
|
||
|
|
sTp = e - rew - spread; sSl = e + risk - spread
|
||
|
|
lW=lL=sW=sL=False
|
||
|
|
hs = high[j+1:last+1]; ls = low[j+1:last+1]
|
||
|
|
for k in range(len(hs)):
|
||
|
|
hi=hs[k]; lo=ls[k]
|
||
|
|
if not (lW or lL):
|
||
|
|
if lo <= lSl: lL=True
|
||
|
|
elif hi >= lTp: lW=True
|
||
|
|
if not (sW or sL):
|
||
|
|
if hi >= sSl: sL=True
|
||
|
|
elif lo <= sTp: sW=True
|
||
|
|
if (lW or lL) and (sW or sL): break
|
||
|
|
valid[j]=True
|
||
|
|
if lW and not sW: lab[j]=0
|
||
|
|
elif sW and not lW: lab[j]=1
|
||
|
|
else:
|
||
|
|
lab[j]=2
|
||
|
|
timed[j] = not (lW or lL or sW or sL)
|
||
|
|
return lab, timed, valid
|